minstrel/internal/library/scanner.go

package library

import (
	"context"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"math"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/dhowden/tag"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/jackc/pgx/v5/pgxpool"

	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
	syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync"
)

// audioExtensions is the set the scanner indexes. Keep in sync with
// `internal/api/media.go` MIME detection — the stream handler must be
// able to serve every extension the scanner indexes, and there is no
// point in adding extensions to the stream handler that the scanner
// will silently skip. Drift #571 caught the divergence after .opus,
// .aac, and .wav were added to media.go but not here.
var audioExtensions = map[string]bool{
	".mp3":  true,
	".m4a":  true,
	".flac": true,
	".ogg":  true,
	".opus": true,
	".aac":  true,
	".wav":  true,
}

// tagReadVersion is the version of this package's tag-extraction logic. Rows
// whose tracks.tag_read_version is lower get their tags re-read on the next
// scan even when the file itself hasn't changed, so a fix reaches an existing
// library without the operator rebuilding it (migration 0054).
//
// Bump this whenever a change to tag extraction should reach already-indexed
// files, and say why below.
//
//	1: genre read from the ID3v2 TCON frame directly and stored ";"-delimited.
//	   dhowden/tag welds null-separated multi-values into one token
//	   ("Alternative Rock" + "Rock" -> "Alternative RockRock"), which corrupted
//	   the genre browse axis and polluted the taste profile's tag vocabulary,
//	   and left bare ID3v1 numeric references unresolved (#2499).
//	2: acronym and apostrophe casing repaired on genre values (#2468) — "Edm" ->
//	   "EDM", "Children'S Music" -> "Children's Music". Bumped rather than left
//	   to new files only because taste_profile.sql reads tracks.genre directly,
//	   so a half-repaired library would carry both spellings as separate tags.
const tagReadVersion int16 = 2

type Stats struct {
	Scanned int `json:"scanned"`
	Added   int `json:"added"`
	Updated int `json:"updated"`
	Skipped int `json:"skipped"`
	Errored int `json:"errored"`
	// Missing / Restored come from the reconcile pass, not the walk (#2523):
	// rows whose file the walk didn't find, and rows whose file came back.
	// Only a full Scan sets these — see reconcileMissing.
	Missing  int `json:"missing"`
	Restored int `json:"restored"`
}

type Scanner struct {
	pool   *pgxpool.Pool
	logger *slog.Logger
	paths  []string
	// fingerprint derives a file's acoustic identity (M400). A field so an
	// integration test can substitute a deterministic one: CI has no real audio
	// to fingerprint, and what the test pins is WHEN the scan fingerprints, not
	// what the tools print. Call it through fingerprintFile.
	fingerprint func(ctx context.Context, path string, opts fingerprintOptions) fingerprintResult
	// settings is the operator's fingerprinting policy (#3913), shared with the
	// workers and the admin API. Nil fingerprints with the defaults.
	settings *FingerprintSettingsService
}

func New(pool *pgxpool.Pool, logger *slog.Logger, paths []string, settings *FingerprintSettingsService) *Scanner {
	return &Scanner{pool: pool, logger: logger, paths: paths, fingerprint: computeFingerprint, settings: settings}
}

