feat(webhook): persist incoming provider messages

docs/CHATWOOT_PARITY_DEVELOPMENT_PLAN.md

@@ -60,6 +60,7 @@ This ledger records the committed parity checkpoints that future slices should b
 | `6aa62c6 docs: consolidate chatwoot parity roadmap` | Promoted Hermes-era plans into this master tracker; locked user decisions; added milestone, slice, enterprise, and webhook provider tracking. | Documentation-only checkpoint. | B1/P6.7 selected as next implementation slice. |
 | `9e3f561 feat(webhook): align chatwoot ingress routes` | Added Chatwoot public webhook paths for Twitter, Telegram, LINE, SMS/Twilio, WhatsApp, Instagram, TikTok, and Shopify; removed generic fallback success masking. | `go test ./...`; route dump regenerated with `TOTAL: 801`; route parity stayed `251 exact, 0 missing`. | Provider-specific lookup and verification moved to review. |
 | `bc7da9e feat(webhook): implement instagram and shopify ingress` | Implemented Instagram verify/event handling, Shopify HMAC/redact/event forwarding, and WhatsApp verify-token/app-secret signature corrections. | Focused webhook tests, `go test ./...`, route dump `TOTAL: 801`, route parity `251 exact, 0 missing`, `git diff --check`. | Remaining P6.7 work is durable incoming-message persistence and provider dispatch parity. |
+| Working tree | Added provider incoming-message persistence boundary and wired Telegram, LINE, SMS/Twilio, WhatsApp, Facebook/Instagram, and TikTok parsed incoming messages into ContactInbox, Conversation, and Message storage. | Focused webhook/channel tests passed; full `go test ./...` passed. | Continue P6.7 review with delivery/read receipt status updates and async dispatch/events. |
 
 ## Next Slice Contract
 
@@ -73,6 +74,12 @@ Next implementation slice: finish B1/P6.7 webhook ingress by replacing provider
 | N4 | Regenerate route artifacts only if routes change; otherwise preserve `TOTAL: 801` and `251 exact, 0 missing`. | `cmd/dump_routes`, `cmd/route_parity`. | Route commands run when applicable. |
 | N5 | Update this tracker in the same commit with P6.7 provider statuses and a progress-log entry. | This document. | `git diff --check`; `go test ./...` for Go changes. |
 
+Current N1/N2 implementation checkpoint:
+
+- Added `IncomingPersister` as the shared durable webhook boundary. It dedupes messages by `inbox_id + source_id`, resolves or creates `ContactInbox` by `inbox_id + sender source_id`, reuses the latest open conversation for the contact/inbox, and creates incoming `Message` records with content attributes and provider metadata.
+- Wired parsed incoming messages from Telegram, LINE, Twilio SMS, WhatsApp, Facebook/Instagram, and TikTok into the persister. Provider verification and provider-safe acknowledgement behavior remain in their existing handlers.
+- Added regression coverage for direct persistence, duplicate suppression, and Telegram webhook-to-message persistence. Broader provider-specific persistence assertions remain required before P6.7 is `Done`.
+
 ## Immediate Execution Queue
 
 This is the ordered queue for the next implementation slices. Do not skip the route and test gates even when working on deeper business behavior.
@@ -589,3 +596,4 @@ Verification milestone gates:
 - 2026-06-04: Consolidated the Hermes-era planning into this master tracker. Added the locked decision ledger, end-to-end milestone map, ordered slice backlog, enterprise work package breakdown, and detailed P6.7 provider webhook ingress checklist. Current next implementation slice is B1/P6.7 webhook ingress.
 - 2026-06-04: Started P6.7 webhook ingress parity. Added Chatwoot public webhook paths for Twitter, Telegram, LINE, SMS/Twilio, WhatsApp, Instagram, TikTok, and Shopify; removed the generic success fallback so unsupported providers no longer return placeholder success. Telegram, LINE, Twilio SMS, and TikTok handlers now resolve inboxes through provider channel records instead of inbox-id placeholders; TikTok model column naming now matches existing repository queries. Added provider lookup tests and router boot coverage. Regenerated route dump: `TOTAL: 801`; tracked route parity remains `251 exact, 0 missing`.
 - 2026-06-04: Continued P6.7 webhook ingress parity. Instagram `/webhooks/instagram` now performs Chatwoot-style global verify-token challenge handling, verifies Meta signatures against env/channel app secrets, resolves Instagram inboxes from webhook sender/recipient IDs, and dispatches parsed DM/comment events through the existing Meta pipeline boundary. Shopify `/webhooks/shopify` now verifies `X-Shopify-Hmac-SHA256` with `SHOPIFY_CLIENT_SECRET`, handles `shop/redact` by deleting matching Shopify integration hooks, and forwards supported topics to the existing Shopify event processor. WhatsApp verification now queries channels by verify token directly, and Cloud API signature verification uses app secrets from provider config/env instead of access tokens. Added focused webhook tests; route dump remains `TOTAL: 801` and tracked route parity remains `251 exact, 0 missing`.
+- 2026-06-04: Added the P6.7 incoming persistence boundary. Provider webhook handlers now persist normalized incoming messages into `contacts`, `contact_inboxes`, open `conversations`, and incoming `messages` instead of only parsing/logging them. Telegram, LINE, Twilio SMS, WhatsApp, Facebook/Instagram, and TikTok are wired through the shared persister; duplicates are skipped by `inbox_id + source_id`. Added direct persister coverage and Telegram webhook persistence coverage. Focused webhook/channel tests and full `go test ./...` passed.

internal/app/bootstrap.go

