memby/server/internal/adminevents/adminevents.go

// Package adminevents is the gateway's administrative event bus: one place a backend
// service says "this happened" and every reader — the console's notification bell, an
// outgoing Discord webhook, anything added later — hears about it without the publisher
// knowing they exist.
//
// The point of it being generic is that authentication code must not know about Discord.
// handleLogin publishes a sign-in; it does not format a webhook payload, decide whether
// the operator wants one, or handle the failure when the webhook is down. That separation
// is what lets a second integration be a file that subscribes rather than an edit to
// every publisher.
package adminevents

import (
	"context"
	"encoding/json"
	"log/slog"
	"sync"
	"time"

	"github.com/ponzischeme89/memby/server/internal/store"
)

// Event is one administrative occurrence. It is store.AdminEvent so that what is
// broadcast is exactly what was stored, ids and timestamps included — a subscriber shown
// a different shape from the one the console later reads back is a subscriber whose view
// can disagree with the feed.
type Event = store.AdminEvent

// Event types. These are routing keys: an integration selects the ones it wants and the
// console groups by them, so they are stable strings rather than an enum somebody would
// be tempted to renumber.
//
// Adding one is a constant here and a call to Publish. Nothing has to be taught about it:
// the console's type filter is built from what has actually been published, and an
// integration configured to send "everything" sends it.
const (
	TypeLogin                = "auth.login"
	TypeLoginFailed          = "auth.login_failed"
	TypeLogout               = "auth.logout"
	TypeDeviceRegistered     = "device.registered"
	TypeDeviceRemoved        = "device.removed"
	TypeDeviceRenamed        = "device.renamed"
	TypeAdminSignIn          = "admin.sign_in"
	TypeServerStarted        = "server.started"
	TypeMaintenanceChanged   = "server.maintenance"
	TypeTaskCompleted        = "task.completed"
	TypeTaskFailed           = "task.failed"
	TypeIntegrationFailed    = "integration.failed"
	TypeLibrarySync          = "library.sync"
	TypeEmbyUnreachable      = "emby.unreachable"
	TypeEmbyRecovered        = "emby.recovered"
	TypeMediaReportCreated   = "media.report.created"
	TypeReplacementRequested = "media.replacement.requested"
	TypeReplacementFailed    = "media.replacement.failed"
	TypeMediaReportResolved  = "media.report.resolved"
)

// Severity re-exports the store's vocabulary so a publisher needs only this package.
const (
	SeverityInfo    = store.SeverityInfo
	SeverityWarning = store.SeverityWarning
	SeverityError   = store.SeverityError
)

// Sink is a synchronous reader — an integration dispatcher, typically. Sinks are called
// on the bus's own goroutine after the event is persisted, and are expected to return
// promptly: anything that talks to a network does its own queueing behind this.
type Sink interface {
	DeliverAdminEvent(ctx context.Context, event Event)
}

// subscriberBuffer is how far behind a live reader may fall before it starts losing
// events. A browser tab on a slow connection must not be able to hold up a sign-in, so
// the drop is deliberate — and it is recoverable, because the console's stream reconnects
// by asking for everything after the last id it saw.
const subscriberBuffer = 64

type subscriber struct {
	events chan Event
}

// Bus is the publisher. It is safe for concurrent use and every one of its operations is
// best-effort: an event that cannot be stored is logged and dropped, never returned as an
// error to the thing that was merely reporting it.
type Bus struct {
	store *store.Store
	log   *slog.Logger

	mu          sync.RWMutex
	subscribers map[*subscriber]struct{}
	sinks       []Sink
}

func New(st *store.Store, log *slog.Logger) *Bus {
	return &Bus{store: st, log: log, subscribers: map[*subscriber]struct{}{}}
}

// AddSink registers a reader that wants every event. Called during start-up wiring,
// before anything can publish.
func (b *Bus) AddSink(sink Sink) {
	if b == nil || sink == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.sinks = append(b.sinks, sink)
}

// Publish stores the event and hands it to every reader.
//
// The context is detached from the caller's before the write. An event is a *record of
// something that already happened*, and the commonest publisher is a request handler —
// so hanging the write off the request context would lose exactly the events belonging to
// a client that disconnected, which are frequently the interesting ones. The deadline
// below is what stops that detachment turning into an unbounded write.
func (b *Bus) Publish(ctx context.Context, event Event) Event {
	if b == nil {
		return event
	}
	if event.Severity == "" {
		event.Severity = SeverityInfo
	}
	if b.store != nil {
		writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
		stored, err := b.store.RecordAdminEvent(writeCtx, event)
		cancel()
		if err != nil {
			b.log.Warn("admin event not recorded", "type", event.Type, "error", err)
			// Still broadcast it. A bell that goes quiet because Postgres hiccuped is
			// worse than one showing an event that will not survive a refresh.
			if event.OccurredAt.IsZero() {
				event.OccurredAt = time.Now()
			}
		} else {
			event = stored
		}
	} else if event.OccurredAt.IsZero() {
		event.OccurredAt = time.Now()
	}

	b.mu.RLock()
	sinks := b.sinks
	for sub := range b.subscribers {
		select {
		case sub.events <- event:
		default:
			// Behind. See subscriberBuffer: the reader recovers by id on reconnect.
		}
	}
	b.mu.RUnlock()

	for _, sink := range sinks {
		sink.DeliverAdminEvent(context.WithoutCancel(ctx), event)
	}
	return event
}

// Subscribe opens a live feed. The returned function must be called when the reader goes
// away, or the bus keeps writing into a channel nobody reads and the subscriber leaks for
// the life of the process.
func (b *Bus) Subscribe() (<-chan Event, func()) {
	if b == nil {
		closed := make(chan Event)
		close(closed)
		return closed, func() {}
	}
	sub := &subscriber{events: make(chan Event, subscriberBuffer)}
	b.mu.Lock()
	b.subscribers[sub] = struct{}{}
	b.mu.Unlock()
	return sub.events, func() {
		b.mu.Lock()
		delete(b.subscribers, sub)
		b.mu.Unlock()
	}
}

// Subscribers reports how many live readers are attached. The console shows it, which is
// the cheapest way to tell "the stream is not delivering" from "nothing has happened".
func (b *Bus) Subscribers() int {
	if b == nil {
		return 0
	}
	b.mu.RLock()
	defer b.mu.RUnlock()
	return len(b.subscribers)
}

// Meta builds a metadata document, dropping anything that will not encode rather than
// failing the publish. Metadata is by definition the part no reader depends on.
func Meta(pairs map[string]any) json.RawMessage {
	if len(pairs) == 0 {
		return nil
	}
	encoded, err := json.Marshal(pairs)
	if err != nil {
		return nil
	}
	return encoded
}