gochat/tests/e2e/channel_incoming_e2e_test.go

package e2e

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/suite"
	"gorm.io/gorm"

	"github.com/gochat/gochat/internal/channel"
	"github.com/gochat/gochat/internal/model"

	// Blank import to trigger init() — registers FB, IG, Email, Telegram providers in global channel registry
	_ "github.com/gochat/gochat/internal/channel/provider"
)

// --- GORM-backed adapters for channel repository interfaces ---
// The IncomingMessageProcessor uses abstract interfaces (AccountRepository, ContactRepository,
// ConversationRepository, MessageRepository) that don't match the existing GORM repos directly.
// These thin adapters bridge the gap by implementing the channel interfaces over raw GORM DB.

// channelAccountAdapter implements channel.AccountRepository over GORM.
type channelAccountAdapter struct {
	db *gorm.DB
}

func (a *channelAccountAdapter) FindByID(ctx context.Context, id uint) (*model.Account, error) {
	var account model.Account
	if err := a.db.WithContext(ctx).First(&account, id).Error; err != nil {
		return nil, err
	}
	return &account, nil
}

// channelContactAdapter implements channel.ContactRepository over GORM.
// FindBySourceIDAndInboxID looks up a Contact via the contact_inboxes join table,
// matching source_id + inbox_id — the Chatwoot pattern for cross-channel contact identity.
type channelContactAdapter struct {
	db *gorm.DB
}

func (a *channelContactAdapter) FindBySourceIDAndInboxID(ctx context.Context, sourceID string, inboxID uint) (*model.Contact, error) {
	var ci model.ContactInbox
	if err := a.db.WithContext(ctx).
		Where("source_id = ? AND inbox_id = ?", sourceID, inboxID).
		First(&ci).Error; err != nil {
		return nil, err
	}
	var contact model.Contact
	if err := a.db.WithContext(ctx).First(&contact, ci.ContactID).Error; err != nil {
		return nil, err
	}
	return &contact, nil
}

func (a *channelContactAdapter) Create(ctx context.Context, contact *model.Contact) (*model.Contact, error) {
	if err := a.db.WithContext(ctx).Create(contact).Error; err != nil {
		return nil, err
	}
	return contact, nil
}

func (a *channelContactAdapter) Update(ctx context.Context, contact *model.Contact) (*model.Contact, error) {
	if err := a.db.WithContext(ctx).Save(contact).Error; err != nil {
		return nil, err
	}
	return contact, nil
}

// CreateContactInbox creates a ContactInbox join record binding a Contact to an Inbox
// with the given source_id. Reference: Chatwoot Contacts::InboxCreateService creates
// both Contact + ContactInbox in a single operation. The pipeline's ContactResolutionStage
// currently only creates the Contact, so we handle the ContactInbox creation separately
// in our processIncoming helper to make E2E tests realistic.
func (a *channelContactAdapter) CreateContactInbox(ctx context.Context, contactInbox *model.ContactInbox) (*model.ContactInbox, error) {
	if err := a.db.WithContext(ctx).Create(contactInbox).Error; err != nil {
		return nil, err
	}
	return contactInbox, nil
}

// channelConversationAdapter implements channel.ConversationRepository over GORM.
type channelConversationAdapter struct {
	db *gorm.DB
}

func (a *channelConversationAdapter) FindOpenByContactIDAndInboxID(ctx context.Context, contactID uint, inboxID uint) (*model.Conversation, error) {
	var conv model.Conversation
	if err := a.db.WithContext(ctx).
		Where("contact_id = ? AND inbox_id = ? AND status = ?", contactID, inboxID, string(model.ConversationStatusOpen)).
		Order("id DESC"). // pick the most recent open conversation
		First(&conv).Error; err != nil {
		return nil, err
	}
	return &conv, nil
}

func (a *channelConversationAdapter) Create(ctx context.Context, conversation *model.Conversation) (*model.Conversation, error) {
	if err := a.db.WithContext(ctx).Create(conversation).Error; err != nil {
		return nil, err
	}
	return conversation, nil
}