@@ -415,7 +415,7 @@ func Bootstrap(env string) (*App, error) {
 
 	// Create WhatsApp webhook handler (Gin HTTP handler for WA webhook endpoints)
 	waWebhook := whatsappchannel.NewWebhookHandler(waProvider)
-	whatsappWebhookHandler := webhook.NewWhatsAppWebhookHandler(waProvider, waWebhook)
+	whatsappWebhookHandler := webhook.NewWhatsAppWebhookHandler(waProvider, waWebhook, db)
 
 	// Step 8d: Wire TikTok channel provider (Business API)
 	// TikTok requires service/repo/pipeline deps (like WhatsApp), so wire here in bootstrap.

internal/channel/tiktok/pipeline.go

@@ -48,21 +48,19 @@ func NewIncomingProcessor(service *TikTokService, repo *Repository) *IncomingPro
 
 // ProcessUpdate handles a TikTok webhook event by transforming it into an IncomingMessage
 // and then processing it through gochat's pipeline stages.
-func (p *IncomingProcessor) ProcessUpdate(ctx context.Context, inbox *model.Inbox, rawEvent TikTokWebhookEvent) error {
+func (p *IncomingProcessor) ProcessUpdate(ctx context.Context, inbox *model.Inbox, rawEvent TikTokWebhookEvent) (*channelpkg.IncomingMessage, error) {
 	incomingMsg, err := p.transformToIncomingMessage(ctx, inbox, rawEvent)
 	if err != nil {
-		return fmt.Errorf("tiktok incoming pipeline: transform failed: %w", err)
+		return nil, fmt.Errorf("tiktok incoming pipeline: transform failed: %w", err)
 	}
 	if incomingMsg == nil {
-		return nil // no processable content
+		return nil, nil // no processable content
 	}
 
 	applogger.L().Infof("TikTok incoming: channel=%s sender=%s source_id=%s inbox=%d",
 		incomingMsg.ChannelType, incomingMsg.SenderID, incomingMsg.SourceID, inbox.ID)
 
-	// The IncomingMessage is returned to the provider for full pipeline processing
-	// (contact resolution, conversation creation, message persistence, event dispatch)
-	return nil
+	return incomingMsg, nil
 }
 
 // transformToIncomingMessage converts a raw TikTok webhook event into an IncomingMessage.
@@ -104,8 +102,8 @@ func (p *IncomingProcessor) transformMessageReceived(ctx context.Context, inbox
 		Content:     payload.Content,
 		ContentType: mapContentType(payload.ContentType),
 
-		InboxID:   inbox.ID,
-		AccountID: inbox.AccountID,
+		InboxID:    inbox.ID,
+		AccountID:  inbox.AccountID,
 		ReceivedAt: time.Unix(payload.Timestamp, 0),
 
 		Extra: channelpkg.ChannelConfig{
@@ -163,4 +161,4 @@ func mapContentType(tiktokType string) channelpkg.ContentType {
 // getChannelFromInbox resolves the ChannelTikTok model from an Inbox.
 func (p *IncomingProcessor) getChannelFromInbox(ctx context.Context, inbox *model.Inbox) (*channelmodel.ChannelTikTok, error) {
 	return p.repo.GetByInboxID(ctx, inbox.ID)
-}
+}

internal/channel/twilio/webhook_handler.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"net/http"
 
+	channelpkg "github.com/gochat/gochat/internal/channel"
 	"github.com/gochat/gochat/internal/model"
 	applogger "github.com/gochat/gochat/pkg/logger"
 )
@@ -43,11 +44,23 @@ func NewWebhookHandler(pipeline *IncomingProcessor, service *TwilioService) *Web
 // HandleInboundSMS processes an incoming SMS webhook from Twilio.
 // Twilio sends form-encoded data with fields like From, To, Body, etc.
 func (h *WebhookHandler) HandleInboundSMS(w http.ResponseWriter, r *http.Request, inbox *model.Inbox) {
-	if err := r.ParseForm(); err != nil {
-		applogger.L().Errorf("Twilio HandleInboundSMS: failed to parse form: %v", err)
-		http.Error(w, "Failed to parse form data", http.StatusBadRequest)
+	incomingMsg, err := h.ProcessInboundSMS(r, inbox)
+	if err != nil {
+		applogger.L().Errorf("Twilio HandleInboundSMS: pipeline process failed: %v", err)
+		writeTwiMLResponse(w, "")
 		return
 	}
+	if incomingMsg != nil {
+		applogger.L().Debugf("Twilio HandleInboundSMS: processed source_id=%s", incomingMsg.SourceID)
+	}
+	writeTwiMLResponse(w, "")
+}
+
+// ProcessInboundSMS parses an inbound Twilio webhook and returns the normalized incoming message.
+func (h *WebhookHandler) ProcessInboundSMS(r *http.Request, inbox *model.Inbox) (*channelpkg.IncomingMessage, error) {
+	if err := r.ParseForm(); err != nil {
+		return nil, fmt.Errorf("failed to parse form data: %w", err)
+	}
 
 	// Parse form data into InboundSMS struct
 	sms := InboundSMS{
@@ -83,21 +96,9 @@ func (h *WebhookHandler) HandleInboundSMS(w http.ResponseWriter, r *http.Request
 	ctx := r.Context()
 	incomingMsg, err := h.pipeline.ProcessInboundSMS(ctx, inbox, sms)
 	if err != nil {
-		applogger.L().Errorf("Twilio HandleInboundSMS: pipeline process failed: %v", err)
-		// Still respond with TwiML so Twilio doesn't retry
-		writeTwiMLResponse(w, "")
-		return
+		return nil, err
 	}
-	if incomingMsg == nil {
-		writeTwiMLResponse(w, "")
-		return
-	}
-
-	applogger.L().Debugf("Twilio HandleInboundSMS: processed from=%s", sms.From)
-
-	// Respond with empty TwiML — we don't auto-reply via webhook response
-	// Outbound replies are sent asynchronously through the SendMessage flow
-	writeTwiMLResponse(w, "")
+	return incomingMsg, nil
 }
 
 // HandleDeliveryStatus processes a Twilio delivery status callback.
@@ -139,8 +140,8 @@ func (h *WebhookHandler) HandleDeliveryStatus(w http.ResponseWriter, r *http.Req
 
 // TwiMLResponse represents a Twilio Markup Language response.
 type TwiMLResponse struct {
-	XMLName xml.Name    `xml:"Response"`
-	Message []TwiMLMsg  `xml:"Message,omitempty"`
+	XMLName xml.Name   `xml:"Response"`
+	Message []TwiMLMsg `xml:"Message,omitempty"`
 }
 
 // TwiMLMsg represents a TwiML <Message> element.

internal/channel/whatsapp/provider.go

@@ -159,12 +159,12 @@ func (p *WhatsAppProvider) ValidateConfig(ctx context.Context, config channel.Ch
 // DefaultConfig returns default configuration for WhatsApp channel.
 func (p *WhatsAppProvider) DefaultConfig() channel.ChannelConfig {
 	return channel.ChannelConfig{
-		"provider":              "whatsapp_cloud",
-		"phone_number_id":       "",
-		"business_account_id":   "",
-		"access_token":          "",
-		"webhook_verify_token":  "",
-		"webhook_url":           "",
+		"provider":             "whatsapp_cloud",
+		"phone_number_id":      "",
+		"business_account_id":  "",
+		"access_token":         "",
+		"webhook_verify_token": "",
+		"webhook_url":          "",
 	}
 }
 
@@ -221,6 +221,18 @@ func (p *WhatsAppProvider) OnDestroy(ctx context.Context, inbox *model.Inbox, co
 // ProcessIncoming transforms raw WhatsApp webhook payload into IncomingMessage.
 // Reference: Chatwoot's WebhooksController + IncomingMessageService
 func (p *WhatsAppProvider) ProcessIncoming(ctx context.Context, inbox *model.Inbox, rawPayload []byte) (*channel.IncomingMessage, error) {
+	messages, err := p.ProcessIncomingMessages(ctx, inbox, rawPayload)
+	if err != nil {
+		return nil, err
+	}
+	if len(messages) == 0 {
+		return nil, nil
+	}
+	return messages[0], nil
+}
+
+// ProcessIncomingMessages transforms a raw WhatsApp webhook payload into all normalized messages.
+func (p *WhatsAppProvider) ProcessIncomingMessages(ctx context.Context, inbox *model.Inbox, rawPayload []byte) ([]*channel.IncomingMessage, error) {
 	event := &WAWebhookEvent{}
 	if err := json.Unmarshal(rawPayload, event); err != nil {
 		return nil, fmt.Errorf("failed to parse WhatsApp webhook payload: %w", err)
@@ -230,12 +242,7 @@ func (p *WhatsAppProvider) ProcessIncoming(ctx context.Context, inbox *model.Inb
 	if err != nil {
 		return nil, fmt.Errorf("failed to process WhatsApp incoming pipeline: %w", err)
 	}
-
-	if len(messages) == 0 {
-		return nil, nil
-	}
-
-	return messages[0], nil
+	return messages, nil
 }
 
 // ValidateWebhookRequest verifies WhatsApp webhook callback authenticity.
@@ -631,4 +638,4 @@ func (p *WhatsAppProvider) get360DialogContactProfile(ctx context.Context, phone
 			"wa_id": phone,
 		},
 	}, nil
-}
+}

internal/channel/whatsapp/webhook_handler.go

@@ -15,6 +15,7 @@ package whatsapp
 // 3. Return 200 OK immediately (WhatsApp expects fast response)
 
 import (
+	"context"
 	"crypto/hmac"
 	"crypto/sha256"
 	"encoding/hex"
@@ -26,6 +27,7 @@ import (
 
 	"github.com/gin-gonic/gin"
 
+	"github.com/gochat/gochat/internal/channel"
 	"github.com/gochat/gochat/internal/model"
 	channelmodel "github.com/gochat/gochat/internal/model/channel"
 	applogger "github.com/gochat/gochat/pkg/logger"
@@ -33,7 +35,12 @@ import (
 
 // WebhookHandler processes WhatsApp webhook requests.
 type WebhookHandler struct {
-	provider *WhatsAppProvider
+	provider  *WhatsAppProvider
+	persister IncomingPersister
+}
+
+type IncomingPersister interface {
+	PersistIncoming(ctx context.Context, inbox *model.Inbox, msg *channel.IncomingMessage) (interface{}, error)
 }
 
 // NewWebhookHandler creates a WhatsApp webhook handler.
@@ -43,6 +50,11 @@ func NewWebhookHandler(provider *WhatsAppProvider) *WebhookHandler {
 	}
 }
 
+// SetIncomingPersister wires the durable message persistence boundary used after parsing.
+func (h *WebhookHandler) SetIncomingPersister(persister IncomingPersister) {
+	h.persister = persister
+}
+
 // HandleVerification handles GET requests for WhatsApp webhook verification.
 // Meta Cloud API sends: hub.mode=subscribe, hub.verify_token=<token>, hub.challenge=<string>
 // We respond with hub.challenge if verify_token matches the channel's WebhookVerifyToken.
@@ -129,9 +141,18 @@ func (h *WebhookHandler) HandleWebhookEvent(c *gin.Context) {
 
 	// Delegate to provider's ProcessIncoming for full pipeline processing
 	if h.provider != nil {
-		_, processErr := h.provider.ProcessIncoming(c.Request.Context(), inbox, body)
+		messages, processErr := h.provider.ProcessIncomingMessages(c.Request.Context(), inbox, body)
 		if processErr != nil {
 			applogger.L().Error("WhatsApp webhook: message processing failed", "error", processErr)
+		} else {
+			for _, incomingMsg := range messages {
+				if h.persister == nil || incomingMsg == nil {
+					continue
+				}
+				if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+					applogger.L().Error("WhatsApp webhook: message persistence failed", "source_id", incomingMsg.SourceID, "error", persistErr)
+				}
+			}
 		}
 	}
 

internal/handler/webhook/facebook_webhook.go

@@ -47,6 +47,7 @@ type FacebookWebhookHandler struct {
 	igProvider    *fbchannel.InstagramProvider
 	webhookParser *fbchannel.WebhookParser
 	db            *gorm.DB
+	persister     *IncomingPersister
 }
 
 // NewFacebookWebhookHandler creates a Facebook/Instagram webhook handler for Gin integration.
@@ -60,6 +61,7 @@ func NewFacebookWebhookHandler(
 		igProvider:    igProvider,
 		webhookParser: fbchannel.NewWebhookParser(),
 		db:            db,
+		persister:     NewIncomingPersister(db),
 	}
 }
 
@@ -218,9 +220,12 @@ func (h *FacebookWebhookHandler) HandleFacebookWebhook(c *gin.Context) {
 				continue
 			}
 			if commentMsg != nil {
-				applogger.L().Infof("Facebook webhook: IG comment processed (inbox_id=%d, source_id=%s, type=%s)",
+				if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, commentMsg); persistErr != nil {
+					applogger.L().Errorf("Facebook webhook: IG comment persist failed (source_id=%s): %v", commentMsg.SourceID, persistErr)
+					continue
+				}
+				applogger.L().Infof("Facebook webhook: IG comment persisted (inbox_id=%d, source_id=%s, type=%s)",
 					commentMsg.InboxID, commentMsg.SourceID, event.EventType)
-				// TODO: Push to message broker/dispatcher for persistence + notification
 			}
 			continue
 		}
@@ -257,10 +262,12 @@ func (h *FacebookWebhookHandler) HandleFacebookWebhook(c *gin.Context) {
 		}
 
 		if incomingMsg != nil {
-			applogger.L().Infof("Facebook webhook: message extracted (inbox_id=%d, source_id=%s, type=%s)",
+			if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+				applogger.L().Errorf("Facebook webhook: persist message failed (source_id=%s): %v", incomingMsg.SourceID, persistErr)
+				continue
+			}
+			applogger.L().Infof("Facebook webhook: message persisted (inbox_id=%d, source_id=%s, type=%s)",
 				incomingMsg.InboxID, incomingMsg.SourceID, event.EventType)
-			// TODO: Push to message broker/dispatcher for persistence + notification
-			// Reference: Chatwoot pushes to IncomingMessageService → Conversation + Message creation
 		}
 	}
 
@@ -430,7 +437,11 @@ func (h *FacebookWebhookHandler) processInstagramEvent(c *gin.Context, inbox *mo
 			applogger.L().Errorf("Instagram webhook: comment processing failed: %v", err)
 		}
 		if commentMsg != nil {
-			applogger.L().Infof("Instagram webhook: comment processed (inbox_id=%d, source_id=%s, type=%s)", commentMsg.InboxID, commentMsg.SourceID, event.EventType)
+			if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, commentMsg); persistErr != nil {
+				applogger.L().Errorf("Instagram webhook: comment persist failed (source_id=%s): %v", commentMsg.SourceID, persistErr)
+				return
+			}
+			applogger.L().Infof("Instagram webhook: comment persisted (inbox_id=%d, source_id=%s, type=%s)", commentMsg.InboxID, commentMsg.SourceID, event.EventType)
 		}
 		return
 	}
@@ -444,7 +455,11 @@ func (h *FacebookWebhookHandler) processInstagramEvent(c *gin.Context, inbox *mo
 		return
 	}
 	if incomingMsg != nil {
-		applogger.L().Infof("Instagram webhook: message extracted (inbox_id=%d, source_id=%s, type=%s)", incomingMsg.InboxID, incomingMsg.SourceID, event.EventType)
+		if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+			applogger.L().Errorf("Instagram webhook: persist message failed (source_id=%s): %v", incomingMsg.SourceID, persistErr)
+			return
+		}
+		applogger.L().Infof("Instagram webhook: message persisted (inbox_id=%d, source_id=%s, type=%s)", incomingMsg.InboxID, incomingMsg.SourceID, event.EventType)
 	}
 }
 

internal/handler/webhook/incoming_persister.go

@@ -0,0 +1,254 @@
+package webhook
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"gorm.io/datatypes"
+	"gorm.io/gorm"
+
+	"github.com/gochat/gochat/internal/channel"
+	"github.com/gochat/gochat/internal/model"
+)
+
+// IncomingPersister is the durable boundary after provider-specific webhook parsing.
+// Reference: Chatwoot IncomingMessageService creates ContactInbox, Conversation, and Message.
+type IncomingPersister struct {
+	db *gorm.DB
+}
+
+type IncomingPersistResult struct {
+	Contact      *model.Contact
+	ContactInbox *model.ContactInbox
+	Conversation *model.Conversation
+	Message      *model.Message
+	Duplicate    bool
+}
+
+func NewIncomingPersister(db *gorm.DB) *IncomingPersister {
+	if db == nil {
+		return nil
+	}
+	return &IncomingPersister{db: db}
+}
+
+func (p *IncomingPersister) PersistIncoming(ctx context.Context, inbox *model.Inbox, msg *channel.IncomingMessage) (*IncomingPersistResult, error) {
+	if p == nil || p.db == nil || inbox == nil || msg == nil {
+		return nil, nil
+	}
+	if msg.SourceID == "" {
+		return nil, 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")
+	}
+	if msg.Content == "" && len(msg.Attachments) == 0 {
+		return nil, fmt.Errorf("incoming message has no content or attachments")
+	}
+
+	var result IncomingPersistResult
+	err := p.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		var existingMessage model.Message
+		if err := tx.Where("inbox_id = ? AND source_id = ?", inbox.ID, msg.SourceID).First(&existingMessage).Error; err == nil {
+			result.Message = &existingMessage
+			result.Duplicate = true
+			return nil
+		} else if err != gorm.ErrRecordNotFound {
+			return err
+		}
+
+		contact, contactInbox, err := p.resolveOrCreateContactInbox(ctx, tx, inbox, msg, senderID)
+		if err != nil {
+			return err
+		}
+		result.Contact = contact
+		result.ContactInbox = contactInbox
+
+		conversation, err := p.resolveOrCreateConversation(ctx, tx, inbox, contact, contactInbox, msg)
+		if err != nil {
+			return err
+		}
+		result.Conversation = conversation
+
+		message, err := p.createMessage(ctx, tx, inbox, conversation, contact, msg)
+		if err != nil {
+			return err
+		}
+		result.Message = message
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &result, nil
+}
+
+func (p *IncomingPersister) resolveOrCreateContactInbox(ctx context.Context, tx *gorm.DB, inbox *model.Inbox, msg *channel.IncomingMessage, senderID string) (*model.Contact, *model.ContactInbox, 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
+		updates := map[string]interface{}{}
+		if msg.SenderName != "" && contact.Name != msg.SenderName {
+			updates["name"] = msg.SenderName
+		}
+		if len(updates) > 0 {
+			if err := tx.WithContext(ctx).Model(&contact).Updates(updates).Error; err != nil {
+				return nil, nil, err
+			}
+			if err := tx.WithContext(ctx).First(&contact, contact.ID).Error; err != nil {
+				return nil, nil, err
+			}
+		}
+		return &contact, &contactInbox, nil
+	} else if err != gorm.ErrRecordNotFound {
+		return nil, nil, err
+	}
+
+	name := msg.SenderName
+	if name == "" {
+		name = senderID
+	}
+	attrs := mergeChannelConfig(msg.SenderExtra, map[string]interface{}{
+		"source_id":    senderID,
+		"channel_type": string(msg.ChannelType),
+	})
+	contact := model.Contact{
+		AccountID:            inbox.AccountID,
+		Name:                 name,
+		Identifier:           senderID,
+		SourceID:             string(msg.ChannelType),
+		AdditionalAttributes: mustJSON(attrs),
+		CustomAttributes:     datatypes.JSON("{}"),
+	}
+	if err := tx.WithContext(ctx).Create(&contact).Error; err != nil {
+		return nil, nil, err
+	}
+
+	contactInbox = model.ContactInbox{
+		ContactID:   contact.ID,
+		InboxID:     inbox.ID,
+		SourceID:    senderID,
+		PubsubToken: uuid.NewString(),
+	}
+	if err := tx.WithContext(ctx).Create(&contactInbox).Error; err != nil {
+		return nil, nil, err
+	}
+	return &contact, &contactInbox, 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) {
+	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).
+		Order("id DESC").First(&conversation).Error; err == nil {
+		now := time.Now().Unix()
+		_ = tx.WithContext(ctx).Model(&conversation).Updates(map[string]interface{}{
+			"last_activity_at": now,
+			"last_message_at":  now,
+		}).Error
+		return &conversation, nil
+	} else if err != gorm.ErrRecordNotFound {
+		return nil, err
+	}
+
+	now := time.Now().Unix()
+	channelType := inbox.ChannelType
+	if msg.ChannelType != "" {
+		channelType = string(msg.ChannelType)
+	}
+	conversation = model.Conversation{
+		AccountID:            inbox.AccountID,
+		InboxID:              inbox.ID,
+		ContactID:            contact.ID,
+		ContactInboxID:       &contactInbox.ID,
+		Status:               string(model.ConversationStatusOpen),
+		Priority:             "none",
+		ChannelType:          channelType,
+		Channel:              inbox.ChannelType,
+		LastActivityAt:       &now,
+		LastMessageAt:        &now,
+		LastNonSysMsgAt:      &now,
+		AdditionalAttributes: mustJSON(mergeChannelConfig(msg.ConversationExtra, map[string]interface{}{"external_conversation_id": msg.ConversationID})),
+		CustomAttributes:     datatypes.JSON("{}"),
+	}
+	if err := tx.WithContext(ctx).Create(&conversation).Error; err != nil {
+		return nil, err
+	}
+	return &conversation, 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) {
+	contentAttrs := map[string]interface{}{}
+	if len(msg.Attachments) > 0 {
+		contentAttrs["attachments"] = msg.Attachments
+	}
+	if msg.ReplyToID != "" {
+		contentAttrs["in_reply_to"] = msg.ReplyToID
+	}
+	message := model.Message{
+		ConversationID:       conversation.ID,
+		AccountID:            inbox.AccountID,
+		InboxID:              inbox.ID,
+		SenderID:             &contact.ID,
+		SenderType:           string(model.SenderTypeContact),
+		Content:              msg.Content,
+		ContentType:          mapIncomingContentType(msg.ContentType),
+		Status:               string(model.MessageStatusSent),
+		MessageType:          string(model.MessageTypeIncoming),
+		SourceID:             msg.SourceID,
+		ContentAttributes:    mustJSON(contentAttrs),
+		AdditionalAttributes: mustJSON(msg.Extra),
+	}
+	if err := tx.WithContext(ctx).Create(&message).Error; err != nil {
+		return nil, err
+	}
+	return &message, nil
+}
+
+func mapIncomingContentType(contentType channel.ContentType) string {
+	switch contentType {
+	case channel.ContentImage:
+		return string(model.MessageContentTypeImage)
+	case channel.ContentAudio:
+		return string(model.MessageContentTypeAudio)
+	case channel.ContentVideo:
+		return string(model.MessageContentTypeVideo)
+	case channel.ContentFile:
+		return string(model.MessageContentTypeFile)
+	case channel.ContentLocation:
+		return string(model.MessageContentTypeLocation)
+	default:
+		return string(model.MessageContentTypeText)
+	}
+}
+
+func mergeChannelConfig(base channel.ChannelConfig, extra map[string]interface{}) map[string]interface{} {
+	merged := map[string]interface{}{}
+	for k, v := range base {
+		merged[k] = v
+	}
+	for k, v := range extra {
+		if v != "" && v != nil {
+			merged[k] = v
+		}
+	}
+	return merged
+}
+
+func mustJSON(value interface{}) datatypes.JSON {
+	if value == nil {
+		return datatypes.JSON("{}")
+	}
+	data, err := json.Marshal(value)
+	if err != nil || len(data) == 0 {
+		return datatypes.JSON("{}")
+	}
+	return datatypes.JSON(data)
+}

