feat(search): index message delivery

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) {

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) {

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 {