Vellum/internal/scanner/scan.go

// Package scanner walks a library's root and reconciles it with the database:
// files that appeared become items, files that changed are re-upserted with
// fresh codec facts, and files that vanished are flagged missing rather than
// deleted.
//
// It is separate from internal/library because the two have different callers.
// A scan runs in a background job with no user and no HTTP request, tolerates a
// partially readable tree, and must never delete; library CRUD is an admin
// surface where a bad request is a 400 and a bad path is a 404. Scanner imports
// library (for Library and the media-kind registry), never the other way round.
package scanner

import (
	"context"
	"database/sql"
	"fmt"
	"io/fs"
	"log/slog"
	"path"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"time"

	"github.com/mithral/vellum/internal/core"
	"github.com/mithral/vellum/internal/jobs"
	"github.com/mithral/vellum/internal/library"
	"github.com/mithral/vellum/internal/natsort"
	"github.com/mithral/vellum/internal/probe"
)

const (
	// scanBatchSize is the number of media-item groups reconciled per
	// transaction. Chunking keeps the SQLite write lock held for seconds at a
	// time rather than minutes on a large library, so other writers (progress
	// saves, session touches) are not starved.
	scanBatchSize = 500
	// maxScanErrors bounds the errors retained in a ScanResult. A pathological
	// tree can produce an error per file; we keep the first few and count the
	// rest rather than balloon the result and the job row.
	maxScanErrors = 100
)

// ScanResult summarises one scan pass.
type ScanResult struct {
	LibraryID        int64    `json:"library_id"`
	Scanned          int      `json:"scanned"`
	Added            int      `json:"added"`
	Updated          int      `json:"updated"`
	Restored         int      `json:"restored"`
	Missing          int      `json:"missing"`
	Unchanged        int      `json:"unchanged"`
	Duration         int64    `json:"duration_ms"`
	Errors           []string `json:"errors,omitempty"`
	ErrorsSuppressed int      `json:"errors_suppressed,omitempty"`
}

// addError records a scan error while keeping res.Errors bounded; errors past
// maxScanErrors are counted instead of stored.
func (r *ScanResult) addError(msg string) {
	if len(r.Errors) < maxScanErrors {
		r.Errors = append(r.Errors, msg)
		return
	}
	r.ErrorsSuppressed++
}

// candidate is a supported file found on disk.
type candidate struct {
	relPath string
	size    int64
	mtime   int64
	inode   int64
	ext     string
}

// knownFile is an existing media_files row for this library.
type knownFile struct {
	itemID  int64
	fileID  int64
	size    int64
	mtime   int64
	missing bool
}

// group is one media item: either a single file or a directory of tracks.
type group struct {
	key   string // media_items.path (unique within the library)
	title string
	files []candidate
}