internal/handler/webhook/line_webhook.go

@@ -27,6 +27,7 @@ type LineWebhookHandler struct {
 	pipeline    *linechannel.IncomingProcessor
 	service     *linechannel.LineService
 	db          *gorm.DB
+	persister   *IncomingPersister
 }
 
 // NewLineWebhookHandler creates a LINE webhook handler for Gin integration.
@@ -36,6 +37,7 @@ func NewLineWebhookHandler(lineWebhook *linechannel.WebhookHandler, pipeline *li
 		pipeline:    pipeline,
 		service:     service,
 		db:          db,
+		persister:   NewIncomingPersister(db),
 	}
 }
 
@@ -100,8 +102,11 @@ func (h *LineWebhookHandler) HandleLineWebhook(c *gin.Context) {
 			continue
 		}
 		if incomingMsg != nil {
-			applogger.L().Debugf("LINE webhook: processed event type=%s sender=%s",
-				event.Type, incomingMsg.SenderID)
+			if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+				applogger.L().Errorf("LINE webhook: persist event failed for inbox %d source_id=%s: %v", inbox.ID, incomingMsg.SourceID, persistErr)
+				continue
+			}
+			applogger.L().Debugf("LINE webhook: persisted event type=%s sender=%s", event.Type, incomingMsg.SenderID)
 		}
 	}
 