func (a *channelConversationAdapter) Update(ctx context.Context, conversation *model.Conversation) (*model.Conversation, error) {
	if err := a.db.WithContext(ctx).Save(conversation).Error; err != nil {
		return nil, err
	}
	return conversation, nil
}

// channelMessageAdapter implements channel.MessageRepository over GORM.
type channelMessageAdapter struct {
	db *gorm.DB
}

func (a *channelMessageAdapter) FindBySourceIDAndInboxID(ctx context.Context, sourceID string, inboxID uint) (*model.Message, error) {
	var msg model.Message
	if err := a.db.WithContext(ctx).
		Where("source_id = ? AND inbox_id = ?", sourceID, inboxID).
		First(&msg).Error; err != nil {
		return nil, err
	}
	return &msg, nil
}

func (a *channelMessageAdapter) Create(ctx context.Context, message *model.Message) (*model.Message, error) {
	if err := a.db.WithContext(ctx).Create(message).Error; err != nil {
		return nil, err
	}
	return message, nil
}

func (a *channelMessageAdapter) Update(ctx context.Context, message *model.Message) (*model.Message, error) {
	if err := a.db.WithContext(ctx).Save(message).Error; err != nil {
		return nil, err
	}
	return message, nil
}

// --- Channel Incoming E2E Test Suite ---
// ChannelIncomingE2ETestSuite tests the full incoming message pipeline end-to-end:
//   1. Set up Account + Inbox (with EnableAutoAssignment=true) for a channel type
//   2. Construct an IncomingMessage simulating an external webhook payload
//   3. Run it through the IncomingMessageProcessor pipeline
//   4. Verify Contact, Conversation, Message, and ContactInbox records in the DB
//
// Reference: Chatwoot IncomingMessageService pipeline pattern:
//   parse webhook → create/update Contact → find/create Conversation → create Message → fire Wisper events
type ChannelIncomingE2ETestSuite struct {
	E2ETestSuite
	account          *model.Account
	processor        *channel.IncomingMessageProcessor
	contactAdapter   *channelContactAdapter
	ctx              context.Context
}

// SetupSuite initializes the processor with GORM-backed adapters once for the suite.
func (s *ChannelIncomingE2ETestSuite) SetupSuite() {
	// Run the parent SetupSuite (creates DB, server, etc.)
	s.E2ETestSuite.SetupSuite()

	// Build the IncomingMessageProcessor with GORM-backed adapters
	accountRepo := &channelAccountAdapter{db: s.DB()}
	s.contactAdapter = &channelContactAdapter{db: s.DB()}
	conversationRepo := &channelConversationAdapter{db: s.DB()}
	messageRepo := &channelMessageAdapter{db: s.DB()}

	s.processor = channel.NewIncomingMessageProcessor(
		accountRepo, s.contactAdapter, conversationRepo, messageRepo,
	)
	s.ctx = context.Background()
}

// SetupTest clears the DB and creates a fresh account for each test.
func (s *ChannelIncomingE2ETestSuite) SetupTest() {
	s.ClearDatabase()
	s.account = s.CreateTestAccount("Channel Test Org")
}

// --- Helper: create an Inbox with EnableAutoAssignment=true ---
// The ValidateStage rejects messages on inboxes where EnableAutoAssignment is false.
func (s *ChannelIncomingE2ETestSuite) createChannelInbox(name, channelType string, accountID uint) *model.Inbox {
	inbox := &model.Inbox{
		AccountID:           accountID,
		Name:                name,
		ChannelType:         channelType,
		ChannelID:           1,
		EnableAutoAssignment: true,
		Enabled:             true,
	}
	err := s.DB().Create(inbox).Error
	s.Require().NoError(err)
	return inbox
}

