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
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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))
|
||||
}
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
// ============================================
|
||||
|
||||
Reference in New Issue
Block a user