minstrel/internal/eventbus/eventbus.go

// 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{}
	closed      bool
}

// 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()
	if b.closed {
		// A subscriber arriving during shutdown ends at once, like the
		// ones Close already ended.
		b.mu.Unlock()
		close(ch)
		return ch, func() {}
	}
	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
}

// Close ends every subscription by closing its channel, and any later
// Subscribe gets a channel that is already closed. The server calls it as
// shutdown begins: an SSE stream otherwise runs until its client leaves,
// and holds the graceful shutdown open for its whole deadline. Publish
// after Close is a no-op. Safe to call more than once.
func (b *Bus) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.closed = true
	for ch := range b.subscribers {
		delete(b.subscribers, ch)
		close(ch)
	}
}

// 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)
}