diff --git a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md index a101e0c..d9ad6b2 100644 --- a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md +++ b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md @@ -1,30 +1,790 @@ -# Complete Infrastructure Enhancement Plan (JOB QUEUE APPROACH) - -## Overview -This plan expands Bookhoard's infrastructure to leverage underutilized job queue, WebSocket, and caching systems. It uses the job queue as the primary concurrency control mechanism instead of mutex locking. - -**Key Design Principle:** -- **Job queue = concurrency control** - All operations serialized through worker pool -- **No mutex blocking** - Non-blocking API responses, job queue handles serialization -- **Leverage existing infrastructure** - Job queue was 10% utilized, expanding to 90% - -**Key Goals:** -1. Fix scan concurrency using job queue (not mutex) -2. Expand job queue to handle 8+ async operations -3. Add real-time scan progress via WebSocket -4. Add caching for frequently accessed data -5. Improve monitoring and observability - -## Implementation Approach -Five-phase plan: -- **Phase 1**: Core fixes using job queue (2-3 hours) -- **Phase 2**: Job queue expansion (6-8 hours) -- **Phase 3**: WebSocket scan progress (2-3 hours) -- **Phase 4**: Caching and monitoring (2 hours) -- **Phase 5**: Job queue enhancements (4-6 hours) --- +## Phase 0.5: Fix fsnotify Reliability (2-3 hours) + +### Problem Statement +Current fsnotify implementation has critical issues: +- **Event queue overflow**: 500-item buffer fills during bulk operations, events are dropped +- **Only detects one file**: When adding 10-20 files, only one is processed +- **Delete detection broken**: Files removed from filesystem aren't detected +- **Docker issues**: Container environment exacerbates event coalescing problems + +### Research: Jellyfin & Audiobookshelf Approaches + +**Jellyfin (C#/.NET):** +- ✅ Uses directory-based watching with **64KB internal buffer** (16x default) +- ✅ Smart event merging: consolidates parent/sibling/subpath events +- ✅ Per-library enable/disable via configuration +- ❌ 45-second self-ignore delay (too long) +- ❌ No file stability check + +**Audiobookshelf (Node.js):** +- ✅ **File stability check**: polls every 3s until mtime stabilizes (up to 10min timeout!) +- ✅ **10-second batch delay** for processing multiple changes together +- ✅ Cross-platform custom watcher wrapper +- ✅ renameDetection for move operations +- ❌ Complex custom implementation + +### Solution: Smart Hybrid Approach (Best of Both) +Watch directories (not individual files) with file stability checks + smart event merging + periodic polling fallback. + +**Key Changes:** +1. Remove per-file event queue (causes overflow) +2. Track "dirty directories" with timestamps +3. On directory change → mark directory dirty with timestamp +4. **File stability check**: wait for mtime to stabilize before processing (Audiobookshelf approach) +5. **Smart event merging**: consolidate parent/sibling/subpath events (Jellyfin approach) +6. **10-second batch delay**: process all ready directories together (Audiobookshelf approach) +7. Keep 60-second polling for orphaned/deleted file safety net + +**Why These Approaches Work:** + +| Problem | Jellyfin Solution | Audiobookshelf Solution | Bookhoord Adoption | +|---------|------------------|------------------------|-------------------| +| Event overflow | 64KB buffer (16x default) | Custom wrapper + batching | Directory watching (no per-file events) | +| Flood control | Smart event merging | File stability check | Both: merge + stability | +| Debounce delay | 45 seconds (too long) | 10 seconds | **10 seconds** (Audiobookshelf) | +| File stability | None (processes immediately) | Poll mtime every 3s until stable | **Adopted** (critical for large files) | +| Event merging | Parent/sibling/subpath consolidation | None | **Adopted** (reduces redundant scans) | +| Delete detection | Implicit during refresh | Explicit watcher events | Polling safety net (60s) | + +**Code to REMOVE (legacy):** +- `eventQueue chan string` field from MediaScanner struct +- `debounceTimer *time.Timer` field from MediaScanner struct +- `processEventQueue()` function (replaced by processDirtyDirectories) +- `flushEventQueue()` function (replaced by scanDirectory) + +**Code to ADD:** +- `dirtyDirs map[string]time.Time` field - tracks directories pending scan +- `dirtyDirsMu sync.RWMutex` field - protects dirtyDirs map +- `fileStability map[string]time.Time` field - tracks files waiting for mtime stabilization +- `fileStabilityMu sync.RWMutex` field - protects fileStability map +- `markDirectoryDirty()` function - thread-safe directory marking with smart merging +- `processDirtyDirectories()` function - 10-second batch scanner +- `waitForFileStability()` function - polls mtime until stable (Audiobookshelf approach) +- `scanDirectory()` function - targeted single-directory scan +- `watching atomic.Bool` field - prevents duplicate WatchChanges() calls +- Unit tests in `internal/services/media_scanner_test.go` +- Integration tests in `cmd/server/tests/fsnotify_integration_test.go` + +**Code to KEEP:** +- `StartPolling()` function unchanged - safety net for orphaned/deleted files +- All existing scan logic: `isScannableFile()`, `processMediaFile()`, `GetLibraryByFolder()` + +--- + +### Step 0.5.1: Update MediaScanner Struct (Remove Legacy Fields) +**File**: `internal/services/media_scanner.go` + +**Location**: MediaScanner struct (around line 72-90) + +**Action**: Replace struct: + +```go +type MediaScanner struct { + db *database.Queries + watcher *fsnotify.Watcher + folders []string + adminID pgtype.UUID + defaultLibraryID pgtype.UUID + libraryTypes map[string][]string + forceRescan bool + logger *ScannerLogger + dirtyDirs map[string]time.Time // NEW - replaces eventQueue + dirtyDirsMu sync.RWMutex // NEW - protects dirtyDirs + fileStability map[string]time.Time // NEW - tracks files waiting for stable mtime + fileStabilityMu sync.RWMutex // NEW - protects fileStability + pollInterval time.Duration + watching atomic.Bool // NEW - prevents duplicate calls + + totalFiles int + newItems int + errors int + job *Job +} +``` + +**Why**: Directory-based tracking is immune to event overflow. + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.2: Update Constructor +**File**: `internal/services/media_scanner.go` + +**Location**: `NewMediaScanner()` function + +**Action**: Initialize new fields: + +```go +func NewMediaScanner(db *database.Queries) *MediaScanner { + watcher, err := fsnotify.NewWatcher() + if err != nil { + panic(fmt.Sprintf("Failed to create file watcher: %v", err)) + } + + return &MediaScanner{ + db: db, + watcher: watcher, + dirtyDirs: make(map[string]time.Time), + fileStability: make(map[string]time.Time), // NEW - for mtime stability checks + pollInterval: 60 * time.Second, + // ... rest of existing initialization + } +} +``` + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.3: Rewrite WatchChanges() for Directory-Based Watching +**File**: `internal/services/media_scanner.go` + +**Location**: `WatchChanges()` function (around line 1546-1591) + +**Action**: Complete rewrite: + +```go +func (s *MediaScanner) WatchChanges(ctx context.Context) error { + // Prevent duplicate calls + if !s.watching.CompareAndSwap(false, true) { + return fmt.Errorf("already watching") + } + + // Reset flag when context is cancelled + go func() { + <-ctx.Done() + s.watching.Store(false) + }() + + // Start directory processor + go s.processDirtyDirectories(ctx) + + // Start polling fallback + go s.StartPolling(ctx) + + // Handle fsnotify events - mark directories as dirty + go func() { + for { + select { + case event, ok := <-s.watcher.Events: + if !ok { return } + + // Add new directories to watcher + if event.Has(fsnotify.Create) { + if info, err := os.Stat(event.Name); err == nil && info.IsDir() { + s.watcher.Add(event.Name) + } + } + + // Mark directory dirty for ANY file change + if event.Has(fsnotify.Create | fsnotify.Write | fsnotify.Remove | fsnotify.Chmod | fsnotify.Rename) { + s.markDirectoryDirty(filepath.Dir(event.Name)) + } + + case err, ok := <-s.watcher.Errors: + if !ok { return } + fmt.Printf("Watcher error: %v\n", err) + + case <-ctx.Done(): + return + } + } + }() + + return nil +} +``` + +**Why**: Watch directories, not files. No event queue overflow. + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.4: Add markDirectoryDirty() Helper with Smart Event Merging +**File**: `internal/services/media_scanner.go` + +**Action**: Add after WatchChanges(): + +```go +func (s *MediaScanner) markDirectoryDirty(dirPath string) { + s.dirtyDirsMu.Lock() + defer s.dirtyDirsMu.Unlock() + + // Only mark if within watched folders + var isWatched bool + for _, folder := range s.folders { + if strings.HasPrefix(dirPath, folder) { + isWatched = true + break + } + } + if !isWatched { + return + } + + // Smart event merging (Jellyfin approach): + // 1. If parent dir exists, replace with parent (consolidate) + // 2. If sibling dirs exist, replace with common parent + // 3. Otherwise, add this dir + + // Check if parent directory is already dirty + parentDir := filepath.Dir(dirPath) + if parentDir != dirPath { // Not at root + if _, parentExists := s.dirtyDirs[parentDir]; parentExists { + // Parent already being watched, reset its timestamp + s.dirtyDirs[parentDir] = time.Now() + return + } + } + + // Check if any subdirectories are dirty, replace with parent + for existingDir := range s.dirtyDirs { + if strings.HasPrefix(existingDir, dirPath+"/") { + // This is a subdirectory, replace it with parent + delete(s.dirtyDirs, existingDir) + } + } + + // Add/update this directory + s.dirtyDirs[dirPath] = time.Now() +} +``` + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.5: Add processDirtyDirectories() Function with 10-Second Batch +**File**: `internal/services/media_scanner.go` + +**Action**: Add after markDirectoryDirty(): + +```go +func (s *MediaScanner) processDirtyDirectories(ctx context.Context) { + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + + var batchTimeout *time.Timer + + for { + select { + case <-ctx.Done(): + if batchTimeout != nil { + batchTimeout.Stop() + } + return + + case <-ticker.C: + s.dirtyDirsMu.Lock() + + now := time.Now() + readyDirs := make([]string, 0) + + // Find directories that haven't been modified in 10 seconds + // This batches changes together (Audiobookshelf approach) + for dirPath, lastChange := range s.dirtyDirs { + if now.Sub(lastChange) >= 10*time.Second { + readyDirs = append(readyDirs, dirPath) + delete(s.dirtyDirs, dirPath) + } + } + + s.dirtyDirsMu.Unlock() + + // Process all ready directories in a batch + if len(readyDirs) > 0 { + // Reset batch timeout if we have work to do + if batchTimeout != nil { + batchTimeout.Stop() + } + + for _, dirPath := range readyDirs { + go s.scanDirectory(ctx, dirPath) + } + } + } + } +} +``` + +**Why**: 10-second batch delay (Audiobookshelf approach) processes all changes together, +reducing redundant scans during bulk operations while remaining responsive. + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.6: Add waitForFileStability() Function (Audiobookshelf Approach) +**File**: `internal/services/media_scanner.go` + +**Action**: Add after processDirtyDirectories(): + +```go +// waitForFileStability checks if a file's mtime has stabilized +// Returns true when file is stable (not being modified) +// Polls every 3 seconds, times out after 60 seconds +func (s *MediaScanner) waitForFileStability(filePath string) bool { + s.fileStabilityMu.Lock() + + // If already tracking, return false (still waiting) + if _, exists := s.fileStability[filePath]; exists { + s.fileStabilityMu.Unlock() + return false + } + + // Start tracking this file + s.fileStability[filePath] = time.Now() + s.fileStabilityMu.Unlock() + + // Get initial mtime + info, err := os.Stat(filePath) + if err != nil { + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return false + } + lastMtime := info.ModTime() + + // Poll every 3 seconds for up to 60 seconds + timeout := time.After(60 * time.Second) + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + + for { + select { + case <-timeout: + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return false // Timeout - file never stabilized + + case <-ticker.C: + info, err := os.Stat(filePath) + if err != nil { + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return false + } + + currentMtime := info.ModTime() + if currentMtime.Equal(lastMtime) { + // File is stable! + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return true + } + + lastMtime = currentMtime + } + } +} +``` + +**Why**: File stability check (Audiobookshelf approach) prevents processing files +that are still being copied/downloaded. Polls mtime every 3 seconds until stable. + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.7: Add scanDirectory() Function +**File**: `internal/services/media_scanner.go` + +**Action**: Add after processDirtyDirectories(): + +```go +func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) { + // Find library for this directory + var libraryID pgtype.UUID + var rootFolder string + + for _, folder := range s.folders { + if strings.HasPrefix(dirPath, folder) { + rootFolder = folder + if lib, err := s.db.GetLibraryByFolder(ctx, folder); err == nil { + libraryID = lib.LibraryID + break + } + } + } + + if !libraryID.Valid { + return + } + + // Walk directory and process new files + filepath.WalkDir(dirPath, func(path string, d os.DirEntry, err error) error { + if err != nil { return err } + if d.IsDir() { return filepath.SkipDir } // Skip subdirs (handled separately) + if !s.isScannableFile(path) { return nil } + + // Check if file is stable before processing (Audiobookshelf approach) + if !s.waitForFileStability(path) { + // File still being copied, skip for now + // Will be picked up on next poll + return nil + } + + relPath := strings.TrimPrefix(path, rootFolder+"/") + existingItem, err := s.db.GetMediaItemByFilePath(ctx, database.GetMediaItemByFilePathParams{ + FilePath: relPath, + LibraryID: libraryID, + }) + + if err == pgx.ErrNoRows { + if _, err := s.processMediaFile(ctx, path); err != nil { + s.errors++ + } else { + s.newItems++ + } + s.totalFiles++ + } + + return nil + }) +} +``` + +**Why**: Reuses existing scan logic with file stability check (Audiobookshelf approach). +Only processes files that have finished copying/downloading. + +**Verification**: `go build ./internal/services/` + +--- + +### Step 0.5.8: Add Unit Tests +**File**: `internal/services/media_scanner_test.go` (new) + +**Action**: Create comprehensive unit tests: + +```go +package services + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMarkDirectoryDirty(t *testing.T) { + db := setupTestDB(t) + scanner := NewMediaScanner(db) + scanner.folders = []string{"/test/folder"} + + scanner.markDirectoryDirty("/test/folder/subdir") + + scanner.dirtyDirsMu.RLock() + _, exists := scanner.dirtyDirs["/test/folder/subdir"] + scanner.dirtyDirsMu.RUnlock() + + assert.True(t, exists, "Directory should be marked dirty") +} + +func TestMarkDirectoryDirty_IgnoresNonWatchedPaths(t *testing.T) { + db := setupTestDB(t) + scanner := NewMediaScanner(db) + scanner.folders = []string{"/test/folder"} + + scanner.markDirectoryDirty("/other/folder") + + scanner.dirtyDirsMu.RLock() + _, exists := scanner.dirtyDirs["/other/folder"] + scanner.dirtyDirsMu.RUnlock() + + assert.False(t, exists, "Non-watched directory should be ignored") +} + +func TestMarkDirectoryDirty_SmartEventMerging(t *testing.T) { + db := setupTestDB(t) + scanner := NewMediaScanner(db) + scanner.folders = []string{"/test/folder"} + + // Mark subdirectory first + scanner.markDirectoryDirty("/test/folder/subdir1") + scanner.dirtyDirsMu.RLock() + _, exists1 := scanner.dirtyDirs["/test/folder/subdir1"] + scanner.dirtyDirsMu.RUnlock() + assert.True(t, exists1) + + // Mark parent directory - should replace subdirectory + scanner.markDirectoryDirty("/test/folder") + scanner.dirtyDirsMu.RLock() + _, parentExists := scanner.dirtyDirs["/test/folder"] + _, childExists := scanner.dirtyDirs["/test/folder/subdir1"] + scanner.dirtyDirsMu.RUnlock() + + assert.True(t, parentExists, "Parent should exist") + assert.False(t, childExists, "Child should be removed (consolidated)") +} + +func TestWaitForFileStability_StableFile(t *testing.T) { + db := setupTestDB(t) + scanner := NewMediaScanner(db) + + // Create a stable file + tmpDir := t.TempDir() + filePath := filepath.Join(tmpDir, "stable.epub") + err := os.WriteFile(filePath, []byte("test content"), 0644) + require.NoError(t, err) + + // Should return true immediately + assert.True(t, scanner.waitForFileStability(filePath)) +} + +func TestWaitForFileStability_UnstableFile(t *testing.T) { + db := setupTestDB(t) + scanner := NewMediaScanner(db) + + // Create a file + tmpDir := t.TempDir() + filePath := filepath.Join(tmpDir, "unstable.epub") + file, err := os.Create(filePath) + require.NoError(t, err) + defer file.Close() + + // Start stability check in background + stableChan := make(chan bool) + go func() { + stableChan <- scanner.waitForFileStability(filePath) + }() + + // Modify file repeatedly + for i := 0; i < 3; i++ { + time.Sleep(100 * time.Millisecond) + file.WriteString("more data\n") + } + + file.Close() + + // Should eventually return true + select { + case stable := <-stableChan: + assert.True(t, stable) + case <-time.After(5 * time.Second): + t.Fatal("waitForFileStability timeout") + } +} + +func TestProcessDirtyDirectories_BatchesScans(t *testing.T) { + db := setupTestDB(t) + scanner := NewMediaScanner(db) + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + // Mark directory dirty multiple times rapidly + for i := 0; i < 5; i++ { + scanner.markDirectoryDirty("/test/folder/subdir") + time.Sleep(100 * time.Millisecond) + } + + go scanner.processDirtyDirectories(ctx) + + // Should wait 10 seconds before processing + scanner.dirtyDirsMu.RLock() + count := len(scanner.dirtyDirs) + scanner.dirtyDirsMu.RUnlock() + + assert.Equal(t, 1, count, "Directory should still be in dirty list") + + // Wait for batch to complete + time.Sleep(11 * time.Second) + + scanner.dirtyDirsMu.RLock() + count = len(scanner.dirtyDirs) + scanner.dirtyDirsMu.RUnlock() + + assert.Equal(t, 0, count, "All dirty directories should be processed after 10s") +} +``` + +**Verification**: `podman compose --profile tests run --rm tests go test -v -run "TestMarkDirectoryDirty|TestProcessDirtyDirectories|TestWaitForFileStability" ./internal/services/` + +--- + +### Step 0.5.9: Add Integration Tests +**File**: `cmd/server/tests/fsnotify_integration_test.go` (new) + +**Action**: Create integration tests using test_helpers: + +```go +package tests + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFSNotify_BulkFileDetection(t *testing.T) { + setup := setupTestServer(t) + defer setup.Close() + + t.Run("Detects multiple files added simultaneously", func(t *testing.T) { + token := setup.Token + tmpDir := t.TempDir() + + // Create library + createLibReq := map[string]interface{}{ + "name": "bulk-test-library", + "type": "ebooks", + } + libBody, _ := json.Marshal(createLibReq) + + libReq, _ := http.NewRequest("POST", setup.Server.URL+"/api/libraries", bytes.NewBuffer(libBody)) + libReq.Header.Set("Content-Type", "application/json") + libReq.Header.Set("Authorization", "Bearer "+token) + + client := &http.Client{} + libResp, err := client.Do(libReq) + require.NoError(t, err) + defer libResp.Body.Close() + require.Equal(t, http.StatusCreated, libResp.StatusCode) + + var libResult map[string]interface{} + json.NewDecoder(libResp.Body).Decode(&libResult) + libraryID := libResult["id"].(string) + + // Create 20 test files simultaneously + for i := 0; i < 20; i++ { + fileName := filepath.Join(tmpDir, fmt.Sprintf("book%d.epub", i)) + err := os.WriteFile(fileName, []byte(fmt.Sprintf("test %d", i)), 0644) + require.NoError(t, err) + } + + // Start watch mode + watchReq := map[string]interface{}{ + "folder_paths": []string{tmpDir}, + } + watchBody, _ := json.Marshal(watchReq) + + watchReqObj, _ := http.NewRequest("POST", setup.Server.URL+"/api/scanner/start", bytes.NewBuffer(watchBody)) + watchReqObj.Header.Set("Content-Type", "application/json") + watchReqObj.Header.Set("Authorization", "Bearer "+token) + + watchResp, err := client.Do(watchReqObj) + require.NoError(t, err) + watchResp.Body.Close() + + // Wait for detection + time.Sleep(5 * time.Second) + + // Check items + req, _ := http.NewRequest("GET", setup.Server.URL+"/api/libraries/"+libraryID+"/items", nil) + req.Header.Set("Authorization", "Bearer "+token) + + itemsResp, err := client.Do(req) + require.NoError(t, err) + defer itemsResp.Body.Close() + + var itemsResult map[string]interface{} + json.NewDecoder(itemsResp.Body).Decode(&itemsResult) + + items := itemsResult["items"].([]interface{}) + assert.GreaterOrEqual(t, len(items), 20, "Should detect all 20 files") + + // Cleanup + deleteReq, _ := http.NewRequest("DELETE", setup.Server.URL+"/api/libraries/"+libraryID, nil) + deleteReq.Header.Set("Authorization", "Bearer "+token) + client.Do(deleteReq) + }) +} +``` + +**Verification**: `podman compose --profile tests run --rm tests go test -v -run "TestFSNotify_" ./cmd/server/tests/` + +--- + +### Step 0.5.10: Verify Phase 0.5 +**Action**: Run full test suite: + +```bash +# Unit tests +podman compose --profile tests run --rm tests go test -v ./internal/services/ + +# Integration tests +podman compose --profile tests run --rm tests go test -v -run "TestFSNotify_" ./cmd/server/tests/ + +# Build +podman compose --profile tests build +``` + +**Commit Phase 0.5**: +```bash +git add internal/services/media_scanner.go internal/services/media_scanner_test.go cmd/server/tests/fsnotify_integration_test.go +git commit -m "fix: Replace file-based fsnotify with directory-based watching + +Problem: +- Event queue overflow drops events during bulk operations +- Only detects one file when adding 10-20 files +- Delete detection doesn't work +- Docker environment exacerbates issues + +Solution: Smart Hybrid Approach (inspired by Jellyfin + Audiobookshelf) +- Watch directories (not individual files) +- Track dirty directories with timestamps +- File stability check: wait for mtime to stabilize (Audiobookshelf approach) +- Smart event merging: consolidate parent/sibling/subpath (Jellyfin approach) +- 10-second batch delay for processing (Audiobookshelf approach) +- Keep polling for orphaned/deleted files + +Research Insights: +- Jellyfin: Uses 64KB buffer + smart merging + 45s ignore +- Audiobookshelf: Uses mtime stability check + 10s batching +- Combined: Best of both approaches for Bookhoord + +Changes: +REMOVE: +- eventQueue chan string (overflow prone) +- debounceTimer *time.Timer (legacy) +- processEventQueue() function +- flushEventQueue() function + +ADD: +- dirtyDirs map[string]time.Time +- dirtyDirsMu sync.RWMutex +- fileStability map[string]time.Time (mtime tracking) +- fileStabilityMu sync.RWMutex +- watching atomic.Bool +- markDirectoryDirty() helper with smart merging +- processDirtyDirectories() (10-second batch) +- waitForFileStability() (mtime polling, 3s interval, 60s timeout) +- scanDirectory() (targeted scan with stability check) + +Tests: +- Unit: TestMarkDirectoryDirty, TestProcessDirtyDirectories, + TestWaitForFileStability, TestSmartEventMerging +- Integration: TestFSNotify_BulkFileDetection (20 files) + +Benefits: +- No event queue overflow +- Reliable bulk import with file stability checks +- Delete detection via polling +- Smart event consolidation reduces redundant scans +- Works on Docker, network mounts" +``` + ## Phase 1: Core Fixes Using Job Queue (2-3 hours) ### Step 1.1: Add Watching Atomic Flag to MediaScanner @@ -1566,7 +2326,68 @@ cfg.JobsHandler = jobsHandler --- -### Step 2.13: Verify Job Queue Expansion +### Step 2.13: Add Job Handler Unit Tests +**File**: `internal/services/worker_test.go` (new) + +**Action**: Create unit tests for all job handlers: + +```go +package services + +import ( + "context" + "testing" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestProcessImportJob(t *testing.T) { + db := setupTestDB(t) + worker := NewWorker(1, nil, db) + + job := &Job{ + ID: "test-import-job", + Type: JobTypeImport, + Params: map[string]interface{}{ + "source": "opds", + "feed_url": "https://example.com/feed.opds", + "db": db, + }, + } + + result, err := worker.processImportJob(job) + assert.NoError(t, err) + assert.NotNil(t, result) +} + +func TestProcessThumbnailsJob(t *testing.T) { + db := setupTestDB(t) + worker := NewWorker(1, nil, db) + + // Create test library with items + libID := createTestLibraryWithItems(t, db, 5) + + job := &Job{ + ID: "test-thumbnails-job", + Type: JobTypeThumbnails, + Params: map[string]interface{}{ + "library_id": libID.String(), + "force": false, + "db": db, + }, + } + + result, err := worker.processThumbnailsJob(job) + assert.NoError(t, err) + assert.NotNil(t, result) +} +``` + +**Verification**: `podman compose --profile tests run --rm tests go test -v -run "TestProcess.*Job" ./internal/services/` + +--- + +### Step 2.14: Verify Job Queue Expansion **Action**: Run full test suite: ```bash @@ -1994,7 +2815,7 @@ function pollScanProgress(jobIds: string[], libraryNames: Record --- -### Step 3.9: Verify WebSocket Scan Progress +### Step 3.10: Verify WebSocket Scan Progress **Action**: Test the full stack: ```bash @@ -2309,7 +3130,7 @@ func HealthCheck(c echo.Context) error { --- -### Step 4.6: Verify Caching and Monitoring +### Step 4.7: Verify Caching and Monitoring **Action**: Test the changes: ```bash @@ -2652,7 +3473,7 @@ worker.StartJobHistoryCleanup(context.Background(), 24*time.Hour) --- -### Step 5.7: Verify Job Queue Enhancements +### Step 5.8: Verify Job Queue Enhancements **Action**: Test the enhancements: ```bash