diff --git a/internal/api/api.go b/internal/api/api.go index e4cde55f..32de80c8 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -14,6 +14,7 @@ import ( "git.fabledsword.com/bvandeusen/minstrel/internal/auth" "git.fabledsword.com/bvandeusen/minstrel/internal/config" "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" + "git.fabledsword.com/bvandeusen/minstrel/internal/eventbus" "git.fabledsword.com/bvandeusen/minstrel/internal/library" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrquarantine" @@ -27,7 +28,7 @@ import ( // Mount attaches /api/* handlers to r. Public endpoints (login) are outside // RequireUser; everything else is gated by the middleware. The events writer // is shared with the Subsonic mount so /rest/scrobble feeds the same store. -func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playevents.Writer, recCfg config.RecommendationConfig, lidarrCfg *lidarrconfig.Service, lidarrReqs *lidarrrequests.Service, lidarrQuar *lidarrquarantine.Service, tracksSvc *tracks.Service, playlistsSvc *playlists.Service, coverEnricher *coverart.Enricher, coverBackfillCap int, coverSettings *coverart.SettingsService, scanner *library.Scanner, scanCfg library.RunScanConfig, scheduler *library.Scheduler, dataDir string, sender mailer.Sender) { +func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playevents.Writer, recCfg config.RecommendationConfig, lidarrCfg *lidarrconfig.Service, lidarrReqs *lidarrrequests.Service, lidarrQuar *lidarrquarantine.Service, tracksSvc *tracks.Service, playlistsSvc *playlists.Service, coverEnricher *coverart.Enricher, coverBackfillCap int, coverSettings *coverart.SettingsService, scanner *library.Scanner, scanCfg library.RunScanConfig, scheduler *library.Scheduler, dataDir string, sender mailer.Sender, bus *eventbus.Bus) { rng := rand.New(rand.NewSource(rand.Int63())) h := &handlers{ pool: pool, logger: logger, events: events, recCfg: recCfg, @@ -45,6 +46,7 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev scheduler: scheduler, dataDir: dataDir, mailer: sender, + eventbus: bus, } r.Route("/api", func(api chi.Router) { @@ -79,6 +81,7 @@ func Mount(r chi.Router, pool *pgxpool.Pool, logger *slog.Logger, events *playev authed.Get("/discover/suggestions", h.handleListSuggestions) authed.Get("/home", h.handleGetHome) authed.Post("/events", h.handleEvents) + authed.Get("/events/stream", h.handleEventsStream) authed.Post("/likes/tracks/{id}", h.handleLikeTrack) authed.Delete("/likes/tracks/{id}", h.handleUnlikeTrack) authed.Post("/likes/albums/{id}", h.handleLikeAlbum) @@ -191,4 +194,5 @@ type handlers struct { scheduler *library.Scheduler dataDir string mailer mailer.Sender + eventbus *eventbus.Bus } diff --git a/internal/api/events_stream.go b/internal/api/events_stream.go new file mode 100644 index 00000000..ff58deb0 --- /dev/null +++ b/internal/api/events_stream.go @@ -0,0 +1,135 @@ +// 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: +// data: +// +// where follows "domain.action" naming and +// 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 { + // Bus closed our subscription (rare; only on full shutdown). + 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 +} diff --git a/internal/eventbus/eventbus.go b/internal/eventbus/eventbus.go new file mode 100644 index 00000000..1f3504e6 --- /dev/null +++ b/internal/eventbus/eventbus.go @@ -0,0 +1,96 @@ +// Package eventbus is an in-process publish/subscribe bus for real-time +// events the server emits to subscribed clients (today: Flutter via SSE). +// +// Writers (likes handler, requests reconciler, scanner, etc.) call Publish +// to announce a state change. Each connected SSE client holds a +// subscription via Subscribe and forwards events to its HTTP response. +// +// The bus is intentionally simple: +// +// - Per-subscriber bounded channel buffer. If a subscriber's buffer is +// full when an event arrives, the event is dropped FOR THAT SUBSCRIBER +// ONLY — writers never block, and other subscribers are unaffected. +// - No persistence. A client that connects after an event was emitted +// will not see that event. SSE clients are expected to fetch a fresh +// snapshot via the normal /api/* endpoints on (re)connect. +// - No cross-process broadcast. v1 runs as a single Go binary; a future +// multi-instance deployment will need a real broker (NATS / Redis +// pub-sub / etc.) but that's out of scope here. +package eventbus + +import ( + "sync" +) + +// Event is a single broadcast message. Kind names follow a "domain.action" +// convention (e.g. "track.liked", "request.status_changed"). UserID, when +// non-empty, scopes the event to a specific user — subscribers receive +// only events with UserID == their auth user OR with empty UserID (which +// means "broadcast to everyone subscribed"). +type Event struct { + Kind string `json:"kind"` + UserID string `json:"user_id,omitempty"` + Data map[string]any `json:"data"` +} + +// Bus is a fan-out broadcaster. Safe for concurrent use; the zero value is +// not usable — call New(). +type Bus struct { + mu sync.RWMutex + subscribers map[chan Event]struct{} +} + +// New returns a Bus with no subscribers. +func New() *Bus { + return &Bus{subscribers: map[chan Event]struct{}{}} +} + +// Publish fans an event out to every subscriber. Non-blocking per +// subscriber: if a subscriber's buffer is full, the event is dropped for +// that subscriber rather than stalling the writer. +func (b *Bus) Publish(e Event) { + b.mu.RLock() + defer b.mu.RUnlock() + for ch := range b.subscribers { + select { + case ch <- e: + default: + // Subscriber buffer full — drop. The subscriber will resync + // via a normal API fetch when the client decides to. + } + } +} + +// Subscribe registers a new subscriber and returns its receive channel +// plus an unsubscribe func. The buffer argument controls how many events +// can be queued before drops occur; 16–64 is a reasonable range for an +// SSE client that processes events promptly. +// +// Callers MUST invoke the returned unsubscribe func when done (typically +// in a defer), or the subscription leaks. +func (b *Bus) Subscribe(buffer int) (<-chan Event, func()) { + if buffer <= 0 { + buffer = 16 + } + ch := make(chan Event, buffer) + b.mu.Lock() + b.subscribers[ch] = struct{}{} + b.mu.Unlock() + unsub := func() { + b.mu.Lock() + if _, ok := b.subscribers[ch]; ok { + delete(b.subscribers, ch) + close(ch) + } + b.mu.Unlock() + } + return ch, unsub +} + +// SubscriberCount returns the current number of active subscribers. For +// tests + observability; not load-bearing in hot paths. +func (b *Bus) SubscriberCount() int { + b.mu.RLock() + defer b.mu.RUnlock() + return len(b.subscribers) +} diff --git a/internal/eventbus/eventbus_test.go b/internal/eventbus/eventbus_test.go new file mode 100644 index 00000000..65898b77 --- /dev/null +++ b/internal/eventbus/eventbus_test.go @@ -0,0 +1,150 @@ +package eventbus + +import ( + "sync" + "testing" + "time" +) + +func TestPublish_DeliversToSubscriber(t *testing.T) { + t.Parallel() + b := New() + ch, unsub := b.Subscribe(4) + defer unsub() + + b.Publish(Event{Kind: "test.event", UserID: "u1", Data: map[string]any{"x": 1}}) + + select { + case got := <-ch: + if got.Kind != "test.event" || got.UserID != "u1" { + t.Fatalf("unexpected event: %+v", got) + } + if got.Data["x"] != 1 { + t.Fatalf("payload mismatch: %+v", got.Data) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for event") + } +} + +func TestPublish_FanOutToMultipleSubscribers(t *testing.T) { + t.Parallel() + b := New() + ch1, unsub1 := b.Subscribe(4) + ch2, unsub2 := b.Subscribe(4) + defer unsub1() + defer unsub2() + + b.Publish(Event{Kind: "broadcast"}) + + for i, ch := range []<-chan Event{ch1, ch2} { + select { + case got := <-ch: + if got.Kind != "broadcast" { + t.Fatalf("ch%d: kind=%q", i, got.Kind) + } + case <-time.After(time.Second): + t.Fatalf("ch%d: timed out", i) + } + } +} + +func TestPublish_DropsWhenSubscriberBufferFull(t *testing.T) { + t.Parallel() + b := New() + _, unsub := b.Subscribe(2) // tiny buffer + defer unsub() + + // Without draining the channel, push more events than the buffer holds. + // Drops are silent; the assertion is that Publish does not block. + done := make(chan struct{}) + go func() { + for i := 0; i < 100; i++ { + b.Publish(Event{Kind: "flood"}) + } + close(done) + }() + + select { + case <-done: + // Good — Publish returned without blocking on the laggard. + case <-time.After(2 * time.Second): + t.Fatal("Publish blocked on a slow subscriber") + } +} + +func TestUnsubscribe_RemovesFromBusAndClosesChannel(t *testing.T) { + t.Parallel() + b := New() + ch, unsub := b.Subscribe(4) + + if got := b.SubscriberCount(); got != 1 { + t.Fatalf("after Subscribe: SubscriberCount=%d, want 1", got) + } + + unsub() + + if got := b.SubscriberCount(); got != 0 { + t.Fatalf("after unsub: SubscriberCount=%d, want 0", got) + } + + // Channel is closed: receive returns zero value with ok=false. + if _, ok := <-ch; ok { + t.Fatal("expected channel to be closed after unsubscribe") + } +} + +func TestUnsubscribe_Idempotent(t *testing.T) { + t.Parallel() + b := New() + _, unsub := b.Subscribe(4) + unsub() + unsub() // must not panic / double-close + if got := b.SubscriberCount(); got != 0 { + t.Fatalf("SubscriberCount=%d, want 0", got) + } +} + +func TestPublish_ConcurrentWritersAndSubscribers(t *testing.T) { + t.Parallel() + b := New() + + const ( + subs = 8 + perWriter = 50 + writers = 4 + bufferSize = 100 + ) + + var subWG sync.WaitGroup + subWG.Add(subs) + stops := make([]func(), 0, subs) + for i := 0; i < subs; i++ { + ch, unsub := b.Subscribe(bufferSize) + stops = append(stops, unsub) + go func() { + defer subWG.Done() + for range ch { + // Drain until channel closes. + } + }() + } + + var pubWG sync.WaitGroup + pubWG.Add(writers) + for w := 0; w < writers; w++ { + go func() { + defer pubWG.Done() + for i := 0; i < perWriter; i++ { + b.Publish(Event{Kind: "concurrent"}) + } + }() + } + pubWG.Wait() + + // Tear down subscribers — channels close, goroutines exit. + for _, stop := range stops { + stop() + } + subWG.Wait() +} diff --git a/internal/server/server.go b/internal/server/server.go index 699418ac..a4e3e17a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -18,6 +18,7 @@ import ( "git.fabledsword.com/bvandeusen/minstrel/internal/auth" "git.fabledsword.com/bvandeusen/minstrel/internal/config" "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" + "git.fabledsword.com/bvandeusen/minstrel/internal/eventbus" "git.fabledsword.com/bvandeusen/minstrel/internal/library" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarr" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig" @@ -117,7 +118,13 @@ func (s *Server) Router() http.Handler { tracksSvc := tracks.NewService(s.Pool, s.Logger, lidarrUnmonitorAdapter{fn: lidarrClientFn}, s.DataDir) playlistsSvc := playlists.NewService(s.Pool, s.Logger, s.DataDir) smtpSender := mailer.NewSMTPSender(s.Pool, s.Logger.With("component", "mailer")) - api.Mount(r, s.Pool, s.Logger, writer, s.RecommendationCfg, lidarrCfg, lidarrReqs, lidarrQuar, tracksSvc, playlistsSvc, s.CoverEnricher, s.CoverArtBackfillCap, s.CoverSettings, s.LibraryScanner, s.ScanCfg, s.Scheduler, s.DataDir, smtpSender) + // Live-event bus for SSE subscribers (#392). Constructed per-process; + // future producers in playevents / lidarrrequests / scanner / etc. + // will publish into the same instance. Slice 1 ships with the + // subscriber endpoint (/api/events/stream) only — producers wire up + // in later slices. + bus := eventbus.New() + api.Mount(r, s.Pool, s.Logger, writer, s.RecommendationCfg, lidarrCfg, lidarrReqs, lidarrQuar, tracksSvc, playlistsSvc, s.CoverEnricher, s.CoverArtBackfillCap, s.CoverSettings, s.LibraryScanner, s.ScanCfg, s.Scheduler, s.DataDir, smtpSender, bus) // /api/admin/scan is the only admin route owned by the server package // (it needs the Scanner). Register it as a single inline-middleware // route — using r.Route("/api/admin", ...) here would create a second