internal/handler/webhook/telegram_webhook.go

@@ -36,6 +36,7 @@ type TelegramWebhookHandler struct {
 	provider   *channelprovider.TelegramProvider
 	telWebhook *telegramchannel.WebhookHandler
 	db         *gorm.DB
+	persister  *IncomingPersister
 }
 
 // NewTelegramWebhookHandler creates a Telegram webhook handler for Gin integration.
@@ -48,6 +49,7 @@ func NewTelegramWebhookHandler(
 		provider:   provider,
 		telWebhook: telWebhook,
 		db:         db,
+		persister:  NewIncomingPersister(db),
 	}
 }
 
@@ -115,9 +117,12 @@ func (h *TelegramWebhookHandler) HandleTelegramWebhook(c *gin.Context) {
 	if err != nil {
 		applogger.L().Errorf("Telegram webhook: message processing failed (update_id=%d): %v", update.UpdateID, err)
 	} else if incomingMsg != nil {
-		applogger.L().Infof("Telegram webhook: message processed (inbox_id=%d, source_id=%s)",
-			incomingMsg.InboxID, incomingMsg.SourceID)
-		// TODO: Push to message broker/dispatcher for persistence + notification
+		if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+			applogger.L().Errorf("Telegram webhook: persist message failed (update_id=%d source_id=%s): %v", update.UpdateID, incomingMsg.SourceID, persistErr)
+		} else {
+			applogger.L().Infof("Telegram webhook: message persisted (inbox_id=%d, source_id=%s)",
+				incomingMsg.InboxID, incomingMsg.SourceID)
+		}
 	}
 
 	// Always return 200 OK to Telegram — it retries if not 200

