diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index 5bc0d4d6..f0edb417 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -152,6 +152,10 @@ func run() error { // Constructed before any service that publishes so they all share the // same instance. bus := eventbus.New() + // Scan-run lifecycle events use a package-level setter rather than + // threading the bus through RunScan + TryStartScan + the Scheduler + // + every test caller. Per-process singleton, set once at startup. + library.SetEventBus(bus) lidarrReconciler := lidarrrequests.NewReconciler(pool, lidarrCfg, logger.With("component", "lidarr"), bus) go lidarrReconciler.Run(ctx) diff --git a/internal/library/eventbus.go b/internal/library/eventbus.go new file mode 100644 index 00000000..6a1c805a --- /dev/null +++ b/internal/library/eventbus.go @@ -0,0 +1,81 @@ +package library + +import ( + "sync" + + "github.com/jackc/pgx/v5/pgtype" + + "git.fabledsword.com/bvandeusen/minstrel/internal/eventbus" +) + +// Package-level bus pointer for scan lifecycle events (#392). Set once +// at startup via SetEventBus; reads are nil-safe so test contexts that +// never call SetEventBus simply skip publishing. +// +// A package-global is the right shape here because: +// +// - There is exactly one bus per process (the same instance powers +// the SSE subscriber endpoint and every publisher across the codebase). +// - Threading it through RunScan + TryStartScan + the Scheduler + +// every test caller would touch ~10 call sites without changing +// behavior at the boundaries that don't care. +// - Other Go stdlib packages use the same pattern (log.SetDefault, +// etc.) for process-wide singletons. +var ( + busMu sync.RWMutex + bus *eventbus.Bus +) + +// SetEventBus wires the live-event bus into the library package so scan +// runs emit scan.run_started / scan.run_finished events. Call once during +// process startup; subsequent calls overwrite the previous bus. +func SetEventBus(b *eventbus.Bus) { + busMu.Lock() + defer busMu.Unlock() + bus = b +} + +// publishScanEvent broadcasts a scan lifecycle event. No-op when no bus +// has been set. Scan events are always broadcast (empty UserID) because +// only admins see the scan status card; non-admin clients' invalidation +// dispatcher will receive the event but won't have any provider keyed on +// scan status, so the receive is harmless. +func publishScanEvent(kind string, runID pgtype.UUID, data map[string]any) { + busMu.RLock() + b := bus + busMu.RUnlock() + if b == nil { + return + } + payload := map[string]any{"run_id": uuidToBusString(runID)} + for k, v := range data { + payload[k] = v + } + b.Publish(eventbus.Event{ + Kind: kind, + UserID: "", // broadcast + Data: payload, + }) +} + +// uuidToBusString renders a pgtype.UUID in canonical 8-4-4-4-12 form. +// Inlined to avoid a back-edge dependency on internal/api or another +// helper-bearing package. +func uuidToBusString(u pgtype.UUID) string { + if !u.Valid { + return "" + } + const hex = "0123456789abcdef" + out := make([]byte, 36) + pos := 0 + for i, b := range u.Bytes { + if i == 4 || i == 6 || i == 8 || i == 10 { + out[pos] = '-' + pos++ + } + out[pos] = hex[b>>4] + out[pos+1] = hex[b&0xf] + pos += 2 + } + return string(out) +} diff --git a/internal/library/scanrun.go b/internal/library/scanrun.go index 97e13a7f..4b8a5db3 100644 --- a/internal/library/scanrun.go +++ b/internal/library/scanrun.go @@ -83,6 +83,7 @@ func RunScan( return pgtype.UUID{}, fmt.Errorf("insert scan_run: %w", err) } logger.Info("scan run started", "id", row.ID) + publishScanEvent("scan.run_started", row.ID, nil) var firstErr error captureErr := func(stage string, stageErr error) { @@ -228,5 +229,8 @@ func RunScan( } logger.Info("scan run complete", "id", row.ID, "error", errMsg) + publishScanEvent("scan.run_finished", row.ID, map[string]any{ + "error_message": errMsg, + }) return row.ID, firstErr }