// Scan walks every configured root and upserts any audio file whose mtime is
// newer than the existing row's updated_at. Walk errors and per-file errors
// are logged + counted; the scan keeps going.
//
// It then reconciles: rows whose file the walk never saw get marked missing,
// and rows whose file has come back get un-marked (#2523). Only a FULL scan may
// do this — the walk's set of seen paths is the evidence, and a partial
// (watcher-driven) scan has no basis for concluding anything about files it
// didn't look at. That's why ScanFiles does not reconcile.
//
// progressCb (may be nil) receives the current Stats snapshot after each
// processed file. Used by the orchestrator to drive partial-tally writes.
func (s *Scanner) Scan(ctx context.Context, progressCb func(Stats)) (Stats, error) {
	var stats Stats
	q := dbq.New(s.pool)
	start := time.Now()

	// PHASE 1 — enumerate. Collect every audio path without touching tags or
	// ffprobe. Cheap: WalkDir already stats each entry, so this adds a directory
	// traversal and nothing else.
	//
	// The order matters and is the whole reason enumeration is separate.
	// Reconcile has to mark disappeared rows BEFORE any file is processed,
	// because move detection (#2528) can only adopt a row that is already marked
	// missing. A rename performed while the server was down surfaces the deletion
	// and the addition in the SAME scan — so if reconcile ran at the end, the new
	// path would insert a fresh row first and the fork would be permanent.
	paths, walkErrs := s.enumerate(ctx, progressCb, &stats)
	stats.Errored += walkErrs
	if err := ctx.Err(); err != nil {
		return stats, err
	}

	// PHASE 2 — reconcile. Only ever on a COMPLETE enumeration: a cancelled walk
	// has a partial view and would mark everything it hadn't reached.
	seen := make(map[string]struct{}, len(paths))
	for _, p := range paths {
		seen[p] = struct{}{}
	}
	if err := s.reconcileMissing(ctx, q, seen, &stats); err != nil {
		// Not fatal. The guards deliberately refuse to act on ambiguous
		// evidence, and that refusal arrives here as an error.
		//
		// The consequence is named explicitly because it is not obvious: move
		// detection (#2528) can only adopt a row that is already marked missing,
		// so a refused reconcile also means renamed files insert fresh rows and
		// fork their history. That's the pre-#2528 behaviour rather than a new
		// failure, but it's worth knowing which scan it happened on. It bites
		// hardest when a large fraction of a small library is reorganised at
		// once, which trips the mark cap.
		s.logger.Warn("library scan: reconcile skipped — moved files will fork rather than adopt",
			"err", err)
	}

	// PHASE 3 — process, in walk order so logs and cover-art batching stay
	// grouped by directory rather than following map iteration order.
	for _, path := range paths {
		if ctx.Err() != nil {
			break
		}
		if _, _, err := s.scanFile(ctx, q, path, &stats); err != nil {
			s.logger.Warn("library scan file error", "path", path, "err", err)
			stats.Errored++
		}
		if progressCb != nil {
			progressCb(stats)
		}
	}

	s.logger.Info("library scan complete",
		"scanned", stats.Scanned,
		"added", stats.Added,
		"updated", stats.Updated,
		"skipped", stats.Skipped,
		"errored", stats.Errored,
		"missing", stats.Missing,
		"restored", stats.Restored,
		"duration_ms", time.Since(start).Milliseconds(),
	)
	if err := ctx.Err(); err != nil {
		return stats, err
	}
	return stats, nil
}

// enumerate walks every configured root and returns the audio paths found, in
// walk order, plus a count of walk errors.
//
// A path is recorded even if it will later fail to parse: an unreadable file is a
// broken file, not a missing one, and letting reconcile mark it missing would
// hide it from the operator behind the wrong explanation.
func (s *Scanner) enumerate(
	ctx context.Context, progressCb func(Stats), stats *Stats,
) ([]string, int) {
	paths := make([]string, 0, 8192)
	errs := 0
	for _, root := range s.paths {
		// WalkDir's own error return is folded into the per-entry handler below,
		// so a bad root is counted rather than aborting the whole scan — one
		// unreadable root shouldn't discard the others' results.
		_ = filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
			if ctx.Err() != nil {
				return fs.SkipAll
			}
			if err != nil {
				s.logger.Warn("library scan walk error", "path", path, "err", err)
				errs++
				if progressCb != nil {
					progressCb(*stats)
				}
				return nil
			}
			if d.IsDir() {
				return nil
			}
			if !audioExtensions[strings.ToLower(filepath.Ext(path))] {
				return nil
			}
			paths = append(paths, path)
			return nil
		})
	}
	return paths, errs
}

