From dd6d0191073a2a13e7cd31930739433fdce060ac Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 5 Jun 2026 20:08:15 +0800 Subject: [PATCH] feat(captain): queue response embedding jobs --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 40 ++-- internal/app/bootstrap.go | 1 + internal/service/captain_document_service.go | 161 ++++++++++++++ .../service/captain_document_service_test.go | 199 +++++++++++++++++- internal/service/captain_document_worker.go | 47 ++++- 5 files changed, 424 insertions(+), 24 deletions(-) diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index a129d32d..84441ed5 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 `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, 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. +- Current tracking checkpoint: 2026-06-05 after `89cdb2a feat(captain): queue document crawl jobs`, with this implementation checkpoint prepared as `feat(captain): queue response embedding jobs`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(captain): queue response embedding jobs`. +- Latest documentation-only checkpoint: `2923aae 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, crawl/parser, schedule-sync, response-builder, and embedding-update 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 Copilot/conversation response 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`. @@ -45,7 +45,7 @@ Next ordered checkpoints: | Order | Slice | Required outcome | Primary verification | | --- | --- | --- | --- | -| 1 | P5.11 remaining Captain/Copilot jobs | Queue Captain crawl/response/embedding and Copilot response work behind fakeable provider gates. | Captain/Copilot worker tests and disabled/failure-state fixtures. | +| 1 | P5.11 remaining Captain/Copilot jobs | Queue remaining Copilot/conversation response work behind fakeable provider gates. | Copilot worker tests and disabled/failure-state fixtures. | | 2 | P5.13 analytics aggregation | Replace frontend-visible placeholder report values with scheduled or cached real aggregations. | Report service/handler fixtures and freshness/idempotency tests. | | 3 | Phase 2/3 audit pass | Route/controller/serializer drift found by B12 or new reference inspection is captured as named slices, not free-form TODOs. | Regenerated route parity artifacts and fixture-backed serializer tests. | | 4 | Phase 6 placeholder burn-down | Remaining account/contact/conversation/message/inbox placeholder handlers are either real Chatwoot-compatible flows or explicitly tracked as unsupported reference gaps. | `rg` placeholder audit, route smoke, and endpoint-family tests. | @@ -54,7 +54,7 @@ Next ordered checkpoints: This checkpoint is intended to make the development plan complete enough to track without reading Hermes notes first. -- The next active implementation slice is Phase 5 remaining job parity, starting with remaining P5.11 Captain/Copilot jobs. B12 has repeatable API/browser/enterprise smoke harnesses in Review; optional live failures should be converted into named slices instead of blocking Phase 5 job work. +- The next active implementation slice is Phase 5 remaining job parity, starting with P5.11c Copilot/conversation response jobs. B12 has repeatable API/browser/enterprise smoke harnesses in Review; optional live failures should be converted into named slices instead of blocking Phase 5 job work. - The Hermes search plan is fully represented by Phase 1/B6. Future search changes must be Meilisearch-first and must not reintroduce production DB fallback. - The Hermes automation/macro/CSAT plan is fully represented by B8/B9 and Phase 5. Durable delayed execution and channel-specific template delivery remain visible Phase 5 work, not hidden Done work. - Enterprise scope is fixed: SLA, Audit, CustomRole, AgentCapacity, Captain/Copilot, CSAT, InboxLimit, automation, macros, assignment policies, and related limits/workflows are in scope; SSO/SAML/LDAP/OIDC are out of scope. @@ -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 harnesses exist and optional live runs can expose follow-up slices | -| Phase 5 | Background jobs and integrations | Doing | P5.1/P5.2/P5.3/P5.4/P5.5/P5.6/P5.7/P5.8/P5.9/P5.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, inbound webhook persistence, outbound/provider delivery status, Captain document sync, conversation maintenance, message status update, and account bulk-action cores are in Review; remaining Captain/Copilot and analytics aggregation jobs remain 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.9/P5.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, inbound webhook persistence, outbound/provider delivery status, Captain document sync/crawl/response/embedding, conversation maintenance, message status update, and account bulk-action cores are in Review; remaining Copilot/conversation response and analytics aggregation jobs remain 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 | @@ -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 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 ./...`. | +| 1 | P5.11 Captain/Copilot jobs | Captain document sync, crawl/parser/schedule, response-builder, and embedding jobs are queued; Copilot/conversation response jobs remain open. | Add fakeable durable jobs for remaining Copilot/conversation async work and disabled-provider states. | Focused 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. | @@ -101,7 +101,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 | 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.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. | Review by `feat(captain): queue response embedding jobs`; tests cover response reset/create, embedding-disabled retry, fake embedding success, account scope, 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. | | Phase 2/3 drift audit | `cmd/route_parity`, `docs/parity/*`, serializer tests | `reference/chatwoot/config/routes.rb`, controller Jbuilder views, reused frontend API clients | Convert any smoke/reference mismatch into a named route, controller, or serializer slice. Static route extraction remains acceptable until Ruby/Bundler is available. | Regenerated route parity shows 0 missing tracked frontend routes; new serializer fixtures cover the drift. | @@ -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 response embedding jobs` | Advances P5.11b with durable Captain document response building and embedding update fan-out. Successful document sync/parser content updates enqueue `captain:document_response_builder`, worker replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses from a fakeable FAQ backend, and enqueues `captain:llm_update_embedding` jobs for created responses. Embedding replay reloads account-scoped responses, uses a fakeable embedding backend or configured LLM provider, and surfaces missing provider config as retryable worker failures. | `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.11b moves to Review; continue P5.11c Copilot/conversation response jobs, then P5.13 analytics aggregation. | | `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. | @@ -512,7 +513,7 @@ Upcoming enterprise task boards: | B10 | B10.4 | Review remaining InboxLimit/account-limit create-path enforcement outside capacity policies. | Chatwoot enterprise account/inbox limit policies. | Inbox/channel create tests for over-limit behavior and frontend-readable errors. | Done by `feat(inboxes): enforce chatwoot inbox limits` | | B11 | B11.1 | Align Captain assistant CRUD, inbox bindings, responses, documents, scenarios, and custom tools payloads. | Captain controllers/services/frontend clients under `reference/chatwoot`. | Handler/service fixtures for every Captain dashboard client path. | Review; assistant CRUD/tools/inbox binding, scenarios, custom tools, documents, assistant responses, bulk actions, and custom-tool test payloads are landed | | B11 | B11.2 | Align Copilot threads, messages, tasks, preferences, playground/tool-call behavior, and disabled-state feature gates. | Copilot controllers/services/frontend clients under `reference/chatwoot`. | Copilot handler/service tests for persistence, disabled LLM state, and frontend payloads. | Review; thread/message payloads, account/user scoping, assistant scope, and no-LLM fallback are landed; tasks/preferences/tool-call/playground depth remains in B11.3 follow-up | -| B11 | B11.3 | Add document sync/embedding/LLM job boundaries where external dependencies are required and finish remaining Copilot task/preference/tool-call/streaming depth. | Captain/Copilot jobs, document services, Copilot controllers/services/frontend clients. | Worker tests or explicit feature-gated fallback tests plus Copilot task/preference/tool-call fixtures. | Review; B11.3a preferences, B11.3b playground, B11.3c document sync gate, B11.3d task/tool-call persistence, and B11.3e streaming fallback landed. Remaining durable crawl/response/embedding/Copilot jobs are P5.11 | +| B11 | B11.3 | Add document sync/embedding/LLM job boundaries where external dependencies are required and finish remaining Copilot task/preference/tool-call/streaming depth. | Captain/Copilot jobs, document services, Copilot controllers/services/frontend clients. | Worker tests or explicit feature-gated fallback tests plus Copilot task/preference/tool-call fixtures. | Review; B11.3a preferences, B11.3b playground, B11.3c document sync gate, B11.3d task/tool-call persistence, and B11.3e streaming fallback landed. Remaining durable Copilot/conversation response jobs are P5.11 | | B12 | B12.1 | Add a repeatable command to run the reused Chatwoot frontend against GoChat. | `reference/chatwoot` frontend boot/auth/API clients. | Smoke command documented and runnable locally. | Done by `test(parity): add frontend smoke harness` | | B12 | B12.2 | Cover login, inbox list/settings, conversation list/detail/message send, contact/company views, widget init/message, public CSAT, SLA/CSAT reports, and enterprise admin screens. | Dashboard route usage and frontend stores/API modules. | Smoke report checked into `docs/parity/` with pass/fail gaps. | Review; B12.2a API assertion command and B12.2b browser smoke harness landed; live run is environment-dependent | | B12 | B12.3 | Add enterprise smoke coverage for SLA reports/download, CSAT reports/download, automation/macros, audit/custom roles, capacity, Captain, and Copilot. | Enterprise dashboard clients and `reference/chatwoot` enterprise screens. | Enterprise smoke command records pass/fail and names failed route/serializer owners. | Review; B12.3a enterprise API assertions and B12.3b enterprise browser navigation landed. Optional live execution is environment-dependent | @@ -1638,7 +1639,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_*` now cover Captain document sync, crawl/parser, and schedule-sync jobs; P5.11 still needs response/embedding/Copilot response jobs. +- `internal/service/captain_*` and `internal/service/copilot_*` now cover Captain document sync, crawl/parser, schedule-sync, response-builder, and embedding-update jobs; P5.11 still needs Copilot/conversation response jobs. - `internal/service/analytics_service.go` has placeholder analytics/report paths and is the owner for P5.13 aggregation work. Checklist: @@ -1646,7 +1647,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 response/embedding/Copilot jobs, delayed automation actions, and report aggregation. +- [ ] Finish durable job dispatch for remaining Copilot/conversation response 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. @@ -1669,7 +1670,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: 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.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 Review by `feat(captain): queue response embedding jobs`; 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 | @@ -1679,7 +1680,7 @@ P5.1 current checkpoint: - Replaced the `WorkerPool` stub with enqueue/schedule APIs, handler registration, `ProcessOne`, start/stop worker loops, PostgreSQL `FOR UPDATE SKIP LOCKED` claiming, retry/backoff, dead-letter state, and stale-lock recovery for worker restart pickup. - Kept older `NewWorkerPool()` construction as a no-op-compatible path while adding `NewWorkerPoolWithOptions(db, ...)` for durable wiring. - Added focused tests proving enqueue/idempotency, due job completion, retry to dead-letter, queue/schedule filtering, and stale running job requeue. -- Remaining Phase 5 work: delayed automation scheduled-item execution, Captain crawl/response/embedding/Copilot response jobs, and analytics/report aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution, Copilot/conversation response jobs, and analytics/report aggregation. P5.1 verification: @@ -1849,9 +1850,11 @@ P5.11 current checkpoint: - `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. +- `captain:document_response_builder` now mirrors Chatwoot enterprise `Captain::Documents::ResponseBuilderJob` on the `low` queue. Successful document sync/parser content updates enqueue response generation, worker replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses from a fakeable FAQ backend, and keeps missing FAQ provider config as an observable retry error. +- `captain:llm_update_embedding` now mirrors `Captain::Llm::UpdateEmbeddingJob` for assistant responses. Response-builder replay fans out embedding jobs for created responses, embedding replay reloads the account-scoped response, uses a fakeable embedding backend or the configured LLM provider, and records disabled/missing provider failures through `background_jobs` retry state. - 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/crawl/schedule requests are replayable after process restart. -- Remaining P5.11 work: response-building jobs, embedding update fan-out, and Copilot response jobs. +- Bootstrap registers Captain document jobs on the shared WorkerPool and wires the assistant-response repository so production sync/crawl/schedule/response/embedding requests are replayable after process restart. +- Remaining P5.11 work: Copilot/conversation response jobs. P5.11 verification: @@ -1990,7 +1993,8 @@ 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: P5.11b Captain response/embedding checkpoint prepared as `feat(captain): queue response embedding jobs`; successful document sync/parser content updates now enqueue `captain:document_response_builder`, response-builder replay resets only unedited document responses, preserves edited responses, creates approved `Captain::Document` assistant responses through a fakeable FAQ backend, and fans out `captain:llm_update_embedding` jobs. Embedding replay reloads account-scoped responses and uses a fakeable embedding backend or configured LLM provider, with missing providers surfacing as retryable worker failures. Focused Captain document worker tests, service/worker/app package tests, full `go test ./...`, and `git diff --check` passed. Remaining P5.11 follow-up is Copilot/conversation response jobs. +- 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 now Copilot/conversation response jobs after P5.11b. - 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. @@ -2094,6 +2098,6 @@ 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. Remaining Phase 5 follow-up is Captain/Copilot 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. Remaining Phase 5 follow-up is Captain/Copilot 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. Remaining Phase 5 follow-up is Captain crawl/response/embedding/Copilot response 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. Remaining Phase 5 follow-up is now Copilot/conversation response jobs and analytics aggregation after P5.11a/P5.11b. - 2026-06-05: P5.12 durable message status checkpoint prepared as `feat(conversations): queue message status updates`; widget and public inbox last-seen updates now enqueue `conversation:update_message_status` jobs on the `deferred` queue, and worker replay marks eligible non-incoming `sent/delivered` messages as `read` or `delivered` up to the viewer timestamp while ignoring invalid statuses and missing conversations. Focused service/worker/app tests passed. Remaining Phase 5 follow-up is Captain/Copilot jobs and analytics aggregation. - 2026-06-05: P5.12 durable account bulk action checkpoint prepared as `feat(conversations): queue bulk actions`; account conversation bulk actions now accept Chatwoot frontend payloads without `action_name`, enqueue `conversation:bulk_action` on the `medium` queue, and worker replay applies display-ID-scoped status/team/assignee/snooze/label updates without crossing account boundaries. Focused service/API/worker/app tests passed. Remaining Phase 5 follow-up is Captain/Copilot jobs and analytics aggregation. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 59591863..ec3ab358 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -581,6 +581,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.SetResponseRepo(captainAssistantResponseRepo) 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) diff --git a/internal/service/captain_document_service.go b/internal/service/captain_document_service.go index ceed8952..55648043 100644 --- a/internal/service/captain_document_service.go +++ b/internal/service/captain_document_service.go @@ -15,6 +15,7 @@ import ( "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" + "github.com/pgvector/pgvector-go" "gorm.io/gorm" ) @@ -23,10 +24,13 @@ import ( type CaptainDocumentService struct { documentRepo *repository.CaptainDocumentRepo assistantRepo *repository.CaptainAssistantRepo + responseRepo *repository.CaptainAssistantResponseRepo llmProvider llm.Provider syncBackend CaptainDocumentSyncBackend crawlBackend CaptainDocumentCrawlBackend pageParser CaptainDocumentPageParserBackend + faqBackend CaptainDocumentFAQBackend + embeddings CaptainDocumentEmbeddingBackend worker *worker.WorkerPool } @@ -53,6 +57,19 @@ type CaptainDocumentPageParserBackend interface { ParseCaptainDocumentPage(ctx context.Context, pageLink string) (*CaptainDocumentSyncResult, error) } +type CaptainDocumentFAQBackend interface { + GenerateCaptainDocumentFAQs(ctx context.Context, doc *model.CaptainDocument) ([]CaptainDocumentFAQ, error) +} + +type CaptainDocumentFAQ struct { + Question string + Answer string +} + +type CaptainDocumentEmbeddingBackend interface { + GenerateCaptainEmbedding(ctx context.Context, accountID uint, content string) (pgvector.Vector, error) +} + // NewCaptainDocumentService creates a new CaptainDocumentService. func NewCaptainDocumentService( documentRepo *repository.CaptainDocumentRepo, @@ -81,6 +98,18 @@ func (s *CaptainDocumentService) SetPageParserBackend(pageParser CaptainDocument s.pageParser = pageParser } +func (s *CaptainDocumentService) SetResponseRepo(responseRepo *repository.CaptainAssistantResponseRepo) { + s.responseRepo = responseRepo +} + +func (s *CaptainDocumentService) SetFAQBackend(faqBackend CaptainDocumentFAQBackend) { + s.faqBackend = faqBackend +} + +func (s *CaptainDocumentService) SetEmbeddingBackend(embeddingBackend CaptainDocumentEmbeddingBackend) { + s.embeddings = embeddingBackend +} + func (s *CaptainDocumentService) SetWorkerPool(wp *worker.WorkerPool) { s.worker = wp RegisterCaptainDocumentJobs(wp, s) @@ -425,6 +454,9 @@ func (s *CaptainDocumentService) ParseCrawledPage(ctx context.Context, accountID } else if err := s.documentRepo.Update(ctx, doc); err != nil { return nil, fmt.Errorf("update crawled document: %w", err) } + if err := s.enqueueDocumentResponseBuilder(ctx, accountID, doc.ID); err != nil { + return nil, err + } return s.documentRepo.GetByAccountAndID(ctx, accountID, doc.ID) } @@ -472,9 +504,138 @@ func (s *CaptainDocumentService) SyncDocumentByAccount(ctx context.Context, acco if err := s.documentRepo.Update(ctx, doc); err != nil { return nil, fmt.Errorf("update synced document: %w", err) } + if err := s.enqueueDocumentResponseBuilder(ctx, accountID, id); err != nil { + return nil, err + } return s.documentRepo.GetByAccountAndID(ctx, accountID, id) } +func (s *CaptainDocumentService) enqueueDocumentResponseBuilder(ctx context.Context, accountID, id uint) error { + if s.worker == nil || s.responseRepo == nil { + return nil + } + _, err := s.worker.Enqueue(ctx, TaskTypeCaptainDocumentResponseBuilder, captainDocumentResponseBuilderJob{AccountID: accountID, DocumentID: id}, + worker.WithQueue("low"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:document_response_builder:%d:%d", accountID, id)), + ) + return err +} + +func (s *CaptainDocumentService) BuildResponsesForDocumentByAccount(ctx context.Context, accountID, id uint) ([]model.CaptainAssistantResponse, error) { + if s.responseRepo == nil { + return nil, fmt.Errorf("captain response repository is required") + } + doc, err := s.documentRepo.GetByAccountAndID(ctx, accountID, id) + if err != nil { + return nil, fmt.Errorf("document not found: %w", err) + } + if strings.TrimSpace(doc.Content) == "" || doc.Status != model.DocumentStatusCompleted { + return nil, fmt.Errorf("document is not ready for response building") + } + if err := s.responseRepo.DB().WithContext(ctx). + Where("account_id = ? AND documentable_id = ? AND documentable_type IN ? AND edited = ?", accountID, doc.ID, []string{"Captain::Document", "CaptainDocument"}, false). + Delete(&model.CaptainAssistantResponse{}).Error; err != nil { + return nil, fmt.Errorf("reset previous document responses: %w", err) + } + if s.faqBackend == nil { + return nil, fmt.Errorf("faq generation disabled") + } + faqs, err := s.faqBackend.GenerateCaptainDocumentFAQs(ctx, doc) + if err != nil { + return nil, fmt.Errorf("generate document faqs: %w", err) + } + + created := make([]model.CaptainAssistantResponse, 0, len(faqs)) + for _, faq := range faqs { + question := strings.TrimSpace(faq.Question) + answer := strings.TrimSpace(faq.Answer) + if question == "" || answer == "" { + continue + } + documentID := doc.ID + resp := &model.CaptainAssistantResponse{ + AccountID: accountID, + AssistantID: doc.AssistantID, + DocumentableID: &documentID, + DocumentableType: "Captain::Document", + Question: question, + Answer: answer, + Status: model.ResponseStatusApproved, + Edited: false, + } + if err := s.responseRepo.Create(ctx, resp); err != nil { + return created, fmt.Errorf("create document response: %w", err) + } + created = append(created, *resp) + if err := s.enqueueResponseEmbedding(ctx, accountID, resp.ID, fmt.Sprintf("%s: %s", question, answer)); err != nil { + return created, err + } + } + return created, nil +} + +func (s *CaptainDocumentService) enqueueResponseEmbedding(ctx context.Context, accountID, responseID uint, content string) error { + if s.worker == nil { + return nil + } + _, err := s.worker.Enqueue(ctx, TaskTypeCaptainLLMUpdateEmbedding, captainLLMUpdateEmbeddingJob{AccountID: accountID, ResponseID: responseID, Content: content}, + worker.WithQueue("low"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("captain:llm_update_embedding:response:%d:%d", accountID, responseID)), + ) + return err +} + +func (s *CaptainDocumentService) UpdateAssistantResponseEmbeddingByAccount(ctx context.Context, accountID, responseID uint, content string) (*model.CaptainAssistantResponse, error) { + if s.responseRepo == nil { + return nil, fmt.Errorf("captain response repository is required") + } + resp, err := s.responseRepo.GetByAccountAndID(ctx, accountID, responseID) + if err != nil { + return nil, fmt.Errorf("response not found: %w", err) + } + if strings.TrimSpace(content) == "" { + content = fmt.Sprintf("%s: %s", resp.Question, resp.Answer) + } + embedding, err := s.generateResponseEmbedding(ctx, accountID, content) + if err != nil { + return nil, err + } + resp.Embedding = embedding + if s.responseRepo.DB().Dialector != nil && s.responseRepo.DB().Dialector.Name() == "sqlite" { + if err := s.responseRepo.DB().WithContext(ctx).Omit("Embedding").Save(resp).Error; err != nil { + return nil, fmt.Errorf("update response embedding: %w", err) + } + return resp, nil + } + if err := s.responseRepo.Update(ctx, resp); err != nil { + return nil, fmt.Errorf("update response embedding: %w", err) + } + return s.responseRepo.GetByAccountAndID(ctx, accountID, responseID) +} + +func (s *CaptainDocumentService) generateResponseEmbedding(ctx context.Context, accountID uint, content string) (pgvector.Vector, error) { + if s.embeddings != nil { + return s.embeddings.GenerateCaptainEmbedding(ctx, accountID, content) + } + if s.llmProvider == nil { + return pgvector.Vector{}, fmt.Errorf("embedding generation disabled") + } + result, err := s.llmProvider.CreateEmbedding(ctx, llm.EmbeddingRequest{Model: "text-embedding-3-small", Input: []string{content}}) + if err != nil { + return pgvector.Vector{}, fmt.Errorf("generate response embedding: %w", err) + } + if len(result.Data) == 0 { + return pgvector.Vector{}, fmt.Errorf("no embedding returned") + } + values := make([]float32, len(result.Data[0].Embedding)) + for i, value := range result.Data[0].Embedding { + values[i] = float32(value) + } + return pgvector.NewVector(values), nil +} + func (s *CaptainDocumentService) markDocumentSyncStarted(ctx context.Context, doc *model.CaptainDocument) error { now := time.Now().Unix() doc.LastSyncAttemptedAt = &now diff --git a/internal/service/captain_document_service_test.go b/internal/service/captain_document_service_test.go index 15a27816..86b241b5 100644 --- a/internal/service/captain_document_service_test.go +++ b/internal/service/captain_document_service_test.go @@ -10,6 +10,7 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/worker" + "github.com/pgvector/pgvector-go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "gorm.io/driver/sqlite" @@ -21,14 +22,17 @@ 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{}, &model.BackgroundJob{})) + require.NoError(t, db.AutoMigrate(&model.Account{}, &model.CaptainAssistant{}, &model.CaptainDocument{}, &model.CaptainAssistantResponse{}, &model.BackgroundJob{})) t.Cleanup(func() { sqlDB, _ := db.DB() sqlDB.Close() }) documentRepo := repository.NewCaptainDocumentRepo(db) assistantRepo := repository.NewCaptainAssistantRepo(db) - return db, NewCaptainDocumentService(documentRepo, nil, assistantRepo) + responseRepo := repository.NewCaptainAssistantResponseRepo(db) + svc := NewCaptainDocumentService(documentRepo, nil, assistantRepo) + svc.SetResponseRepo(responseRepo) + return db, svc } func seedCaptainDocumentSyncFixture(t *testing.T, db *gorm.DB) (*model.Account, *model.Account, *model.CaptainDocument) { @@ -266,6 +270,171 @@ func TestEnqueueCaptainDocumentScheduleSyncsUsesDailyIdempotency(t *testing.T) { assert.Equal(t, int64(1), count) } +func TestCaptainDocumentService_ResponseBuilderCreatesResponsesAndEmbeddingJobs(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + doc.Content = "How refunds work. How billing works." + doc.Status = model.DocumentStatusCompleted + require.NoError(t, db.Save(doc).Error) + uneditedDocID := doc.ID + unedited := &model.CaptainAssistantResponse{ + AccountID: account.ID, + AssistantID: doc.AssistantID, + DocumentableID: &uneditedDocID, + DocumentableType: "Captain::Document", + Question: "old", + Answer: "old answer", + Status: model.ResponseStatusApproved, + Edited: false, + } + edited := &model.CaptainAssistantResponse{ + AccountID: account.ID, + AssistantID: doc.AssistantID, + DocumentableID: &uneditedDocID, + DocumentableType: "Captain::Document", + Question: "keep", + Answer: "edited answer", + Status: model.ResponseStatusApproved, + Edited: true, + } + require.NoError(t, db.Create(unedited).Error) + require.NoError(t, db.Create(edited).Error) + svc.SetFAQBackend(&captainDocumentFakeFAQBackend{faqs: []CaptainDocumentFAQ{ + {Question: "How do refunds work?", Answer: "Refunds take five days."}, + {Question: "How does billing work?", Answer: "Billing is monthly."}, + }}) + now := time.Date(2026, 6, 6, 2, 0, 0, 0, time.UTC) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return now })) + svc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeCaptainDocumentResponseBuilder, captainDocumentResponseBuilderJob{AccountID: account.ID, DocumentID: doc.ID}, worker.WithQueue("low"), worker.WithMaxAttempts(3)) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + var deleted model.CaptainAssistantResponse + assert.Error(t, db.First(&deleted, unedited.ID).Error) + var kept model.CaptainAssistantResponse + require.NoError(t, db.First(&kept, edited.ID).Error) + assert.True(t, kept.Edited) + + var responses []model.CaptainAssistantResponse + require.NoError(t, db.Where("account_id = ? AND documentable_id = ? AND edited = ?", account.ID, doc.ID, false).Order("id ASC").Find(&responses).Error) + require.Len(t, responses, 2) + assert.Equal(t, "Captain::Document", responses[0].DocumentableType) + assert.Equal(t, model.ResponseStatusApproved, responses[0].Status) + + var embeddingJobs []model.BackgroundJob + require.NoError(t, db.Where("job_type = ? AND queue = ?", TaskTypeCaptainLLMUpdateEmbedding, "low").Find(&embeddingJobs).Error) + require.Len(t, embeddingJobs, 2) + assert.Contains(t, string(embeddingJobs[0].Payload), fmt.Sprintf("\"response_id\":%d", responses[0].ID)) +} + +func TestCaptainDocumentService_EmbeddingJobUpdatesResponse(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + documentID := doc.ID + resp := &model.CaptainAssistantResponse{ + AccountID: account.ID, + AssistantID: doc.AssistantID, + DocumentableID: &documentID, + DocumentableType: "Captain::Document", + Question: "What is Captain?", + Answer: "Captain answers customers.", + Status: model.ResponseStatusApproved, + } + require.NoError(t, db.Create(resp).Error) + embeddings := &captainDocumentFakeEmbeddingBackend{embedding: pgvector.NewVector([]float32{0.1, 0.2, 0.3})} + svc.SetEmbeddingBackend(embeddings) + wp := worker.NewWorkerPool(db) + svc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeCaptainLLMUpdateEmbedding, captainLLMUpdateEmbeddingJob{AccountID: account.ID, ResponseID: resp.ID}, worker.WithQueue("low"), worker.WithMaxAttempts(3)) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + assert.True(t, processed) + + assert.Equal(t, fmt.Sprintf("%s: %s", resp.Question, resp.Answer), embeddings.content) + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeCaptainLLMUpdateEmbedding).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusCompleted, job.Status) +} + +func TestCaptainDocumentService_EmbeddingJobRetriesWhenProviderDisabled(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + documentID := doc.ID + resp := &model.CaptainAssistantResponse{ + AccountID: account.ID, + AssistantID: doc.AssistantID, + DocumentableID: &documentID, + DocumentableType: "Captain::Document", + Question: "What is Captain?", + Answer: "Captain answers customers.", + Status: model.ResponseStatusApproved, + } + require.NoError(t, db.Create(resp).Error) + now := time.Date(2026, 6, 6, 2, 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(), TaskTypeCaptainLLMUpdateEmbedding, captainLLMUpdateEmbeddingJob{AccountID: account.ID, ResponseID: resp.ID}, 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 = ?", TaskTypeCaptainLLMUpdateEmbedding).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + assert.Contains(t, job.LastError, "embedding generation disabled") +} + +func TestCaptainDocumentService_ResponseBuilderRetriesWhenFAQDisabled(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + account, _, doc := seedCaptainDocumentSyncFixture(t, db) + doc.Content = "ready content" + doc.Status = model.DocumentStatusCompleted + require.NoError(t, db.Save(doc).Error) + now := time.Date(2026, 6, 6, 2, 30, 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(), TaskTypeCaptainDocumentResponseBuilder, captainDocumentResponseBuilderJob{AccountID: account.ID, DocumentID: doc.ID}, 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 = ?", TaskTypeCaptainDocumentResponseBuilder).First(&job).Error) + assert.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + assert.Contains(t, job.LastError, "faq generation disabled") +} + +func TestCaptainDocumentService_ResponseBuilderScopesAccount(t *testing.T) { + db, svc := setupCaptainDocumentServiceTest(t) + _, otherAccount, doc := seedCaptainDocumentSyncFixture(t, db) + doc.Content = "ready content" + doc.Status = model.DocumentStatusCompleted + require.NoError(t, db.Save(doc).Error) + svc.SetFAQBackend(&captainDocumentFakeFAQBackend{faqs: []CaptainDocumentFAQ{{Question: "Q", Answer: "A"}}}) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithBackoff(func(attempt int) time.Duration { return time.Minute })) + svc.SetWorkerPool(wp) + + _, err := wp.Enqueue(context.Background(), TaskTypeCaptainDocumentResponseBuilder, captainDocumentResponseBuilderJob{AccountID: otherAccount.ID, DocumentID: doc.ID}, 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 count int64 + require.NoError(t, db.Model(&model.CaptainAssistantResponse{}).Where("documentable_id = ?", doc.ID).Count(&count).Error) + assert.Equal(t, int64(0), count) +} + type captainDocumentFakeSyncBackend struct { result *CaptainDocumentSyncResult err error @@ -305,3 +474,29 @@ func (b *captainDocumentFakePageParserBackend) ParseCaptainDocumentPage(ctx cont } return b.pages[pageLink], nil } + +type captainDocumentFakeFAQBackend struct { + faqs []CaptainDocumentFAQ + err error +} + +func (b *captainDocumentFakeFAQBackend) GenerateCaptainDocumentFAQs(ctx context.Context, doc *model.CaptainDocument) ([]CaptainDocumentFAQ, error) { + if b.err != nil { + return nil, b.err + } + return b.faqs, nil +} + +type captainDocumentFakeEmbeddingBackend struct { + embedding pgvector.Vector + err error + content string +} + +func (b *captainDocumentFakeEmbeddingBackend) GenerateCaptainEmbedding(ctx context.Context, accountID uint, content string) (pgvector.Vector, error) { + b.content = content + if b.err != nil { + return pgvector.Vector{}, b.err + } + return b.embedding, nil +} diff --git a/internal/service/captain_document_worker.go b/internal/service/captain_document_worker.go index 4e87edc7..ae2f94d9 100644 --- a/internal/service/captain_document_worker.go +++ b/internal/service/captain_document_worker.go @@ -12,10 +12,12 @@ import ( ) const ( - TaskTypeCaptainDocumentSync = "captain:document_sync" - TaskTypeCaptainDocumentCrawl = "captain:document_crawl" - TaskTypeCaptainDocumentPageCrawlParse = "captain:document_page_crawl_parse" - TaskTypeCaptainDocumentScheduleSyncs = "captain:documents_schedule_syncs" + TaskTypeCaptainDocumentSync = "captain:document_sync" + TaskTypeCaptainDocumentCrawl = "captain:document_crawl" + TaskTypeCaptainDocumentPageCrawlParse = "captain:document_page_crawl_parse" + TaskTypeCaptainDocumentScheduleSyncs = "captain:documents_schedule_syncs" + TaskTypeCaptainDocumentResponseBuilder = "captain:document_response_builder" + TaskTypeCaptainLLMUpdateEmbedding = "captain:llm_update_embedding" ) const captainDocumentScheduleInterval = 24 * time.Hour @@ -40,6 +42,17 @@ type captainDocumentScheduleSyncsJob struct { PlanName string `json:"plan_name,omitempty"` } +type captainDocumentResponseBuilderJob struct { + AccountID uint `json:"account_id"` + DocumentID uint `json:"document_id"` +} + +type captainLLMUpdateEmbeddingJob struct { + AccountID uint `json:"account_id"` + ResponseID uint `json:"response_id"` + Content string `json:"content,omitempty"` +} + var captainDocumentRegistrations sync.Map // RegisterCaptainDocumentJobs wires Captain::Documents::PerformSyncJob into @@ -56,6 +69,8 @@ func RegisterCaptainDocumentJobs(wp *worker.WorkerPool, svc *CaptainDocumentServ wp.Register(TaskTypeCaptainDocumentCrawl, svc.performDocumentCrawlJob) wp.Register(TaskTypeCaptainDocumentPageCrawlParse, svc.performDocumentPageCrawlParseJob) wp.Register(TaskTypeCaptainDocumentScheduleSyncs, svc.performDocumentScheduleSyncsJob) + wp.Register(TaskTypeCaptainDocumentResponseBuilder, svc.performDocumentResponseBuilderJob) + wp.Register(TaskTypeCaptainLLMUpdateEmbedding, svc.performCaptainLLMUpdateEmbeddingJob) } func EnqueueCaptainDocumentScheduleSyncs(ctx context.Context, wp *worker.WorkerPool, scheduledAt time.Time) (*model.BackgroundJob, error) { @@ -127,3 +142,27 @@ func (s *CaptainDocumentService) performDocumentScheduleSyncsJob(ctx context.Con _, err := EnqueueCaptainDocumentScheduleSyncs(ctx, s.worker, time.Now().Add(captainDocumentScheduleInterval)) return err } + +func (s *CaptainDocumentService) performDocumentResponseBuilderJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainDocumentResponseBuilderJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain document response builder job: %w", err) + } + if payload.AccountID == 0 || payload.DocumentID == 0 { + return fmt.Errorf("invalid captain document response builder job payload: %#v", payload) + } + _, err := s.BuildResponsesForDocumentByAccount(ctx, payload.AccountID, payload.DocumentID) + return err +} + +func (s *CaptainDocumentService) performCaptainLLMUpdateEmbeddingJob(ctx context.Context, job *model.BackgroundJob) error { + var payload captainLLMUpdateEmbeddingJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal captain llm update embedding job: %w", err) + } + if payload.AccountID == 0 || payload.ResponseID == 0 { + return fmt.Errorf("invalid captain llm update embedding job payload: %#v", payload) + } + _, err := s.UpdateAssistantResponseEmbeddingByAccount(ctx, payload.AccountID, payload.ResponseID, payload.Content) + return err +}