From 24b9f02be0fbe3c1a4470fe8ffb20fe6676fc820 Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 5 Jun 2026 17:51:49 +0800 Subject: [PATCH] feat(captain): queue document syncs --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 29 ++++++++-- internal/app/bootstrap.go | 1 + .../api/v1/captain_document_handler.go | 2 +- internal/service/captain_document_service.go | 21 ++++++++ .../service/captain_document_service_test.go | 53 ++++++++++++++++++- internal/service/captain_document_worker.go | 45 ++++++++++++++++ 6 files changed, 145 insertions(+), 6 deletions(-) create mode 100644 internal/service/captain_document_worker.go diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 308ac914..592a9ea6 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -17,9 +17,9 @@ Build GoChat as a Go backend that can directly reuse the frontend from `referenc ## Current Baseline - Plan freeze checkpoint: 2026-06-05 docs-only tracker landing after `13cb750 feat(captain): align task payload persistence`. -- Latest implementation checkpoint: this checkpoint, prepared as `feat(csat): queue channel templates`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(captain): queue document syncs`. - Latest documentation checkpoint before this freeze: `3263ed9 docs: land copilot task execution plan`; this document now carries the active follow-up plan directly. -- Worktree status at this implementation checkpoint: B11.1a aligns Captain assistant CRUD/tools/inbox bindings; B11.1b aligns Captain scenarios and custom tools; B11.1c aligns Captain documents, assistant responses, bulk actions, and custom-tool test payloads; B11.2 aligns Copilot thread/message create/list/get/delete payloads, account/user scoping, and no-LLM fallback persistence; B11.3a aligns Captain preferences show/update payloads and account-level model/feature storage; B11.3b aligns Captain playground request/response payloads, account scoping, v2 history handling, and no-LLM fallback; B11.3c adds the fakeable Captain document sync backend gate with disabled, failed, and fake-success states; B11.3d aligns Captain task request/response payloads, no-provider disabled states, follow-up context, suggestion persistence, and Copilot message tool-call key validation; B11.3e aligns Captain stream DTOs/disabled SSE fallbacks and Copilot push-event payload shapes; B12.1 adds the reusable GoChat server/seed entrypoint plus a Meilisearch-first reused Chatwoot frontend smoke harness and report; B12.2a adds API smoke assertions for auth/profile, inbox, conversation/messages, contact/company, widget config/message, and public CSAT; B12.2b adds a zero-dependency Chrome DevTools browser smoke that loads the reused Chatwoot login and dashboard entrypoints through Vite and checks browser auth/dashboard API requests; B12.3a adds enterprise API smoke assertions for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot; B12.3b adds reused-frontend enterprise browser route navigation for SLA, CSAT, automation, macros, audit logs, custom roles, capacity, Captain, and Copilot request coverage; P5.1 adds the PostgreSQL-backed durable `background_jobs` model/migration plus WorkerPool enqueue, schedule, retry/backoff, dead-letter, idempotency, stale-lock recovery, and focused tests; P5.2 wires `channel.Dispatcher` and `dispatch.EventDispatcher` async paths into durable event jobs with worker replay tests; P5.3 queues Meilisearch write-side index/delete jobs for conversations, messages, contacts, companies, and articles while keeping search reads Meilisearch-first; P5.4 queues automation webhook and email transcript side effects as durable jobs while preserving fakeable delivery boundaries; P5.5 queues Chatwoot-style macro execute fan-out through durable `automation:macro_execution` jobs; P5.6 queues resolve-triggered CSAT survey sends and WhatsApp/Twilio CSAT template creation through durable jobs; P5.7 queues Chatwoot enterprise SLA account scans and applied-SLA evaluation jobs through the durable worker; P5.8 queues Chatwoot-style contact export artifact generation through durable `contact:export` jobs; P5.12 queues scheduled item fan-out, one-off campaigns, snoozed conversation reopening, and account auto-resolution through durable jobs. Next active implementation slice is Captain/Copilot jobs, provider webhook/outbound job fan-out, update-message-status/bulk-action jobs, and analytics aggregation. +- Worktree status at this implementation checkpoint: B11.1a aligns Captain assistant CRUD/tools/inbox bindings; B11.1b aligns Captain scenarios and custom tools; B11.1c aligns Captain documents, assistant responses, bulk actions, and custom-tool test payloads; B11.2 aligns Copilot thread/message create/list/get/delete payloads, account/user scoping, and no-LLM fallback persistence; B11.3a aligns Captain preferences show/update payloads and account-level model/feature storage; B11.3b aligns Captain playground request/response payloads, account scoping, v2 history handling, and no-LLM fallback; B11.3c adds the fakeable Captain document sync backend gate with disabled, failed, and fake-success states; B11.3d aligns Captain task request/response payloads, no-provider disabled states, follow-up context, suggestion persistence, and Copilot message tool-call key validation; B11.3e aligns Captain stream DTOs/disabled SSE fallbacks and Copilot push-event payload shapes; B12.1 adds the reusable GoChat server/seed entrypoint plus a Meilisearch-first reused Chatwoot frontend smoke harness and report; B12.2a adds API smoke assertions for auth/profile, inbox, conversation/messages, contact/company, widget config/message, and public CSAT; B12.2b adds a zero-dependency Chrome DevTools browser smoke that loads the reused Chatwoot login and dashboard entrypoints through Vite and checks browser auth/dashboard API requests; B12.3a adds enterprise API smoke assertions for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot; B12.3b adds reused-frontend enterprise browser route navigation for SLA, CSAT, automation, macros, audit logs, custom roles, capacity, Captain, and Copilot request coverage; P5.1 adds the PostgreSQL-backed durable `background_jobs` model/migration plus WorkerPool enqueue, schedule, retry/backoff, dead-letter, idempotency, stale-lock recovery, and focused tests; P5.2 wires `channel.Dispatcher` and `dispatch.EventDispatcher` async paths into durable event jobs with worker replay tests; P5.3 queues Meilisearch write-side index/delete jobs for conversations, messages, contacts, companies, and articles while keeping search reads Meilisearch-first; P5.4 queues automation webhook and email transcript side effects as durable jobs while preserving fakeable delivery boundaries; P5.5 queues Chatwoot-style macro execute fan-out through durable `automation:macro_execution` jobs; P5.6 queues resolve-triggered CSAT survey sends and WhatsApp/Twilio CSAT template creation through durable jobs; P5.7 queues Chatwoot enterprise SLA account scans and applied-SLA evaluation jobs through the durable worker; P5.8 queues Chatwoot-style contact export artifact generation through durable `contact:export` jobs; P5.11 queues Captain document sync requests through durable `captain:document_sync` jobs; P5.12 queues scheduled item fan-out, one-off campaigns, snoozed conversation reopening, and account auto-resolution through durable jobs. Next active implementation slice is remaining Captain/Copilot jobs, provider webhook/outbound job fan-out, update-message-status/bulk-action jobs, and analytics aggregation. - `go test ./...` passes. - Route dump succeeds with `TOTAL: 830` after adding the Chatwoot-compatible applied-SLA index route. - Route parity artifacts now exist under `docs/parity/` and are generated by `cmd/route_parity`. @@ -78,7 +78,7 @@ Open work after the current checkpoint: | Phase 2 | Route and controller parity audit | Doing | Ruby/Bundler unavailable, so Chatwoot route extraction currently uses static `routes.rb` fallback | | Phase 3 | Data and serializer parity | Doing | JSON fixture coverage is partial and still endpoint-family based | | Phase 4 | Enterprise feature completion | Doing | B7, B8, B9, B10, and B11 are in Review; B12 reused frontend smoke is the next broad verification gate | -| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, and conversation maintenance cores are in Review; Captain/provider/outbound job integration remains open | +| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, Captain document sync, and conversation maintenance cores are in Review; remaining Captain/Copilot/provider/outbound job integration remains open | | Phase 6 | Core placeholder burn-down | Doing | account/contact/conversation/message/inbox placeholder groups remain broad | | Phase 7 | Verification harness | Review | B12.1 boot/readiness, B12.2a API assertions, B12.2b browser smoke harness, B12.3a enterprise API assertions, and B12.3b enterprise browser route navigation exist; optional live Meilisearch/full-browser runs remain environment-dependent | @@ -101,6 +101,7 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | +| `feat(captain): queue document syncs` | Advances P5.11 with a durable Chatwoot `Captain::Documents::PerformSyncJob` equivalent. `POST /captain/documents/:id/sync` still returns `202` after marking the account-scoped document as syncing, but now enqueues `captain:document_sync` on the `low` queue when a WorkerPool is configured. Worker replay calls the existing fakeable `CaptainDocumentSyncBackend`, persists synced/failed/disabled states through `SyncDocumentByAccount`, and lets missing documents or malformed payloads retry through `background_jobs`. | `go test ./internal/service -run 'CaptainDocumentService' -count=1`; `go test ./internal/handler/api/v1 -run 'CaptainDocument\|CaptainResource' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | Move P5.11 document sync core to Review; continue Captain crawl/response-building/embedding/Copilot response jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. | | `feat(csat): queue channel templates` | Advances P5.6 with durable WhatsApp/Twilio CSAT template creation. `POST /inboxes/:inbox_id/csat_template` accepts Chatwoot nested `template` params, persists the singular template as pending, and enqueues `csat:template_create` when a WorkerPool is configured. Worker replay calls a fakeable provider boundary, records WhatsApp template metadata or Twilio content/approval SIDs into `inbox.csat_config`, updates template status, and retries provider failures through `background_jobs`. | `go test ./internal/service -run 'CsatTemplate' -count=1`; `go test ./internal/handler/api/v1 -run 'InboxCsatTemplate' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.6 section. | Move P5.6 channel templates to Review; continue Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. | | `feat(conversations): queue maintenance jobs` | Advances P5.12 and the scheduled-items portion of P5.5. `scheduled:trigger_items` now mirrors Chatwoot `TriggerScheduledItemsJob` by queueing due one-off campaigns, snoozed conversation reopening, account auto-resolution scheduling, and the next hourly root trigger. One-off campaigns claim active rows as `processing` before execution and complete after artifact creation, so duplicate worker replays do not duplicate campaign messages. | `go test ./internal/service -run 'ConversationMaintenanceJobs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | Move P5.12 conversation maintenance core to Review; continue update-message-status/bulk-action job depth plus CSAT channel templates, Captain/Copilot, provider webhook/outbound delivery, and analytics aggregation. | | `feat(crm): queue contact exports` | Advances P5.8 with a durable Chatwoot `Account::ContactsExportJob` equivalent. Contact export requests now create a pending `contact_exports` artifact and enqueue `contact:export` on the `low` queue when a WorkerPool is configured; worker replay generates the CSV artifact, marks completion, creates the persisted notification, and invokes the fakeable completion mailer. No-worker construction keeps the synchronous fallback for focused tests. | `go test ./internal/service -run 'ContactService_ExportContacts\|ContactExportJob' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.8 section. | Move P5.8 to Review; continue delayed automation scheduled items, CSAT channel templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. | @@ -1621,7 +1622,7 @@ Tracking table: | P5.8 | Queue contact export artifact generation and completion notification/email. | `account/contacts_export_job.rb` | contact export service and mailer boundary | Export API returns immediately, artifact generation is durable, completion notification/email is fakeable, and download remains stable after restart. | Review by `feat(crm): queue contact exports` | | P5.9 | Queue inbound provider webhook processing where Chatwoot defers work. | `webhooks/*_events_job.rb`, `webhooks/*_delivery_job.rb` | webhook handlers, provider services | Provider HTTP ack behavior remains Chatwoot-compatible while persistence/dispatch runs through retryable jobs with signature validation preserved. | Todo | | P5.10 | Queue outbound message delivery and delivery-status updates. | `send_reply_job.rb`, provider delivery/status jobs | message send/channel services, delivery status handler | Outgoing message creation and provider delivery are separated; retries update message/delivery status exactly once. | Todo | -| P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot jobs | `internal/service/captain_document_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses and Copilot message persistence survive worker restart. | Todo | +| P5.11 | Queue Captain document sync, crawl, response building, embeddings, and Copilot responses. | Captain document/crawl/response/embedding/Copilot jobs | `internal/service/captain_document_service.go`, Captain/Copilot services | Existing fakeable disabled/failure gates run under durable jobs; document statuses and Copilot message persistence survive worker restart. | Doing: Captain document sync Review by `feat(captain): queue document syncs`; crawl/response/embedding/Copilot response jobs Todo | | P5.12 | Queue conversation maintenance jobs. | `trigger_scheduled_items_job.rb`, `campaigns/trigger_oneoff_campaign_job.rb`, `conversations/resolution_job.rb`, `reopen_snoozed_conversations_job.rb`, `update_message_status_job.rb`, `bulk_actions_job.rb` | conversation service/handlers | Auto-resolution, snooze reopen, status updates, and bulk actions are scheduled/retryable with idempotent tests. | Doing: scheduled items, one-off campaigns, snooze reopen, and auto-resolution Review by `feat(conversations): queue maintenance jobs`; update-message-status and bulk-action jobs Todo | | P5.13 | Replace placeholder analytics/report builders that need background aggregation. | reporting jobs/services and report controllers | `internal/service/analytics_service.go`, reporting services | Frontend-visible reports no longer use placeholder values; any expensive aggregation is scheduled or cached with freshness rules. | Todo | @@ -1747,6 +1748,25 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache git diff --check ``` +P5.11 current checkpoint: + +- `captain:document_sync` now mirrors Chatwoot enterprise `Captain::Documents::PerformSyncJob` on the `low` queue. `POST /captain/documents/:id/sync` marks the account-scoped document as syncing and enqueues durable replay when a WorkerPool is configured. +- Worker replay unmarshals the persisted account/document IDs and calls `SyncDocumentByAccount`, reusing the B11.3c fakeable sync backend boundary for disabled provider config, backend failures, explicit backend error codes, successful content/title/fingerprint updates, and account scoping. +- Missing documents or malformed payloads fail the job so retry/backoff/dead-letter state stays observable in `background_jobs`, matching the Chatwoot background-job failure surface instead of silently swallowing bad work. +- No-worker construction keeps the previous mark-syncing fallback for focused tests and local paths that do not start the durable worker. +- Bootstrap registers Captain document jobs on the shared WorkerPool so production sync requests are replayable after process restart. +- Remaining P5.11 work: Captain crawl jobs, response-building jobs, embedding update fan-out, and Copilot response jobs. + +P5.11 verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'CaptainDocumentService' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/handler/api/v1 -run 'CaptainDocument\|CaptainResource' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1 +env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./... +git diff --check +``` + P5.12 current checkpoint: - `scheduled:trigger_items` now mirrors Chatwoot `TriggerScheduledItemsJob` on the `scheduled_jobs` queue. Bootstrap registers the handlers and seeds the first root trigger; the root trigger self-schedules the next hourly run with an idempotency bucket. @@ -1964,3 +1984,4 @@ Verification milestone gates: - 2026-06-05: P5.8 durable contact export checkpoint prepared as `feat(crm): queue contact exports`; contact export requests now enqueue `contact:export` jobs on the low queue when a WorkerPool is configured, while worker replay generates the persisted CSV artifact, completion notification, and fakeable completion email. Focused service/API/worker/app tests passed. Next slice is delayed automation scheduled items, WhatsApp/Twilio CSAT templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. - 2026-06-05: P5.12 durable conversation maintenance checkpoint prepared as `feat(conversations): queue maintenance jobs`; the Chatwoot scheduled-items chain now runs through `scheduled:trigger_items`, due one-off campaigns queue `campaign:trigger_oneoff`, snoozed conversations reopen through `conversation:reopen_snoozed`, and account auto-resolution fans out through `account:conversations_resolution_scheduler` and `conversation:resolution`. Focused service/worker/app tests passed. Next slice is CSAT WhatsApp/Twilio templates, Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. - 2026-06-05: P5.6 durable CSAT template checkpoint prepared as `feat(csat): queue channel templates`; inbox CSAT template creation now accepts Chatwoot nested template payloads, queues `csat:template_create` jobs when a WorkerPool is configured, and worker replay records WhatsApp template IDs or Twilio content/approval SIDs through a fakeable provider boundary. Focused service/API/worker/app tests passed. Next slice is Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. +- 2026-06-05: P5.11 durable Captain document sync checkpoint prepared as `feat(captain): queue document syncs`; Captain document sync requests now mark the document syncing and enqueue `captain:document_sync` jobs on the `low` queue, while worker replay uses the existing fakeable sync backend to persist disabled, failed, or synced document states. Focused service/API/worker/app tests passed. Next slice is remaining Captain crawl/response/embedding/Copilot response jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 05dfc6d3..69ceb425 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -574,6 +574,7 @@ func Bootstrap(env string) (*App, error) { // Captain services (P10 M10 — Captain AI + Copilot) captainAssistantService := service.NewCaptainAssistantService(captainAssistantRepo, captainInboxRepo, captainDocumentRepo, captainAssistantResponseRepo, llmProvider) captainDocumentService := service.NewCaptainDocumentService(captainDocumentRepo, llmProvider, captainAssistantRepo) + captainDocumentService.SetWorkerPool(workerPool) captainScenarioService := service.NewCaptainScenarioService(captainScenarioRepo, captainAssistantRepo) captainCustomToolService := service.NewCaptainCustomToolService(captainCustomToolRepo) copilotService := service.NewCopilotService(copilotThreadRepo, copilotMessageRepo, copilotSuggestionRepo, llmProvider, captainAssistantRepo) diff --git a/internal/handler/api/v1/captain_document_handler.go b/internal/handler/api/v1/captain_document_handler.go index e521bbd6..a7901c50 100644 --- a/internal/handler/api/v1/captain_document_handler.go +++ b/internal/handler/api/v1/captain_document_handler.go @@ -194,7 +194,7 @@ func (h *CaptainDocumentHandler) SyncDocument(c *gin.Context) { return } - if _, err := h.svc.MarkSyncing(c.Request.Context(), accountID, id); err != nil { + if _, err := h.svc.RequestSyncDocumentByAccount(c.Request.Context(), accountID, id); err != nil { applogger.L().Errorf("SyncDocument: %v", err) response.AbortWithStatusError(c, captainAssistantErrorStatus(err), response.ErrInternal, "failed to sync document") return diff --git a/internal/service/captain_document_service.go b/internal/service/captain_document_service.go index d6d31777..852cf753 100644 --- a/internal/service/captain_document_service.go +++ b/internal/service/captain_document_service.go @@ -12,6 +12,7 @@ import ( "github.com/gochat/gochat/internal/llm" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" ) @@ -22,6 +23,7 @@ type CaptainDocumentService struct { assistantRepo *repository.CaptainAssistantRepo llmProvider llm.Provider syncBackend CaptainDocumentSyncBackend + worker *worker.WorkerPool } type CaptainDocumentSyncBackend interface { @@ -54,6 +56,11 @@ func (s *CaptainDocumentService) SetSyncBackend(syncBackend CaptainDocumentSyncB s.syncBackend = syncBackend } +func (s *CaptainDocumentService) SetWorkerPool(wp *worker.WorkerPool) { + s.worker = wp + RegisterCaptainDocumentJobs(wp, s) +} + // --- Request DTOs --- // CreateDocumentRequest is the DTO for creating a document. @@ -216,6 +223,20 @@ func (s *CaptainDocumentService) MarkSyncing(ctx context.Context, accountID, id return s.documentRepo.GetByAccountAndID(ctx, accountID, id) } +func (s *CaptainDocumentService) RequestSyncDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) { + doc, err := s.MarkSyncing(ctx, accountID, id) + if err != nil { + return nil, err + } + if s.worker == nil { + return doc, nil + } + if _, err := s.worker.Enqueue(ctx, TaskTypeCaptainDocumentSync, captainDocumentSyncJob{AccountID: accountID, DocumentID: id}, worker.WithQueue("low"), worker.WithMaxAttempts(3)); err != nil { + return nil, err + } + return doc, nil +} + func (s *CaptainDocumentService) SyncDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) { doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id) if err != nil { diff --git a/internal/service/captain_document_service_test.go b/internal/service/captain_document_service_test.go index 52157acf..19b8707c 100644 --- a/internal/service/captain_document_service_test.go +++ b/internal/service/captain_document_service_test.go @@ -5,9 +5,11 @@ import ( "errors" "fmt" "testing" + "time" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" + "github.com/gochat/gochat/internal/worker" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gorm.io/driver/sqlite" @@ -19,7 +21,7 @@ func setupCaptainDocumentServiceTest(t *testing.T) (*gorm.DB, *CaptainDocumentSe dbName := fmt.Sprintf("file:%s?mode=memory&cache=private", t.Name()) db, err := gorm.Open(sqlite.Open(dbName), &gorm.Config{}) require.NoError(t, err) - require.NoError(t, db.AutoMigrate(&model.Account{}, &model.CaptainAssistant{}, &model.CaptainDocument{})) + require.NoError(t, db.AutoMigrate(&model.Account{}, &model.CaptainAssistant{}, &model.CaptainDocument{}, &model.BackgroundJob{})) t.Cleanup(func() { sqlDB, _ := db.DB() sqlDB.Close() @@ -104,6 +106,55 @@ func TestCaptainDocumentService_SyncDocumentByAccountScopesDocument(t *testing.T require.Error(t, err) } +func TestCaptainDocumentService_RequestSyncQueuesDurableJob(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + backend := &captainDocumentFakeSyncBackend{result: &CaptainDocumentSyncResult{Title: "Durable Help", Content: "fresh durable content"}} + svc.SetSyncBackend(backend) + now := time.Date(2026, 6, 5, 23, 0, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + svc.SetWorkerPool(wp) + + queued, err := svc.RequestSyncDocumentByAccount(context.Background(), account.ID, doc.ID) + require.NoError(t, err) + assert.Equal(t, model.DocumentSyncStatusPending, queued.SyncStatus) + assert.Equal(t, uint(0), backend.documentID) + + var jobCount int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ? AND status = ?", TaskTypeCaptainDocumentSync, "low", model.BackgroundJobStatusQueued).Count(&jobCount).Error) + assert.Equal(t, int64(1), jobCount) + + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + assert.Equal(t, doc.ID, backend.documentID) + + var synced model.CaptainDocument + require.NoError(t, db.First(&synced, doc.ID).Error) + assert.Equal(t, "Durable Help", synced.Name) + assert.Equal(t, "fresh durable content", synced.Content) + assert.Equal(t, model.DocumentSyncStatusSynced, synced.SyncStatus) + assert.Empty(t, synced.LastSyncErrorCode) +} + +func TestCaptainDocumentService_DocumentSyncJobRetriesMissingDocument(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + now := time.Date(2026, 6, 5, 23, 15, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now }), worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + svc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeCaptainDocumentSync, captainDocumentSyncJob{AccountID: 999, DocumentID: 9999}, worker.WithQueue("low"), worker.WithMaxAttempts(3)) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.Error(t, err) + assert.True(t, processed) + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainDocumentSync).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + assert.NotEmpty(t, job.LastError) +} + type captainDocumentFakeSyncBackend struct { result *CaptainDocumentSyncResult err error diff --git a/internal/service/captain_document_worker.go b/internal/service/captain_document_worker.go new file mode 100644 index 00000000..4c27bead --- /dev/null +++ b/internal/service/captain_document_worker.go @@ -0,0 +1,45 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" +) + +const TaskTypeCaptainDocumentSync = "captain:document_sync" + +type captainDocumentSyncJob struct { + AccountID uint `json:"account_id"` + DocumentID uint `json:"document_id"` +} + +var captainDocumentRegistrations sync.Map + +// RegisterCaptainDocumentJobs wires Captain::Documents::PerformSyncJob into +// the durable worker. The sync backend remains fakeable for tests and disabled +// deployments. +func RegisterCaptainDocumentJobs(wp *worker.WorkerPool, svc *CaptainDocumentService) { + if wp == nil || svc == nil { + return + } + if _, loaded := captainDocumentRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + wp.Register(TaskTypeCaptainDocumentSync, svc.performDocumentSyncJob) +} + +func (s *CaptainDocumentService) performDocumentSyncJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainDocumentSyncJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain document sync job: %w", err) + } + if payload.AccountID == 0 || payload.DocumentID == 0 { + return fmt.Errorf("invalid captain document sync job payload: %#v", payload) + } + _, err := s.SyncDocumentByAccount(ctx, payload.AccountID, payload.DocumentID) + return err +}