// --- Helper: run the pipeline and return the result ---
// After the pipeline completes, if a new Contact was created, we also create a ContactInbox
// join record. This mirrors Chatwoot's Contacts::InboxCreateService which creates both
// Contact + ContactInbox in one operation. The pipeline's ContactResolutionStage currently
// only creates the Contact, so we handle the ContactInbox here.
func (s *ChannelIncomingE2ETestSuite) processIncoming(inbox *model.Inbox, msg *channel.IncomingMessage) (*channel.PipelineContext, error) {
	pipelineCtx := &channel.PipelineContext{
		Inbox:           inbox,
		IncomingMessage: msg,
	}
	result, err := s.processor.Process(s.ctx, pipelineCtx)
	if err != nil {
		return result, err
	}

	// Ensure ContactInbox exists for the resolved Contact.
	// Without this, subsequent FindBySourceIDAndInboxID calls would fail
	// because no join record exists.
	if result.Contact != nil {
		var existingCI model.ContactInbox
		ciErr := s.DB().WithContext(s.ctx).
			Where("contact_id = ? AND inbox_id = ? AND source_id = ?", result.Contact.ID, inbox.ID, msg.SenderID).
			First(&existingCI).Error
		if ciErr == gorm.ErrRecordNotFound {
			_, ciCreateErr := s.contactAdapter.CreateContactInbox(s.ctx, &model.ContactInbox{
				ContactID: result.Contact.ID,
				InboxID:   inbox.ID,
				SourceID:  msg.SenderID,
			})
			if ciCreateErr != nil {
				return result, fmt.Errorf("failed to create ContactInbox: %w", ciCreateErr)
			}
		}
	}
	return result, nil
}

// --- Helper: build a minimal IncomingMessage for a channel type ---
func buildIncomingMessage(channelType channel.ChannelType, inboxID, accountID uint, senderID, senderName, content, sourceID string) *channel.IncomingMessage {
	return &channel.IncomingMessage{
		ChannelType:  channelType,
		SourceID:     sourceID,
		SenderID:     senderID,
		SenderName:   senderName,
		SenderType:   channel.SenderContact,
		Content:      content,
		ContentType:  channel.ContentText,
		InboxID:      inboxID,
		AccountID:    accountID,
		ReceivedAt:   time.Now(),
	}
}

// ====================== FACEBOOK TESTS ======================

// TestFacebookIncomingCreatesConversation verifies that a Facebook Messenger message
// flowing through the pipeline creates Contact, Conversation, and Message records.
func (s *ChannelIncomingE2ETestSuite) TestFacebookIncomingCreatesConversation() {
	// Verify FB provider is registered
	assert.True(s.T(), channel.IsRegistered(channel.ChannelFacebook),
		"Facebook provider should be registered in the global channel registry")

	// Create a FB inbox
	inbox := s.createChannelInbox("FB Support Inbox", string(channel.ChannelFacebook), s.account.ID)

	// Simulate a FB Messenger incoming message
	msg := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_psid_12345", // FB Page-Specific User ID
		"John FB User",
		"Hello from Facebook Messenger!",
		"mid_fb_msg_001",
	)

	// Run the pipeline
	result, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err, "Facebook incoming pipeline should complete without error")
	s.Require().NotNil(result)

	// Verify Contact was created
	s.Require().NotNil(result.Contact, "Pipeline should have created a Contact")
	assert.Equal(s.T(), s.account.ID, result.Contact.AccountID)
	assert.Equal(s.T(), "John FB User", result.Contact.Name)
	assert.Equal(s.T(), "fb_sender_psid_12345", result.Contact.Identifier)

	// Verify Conversation was created
	s.Require().NotNil(result.Conversation, "Pipeline should have created a Conversation")
	assert.Equal(s.T(), s.account.ID, result.Conversation.AccountID)
	assert.Equal(s.T(), inbox.ID, result.Conversation.InboxID)
	assert.Equal(s.T(), result.Contact.ID, result.Conversation.ContactID)
	assert.Equal(s.T(), string(model.ConversationStatusOpen), result.Conversation.Status)
	assert.Equal(s.T(), string(channel.ChannelFacebook), result.Conversation.ChannelType)

	// Verify Message was created
	s.Require().NotNil(result.Message, "Pipeline should have created a Message")
	assert.Equal(s.T(), result.Conversation.ID, result.Message.ConversationID)
	assert.Equal(s.T(), s.account.ID, result.Message.AccountID)
	assert.Equal(s.T(), inbox.ID, result.Message.InboxID)
	assert.Equal(s.T(), "Hello from Facebook Messenger!", result.Message.Content)
	assert.Equal(s.T(), string(model.MessageContentTypeText), result.Message.ContentType)
	assert.Equal(s.T(), string(model.MessageTypeIncoming), result.Message.MessageType)
	assert.Equal(s.T(), "mid_fb_msg_001", result.Message.SourceID)
	assert.False(s.T(), result.Message.Private)

	// Verify records persisted in DB
	var dbContact model.Contact
	s.Require().NoError(s.DB().First(&dbContact, result.Contact.ID).Error)
	assert.Equal(s.T(), "John FB User", dbContact.Name)

	var dbConv model.Conversation
	s.Require().NoError(s.DB().First(&dbConv, result.Conversation.ID).Error)
	assert.Equal(s.T(), string(model.ConversationStatusOpen), dbConv.Status)

	var dbMsg model.Message
	s.Require().NoError(s.DB().First(&dbMsg, result.Message.ID).Error)
	assert.Equal(s.T(), "Hello from Facebook Messenger!", dbMsg.Content)
}

