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