// scanFile upserts a single audio file. Returns the album ID the track
// belongs to and whether the file was added/updated (false = skipped as
// unchanged), so watcher-driven callers can enrich just the changed albums.
func (s *Scanner) scanFile(
	ctx context.Context, q *dbq.Queries, path string, stats *Stats,
) (pgtype.UUID, bool, error) {
	stats.Scanned++

	info, err := os.Stat(path)
	if err != nil {
		return pgtype.UUID{}, false, fmt.Errorf("stat: %w", err)
	}
	mtime := info.ModTime()

	existing, err := q.GetTrackByPath(ctx, path)
	knownTrack := err == nil
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return pgtype.UUID{}, false, fmt.Errorf("lookup: %w", err)
	}
	// Incremental skip: only when the file hasn't changed AND we already have a
	// real duration AND the row's tag-derived columns were written by the
	// current extraction logic. The duration clause lets older scans that
	// recorded duration_ms=0 (before ffprobe was wired) get backfilled without
	// forcing the operator to wipe the library; the tag-version clause does the
	// same job for tag-extraction fixes (#2499). Once both are current,
	// subsequent scans short-circuit as before.
	unchanged := knownTrack && !existing.UpdatedAt.Time.Before(mtime)
	if unchanged && existing.DurationMs > 0 && existing.TagReadVersion >= tagReadVersion {
		stats.Skipped++
		return pgtype.UUID{}, false, nil
	}

	f, err := os.Open(path)
	if err != nil {
		return pgtype.UUID{}, false, fmt.Errorf("open: %w", err)
	}
	defer func() { _ = f.Close() }()

	meta, err := tag.ReadFrom(f)
	if err != nil {
		return pgtype.UUID{}, false, fmt.Errorf("tag read: %w", err)
	}
	albumMBID, artistMBID := extractMBIDs(meta)
	recordingMBID := extractRecordingMBID(meta)

	artistName := meta.Artist()
	if artistName == "" {
		artistName = "Unknown Artist"
	}
	albumTitle := meta.Album()
	if albumTitle == "" {
		albumTitle = "Unknown Album"
	}
	trackTitle := meta.Title()
	if trackTitle == "" {
		trackTitle = strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
	}

	artist, err := s.resolveArtist(ctx, q, artistName, artistMBID)
	if err != nil {
		return pgtype.UUID{}, false, fmt.Errorf("artist: %w", err)
	}
	album, err := s.resolveAlbum(ctx, q, artist.ID, albumTitle, meta.Year(), albumMBID)
	if err != nil {
		return pgtype.UUID{}, false, fmt.Errorf("album: %w", err)
	}

	trackNum, _ := meta.Track()
	discNum, _ := meta.Disc()

	// An unchanged file being re-read only to refresh tag-derived columns
	// doesn't need another ffprobe: the stored duration is still accurate, and
	// the file's bytes haven't moved. This keeps a library-wide tag-repair pass
	// (a tagReadVersion bump) bound by tag reads rather than costing one
	// fork+exec per file.
	var durationMs int32
	if unchanged && existing.DurationMs > 0 {
		durationMs = existing.DurationMs
	} else {
		probed, perr := probeDurationMs(ctx, path)
		if perr != nil {
			// Missing duration is degraded UX (clients can't scrub) but not a
			// blocker for ingestion. Record the file with 0ms; the next scan
			// will retry via the backfill clause in the skip check above.
			s.logger.Warn("library scan: ffprobe failed", "path", path, "err", perr)
		}
		durationMs = probed
	}

	// Fingerprint only bytes this row has not seen: a new path, or a file whose
	// mtime moved past the row's. An unchanged file re-read for a tag repair
	// keeps its stored fingerprint, for the same reason it keeps its duration
	// above — a tagReadVersion bump must stay bound by tag reads, not become a
	// decode of the whole library.
	//
	// An unchanged file with NO fingerprint yet is the backfill's job (#3908),
	// deliberately not the scan's. Folding it into the skip check would make
	// the first scan after an upgrade re-decode every track and push a sync
	// change to every client for each one.
	//
	// Computed before move adoption so adoption can match on the audio hash
	// (#3914); stored after the upsert, once the row id is known.
	//
	// With fingerprinting switched off (#3913) the scan decodes nothing, but it
	// still takes the stream hash: a demux rather than a decode, and the only
	// thing that recognises a moved file with no MBID.
	var fp fingerprintResult
	fpCfg := s.settings.Get()
	fingerprinted := !unchanged
	if fingerprinted {
		fp = s.fingerprintFile(ctx, path, fingerprintOptions{
			lengthSec:   fpCfg.ChromaprintLengthSec,
			chromaprint: fpCfg.Enabled,
		})
	}

	// A path we've never seen might not be a new track — it might be one that
	// moved or was renamed (#2528). Adopting re-points the existing row at this
	// path and clears its missing mark, so the UpsertTrack below conflicts on
	// file_path and updates THAT row: same track id, likes and play history
	// intact. Without this, renumbering an album forks every track on it.
	//
	// Runs after fingerprinting because, for a file with no MBID, adoption matches
	// on its audio hash (#3914), and only for genuinely unknown paths — a known
	// path is already the row we're going to update.
	if !knownTrack {
		if s.adoptMovedTrack(ctx, q, path, fp.streamSHA256, recordingMBID) {
			// Count it as an update: the row existed, and reporting it as Added
			// would overstate library growth on every reorganisation.
			knownTrack = true
		}
	}

	params := dbq.UpsertTrackParams{
		Title:      trackTitle,
		AlbumID:    album.ID,
		ArtistID:   artist.ID,
		DurationMs: durationMs,
		FilePath:   path,
		FileSize:   info.Size(),
		FileFormat: strings.TrimPrefix(strings.ToLower(filepath.Ext(path)), "."),
		// Stamped so a future extraction fix can find this row again.
		TagReadVersion: tagReadVersion,
	}
	if trackNum > 0 {
		v := int32(trackNum)
		params.TrackNumber = &v
	}
	if discNum > 0 {
		v := int32(discNum)
		params.DiscNumber = &v
	}
	if genres, fellBack := extractGenres(meta, f); len(genres) > 0 {
		if fellBack {
			// dhowden/tag's welded value — see genre.go. Logged because the
			// stored genre for this file is the old, corrupt shape.
			s.logger.Warn("library scan: genre frame unreadable, using fallback",
				"path", path, "genre", meta.Genre())
		}
		g := strings.Join(genres, genreDelimiter)
		params.Genre = &g
	}
	// Recording MBID feeds the ListenBrainz similarity pipeline.
	// UpsertTrack heals mbid on the file_path conflict, so a re-scan
	// of a previously-untagged-into-DB track backfills it for free.
	if recordingMBID != "" {
		m := recordingMBID
		params.Mbid = &m
	}

	track, err := q.UpsertTrack(ctx, params)
	if err != nil {
		return pgtype.UUID{}, false, fmt.Errorf("upsert track: %w", err)
	}
	if err := syncpkg.LogChange(ctx, s.pool, syncpkg.EntityTrack,
		syncpkg.FormatUUID(track.ID), syncpkg.OpUpsert); err != nil {
		// Best-effort: log but don't fail the scan. The next scan that
		// touches this track will re-emit the change.
		s.logger.Warn("library scan: LogChange track upsert failed", "track_id", track.ID, "err", err)
	}
	if fingerprinted {
		if fpCfg.Enabled {
			storeFingerprint(ctx, q, s.logger, track.ID, path, fp, fpCfg.ChromaprintLengthSec)
		} else if err := q.DeleteTrackFingerprint(ctx, track.ID); err != nil {
			// Off, nothing is stored — but a row describing the previous bytes
			// must not outlive them, or the sweep would compare audio that is gone.
			s.logger.Warn("fingerprint: clearing stale fingerprint failed", "path", path, "err", err)
		}
	}

	if knownTrack {
		stats.Updated++
	} else {
		stats.Added++
	}
	return album.ID, true, nil
}

