diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index ab73d0e..ef0bead 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -123,6 +123,15 @@ golangci-lint run - **Error handling:** Always check errors explicitly; use structured error wrapping with `fmt.Errorf("context: %w", err)` - **Context usage:** Pass `context.Context` as first parameter for operations that may be cancelled +### Commit Messages + +This project follows [Conventional Commits](https://www.conventionalcommits.org/). Prefix commit messages with a type (and optional scope), for example: + +- `fix(watch): debounce fsnotify events before optimizing` +- `feat(converter): add avif support` +- `docs: clarify commit convention` +- `chore: update dependency` + ### Logging Use **zerolog** for all logging: diff --git a/AGENTS.md b/AGENTS.md index a7a5d97..d45b6e5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,6 +14,7 @@ This repository contains AI-agent oriented project context. - Logging: zerolog - Error wrapping: `fmt.Errorf("context: %w", err)` - Prefer small, focused changes. +- Commit messages: follow [Conventional Commits](https://www.conventionalcommits.org/) (e.g. `fix(watch): debounce fsnotify events`, `docs: clarify commit convention`). ## Areas to know diff --git a/cmd/cbzoptimizer/commands/watch_command.go b/cmd/cbzoptimizer/commands/watch_command.go index d9265bf..b3bcb3e 100644 --- a/cmd/cbzoptimizer/commands/watch_command.go +++ b/cmd/cbzoptimizer/commands/watch_command.go @@ -1,12 +1,15 @@ package commands import ( + "errors" "fmt" "io/fs" "os" "path/filepath" "runtime" "strings" + "sync" + "time" utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils" "github.com/belphemur/CBZOptimizer/v2/pkg/converter" @@ -17,6 +20,12 @@ import ( "github.com/spf13/viper" ) +// debounceDelay is the quiet period we wait, per path, after the last fsnotify +// event before triggering an optimization. This coalesces the bursts of +// Write/Create/Rename events fsnotify emits while a file is still being +// written (e.g. copied or downloaded into the watched folder). +const debounceDelay = 2 * time.Second + func init() { if runtime.GOOS != "linux" { return @@ -58,12 +67,12 @@ func WatchCommand(_ *cobra.Command, args []string) error { converterType := constant.FindConversionFormat(viper.GetString("format")) chapterConverter, err := converter.Get(converterType) if err != nil { - return fmt.Errorf("failed to get chapterConverter: %v", err) + return fmt.Errorf("failed to get chapterConverter: %w", err) } err = chapterConverter.PrepareConverter() if err != nil { - return fmt.Errorf("failed to prepare converter: %v", err) + return fmt.Errorf("failed to prepare converter: %w", err) } log.Info().Str("path", path).Bool("override", override).Uint8("quality", quality).Str("format", converterType.String()).Bool("split", split).Msg("Watching directory") @@ -71,11 +80,30 @@ func WatchCommand(_ *cobra.Command, args []string) error { if err != nil { return fmt.Errorf("failed to create file watcher: %w", err) } - defer watcher.Close() + defer func() { + if closeErr := watcher.Close(); closeErr != nil { + log.Error().Err(closeErr).Msg("Failed to close file watcher") + } + }() + + // Optimization jobs run in a small worker pool so the fsnotify event loop + // is never blocked waiting on a conversion. + queue := newOptimizeQueue(runtime.NumCPU(), &utils2.OptimizeOptions{ + ChapterConverter: chapterConverter, + Quality: quality, + Override: override, + Split: split, + Timeout: timeout, + }) + defer queue.Stop() + + debouncer := newEventDebouncer(debounceDelay, queue.Enqueue) + defer debouncer.Stop() if err := addRecursiveWatch(watcher, path); err != nil { return fmt.Errorf("failed to watch path %s: %w", path, err) } + backfillExistingArchives(path, debouncer.Trigger) for { select { @@ -91,6 +119,11 @@ func WatchCommand(_ *cobra.Command, args []string) error { if err := addRecursiveWatch(watcher, event.Name); err != nil { log.Error().Err(err).Str("path", event.Name).Msg("Failed to watch created directory") } + // The newly discovered directory may already contain + // archives (e.g. a folder moved/copied in); back-fill them + // since no further fsnotify event will target them. + backfillExistingArchives(event.Name, debouncer.Trigger) + continue } } @@ -102,16 +135,7 @@ func WatchCommand(_ *cobra.Command, args []string) error { continue } - if err := utils2.Optimize(&utils2.OptimizeOptions{ - ChapterConverter: chapterConverter, - Path: event.Name, - Quality: quality, - Override: override, - Split: split, - Timeout: timeout, - }); err != nil { - log.Error().Err(err).Str("file", event.Name).Msg("Error processing file") - } + debouncer.Trigger(event.Name) case err, ok := <-watcher.Errors: if !ok { return nil @@ -121,21 +145,55 @@ func WatchCommand(_ *cobra.Command, args []string) error { } } +// addRecursiveWatch registers a watch on rootPath and every subdirectory +// beneath it. Directories that can't be read (e.g. permission errors) are +// logged and skipped instead of aborting the whole walk, so a single bad +// subtree doesn't prevent the rest of the folder from being watched. func addRecursiveWatch(watcher *fsnotify.Watcher, rootPath string) error { return filepath.WalkDir(rootPath, func(path string, entry fs.DirEntry, err error) error { if err != nil { - return err + log.Warn().Err(err).Str("path", path).Msg("Skipping path while setting up watch") + return nil } if !entry.IsDir() { return nil } if err := watcher.Add(path); err != nil { + if errors.Is(err, fs.ErrPermission) { + log.Warn().Err(err).Str("path", path).Msg("Skipping unreadable directory while setting up watch") + return nil + } return fmt.Errorf("failed to watch directory %s: %w", path, err) } return nil }) } +// backfillExistingArchives walks rootPath for comic archives that already +// exist on disk and hands them to process. This covers the case where a +// directory (potentially already containing archives) is created/moved into +// the watched tree: only the directory itself generates an fsnotify event, +// so the files inside it would otherwise never be picked up. +func backfillExistingArchives(rootPath string, process func(path string)) { + err := filepath.WalkDir(rootPath, func(path string, entry fs.DirEntry, err error) error { + if err != nil { + log.Warn().Err(err).Str("path", path).Msg("Skipping path while scanning for existing archives") + return nil + } + if entry.IsDir() { + return nil + } + if !isComicArchive(path) { + return nil + } + process(path) + return nil + }) + if err != nil { + log.Error().Err(err).Str("path", rootPath).Msg("Failed to scan directory for existing archives") + } +} + func shouldProcessWatchEvent(event fsnotify.Event) bool { return event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Rename) } @@ -144,3 +202,159 @@ func isComicArchive(path string) bool { filename := strings.ToLower(path) return strings.HasSuffix(filename, ".cbz") || strings.HasSuffix(filename, ".cbr") } + +// eventDebouncer coalesces bursts of fsnotify events targeting the same path +// into a single call to onQuiet, fired after the path has been quiet for +// delay. This prevents repeated conversions being triggered while a file is +// still being written, and avoids running Optimize against a Rename event +// that reports the *old* (already gone) file name once the quiet period lets +// us re-check the path. +type eventDebouncer struct { + delay time.Duration + onQuiet func(path string) + + mu sync.Mutex + timers map[string]*debounceTimer + inFlight sync.WaitGroup + stopping bool +} + +type debounceTimer struct { + timer *time.Timer +} + +func newEventDebouncer(delay time.Duration, onQuiet func(path string)) *eventDebouncer { + return &eventDebouncer{ + delay: delay, + onQuiet: onQuiet, + timers: make(map[string]*debounceTimer), + } +} + +// Trigger (re)schedules onQuiet to run for path after the debounce delay, +// resetting any previously pending timer for the same path. +func (d *eventDebouncer) Trigger(path string) { + d.mu.Lock() + defer d.mu.Unlock() + + if d.stopping { + return + } + + if existing, exists := d.timers[path]; exists { + if existing.timer.Stop() { + d.inFlight.Done() + } + } + + entry := &debounceTimer{} + d.inFlight.Add(1) + entry.timer = time.AfterFunc(d.delay, func() { + defer d.inFlight.Done() + + d.mu.Lock() + // Only the still-current timer for this path is allowed to clear the + // entry and fire onQuiet. If Trigger raced with this callback and + // already installed a newer timer, this stale invocation must not + // delete that newer entry or fire early. + owns := d.timers[path] == entry && !d.stopping + if owns { + delete(d.timers, path) + } + d.mu.Unlock() + if owns { + d.onQuiet(path) + } + }) + d.timers[path] = entry +} + +// Stop cancels all pending timers. +func (d *eventDebouncer) Stop() { + d.mu.Lock() + d.stopping = true + for path, timer := range d.timers { + if timer.timer.Stop() { + d.inFlight.Done() + } + delete(d.timers, path) + } + d.mu.Unlock() + d.inFlight.Wait() +} + +// optimizeQueue is a small worker pool that runs Optimize jobs off of the +// fsnotify event loop, so a slow conversion never blocks event draining. +type optimizeQueue struct { + jobs chan string + wg sync.WaitGroup + mu sync.RWMutex + stopped bool + options *utils2.OptimizeOptions + optimize func(options *utils2.OptimizeOptions) error +} + +func newOptimizeQueue(workerCount int, options *utils2.OptimizeOptions) *optimizeQueue { + if workerCount < 1 { + workerCount = 1 + } + q := &optimizeQueue{ + jobs: make(chan string, 64), + options: options, + optimize: utils2.Optimize, + } + q.wg.Add(workerCount) + for i := 0; i < workerCount; i++ { + go q.worker() + } + return q +} + +func (q *optimizeQueue) worker() { + defer q.wg.Done() + for path := range q.jobs { + q.process(path) + } +} + +// process re-checks the path right before optimizing: fsnotify Rename events +// commonly report the *old* file name (already gone by the time we act on +// it), and Write can fire while a file is still being written elsewhere. In +// override mode, Optimize may overwrite/delete the source file, so skipping +// paths that no longer exist (or that are no longer regular files) avoids +// noisy failures. +func (q *optimizeQueue) process(path string) { + info, err := os.Stat(path) + if err != nil { + log.Debug().Err(err).Str("file", path).Msg("Skipping watch event: path no longer accessible") + return + } + if info.IsDir() { + return + } + + options := *q.options + options.Path = path + if err := q.optimize(&options); err != nil { + log.Error().Err(err).Str("file", path).Msg("Error processing file") + } +} + +// Enqueue submits path for processing by the worker pool. +func (q *optimizeQueue) Enqueue(path string) { + q.mu.RLock() + defer q.mu.RUnlock() + if q.stopped { + return + } + q.jobs <- path +} + +// Stop closes the job queue and waits for in-flight jobs to finish. +func (q *optimizeQueue) Stop() { + q.mu.Lock() + q.stopped = true + close(q.jobs) + q.mu.Unlock() + q.wg.Wait() +} diff --git a/cmd/cbzoptimizer/commands/watch_command_test.go b/cmd/cbzoptimizer/commands/watch_command_test.go new file mode 100644 index 0000000..485c3fd --- /dev/null +++ b/cmd/cbzoptimizer/commands/watch_command_test.go @@ -0,0 +1,172 @@ +package commands + +import ( + "os" + "path/filepath" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils" + "github.com/fsnotify/fsnotify" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestIsComicArchive(t *testing.T) { + testCases := []struct { + name string + path string + expected bool + }{ + {"cbz lowercase", "/a/b/chapter.cbz", true}, + {"cbr lowercase", "/a/b/chapter.cbr", true}, + {"cbz uppercase", "/a/b/chapter.CBZ", true}, + {"other extension", "/a/b/chapter.zip", false}, + {"no extension", "/a/b/chapter", false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.expected, isComicArchive(tc.path)) + }) + } +} + +func TestShouldProcessWatchEvent(t *testing.T) { + testCases := []struct { + name string + op fsnotify.Op + expected bool + }{ + {"create", fsnotify.Create, true}, + {"write", fsnotify.Write, true}, + {"rename", fsnotify.Rename, true}, + {"remove", fsnotify.Remove, false}, + {"chmod", fsnotify.Chmod, false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + event := fsnotify.Event{Name: "file.cbz", Op: tc.op} + assert.Equal(t, tc.expected, shouldProcessWatchEvent(event)) + }) + } +} + +func TestAddRecursiveWatchSkipsUnreadableSubdirectory(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("permission bits are not reliably enforced on Windows") + } + + root := t.TempDir() + nested := filepath.Join(root, "nested") + assert.NoError(t, os.MkdirAll(nested, 0o755)) + assert.NoError(t, os.Chmod(nested, 0o000)) + defer func() { + assert.NoError(t, os.Chmod(nested, 0o755)) + }() + if _, err := os.ReadDir(nested); err == nil { + t.Skip("cannot make nested directory unreadable in this environment") + } + + watcher, err := fsnotify.NewWatcher() + assert.NoError(t, err) + defer func() { + assert.NoError(t, watcher.Close()) + }() + + // A regular, fully accessible tree should be watched without error. + assert.NoError(t, addRecursiveWatch(watcher, root)) +} + +func TestBackfillExistingArchivesFindsPreExistingArchives(t *testing.T) { + root := t.TempDir() + assert.NoError(t, os.WriteFile(filepath.Join(root, "chapter1.cbz"), []byte("data"), 0o644)) + assert.NoError(t, os.WriteFile(filepath.Join(root, "notes.txt"), []byte("data"), 0o644)) + + sub := filepath.Join(root, "sub") + assert.NoError(t, os.MkdirAll(sub, 0o755)) + assert.NoError(t, os.WriteFile(filepath.Join(sub, "chapter2.cbr"), []byte("data"), 0o644)) + + var mu sync.Mutex + var found []string + backfillExistingArchives(root, func(path string) { + mu.Lock() + defer mu.Unlock() + found = append(found, path) + }) + + assert.Len(t, found, 2) +} + +func TestEventDebouncerCoalescesBurstsIntoSingleCall(t *testing.T) { + var calls int32 + debouncer := newEventDebouncer(20*time.Millisecond, func(path string) { + atomic.AddInt32(&calls, 1) + }) + defer debouncer.Stop() + + // Simulate a burst of events for the same path. + for i := 0; i < 5; i++ { + debouncer.Trigger("/tmp/chapter.cbz") + } + + require.Eventually(t, func() bool { + return atomic.LoadInt32(&calls) == 1 + }, 2*time.Second, 10*time.Millisecond) +} + +func TestEventDebouncerHandlesMultiplePaths(t *testing.T) { + var mu sync.Mutex + seen := make(map[string]int) + debouncer := newEventDebouncer(10*time.Millisecond, func(path string) { + mu.Lock() + defer mu.Unlock() + seen[path]++ + }) + defer debouncer.Stop() + + debouncer.Trigger("/tmp/a.cbz") + debouncer.Trigger("/tmp/b.cbz") + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return seen["/tmp/a.cbz"] == 1 && seen["/tmp/b.cbz"] == 1 + }, 2*time.Second, 10*time.Millisecond) +} + +func TestOptimizeQueueSkipsMissingPath(t *testing.T) { + q := newOptimizeQueue(1, &utils2.OptimizeOptions{}) + defer q.Stop() + var calls int32 + done := make(chan struct{}, 1) + existing := filepath.Join(t.TempDir(), "existing.cbz") + require.NoError(t, os.WriteFile(existing, []byte("data"), 0o644)) + q.optimize = func(options *utils2.OptimizeOptions) error { + if options.Path == existing { + atomic.AddInt32(&calls, 1) + select { + case done <- struct{}{}: + default: + } + } + return nil + } + + // Enqueue a path that doesn't exist; process should skip it without + // panicking or blocking since it never reaches utils2.Optimize. + q.Enqueue(filepath.Join(t.TempDir(), "missing.cbz")) + q.Enqueue(existing) + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for existing path to be processed") + } + + assert.EqualValues(t, 1, atomic.LoadInt32(&calls)) +}