From 8eabadc3fcd21b37db855b3ad86f41fee13f1f52 Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 7 Jun 2026 15:23:29 +0800 Subject: [PATCH] feat(search): index provider webhooks --- docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md | 10 ++- internal/app/bootstrap.go | 6 ++ internal/handler/webhook/facebook_webhook.go | 7 ++ .../handler/webhook/incoming_persister.go | 84 ++++++++++++++++++- internal/handler/webhook/line_webhook.go | 7 ++ internal/handler/webhook/telegram_webhook.go | 7 ++ internal/handler/webhook/tiktok_webhook.go | 7 ++ internal/handler/webhook/twilio_webhook.go | 7 ++ .../handler/webhook/webhook_lookup_test.go | 54 ++++++++++-- internal/handler/webhook/whatsapp_webhook.go | 7 ++ 10 files changed, 183 insertions(+), 13 deletions(-) diff --git a/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md b/docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md index f5d386de..cc5f9ef4 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.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. +- Current tracking checkpoint: 2026-06-07 P5.3h provider webhook search indexing, prepared as `feat(search): index provider webhooks`. +- Latest implementation checkpoint: this checkpoint, prepared as `feat(search): index provider webhooks`. +- Latest documentation/tooling checkpoint: this tracker update records Meilisearch reindex fan-out after provider webhook incoming message persistence and delivery/read status 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: 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. +- Worktree status at this implementation checkpoint: provider webhook persistence now synchronizes Meilisearch after incoming contact/conversation/message creates or updates, queued incoming webhook replay, direct provider message-status updates, and queued contact-wide delivery/read status updates. This retains P5.3g message delivery indexing, 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.3h provider webhook search indexing | Implemented for Meilisearch provider-webhook parity: `IncomingPersister` indexes contacts, conversations, and messages after non-duplicate incoming persistence; indexes direct and durable provider message status updates; indexes every message touched by contact-wide status/read receipts; Facebook, Telegram, WhatsApp, TikTok, LINE, and Twilio webhook handlers receive the durable search indexer from bootstrap. | Keep in Review; reopen from live Meilisearch gate, provider webhook smoke, or a fresh incoming/status mutation path that changes searchable contact/conversation/message fields without indexing. | Focused webhook persister search tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. | | 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. | @@ -2808,3 +2809,4 @@ Verification milestone gates: - 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. +- 2026-06-07: P5.3h provider webhook search-index checkpoint prepared as `feat(search): index provider webhooks`; audited GoChat `IncomingPersister` incoming persistence, provider message status, contact-wide read receipt jobs, and provider webhook handler bootstrap wiring against the P5.3 Meilisearch durability contract. GoChat now indexes contacts/conversations/messages after provider incoming message persistence, skips duplicate indexing, indexes durable incoming replay, indexes direct/durable provider delivery status updates, indexes every contact-wide read-receipt message mutation, and wires Facebook/Telegram/WhatsApp/TikTok/LINE/Twilio webhook handlers to the durable search indexer. Focused webhook persister search tests passed; full `go test ./...` passed outside the restricted socket sandbox; `git diff --check` passed. No route artifacts change. diff --git a/internal/app/bootstrap.go b/internal/app/bootstrap.go index dad67571..b215f9e5 100644 --- a/internal/app/bootstrap.go +++ b/internal/app/bootstrap.go @@ -682,6 +682,12 @@ func Bootstrap(env string) (*App, error) { automationRuleService.SetSearchIndexer(searchIndexer) macroService.SetSearchIndexer(searchIndexer) botRuleService.SetSearchIndexer(searchIndexer) + facebookWebhookHandler.WithSearchIndexer(searchIndexer) + telegramWebhookHandler.WithSearchIndexer(searchIndexer) + whatsappWebhookHandler.WithSearchIndexer(searchIndexer) + tiktokWebhookHandler.WithSearchIndexer(searchIndexer) + lineWebhookHandler.WithSearchIndexer(searchIndexer) + twilioWebhookHandler.WithSearchIndexer(searchIndexer) // Event-triggered automation/AgentBot listeners are registered after the durable // search indexer exists so action side effects keep Meilisearch current. channelDispatcher.Register(automation.NewAgentBotRuleListenerWithSearchIndexer(&dbProvider{db: db}, searchIndexer)) diff --git a/internal/handler/webhook/facebook_webhook.go b/internal/handler/webhook/facebook_webhook.go index 3a1de524..0b8e5405 100644 --- a/internal/handler/webhook/facebook_webhook.go +++ b/internal/handler/webhook/facebook_webhook.go @@ -76,6 +76,13 @@ func (h *FacebookWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *Facebook return h } +func (h *FacebookWebhookHandler) WithSearchIndexer(indexer IncomingSearchIndexer) *FacebookWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetSearchIndexer(indexer) + } + return h +} + // HandleFacebookVerification handles the GET webhook verification request from Facebook. // URL pattern: /webhooks/facebook/:inbox_id // Method: GET diff --git a/internal/handler/webhook/incoming_persister.go b/internal/handler/webhook/incoming_persister.go index 05afb0c8..ce103917 100644 --- a/internal/handler/webhook/incoming_persister.go +++ b/internal/handler/webhook/incoming_persister.go @@ -21,6 +21,13 @@ type IncomingPersister struct { db *gorm.DB dispatcher *channel.Dispatcher worker *worker.WorkerPool + indexer IncomingSearchIndexer +} + +type IncomingSearchIndexer interface { + IndexConversation(ctx context.Context, conversation *model.Conversation) error + IndexMessage(ctx context.Context, message *model.Message) error + IndexContact(ctx context.Context, contact *model.Contact) error } type IncomingPersistResult struct { @@ -42,6 +49,14 @@ func (p *IncomingPersister) SetWorkerPool(wp *worker.WorkerPool) *IncomingPersis return p } +func (p *IncomingPersister) SetSearchIndexer(indexer IncomingSearchIndexer) *IncomingPersister { + if p == nil { + return p + } + p.indexer = indexer + return p +} + func NewIncomingPersister(db *gorm.DB, dispatcher ...*channel.Dispatcher) *IncomingPersister { if db == nil { return nil @@ -131,6 +146,7 @@ func (p *IncomingPersister) performPersistIncoming(ctx context.Context, inbox *m if err != nil { return nil, err } + p.indexIncomingResult(ctx, &result) p.dispatchIncomingEvents(ctx, inbox, &result) return &result, nil } @@ -156,7 +172,8 @@ func (p *IncomingPersister) performMessageStatusUpdate(ctx context.Context, inbo return nil } - return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + var updatedMessage *model.Message + if err := p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var message model.Message if err := tx.Where("inbox_id = ? AND source_id = ?", inbox.ID, sourceID).First(&message).Error; err != nil { if err == gorm.ErrRecordNotFound { @@ -182,14 +199,22 @@ func (p *IncomingPersister) performMessageStatusUpdate(ctx context.Context, inbo } if contactID == 0 { p.dispatchMessageStatusEvent(ctx, inbox, &message) + copy := message + updatedMessage = © return nil } if err := p.upsertDeliveryStatus(ctx, tx, &message, contactID, status, occurredAt); err != nil { return err } p.dispatchMessageStatusEvent(ctx, inbox, &message) + copy := message + updatedMessage = © return nil - }) + }); err != nil { + return err + } + p.indexMessage(ctx, updatedMessage) + return nil } // UpdateContactConversationMessagesStatus applies read receipts that only identify the contact/conversation. @@ -208,7 +233,8 @@ func (p *IncomingPersister) performContactMessagesStatusUpdate(ctx context.Conte return nil } - return p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + updatedMessages := []model.Message{} + if err := p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var contactInbox model.ContactInbox if err := tx.Where("inbox_id = ? AND source_id = ?", inbox.ID, contactSourceID).First(&contactInbox).Error; err != nil { if err == gorm.ErrRecordNotFound { @@ -244,9 +270,59 @@ func (p *IncomingPersister) performContactMessagesStatusUpdate(ctx context.Conte return err } p.dispatchMessageStatusEvent(ctx, inbox, &messages[i]) + updatedMessages = append(updatedMessages, messages[i]) } return nil - }) + }); err != nil { + return err + } + for i := range updatedMessages { + p.indexMessage(ctx, &updatedMessages[i]) + } + return nil +} + +func (p *IncomingPersister) indexIncomingResult(ctx context.Context, result *IncomingPersistResult) { + if p == nil || p.indexer == nil || result == nil || result.Duplicate { + return + } + if result.Contact != nil { + p.indexContact(ctx, result.Contact) + } + if result.Conversation != nil { + p.indexConversation(ctx, result.Conversation) + } + if result.Message != nil { + p.indexMessage(ctx, result.Message) + } +} + +func (p *IncomingPersister) indexContact(ctx context.Context, contact *model.Contact) { + if p == nil || p.indexer == nil || contact == nil { + return + } + if err := p.indexer.IndexContact(ctx, contact); err != nil { + // Provider webhooks should not fail because search is temporarily unavailable. + _ = err + } +} + +func (p *IncomingPersister) indexConversation(ctx context.Context, conversation *model.Conversation) { + if p == nil || p.indexer == nil || conversation == nil { + return + } + if err := p.indexer.IndexConversation(ctx, conversation); err != nil { + _ = err + } +} + +func (p *IncomingPersister) indexMessage(ctx context.Context, message *model.Message) { + if p == nil || p.indexer == nil || message == nil { + return + } + if err := p.indexer.IndexMessage(ctx, message); err != nil { + _ = err + } } func (p *IncomingPersister) deliveryStatusContactID(ctx context.Context, tx *gorm.DB, message *model.Message) (uint, error) { diff --git a/internal/handler/webhook/line_webhook.go b/internal/handler/webhook/line_webhook.go index 8d585eb5..e8ffcb7d 100644 --- a/internal/handler/webhook/line_webhook.go +++ b/internal/handler/webhook/line_webhook.go @@ -50,6 +50,13 @@ func (h *LineWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *LineWebhookH return h } +func (h *LineWebhookHandler) WithSearchIndexer(indexer IncomingSearchIndexer) *LineWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetSearchIndexer(indexer) + } + 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 1332c47e..ab2c8f9f 100644 --- a/internal/handler/webhook/telegram_webhook.go +++ b/internal/handler/webhook/telegram_webhook.go @@ -63,6 +63,13 @@ func (h *TelegramWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *Telegram return h } +func (h *TelegramWebhookHandler) WithSearchIndexer(indexer IncomingSearchIndexer) *TelegramWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetSearchIndexer(indexer) + } + return h +} + // HandleTelegramWebhook processes an incoming Telegram webhook Gin request. // URL pattern: /webhooks/telegram/:bot_token // Method: POST diff --git a/internal/handler/webhook/tiktok_webhook.go b/internal/handler/webhook/tiktok_webhook.go index 3bc9db7b..9556895c 100644 --- a/internal/handler/webhook/tiktok_webhook.go +++ b/internal/handler/webhook/tiktok_webhook.go @@ -54,6 +54,13 @@ func (h *TikTokWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *TikTokWebh return h } +func (h *TikTokWebhookHandler) WithSearchIndexer(indexer IncomingSearchIndexer) *TikTokWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetSearchIndexer(indexer) + } + return h +} + // HandleTikTokWebhook processes incoming TikTok webhook HTTP requests. func (h *TikTokWebhookHandler) HandleTikTokWebhook(c *gin.Context) { body, err := io.ReadAll(c.Request.Body) diff --git a/internal/handler/webhook/twilio_webhook.go b/internal/handler/webhook/twilio_webhook.go index 1d5bf1ce..a7d5ff9e 100644 --- a/internal/handler/webhook/twilio_webhook.go +++ b/internal/handler/webhook/twilio_webhook.go @@ -41,6 +41,13 @@ func (h *TwilioWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *TwilioWebh return h } +func (h *TwilioWebhookHandler) WithSearchIndexer(indexer IncomingSearchIndexer) *TwilioWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetSearchIndexer(indexer) + } + return h +} + // NewTwilioWebhookHandler creates a Twilio SMS webhook handler for Gin integration. func NewTwilioWebhookHandler(twilioWebhook *twiliochannel.WebhookHandler, db *gorm.DB, dispatcher ...*channel.Dispatcher) *TwilioWebhookHandler { return &TwilioWebhookHandler{ diff --git a/internal/handler/webhook/webhook_lookup_test.go b/internal/handler/webhook/webhook_lookup_test.go index 499e5946..04e4ee47 100644 --- a/internal/handler/webhook/webhook_lookup_test.go +++ b/internal/handler/webhook/webhook_lookup_test.go @@ -35,6 +35,27 @@ type recordingListener struct { events []*channel.ChannelEvent } +type recordingIncomingSearchIndexer struct { + contacts []model.Contact + conversations []model.Conversation + messages []model.Message +} + +func (r *recordingIncomingSearchIndexer) IndexContact(ctx context.Context, contact *model.Contact) error { + r.contacts = append(r.contacts, *contact) + return nil +} + +func (r *recordingIncomingSearchIndexer) IndexConversation(ctx context.Context, conversation *model.Conversation) error { + r.conversations = append(r.conversations, *conversation) + return nil +} + +func (r *recordingIncomingSearchIndexer) IndexMessage(ctx context.Context, message *model.Message) error { + r.messages = append(r.messages, *message) + return nil +} + type webhookAutomationDBProvider struct { db *gorm.DB } @@ -175,8 +196,9 @@ func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) { dispatcher := channel.NewDispatcher() listener := &recordingListener{} dispatcher.Register(listener) + indexer := &recordingIncomingSearchIndexer{} wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low")) - persister := NewIncomingPersister(db, dispatcher) + persister := NewIncomingPersister(db, dispatcher).SetSearchIndexer(indexer) msg := &channel.IncomingMessage{ ChannelType: channel.ChannelTelegram, @@ -232,6 +254,9 @@ func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) { if !listenerSaw(listener, channel.EventMessageStatusUpdated) { t.Fatalf("expected message.status_updated event, got %#v", listener.events) } + if len(indexer.messages) != 2 || indexer.messages[1].ID != updated.ID || indexer.messages[1].Status != string(model.MessageStatusDelivered) { + t.Fatalf("expected delivered message to be indexed, got %#v", indexer.messages) + } } func TestIncomingPersisterStatusJobDoesNotDowngradeRead(t *testing.T) { @@ -250,7 +275,8 @@ func TestIncomingPersisterStatusJobDoesNotDowngradeRead(t *testing.T) { t.Fatalf("create message: %v", err) } wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low")) - persister := NewIncomingPersister(db).SetWorkerPool(wp) + indexer := &recordingIncomingSearchIndexer{} + persister := NewIncomingPersister(db).SetSearchIndexer(indexer).SetWorkerPool(wp) if err := persister.UpdateMessageStatus(t.Context(), &inbox, "provider-read-1", model.MessageStatusDelivered, nil); err != nil { t.Fatalf("enqueue downgrade: %v", err) @@ -330,7 +356,8 @@ func TestIncomingPersisterQueuesContactMessagesStatusUpdateWithWorker(t *testing alreadyRead := createWebhookStatusMessage(t, db, inbox, conversation, "read-before", model.MessageStatusRead, cutoff.Add(-20*time.Second)) afterCutoff := createWebhookStatusMessage(t, db, inbox, conversation, "sent-after", model.MessageStatusSent, cutoff.Add(time.Minute)) wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low")) - persister := NewIncomingPersister(db).SetWorkerPool(wp) + indexer := &recordingIncomingSearchIndexer{} + persister := NewIncomingPersister(db).SetSearchIndexer(indexer).SetWorkerPool(wp) if err := persister.UpdateContactConversationMessagesStatus(t.Context(), &inbox, "meta-user-1", model.MessageStatusRead, &cutoff); err != nil { t.Fatalf("enqueue contact read: %v", err) @@ -355,6 +382,9 @@ func TestIncomingPersisterQueuesContactMessagesStatusUpdateWithWorker(t *testing if deliveryCount != 3 { t.Fatalf("expected 3 read delivery statuses, got %d", deliveryCount) } + if len(indexer.messages) != 3 { + t.Fatalf("expected 3 updated messages to be indexed, got %#v", indexer.messages) + } } func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) { @@ -363,7 +393,8 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) { dispatcher := channel.NewDispatcher() listener := &recordingListener{} dispatcher.Register(listener) - persister := NewIncomingPersister(db, dispatcher) + indexer := &recordingIncomingSearchIndexer{} + persister := NewIncomingPersister(db, dispatcher).SetSearchIndexer(indexer) msg := &channel.IncomingMessage{ ChannelType: channel.ChannelTelegram, @@ -387,6 +418,9 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) { if result.Message.Content != "hello" || result.Message.SourceID != "tg-msg-1" { t.Fatalf("unexpected message: %#v", result.Message) } + if len(indexer.contacts) != 1 || len(indexer.conversations) != 1 || len(indexer.messages) != 1 { + t.Fatalf("expected first incoming result indexed, contacts=%#v conversations=%#v messages=%#v", indexer.contacts, indexer.conversations, indexer.messages) + } duplicate, err := persister.PersistIncoming(t.Context(), &inbox, msg) if err != nil { @@ -395,6 +429,9 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) { if duplicate == nil || !duplicate.Duplicate { t.Fatalf("expected duplicate result, got %#v", duplicate) } + if len(indexer.messages) != 1 { + t.Fatalf("expected duplicate to skip indexing, got %#v", indexer.messages) + } msg.SourceID = "tg-msg-2" msg.Content = "second" @@ -405,6 +442,9 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) { if second.Contact.ID != result.Contact.ID || second.Conversation.ID != result.Conversation.ID { t.Fatalf("expected contact/conversation reuse: first=%#v second=%#v", result, second) } + if len(indexer.messages) != 2 || indexer.messages[1].ID != second.Message.ID { + t.Fatalf("expected second incoming message indexed, got %#v", indexer.messages) + } var messageCount int64 if err := db.Model(&model.Message{}).Where("inbox_id = ?", inbox.ID).Count(&messageCount).Error; err != nil { @@ -426,8 +466,9 @@ func TestIncomingPersisterQueuesIncomingMessageWithWorker(t *testing.T) { dispatcher := channel.NewDispatcher() listener := &recordingListener{} dispatcher.Register(listener) + indexer := &recordingIncomingSearchIndexer{} wp := worker.NewWorkerPool(db) - persister := NewIncomingPersister(db, dispatcher).SetWorkerPool(wp) + persister := NewIncomingPersister(db, dispatcher).SetSearchIndexer(indexer).SetWorkerPool(wp) msg := &channel.IncomingMessage{ ChannelType: channel.ChannelTelegram, @@ -464,6 +505,9 @@ func TestIncomingPersisterQueuesIncomingMessageWithWorker(t *testing.T) { t.Fatalf("process incoming job processed=%v err=%v", processed, err) } assertPersistedMessage(t, db, inbox.ID, "tg-inbound-job-1", "persist me later") + if len(indexer.messages) != 1 || indexer.messages[0].SourceID != "tg-inbound-job-1" { + t.Fatalf("expected durable incoming message to be indexed, got %#v", indexer.messages) + } 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) diff --git a/internal/handler/webhook/whatsapp_webhook.go b/internal/handler/webhook/whatsapp_webhook.go index 186b927c..dad427ab 100644 --- a/internal/handler/webhook/whatsapp_webhook.go +++ b/internal/handler/webhook/whatsapp_webhook.go @@ -78,6 +78,13 @@ func (h *WhatsAppWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *WhatsApp return h } +func (h *WhatsAppWebhookHandler) WithSearchIndexer(indexer IncomingSearchIndexer) *WhatsAppWebhookHandler { + if h != nil && h.persister != nil { + h.persister.SetSearchIndexer(indexer) + } + return h +} + type whatsAppPersisterAdapter struct { persister *IncomingPersister }