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