internal/handler/webhook/tiktok_webhook.go

@@ -25,6 +25,7 @@ type TikTokWebhookHandler struct {
 	tiktokWebhook *tiktokchannel.WebhookHandler
 	pipeline      *tiktokchannel.IncomingProcessor
 	db            *gorm.DB
+	persister     *IncomingPersister
 }
 
 // NewTikTokWebhookHandler creates a Gin-compatible TikTok webhook handler.
@@ -33,6 +34,7 @@ func NewTikTokWebhookHandler(tiktokWebhook *tiktokchannel.WebhookHandler, pipeli
 		tiktokWebhook: tiktokWebhook,
 		pipeline:      pipeline,
 		db:            db,
+		persister:     NewIncomingPersister(db),
 	}
 }
 
@@ -74,11 +76,17 @@ func (h *TikTokWebhookHandler) HandleTikTokWebhook(c *gin.Context) {
 	}
 
 	// Process the event via the pipeline
-	if err := h.pipeline.ProcessUpdate(c.Request.Context(), inbox, *event); err != nil {
+	incomingMsg, err := h.pipeline.ProcessUpdate(c.Request.Context(), inbox, *event)
+	if err != nil {
 		applogger.L().Errorf("TikTok webhook: process event failed for inbox %d: %v", inbox.ID, err)
 		c.JSON(http.StatusOK, gin.H{"status": "ignored"})
 		return
 	}
+	if incomingMsg != nil {
+		if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+			applogger.L().Errorf("TikTok webhook: persist event failed for inbox %d source_id=%s: %v", inbox.ID, incomingMsg.SourceID, persistErr)
+		}
+	}
 
 	c.JSON(http.StatusOK, gin.H{"status": "processed"})
 }