// ScanFiles processes a specific set of audio file paths (watcher-driven),
// applying the same upsert + delta-skip logic as a full Scan. Non-audio or
// unreadable paths are skipped (logged), never fatal. Returns the distinct
// album IDs whose tracks were added or updated, so the caller can enrich
// just those albums inline rather than waiting for a batch pass.
func (s *Scanner) ScanFiles(ctx context.Context, paths []string) ([]pgtype.UUID, error) {
	q := dbq.New(s.pool)
	seen := make(map[[16]byte]struct{})
	changed := make([]pgtype.UUID, 0)
	var stats Stats
	for _, path := range paths {
		if ctx.Err() != nil {
			return changed, ctx.Err()
		}
		if !audioExtensions[strings.ToLower(filepath.Ext(path))] {
			continue
		}
		albumID, didChange, err := s.scanFile(ctx, q, path, &stats)
		if err != nil {
			s.logger.Warn("library watch: scan file error", "path", path, "err", err)
			continue
		}
		if didChange && albumID.Valid {
			if _, ok := seen[albumID.Bytes]; !ok {
				seen[albumID.Bytes] = struct{}{}
				changed = append(changed, albumID)
			}
		}
	}
	if stats.Added > 0 || stats.Updated > 0 {
		s.logger.Info("library watch: scan batch",
			"added", stats.Added,
			"updated", stats.Updated,
			"skipped", stats.Skipped,
			"errored", stats.Errored,
		)
	}
	return changed, nil
}

