gochat/internal/channel/incoming.go

package channel

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/gochat/gochat/internal/model"
	applogger "github.com/gochat/gochat/pkg/logger"
)

// --- Pipeline Stage Interface & Context ---

// PipelineStage is the interface for a single step in the incoming message
// processing pipeline. Each stage receives the accumulated PipelineContext
// and returns a modified context or an error to abort the pipeline.
// Reference: Chatwoot IncomingMessageService pipeline pattern:
//   parse webhook → create/update Contact → find/create Conversation →
//   create Message → fire Wisper events
type PipelineStage interface {
	// Name returns a human-readable stage name for logging.
	Name() string

	// Process executes the pipeline stage, mutating pipelineCtx as needed.
	// Returns the (possibly modified) PipelineContext, or an error to halt.
	Process(ctx context.Context, pipelineCtx *PipelineContext) (*PipelineContext, error)
}

// PipelineContext carries state through the incoming message pipeline.
// Each stage reads from and writes to this context, progressively building
// up the Contact, Conversation, and Message as the pipeline advances.
type PipelineContext struct {
	// Input fields (set before pipeline starts)
	IncomingMessage *IncomingMessage `json:"incoming_message"`
	Inbox           *model.Inbox     `json:"inbox"`

	// Accumulated fields (populated by pipeline stages)
	Account      *model.Account     `json:"account,omitempty"`
	Contact      *model.Contact     `json:"contact,omitempty"`
	Conversation *model.Conversation `json:"conversation,omitempty"`
	Message      *model.Message     `json:"message,omitempty"`
}

// --- Repository Interfaces ---
// These are abstracted interfaces so that the processor doesn't depend on
// concrete GORM repositories. Wire up implementations via DI.

// AccountRepository provides account lookups for message processing.
type AccountRepository interface {
	FindByID(ctx context.Context, id uint) (*model.Account, error)
}

// ContactRepository provides contact CRUD for the incoming pipeline.
// Reference: Chatwoot uses source_id + inbox_id pattern to uniquely identify
// contacts across channels (Contact.find_by_source_id + inbox_id).
type ContactRepository interface {
	FindBySourceIDAndInboxID(ctx context.Context, sourceID string, inboxID uint) (*model.Contact, error)
	Create(ctx context.Context, contact *model.Contact) (*model.Contact, error)
	Update(ctx context.Context, contact *model.Contact) (*model.Contact, error)
}

// ConversationRepository provides conversation lookup/creation.
// Reference: Chatwoot Conversation.find_or_create_by(contact_id + inbox_id + status=open).
type ConversationRepository interface {
	FindOpenByContactIDAndInboxID(ctx context.Context, contactID uint, inboxID uint) (*model.Conversation, error)
	Create(ctx context.Context, conversation *model.Conversation) (*model.Conversation, error)
	Update(ctx context.Context, conversation *model.Conversation) (*model.Conversation, error)
}

// MessageRepository provides message persistence.
// Reference: Chatwoot Message.create! with conversation_id + account_id + inbox_id.
type MessageRepository interface {
	Create(ctx context.Context, message *model.Message) (*model.Message, error)
	Update(ctx context.Context, message *model.Message) (*model.Message, error)
	FindBySourceIDAndInboxID(ctx context.Context, sourceID string, inboxID uint) (*model.Message, error)
}

// --- IncomingMessageProcessor ---

// IncomingMessageProcessor orchestrates the incoming message pipeline.
// It holds an ordered list of PipelineStage implementations and runs them
// sequentially, aborting on the first error.
//
// Reference: Chatwoot IncomingMessageService — per-channel services that
// parse webhook payload → create/update Contact → find/create Conversation →
// create Message → fire events. GoChat unifies this into a single pipeline
// with provider-specific logic only in the webhook parsing layer.
type IncomingMessageProcessor struct {
	stages []PipelineStage
}

// NewIncomingMessageProcessor creates a processor with the standard pipeline stages.
func NewIncomingMessageProcessor(
	accountRepo AccountRepository,
	contactRepo ContactRepository,
	conversationRepo ConversationRepository,
	messageRepo MessageRepository,
) *IncomingMessageProcessor {
	stages := []PipelineStage{
		&ValidateStage{},
		&ContactResolutionStage{contactRepo: contactRepo},
		&ConversationResolutionStage{conversationRepo: conversationRepo},
		&MessagePersistenceStage{messageRepo: messageRepo},
		&EventDispatchStage{accountRepo: accountRepo},
	}
	return &IncomingMessageProcessor{stages: stages}
}

// Process runs all pipeline stages sequentially.
// If any stage returns an error, the pipeline aborts and the error is propagated.
func (p *IncomingMessageProcessor) Process(ctx context.Context, pipelineCtx *PipelineContext) (*PipelineContext, error) {
	for _, stage := range p.stages {
		applogger.L().Infof("Incoming pipeline: executing stage %s", stage.Name())
		result, err := stage.Process(ctx, pipelineCtx)
		if err != nil {
			applogger.L().Errorf("Incoming pipeline: stage %s failed: %v", stage.Name(), err)
			return nil, fmt.Errorf("stage %s: %w", stage.Name(), err)
		}
		pipelineCtx = result
	}
	applogger.L().Infof("Incoming pipeline: all stages completed successfully")
	return pipelineCtx, nil
}

