Merge pull request #207 from Belphemur/copilot/fix-comments-in-pr-204

fix(watch): debounce fsnotify events, backfill archives, resilient recursive watch

.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:

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
 

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{
+			debouncer.Trigger(event.Name)
-				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")
-			}
 		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()
+}

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))
+}