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

feat(webhook): dispatch persisted provider events

Browse Source
This commit is contained in:
rogee
2026-06-05 00:12:04 +08:00
parent 66ecabb984
commit 06b999becd
10 changed files with 178 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/handler/webhook/facebook_webhook.go
+2 -1
View File
@@ -57,13 +57,14 @@ func NewFacebookWebhookHandler(
fbProvider *fbchannel.FacebookProvider,
igProvider *fbchannel.InstagramProvider,
db *gorm.DB,
dispatcher ...*channel.Dispatcher,
) *FacebookWebhookHandler {
return &FacebookWebhookHandler{
fbProvider: fbProvider,
igProvider: igProvider,
webhookParser: fbchannel.NewWebhookParser(),
db: db,
persister: NewIncomingPersister(db),
persister: NewIncomingPersister(db, dispatcher...),
}
}
internal/handler/webhook/incoming_persister.go
+115 -25
View File
@@ -17,22 +17,29 @@ import (
// IncomingPersister is the durable boundary after provider-specific webhook parsing.
// Reference: Chatwoot IncomingMessageService creates ContactInbox, Conversation, and Message.
type IncomingPersister struct {
db *gorm.DB
db *gorm.DB
dispatcher *channel.Dispatcher
}
type IncomingPersistResult struct {
Contact *model.Contact
ContactInbox *model.ContactInbox
Conversation *model.Conversation
Message *model.Message
Duplicate bool
Contact *model.Contact
ContactInbox *model.ContactInbox
Conversation *model.Conversation
Message *model.Message
Duplicate bool
ContactCreated bool
ConversationCreated bool
}
func NewIncomingPersister(db *gorm.DB) *IncomingPersister {
func NewIncomingPersister(db *gorm.DB, dispatcher ...*channel.Dispatcher) *IncomingPersister {
if db == nil {
return nil
}
return &IncomingPersister{db: db}
p := &IncomingPersister{db: db}
if len(dispatcher) > 0 {
p.dispatcher = dispatcher[0]
}
return p
}
func (p *IncomingPersister) PersistIncoming(ctx context.Context, inbox *model.Inbox, msg *channel.IncomingMessage) (*IncomingPersistResult, error) {
@@ -64,18 +71,20 @@ func (p *IncomingPersister) PersistIncoming(ctx context.Context, inbox *model.In
return err
}
contact, contactInbox, err := p.resolveOrCreateContactInbox(ctx, tx, inbox, msg, senderID)
contact, contactInbox, contactCreated, err := p.resolveOrCreateContactInbox(ctx, tx, inbox, msg, senderID)
if err != nil {
return err
}
result.Contact = contact
result.ContactInbox = contactInbox
result.ContactCreated = contactCreated
conversation, err := p.resolveOrCreateConversation(ctx, tx, inbox, contact, contactInbox, msg)
conversation, conversationCreated, err := p.resolveOrCreateConversation(ctx, tx, inbox, contact, contactInbox, msg)
if err != nil {
return err
}
result.Conversation = conversation
result.ConversationCreated = conversationCreated
message, err := p.createMessage(ctx, tx, inbox, conversation, contact, msg)
if err != nil {
@@ -87,6 +96,7 @@ func (p *IncomingPersister) PersistIncoming(ctx context.Context, inbox *model.In
if err != nil {
return nil, err
}
p.dispatchIncomingEvents(ctx, inbox, &result)
return &result, nil
}
@@ -107,10 +117,16 @@ func (p *IncomingPersister) UpdateMessageStatus(ctx context.Context, inbox *mode
if err := tx.Model(&message).Update("status", string(status)).Error; err != nil {
return err
}
message.Status = string(status)
if message.SenderID == nil {
p.dispatchMessageStatusEvent(ctx, inbox, &message)
return nil
}
return p.upsertDeliveryStatus(ctx, tx, &message, *message.SenderID, status, occurredAt)
if err := p.upsertDeliveryStatus(ctx, tx, &message, *message.SenderID, status, occurredAt); err != nil {
return err
}
p.dispatchMessageStatusEvent(ctx, inbox, &message)
return nil
})
}
@@ -136,7 +152,19 @@ func (p *IncomingPersister) UpdateContactConversationMessagesStatus(ctx context.
if occurredAt != nil {
query = query.Where("messages.created_at <= ?", *occurredAt)
}
return query.Update("status", string(status)).Error
var messages []model.Message
if err := query.Select("messages.*").Find(&messages).Error; err != nil {
return err
}
if err := query.Update("status", string(status)).Error; err != nil {
return err
}
for i := range messages {
messages[i].Status = string(status)
p.dispatchMessageStatusEvent(ctx, inbox, &messages[i])
}
return nil
})
}
@@ -164,7 +192,69 @@ func (p *IncomingPersister) upsertDeliveryStatus(ctx context.Context, tx *gorm.D
return tx.WithContext(ctx).Save(&delivery).Error
}
func (p *IncomingPersister) resolveOrCreateContactInbox(ctx context.Context, tx *gorm.DB, inbox *model.Inbox, msg *channel.IncomingMessage, senderID string) (*model.Contact, *model.ContactInbox, error) {
func (p *IncomingPersister) dispatchIncomingEvents(ctx context.Context, inbox *model.Inbox, result *IncomingPersistResult) {
if p.dispatcher == nil || result == nil || result.Duplicate {
return
}
if result.ContactCreated && result.Contact != nil {
p.dispatch(ctx, channel.EventContactCreated, inbox, result, nil)
}
if result.Conversation != nil {
if result.ConversationCreated {
p.dispatch(ctx, channel.EventConversationCreated, inbox, result, nil)
p.dispatch(ctx, channel.EventConversationOpened, inbox, result, nil)
} else {
p.dispatch(ctx, channel.EventConversationUpdated, inbox, result, nil)
}
}
if result.Message != nil {
p.dispatch(ctx, channel.EventMessageCreated, inbox, result, result.Message)
p.dispatch(ctx, channel.EventMessageIncoming, inbox, result, result.Message)
}
}
func (p *IncomingPersister) dispatchMessageStatusEvent(ctx context.Context, inbox *model.Inbox, message *model.Message) {
if p.dispatcher == nil || inbox == nil || message == nil {
return
}
result := &IncomingPersistResult{Message: message}
p.dispatch(ctx, channel.EventMessageStatusUpdated, inbox, result, message)
}
func (p *IncomingPersister) dispatch(ctx context.Context, eventType channel.EventType, inbox *model.Inbox, result *IncomingPersistResult, message *model.Message) {
if p.dispatcher == nil || inbox == nil {
return
}
channelType := channel.ChannelType(inbox.ChannelType)
event := channel.NewChannelEvent(eventType, channelType, inbox.AccountID, inbox.ID)
if result != nil {
if result.Conversation != nil {
event.ConversationID = result.Conversation.ID
event.Data["conversation"] = result.Conversation
}
if result.Contact != nil {
event.ContactID = result.Contact.ID
event.Data["contact"] = result.Contact
}
if result.Message != nil {
event.Data["message"] = result.Message
}
}
if message != nil {
event.ConversationID = message.ConversationID
event.Data["message"] = message
if message.SenderID != nil {
event.ContactID = *message.SenderID
}
}
event.Data["inbox"] = inbox
if err := p.dispatcher.DispatchAsync(ctx, event); err != nil {
// Chatwoot's async side effects should not make provider webhooks fail.
_ = err
}
}
func (p *IncomingPersister) resolveOrCreateContactInbox(ctx context.Context, tx *gorm.DB, inbox *model.Inbox, msg *channel.IncomingMessage, senderID string) (*model.Contact, *model.ContactInbox, bool, error) {
var contactInbox model.ContactInbox
if err := tx.WithContext(ctx).Preload("Contact").Where("inbox_id = ? AND source_id = ?", inbox.ID, senderID).First(&contactInbox).Error; err == nil {
contact := contactInbox.Contact
@@ -174,15 +264,15 @@ func (p *IncomingPersister) resolveOrCreateContactInbox(ctx context.Context, tx
}
if len(updates) > 0 {
if err := tx.WithContext(ctx).Model(&contact).Updates(updates).Error; err != nil {
return nil, nil, err
return nil, nil, false, err
}
if err := tx.WithContext(ctx).First(&contact, contact.ID).Error; err != nil {
return nil, nil, err
return nil, nil, false, err
}
}
return &contact, &contactInbox, nil
return &contact, &contactInbox, false, nil
} else if err != gorm.ErrRecordNotFound {
return nil, nil, err
return nil, nil, false, err
}
name := msg.SenderName
@@ -202,7 +292,7 @@ func (p *IncomingPersister) resolveOrCreateContactInbox(ctx context.Context, tx
CustomAttributes: datatypes.JSON("{}"),
}
if err := tx.WithContext(ctx).Create(&contact).Error; err != nil {
return nil, nil, err
return nil, nil, false, err
}
contactInbox = model.ContactInbox{
@@ -212,12 +302,12 @@ func (p *IncomingPersister) resolveOrCreateContactInbox(ctx context.Context, tx
PubsubToken: uuid.NewString(),
}
if err := tx.WithContext(ctx).Create(&contactInbox).Error; err != nil {
return nil, nil, err
return nil, nil, false, err
}
return &contact, &contactInbox, nil
return &contact, &contactInbox, true, nil
}
func (p *IncomingPersister) resolveOrCreateConversation(ctx context.Context, tx *gorm.DB, inbox *model.Inbox, contact *model.Contact, contactInbox *model.ContactInbox, msg *channel.IncomingMessage) (*model.Conversation, error) {
func (p *IncomingPersister) resolveOrCreateConversation(ctx context.Context, tx *gorm.DB, inbox *model.Inbox, contact *model.Contact, contactInbox *model.ContactInbox, msg *channel.IncomingMessage) (*model.Conversation, bool, error) {
var conversation model.Conversation
if err := tx.WithContext(ctx).
Where("account_id = ? AND inbox_id = ? AND contact_id = ? AND status = ?", inbox.AccountID, inbox.ID, contact.ID, model.ConversationStatusOpen).
@@ -227,9 +317,9 @@ func (p *IncomingPersister) resolveOrCreateConversation(ctx context.Context, tx
"last_activity_at": now,
"last_message_at": now,
}).Error
return &conversation, nil
return &conversation, false, nil
} else if err != gorm.ErrRecordNotFound {
return nil, err
return nil, false, err
}
now := time.Now().Unix()
@@ -253,9 +343,9 @@ func (p *IncomingPersister) resolveOrCreateConversation(ctx context.Context, tx
CustomAttributes: datatypes.JSON("{}"),
}
if err := tx.WithContext(ctx).Create(&conversation).Error; err != nil {
return nil, err
return nil, false, err
}
return &conversation, nil
return &conversation, true, nil
}
func (p *IncomingPersister) createMessage(ctx context.Context, tx *gorm.DB, inbox *model.Inbox, conversation *model.Conversation, contact *model.Contact, msg *channel.IncomingMessage) (*model.Message, error) {
internal/handler/webhook/line_webhook.go
+3 -2
View File
@@ -12,6 +12,7 @@ import (
"io"
"net/http"
"github.com/gochat/gochat/internal/channel"
linechannel "github.com/gochat/gochat/internal/channel/line"
"github.com/gochat/gochat/internal/model"
channelmodel "github.com/gochat/gochat/internal/model/channel"
@@ -31,13 +32,13 @@ type LineWebhookHandler struct {
}
// NewLineWebhookHandler creates a LINE webhook handler for Gin integration.
func NewLineWebhookHandler(lineWebhook *linechannel.WebhookHandler, pipeline *linechannel.IncomingProcessor, service *linechannel.LineService, db *gorm.DB) *LineWebhookHandler {
func NewLineWebhookHandler(lineWebhook *linechannel.WebhookHandler, pipeline *linechannel.IncomingProcessor, service *linechannel.LineService, db *gorm.DB, dispatcher ...*channel.Dispatcher) *LineWebhookHandler {
return &LineWebhookHandler{
lineWebhook: lineWebhook,
pipeline: pipeline,
service: service,
db: db,
persister: NewIncomingPersister(db),
persister: NewIncomingPersister(db, dispatcher...),
}
}
internal/handler/webhook/telegram_webhook.go
+3 -1
View File
@@ -8,6 +8,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/gochat/gochat/internal/channel"
channelprovider "github.com/gochat/gochat/internal/channel/provider"
telegramchannel "github.com/gochat/gochat/internal/channel/telegram"
"github.com/gochat/gochat/internal/model"
@@ -44,12 +45,13 @@ func NewTelegramWebhookHandler(
provider *channelprovider.TelegramProvider,
telWebhook *telegramchannel.WebhookHandler,
db *gorm.DB,
dispatcher ...*channel.Dispatcher,
) *TelegramWebhookHandler {
return &TelegramWebhookHandler{
provider: provider,
telWebhook: telWebhook,
db: db,
persister: NewIncomingPersister(db),
persister: NewIncomingPersister(db, dispatcher...),
}
}
internal/handler/webhook/tiktok_webhook.go
+3 -2
View File
@@ -11,6 +11,7 @@ import (
"io"
"net/http"
"github.com/gochat/gochat/internal/channel"
tiktokchannel "github.com/gochat/gochat/internal/channel/tiktok"
"github.com/gochat/gochat/internal/model"
channelmodel "github.com/gochat/gochat/internal/model/channel"
@@ -29,12 +30,12 @@ type TikTokWebhookHandler struct {
}
// NewTikTokWebhookHandler creates a Gin-compatible TikTok webhook handler.
func NewTikTokWebhookHandler(tiktokWebhook *tiktokchannel.WebhookHandler, pipeline *tiktokchannel.IncomingProcessor, db *gorm.DB) *TikTokWebhookHandler {
func NewTikTokWebhookHandler(tiktokWebhook *tiktokchannel.WebhookHandler, pipeline *tiktokchannel.IncomingProcessor, db *gorm.DB, dispatcher ...*channel.Dispatcher) *TikTokWebhookHandler {
return &TikTokWebhookHandler{
tiktokWebhook: tiktokWebhook,
pipeline: pipeline,
db: db,
persister: NewIncomingPersister(db),
persister: NewIncomingPersister(db, dispatcher...),
}
}
internal/handler/webhook/twilio_webhook.go
+3 -2
View File
@@ -14,6 +14,7 @@ import (
"fmt"
"net/http"
"github.com/gochat/gochat/internal/channel"
twiliochannel "github.com/gochat/gochat/internal/channel/twilio"
"github.com/gochat/gochat/internal/model"
channelmodel "github.com/gochat/gochat/internal/model/channel"
@@ -31,11 +32,11 @@ type TwilioWebhookHandler struct {
}
// NewTwilioWebhookHandler creates a Twilio SMS webhook handler for Gin integration.
func NewTwilioWebhookHandler(twilioWebhook *twiliochannel.WebhookHandler, db *gorm.DB) *TwilioWebhookHandler {
func NewTwilioWebhookHandler(twilioWebhook *twiliochannel.WebhookHandler, db *gorm.DB, dispatcher ...*channel.Dispatcher) *TwilioWebhookHandler {
return &TwilioWebhookHandler{
twilioWebhook: twilioWebhook,
db: db,
persister: NewIncomingPersister(db),
persister: NewIncomingPersister(db, dispatcher...),
}
}
internal/handler/webhook/webhook_lookup_test.go
+37 -2
View File
@@ -2,6 +2,7 @@ package webhook
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/base64"
@@ -21,6 +22,17 @@ import (
"gorm.io/gorm"
)
type recordingListener struct {
events []*channel.ChannelEvent
}
func (l *recordingListener) Name() string { return "recording-listener" }
func (l *recordingListener) OnEvent(ctx context.Context, event *channel.ChannelEvent) error {
l.events = append(l.events, event)
return nil
}
func newWebhookLookupTestDB(t *testing.T) *gorm.DB {
t.Helper()
@@ -51,7 +63,10 @@ func newWebhookLookupTestDB(t *testing.T) *gorm.DB {
func TestIncomingPersisterUpdatesMessageStatus(t *testing.T) {
db := newWebhookLookupTestDB(t)
inbox := seedWebhookInbox(t, db, "telegram")
persister := NewIncomingPersister(db)
dispatcher := channel.NewDispatcher()
listener := &recordingListener{}
dispatcher.Register(listener)
persister := NewIncomingPersister(db, dispatcher)
msg := &channel.IncomingMessage{
ChannelType: channel.ChannelTelegram,
@@ -86,12 +101,18 @@ func TestIncomingPersisterUpdatesMessageStatus(t *testing.T) {
if delivery.Status != model.MessageStatusRead {
t.Fatalf("expected delivery read, got %s", delivery.Status)
}
if !listenerSaw(listener, channel.EventMessageStatusUpdated) {
t.Fatalf("expected message.status_updated event, got %#v", listener.events)
}
}
func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) {
db := newWebhookLookupTestDB(t)
inbox := seedWebhookInbox(t, db, "telegram")
persister := NewIncomingPersister(db)
dispatcher := channel.NewDispatcher()
listener := &recordingListener{}
dispatcher.Register(listener)
persister := NewIncomingPersister(db, dispatcher)
msg := &channel.IncomingMessage{
ChannelType: channel.ChannelTelegram,
@@ -141,6 +162,20 @@ func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) {
if messageCount != 2 {
t.Fatalf("expected 2 persisted messages after duplicate skip, got %d", messageCount)
}
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 listenerSaw(listener *recordingListener, eventType channel.EventType) bool {
for _, event := range listener.events {
if event.Type == eventType {
return true
}
}
return false
}
func shopifyHMAC(secret string, body []byte) string {
internal/handler/webhook/whatsapp_webhook.go
+2 -2
View File
@@ -58,11 +58,11 @@ type WhatsAppWebhookHandler struct {
// sub-package's WebhookHandler. The provider is stored for future use (e.g.
// health checks or direct API calls) while waWebhook is the core handler that
// processes verification and event requests.
func NewWhatsAppWebhookHandler(provider *whatsapp.WhatsAppProvider, waWebhook *whatsapp.WebhookHandler, db *gorm.DB) *WhatsAppWebhookHandler {
func NewWhatsAppWebhookHandler(provider *whatsapp.WhatsAppProvider, waWebhook *whatsapp.WebhookHandler, db *gorm.DB, dispatcher ...*channel.Dispatcher) *WhatsAppWebhookHandler {
h := &WhatsAppWebhookHandler{
provider: provider,
waWebhook: waWebhook,
persister: NewIncomingPersister(db),
persister: NewIncomingPersister(db, dispatcher...),
}
if waWebhook != nil && h.persister != nil {
waWebhook.SetIncomingPersister(whatsAppPersisterAdapter{persister: h.persister})