feat(api): library sources, scanner and catalog (M2)

- schema for sources, path patterns, media files, artists, albums,
  tracks, genres (migration 00003)
- path patterns (RE2 with named groups, skip), merge with tags by
  source priority, fallbacks from the file name and folder, name
  normalization and multi-artist split
- scanner: walk, skip unchanged files by size and mtime, ffprobe in 8
  workers with a single catalog writer, renames by content hash keep
  the track, duplicates fold into one track with the best file as
  primary, missing files are kept 30 days
- storage-agent mounts NFS with validated server, export and an option
  whitelist, no shell; test mount counts audio files
- admin API: sources CRUD, connection test, patterns with preview,
  scan, library status, file errors, host folder picker; scheduled
  scans and health checks; mounts are restored on start
- catalog API: artist, artist albums, album with tracks, track
- api image is alpine with ffmpeg for ffprobe

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

services/api/internal/library/scanner.go

@@ -0,0 +1,504 @@
+package library
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/cespare/xxhash/v2"
+	"github.com/google/uuid"
+
+	"pipu/api/internal/library/meta"
+)
+
+// UnknownAlbum — альбом файла без тегов в корне источника.
+const UnknownAlbum = "Unknown album"
+
+// MissingRetention — сколько пропавший файл хранится, прежде чем трек удалится.
+const MissingRetention = 30 * 24 * time.Hour
+
+// Progress — состояние скана одного источника (для админки).
+type Progress struct {
+	SourceID   uuid.UUID
+	Running    bool
+	Full       bool
+	StartedAt  time.Time
+	FinishedAt *time.Time
+	Found      int // аудиофайлов найдено обходом
+	ToProcess  int // требуют чтения
+	Processed  int
+	Added      int
+	Updated    int
+	Moved      int
+	Missing    int
+	Errors     int
+	Error      string // фатальная ошибка скана
+}
+
+// ProgressBoard хранит прогресс сканов в памяти процесса.
+type ProgressBoard struct {
+	mu sync.Mutex
+	m  map[uuid.UUID]*Progress
+}
+
+// NewProgressBoard создаёт пустую доску.
+func NewProgressBoard() *ProgressBoard { return &ProgressBoard{m: map[uuid.UUID]*Progress{}} }
+
+// Get — копия прогресса источника.
+func (b *ProgressBoard) Get(id uuid.UUID) (Progress, bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	p, ok := b.m[id]
+	if !ok {
+		return Progress{}, false
+	}
+	return *p, true
+}
+
+// All — копии прогресса всех источников.
+func (b *ProgressBoard) All() []Progress {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	out := make([]Progress, 0, len(b.m))
+	for _, p := range b.m {
+		out = append(out, *p)
+	}
+	return out
+}
+
+// start регистрирует новый скан; false — скан этого источника уже идёт.
+func (b *ProgressBoard) start(id uuid.UUID, full bool, now time.Time) bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if p, ok := b.m[id]; ok && p.Running {
+		return false
+	}
+	b.m[id] = &Progress{SourceID: id, Running: true, Full: full, StartedAt: now}
+	return true
+}
+
+func (b *ProgressBoard) update(id uuid.UUID, fn func(*Progress)) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if p, ok := b.m[id]; ok {
+		fn(p)
+	}
+}
+
+// Scanner сканирует источники и обновляет каталог.
+type Scanner struct {
+	store   *Store
+	prober  Prober
+	log     *slog.Logger
+	board   *ProgressBoard
+	workers int
+	now     func() time.Time
+	// AfterScan вызывается после каждого успешного скана (переиндексация поиска).
+	AfterScan func(ctx context.Context)
+}
+
+// NewScanner создаёт сканер с 8 воркерами чтения.
+func NewScanner(store *Store, prober Prober, board *ProgressBoard, log *slog.Logger) *Scanner {
+	return &Scanner{store: store, prober: prober, log: log, board: board, workers: 8, now: func() time.Time { return time.Now().UTC() }}
+}
+
+// walked — аудиофайл, найденный обходом.
+type walked struct {
+	rel   string
+	size  int64
+	mtime time.Time
+	hash  []byte
+}
+
+// job — файл, который нужно прочитать.
+type job struct {
+	walked
+	fileID   uuid.UUID  // Nil — новый
+	oldTrack *uuid.UUID // трек, к которому файл был привязан
+}
+
+type probed struct {
+	job
+	probe Probe
+	err   error
+}
+
+// ErrScanRunning — скан источника уже идёт.
+var ErrScanRunning = errors.New("library: scan already running")
+
+// Start запускает скан в фоне. root — каталог источника в файловой системе api.
+func (sc *Scanner) Start(src Source, root string, full bool) error {
+	if !sc.board.start(src.ID, full, sc.now()) {
+		return ErrScanRunning
+	}
+	go func() {
+		ctx := context.Background()
+		err := sc.run(ctx, src, root, full)
+		finished := sc.now()
+		sc.board.update(src.ID, func(p *Progress) {
+			p.Running, p.FinishedAt = false, &finished
+			if err != nil {
+				p.Error = err.Error()
+			}
+		})
+		if err != nil {
+			sc.log.Error("scan failed", "source_id", src.ID, "err", err)
+			return
+		}
+		if sc.AfterScan != nil {
+			sc.AfterScan(ctx)
+		}
+	}()
+	return nil
+}
+
+func (sc *Scanner) run(ctx context.Context, src Source, root string, full bool) error {
+	start := sc.now()
+	rows, err := sc.store.Patterns(ctx, src.ID)
+	if err != nil {
+		return err
+	}
+	var exprs []string
+	for _, r := range rows {
+		if r.IsEnabled {
+			exprs = append(exprs, r.Regex)
+		}
+	}
+	patterns, err := meta.CompilePatterns(exprs)
+	if err != nil {
+		return fmt.Errorf("library: patterns: %w", err)
+	}
+
+	existing, err := sc.store.Files(ctx, src.ID)
+	if err != nil {
+		return err
+	}
+	found, err := walk(root, func(rel string) bool { return meta.MatchPath(patterns, rel).Skip })
+	if err != nil {
+		return err
+	}
+	sc.board.update(src.ID, func(p *Progress) { p.Found = len(found) })
+
+	var (
+		jobs     []job
+		newFiles []walked
+		seen     = make(map[string]bool, len(found))
+	)
+	for _, w := range found {
+		seen[w.rel] = true
+		f, ok := existing[w.rel]
+		switch {
+		case !ok:
+			newFiles = append(newFiles, w)
+		case !full && f.Size == w.size && f.Mtime.Equal(w.mtime):
+			if f.Missing {
+				if err := sc.store.SetPresent(ctx, f.ID); err != nil {
+					return err
+				}
+			}
+		default:
+			jobs = append(jobs, job{walked: w, fileID: f.ID, oldTrack: f.TrackID})
+		}
+	}
+	var gone []FileRow
+	for rel, f := range existing {
+		if !seen[rel] {
+			gone = append(gone, f)
+		}
+	}
+
+	// Переименования: новый путь с тем же содержимым, что у пропавшего файла.
+	if len(gone) > 0 {
+		for i := range newFiles {
+			if h, err := contentHash(filepath.Join(root, filepath.FromSlash(newFiles[i].rel)), newFiles[i].size); err == nil {
+				newFiles[i].hash = h
+			}
+		}
+	}
+	moves, rest, stillGone := matchRenames(newFiles, gone)
+	for _, mv := range moves {
+		if err := sc.store.MoveFile(ctx, mv.from.ID, mv.to.rel, mv.to.size, mv.to.mtime); err != nil {
+			return err
+		}
+	}
+	for _, w := range rest {
+		jobs = append(jobs, job{walked: w})
+	}
+	sc.board.update(src.ID, func(p *Progress) { p.ToProcess, p.Moved = len(jobs), len(moves) })
+
+	if err := sc.process(ctx, src, root, patterns, jobs); err != nil {
+		return err
+	}
+
+	var goneIDs []uuid.UUID
+	affected := map[uuid.UUID]bool{}
+	for _, f := range stillGone {
+		goneIDs = append(goneIDs, f.ID)
+		if f.TrackID != nil {
+			affected[*f.TrackID] = true
+		}
+	}
+	if err := sc.store.MarkMissing(ctx, goneIDs, start); err != nil {
+		return err
+	}
+	if _, err := sc.store.PurgeMissing(ctx, src.ID, start.Add(-MissingRetention)); err != nil {
+		return err
+	}
+	for id := range affected {
+		if err := sc.store.RefreshPrimaryFile(ctx, id); err != nil {
+			return err
+		}
+	}
+	sc.board.update(src.ID, func(p *Progress) { p.Missing = len(goneIDs) })
+	if err := sc.store.CleanupOrphans(ctx); err != nil {
+		return err
+	}
+	return sc.store.SetLastScan(ctx, src.ID, sc.now())
+}
+
+// process читает файлы параллельно, а каталог обновляет один писатель —
+// так дубли и новые исполнители не создаются дважды в гонке.
+func (sc *Scanner) process(ctx context.Context, src Source, root string, patterns []meta.Pattern, jobs []job) error {
+	in := make(chan job)
+	out := make(chan probed)
+	var wg sync.WaitGroup
+	for range max(1, sc.workers) {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := range in {
+				abs := filepath.Join(root, filepath.FromSlash(j.rel))
+				p, err := sc.prober.Probe(ctx, abs)
+				if j.hash == nil {
+					j.hash, _ = contentHash(abs, j.size)
+				}
+				out <- probed{job: j, probe: p, err: err}
+			}
+		}()
+	}
+	go func() {
+		for _, j := range jobs {
+			in <- j
+		}
+		close(in)
+		wg.Wait()
+		close(out)
+	}()
+
+	var firstErr error
+	for r := range out {
+		if firstErr != nil {
+			continue // дочитываем канал, чтобы воркеры завершились
+		}
+		added, err := sc.apply(ctx, src, patterns, r)
+		if err != nil {
+			firstErr = err
+			continue
+		}
+		sc.board.update(src.ID, func(p *Progress) {
+			p.Processed++
+			switch {
+			case r.err != nil:
+				p.Errors++
+			case added:
+				p.Added++
+			default:
+				p.Updated++
+			}
+		})
+	}
+	return firstErr
+}
+
+// apply записывает прочитанный файл в каталог. added — файл новый.
+func (sc *Scanner) apply(ctx context.Context, src Source, patterns []meta.Pattern, r probed) (bool, error) {
+	data := FileData{RelPath: r.rel, Size: r.size, Mtime: r.mtime, ContentHash: r.hash, Probe: r.probe}
+	added := r.fileID == uuid.Nil
+	if r.err != nil {
+		data.Probe = Probe{Format: strings.TrimPrefix(strings.ToLower(filepath.Ext(r.rel)), ".")}
+		data.ScanError = r.err.Error()
+		if _, err := sc.store.SaveFile(ctx, src.ID, r.fileID, data, nil); err != nil {
+			return false, err
+		}
+		if r.oldTrack != nil {
+			return added, sc.store.RefreshPrimaryFile(ctx, *r.oldTrack)
+		}
+		return added, nil
+	}
+
+	pm := meta.MatchPath(patterns, r.rel)
+	m := meta.Merge(r.probe.Tags, meta.FromPath(pm.Fields), src.TagsFirst, r.rel)
+	if m.Album == "" {
+		m.Album = UnknownAlbum
+	}
+	albumArtist, err := sc.store.EnsureArtist(ctx, m.AlbumArtist)
+	if err != nil {
+		return false, err
+	}
+	albumID, err := sc.store.EnsureAlbum(ctx, albumArtist, m.Album, m.Year)
+	if err != nil {
+		return false, err
+	}
+	in := TrackInput{
+		AlbumID: albumID, Title: m.Title, Disc: m.Disc, Track: m.Track, Year: m.Year, DurationMs: r.probe.DurationMs,
+	}
+	trackID, dup, err := sc.store.FindTrack(ctx, in)
+	if err != nil {
+		return false, err
+	}
+	if !dup {
+		// Перечитанный файл сохраняет свой трек (лайки, история); новый — получает новый.
+		if r.oldTrack != nil {
+			trackID = *r.oldTrack
+		}
+		if trackID, err = sc.store.SaveTrack(ctx, trackID, in); err != nil {
+			return false, err
+		}
+		credits := meta.SplitArtists(m.Artist, func(k string) bool { return sc.store.ArtistKeyExists(ctx, k) })
+		ids := make([]uuid.UUID, 0, len(credits))
+		roles := make([]string, 0, len(credits))
+		for _, c := range credits {
+			id, err := sc.store.EnsureArtist(ctx, c.Name)
+			if err != nil {
+				return false, err
+			}
+			ids, roles = append(ids, id), append(roles, c.Role)
+		}
+		if err := sc.store.SetTrackArtists(ctx, trackID, ids, roles); err != nil {
+			return false, err
+		}
+		if err := sc.store.SetTrackGenres(ctx, trackID, m.Genres); err != nil {
+			return false, err
+		}
+	}
+	if _, err := sc.store.SaveFile(ctx, src.ID, r.fileID, data, &trackID); err != nil {
+		return false, err
+	}
+	if err := sc.store.RefreshPrimaryFile(ctx, trackID); err != nil {
+		return false, err
+	}
+	if r.oldTrack != nil && *r.oldTrack != trackID {
+		if err := sc.store.RefreshPrimaryFile(ctx, *r.oldTrack); err != nil {
+			return false, err
+		}
+	}
+	return added, nil
+}
+
+// walk обходит дерево источника и возвращает аудиофайлы. Скрытые файлы и каталоги пропускаются;
+// skip — пути, которые шаблоны велят игнорировать. Ошибки отдельных каталогов не прерывают обход.
+func walk(root string, skip func(rel string) bool) ([]walked, error) {
+	if _, err := os.Stat(root); err != nil {
+		return nil, fmt.Errorf("library: source root: %w", err)
+	}
+	var out []walked
+	err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+		if err != nil {
+			if p == root {
+				return err
+			}
+			return nil
+		}
+		name := d.Name()
+		if p != root && strings.HasPrefix(name, ".") {
+			if d.IsDir() {
+				return filepath.SkipDir
+			}
+			return nil
+		}
+		if d.IsDir() || !IsAudio(name) {
+			return nil
+		}
+		rel, err := filepath.Rel(root, p)
+		if err != nil {
+			return nil
+		}
+		rel = filepath.ToSlash(rel)
+		if skip(rel) {
+			return nil
+		}
+		info, err := d.Info()
+		if err != nil {
+			return nil
+		}
+		// PostgreSQL хранит время с точностью до микросекунд.
+		out = append(out, walked{rel: rel, size: info.Size(), mtime: info.ModTime().UTC().Truncate(time.Microsecond)})
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("library: walk: %w", err)
+	}
+	return out, nil
+}
+
+const hashChunk = 1 << 20
+
+// contentHash — xxhash64 от размера, первого и последнего мегабайта: быстро даже по NFS
+// и не меняется при переименовании.
+func contentHash(abs string, size int64) ([]byte, error) {
+	f, err := os.Open(abs)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = f.Close() }()
+	h := xxhash.New()
+	var sz [8]byte
+	binary.LittleEndian.PutUint64(sz[:], uint64(size))
+	_, _ = h.Write(sz[:])
+	if _, err := io.CopyN(h, f, hashChunk); err != nil && !errors.Is(err, io.EOF) {
+		return nil, err
+	}
+	if size > 2*hashChunk {
+		if _, err := f.Seek(-hashChunk, io.SeekEnd); err != nil {
+			return nil, err
+		}
+		if _, err := io.CopyN(h, f, hashChunk); err != nil && !errors.Is(err, io.EOF) {
+			return nil, err
+		}
+	}
+	return h.Sum(nil), nil
+}
+
+type move struct {
+	from FileRow
+	to   walked
+}
+
+// matchRenames сопоставляет новые пути с пропавшими файлами по содержимому (хэш и размер).
+// Каждый пропавший файл используется не больше одного раза.
+func matchRenames(newFiles []walked, gone []FileRow) (moves []move, rest []walked, stillGone []FileRow) {
+	byHash := map[string][]int{}
+	for i, f := range gone {
+		if len(f.ContentHash) > 0 {
+			k := fmt.Sprintf("%x:%d", f.ContentHash, f.Size)
+			byHash[k] = append(byHash[k], i)
+		}
+	}
+	used := make([]bool, len(gone))
+	for _, w := range newFiles {
+		k := fmt.Sprintf("%x:%d", w.hash, w.size)
+		if len(w.hash) > 0 && len(byHash[k]) > 0 {
+			i := byHash[k][0]
+			byHash[k] = byHash[k][1:]
+			used[i] = true
+			moves = append(moves, move{from: gone[i], to: w})
+			continue
+		}
+		rest = append(rest, w)
+	}
+	for i, f := range gone {
+		if !used[i] {
+			stillGone = append(stillGone, f)
+		}
+	}
+	return moves, rest, stillGone
+}