Compare commits

...
6 Commits
5 changed files with 426 additions and 34 deletions
+9
View File
@@ -123,6 +123,15 @@ golangci-lint run
- **Error handling:** Always check errors explicitly; use structured error wrapping with `fmt.Errorf("context: %w", err)` - **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 - **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 ### Logging
Use **zerolog** for all logging: Use **zerolog** for all logging:
+1
View File
@@ -14,6 +14,7 @@ This repository contains AI-agent oriented project context.
- Logging: zerolog - Logging: zerolog
- Error wrapping: `fmt.Errorf("context: %w", err)` - Error wrapping: `fmt.Errorf("context: %w", err)`
- Prefer small, focused changes. - 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 ## Areas to know
+16 -20
View File
@@ -1,35 +1,31 @@
FROM debian:trixie-slim FROM alpine:latest
LABEL authors="Belphemur" LABEL authors="Belphemur"
ARG TARGETPLATFORM ARG TARGETPLATFORM
ARG APP_PATH=/usr/local/bin/CBZOptimizer ARG APP_PATH=/usr/local/bin/CBZOptimizer
ENV USER=abc ENV USER=abc
ENV CONFIG_FOLDER=/config ENV CONFIG_FOLDER=/config
ENV PUID=99 ENV PUID=99
ENV DEBIAN_FRONTEND=noninteractive # libwebp-tools (cwebp) is installed via apk below into /usr/bin; point go-webpbin
# at it directly so it doesn't try to download a glibc prebuilt binary.
ENV VENDOR_PATH=/usr/bin
RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \ RUN adduser \
--mount=type=cache,target=/var/lib/apt,sharing=locked \ -S \
apt-get update && apt-get install -y --no-install-recommends adduser && \ -D \
addgroup --system users && \ -H \
adduser \ -h "${CONFIG_FOLDER}" \
--system \ -u "${PUID}" \
--home "${CONFIG_FOLDER}" \ -G users \
--uid "${PUID}" \ -s /bin/bash \
--ingroup users \ "${USER}"
--disabled-password \
"${USER}" && \
apt-get purge -y --auto-remove adduser
COPY ${TARGETPLATFORM}/CBZOptimizer ${APP_PATH} COPY ${TARGETPLATFORM}/CBZOptimizer ${APP_PATH}
RUN --mount=type=cache,target=/var/cache/apt,sharing=locked \ RUN apk add --no-cache \
--mount=type=cache,target=/var/lib/apt,sharing=locked \
apt-get update && \
apt-get full-upgrade -y && \
apt-get install -y --no-install-recommends \
bash \ bash \
ca-certificates \ ca-certificates \
bash-completion && \ bash-completion \
libwebp-tools && \
chmod +x ${APP_PATH} && \ chmod +x ${APP_PATH} && \
${APP_PATH} completion bash > /etc/bash_completion.d/CBZOptimizer.bash ${APP_PATH} completion bash > /etc/bash_completion.d/CBZOptimizer.bash
+228 -14
View File
@@ -1,12 +1,15 @@
package commands package commands
import ( import (
"errors"
"fmt" "fmt"
"io/fs" "io/fs"
"os" "os"
"path/filepath" "path/filepath"
"runtime" "runtime"
"strings" "strings"
"sync"
"time"
utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils" utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils"
"github.com/belphemur/CBZOptimizer/v2/pkg/converter" "github.com/belphemur/CBZOptimizer/v2/pkg/converter"
@@ -17,6 +20,12 @@ import (
"github.com/spf13/viper" "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() { func init() {
if runtime.GOOS != "linux" { if runtime.GOOS != "linux" {
return return
@@ -58,12 +67,12 @@ func WatchCommand(_ *cobra.Command, args []string) error {
converterType := constant.FindConversionFormat(viper.GetString("format")) converterType := constant.FindConversionFormat(viper.GetString("format"))
chapterConverter, err := converter.Get(converterType) chapterConverter, err := converter.Get(converterType)
if err != nil { if err != nil {
return fmt.Errorf("failed to get chapterConverter: %v", err) return fmt.Errorf("failed to get chapterConverter: %w", err)
} }
err = chapterConverter.PrepareConverter() err = chapterConverter.PrepareConverter()
if err != nil { 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") 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 { if err != nil {
return fmt.Errorf("failed to create file watcher: %w", err) 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 { if err := addRecursiveWatch(watcher, path); err != nil {
return fmt.Errorf("failed to watch path %s: %w", path, err) return fmt.Errorf("failed to watch path %s: %w", path, err)
} }
backfillExistingArchives(path, debouncer.Trigger)
for { for {
select { select {
@@ -91,6 +119,11 @@ func WatchCommand(_ *cobra.Command, args []string) error {
if err := addRecursiveWatch(watcher, event.Name); err != nil { if err := addRecursiveWatch(watcher, event.Name); err != nil {
log.Error().Err(err).Str("path", event.Name).Msg("Failed to watch created directory") 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 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: case err, ok := <-watcher.Errors:
if !ok { if !ok {
return nil 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 { func addRecursiveWatch(watcher *fsnotify.Watcher, rootPath string) error {
return filepath.WalkDir(rootPath, func(path string, entry fs.DirEntry, err error) error { return filepath.WalkDir(rootPath, func(path string, entry fs.DirEntry, err error) error {
if err != nil { if err != nil {
return err log.Warn().Err(err).Str("path", path).Msg("Skipping path while setting up watch")
return nil
} }
if !entry.IsDir() { if !entry.IsDir() {
return nil return nil
} }
if err := watcher.Add(path); err != 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 fmt.Errorf("failed to watch directory %s: %w", path, err)
} }
return nil 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 { func shouldProcessWatchEvent(event fsnotify.Event) bool {
return event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Rename) 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) filename := strings.ToLower(path)
return strings.HasSuffix(filename, ".cbz") || strings.HasSuffix(filename, ".cbr") 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()
}
@@ -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))
}