// Scan walks a library's root and reconciles it with the database.
//
// Files that changed are re-upserted; files that vanished are marked missing
// rather than deleted, so their progress survives a temporarily unavailable
// mount. A scan never fails because of a single unreadable file — it records
// the error and continues, because a partial library is better than none.
func Scan(ctx context.Context, db *sql.DB, lib library.Library, log *slog.Logger) (ScanResult, error) {
	start := time.Now()
	res := ScanResult{LibraryID: lib.ID}

	exts := library.ExtensionsFor(lib.Kind)
	if len(exts) == 0 {
		return res, fmt.Errorf("library %d: no extensions registered for kind %q", lib.ID, lib.Kind)
	}

	knownFiles, err := loadKnownFiles(ctx, db, lib.ID)
	if err != nil {
		return res, err
	}
	knownItems, err := loadKnownItems(ctx, db, lib.ID)
	if err != nil {
		return res, err
	}

	found, errs := walkRoot(lib.RootPath, exts)
	for _, e := range errs {
		res.addError(e)
	}

	groups := buildGroups(lib, found)

	now := core.Now()
	seenFiles := make(map[string]bool, len(found))

	// report writes the running result back to the job (if any) so the UI can
	// poll for progress. Best-effort: a failed report must not fail the scan.
	report := func() {
		if id, ok := jobs.JobID(ctx); ok {
			_ = jobs.SetResult(ctx, db, id, res)
		}
	}

	groupKeys := sortedGroupKeys(groups)
	for start := 0; start < len(groupKeys); start += scanBatchSize {
		end := start + scanBatchSize
		if end > len(groupKeys) {
			end = len(groupKeys)
		}
		batch := groupKeys[start:end]

		// Probe the changed media files in this batch BEFORE opening the write
		// transaction, so a slow ffprobe (a stalled mount, a huge file) can
		// never hold SQLite's write lock.
		//
		// The probe map is deliberately built from the same changed-only test
		// the transaction uses below. Probing on the unchanged path would turn
		// every rescan into a full ffprobe sweep of the entire library; the
		// size+mtime skip is what keeps rescans cheap, and the probe must live
		// on the changed side of it.
		probed := probeBatch(ctx, lib, groups, batch, knownFiles, &res)

		err := commitBatch(ctx, db, func(tx *sql.Tx) error {
			for _, key := range batch {
				g := groups[key]

				itemID, err := upsertItem(ctx, tx, lib, g.key, g.title)
				if err != nil {
					res.addError(fmt.Sprintf("%s: %v", g.key, err))
					continue
				}

				for _, c := range g.files {
					res.Scanned++
					seenFiles[c.relPath] = true

					prev, exists := knownFiles[c.relPath]
					if exists && !prev.missing && prev.size == c.size && prev.mtime == c.mtime {
						res.Unchanged++
						continue
					}

					info, wasProbed := probed[c.relPath]
					if err := upsertFile(ctx, tx, itemID, c, info, wasProbed, now); err != nil {
						res.addError(fmt.Sprintf("%s: %v", c.relPath, err))
						continue
					}
					switch {
					case !exists:
						res.Added++
					case prev.missing:
						res.Restored++
					default:
						res.Updated++
					}
				}
			}
			return nil
		})
		if err != nil {
			return res, err
		}
		report()
	}

	// Items whose group is gone entirely.
	missingItemKeys := make([]string, 0)
	for key := range knownItems {
		if _, ok := groups[key]; ok {
			continue
		}
		missingItemKeys = append(missingItemKeys, key)
	}
	sort.Strings(missingItemKeys)

	for start := 0; start < len(missingItemKeys); start += scanBatchSize {
		end := start + scanBatchSize
		if end > len(missingItemKeys) {
			end = len(missingItemKeys)
		}
		batch := missingItemKeys[start:end]

		err := commitBatch(ctx, db, func(tx *sql.Tx) error {
			for _, key := range batch {
				if err := markItemMissing(ctx, tx, lib.ID, key, now); err != nil {
					res.addError(fmt.Sprintf("%s: %v", key, err))
					continue
				}
				res.Missing++
			}
			return nil
		})
		if err != nil {
			return res, err
		}
		report()
	}

	// Individual files that vanished but whose item remains.
	missingFilePaths := make([]string, 0)
	for rel := range knownFiles {
		if seenFiles[rel] {
			continue
		}
		missingFilePaths = append(missingFilePaths, rel)
	}
	sort.Strings(missingFilePaths)

	for start := 0; start < len(missingFilePaths); start += scanBatchSize {
		end := start + scanBatchSize
		if end > len(missingFilePaths) {
			end = len(missingFilePaths)
		}
		batch := missingFilePaths[start:end]

		err := commitBatch(ctx, db, func(tx *sql.Tx) error {
			for _, rel := range batch {
				if err := markFileMissing(ctx, tx, rel, now); err != nil {
					res.addError(fmt.Sprintf("%s: %v", rel, err))
				}
			}
			return nil
		})
		if err != nil {
			return res, err
		}
		report()
	}

	res.Duration = time.Since(start).Milliseconds()
	log.Info("scan complete",
		"library", lib.Name, "kind", lib.Kind, "grouping", library.GroupModeFor(lib.Kind),
		"scanned", res.Scanned, "added", res.Added, "updated", res.Updated,
		"restored", res.Restored, "missing", res.Missing, "unchanged", res.Unchanged,
		"duration_ms", res.Duration, "errors", len(res.Errors),
		"errors_suppressed", res.ErrorsSuppressed)
	return res, nil
}

// commitBatch runs fn in a single write transaction and commits it. Per-row
// errors are handled inside fn; a returned error (begin or commit failure)
// aborts the scan.
func commitBatch(ctx context.Context, db *sql.DB, fn func(tx *sql.Tx) error) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	if err := fn(tx); err != nil {
		tx.Rollback() //nolint:errcheck
		return err
	}
	if err := tx.Commit(); err != nil {
		return fmt.Errorf("library: commit scan: %w", err)
	}
	return nil
}

// buildGroups turns found files into items. Directory-grouped kinds (audiobooks,
// music) collapse a folder of tracks into one item; everything else is one item
// per file.
func buildGroups(lib library.Library, found map[string]candidate) map[string]group {
	mode := library.GroupModeFor(lib.Kind)
	groups := make(map[string]group)

	for _, rel := range sortedCandidateKeys(found) {
		c := found[rel]

		var key, title string
		if mode == library.GroupByDirectory {
			dir := path.Dir(rel)
			if dir == "." {
				key, title = ".", lib.Name
			} else {
				key, title = dir, library.CleanTitle(path.Base(dir))
			}
		} else {
			key, title = rel, library.CleanTitle(rel)
		}

		g := groups[key]
		g.key, g.title = key, title
		g.files = append(g.files, c)
		groups[key] = g
	}

	// Natural order so track 2 precedes track 10.
	for k, g := range groups {
		sort.Slice(g.files, func(i, j int) bool {
			return natsort.Less(g.files[i].relPath, g.files[j].relPath)
		})
		groups[k] = g
	}
	return groups
}