// --- Pipeline Stages ---

// ValidateStage validates the incoming message and inbox for required fields.
// Ensures that the message has content, source_id, and sender info, and that
// the inbox is enabled and has a valid channel type.
type ValidateStage struct{}

func (s *ValidateStage) Name() string { return "validate" }

func (s *ValidateStage) Process(ctx context.Context, pc *PipelineContext) (*PipelineContext, error) {
	msg := pc.IncomingMessage
	if msg == nil {
		return nil, fmt.Errorf("incoming message is nil")
	}
	if msg.SourceID == "" {
		return nil, fmt.Errorf("incoming message missing source_id")
	}
	if msg.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")
	}

	inbox := pc.Inbox
	if inbox == nil {
		return nil, fmt.Errorf("inbox is nil")
	}
	if !inbox.EnableAutoAssignment {
		return nil, fmt.Errorf("inbox %d is disabled", inbox.ID)
	}
	if inbox.ChannelType == "" {
		return nil, fmt.Errorf("inbox %d has no channel_type", inbox.ID)
	}

	// Verify the channel provider is registered
	provider, err := Get(ChannelType(inbox.ChannelType))
	if err != nil {
		return nil, fmt.Errorf("no provider registered for channel type %s: %w", inbox.ChannelType, err)
	}

	applogger.L().Infof("ValidateStage: message valid, provider=%s", provider.Name())
	return pc, nil
}

// ContactResolutionStage finds or creates the Contact for the incoming message sender.
// Reference: Chatwoot IncomingMessageService pattern — find contact by source_id +
// inbox_id (polymorphic channel association), or create a new one from the sender profile.
type ContactResolutionStage struct {
	contactRepo ContactRepository
}

func (s *ContactResolutionStage) Name() string { return "contact_resolution" }

func (s *ContactResolutionStage) Process(ctx context.Context, pc *PipelineContext) (*PipelineContext, error) {
	msg := pc.IncomingMessage
	inbox := pc.Inbox

	// Try to find existing contact by source_id + inbox_id
	// Reference: Chatwoot ContactIdentifies where source_id = sender_id AND inbox_id
	existing, err := s.contactRepo.FindBySourceIDAndInboxID(ctx, msg.SenderID, inbox.ID)
	if err == nil && existing != nil {
		applogger.L().Infof("ContactResolutionStage: found existing contact id=%d for source_id=%s",
			existing.ID, msg.SenderID)

		// Update contact name if the message provides a newer display name
		if msg.SenderName != "" && existing.Name != msg.SenderName {
			existing.Name = msg.SenderName
			updated, updateErr := s.contactRepo.Update(ctx, existing)
			if updateErr != nil {
				applogger.L().Warnf("ContactResolutionStage: failed to update contact name: %v", updateErr)
			} else {
				existing = updated
			}
		}

		pc.Contact = existing
		return pc, nil
	}

	// Create new contact
	// Reference: Chatwoot Contacts::InboxCreateService — creates Contact + ContactInbox
	contact := &model.Contact{
		AccountID: inbox.AccountID,
		Name:      msg.SenderName,
		// SourceID would be stored in a separate ContactInbox join table in Chatwoot;
		// in GoChat we use the Identifier field for the external source ID.
		Identifier: msg.SenderID,
	}
	if msg.SenderName == "" {
		contact.Name = fmt.Sprintf("Contact_%s", msg.SenderID)
	}

	created, err := s.contactRepo.Create(ctx, contact)
	if err != nil {
		return nil, fmt.Errorf("failed to create contact for source_id=%s: %w", msg.SenderID, err)
	}

	applogger.L().Infof("ContactResolutionStage: created new contact id=%d for source_id=%s",
		created.ID, msg.SenderID)
	pc.Contact = created
	return pc, nil
}

// ConversationResolutionStage finds or creates the Conversation for the contact + inbox.
// Reference: Chatwoot Conversation.find_or_create_by(contact_id, inbox_id, status=open).
// When a contact already has an open conversation in the same inbox, we reuse it.
// Otherwise we create a new one.
type ConversationResolutionStage struct {
	conversationRepo ConversationRepository
}

func (s *ConversationResolutionStage) Name() string { return "conversation_resolution" }

