package main import ( "context" "errors" "flag" "fmt" "log/slog" "net/http" "os" "os/signal" "syscall" "time" "github.com/jackc/pgx/v5/pgxpool" "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/gc" "git.fabledsword.com/bvandeusen/minstrel/internal/library" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarr" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig" "git.fabledsword.com/bvandeusen/minstrel/internal/lidarrrequests" "git.fabledsword.com/bvandeusen/minstrel/internal/logging" "git.fabledsword.com/bvandeusen/minstrel/internal/playlists" "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble" "git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz" "git.fabledsword.com/bvandeusen/minstrel/internal/server" "git.fabledsword.com/bvandeusen/minstrel/internal/similarity" "git.fabledsword.com/bvandeusen/minstrel/internal/subsonic" syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" ) func main() { if len(os.Args) > 1 { switch os.Args[1] { case "admin": if err := runAdmin(os.Args[2:]); err != nil { fmt.Fprintf(os.Stderr, "minstrel admin: %v\n", err) os.Exit(1) } return case "migrate": if err := runMigrate(os.Args[2:]); err != nil { fmt.Fprintf(os.Stderr, "minstrel migrate: %v\n", err) os.Exit(1) } return } } if err := run(); err != nil { fmt.Fprintf(os.Stderr, "minstrel: %v\n", err) os.Exit(1) } } // safetyNetScanInterval backstops the fsnotify watcher with a low-frequency // full delta walk, catching anything the watcher missed (inotify watch-limit // exhaustion, dropped events). Fixed — there is no operator configuration. const safetyNetScanInterval = 12 * time.Hour // runSafetyNetScans ticks a delta RunScan at safetyNetScanInterval until ctx // is cancelled, deferring to TryStartScan's in-flight guard so it never // collides with a manual, startup, or watcher-driven scan. func runSafetyNetScans(ctx context.Context, pool *pgxpool.Pool, scanner *library.Scanner, enricher *coverart.Enricher, logger *slog.Logger, scanCfg library.RunScanConfig, ) { ticker := time.NewTicker(safetyNetScanInterval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: started, existing, err := library.TryStartScan(ctx, pool, scanner, enricher, logger, scanCfg) if err != nil { logger.Error("safety-net scan: try start failed", "err", err) } else if !started && existing != nil { logger.Info("safety-net scan skipped — prior still in flight", "in_flight_id", existing.ID) } } } } func run() error { configPath := flag.String("config", os.Getenv("MINSTREL_CONFIG"), "path to YAML config file") flag.Parse() cfg, err := config.Load(*configPath) if err != nil { return fmt.Errorf("load config: %w", err) } logger, err := logging.New(os.Stdout, cfg.Log.Level, cfg.Log.Format) if err != nil { return fmt.Errorf("init logger: %w", err) } logger.Info("minstrel starting", "address", cfg.Server.Address, "log_level", cfg.Log.Level, "log_format", cfg.Log.Format, ) ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() if err := db.Migrate(cfg.Database.URL, logger); err != nil { return fmt.Errorf("migrate: %w", err) } pool, err := db.Open(ctx, cfg.Database.URL) if err != nil { return fmt.Errorf("open db: %w", err) } defer pool.Close() scanner := library.New(pool, logger, cfg.Library.ScanPaths) contact := cfg.Library.ContactEmail if contact == "" { contact = "https://git.fabledsword.com/bvandeusen/minstrel" } // Swap in the production User-Agent + MinPeriod on the registered // MBCAA provider. init() registered it with default config; this // updates the running instance. coverart.NewMBCAAProviderFromConfig(coverart.FetcherConfig{ UserAgent: fmt.Sprintf("Minstrel/dev (%s)", contact), MinPeriod: time.Second, }) coverSettings, err := coverart.NewSettingsService(ctx, pool, logger.With("component", "coverart")) if err != nil { logger.Error("coverart settings service init failed", "err", err) os.Exit(1) } if newVer, bumped, berr := coverSettings.BumpVersionIfProvidersChanged(ctx); berr != nil { logger.Warn("coverart: provider-hash boot check failed", "err", berr) } else if bumped { logger.Info("coverart: registered provider set changed; version bumped", "new_version", newVer) } coverEnricher := coverart.NewEnricher(pool, logger.With("component", "coverart"), coverSettings) coverEnricher.DataDir = cfg.Storage.DataDir // One unified scan chain: library walk → MBID backfill → cover enrich. // Boot-time scan and manual-trigger scans share this path; results land // in scan_runs for the admin overview. if cfg.Library.ScanOnStartup && len(cfg.Library.ScanPaths) > 0 { go func() { if _, err := library.RunScan(ctx, pool, scanner, coverEnricher, logger.With("component", "scan_run"), library.RunScanConfig{ BackfillCap: 5000, EnrichCap: -1, // #388: no global cap; per-provider rate limits ArtistEnrichCap: -1, DataDir: cfg.Storage.DataDir, }, ); err != nil { logger.Warn("startup scan run failed", "err", err) } }() } else { // No startup file walk, but still run backfill + enrich so cover // progress doesn't stall just because the operator disabled // scan-on-startup. go func() { if _, err := library.RunScan(ctx, pool, nil, coverEnricher, logger.With("component", "scan_run"), library.RunScanConfig{ BackfillCap: 5000, EnrichCap: -1, // #388: no global cap; per-provider rate limits ArtistEnrichCap: -1, DataDir: cfg.Storage.DataDir, }, ); err != nil { logger.Warn("boot-only scan run failed", "err", err) } }() } // Start the ListenBrainz scrobble worker. Per spec §M4a, runs every 30s // and drains up to 50 pending rows per tick. Per-user gating happens // inside the worker (rows from disabled users are skipped). scrobbleWorker := scrobble.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "scrobble")) go scrobbleWorker.Run(ctx) // Start the similarity ingest worker. Per spec §M4b, runs every 1h, drains // up to 5 played tracks + 5 played artists per tick, weekly re-fetch cap // per row. Public LB endpoints — no token required. similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity")) go similarityWorker.Run(ctx) // Start the GC worker. Runs every 1h and sweeps lifecycle tables // that have no writer-side close path or retention policy: // orphan play_events, stale play_sessions, expired // scrobble_queue failures, stuck system_playlist_runs, expired // password_resets. Each sweep is idempotent — a row that's // already clean is a no-op. Addresses drift audit findings // #565 #566 #567 #574 #575 (Scribe parent #552). gcWorker := gc.NewWorker(pool, logger.With("component", "gc")) go gcWorker.Run(ctx) // Start the Lidarr reconciler worker. Per spec §M5a, polls pending Lidarr // import requests and reconciles them against the library. Short-circuits // to no-op when lidarr_config.enabled = false. lidarrCfg := lidarrconfig.New(pool) // 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() // 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) // Per-tick Lidarr client factory: re-reads config so an admin save // takes effect without a restart, returning nil while Lidarr is // disabled/unconfigured. Mirrors server.go's lidarrClientFn; the // reconciler uses it to (re)send unconfirmed adds. lidarrClientFn := func() *lidarr.Client { c, cerr := lidarrCfg.Get(ctx) if cerr != nil || !c.Enabled || c.BaseURL == "" || c.APIKey == "" { return nil } return lidarr.NewClient(c.BaseURL, c.APIKey) } lidarrReconciler := lidarrrequests.NewReconciler(pool, lidarrCfg, lidarrClientFn, logger.With("component", "lidarr"), bus) go lidarrReconciler.Run(ctx) // library_changes compactor (#357 follow-up). Daily tick; deletes // rows older than the configured retention so the change-log table // doesn't grow unbounded. Clients that drop offline longer than // retention hit the /api/library/sync 410 fallback and resync. libraryChangesCompactor := syncpkg.NewCompactor(pool, logger.With("component", "library_changes_compactor")) go libraryChangesCompactor.Run(ctx) // Per-user system-playlist scheduler (#392 Half B). Fires each // active user's daily build at 03:00 in their stored timezone. // Replaces the 24h-anchored cron loop (removed in the next commit // of this arc). playlistScheduler, err := playlists.NewScheduler( pool, logger.With("component", "playlist_scheduler"), cfg.Storage.DataDir, bus, ) if err != nil { return fmt.Errorf("init playlist scheduler: %w", err) } if err := playlistScheduler.Start(ctx); err != nil { return fmt.Errorf("start playlist scheduler: %w", err) } defer playlistScheduler.Stop() // Ensure DataDir exists before any service tries to write into it. // Fatal on failure: a non-writable data_dir silently breaks every // downstream cache (playlist covers, artist art, album-cover fallback) // and produces "0 successes" symptoms that are hard to diagnose. if cfg.Storage.DataDir != "" { if err := os.MkdirAll(cfg.Storage.DataDir, 0o755); err != nil { logger.Error("data_dir create failed; refusing to start", "path", cfg.Storage.DataDir, "err", err) os.Exit(1) } } scanCfg := library.RunScanConfig{ BackfillCap: 5000, EnrichCap: -1, // #388: no global cap; per-provider rate limits ArtistEnrichCap: -1, DataDir: cfg.Storage.DataDir, } // Filesystem watcher: near-instant pickup of new/changed files via // fsnotify, scanning just the affected paths + enriching their albums // inline. Replaces the removed configurable scan scheduler. watcher := library.NewWatcher(scanner, coverEnricher, logger.With("component", "watcher"), cfg.Library.ScanPaths) go func() { if werr := watcher.Run(ctx); werr != nil { logger.Warn("library watcher exited", "err", werr) } }() // Safety-net delta walk backstops the watcher (inotify limits / dropped // events) at a fixed low frequency. No operator config. go runSafetyNetScans(ctx, pool, scanner, coverEnricher, logger.With("component", "scan_safetynet"), scanCfg) srv := server.New(logger, pool, scanner, subsonic.Config{ AllowPlaintextPassword: cfg.Subsonic.AllowPlaintextPassword, }, cfg.Events, cfg.Recommendation, cfg.Storage.DataDir, cfg.Branding, coverEnricher, coverSettings, scanner, scanCfg) srv.Bus = bus srv.PlaylistScheduler = playlistScheduler srv.StreamSecret = cfg.StreamSecret httpServer := &http.Server{ Addr: cfg.Server.Address, Handler: srv.Router(), ReadHeaderTimeout: 10 * time.Second, } errCh := make(chan error, 1) go func() { logger.Info("listening", "address", cfg.Server.Address) if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { errCh <- err } close(errCh) }() select { case <-ctx.Done(): logger.Info("shutdown signal received") case err := <-errCh: if err != nil { return fmt.Errorf("server: %w", err) } } shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := httpServer.Shutdown(shutdownCtx); err != nil { return fmt.Errorf("shutdown: %w", err) } logger.Info("minstrel stopped") return nil }