func (s *Scanner) resolveArtist(ctx context.Context, q *dbq.Queries, name, mbid string) (dbq.Artist, error) {
	existing, err := q.GetArtistByName(ctx, name)
	if err == nil {
		// Heal: backfill mbid on a previously-imported row if we have one now.
		if mbid != "" && (existing.Mbid == nil || *existing.Mbid == "") {
			m := mbid
			if uerr := q.SetArtistMbidIfNull(ctx, dbq.SetArtistMbidIfNullParams{
				ID:   existing.ID,
				Mbid: &m,
			}); uerr != nil {
				if isUniqueViolation(uerr) {
					// Another artist row already owns this MBID — two rows that
					// should be merged (usually two spellings of one name).
					// Expected, not a fault: leave NULL and let the operator
					// merge. Mirrors resolveAlbum, which has always handled it
					// this way — without this branch the identical benign
					// condition logged a generic warning plus a Postgres ERROR
					// line on every scan, which teaches an operator to ignore
					// database errors (#2524).
					s.logger.Info("library scan: duplicate artist mbid (canonical row already owns it)",
						"artist_id", existing.ID, "artist", name, "mbid", mbid)
				} else {
					s.logger.Warn("library scan: heal artist mbid failed",
						"artist_id", existing.ID, "err", uerr)
				}
			} else {
				existing.Mbid = &m
			}
		}
		return existing, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return dbq.Artist{}, err
	}
	params := dbq.UpsertArtistParams{
		Name:     name,
		SortName: sortKey(name),
	}
	if mbid != "" {
		m := mbid
		params.Mbid = &m
	}
	artist, err := q.UpsertArtist(ctx, params)
	if err != nil {
		return dbq.Artist{}, err
	}
	if err := syncpkg.LogChange(ctx, s.pool, syncpkg.EntityArtist,
		syncpkg.FormatUUID(artist.ID), syncpkg.OpUpsert); err != nil {
		s.logger.Warn("library scan: LogChange artist upsert failed", "artist_id", artist.ID, "err", err)
	}
	return artist, nil
}

