gochat/internal/ws/sse_registry.go

// Package ws provides SSE (Server-Sent Events) channel registry and management
// for clients that prefer unidirectional real-time updates over WebSocket.
// This mirrors Chatwoot's ActionCable events pushed via SSE as a simpler fallback.
package ws

import (
	"encoding/json"
	"fmt"
	"sync"
)

// SSEChannel represents a single SSE client subscription.
// Each SSE client subscribes to events for a specific account and optionally
// specific conversations, matching Chatwoot's room subscription model.
//
// Reference: Chatwoot ActionCable — AccountChannel and ConversationChannel
// subscriptions are replicated here via SSE event filtering.
type SSEChannel struct {
	ID              string          // unique channel ID
	AccountID       uint           // account this client subscribes to
	UserID          uint           // authenticated user
	ConversationIDs map[uint]bool  // optional focused conversation subscriptions
	Events          chan SSEEvent  // buffered channel for outgoing SSE events
	Closed          bool           // marks channel as closed after disconnect
}

// SSEEvent represents a single SSE push event, formatted for the
// text/event-stream protocol:
//
//   event: <type>
//   data: <json>
//
// Reference: Chatwoot ActionCable event format — type/payload pairs
// broadcast to subscribed WebSocket clients. SSE uses the same event
// types but formatted per the SSE specification.
type SSEEvent struct {
	Type    string      `json:"type"`     // e.g. "message.created", "conversation.updated"
	Payload interface{} `json:"payload"`  // event data (typically a struct or map)
}

// FormatSSE formats an SSEEvent into the text/event-stream wire format:
//
//   event: <type>\n
//   data: <json>\n\n
func FormatSSE(event SSEEvent) (string, error) {
	payloadJSON, err := json.Marshal(event.Payload)
	if err != nil {
		return "", fmt.Errorf("marshal SSE payload: %w", err)
	}
	return fmt.Sprintf("event: %s\ndata: %s\n\n", event.Type, payloadJSON), nil
}

// SSERegistry manages all active SSE client subscriptions.
// It routes events from the EventPublisher to the appropriate SSE channels,
// mirroring the Hub's room-based routing but for unidirectional SSE clients.
//
// Architecture: EventPublisher → SSERegistry.SendToAccount/SendToConversation
// → SSEChannel.Events → HTTP handler flushes to client
type SSERegistry struct {
	mu      sync.RWMutex
	channels map[string]*SSEChannel  // channel ID → SSEChannel
	accounts map[uint]map[string]bool // account ID → set of channel IDs
}

// NewSSERegistry creates a new SSE channel registry.
func NewSSERegistry() *SSERegistry {
	return &SSERegistry{
		channels: make(map[string]*SSEChannel),
		accounts: make(map[uint]map[string]bool),
	}
}

// Subscribe creates a new SSE channel subscription for the given account.
// Returns the SSEChannel which the HTTP handler reads from.
func (r *SSERegistry) Subscribe(channelID string, accountID uint, userID uint) *SSEChannel {
	r.mu.Lock()
	defer r.mu.Unlock()

	ch := &SSEChannel{
		ID:              channelID,
		AccountID:       accountID,
		UserID:          userID,
		ConversationIDs: make(map[uint]bool),
		Events:          make(chan SSEEvent, 64),
	}

	r.channels[channelID] = ch

	if _, ok := r.accounts[accountID]; !ok {
		r.accounts[accountID] = make(map[string]bool)
	}
	r.accounts[accountID][channelID] = true

	return ch
}

// SubscribeConversation adds a conversation filter to an existing SSE channel.
// After this, the client will also receive events targeted at that conversation.
func (r *SSERegistry) SubscribeConversation(channelID string, conversationID uint) {
	r.mu.Lock()
	defer r.mu.Unlock()

	ch, ok := r.channels[channelID]
	if !ok {
		return
	}
	ch.ConversationIDs[conversationID] = true
}

// UnsubscribeConversation removes a conversation filter from an SSE channel.
func (r *SSERegistry) UnsubscribeConversation(channelID string, conversationID uint) {
	r.mu.Lock()
	defer r.mu.Unlock()

	ch, ok := r.channels[channelID]
	if !ok {
		return
	}
	delete(ch.ConversationIDs, conversationID)
}

// Unsubscribe removes an SSE channel subscription and closes the events channel.
func (r *SSERegistry) Unsubscribe(channelID string) {
	r.mu.Lock()
	defer r.mu.Unlock()

	ch, ok := r.channels[channelID]
	if !ok {
		return
	}

	ch.Closed = true
	close(ch.Events)

	delete(r.channels, channelID)

	if accountChannels, ok := r.accounts[ch.AccountID]; ok {
		delete(accountChannels, channelID)
		if len(accountChannels) == 0 {
			delete(r.accounts, ch.AccountID)
		}
	}
}

// SendToAccount sends an SSE event to all channels subscribed to the given account.
// This mirrors Hub.SendToAccount for WebSocket delivery.
func (r *SSERegistry) SendToAccount(accountID uint, event SSEEvent) {
	r.mu.RLock()
	defer r.mu.RUnlock()

	if channelIDs, ok := r.accounts[accountID]; ok {
		for channelID := range channelIDs {
			ch, ok := r.channels[channelID]
			if !ok || ch.Closed {
				continue
			}
			select {
			case ch.Events <- event:
			default:
				// Drop event for slow SSE client (matching WebSocket Hub behavior)
			}
		}
	}
}

// SendToConversation sends an SSE event to channels subscribed to a specific conversation
// within an account. Also sends to channels subscribed to the account (no conversation filter).
func (r *SSERegistry) SendToConversation(accountID uint, conversationID uint, event SSEEvent) {
	r.mu.RLock()
	defer r.mu.RUnlock()

	if channelIDs, ok := r.accounts[accountID]; ok {
		for channelID := range channelIDs {
			ch, ok := r.channels[channelID]
			if !ok || ch.Closed {
				continue
			}
			// Send to channels that are subscribed to this conversation
			// or that have no conversation filters (broad account subscription)
			if len(ch.ConversationIDs) == 0 || ch.ConversationIDs[conversationID] {
				select {
				case ch.Events <- event:
				default:
					// Drop event for slow SSE client
				}
			}
		}
	}
}

// ChannelCount returns the number of active SSE channels for a given account.
func (r *SSERegistry) ChannelCount(accountID uint) int {
	r.mu.RLock()
	defer r.mu.RUnlock()

	if channels, ok := r.accounts[accountID]; ok {
		return len(channels)
	}
	return 0
}

// TotalChannelCount returns the total number of active SSE channels across all accounts.
func (r *SSERegistry) TotalChannelCount() int {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return len(r.channels)
}