// TestFacebookIncomingReusesConversation verifies that a second message from the same
// sender reuses the existing open conversation rather than creating a new one.
func (s *ChannelIncomingE2ETestSuite) TestFacebookIncomingReusesConversation() {
	inbox := s.createChannelInbox("FB Reuse Inbox", string(channel.ChannelFacebook), s.account.ID)

	// First message — creates conversation
	msg1 := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_reuse_001",
		"Reuse FB User",
		"First message",
		"mid_fb_reuse_001",
	)
	result1, err := s.processIncoming(inbox, msg1)
	s.Require().NoError(err)
	s.Require().NotNil(result1.Conversation)
	conv1ID := result1.Conversation.ID

	// Second message from same sender — should reuse the same conversation
	msg2 := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_reuse_001", // same sender
		"Reuse FB User",
		"Second message",
		"mid_fb_reuse_002",
	)
	result2, err := s.processIncoming(inbox, msg2)
	s.Require().NoError(err)
	s.Require().NotNil(result2.Conversation)

	// Should reuse the same conversation
	assert.Equal(s.T(), conv1ID, result2.Conversation.ID,
		"Second message should reuse the existing open conversation")

	// Should also reuse the same contact
	assert.Equal(s.T(), result1.Contact.ID, result2.Contact.ID,
		"Second message should reuse the existing contact")

	// Verify two messages exist in DB for this conversation
	var msgs []model.Message
	s.DB().Where("conversation_id = ?", conv1ID).Order("id ASC").Find(&msgs)
	assert.Equal(s.T(), 2, len(msgs), "Should have 2 messages in the conversation")
	assert.Equal(s.T(), "First message", msgs[0].Content)
	assert.Equal(s.T(), "Second message", msgs[1].Content)
}

// TestFacebookIncomingDuplicateMessageSkipped verifies that a message with the same
// source_id is deduplicated (the pipeline's MessagePersistenceStage skips duplicates).
func (s *ChannelIncomingE2ETestSuite) TestFacebookIncomingDuplicateMessageSkipped() {
	inbox := s.createChannelInbox("FB Dedup Inbox", string(channel.ChannelFacebook), s.account.ID)

	msg := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_dedup_001",
		"Dedup FB User",
		"Original message",
		"mid_fb_dedup_001", // unique source_id
	)

	result1, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err)
	msg1ID := result1.Message.ID

	// Send the same source_id again — should be deduplicated
	result2, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err)
	assert.Equal(s.T(), msg1ID, result2.Message.ID,
		"Duplicate message (same source_id) should return the existing message, not create a new one")

	// Verify only 1 message in DB
	var count int64
	s.DB().Model(&model.Message{}).Where("source_id = ? AND inbox_id = ?", "mid_fb_dedup_001", inbox.ID).Count(&count)
	assert.Equal(s.T(), int64(1), count, "Only one message should exist for the same source_id")
}

