minstrel/cmd/minstrel/main.go

package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"slices"
	"syscall"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"

	"git.fabledsword.com/bvandeusen/minstrel/internal/acoustid"
	"git.fabledsword.com/bvandeusen/minstrel/internal/backup"
	"git.fabledsword.com/bvandeusen/minstrel/internal/buildinfo"
	"git.fabledsword.com/bvandeusen/minstrel/internal/catalogue"
	"git.fabledsword.com/bvandeusen/minstrel/internal/config"
	"git.fabledsword.com/bvandeusen/minstrel/internal/coplay"
	"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/importer"
	"git.fabledsword.com/bvandeusen/minstrel/internal/library"
	"git.fabledsword.com/bvandeusen/minstrel/internal/lifecycle"
	"git.fabledsword.com/bvandeusen/minstrel/internal/logging"
	"git.fabledsword.com/bvandeusen/minstrel/internal/mailer"
	"git.fabledsword.com/bvandeusen/minstrel/internal/notifications"
	"git.fabledsword.com/bvandeusen/minstrel/internal/playlists"
	"git.fabledsword.com/bvandeusen/minstrel/internal/reacquisition"
	"git.fabledsword.com/bvandeusen/minstrel/internal/recsettings"
	"git.fabledsword.com/bvandeusen/minstrel/internal/requests"
	"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble"
	"git.fabledsword.com/bvandeusen/minstrel/internal/scrobble/listenbrainz"
	"git.fabledsword.com/bvandeusen/minstrel/internal/secret"
	"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"
	"git.fabledsword.com/bvandeusen/minstrel/internal/tags"
	"git.fabledsword.com/bvandeusen/minstrel/internal/wanted"
)

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
		case "restore":
			if err := runRestore(os.Args[2:]); err != nil {
				fmt.Fprintf(os.Stderr, "minstrel restore: %v\n", err)
				os.Exit(1)
			}
			return
		}
	}
	err := run()
	if errors.Is(err, errRestart) {
		err = restartInPlace()
	}
	if err != nil {
		fmt.Fprintf(os.Stderr, "minstrel: %v\n", err)
		os.Exit(1)
	}
}

// errRestart is run's answer when the server stopped in order to start again:
// after a web restore, so the restored settings are read at boot.
var errRestart = errors.New("restart requested")

// restartInPlace replaces this process with a fresh start of the same binary,
// arguments and environment. The process id stays the same, so a container
// keeps running without needing a restart policy.
func restartInPlace() error {
	exe, err := os.Executable()
	if err != nil {
		return fmt.Errorf("restart: %w", err)
	}
	// Exec returns only when it failed.
	return fmt.Errorf("restart: %w", syscall.Exec(exe, os.Args, os.Environ()))
}

