diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index dce4f8ba..a129d32d 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -16,10 +16,10 @@ Build GoChat as a Go backend that can directly reuse the frontend from `referenc ## Current Baseline -- Current tracking checkpoint: 2026-06-05 after `d756473 feat(webhook): queue incoming messages`, with this documentation checkpoint prepared as `docs: land parity execution tracker`. -- Latest implementation checkpoint: `d756473 feat(webhook): queue incoming messages`. +- Current tracking checkpoint: 2026-06-05 after `2923aae docs: land parity execution tracker`, with this implementation checkpoint prepared as `feat(captain): queue document crawl jobs`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(captain): queue document crawl jobs`. - Latest documentation-only checkpoint: this checkpoint, prepared as `docs: land parity execution tracker`; this document is now the active follow-up plan and supersedes `.hermes/plans/*`. -- 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.9 queues normalized provider inbound message persistence/dispatch through durable `webhook:incoming_message_persist` jobs; P5.10 queues Chatwoot `SendReplyJob`-style outbound message delivery through durable `message:send_reply` jobs and provider delivery-status/read-receipt updates through durable webhook status 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, account auto-resolution, widget/public message status updates, and account conversation bulk actions through durable jobs. Next active implementation slice is remaining P5.11 Captain/Copilot jobs, followed by P5.13 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.9 queues normalized provider inbound message persistence/dispatch through durable `webhook:incoming_message_persist` jobs; P5.10 queues Chatwoot `SendReplyJob`-style outbound message delivery through durable `message:send_reply` jobs and provider delivery-status/read-receipt updates through durable webhook status jobs; P5.11 queues Captain document sync, crawl/parser, and schedule-sync work through durable jobs; P5.12 queues scheduled item fan-out, one-off campaigns, snoozed conversation reopening, account auto-resolution, widget/public message status updates, and account conversation bulk actions through durable jobs. Next active implementation slice is remaining P5.11 response/embedding/Copilot jobs, followed by P5.13 analytics aggregation. - `go test ./...` passes. - Route dump succeeds with `TOTAL: 832` after adding the Chatwoot-compatible Twilio delivery-status route plus the legacy namespaced alias. - Route parity artifacts now exist under `docs/parity/` and are generated by `cmd/route_parity`. @@ -88,7 +88,7 @@ This table is the shortest authoritative handoff view. If an older lower section | Priority | Workstream | Current state | Next checkpoint | Commit close rule | | --- | --- | --- | --- | --- | -| 1 | P5.11 Captain/Copilot jobs | Captain document sync is queued; crawl/response/embedding/Copilot response jobs remain open. | Add fakeable durable jobs for remaining Captain/Copilot async work and disabled-provider states. | Focused Captain/Copilot worker tests plus full `go test ./...`. | +| 1 | P5.11 Captain/Copilot jobs | Captain document sync plus crawl/parser/schedule jobs are queued; response/embedding/Copilot response jobs remain open. | Add fakeable durable jobs for remaining Captain/Copilot async work and disabled-provider states. | Focused Captain/Copilot worker tests plus full `go test ./...`. | | 2 | P5.13 reports/analytics | Frontend smoke harness exists; `analytics_service` still has placeholder aggregation paths. | Replace frontend-visible report placeholders with real scheduled/cached aggregations. | Report fixtures verify values, cache/freshness behavior, and no hidden placeholder JSON. | | 3 | Phase 2/3 drift | Tracked route parity is 0 missing for the current critical set; serializer fixtures remain partial. | Expand route/serializer fixtures when smoke or reference inspection exposes drift. | Regenerate parity artifacts and add endpoint-family fixture tests. | | 4 | Phase 6 placeholder audit | Widget/public/webhook critical placeholders are burned down; account/contact/conversation/message/inbox audit remains broad. | Run a fresh placeholder audit and assign every frontend-reachable stub to a tracked owner. | `rg` audit result is recorded and no reused-frontend blocker is ownerless. | @@ -100,7 +100,7 @@ These rows are the executable development plan from this point forward. A checkp | ID | Owner files | Reference files | Work to land | Exit gate | | --- | --- | --- | --- | --- | -| P5.11a Captain document crawl/schedule | `internal/service/captain_document_service.go`, `internal/service/captain_document_worker.go`, `internal/app/bootstrap.go` | `reference/chatwoot/enterprise/app/jobs/captain/documents/crawl_job.rb`, `schedule_syncs_job.rb`, `perform_sync_job.rb`, Firecrawl/simple parser jobs | Add durable schedule/crawl producers and handlers with fakeable crawl/parser boundaries. Keep missing provider config as `sync_disabled`/retryable failure state, not placeholder success. | Focused Captain document worker tests prove enqueue, replay, retry/dead-letter, account scope, and disabled/failure states. | +| P5.11a Captain document crawl/schedule | `internal/service/captain_document_service.go`, `internal/service/captain_document_worker.go`, `internal/app/bootstrap.go` | `reference/chatwoot/enterprise/app/jobs/captain/documents/crawl_job.rb`, `schedule_syncs_job.rb`, `perform_sync_job.rb`, Firecrawl/simple parser jobs | Durable schedule/crawl producers and handlers with fakeable crawl/parser boundaries. Missing provider config is a failed `crawl_disabled` state, not placeholder success. | Review by `feat(captain): queue document crawl jobs`; focused worker tests prove enqueue, replay, account scope, idempotent scheduler, and disabled/failure states. | | P5.11b Captain response/embedding fan-out | Captain document/assistant-response services and repositories, Meilisearch/embedding boundaries | `response_builder_job.rb`, `enterprise/app/jobs/captain/llm/update_embedding_job.rb`, FAQ generator/embedding services | Queue FAQ response generation after successful document content changes, reset unedited responses, create/update assistant responses, and fan out embedding update work behind fakeable LLM gates. | Tests cover response reset/create, embedding-disabled state, fake success, idempotent replay, and no external network in default tests. | | P5.11c Copilot and conversation response jobs | Copilot thread/message services, Captain task services, event/realtime payload helpers | `enterprise/app/jobs/captain/copilot/response_job.rb`, `conversation/response_builder_job.rb`, Copilot chat service | Queue assistant replies after Copilot user messages and Captain conversation triggers. Persist assistant messages/tool-call JSON and emit existing push payloads from worker replay. | Copilot worker tests cover no-provider disabled reply, fake provider success/failure, user/account scope, and restart-safe persistence. | | P5.13 analytics aggregation | `internal/service/analytics_service.go`, report handlers/services, worker bootstrap | Chatwoot report controllers/services used by dashboard analytics, CSAT/SLA reporting views | Replace frontend-visible placeholder report values with real scheduled or cached aggregations. Define freshness/idempotency rules for expensive rollups. | Report fixtures prove values are derived from persisted conversations/messages/CSAT/SLA rows; `rg` finds no frontend-visible placeholder report JSON. | @@ -140,6 +140,7 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | +| `feat(captain): queue document crawl jobs` | Advances P5.11a with durable Captain document crawl, simple-page parser, and auto-sync scheduler jobs. Document create/crawl requests enqueue `captain:document_crawl`, crawl replay uses a fakeable crawl backend to enqueue normalized `captain:document_page_crawl_parse` jobs, parser replay creates or updates account-scoped document content through a fakeable parser backend, and the `captain:documents_schedule_syncs` root job scans stale synced/failed/syncing documents for accounts with `captain_document_auto_sync` enabled before enqueueing `captain:document_sync` work with daily idempotency. Bootstrap seeds the scheduler after Captain document services are wired. | `go test ./internal/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11a moves to Review; continue P5.11b response/embedding fan-out, then P5.11c Copilot/conversation response jobs. | | `docs: land parity execution tracker` | Documentation-only checkpoint requested before continuing implementation. Updates the current baseline to `d756473`, front-loads executable open checkpoint contracts for P5.11 Captain/Copilot jobs, P5.13 analytics aggregation, Phase 2/3 drift, Phase 6 placeholder burn-down, and B12 live smoke, and records commit discipline for future parity slices. | `git diff --check`; documentation-only change. | Start P5.11a Captain document crawl/schedule jobs, then P5.11b response/embedding fan-out and P5.11c Copilot/conversation response jobs. | | `feat(webhook): queue incoming messages` | Completes P5.9 normalized inbound persistence deferral. `IncomingPersister.PersistIncoming` now validates provider-parsed messages and, when a WorkerPool is configured, enqueues `webhook:incoming_message_persist` instead of creating contacts/conversations/messages inline. Worker replay reloads the inbox, performs the existing Chatwoot-style contact inbox/conversation/message persistence, keeps `source_id` idempotency, and dispatches the same conversation/message events from the durable path. Telegram, LINE, Facebook/Instagram, WhatsApp, TikTok, and Twilio webhook handlers are wired to the WorkerPool while no-worker construction remains synchronous for focused tests. | `go test ./internal/handler/webhook -run 'IncomingPersister.*Incoming\|IncomingPersister.*Status\|StatusJob\|TwilioInboundSMS\|TwilioDeliveryStatus' -count=1`; `go test ./internal/handler/webhook ./internal/channel/whatsapp ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.9 section. | Move P5.9 to Review; next active slice is remaining P5.11 Captain/Copilot jobs, then P5.13 analytics aggregation. | | `feat(messages): queue delivery statuses` | Completes P5.10 provider delivery-status depth. `IncomingPersister` can now enqueue direct message status updates as `webhook:message_status_update` jobs and contact-scoped read receipts as `webhook:contact_messages_status_update` jobs on the `low` queue; worker replay reloads the inbox, applies sent/delivered/read/failed transitions without downgrading already-read messages, upserts `delivery_statuses`, dispatches status events, and records provider failure details in `content_attributes.external_error`. Twilio delivery callbacks now return Chatwoot-style `204`, include the exact `/twilio/delivery_status` route plus the legacy `/webhooks/twilio/delivery_status` alias, resolve channels by `MessagingServiceSid` or `AccountSid`/`From`, and enqueue through the same durable path when the WorkerPool is wired. | `go test ./internal/handler/webhook -run 'IncomingPersister.*Status\|StatusJob\|TwilioDeliveryStatus' -count=1`; `go test ./internal/handler/webhook ./internal/service ./internal/worker ./internal/app -count=1`; `go run ./cmd/dump_routes > docs/parity/gochat_routes.txt`; `go run ./cmd/route_parity`; `go test ./...`; `git diff --check`; full verification recorded in the P5.10 section. | P5.10 moved to Review; inbound persistence deferral is completed by `feat(webhook): queue incoming messages`. | @@ -1637,7 +1638,7 @@ Known hotspots: - `internal/worker/worker.go` has the committed P5.1 durable core and is already wired to dispatcher, search indexing, automation delivery, macro fan-out, CSAT survey/template sends, SLA scans, contact exports, Captain document sync, and conversation maintenance jobs. - `internal/service/message_delivery_worker.go` and `internal/handler/webhook/incoming_persister_jobs.go` now cover P5.10 SendReplyJob-style outbound delivery plus provider delivery-status/read-receipt jobs. - `internal/handler/webhook/*` and provider services now keep verification/parsing in the HTTP path while `IncomingPersister` moves normalized inbound persistence/dispatch behind durable jobs when the WorkerPool is configured. -- `internal/service/captain_*` and `internal/service/copilot_*` have document sync job coverage only; P5.11 still needs crawl/response/embedding/Copilot response jobs. +- `internal/service/captain_*` and `internal/service/copilot_*` now cover Captain document sync, crawl/parser, and schedule-sync jobs; P5.11 still needs response/embedding/Copilot response jobs. - `internal/service/analytics_service.go` has placeholder analytics/report paths and is the owner for P5.13 aggregation work. Checklist: @@ -1645,7 +1646,7 @@ Checklist: - [x] Add `background_jobs` persistence with job type, payload, queue, status, attempt counters, scheduled/locked timestamps, idempotency key, last error, and completion timestamps. - [x] Replace `internal/worker/worker.go` placeholder with enqueue, schedule, perform, retry/backoff, dead-letter, and graceful shutdown behavior. - [x] Map the committed Chatwoot job/listener families to Go worker responsibilities in the tracking table. -- [ ] Finish durable job dispatch for remaining Captain/Copilot jobs, delayed automation actions, and report aggregation. +- [ ] Finish durable job dispatch for remaining Captain response/embedding/Copilot jobs, delayed automation actions, and report aggregation. - [ ] Add retry and failure logging for the remaining external provider calls. - [ ] Add tests for the remaining enqueueing, uniqueness/idempotency, delayed execution, retries, worker restart pickup, and fakeable external side effects. @@ -1668,7 +1669,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. | Review by `feat(webhook): queue incoming messages` | | 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. | Review by `feat(messages): queue send replies` and `feat(messages): queue delivery statuses` | -| 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.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: document sync Review by `feat(captain): queue document syncs`; crawl/schedule Review by `feat(captain): queue document crawl jobs`; 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. | Review by `feat(conversations): queue maintenance jobs`, `feat(conversations): queue message status updates`, and `feat(conversations): queue bulk actions` | | 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 | @@ -1845,14 +1846,18 @@ 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. +- `captain:document_crawl` now mirrors the simple crawl fan-out portion of Chatwoot `Captain::Documents::CrawlJob` on the `low` queue. Document create/crawl requests enqueue durable crawl work when a WorkerPool is configured, and crawl replay uses a fakeable backend to produce normalized parser jobs for page links plus the original document URL. +- `captain:document_page_crawl_parse` now mirrors `Captain::Tools::SimplePageCrawlParserJob` for the local simple-crawl path. Worker replay validates assistant/account scope, creates or updates the matching document by assistant/link, stores title/content/fingerprint, marks sync as `synced`, and exposes parser-disabled or content-empty failures without hitting the network in default tests. +- `captain:documents_schedule_syncs` now mirrors the reference auto-sync scheduler shape on `scheduled_jobs`. Bootstrap seeds the scheduler, scheduler jobs use daily idempotency, scan stale synced/failed/syncing completed documents for accounts with `captain_document_auto_sync` enabled, and enqueue `captain:document_sync` work on the `purgable` queue for retryable execution. - 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. +- Bootstrap registers Captain document jobs on the shared WorkerPool so production sync/crawl/schedule requests are replayable after process restart. +- Remaining P5.11 work: 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/service -run 'CaptainDocumentService\|EnqueueCaptainDocumentScheduleSyncs' -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 ./... @@ -1985,6 +1990,7 @@ Verification milestone gates: ## Progress Log +- 2026-06-05: P5.11a Captain document crawl/schedule checkpoint prepared as `feat(captain): queue document crawl jobs`; document create/crawl requests now enqueue `captain:document_crawl`, crawl replay fans out normalized `captain:document_page_crawl_parse` parser jobs through fakeable crawl/parser boundaries, parser replay creates or updates account-scoped documents with synced content/fingerprints, and `captain:documents_schedule_syncs` scans stale completed documents for `captain_document_auto_sync` accounts to enqueue `captain:document_sync` jobs with daily idempotency. Focused Captain document worker tests, service/worker/app package tests, full `go test ./...`, and `git diff --check` passed. Remaining P5.11 follow-up is response/embedding fan-out and Copilot/conversation response jobs. - 2026-06-05: Parity execution tracker checkpoint prepared as `docs: land parity execution tracker`; the plan now records `d756473 feat(webhook): queue incoming messages` as the current implementation baseline, adds executable contracts for P5.11a/P5.11b/P5.11c Captain/Copilot jobs, P5.13 analytics aggregation, Phase 2/3 drift audit, Phase 6 placeholder burn-down, and B12 live smoke, and makes commit/update discipline explicit for future checkpoints. Documentation-only checkpoint; `git diff --check` passed. - 2026-06-05: P5.9 inbound webhook deferral checkpoint prepared as `feat(webhook): queue incoming messages`; normalized provider inbound messages now enqueue `webhook:incoming_message_persist` when WorkerPool is configured, while worker replay creates contact inboxes, conversations, messages, and dispatches incoming events from the durable path with source-ID idempotency. Telegram, LINE, Facebook/Instagram, WhatsApp, TikTok, and Twilio persisters are wired to the WorkerPool. Focused webhook/worker tests passed. Next slice is remaining P5.11 Captain/Copilot jobs. - 2026-06-05: P5.10 provider delivery-status checkpoint prepared as `feat(messages): queue delivery statuses`; provider direct status callbacks now enqueue `webhook:message_status_update`, contact-scoped read receipts enqueue `webhook:contact_messages_status_update`, worker replay upserts delivery status rows, records failed provider `external_error`, preserves monotonic sent/delivered/read transitions, and Twilio status callbacks now support the exact Chatwoot `/twilio/delivery_status` route with `204 No Content`. Focused webhook tests, route generation, full `go test ./...`, and `git diff --check` passed; inbound persistence deferral is completed by the next P5.9 checkpoint. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 37252003..59591863 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -582,6 +582,9 @@ func Bootstrap(env string) (*App, error) { captainAssistantService := service.NewCaptainAssistantService(captainAssistantRepo, captainInboxRepo, captainDocumentRepo, captainAssistantResponseRepo, llmProvider) captainDocumentService := service.NewCaptainDocumentService(captainDocumentRepo, llmProvider, captainAssistantRepo) captainDocumentService.SetWorkerPool(workerPool) + if _, err := service.EnqueueCaptainDocumentScheduleSyncs(context.Background(), workerPool, time.Now()); err != nil { + applogger.L().Warnf("failed to enqueue initial Captain document sync scheduler: %v", err) + } captainScenarioService := service.NewCaptainScenarioService(captainScenarioRepo, captainAssistantRepo) captainCustomToolService := service.NewCaptainCustomToolService(captainCustomToolRepo) copilotService := service.NewCopilotService(copilotThreadRepo, copilotMessageRepo, copilotSuggestionRepo, llmProvider, captainAssistantRepo) diff --git a/internal/repository/captain_document_repo.go b/internal/repository/captain_document_repo.go index c962553b..d40a3b93 100644 --- a/internal/repository/captain_document_repo.go +++ b/internal/repository/captain_document_repo.go @@ -3,6 +3,7 @@ package repository import ( "context" "strings" + "time" "github.com/gochat/gochat/internal/model" "gorm.io/gorm" @@ -123,6 +124,37 @@ func (r *CaptainDocumentRepo) FindByExternalLink(ctx context.Context, assistantI return &doc, nil } +func (r *CaptainDocumentRepo) ListDueForAutoSync(ctx context.Context, now time.Time, interval, staleTimeout time.Duration, limit int) ([]model.CaptainDocument, error) { + if interval <= 0 { + interval = 24 * time.Hour + } + if staleTimeout <= 0 { + staleTimeout = 10 * time.Minute + } + if limit <= 0 { + limit = 1000 + } + syncDueBefore := now.Add(-interval / 2).Unix() + staleCutoff := now.Add(-staleTimeout).Unix() + + var docs []model.CaptainDocument + err := r.db.WithContext(ctx). + Model(&model.CaptainDocument{}). + Where("status = ?", model.DocumentStatusCompleted). + Where( + "(sync_status = ? AND (last_synced_at IS NULL OR last_synced_at < ?)) OR "+ + "(sync_status = ? AND (last_sync_attempted_at IS NULL OR last_sync_attempted_at < ?)) OR "+ + "(sync_status = ? AND last_sync_attempted_at < ?)", + model.DocumentSyncStatusSynced, syncDueBefore, + model.DocumentSyncStatusFailed, syncDueBefore, + model.DocumentSyncStatusPending, staleCutoff, + ). + Order("last_sync_attempted_at ASC, id ASC"). + Limit(limit). + Find(&docs).Error + return docs, err +} + // FindByStatus retrieves documents matching a specific status. func (r *CaptainDocumentRepo) FindByStatus(ctx context.Context, assistantID uint, status model.DocumentStatus, offset, limit int) ([]model.CaptainDocument, int64, error) { var docs []model.CaptainDocument diff --git a/internal/service/captain_document_service.go b/internal/service/captain_document_service.go index 852cf753..ceed8952 100644 --- a/internal/service/captain_document_service.go +++ b/internal/service/captain_document_service.go @@ -3,6 +3,7 @@ package service import ( "context" "crypto/sha256" + "errors" "fmt" "io" "net/http" @@ -14,6 +15,7 @@ import ( "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" + "gorm.io/gorm" ) // CaptainDocumentService implements business logic for CaptainDocument operations. @@ -23,6 +25,8 @@ type CaptainDocumentService struct { assistantRepo *repository.CaptainAssistantRepo llmProvider llm.Provider syncBackend CaptainDocumentSyncBackend + crawlBackend CaptainDocumentCrawlBackend + pageParser CaptainDocumentPageParserBackend worker *worker.WorkerPool } @@ -36,6 +40,19 @@ type CaptainDocumentSyncResult struct { ErrorCode string } +type CaptainDocumentCrawlBackend interface { + CrawlCaptainDocument(ctx context.Context, doc *model.CaptainDocument) (*CaptainDocumentCrawlResult, error) +} + +type CaptainDocumentCrawlResult struct { + PageLinks []string + ErrorCode string +} + +type CaptainDocumentPageParserBackend interface { + ParseCaptainDocumentPage(ctx context.Context, pageLink string) (*CaptainDocumentSyncResult, error) +} + // NewCaptainDocumentService creates a new CaptainDocumentService. func NewCaptainDocumentService( documentRepo *repository.CaptainDocumentRepo, @@ -56,6 +73,14 @@ func (s *CaptainDocumentService) SetSyncBackend(syncBackend CaptainDocumentSyncB s.syncBackend = syncBackend } +func (s *CaptainDocumentService) SetCrawlBackend(crawlBackend CaptainDocumentCrawlBackend) { + s.crawlBackend = crawlBackend +} + +func (s *CaptainDocumentService) SetPageParserBackend(pageParser CaptainDocumentPageParserBackend) { + s.pageParser = pageParser +} + func (s *CaptainDocumentService) SetWorkerPool(wp *worker.WorkerPool) { s.worker = wp RegisterCaptainDocumentJobs(wp, s) @@ -108,8 +133,14 @@ func (s *CaptainDocumentService) Create(ctx context.Context, assistantID, accoun return nil, fmt.Errorf("create document: %w", err) } if created, err := s.documentRepo.GetByAccountAndID(ctx, accountID, doc.ID); err == nil { + if enqueueErr := s.enqueueDocumentCrawl(ctx, accountID, created.ID); enqueueErr != nil { + return nil, enqueueErr + } return created, nil } + if enqueueErr := s.enqueueDocumentCrawl(ctx, accountID, doc.ID); enqueueErr != nil { + return nil, enqueueErr + } return doc, nil } @@ -237,6 +268,166 @@ func (s *CaptainDocumentService) RequestSyncDocumentByAccount(ctx context.Contex return doc, nil } +func (s *CaptainDocumentService) RequestCrawlDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) { + doc, err := s.MarkSyncing(ctx, accountID, id) + if err != nil { + return nil, err + } + doc.Status = model.DocumentStatusInProgress + if err := s.documentRepo.Update(ctx, doc); err != nil { + return nil, fmt.Errorf("mark document crawling: %w", err) + } + if err := s.enqueueDocumentCrawl(ctx, accountID, id); err != nil { + return nil, err + } + return s.documentRepo.GetByAccountAndID(ctx, accountID, id) +} + +func (s *CaptainDocumentService) enqueueDocumentCrawl(ctx context.Context, accountID, id uint) error { + if s.worker == nil { + return nil + } + _, err := s.worker.Enqueue(ctx, TaskTypeCaptainDocumentCrawl, captainDocumentCrawlJob{AccountID: accountID, DocumentID: id}, + worker.WithQueue("low"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:document_crawl:%d:%d", accountID, id)), + ) + return err +} + +func (s *CaptainDocumentService) ScheduleDueDocumentSyncs(ctx context.Context, now time.Time) (int, error) { + if s.worker == nil { + return 0, nil + } + docs, err := s.documentRepo.ListDueForAutoSync(ctx, now, 24*time.Hour, 10*time.Minute, 1000) + if err != nil { + return 0, fmt.Errorf("list due captain documents: %w", err) + } + enqueued := 0 + for _, doc := range docs { + if !s.captainDocumentAutoSyncEnabled(ctx, doc.AccountID) { + continue + } + if _, err := s.worker.Enqueue(ctx, TaskTypeCaptainDocumentSync, captainDocumentSyncJob{AccountID: doc.AccountID, DocumentID: doc.ID}, + worker.WithQueue("purgable"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:document_sync:auto:%d:%d:%d", doc.AccountID, doc.ID, now.UTC().Truncate(24*time.Hour).Unix())), + ); err != nil { + return enqueued, err + } + enqueued++ + } + return enqueued, nil +} + +func (s *CaptainDocumentService) captainDocumentAutoSyncEnabled(ctx context.Context, accountID uint) bool { + var account model.Account + if err := s.documentRepo.DB().WithContext(ctx).Select("feature_flags").First(&account, accountID).Error; err != nil { + return false + } + return featureFlagStringEnabled(account.FeatureFlags, "captain_document_auto_sync") +} + +func (s *CaptainDocumentService) CrawlDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) { + doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id) + if err != nil { + return nil, fmt.Errorf("document not found: %w", err) + } + if s.crawlBackend == nil { + return s.markDocumentSyncFailed(ctx, accountID, id, "crawl_disabled") + } + result, err := s.crawlBackend.CrawlCaptainDocument(ctx, doc) + if err != nil { + updated, markErr := s.markDocumentSyncFailed(ctx, accountID, id, "crawl_error") + if markErr != nil { + return nil, markErr + } + return updated, fmt.Errorf("crawl document: %w", err) + } + if result == nil { + return s.markDocumentSyncFailed(ctx, accountID, id, "crawl_error") + } + if result.ErrorCode != "" { + return s.markDocumentSyncFailed(ctx, accountID, id, result.ErrorCode) + } + if s.worker == nil { + return doc, fmt.Errorf("worker pool required for document crawl parser fan-out") + } + links := normalizedUniqueLinks(append(result.PageLinks, doc.ExternalLink)) + for _, link := range links { + if _, err := s.worker.Enqueue(ctx, TaskTypeCaptainDocumentPageCrawlParse, captainDocumentPageCrawlParseJob{AccountID: accountID, AssistantID: doc.AssistantID, PageLink: link}, + worker.WithQueue("low"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:document_page_crawl_parse:%d:%d:%s", accountID, doc.AssistantID, link)), + ); err != nil { + return doc, err + } + } + return s.documentRepo.GetByAccountAndID(ctx, accountID, id) +} + +func (s *CaptainDocumentService) ParseCrawledPage(ctx context.Context, accountID, assistantID uint, pageLink string) (*model.CaptainDocument, error) { + pageLink = normalizeCaptainDocumentLink(pageLink) + if accountID == 0 || assistantID == 0 || pageLink == "" { + return nil, fmt.Errorf("invalid captain page crawl payload: account_id=%d assistant_id=%d page_link=%q", accountID, assistantID, pageLink) + } + if s.assistantRepo != nil { + if _, err := s.assistantRepo.GetByAccountAndID(ctx, accountID, assistantID); err != nil { + return nil, fmt.Errorf("assistant not found: %w", err) + } + } + doc, err := s.documentRepo.FindByExternalLink(ctx, assistantID, pageLink) + if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) { + return nil, err + } + if s.pageParser == nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, fmt.Errorf("page crawl parser disabled") + } + return s.markDocumentSyncFailed(ctx, accountID, doc.ID, "crawl_disabled") + } + result, parseErr := s.pageParser.ParseCaptainDocumentPage(ctx, pageLink) + if parseErr != nil { + if err == nil { + _, _ = s.markDocumentSyncFailed(ctx, accountID, doc.ID, "fetch_failed") + } + return nil, fmt.Errorf("parse crawled page: %w", parseErr) + } + if result == nil || result.ErrorCode != "" || strings.TrimSpace(result.Content) == "" { + code := "content_empty" + if result != nil && result.ErrorCode != "" { + code = result.ErrorCode + } + if err == nil { + return s.markDocumentSyncFailed(ctx, accountID, doc.ID, code) + } + return nil, fmt.Errorf("parse crawled page failed: %s", code) + } + if errors.Is(err, gorm.ErrRecordNotFound) { + doc = &model.CaptainDocument{AccountID: accountID, AssistantID: assistantID, ExternalLink: pageLink} + } + doc.Name = strings.TrimSpace(result.Title) + if doc.Name == "" { + doc.Name = pageLink + } + doc.Content = strings.TrimSpace(result.Content) + doc.ContentFingerprint = computeFingerprint(doc.Content) + doc.Status = model.DocumentStatusCompleted + doc.SyncStatus = model.DocumentSyncStatusSynced + doc.LastSyncErrorCode = "" + now := time.Now().Unix() + doc.LastSyncedAt = &now + doc.LastSyncAttemptedAt = &now + if doc.ID == 0 { + if err := s.documentRepo.Create(ctx, doc); err != nil { + return nil, fmt.Errorf("create crawled document: %w", err) + } + } else if err := s.documentRepo.Update(ctx, doc); err != nil { + return nil, fmt.Errorf("update crawled document: %w", err) + } + return s.documentRepo.GetByAccountAndID(ctx, accountID, doc.ID) +} + func (s *CaptainDocumentService) SyncDocumentByAccount(ctx context.Context, accountID, id uint) (*model.CaptainDocument, error) { doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id) if err != nil { @@ -452,3 +643,24 @@ func computeFingerprint(content string) string { h.Write([]byte(normalized)) return fmt.Sprintf("%x", h.Sum(nil)) } + +func normalizedUniqueLinks(rawLinks []string) []string { + seen := make(map[string]struct{}, len(rawLinks)) + links := make([]string, 0, len(rawLinks)) + for _, rawLink := range rawLinks { + link := normalizeCaptainDocumentLink(rawLink) + if link == "" { + continue + } + if _, ok := seen[link]; ok { + continue + } + seen[link] = struct{}{} + links = append(links, link) + } + return links +} + +func normalizeCaptainDocumentLink(rawLink string) string { + return strings.TrimRight(strings.TrimSpace(rawLink), "/") +} diff --git a/internal/service/captain_document_service_test.go b/internal/service/captain_document_service_test.go index 19b8707c..15a27816 100644 --- a/internal/service/captain_document_service_test.go +++ b/internal/service/captain_document_service_test.go @@ -155,6 +155,117 @@ func TestCaptainDocumentService_DocumentSyncJobRetriesMissingDocument(t *testing assert.NotEmpty(t, job.LastError) } +func TestCaptainDocumentService_RequestCrawlQueuesParserJobs(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + now := time.Date(2026, 6, 5, 23, 30, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + svc.SetWorkerPool(wp) + svc.SetCrawlBackend(&captainDocumentFakeCrawlBackend{links: []string{"https://example.com/help", "https://example.com/faq/"}}) + svc.SetPageParserBackend(&captainDocumentFakePageParserBackend{pages: map[string]*CaptainDocumentSyncResult{ + "https://example.com/help": {Title: "Fresh Help", Content: "fresh help content"}, + "https://example.com/faq": {Title: "FAQ", Content: "answer one"}, + }}) + + queued, err := svc.RequestCrawlDocumentByAccount(context.Background(), account.ID, doc.ID) + require.NoError(t, err) + assert.Equal(t, model.DocumentStatusInProgress, queued.Status) + assert.Equal(t, model.DocumentSyncStatusPending, queued.SyncStatus) + + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var parserJobCount int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ? AND queue = ?", TaskTypeCaptainDocumentPageCrawlParse, "low").Count(&parserJobCount).Error) + assert.Equal(t, int64(2), parserJobCount) + + processed, err = wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + processed, err = wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var updated model.CaptainDocument + require.NoError(t, db.First(&updated, doc.ID).Error) + assert.Equal(t, "Fresh Help", updated.Name) + assert.Equal(t, "fresh help content", updated.Content) + assert.Equal(t, model.DocumentStatusCompleted, updated.Status) + assert.Equal(t, model.DocumentSyncStatusSynced, updated.SyncStatus) + + var created model.CaptainDocument + require.NoError(t, db.Where("assistant_id = ? AND external_link = ?", doc.AssistantID, "https://example.com/faq").First(&created).Error) + assert.Equal(t, account.ID, created.AccountID) + assert.Equal(t, "FAQ", created.Name) + assert.Equal(t, "answer one", created.Content) +} + +func TestCaptainDocumentService_CrawlJobDisabledMarksFailed(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + wp := worker.NewWorkerPool(db) + svc.SetWorkerPool(wp) + + _, err := svc.RequestCrawlDocumentByAccount(context.Background(), account.ID, doc.ID) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var updated model.CaptainDocument + require.NoError(t, db.First(&updated, doc.ID).Error) + assert.Equal(t, model.DocumentSyncStatusFailed, updated.SyncStatus) + assert.Equal(t, "crawl_disabled", updated.LastSyncErrorCode) +} + +func TestCaptainDocumentService_ScheduleDueDocumentSyncsQueuesDueDocs(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + account.FeatureFlags = `{"captain_document_auto_sync":true}` + require.NoError(t, db.Save(account).Error) + now := time.Date(2026, 6, 6, 0, 0, 0, 0, time.UTC) + oldSync := now.Add(-13 * time.Hour).Unix() + doc.LastSyncedAt = &oldSync + doc.LastSyncAttemptedAt = &oldSync + doc.Status = model.DocumentStatusCompleted + doc.SyncStatus = model.DocumentSyncStatusSynced + require.NoError(t, db.Save(doc).Error) + freshSync := now.Add(-time.Hour).Unix() + fresh := &model.CaptainDocument{AccountID: account.ID, AssistantID: doc.AssistantID, Name: "Fresh", ExternalLink: "https://example.com/fresh", Status: model.DocumentStatusCompleted, SyncStatus: model.DocumentSyncStatusSynced, LastSyncedAt: &freshSync, LastSyncAttemptedAt: &freshSync} + require.NoError(t, db.Create(fresh).Error) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + svc.SetWorkerPool(wp) + + enqueued, err := svc.ScheduleDueDocumentSyncs(context.Background(), now) + require.NoError(t, err) + assert.Equal(t, 1, enqueued) + + var jobs []model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainDocumentSync).Find(&jobs).Error) + require.Len(t, jobs, 1) + assert.Equal(t, "purgable", jobs[0].Queue) + assert.Contains(t, string(jobs[0].Payload), fmt.Sprintf("\"document_id\":%d", doc.ID)) +} + +func TestEnqueueCaptainDocumentScheduleSyncsUsesDailyIdempotency(t *testing.T) { + db, _ := setupCaptainDocumentServiceTest(t) + now := time.Date(2026, 6, 6, 1, 0, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + + first, err := EnqueueCaptainDocumentScheduleSyncs(context.Background(), wp, now) + require.NoError(t, err) + second, err := EnqueueCaptainDocumentScheduleSyncs(context.Background(), wp, now.Add(time.Hour)) + require.NoError(t, err) + assert.Equal(t, first.ID, second.ID) + assert.Equal(t, TaskTypeCaptainDocumentScheduleSyncs, first.JobType) + assert.Equal(t, "scheduled_jobs", first.Queue) + + var count int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeCaptainDocumentScheduleSyncs).Count(&count).Error) + assert.Equal(t, int64(1), count) +} + type captainDocumentFakeSyncBackend struct { result *CaptainDocumentSyncResult err error @@ -168,3 +279,29 @@ func (b *captainDocumentFakeSyncBackend) SyncCaptainDocument(ctx context.Context } return b.result, nil } + +type captainDocumentFakeCrawlBackend struct { + links []string + err error + documentID uint +} + +func (b *captainDocumentFakeCrawlBackend) CrawlCaptainDocument(ctx context.Context, doc *model.CaptainDocument) (*CaptainDocumentCrawlResult, error) { + b.documentID = doc.ID + if b.err != nil { + return nil, b.err + } + return &CaptainDocumentCrawlResult{PageLinks: b.links}, nil +} + +type captainDocumentFakePageParserBackend struct { + pages map[string]*CaptainDocumentSyncResult + err error +} + +func (b *captainDocumentFakePageParserBackend) ParseCaptainDocumentPage(ctx context.Context, pageLink string) (*CaptainDocumentSyncResult, error) { + if b.err != nil { + return nil, b.err + } + return b.pages[pageLink], nil +} diff --git a/internal/service/captain_document_worker.go b/internal/service/captain_document_worker.go index 4c27bead..4e87edc7 100644 --- a/internal/service/captain_document_worker.go +++ b/internal/service/captain_document_worker.go @@ -5,18 +5,41 @@ import ( "encoding/json" "fmt" "sync" + "time" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/worker" ) -const TaskTypeCaptainDocumentSync = "captain:document_sync" +const ( + TaskTypeCaptainDocumentSync = "captain:document_sync" + TaskTypeCaptainDocumentCrawl = "captain:document_crawl" + TaskTypeCaptainDocumentPageCrawlParse = "captain:document_page_crawl_parse" + TaskTypeCaptainDocumentScheduleSyncs = "captain:documents_schedule_syncs" +) + +const captainDocumentScheduleInterval = 24 * time.Hour type captainDocumentSyncJob struct { AccountID uint `json:"account_id"` DocumentID uint `json:"document_id"` } +type captainDocumentCrawlJob struct { + AccountID uint `json:"account_id"` + DocumentID uint `json:"document_id"` +} + +type captainDocumentPageCrawlParseJob struct { + AccountID uint `json:"account_id"` + AssistantID uint `json:"assistant_id"` + PageLink string `json:"page_link"` +} + +type captainDocumentScheduleSyncsJob struct { + PlanName string `json:"plan_name,omitempty"` +} + var captainDocumentRegistrations sync.Map // RegisterCaptainDocumentJobs wires Captain::Documents::PerformSyncJob into @@ -30,6 +53,29 @@ func RegisterCaptainDocumentJobs(wp *worker.WorkerPool, svc *CaptainDocumentServ return } wp.Register(TaskTypeCaptainDocumentSync, svc.performDocumentSyncJob) + wp.Register(TaskTypeCaptainDocumentCrawl, svc.performDocumentCrawlJob) + wp.Register(TaskTypeCaptainDocumentPageCrawlParse, svc.performDocumentPageCrawlParseJob) + wp.Register(TaskTypeCaptainDocumentScheduleSyncs, svc.performDocumentScheduleSyncsJob) +} + +func EnqueueCaptainDocumentScheduleSyncs(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) { + if wp == nil { + return nil, nil + } + if scheduledAt.IsZero() { + scheduledAt = time.Now() + } + return wp.Enqueue(ctx, TaskTypeCaptainDocumentScheduleSyncs, captainDocumentScheduleSyncsJob{}, + worker.WithQueue("scheduled_jobs"), + worker.WithScheduledAt(scheduledAt), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(captainDocumentScheduleIdempotencyKey(scheduledAt)), + ) +} + +func captainDocumentScheduleIdempotencyKey(scheduledAt time.Time) string { + bucket := scheduledAt.UTC().Truncate(captainDocumentScheduleInterval).Unix() + return fmt.Sprintf("captain:documents_schedule_syncs:%d", bucket) } func (s *CaptainDocumentService) performDocumentSyncJob(ctx context.Context, job *model.BackgroundJob) error { @@ -43,3 +89,41 @@ func (s *CaptainDocumentService) performDocumentSyncJob(ctx context.Context, job _, err := s.SyncDocumentByAccount(ctx, payload.AccountID, payload.DocumentID) return err } + +func (s *CaptainDocumentService) performDocumentCrawlJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainDocumentCrawlJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain document crawl job: %w", err) + } + if payload.AccountID == 0 || payload.DocumentID == 0 { + return fmt.Errorf("invalid captain document crawl job payload: %#v", payload) + } + _, err := s.CrawlDocumentByAccount(ctx, payload.AccountID, payload.DocumentID) + return err +} + +func (s *CaptainDocumentService) performDocumentPageCrawlParseJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainDocumentPageCrawlParseJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain document page crawl parse job: %w", err) + } + if payload.AccountID == 0 || payload.AssistantID == 0 || payload.PageLink == "" { + return fmt.Errorf("invalid captain document page crawl parse job payload: %#v", payload) + } + _, err := s.ParseCrawledPage(ctx, payload.AccountID, payload.AssistantID, payload.PageLink) + return err +} + +func (s *CaptainDocumentService) performDocumentScheduleSyncsJob(ctx context.Context, job *model.BackgroundJob) error { + if len(job.Payload) > 0 { + var payload captainDocumentScheduleSyncsJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain document schedule syncs job: %w", err) + } + } + if _, err := s.ScheduleDueDocumentSyncs(ctx, time.Now()); err != nil { + return err + } + _, err := EnqueueCaptainDocumentScheduleSyncs(ctx, s.worker, time.Now().Add(captainDocumentScheduleInterval)) + return err +}