// ====================== INSTAGRAM TESTS ======================

// TestInstagramIncomingCreatesConversation verifies that an Instagram DM flowing
// through the pipeline creates Contact, Conversation, and Message records.
func (s *ChannelIncomingE2ETestSuite) TestInstagramIncomingCreatesConversation() {
	assert.True(s.T(), channel.IsRegistered(channel.ChannelInstagram),
		"Instagram provider should be registered in the global channel registry")

	inbox := s.createChannelInbox("IG Support Inbox", string(channel.ChannelInstagram), s.account.ID)

	msg := buildIncomingMessage(
		channel.ChannelInstagram,
		inbox.ID,
		s.account.ID,
		"ig_sender_12345",
		"Jane IG User",
		"Hello from Instagram DM!",
		"mid_ig_msg_001",
	)

	result, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err, "Instagram incoming pipeline should complete without error")
	s.Require().NotNil(result)

	// Verify Contact
	s.Require().NotNil(result.Contact)
	assert.Equal(s.T(), s.account.ID, result.Contact.AccountID)
	assert.Equal(s.T(), "Jane IG User", result.Contact.Name)
	assert.Equal(s.T(), "ig_sender_12345", result.Contact.Identifier)

	// Verify Conversation
	s.Require().NotNil(result.Conversation)
	assert.Equal(s.T(), string(channel.ChannelInstagram), result.Conversation.ChannelType)
	assert.Equal(s.T(), string(model.ConversationStatusOpen), result.Conversation.Status)

	// Verify Message
	s.Require().NotNil(result.Message)
	assert.Equal(s.T(), "Hello from Instagram DM!", result.Message.Content)
	assert.Equal(s.T(), "mid_ig_msg_001", result.Message.SourceID)
}

// TestInstagramIncomingReusesConversation verifies conversation reuse for Instagram.
func (s *ChannelIncomingE2ETestSuite) TestInstagramIncomingReusesConversation() {
	inbox := s.createChannelInbox("IG Reuse Inbox", string(channel.ChannelInstagram), s.account.ID)

	msg1 := buildIncomingMessage(
		channel.ChannelInstagram,
		inbox.ID,
		s.account.ID,
		"ig_sender_reuse_001",
		"Reuse IG User",
		"First IG message",
		"mid_ig_reuse_001",
	)
	result1, err := s.processIncoming(inbox, msg1)
	s.Require().NoError(err)
	conv1ID := result1.Conversation.ID

	msg2 := buildIncomingMessage(
		channel.ChannelInstagram,
		inbox.ID,
		s.account.ID,
		"ig_sender_reuse_001",
		"Reuse IG User",
		"Second IG message",
		"mid_ig_reuse_002",
	)
	result2, err := s.processIncoming(inbox, msg2)
	s.Require().NoError(err)

	assert.Equal(s.T(), conv1ID, result2.Conversation.ID,
		"Second IG message should reuse the existing open conversation")
	assert.Equal(s.T(), result1.Contact.ID, result2.Contact.ID,
		"Same IG sender should reuse the existing contact")
}

// ====================== EMAIL TESTS ======================

