gochat/internal/dispatch/dispatcher.go

package dispatch

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

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

const TaskTypeEventListenerDispatch = "event:listener_dispatch"

// EventDispatcher wraps the channel.Dispatcher and adds:
//   - Sync/async split: sync listeners run immediately; async listeners are queued
//   - Event name routing: listeners subscribe to specific event names
//   - Singleton pattern: one EventDispatcher per application
//
// Reference: Chatwoot Dispatcher pattern
//   - Chatwoot::Dispatcher is a singleton that dispatches events to SyncDispatcher
//     (immediate) and AsyncDispatcher (Sidekiq job).
//   - SyncDispatcher listeners: ActionCableListener, AgentBotListener
//   - AsyncDispatcher listeners: AutomationRuleListener, CampaignListener,
//     CsatSurveyListener, HookListener, InstallationWebhookListener,
//     NotificationListener, ParticipationListener, ReportingEventListener, WebhookListener
//
// In gochat, we wrap the existing channel.Dispatcher to preserve its simple
// EventListener interface and add the sync/async + event-name-routing layer.

// ListenerMode determines whether a listener runs synchronously or asynchronously.
type ListenerMode int

const (
	// SyncMode means the listener runs in the same goroutine as the dispatcher call.
	// Use for lightweight, latency-sensitive listeners (e.g. ActionCable, AgentBot).
	SyncMode ListenerMode = iota

	// AsyncMode means the listener runs in a background goroutine.
	// Use for heavier listeners (e.g. AutomationRule, Notification, Webhook).
	AsyncMode
)

// listenerEntry tracks a registered listener together with its mode and
// the event names it subscribes to. If eventNames is empty, the listener
// receives all events (wildcard).
type listenerEntry struct {
	listener   channel.EventListener
	mode       ListenerMode
	eventNames []string // empty = wildcard (all events)
}

// EventDispatcher is the enhanced central dispatch hub.
// It wraps a channel.Dispatcher and adds sync/async routing and
// event-name filtering.
type EventDispatcher struct {
	channelDispatcher *channel.Dispatcher
	registry          *ListenerRegistry
	entries           map[string]listenerEntry // listener name → entry
	worker            *worker.WorkerPool
}

// NewEventDispatcher creates a new EventDispatcher wrapping the given
// channel.Dispatcher.
func NewEventDispatcher(cd *channel.Dispatcher) *EventDispatcher {
	return &EventDispatcher{
		channelDispatcher: cd,
		registry:          NewListenerRegistry(),
		entries:           make(map[string]listenerEntry),
	}
}

func (ed *EventDispatcher) SetWorkerPool(wp *worker.WorkerPool) {
	ed.worker = wp
	if wp != nil {
		wp.Register(TaskTypeEventListenerDispatch, ed.performListenerJob)
	}
}

// RegisterSync adds a sync-mode listener for the given event names.
// If eventNames is empty, the listener receives all events.
func (ed *EventDispatcher) RegisterSync(listener channel.EventListener, eventNames ...string) {
	ed.entries[listener.Name()] = listenerEntry{
		listener:   listener,
		mode:       SyncMode,
		eventNames: eventNames,
	}
	ed.registry.Subscribe(listener.Name(), eventNames...)
	// Also register with the underlying channel.Dispatcher so Dispatch() calls work.
	ed.channelDispatcher.Register(listener)
}

// RegisterAsync adds an async-mode listener for the given event names.
// If eventNames is empty, the listener receives all events.
func (ed *EventDispatcher) RegisterAsync(listener channel.EventListener, eventNames ...string) {
	ed.entries[listener.Name()] = listenerEntry{
		listener:   listener,
		mode:       AsyncMode,
		eventNames: eventNames,
	}
	ed.registry.Subscribe(listener.Name(), eventNames...)
	ed.channelDispatcher.Register(listener)
}

// Dispatch sends an event to all matching listeners.
// Sync listeners run immediately in the calling goroutine.
// Async listeners are dispatched via a background goroutine.
// Returns the first error encountered from sync listeners (async errors are logged only).
func (ed *EventDispatcher) Dispatch(ctx context.Context, event *channel.ChannelEvent) error {
	subscribers := ed.registry.SubscribersForEvent(string(event.Type))
	var firstSyncErr error

	for _, name := range subscribers {
		entry, ok := ed.entries[name]
		if !ok {
			continue
		}

		if entry.mode == SyncMode {
			if err := entry.listener.OnEvent(ctx, event); err != nil {
				applogger.L().Errorf("sync listener %s error on event %s: %v", entry.listener.Name(), event.Type, err)
				if firstSyncErr == nil {
					firstSyncErr = err
				}
			}
		} else {
			ed.dispatchListenerAsync(ctx, entry.listener.Name(), event, "async listener")
		}
	}

	return firstSyncErr
}

// DispatchAsync is a convenience method that always dispatches in async mode,
// running all listeners (both sync and async registrations) in background goroutines.
// This is useful for fire-and-forget event publishing.
func (ed *EventDispatcher) DispatchAsync(ctx context.Context, event *channel.ChannelEvent) {
	subscribers := ed.registry.SubscribersForEvent(string(event.Type))
	for _, name := range subscribers {
		entry, ok := ed.entries[name]
		if !ok {
			continue
		}
		ed.dispatchListenerAsync(ctx, entry.listener.Name(), event, "async dispatch")
	}
}

type listenerJobPayload struct {
	ListenerName string                `json:"listener_name"`
	Event        *channel.ChannelEvent `json:"event"`
}

func (ed *EventDispatcher) dispatchListenerAsync(ctx context.Context, listenerName string, event *channel.ChannelEvent, logPrefix string) {
	if ed.worker != nil {
		payload := listenerJobPayload{ListenerName: listenerName, Event: event}
		if _, err := ed.worker.Enqueue(ctx, TaskTypeEventListenerDispatch, payload, worker.WithQueue("events"), worker.WithMaxAttempts(3)); err != nil {
			applogger.L().Errorf("%s %s enqueue error on event %s: %v", logPrefix, listenerName, event.Type, err)
		}
		return
	}
	entry := ed.entries[listenerName]
	go func(l channel.EventListener, e *channel.ChannelEvent) {
		asyncCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
		defer cancel()
		if err := l.OnEvent(asyncCtx, e); err != nil {
			applogger.L().Errorf("%s: listener %s error on event %s: %v", logPrefix, l.Name(), e.Type, err)
		}
	}(entry.listener, event)
}

func (ed *EventDispatcher) performListenerJob(ctx context.Context, job *model.BackgroundJob) error {
	var payload listenerJobPayload
	if err := json.Unmarshal(job.Payload, &payload); err != nil {
		return fmt.Errorf("unmarshal listener dispatch payload: %w", err)
	}
	entry, ok := ed.entries[payload.ListenerName]
	if !ok {
		return fmt.Errorf("listener %q not registered", payload.ListenerName)
	}
	if payload.Event == nil {
		return fmt.Errorf("listener %q job missing event", payload.ListenerName)
	}
	return entry.listener.OnEvent(ctx, payload.Event)
}

// ChannelDispatcher returns the underlying channel.Dispatcher for direct access
// if needed (e.g. for channel-level dispatch without the enhanced routing).
func (ed *EventDispatcher) ChannelDispatcher() *channel.Dispatcher {
	return ed.channelDispatcher
}

// Registry returns the listener registry for inspection/testing.
func (ed *EventDispatcher) Registry() *ListenerRegistry {
	return ed.registry
}