From 54bfd778db0413435fe857e470873a1c54504e9a Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Thu, 5 Mar 2026 16:28:40 -0500 Subject: [PATCH] Refactor MediaScanner for improved file watching and job queue integration - Replace event queue with dirty directories tracking (Jellyfin approach) - Add file stability checking to wait for file writes to complete - Add initial scan on startup to detect existing files - Integrate with Worker job queue for directory scanning - Change WatchChanges to return error and use atomic.Bool for state - Add scan_mutex to prevent concurrent scans - Add Close method with proper cleanup of resources - Enhance polling with configurable interval --- internal/services/media_scanner.go | 436 ++++++++++++++++++++++++----- 1 file changed, 361 insertions(+), 75 deletions(-) diff --git a/internal/services/media_scanner.go b/internal/services/media_scanner.go index 12f2f9a..722e208 100644 --- a/internal/services/media_scanner.go +++ b/internal/services/media_scanner.go @@ -23,12 +23,15 @@ import ( "regexp" "strconv" "strings" + "sync" + "sync/atomic" "time" "bookhoard/internal/sevenzip" epub "github.com/ArcadiaLin/go-epub" "github.com/fsnotify/fsnotify" + "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "github.com/nwaples/rardecode" @@ -79,9 +82,13 @@ type MediaScanner struct { libraryTypes map[string][]string forceRescan bool logger *ScannerLogger - eventQueue chan string - debounceTimer *time.Timer + dirtyDirs map[string]time.Time + dirtyDirsMu sync.RWMutex + fileStability map[string]*atomic.Bool + fileStabilityMu sync.RWMutex + scan_mutex sync.Mutex pollInterval time.Duration + watching atomic.Bool totalFiles int newItems int @@ -99,12 +106,15 @@ func NewMediaScanner(db *database.Queries) *MediaScanner { return &MediaScanner{ db: db, watcher: watcher, + dirtyDirs: make(map[string]time.Time), + fileStability: make(map[string]*atomic.Bool), + pollInterval: 60 * time.Second, + watching: atomic.Bool{}, folders: []string{}, adminID: pgtype.UUID{}, defaultLibraryID: pgtype.UUID{Valid: false}, libraryTypes: make(map[string][]string), logger: NewScannerLogger(), - eventQueue: make(chan string, 500), } } @@ -1543,11 +1553,27 @@ func (s *MediaScanner) getMimeType(path string) string { return "application/octet-stream" } -func (s *MediaScanner) WatchChanges(ctx context.Context) { - // Start the debounced event processor - go s.processEventQueue(ctx) +func (s *MediaScanner) WatchChanges(ctx context.Context) error { + // Prevent duplicate calls + if !s.watching.CompareAndSwap(false, true) { + return fmt.Errorf("already watching") + } + + // Reset flag when context is cancelled + go func() { + <-ctx.Done() + s.watching.Store(false) + }() + + // Perform initial scan of all root folders + go s.performInitialScan(ctx) + + // Start directory processor + go s.processDirtyDirectories(ctx) + // Start polling fallback go s.StartPolling(ctx) + // Handle fsnotify events - queue them for debouncing go func() { for { @@ -1556,83 +1582,31 @@ func (s *MediaScanner) WatchChanges(ctx context.Context) { if !ok { return } - fmt.Printf("[FSNOTIFY] Event: %s %s\n", event.Name, event.Op.String()) + // Handle new directories - add them to the watcher if event.Has(fsnotify.Create) { - info, err := os.Stat(event.Name) - if err == nil && info.IsDir() { - if err := s.watcher.Add(event.Name); err != nil { - fmt.Printf("Warning: failed to watch new directory %s: %v\n", event.Name, err) - } else { - fmt.Printf("Now watching new directory: %s\n", event.Name) - } + if info, err := os.Stat(event.Name); err == nil && info.IsDir() { + s.watcher.Add(event.Name) } } - // Queue file events for debounced processing - if (event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Remove)) && s.isScannableFile(event.Name) { - fmt.Printf("[FSNOTIFY] Queuing scannable file: %s\n", event.Name) - select { - case s.eventQueue <- event.Name: - // Event queued - default: - fmt.Printf("Warning: event queue full, dropping event for %s\n", 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 } } }() -} -func (s *MediaScanner) processEventQueue(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case path := <-s.eventQueue: - // Reset debounce timer - wait for more events - if s.debounceTimer != nil { - s.debounceTimer.Stop() - } - s.debounceTimer = time.AfterFunc(3*time.Second, func() { - s.flushEventQueue(ctx, path) - }) - } - } -} -func (s *MediaScanner) flushEventQueue(ctx context.Context, initialPath string) { - // Collect all pending events from queue - paths := make(map[string]bool) - paths[initialPath] = true - // Drain remaining events (with short timeout to batch them) - timeout := time.After(500 * time.Millisecond) -DrainLoop: - for { - select { - case path := <-s.eventQueue: - paths[path] = true - case <-timeout: - break DrainLoop - } - } - fmt.Printf("Processing %d file events after debounce\n", len(paths)) - // Process each unique path - for path := range paths { - // Determine if file exists or was deleted - _, err := os.Stat(path) - if os.IsNotExist(err) { - // File was deleted - s.handleFileDelete(ctx, path) - } else if err == nil { - // File exists (new or modified) - s.handleFileAdd(ctx, path) - } - } + return nil } func (s *MediaScanner) handleFileAdd(ctx context.Context, filePath string) { fmt.Printf("New/modified media file detected: %s\n", filePath) @@ -1685,6 +1659,325 @@ func (s *MediaScanner) handleFileDelete(ctx context.Context, filePath string) { } } +func (s *MediaScanner) markDirectoryDirty(dirPath string) { + s.dirtyDirsMu.Lock() + defer s.dirtyDirsMu.Unlock() + + // Only mark if within watched folders + var isWatched bool + for _, folder := range s.folders { + if strings.HasPrefix(dirPath, folder) { + isWatched = true + break + } + } + if !isWatched { + return + } + + // Smart event merging (Jellyfin approach): + // 1. If parent dir exists, replace with parent (consolidate) + // 2. If sibling dirs exist, replace with common parent + // 3. Otherwise, add this dir + + // Check if parent directory is already dirty + parentDir := filepath.Dir(dirPath) + if parentDir != dirPath { // Not at root + if _, parentExists := s.dirtyDirs[parentDir]; parentExists { + // Parent already being watched, reset its timestamp + s.dirtyDirs[parentDir] = time.Now() + return + } + } + + // Check if any subdirectories are dirty, replace with parent + for existingDir := range s.dirtyDirs { + if strings.HasPrefix(existingDir, dirPath+"/") { + // This is a subdirectory, replace it with parent + delete(s.dirtyDirs, existingDir) + } + } + + // NEW: Check for sibling directories and consolidate to parent + parentDir = filepath.Dir(dirPath) + for existingDir := range s.dirtyDirs { + existingParent := filepath.Dir(existingDir) + if existingParent == parentDir && existingParent != dirPath && existingParent != "." { + // Found a sibling! Both should be replaced with parent + delete(s.dirtyDirs, existingDir) + s.dirtyDirs[parentDir] = time.Now() + return + } + } + + // Add/update this directory + s.dirtyDirs[dirPath] = time.Now() +} + +func (s *MediaScanner) processDirtyDirectories(ctx context.Context) { + ticker := time.NewTicker(1 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + + case <-ticker.C: + s.dirtyDirsMu.Lock() + + now := time.Now() + readyDirs := make([]string, 0) + + // Find directories that haven't been modified in 10 seconds + // This batches changes together (Audiobookshelf approach) + for dirPath, lastChange := range s.dirtyDirs { + if now.Sub(lastChange) >= 10*time.Second { + readyDirs = append(readyDirs, dirPath) + delete(s.dirtyDirs, dirPath) + } + } + + s.dirtyDirsMu.Unlock() + + // Process all ready directories in a batch via job queue + // Job queue serializes scans - prevents concurrent directory access + if len(readyDirs) > 0 { + for _, dirPath := range readyDirs { + // Create directory scan job with correct params for processDirectoryScanJob() + job := &Job{ + ID: uuid.New().String(), + Type: JobTypeDirectoryScan, + Params: map[string]interface{}{ + "directory": dirPath, + "db": s.db, + }, + Status: JobStatusPending, + } + + // Enqueue via global worker singleton + if WorkerInstance != nil { + WorkerInstance.Enqueue(job) + fmt.Printf("Enqueued directory scan job: %s\n", dirPath) + } else { + fmt.Printf("Warning: Worker not initialized, skipping directory scan: %s\n", dirPath) + } + } + } + } + } +} + +// waitForFileStability checks if a file's mtime has stabilized +// Returns true when file is stable (not being modified) +// Polls every 3 seconds, times out after 60 seconds +// Uses atomic.Bool to prevent race conditions with concurrent checks +func (s *MediaScanner) waitForFileStability(filePath string) bool { + s.fileStabilityMu.Lock() + + // Check if already being checked (atomic.Bool prevents race condition) + tracking, exists := s.fileStability[filePath] + if exists { + s.fileStabilityMu.Unlock() + // Another goroutine is already checking this file + if tracking.Load() { + return false // Still being checked + } + // Tracking exists but completed, remove stale entry + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + } + + // Start tracking with atomic.Bool set to true (checking in progress) + trackingFlag := &atomic.Bool{} + trackingFlag.Store(true) + s.fileStability[filePath] = trackingFlag + s.fileStabilityMu.Unlock() + + // Get initial mtime + info, err := os.Stat(filePath) + if err != nil { + // Clean up tracking entry if file doesn't exist + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return false + } + lastMtime := info.ModTime() + + // Poll every 3 seconds for up to 60 seconds + timeout := time.After(60 * time.Second) + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + + for { + select { + case <-timeout: + // Timeout - mark as done and clean up + trackingFlag.Store(false) + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return false // File never stabilized + + case <-ticker.C: + info, err := os.Stat(filePath) + if err != nil { + // File deleted - mark as done and clean up + trackingFlag.Store(false) + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return false + } + + currentMtime := info.ModTime() + if currentMtime.Equal(lastMtime) { + // File is stable! Mark as done and clean up + trackingFlag.Store(false) + s.fileStabilityMu.Lock() + delete(s.fileStability, filePath) + s.fileStabilityMu.Unlock() + return true + } + + lastMtime = currentMtime + } + } +} + +func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) { + // Prevent concurrent scans of ANY directory + // Simple mutex is enough - job queue already serializes by directory + s.scan_mutex.Lock() + defer s.scan_mutex.Unlock() + + // Find library for this directory + var libraryID pgtype.UUID + var rootFolder string + + for _, folder := range s.folders { + if strings.HasPrefix(dirPath, folder) { + rootFolder = folder + if lib, err := s.db.GetLibraryByFolder(ctx, folder); err == nil { + libraryID = lib.LibraryID + break + } + } + } + + // Check if libraryID is valid before proceeding + if !libraryID.Valid { + return + } + + // Walk directory and process new files + filepath.WalkDir(dirPath, func(path string, d os.DirEntry, err error) error { + if err != nil { + return err + } + if d.IsDir() { + return filepath.SkipDir + } + if !s.isScannableFile(path) { + return nil + } + + // Check if file is stable before processing (Audiobookshelf approach) + if !s.waitForFileStability(path) { + return nil + } + + relPath := strings.TrimPrefix(path, rootFolder+"/") + _, 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 + }) +} + +// performInitialScan scans all root folders on startup +// This ensures existing files are detected before watching begins +func (s *MediaScanner) performInitialScan(ctx context.Context) { + fmt.Printf("Performing initial scan of root folders...\n") + + for _, folder := range s.folders { + // Skip if folder doesn't exist + if _, err := os.Stat(folder); os.IsNotExist(err) { + fmt.Printf("Skipping nonexistent folder: %s\n", folder) + continue + } + + // Submit scan job to worker (non-blocking) + job := &Job{ + ID: uuid.New().String(), + Type: JobTypeDirectoryScan, + Params: map[string]interface{}{ + "directory": folder, + "db": s.db, + }, + Status: JobStatusPending, + } + + if WorkerInstance != nil { + WorkerInstance.Enqueue(job) + fmt.Printf("Enqueued initial scan job: %s\n", folder) + } else { + fmt.Printf("Warning: Worker not initialized, skipping initial scan: %s\n", folder) + } + } + + fmt.Printf("Initial scan jobs enqueued\n") +} + +func (s *MediaScanner) Close() error { + fmt.Printf("Cleaning up scanner resources...\n") + + // Stop watching + if s.watcher != nil { + s.watcher.Close() + } + + // Clean up fileStability map to prevent memory leaks + s.fileStabilityMu.Lock() + s.fileStability = make(map[string]*atomic.Bool) + s.fileStabilityMu.Unlock() + + // Clear dirty directories + s.dirtyDirsMu.Lock() + s.dirtyDirs = make(map[string]time.Time) + s.dirtyDirsMu.Unlock() + + // Wait for in-progress scan to complete (with timeout) + timeout := time.After(5 * time.Second) + done := make(chan struct{}) + + go func() { + s.scan_mutex.Lock() + s.scan_mutex.Unlock() + close(done) + }() + + select { + case <-done: + fmt.Printf("Scanner cleanup complete\n") + case <-timeout: + fmt.Printf("Timeout waiting for scan to complete\n") + } + + return nil +} func (s *MediaScanner) StartPolling(ctx context.Context) { interval := s.GetPollInterval() if interval <= 0 { @@ -1778,13 +2071,6 @@ func (s *MediaScanner) SyncFilesystemWithDatabase(ctx context.Context) error { return nil } -func (s *MediaScanner) Close() error { - if s.watcher != nil { - return s.watcher.Close() - } - return nil -} - // ============================================ // SCANNER ENHANCEMENTS // ============================================