// TestEmailIncomingCreatesConversation verifies that an email message flowing
// through the pipeline creates Contact, Conversation, and Message records.
func (s *ChannelIncomingE2ETestSuite) TestEmailIncomingCreatesConversation() {
	assert.True(s.T(), channel.IsRegistered(channel.ChannelEmail),
		"Email provider should be registered in the global channel registry")

	inbox := s.createChannelInbox("Email Support Inbox", string(channel.ChannelEmail), s.account.ID)

	msg := buildIncomingMessage(
		channel.ChannelEmail,
		inbox.ID,
		s.account.ID,
		"customer@example.com", // email sender — source_id is the email address
		"Email Customer",
		"I have a question about my order",
		"msg_email_001",
	)

	result, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err, "Email incoming pipeline should complete without error")
	s.Require().NotNil(result)

	// Verify Contact
	s.Require().NotNil(result.Contact)
	assert.Equal(s.T(), s.account.ID, result.Contact.AccountID)
	assert.Equal(s.T(), "Email Customer", result.Contact.Name)
	assert.Equal(s.T(), "customer@example.com", result.Contact.Identifier)

	// Verify Conversation
	s.Require().NotNil(result.Conversation)
	assert.Equal(s.T(), string(channel.ChannelEmail), result.Conversation.ChannelType)
	assert.Equal(s.T(), string(model.ConversationStatusOpen), result.Conversation.Status)

	// Verify Message
	s.Require().NotNil(result.Message)
	assert.Equal(s.T(), "I have a question about my order", result.Message.Content)
	assert.Equal(s.T(), "msg_email_001", result.Message.SourceID)
}

// TestEmailIncomingReusesConversation verifies conversation reuse for Email.
func (s *ChannelIncomingE2ETestSuite) TestEmailIncomingReusesConversation() {
	inbox := s.createChannelInbox("Email Reuse Inbox", string(channel.ChannelEmail), s.account.ID)

	msg1 := buildIncomingMessage(
		channel.ChannelEmail,
		inbox.ID,
		s.account.ID,
		"reuse@example.com",
		"Reuse Email User",
		"First email",
		"mid_email_reuse_001",
	)
	result1, err := s.processIncoming(inbox, msg1)
	s.Require().NoError(err)
	conv1ID := result1.Conversation.ID

	msg2 := buildIncomingMessage(
		channel.ChannelEmail,
		inbox.ID,
		s.account.ID,
		"reuse@example.com",
		"Reuse Email User",
		"Second email",
		"mid_email_reuse_002",
	)
	result2, err := s.processIncoming(inbox, msg2)
	s.Require().NoError(err)

	assert.Equal(s.T(), conv1ID, result2.Conversation.ID,
		"Second email from same sender should reuse the existing open conversation")
}

// TestEmailIncomingDuplicateMessageSkipped verifies deduplication for email messages.
func (s *ChannelIncomingE2ETestSuite) TestEmailIncomingDuplicateMessageSkipped() {
	inbox := s.createChannelInbox("Email Dedup Inbox", string(channel.ChannelEmail), s.account.ID)

	msg := buildIncomingMessage(
		channel.ChannelEmail,
		inbox.ID,
		s.account.ID,
		"dedup@example.com",
		"Dedup Email User",
		"Original email message",
		"mid_email_dedup_001",
	)

	result1, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err)
	msg1ID := result1.Message.ID

	result2, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err)
	assert.Equal(s.T(), msg1ID, result2.Message.ID,
		"Duplicate email (same source_id) should return the existing message")
}

// ====================== WHATSAPP TESTS (SKIPPED) ======================

// TestWhatsAppIncomingSkipped verifies that WhatsApp is correctly skipped
// because no provider is registered.
func (s *ChannelIncomingE2ETestSuite) TestWhatsAppIncomingSkipped() {
	// WhatsApp provider is NOT registered — only the channel type constant exists
	assert.False(s.T(), channel.IsRegistered(channel.ChannelWhatsApp),
		"WhatsApp provider should NOT be registered (no implementation)")

	// Attempting to process a WhatsApp message through the pipeline should fail
	// at the ValidateStage because the provider lookup fails
	inbox := s.createChannelInbox("WA Inbox", string(channel.ChannelWhatsApp), s.account.ID)

	msg := buildIncomingMessage(
		channel.ChannelWhatsApp,
		inbox.ID,
		s.account.ID,
		"wa_sender_12345",
		"WA User",
		"Hello from WhatsApp!",
		"mid_wa_msg_001",
	)

	_, err := s.processIncoming(inbox, msg)
	s.Require().Error(err, "WhatsApp pipeline should fail because no provider is registered")
	assert.Contains(s.T(), err.Error(), "no provider",
		"Error should mention missing provider for WhatsApp")
}

