diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index fdb15eac..bf4f01f9 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 `1a4b4a2 feat(messages): queue send replies`, with this checkpoint prepared as `feat(messages): queue delivery statuses`. -- Latest implementation checkpoint: this checkpoint, prepared as `feat(messages): queue delivery statuses`. +- Current tracking checkpoint: 2026-06-05 after `dec8854 feat(messages): queue delivery statuses`, with this checkpoint prepared as `feat(webhook): queue incoming messages`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(webhook): queue incoming messages`. - Latest documentation-only checkpoint: this checkpoint, prepared as `docs: land active parity tracking board`; 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.10 now 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 P5.9 inbound webhook deferral, followed by remaining P5.11 Captain/Copilot jobs and 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 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. - `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,17 +45,16 @@ Next ordered checkpoints: | Order | Slice | Required outcome | Primary verification | | --- | --- | --- | --- | -| 1 | P5.9 inbound jobs | Move provider webhook persistence/dispatch that Chatwoot defers into durable jobs while keeping HTTP acknowledgement behavior compatible. | Provider webhook handler tests plus worker replay/idempotency tests. | -| 2 | 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. | -| 3 | P5.13 analytics aggregation | Replace frontend-visible placeholder report values with scheduled or cached real aggregations. | Report service/handler fixtures and freshness/idempotency tests. | -| 4 | 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. | -| 5 | 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. | +| 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. | +| 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. | ## Handoff Contract 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 P5.9 inbound webhook deferral. 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 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 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. @@ -65,7 +64,7 @@ Open work after the current checkpoint: | Area | Next concrete action | Tracking location | Done boundary | | --- | --- | --- | --- | -| Phase 5 jobs | Finish outbound delivery, inbound webhook deferral, remaining Captain/Copilot jobs, and analytics aggregation on top of the committed durable worker. | `Phase 5: Background Jobs And Integrations` | Worker tests prove enqueue, retry, idempotency, and fakeable external effects. | +| Phase 5 jobs | Finish remaining Captain/Copilot jobs and analytics aggregation on top of the committed durable worker. | `Phase 5: Background Jobs And Integrations` | Worker tests prove enqueue, retry, idempotency, and fakeable external effects. | | B12 smoke | Run optional live API/browser/enterprise smoke in a full PostgreSQL/Redis/Meilisearch/Vite environment and convert failures into named slices. | `B12 reused frontend verification breakdown` | `docs/parity/frontend_smoke_report.md` records checked pass/fail results and maps failures to slices. | | Phase 2/3 drift | Expand tracked route/serializer fixtures when B12 exposes frontend-critical gaps. | `Phase 2`, `Phase 3`, `docs/parity/` | Route parity remains 0 missing for tracked frontend routes; serializers have reference fixtures. | | Phase 6 placeholders | Re-run placeholder audit and burn down any frontend-reachable stub. | `Phase 6: Core Product Placeholder Burn-down` | Stub list has no reused-frontend critical path without a named owner. | @@ -79,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.10/P5.11/P5.12 durable worker, event dispatch, search indexing, automation delivery, macro, CSAT survey/template, SLA scan, contact export, outbound/provider delivery status, Captain document sync, conversation maintenance, message status update, and account bulk-action cores are in Review; remaining inbound webhook, 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, conversation maintenance, message status update, and account bulk-action cores are in Review; remaining Captain/Copilot 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 | @@ -89,12 +88,11 @@ This table is the shortest authoritative handoff view. If an older lower section | Priority | Workstream | Current state | Next checkpoint | Commit close rule | | --- | --- | --- | --- | --- | -| 1 | P5.9 inbound webhook deferral | Provider ingress routes verify, persist, and dispatch synchronously. | Preserve provider HTTP ack behavior while moving Chatwoot-deferred persistence/dispatch into retryable jobs. | Provider HTTP tests plus worker replay/idempotency tests must pass. | -| 2 | 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 ./...`. | -| 3 | 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. | -| 4 | 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. | -| 5 | 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. | -| 6 | B12 optional live smoke | API/browser/enterprise smoke commands are checked in; live runs need PostgreSQL, Redis, Meilisearch, Vite, and Chrome. | Run full live smoke when environment is available and map failures to the board. | `docs/parity/frontend_smoke_report.md` records pass/fail and linked owners. | +| 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 ./...`. | +| 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. | +| 5 | B12 optional live smoke | API/browser/enterprise smoke commands are checked in; live runs need PostgreSQL, Redis, Meilisearch, Vite, and Chrome. | Run full live smoke when environment is available and map failures to the board. | `docs/parity/frontend_smoke_report.md` records pass/fail and linked owners. | ## Tracking Artifacts @@ -115,14 +113,15 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | -| `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. | Move P5.10 to Review; next active slice is P5.9 inbound webhook deferral, then remaining P5.11 Captain/Copilot jobs and P5.13 analytics aggregation. | +| `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`. | | `feat(messages): queue send replies` | Advances P5.10 with a durable Chatwoot `SendReplyJob` equivalent for outbound message delivery. `MessageService.Create` now enqueues `message:send_reply` on the `high` queue when a WorkerPool is configured, keeps the legacy `message.outgoing` dispatch fallback only for no-worker construction, and worker replay reloads message/conversation/inbox/contact, skips already-sent/private/voice-call/non-outgoing messages, calls the registered channel provider, persists `source_id`, and records retryable provider failures as `failed` with `content_attributes.external_error`. | `go test ./internal/service -run 'MessageDeliveryWorker\|MessageService' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.10 section. | Send-reply core moved to Review; provider delivery-status depth is completed by `feat(messages): queue delivery statuses`. | | `docs: land active parity tracking board` | Documentation-only checkpoint that reconciles the current state after `cda7d02`. Adds the front-loaded Active Remaining Work Board, makes Phase 5 the active slice, records P5.10/P5.9/P5.11/P5.13 as the next implementation order, and updates stale B12/M6/M7 tracker rows to Review/Doing. | `git diff --check`; documentation-only change. | Start P5.10 outbound message delivery from the new active board. | -| `feat(conversations): queue bulk actions` | Advances P5.12 with a durable Chatwoot `BulkActionsJob` equivalent for account conversation bulk actions. `POST /bulk_actions` now accepts frontend Chatwoot payloads without `action_name`, including `fields`, nested `labels.add/remove`, `snoozed_until`, and display-ID `ids`; when a WorkerPool is configured it enqueues `conversation:bulk_action` on the `medium` queue and returns empty `200 OK`. Worker replay scopes by account and display IDs, merges labels, applies status/team/assignee/snooze updates, and leaves same-display conversations in other accounts untouched. | `go test ./internal/service -run 'ConversationMaintenanceJobs.*BulkAction\|ConversationMaintenanceJobs' -count=1`; `go test ./internal/handler/api/v1 -run 'BulkActionHandler' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | Move P5.12 account bulk actions to Review; continue remaining Captain/Copilot jobs, provider webhook/outbound delivery, inbound webhook job deferral, and analytics aggregation. | -| `feat(conversations): queue message status updates` | Advances P5.12 with a durable Chatwoot `Conversations::UpdateMessageStatusJob` equivalent. Widget and public inbox `update_last_seen` paths now enqueue `conversation:update_message_status` on the `deferred` queue when a WorkerPool is configured. Worker replay validates `read`/`delivered`, skips missing conversations, and updates only non-incoming messages created at or before the viewing timestamp while preserving already-read and post-cutoff messages. | `go test ./internal/service -run 'ConversationMaintenanceJobs\|WidgetService_UpdateLastSeenQueuesMessageStatusJob' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | Move P5.12 update-message-status depth to Review; continue durable bulk-action fan-out, remaining Captain/Copilot jobs, provider webhook/outbound delivery, and analytics aggregation. | -| `feat(captain): queue document syncs` | Advances P5.11 with a durable Chatwoot `Captain::Documents::PerformSyncJob` equivalent. `POST /captain/documents/:id/sync` still returns `202` after marking the account-scoped document as syncing, but now enqueues `captain:document_sync` on the `low` queue when a WorkerPool is configured. Worker replay calls the existing fakeable `CaptainDocumentSyncBackend`, persists synced/failed/disabled states through `SyncDocumentByAccount`, and lets missing documents or malformed payloads retry through `background_jobs`. | `go test ./internal/service -run 'CaptainDocumentService' -count=1`; `go test ./internal/handler/api/v1 -run 'CaptainDocument\|CaptainResource' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | Move P5.11 document sync core to Review; continue Captain crawl/response-building/embedding/Copilot response jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. | -| `feat(csat): queue channel templates` | Advances P5.6 with durable WhatsApp/Twilio CSAT template creation. `POST /inboxes/:inbox_id/csat_template` accepts Chatwoot nested `template` params, persists the singular template as pending, and enqueues `csat:template_create` when a WorkerPool is configured. Worker replay calls a fakeable provider boundary, records WhatsApp template metadata or Twilio content/approval SIDs into `inbox.csat_config`, updates template status, and retries provider failures through `background_jobs`. | `go test ./internal/service -run 'CsatTemplate' -count=1`; `go test ./internal/handler/api/v1 -run 'InboxCsatTemplate' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.6 section. | Move P5.6 channel templates to Review; continue Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. | -| `feat(conversations): queue maintenance jobs` | Advances P5.12 and the scheduled-items portion of P5.5. `scheduled:trigger_items` now mirrors Chatwoot `TriggerScheduledItemsJob` by queueing due one-off campaigns, snoozed conversation reopening, account auto-resolution scheduling, and the next hourly root trigger. One-off campaigns claim active rows as `processing` before execution and complete after artifact creation, so duplicate worker replays do not duplicate campaign messages. | `go test ./internal/service -run 'ConversationMaintenanceJobs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | Move P5.12 conversation maintenance core to Review; continue update-message-status/bulk-action job depth plus CSAT channel templates, Captain/Copilot, provider webhook/outbound delivery, and analytics aggregation. | +| `feat(conversations): queue bulk actions` | Advances P5.12 with a durable Chatwoot `BulkActionsJob` equivalent for account conversation bulk actions. `POST /bulk_actions` now accepts frontend Chatwoot payloads without `action_name`, including `fields`, nested `labels.add/remove`, `snoozed_until`, and display-ID `ids`; when a WorkerPool is configured it enqueues `conversation:bulk_action` on the `medium` queue and returns empty `200 OK`. Worker replay scopes by account and display IDs, merges labels, applies status/team/assignee/snooze updates, and leaves same-display conversations in other accounts untouched. | `go test ./internal/service -run 'ConversationMaintenanceJobs.*BulkAction\|ConversationMaintenanceJobs' -count=1`; `go test ./internal/handler/api/v1 -run 'BulkActionHandler' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | P5.12 account bulk actions moved to Review; continue remaining Captain/Copilot jobs and analytics aggregation. | +| `feat(conversations): queue message status updates` | Advances P5.12 with a durable Chatwoot `Conversations::UpdateMessageStatusJob` equivalent. Widget and public inbox `update_last_seen` paths now enqueue `conversation:update_message_status` on the `deferred` queue when a WorkerPool is configured. Worker replay validates `read`/`delivered`, skips missing conversations, and updates only non-incoming messages created at or before the viewing timestamp while preserving already-read and post-cutoff messages. | `go test ./internal/service -run 'ConversationMaintenanceJobs\|WidgetService_UpdateLastSeenQueuesMessageStatusJob' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | P5.12 update-message-status depth moved to Review; continue remaining Captain/Copilot jobs and analytics aggregation. | +| `feat(captain): queue document syncs` | Advances P5.11 with a durable Chatwoot `Captain::Documents::PerformSyncJob` equivalent. `POST /captain/documents/:id/sync` still returns `202` after marking the account-scoped document as syncing, but now enqueues `captain:document_sync` on the `low` queue when a WorkerPool is configured. Worker replay calls the existing fakeable `CaptainDocumentSyncBackend`, persists synced/failed/disabled states through `SyncDocumentByAccount`, and lets missing documents or malformed payloads retry through `background_jobs`. | `go test ./internal/service -run 'CaptainDocumentService' -count=1`; `go test ./internal/handler/api/v1 -run 'CaptainDocument\|CaptainResource' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.11 section. | P5.11 document sync core moved to Review; continue Captain crawl/response-building/embedding/Copilot response jobs and analytics aggregation. | +| `feat(csat): queue channel templates` | Advances P5.6 with durable WhatsApp/Twilio CSAT template creation. `POST /inboxes/:inbox_id/csat_template` accepts Chatwoot nested `template` params, persists the singular template as pending, and enqueues `csat:template_create` when a WorkerPool is configured. Worker replay calls a fakeable provider boundary, records WhatsApp template metadata or Twilio content/approval SIDs into `inbox.csat_config`, updates template status, and retries provider failures through `background_jobs`. | `go test ./internal/service -run 'CsatTemplate' -count=1`; `go test ./internal/handler/api/v1 -run 'InboxCsatTemplate' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.6 section. | P5.6 channel templates moved to Review; continue Captain/Copilot jobs and analytics aggregation. | +| `feat(conversations): queue maintenance jobs` | Advances P5.12 and the scheduled-items portion of P5.5. `scheduled:trigger_items` now mirrors Chatwoot `TriggerScheduledItemsJob` by queueing due one-off campaigns, snoozed conversation reopening, account auto-resolution scheduling, and the next hourly root trigger. One-off campaigns claim active rows as `processing` before execution and complete after artifact creation, so duplicate worker replays do not duplicate campaign messages. | `go test ./internal/service -run 'ConversationMaintenanceJobs' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.12 section. | P5.12 conversation maintenance core moved to Review; continue Captain/Copilot jobs and analytics aggregation. | | `feat(crm): queue contact exports` | Advances P5.8 with a durable Chatwoot `Account::ContactsExportJob` equivalent. Contact export requests now create a pending `contact_exports` artifact and enqueue `contact:export` on the `low` queue when a WorkerPool is configured; worker replay generates the CSV artifact, marks completion, creates the persisted notification, and invokes the fakeable completion mailer. No-worker construction keeps the synchronous fallback for focused tests. | `go test ./internal/service -run 'ContactService_ExportContacts\|ContactExportJob' -count=1`; `go test ./internal/service ./internal/handler/api/v1 ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.8 section. | Move P5.8 to Review; continue delayed automation scheduled items, CSAT channel templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. | | `feat(sla): queue applied sla scans` | Advances P5.7 with the durable Chatwoot enterprise SLA chain. `sla:trigger_accounts` scans accounts with SLA policies, queues `sla:process_account` jobs, self-schedules the next five-minute scan, and `sla:process_account` queues `sla:process_applied` jobs for active/active_with_misses AppliedSLA rows; applied jobs call the existing idempotent `AppliedSlaService.Evaluate`. Bootstrap registers the handlers and seeds the first root scan. | `go test ./internal/service -run 'SlaProcessingJobs\|AppliedSlaService' -count=1`; `go test ./internal/service ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.7 section. | Move P5.7 to Review; continue delayed automation scheduled items, CSAT channel templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. | | `feat(search): queue index updates durably` | Advances P5.3 with a durable `SearchIndexer` wrapper. Service-layer create/update/delete hooks now enqueue `search:index` jobs for conversations, messages, contacts, companies, and articles when a WorkerPool is configured; worker replay reloads the current account-scoped record before calling the real Meilisearch-backed `SearchService`, turns missing index records into delete jobs, and leaves search reads pointed at the live search service. | `go test ./internal/service -run 'DurableSearchIndexer\|SearchIndexHooks' -count=1`; `go test ./internal/service ./internal/search ./internal/worker ./internal/app -count=1`; `go test ./...`; `git diff --check`; full verification recorded in the P5.3 section. | Move P5.3 to Review; continue delayed automation scheduled items and provider/template jobs. | @@ -1609,7 +1608,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 have synchronous verified ingress/persistence; P5.9 must move Chatwoot-deferred webhook work behind durable jobs without changing provider acknowledgement semantics. +- `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/analytics_service.go` has placeholder analytics/report paths and is the owner for P5.13 aggregation work. @@ -1618,7 +1617,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 inbound provider webhook deferral, remaining Captain/Copilot jobs, delayed automation actions, and report aggregation. +- [ ] Finish durable job dispatch for remaining Captain/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. @@ -1639,7 +1638,7 @@ Tracking table: | P5.6 | Queue CSAT survey sends and channel-specific templates. | CSAT listener/services, WhatsApp/Twilio template services/jobs | `internal/automation/csat_survey_listener.go`, `internal/service/csat_template_service.go`, channel send services | Resolve-triggered CSAT send is durable; WhatsApp/Twilio template delivery and failure states are fakeable and observable. | Review by `feat(automation): queue macro and csat jobs` and `feat(csat): queue channel templates` | | P5.7 | Queue SLA account-wide scans and applied-SLA processing. | `enterprise/app/jobs/sla/trigger_slas_for_accounts_job.rb`, `process_account_applied_slas_job.rb`, `process_applied_sla_job.rb` | `internal/service/applied_sla_service.go`, SLA event services | Scheduler enqueues account scans and per-SLA processing; threshold transitions remain idempotent and notification fan-out is retryable. | Review by `feat(sla): queue applied sla scans` | | P5.8 | Queue contact export artifact generation and completion notification/email. | `account/contacts_export_job.rb` | contact export service and mailer boundary | Export API returns immediately, artifact generation is durable, completion notification/email is fakeable, and download remains stable after restart. | Review by `feat(crm): queue contact exports` | -| P5.9 | Queue inbound provider webhook processing where Chatwoot defers work. | `webhooks/*_events_job.rb`, `webhooks/*_delivery_job.rb` | webhook handlers, provider services | Provider HTTP ack behavior remains Chatwoot-compatible while persistence/dispatch runs through retryable jobs with signature validation preserved. | Todo | +| P5.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.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` | @@ -1651,7 +1650,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, inbound provider webhook deferral, Captain crawl/response/embedding/Copilot response jobs, and analytics/report aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution, Captain crawl/response/embedding/Copilot response jobs, and analytics/report aggregation. P5.1 verification: @@ -1667,7 +1666,7 @@ P5.2 current checkpoint: - `dispatch.EventDispatcher` can now attach a `WorkerPool`; sync listeners still execute inline, while async listeners and `DispatchAsync` subscribers enqueue durable per-listener `event:listener_dispatch` jobs. - Worker replay resolves the listener by name and invokes `OnEvent` with the serialized event payload. Missing listeners or malformed payloads fail the job, preserving retry/dead-letter visibility. - `WorkerPool` default queue handling was broadened to process all queues unless explicitly filtered, so `events` jobs are processed by the default worker. -- Remaining integration work is feature-specific: delayed automation/macros, inbound provider webhook deferral, Captain/Copilot, and analytics aggregation must register producers/handlers on this durable path. +- Remaining integration work is feature-specific: delayed automation/macros, Captain/Copilot, and analytics aggregation must register producers/handlers on this durable path. P5.2 verification: @@ -1738,7 +1737,7 @@ P5.7 current checkpoint: - `sla:process_account` mirrors `Sla::ProcessAccountAppliedSlasJob`: it finds active and active_with_misses AppliedSLA rows for the account and queues `sla:process_applied` jobs. - `sla:process_applied` mirrors `Sla::ProcessAppliedSlaJob`: it calls the existing `AppliedSlaService.Evaluate`, preserving idempotent miss events, notification fan-out, retry/backoff, and dead-letter visibility through `background_jobs`. - Bootstrap registers the SLA job handlers and enqueues the initial root scan when the app boots. -- Remaining Phase 5 work: delayed automation scheduled-item execution, inbound provider webhook deferral, Captain/Copilot durable jobs, and analytics aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution, Captain/Copilot durable jobs, and analytics aggregation. P5.7 verification: @@ -1756,7 +1755,7 @@ P5.8 current checkpoint: - Completed export jobs are idempotent: a duplicate replay of an already-completed export is a no-op, so completion notifications and mail are not duplicated. - Failed worker execution records the export error and leaves the background job retry/dead-letter state observable through `background_jobs`. - No-worker construction still uses the synchronous fallback for focused tests and local paths that do not start the durable worker. -- Remaining Phase 5 work: delayed automation scheduled-item execution, inbound provider webhook deferral, Captain/Copilot durable jobs, and analytics aggregation. +- Remaining Phase 5 work: delayed automation scheduled-item execution, Captain/Copilot durable jobs, and analytics aggregation. P5.8 verification: @@ -1767,6 +1766,25 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache git diff --check ``` +P5.9 current checkpoint: + +- `webhook:incoming_message_persist` now mirrors the persistence/dispatch deferral used by Chatwoot provider event jobs such as `Webhooks::TelegramEventsJob`, `Webhooks::LineEventsJob`, `Webhooks::WhatsappEventsJob`, `Webhooks::TwilioEventsJob`, `Webhooks::FacebookEventsJob`, and `Webhooks::TiktokEventsJob`. +- Provider HTTP handlers still perform request acknowledgement, signature validation, inbox lookup, and provider-specific payload parsing on the request path, preserving current acknowledgement behavior. +- When a WorkerPool is configured, `IncomingPersister.PersistIncoming` validates the normalized `channel.IncomingMessage`, enqueues `webhook:incoming_message_persist`, and returns before creating contact inboxes, conversations, messages, or dispatching automation/realtime events. +- Worker replay reloads the inbox, runs the existing Chatwoot-style `ContactInbox`/`Conversation`/`Message` persistence path, keeps source-ID dedupe behavior, and dispatches the same `contact.created`, `conversation.created/opened/updated`, `message.created`, and `message.incoming` events from the durable path. +- Incoming jobs use idempotency keys by inbox/source ID; WhatsApp and Twilio inbound messages use the `low` queue to match their reference jobs, while Telegram, LINE, Facebook/Instagram, and TikTok use the default queue. +- Bootstrap wires Telegram, LINE, Facebook/Instagram, WhatsApp, TikTok, and Twilio persisters to the WorkerPool. No-worker construction remains synchronous for existing focused unit tests and local non-worker paths. +- Remaining P5.9 work: none known for normalized inbound persistence/dispatch deferral; any future provider-specific raw event replay gap should be tracked as a separate Phase 2/3 drift slice if reference inspection or smoke exposes one. + +P5.9 verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/handler/webhook -run 'IncomingPersister.*Incoming\|IncomingPersister.*Status\|StatusJob\|TwilioInboundSMS\|TwilioDeliveryStatus' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/handler/webhook ./internal/channel/whatsapp ./internal/worker ./internal/app -count=1 +env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./... +git diff --check +``` + P5.10 current checkpoint: - `message:send_reply` now mirrors Chatwoot `SendReplyJob` on the `high` queue for the send-reply core. Outgoing message creation enqueues provider delivery when a WorkerPool is configured, while no-worker service construction keeps the older `message.outgoing` dispatch fallback. @@ -1779,7 +1797,7 @@ P5.10 current checkpoint: - Delivery-status replay upserts `delivery_statuses`, dispatches `message.status_updated`, records failed provider details in `content_attributes.external_error`, clears stale external errors on non-failed updates, and prevents stale `sent`/`delivered` receipts from downgrading already-`read` messages. - Twilio status callbacks now match the reference `Twilio::DeliveryStatusController` acknowledgement with `204 No Content`, support the exact `/twilio/delivery_status` route plus the legacy `/webhooks/twilio/delivery_status` alias, resolve channels by `MessagingServiceSid` or `AccountSid`/`From`, map `undelivered` to `failed`, and persist Twilio `ErrorCode`/`ErrorMessage` as `external_error`. - Bootstrap wires Facebook/Instagram, WhatsApp, TikTok, and Twilio webhook persisters to the WorkerPool, so provider status callbacks are replayable after process restart. No-worker construction still performs status updates inline for focused unit tests. -- Remaining P5.10 work: none known for the named outbound delivery/status row; follow-up provider ingress persistence deferral is tracked under P5.9. +- Remaining P5.10 work: none known for the named outbound delivery/status row. P5.10 verification: @@ -1824,7 +1842,7 @@ P5.12 current checkpoint: - Message-status replay validates only `read` and `delivered`, no-ops missing conversations, and updates non-incoming `sent`/`delivered` messages created before the viewer timestamp while skipping incoming, already-read, and post-cutoff messages. - `conversation:bulk_action` now mirrors Chatwoot `BulkActionsJob` on the `medium` queue for account conversation bulk actions. The account `POST /bulk_actions` handler accepts the reused frontend payload shape with `fields`, nested `labels.add/remove`, `snoozed_until`, and display-ID `ids`, enqueues the job when a WorkerPool is configured, and returns empty `200 OK` like Chatwoot. - Bulk-action replay scopes records by account and conversation display IDs, merges label adds/removals without duplicating labels, applies status/team/assignee/snooze fields, and leaves same-display conversations in other accounts untouched. -- Remaining P5.12 work: none known for the named maintenance jobs in this row; follow-up provider ingress jobs remain P5.9. +- Remaining P5.12 work: none known for the named maintenance jobs in this row. P5.12 verification: @@ -1939,7 +1957,8 @@ Verification milestone gates: ## Progress Log -- 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. Next slice is P5.9 inbound webhook deferral. +- 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. - 2026-06-05: P5.10 durable send-reply checkpoint prepared as `feat(messages): queue send replies`; outgoing message creation now enqueues `message:send_reply` on the `high` queue when a WorkerPool is configured, worker replay reloads account-scoped message/conversation/inbox/contact records, skips duplicate/private/voice-call/non-deliverable messages, calls the registered channel provider, persists `source_id`, and records retryable provider failures as message `failed` with `content_attributes.external_error`. Focused service/worker/app tests passed; provider delivery-status depth is completed by the next P5.10 checkpoint. - 2026-06-05: Active parity tracker checkpoint prepared as `docs: land active parity tracking board`; the plan now front-loads the current authoritative remaining-work board, points the next implementation slice at P5.10 outbound message delivery, records P5.9 inbound job deferral, P5.11 remaining Captain/Copilot jobs, P5.13 analytics aggregation, Phase 2/3 drift, Phase 6 placeholder audit, and B12 optional live smoke as tracked owners, and reconciles stale B12/M6/M7 status rows. Documentation-only checkpoint; `git diff --check` passed. - 2026-06-04: Baseline stabilized and committed as `42cdab8 chore: stabilize chatwoot parity baseline`; `go test ./...` passed and route dump reported `TOTAL: 704`. @@ -2038,8 +2057,8 @@ Verification milestone gates: - 2026-06-05: P5.5/P5.6 durable macro and CSAT checkpoint prepared as `feat(automation): queue macro and csat jobs`; macro execute now enqueues `automation:macro_execution` jobs for display-ID fan-out when a WorkerPool is configured, and resolved-conversation CSAT sends now enqueue idempotent `csat:survey_send` jobs. Focused automation/worker/app tests passed. Next slice is delayed automation scheduled items and WhatsApp/Twilio CSAT template delivery. - 2026-06-05: P5.7 durable SLA scan checkpoint prepared as `feat(sla): queue applied sla scans`; the Chatwoot enterprise SLA chain now runs through `sla:trigger_accounts`, `sla:process_account`, and `sla:process_applied` jobs, self-schedules five-minute root scans, queues account/applied-SLA fan-out, and reuses idempotent `AppliedSlaService.Evaluate` for miss events and notifications. Focused SLA/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.8 durable contact export checkpoint prepared as `feat(crm): queue contact exports`; contact export requests now enqueue `contact:export` jobs on the low queue when a WorkerPool is configured, while worker replay generates the persisted CSV artifact, completion notification, and fakeable completion email. Focused service/API/worker/app tests passed. Next slice is delayed automation scheduled items, WhatsApp/Twilio CSAT templates, Captain/Copilot jobs, provider delivery, and conversation maintenance. -- 2026-06-05: P5.12 durable conversation maintenance checkpoint prepared as `feat(conversations): queue maintenance jobs`; the Chatwoot scheduled-items chain now runs through `scheduled:trigger_items`, due one-off campaigns queue `campaign:trigger_oneoff`, snoozed conversations reopen through `conversation:reopen_snoozed`, and account auto-resolution fans out through `account:conversations_resolution_scheduler` and `conversation:resolution`. Focused service/worker/app tests passed. Next slice is CSAT WhatsApp/Twilio templates, Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. -- 2026-06-05: P5.6 durable CSAT template checkpoint prepared as `feat(csat): queue channel templates`; inbox CSAT template creation now accepts Chatwoot nested template payloads, queues `csat:template_create` jobs when a WorkerPool is configured, and worker replay records WhatsApp template IDs or Twilio content/approval SIDs through a fakeable provider boundary. Focused service/API/worker/app tests passed. Next slice is Captain/Copilot jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. -- 2026-06-05: P5.11 durable Captain document sync checkpoint prepared as `feat(captain): queue document syncs`; Captain document sync requests now mark the document syncing and enqueue `captain:document_sync` jobs on the `low` queue, while worker replay uses the existing fakeable sync backend to persist disabled, failed, or synced document states. Focused service/API/worker/app tests passed. Next slice is remaining Captain crawl/response/embedding/Copilot response jobs, provider webhook/outbound delivery, update-message-status/bulk-action jobs, and analytics aggregation. -- 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. Next slice is durable account bulk actions, remaining Captain/Copilot jobs, provider webhook/outbound delivery, 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. Next slice is remaining Captain/Copilot jobs, provider webhook/outbound delivery, inbound webhook job deferral, and analytics aggregation. +- 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.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 7726e035..37252003 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -409,6 +409,7 @@ func Bootstrap(env string) (*App, error) { // Create Telegram webhook handler (Gin HTTP handler for Telegram webhook endpoint) telWebhook := telegramchannel.NewWebhookHandler(tgProvider) telegramWebhookHandler := webhook.NewTelegramWebhookHandler(tgProvider, telWebhook, db, channelDispatcher) + telegramWebhookHandler.WithWorkerPool(workerPool) // Step 8c: Wire WhatsApp channel provider (Cloud API / 360dialog) // Reference: Chatwoot registers WhatsApp channel providers in config/initializers/channels.rb @@ -461,6 +462,7 @@ func Bootstrap(env string) (*App, error) { // Create LINE webhook handler (Gin HTTP handler for LINE webhook endpoints) lineWebhook := linechannel.NewWebhookHandler(linePipeline, lineService) lineWebhookHandler := webhook.NewLineWebhookHandler(lineWebhook, linePipeline, lineService, db, channelDispatcher) + lineWebhookHandler.WithWorkerPool(workerPool) // Step 8f: Wire Twilio SMS channel provider (Twilio SMS API) // Twilio SMS requires service/repo/pipeline deps (like WhatsApp), so wire here in bootstrap. diff --git a/internal/handler/webhook/incoming_persister.go b/internal/handler/webhook/incoming_persister.go index 17d94800..05afb0c8 100644 --- a/internal/handler/webhook/incoming_persister.go +++ b/internal/handler/webhook/incoming_persister.go @@ -57,18 +57,42 @@ func (p *IncomingPersister) PersistIncoming(ctx context.Context, inbox *model.In if p == nil || p.db == nil || inbox == nil || msg == nil { return nil, nil } + if err := validateIncomingMessage(p, inbox, msg); err != nil { + return nil, err + } + if p.worker != nil { + return nil, p.enqueueIncomingMessagePersist(ctx, inbox.ID, msg) + } + return p.performPersistIncoming(ctx, inbox, msg) +} + +func validateIncomingMessage(p *IncomingPersister, inbox *model.Inbox, msg *channel.IncomingMessage) error { + if p == nil || p.db == nil || inbox == nil || msg == nil { + return nil + } if msg.SourceID == "" { - return nil, fmt.Errorf("incoming message missing source_id") + return fmt.Errorf("incoming message missing source_id") } senderID := msg.SenderID if senderID == "" { senderID = msg.ConversationID } if senderID == "" { - return nil, fmt.Errorf("incoming message missing sender_id") + return fmt.Errorf("incoming message missing sender_id") } if msg.Content == "" && len(msg.Attachments) == 0 { - return nil, fmt.Errorf("incoming message has no content or attachments") + return fmt.Errorf("incoming message has no content or attachments") + } + return nil +} + +func (p *IncomingPersister) performPersistIncoming(ctx context.Context, inbox *model.Inbox, msg *channel.IncomingMessage) (*IncomingPersistResult, error) { + if p == nil || p.db == nil || inbox == nil || msg == nil { + return nil, nil + } + senderID := msg.SenderID + if senderID == "" { + senderID = msg.ConversationID } var result IncomingPersistResult diff --git a/internal/handler/webhook/incoming_persister_jobs.go b/internal/handler/webhook/incoming_persister_jobs.go index 99f4231e..03e6369a 100644 --- a/internal/handler/webhook/incoming_persister_jobs.go +++ b/internal/handler/webhook/incoming_persister_jobs.go @@ -10,15 +10,22 @@ import ( "gorm.io/datatypes" "gorm.io/gorm" + "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/worker" ) const ( + TaskTypeProviderIncomingMessagePersist = "webhook:incoming_message_persist" TaskTypeProviderMessageStatusUpdate = "webhook:message_status_update" TaskTypeProviderContactMessagesStatusUpdate = "webhook:contact_messages_status_update" ) +type providerIncomingMessagePersistJob struct { + InboxID uint `json:"inbox_id"` + Message channel.IncomingMessage `json:"message"` +} + type providerMessageStatusUpdateJob struct { InboxID uint `json:"inbox_id"` SourceID string `json:"source_id"` @@ -38,10 +45,24 @@ func (p *IncomingPersister) registerJobs(wp *worker.WorkerPool) { if p == nil || wp == nil { return } + wp.Register(TaskTypeProviderIncomingMessagePersist, p.performIncomingMessagePersistJob) wp.Register(TaskTypeProviderMessageStatusUpdate, p.performMessageStatusUpdateJob) wp.Register(TaskTypeProviderContactMessagesStatusUpdate, p.performContactMessagesStatusUpdateJob) } +func (p *IncomingPersister) enqueueIncomingMessagePersist(ctx context.Context, inboxID uint, msg *channel.IncomingMessage) error { + if p == nil || p.worker == nil || inboxID == 0 || msg == nil { + return nil + } + queue := incomingMessageQueue(msg.ChannelType) + _, err := p.worker.Enqueue(ctx, TaskTypeProviderIncomingMessagePersist, providerIncomingMessagePersistJob{InboxID: inboxID, Message: *msg}, + worker.WithQueue(queue), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("webhook:incoming_message:%d:%s", inboxID, msg.SourceID)), + ) + return err +} + func (p *IncomingPersister) enqueueMessageStatusUpdate(ctx context.Context, inboxID uint, sourceID string, status model.MessageStatus, occurredAt *time.Time, externalError string) error { if p == nil || p.worker == nil || inboxID == 0 || sourceID == "" { return nil @@ -103,6 +124,28 @@ func (p *IncomingPersister) performMessageStatusUpdateJob(ctx context.Context, j return p.performMessageStatusUpdate(ctx, inbox, payload.SourceID, payload.Status, unixNanoToTime(payload.OccurredAtUnixNano), payload.ExternalError) } +func (p *IncomingPersister) performIncomingMessagePersistJob(ctx context.Context, job *model.BackgroundJob) error { + var payload providerIncomingMessagePersistJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal provider incoming message job: %w", err) + } + if payload.InboxID == 0 { + return fmt.Errorf("invalid provider incoming message job payload: %#v", payload) + } + inbox, err := p.loadInboxForStatusJob(ctx, payload.InboxID) + if err != nil { + return err + } + if inbox == nil { + return nil + } + if err := validateIncomingMessage(p, inbox, &payload.Message); err != nil { + return err + } + _, err = p.performPersistIncoming(ctx, inbox, &payload.Message) + return err +} + func (p *IncomingPersister) performContactMessagesStatusUpdateJob(ctx context.Context, job *model.BackgroundJob) error { var payload providerContactMessagesStatusUpdateJob if err := json.Unmarshal(job.Payload, &payload); err != nil { @@ -141,6 +184,15 @@ func validProviderMessageStatus(status model.MessageStatus) bool { } } +func incomingMessageQueue(channelType channel.ChannelType) string { + switch channelType { + case channel.ChannelWhatsApp, channel.ChannelTwilioSMS, channel.ChannelTwilioWA: + return "low" + default: + return model.DefaultBackgroundJobQueue + } +} + func validProviderMessageStatusTransition(current, next model.MessageStatus) bool { if !validProviderMessageStatus(next) || current == next { return validProviderMessageStatus(next) diff --git a/internal/handler/webhook/line_webhook.go b/internal/handler/webhook/line_webhook.go index aa33e166..8d585eb5 100644 --- a/internal/handler/webhook/line_webhook.go +++ b/internal/handler/webhook/line_webhook.go @@ -16,6 +16,7 @@ import ( linechannel "github.com/gochat/gochat/internal/channel/line" "github.com/gochat/gochat/internal/model" channelmodel "github.com/gochat/gochat/internal/model/channel" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" "github.com/gin-gonic/gin" @@ -42,6 +43,13 @@ func NewLineWebhookHandler(lineWebhook *linechannel.WebhookHandler, pipeline *li } } +func (h *LineWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *LineWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetWorkerPool(wp) + } + return h +} + // HandleLineWebhook processes an incoming LINE webhook Gin request. func (h *LineWebhookHandler) HandleLineWebhook(c *gin.Context) { lineChannelID := c.Param("line_channel_id") diff --git a/internal/handler/webhook/telegram_webhook.go b/internal/handler/webhook/telegram_webhook.go index 3e4290e5..1332c47e 100644 --- a/internal/handler/webhook/telegram_webhook.go +++ b/internal/handler/webhook/telegram_webhook.go @@ -13,6 +13,7 @@ import ( telegramchannel "github.com/gochat/gochat/internal/channel/telegram" "github.com/gochat/gochat/internal/model" channelmodel "github.com/gochat/gochat/internal/model/channel" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" "gorm.io/gorm" ) @@ -55,6 +56,13 @@ func NewTelegramWebhookHandler( } } +func (h *TelegramWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *TelegramWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetWorkerPool(wp) + } + return h +} + // HandleTelegramWebhook processes an incoming Telegram webhook Gin request. // URL pattern: /webhooks/telegram/:bot_token // Method: POST diff --git a/internal/handler/webhook/webhook_lookup_test.go b/internal/handler/webhook/webhook_lookup_test.go index 81100b53..b2696175 100644 --- a/internal/handler/webhook/webhook_lookup_test.go +++ b/internal/handler/webhook/webhook_lookup_test.go @@ -176,7 +176,7 @@ func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) { listener := &recordingListener{} dispatcher.Register(listener) wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low")) - persister := NewIncomingPersister(db, dispatcher).SetWorkerPool(wp) + persister := NewIncomingPersister(db, dispatcher) msg := &channel.IncomingMessage{ ChannelType: channel.ChannelTelegram, @@ -193,6 +193,7 @@ func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) { if err != nil { t.Fatalf("persist incoming: %v", err) } + persister.SetWorkerPool(wp) occurredAt := time.Now().UTC().Add(-time.Minute) if err := persister.UpdateMessageStatus(t.Context(), &inbox, "tg-status-job-1", model.MessageStatusDelivered, &occurredAt); err != nil { @@ -419,6 +420,100 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) { } } +func TestIncomingPersisterQueuesIncomingMessageWithWorker(t *testing.T) { + db := newWebhookLookupTestDB(t) + inbox := seedWebhookInbox(t, db, "telegram") + dispatcher := channel.NewDispatcher() + listener := &recordingListener{} + dispatcher.Register(listener) + wp := worker.NewWorkerPool(db) + persister := NewIncomingPersister(db, dispatcher).SetWorkerPool(wp) + + msg := &channel.IncomingMessage{ + ChannelType: channel.ChannelTelegram, + SourceID: "tg-inbound-job-1", + SenderID: "tg-inbound-user-1", + SenderName: "Inbound Worker User", + SenderType: channel.SenderContact, + Content: "persist me later", + ContentType: channel.ContentText, + InboxID: inbox.ID, + AccountID: inbox.AccountID, + } + result, err := persister.PersistIncoming(t.Context(), &inbox, msg) + if err != nil { + t.Fatalf("enqueue incoming: %v", err) + } + if result != nil { + t.Fatalf("expected async enqueue result to be nil, got %#v", result) + } + var queued model.BackgroundJob + if err := db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeProviderIncomingMessagePersist, model.DefaultBackgroundJobQueue, model.BackgroundJobStatusQueued).First(&queued).Error; err != nil { + t.Fatalf("expected queued incoming job: %v", err) + } + var beforeCount int64 + if err := db.Model(&model.Message{}).Where("source_id = ?", "tg-inbound-job-1").Count(&beforeCount).Error; err != nil { + t.Fatalf("count before messages: %v", err) + } + if beforeCount != 0 { + t.Fatalf("expected no message before worker, got %d", beforeCount) + } + + processed, err := wp.ProcessOne(t.Context()) + if err != nil || !processed { + t.Fatalf("process incoming job processed=%v err=%v", processed, err) + } + assertPersistedMessage(t, db, inbox.ID, "tg-inbound-job-1", "persist me later") + for _, eventType := range []channel.EventType{channel.EventContactCreated, channel.EventConversationCreated, channel.EventConversationOpened, channel.EventMessageCreated, channel.EventMessageIncoming} { + if !listenerSaw(listener, eventType) { + t.Fatalf("expected event %s, got %#v", eventType, listener.events) + } + } +} + +func TestIncomingPersisterIncomingMessageJobIsIdempotent(t *testing.T) { + db := newWebhookLookupTestDB(t) + inbox := seedWebhookInbox(t, db, "telegram") + wp := worker.NewWorkerPool(db) + persister := NewIncomingPersister(db).SetWorkerPool(wp) + msg := &channel.IncomingMessage{ + ChannelType: channel.ChannelTelegram, + SourceID: "tg-inbound-idempotent-1", + SenderID: "tg-inbound-idempotent-user", + SenderName: "Idempotent User", + SenderType: channel.SenderContact, + Content: "only once", + ContentType: channel.ContentText, + InboxID: inbox.ID, + AccountID: inbox.AccountID, + } + + if _, err := persister.PersistIncoming(t.Context(), &inbox, msg); err != nil { + t.Fatalf("enqueue first incoming: %v", err) + } + if _, err := persister.PersistIncoming(t.Context(), &inbox, msg); err != nil { + t.Fatalf("enqueue duplicate incoming: %v", err) + } + var jobCount int64 + if err := db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeProviderIncomingMessagePersist).Count(&jobCount).Error; err != nil { + t.Fatalf("count jobs: %v", err) + } + if jobCount != 1 { + t.Fatalf("expected one idempotent job, got %d", jobCount) + } + processed, err := wp.ProcessOne(t.Context()) + if err != nil || !processed { + t.Fatalf("process incoming job processed=%v err=%v", processed, err) + } + var messageCount int64 + if err := db.Model(&model.Message{}).Where("source_id = ?", "tg-inbound-idempotent-1").Count(&messageCount).Error; err != nil { + t.Fatalf("count messages: %v", err) + } + if messageCount != 1 { + t.Fatalf("expected one persisted message, got %d", messageCount) + } +} + func listenerSaw(listener *recordingListener, eventType channel.EventType) bool { for _, event := range listener.events { if event.Type == eventType { @@ -710,6 +805,56 @@ func TestTwilioWebhookPersistsIncomingMessage(t *testing.T) { assertPersistedMessage(t, db, inbox.ID, "SMIN1", "hello sms") } +func TestTwilioInboundSMSQueuesIncomingMessageWithWorker(t *testing.T) { + gin.SetMode(gin.TestMode) + db := newWebhookLookupTestDB(t) + inbox := seedWebhookInbox(t, db, "twilio_sms") + channelRecord := channelmodel.ChannelTwilioSMS{AccountID: 1, InboxID: inbox.ID, AccountSID: "AC123", PhoneNumber: "+15551234567"} + if err := db.Create(&channelRecord).Error; err != nil { + t.Fatalf("create twilio channel: %v", err) + } + + twilioRepo := twiliochannel.NewRepository(db) + twilioService := twiliochannel.NewTwilioService(twilioRepo) + twilioPipeline := twiliochannel.NewIncomingProcessor(twilioService) + twilioWebhook := twiliochannel.NewWebhookHandler(twilioPipeline, twilioService) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low")) + h := NewTwilioWebhookHandler(twilioWebhook, db).WithWorkerPool(wp) + r := gin.New() + r.POST("/webhooks/sms/:phone_number", h.HandleTwilioInboundSMS) + form := url.Values{} + form.Set("MessageSid", "SMINASYNC1") + form.Set("AccountSid", "AC123") + form.Set("From", "+15550002222") + form.Set("To", "+15551234567") + form.Set("Body", "queued sms") + req := httptest.NewRequest(http.MethodPost, "/webhooks/sms/+15551234567", strings.NewReader(form.Encode())) + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + w := httptest.NewRecorder() + + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String()) + } + var queued model.BackgroundJob + if err := db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeProviderIncomingMessagePersist, "low", model.BackgroundJobStatusQueued).First(&queued).Error; err != nil { + t.Fatalf("expected queued twilio incoming job: %v", err) + } + var beforeCount int64 + if err := db.Model(&model.Message{}).Where("source_id = ?", "SMINASYNC1").Count(&beforeCount).Error; err != nil { + t.Fatalf("count before messages: %v", err) + } + if beforeCount != 0 { + t.Fatalf("expected no message before worker, got %d", beforeCount) + } + processed, err := wp.ProcessOne(t.Context()) + if err != nil || !processed { + t.Fatalf("process twilio incoming job processed=%v err=%v", processed, err) + } + assertPersistedMessage(t, db, inbox.ID, "SMINASYNC1", "queued sms") +} + func TestTwilioDeliveryStatusUpdatesExistingMessage(t *testing.T) { gin.SetMode(gin.TestMode) db := newWebhookLookupTestDB(t)