func (s *Scanner) resolveAlbum(ctx context.Context, q *dbq.Queries, artistID pgtype.UUID, title string, year int, mbid string) (dbq.Album, error) {
	existing, err := q.GetAlbumByArtistAndTitle(ctx, dbq.GetAlbumByArtistAndTitleParams{ArtistID: artistID, Title: title})
	if err == nil {
		// Heal: backfill mbid on a previously-imported row if we have one now.
		if mbid != "" && (existing.Mbid == nil || *existing.Mbid == "") {
			m := mbid
			if uerr := q.SetAlbumMbidIfNull(ctx, dbq.SetAlbumMbidIfNullParams{
				ID:   existing.ID,
				Mbid: &m,
			}); uerr != nil {
				if isUniqueViolation(uerr) {
					// Another album row already owns this MBID — duplicate
					// release in the DB. Leave NULL; operator merges later.
					s.logger.Info("library scan: duplicate album mbid (canonical row already owns it)",
						"album_id", existing.ID, "mbid", mbid)
				} else {
					s.logger.Warn("library scan: heal album mbid failed",
						"album_id", existing.ID, "err", uerr)
				}
			} else {
				existing.Mbid = &m
			}
		}
		return existing, nil
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return dbq.Album{}, err
	}
	params := dbq.UpsertAlbumParams{
		Title:     title,
		SortTitle: sortKey(title),
		ArtistID:  artistID,
	}
	if d, ok := releaseDateFromYear(year); ok {
		params.ReleaseDate = d
	} else if year != 0 {
		// year=0 is the no-tag case; anything else getting rejected is a tag
		// we couldn't trust (typo, OOB number, …). Log it so users can chase
		// down the file but don't fail the album insert over a soft field.
		s.logger.Warn("library scan: dropping invalid release year",
			"year", year, "album", title)
	}
	if mbid != "" {
		m := mbid
		params.Mbid = &m
	}
	album, err := q.UpsertAlbum(ctx, params)
	if err != nil {
		return dbq.Album{}, err
	}
	if err := syncpkg.LogChange(ctx, s.pool, syncpkg.EntityAlbum,
		syncpkg.FormatUUID(album.ID), syncpkg.OpUpsert); err != nil {
		s.logger.Warn("library scan: LogChange album upsert failed", "album_id", album.ID, "err", err)
	}
	return album, nil
}

// releaseDateFromYear converts a tag-supplied year into a Postgres date,
// returning ok=false if the year is outside what we'll accept. We're
// deliberately strict (1..9999): Postgres' date type goes much wider, but
// 5-digit years from ID3 tags are always garbage in practice and trigger
// SQLSTATE 22008 (datetime field overflow) at insert time.
func releaseDateFromYear(year int) (pgtype.Date, bool) {
	if year < 1 || year > 9999 {
		return pgtype.Date{}, false
	}
	return pgtype.Date{
		Time:  time.Date(year, 1, 1, 0, 0, 0, 0, time.UTC),
		Valid: true,
	}, true
}

// sortKey drops a leading "The " for sortable ordering. Non-English articles
// (Los, Die, Les) can be added if users ask — keeping the rule obvious for now.
func sortKey(s string) string {
	if len(s) >= 4 && strings.EqualFold(s[:4], "the ") {
		return s[4:]
	}
	return s
}

// probeTimeout bounds how long ffprobe is allowed to inspect a single file.
// 10s is generous for local mp3/flac — hits are usually <100ms — but caps
// blast radius if a pathological file or a slow network mount stalls a scan.
const probeTimeout = 10 * time.Second

// probeDurationMs shells out to ffprobe to extract a track's duration. We
// rely on ffmpeg being in the image (see Dockerfile). The CLI is slow per
// call (fork+exec) but scans are batch-mode; this is simpler than pulling
// a Go-side decoder library and handles every format ffmpeg does.
func probeDurationMs(ctx context.Context, path string) (int32, error) {
	probeCtx, cancel := context.WithTimeout(ctx, probeTimeout)
	defer cancel()
	cmd := exec.CommandContext(probeCtx, "ffprobe",
		"-v", "error",
		"-show_entries", "format=duration",
		"-of", "default=noprint_wrappers=1:nokey=1",
		path,
	)
	out, err := cmd.Output()
	if err != nil {
		return 0, fmt.Errorf("ffprobe: %w", err)
	}
	seconds, err := strconv.ParseFloat(strings.TrimSpace(string(out)), 64)
	if err != nil {
		return 0, fmt.Errorf("parse ffprobe output %q: %w", out, err)
	}
	if seconds <= 0 || math.IsNaN(seconds) || math.IsInf(seconds, 0) {
		return 0, fmt.Errorf("invalid duration: %v", seconds)
	}
	ms := seconds * 1000
	if ms > math.MaxInt32 {
		ms = math.MaxInt32
	}
	return int32(ms), nil
}