// ====================== CROSS-CHANNEL TESTS ======================

// TestCrossChannelSameContactDifferentInboxes verifies that the same external sender
// (e.g., same person on FB and IG) creates different contacts in different inboxes,
// each with their own conversation.
// Reference: Chatwoot uses source_id + inbox_id as the contact identity composite key,
// so the same person on different channels gets separate Contact records.
func (s *ChannelIncomingE2ETestSuite) TestCrossChannelSameContactDifferentInboxes() {
	fbInbox := s.createChannelInbox("Cross FB Inbox", string(channel.ChannelFacebook), s.account.ID)
	igInbox := s.createChannelInbox("Cross IG Inbox", string(channel.ChannelInstagram), s.account.ID)

	// Same person "Jane Doe" sends from FB
	fbMsg := buildIncomingMessage(
		channel.ChannelFacebook,
		fbInbox.ID,
		s.account.ID,
		"cross_sender_jane_fb",
		"Jane Doe",
		"Message from FB",
		"mid_cross_fb_001",
	)
	fbResult, err := s.processIncoming(fbInbox, fbMsg)
	s.Require().NoError(err)

	// Same person "Jane Doe" sends from IG
	igMsg := buildIncomingMessage(
		channel.ChannelInstagram,
		igInbox.ID,
		s.account.ID,
		"cross_sender_jane_ig",
		"Jane Doe",
		"Message from IG",
		"mid_cross_ig_001",
	)
	igResult, err := s.processIncoming(igInbox, igMsg)
	s.Require().NoError(err)

	// Different source IDs → different contacts (Chatwoot identity pattern)
	assert.NotEqual(s.T(), fbResult.Contact.ID, igResult.Contact.ID,
		"Same person on different channels should create separate Contact records")

	// Different inboxes → different conversations
	assert.NotEqual(s.T(), fbResult.Conversation.ID, igResult.Conversation.ID,
		"Messages in different inboxes should create separate Conversations")

	// Verify channel types
	assert.Equal(s.T(), string(channel.ChannelFacebook), fbResult.Conversation.ChannelType)
	assert.Equal(s.T(), string(channel.ChannelInstagram), igResult.Conversation.ChannelType)
}

// TestPipelineValidationRejectsEmptyContent verifies that the ValidateStage
// rejects messages with no content and no attachments.
func (s *ChannelIncomingE2ETestSuite) TestPipelineValidationRejectsEmptyContent() {
	inbox := s.createChannelInbox("Validation Inbox", string(channel.ChannelFacebook), s.account.ID)

	// Message with empty content and no attachments
	msg := &channel.IncomingMessage{
		ChannelType:  channel.ChannelFacebook,
		SourceID:     "mid_validation_empty",
		SenderID:     "fb_sender_empty",
		SenderName:   "Empty Sender",
		SenderType:   channel.SenderContact,
		Content:      "", // empty content
		ContentType:  channel.ContentText,
		Attachments:  nil, // no attachments
		InboxID:      inbox.ID,
		AccountID:    s.account.ID,
		ReceivedAt:   time.Now(),
	}

	_, err := s.processIncoming(inbox, msg)
	s.Require().Error(err, "Pipeline should reject messages with no content or attachments")
	assert.Contains(s.T(), err.Error(), "no content",
		"Error should mention missing content/attachments")
}

