feat(messages): queue delivery statuses

2026-06-05 19:05:45 +08:00
parent 1a4b4a2d29
commit dec8854ec9
11 changed files with 711 additions and 57 deletions
@@ -26,6 +26,7 @@ import (
 	whatsappchannel "github.com/gochat/gochat/internal/channel/whatsapp"
 	"github.com/gochat/gochat/internal/model"
 	channelmodel "github.com/gochat/gochat/internal/model/channel"
+	"github.com/gochat/gochat/internal/worker"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
 )
@@ -62,6 +63,7 @@ func newWebhookLookupTestDB(t *testing.T) *gorm.DB {
 		&model.Conversation{},
 		&model.Message{},
 		&model.DeliveryStatus{},
+		&model.BackgroundJob{},
 		&automation.AutomationRule{},
 		&automation.AutomationExecution{},
 		&channelmodel.ChannelTelegram{},
@@ -167,6 +169,193 @@ func TestIncomingPersisterUpdatesMessageStatus(t *testing.T) {
 	}
 }

+func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) {
+	db := newWebhookLookupTestDB(t)
+	inbox := seedWebhookInbox(t, db, "telegram")
+	dispatcher := channel.NewDispatcher()
+	listener := &recordingListener{}
+	dispatcher.Register(listener)
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low"))
+	persister := NewIncomingPersister(db, dispatcher).SetWorkerPool(wp)
+
+	msg := &channel.IncomingMessage{
+		ChannelType: channel.ChannelTelegram,
+		SourceID:    "tg-status-job-1",
+		SenderID:    "tg-user-status-job",
+		SenderName:  "Status Worker User",
+		SenderType:  channel.SenderContact,
+		Content:     "status me async",
+		ContentType: channel.ContentText,
+		InboxID:     inbox.ID,
+		AccountID:   inbox.AccountID,
+	}
+	result, err := persister.PersistIncoming(t.Context(), &inbox, msg)
+	if err != nil {
+		t.Fatalf("persist incoming: %v", err)
+	}
+
+	occurredAt := time.Now().UTC().Add(-time.Minute)
+	if err := persister.UpdateMessageStatus(t.Context(), &inbox, "tg-status-job-1", model.MessageStatusDelivered, &occurredAt); err != nil {
+		t.Fatalf("enqueue status: %v", err)
+	}
+	var queued model.BackgroundJob
+	if err := db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeProviderMessageStatusUpdate, "low", model.BackgroundJobStatusQueued).First(&queued).Error; err != nil {
+		t.Fatalf("expected queued status job: %v", err)
+	}
+	var before model.Message
+	if err := db.First(&before, result.Message.ID).Error; err != nil {
+		t.Fatalf("load before message: %v", err)
+	}
+	if before.Status != string(model.MessageStatusSent) {
+		t.Fatalf("expected status to remain sent before worker, got %s", before.Status)
+	}
+
+	processed, err := wp.ProcessOne(t.Context())
+	if err != nil || !processed {
+		t.Fatalf("process status job processed=%v err=%v", processed, err)
+	}
+	var updated model.Message
+	if err := db.First(&updated, result.Message.ID).Error; err != nil {
+		t.Fatalf("load updated message: %v", err)
+	}
+	if updated.Status != string(model.MessageStatusDelivered) {
+		t.Fatalf("expected delivered status, got %s", updated.Status)
+	}
+	var delivery model.DeliveryStatus
+	if err := db.Where("message_id = ? AND contact_id = ?", updated.ID, result.Contact.ID).First(&delivery).Error; err != nil {
+		t.Fatalf("expected delivery status: %v", err)
+	}
+	if delivery.Status != model.MessageStatusDelivered || delivery.DeliveredAt == nil {
+		t.Fatalf("expected delivered timestamp, got %#v", delivery)
+	}
+	if !listenerSaw(listener, channel.EventMessageStatusUpdated) {
+		t.Fatalf("expected message.status_updated event, got %#v", listener.events)
+	}
+}
+
+func TestIncomingPersisterStatusJobDoesNotDowngradeRead(t *testing.T) {
+	db := newWebhookLookupTestDB(t)
+	inbox := seedWebhookInbox(t, db, "telegram")
+	contact := model.Contact{AccountID: inbox.AccountID, Name: "Read User"}
+	if err := db.Create(&contact).Error; err != nil {
+		t.Fatalf("create contact: %v", err)
+	}
+	conversation := model.Conversation{AccountID: inbox.AccountID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusOpen), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType}
+	if err := db.Create(&conversation).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+	message := model.Message{ConversationID: conversation.ID, AccountID: inbox.AccountID, InboxID: inbox.ID, SenderID: &contact.ID, SenderType: string(model.SenderTypeContact), Content: "already read", ContentType: string(model.MessageContentTypeText), MessageType: string(model.MessageTypeOutgoing), Status: string(model.MessageStatusRead), SourceID: "provider-read-1"}
+	if err := db.Create(&message).Error; err != nil {
+		t.Fatalf("create message: %v", err)
+	}
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low"))
+	persister := NewIncomingPersister(db).SetWorkerPool(wp)
+
+	if err := persister.UpdateMessageStatus(t.Context(), &inbox, "provider-read-1", model.MessageStatusDelivered, nil); err != nil {
+		t.Fatalf("enqueue downgrade: %v", err)
+	}
+	processed, err := wp.ProcessOne(t.Context())
+	if err != nil || !processed {
+		t.Fatalf("process downgrade job processed=%v err=%v", processed, err)
+	}
+	var updated model.Message
+	if err := db.First(&updated, message.ID).Error; err != nil {
+		t.Fatalf("load message: %v", err)
+	}
+	if updated.Status != string(model.MessageStatusRead) {
+		t.Fatalf("expected read to remain read, got %s", updated.Status)
+	}
+}
+
+func TestIncomingPersisterStatusJobStoresExternalError(t *testing.T) {
+	db := newWebhookLookupTestDB(t)
+	inbox := seedWebhookInbox(t, db, "twilio_sms")
+	contact := model.Contact{AccountID: inbox.AccountID, Name: "SMS User"}
+	if err := db.Create(&contact).Error; err != nil {
+		t.Fatalf("create contact: %v", err)
+	}
+	conversation := model.Conversation{AccountID: inbox.AccountID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusOpen), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType}
+	if err := db.Create(&conversation).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+	message := model.Message{ConversationID: conversation.ID, AccountID: inbox.AccountID, InboxID: inbox.ID, SenderID: &contact.ID, SenderType: string(model.SenderTypeContact), Content: "failed", ContentType: string(model.MessageContentTypeText), MessageType: string(model.MessageTypeOutgoing), Status: string(model.MessageStatusSent), SourceID: "SMFAIL1"}
+	if err := db.Create(&message).Error; err != nil {
+		t.Fatalf("create message: %v", err)
+	}
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low"))
+	persister := NewIncomingPersister(db).SetWorkerPool(wp)
+
+	if err := persister.UpdateMessageStatusWithError(t.Context(), &inbox, "SMFAIL1", model.MessageStatusFailed, nil, "30003 - unreachable handset"); err != nil {
+		t.Fatalf("enqueue failure: %v", err)
+	}
+	processed, err := wp.ProcessOne(t.Context())
+	if err != nil || !processed {
+		t.Fatalf("process failure job processed=%v err=%v", processed, err)
+	}
+	var updated model.Message
+	if err := db.First(&updated, message.ID).Error; err != nil {
+		t.Fatalf("load failed message: %v", err)
+	}
+	if updated.Status != string(model.MessageStatusFailed) {
+		t.Fatalf("expected failed status, got %s", updated.Status)
+	}
+	attrs := map[string]any{}
+	if err := json.Unmarshal(updated.ContentAttributes, &attrs); err != nil {
+		t.Fatalf("unmarshal content attrs: %v", err)
+	}
+	if attrs["external_error"] != "30003 - unreachable handset" {
+		t.Fatalf("expected external_error, got %#v", attrs)
+	}
+}
+
+func TestIncomingPersisterQueuesContactMessagesStatusUpdateWithWorker(t *testing.T) {
+	db := newWebhookLookupTestDB(t)
+	inbox := seedWebhookInbox(t, db, "facebook")
+	contact := model.Contact{AccountID: inbox.AccountID, Name: "Meta User"}
+	if err := db.Create(&contact).Error; err != nil {
+		t.Fatalf("create contact: %v", err)
+	}
+	contactInbox := model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, SourceID: "meta-user-1", PubsubToken: "pub-meta"}
+	if err := db.Create(&contactInbox).Error; err != nil {
+		t.Fatalf("create contact inbox: %v", err)
+	}
+	conversation := model.Conversation{AccountID: inbox.AccountID, InboxID: inbox.ID, ContactID: contact.ID, ContactInboxID: &contactInbox.ID, Status: string(model.ConversationStatusOpen), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType}
+	if err := db.Create(&conversation).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+	cutoff := time.Now().UTC()
+	beforeSent := createWebhookStatusMessage(t, db, inbox, conversation, "sent-before", model.MessageStatusSent, cutoff.Add(-time.Minute))
+	beforeDelivered := createWebhookStatusMessage(t, db, inbox, conversation, "delivered-before", model.MessageStatusDelivered, cutoff.Add(-30*time.Second))
+	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)
+
+	if err := persister.UpdateContactConversationMessagesStatus(t.Context(), &inbox, "meta-user-1", model.MessageStatusRead, &cutoff); err != nil {
+		t.Fatalf("enqueue contact read: %v", err)
+	}
+	var queued model.BackgroundJob
+	if err := db.Where("job_type = ? AND queue = ?", TaskTypeProviderContactMessagesStatusUpdate, "low").First(&queued).Error; err != nil {
+		t.Fatalf("expected queued contact status job: %v", err)
+	}
+	processed, err := wp.ProcessOne(t.Context())
+	if err != nil || !processed {
+		t.Fatalf("process contact status job processed=%v err=%v", processed, err)
+	}
+	assertWebhookMessageStatus(t, db, beforeSent.ID, model.MessageStatusRead)
+	assertWebhookMessageStatus(t, db, beforeDelivered.ID, model.MessageStatusRead)
+	assertWebhookMessageStatus(t, db, alreadyRead.ID, model.MessageStatusRead)
+	assertWebhookMessageStatus(t, db, afterCutoff.ID, model.MessageStatusSent)
+
+	var deliveryCount int64
+	if err := db.Model(&model.DeliveryStatus{}).Where("contact_id = ? AND status = ?", contact.ID, model.MessageStatusRead).Count(&deliveryCount).Error; err != nil {
+		t.Fatalf("count delivery statuses: %v", err)
+	}
+	if deliveryCount != 3 {
+		t.Fatalf("expected 3 read delivery statuses, got %d", deliveryCount)
+	}
+}
+
 func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) {
 	db := newWebhookLookupTestDB(t)
 	inbox := seedWebhookInbox(t, db, "telegram")
@@ -239,6 +428,37 @@ func listenerSaw(listener *recordingListener, eventType channel.EventType) bool
 	return false
 }

