gochat/backend/internal/handler/api/v1/sse_event_handler.go

// Package v1 provides the SSE (Server-Sent Events) handler for real-time event push.
// This is the SSE fallback for clients that don't need bidirectional WebSocket
// communication but still want real-time updates (one-way server push).
//
// Endpoint: GET /api/v1/accounts/:account_id/events
// Content-Type: text/event-stream
//
// SSE event format (matching Chatwoot ActionCable event types):
//   event: message.created
//   data: {"id":1,"content":"hello"}
//
// Reference: Chatwoot ActionCable — this endpoint provides the same real-time
// events as WebSocket but via SSE protocol for simpler client integration.
package v1

import (
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"net/http"
	"strconv"

	"github.com/gin-gonic/gin"

	wspkg "github.com/gochat/gochat/internal/ws"
	applogger "github.com/gochat/gochat/pkg/logger"
)

// SSEEventHandler handles Server-Sent Events for real-time event push.
// It provides an SSE fallback endpoint matching Chatwoot's ActionCable
// events, for clients that prefer unidirectional HTTP streaming over WebSocket.
type SSEEventHandler struct {
	registry *wspkg.SSERegistry
}

// NewSSEEventHandler creates a new SSE event handler with the given registry.
func NewSSEEventHandler(registry *wspkg.SSERegistry) *SSEEventHandler {
	return &SSEEventHandler{
		registry: registry,
	}
}

// StreamEvents handles the SSE streaming endpoint.
// GET /api/v1/accounts/:account_id/events
//
// The handler:
//   1. Validates the authenticated user has access to the account
//   2. Subscribes the SSE client to the account's event stream
//   3. Flushes events as they arrive in the text/event-stream format
//   4. Handles client disconnect by unsubscribing
//
// SSE protocol:
//   - Content-Type: text/event-stream
//   - Each event: "event: <type>\ndata: <json>\n\n"
//   - Keep-alive comments: ": keepalive\n\n" (every 30s)
func (h *SSEEventHandler) StreamEvents(c *gin.Context) {
	// Extract account ID from URL
	accountIDStr := c.Param("account_id")
	accountID, err := strconv.ParseUint(accountIDStr, 10, 32)
	if err != nil {
		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid account_id"})
		return
	}

	// Get authenticated user ID from context (set by AuthMiddleware)
	userID, exists := c.Get("user_id")
	if !exists {
		c.JSON(http.StatusUnauthorized, gin.H{"error": "unauthorized"})
		return
	}
	userIDUint, ok := userID.(uint)
	if !ok {
		// user_id may be set as float64 from JSON claims
		if f, ok := userID.(float64); ok {
			userIDUint = uint(f)
		} else {
			c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid user_id"})
			return
		}
	}

	// Set SSE headers
	c.Header("Content-Type", "text/event-stream")
	c.Header("Cache-Control", "no-cache")
	c.Header("Connection", "keep-alive")
	c.Header("X-Accel-Buffering", "no") // Disable nginx buffering
	c.Status(http.StatusOK)

	// Generate unique channel ID
	channelID := generateSSEChannelID()

	// Subscribe to the SSE registry
	channel := h.registry.Subscribe(channelID, uint(accountID), userIDUint)

	applogger.L().Infof("sse: client connected (channel=%s, account=%d, user=%d)",
		channelID, accountID, userIDUint)

	// Flush headers to client
	c.Writer.Flush()

	// Stream events until client disconnects
	defer h.registry.Unsubscribe(channelID)

	// Send initial connection confirmation
	c.Writer.WriteString(fmt.Sprintf("event: connected\ndata: {\"channel_id\":\"%s\"}\n\n", channelID))
	c.Writer.Flush()

	for {
		select {
		case event, ok := <-channel.Events:
			if !ok {
				// Channel closed (unsubscribed)
				return
			}
			sseFormatted, err := wspkg.FormatSSE(event)
			if err != nil {
				applogger.L().Warnf("sse: failed to format event: %v", err)
				continue
			}
			_, err = c.Writer.WriteString(sseFormatted)
			if err != nil {
				// Client disconnected
				applogger.L().Infof("sse: client disconnected (channel=%s)", channelID)
				return
			}
			c.Writer.Flush()

		case <-c.Request.Context().Done():
			// HTTP connection closed by client
			applogger.L().Infof("sse: client context done (channel=%s)", channelID)
			return
		}
	}
}

// generateSSEChannelID creates a unique SSE channel ID using crypto/rand.
func generateSSEChannelID() string {
	b := make([]byte, 16)
	_, _ = rand.Read(b)
	return "sse_" + hex.EncodeToString(b)
}