feat(#392): lidarr reconciler publishes request.status_changed on complete

Slice 3b — extends event publishing to background workers.

When the reconciler matches an approved Lidarr request against the
library and flips it to 'completed', it now publishes a
request.status_changed event scoped to the original requester so their
/requests page invalidates without polling. Three call sites — one per
kind (artist / album / track) — capture the completed row instead of
discarding it so the publish has the user_id.

Bus plumbing: the reconciler runs in cmd/minstrel/main.go which
constructs services before server.New is called. Moved bus
construction into main; the Server struct gained a Bus field that
Router() reads, falling back to a fresh local instance when nil
(test contexts). Result: one bus per process shared by every
publisher and the SSE subscriber endpoint.

reconciler.go inlines a uuid->string helper rather than reaching into
internal/api for one — avoids a back-edge dependency.

Test compat: 9 NewReconciler call sites in reconciler_integration_test.go
get `nil` for the new bus param. The reconciler's publishCompleted helper
is a no-op when the bus is nil so tests that don't care about events
keep passing.

Scanner producer (scan.progress) deferred to slice 3c — needs more
thought about which lifecycle transitions warrant events vs. flood the
stream.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

cmd/minstrel/main.go

@@ -14,6 +14,7 @@ import (
 	"git.fabledsword.com/bvandeusen/minstrel/internal/config"
 	"git.fabledsword.com/bvandeusen/minstrel/internal/coverart"
 	"git.fabledsword.com/bvandeusen/minstrel/internal/db"
+	"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/lidarrrequests"
@@ -146,7 +147,12 @@ func run() error {
 	// import requests and reconciles them against the library. Short-circuits
 	// to no-op when lidarr_config.enabled = false.
 	lidarrCfg := lidarrconfig.New(pool)
-	lidarrReconciler := lidarrrequests.NewReconciler(pool, lidarrCfg, logger.With("component", "lidarr"))
+	// Live-event bus shared between SSE subscribers (api.Mount) and
+	// background workers that publish (reconciler today; scanner later).
+	// Constructed before any service that publishes so they all share the
+	// same instance.
+	bus := eventbus.New()
+	lidarrReconciler := lidarrrequests.NewReconciler(pool, lidarrCfg, logger.With("component", "lidarr"), bus)
 	go lidarrReconciler.Run(ctx)
 
 	// Ensure DataDir exists before any service tries to write into it.
@@ -173,6 +179,7 @@ func run() error {
 	srv := server.New(logger, pool, scanner, subsonic.Config{
 		AllowPlaintextPassword: cfg.Subsonic.AllowPlaintextPassword,
 	}, cfg.Events, cfg.Recommendation, cfg.Storage.DataDir, cfg.Branding, coverEnricher, cfg.Library.CoverArtBackfillCap, coverSettings, scanner, scanCfg, scheduler)
+	srv.Bus = bus
 	playlists.StartSystemPlaylistCron(ctx, pool, logger.With("component", "system_playlist_cron"), cfg.Storage.DataDir)
 	httpServer := &http.Server{
 		Addr:              cfg.Server.Address,

internal/lidarrrequests/reconciler.go

@@ -10,6 +10,7 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
+	"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
 	"git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig"
 )
 
@@ -21,22 +22,81 @@ type Reconciler struct {
 	pool      *pgxpool.Pool
 	lidarrCfg *lidarrconfig.Service
 	logger    *slog.Logger
+	bus       *eventbus.Bus
 	tick      time.Duration
 	batch     int32
 }
 
 // NewReconciler constructs a Reconciler with production defaults:
-// 5-minute tick, batch size 50.
+// 5-minute tick, batch size 50. Pass nil for bus when no SSE publishing
-func NewReconciler(pool *pgxpool.Pool, cfg *lidarrconfig.Service, logger *slog.Logger) *Reconciler {
+// is desired (tests, or environments without the event stream enabled).
+func NewReconciler(pool *pgxpool.Pool, cfg *lidarrconfig.Service, logger *slog.Logger, bus *eventbus.Bus) *Reconciler {
 	return &Reconciler{
 		pool:      pool,
 		lidarrCfg: cfg,
 		logger:    logger,
+		bus:       bus,
 		tick:      5 * time.Minute,
 		batch:     50,
 	}
 }
 
+// publishCompleted broadcasts a request.status_changed event scoped to
+// the original requester. No-op when bus is nil.
+func (r *Reconciler) publishCompleted(row dbq.LidarrRequest) {
+	if r.bus == nil {
+		return
+	}
+	r.bus.Publish(eventbus.Event{
+		Kind:   "request.status_changed",
+		UserID: formatUUIDForBus(row.UserID),
+		Data: map[string]any{
+			"request_id": formatUUIDForBus(row.ID),
+			"status":     "completed",
+			"kind":       string(row.Kind),
+		},
+	})
+}
+
+// formatUUIDForBus renders a pgtype.UUID as the canonical 8-4-4-4-12 hex
+// string. Mirrors helpers in internal/api/convert.go and other packages;
+// inlined here to avoid a back-edge dependency from lidarrrequests onto
+// the api layer.
+func formatUUIDForBus(u pgtype.UUID) string {
+	if !u.Valid {
+		return ""
+	}
+	return string([]byte{
+		hex(u.Bytes[0]>>4), hex(u.Bytes[0]&0xf),
+		hex(u.Bytes[1]>>4), hex(u.Bytes[1]&0xf),
+		hex(u.Bytes[2]>>4), hex(u.Bytes[2]&0xf),
+		hex(u.Bytes[3]>>4), hex(u.Bytes[3]&0xf),
+		'-',
+		hex(u.Bytes[4]>>4), hex(u.Bytes[4]&0xf),
+		hex(u.Bytes[5]>>4), hex(u.Bytes[5]&0xf),
+		'-',
+		hex(u.Bytes[6]>>4), hex(u.Bytes[6]&0xf),
+		hex(u.Bytes[7]>>4), hex(u.Bytes[7]&0xf),
+		'-',
+		hex(u.Bytes[8]>>4), hex(u.Bytes[8]&0xf),
+		hex(u.Bytes[9]>>4), hex(u.Bytes[9]&0xf),
+		'-',
+		hex(u.Bytes[10]>>4), hex(u.Bytes[10]&0xf),
+		hex(u.Bytes[11]>>4), hex(u.Bytes[11]&0xf),
+		hex(u.Bytes[12]>>4), hex(u.Bytes[12]&0xf),
+		hex(u.Bytes[13]>>4), hex(u.Bytes[13]&0xf),
+		hex(u.Bytes[14]>>4), hex(u.Bytes[14]&0xf),
+		hex(u.Bytes[15]>>4), hex(u.Bytes[15]&0xf),
+	})
+}
+
+func hex(b byte) byte {
+	if b < 10 {
+		return '0' + b
+	}
+	return 'a' + (b - 10)
+}
+
 // Run blocks until ctx is cancelled, ticking every r.tick. Errors from
 // tickOnce are logged at WARN and never propagated.
 func (r *Reconciler) Run(ctx context.Context) {
@@ -114,13 +174,17 @@ func (r *Reconciler) reconcileArtist(ctx context.Context, q *dbq.Queries, row db
 		}
 		return err
 	}
-	_, err = q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
+	completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
 		ID:              row.ID,
 		MatchedArtistID: artistID,
 		MatchedAlbumID:  pgtype.UUID{},
 		MatchedTrackID:  pgtype.UUID{},
 	})
-	return err
+	if err != nil {
+		return err
+	}
+	r.publishCompleted(completed)
+	return nil
 }
 
 func (r *Reconciler) reconcileAlbum(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error {
@@ -138,13 +202,17 @@ func (r *Reconciler) reconcileAlbum(ctx context.Context, q *dbq.Queries, row dbq
 		}
 		return err
 	}
-	_, err = q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
+	completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
 		ID:              row.ID,
 		MatchedAlbumID:  albumID,
 		MatchedArtistID: pgtype.UUID{},
 		MatchedTrackID:  pgtype.UUID{},
 	})
-	return err
+	if err != nil {
+		return err
+	}
+	r.publishCompleted(completed)
+	return nil
 }
 
 func (r *Reconciler) reconcileTrack(ctx context.Context, q *dbq.Queries, row dbq.LidarrRequest) error {
@@ -178,13 +246,17 @@ func (r *Reconciler) reconcileTrack(ctx context.Context, q *dbq.Queries, row dbq
 		return err
 	}
 
-	_, err = q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
+	completed, err := q.CompleteLidarrRequest(ctx, dbq.CompleteLidarrRequestParams{
 		ID:              row.ID,
 		MatchedAlbumID:  albumID,
 		MatchedTrackID:  trackID,
 		MatchedArtistID: pgtype.UUID{},
 	})
-	return err
+	if err != nil {
+		return err
+	}
+	r.publishCompleted(completed)
+	return nil
 }
 
 func isNoRows(err error) bool {

internal/lidarrrequests/reconciler_integration_test.go

@@ -121,7 +121,7 @@ func TestReconciler_MatchesArtistByMBID(t *testing.T) {
 		Kind: "artist", LidarrArtistMBID: artistMBID, ArtistName: "Boards of Canada",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -158,7 +158,7 @@ func TestReconciler_MatchesAlbumByMBID(t *testing.T) {
 		LidarrAlbumMBID: albumMBID, AlbumTitle: "Test Album",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -200,7 +200,7 @@ func TestReconciler_MatchesTrackViaAlbumMBID(t *testing.T) {
 		LidarrTrackMBID: trackMBID, TrackTitle: "Track One",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -234,7 +234,7 @@ func TestReconciler_NoMatchLeavesPending(t *testing.T) {
 		Kind: "artist", LidarrArtistMBID: "nonexistent-mbid-xyz", ArtistName: "Ghost Artist",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -278,7 +278,7 @@ func TestReconciler_AlreadyCompletedRowNotReprocessed(t *testing.T) {
 		t.Fatalf("CompleteLidarrRequest setup: %v", err)
 	}
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -313,7 +313,7 @@ func TestReconciler_DisabledIsNoOp(t *testing.T) {
 		Kind: "artist", LidarrArtistMBID: artistMBID, ArtistName: "No-Op Artist",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -345,7 +345,7 @@ func TestReconciler_RunRunsAtLeastOneTick(t *testing.T) {
 		Kind: "artist", LidarrArtistMBID: mbid, ArtistName: "Run Artist",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	rec.tick = 25 * time.Millisecond
 
 	done := make(chan struct{})
@@ -399,7 +399,7 @@ func TestReconciler_TrackKindNoTracksInAlbumYet(t *testing.T) {
 		LidarrTrackMBID: "t-mbid-not-yet", TrackTitle: "Some Track",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}
@@ -428,7 +428,7 @@ func TestReconciler_AlbumKindNoMatchInAlbumsTable(t *testing.T) {
 		LidarrAlbumMBID: "al-not-in-library", AlbumTitle: "Nope Album",
 	})
 
-	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger())
+	rec := NewReconciler(pool, lidarrconfig.New(pool), newTestLogger(), nil)
 	if err := rec.tickOnce(ctx); err != nil {
 		t.Fatalf("tickOnce: %v", err)
 	}

internal/server/server.go

@@ -80,6 +80,10 @@ type Server struct {
 	LibraryScanner      *library.Scanner
 	ScanCfg             library.RunScanConfig
 	Scheduler           *library.Scheduler
+	// Bus is the live-event bus shared with background workers (the
+	// lidarr reconciler, scan scheduler) constructed in cmd/minstrel/main.go.
+	// When nil, Router() constructs a local fallback (test contexts).
+	Bus *eventbus.Bus
 }
 
 func New(logger *slog.Logger, pool *pgxpool.Pool, scanner ScanTrigger, subCfg subsonic.Config, eventsCfg config.EventsConfig, recCfg config.RecommendationConfig, dataDir string, brandingCfg config.BrandingConfig, coverEnricher *coverart.Enricher, coverArtBackfillCap int, coverSettings *coverart.SettingsService, libraryScanner *library.Scanner, scanCfg library.RunScanConfig, scheduler *library.Scheduler) *Server {
@@ -119,11 +123,16 @@ func (s *Server) Router() http.Handler {
 		playlistsSvc := playlists.NewService(s.Pool, s.Logger, s.DataDir)
 		smtpSender := mailer.NewSMTPSender(s.Pool, s.Logger.With("component", "mailer"))
 		// Live-event bus for SSE subscribers (#392). Constructed per-process;
-		// future producers in playevents / lidarrrequests / scanner / etc.
+		// producers in playevents / lidarrrequests / scanner publish into
-		// will publish into the same instance. Slice 1 ships with the
+		// the same instance. Background workers (lidarr reconciler, scan
-		// subscriber endpoint (/api/events/stream) only — producers wire up
+		// scheduler) live in cmd/minstrel/main.go where they're constructed
-		// in later slices.
+		// before Router() runs; they read s.Bus directly so they share this
-		bus := eventbus.New()
+		// process's bus. When Router() runs before main set the bus (tests,
+		// or future contexts) we fall back to a fresh local instance.
+		bus := s.Bus
+		if bus == nil {
+			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