func (s *ConversationResolutionStage) Process(ctx context.Context, pc *PipelineContext) (*PipelineContext, error) {
	inbox := pc.Inbox
	contact := pc.Contact

	// Try to find an existing open conversation for this contact + inbox
	existing, err := s.conversationRepo.FindOpenByContactIDAndInboxID(ctx, contact.ID, inbox.ID)
	if err == nil && existing != nil {
		applogger.L().Infof("ConversationResolutionStage: found existing open conversation id=%d",
			existing.ID)
		pc.Conversation = existing
		return pc, nil
	}

	// Create a new conversation
	// Reference: Chatwoot Conversations::CreateService
	now := time.Now().Unix()
	conversation := &model.Conversation{
		AccountID:     inbox.AccountID,
		InboxID:       inbox.ID,
		ContactID:     contact.ID,
		Status:        string(model.ConversationStatusOpen),
		ChannelType:   inbox.ChannelType,
		LastMessageAt: &now,
	}

	created, err := s.conversationRepo.Create(ctx, conversation)
	if err != nil {
		return nil, fmt.Errorf("failed to create conversation for contact_id=%d inbox_id=%d: %w",
			contact.ID, inbox.ID, err)
	}

	applogger.L().Infof("ConversationResolutionStage: created new conversation id=%d",
		created.ID)
	pc.Conversation = created
	return pc, nil
}

// MessagePersistenceStage creates the Message record from the incoming message data.
// Reference: Chatwoot Message.create! — maps the parsed IncomingMessage to a
// persisted Message model with conversation_id, account_id, inbox_id.
type MessagePersistenceStage struct {
	messageRepo MessageRepository
}

func (s *MessagePersistenceStage) Name() string { return "message_persistence" }

func (s *MessagePersistenceStage) Process(ctx context.Context, pc *PipelineContext) (*PipelineContext, error) {
	msg := pc.IncomingMessage
	inbox := pc.Inbox
	contact := pc.Contact
	conversation := pc.Conversation

	// Check for duplicate message by source_id + inbox_id
	// Reference: Chatwoot deduplication by message source_id
	existing, err := s.messageRepo.FindBySourceIDAndInboxID(ctx, msg.SourceID, inbox.ID)
	if err == nil && existing != nil {
		applogger.L().Infof("MessagePersistenceStage: duplicate message found id=%d source_id=%s, skipping",
			existing.ID, msg.SourceID)
		pc.Message = existing
		return pc, nil
	}

	// Map ContentType to model.ContentType
	contentType := mapContentType(msg.ContentType)

	// Map SenderType
	senderType := mapSenderType(msg.SenderType)

	message := &model.Message{
		ConversationID: conversation.ID,
		AccountID:      inbox.AccountID,
		InboxID:        inbox.ID,
		SenderType:     senderType,
		SenderID:       &contact.ID,
		Content:        msg.Content,
		ContentType:    contentType,
		MessageType:    string(model.MessageTypeIncoming),
		SourceID:       msg.SourceID,
		Private:        false,
	}

	created, err := s.messageRepo.Create(ctx, message)
	if err != nil {
		return nil, fmt.Errorf("failed to persist message source_id=%s: %w", msg.SourceID, err)
	}

	applogger.L().Infof("MessagePersistenceStage: created message id=%d source_id=%s",
		created.ID, msg.SourceID)
	pc.Message = created
	return pc, nil
}

// EventDispatchStage resolves the Account and prepares event dispatch data.
// The actual event publishing is done by the MessageBroker after the pipeline
// completes, but this stage ensures the Account is loaded and available.
// Reference: Chatwoot Dispatcher fires Wisper events after pipeline completion.
type EventDispatchStage struct {
	accountRepo AccountRepository
}

func (s *EventDispatchStage) Name() string { return "event_dispatch" }

func (s *EventDispatchStage) Process(ctx context.Context, pc *PipelineContext) (*PipelineContext, error) {
	inbox := pc.Inbox

	account, err := s.accountRepo.FindByID(ctx, inbox.AccountID)
	if err != nil {
		return nil, fmt.Errorf("failed to load account id=%d: %w", inbox.AccountID, err)
	}

	pc.Account = account
	applogger.L().Infof("EventDispatchStage: account resolved id=%d name=%s",
		account.ID, account.Name)
	return pc, nil
}

// --- Helper Functions ---

// mapContentType maps channel ContentType to model content type string.
func mapContentType(ct ContentType) string {
	switch ct {
	case ContentText:
		return string(model.MessageContentTypeText)
	case ContentImage:
		return "image"
	case ContentFile:
		return "file"
	case ContentAudio:
		return "audio"
	case ContentVideo:
		return "video"
	case ContentLocation:
		return "location"
	case ContentEmail:
		return "email"
	case ContentTemplate:
		return "template"
	default:
		return string(model.MessageContentTypeText)
	}
}

// mapSenderType maps channel SenderType to model sender type string.
func mapSenderType(st SenderType) string {
	switch st {
	case SenderContact:
		return "Contact"
	case SenderAgent:
		return "User"
	case SenderSystem:
		return "System"
	default:
		return "Contact"
	}
}

// marshalJSON is a helper for JSON encoding used in broker event publishing.
func marshalJSON(v interface{}) ([]byte, error) {
	return json.Marshal(v)
}