diff --git a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md index d9ad6b2..9682ad1 100644 --- a/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md +++ b/COMPLETE_INFRASTRUCTURE_ENHANCEMENT_PLAN.md @@ -785,174 +785,9 @@ Benefits: - Works on Docker, network mounts" ``` -## Phase 1: Core Fixes Using Job Queue (2-3 hours) +## Phase 1: Job Queue Expansion (6-8 hours) -### Step 1.1: Add Watching Atomic Flag to MediaScanner -**File**: `internal/services/media_scanner.go` - -**Location**: Add to MediaScanner struct (around line 45-70) - -**Action**: Add atomic field: - -```go -watching atomic.Bool // Prevents duplicate WatchChanges() calls -``` - -**Why**: `WatchChanges()` starts 3 goroutines with no tracking. If called twice, you get duplicate goroutines running → memory leak, CPU waste, duplicated polling. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.2: Protect WatchChanges() from Duplicate Calls -**File**: `internal/services/media_scanner.go` - -**Location**: `WatchChanges()` function (lines 1546-1591) - -**Current code** (line 1547): -```go -func (s *MediaScanner) WatchChanges(ctx context.Context) { -``` - -**Action**: Add check at start of function: - -```go -func (s *MediaScanner) WatchChanges(ctx context.Context) error { - // Prevent duplicate calls (which would launch duplicate goroutines) - if !s.watching.CompareAndSwap(false, true) { - return fmt.Errorf("already watching") - } - - // Reset flag when context is cancelled - go func() { - <-ctx.Done() - s.watching.Store(false) - }() - - // Start the debounced event processor - go s.processEventQueue(ctx) - - // Start polling fallback - go s.StartPolling(ctx) - - // Handle fsnotify events - queue them for debouncing - go func() { - // ... existing event handling code ... - }() - - return nil -} -``` - -**Note**: Change return type from `void` to `error`. All callers will need to handle the error. - -**Why**: Prevents goroutine leaks. Returns error if already watching (callers can decide whether to log or ignore). - -**Verification**: -1. Run `go build ./internal/services/` to ensure compiles -2. Check all callers of `WatchChanges()` to ensure error is handled (or logged) - ---- - -### Step 1.3: Fix 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 1.4: Fix Test Expectation (60 → 300) -**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**: Change expectation to 300 (5 minutes): - -```go -assert.Equal(t, float64(300), response["scan_poll_interval_seconds"]) -``` - -**Why**: Database default is now 300 (5 min polling interval). Test should match reality. Combined with Step 1.3, this ensures test passes reliably. - -**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 1.5: Fix Default Value Inconsistency -**File**: `internal/services/media_scanner.go` - -**Location**: `GetPollInterval()` function (lines 111-127) - -**Current code** (line 121): -```go -return 30 * time.Second // ← Different from handler default (60) -``` - -**Action**: Change to match handler default: - -```go -return 60 * time.Second // ← Matches handler and schema default -``` - -**Why**: -- Handler defaults to 60s (system_settings.go:79) -- Schema initializes to 60s (schema.sql:46) -- Scanner should also default to 60s for consistency -- Reduces confusion - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.6: Add JobTypeSetFolders to Worker +### Step 1.1: Add All Job Type Constants **File**: `internal/services/worker.go` **Location**: JobType constants (lines 24-28) @@ -964,485 +799,11 @@ const ( ) ``` -**Action**: Add new job type: - -```go -const ( - JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" // NEW -) -``` - -**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 1.7: 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) -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 -} -``` - -**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 1.8: Update StartScanner Handler to Use Job Queue -**File**: `internal/handlers/scanner.go` - -**Location**: `StartScanner()` function (lines 72-142) - -**Current code** (lines 120-131): -```go -h.mu.Lock() -defer h.mu.Unlock() - -if err := h.scanner.SetFolders(req.FolderPaths); err != nil { - return c.JSON(http.StatusBadRequest, map[string]string{ - "error": "Failed to set folders", - }) -} - -h.scanner.WatchChanges(h.watchModeCtx) -``` - -**Action**: Replace blocking SetFolders() call with async job: - -```go -// Create job to configure folders asynchronously -job := &services.Job{ - ID: uuid.New().String(), - Type: services.JobTypeSetFolders, - Params: map[string]interface{}{ - "folders": req.FolderPaths, - "db": h.db, - }, - Status: services.JobStatusPending, -} - -// Enqueue the job -if err := h.worker.EnqueueJob(job); err != nil { - return c.JSON(http.StatusInternalServerError, map[string]string{ - "error": fmt.Sprintf("Failed to enqueue folder configuration job: %v", err), - }) -} - -// Start watch mode (non-blocking, starts goroutines) -if err := h.scanner.WatchChanges(h.watchModeCtx); err != nil { - // Log but don't fail - already watching is OK - fmt.Printf("WatchChanges warning: %v\n", err) -} - -// Return immediately with job ID -return c.JSON(http.StatusAccepted, map[string]interface{}{ - "message": "Scanner started - folder configuration enqueued", - "job_id": job.ID, - "status": "pending", -}) -``` - -**Note**: Remove `h.mu.Lock()` and `defer h.mu.Unlock()` - no longer needed since SetFolders() is async. - -**Why**: -- API returns immediately instead of blocking on SetFolders() -- Folder configuration happens in background job -- User can check job status with `/api/scanner/status/:jobId` -- No deadlocks with running scans - -**Verification**: Run `go build ./internal/handlers/` to ensure compiles. - ---- - -### Step 1.9: Update StartWatchModeForLibrary to Use Job Queue -**File**: `internal/handlers/scanner.go` - -**Location**: `StartWatchModeForLibrary()` function (lines 179-237) - -**Current code** (lines 202-207): -```go -scanner := services.NewMediaScanner(h.db) - -if err := scanner.SetFolders(folderPaths); err != nil { - return fmt.Errorf("failed to set folders: %w", err) -} -``` - -**Action**: Replace with async job: - -```go -// Create scanner for this library -scanner := services.NewMediaScanner(h.db) - -// Enqueue folder configuration as a job -job := &services.Job{ - ID: uuid.New().String(), - Type: services.JobTypeSetFolders, - Params: map[string]interface{}{ - "folders": folderPaths, - "db": h.db, - }, - Status: services.JobStatusPending, -} - -if err := h.worker.EnqueueJob(job); err != nil { - return fmt.Errorf("failed to enqueue folder configuration job: %w", err) -} -``` - -**Why**: Same benefits as Step 1.8 - non-blocking, async, no deadlocks. - -**Verification**: Run `go build ./internal/handlers/` to ensure compiles. - ---- - -### Step 1.10: Handle WatchChanges() Return Value -**File**: `internal/handlers/scanner.go` - -**Location**: Both `StartScanner()` (line 131) and `StartWatchModeForLibrary()` (line 210) - -**Current code**: -```go -h.scanner.WatchChanges(h.watchModeCtx) -``` - -**Action**: Handle the error return value: - -```go -if err := h.scanner.WatchChanges(h.watchModeCtx); err != nil { - // Log but don't fail - already watching is OK - fmt.Printf("WatchChanges warning: %v\n", err) -} -``` - -**Why**: `WatchChanges()` now returns an error if already watching. This is not a fatal error - it's actually fine (already have goroutines running). Just log it. - -**Verification**: Run `go build ./internal/handlers/` to ensure compiles. - ---- - -### Step 1.11: Fix Polling to Check Job Queue -**File**: `internal/services/media_scanner.go` - -**Location**: `StartPolling()` function (lines 1688-1713) - -**Current code** (lines 1705-1711): -```go -case <-ticker.C: - interval = s.GetPollInterval() - fmt.Printf("Running polling fallback sync (interval: %v)...\n", interval) - if err := s.SyncFilesystemWithDatabase(ctx); err != nil { - fmt.Printf("Polling sync error: %v\n", err) - } -``` - -**Problem**: Calling `SyncFilesystemWithDatabase()` directly could conflict with manual scans. - -**Solution**: Make polling use job queue instead: - -```go -case <-ticker.C: - interval = s.GetPollInterval() - - // Check if we have a worker reference - // Note: MediaScanner doesn't have worker reference, need to add it - // For now, we'll skip polling if a recent scan job completed recently - - // Simple approach: Skip this poll if less than interval/2 since last scan - // This prevents pile-up without needing mutex - - // Run polling sync - fmt.Printf("Running polling fallback sync (interval: %v)...\n", interval) - if err := s.SyncFilesystemWithDatabase(ctx); err != nil { - fmt.Printf("Polling sync error: %v\n", err) - } -``` - -**Better Solution** (requires adding worker reference to scanner): -```go -// Add to MediaScanner struct: -worker *Worker // NEW - -// In StartPolling(): -case <-ticker.C: - interval = s.GetPollInterval() - - // Create polling sync job - job := &services.Job{ - ID: uuid.New().String(), - Type: services.JobTypeScan, - Params: map[string]interface{}{ - "scan_type": "polling", - "db": s.db, - }, - Status: services.JobStatusPending, - } - - // Try to enqueue - will skip if queue is full - select { - case s.worker.jobQueue <- job: - fmt.Printf("Polling scan enqueued\n") - default: - // Queue full, skip this poll tick - fmt.Printf("Polling skipped: worker queue full (scan already in progress)\n") - } -``` - -**Why**: Polling scans go through job queue, naturally serialized with manual scans. - -**Verification**: Run `go build ./internal/services/` to ensure compiles. - ---- - -### Step 1.12: Add Test for Job Queue Serialization -**File**: `cmd/server/tests/scan_settings_integration_test.go` - -**Location**: After existing tests (end of file, around line 179) - -**Action**: Add new test to verify job queue prevents concurrent scans: - -```go -func TestScanSettings_JobQueueSerialization(t *testing.T) { - setup := setupTestServer(t) - defer setup.Close() - - t.Run("Concurrent scan requests are serialized by job queue", func(t *testing.T) { - token := setup.Token - - // Create a test library first - createLibReq := map[string]interface{}{ - "name": "concurrent-test-library", - "description": "Test library for job queue", - "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) - - // Start first scan - scanReq1, _ := http.NewRequest("POST", setup.Server.URL+"/api/libraries/"+libraryID+"/scan", nil) - scanReq1.Header.Set("Authorization", "Bearer "+token) - - // Immediately try second scan - scanReq2, _ := http.NewRequest("POST", setup.Server.URL+"/api/libraries/"+libraryID+"/scan", nil) - scanReq2.Header.Set("Authorization", "Bearer "+token) - - done1 := make(chan bool) - done2 := make(chan bool) - - // Start first scan in background - go func() { - resp, _ := client.Do(scanReq1) - if resp != nil { - resp.Body.Close() - } - done1 <- true - }() - - // Give first scan time to enqueue - time.Sleep(100 * time.Millisecond) - - // Second scan should enqueue (not block) - start2 := time.Now() - go func() { - resp, _ := client.Do(scanReq2) - if resp != nil { - resp.Body.Close() - } - done2 <- true - }() - - // Both scans should complete (serialized by job queue) - <-done1 - <-done2 - - // If we got here without issues, job queue is working - assert.True(t, true, "Job queue serializes scans correctly") - - // Cleanup: Delete test library - deleteReq, _ := http.NewRequest("DELETE", setup.Server.URL+"/api/libraries/"+libraryID, nil) - deleteReq.Header.Set("Authorization", "Bearer "+token) - client.Do(deleteReq) - }) -} -``` - -**Why**: Ensures the job queue properly serializes scan operations. - -**Verification**: Run new test to confirm it passes: -```bash -podman compose --profile tests run --rm tests go test -v -run "TestScanSettings_JobQueueSerialization" ./cmd/server/tests/ -``` - ---- - -### Step 1.13: Verify All Phase 1 Changes -**Action**: Run full test suite for affected files - -```bash -# Test scanner service -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 integration -podman compose --profile tests run --rm tests go test -v -run "TestScanSettings" ./cmd/server/tests/ - -# Ensure full project builds -podman compose --profile tests build -``` - -**Commit Phase 1**: -```bash -git add internal/services/media_scanner.go internal/services/worker.go internal/handlers/scanner.go cmd/server/tests/test_helpers.go cmd/server/tests/scan_settings_integration_test.go -git commit -m "fix: Use job queue for concurrency control (no mutex) - -MediaScanner improvements: -- Add watching atomic flag to prevent duplicate WatchChanges() calls -- WatchChanges() now returns error (prevents goroutine leaks) -- Fix default value: 30s → 60s (matches handler/schema) - -Worker improvements: -- Add JobTypeSetFolders for async folder configuration -- Add processSetFoldersJob() handler -- Folder changes now go through job queue (non-blocking) - -Handler improvements: -- StartScanner() uses job queue for SetFolders() instead of blocking -- StartWatchModeForLibrary() uses job queue for SetFolders() -- Remove h.mu.Lock() from handlers (no longer needed) -- Handle WatchChanges() error return (log if already watching) - -Test improvements: -- Add system_settings snapshot/restore to setupTestServer() -- Ensures test isolation and preserves dev database state -- Fix test expectation: 60 → 300 (5 min polling interval) -- Add job queue serialization test - -Benefits: -- No mutex complexity - job queue handles serialization -- Non-blocking API responses (folder config via job queue) -- Prevents concurrent scans (job queue serializes everything) -- Prevents goroutine leaks from duplicate WatchChanges() calls -- Better test isolation (settings restored after tests) -- Consistent default values (all components use 60s) - -Key Design Decision: -- Job queue is the concurrency control mechanism -- All operations (scans, folder changes) are serialized by worker pool -- No mutex blocking - job queue prevents conflicts naturally -- Non-blocking APIs - jobs return immediately with job ID - -Files modified: -- internal/services/media_scanner.go (atomic flag, default value) -- internal/services/worker.go (JobTypeSetFolders, handler) -- internal/handlers/scanner.go (async folder config) -- cmd/server/tests/test_helpers.go (settings snapshot) -- cmd/server/tests/scan_settings_integration_test.go (fix + new test) - -Related: fsnotify unreliability requires polling as fallback" -``` - ---- - -## Phase 2: Job Queue Expansion (6-8 hours) - -### Step 2.1: Add All Job Type Constants -**File**: `internal/services/worker.go` - -**Location**: JobType constants (lines 24-28) - -**Current code**: -```go -const ( - JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" -) -``` - **Action**: Add all new job types: ```go const ( JobTypeScan JobType = "scan" - JobTypeSetFolders JobType = "set_folders" JobTypeImport JobType = "import" // NEW JobTypeConvert JobType = "convert" // NEW JobTypeThumbnails JobType = "thumbnails" // NEW @@ -1453,13 +814,13 @@ const ( ) ``` -**Why**: Defines all job types the system will support. Job queue is massively underutilized (only 2 types). +**Why**: Defines all job types the system will support. Job queue is massively underutilized (only 1 type). **Verification**: Run `go build ./internal/services/` to ensure compiles. --- -### Step 2.2: Add Job Handlers to Switch Statement +### Step 1.2: Add Job Handlers to Switch Statement **File**: `internal/services/worker.go` **Location**: `processJob()` function (around line 127-132) @@ -1469,8 +830,6 @@ const ( switch job.Type { case JobTypeScan: result, err = w.processScanJob(job) -case JobTypeSetFolders: - result, err = w.processSetFoldersJob(job) default: err = fmt.Errorf("unknown job type: %s", job.Type) } @@ -1482,8 +841,6 @@ default: switch job.Type { case JobTypeScan: result, err = w.processScanJob(job) -case JobTypeSetFolders: - result, err = w.processSetFoldersJob(job) case JobTypeImport: result, err = w.processImportJob(job) case JobTypeConvert: @@ -1509,10 +866,10 @@ default: --- -### Step 2.3: Implement Import Job Handler +### Step 1.3: Implement Import Job Handler **File**: `internal/services/worker.go` -**Location**: Add new function after `processSetFoldersJob()` (around line 282) +**Location**: Add new function after `processScanJob()` (around line 248) **Action**: Add import handler: @@ -1620,7 +977,7 @@ func (w *Worker) processImportJob(job *Job) (interface{}, error) { --- -### Step 2.4: Implement Convert Job Handler +### Step 1.4: Implement Convert Job Handler **File**: `internal/services/worker.go` **Location**: Add new function after `processImportJob()` @@ -1705,7 +1062,7 @@ func (w *Worker) processConvertJob(job *Job) (interface{}, error) { --- -### Step 2.5: Implement Thumbnails Job Handler +### Step 1.5: Implement Thumbnails Job Handler **File**: `internal/services/worker.go` **Location**: Add new function after `processConvertJob()` @@ -1803,7 +1160,7 @@ func (w *Worker) processThumbnailsJob(job *Job) (interface{}, error) { --- -### Step 2.6: Implement Reindex Job Handler +### Step 1.6: Implement Reindex Job Handler **File**: `internal/services/worker.go` **Location**: Add new function after `processThumbnailsJob()` @@ -1870,7 +1227,7 @@ func (w *Worker) processReindexJob(job *Job) (interface{}, error) { --- -### Step 2.7: Implement Backup Job Handler +### Step 1.7: Implement Backup Job Handler **File**: `internal/services/worker.go` **Location**: Add new function after `processReindexJob()` @@ -1938,7 +1295,7 @@ func (w *Worker) processBackupJob(job *Job) (interface{}, error) { --- -### Step 2.8: Implement Analytics Job Handler +### Step 1.8: Implement Analytics Job Handler **File**: `internal/services/worker.go` **Location**: Add new function after `processBackupJob()` @@ -2084,7 +1441,7 @@ func getTopN(m map[string]int, n int) map[string]int { --- -### Step 2.9: Implement Sync Job Handler +### Step 1.9: Implement Sync Job Handler **File**: `internal/services/worker.go` **Location**: Add new function after `processAnalyticsJob()` @@ -2156,7 +1513,7 @@ func (w *Worker) processSyncJob(job *Job) (interface{}, error) { --- -### Step 2.10: Create Job Management Handler +### Step 1.10: Create Job Management Handler **File**: `internal/handlers/jobs.go` (new file) **Action**: Create new handler for job management: @@ -2260,7 +1617,7 @@ func (h *JobsHandler) GetJobStatus(c echo.Context) error { --- -### Step 2.11: Register Job Routes +### Step 1.11: Register Job Routes **File**: `internal/router/router.go` **Location**: Add jobs handler to Config struct (around line 39-64) @@ -2301,7 +1658,7 @@ jobsGroup.GET("/:jobId", cfg.JobsHandler.GetJobStatus) --- -### Step 2.12: Initialize JobsHandler in main.go +### Step 1.12: Initialize JobsHandler in main.go **File**: `cmd/server/main.go` **Location**: Around line 92 (before systemSettingsHandler creation) @@ -2326,7 +1683,7 @@ cfg.JobsHandler = jobsHandler --- -### Step 2.13: Add Job Handler Unit Tests +### Step 1.13: Add Job Handler Unit Tests **File**: `internal/services/worker_test.go` (new) **Action**: Create unit tests for all job handlers: @@ -2387,7 +1744,7 @@ func TestProcessThumbnailsJob(t *testing.T) { --- -### Step 2.14: Verify Job Queue Expansion +### Step 1.14: Verify Job Queue Expansion **Action**: Run full test suite: ```bash @@ -2407,10 +1764,9 @@ 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 8 async operations +git commit -m "feat: Expand job queue to handle 7 async operations Job Types Added: -- JobTypeSetFolders: Async folder configuration - JobTypeImport: Import from OPDS feeds or Calibre - JobTypeConvert: Convert EPUB to KEPUB - JobTypeThumbnails: Generate missing book covers @@ -2447,9 +1803,9 @@ Files modified: --- -## Phase 3: WebSocket Scan Progress (2-3 hours) +## Phase 2: WebSocket Scan Progress (2-3 hours) -### Step 3.1: Add Scan Progress Message Types +### Step 2.1: Add Scan Progress Message Types **File**: `internal/sync/websocket.go` **Location**: Message type constants (around line 20-30) @@ -2488,7 +1844,7 @@ const ( --- -### Step 3.2: Pass Connection Manager to Worker +### Step 2.2: Pass Connection Manager to Worker **File**: `internal/services/worker.go` **Location**: Worker struct definition (around line 14-20) @@ -2537,7 +1893,7 @@ func NewWorker(numWorkers int, connManager *ConnectionManager) *Worker { --- -### Step 3.3: Update Worker Initialization in main.go +### Step 2.3: Update Worker Initialization in main.go **File**: `cmd/server/main.go` **Location**: Where worker is created (around line 30-50) @@ -2563,7 +1919,7 @@ worker := services.NewWorker(3, connManager) --- -### Step 3.4: Add User ID to Job +### Step 2.4: Add User ID to Job **File**: `internal/services/worker.go` **Location**: Job struct (around line 30-48) @@ -2608,7 +1964,7 @@ type Job struct { --- -### Step 3.5: Broadcast Scan Progress from Worker +### Step 2.5: Broadcast Scan Progress from Worker **File**: `internal/services/worker.go` **Location**: `processScanJob()` function (around line 176-247) @@ -2671,7 +2027,7 @@ job.ProgressCallback = func(progress float64, filesScanned, newItems, errors int --- -### Step 3.6: Add User ID to Scan Jobs +### Step 2.6: Add User ID to Scan Jobs **File**: `internal/handlers/scanner.go` **Location**: `ScanLibrary()` function (around line 33-110) @@ -2721,7 +2077,7 @@ job := &services.Job{ --- -### Step 3.7: Add Frontend WebSocket Scan Progress Listener +### Step 2.7: Add Frontend WebSocket Scan Progress Listener **File**: `web/src/admin.ts` or appropriate TypeScript file **Location**: After existing WebSocket connection setup @@ -2787,7 +2143,7 @@ function showScanComplete(data: any) { --- -### Step 3.8: Remove Scan Progress Polling (Optional) +### Step 2.8: Remove Scan Progress Polling (Optional) **File**: `web/src/admin.ts` **Location**: `pollScanProgress()` function (around line 210-277) @@ -2815,7 +2171,7 @@ function pollScanProgress(jobIds: string[], libraryNames: Record --- -### Step 3.10: Verify WebSocket Scan Progress +### Step 2.10: Verify WebSocket Scan Progress **Action**: Test the full stack: ```bash @@ -2864,9 +2220,9 @@ Note: Can reduce or remove frontend polling since WebSocket provides real-time u --- -## Phase 4: Caching and Monitoring (2 hours) +## Phase 3: Caching and Monitoring (2 hours) -### Step 4.1: Create Settings Cache +### Step 3.1: Create Settings Cache **File**: `internal/services/cache.go` (new file) **Action**: Create settings cache implementation: @@ -2937,7 +2293,7 @@ func (c *SettingsCache) InvalidateKey(key string) { --- -### Step 4.2: Add Cache to MediaScanner +### Step 3.2: Add Cache to MediaScanner **File**: `internal/services/media_scanner.go` **Location**: MediaScanner struct (around line 45-70) @@ -2976,7 +2332,7 @@ func NewMediaScanner(db *database.Queries) *MediaScanner { --- -### Step 4.3: Use Cache in GetPollInterval() +### Step 3.3: Use Cache in GetPollInterval() **File**: `internal/services/media_scanner.go` **Location**: `GetPollInterval()` function (lines 111-127) @@ -3032,7 +2388,7 @@ func (s *MediaScanner) GetPollInterval() time.Duration { --- -### Step 4.4: Use Cache in GetAutoScanEnabled() +### Step 3.4: Use Cache in GetAutoScanEnabled() **File**: `internal/services/media_scanner.go` **Location**: Find or add `GetAutoScanEnabled()` method @@ -3068,7 +2424,7 @@ func (s *MediaScanner) GetAutoScanEnabled() bool { --- -### Step 4.5: Extend /health Endpoint +### Step 3.5: Extend /health Endpoint **File**: `internal/router/frontend.go` **Location**: Health check handler (around line 833-847) @@ -3130,7 +2486,7 @@ func HealthCheck(c echo.Context) error { --- -### Step 4.7: Verify Caching and Monitoring +### Step 3.7: Verify Caching and Monitoring **Action**: Test the changes: ```bash @@ -3178,9 +2534,9 @@ is sufficient for now." --- -## Phase 5: Job Queue Enhancements (4-6 hours) +## Phase 4: Job Queue Enhancements (4-6 hours) -### Step 5.1: Add Job Priority Field +### Step 4.1: Add Job Priority Field **File**: `internal/services/worker.go` **Location**: Job struct (around line 30-48) @@ -3227,7 +2583,7 @@ type Job struct { --- -### Step 5.2: Add Job History Table +### Step 4.2: Add Job History Table **File**: `internal/database/queries.sql` **Location**: Add new table at end @@ -3268,7 +2624,7 @@ $$ LANGUAGE plpgsql; --- -### Step 5.3: Add Job History Queries +### Step 4.3: Add Job History Queries **File**: `internal/database/queries.sql` **Location**: Add new queries at end @@ -3299,7 +2655,7 @@ SELECT cleanup_old_jobs(); --- -### Step 5.4: Save Job Results to Database +### Step 4.4: Save Job Results to Database **File**: `internal/services/worker.go` **Location**: After job completion in `processJob()` @@ -3372,7 +2728,7 @@ func (w *Worker) processJob(job *Job) { --- -### Step 5.5: Add Job History API Endpoint +### Step 4.5: Add Job History API Endpoint **File**: `internal/handlers/jobs.go` **Location**: After `GetJobStatus()` @@ -3431,7 +2787,7 @@ jobsGroup.GET("/history", cfg.JobsHandler.GetJobHistory) --- -### Step 5.6: Add Job History Cleanup Task +### Step 4.6: Add Job History Cleanup Task **File**: `internal/services/worker.go` **Location**: Add periodic cleanup function @@ -3473,7 +2829,7 @@ worker.StartJobHistoryCleanup(context.Background(), 24*time.Hour) --- -### Step 5.8: Verify Job Queue Enhancements +### Step 4.8: Verify Job Queue Enhancements **Action**: Test the enhancements: ```bash @@ -3534,18 +2890,18 @@ would require more significant refactoring." ## Summary ### Total Time Estimate -- **Phase 1**: 2-3 hours (core fixes using job queue) -- **Phase 2**: 6-8 hours (job queue expansion) -- **Phase 3**: 2-3 hours (WebSocket scan progress) -- **Phase 4**: 2 hours (caching and monitoring) -- **Phase 5**: 4-6 hours (job queue enhancements) +- **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. **No mutex complexity** - Job queue handles serialization -2. **8 async job types** - Import, convert, thumbnails, reindex, backup, analytics, sync, setfolders +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 @@ -3556,7 +2912,7 @@ would require more significant refactoring." ### Key Design Decision **Job queue = concurrency control** -- All scans (manual, polling, folder config) go through job queue +- 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 @@ -3585,32 +2941,30 @@ The job queue **IS** the concurrency control mechanism. Much simpler and cleaner ### Files Modified -**Phase 1** (2-3 hours): -- `internal/services/media_scanner.go` (atomic flag, default value) -- `internal/services/worker.go` (JobTypeSetFolders, processSetFoldersJob) -- `internal/handlers/scanner.go` (async folder config, no blocking) -- `cmd/server/tests/test_helpers.go` (settings snapshot) -- `cmd/server/tests/scan_settings_integration_test.go` (fix + new test) +**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 2** (6-8 hours): +**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 3** (2-3 hours): +**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 4** (2 hours): +**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 5** (4-6 hours): +**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)