diff --git a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md deleted file mode 100644 index 97d6f85..0000000 --- a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md +++ /dev/null @@ -1,3488 +0,0 @@ - ---- - -## Phase 0.5: Fix fsnotify Reliability with Directory Watching + Job Queue Concurrency (3-4 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 -- **No concurrency control**: Multiple scans can run simultaneously, causing conflicts -- **File stability race condition**: Multiple goroutines can check same file simultaneously -- **Unbounded goroutine spawn**: Each directory scan spawns goroutine without limit - -### 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: Clean Phase 0.5 + Phase 1 Robustness -Watch directories (not individual files) with file stability checks + smart event merging + periodic polling fallback. Use job queue ONLY for concurrency control (serialize scans), not for scanning logic. - -**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. **Job queue for all scans**: directory scans submitted as jobs, serialized by worker pool -8. **JobTypeSetFolders**: Async folder configuration via job queue -9. **Atomic file stability tracking**: Prevents race conditions with concurrent checks -10. 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]*atomic.Bool` field - tracks files waiting for mtime stabilization (pointer to atomic.Bool, not value type) -- `fileStabilityMu sync.RWMutex` field - protects fileStability map -- `scan_mutex sync.Mutex` field - prevents concurrent scans (simple, effective, works with job queue) -- `watching atomic.Bool` field - prevents duplicate WatchChanges() calls -- `markDirectoryDirty()` function - thread-safe directory marking with smart merging -- `processDirtyDirectories()` function - 10-second batch scanner, submits jobs to global WorkerInstance -- `waitForFileStability()` function - polls mtime until stable (uses *atomic.Bool, no race condition) -- `performInitialScan()` function - scans all root folders on startup (submitted as jobs, not direct calls) -- `JobTypeDirectoryScan` constant - for directory scanning via job queue -- `processDirectoryScanJob()` function in Worker - handles directory scan jobs -- `WorkerInstance *Worker` global variable - provides access to worker for MediaScanner (no circular dependency) -- `Enqueue()` method in Worker - non-blocking job submission with safe fallback -- `JobTypeSetFolders` constant - for async folder configuration -- `processSetFoldersJob()` function - handles folder configuration asynchronously -- Test isolation in setupTestServer() - snapshot/restore system_settings -- Fix GetPollInterval() default value: 30s → 60s (matches real-time expectations) -- 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]*atomic.Bool // NEW - tracks files waiting (atomic prevents races) - fileStabilityMu sync.RWMutex // NEW - protects fileStability - scan_mutex sync.Mutex // NEW - prevents concurrent directory scans (simple, effective) - 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]*atomic.Bool), - pollInterval: 60 * time.Second, - watching: atomic.Bool{}, - // ... rest of existing initialization - } -} -``` - -**Why**: Directory-based tracking is immune to event overflow. Fixed default value inconsistency. - -**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) - }() - - // Perform initial scan of all root folders - go s.performInitialScan(ctx) - - // 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.3.5: Add JobTypeSetFolders to Worker -**File**: `internal/services/worker.go` - -**Location**: JobType constants (lines 24-28) - -**Current code**: -```go -const ( - JobTypeScan JobType = "scan" -) -``` - -**Action**: Add new job type: - -```go -const ( - JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" - JobTypeDirectoryScan JobType = "directory_scan" // NEW - Phase 0.5 directory scanning -) -``` - -**Why**: Enables folder configuration changes to go through the job queue instead of blocking API handlers. Prevents deadlocks and provides better UX. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 0.5.3.6: Add processSetFoldersJob Handler -**File**: `internal/services/worker.go` - -**Location**: In the switch statement in `processJob()` (around line 127-132) - -**Current code**: -```go -switch job.Type { -case JobTypeScan: - result, err = w.processScanJob(job) -default: - err = fmt.Errorf("unknown job type: %s", job.Type) -} -``` - -**Action**: Add handler for new job type: - -```go -switch job.Type { -case JobTypeScan: - result, err = w.processScanJob(job) -case JobTypeSetFolders: - result, err = w.processSetFoldersJob(job) -case JobTypeDirectoryScan: - result, err = w.processDirectoryScanJob(job) -default: - err = fmt.Errorf("unknown job type: %s", job.Type) -} -``` - -**Then add the handler function** (after `processScanJob()`, around line 248): - -```go -func (w *Worker) processSetFoldersJob(job *Job) (interface{}, error) { - // Extract parameters - foldersParam, ok := job.Params["folders"] - if !ok { - return nil, fmt.Errorf("folders parameter required") - } - - folders, ok := foldersParam.([]string) - if !ok { - return nil, fmt.Errorf("folders must be a string array") - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - // Create scanner and configure folders - scanner := NewMediaScanner(db) - if err := scanner.SetFolders(folders); err != nil { - return nil, fmt.Errorf("failed to set folders: %w", err) - } - - // Return success result - return map[string]interface{}{ - "message": "folders configured successfully", - "folders": folders, - }, nil -} - -func (w *Worker) processDirectoryScanJob(job *Job) (interface{}, error) { - // Extract parameters - directoryParam, ok := job.Params["directory"] - if !ok { - return nil, fmt.Errorf("directory parameter required") - } - - directory, ok := directoryParam.(string) - if !ok { - return nil, fmt.Errorf("directory must be a string") - } - - dbParam, ok := job.Params["db"] - if !ok { - return nil, fmt.Errorf("db parameter required") - } - - db, ok := dbParam.(*database.Queries) - if !ok { - return nil, fmt.Errorf("db must be *database.Queries") - } - - // Create temporary scanner instance for this job - scanner := NewMediaScanner(db) - - // Call scanDirectory() directly - // Job queue provides concurrency control - no need for activeScans map - ctx := context.Background() - scanner.scanDirectory(ctx, directory) - - // Return scan results - return map[string]interface{}{ - "message": fmt.Sprintf("Scanned directory: %s", directory), - "totalFiles": scanner.totalFiles, - "newItems": scanner.newItems, - "errors": scanner.errors, - }, nil -} -``` - -**Why**: -- Makes SetFolders() async via job queue -- Non-blocking API responses -- Folder changes wait behind scans naturally -- User gets job ID for status tracking -- No deadlocks or blocking - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 0.5.3.7: Add WorkerInstance Global and Enqueue Method -**File**: `internal/services/worker.go` - -**Location**: After the JobType constants (around line 24-28) - -**Action**: Add global variable and helper method: - -```go -var WorkerInstance *Worker - -func (w *Worker) Enqueue(job *Job) { - select { - case w.jobQueue <- job: - default: - fmt.Printf("Worker queue full, rejecting job: %s\n", job.ID) - } -} -``` - -**Then in main() or worker initialization**: -```go -worker := NewWorker(/* params */) -WorkerInstance = worker -go worker.Start(context.Background()) -``` - -**Why**: -- Provides global access to worker for MediaScanner -- Enqueue() is non-blocking with safe fallback -- No circular dependency between scanner and worker -- Clean separation of concerns - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 0.5.3.8: Initialize WorkerInstance in main() -**File**: `cmd/server/main.go` - -**Location**: In main() function, where worker is created (around line 50-100) - -**Action**: Initialize global WorkerInstance after creating worker: - -```go -// Find existing worker creation -worker := NewWorker(numWorkers, connManager, db) - -// Add this line immediately after -WorkerInstance = worker - -// Existing code -go worker.Start(ctx) -``` - -**Why**: MediaScanner needs access to worker for submitting directory scan jobs. Global variable is set during initialization, avoiding circular dependency between scanner and worker. - -**Verification**: Run `podman compose build` to ensure server starts correctly. - ---- - -### 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) - } - } - - // NEW: Check for sibling directories and consolidate to parent - parentDir = filepath.Dir(dirPath) - for existingDir := range s.dirtyDirs { - existingParent := filepath.Dir(existingDir) - if existingParent == parentDir && existingParent != dirPath && existingParent != "." { - // Found a sibling! Both should be replaced with parent - delete(s.dirtyDirs, existingDir) - s.dirtyDirs[parentDir] = time.Now() - return - } - } - - // Add/update this directory - s.dirtyDirs[dirPath] = time.Now() -} -``` - -**Verification**: `go build ./internal/services/` - ---- - -### Step 0.5.5: Add processDirtyDirectories() Function with Job Queue Integration -**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() - - for { - select { - case <-ctx.Done(): - 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 via job queue - // Job queue serializes scans - prevents concurrent directory access - if len(readyDirs) > 0 { - for _, dirPath := range readyDirs { - // Create directory scan job with correct params for processDirectoryScanJob() - job := &Job{ - ID: uuid.New().String(), - Type: JobTypeDirectoryScan, - Params: map[string]interface{}{ - "directory": dirPath, - "db": s.db, - }, - Status: JobStatusPending, - } - - // Enqueue via global worker singleton - if WorkerInstance != nil { - WorkerInstance.Enqueue(job) - fmt.Printf("Enqueued directory scan job: %s\n", dirPath) - } else { - fmt.Printf("Warning: Worker not initialized, skipping directory scan: %s\n", dirPath) - } - } - } - } - } -} -``` - -**Why**: 10-second batch delay (Audiobookshelf approach) processes all changes together, -reducing redundant scans during bulk operations. Job queue integration ensures: -- No concurrent directory scans (serialized by worker pool) -- No unbounded goroutine spawn -- Natural backpressure when worker queue is full -- Clean separation: job queue handles concurrency, scanner handles logic - -**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 -// Uses atomic.Bool to prevent race conditions with concurrent checks -func (s *MediaScanner) waitForFileStability(filePath string) bool { - s.fileStabilityMu.Lock() - - // Check if already being checked (atomic.Bool prevents race condition) - tracking, exists := s.fileStability[filePath] - if exists { - s.fileStabilityMu.Unlock() - // Another goroutine is already checking this file - if tracking.Load() { - return false // Still being checked - } - // Tracking exists but completed, remove stale entry - s.fileStabilityMu.Lock() - delete(s.fileStability, filePath) - } - - // Start tracking with atomic.Bool set to true (checking in progress) - trackingFlag := &atomic.Bool{} - trackingFlag.Store(true) - s.fileStability[filePath] = trackingFlag - s.fileStabilityMu.Unlock() - - // Get initial mtime - info, err := os.Stat(filePath) - if err != nil { - // Clean up tracking entry if file doesn't exist - 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: - // Timeout - mark as done and clean up - trackingFlag.Store(false) - s.fileStabilityMu.Lock() - delete(s.fileStability, filePath) - s.fileStabilityMu.Unlock() - return false // File never stabilized - - case <-ticker.C: - info, err := os.Stat(filePath) - if err != nil { - // File deleted - mark as done and clean up - trackingFlag.Store(false) - s.fileStabilityMu.Lock() - delete(s.fileStability, filePath) - s.fileStabilityMu.Unlock() - return false - } - - currentMtime := info.ModTime() - if currentMtime.Equal(lastMtime) { - // File is stable! Mark as done and clean up - trackingFlag.Store(false) - 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. - -**FIXES**: -- Uses atomic.Bool to prevent race condition when multiple goroutines check same file -- First goroutine to check sets atomic.Bool to true -- Subsequent goroutines see true and return false immediately -- No double-unlock bug (no defer unlock, explicit cleanup only) -- Entries cleaned up on timeout, error, or success (no memory leaks) -- Lock released during polling to allow concurrent checks for different files - -**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) { - // Prevent concurrent scans of ANY directory - // Simple mutex is enough - job queue already serializes by directory - s.scan_mutex.Lock() - defer s.scan_mutex.Unlock() - - // 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 - } - } - } - - // Check if libraryID is valid before proceeding - 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 } - if !s.isScannableFile(path) { return nil } - - // Check if file is stable before processing (Audiobookshelf approach) - if !s.waitForFileStability(path) { - 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. Simple mutex provides -double-protection against concurrent scans, even though job queue already serializes. - -**Verification**: `go build ./internal/services/` - ---- - -### Step 0.5.8: Add performInitialScan() Function -**File**: `internal/services/media_scanner.go` - -**Action**: Add after scanDirectory(): - -```go -// performInitialScan scans all root folders on startup -// This ensures existing files are detected before watching begins -func (s *MediaScanner) performInitialScan(ctx context.Context) { - fmt.Printf("Performing initial scan of root folders...\n") - - for _, folder := range s.folders { - // Skip if folder doesn't exist - if _, err := os.Stat(folder); os.IsNotExist(err) { - fmt.Printf("Skipping nonexistent folder: %s\n", folder) - continue - } - - // Submit scan job to worker (non-blocking) - job := &Job{ - ID: uuid.New().String(), - Type: JobTypeDirectoryScan, - Params: map[string]interface{}{ - "directory": folder, - "db": s.db, - }, - Status: JobStatusPending, - } - - if WorkerInstance != nil { - WorkerInstance.Enqueue(job) - fmt.Printf("Enqueued initial scan job: %s\n", folder) - } else { - fmt.Printf("Warning: Worker not initialized, skipping initial scan: %s\n", folder) - } - } - - fmt.Printf("Initial scan jobs enqueued\n") -} -``` - -**Why**: Ensures existing files are detected when watching starts, not just new changes. -Submits jobs to worker instead of calling scanDirectory() directly, ensuring all scans -go through the job queue for proper concurrency control. - -**Verification**: `go build ./internal/services/` - ---- - -### Step 0.5.9: Add Close() Method for Cleanup -**File**: `internal/services/media_scanner.go` - -**Action**: Add after performInitialScan(): - -```go -func (s *MediaScanner) Close() error { - fmt.Printf("Cleaning up scanner resources...\n") - - // Stop watching - if s.watcher != nil { - s.watcher.Close() - } - - // Clean up fileStability map to prevent memory leaks - s.fileStabilityMu.Lock() - s.fileStability = make(map[string]*atomic.Bool) - s.fileStabilityMu.Unlock() - - // Clear dirty directories - s.dirtyDirsMu.Lock() - s.dirtyDirs = make(map[string]time.Time) - s.dirtyDirsMu.Unlock() - - // Wait for in-progress scan to complete (with timeout) - timeout := time.After(5 * time.Second) - done := make(chan struct{}) - - go func() { - s.scan_mutex.Lock() - s.scan_mutex.Unlock() - close(done) - }() - - select { - case <-done: - fmt.Printf("Scanner cleanup complete\n") - case <-timeout: - fmt.Printf("Timeout waiting for scan to complete\n") - } - - return nil -} -``` - -**Why**: Ensures clean shutdown without resource leaks. Cleans up: -- fileStability map (prevents memory leaks) -- dirtyDirs map -- Waits for in-progress scan to complete (simpler than activeScans map) - -**Verification**: `go build ./internal/services/` - ---- - -### Step 0.5.10: 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(), 20*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 - // FIXED: More generous timeout to prevent flaky tests - time.Sleep(15 * 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.11: Add Test Isolation with Snapshot/Restore -**File**: `cmd/server/tests/test_helpers.go` - -**Location**: In `setupTestServer()` function - -**Action 1**: Snapshot original system_settings before test modifications (around line 519, after deleting users/libraries, before creating admin user): - -```go -// Snapshot current system_settings to restore after test -originalSettings := make(map[string]string) -settings, err := queries.GetAllSystemSettings(ctx) -if err == nil { - for _, setting := range settings { - originalSettings[setting.SettingKey] = setting.SettingValue - } -} -``` - -**Note**: Use `GetAllSystemSettings()` (not `ListSystemSettings()` - that function doesn't exist). - -**Action 2**: Add cleanup to restore settings (around line 591, before return statement): - -```go -// Register cleanup function to run automatically when test completes -t.Cleanup(func() { - // Restore original system_settings - for key, value := range originalSettings { - // Use background context since test context might be cancelled - queries.UpdateSystemSetting(context.Background(), database.UpdateSystemSettingParams{ - SettingKey: key, - SettingValue: value, - }) - } -}) -``` - -**Why**: -- Tests can modify settings during execution -- Original state always restored after test completes -- Dev database preserved -- Tests don't depend on execution order -- No risk of test pollution - -**Verification**: Run a test that modifies settings, check that settings are restored after test completes. - ---- - -### Step 0.5.12: Fix Test Expectation (Keep 60s) -**File**: `cmd/server/tests/scan_settings_integration_test.go` - -**Location**: Line 34 - -**Current code**: -```go -assert.Equal(t, float64(60), response["scan_poll_interval_seconds"]) -``` - -**Action**: Keep expectation at 60 seconds (no change needed): -```go -assert.Equal(t, float64(60), response["scan_poll_interval_seconds"]) -``` - -**Why**: Real-time fsnotify watching + 60s polling provides fast feedback. Test already expects 60s, which is correct. - -**Verification**: Run the failing test to confirm it now passes: -```bash -podman compose --profile tests run --rm tests go test -v -run "TestScanSettings_GetSettings/Get_settings_as_admin" ./cmd/server/tests/ -``` - ---- - -### Step 0.5.13: 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() - - // FIXED: Wait for detection - 12 seconds accounts for 10s batch + processing time - // Original 5 seconds was too short, causing test failures - time.Sleep(12 * 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.14: 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/worker.go internal/services/media_scanner_test.go cmd/server/tests/fsnotify_integration_test.go cmd/server/tests/test_helpers.go cmd/server/tests/scan_settings_integration_test.go -git commit -m "fix: Replace file-based fsnotify with directory-based watching + job queue concurrency - -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 -- No concurrency control: Multiple scans can run simultaneously -- File stability race condition: Multiple goroutines check same file -- Unbounded goroutine spawn: Each directory scan spawns goroutine without limit -- Default value inconsistency: GetPollInterval() returns 30s instead of 60s -- Test isolation issues: Settings not restored after tests - -Solution: Smart Hybrid Approach + Job Queue Concurrency Control -- 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) -- Job queue for all directory scans: Serialized by worker pool (no concurrent scans) -- JobTypeSetFolders: Async folder configuration via job queue -- Atomic file stability tracking: Prevents race conditions -- Fixed default value: 60s (matches handler/schema) -- Test isolation: Snapshot/restore system_settings - -Research Insights: -- Jellyfin: Uses 64KB buffer + smart merging + 45s ignore -- Audiobookshelf: Uses mtime stability check + 10s batching -- Old Phase 1: Job queue as concurrency control mechanism -- Combined: Best of all approaches for Bookhoord - -Changes: -REMOVE: -- eventQueue chan string (overflow prone) -- debounceTimer *time.Timer (legacy) -- processEventQueue() function -- flushEventQueue() function - -ADD: -- dirtyDirs map[string]time.Time (directory tracking) -- dirtyDirsMu sync.RWMutex (protects dirtyDirs) -- fileStability map[string]*atomic.Bool (mtime tracking, pointer to atomic.Bool prevents races) -- fileStabilityMu sync.RWMutex (protects fileStability) -- scan_mutex sync.Mutex (prevents concurrent scans) -- watching atomic.Bool (prevents duplicate WatchChanges() calls) -- markDirectoryDirty() helper with smart merging (parent + sibling consolidation) -- processDirtyDirectories() (10-second batch, submits jobs to WorkerInstance) -- waitForFileStability() (mtime polling, 3s interval, 60s timeout, uses *atomic.Bool) -- JobTypeDirectoryScan constant (directory scanning via job queue) -- processDirectoryScanJob() handler (handles directory scan jobs in worker) -- WorkerInstance *Worker global (no circular dependency between scanner and worker) -- Enqueue() method in Worker (non-blocking job submission) -- JobTypeSetFolders constant (async folder configuration) -- processSetFoldersJob() handler (handles folder configuration asynchronously) -- Test isolation in setupTestServer() (snapshot/restore system_settings) - -FIX: -- File stability uses *atomic.Bool (pointer) to prevent race conditions -- Default value fixed: GetPollInterval() returns 60s (real-time + fast safety net) -- Integration test timing: 5s → 12s (accounts for batch + processing) - -Concurrency Control: -- Job queue serializes all directory scans (no concurrent access) -- No unbounded goroutine spawn (worker pool limits concurrency) -- Atomic file stability checks prevent duplicate entries -- No race conditions in fileStability map -- No memory leaks from orphaned map entries -- Folder configuration is async (non-blocking API) -- No circular dependency (WorkerInstance global instead of struct field) - -Tests: -- Unit: TestMarkDirectoryDirty, TestProcessDirtyDirectories, - TestWaitForFileStability, TestSmartEventMerging -- Integration: TestFSNotify_BulkFileDetection (20 files, 12s wait) -- Test Isolation: Snapshot/restore system_settings in setupTestServer() -- Test Expectation: 60s polling interval (real-time + fast safety net) - -Benefits: -- No event queue overflow (directory watching eliminates per-file events) -- Reliable bulk import with file stability checks -- Delete detection via 60s polling (fast safety net) -- Smart event consolidation reduces redundant scans -- Job queue prevents concurrent scans (serialized by worker pool) -- No race conditions in fileStability map (atomic operations) -- No memory leaks from orphaned map entries (proper cleanup) -- No unbounded goroutine spawn (worker pool limits) -- No circular dependency (WorkerInstance global) -- Works on Docker, network mounts -- Test isolation preserves dev database state -- Consistent 60s polling interval across all components - -Files modified: -- internal/services/media_scanner.go (directory watching, file stability, atomic tracking, scan_mutex) -- internal/services/worker.go (JobTypeSetFolders, JobTypeDirectoryScan, processSetFoldersJob, processDirectoryScanJob, WorkerInstance, Enqueue) -- cmd/server/main.go (initialize WorkerInstance) -- internal/services/media_scanner_test.go (unit tests) -- cmd/server/tests/fsnotify_integration_test.go (bulk file detection test) -- cmd/server/tests/test_helpers.go (test isolation: snapshot/restore) -- cmd/server/tests/scan_settings_integration_test.go (test expects 60s polling)" -``` - -## Phase 1: Job Queue Expansion (6-8 hours) - -### Step 1.1: Add NEW Job Type Constants -**File**: `internal/services/worker.go` - -**Location**: JobType constants (lines 24-28) - -**Current code** (after Phase 0.5): -```go -const ( - JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" - JobTypeDirectoryScan JobType = "directory_scan" -) -``` - -**Action**: Add NEW job types for Phase 1: - -```go -const ( - JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" - JobTypeDirectoryScan JobType = "directory_scan" - JobTypeImport JobType = "import" // NEW - Phase 1 - JobTypeConvert JobType = "convert" // NEW - Phase 1 - JobTypeThumbnails JobType = "thumbnails" // NEW - Phase 1 - JobTypeReindex JobType = "reindex" // NEW - Phase 1 - JobTypeBackup JobType = "backup" // NEW - Phase 1 - JobTypeAnalytics JobType = "analytics" // NEW - Phase 1 - JobTypeSync JobType = "sync" // NEW - Phase 1 -) -``` - -**Why**: Phase 0.5 added JobTypeScan, JobTypeSetFolders, and JobTypeDirectoryScan. Phase 1 expands the job queue to support more async operations: import, conversion, thumbnails, reindexing, backup, analytics, and sync. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.2: Add NEW Job Handlers to Switch Statement -**File**: `internal/services/worker.go` - -**Location**: `processJob()` function (around line 127-132) - -**Current code** (after Phase 0.5): -```go -switch job.Type { -case JobTypeScan: - result, err = w.processScanJob(job) -case JobTypeSetFolders: - result, err = w.processSetFoldersJob(job) -case JobTypeDirectoryScan: - result, err = w.processDirectoryScanJob(job) -default: - err = fmt.Errorf("unknown job type: %s", job.Type) -} -``` - -**Action**: Add NEW handlers for Phase 1 job types: - -```go -switch job.Type { -case JobTypeScan: - result, err = w.processScanJob(job) -case JobTypeSetFolders: - result, err = w.processSetFoldersJob(job) -case JobTypeDirectoryScan: - result, err = w.processDirectoryScanJob(job) -case JobTypeImport: - result, err = w.processImportJob(job) -case JobTypeConvert: - result, err = w.processConvertJob(job) -case JobTypeThumbnails: - result, err = w.processThumbnailsJob(job) -case JobTypeReindex: - result, err = w.processReindexJob(job) -case JobTypeBackup: - result, err = w.processBackupJob(job) -case JobTypeAnalytics: - result, err = w.processAnalyticsJob(job) -case JobTypeSync: - result, err = w.processSyncJob(job) -default: - err = fmt.Errorf("unknown job type: %s", job.Type) -} -``` - -**Why**: Phase 0.5 added handlers for JobTypeScan, JobTypeSetFolders, and JobTypeDirectoryScan. Phase 1 adds handlers for the new job types: import, conversion, thumbnails, reindexing, backup, analytics, and sync. - -**Verification**: Run `go build ./internal/services/` to ensure compiles (will fail until handlers are implemented). - ---- - -### Step 1.3: Implement Import Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processScanJob()` (around line 248) - -**Action**: Add import handler: - -```go -func (w *Worker) processImportJob(job *Job) (interface{}, error) { - // Extract parameters - sourceParam, ok := job.Params["source"] - if !ok { - return nil, fmt.Errorf("source parameter required") - } - - source, ok := sourceParam.(string) - if !ok { - return nil, fmt.Errorf("source must be a string") - } - - libraryIDParam, ok := job.Params["library_id"] - if !ok { - return nil, fmt.Errorf("library_id parameter required") - } - - libraryID, ok := libraryIDParam.(string) - if !ok { - return nil, fmt.Errorf("library_id must be a string") - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - ctx := context.Background() - - // Import based on source type - var result map[string]interface{} - - switch source { - case "opds": - // Import from OPDS feed - feedURLParam, ok := job.Params["feed_url"] - if !ok { - return nil, fmt.Errorf("feed_url parameter required for OPDS import") - } - - feedURL, ok := feedURLParam.(string) - if !ok { - return nil, fmt.Errorf("feed_url must be a string") - } - - // Fetch OPDS feed - client := &http.Client{Timeout: 30 * time.Second} - resp, err := client.Get(feedURL) - if err != nil { - return nil, fmt.Errorf("failed to fetch OPDS feed: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("OPDS feed returned status %d", resp.StatusCode) - } - - // Parse OPDS feed (simplified - would need OPDS parser library) - // For now, just return the feed URL as the result - result = map[string]interface{}{ - "message": "OPDS import initiated", - "source": "opds", - "feed_url": feedURL, - "library_id": libraryID, - "note": "OPDS parsing not yet implemented", - } - - case "calibre": - // Import from Calibre library - calibreDBParam, ok := job.Params["calibre_db_path"] - if !ok { - return nil, fmt.Errorf("calibre_db_path parameter required for Calibre import") - } - - calibreDBPath, ok := calibreDBParam.(string) - if !ok { - return nil, fmt.Errorf("calibre_db_path must be a string") - } - - // Import from Calibre database (requires SQLite access) - // For now, just return the path as the result - result = map[string]interface{}{ - "message": "Calibre import initiated", - "source": "calibre", - "calibre_db_path": calibreDBPath, - "library_id": libraryID, - "note": "Calibre import not yet implemented", - } - - default: - return nil, fmt.Errorf("unsupported import source: %s (supported: opds, calibre)", source) - } - - return result, nil -} -``` - -**Why**: Foundation for importing books from OPDS feeds or Calibre libraries. Note: Full implementation would require OPDS parser and Calibre SQLite reader. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.4: Implement Convert Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processImportJob()` - -**Action**: Add conversion handler: - -```go -func (w *Worker) processConvertJob(job *Job) (interface{}, error) { - // Extract parameters - mediaIDParam, ok := job.Params["media_id"] - if !ok { - return nil, fmt.Errorf("media_id parameter required") - } - - mediaID, ok := mediaIDParam.(string) - if !ok { - return nil, fmt.Errorf("media_id must be a string") - } - - targetFormatParam, ok := job.Params["target_format"] - if !ok { - return nil, fmt.Errorf("target_format parameter required") - } - - targetFormat, ok := targetFormatParam.(string) - if !ok { - return nil, fmt.Errorf("target_format must be a string") - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - // Validate target format - if targetFormat != "kepub" { - return nil, fmt.Errorf("unsupported target format: %s (only 'kepub' supported)", targetFormat) - } - - ctx := context.Background() - - // Get media item - item, err := db.GetMediaItem(ctx, uuid.MustParse(mediaID)) - if err != nil { - return nil, fmt.Errorf("failed to get media item: %w", err) - } - - // Update progress - if job.ProgressCallback != nil { - job.ProgressCallback(0.0, 0, 0, 0) - } - - // Check if EPUB - if !strings.HasSuffix(strings.ToLower(item.FilePath), ".epub") { - return nil, fmt.Errorf("only EPUB files can be converted to KEPUB") - } - - // Perform conversion - // Note: This would call the actual conversion utility - // For now, return success with the converted path - - convertedPath := strings.TrimSuffix(item.FilePath, ".epub") + ".kepub.epub" - - // Update progress to complete - if job.ProgressCallback != nil { - job.ProgressCallback(1.0, 1, 1, 0) - } - - return map[string]interface{}{ - "message": "conversion completed", - "media_id": mediaID, - "source_format": "epub", - "target_format": targetFormat, - "converted_path": convertedPath, - }, nil -} -``` - -**Why**: Converts EPUB to KEPUB format for Kobo devices. Full implementation would integrate with existing conversion tools. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.5: Implement Thumbnails Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processConvertJob()` - -**Action**: Add thumbnail generation handler: - -```go -func (w *Worker) processThumbnailsJob(job *Job) (interface{}, error) { - // Extract parameters - libraryIDParam, ok := job.Params["library_id"] - if !ok { - return nil, fmt.Errorf("library_id parameter required") - } - - libraryID, ok := libraryIDParam.(string) - if !ok { - return nil, fmt.Errorf("library_id must be a string") - } - - forceParam, forceOk := job.Params["force"] - force := false - if forceOk { - force, ok = forceParam.(bool) - if !ok { - return nil, fmt.Errorf("force must be a boolean") - } - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - ctx := context.Background() - - // Get all items in library - items, err := db.ListMediaItemsByLibrary(ctx, uuid.MustParse(libraryID)) - if err != nil { - return nil, fmt.Errorf("failed to query library items: %w", err) - } - - // Set up progress tracking - totalItems := len(items) - processedItems := 0 - newThumbnails := 0 - errors := 0 - - updateProgress := func() { - if job.ProgressCallback != nil { - progress := float64(processedItems) / float64(totalItems) - job.ProgressCallback(progress, processedItems, newThumbnails, errors) - } - } - - // Process each item - for _, item := range items { - // Check if already has cover image - if !force && item.CoverImage != nil && len(item.CoverImage) > 0 { - processedItems++ - updateProgress() - continue - } - - // Extract thumbnail from file - // Note: This would call the actual thumbnail extraction - // For now, just simulate the operation - - // Simulate thumbnail extraction - processedItems++ - - // In real implementation: - // - Open file (EPUB, PDF, comic) - // - Extract cover image - // - Resize/compress - // - Store in database - // - If successful: newThumbnails++ - - updateProgress() - } - - return map[string]interface{}{ - "message": "thumbnail generation completed", - "library_id": libraryID, - "total_items": totalItems, - "processed": processedItems, - "new_thumbnails": newThumbnails, - "errors": errors, - }, nil -} -``` - -**Why**: Generates missing book covers. Useful for libraries without embedded covers. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.6: Implement Reindex Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processThumbnailsJob()` - -**Action**: Add search index rebuild handler: - -```go -func (w *Worker) processReindexJob(job *Job) (interface{}, error) { - // Extract parameters - forceParam, forceOk := job.Params["force"] - force := false - if forceOk { - force, ok = forceParam.(bool) - if !ok { - return nil, fmt.Errorf("force must be a boolean") - } - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - ctx := context.Background() - - // Get all media items - items, err := db.ListAllMediaItems(ctx) - if err != nil { - return nil, fmt.Errorf("failed to query media items: %w", err) - } - - // Set up progress tracking - totalItems := len(items) - processedItems := 0 - - updateProgress := func() { - if job.ProgressCallback != nil { - progress := float64(processedItems) / float64(totalItems) - job.ProgressCallback(progress, processedItems, 0, 0) - } - } - - // Reindex each item - for _, item := range items { - // Update full-text search index - // Note: This depends on your search implementation - // For now, just track progress - - processedItems++ - updateProgress() - } - - return map[string]interface{}{ - "message": "search index rebuilt", - "total_items": totalItems, - "indexed": processedItems, - }, nil -} -``` - -**Why**: Rebuilds search index for all media items. Useful after bulk imports or schema changes. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.7: Implement Backup Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processReindexJob()` - -**Action**: Add database backup handler: - -```go -func (w *Worker) processBackupJob(job *Job) (interface{}, error) { - // Extract parameters - backupTypeParam, ok := job.Params["backup_type"] - if !ok { - return nil, fmt.Errorf("backup_type parameter required") - } - - backupType, ok := backupTypeParam.(string) - if !ok { - return nil, fmt.Errorf("backup_type must be a string") - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - // Validate backup type - if backupType != "full" && backupType != "schema_only" { - return nil, fmt.Errorf("backup_type must be 'full' or 'schema_only'") - } - - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) - defer cancel() - - var backupPath string - var timestamp string - - if backupType == "schema_only" { - // Dump schema - timestamp = time.Now().Format("20060102_150405") - backupPath = fmt.Sprintf("/backups/schema_%s.sql", timestamp) - - // Note: This would call pg_dump to dump schema - // For now, just return the path - - } else { - // Full backup - timestamp = time.Now().Format("20060102_150405") - backupPath = fmt.Sprintf("/backups/full_%s.sql", timestamp) - - // Note: This would call pg_dump to dump full database - // For now, just return the path - } - - return map[string]interface{}{ - "message": "backup completed", - "backup_type": backupType, - "backup_path": backupPath, - "timestamp": timestamp, - }, nil -} -``` - -**Why**: Creates database backups. Full implementation would call `pg_dump`. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.8: Implement Analytics Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processBackupJob()` - -**Action**: Add analytics report handler: - -```go -func (w *Worker) processAnalyticsJob(job *Job) (interface{}, error) { - // Extract parameters - reportTypeParam, ok := job.Params["report_type"] - if !ok { - return nil, fmt.Errorf("report_type parameter required") - } - - reportType, ok := reportTypeParam.(string) - if !ok { - return nil, fmt.Errorf("report_type must be a string") - } - - libraryIDParam, libOk := job.Params["library_id"] - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - ctx := context.Background() - - var result interface{} - - switch reportType { - case "library_stats": - // Library statistics - var libraryID uuid.UUID - if libOk { - libraryID, ok = libraryIDParam.(string) - if !ok { - return nil, fmt.Errorf("library_id must be a string") - } - } - - // Query library stats - if libOk { - items, err := db.ListMediaItemsByLibrary(ctx, uuid.MustParse(libraryID)) - if err != nil { - return nil, fmt.Errorf("failed to query library items: %w", err) - } - - // Calculate stats - totalSize := int64(0) - formats := make(map[string]int) - authors := make(map[string]int) - - for _, item := range items { - totalSize += item.FileSize - ext := strings.ToLower(filepath.Ext(item.FilePath)) - formats[ext]++ - if item.Author != "" { - authors[item.Author]++ - } - } - - result = map[string]interface{}{ - "report_type": "library_stats", - "library_id": libraryID, - "total_items": len(items), - "total_size": totalSize, - "formats": formats, - "authors": authors, - "top_authors": getTopN(authors, 10), - } - } - - case "system_stats": - // System-wide statistics - libraries, err := db.ListLibraries(ctx) - if err != nil { - return nil, fmt.Errorf("failed to query libraries: %w", err) - } - - items, err := db.ListAllMediaItems(ctx) - if err != nil { - return nil, fmt.Errorf("failed to query items: %w", err) - } - - // Calculate system stats - totalSize := int64(0) - formats := make(map[string]int) - - for _, item := range items { - totalSize += item.FileSize - ext := strings.ToLower(filepath.Ext(item.FilePath)) - formats[ext]++ - } - - result = map[string]interface{}{ - "report_type": "system_stats", - "total_libraries": len(libraries), - "total_items": len(items), - "total_size": totalSize, - "formats": formats, - } - - default: - return nil, fmt.Errorf("unsupported report_type: %s (supported: library_stats, system_stats)", reportType) - } - - return result, nil -} - -// Helper function to get top N items from a map -func getTopN(m map[string]int, n int) map[string]int { - type kv struct { - key string - value int - } - - var ss []kv - for k, v := range m { - ss = append(ss, kv{k, v}) - } - - sort.Slice(ss, func(i, j int) bool { - return ss[i].value > ss[j].value - }) - - if len(ss) > n { - ss = ss[:n] - } - - result := make(map[string]int) - for _, kv := range ss { - result[kv.key] = kv.value - } - - return result -} -``` - -**Why**: Generates analytics reports for library and system statistics. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.9: Implement Sync Job Handler -**File**: `internal/services/worker.go` - -**Location**: Add new function after `processAnalyticsJob()` - -**Action**: Add device sync trigger handler: - -```go -func (w *Worker) processSyncJob(job *Job) (interface{}, error) { - // Extract parameters - deviceIDParam, ok := job.Params["device_id"] - if !ok { - return nil, fmt.Errorf("device_id parameter required") - } - - deviceID, ok := deviceIDParam.(string) - if !ok { - return nil, fmt.Errorf("device_id must be a string") - } - - libraryIDParam, ok := job.Params["library_id"] - if !ok { - return nil, fmt.Errorf("library_id parameter required") - } - - libraryID, ok := libraryIDParam.(string) - if !ok { - return nil, fmt.Errorf("library_id must be a string") - } - - db, ok := job.Params["db"].(*database.Queries) - if !ok { - return nil, fmt.Errorf("database parameter required") - } - - ctx := context.Background() - - // Get device info - device, err := db.GetDevice(ctx, uuid.MustParse(deviceID)) - if err != nil { - return nil, fmt.Errorf("failed to get device: %w", err) - } - - // Trigger sync by adding to sync queue - syncItem := database.AddToSyncQueueParams{ - DeviceID: uuid.MustParse(deviceID), - LibraryID: uuid.MustParse(libraryID), - SyncType: database.SyncTypeProgress, - Priority: 5, // Medium priority - } - - _, err = db.AddToSyncQueue(ctx, syncItem) - if err != nil { - return nil, fmt.Errorf("failed to add to sync queue: %w", err) - } - - return map[string]interface{}{ - "message": "sync triggered", - "device_id": deviceID, - "device_name": device.DeviceName, - "library_id": libraryID, - "sync_type": "progress", - }, nil -} -``` - -**Why**: Triggers device sync operations via the existing sync queue system. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.10: Create Job Management Handler -**File**: `internal/handlers/jobs.go` (new file) - -**Action**: Create new handler for job management: - -```go -package handlers - -import ( - "net/http" - "github.com/labstack/echo/v4" - "github.com/google/uuid" - - "bookhoard/internal/database" - "bookhoard/internal/services" -) - -type JobsHandler struct { - db *database.Queries - worker *services.Worker -} - -func NewJobsHandler(db *database.Queries, worker *services.Worker) *JobsHandler { - return &JobsHandler{ - db: db, - worker: worker, - } -} - -// CreateJob creates a new job based on type -func (h *JobsHandler) CreateJob(c echo.Context) error { - var req struct { - Type string `json:"type"` - Params map[string]interface{} `json:"params"` - } - - if err := c.Bind(&req); err != nil { - return c.JSON(http.StatusBadRequest, map[string]string{ - "error": "Invalid request body", - }) - } - - // Validate job type - var jobType services.JobType - switch req.Type { - case "import", "convert", "thumbnails", "reindex", "backup", "analytics", "sync": - jobType = services.JobType(req.Type) - default: - return c.JSON(http.StatusBadRequest, map[string]string{ - "error": "Invalid job type", - }) - } - - // Add database to params - req.Params["db"] = h.db - - // Get user ID from context - userID := c.Get("user_id").(string) - - // Create job - job := &services.Job{ - ID: uuid.New().String(), - Type: jobType, - UserID: userID, - Params: req.Params, - Status: services.JobStatusPending, - } - - // Enqueue job - if err := h.worker.EnqueueJob(job); err != nil { - return c.JSON(http.StatusInternalServerError, map[string]string{ - "error": "Failed to enqueue job", - }) - } - - return c.JSON(http.StatusAccepted, map[string]interface{}{ - "message": "Job created", - "job_id": job.ID, - "type": req.Type, - "status": "pending", - }) -} - -// GetJobStatus returns the status of a specific job -func (h *JobsHandler) GetJobStatus(c echo.Context) error { - jobID := c.Param("jobId") - - result, exists := h.worker.GetJobStatus(jobID) - if !exists { - return c.JSON(http.StatusNotFound, map[string]string{ - "error": "Job not found", - }) - } - - return c.JSON(http.StatusOK, result) -} -``` - -**Why**: Provides REST API for creating and managing jobs. - -**Verification**: Run `go build ./internal/handlers/` to ensure compiles. - ---- - -### Step 1.11: Register Job Routes -**File**: `internal/router/router.go` - -**Location**: Add jobs handler to Config struct (around line 39-64) - -**Action**: Add to Config struct: - -```go -type Config struct { - Echo *echo.Echo - Queries *database.Queries - Cfg *config.Config - AuthHandler *handlers.AuthHandler - LibraryHandler *handlers.LibraryHandler - SystemSettingsHandler *handlers.SystemSettingsHandler - ScannerHandler *handlers.Handler - JobsHandler *handlers.JobsHandler // NEW - // ... other handlers -} -``` - -**Location**: Register routes (around line 200+) - -**Action**: Add job routes: - -```go -// Job management routes (admin-only) -jobsGroup := apiGroup.Group("/jobs") -jobsGroup.Use(middleware.AuthMiddleware) -jobsGroup.Use(middleware.AdminMiddleware) - -jobsGroup.POST("", cfg.JobsHandler.CreateJob) -jobsGroup.GET("/:jobId", cfg.JobsHandler.GetJobStatus) -``` - -**Why**: Makes job management API accessible. - -**Verification**: Run `go build ./internal/router/` to ensure compiles. - ---- - -### Step 1.12: Initialize JobsHandler in main.go -**File**: `cmd/server/main.go` - -**Location**: Around line 92 (before systemSettingsHandler creation) - -**Action**: Create jobs handler: - -```go -jobsHandler := handlers.NewJobsHandler(queries, worker) -``` - -**Location**: Add to router Config (around line 200) - -**Action**: Add to config: - -```go -cfg.JobsHandler = jobsHandler -``` - -**Why**: Makes jobs handler available to router. - -**Verification**: Run `go build ./cmd/server` to ensure compiles. - ---- - -### Step 1.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 1.14: Verify Job Queue Expansion -**Action**: Run full test suite: - -```bash -# Test worker with new job types -podman compose --profile tests run --rm tests go test -v ./internal/services/ - -# Test handlers -podman compose --profile tests run --rm tests go test -v ./internal/handlers/ - -# Test router -podman compose --profile tests run --rm tests go test -v ./internal/router/ - -# Ensure full project builds -podman compose --profile tests build -``` - -**Commit Phase 2**: -```bash -git add internal/services/worker.go internal/handlers/jobs.go internal/router/router.go cmd/server/main.go -git commit -m "feat: Expand job queue to handle 7 async operations - -Job Types Added: -- JobTypeImport: Import from OPDS feeds or Calibre -- JobTypeConvert: Convert EPUB to KEPUB -- JobTypeThumbnails: Generate missing book covers -- JobTypeReindex: Rebuild search index -- JobTypeBackup: Create database backups -- JobTypeAnalytics: Generate library/system reports -- JobTypeSync: Trigger device sync operations - -Infrastructure: -- Add JobsHandler for job management API -- Add POST /api/jobs endpoint for job creation -- Add GET /api/jobs/:jobId endpoint for job status -- All jobs support progress tracking and status queries - -Benefits: -- Leverages underutilized job queue infrastructure -- Provides unified async operation handling -- Non-blocking API responses for long-running tasks -- Real-time progress tracking for all operations -- Extensible for future job types - -Note: Import/Convert/Thumbnail jobs require additional implementation: -- OPDS parser library needed -- Calibre SQLite reader needed -- Conversion utility integration needed -- Thumbnail extraction implementation needed - -Files modified: -- internal/services/worker.go (7 new job handlers) -- internal/handlers/jobs.go (new file) -- internal/router/router.go (job routes) -- cmd/server/main.go (jobs handler initialization)" -``` - ---- - -## Phase 2: WebSocket Scan Progress (2-3 hours) - -### Step 2.1: Add Scan Progress Message Types -**File**: `internal/sync/websocket.go` - -**Location**: Message type constants (around line 20-30) - -**Current code**: -```go -const ( - MessageTypeProgressUpdate = "progress_update" - MessageTypeAnnotationUpdate = "annotation_update" - MessageTypeConflict = "conflict" - MessageTypeSyncComplete = "sync_complete" - MessageTypeHeartbeat = "heartbeat" - MessageTypeInitial = "initial_state" -) -``` - -**Action**: Add scan progress message types: - -```go -const ( - MessageTypeProgressUpdate = "progress_update" - MessageTypeAnnotationUpdate = "annotation_update" - MessageTypeConflict = "conflict" - MessageTypeSyncComplete = "sync_complete" - MessageTypeHeartbeat = "heartbeat" - MessageTypeInitial = "initial_state" - MessageTypeScanProgress = "scan_progress" // NEW - MessageTypeScanComplete = "scan_complete" // NEW - MessageTypeScanError = "scan_error" // NEW -) -``` - -**Why**: Defines message types for real-time scan progress updates via WebSocket. - -**Verification**: Run `go build ./internal/sync/` to ensure compiles. - ---- - -### Step 2.2: Pass Connection Manager to Worker -**File**: `internal/services/worker.go` - -**Location**: Worker struct definition (around line 14-20) - -**Current struct**: -```go -type Worker struct { - jobQueue chan *Job - results map[string]*JobResult - mu sync.RWMutex - wg sync.WaitGroup - ctx context.Context - cancel context.CancelFunc - shuttingDown atomic.Bool -} -``` - -**Action**: Add connection manager field: - -```go -type Worker struct { - jobQueue chan *Job - results map[string]*JobResult - connManager *ConnectionManager // NEW - mu sync.RWMutex - wg sync.WaitGroup - ctx context.Context - cancel context.CancelFunc - shuttingDown atomic.Bool -} -``` - -**Action**: Update constructor to accept connection manager: - -```go -func NewWorker(numWorkers int, connManager *ConnectionManager) *Worker { - // ... existing code ... - w.connManager = connManager - return w -} -``` - -**Why**: Worker needs connection manager to broadcast scan progress via WebSocket. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 2.3: Update Worker Initialization in main.go -**File**: `cmd/server/main.go` - -**Location**: Where worker is created (around line 30-50) - -**Current code**: -```go -worker := services.NewWorker(3) -``` - -**Action**: Pass connection manager: - -```go -// After connection manager is created -connManager := wsync.NewConnectionManager() - -// Pass to worker -worker := services.NewWorker(3, connManager) -``` - -**Why**: Provides worker with WebSocket connection manager for broadcasting scan progress. - -**Verification**: Run `go build ./cmd/server` to ensure compiles. - ---- - -### Step 2.4: Add User ID to Job -**File**: `internal/services/worker.go` - -**Location**: Job struct (around line 30-48) - -**Current struct**: -```go -type Job struct { - ID string - Type JobType - Params map[string]interface{} - Status JobStatus - CreatedAt time.Time - StartedAt *time.Time - CompletedAt *time.Time - Error error - Result interface{} - Context context.Context - ProgressCallback func(progress float64, filesScanned, newItems, errors int) -} -``` - -**Action**: Add UserID field: - -```go -type Job struct { - ID string - Type JobType - UserID string // NEW - for WebSocket targeting - Params map[string]interface{} - Status JobStatus - CreatedAt time.Time - StartedAt *time.Time - CompletedAt *time.Time - Error error - Result interface{} - Context context.Context - ProgressCallback func(progress float64, filesScanned, newItems, errors int) -} -``` - -**Why**: Worker needs to know which user to broadcast scan progress to. - ---- - -### Step 2.5: Broadcast Scan Progress from Worker -**File**: `internal/services/worker.go` - -**Location**: `processScanJob()` function (around line 176-247) - -**Current code** (around line 210): -```go -scanner.job = job - -// Set up progress callback -job.ProgressCallback = func(progress float64, filesScanned, newItems, errors int) { - w.mu.Lock() - defer w.mu.Unlock() - - if result, exists := w.results[job.ID]; exists { - result.Progress = progress - result.FilesScanned = filesScanned - result.NewItems = newItems - result.Errors = errors - } -} -``` - -**Action**: Enhance progress callback to broadcast via WebSocket: - -```go -scanner.job = job - -// Set up progress callback -job.ProgressCallback = func(progress float64, filesScanned, newItems, errors int) { - w.mu.Lock() - defer w.mu.Unlock() - - // Update job result - if result, exists := w.results[job.ID]; exists { - result.Progress = progress - result.FilesScanned = filesScanned - result.NewItems = newItems - result.Errors = errors - } - - // Broadcast via WebSocket to user - if w.connManager != nil && job.UserID != "" { - msg := wsync.BroadcastMessage{ - Type: wsync.MessageTypeScanProgress, - Data: map[string]interface{}{ - "job_id": job.ID, - "progress": progress, - "files_scanned": filesScanned, - "new_items": newItems, - "errors": errors, - }, - } - - w.connManager.BroadcastToUser(job.UserID, msg) - } -} -``` - -**Why**: Real-time scan progress updates pushed to user's connected devices via WebSocket. - ---- - -### Step 2.6: Add User ID to Scan Jobs -**File**: `internal/handlers/scanner.go` - -**Location**: `ScanLibrary()` function (around line 33-110) - -**Current code** (around line 60-85): -```go -job := &services.Job{ - ID: uuid.New().String(), - Type: services.JobTypeScan, - Params: map[string]interface{}{ - "library_id": libraryID, - "folders": folders, - "admin_id": adminID, - "db": h.db, - "force": force, - }, - Status: services.JobStatusPending, -} -``` - -**Action**: Add user ID from context: - -```go -// Get user ID from context -userID := c.Get("user_id").(string) - -job := &services.Job{ - ID: uuid.New().String(), - Type: services.JobTypeScan, - UserID: userID, // NEW - Params: map[string]interface{}{ - "library_id": libraryID, - "folders": folders, - "admin_id": adminID, - "db": h.db, - "force": force, - }, - Status: services.JobStatusPending, -} -``` - -**Action**: Do the same for any other job creation (import, convert, etc.). - -**Why**: Worker knows which user to broadcast progress to. - -**Verification**: Run `go build ./internal/handlers/` to ensure compiles. - ---- - -### Step 2.7: Add Frontend WebSocket Scan Progress Listener -**File**: `web/src/admin.ts` or appropriate TypeScript file - -**Location**: After existing WebSocket connection setup - -**Action**: Add scan progress message handler: - -```typescript -// In WebSocket connection setup -ws.onmessage = (event) => { - const message = JSON.parse(event.data); - - switch (message.type) { - case 'scan_progress': - // Update scan progress UI - updateScanProgress(message.data); - break; - - case 'scan_complete': - // Scan completed - showScanComplete(message.data); - // Stop polling - stopScanStatusPolling(); - break; - - case 'scan_error': - // Scan error - showScanError(message.data); - break; - - // ... existing message handlers ... - } -}; - -function updateScanProgress(data: any) { - // Update progress bar - const progressBar = document.getElementById('scan-progress-bar'); - if (progressBar) { - progressBar.style.width = `${data.progress * 100}%`; - } - - // Update stats - const progressText = document.getElementById('scan-progress-text'); - if (progressText) { - progressText.textContent = `${data.files_scanned} files scanned (${data.new_items} new)`; - } -} - -function showScanComplete(data: any) { - // Hide progress bar - const progressSection = document.getElementById('scan-progress'); - if (progressSection) { - progressSection.classList.add('hidden'); - } - - // Show completion message - console.log('Scan complete:', data); -} -``` - -**Why**: Frontend receives real-time scan progress updates instead of polling every 2 seconds. - -**Verification**: Run `npm run build:ts` to compile TypeScript. - ---- - -### Step 2.8: Remove Scan Progress Polling (Optional) -**File**: `web/src/admin.ts` - -**Location**: `pollScanProgress()` function (around line 210-277) - -**Action**: You can now remove or reduce polling frequency since WebSocket provides real-time updates: - -```typescript -// Option 1: Remove polling entirely (relying on WebSocket) -function pollScanProgress(jobIds: string[], libraryNames: Record): void { - // WebSocket handles updates now - no polling needed - console.log('Scan progress via WebSocket'); -} - -// Option 2: Keep polling as fallback (less frequent) -function pollScanProgress(jobIds: string[], libraryNames: Record): void { - const interval = setInterval(async () => { - // ... existing polling code ... - }, 10000); // Reduce to 10 seconds (fallback only) -} -``` - -**Why**: WebSocket provides real-time updates, reducing need for frequent polling. Can keep polling as fallback. - -**Verification**: Run `npm run build:ts` to compile TypeScript. - ---- - -### Step 2.10: Verify WebSocket Scan Progress -**Action**: Test the full stack: - -```bash -# Build everything -podman compose --profile tests build - -# Start container -podman compose --profile tests up -d - -# Test WebSocket connection -# Open browser console, trigger scan, verify real-time progress updates -``` - -**Commit Phase 3**: -```bash -git add internal/sync/websocket.go internal/services/worker.go internal/handlers/scanner.go web/src/admin.ts cmd/server/main.go -git commit -m "feat: Add real-time scan progress via WebSocket - -WebSocket Enhancements: -- Add MessageTypeScanProgress, MessageTypeScanComplete, MessageTypeScanError -- Pass connection manager to worker for broadcast capability -- Add UserID to Job struct for user-targeted broadcasts -- Broadcast scan progress from worker progress callback -- Frontend receives real-time updates instead of polling every 2 seconds - -Benefits: -- Instant scan progress updates (no 2-second polling delay) -- Reduced server load from fewer HTTP requests -- Better user experience with real-time feedback -- Leverages existing WebSocket infrastructure - -Architecture: -- Worker broadcasts to user's connected devices -- Frontend listens for scan_progress messages -- Optional: Keep polling as fallback at reduced frequency (10s) - -Files modified: -- internal/sync/websocket.go (message types) -- internal/services/worker.go (connManager, broadcasting) -- internal/handlers/scanner.go (add user_id to jobs) -- web/src/admin.ts (WebSocket message handlers) -- cmd/server/main.go (pass connManager to worker) - -Note: Can reduce or remove frontend polling since WebSocket provides real-time updates" -``` - ---- - -## Phase 3: Caching and Monitoring (2 hours) - -### Step 3.1: Create Settings Cache -**File**: `internal/services/cache.go` (new file) - -**Action**: Create settings cache implementation: - -```go -package services - -import ( - "sync" - "time" -) - -type SettingsCache struct { - data map[string]string - mu sync.RWMutex - ttl time.Duration - lastUpdate time.Time -} - -func NewSettingsCache(ttl time.Duration) *SettingsCache { - return &SettingsCache{ - data: make(map[string]string), - ttl: ttl, - lastUpdate: time.Now(), - } -} - -func (c *SettingsCache) Get(key string) (string, bool) { - c.mu.RLock() - defer c.mu.RUnlock() - - // Check if cache is expired - if time.Since(c.lastUpdate) > c.ttl { - return "", false - } - - val, ok := c.data[key] - return val, ok -} - -func (c *SettingsCache) Set(key, value string) { - c.mu.Lock() - defer c.mu.Unlock() - - c.data[key] = value - c.lastUpdate = time.Now() -} - -func (c *SettingsCache) Invalidate() { - c.mu.Lock() - defer c.mu.Unlock() - - c.data = make(map[string]string) - c.lastUpdate = time.Time{} -} - -func (c *SettingsCache) InvalidateKey(key string) { - c.mu.Lock() - defer c.mu.Unlock() - - delete(c.data, key) -} -``` - -**Why**: In-memory cache for frequently accessed system settings with TTL-based expiration. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 3.2: Add Cache to MediaScanner -**File**: `internal/services/media_scanner.go` - -**Location**: MediaScanner struct (around line 45-70) - -**Action**: Add settings cache field: - -```go -type MediaScanner struct { - // ... existing fields ... - settingsCache *SettingsCache // NEW -} -``` - -**Location**: Constructor `NewMediaScanner()` (around line 93-108) - -**Action**: Initialize cache: - -```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, - settingsCache: NewSettingsCache(30 * time.Second), // 30 second TTL - eventQueue: make(chan string, 500), - // ... rest of existing initialization - } -} -``` - -**Why**: Scanner uses cached settings instead of querying database every time. - ---- - -### Step 3.3: Use Cache in GetPollInterval() -**File**: `internal/services/media_scanner.go` - -**Location**: `GetPollInterval()` function (lines 111-127) - -**Current code**: -```go -func (s *MediaScanner) GetPollInterval() time.Duration { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - setting, err := s.db.GetSystemSetting(ctx, "scan_poll_interval_seconds") - if err != nil || setting == "" { - return 60 * time.Second - } - // ... convert to duration ... -} -``` - -**Action**: Use cache first: - -```go -func (s *MediaScanner) GetPollInterval() time.Duration { - // Check cache first - if cached, ok := s.settingsCache.Get("scan_poll_interval_seconds"); ok { - if seconds, err := strconv.Atoi(cached); err == nil { - return time.Duration(seconds) * time.Second - } - } - - // Cache miss - query database - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - setting, err := s.db.GetSystemSetting(ctx, "scan_poll_interval_seconds") - if err != nil || setting == "" { - return 60 * time.Second - } - - // Store in cache - s.settingsCache.Set("scan_poll_interval_seconds", setting) - - // Convert to duration - seconds, err := strconv.Atoi(setting) - if err != nil { - return 60 * time.Second - } - - return time.Duration(seconds) * time.Second -} -``` - -**Why**: Reduces database queries. Cache invalidates after 30 seconds. - ---- - -### Step 3.4: Use Cache in GetAutoScanEnabled() -**File**: `internal/services/media_scanner.go` - -**Location**: Find or add `GetAutoScanEnabled()` method - -**Action**: Add cached method: - -```go -func (s *MediaScanner) GetAutoScanEnabled() bool { - // Check cache first - if cached, ok := s.settingsCache.Get("auto_scan_enabled"); ok { - return strings.ToLower(cached) == "true" - } - - // Cache miss - query database - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - - setting, err := s.db.GetSystemSetting(ctx, "auto_scan_enabled") - if err != nil || setting == "" { - return true // Default to enabled - } - - // Store in cache - s.settingsCache.Set("auto_scan_enabled", setting) - - return strings.ToLower(setting) == "true" -} -``` - -**Why**: Reduces database queries for auto_scan_enabled setting. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 3.5: Extend /health Endpoint -**File**: `internal/router/frontend.go` - -**Location**: Health check handler (around line 833-847) - -**Current code**: -```go -func HealthCheck(c echo.Context) error { - ctx, cancel := context.WithTimeout(c.Request().Context(), 2*time.Second) - defer cancel() - - // Check database - if err := queries.Ping(ctx); err != nil { - return c.JSON(http.StatusServiceUnavailable, map[string]interface{}{ - "status": "unhealthy", - "error": err.Error(), - }) - } - - return c.JSON(http.StatusOK, map[string]interface{}{ - "status": "healthy", - "database": "connected", - }) -} -``` - -**Action**: Add scan health information: - -```go -func HealthCheck(c echo.Context) error { - ctx, cancel := context.WithTimeout(c.Request().Context(), 2*time.Second) - defer cancel() - - // Check database - if err := queries.Ping(ctx); err != nil { - return c.JSON(http.StatusServiceUnavailable, map[string]interface{}{ - "status": "unhealthy", - "error": err.Error(), - }) - } - - // Get scan health information - // Note: Would need to pass worker to this handler - // For now, return basic health - - return c.JSON(http.StatusOK, map[string]interface{}{ - "status": "healthy", - "database": "connected", - "scan": map[string]interface{}{ - "scan_in_progress": false, // Would check worker results - "active_jobs": 0, // Would count running jobs - }, - }) -} -``` - -**Note**: Full implementation would require passing worker to health check handler. For now, this is a placeholder. - -**Verification**: Run `go build ./internal/router/` to ensure compiles. - ---- - -### Step 3.7: Verify Caching and Monitoring -**Action**: Test the changes: - -```bash -# Build everything -podman compose --profile tests build - -# Test health endpoint -curl http://localhost:8765/health - -# Verify settings are cached (check database query logs) -``` - -**Commit Phase 4**: -```bash -git add internal/services/cache.go internal/services/media_scanner.go internal/router/frontend.go -git commit -m "feat: Add settings cache and enhance health monitoring - -Caching: -- Add SettingsCache with TTL (30 seconds) -- Cache scan_poll_interval_seconds and auto_scan_enabled settings -- Reduces database queries for frequently accessed settings -- Cache invalidates automatically after TTL - -Monitoring: -- Extend /health endpoint to include scan health information -- Add scan_in_progress status -- Add active_jobs count -- Foundation for comprehensive monitoring - -Benefits: -- Reduces database load (cached settings) -- Faster response times for settings queries -- Better visibility into system health -- Foundation for monitoring dashboards - -Files modified: -- internal/services/cache.go (new file) -- internal/services/media_scanner.go (cache integration) -- internal/router/frontend.go (enhanced health check) - -Note: Full cache invalidation on settings update would require -scanner reference in SystemSettingsHandler. TTL-based expiration -is sufficient for now." -``` - ---- - -## Phase 4: Job Queue Enhancements (4-6 hours) - -### Step 4.1: Add Job Priority Field -**File**: `internal/services/worker.go` - -**Location**: Job struct (around line 30-48) - -**Current struct**: -```go -type Job struct { - ID string - Type JobType - UserID string - Params map[string]interface{} - Status JobStatus - CreatedAt time.Time - StartedAt *time.Time - CompletedAt *time.Time - Error error - Result interface{} - Context context.Context - ProgressCallback func(progress float64, filesScanned, newItems, errors int) -} -``` - -**Action**: Add priority field: - -```go -type Job struct { - ID string - Type JobType - UserID string - Priority int // NEW - 0=low, 5=medium, 10=high - Params map[string]interface{} - Status JobStatus - CreatedAt time.Time - StartedAt *time.Time - CompletedAt *time.Time - Error error - Result interface{} - Context context.Context - ProgressCallback func(progress float64, filesScanned, newItems, errors int) -} -``` - -**Why**: Allows higher priority jobs to be processed first. - ---- - -### Step 4.2: Add Job History Table -**File**: `internal/database/queries.sql` - -**Location**: Add new table at end - -**Action**: Add job history table: - -```sql --- Job history table -CREATE TABLE IF NOT EXISTS job_history ( - id UUID PRIMARY KEY DEFAULT gen_random_uuid(), - job_id TEXT NOT NULL UNIQUE, - job_type TEXT NOT NULL, - user_id TEXT, - status TEXT NOT NULL, - params JSONB, - result JSONB, - error_message TEXT, - created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), - started_at TIMESTAMP WITH TIME ZONE, - completed_at TIMESTAMP WITH TIME ZONE, - expires_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() + INTERVAL '7 days' -); - --- Index for looking up jobs -CREATE INDEX idx_job_history_job_id ON job_history(job_id); -CREATE INDEX idx_job_history_user_id ON job_history(user_id); -CREATE INDEX idx_job_history_created_at ON job_history(created_at DESC); - --- Clean up old jobs (run via cron or scheduled task) -CREATE OR REPLACE FUNCTION cleanup_old_jobs() RETURNS void AS $$ -BEGIN - DELETE FROM job_history WHERE completed_at < NOW() - INTERVAL '7 days'; -END; -$$ LANGUAGE plpgsql; -``` - -**Why**: Persist job history to database. Survives server restarts. Allows audit trail. - ---- - -### Step 4.3: Add Job History Queries -**File**: `internal/database/queries.sql` - -**Location**: Add new queries at end - -**Action**: Add job history queries: - -```sql --- name: CreateJobHistory :one -INSERT INTO job_history ( - job_id, job_type, user_id, status, params, result, error_message, expires_at -) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8 -) RETURNING *; - --- name: GetJobHistoryByUser :many -SELECT id, job_id, job_type, user_id, status, params, result, error_message, created_at, started_at, completed_at, expires_at -FROM job_history -WHERE user_id = $1 - AND expires_at > NOW() -ORDER BY created_at DESC -LIMIT $2 OFFSET $3; - --- name: CleanupOldJobs :exec -SELECT cleanup_old_jobs(); -``` - -**Why**: Provides database operations for job persistence. - ---- - -### Step 4.4: Save Job Results to Database -**File**: `internal/services/worker.go` - -**Location**: After job completion in `processJob()` - -**Action**: Save to database. First, update Job struct with database field: - -```go -type Worker struct { - jobQueue chan *Job - results map[string]*JobResult - connManager *ConnectionManager - db *database.Queries // NEW - mu sync.RWMutex - wg sync.WaitGroup - ctx context.Context - cancel context.CancelFunc - shuttingDown atomic.Bool -} -``` - -**Action**: Update constructor: - -```go -func NewWorker(numWorkers int, connManager *ConnectionManager, db *database.Queries) *Worker { - // ... existing code ... - w.db = db - return w -} -``` - -**Action**: Save job to database in `processJob()`: - -```go -func (w *Worker) processJob(job *Job) { - var result interface{} - var err error - - // ... process job ... - - // Save to database at the end - if w.db != nil { - ctx := context.Background() - - paramsJSON, _ := json.Marshal(job.Params) - resultJSON, _ := json.Marshal(result) - - expiresAt := time.Now().Add(7 * 24 * time.Hour) // 7 days - - _, dbErr := w.db.CreateJobHistory(ctx, database.CreateJobHistoryParams{ - JobID: job.ID, - JobType: string(job.Type), - UserID: job.UserID, - Status: string(status), - Params: paramsJSON, - Result: resultJSON, - ErrorMessage: errMsg, - ExpiresAt: expiresAt, - }) - - if dbErr != nil { - fmt.Printf("Failed to save job history: %v\n", dbErr) - } - } - - return result, err -} -``` - -**Why**: Persistent job history for audit and debugging. - ---- - -### Step 4.5: Add Job History API Endpoint -**File**: `internal/handlers/jobs.go` - -**Location**: After `GetJobStatus()` - -**Action**: Add job history endpoint: - -```go -// GetJobHistory returns job history for a user -func (h *JobsHandler) GetJobHistory(c echo.Context) error { - userID := c.Get("user_id").(string) - - // Parse query parameters - limit := 100 - if limitParam := c.QueryParam("limit"); limitParam != "" { - if l, err := strconv.Atoi(limitParam); err == nil { - limit = l - } - } - - offset := 0 - if offsetParam := c.QueryParam("offset"); offsetParam != "" { - if o, err := strconv.Atoi(offsetParam); err == nil { - offset = o - } - } - - ctx := context.Background() - - // Get job history - history, err := h.db.GetJobHistoryByUser(ctx, database.GetJobHistoryByUserParams{ - UserID: userID, - Limit: int32(limit), - Offset: int32(offset), - }) - - if err != nil { - return c.JSON(http.StatusInternalServerError, map[string]string{ - "error": "Failed to fetch job history", - }) - } - - return c.JSON(http.StatusOK, map[string]interface{}{ - "history": history, - "count": len(history), - }) -} -``` - -**Action**: Add route in router: - -```go -jobsGroup.GET("/history", cfg.JobsHandler.GetJobHistory) -``` - -**Why**: Allows users to view their job history (imports, conversions, etc.). - ---- - -### Step 4.6: Add Job History Cleanup Task -**File**: `internal/services/worker.go` - -**Location**: Add periodic cleanup function - -**Action**: Add cleanup goroutine: - -```go -func (w *Worker) StartJobHistoryCleanup(ctx context.Context, interval time.Duration) { - ticker := time.NewTicker(interval) - - go func() { - for { - select { - case <-ticker.C: - if w.db != nil { - _, err := w.db.CleanupOldJobs(context.Background()) - if err != nil { - fmt.Printf("Failed to cleanup old jobs: %v\n", err) - } else { - fmt.Printf("Cleaned up old job history entries\n") - } - } - case <-ctx.Done(): - return - } - } - } -} -``` - -**Action**: Start in main.go: - -```go -// Start job history cleanup (runs daily) -worker.StartJobHistoryCleanup(context.Background(), 24*time.Hour) -``` - -**Why**: Automatically removes job history older than 7 days. - ---- - -### Step 4.8: Verify Job Queue Enhancements -**Action**: Test the enhancements: - -```bash -# Build everything -podman compose --profile tests build - -# Run database migration -podman compose exec db psql -U bookhoard_user -d bookhoard_db -f /docker/schema/schema.sql - -# Test job creation and retry -# Test job history persistence -# Test job cleanup -``` - -**Commit Phase 5**: -```bash -git add internal/services/worker.go internal/database/queries.sql internal/handlers/jobs.go internal/router/router.go cmd/server/main.go -git commit -m "feat: Add job queue priority, persistence, and history - -Job Queue Enhancements: -- Add Priority field to Job struct (0=low, 5=medium, 10=high) -- Add job history table for persistence -- Save job results to database (survives restarts) -- Add automatic cleanup of old jobs (7 day retention) -- Add job history API endpoint for users - -Database Changes: -- Add job_history table -- Add cleanup_old_jobs() function -- Add indexes for job lookup -- Add job history queries - -API Changes: -- GET /api/jobs/history - List user's job history -- Query parameters: limit, offset - -Benefits: -- Jobs survive server restarts -- Audit trail of async operations -- Historical job data for analytics -- Automatic cleanup prevents database bloat - -Files modified: -- internal/services/worker.go (priority, persistence) -- internal/database/queries.sql (job_history table and queries) -- internal/handlers/jobs.go (job history endpoint) -- internal/router/router.go (history route) -- cmd/server/main.go (cleanup task, db param) - -Note: True priority queue requires restructuring jobQueue channel -or using a priority queue library. Current implementation adds -Priority field but processes jobs in FIFO order. Priority processing -would require more significant refactoring." -``` - ---- - -## Summary - -### Total Time Estimate -- **Phase 0.5**: 2-3 hours (fix fsnotify reliability) -- **Phase 1**: 6-8 hours (job queue expansion) -- **Phase 2**: 2-3 hours (WebSocket scan progress) -- **Phase 3**: 2 hours (caching and monitoring) -- **Phase 4**: 4-6 hours (job queue enhancements) - -**Total: 16-22 hours of development time** - -### What You Get - -1. **Reliable fsnotify** - Directory-based watching prevents event overflow -2. **7 async job types** - Import, convert, thumbnails, reindex, backup, analytics, sync -3. **Real-time scan progress** - WebSocket instead of polling -4. **Cached settings** - Reduced database load -5. **Enhanced monitoring** - /health endpoint shows scan status -6. **Job persistence** - Jobs survive restarts -7. **Job history** - Audit trail of async operations -8. **Automatic cleanup** - Old jobs removed after 7 days - -### Key Design Decision - -**Job queue = concurrency control** -- All scans (manual, polling) go through job queue -- Worker pool (3 workers) processes jobs one at a time -- No concurrent scans possible - job queue serializes everything -- Non-blocking APIs - jobs return immediately with job ID -- User can poll `/api/jobs/:jobId` for status - -### Infrastructure Reuse - -This plan maximizes reuse of existing infrastructure: -- ✅ Job queue (was 10% utilized, now 90%) -- ✅ WebSocket (was only for sync, now also scans) -- ✅ Worker pool (same workers handle all job types) -- ✅ Progress callbacks (same pattern for all jobs) -- ✅ Status API pattern (consistent across all operations) -- ✅ Sync queue retry/priority patterns (can apply to job queue) - -### What Was Removed (Compared to Mutex Plan) - -- ❌ No scanMutex / scanMu / RWMutex -- ❌ No atomic `scanInProgress` flag (job queue handles this) -- ❌ No TryLock() in polling (job queue serializes) -- ❌ No lock/unlock in ScanFolders() (job queue serializes) -- ❌ No GetStats() locking (job queue prevents conflicts) -- ❌ No IsScanInProgress() method (check worker instead) - -The job queue **IS** the concurrency control mechanism. Much simpler and cleaner than mutex approach! - -### Files Modified - -**Phase 0.5** (2-3 hours): -- `internal/services/media_scanner.go` (directory-based watching, file stability checks) -- `internal/services/media_scanner_test.go` (new unit tests) -- `cmd/server/tests/fsnotify_integration_test.go` (new integration tests) - -**Phase 1** (6-8 hours): -- `internal/services/worker.go` (7 new job handlers) -- `internal/handlers/jobs.go` (new file) -- `internal/router/router.go` (job routes) -- `cmd/server/main.go` (jobs handler initialization) - -**Phase 2** (2-3 hours): -- `internal/sync/websocket.go` (message types) -- `internal/services/worker.go` (connManager, broadcasting) -- `internal/handlers/scanner.go` (add user_id to jobs) -- `web/src/admin.ts` (WebSocket message handlers) -- `cmd/server/main.go` (pass connManager to worker) - -**Phase 3** (2 hours): -- `internal/services/cache.go` (new file) -- `internal/services/media_scanner.go` (cache integration) -- `internal/router/frontend.go` (enhanced health check) - -**Phase 4** (4-6 hours): -- `internal/services/worker.go` (priority, persistence) -- `internal/database/queries.sql` (job_history table) -- `internal/handlers/jobs.go` (job history endpoint) -- `internal/router/router.go` (history route) -- `cmd/server/main.go` (cleanup task, db param) - -Your job queue infrastructure is now fully utilized! diff --git a/ECHO_V5_MIGRATION.md b/ECHO_V5_MIGRATION.md new file mode 100644 index 0000000..92e7ab2 --- /dev/null +++ b/ECHO_V5_MIGRATION.md @@ -0,0 +1,338 @@ +# Echo v5 Migration Plan + +## Overview + +This document outlines the steps required to migrate from Echo v4 to Echo v5, including API changes, type signature updates, and middleware modifications. + +## Errors Identified + +### 1. Deprecated Middleware + +**Error:** `echomiddleware.Logger undefined` + +**Solution:** Replace `echomiddleware.Logger()` with `echomiddleware.RequestLogger()` + +**Files affected:** +- `cmd/server/main.go` (line 142) +- `internal/router/router.go` (line 144) + +### 2. Removed API Methods + +**Error:** `a.echo.Close undefined (type *echo.Echo has no field or method Close)` + +**Solution:** Remove the `echo.Close()` call as v5 uses different graceful shutdown mechanism + +**Files affected:** +- `internal/app/app.go` (line 86) + +### 3. Type Signature & Response API Changes + +**Error:** Multiple type mismatches and Response API changes in Echo v5 + +**Root Cause:** Echo v5 uses `*echo.Context` (pointer) for handlers but some code uses `echo.Context` (value). Response wrapper API also changed. + +**Solution:** +- Update all middleware to use `*echo.Context` +- Update helper functions to accept `*echo.Context` +- Fix Response wrapper usage for Echo v5 API + +**Files affected:** +- `internal/middleware/device_auth.go` (lines 38, 170, 212) +- `internal/middleware/error_handler.go` (lines 44, 69, 82, 84, 87, 89) +- `internal/middleware/rate_limiter.go` (line 102) +- `internal/middleware/request_tracing.go` (lines 48, 58, 60) +- `internal/middleware/security.go` (line 14) + +--- + +## Step-by-Step Fixes + +### Step 1: Fix Deprecated Logger Middleware + +#### File: `cmd/server/main.go` + +**Line 142:** +```go +// BEFORE: +e.Use(echomiddleware.Logger()) + +// AFTER: +e.Use(echomiddleware.RequestLogger()) +``` + +#### File: `internal/router/router.go` + +**Line ~144:** +```go +// BEFORE: +e.Use(echomiddleware.Logger()) + +// AFTER: +e.Use(echomiddleware.RequestLogger()) +``` + +--- + +### Step 2: Fix Removed API Methods + +#### File: `internal/app/app.go` + +**Lines 26-32 - Initialize server field:** +```go +// REPLACE lines 26-32 with: +func New(echo *echo.Echo) *App { + return &App{ + echo: echo, + server: nil, // Will be set in StartServer() + shutdownTimeout: 30 * time.Second, + shutdownDone: make(chan struct{}), + } +} +``` + +**Add StartServer method after New() (after line 32):** +```go +// StartServer creates HTTP server and starts listening +func (a *App) StartServer(addr string) error { + a.server = &http.Server{ + Addr: addr, + Handler: a.echo, + } + + // Start HTTP server in background + go func() { + if err := a.server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("Server failed to start: %v", err) + } + }() + + return nil +} +``` + +**Lines 82-94 - Replace goroutine in Shutdown():** +```go +// REPLACE lines 82-94 with: + // Stop accepting new connections and shutdown HTTP server + log.Println("Stopping HTTP server...") + if a.server != nil { + ctx, cancel := context.WithTimeout(context.Background(), a.shutdownTimeout) + defer cancel() + + if err := a.server.Shutdown(ctx); err != nil { + log.Printf("Error stopping HTTP server: %v", err) + } + } + + log.Println("All services stopped") +``` + +#### File: `cmd/server/main.go` + +**Lines 197-214 - Replace server startup:** +```go +// REPLACE lines 197-214 with: + // ======================================================================== + // START SERVER (managed by app lifecycle) + // ======================================================================== + + log.Printf("Starting server on port %s", cfg.ServerPort) + + // Start HTTP server + if err := application.StartServer(":" + cfg.ServerPort); err != nil { + log.Fatalf("Failed to start server: %v", err) + } + + // Start application lifecycle (blocks until shutdown signal) + if err := application.Start(); err != nil { + log.Fatalf("Application error: %v", err) + } +``` + +--- + +### Step 3: Fix Middleware Type Signatures + +#### Pattern for Echo v5 Middleware + +**Echo v5 Middleware Pattern:** +```go +func MiddlewareFunc(next echo.HandlerFunc) echo.HandlerFunc { + return func(c *echo.Context) error { // ← POINTER (required in v5) + // ... middleware logic + return next(c) // ← c is already a pointer, no & needed + } +} +``` + +#### File: `internal/middleware/device_auth.go` + +**Line 38:** +```go +// BEFORE: +return func(c echo.Context) error { + +// AFTER: +return func(c *echo.Context) error { +``` + +**Line 170:** +```go +// BEFORE: +return func(c echo.Context) error { + +// AFTER: +return func(c *echo.Context) error { +``` + +**Line 212:** +```go +// BEFORE: +return func(c echo.Context) error { + +// AFTER: +return func(c *echo.Context) error { +``` + +#### File: `internal/middleware/error_handler.go` + +**Line 44:** +```go +// BEFORE: +func RespondWithError(c echo.Context, code int, message string, err error) error { + +// AFTER: +func RespondWithError(c *echo.Context, code int, message string, err error) error { +``` + +**Line 69:** +```go +// BEFORE: +func RespondWithHTTPError(c echo.Context, httpErr *HTTPError) error { + +// AFTER: +func RespondWithHTTPError(c *echo.Context, httpErr *HTTPError) error { +``` + +**Line 82:** +```go +// BEFORE: +func WrapHandler(fn func(c echo.Context) error) echo.HandlerFunc { + return func(c *echo.Context) error { + err := fn(c) + +// AFTER: +func WrapHandler(fn func(*echo.Context) error) echo.HandlerFunc { + return func(c *echo.Context) error { + err := fn(c) // c is already *echo.Context +``` + +**Lines 84, 87, 89:** No change - calls to helpers will now work with updated signatures + +#### File: `internal/middleware/rate_limiter.go` + +**Line 102:** +```go +// BEFORE: +return func(c echo.Context) error { + +// AFTER: +return func(c *echo.Context) error { +``` + +#### File: `internal/middleware/request_tracing.go` + +**Line 48:** +```go +// BEFORE: +return func(c echo.Context) error { + +// AFTER: +return func(c *echo.Context) error { +``` + +**Lines 57-61:** +```go +// BEFORE: +recorder := &responseWriter{ + ResponseWriter: c.Response().Writer, +} +c.Response().Writer = recorder + +// AFTER: +recorder := &responseWriter{ + ResponseWriter: *c.Response(), // Dereference: *echo.Response -> http.ResponseWriter +} +``` + +**Note:** `c.Response().Status` (line 108) should work - Status field exists in Echo v5 + +#### File: `internal/middleware/security.go` + +**Line 14:** +```go +// BEFORE: +return func(c echo.Context) error { + +// AFTER: +return func(c *echo.Context) error { +``` + +--- + +## WebSocket Fix + +After Echo v5 migration, the WebSocket handler should now work natively: + +### File: `internal/handlers/websocket.go` + +The existing code should now work: +```go +ws, err := upgrader.Upgrade(c.Response(), c.Request(), nil) +``` + +This previously failed with "response does not implement http.Hijacker" but Echo v5's Response now supports the `rwUnwrapper` interface needed for WebSocket upgrades. + +--- + +## Verification + +After making all changes: + +1. **Build the project:** + ```bash + go build ./cmd/server + ``` + +2. **Run tests:** + ```bash + go test ./cmd/server/tests/... -v + ``` + +3. **Test WebSocket manually:** + - Get JWT token: `curl -X POST http://localhost:8765/api/auth/login -H "Content-Type: application/json" -d '{"login":"testuser@tests.bookhoard.internal","password":"Test@Pass123!"}' | jq -r '.access_token'` + - Connect in browser console: `new WebSocket('ws://localhost:8765/ws/sync?token=YOUR_TOKEN')` + +--- + +## Rollback Plan + +If Echo v5 migration fails: + +1. Revert go.mod changes: + ```go + // Change back to v4: + github.com/labstack/echo/v4 v4.15.1 + github.com/labstack/echo-jwt/v4 v4.4.0 + ``` + +2. Restore import statements in all modified files +3. Revert all code changes + +--- + +## References + +- [Echo v5 Changelog](https://github.com/labstack/echo/blob/master/CHANGELOG.md) +- [Echo v5 Middleware Documentation](https://echo.labstack.com/docs/middleware) +- [Echo WebSocket Cookbook](https://echo.labstack.com/docs/cookbook/websocket)