From c39a9ca18ff8948f32b43bbca07b8922b15967f1 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 21:43:27 -0400 Subject: [PATCH] feat(server): targeted ScanFiles + fsnotify library watcher Add Scanner.ScanFiles (watcher-driven targeted scan returning changed album IDs) and a recursive fsnotify Watcher that debounces filesystem events and enriches just the affected albums inline. Pure classifyEvent/drainPending seams unit-tested; ScanFiles covered in the scanner integration test. --- go.mod | 2 + go.sum | 2 + internal/library/scanner.go | 67 +++++++-- internal/library/scanner_test.go | 39 +++++ internal/library/watcher.go | 249 +++++++++++++++++++++++++++++++ internal/library/watcher_test.go | 59 ++++++++ 6 files changed, 407 insertions(+), 11 deletions(-) create mode 100644 internal/library/watcher.go create mode 100644 internal/library/watcher_test.go diff --git a/go.mod b/go.mod index 873b5d37..28a4ff66 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.25.0 require ( github.com/dhowden/tag v0.0.0-20240417053706-3d75831295e8 + github.com/fsnotify/fsnotify v1.10.1 github.com/go-chi/chi/v5 v5.2.5 github.com/go-co-op/gocron/v2 v2.21.2 github.com/golang-migrate/migrate/v4 v4.19.1 @@ -25,5 +26,6 @@ require ( github.com/robfig/cron/v3 v3.0.1 // indirect github.com/rogpeppe/go-internal v1.14.1 // indirect golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.44.0 // indirect golang.org/x/text v0.37.0 // indirect ) diff --git a/go.sum b/go.sum index 3ebffbc5..cae66c76 100644 --- a/go.sum +++ b/go.sum @@ -23,6 +23,8 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx59Ho= +github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo= github.com/go-chi/chi/v5 v5.2.5 h1:Eg4myHZBjyvJmAFjFvWgrqDTXFyOzjj7YIm3L3mu6Ug= github.com/go-chi/chi/v5 v5.2.5/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0= github.com/go-co-op/gocron/v2 v2.21.2 h1:bD8/YwkojYHgXFr3iEulL148KBdTbKVxUZzFKpXcdbY= diff --git a/internal/library/scanner.go b/internal/library/scanner.go index c6b7c6e7..49689651 100644 --- a/internal/library/scanner.go +++ b/internal/library/scanner.go @@ -87,7 +87,7 @@ func (s *Scanner) Scan(ctx context.Context, progressCb func(Stats)) (Stats, erro if !audioExtensions[strings.ToLower(filepath.Ext(path))] { return nil } - if err := s.scanFile(ctx, q, path, &stats); err != nil { + if _, _, err := s.scanFile(ctx, q, path, &stats); err != nil { s.logger.Warn("library scan file error", "path", path, "err", err) stats.Errored++ } @@ -114,19 +114,24 @@ func (s *Scanner) Scan(ctx context.Context, progressCb func(Stats)) (Stats, erro return stats, nil } -func (s *Scanner) scanFile(ctx context.Context, q *dbq.Queries, path string, stats *Stats) error { +// 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 fmt.Errorf("stat: %w", err) + 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 fmt.Errorf("lookup: %w", err) + 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. The second clause lets older scans that recorded @@ -135,18 +140,18 @@ func (s *Scanner) scanFile(ctx context.Context, q *dbq.Queries, path string, sta // scans short-circuit as before. if knownTrack && !existing.UpdatedAt.Time.Before(mtime) && existing.DurationMs > 0 { stats.Skipped++ - return nil + return pgtype.UUID{}, false, nil } f, err := os.Open(path) if err != nil { - return fmt.Errorf("open: %w", err) + return pgtype.UUID{}, false, fmt.Errorf("open: %w", err) } defer func() { _ = f.Close() }() meta, err := tag.ReadFrom(f) if err != nil { - return fmt.Errorf("tag read: %w", err) + return pgtype.UUID{}, false, fmt.Errorf("tag read: %w", err) } albumMBID, artistMBID := extractMBIDs(meta) recordingMBID := extractRecordingMBID(meta) @@ -166,11 +171,11 @@ func (s *Scanner) scanFile(ctx context.Context, q *dbq.Queries, path string, sta artist, err := s.resolveArtist(ctx, q, artistName, artistMBID) if err != nil { - return fmt.Errorf("artist: %w", err) + 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 fmt.Errorf("album: %w", err) + return pgtype.UUID{}, false, fmt.Errorf("album: %w", err) } trackNum, _ := meta.Track() @@ -213,7 +218,7 @@ func (s *Scanner) scanFile(ctx context.Context, q *dbq.Queries, path string, sta track, err := q.UpsertTrack(ctx, params) if err != nil { - return fmt.Errorf("upsert track: %w", err) + 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 { @@ -227,7 +232,47 @@ func (s *Scanner) scanFile(ctx context.Context, q *dbq.Queries, path string, sta } else { stats.Added++ } - return nil + 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) { diff --git a/internal/library/scanner_test.go b/internal/library/scanner_test.go index 80f8df54..f894fa34 100644 --- a/internal/library/scanner_test.go +++ b/internal/library/scanner_test.go @@ -121,6 +121,45 @@ func TestScanner_Integration(t *testing.T) { if stats2.Added != 0 || stats2.Updated != 0 || stats2.Skipped != 4 { t.Errorf("incremental scan stats = %+v, want Skipped=4", stats2) } + + // ScanFiles (watcher-driven targeted scan): a newly-added file is picked + // up by path without walking the whole tree, and its album ID is returned + // so the watcher can enrich just that album. A non-audio path in the batch + // is silently ignored. + newFile := filepath.Join(root, "artistX/albumA/03.mp3") + writeTestMP3(t, newFile, map[string]string{ + "TIT2": "Track Three", "TPE1": "Artist X", "TALB": "Album A", "TRCK": "3", "TYER": "2020", + }) + notAudio := filepath.Join(root, "artistX/albumA/cover.txt") + if err := os.WriteFile(notAudio, []byte("nope"), 0o644); err != nil { + t.Fatalf("write non-audio: %v", err) + } + changed, err := scanner.ScanFiles(ctx, []string{newFile, notAudio}) + if err != nil { + t.Fatalf("ScanFiles: %v", err) + } + if len(changed) != 1 || !changed[0].Valid { + t.Fatalf("ScanFiles changed albums = %+v, want exactly 1 valid", changed) + } + + // The targeted scan upserts only the named file; the rest of the library + // is untouched (no full walk). + if _, err := q.GetTrackByPath(ctx, newFile); err != nil { + t.Fatalf("GetTrackByPath after ScanFiles: %v", err) + } + + // Re-running ScanFiles on the now-unchanged file (duration probed) skips it, + // so no album is reported as changed. + if _, err := pool.Exec(ctx, "UPDATE tracks SET duration_ms = 1000 WHERE duration_ms = 0"); err != nil { + t.Fatalf("seed durations (2): %v", err) + } + changed2, err := scanner.ScanFiles(ctx, []string{newFile}) + if err != nil { + t.Fatalf("ScanFiles (rescan): %v", err) + } + if len(changed2) != 0 { + t.Errorf("ScanFiles on unchanged file changed = %+v, want none", changed2) + } } // writeTestMP3 emits a file with only an ID3v2.3 tag payload plus a few diff --git a/internal/library/watcher.go b/internal/library/watcher.go new file mode 100644 index 00000000..94a66900 --- /dev/null +++ b/internal/library/watcher.go @@ -0,0 +1,249 @@ +package library + +import ( + "context" + "io/fs" + "log/slog" + "os" + "path/filepath" + "strings" + "time" + + "github.com/fsnotify/fsnotify" + + "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" +) + +// watcherDebounce is the quiet period after the last filesystem event before +// the watcher processes the pending batch. Copying an album emits a burst of +// write events; waiting for quiet coalesces the whole album into one scan. +const watcherDebounce = 3 * time.Second + +// watchAction is the pure decision for a single filesystem event. +type watchAction int + +const ( + watchIgnore watchAction = iota // not a relevant change + watchEnqueueFile // an audio file was created/written + watchScanDir // a directory appeared; scan it recursively +) + +// classifyEvent decides what to do with a filesystem event, separated from I/O +// so it is unit-testable. Only Create/Write matter (removals are not pruned +// here -- see the watcher's design non-goals). A newly-created directory is +// scanned recursively because fsnotify does not emit events for files that +// already existed inside a directory that was moved/copied in wholesale. +func classifyEvent(op fsnotify.Op, isDir, isAudio bool) watchAction { + if op&(fsnotify.Create|fsnotify.Write) == 0 { + return watchIgnore + } + if isDir { + if op&fsnotify.Create != 0 { + return watchScanDir + } + return watchIgnore + } + if isAudio { + return watchEnqueueFile + } + return watchIgnore +} + +// Watcher delivers near-instant library updates by watching the scan roots +// with fsnotify and scanning just the changed files (plus inline cover-art +// enrichment for the affected albums), instead of waiting for a full scan. +// It is recursive: a watch is added per directory, and new directories get a +// watch as they appear. inotify watch-limit exhaustion on huge libraries is +// logged, not fatal -- the periodic safety-net scan covers anything missed. +type Watcher struct { + scanner *Scanner + enricher *coverart.Enricher + logger *slog.Logger + roots []string + debounce time.Duration +} + +func NewWatcher( + scanner *Scanner, enricher *coverart.Enricher, logger *slog.Logger, roots []string, +) *Watcher { + return &Watcher{ + scanner: scanner, + enricher: enricher, + logger: logger, + roots: roots, + debounce: watcherDebounce, + } +} + +// Run blocks until ctx is cancelled. Returns a non-nil error only when the +// fsnotify watcher cannot be created at all; per-path watch failures and +// fsnotify errors are logged and tolerated. +func (w *Watcher) Run(ctx context.Context) error { + if len(w.roots) == 0 { + w.logger.Info("library watcher: no scan paths configured, watcher disabled") + return nil + } + fsw, err := fsnotify.NewWatcher() + if err != nil { + w.logger.Error("library watcher: create failed", "err", err) + return err + } + defer func() { _ = fsw.Close() }() + + for _, root := range w.roots { + w.addRecursive(fsw, root) + } + w.logger.Info("library watcher: started", "roots", len(w.roots)) + + batches := make(chan []string, batchQueueDepth) + go w.processLoop(ctx, batches) + w.eventLoop(ctx, fsw, batches) + return nil +} + +// eventLoop reads fsnotify events, coalesces them via a debounce timer, and +// hands drained batches to the process loop. Returns on ctx cancellation. +func (w *Watcher) eventLoop(ctx context.Context, fsw *fsnotify.Watcher, batches chan<- []string) { + pending := make(map[string]struct{}) + timer := time.NewTimer(w.debounce) + timer.Stop() + var timerC <-chan time.Time + for { + select { + case <-ctx.Done(): + timer.Stop() + w.logger.Info("library watcher: stopped") + return + case event, ok := <-fsw.Events: + if !ok { + return + } + if w.handleEvent(fsw, event, pending) { + timer.Reset(w.debounce) + timerC = timer.C + } + case err, ok := <-fsw.Errors: + if ok { + w.logger.Warn("library watcher: fsnotify error", "err", err) + } + case <-timerC: + timerC = nil + paths := drainPending(pending) + if len(paths) == 0 { + continue + } + select { + case batches <- paths: + case <-ctx.Done(): + return + } + } + } +} + +// handleEvent applies one event to the pending set and reports whether the +// debounce timer should be (re)armed. +func (w *Watcher) handleEvent( + fsw *fsnotify.Watcher, event fsnotify.Event, pending map[string]struct{}, +) bool { + info, err := os.Stat(event.Name) + isDir := err == nil && info.IsDir() + isAudio := audioExtensions[strings.ToLower(filepath.Ext(event.Name))] + switch classifyEvent(event.Op, isDir, isAudio) { + case watchScanDir: + w.addRecursive(fsw, event.Name) + return w.enqueueDirAudio(event.Name, pending) + case watchEnqueueFile: + pending[event.Name] = struct{}{} + return true + case watchIgnore: + return false + default: + return false + } +} + +// processLoop serializes batch processing so a slow remote enrichment never +// blocks the fsnotify event loop. +func (w *Watcher) processLoop(ctx context.Context, batches <-chan []string) { + for { + select { + case <-ctx.Done(): + return + case paths, ok := <-batches: + if !ok { + return + } + w.process(ctx, paths) + } + } +} + +// process scans the changed files then enriches just the affected albums. +func (w *Watcher) process(ctx context.Context, paths []string) { + changed, err := w.scanner.ScanFiles(ctx, paths) + if err != nil { + w.logger.Warn("library watcher: scan batch failed", "err", err) + } + if w.enricher != nil { + for _, albumID := range changed { + if ctx.Err() != nil { + return + } + if eerr := w.enricher.EnrichAlbum(ctx, albumID); eerr != nil { + w.logger.Warn("library watcher: enrich album failed", "err", eerr) + } + } + } + if len(changed) > 0 { + w.logger.Info("library watcher: applied", "files", len(paths), "albums", len(changed)) + } +} + +// addRecursive adds an fsnotify watch for root and every subdirectory under it. +// Per-directory failures (inotify limits, permissions) are logged and skipped. +func (w *Watcher) addRecursive(fsw *fsnotify.Watcher, root string) { + _ = filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil || !d.IsDir() { + return nil //nolint:nilerr // tolerate unreadable subtrees + } + if aerr := fsw.Add(path); aerr != nil { + w.logger.Warn("library watcher: add watch failed", "path", path, "err", aerr) + } + return nil + }) +} + +// enqueueDirAudio enqueues every audio file under dir (used when a directory +// appears wholesale, since fsnotify emits no events for its existing children). +func (w *Watcher) enqueueDirAudio(dir string, pending map[string]struct{}) bool { + added := false + _ = filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil || d.IsDir() { + return nil //nolint:nilerr // tolerate unreadable subtrees + } + if audioExtensions[strings.ToLower(filepath.Ext(path))] { + pending[path] = struct{}{} + added = true + } + return nil + }) + return added +} + +// drainPending returns the pending paths and clears the set. +func drainPending(pending map[string]struct{}) []string { + if len(pending) == 0 { + return nil + } + paths := make([]string, 0, len(pending)) + for p := range pending { + paths = append(paths, p) + } + clear(pending) + return paths +} + +// batchQueueDepth bounds how many debounced batches can be in flight before the +// event loop would block; generous since each batch is processed quickly. +const batchQueueDepth = 16 diff --git a/internal/library/watcher_test.go b/internal/library/watcher_test.go new file mode 100644 index 00000000..b971e480 --- /dev/null +++ b/internal/library/watcher_test.go @@ -0,0 +1,59 @@ +package library + +import ( + "sort" + "testing" + + "github.com/fsnotify/fsnotify" +) + +func TestClassifyEvent(t *testing.T) { + tests := []struct { + name string + op fsnotify.Op + isDir bool + isAudio bool + want watchAction + }{ + {"create audio file enqueues", fsnotify.Create, false, true, watchEnqueueFile}, + {"write audio file enqueues", fsnotify.Write, false, true, watchEnqueueFile}, + {"create non-audio file ignored", fsnotify.Create, false, false, watchIgnore}, + {"create directory scans it", fsnotify.Create, true, false, watchScanDir}, + {"write on directory ignored", fsnotify.Write, true, false, watchIgnore}, + {"remove audio file ignored (no pruning)", fsnotify.Remove, false, true, watchIgnore}, + {"rename audio file ignored", fsnotify.Rename, false, true, watchIgnore}, + {"chmod audio file ignored", fsnotify.Chmod, false, true, watchIgnore}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := classifyEvent(tt.op, tt.isDir, tt.isAudio); got != tt.want { + t.Fatalf("classifyEvent(%v, dir=%v, audio=%v) = %v, want %v", + tt.op, tt.isDir, tt.isAudio, got, tt.want) + } + }) + } +} + +func TestDrainPending(t *testing.T) { + t.Run("empty set returns nil", func(t *testing.T) { + if got := drainPending(map[string]struct{}{}); got != nil { + t.Fatalf("drainPending(empty) = %v, want nil", got) + } + }) + + t.Run("returns paths and clears the set", func(t *testing.T) { + pending := map[string]struct{}{ + "/music/a.mp3": {}, + "/music/b.flac": {}, + } + got := drainPending(pending) + sort.Strings(got) + want := []string{"/music/a.mp3", "/music/b.flac"} + if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Fatalf("drainPending = %v, want %v", got, want) + } + if len(pending) != 0 { + t.Fatalf("pending not cleared: %v", pending) + } + }) +}