diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index 11a0e375..5bc0d4d6 100644 --- a/cmd/minstrel/main.go +++ b/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, diff --git a/internal/lidarrrequests/reconciler.go b/internal/lidarrrequests/reconciler.go index 9b4ba12e..62ad1d75 100644 --- a/internal/lidarrrequests/reconciler.go +++ b/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. -func NewReconciler(pool *pgxpool.Pool, cfg *lidarrconfig.Service, logger *slog.Logger) *Reconciler { +// 5-minute tick, batch size 50. Pass nil for bus when no SSE publishing +// 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 { diff --git a/internal/lidarrrequests/reconciler_integration_test.go b/internal/lidarrrequests/reconciler_integration_test.go index ff13471f..28bbd50a 100644 --- a/internal/lidarrrequests/reconciler_integration_test.go +++ b/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) } diff --git a/internal/server/server.go b/internal/server/server.go index a4e3e17a..8a49630a 100644 --- a/internal/server/server.go +++ b/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. - // will publish into the same instance. Slice 1 ships with the - // subscriber endpoint (/api/events/stream) only — producers wire up - // in later slices. - bus := eventbus.New() + // producers in playevents / lidarrrequests / scanner publish into + // the same instance. Background workers (lidarr reconciler, scan + // scheduler) live in cmd/minstrel/main.go where they're constructed + // before Router() runs; they read s.Bus directly so they share this + // 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