internal/handler/webhook/twilio_webhook.go

@@ -27,6 +27,7 @@ import (
 type TwilioWebhookHandler struct {
 	twilioWebhook *twiliochannel.WebhookHandler
 	db            *gorm.DB
+	persister     *IncomingPersister
 }
 
 // NewTwilioWebhookHandler creates a Twilio SMS webhook handler for Gin integration.
@@ -34,6 +35,7 @@ func NewTwilioWebhookHandler(twilioWebhook *twiliochannel.WebhookHandler, db *go
 	return &TwilioWebhookHandler{
 		twilioWebhook: twilioWebhook,
 		db:            db,
+		persister:     NewIncomingPersister(db),
 	}
 }
 
@@ -54,9 +56,18 @@ func (h *TwilioWebhookHandler) HandleTwilioInboundSMS(c *gin.Context) {
 		return
 	}
 
-	// Dispatch to the channel-level webhook handler
-	// Twilio expects a TwiML XML response, not JSON
-	h.twilioWebhook.HandleInboundSMS(c.Writer, c.Request, inbox)
+	incomingMsg, err := h.twilioWebhook.ProcessInboundSMS(c.Request, inbox)
+	if err != nil {
+		applogger.L().Errorf("Twilio webhook: process inbound SMS failed for inbox %d: %v", inbox.ID, err)
+		c.Data(http.StatusOK, "application/xml", []byte("<Response></Response>"))
+		return
+	}
+	if incomingMsg != nil {
+		if _, persistErr := h.persister.PersistIncoming(c.Request.Context(), inbox, incomingMsg); persistErr != nil {
+			applogger.L().Errorf("Twilio webhook: persist inbound SMS failed for inbox %d source_id=%s: %v", inbox.ID, incomingMsg.SourceID, persistErr)
+		}
+	}
+	c.Data(http.StatusOK, "application/xml", []byte("<Response></Response>"))
 }
 
 // HandleTwilioDeliveryStatus processes a Twilio delivery status callback.

