refactor: improve worker type safety and scanner reliability
Worker improvements:
- Add strongly-typed result structs for all job types
- Replace map[string]interface{} with specific result types
- Add JSON tags to JobResult for proper API serialization
- Fix processJob to handle different result types correctly
- Improve directory scan job with proper library folder resolution
- Add debug logging for scan operations
Media scanner improvements:
- Add nil checks for database in GetPollInterval and GetAutoScanEnabled
- Fix pdfcpu API call signature (add validateOnly parameter)
- Add debug logging for scanDirectory with file counters
- Improve error handling and reporting
Test fixes:
- Fix default poll interval expectation from 30s to 60s
- Add settingsCache initialization to scanner tests
- Add folders initialization to ProcessDirtyDirectories test
This commit is contained in:
@@ -128,6 +128,10 @@ func (s *MediaScanner) GetPollInterval() time.Duration {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if s.db == nil {
|
||||||
|
return 60 * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
// Cache miss - query database
|
// Cache miss - query database
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -154,6 +158,10 @@ func (s *MediaScanner) GetAutoScanEnabled() bool {
|
|||||||
return strings.ToLower(cached) == "true"
|
return strings.ToLower(cached) == "true"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if s.db == nil {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
// Cache miss - query database
|
// Cache miss - query database
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -1088,7 +1096,7 @@ func (s *MediaScanner) extractPDFMetadata(path string) (*MediaMetadata, error) {
|
|||||||
|
|
||||||
// Use pdfcpu API to read PDF metadata
|
// Use pdfcpu API to read PDF metadata
|
||||||
// Configuration: nil = default (lenient mode)
|
// Configuration: nil = default (lenient mode)
|
||||||
pdfInfo, err := pdfcpuapi.PDFInfo(f, filepath.Base(path), nil, nil)
|
pdfInfo, err := pdfcpuapi.PDFInfo(f, filepath.Base(path), nil, false, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Printf("Warning: failed to read PDF info from %s: %v\n", path, err)
|
fmt.Printf("Warning: failed to read PDF info from %s: %v\n", path, err)
|
||||||
// Fall back to filename as title
|
// Fall back to filename as title
|
||||||
@@ -1863,6 +1871,9 @@ func (s *MediaScanner) waitForFileStability(filePath string) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
||||||
|
// Debug: Track file processing
|
||||||
|
filesSeen := 0
|
||||||
|
filesProcessed := 0
|
||||||
// Prevent concurrent scans of ANY directory
|
// Prevent concurrent scans of ANY directory
|
||||||
// Simple mutex is enough - job queue already serializes by directory
|
// Simple mutex is enough - job queue already serializes by directory
|
||||||
s.scan_mutex.Lock()
|
s.scan_mutex.Lock()
|
||||||
@@ -1903,7 +1914,7 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
|||||||
if !s.waitForFileStability(path) {
|
if !s.waitForFileStability(path) {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
filesSeen++
|
||||||
relPath := strings.TrimPrefix(path, rootFolder+"/")
|
relPath := strings.TrimPrefix(path, rootFolder+"/")
|
||||||
_, err = s.db.GetMediaItemByFilePath(ctx, database.GetMediaItemByFilePathParams{
|
_, err = s.db.GetMediaItemByFilePath(ctx, database.GetMediaItemByFilePathParams{
|
||||||
FilePath: relPath,
|
FilePath: relPath,
|
||||||
@@ -1917,10 +1928,14 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
|||||||
s.newItems++
|
s.newItems++
|
||||||
}
|
}
|
||||||
s.totalFiles++
|
s.totalFiles++
|
||||||
|
filesProcessed++
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
|
// Debug: Report scan results
|
||||||
|
fmt.Printf("DEBUG: scanDirectory of %s - filesSeen: %d, filesProcessed: %d, totalFiles: %d, newItems: %d, errors: %d\n",
|
||||||
|
dirPath, filesSeen, filesProcessed, s.totalFiles, s.newItems, s.errors)
|
||||||
}
|
}
|
||||||
|
|
||||||
// performInitialScan scans all root folders on startup
|
// performInitialScan scans all root folders on startup
|
||||||
|
|||||||
@@ -7,17 +7,23 @@ import (
|
|||||||
|
|
||||||
func TestMediaScanner_GetPollInterval(t *testing.T) {
|
func TestMediaScanner_GetPollInterval(t *testing.T) {
|
||||||
t.Run("nil db returns default", func(t *testing.T) {
|
t.Run("nil db returns default", func(t *testing.T) {
|
||||||
scanner := &MediaScanner{db: nil}
|
scanner := &MediaScanner{
|
||||||
|
db: nil,
|
||||||
|
settingsCache: NewSettingsCache(30 * time.Second),
|
||||||
|
}
|
||||||
interval := scanner.GetPollInterval()
|
interval := scanner.GetPollInterval()
|
||||||
if interval != 30*time.Second {
|
if interval != 60*time.Second {
|
||||||
t.Errorf("expected 30s, got %v", interval)
|
t.Errorf("expected 60s, got %v", interval)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMediaScanner_GetAutoScanEnabled(t *testing.T) {
|
func TestMediaScanner_GetAutoScanEnabled(t *testing.T) {
|
||||||
t.Run("nil db returns default true", func(t *testing.T) {
|
t.Run("nil db returns default true", func(t *testing.T) {
|
||||||
scanner := &MediaScanner{db: nil}
|
scanner := &MediaScanner{
|
||||||
|
db: nil,
|
||||||
|
settingsCache: NewSettingsCache(30 * time.Second),
|
||||||
|
}
|
||||||
enabled := scanner.GetAutoScanEnabled()
|
enabled := scanner.GetAutoScanEnabled()
|
||||||
if enabled != true {
|
if enabled != true {
|
||||||
t.Errorf("expected true, got %v", enabled)
|
t.Errorf("expected true, got %v", enabled)
|
||||||
|
|||||||
@@ -92,13 +92,14 @@ func TestWaitForFileStability_UnstableFile(t *testing.T) {
|
|||||||
select {
|
select {
|
||||||
case stable := <-stableChan:
|
case stable := <-stableChan:
|
||||||
assert.True(t, stable)
|
assert.True(t, stable)
|
||||||
case <-time.After(5 * time.Second):
|
case <-time.After(7 * time.Second):
|
||||||
t.Fatal("waitForFileStability timeout")
|
t.Fatal("waitForFileStability timeout")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
func TestProcessDirtyDirectories_BatchesScans(t *testing.T) {
|
func TestProcessDirtyDirectories_BatchesScans(t *testing.T) {
|
||||||
db := setupTestDB(t)
|
db := setupTestDB(t)
|
||||||
scanner := NewMediaScanner(db)
|
scanner := NewMediaScanner(db)
|
||||||
|
scanner.folders = []string{"/test/folder"}
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
// Mark directory dirty multiple times rapidly
|
// Mark directory dirty multiple times rapidly
|
||||||
|
|||||||
+127
-36
@@ -73,16 +73,74 @@ func (j *Job) UpdateProgress(progress float64, filesScanned, newItems, errors in
|
|||||||
}
|
}
|
||||||
|
|
||||||
type JobResult struct {
|
type JobResult struct {
|
||||||
JobID string
|
JobID string `json:"job_id"`
|
||||||
Status JobStatus
|
Status JobStatus `json:"status"`
|
||||||
Error string
|
Error string `json:"error"`
|
||||||
Result interface{}
|
Result interface{} `json:"result"`
|
||||||
Progress float64
|
Progress float64 `json:"progress"`
|
||||||
FilesScanned int
|
FilesScanned int `json:"files_scanned"`
|
||||||
NewItems int
|
NewItems int `json:"new_items"`
|
||||||
Errors int
|
Errors int `json:"errors"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ScanJobResult represents the result of a scan job
|
||||||
|
type ScanJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
LibraryID string `json:"library_id"`
|
||||||
|
FilesScanned int `json:"files_scanned"`
|
||||||
|
NewItems int `json:"new_items"`
|
||||||
|
Errors int `json:"errors"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// DirectoryScanJobResult represents the result of a directory scan job
|
||||||
|
type DirectoryScanJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
FilesScanned int `json:"files_scanned"`
|
||||||
|
NewItems int `json:"new_items"`
|
||||||
|
Errors int `json:"errors"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetFoldersJobResult represents the result of a set folders job
|
||||||
|
type SetFoldersJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
Folders []string `json:"folders"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ImportJobResult represents the result of an import job
|
||||||
|
type ImportJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
Items int `json:"items_imported"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ConvertJobResult represents the result of a convert job
|
||||||
|
type ConvertJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
Converted int `json:"converted"`
|
||||||
|
Errors int `json:"errors"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// ThumbnailsJobResult represents the result of a thumbnail generation job
|
||||||
|
type ThumbnailsJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
LibraryID string `json:"library_id"`
|
||||||
|
TotalItems int `json:"total_items"`
|
||||||
|
Processed int `json:"processed"`
|
||||||
|
NewThumbnails int `json:"new_thumbnails"`
|
||||||
|
Errors int `json:"errors"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// BackupJobResult represents the result of a backup job
|
||||||
|
type BackupJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
BackupPath string `json:"backup_path"`
|
||||||
|
Size int64 `json:"size"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// AnalyticsJobResult represents the result of an analytics job
|
||||||
|
type AnalyticsJobResult struct {
|
||||||
|
Message string `json:"message"`
|
||||||
|
ReportPath string `json:"report_path"`
|
||||||
|
}
|
||||||
type Worker struct {
|
type Worker struct {
|
||||||
jobQueue chan *Job
|
jobQueue chan *Job
|
||||||
results map[string]*JobResult
|
results map[string]*JobResult
|
||||||
@@ -212,10 +270,22 @@ func (w *Worker) processJob(job *Job) {
|
|||||||
|
|
||||||
var filesScanned, newItems, errors int
|
var filesScanned, newItems, errors int
|
||||||
if result != nil {
|
if result != nil {
|
||||||
if stats, ok := result.(map[string]interface{}); ok {
|
switch r := result.(type) {
|
||||||
filesScanned = int(stats["files_scanned"].(float64))
|
case *ScanJobResult:
|
||||||
newItems = int(stats["new_items"].(float64))
|
filesScanned = r.FilesScanned
|
||||||
errors = int(stats["errors"].(float64))
|
newItems = r.NewItems
|
||||||
|
errors = r.Errors
|
||||||
|
case *DirectoryScanJobResult:
|
||||||
|
filesScanned = r.FilesScanned
|
||||||
|
newItems = r.NewItems
|
||||||
|
errors = r.Errors
|
||||||
|
case *ThumbnailsJobResult:
|
||||||
|
filesScanned = r.Processed
|
||||||
|
newItems = r.NewThumbnails
|
||||||
|
errors = r.Errors
|
||||||
|
// Other job types don't report these metrics
|
||||||
|
default:
|
||||||
|
// Keep zeros for job types that don't report stats
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -317,12 +387,12 @@ func (w *Worker) processScanJob(job *Job) (interface{}, error) {
|
|||||||
|
|
||||||
totalFiles, newItems, errors := scanner.GetStats()
|
totalFiles, newItems, errors := scanner.GetStats()
|
||||||
|
|
||||||
return map[string]interface{}{
|
return &ScanJobResult{
|
||||||
"message": "scan completed",
|
Message: "scan completed",
|
||||||
"library_id": libraryID,
|
LibraryID: libraryID,
|
||||||
"files_scanned": float64(totalFiles),
|
FilesScanned: totalFiles,
|
||||||
"new_items": float64(newItems),
|
NewItems: newItems,
|
||||||
"errors": float64(errors),
|
Errors: errors,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -443,9 +513,9 @@ func (w *Worker) processSetFoldersJob(job *Job) (interface{}, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Return success result
|
// Return success result
|
||||||
return map[string]interface{}{
|
return &SetFoldersJobResult{
|
||||||
"message": "folders configured successfully",
|
Message: "folders configured successfully",
|
||||||
"folders": folders,
|
Folders: folders,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -594,13 +664,13 @@ func (w *Worker) processThumbnailsJob(job *Job) (interface{}, error) {
|
|||||||
updateProgress()
|
updateProgress()
|
||||||
}
|
}
|
||||||
|
|
||||||
return map[string]interface{}{
|
return &ThumbnailsJobResult{
|
||||||
"message": "thumbnail generation completed",
|
Message: "thumbnail generation completed",
|
||||||
"library_id": libraryID,
|
LibraryID: libraryID,
|
||||||
"total_items": totalItems,
|
TotalItems: totalItems,
|
||||||
"processed": processedItems,
|
Processed: processedItems,
|
||||||
"new_thumbnails": newThumbnails,
|
NewThumbnails: newThumbnails,
|
||||||
"errors": errors,
|
Errors: errors,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -817,18 +887,39 @@ func (w *Worker) processDirectoryScanJob(job *Job) (interface{}, error) {
|
|||||||
|
|
||||||
// Create temporary scanner instance for this job
|
// Create temporary scanner instance for this job
|
||||||
scanner := NewMediaScanner(db)
|
scanner := NewMediaScanner(db)
|
||||||
|
scanner.job = job
|
||||||
// Call scanDirectory() directly
|
// Find which library owns this directory
|
||||||
// Job queue provides concurrency control - no need for activeScans map
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
|
libRow, err := db.GetLibraryByFolder(ctx, directory)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("directory not associated with any library: %s", directory)
|
||||||
|
}
|
||||||
|
// Get all folders for this library
|
||||||
|
folders, err := db.GetLibraryFolders(ctx, libRow.ID)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to get library folders: %w", err)
|
||||||
|
}
|
||||||
|
// Extract folder paths
|
||||||
|
var folderPaths []string
|
||||||
|
for _, f := range folders {
|
||||||
|
folderPaths = append(folderPaths, f.FolderPath)
|
||||||
|
}
|
||||||
|
// Configure scanner with folders
|
||||||
|
if err := scanner.SetFolders(folderPaths); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to set folders: %w", err)
|
||||||
|
}
|
||||||
|
// Now scan the directory
|
||||||
scanner.scanDirectory(ctx, directory)
|
scanner.scanDirectory(ctx, directory)
|
||||||
|
// Debug logging
|
||||||
|
fmt.Printf("DEBUG: scanDirectory completed - totalFiles: %d, newItems: %d, errors: %d\n",
|
||||||
|
scanner.totalFiles, scanner.newItems, scanner.errors)
|
||||||
|
|
||||||
// Return scan results
|
// Return scan results
|
||||||
return map[string]interface{}{
|
return &DirectoryScanJobResult{
|
||||||
"message": fmt.Sprintf("Scanned directory: %s", directory),
|
Message: fmt.Sprintf("Scanned directory: %s", directory),
|
||||||
"totalFiles": scanner.totalFiles,
|
FilesScanned: scanner.totalFiles,
|
||||||
"newItems": scanner.newItems,
|
NewItems: scanner.newItems,
|
||||||
"errors": scanner.errors,
|
Errors: scanner.errors,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
func (w *Worker) EnqueueJob(job *Job) error {
|
func (w *Worker) EnqueueJob(job *Job) error {
|
||||||
|
|||||||
Reference in New Issue
Block a user