minstrel/internal/api/events_stream.go

// Live event stream — Server-Sent Events at GET /api/events/stream.
//
// Clients (today: Flutter) subscribe to be notified of state changes
// elsewhere in the system (likes, request status flips, quarantine
// resolutions, scan progress, playlist mutations) so static screens
// can invalidate cached data without explicit user action.
//
// Wire-format: standard SSE. Each event has the shape
//
//	event: <kind>
//	data: <json payload>
//
// where <kind> follows "domain.action" naming and <json payload>
// matches `eventbus.Event`. A heartbeat comment line is emitted every
// 15 seconds to keep idle connections alive through proxies / NATs.
//
// Scoping: events with `user_id` matching the connected user OR with
// empty `user_id` (broadcast) are forwarded. Other users' events are
// dropped silently before reaching the wire.

package api

import (
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"github.com/jackc/pgx/v5/pgtype"

	"git.fabledsword.com/bvandeusen/minstrel/internal/auth"
	"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
)

// userUUIDString renders a pgtype.UUID in canonical 8-4-4-4-12 form so
// the SSE handler can compare against eventbus.Event.UserID strings.
// Other packages keep their own copies of this same pattern; consolidating
// is out of scope for this slice.
func userUUIDString(u pgtype.UUID) string {
	if !u.Valid {
		return ""
	}
	return fmt.Sprintf("%x-%x-%x-%x-%x",
		u.Bytes[0:4], u.Bytes[4:6], u.Bytes[6:8], u.Bytes[8:10], u.Bytes[10:16])
}

const (
	// sseSubscriberBuffer bounds per-connection event queue depth. Events
	// beyond this are dropped (logged at debug); the client resyncs via
	// normal /api/* fetches when reconnecting or focusing.
	sseSubscriberBuffer = 64
	// sseHeartbeatInterval keeps proxy connections alive. SSE spec allows
	// a comment line (starts with ":") as a no-op heartbeat.
	sseHeartbeatInterval = 15 * time.Second
)

func (h *handlers) handleEventsStream(w http.ResponseWriter, r *http.Request) {
	user, ok := auth.UserFromContext(r.Context())
	if !ok {
		http.Error(w, "unauthorized", http.StatusUnauthorized)
		return
	}

	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	// Standard SSE headers. X-Accel-Buffering disables nginx's response
	// buffering when Minstrel sits behind it; harmless on direct
	// connections.
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")
	w.Header().Set("X-Accel-Buffering", "no")
	w.WriteHeader(http.StatusOK)

	// Initial comment so the client sees the connection open immediately
	// even if no real event lands for a while.
	if _, err := fmt.Fprintf(w, ": connected\n\n"); err != nil {
		return
	}
	flusher.Flush()

	ch, unsub := h.eventbus.Subscribe(sseSubscriberBuffer)
	defer unsub()

	heartbeat := time.NewTicker(sseHeartbeatInterval)
	defer heartbeat.Stop()

	userIDStr := userUUIDString(user.ID)
	ctx := r.Context()

	for {
		select {
		case <-ctx.Done():
			return

		case <-heartbeat.C:
			if _, err := fmt.Fprintf(w, ": heartbeat\n\n"); err != nil {
				return
			}
			flusher.Flush()

		case e, ok := <-ch:
			if !ok {
				// The bus closed at shutdown. Ending here lets the
				// server drain; the client reconnects to the next one.
				return
			}
			// Scope filter: forward only events scoped to this user or
			// broadcast events (empty UserID).
			if e.UserID != "" && e.UserID != userIDStr {
				continue
			}
			if err := writeSSEEvent(w, e); err != nil {
				return
			}
			flusher.Flush()
		}
	}
}

// writeSSEEvent serializes one bus event in SSE wire format. JSON
// marshal errors collapse the event to an "error" kind so the client
// at least sees something rather than silent drops.
func writeSSEEvent(w http.ResponseWriter, e eventbus.Event) error {
	payload, err := json.Marshal(e)
	if err != nil {
		_, werr := fmt.Fprintf(w, "event: error\ndata: {\"msg\":\"marshal failed\"}\n\n")
		return werr
	}
	_, err = fmt.Fprintf(w, "event: %s\ndata: %s\n\n", e.Kind, payload)
	return err
}