internal/handler/webhook/webhook_lookup_test.go

@@ -9,9 +9,12 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/gin-gonic/gin"
+	"github.com/gochat/gochat/internal/channel"
+	channelprovider "github.com/gochat/gochat/internal/channel/provider"
 	"github.com/gochat/gochat/internal/model"
 	channelmodel "github.com/gochat/gochat/internal/model/channel"
 	"gorm.io/driver/sqlite"
@@ -21,12 +24,17 @@ import (
 func newWebhookLookupTestDB(t *testing.T) *gorm.DB {
 	t.Helper()
 
-	db, err := gorm.Open(sqlite.Open("file::memory:?cache=shared"), &gorm.Config{})
+	dsn := "file:" + strings.NewReplacer("/", "_", " ", "_", ":", "_").Replace(t.Name()) + "?mode=memory&cache=shared"
+	db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{})
 	if err != nil {
 		t.Fatalf("open sqlite: %v", err)
 	}
 	if err := db.AutoMigrate(
 		&model.Inbox{},
+		&model.Contact{},
+		&model.ContactInbox{},
+		&model.Conversation{},
+		&model.Message{},
 		&channelmodel.ChannelTelegram{},
 		&channelmodel.ChannelLINE{},
 		&channelmodel.ChannelTwilioSMS{},
@@ -39,6 +47,61 @@ func newWebhookLookupTestDB(t *testing.T) *gorm.DB {
 	return db
 }
 
+func TestIncomingPersisterCreatesConversationMessageAndDedupes(t *testing.T) {
+	db := newWebhookLookupTestDB(t)
+	inbox := seedWebhookInbox(t, db, "telegram")
+	persister := NewIncomingPersister(db)
+
+	msg := &channel.IncomingMessage{
+		ChannelType: channel.ChannelTelegram,
+		SourceID:    "tg-msg-1",
+		SenderID:    "tg-user-1",
+		SenderName:  "Ada Lovelace",
+		SenderType:  channel.SenderContact,
+		Content:     "hello",
+		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)
+	}
+	if result.Contact == nil || result.ContactInbox == nil || result.Conversation == nil || result.Message == nil {
+		t.Fatalf("expected full persistence result: %#v", result)
+	}
+	if result.Message.Content != "hello" || result.Message.SourceID != "tg-msg-1" {
+		t.Fatalf("unexpected message: %#v", result.Message)
+	}
+
+	duplicate, err := persister.PersistIncoming(t.Context(), &inbox, msg)
+	if err != nil {
+		t.Fatalf("persist duplicate: %v", err)
+	}
+	if duplicate == nil || !duplicate.Duplicate {
+		t.Fatalf("expected duplicate result, got %#v", duplicate)
+	}
+
+	msg.SourceID = "tg-msg-2"
+	msg.Content = "second"
+	second, err := persister.PersistIncoming(t.Context(), &inbox, msg)
+	if err != nil {
+		t.Fatalf("persist second: %v", err)
+	}
+	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)
+	}
+
+	var messageCount int64
+	if err := db.Model(&model.Message{}).Where("inbox_id = ?", inbox.ID).Count(&messageCount).Error; err != nil {
+		t.Fatalf("count messages: %v", err)
+	}
+	if messageCount != 2 {
+		t.Fatalf("expected 2 persisted messages after duplicate skip, got %d", messageCount)
+	}
+}
+
 func shopifyHMAC(secret string, body []byte) string {
 	mac := hmac.New(sha256.New, []byte(secret))
 	mac.Write(body)
@@ -90,6 +153,42 @@ func TestTelegramWebhookLookupInboxByBotToken(t *testing.T) {
 	}
 }
 
