gochat/internal/channel/broker.go

package channel

import (
	"context"
	"fmt"

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

// Broker is the central message routing interface that coordinates incoming
// and outgoing message flow through registered ChannelProviders.
// Reference: Chatwoot Dispatcher — singleton that delegates to SyncDispatcher
// + AsyncDispatcher, which uses Wisper pub/sub to broadcast events to Listener
// instances (WebhookListener, NotificationListener, etc).
type Broker interface {
	// HandleIncoming processes an inbound message from a channel provider.
	// The message is routed through the IncomingMessageProcessor pipeline
	// (validate → resolve contact → resolve conversation → persist → fire events).
	HandleIncoming(ctx context.Context, inbox *model.Inbox, msg *IncomingMessage) error

	// HandleOutgoing processes an outbound message request.
	// The message is routed through the OutgoingMessageProcessor pipeline
	// (validate → resolve config → send via provider → update source ID → fire events).
	HandleOutgoing(ctx context.Context, inbox *model.Inbox, message *model.Message, contact *model.Contact) error
}

// MessageBroker is the concrete implementation of Broker.
// It holds references to the channel registry, the incoming/outgoing
// processors, and the event bus for cross-instance event broadcasting.
//
// Reference: Chatwoot Dispatcher pattern:
//   - Incoming: per-channel IncomingMessageService parses webhook payload →
//     create/update Contact → find/create Conversation → create Message → fire events
//   - Outgoing: per-channel SendOn*Service (parent Base::SendOnChannelService) sends
//     message via external API → update source_id → fire events
type MessageBroker struct {
	registry  *ChannelRegistry
	incoming  *IncomingMessageProcessor
	outgoing  *OutgoingMessageProcessor
	eventBus  *pubsub.EventBus
}

// NewMessageBroker creates a new MessageBroker with the given dependencies.
func NewMessageBroker(
	registry *ChannelRegistry,
	incoming *IncomingMessageProcessor,
	outgoing *OutgoingMessageProcessor,
	eventBus *pubsub.EventBus,
) *MessageBroker {
	return &MessageBroker{
		registry:  registry,
		incoming:  incoming,
		outgoing:  outgoing,
		eventBus:  eventBus,
	}
}

// HandleIncoming processes an inbound message from a channel provider.
// It delegates to the IncomingMessageProcessor pipeline and then publishes
// the resulting events to the EventBus for cross-instance propagation.
func (b *MessageBroker) HandleIncoming(ctx context.Context, inbox *model.Inbox, msg *IncomingMessage) error {
	applogger.L().Infof("Broker: handling incoming message from %s inbox=%d source_id=%s",
		inbox.ChannelType, inbox.ID, msg.SourceID)

	// Build the pipeline context with inbox + incoming message
	pipelineCtx := &PipelineContext{
		Inbox:           inbox,
		IncomingMessage: msg,
	}

	// Run through the incoming pipeline
	result, err := b.incoming.Process(ctx, pipelineCtx)
	if err != nil {
		applogger.L().Errorf("Broker: incoming pipeline failed: inbox=%d err=%v", inbox.ID, err)
		return fmt.Errorf("incoming pipeline failed for inbox %d: %w", inbox.ID, err)
	}

	// Publish events to the event bus for cross-module propagation.
	// Reference: Chatwoot Dispatcher fires Wisper events after message creation,
	// which are consumed by NotificationListener, WebhookListener, etc.
	if result.Message != nil {
		topic := pubsub.TopicMessageCreated
		payload, encodeErr := encodeEventPayload(EventMessageCreated, result.Message)
		if encodeErr != nil {
			applogger.L().Errorf("Broker: encode event payload failed: %v", encodeErr)
		} else if publishErr := b.eventBus.Publish(topic, payload); publishErr != nil {
			applogger.L().Errorf("Broker: publish event failed: topic=%s err=%v", topic, publishErr)
		}
	}

	if result.Conversation != nil {
		topic := pubsub.TopicConversationUpdated
		payload, encodeErr := encodeEventPayload(EventConversationUpdated, result.Conversation)
		if encodeErr != nil {
			applogger.L().Errorf("Broker: encode conversation event payload failed: %v", encodeErr)
		} else if publishErr := b.eventBus.Publish(topic, payload); publishErr != nil {
			applogger.L().Errorf("Broker: publish conversation event failed: topic=%s err=%v", topic, publishErr)
		}
	}

	if result.Contact != nil {
		channelTopic := pubsub.FormatTopic(pubsub.TopicChannelEvent, inbox.ChannelType)
		payload, encodeErr := encodeEventPayload(EventContactCreated, result.Contact)
		if encodeErr != nil {
			applogger.L().Errorf("Broker: encode contact event payload failed: %v", encodeErr)
		} else if publishErr := b.eventBus.Publish(channelTopic, payload); publishErr != nil {
			applogger.L().Errorf("Broker: publish contact event failed: topic=%s err=%v", channelTopic, publishErr)
		}
	}

	applogger.L().Infof("Broker: incoming message processed successfully: message_id=%d conversation_id=%d",
		result.Message.ID, result.Conversation.ID)
	return nil
}

// HandleOutgoing processes an outbound message request.
// It delegates to the OutgoingMessageProcessor pipeline and then publishes
// the resulting events to the EventBus.
func (b *MessageBroker) HandleOutgoing(ctx context.Context, inbox *model.Inbox, message *model.Message, contact *model.Contact) error {
	applogger.L().Infof("Broker: handling outgoing message to %s inbox=%d message_id=%d",
		inbox.ChannelType, inbox.ID, message.ID)

	// Build the outgoing pipeline context
	outCtx := &OutgoingPipelineContext{
		Inbox:    inbox,
		Message:  message,
		Contact:  contact,
	}

	// Run through the outgoing pipeline
	result, err := b.outgoing.Process(ctx, outCtx)
	if err != nil {
		applogger.L().Errorf("Broker: outgoing pipeline failed: inbox=%d message_id=%d err=%v",
			inbox.ID, message.ID, err)
		return fmt.Errorf("outgoing pipeline failed for inbox %d message %d: %w", inbox.ID, message.ID, err)
	}

	// Publish outgoing message event
	if result.Message != nil {
		topic := pubsub.TopicMessageUpdated
		payload, encodeErr := encodeEventPayload(EventMessageOutgoing, result.Message)
		if encodeErr != nil {
			applogger.L().Errorf("Broker: encode outgoing event payload failed: %v", encodeErr)
		} else if publishErr := b.eventBus.Publish(topic, payload); publishErr != nil {
			applogger.L().Errorf("Broker: publish outgoing event failed: topic=%s err=%v", topic, publishErr)
		}
	}

	applogger.L().Infof("Broker: outgoing message processed successfully: message_id=%d source_id=%s",
		result.Message.ID, result.Message.SourceID)
	return nil
}

// encodeEventPayload marshals an event type + data into JSON bytes for pub/sub.
func encodeEventPayload(eventType EventType, data interface{}) ([]byte, error) {
	// We import encoding/json inline in this function to keep the event
	// encoding simple. In production, this could be replaced with a more
	// efficient serialization (protobuf, msgpack, etc.)
	payload := struct {
		EventType EventType   `json:"event_type"`
		Data      interface{} `json:"data"`
	}{
		EventType: eventType,
		Data:      data,
	}
	// Use stdlib json encoding
	return marshalJSON(payload)
}