+func createWebhookStatusMessage(t *testing.T, db *gorm.DB, inbox model.Inbox, conversation model.Conversation, sourceID string, status model.MessageStatus, createdAt time.Time) model.Message {
+	t.Helper()
+	message := model.Message{
+		ConversationID: conversation.ID,
+		AccountID:      inbox.AccountID,
+		InboxID:        inbox.ID,
+		Content:        sourceID,
+		ContentType:    string(model.MessageContentTypeText),
+		MessageType:    string(model.MessageTypeOutgoing),
+		Status:         string(status),
+		SourceID:       sourceID,
+	}
+	message.CreatedAt = createdAt
+	message.UpdatedAt = createdAt
+	if err := db.Create(&message).Error; err != nil {
+		t.Fatalf("create status message: %v", err)
+	}
+	return message
+}
+
+func assertWebhookMessageStatus(t *testing.T, db *gorm.DB, messageID uint, want model.MessageStatus) {
+	t.Helper()
+	var message model.Message
+	if err := db.First(&message, messageID).Error; err != nil {
+		t.Fatalf("load message %d: %v", messageID, err)
+	}
+	if message.Status != string(want) {
+		t.Fatalf("message %d expected %s, got %s", messageID, want, message.Status)
+	}
+}
+
 func shopifyHMAC(secret string, body []byte) string {
 	mac := hmac.New(sha256.New, []byte(secret))
 	mac.Write(body)
@@ -524,8 +744,8 @@ func TestTwilioDeliveryStatusUpdatesExistingMessage(t *testing.T) {

 	r.ServeHTTP(w, req)

-	if w.Code != http.StatusOK {
-		t.Fatalf("expected 200, got %d", w.Code)
+	if w.Code != http.StatusNoContent {
+		t.Fatalf("expected 204, got %d", w.Code)
 	}
 	var updated model.Message
 	if err := db.First(&updated, message.ID).Error; err != nil {
@@ -536,6 +756,76 @@ func TestTwilioDeliveryStatusUpdatesExistingMessage(t *testing.T) {
 	}
 }

+func TestTwilioDeliveryStatusQueuesExactRouteWithWorker(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", MessagingServiceSID: "MG123"}
+	if err := db.Create(&channelRecord).Error; err != nil {
+		t.Fatalf("create twilio channel: %v", err)
+	}
+	contact := model.Contact{AccountID: inbox.AccountID, Name: "SMS Contact", Identifier: "+15550001111"}
+	if err := db.Create(&contact).Error; err != nil {
+		t.Fatalf("create contact: %v", err)
+	}
+	conversation := model.Conversation{AccountID: inbox.AccountID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusOpen), ChannelType: inbox.ChannelType, Channel: inbox.ChannelType}
+	if err := db.Create(&conversation).Error; err != nil {
+		t.Fatalf("create conversation: %v", err)
+	}
+	message := model.Message{ConversationID: conversation.ID, AccountID: inbox.AccountID, InboxID: inbox.ID, SenderID: &contact.ID, SenderType: string(model.SenderTypeContact), Content: "out", ContentType: string(model.MessageContentTypeText), MessageType: string(model.MessageTypeOutgoing), Status: string(model.MessageStatusSent), SourceID: "SMFAILED"}
+	if err := db.Create(&message).Error; err != nil {
+		t.Fatalf("create message: %v", err)
+	}
+	wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low"))
+	h := NewTwilioWebhookHandler(nil, db).WithWorkerPool(wp)
+	r := gin.New()
+	r.POST("/twilio/delivery_status", h.HandleTwilioDeliveryStatus)
+	form := url.Values{}
+	form.Set("MessagingServiceSid", "MG123")
+	form.Set("MessageSid", "SMFAILED")
+	form.Set("MessageStatus", "undelivered")
+	form.Set("ErrorCode", "30003")
+	form.Set("ErrorMessage", "Unreachable handset")
+	req := httptest.NewRequest(http.MethodPost, "/twilio/delivery_status", 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.StatusNoContent {
+		t.Fatalf("expected 204, got %d", w.Code)
+	}
+	var queued model.BackgroundJob
+	if err := db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeProviderMessageStatusUpdate, "low", model.BackgroundJobStatusQueued).First(&queued).Error; err != nil {
+		t.Fatalf("expected queued twilio status job: %v", err)
+	}
+	var before model.Message
+	if err := db.First(&before, message.ID).Error; err != nil {
+		t.Fatalf("load before message: %v", err)
+	}
+	if before.Status != string(model.MessageStatusSent) {
+		t.Fatalf("expected message to remain sent before worker, got %s", before.Status)
+	}
+	processed, err := wp.ProcessOne(t.Context())
+	if err != nil || !processed {
+		t.Fatalf("process twilio status job processed=%v err=%v", processed, err)
+	}
+	var updated model.Message
+	if err := db.First(&updated, message.ID).Error; err != nil {
+		t.Fatalf("load updated message: %v", err)
+	}
+	if updated.Status != string(model.MessageStatusFailed) {
+		t.Fatalf("expected failed status, got %s", updated.Status)
+	}
+	attrs := map[string]any{}
+	if err := json.Unmarshal(updated.ContentAttributes, &attrs); err != nil {
+		t.Fatalf("unmarshal attrs: %v", err)
+	}
+	if attrs["external_error"] != "30003 - Unreachable handset" {
+		t.Fatalf("expected twilio external error, got %#v", attrs)
+	}
+}
+
 func TestWhatsAppWebhookPersistsIncomingMessage(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 	db := newWebhookLookupTestDB(t)