+func TestTelegramWebhookPersistsIncomingMessage(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db := newWebhookLookupTestDB(t)
+	inbox := seedWebhookInbox(t, db, "telegram")
+	channelRecord := channelmodel.ChannelTelegram{
+		AccountID: 1,
+		InboxID:   inbox.ID,
+		BotToken:  "123:secret-token",
+		BotName:   "support_bot",
+	}
+	if err := db.Create(&channelRecord).Error; err != nil {
+		t.Fatalf("create telegram channel: %v", err)
+	}
+
+	body := []byte(`{"update_id":1001,"message":{"message_id":2002,"from":{"id":3003,"first_name":"Ada","last_name":"Lovelace","username":"ada"},"chat":{"id":3003,"type":"private"},"date":1710000000,"text":"hello telegram"}}`)
+	h := NewTelegramWebhookHandler(channelprovider.NewTelegramProvider(), nil, db)
+	r := gin.New()
+	r.POST("/webhooks/telegram/:bot_token", h.HandleTelegramWebhook)
+	req := httptest.NewRequest(http.MethodPost, "/webhooks/telegram/123:secret-token", bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	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 message model.Message
+	if err := db.Where("inbox_id = ? AND source_id = ?", inbox.ID, "2002").First(&message).Error; err != nil {
+		t.Fatalf("expected telegram message persisted: %v", err)
+	}
+	if message.Content != "hello telegram" || message.MessageType != string(model.MessageTypeIncoming) {
+		t.Fatalf("unexpected message: %#v", message)
+	}
+}
+
 func TestLineWebhookLookupInboxByLineChannelID(t *testing.T) {
 	db := newWebhookLookupTestDB(t)
 	inbox := seedWebhookInbox(t, db, "line")

internal/handler/webhook/whatsapp_webhook.go

@@ -5,9 +5,14 @@
 package webhook
 
 import (
-	"github.com/gin-gonic/gin"
+	"context"
 
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm"
+
+	"github.com/gochat/gochat/internal/channel"
 	"github.com/gochat/gochat/internal/channel/whatsapp"
+	"github.com/gochat/gochat/internal/model"
 )
 
 // WhatsAppWebhookHandler is a Gin adapter that wraps the WhatsApp
@@ -17,45 +22,59 @@ import (
 //
 // Chatwoot-style routes:
 //
-//   GET  /webhooks/whatsapp/:phone_number        → HandleWhatsAppVerification
-//   POST /webhooks/whatsapp/:phone_number        → HandleWhatsAppWebhook
+//	GET  /webhooks/whatsapp/:phone_number        → HandleWhatsAppVerification
+//	POST /webhooks/whatsapp/:phone_number        → HandleWhatsAppWebhook
 //
 // WhatsApp Cloud API webhook verification (GET):
 //
-//   The Meta platform sends a GET request with query parameters:
-//     - hub.mode          = "subscribe"
-//     - hub.verify_token  = the token configured in the Meta dashboard
-//     - hub.challenge     = a string the endpoint must echo back verbatim
+//	The Meta platform sends a GET request with query parameters:
+//	  - hub.mode          = "subscribe"
+//	  - hub.verify_token  = the token configured in the Meta dashboard
+//	  - hub.challenge     = a string the endpoint must echo back verbatim
 //
-//   The handler validates hub.verify_token against the stored token and
-//   returns hub.challenge as the response body with HTTP 200, or returns
-//   HTTP 403 on mismatch.
+//	The handler validates hub.verify_token against the stored token and
+//	returns hub.challenge as the response body with HTTP 200, or returns
+//	HTTP 403 on mismatch.
 //
 // WhatsApp Cloud API webhook events (POST):
 //
-//   Meta delivers event payloads as JSON with structure:
-//     {
-//       "object": "whatsapp_business_account",
-//       "entry": [ { "changes": [ ... ] } ]
-//     }
+//	Meta delivers event payloads as JSON with structure:
+//	  {
+//	    "object": "whatsapp_business_account",
+//	    "entry": [ { "changes": [ ... ] } ]
+//	  }
 //
-//   The handler parses the payload, dispatches incoming messages and status
-//   updates, and must respond with HTTP 200 OK within 10 seconds to avoid
-//   Meta retrying delivery.
+//	The handler parses the payload, dispatches incoming messages and status
+//	updates, and must respond with HTTP 200 OK within 10 seconds to avoid
+//	Meta retrying delivery.
 type WhatsAppWebhookHandler struct {
-	provider   *whatsapp.WhatsAppProvider
-	waWebhook  *whatsapp.WebhookHandler
+	provider  *whatsapp.WhatsAppProvider
+	waWebhook *whatsapp.WebhookHandler
+	persister *IncomingPersister
 }
 
 // NewWhatsAppWebhookHandler creates a Gin adapter wrapping the WhatsApp
 // 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) *WhatsAppWebhookHandler {
-	return &WhatsAppWebhookHandler{
+func NewWhatsAppWebhookHandler(provider *whatsapp.WhatsAppProvider, waWebhook *whatsapp.WebhookHandler, db *gorm.DB) *WhatsAppWebhookHandler {
+	h := &WhatsAppWebhookHandler{
 		provider:  provider,
 		waWebhook: waWebhook,
+		persister: NewIncomingPersister(db),
 	}
+	if waWebhook != nil && h.persister != nil {
+		waWebhook.SetIncomingPersister(whatsAppPersisterAdapter{persister: h.persister})
+	}
+	return h
+}
+
+type whatsAppPersisterAdapter struct {
+	persister *IncomingPersister
+}
+
+func (a whatsAppPersisterAdapter) PersistIncoming(ctx context.Context, inbox *model.Inbox, msg *channel.IncomingMessage) (interface{}, error) {
+	return a.persister.PersistIncoming(ctx, inbox, msg)
 }
 
 // HandleWhatsAppVerification handles GET requests for WhatsApp Cloud API
@@ -80,4 +99,4 @@ func (h *WhatsAppWebhookHandler) HandleWhatsAppVerification(c *gin.Context) {
 // Expected Gin route: POST /webhooks/whatsapp/:phone_number
 func (h *WhatsAppWebhookHandler) HandleWhatsAppWebhook(c *gin.Context) {
 	h.waWebhook.HandleWebhookEvent(c)
-}
+}