From 28e72054104e8a59fd16480fc78cb83f4cc49cb9 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 7 Jun 2026 15:11:42 +0800 Subject: [PATCH] feat(search): index message delivery --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 10 ++-- internal/service/message_delivery_worker.go | 49 ++++++++++++++++--- .../service/message_delivery_worker_test.go | 11 +++++ internal/service/message_service.go | 3 +- 4 files changed, 62 insertions(+), 11 deletions(-) diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index 05ae8a39..f5d386de 100644 --- a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md +++ b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md @@ -49,11 +49,11 @@ Hermes task landing checklist: ## Current Baseline -- Current tracking checkpoint: 2026-06-07 P5.3f automation action search indexing, prepared as `feat(search): index automation actions`. -- Latest implementation checkpoint: this checkpoint, prepared as `feat(search): index automation actions`. -- Latest documentation/tooling checkpoint: this tracker update records Meilisearch reindex fan-out after automation, macro, and AgentBot action mutations. +- Current tracking checkpoint: 2026-06-07 P5.3g message delivery search indexing, prepared as `feat(search): index message delivery`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(search): index message delivery`. +- Latest documentation/tooling checkpoint: this tracker update records Meilisearch reindex fan-out after durable outgoing provider delivery success/failure mutations. - Plan landing status: complete for the current known Hermes plans and user-confirmed scope. Future work should update this file directly instead of opening a parallel tracker. -- Worktree status at this implementation checkpoint: automation, macro, and AgentBot actions now synchronize Meilisearch after creating searchable messages or mutating conversation/contact fields, including durable macro execution jobs and event-triggered automation listeners. This retains P5.3e conversation maintenance indexing, P5.3d conversation bulk-action indexing, P5.3c contact label search indexing, P3.103 contact bulk-action parity, and prior checkpoints. Live API/browser/enterprise smoke still needs the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack. +- Worktree status at this implementation checkpoint: durable outgoing message delivery now synchronizes Meilisearch after provider send success updates message `status`/`source_id` and after provider send failure updates `status`/`content_attributes.external_error`. This retains P5.3f automation action indexing, P5.3e conversation maintenance indexing, P5.3d conversation bulk-action indexing, P5.3c contact label search indexing, P3.103 contact bulk-action parity, and prior checkpoints. Live API/browser/enterprise smoke still needs the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack. - Next executable implementation checkpoint: continue Phase 2/3 drift audit for the next reused-frontend mismatch, or run B12 live smoke when the full PostgreSQL/Redis/Meilisearch/GoChat/Vite/Chrome stack is available. Re-run Phase 6 placeholder audit after future route/smoke changes. - `go test ./...` passes when run outside the restricted socket sandbox for the latest implementation baseline; the latest docs/tooling checkpoint verified `scripts/parity_frontend_smoke.sh --check` with workspace-local temp/cache dirs after `/tmp` was full. - Route dump succeeds with `972` registered routes after enterprise account route tracking. @@ -156,6 +156,7 @@ This table is the shortest authoritative handoff view. If an older lower section | Priority | Workstream | Current state | Next checkpoint | Commit close rule | | --- | --- | --- | --- | --- | +| 0 | P5.3g message delivery search indexing | Implemented for Meilisearch outgoing-delivery parity: durable `message:send_reply` replay updates the search document after provider success writes `status`/`source_id` and after provider failure writes `status` plus `content_attributes.external_error`; `MessageService` now propagates its search indexer into already-registered delivery workers. | Keep in Review; reopen from live Meilisearch gate, outgoing delivery smoke, or a fresh delivery/status mutation path that changes searchable message fields without indexing. | Focused message delivery worker search tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | | 0 | P5.3f automation action search indexing | Implemented for Meilisearch automation/macro parity: `send_message`, `send_attachment`, and `add_private_note` index created messages plus parent conversations; status, priority, assignee, team, label, SLA, and mute actions reindex the updated conversation; mute also reindexes the blocked contact; macro worker replay and event-triggered automation/AgentBot listeners receive the durable search indexer during bootstrap. | Keep in Review; reopen from live Meilisearch gate, automation/macro smoke, or a fresh action path that mutates searchable conversation/message/contact fields without indexing. | Focused automation action/search propagation tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | | 0 | P5.3e conversation maintenance search indexing | Implemented for Meilisearch maintenance-job parity: scheduled snooze reopen and account auto-resolution reindex updated conversations, message-status maintenance reindexes changed messages plus the parent conversation, and bootstrap registers search-aware maintenance handlers after the durable search indexer is wired. | Keep in Review; reopen from live Meilisearch gate, scheduled maintenance smoke, or a fresh background mutation path that changes searchable conversation/message fields without indexing. | Focused conversation maintenance search tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | | 0 | P5.3d conversation bulk-action search indexing | Implemented for Meilisearch conversation bulk-action parity: `conversation:bulk_action` now reindexes updated account conversations after status, assignee, team, snooze, or label mutations, and the app registers a search-aware bulk-action runner after the durable search indexer is wired. | Keep in Review; reopen from live Meilisearch gate, conversation bulk-action search smoke, or a fresh mutation path that changes searchable conversation fields without indexing. | Focused conversation bulk-action search test passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | @@ -2806,3 +2807,4 @@ Verification milestone gates: - 2026-06-07: P5.3d conversation bulk-action search-index checkpoint prepared as `feat(search): index conversation bulk actions`; audited the Chatwoot account `BulkActionsJob` mutation path and the P5.3 Meilisearch durability contract. GoChat conversation bulk-action worker replay now reindexes every updated account-scoped conversation after bulk status, assignee, team, snooze, or label updates, and bootstrap registers the search-aware bulk-action runner after the durable search indexer is wired. Focused conversation bulk-action search test passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. - 2026-06-07: P5.3e conversation maintenance search-index checkpoint prepared as `feat(search): index conversation maintenance`; audited Chatwoot scheduled conversation maintenance jobs and the P5.3 Meilisearch durability contract. GoChat scheduled snooze reopen and account auto-resolution jobs now reindex updated conversations, deferred outgoing message-status jobs reindex updated messages plus their parent conversation, and bootstrap registers the search-aware maintenance runner after the durable search indexer is wired. Focused conversation maintenance search tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. - 2026-06-07: P5.3f automation action search-index checkpoint prepared as `feat(search): index automation actions`; audited GoChat automation, macro, AgentBot, and durable macro execution action paths against the P5.3 Meilisearch durability contract. GoChat shared action execution now reindexes created outgoing/private/attachment messages plus parent conversations, reindexes conversations after status/priority/assignment/team/label/SLA/mute mutations, reindexes contacts after mute blocks them, keeps conversation label strings current for action label changes, and wires durable search indexing into automation listeners, macro services, AgentBot rules, and macro worker replay. Focused automation action/search propagation tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. +- 2026-06-07: P5.3g message delivery search-index checkpoint prepared as `feat(search): index message delivery`; audited GoChat durable `message:send_reply` worker against the P5.3 Meilisearch durability contract. GoChat now propagates `MessageService` search indexing into message delivery runners registered before the indexer exists, and the worker reindexes messages after provider success persists `sent`/`source_id` or provider failure persists `failed`/`external_error`. Focused message delivery worker search tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. diff --git a/internal/service/message_delivery_worker.go b/internal/service/message_delivery_worker.go index 143ae13e..13751f92 100644 --- a/internal/service/message_delivery_worker.go +++ b/internal/service/message_delivery_worker.go @@ -9,6 +9,7 @@ import ( "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/worker" + "gorm.io/datatypes" "gorm.io/gorm" ) @@ -24,13 +25,29 @@ var messageDeliveryRegistrations sync.Map // 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) { + registerMessageDeliveryJobs(wp, db, dispatcher, nil) +} + +func RegisterMessageDeliverySearchIndexer(wp *worker.WorkerPool, db *gorm.DB, dispatcher *channel.Dispatcher, indexer SearchIndexer) { + registerMessageDeliveryJobs(wp, db, dispatcher, indexer) +} + +func registerMessageDeliveryJobs(wp *worker.WorkerPool, db *gorm.DB, dispatcher *channel.Dispatcher, indexer SearchIndexer) { if wp == nil || db == nil { return } - if _, loaded := messageDeliveryRegistrations.LoadOrStore(wp, struct{}{}); loaded { + runner := &messageDeliveryRunner{db: db, dispatcher: dispatcher, searchIndexer: indexer} + actual, loaded := messageDeliveryRegistrations.LoadOrStore(wp, runner) + if loaded { + registered := actual.(*messageDeliveryRunner) + if dispatcher != nil { + registered.dispatcher = dispatcher + } + if indexer != nil { + registered.searchIndexer = indexer + } return } - runner := &messageDeliveryRunner{db: db, dispatcher: dispatcher} wp.Register(TaskTypeMessageSendReply, runner.performSendReply) } @@ -49,8 +66,9 @@ func EnqueueSendReply(ctx context.Context, wp *worker.WorkerPool, messageID uint } type messageDeliveryRunner struct { - db *gorm.DB - dispatcher *channel.Dispatcher + db *gorm.DB + dispatcher *channel.Dispatcher + searchIndexer SearchIndexer } func (r *messageDeliveryRunner) performSendReply(ctx context.Context, job *model.BackgroundJob) error { @@ -125,7 +143,13 @@ func (r *messageDeliveryRunner) updateMessageDeliverySuccess(ctx context.Context 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) } + message.Status = string(model.MessageStatusSent) + message.ContentAttributes = updates["content_attributes"].(datatypes.JSON) + if result != nil && result.ExternalID != "" { + message.SourceID = result.ExternalID + } r.dispatchMessageStatus(ctx, message, string(model.MessageStatusSent)) + r.indexMessage(ctx, message) return nil } @@ -134,9 +158,22 @@ func (r *messageDeliveryRunner) updateMessageDeliveryFailure(ctx context.Context "status": string(model.MessageStatusFailed), "content_attributes": setMessageExternalError(message.ContentAttributes, string(model.MessageStatusFailed), sendErr.Error()), } - return r.db.WithContext(ctx).Model(message). + if err := r.db.WithContext(ctx).Model(message). Where("id = ?", message.ID). - Updates(updates).Error + Updates(updates).Error; err != nil { + return err + } + message.Status = string(model.MessageStatusFailed) + message.ContentAttributes = updates["content_attributes"].(datatypes.JSON) + r.indexMessage(ctx, message) + return nil +} + +func (r *messageDeliveryRunner) indexMessage(ctx context.Context, message *model.Message) { + if r.searchIndexer == nil || message == nil { + return + } + logSearchIndexError("message", message.ID, r.searchIndexer.IndexMessage(ctx, message)) } func (r *messageDeliveryRunner) dispatchMessageStatus(ctx context.Context, message *model.Message, status string) { diff --git a/internal/service/message_delivery_worker_test.go b/internal/service/message_delivery_worker_test.go index 8491f1aa..688ab197 100644 --- a/internal/service/message_delivery_worker_test.go +++ b/internal/service/message_delivery_worker_test.go @@ -106,6 +106,8 @@ func TestMessageDeliveryWorker_CreateOutgoingQueuesSendReply(t *testing.T) { MessageType: "outgoing", }) require.NoError(t, err) + indexer := &recordingDurableSearchIndexer{} + svc.SetSearchIndexer(indexer) var queued model.BackgroundJob require.NoError(t, db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeMessageSendReply, "high", model.BackgroundJobStatusQueued).First(&queued).Error) @@ -118,6 +120,10 @@ func TestMessageDeliveryWorker_CreateOutgoingQueuesSendReply(t *testing.T) { require.NoError(t, err) require.Equal(t, "external-message-1", updated.SourceID) require.Equal(t, string(model.MessageStatusSent), updated.Status) + require.Len(t, indexer.indexedMessages, 1) + require.Equal(t, message.ID, indexer.indexedMessages[0].ID) + require.Equal(t, "external-message-1", indexer.indexedMessages[0].SourceID) + require.Equal(t, string(model.MessageStatusSent), indexer.indexedMessages[0].Status) provider.mu.Lock() require.Equal(t, 1, provider.count) @@ -164,6 +170,8 @@ func TestMessageDeliveryWorker_RetriesProviderFailuresAndMarksMessageFailed(t *t worker.WithBackoff(func(attempt int) time.Duration { return time.Minute }), ) RegisterMessageDeliveryJobs(wp, db, nil) + indexer := &recordingDurableSearchIndexer{} + RegisterMessageDeliverySearchIndexer(wp, db, nil, indexer) account := createTestAccount(t, db) inbox := createTestInbox(t, db, account.ID, string(testSendReplyChannel)) @@ -190,6 +198,9 @@ func TestMessageDeliveryWorker_RetriesProviderFailuresAndMarksMessageFailed(t *t attrs := map[string]any{} require.NoError(t, json.Unmarshal(updated.ContentAttributes, &attrs)) require.Equal(t, "provider rejected message", attrs["external_error"]) + require.Len(t, indexer.indexedMessages, 1) + require.Equal(t, message.ID, indexer.indexedMessages[0].ID) + require.Equal(t, string(model.MessageStatusFailed), indexer.indexedMessages[0].Status) } func TestMessageDeliveryWorker_LoadsMissingMessagesAsRetryableFailures(t *testing.T) { diff --git a/internal/service/message_service.go b/internal/service/message_service.go index 7d072074..1c088217 100644 --- a/internal/service/message_service.go +++ b/internal/service/message_service.go @@ -36,11 +36,12 @@ func NewMessageService(repo *repository.MessageRepo, dispatcher *channel.Dispatc func (s *MessageService) SetSearchIndexer(indexer SearchIndexer) { s.searchIndexer = indexer + RegisterMessageDeliverySearchIndexer(s.worker, s.repo.DB(), s.dispatcher, indexer) } func (s *MessageService) SetWorkerPool(wp *worker.WorkerPool) { s.worker = wp - RegisterMessageDeliveryJobs(wp, s.repo.DB(), s.dispatcher) + RegisterMessageDeliverySearchIndexer(wp, s.repo.DB(), s.dispatcher, s.searchIndexer) } func (s *MessageService) DB() *gorm.DB {