minstrel/internal/library/watcher.go

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