rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(webhook): queue incoming messages

Browse Source
This commit is contained in:
rogee
2026-06-05 19:21:42 +08:00
parent dec8854ec9
commit d756473d91
7 changed files with 300 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/handler/webhook/incoming_persister.go
+27 -3
View File
@@ -57,18 +57,42 @@ func (p *IncomingPersister) PersistIncoming(ctx context.Context, inbox *model.In
if p == nil || p.db == nil || inbox == nil || msg == nil {
return nil, nil
}
if err := validateIncomingMessage(p, inbox, msg); err != nil {
return nil, err
}
if p.worker != nil {
return nil, p.enqueueIncomingMessagePersist(ctx, inbox.ID, msg)
}
return p.performPersistIncoming(ctx, inbox, msg)
}
func validateIncomingMessage(p *IncomingPersister, inbox *model.Inbox, msg *channel.IncomingMessage) error {
if p == nil || p.db == nil || inbox == nil || msg == nil {
return nil
}
if msg.SourceID == "" {
return nil, fmt.Errorf("incoming message missing source_id")
return fmt.Errorf("incoming message missing source_id")
}
senderID := msg.SenderID
if senderID == "" {
senderID = msg.ConversationID
}
if senderID == "" {
return nil, fmt.Errorf("incoming message missing sender_id")
return fmt.Errorf("incoming message missing sender_id")
}
if msg.Content == "" && len(msg.Attachments) == 0 {
return nil, fmt.Errorf("incoming message has no content or attachments")
return fmt.Errorf("incoming message has no content or attachments")
}
return nil
}
func (p *IncomingPersister) performPersistIncoming(ctx context.Context, inbox *model.Inbox, msg *channel.IncomingMessage) (*IncomingPersistResult, error) {
if p == nil || p.db == nil || inbox == nil || msg == nil {
return nil, nil
}
senderID := msg.SenderID
if senderID == "" {
senderID = msg.ConversationID
}
var result IncomingPersistResult
internal/handler/webhook/incoming_persister_jobs.go
+52
View File
@@ -10,15 +10,22 @@ import (
"gorm.io/datatypes"
"gorm.io/gorm"
"github.com/gochat/gochat/internal/channel"
"github.com/gochat/gochat/internal/model"
"github.com/gochat/gochat/internal/worker"
)
const (
TaskTypeProviderIncomingMessagePersist = "webhook:incoming_message_persist"
TaskTypeProviderMessageStatusUpdate = "webhook:message_status_update"
TaskTypeProviderContactMessagesStatusUpdate = "webhook:contact_messages_status_update"
)
type providerIncomingMessagePersistJob struct {
InboxID uint `json:"inbox_id"`
Message channel.IncomingMessage `json:"message"`
}
type providerMessageStatusUpdateJob struct {
InboxID uint `json:"inbox_id"`
SourceID string `json:"source_id"`
@@ -38,10 +45,24 @@ func (p *IncomingPersister) registerJobs(wp *worker.WorkerPool) {
if p == nil || wp == nil {
return
}
wp.Register(TaskTypeProviderIncomingMessagePersist, p.performIncomingMessagePersistJob)
wp.Register(TaskTypeProviderMessageStatusUpdate, p.performMessageStatusUpdateJob)
wp.Register(TaskTypeProviderContactMessagesStatusUpdate, p.performContactMessagesStatusUpdateJob)
}
func (p *IncomingPersister) enqueueIncomingMessagePersist(ctx context.Context, inboxID uint, msg *channel.IncomingMessage) error {
if p == nil || p.worker == nil || inboxID == 0 || msg == nil {
return nil
}
queue := incomingMessageQueue(msg.ChannelType)
_, err := p.worker.Enqueue(ctx, TaskTypeProviderIncomingMessagePersist, providerIncomingMessagePersistJob{InboxID: inboxID, Message: *msg},
worker.WithQueue(queue),
worker.WithMaxAttempts(3),
worker.WithIdempotencyKey(fmt.Sprintf("webhook:incoming_message:%d:%s", inboxID, msg.SourceID)),
)
return err
}
func (p *IncomingPersister) enqueueMessageStatusUpdate(ctx context.Context, inboxID uint, sourceID string, status model.MessageStatus, occurredAt *time.Time, externalError string) error {
if p == nil || p.worker == nil || inboxID == 0 || sourceID == "" {
return nil
@@ -103,6 +124,28 @@ func (p *IncomingPersister) performMessageStatusUpdateJob(ctx context.Context, j
return p.performMessageStatusUpdate(ctx, inbox, payload.SourceID, payload.Status, unixNanoToTime(payload.OccurredAtUnixNano), payload.ExternalError)
}
func (p *IncomingPersister) performIncomingMessagePersistJob(ctx context.Context, job *model.BackgroundJob) error {
var payload providerIncomingMessagePersistJob
if err := json.Unmarshal(job.Payload, &payload); err != nil {
return fmt.Errorf("unmarshal provider incoming message job: %w", err)
}
if payload.InboxID == 0 {
return fmt.Errorf("invalid provider incoming message job payload: %#v", payload)
}
inbox, err := p.loadInboxForStatusJob(ctx, payload.InboxID)
if err != nil {
return err
}
if inbox == nil {
return nil
}
if err := validateIncomingMessage(p, inbox, &payload.Message); err != nil {
return err
}
_, err = p.performPersistIncoming(ctx, inbox, &payload.Message)
return err
}
func (p *IncomingPersister) performContactMessagesStatusUpdateJob(ctx context.Context, job *model.BackgroundJob) error {
var payload providerContactMessagesStatusUpdateJob
if err := json.Unmarshal(job.Payload, &payload); err != nil {
@@ -141,6 +184,15 @@ func validProviderMessageStatus(status model.MessageStatus) bool {
}
}
func incomingMessageQueue(channelType channel.ChannelType) string {
switch channelType {
case channel.ChannelWhatsApp, channel.ChannelTwilioSMS, channel.ChannelTwilioWA:
return "low"
default:
return model.DefaultBackgroundJobQueue
}
}
func validProviderMessageStatusTransition(current, next model.MessageStatus) bool {
if !validProviderMessageStatus(next) || current == next {
return validProviderMessageStatus(next)
internal/handler/webhook/line_webhook.go
+8
View File
@@ -16,6 +16,7 @@ import (
linechannel "github.com/gochat/gochat/internal/channel/line"
"github.com/gochat/gochat/internal/model"
channelmodel "github.com/gochat/gochat/internal/model/channel"
"github.com/gochat/gochat/internal/worker"
applogger "github.com/gochat/gochat/pkg/logger"
"github.com/gin-gonic/gin"
@@ -42,6 +43,13 @@ func NewLineWebhookHandler(lineWebhook *linechannel.WebhookHandler, pipeline *li
}
}
func (h *LineWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *LineWebhookHandler {
if h != nil && h.persister != nil {
h.persister.SetWorkerPool(wp)
}
return h
}
// HandleLineWebhook processes an incoming LINE webhook Gin request.
func (h *LineWebhookHandler) HandleLineWebhook(c *gin.Context) {
lineChannelID := c.Param("line_channel_id")
internal/handler/webhook/telegram_webhook.go
+8
View File
@@ -13,6 +13,7 @@ import (
telegramchannel "github.com/gochat/gochat/internal/channel/telegram"
"github.com/gochat/gochat/internal/model"
channelmodel "github.com/gochat/gochat/internal/model/channel"
"github.com/gochat/gochat/internal/worker"
applogger "github.com/gochat/gochat/pkg/logger"
"gorm.io/gorm"
)
@@ -55,6 +56,13 @@ func NewTelegramWebhookHandler(
}
}
func (h *TelegramWebhookHandler) WithWorkerPool(wp *worker.WorkerPool) *TelegramWebhookHandler {
if h != nil && h.persister != nil {
h.persister.SetWorkerPool(wp)
}
return h
}
// HandleTelegramWebhook processes an incoming Telegram webhook Gin request.
// URL pattern: /webhooks/telegram/:bot_token
// Method: POST
internal/handler/webhook/webhook_lookup_test.go
+146 -1
View File
@@ -176,7 +176,7 @@ func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) {
listener := &recordingListener{}
dispatcher.Register(listener)
wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low"))
persister := NewIncomingPersister(db, dispatcher).SetWorkerPool(wp)
persister := NewIncomingPersister(db, dispatcher)
msg := &channel.IncomingMessage{
ChannelType: channel.ChannelTelegram,
@@ -193,6 +193,7 @@ func TestIncomingPersisterQueuesMessageStatusUpdateWithWorker(t *testing.T) {
if err != nil {
t.Fatalf("persist incoming: %v", err)
}
persister.SetWorkerPool(wp)
occurredAt := time.Now().UTC().Add(-time.Minute)
if err := persister.UpdateMessageStatus(t.Context(), &inbox, "tg-status-job-1", model.MessageStatusDelivered, &occurredAt); err != nil {
@@ -419,6 +420,100 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) {
}
}
func TestIncomingPersisterQueuesIncomingMessageWithWorker(t *testing.T) {
db := newWebhookLookupTestDB(t)
inbox := seedWebhookInbox(t, db, "telegram")
dispatcher := channel.NewDispatcher()
listener := &recordingListener{}
dispatcher.Register(listener)
wp := worker.NewWorkerPool(db)
persister := NewIncomingPersister(db, dispatcher).SetWorkerPool(wp)
msg := &channel.IncomingMessage{
ChannelType: channel.ChannelTelegram,
SourceID: "tg-inbound-job-1",
SenderID: "tg-inbound-user-1",
SenderName: "Inbound Worker User",
SenderType: channel.SenderContact,
Content: "persist me later",
ContentType: channel.ContentText,
InboxID: inbox.ID,
AccountID: inbox.AccountID,
}
result, err := persister.PersistIncoming(t.Context(), &inbox, msg)
if err != nil {
t.Fatalf("enqueue incoming: %v", err)
}
if result != nil {
t.Fatalf("expected async enqueue result to be nil, got %#v", result)
}
var queued model.BackgroundJob
if err := db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeProviderIncomingMessagePersist, model.DefaultBackgroundJobQueue, model.BackgroundJobStatusQueued).First(&queued).Error; err != nil {
t.Fatalf("expected queued incoming job: %v", err)
}
var beforeCount int64
if err := db.Model(&model.Message{}).Where("source_id = ?", "tg-inbound-job-1").Count(&beforeCount).Error; err != nil {
t.Fatalf("count before messages: %v", err)
}
if beforeCount != 0 {
t.Fatalf("expected no message before worker, got %d", beforeCount)
}
processed, err := wp.ProcessOne(t.Context())
if err != nil || !processed {
t.Fatalf("process incoming job processed=%v err=%v", processed, err)
}
assertPersistedMessage(t, db, inbox.ID, "tg-inbound-job-1", "persist me later")
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)
}
}
}
func TestIncomingPersisterIncomingMessageJobIsIdempotent(t *testing.T) {
db := newWebhookLookupTestDB(t)
inbox := seedWebhookInbox(t, db, "telegram")
wp := worker.NewWorkerPool(db)
persister := NewIncomingPersister(db).SetWorkerPool(wp)
msg := &channel.IncomingMessage{
ChannelType: channel.ChannelTelegram,
SourceID: "tg-inbound-idempotent-1",
SenderID: "tg-inbound-idempotent-user",
SenderName: "Idempotent User",
SenderType: channel.SenderContact,
Content: "only once",
ContentType: channel.ContentText,
InboxID: inbox.ID,
AccountID: inbox.AccountID,
}
if _, err := persister.PersistIncoming(t.Context(), &inbox, msg); err != nil {
t.Fatalf("enqueue first incoming: %v", err)
}
if _, err := persister.PersistIncoming(t.Context(), &inbox, msg); err != nil {
t.Fatalf("enqueue duplicate incoming: %v", err)
}
var jobCount int64
if err := db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeProviderIncomingMessagePersist).Count(&jobCount).Error; err != nil {
t.Fatalf("count jobs: %v", err)
}
if jobCount != 1 {
t.Fatalf("expected one idempotent job, got %d", jobCount)
}
processed, err := wp.ProcessOne(t.Context())
if err != nil || !processed {
t.Fatalf("process incoming job processed=%v err=%v", processed, err)
}
var messageCount int64
if err := db.Model(&model.Message{}).Where("source_id = ?", "tg-inbound-idempotent-1").Count(&messageCount).Error; err != nil {
t.Fatalf("count messages: %v", err)
}
if messageCount != 1 {
t.Fatalf("expected one persisted message, got %d", messageCount)
}
}
func listenerSaw(listener *recordingListener, eventType channel.EventType) bool {
for _, event := range listener.events {
if event.Type == eventType {
@@ -710,6 +805,56 @@ func TestTwilioWebhookPersistsIncomingMessage(t *testing.T) {
assertPersistedMessage(t, db, inbox.ID, "SMIN1", "hello sms")
}
func TestTwilioInboundSMSQueuesIncomingMessageWithWorker(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"}
if err := db.Create(&channelRecord).Error; err != nil {
t.Fatalf("create twilio channel: %v", err)
}
twilioRepo := twiliochannel.NewRepository(db)
twilioService := twiliochannel.NewTwilioService(twilioRepo)
twilioPipeline := twiliochannel.NewIncomingProcessor(twilioService)
twilioWebhook := twiliochannel.NewWebhookHandler(twilioPipeline, twilioService)
wp := worker.NewWorkerPoolWithOptions(db, worker.WithQueues("low"))
h := NewTwilioWebhookHandler(twilioWebhook, db).WithWorkerPool(wp)
r := gin.New()
r.POST("/webhooks/sms/:phone_number", h.HandleTwilioInboundSMS)
form := url.Values{}
form.Set("MessageSid", "SMINASYNC1")
form.Set("AccountSid", "AC123")
form.Set("From", "+15550002222")
form.Set("To", "+15551234567")
form.Set("Body", "queued sms")
req := httptest.NewRequest(http.MethodPost, "/webhooks/sms/+15551234567", 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.StatusOK {
t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
}
var queued model.BackgroundJob
if err := db.Where("job_type = ? AND queue = ? AND status = ?", TaskTypeProviderIncomingMessagePersist, "low", model.BackgroundJobStatusQueued).First(&queued).Error; err != nil {
t.Fatalf("expected queued twilio incoming job: %v", err)
}
var beforeCount int64
if err := db.Model(&model.Message{}).Where("source_id = ?", "SMINASYNC1").Count(&beforeCount).Error; err != nil {
t.Fatalf("count before messages: %v", err)
}
if beforeCount != 0 {
t.Fatalf("expected no message before worker, got %d", beforeCount)
}
processed, err := wp.ProcessOne(t.Context())
if err != nil || !processed {
t.Fatalf("process twilio incoming job processed=%v err=%v", processed, err)
}
assertPersistedMessage(t, db, inbox.ID, "SMINASYNC1", "queued sms")
}
func TestTwilioDeliveryStatusUpdatesExistingMessage(t *testing.T) {
gin.SetMode(gin.TestMode)
db := newWebhookLookupTestDB(t)