gochat/internal/channel/outgoing.go

package channel

import (
	"context"
	"fmt"

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

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

// OutgoingPipelineStage is the interface for a single step in the outgoing
// message processing pipeline. Each stage receives the accumulated
// OutgoingPipelineContext and returns a modified context or an error.
// Reference: Chatwoot SendOn*Service pattern — per-channel outbound services
// inheriting from Base::SendOnChannelService. GoChat unifies into a pipeline.
type OutgoingPipelineStage interface {
	// Name returns a human-readable stage name for logging.
	Name() string

	// Process executes the pipeline stage, mutating outCtx as needed.
	Process(ctx context.Context, outCtx *OutgoingPipelineContext) (*OutgoingPipelineContext, error)
}

// OutgoingPipelineContext carries state through the outgoing message pipeline.
type OutgoingPipelineContext struct {
	// Input fields (set before pipeline starts)
	Inbox   *model.Inbox     `json:"inbox"`
	Message *model.Message   `json:"message"`
	Contact *model.Contact   `json:"contact"`

	// Accumulated fields (populated by pipeline stages)
	ChannelConfig ChannelConfig  `json:"channel_config,omitempty"` // resolved per-inbox config
	SendResult    *SendResult    `json:"send_result,omitempty"`    // result from provider send
	Provider      ChannelProvider `json:"-"`                        // resolved channel provider
}

// --- OutgoingMessageProcessor ---

// OutgoingMessageProcessor orchestrates the outgoing message pipeline.
// It holds an ordered list of OutgoingPipelineStage implementations and runs
// them sequentially, aborting on the first error.
//
// Reference: Chatwoot Base::SendOnChannelService + per-channel SendOn*Service
//   - Validates message & channel
//   - Resolves channel configuration
//   - Sends message via external channel API (Telegram Bot API, FB Graph API, etc.)
//   - Updates message source_id with external ID from API response
//   - Fires Wisper events for notifications, webhooks, etc.
type OutgoingMessageProcessor struct {
	stages []OutgoingPipelineStage
}

// NewOutgoingMessageProcessor creates a processor with the standard pipeline stages.
func NewOutgoingMessageProcessor(
	messageRepo MessageRepository,
	inboxRepo InboxRepository,
) *OutgoingMessageProcessor {
	stages := []OutgoingPipelineStage{
		&ValidateOutgoingStage{},
		&ConfigResolutionStage{inboxRepo: inboxRepo},
		&SendMessageStage{},
		&UpdateMessageStage{messageRepo: messageRepo},
		&EventDispatchOutgoingStage{},
	}
	return &OutgoingMessageProcessor{stages: stages}
}

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

// --- Outgoing Pipeline Stages ---

// ValidateOutgoingStage validates the outgoing message and inbox for required fields.
// Ensures that the message has content, a valid inbox, and a registered provider.
// Reference: Chatwoot SendOnChannelService validates message before sending.
type ValidateOutgoingStage struct{}

func (s *ValidateOutgoingStage) Name() string { return "validate_outgoing" }

func (s *ValidateOutgoingStage) Process(ctx context.Context, oc *OutgoingPipelineContext) (*OutgoingPipelineContext, error) {
	message := oc.Message
	if message == nil {
		return nil, fmt.Errorf("outgoing message is nil")
	}
	if message.Content == "" && message.ContentType != "file" && message.ContentType != "image" {
		return nil, fmt.Errorf("outgoing message id=%d has no content", message.ID)
	}
	if message.MessageType != string(model.MessageTypeOutgoing) {
		return nil, fmt.Errorf("message id=%d is not an outgoing message (type=%s)",
			message.ID, message.MessageType)
	}

	inbox := oc.Inbox
	if inbox == nil {
		return nil, fmt.Errorf("inbox is nil")
	}
	if !inbox.EnableAutoAssignment {
		return nil, fmt.Errorf("inbox %d is disabled", 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)
	}
	oc.Provider = provider

	applogger.L().Infof("ValidateOutgoingStage: message id=%d valid, provider=%s",
		message.ID, provider.Name())
	return oc, nil
}

// ConfigResolutionStage resolves the channel configuration for the inbox.
// It loads the inbox's ChannelConfig and validates it against the provider's
// schema, ensuring the channel is properly configured before sending.
// Reference: Chatwoot channels use per-inbox config stored as JSON in the
// polymorphic channel model (e.g. Channel::Telegram has bot_token, etc.)
type ConfigResolutionStage struct {
	inboxRepo InboxRepository
}

func (s *ConfigResolutionStage) Name() string { return "config_resolution" }

func (s *ConfigResolutionStage) Process(ctx context.Context, oc *OutgoingPipelineContext) (*OutgoingPipelineContext, error) {
	inbox := oc.Inbox
	provider := oc.Provider

	// Load full inbox with channel config from repository
	// The inbox passed in may be a partial object; reload to get config
	fullInbox, err := s.inboxRepo.FindByID(inbox.ID)
	if err != nil {
		return nil, fmt.Errorf("failed to load inbox id=%d: %w", inbox.ID, err)
	}

	// Resolve channel config from inbox
	// Reference: Chatwoot stores per-channel config in the polymorphic channel model.
	// In GoChat we store it in the Inbox model itself via a JSON column or separate
	// config table. For now, we pass it via the pipeline context.
	channelConfig := make(ChannelConfig)

	// If the inbox has additional config stored, merge it here.
	// This is where you'd deserialize the Inbox's config JSON column.
	// For the pipeline abstraction, we just ensure the config is available.
	if fullInbox != nil {
		oc.Inbox = fullInbox
	}

	// Validate config against provider schema
	if err := provider.ValidateConfig(ctx, channelConfig); err != nil {
		return nil, fmt.Errorf("channel config validation failed for inbox %d: %w", inbox.ID, err)
	}

	oc.ChannelConfig = channelConfig
	applogger.L().Infof("ConfigResolutionStage: config resolved and validated for inbox id=%d", inbox.ID)
	return oc, nil
}

// SendMessageStage sends the message via the channel provider's SendMessage method.
// Reference: Chatwoot per-channel SendOn*Service calls the external API
// (Telegram Bot API sendMessage, FB Graph API messages, WhatsApp Business API, etc.)
// Each provider handles its own API protocol, error handling, and retries.
type SendMessageStage struct{}

func (s *SendMessageStage) Name() string { return "send_message" }

func (s *SendMessageStage) Process(ctx context.Context, oc *OutgoingPipelineContext) (*OutgoingPipelineContext, error) {
	provider := oc.Provider
	inbox := oc.Inbox
	message := oc.Message
	contact := oc.Contact

	// Call the provider's SendMessage to send via the external channel API
	// Reference: Chatwoot SendOnChannelService.perform → per-channel API call
	result, err := provider.SendMessage(ctx, inbox, message, contact)
	if err != nil {
		return nil, fmt.Errorf("provider %s failed to send message id=%d: %w",
			provider.Name(), message.ID, err)
	}

	oc.SendResult = result
	applogger.L().Infof("SendMessageStage: message id=%d sent via %s, external_id=%s",
		message.ID, provider.Name(), result.ExternalID)
	return oc, nil
}

// UpdateMessageStage updates the message's SourceID with the external ID
// returned by the channel provider. This links the GoChat message to the
// external channel message for future reference (delivery tracking, replies, etc.)
// Reference: Chatwoot updates message.source_id after successful send.
type UpdateMessageStage struct {
	messageRepo MessageRepository
}

func (s *UpdateMessageStage) Name() string { return "update_message" }

func (s *UpdateMessageStage) Process(ctx context.Context, oc *OutgoingPipelineContext) (*OutgoingPipelineContext, error) {
	message := oc.Message
	sendResult := oc.SendResult

	if sendResult == nil || sendResult.ExternalID == "" {
		applogger.L().Warnf("UpdateMessageStage: no external_id from send result, skipping update")
		return oc, nil
	}

	// Update the message's source_id with the external channel message ID
	// Reference: Chatwoot message.update!(source_id: external_id)
	message.SourceID = sendResult.ExternalID

	updated, err := s.messageRepo.Update(ctx, message)
	if err != nil {
		// Non-critical: the message was already sent successfully,
		// we just couldn't persist the source_id. Log and continue.
		applogger.L().Warnf("UpdateMessageStage: failed to update message source_id: %v", err)
		return oc, nil
	}

	oc.Message = updated
	applogger.L().Infof("UpdateMessageStage: message id=%d updated with source_id=%s",
		updated.ID, updated.SourceID)
	return oc, nil
}

// EventDispatchOutgoingStage prepares event dispatch data for the outgoing message.
// The actual event publishing is done by the MessageBroker after the pipeline
// completes. This stage performs any final state mutations needed before events.
// Reference: Chatwoot fires Wisper events after outgoing message delivery.
type EventDispatchOutgoingStage struct{}

func (s *EventDispatchOutgoingStage) Name() string { return "event_dispatch_outgoing" }

func (s *EventDispatchOutgoingStage) Process(ctx context.Context, oc *OutgoingPipelineContext) (*OutgoingPipelineContext, error) {
	// The MessageBroker handles actual event publishing after the pipeline.
	// This stage is a placeholder for any final pre-dispatch mutations,
	// such as updating conversation last_message_at, or marking the
	// conversation as needing notification delivery.

	if oc.Message != nil {
		// In a full implementation, we'd load the conversation here
		// to update last_message_at and trigger notification events.
		// For the pipeline abstraction, we just ensure the message
		// context is complete for the broker to publish events.
		applogger.L().Infof("EventDispatchOutgoingStage: outgoing pipeline context ready for event dispatch")
	}

	return oc, nil
}