diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 4932c75f..20edcc6f 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 `cda7d02 feat(conversations): queue bulk actions`. -- Latest implementation checkpoint: `cda7d02 feat(conversations): queue bulk actions`. +- Current tracking checkpoint: 2026-06-05 after `cda7d02 feat(conversations): queue bulk actions`, with this checkpoint prepared as `feat(messages): queue send replies`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(messages): queue send replies`. - 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.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.10 outbound message delivery, followed by P5.9 inbound webhook deferral, 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.10 now queues Chatwoot `SendReplyJob`-style outbound message delivery through durable `message:send_reply` 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.10 provider delivery-status job depth, followed by P5.9 inbound webhook deferral, remaining P5.11 Captain/Copilot jobs, and P5.13 analytics aggregation. - `go test ./...` passes. - Route dump succeeds with `TOTAL: 830` after adding the Chatwoot-compatible applied-SLA index route. - Route parity artifacts now exist under `docs/parity/` and are generated by `cmd/route_parity`. @@ -90,7 +90,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.10 outbound message delivery | Not started after durable worker foundations. Existing outgoing message creation still needs Chatwoot `SendReplyJob` separation and duplicate-send review. | Queue outbound provider delivery and delivery-status updates through durable jobs. | Commit code, tests, P5.10 table row, commit ledger row, and progress log together. | +| 1 | P5.10 outbound message delivery | SendReplyJob-style durable provider delivery is implemented; provider delivery-status job depth remains open. | Queue provider delivery-status updates where Chatwoot defers them and keep retry/status transitions idempotent. | Commit code, tests, P5.10 table row, commit ledger row, and progress log together. | | 2 | 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. | | 3 | 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 ./...`. | | 4 | 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. | @@ -117,6 +117,7 @@ This ledger records the committed parity checkpoints that future slices should b | Commit | Scope | Verification summary | Follow-up state | | --- | --- | --- | --- | +| `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. | Move P5.10 send-reply core to Review; continue P5.10 provider delivery-status job depth, then P5.9 inbound webhook deferral, P5.11 Captain/Copilot remaining jobs, and P5.13 analytics aggregation. | | `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. | @@ -1608,7 +1609,7 @@ Reference job families to keep in scope: 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_service.go`, `internal/channel/registry.go`, and provider outbound listeners remain the next P5.10 focus for Chatwoot `SendReplyJob` parity and duplicate-send prevention. +- `internal/service/message_delivery_worker.go`, provider delivery-status webhook services, and message status handlers remain the P5.10 focus. SendReplyJob-style outbound delivery is queued; provider delivery-status job depth remains open. - `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/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. @@ -1640,7 +1641,7 @@ Tracking table: | 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.10 | Queue outbound message delivery and delivery-status updates. | `send_reply_job.rb`, provider delivery/status jobs | message send/channel services, delivery status handler | Outgoing message creation and provider delivery are separated; retries update message/delivery status exactly once. | Todo | +| P5.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. | Doing: send-reply core Review by `feat(messages): queue send replies`; provider delivery-status 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: 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` | | 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 | @@ -1767,6 +1768,25 @@ env TMPDIR=/home/rogee/Projects/gochat/.tmp/test-tmp GOCACHE=/tmp/gochat-gocache 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. +- Worker replay reloads the current message, conversation, inbox, and contact by account scope before calling the registered channel provider through `channel.SendMessageToChannel`. +- The job skips already provider-originated messages with `source_id`, private messages, voice-call bubbles, and non-outgoing/non-template messages, matching `Base::SendOnChannelService#invalid_message?` and preventing duplicate external sends. +- Successful provider sends persist the returned external message ID into `messages.source_id` and clear any stale `external_error` from `content_attributes` while keeping status `sent`. +- Provider failures mark the message `failed`, persist `content_attributes.external_error`, and return an error so durable retry/backoff/dead-letter state remains observable in `background_jobs`. +- Bootstrap wires `MessageService.SetWorkerPool(workerPool)`, so production outgoing replies are replayable after process restart. +- Remaining P5.10 work: queue provider delivery-status updates where Chatwoot defers status callbacks, then prove read/delivered/failed transitions are idempotent and do not downgrade `read` to `delivered`. + +P5.10 verification: + +```bash +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service -run 'MessageDeliveryWorker\|MessageService' -count=1 +env GOCACHE=/tmp/gochat-gocache GOMODCACHE=/tmp/gochat-gomodcache go test ./internal/service ./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.11 current checkpoint: - `captain:document_sync` now mirrors Chatwoot enterprise `Captain::Documents::PerformSyncJob` on the `low` queue. `POST /captain/documents/:id/sync` marks the account-scoped document as syncing and enqueues durable replay when a WorkerPool is configured. @@ -1912,6 +1932,7 @@ Verification milestone gates: ## Progress Log +- 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. Next slice is P5.10 provider delivery-status job depth, then P5.9 inbound webhook deferral. - 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`. - 2026-06-04: Phase 1 search foundation added: Meilisearch config/env defaults, `SearchEngine` contract, Meilisearch HTTP wrapper with bootstrap/settings, DB fallback adapter, document builders, reindex command, and no-live-Meilisearch tests. Verified `go test ./...` in unsandboxed mode because miniredis/httptest need local sockets; route dump still reports `TOTAL: 704`. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index 6968f7a4..efc7d638 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -536,6 +536,7 @@ func Bootstrap(env string) (*App, error) { }) messageService := service.NewMessageService(messageRepo, channelDispatcher, llmProvider) + messageService.SetWorkerPool(workerPool) notificationService := service.NewNotificationService(db, notificationRepo, notificationPrefRepo) // M2 — Tag+Label services diff --git a/internal/service/message_delivery_worker.go b/internal/service/message_delivery_worker.go new file mode 100644 index 00000000..143ae13e --- /dev/null +++ b/internal/service/message_delivery_worker.go @@ -0,0 +1,154 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/gochat/gochat/internal/channel" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "gorm.io/gorm" +) + +const TaskTypeMessageSendReply = "message:send_reply" + +type sendReplyJob struct { + MessageID uint `json:"message_id"` +} + +var messageDeliveryRegistrations sync.Map + +// RegisterMessageDeliveryJobs wires Chatwoot's SendReplyJob equivalent into +// the durable worker. Outgoing provider calls stay behind channel providers so +// tests can fake delivery without real external APIs. +func RegisterMessageDeliveryJobs(wp *worker.WorkerPool, db *gorm.DB, dispatcher *channel.Dispatcher) { + if wp == nil || db == nil { + return + } + if _, loaded := messageDeliveryRegistrations.LoadOrStore(wp, struct{}{}); loaded { + return + } + runner := &messageDeliveryRunner{db: db, dispatcher: dispatcher} + wp.Register(TaskTypeMessageSendReply, runner.performSendReply) +} + +func EnqueueSendReply(ctx context.Context, wp *worker.WorkerPool, messageID uint) (*model.BackgroundJob, error) { + if wp == nil { + return nil, nil + } + if messageID == 0 { + return nil, fmt.Errorf("invalid send reply payload: message_id=%d", messageID) + } + return wp.Enqueue(ctx, TaskTypeMessageSendReply, sendReplyJob{MessageID: messageID}, + worker.WithQueue("high"), + worker.WithMaxAttempts(3), + worker.WithIdempotencyKey(fmt.Sprintf("message:send_reply:%d", messageID)), + ) +} + +type messageDeliveryRunner struct { + db *gorm.DB + dispatcher *channel.Dispatcher +} + +func (r *messageDeliveryRunner) performSendReply(ctx context.Context, job *model.BackgroundJob) error { + var payload sendReplyJob + if err := json.Unmarshal(job.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal send reply job: %w", err) + } + if payload.MessageID == 0 { + return fmt.Errorf("invalid send reply job payload: %#v", payload) + } + + var message model.Message + if err := r.db.WithContext(ctx).First(&message, payload.MessageID).Error; err != nil { + return fmt.Errorf("load send reply message %d: %w", payload.MessageID, err) + } + if !deliverableReplyMessage(message) { + return nil + } + + var conversation model.Conversation + if err := r.db.WithContext(ctx). + Where("id = ? AND account_id = ?", message.ConversationID, message.AccountID). + First(&conversation).Error; err != nil { + return fmt.Errorf("load send reply conversation %d: %w", message.ConversationID, err) + } + + var inbox model.Inbox + if err := r.db.WithContext(ctx). + Where("id = ? AND account_id = ?", message.InboxID, message.AccountID). + First(&inbox).Error; err != nil { + return fmt.Errorf("load send reply inbox %d: %w", message.InboxID, err) + } + + var contact model.Contact + if err := r.db.WithContext(ctx). + Where("id = ? AND account_id = ?", conversation.ContactID, message.AccountID). + First(&contact).Error; err != nil { + return fmt.Errorf("load send reply contact %d: %w", conversation.ContactID, err) + } + + result, err := channel.SendMessageToChannel(ctx, channel.ChannelType(inbox.ChannelType), &inbox, &message, &contact) + if err != nil { + _ = r.updateMessageDeliveryFailure(ctx, &message, err) + return fmt.Errorf("send reply message %d: %w", message.ID, err) + } + if err := r.updateMessageDeliverySuccess(ctx, &message, result); err != nil { + return err + } + return nil +} + +func deliverableReplyMessage(message model.Message) bool { + if message.SourceID != "" || message.Private || message.ContentType == "voice_call" { + return false + } + switch message.MessageType { + case string(model.MessageTypeOutgoing), string(model.MessageTypeTemplate): + return true + default: + return false + } +} + +func (r *messageDeliveryRunner) updateMessageDeliverySuccess(ctx context.Context, message *model.Message, result *channel.SendResult) error { + updates := map[string]any{ + "status": string(model.MessageStatusSent), + "content_attributes": setMessageExternalError(message.ContentAttributes, string(model.MessageStatusSent), ""), + } + if result != nil && result.ExternalID != "" { + updates["source_id"] = result.ExternalID + } + if err := r.db.WithContext(ctx).Model(&model.Message{}).Where("id = ?", message.ID).Updates(updates).Error; err != nil { + return fmt.Errorf("update sent message %d: %w", message.ID, err) + } + r.dispatchMessageStatus(ctx, message, string(model.MessageStatusSent)) + return nil +} + +func (r *messageDeliveryRunner) updateMessageDeliveryFailure(ctx context.Context, message *model.Message, sendErr error) error { + updates := map[string]any{ + "status": string(model.MessageStatusFailed), + "content_attributes": setMessageExternalError(message.ContentAttributes, string(model.MessageStatusFailed), sendErr.Error()), + } + return r.db.WithContext(ctx).Model(message). + Where("id = ?", message.ID). + Updates(updates).Error +} + +func (r *messageDeliveryRunner) dispatchMessageStatus(ctx context.Context, message *model.Message, status string) { + if r.dispatcher == nil { + return + } + event := channel.NewChannelEvent(channel.EventMessageStatusUpdated, channel.ChannelAPI, message.AccountID, message.InboxID) + event.ConversationID = message.ConversationID + if message.SenderID != nil { + event.UserID = *message.SenderID + } + event.Data["message_id"] = message.ID + event.Data["status"] = status + _ = r.dispatcher.Dispatch(ctx, event) +} diff --git a/internal/service/message_delivery_worker_test.go b/internal/service/message_delivery_worker_test.go new file mode 100644 index 00000000..8491f1aa --- /dev/null +++ b/internal/service/message_delivery_worker_test.go @@ -0,0 +1,212 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "github.com/gochat/gochat/internal/channel" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "github.com/stretchr/testify/require" +) + +const testSendReplyChannel channel.ChannelType = "send_reply_test" + +var ( + testSendReplyProviderInstance = &testSendReplyProvider{} + testSendReplyRegisterOnce sync.Once +) + +type testSendReplyProvider struct { + mu sync.Mutex + count int + err error + externalID string + messageID uint + contactID uint +} + +func registerTestSendReplyProvider() *testSendReplyProvider { + testSendReplyRegisterOnce.Do(func() { + _ = channel.Register(testSendReplyProviderInstance) + }) + testSendReplyProviderInstance.reset() + return testSendReplyProviderInstance +} + +func (p *testSendReplyProvider) reset() { + p.mu.Lock() + defer p.mu.Unlock() + p.count = 0 + p.err = nil + p.externalID = "external-message-1" + p.messageID = 0 + p.contactID = 0 +} + +func (p *testSendReplyProvider) Type() channel.ChannelType { return testSendReplyChannel } +func (p *testSendReplyProvider) Name() string { return "Send Reply Test" } +func (p *testSendReplyProvider) Description() string { return "test provider" } +func (p *testSendReplyProvider) ConfigSchema() *channel.ConfigSchemaDefinition { + return &channel.ConfigSchemaDefinition{Type: "object"} +} +func (p *testSendReplyProvider) ValidateConfig(ctx context.Context, config channel.ChannelConfig) error { + return nil +} +func (p *testSendReplyProvider) DefaultConfig() channel.ChannelConfig { return channel.ChannelConfig{} } +func (p *testSendReplyProvider) OnCreate(ctx context.Context, inbox *model.Inbox, config channel.ChannelConfig) (channel.ChannelConfig, error) { + return config, nil +} +func (p *testSendReplyProvider) OnDestroy(ctx context.Context, inbox *model.Inbox, config channel.ChannelConfig) error { + return nil +} +func (p *testSendReplyProvider) ProcessIncoming(ctx context.Context, inbox *model.Inbox, rawPayload []byte) (*channel.IncomingMessage, error) { + return nil, nil +} +func (p *testSendReplyProvider) ValidateWebhookRequest(ctx context.Context, inbox *model.Inbox, request *channel.WebhookRequest) error { + return nil +} +func (p *testSendReplyProvider) SendMessage(ctx context.Context, inbox *model.Inbox, message *model.Message, contact *model.Contact) (*channel.SendResult, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.count++ + p.messageID = message.ID + p.contactID = contact.ID + if p.err != nil { + return nil, p.err + } + return &channel.SendResult{ExternalID: p.externalID, DeliveredAt: time.Now()}, nil +} +func (p *testSendReplyProvider) GetContactProfile(ctx context.Context, inbox *model.Inbox, contactSource string) (*channel.ContactProfile, error) { + return nil, nil +} +func (p *testSendReplyProvider) Capabilities() channel.ChannelCapabilities { + return channel.ChannelCapabilities{SupportsDeliveryStatus: true} +} + +func TestMessageDeliveryWorker_CreateOutgoingQueuesSendReply(t *testing.T) { + provider := registerTestSendReplyProvider() + db, repo, _, svc := setupMessageServiceWithDefaultLLM(t) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 20, 0, 0, 0, time.UTC) })) + svc.SetWorkerPool(wp) + + account := createTestAccount(t, db) + user := createTestUser(t, db, account.ID) + inbox := createTestInbox(t, db, account.ID, string(testSendReplyChannel)) + contact := createTestContact(t, db, account.ID) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + + message, err := svc.Create(context.Background(), account.ID, user.ID, CreateMessageRequest{ + ConversationID: conversation.ID, + Content: "hello from agent", + MessageType: "outgoing", + }) + require.NoError(t, err) + + var queued model.BackgroundJob + require.NoError(t, db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeMessageSendReply, "high", model.BackgroundJobStatusQueued).First(&queued).Error) + + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + require.True(t, processed) + + updated, err := repo.FindByID(context.Background(), message.ID) + require.NoError(t, err) + require.Equal(t, "external-message-1", updated.SourceID) + require.Equal(t, string(model.MessageStatusSent), updated.Status) + + provider.mu.Lock() + require.Equal(t, 1, provider.count) + require.Equal(t, message.ID, provider.messageID) + require.Equal(t, contact.ID, provider.contactID) + provider.mu.Unlock() + +} + +func TestMessageDeliveryWorker_SkipsMessagesAlreadySentToProvider(t *testing.T) { + provider := registerTestSendReplyProvider() + db := setupServiceTestDB(t) + wp := worker.NewWorkerPoolWithOptions(db, worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 20, 15, 0, 0, time.UTC) })) + RegisterMessageDeliveryJobs(wp, db, nil) + + account := createTestAccount(t, db) + inbox := createTestInbox(t, db, account.ID, string(testSendReplyChannel)) + contact := createTestContact(t, db, account.ID) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + message := createTestMessage(t, db, account.ID, inbox.ID, conversation.ID, func(m *model.Message) { + m.MessageType = string(model.MessageTypeOutgoing) + m.SourceID = "existing-external-id" + }) + + _, err := EnqueueSendReply(context.Background(), wp, message.ID) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.NoError(t, err) + require.True(t, processed) + + provider.mu.Lock() + require.Equal(t, 0, provider.count) + provider.mu.Unlock() +} + +func TestMessageDeliveryWorker_RetriesProviderFailuresAndMarksMessageFailed(t *testing.T) { + provider := registerTestSendReplyProvider() + provider.mu.Lock() + provider.err = errors.New("provider rejected message") + provider.mu.Unlock() + db := setupServiceTestDB(t) + wp := worker.NewWorkerPoolWithOptions(db, + worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 20, 30, 0, 0, time.UTC) }), + worker.WithBackoff(func(attempt int) time.Duration { return time.Minute }), + ) + RegisterMessageDeliveryJobs(wp, db, nil) + + account := createTestAccount(t, db) + inbox := createTestInbox(t, db, account.ID, string(testSendReplyChannel)) + contact := createTestContact(t, db, account.ID) + conversation := createTestConversation(t, db, account.ID, inbox.ID, contact.ID) + message := createTestMessage(t, db, account.ID, inbox.ID, conversation.ID, func(m *model.Message) { + m.MessageType = string(model.MessageTypeOutgoing) + m.Status = string(model.MessageStatusSent) + }) + + _, err := EnqueueSendReply(context.Background(), wp, message.ID) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.Error(t, err) + require.True(t, processed) + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeMessageSendReply).First(&job).Error) + require.Equal(t, model.BackgroundJobStatusRetrying, job.Status) + + var updated model.Message + require.NoError(t, db.First(&updated, message.ID).Error) + require.Equal(t, string(model.MessageStatusFailed), updated.Status) + attrs := map[string]any{} + require.NoError(t, json.Unmarshal(updated.ContentAttributes, &attrs)) + require.Equal(t, "provider rejected message", attrs["external_error"]) +} + +func TestMessageDeliveryWorker_LoadsMissingMessagesAsRetryableFailures(t *testing.T) { + db := setupServiceTestDB(t) + wp := worker.NewWorkerPoolWithOptions(db, + worker.WithNow(func() time.Time { return time.Date(2026, 6, 5, 20, 45, 0, 0, time.UTC) }), + worker.WithBackoff(func(attempt int) time.Duration { return time.Minute }), + ) + RegisterMessageDeliveryJobs(wp, db, nil) + + _, err := EnqueueSendReply(context.Background(), wp, 9999) + require.NoError(t, err) + processed, err := wp.ProcessOne(context.Background()) + require.Error(t, err) + require.True(t, processed) + + var job model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeMessageSendReply).First(&job).Error) + require.Equal(t, model.BackgroundJobStatusRetrying, job.Status) +} diff --git a/internal/service/message_service.go b/internal/service/message_service.go index 459ef961..e3891253 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -11,6 +11,7 @@ import ( "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" "github.com/gochat/gochat/internal/search" + "github.com/gochat/gochat/internal/worker" applogger "github.com/gochat/gochat/pkg/logger" pkgvalidator "github.com/gochat/gochat/pkg/validator" @@ -25,6 +26,7 @@ type MessageService struct { dispatcher *channel.Dispatcher searchIndexer SearchIndexer llmProvider llm.Provider + worker *worker.WorkerPool } // NewMessageService creates a new Message service. @@ -36,6 +38,11 @@ func (s *MessageService) SetSearchIndexer(indexer SearchIndexer) { s.searchIndexer = indexer } +func (s *MessageService) SetWorkerPool(wp *worker.WorkerPool) { + s.worker = wp + RegisterMessageDeliveryJobs(wp, s.repo.DB(), s.dispatcher) +} + func (s *MessageService) DB() *gorm.DB { if s == nil || s.repo == nil { return nil @@ -206,7 +213,13 @@ func (s *MessageService) Create(ctx context.Context, accountID uint, userID uint if req.MessageType == "incoming" { s.dispatchMessageEvent(ctx, channel.EventMessageIncoming, message) } else if req.MessageType == "outgoing" { - s.dispatchMessageEvent(ctx, channel.EventMessageOutgoing, message) + if s.worker != nil { + if _, err := EnqueueSendReply(ctx, s.worker, message.ID); err != nil { + return message, err + } + } else { + s.dispatchMessageEvent(ctx, channel.EventMessageOutgoing, message) + } } return message, nil