feat(#392): SSE event stream foundation — eventbus + /api/events/stream

Slice 1 of the #392 hybrid live-refresh work. Ships the in-process pub/sub bus and the SSE subscriber endpoint; no producers wired yet, so the stream emits only heartbeats today. Verifiable in isolation by curl-ing the endpoint with a valid Bearer token — the connection opens, ": heartbeat" lines arrive every 15s, the connection closes cleanly on client disconnect. eventbus.Bus is a small fan-out broadcaster: subscribers register through Subscribe (returns a receive channel + an unsubscribe closure), writers call Publish, and the bus drops events for any subscriber whose buffer is full rather than blocking the writer. No persistence — clients are expected to resync via normal /api/* fetches on (re)connect. The SSE handler emits an initial ": connected" comment so the client sees the connection open immediately, then forwards events whose UserID matches the authenticated user (or is empty for broadcast). Heartbeat comments keep proxy connections alive. Context cancellation cleanly tears down the subscription on client disconnect. Producers (likes, request status, quarantine, scanner, playlist mutations) land in subsequent slices. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 20:44:37 -04:00
parent b466b6494a
commit 170614baf1
5 changed files with 394 additions and 2 deletions
@@ -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)
+}
@@ -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()
+}