// walkRoot collects supported files under root, keyed by slash-separated
// path relative to root.
func walkRoot(root string, exts map[string]bool) (map[string]candidate, []string) {
	out := make(map[string]candidate)
	var errs []string

	err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", p, err))
			if d != nil && d.IsDir() {
				return fs.SkipDir
			}
			return nil
		}

		name := d.Name()
		if d.IsDir() {
			if p != root && strings.HasPrefix(name, ".") {
				return fs.SkipDir
			}
			return nil
		}
		if strings.HasPrefix(name, ".") {
			return nil
		}

		ext := strings.ToLower(filepath.Ext(name))
		if !exts[ext] {
			return nil
		}

		info, err := d.Info()
		if err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", p, err))
			return nil
		}
		if !info.Mode().IsRegular() {
			return nil
		}

		rel, err := filepath.Rel(root, p)
		if err != nil {
			errs = append(errs, fmt.Sprintf("%s: %v", p, err))
			return nil
		}

		c := candidate{
			relPath: filepath.ToSlash(rel),
			size:    info.Size(),
			mtime:   info.ModTime().UnixNano(),
			ext:     strings.TrimPrefix(ext, "."),
		}
		if st, ok := info.Sys().(*syscall.Stat_t); ok {
			c.inode = int64(st.Ino)
		}
		out[c.relPath] = c
		return nil
	})
	if err != nil {
		errs = append(errs, fmt.Sprintf("%s: %v", root, err))
	}
	return out, errs
}

func loadKnownFiles(ctx context.Context, db *sql.DB, libraryID int64) (map[string]knownFile, error) {
	rows, err := db.QueryContext(ctx, `
		SELECT f.path, f.item_id, f.id, f.size, f.mtime, f.missing
		FROM media_files f
		JOIN media_items i ON i.id = f.item_id
		WHERE i.library_id = ?`, libraryID)
	if err != nil {
		return nil, fmt.Errorf("library: load known files: %w", err)
	}
	defer rows.Close()

	out := make(map[string]knownFile)
	for rows.Next() {
		var (
			absPath string
			kf      knownFile
			mtime   string
			missing int
		)
		if err := rows.Scan(&absPath, &kf.itemID, &kf.fileID, &kf.size, &mtime, &missing); err != nil {
			return nil, err
		}
		kf.mtime = parseUnixNano(mtime)
		kf.missing = missing == 1
		out[absPath] = kf
	}
	return out, rows.Err()
}

func loadKnownItems(ctx context.Context, db *sql.DB, libraryID int64) (map[string]int64, error) {
	rows, err := db.QueryContext(ctx,
		`SELECT path, id FROM media_items WHERE library_id = ?`, libraryID)
	if err != nil {
		return nil, fmt.Errorf("library: load known items: %w", err)
	}
	defer rows.Close()

	out := make(map[string]int64)
	for rows.Next() {
		var p string
		var id int64
		if err := rows.Scan(&p, &id); err != nil {
			return nil, err
		}
		out[p] = id
	}
	return out, rows.Err()
}

func upsertItem(ctx context.Context, tx *sql.Tx, lib library.Library, key, title string) (int64, error) {
	var id int64
	err := tx.QueryRowContext(ctx, `
		INSERT INTO media_items (library_id, kind, path, title, sort_title, missing, missing_since)
		VALUES (?, ?, ?, ?, ?, 0, '')
		ON CONFLICT (library_id, path)
		DO UPDATE SET missing = 0, missing_since = ''
		RETURNING id`,
		lib.ID, lib.Kind, key, title, strings.ToLower(title),
	).Scan(&id)
	if err != nil {
		return 0, fmt.Errorf("upsert item: %w", err)
	}
	return id, nil
}