// 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()

	// Stored credentials are sealed under a key derived from the stream
	// secret. Install it before anything reads one, and seal any value still
	// stored in the clear.
	secret.Install(cfg.StreamSecret)
	sealed, err := secret.SealStored(ctx, pool)
	if err != nil {
		return fmt.Errorf("seal stored credentials: %w", err)
	}
	if sealed > 0 {
		logger.Info("sealed stored credentials", "count", sealed)
	}

	// Fingerprinting settings: one instance, shared by the scanner,
	// the fingerprint backfill, the duplicate sweep and the admin API, so a save
	// reaches all of them without a restart. A load failure is logged, not fatal:
	// the service falls back to the shipped defaults.
	fpSettings, fpErr := library.NewFingerprintSettingsService(ctx, pool)
	if fpErr != nil {
		logger.Warn("fingerprint settings: using defaults", "err", fpErr)
	}
	// The library's folders: the ones saved in Settings, or
	// MINSTREL_LIBRARY_SCAN_PATHS until any are. A load failure falls back to
	// the env value.
	roots, rootsSaved, rootsErr := library.LoadRoots(ctx, pool, cfg.Library.ScanPaths)
	switch {
	case rootsErr != nil:
		logger.Warn("library folders: using MINSTREL_LIBRARY_SCAN_PATHS", "err", rootsErr)
	case rootsSaved && len(cfg.Library.ScanPaths) > 0 && !slices.Equal(roots, cfg.Library.ScanPaths):
		logger.Warn("library folders: the ones saved in Settings are used, not MINSTREL_LIBRARY_SCAN_PATHS",
			"settings", roots, "env", cfg.Library.ScanPaths)
	}
	scanner := library.New(pool, logger, roots, fpSettings)

	// Loudness analysis settings: shared by the loudness backfill
	// and the admin API, and falls back to the defaults like the above.
	loudSettings, loudErr := library.NewLoudnessSettingsService(ctx, pool)
	if loudErr != nil {
		logger.Warn("loudness settings: using defaults", "err", loudErr)
	}

	// AcoustID lookup settings: shared by the lookup worker and the
	// admin API. A load failure leaves the lookup off.
	acoustIDSettings, acoustIDErr := library.NewAcoustIDSettingsService(ctx, pool)
	if acoustIDErr != nil {
		logger.Warn("acoustid settings: using defaults (lookup off)", "err", acoustIDErr)
	}
	// One AcoustID client, so the lookup worker and the import's check share
	// its pace (AcoustID allows three requests a second per key).
	acoustIDClient := acoustid.New(acoustid.DefaultBaseURL)
	acoustIDLookup := library.NewAcoustIDLookupWorker(pool, logger.With("component", "acoustid_lookup"),
		acoustIDSettings, acoustIDClient)

	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

	// Recommendation tuning: seeds the shipped defaults on first boot and
	// pushes the daily-mix weights and taste config into package playlists.
	// It must precede the scheduler, so the first builds use the tuned values.
	recSettings, err := recsettings.New(ctx, pool, logger.With("component", "recsettings"))
	if err != nil {
		logger.Error("recommendation settings service init failed", "err", err)
		os.Exit(1)
	}

	// 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(roots) > 0 {
		go func() {
			if _, err := library.RunScan(ctx, pool, scanner, coverEnricher,
				logger.With("component", "scan_run"),
				library.RunScanConfig{
					BackfillCap:     5000,
					EnrichCap:       -1, // no global cap; each provider paces itself
					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
		// art keeps arriving with scan-on-startup off.
		go func() {
			if _, err := library.RunScan(ctx, pool, nil, coverEnricher,
				logger.With("component", "scan_run"),
				library.RunScanConfig{
					BackfillCap:     5000,
					EnrichCap:       -1, // no global cap; each provider paces itself
					ArtistEnrichCap: -1,
					DataDir:         cfg.Storage.DataDir,
				},
			); err != nil {
				logger.Warn("boot-only scan run failed", "err", err)
			}
		}()
	}

	// ListenBrainz scrobble worker: every 30s, drains up to 50 pending rows.
	// Rows from users with scrobbling off are skipped inside the worker.
	scrobbleWorker := scrobble.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "scrobble"))
	go scrobbleWorker.Run(ctx)

	// Similarity ingest worker: hourly, fetches ListenBrainz similarity for up
	// to 5 played tracks and 5 played artists, each re-fetched at most weekly.
	// The endpoints are public, so no token is needed.
	similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity"))
	go similarityWorker.Run(ctx)

	// Household co-play worker: recomputes artist–artist
	// co-occurrence edges (source='user_cooccurrence') from play_events every
	// 6h — a collaborative candidate arm for the radio/mix pools. Pure local
	// SQL, no external calls; empty on single-user servers.
	go coplay.NewWorker(pool, logger.With("component", "coplay")).Run(ctx)

	// Fingerprint backfill: fingerprints the tracks the scan skips (unchanged
	// files, and rows fingerprinted by an older method). A worker of its own
	// rather than a scan stage; see internal/library/fingerprint_backfill.go.
	go library.NewFingerprintBackfillWorker(pool, logger.With("component", "fingerprint_backfill"), fpSettings).Run(ctx)

	// Loudness backfill: measures every track's loudness for
	// normalization, new tracks included; the scan only drops a changed file's
	// measurement. See internal/library/loudness_backfill.go.
	go library.NewLoudnessBackfillWorker(pool, logger.With("component", "loudness_backfill"), loudSettings).Run(ctx)

	// AcoustID lookup: fills the recording MBIDs untagged tracks lack, so the
	// similarity arms can see them. Idle until an admin adds a key and
	// switches it on. See internal/library/acoustid_lookup.go.
	go acoustIDLookup.Run(ctx)

	// Duplicate sweep: proposes groups of tracks holding one
	// recording, from the fingerprints above. Sweeps only when fingerprints have
	// changed since the last sweep.
	go library.NewDuplicateSweepWorker(pool, logger.With("component", "duplicate_sweep"), fpSettings).Run(ctx)

	// Tag-enrichment worker: reconciles the compiled-in
	// tag providers with tag_provider_settings, bumps the sources version if
	// the provider set changed (re-opening settled rows), then drains tracks
	// needing folksonomy tags on a periodic tick. Standalone (not in the file
	// scan chain) because tag lookups need only DB fields — MBID / artist /
	// title — that a scan has already imported.
	tagSettings, err := tags.NewSettingsService(ctx, pool, logger.With("component", "tags"))
	if err != nil {
		logger.Error("tag settings service init failed", "err", err)
		os.Exit(1)
	}
	if newVer, bumped, berr := tagSettings.BumpVersionIfProvidersChanged(ctx); berr != nil {
		logger.Warn("tags: provider-hash boot check failed", "err", berr)
	} else if bumped {
		logger.Info("tags: registered provider set changed; version bumped", "new_version", newVer)
	}
	tagEnricher := tags.NewEnricher(pool, logger.With("component", "tags"), tagSettings)
	go tags.NewWorker(tagEnricher, logger.With("component", "tags")).Run(ctx)

	// GC worker: hourly, sweeps the 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 and expired
	// password_resets. Each sweep is idempotent.
	gcWorker := gc.NewWorker(pool, logger.With("component", "gc"))
	go gcWorker.Run(ctx)

	// Live-event bus shared between SSE subscribers (api.Mount) and the
	// background workers that publish. Built before any of them, so they all
	// share one 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)
	// The notifications inbox's one writer, shared by every
	// background producer started here. The API builds its own over the
	// same pool and bus.
	notifier := notifications.New(pool, bus, logger.With("component", "notifications"))

	// The request reconciler completes approved requests once the library
	// holds what they asked for.
	requestReconciler := requests.NewReconciler(pool, logger.With("component", "requests"), bus)
	requestReconciler.SetNotifier(notifier)
	library.SetNotifier(notifier)
	go requestReconciler.Run(ctx)

	// Release catalogue: re-reads discographies older than a
	// week from MusicBrainz, a batch an hour, through the process-wide
	// MusicBrainz pacer it shares with tag enrichment.
	cat := catalogue.New(pool, nil, logger.With("component", "catalogue"))
	go cat.Run(ctx)

	// The wanted list: reads followed artists' discographies
	// daily for new releases and marks wants held once the library has them.
	go wanted.New(pool, cat, logger.With("component", "wanted")).Run(ctx)

	// Search and grab: every ten minutes, searches the
	// configured indexers for wants due, paced by each source's
	// searches-per-hour, and hands the best release to the download client
	// when auto-grab is on.
	go wanted.NewSearcher(pool, cat, logger.With("component", "search")).Run(ctx)

	// Import with verification: every minute, finished
	// downloads are checked against their album (AcoustID, lengths, tags)
	// and filed into the library with MusicBrainz tags, or refused.
	var printer importer.Printer
	if acoustIDSettings != nil {
		printer = &importer.AcoustIDPrinter{Settings: acoustIDSettings, Client: acoustIDClient}
	}
	imp := importer.New(pool, scanner, cat, printer, scanner.Roots, logger.With("component", "import"))
	imp.DataDir = cfg.Storage.DataDir
	go imp.Run(ctx)

	// Download lifecycle: every ten minutes, a download that
	// failed or that nobody shares is retried, then given up and its album
	// searched again; an imported one seeds, then is removed.
	go lifecycle.New(pool, logger.With("component", "lifecycle")).Run(ctx)

	// Duplicate resolver: classifies the sweep's groups and, with auto-resolve
	// on, merges the copies that are safe to merge.
	// Started after SetNotifier so its summaries reach admins. It asks the
	// catalogue for the track list of an album whose wrong slot it must
	// decide, a few a pass. See internal/library/duplicate_resolve.go.
	go library.NewDuplicateResolveWorker(pool, logger.With("component", "duplicate_resolve"),
		cfg.Storage.DataDir, fpSettings, library.ResolveSources{
			TrackLists: cat.RefreshReleaseGroup, Identify: acoustIDLookup.IdentifyRecordings,
		}).Run(ctx)

	// Missing-file re-acquisition: wants back albums whose files have been
	// gone longer than the grace window, for the search schedule to find.
	// Hourly, because the grace window is measured in hours.
	//
	// A settings-load failure is logged, not fatal: NewSettingsService always
	// returns a usable service holding the defaults.
	reacqSettings, reacqErr := reacquisition.NewSettingsService(
		ctx, pool, logger.With("component", "reacquisition"))
	if reacqErr != nil {
		logger.Warn("reacquisition: using default settings", "err", reacqErr)
	}
	go reacquisition.NewSweeper(pool, reacqSettings, logger.With("component", "reacquisition")).Run(ctx)

	// library_changes compactor: daily, 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)

	// Notifications inbox retention, daily: read rows go after 90
	// days, anything at all after a year.
	go notifications.NewRetention(pool, logger.With("component", "notifications_retention")).Run(ctx)

	// Notifications by email, grouped, never one per event: new
	// music as a daily summary at a local hour, everything else batched an
	// hour (admin-configurable) after the first item.
	go notifications.NewDigest(
		pool,
		mailer.NewSMTPSender(pool, logger.With("component", "mailer")),
		logger.With("component", "notification_digest"),
	).Run(ctx)

	// Per-user system-playlist scheduler: runs each active user's daily
	// build at 03:00 in their stored timezone.
	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, // no global cap; each provider paces itself
		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"))
	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.RecSettings = recSettings
	srv.TagSettings = tagSettings
	srv.FingerprintSettings = fpSettings
	srv.LoudnessSettings = loudSettings
	srv.AcoustIDLookup = acoustIDLookup
	// The sweeper above holds this same instance, so a save from the admin
	// card changes what it does on its next tick.
	srv.ReacqSettings = reacqSettings
	// The server's own backup: the loop runs the nightly schedule, and the
	// admin page starts, tests and configures the same instance.
	backups := backup.NewService(pool, logger.With("component", "backup"), buildinfo.Version)
	backups.Roots = scanner.Roots
	go backups.Loop(ctx)
	srv.Backups = backups
	srv.StreamSecret = cfg.StreamSecret
	restart := make(chan struct{}, 1)
	srv.Restart = func() {
		select {
		case restart <- struct{}{}:
		default:
		}
	}
	httpServer := &http.Server{
		Addr:              cfg.Server.Address,
		Handler:           srv.Router(),
		ReadHeaderTimeout: 10 * time.Second,
		// Closes keep-alive connections nobody is using. Deliberately no
		// ReadTimeout or WriteTimeout: either would cut off audio streams and
		// the SSE event stream. Request bodies get their own deadline in the
		// server's limitRequestBody middleware instead.
		IdleTimeout: 120 * time.Second,
	}
	// Shutdown waits for in-flight requests but never cancels them, and an
	// SSE stream only ends when its client leaves. Closing the bus ends
	// every stream as shutdown begins, so the drain waits only on real work
	// such as an audio stream; clients reconnect to the next instance.
	httpServer.RegisterOnShutdown(bus.Close)

	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)
	}()

	restarting := false
	select {
	case <-ctx.Done():
		logger.Info("shutdown signal received")
	case <-restart:
		logger.Info("restarting to load the restored backup")
		restarting = true
	case err := <-errCh:
		if err != nil {
			return fmt.Errorf("server: %w", err)
		}
	}
	// Ends the background work (scans, workers, schedules) as a signal would.
	stop()

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := httpServer.Shutdown(shutdownCtx); err != nil {
		return fmt.Errorf("shutdown: %w", err)
	}
	// A backup still running is stopped and records that it was cut off.
	backups.Close()
	if restarting {
		logger.Info("minstrel stopped; starting again")
		return errRestart
	}
	logger.Info("minstrel stopped")
	return nil
}