// TestPipelineValidationRejectsMissingSenderID verifies that ValidateStage
// rejects messages without a sender_id.
func (s *ChannelIncomingE2ETestSuite) TestPipelineValidationRejectsMissingSenderID() {
	inbox := s.createChannelInbox("Validation Inbox 2", string(channel.ChannelEmail), s.account.ID)

	msg := &channel.IncomingMessage{
		ChannelType:  channel.ChannelEmail,
		SourceID:     "mid_validation_no_sender",
		SenderID:     "", // missing sender_id
		SenderName:   "No Sender",
		SenderType:   channel.SenderContact,
		Content:      "Some content",
		ContentType:  channel.ContentText,
		InboxID:      inbox.ID,
		AccountID:    s.account.ID,
		ReceivedAt:   time.Now(),
	}

	_, err := s.processIncoming(inbox, msg)
	s.Require().Error(err, "Pipeline should reject messages without sender_id")
	assert.Contains(s.T(), err.Error(), "sender_id",
		"Error should mention missing sender_id")
}

// TestPipelineValidationRejectsDisabledInbox verifies that ValidateStage
// rejects messages on an inbox where EnableAutoAssignment is false.
func (s *ChannelIncomingE2ETestSuite) TestPipelineValidationRejectsDisabledInbox() {
	// Create inbox with EnableAutoAssignment = false
	inbox := &model.Inbox{
		AccountID:            s.account.ID,
		Name:                 "Disabled Inbox",
		ChannelType:          string(channel.ChannelFacebook),
		ChannelID:            1,
		EnableAutoAssignment: false,
		Enabled:              true,
	}
	err := s.DB().Create(inbox).Error
	s.Require().NoError(err)

	msg := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_disabled",
		"Disabled Sender",
		"Message to disabled inbox",
		"mid_disabled_001",
	)

	_, err = s.processIncoming(inbox, msg)
	s.Require().Error(err, "Pipeline should reject messages on disabled inbox")
	assert.Contains(s.T(), err.Error(), "disabled",
		"Error should mention inbox is disabled")
}

// TestContactNameFallback verifies that when SenderName is empty, the pipeline
// generates a fallback name like "Contact_<sender_id>".
func (s *ChannelIncomingE2ETestSuite) TestContactNameFallback() {
	inbox := s.createChannelInbox("Fallback Name Inbox", string(channel.ChannelEmail), s.account.ID)

	msg := buildIncomingMessage(
		channel.ChannelEmail,
		inbox.ID,
		s.account.ID,
		"no_name@example.com",
		"", // empty sender name — should trigger fallback
		"Message with no sender name",
		"mid_no_name_001",
	)

	result, err := s.processIncoming(inbox, msg)
	s.Require().NoError(err)
	s.Require().NotNil(result.Contact)

	// Fallback name should be "Contact_<sender_id>"
	assert.Equal(s.T(), fmt.Sprintf("Contact_%s", "no_name@example.com"), result.Contact.Name,
		"Empty sender name should fall back to Contact_<source_id>")
}

// TestContactNameUpdateOnRepeatMessage verifies that the ContactResolutionStage
// updates the contact's display name when a subsequent message provides a different name.
func (s *ChannelIncomingE2ETestSuite) TestContactNameUpdateOnRepeatMessage() {
	inbox := s.createChannelInbox("Name Update Inbox", string(channel.ChannelFacebook), s.account.ID)

	// First message with name "Original Name"
	msg1 := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_name_update",
		"Original Name",
		"First message",
		"mid_name_update_001",
	)
	result1, err := s.processIncoming(inbox, msg1)
	s.Require().NoError(err)
	assert.Equal(s.T(), "Original Name", result1.Contact.Name)

	// Second message from same sender with updated name "Updated Name"
	msg2 := buildIncomingMessage(
		channel.ChannelFacebook,
		inbox.ID,
		s.account.ID,
		"fb_sender_name_update",
		"Updated Name", // name change
		"Second message",
		"mid_name_update_002",
	)
	result2, err := s.processIncoming(inbox, msg2)
	s.Require().NoError(err)
	assert.Equal(s.T(), "Updated Name", result2.Contact.Name,
		"Contact name should be updated when sender provides a new display name")
}

// TestChannelIncomingE2ESuite runs the Channel Incoming E2E test suite.
func TestChannelIncomingE2ESuite(t *testing.T) {
	suite.Run(t, new(ChannelIncomingE2ETestSuite))
}