func upsertFile(ctx context.Context, tx *sql.Tx, itemID int64, c candidate, info probe.Info, probed bool, now string) error {
	// A non-probed file (an ebook, a photo, or any file on a host without
	// ffprobe) must keep whatever codec facts a previous scan recorded, so its
	// upsert never mentions those columns at all.
	if !probed {
		_, err := tx.ExecContext(ctx, `
			INSERT INTO media_files (item_id, path, size, mtime, inode, container, missing, missing_since)
			VALUES (?, ?, ?, ?, ?, ?, 0, '')
			ON CONFLICT (path)
			DO UPDATE SET item_id = excluded.item_id, size = excluded.size,
			              mtime = excluded.mtime, inode = excluded.inode,
			              container = excluded.container, missing = 0,
			              missing_since = ''`,
			itemID, c.relPath, c.size, fmt.Sprint(c.mtime), c.inode, c.ext)
		if err != nil {
			return fmt.Errorf("upsert file: %w", err)
		}
		return nil
	}

	_, err := tx.ExecContext(ctx, `
		INSERT INTO media_files (item_id, path, size, mtime, inode, container, missing, missing_since,
		                         duration_ms, vcodec, acodec, width, height, probed_at)
		VALUES (?, ?, ?, ?, ?, ?, 0, '', ?, ?, ?, ?, ?, ?)
		ON CONFLICT (path)
		DO UPDATE SET item_id = excluded.item_id, size = excluded.size,
		              mtime = excluded.mtime, inode = excluded.inode,
		              container = excluded.container, missing = 0,
		              missing_since = '',
		              duration_ms = excluded.duration_ms, vcodec = excluded.vcodec,
		              acodec = excluded.acodec, width = excluded.width,
		              height = excluded.height, probed_at = excluded.probed_at`,
		itemID, c.relPath, c.size, fmt.Sprint(c.mtime), c.inode, c.ext,
		info.DurationMS, info.VCodec, info.ACodec, info.Width, info.Height, now)
	if err != nil {
		return fmt.Errorf("upsert file: %w", err)
	}
	return nil
}

// probeBatch probes every new or changed video/audio file in one batch and
// returns the facts keyed by relative path.
//
// Only files the transaction will treat as changed (a size+mtime mismatch, or a
// row that is absent or flagged missing) are probed; unchanged files are absent
// from the result, so no ffprobe runs and no codec column is rewritten for
// them. Non-media kinds (ebooks, comics, photos, games) are never probed: they
// hold no media streams, and running ffprobe over a library of them would be
// pure waste.
//
// It runs before commitBatch opens the write transaction, so the write lock is
// never held across an ffprobe. A probe failure is recorded through the bounded
// error mechanism and the file is written with empty codec fields rather than
// failing the scan.
func probeBatch(ctx context.Context, lib library.Library, groups map[string]group, batch []string, known map[string]knownFile, res *ScanResult) map[string]probe.Info {
	if !probeable(lib.Kind) || !probe.Available() {
		return nil
	}

	out := make(map[string]probe.Info)
	for _, key := range batch {
		for _, c := range groups[key].files {
			if prev, ok := known[c.relPath]; ok && !prev.missing && prev.size == c.size && prev.mtime == c.mtime {
				continue // unchanged: never re-probe
			}

			abs := filepath.Join(lib.RootPath, filepath.FromSlash(c.relPath))
			info, err := probe.File(ctx, abs)
			if err != nil {
				res.addError(fmt.Sprintf("%s: %v", c.relPath, err))
				// Record the attempt with empty fields; the codecs stay empty
				// until the file changes again.
				out[c.relPath] = probe.Info{}
				continue
			}
			out[c.relPath] = info
		}
	}
	return out
}

// probeable reports whether a library kind holds files worth probing for codec
// facts. Only video and the audio kinds have media streams; ebooks, comics,
// photos, and games are never probed.
func probeable(kind string) bool {
	switch kind {
	case "video", "audiobooks", "music":
		return true
	default:
		return false
	}
}

// markItemMissing flags an item whose whole group has vanished. missing_since
// records when it first disappeared, and is written only on that transition: a
// rescan of an already-missing item must not refresh the age, or a file gone
// for months would look "just now missing" on every pass. An empty timestamp
// (a row that went missing before this column existed) is filled once, so the
// age is still real rather than permanently blank.
func markItemMissing(ctx context.Context, tx *sql.Tx, libraryID int64, key, now string) error {
	_, err := tx.ExecContext(ctx, `
		UPDATE media_items SET missing = 1, missing_since = ?
		WHERE library_id = ? AND path = ? AND (missing = 0 OR missing_since = '')`,
		now, libraryID, key)
	return err
}

// markFileMissing flags a file that vanished while its item survived. It obeys
// the same transition rule as markItemMissing: the timestamp is set once and
// never rewritten by a later scan.
func markFileMissing(ctx context.Context, tx *sql.Tx, rel, now string) error {
	_, err := tx.ExecContext(ctx, `
		UPDATE media_files SET missing = 1, missing_since = ?
		WHERE path = ? AND (missing = 0 OR missing_since = '')`,
		now, rel)
	return err
}

func sortedGroupKeys(groups map[string]group) []string {
	keys := make([]string, 0, len(groups))
	for k := range groups {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

func sortedCandidateKeys(found map[string]candidate) []string {
	keys := make([]string, 0, len(found))
	for k := range found {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return keys
}

func parseUnixNano(s string) int64 {
	var n int64
	for i := 0; i < len(s); i++ {
		if s[i] < '0' || s[i] > '9' {
			return 0
		}
		n = n*10 + int64(s[i]-'0')
	}
	return n
}