second commit

2026-06-04 15:44:48 +08:00
parent 4db6efb3a7
commit 8ac150bc7b
1275 changed files with 286124 additions and 0 deletions
@@ -0,0 +1,158 @@
+package dispatch
+
+import (
+	"context"
+	"time"
+
+	"github.com/gochat/gochat/internal/channel"
+	applogger "github.com/gochat/gochat/pkg/logger"
+)
+
+// 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
+}
+
+// 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),
+	}
+}
+
+// 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 {
+			// Async: run in background goroutine
+			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("async listener %s error on event %s: %v", l.Name(), e.Type, err)
+				}
+			}(entry.listener, event)
+		}
+	}
+
+	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
+		}
+		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("async dispatch: listener %s error on event %s: %v", l.Name(), e.Type, err)
+			}
+		}(entry.listener, 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
+}
@@ -0,0 +1,155 @@
+package dispatch
+
+// Event types and helper functions for the enhanced dispatch system.
+//
+// Reference: Chatwoot event naming conventions
+//   - Chatwoot uses Wisper-style event names like "conversation.created",
+//     "message.created", "contact.created", etc.
+//   - In gochat, channel.EventType already defines these constants.
+//   - This file provides additional event types specific to the dispatch layer
+//     and helper functions for event construction.
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/gochat/gochat/internal/channel"
+)
+
+// DispatchEvent defines additional event types specific to the dispatch layer.
+// These complement the channel.EventType constants already defined in
+// channel/event.go.
+type DispatchEvent string
+
+const (
+	// Auto-assignment events (triggered by autoassignment listener)
+	DispatchEventAutoAssign        DispatchEvent = "auto_assignment.assign"
+	DispatchEventAutoAssignFailed  DispatchEvent = "auto_assignment.assign_failed"
+	DispatchEventAutoAssignSkipped DispatchEvent = "auto_assignment.assign_skipped"
+
+	// Reporting events (for analytics and reporting)
+	DispatchEventReportGenerated DispatchEvent = "report.generated"
+	DispatchEventReportFailed    DispatchEvent = "report.failed"
+
+	// CSAT survey events
+	DispatchEventCsatSurveySent    DispatchEvent = "csat_survey.sent"
+	DispatchEventCsatResponseReceived DispatchEvent = "csat_survey.response_received"
+
+	// Campaign events
+	DispatchEventCampaignStarted DispatchEvent = "campaign.started"
+	DispatchEventCampaignCompleted DispatchEvent = "campaign.completed"
+
+	// Webhook/hook events
+	DispatchEventHookTriggered      DispatchEvent = "hook.triggered"
+	DispatchEventWebhookSent        DispatchEvent = "webhook.sent"
+	DispatchEventWebhookFailed      DispatchEvent = "webhook.failed"
+)
+
+// NewChannelEvent creates a ChannelEvent with the given type and data.
+// Timestamp is set to the current Unix milliseconds.
+func NewChannelEvent(eventType string, data map[string]interface{}) *channel.ChannelEvent {
+	if data == nil {
+		data = make(map[string]interface{})
+	}
+	return &channel.ChannelEvent{
+		Type:      channel.EventType(eventType),
+		Data:      data,
+		Timestamp: time.Now().UnixMilli(),
+	}
+}
+
+// NewDispatchEvent creates a ChannelEvent from a DispatchEvent type.
+func NewDispatchEvent(eventType DispatchEvent, data map[string]interface{}) *channel.ChannelEvent {
+	return NewChannelEvent(string(eventType), data)
+}
+
+// EventDataWithConversation creates event data containing a conversation ID.
+// This is a common pattern where events need to reference a specific conversation.
+func EventDataWithConversation(conversationID uint, accountID uint, inboxID uint) map[string]interface{} {
+	return map[string]interface{}{
+		"conversation_id": conversationID,
+		"account_id":      accountID,
+		"inbox_id":        inboxID,
+	}
+}
+
+// EventDataWithAssignment creates event data for an assignment event,
+// including the agent who was assigned.
+func EventDataWithAssignment(conversationID uint, accountID uint, inboxID uint, assigneeID uint) map[string]interface{} {
+	data := EventDataWithConversation(conversationID, accountID, inboxID)
+	data["assignee_id"] = assigneeID
+	return data
+}
+
+// ExtractConversationID extracts a conversation ID from event data.
+// Returns 0 and an error if not found or not a valid uint.
+func ExtractConversationID(data map[string]interface{}) (uint, error) {
+	raw, ok := data["conversation_id"]
+	if !ok {
+		return 0, fmt.Errorf("conversation_id not found in event data")
+	}
+	switch v := raw.(type) {
+	case uint:
+		return v, nil
+	case int:
+		return uint(v), nil
+	case float64:
+		return uint(v), nil
+	default:
+		return 0, fmt.Errorf("unexpected type for conversation_id: %T", raw)
+	}
+}
+
+// ExtractAccountID extracts an account ID from event data.
+func ExtractAccountID(data map[string]interface{}) (uint, error) {
+	raw, ok := data["account_id"]
+	if !ok {
+		return 0, fmt.Errorf("account_id not found in event data")
+	}
+	switch v := raw.(type) {
+	case uint:
+		return v, nil
+	case int:
+		return uint(v), nil
+	case float64:
+		return uint(v), nil
+	default:
+		return 0, fmt.Errorf("unexpected type for account_id: %T", raw)
+	}
+}
+
+// ExtractInboxID extracts an inbox ID from event data.
+func ExtractInboxID(data map[string]interface{}) (uint, error) {
+	raw, ok := data["inbox_id"]
+	if !ok {
+		return 0, fmt.Errorf("inbox_id not found in event data")
+	}
+	switch v := raw.(type) {
+	case uint:
+		return v, nil
+	case int:
+		return uint(v), nil
+	case float64:
+		return uint(v), nil
+	default:
+		return 0, fmt.Errorf("unexpected type for inbox_id: %T", raw)
+	}
+}
+
+// ExtractAssigneeID extracts an assignee (agent) ID from event data.
+func ExtractAssigneeID(data map[string]interface{}) (uint, error) {
+	raw, ok := data["assignee_id"]
+	if !ok {
+		return 0, fmt.Errorf("assignee_id not found in event data")
+	}
+	switch v := raw.(type) {
+	case uint:
+		return v, nil
+	case int:
+		return uint(v), nil
+	case float64:
+		return uint(v), nil
+	default:
+		return 0, fmt.Errorf("unexpected type for assignee_id: %T", raw)
+	}
+}
@@ -0,0 +1,87 @@
+package dispatch
+
+// ListenerRegistry maps event names to lists of listener names.
+// This allows efficient filtering: when an event of type X is dispatched,
+// the dispatcher only calls listeners registered for X.
+//
+// Reference: Chatwoot Wisper event routing pattern
+//   - In Chatwoot, each listener subscribes to specific event names via Wisper.
+//   - The registry stores these subscriptions so the dispatcher can efficiently
+//     route events only to interested listeners.
+
+type ListenerRegistry struct {
+	// subscriptions maps event name → set of listener names.
+	// An event name of "*" means the listener subscribes to all events.
+	subscriptions map[string]map[string]bool // eventName → listenerName → true
+}
+
+// NewListenerRegistry creates an empty registry.
+func NewListenerRegistry() *ListenerRegistry {
+	return &ListenerRegistry{
+		subscriptions: make(map[string]map[string]bool),
+	}
+}
+
+// Subscribe registers a listener for the given event names.
+// If eventNames is empty, the listener subscribes to all events (wildcard).
+func (lr *ListenerRegistry) Subscribe(listenerName string, eventNames ...string) {
+	if len(eventNames) == 0 {
+		eventNames = []string{"*"} // wildcard: receives all events
+	}
+
+	for _, eventName := range eventNames {
+		if lr.subscriptions[eventName] == nil {
+			lr.subscriptions[eventName] = make(map[string]bool)
+		}
+		lr.subscriptions[eventName][listenerName] = true
+	}
+}
+
+// Unsubscribe removes a listener from all event subscriptions.
+func (lr *ListenerRegistry) Unsubscribe(listenerName string) {
+	for eventName, listeners := range lr.subscriptions {
+		delete(listeners, listenerName)
+		if len(listeners) == 0 {
+			delete(lr.subscriptions, eventName)
+		}
+	}
+}
+
+// SubscribersForEvent returns all listener names subscribed to the given event.
+// It includes both event-specific subscribers and wildcard subscribers.
+func (lr *ListenerRegistry) SubscribersForEvent(eventName string) []string {
+	result := make(map[string]bool)
+
+	// Add wildcard subscribers (subscribed to all events)
+	if wildcards, ok := lr.subscriptions["*"]; ok {
+		for name := range wildcards {
+			result[name] = true
+		}
+	}
+
+	// Add event-specific subscribers
+	if subs, ok := lr.subscriptions[eventName]; ok {
+		for name := range subs {
+			result[name] = true
+		}
+	}
+
+	names := make([]string, 0, len(result))
+	for name := range result {
+		names = append(names, name)
+	}
+	return names
+}
+
+// AllSubscriptions returns the full subscription map (for debugging/inspection).
+func (lr *ListenerRegistry) AllSubscriptions() map[string]map[string]bool {
+	return lr.subscriptions
+}
+
+// HasSubscribers checks if any listener is subscribed to the given event.
+func (lr *ListenerRegistry) HasSubscribers(eventName string) bool {
+	if len(lr.subscriptions["*"]) > 0 {
+		return true
+	}
+	return len(lr.subscriptions[eventName]) > 0
+}