fix(scanner): replace mtime polling with recursive fsnotify watching
The root cause of scanner failures in Podman containers was NOT that inotify doesn't work through bind mounts (it does — same kernel, same inodes). The real bug was SetFolders() only watching root directories. Linux has no recursive inotify — every subdirectory must be added individually to the watcher. Changes: - SetFolders() now walks all subdirectories and adds each to the watcher (same approach as Audiobookshelf/Kavita) - Remove broken mtime-based detection: seedDirectoryMtimes, pollDirectoryChanges, detectChangedRoots, checkDirectoryMtimes, SyncFilesystemWithDatabase — all unreliable in container overlay mounts - Replace StartPolling with startBackupScan: enqueues full JobTypeScan every 5 minutes (down from 30) as a safety-net fallback - enqueueLibraryScan() sets job.UserID from admin ID so the worker can broadcast WebSocket messages - performInitialScan() sets job.UserID for the same reason - Add [WATCHER] prefix logging to all fsnotify event loop messages - Add defense-in-depth: fallback to GetFirstAdmin() when library has no created_by_admin_id (NULL from test cleanup) - Fix processDirectoryScanJob to use prefix-match (GetLibraryByFolderPathPrefix) - Fix nil context panic: all jobs now set Context: context.Background() - Remove mtime-related tests; update default interval test from 30m to 5m
This commit is contained in:
+150
-253
@@ -118,11 +118,8 @@ type MediaScanner struct {
|
||||
fileStabilityMu sync.RWMutex
|
||||
scanMutex sync.Mutex
|
||||
scanInProgress atomic.Bool
|
||||
pollInterval time.Duration
|
||||
watching atomic.Bool
|
||||
settingsCache *SettingsCache
|
||||
dirMtimes map[string]time.Time
|
||||
dirMtimesMu sync.RWMutex
|
||||
|
||||
totalFiles int
|
||||
newItems int
|
||||
@@ -157,20 +154,18 @@ func NewMediaScanner(db *database.Queries) *MediaScanner {
|
||||
}
|
||||
|
||||
return &MediaScanner{
|
||||
db: db,
|
||||
watcher: watcher,
|
||||
settingsCache: NewSettingsCache(30 * time.Second),
|
||||
dirtyDirs: make(map[string]time.Time),
|
||||
fileStability: make(map[string]*atomic.Bool),
|
||||
pollInterval: 60 * time.Second,
|
||||
watching: atomic.Bool{},
|
||||
scanInProgress: atomic.Bool{},
|
||||
folders: []string{},
|
||||
adminID: pgtype.UUID{},
|
||||
db: db,
|
||||
watcher: watcher,
|
||||
settingsCache: NewSettingsCache(30 * time.Second),
|
||||
dirtyDirs: make(map[string]time.Time),
|
||||
fileStability: make(map[string]*atomic.Bool),
|
||||
watching: atomic.Bool{},
|
||||
scanInProgress: atomic.Bool{},
|
||||
folders: []string{},
|
||||
adminID: pgtype.UUID{},
|
||||
defaultLibraryID: pgtype.UUID{Valid: false},
|
||||
libraryTypes: make(map[string][]string),
|
||||
logger: NewScannerLogger(),
|
||||
dirMtimes: make(map[string]time.Time),
|
||||
libraryTypes: make(map[string][]string),
|
||||
logger: NewScannerLogger(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -182,7 +177,7 @@ func (s *MediaScanner) GetPollInterval() time.Duration {
|
||||
}
|
||||
|
||||
if s.db == nil {
|
||||
return 30 * time.Minute
|
||||
return 5 * time.Minute
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
@@ -190,14 +185,14 @@ func (s *MediaScanner) GetPollInterval() time.Duration {
|
||||
|
||||
setting, err := s.db.GetSystemSetting(ctx, "scan_poll_interval_seconds")
|
||||
if err != nil || setting == "" {
|
||||
return 30 * time.Minute
|
||||
return 5 * time.Minute
|
||||
}
|
||||
|
||||
s.settingsCache.Set("scan_poll_interval_seconds", setting)
|
||||
|
||||
seconds, err := strconv.Atoi(setting)
|
||||
if err != nil {
|
||||
return 30 * time.Minute
|
||||
return 5 * time.Minute
|
||||
}
|
||||
return time.Duration(seconds) * time.Second
|
||||
}
|
||||
@@ -263,129 +258,119 @@ func (s *MediaScanner) SetFolders(folders []string) error {
|
||||
s.watcher = watcher
|
||||
|
||||
// Build cache of allowed extensions per folder
|
||||
// Uses Go AllowedExtensions map as source of truth (not DB)
|
||||
s.libraryTypes = make(map[string][]string)
|
||||
ctx := context.Background()
|
||||
|
||||
for _, folder := range folders {
|
||||
// Get library for this folder
|
||||
lib, err := s.db.GetLibraryByFolder(ctx, folder)
|
||||
if err != nil {
|
||||
fmt.Printf("Warning: failed to get library for folder %s: %v\n", folder, err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Get library type with allowed extensions
|
||||
libType, err := s.db.GetLibraryType(ctx, lib.LibraryTypeID)
|
||||
if err != nil {
|
||||
fmt.Printf("Warning: failed to get library type for %s: %v\n", folder, err)
|
||||
continue
|
||||
}
|
||||
|
||||
// Cache allowed extensions for this folder
|
||||
s.libraryTypes[folder] = libType.AllowedExtensions
|
||||
fmt.Printf("Scanner: Folder %s (type: %s) allows extensions: %v\n",
|
||||
folder, libType.Name, libType.AllowedExtensions)
|
||||
}
|
||||
|
||||
// Add all folders to watch
|
||||
for _, folder := range folders {
|
||||
if err := s.watcher.Add(folder); err != nil {
|
||||
fmt.Printf("Warning: failed to watch folder %s: %v\n", folder, err)
|
||||
if exts, ok := AllowedExtensions[libType.Name]; ok {
|
||||
s.libraryTypes[folder] = exts
|
||||
fmt.Printf("Scanner: Folder %s (type: %s) allows extensions: %v\n",
|
||||
folder, libType.Name, exts)
|
||||
} else {
|
||||
s.libraryTypes[folder] = libType.AllowedExtensions
|
||||
fmt.Printf("Scanner: Folder %s (type: %s) using DB extensions (no Go map entry): %v\n",
|
||||
folder, libType.Name, libType.AllowedExtensions)
|
||||
}
|
||||
}
|
||||
|
||||
s.seedDirectoryMtimes()
|
||||
// Add all folders and their subdirectories to the watcher (like Audiobookshelf)
|
||||
watchCount := 0
|
||||
for _, folder := range folders {
|
||||
if err := s.watcher.Add(folder); err != nil {
|
||||
fmt.Printf("[WATCHER] Warning: failed to watch root folder %s: %v\n", folder, err)
|
||||
} else {
|
||||
watchCount++
|
||||
}
|
||||
filepath.WalkDir(folder, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !d.IsDir() || path == folder {
|
||||
return nil
|
||||
}
|
||||
if err := s.watcher.Add(path); err != nil {
|
||||
fmt.Printf("[WATCHER] Warning: failed to watch subdirectory %s: %v\n", path, err)
|
||||
} else {
|
||||
watchCount++
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
fmt.Printf("[WATCHER] Now watching %d directories across %d root folders\n", watchCount, len(folders))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *MediaScanner) seedDirectoryMtimes() {
|
||||
s.dirMtimesMu.Lock()
|
||||
defer s.dirMtimesMu.Unlock()
|
||||
|
||||
s.dirMtimes = make(map[string]time.Time)
|
||||
|
||||
for _, folder := range s.folders {
|
||||
if _, err := os.Stat(folder); os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
|
||||
filepath.WalkDir(folder, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !d.IsDir() {
|
||||
return nil
|
||||
}
|
||||
info, err := d.Info()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
s.dirMtimes[path] = info.ModTime()
|
||||
return nil
|
||||
})
|
||||
func (s *MediaScanner) enqueueLibraryScan(rootFolder string) {
|
||||
if s.db == nil {
|
||||
return
|
||||
}
|
||||
|
||||
fmt.Printf("Seeded directory mtime cache with %d directories\n", len(s.dirMtimes))
|
||||
}
|
||||
libRow, err := s.db.GetLibraryByFolderPathPrefix(context.Background(), rootFolder)
|
||||
if err != nil {
|
||||
fmt.Printf("[MTIME-POLL] Warning: could not find library for %s: %v\n", rootFolder, err)
|
||||
return
|
||||
}
|
||||
|
||||
func (s *MediaScanner) pollDirectoryChanges(ctx context.Context) {
|
||||
fmt.Println("Directory mtime poller started (interval: 10s)")
|
||||
folders, err := s.db.GetLibraryFolders(context.Background(), libRow.LibraryID)
|
||||
if err != nil {
|
||||
fmt.Printf("[MTIME-POLL] Warning: could not get folders for library: %v\n", err)
|
||||
return
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
folderPaths := make([]string, len(folders))
|
||||
for i, f := range folders {
|
||||
folderPaths[i] = f.FolderPath
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Println("Directory mtime poller stopped")
|
||||
adminIDStr := ""
|
||||
if libRow.CreatedByAdminID.Valid {
|
||||
adminIDStr = uuid.UUID(libRow.CreatedByAdminID.Bytes).String()
|
||||
}
|
||||
if adminIDStr == "" {
|
||||
fmt.Printf("[MTIME-POLL] Library has no owner, falling back to first admin\n")
|
||||
fallbackAdmin, err := s.db.GetFirstAdmin(context.Background())
|
||||
if err != nil {
|
||||
fmt.Printf("[MTIME-POLL] Warning: no admin found in database, skipping scan\n")
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.checkDirectoryMtimes()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *MediaScanner) checkDirectoryMtimes() {
|
||||
s.dirMtimesMu.Lock()
|
||||
defer s.dirMtimesMu.Unlock()
|
||||
|
||||
changedCount := 0
|
||||
|
||||
for _, folder := range s.folders {
|
||||
if _, err := os.Stat(folder); os.IsNotExist(err) {
|
||||
continue
|
||||
}
|
||||
|
||||
filepath.WalkDir(folder, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !d.IsDir() {
|
||||
return nil
|
||||
}
|
||||
|
||||
info, err := d.Info()
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
currentMtime := info.ModTime()
|
||||
|
||||
cachedMtime, exists := s.dirMtimes[path]
|
||||
if !exists || !cachedMtime.Equal(currentMtime) {
|
||||
s.dirtyDirsMu.Lock()
|
||||
s.dirtyDirs[path] = time.Now()
|
||||
s.dirtyDirsMu.Unlock()
|
||||
changedCount++
|
||||
}
|
||||
|
||||
s.dirMtimes[path] = currentMtime
|
||||
return nil
|
||||
})
|
||||
adminIDStr = uuid.UUID(fallbackAdmin.Bytes).String()
|
||||
}
|
||||
|
||||
if changedCount > 0 {
|
||||
fmt.Printf("[MTIME-POLL] Detected changes in %d director(ies)\n", changedCount)
|
||||
libraryIDStr := uuid.UUID(libRow.LibraryID.Bytes).String()
|
||||
|
||||
job := &Job{
|
||||
ID: uuid.New().String(),
|
||||
Type: JobTypeScan,
|
||||
Status: JobStatusPending,
|
||||
UserID: adminIDStr,
|
||||
Context: context.Background(),
|
||||
Params: map[string]any{
|
||||
"library_id": libraryIDStr,
|
||||
"folders": folderPaths,
|
||||
"admin_id": adminIDStr,
|
||||
"db": s.db,
|
||||
"force": false,
|
||||
},
|
||||
}
|
||||
|
||||
if WorkerInstance != nil {
|
||||
WorkerInstance.Enqueue(job)
|
||||
fmt.Printf("[MTIME-POLL] Enqueued library scan for %s (library: %s)\n", rootFolder, libraryIDStr)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -803,6 +788,7 @@ func (s *MediaScanner) processMediaFile(ctx context.Context, path string) (bool,
|
||||
TagsSearch: tagsSearch,
|
||||
AddedByAdminID: s.adminID,
|
||||
CreatedAt: pgtype.Timestamptz{Time: fileModTime, Valid: true},
|
||||
ImportedAt: pgtype.Timestamptz{Time: time.Now(), Valid: true},
|
||||
MangaType: pgtype.Text{String: metadata.MangaType, Valid: metadata.MangaType != ""},
|
||||
ReadingDirection: pgtype.Text{String: metadata.ReadingDirection, Valid: metadata.ReadingDirection != ""},
|
||||
SeriesCount: pgtype.Int4{Int32: metadata.SeriesCount, Valid: metadata.SeriesCount > 0},
|
||||
@@ -2599,59 +2585,55 @@ func (s *MediaScanner) getMimeType(path string) string {
|
||||
}
|
||||
|
||||
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 (primarily for orphan cleanup)
|
||||
go s.StartPolling(ctx)
|
||||
go s.startBackupScan(ctx)
|
||||
|
||||
// Start directory mtime poller (fast detection of new/changed files)
|
||||
go s.pollDirectoryChanges(ctx)
|
||||
|
||||
// Handle fsnotify events - queue them for debouncing
|
||||
go func() {
|
||||
fmt.Printf("[WATCHER] Event loop started for %d folders\n", len(s.folders))
|
||||
for {
|
||||
select {
|
||||
case event, ok := <-s.watcher.Events:
|
||||
if !ok {
|
||||
fmt.Printf("[WATCHER] Event channel closed\n")
|
||||
return
|
||||
}
|
||||
|
||||
// Handle new directories - add them to the watcher
|
||||
if event.Has(fsnotify.Create) {
|
||||
if info, err := os.Stat(event.Name); 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)
|
||||
fmt.Printf("[WATCHER] Warning: failed to watch new directory %s: %v\n", event.Name, err)
|
||||
} else {
|
||||
fmt.Printf("[WATCHER] Now watching new directory: %s\n", event.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Mark directory dirty for ANY file change
|
||||
if event.Has(fsnotify.Create | fsnotify.Write | fsnotify.Remove | fsnotify.Chmod | fsnotify.Rename) {
|
||||
if event.Has(fsnotify.Create | fsnotify.Write | fsnotify.Remove | fsnotify.Rename) {
|
||||
fmt.Printf("[WATCHER] Event: %s on %s\n", event.Op, event.Name)
|
||||
s.markDirectoryDirty(filepath.Dir(event.Name))
|
||||
}
|
||||
|
||||
case err, ok := <-s.watcher.Errors:
|
||||
if !ok {
|
||||
fmt.Printf("[WATCHER] Error channel closed\n")
|
||||
return
|
||||
}
|
||||
fmt.Printf("Watcher error: %v\n", err)
|
||||
fmt.Printf("[WATCHER] Error: %v\n", err)
|
||||
|
||||
case <-ctx.Done():
|
||||
fmt.Printf("[WATCHER] Event loop stopped\n")
|
||||
return
|
||||
}
|
||||
}
|
||||
@@ -2729,8 +2711,6 @@ func (s *MediaScanner) processDirtyDirectories(ctx context.Context) {
|
||||
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)
|
||||
@@ -2740,30 +2720,23 @@ func (s *MediaScanner) processDirtyDirectories(ctx context.Context) {
|
||||
|
||||
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]any{
|
||||
"directory": dirPath,
|
||||
"db": s.db,
|
||||
},
|
||||
Status: JobStatusPending,
|
||||
}
|
||||
if len(readyDirs) == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
// 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)
|
||||
affectedRoots := make(map[string]bool)
|
||||
for _, dirPath := range readyDirs {
|
||||
for _, folder := range s.folders {
|
||||
if strings.HasPrefix(dirPath, folder) {
|
||||
affectedRoots[folder] = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for rootFolder := range affectedRoots {
|
||||
s.enqueueLibraryScan(rootFolder)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2862,7 +2835,7 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
||||
for _, folder := range s.folders {
|
||||
if strings.HasPrefix(dirPath, folder) {
|
||||
rootFolder = folder
|
||||
if lib, err := s.db.GetLibraryByFolder(ctx, folder); err == nil {
|
||||
if lib, err := s.db.GetLibraryByFolderPathPrefix(ctx, dirPath); err == nil {
|
||||
libraryID = lib.LibraryID
|
||||
break
|
||||
}
|
||||
@@ -2874,15 +2847,12 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
||||
return
|
||||
}
|
||||
|
||||
// Walk directory and process new files
|
||||
// Walk directory and process new files (recurses into subdirectories)
|
||||
if err := filepath.WalkDir(dirPath, func(path string, d os.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if d.IsDir() {
|
||||
if path != dirPath {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if !s.isScannableFile(path) {
|
||||
@@ -2919,36 +2889,34 @@ func (s *MediaScanner) scanDirectory(ctx context.Context, dirPath string) {
|
||||
func (s *MediaScanner) performInitialScan(ctx context.Context) {
|
||||
fmt.Printf("Performing initial scan of root folders...\n")
|
||||
|
||||
for _, folder := range s.folders {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Printf("Initial scan cancelled\n")
|
||||
return
|
||||
default:
|
||||
}
|
||||
// Skip if folder doesn't exist
|
||||
if _, err := os.Stat(folder); os.IsNotExist(err) {
|
||||
fmt.Printf("Skipping nonexistent folder: %s\n", folder)
|
||||
continue
|
||||
}
|
||||
if s.defaultLibraryID.Valid && s.adminID.Valid {
|
||||
folderPaths := s.folders
|
||||
libraryIDStr := uuid.UUID(s.defaultLibraryID.Bytes).String()
|
||||
adminIDStr := uuid.UUID(s.adminID.Bytes).String()
|
||||
|
||||
// Submit scan job to worker (non-blocking)
|
||||
job := &Job{
|
||||
ID: uuid.New().String(),
|
||||
Type: JobTypeDirectoryScan,
|
||||
ID: uuid.New().String(),
|
||||
Type: JobTypeScan,
|
||||
Status: JobStatusPending,
|
||||
UserID: adminIDStr,
|
||||
Context: context.Background(),
|
||||
Params: map[string]any{
|
||||
"directory": folder,
|
||||
"db": s.db,
|
||||
"library_id": libraryIDStr,
|
||||
"folders": folderPaths,
|
||||
"admin_id": adminIDStr,
|
||||
"db": s.db,
|
||||
"force": false,
|
||||
},
|
||||
Status: JobStatusPending,
|
||||
}
|
||||
|
||||
if WorkerInstance != nil {
|
||||
WorkerInstance.Enqueue(job)
|
||||
fmt.Printf("Enqueued initial scan job: %s\n", folder)
|
||||
fmt.Printf("Enqueued initial library scan job\n")
|
||||
} else {
|
||||
fmt.Printf("Warning: Worker not initialized, skipping initial scan: %s\n", folder)
|
||||
fmt.Printf("Warning: Worker not initialized, skipping initial scan\n")
|
||||
}
|
||||
} else {
|
||||
fmt.Printf("Warning: no library/admin ID set, skipping initial scan\n")
|
||||
}
|
||||
|
||||
fmt.Printf("Initial scan jobs enqueued\n")
|
||||
@@ -2994,13 +2962,13 @@ func (s *MediaScanner) Close() error {
|
||||
|
||||
return nil
|
||||
}
|
||||
func (s *MediaScanner) StartPolling(ctx context.Context) {
|
||||
func (s *MediaScanner) startBackupScan(ctx context.Context) {
|
||||
interval := s.GetPollInterval()
|
||||
if interval <= 0 {
|
||||
fmt.Println("Orphan cleanup polling disabled (interval = 0)")
|
||||
fmt.Println("[BACKUP-SCAN] Periodic scan disabled (interval = 0)")
|
||||
return
|
||||
}
|
||||
fmt.Printf("Orphan cleanup polling started with interval: %v\n", interval)
|
||||
fmt.Printf("[BACKUP-SCAN] Periodic scan started with interval: %v\n", interval)
|
||||
|
||||
for {
|
||||
ticker := time.NewTicker(interval)
|
||||
@@ -3008,92 +2976,21 @@ func (s *MediaScanner) StartPolling(ctx context.Context) {
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
fmt.Println("Orphan cleanup polling stopped")
|
||||
fmt.Println("[BACKUP-SCAN] Periodic scan stopped")
|
||||
return
|
||||
case <-ticker.C:
|
||||
interval = s.GetPollInterval()
|
||||
fmt.Printf("[ORPHAN-CLEANUP] Running filesystem sync (interval: %v)...\n", interval)
|
||||
if err := s.SyncFilesystemWithDatabase(ctx); err != nil {
|
||||
fmt.Printf("[ORPHAN-CLEANUP] Sync error: %v\n", err)
|
||||
if !s.GetAutoScanEnabled() {
|
||||
continue
|
||||
}
|
||||
fmt.Printf("[BACKUP-SCAN] Running periodic full scan (interval: %v)...\n", interval)
|
||||
for _, folder := range s.folders {
|
||||
s.enqueueLibraryScan(folder)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *MediaScanner) SyncFilesystemWithDatabase(ctx context.Context) error {
|
||||
fmt.Println("[POLL-SYNC] Starting filesystem sync with database")
|
||||
for _, folder := range s.folders {
|
||||
lib, err := s.db.GetLibraryByFolder(ctx, folder)
|
||||
if err != nil {
|
||||
fmt.Printf("[POLL-SYNC] Warning: failed to get library for folder %s: %v\n", folder, err)
|
||||
continue
|
||||
}
|
||||
libraryID := lib.LibraryID
|
||||
// Get all media items from database for this library
|
||||
dbItems, err := s.db.ListMediaItemsByLibrary(ctx, libraryID)
|
||||
if err != nil {
|
||||
fmt.Printf("[POLL-SYNC] Warning: failed to get library items: %v\n", err)
|
||||
continue
|
||||
}
|
||||
// Build set of existing file paths from filesystem
|
||||
existingPaths := make(map[string]bool)
|
||||
if err := filepath.WalkDir(folder, func(path string, d os.DirEntry, walkErr error) error {
|
||||
if walkErr != nil {
|
||||
return walkErr
|
||||
}
|
||||
if !d.IsDir() && s.isScannableFile(path) {
|
||||
existingPaths[s.getRelativePath(path)] = true
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
fmt.Printf("[POLL-SYNC] Warning: failed to walk directory %s: %v\n", folder, err)
|
||||
continue
|
||||
}
|
||||
// Check for orphaned items (in DB but not on filesystem)
|
||||
for _, item := range dbItems {
|
||||
if item.FilePath != "" && !existingPaths[item.FilePath] {
|
||||
msg := fmt.Sprintf("[POLL-SYNC] Orphaned media item found: ID=%s, Title=%s, Path=%s",
|
||||
item.ID, item.Title, item.FilePath)
|
||||
s.logger.LogDelete(msg)
|
||||
if err := s.db.DeleteMediaItem(ctx, item.ID); err != nil {
|
||||
errMsg := fmt.Sprintf("[POLL-SYNC] ERROR: failed to delete orphaned item %s: %v", item.Title, err)
|
||||
s.logger.LogDelete(errMsg)
|
||||
s.logger.LogError(errMsg)
|
||||
} else {
|
||||
s.logger.LogDelete(fmt.Sprintf("[POLL-SYNC] SUCCESS: deleted orphaned item '%s'", item.Title))
|
||||
}
|
||||
}
|
||||
}
|
||||
// Check for new files (on filesystem but not in DB)
|
||||
// This is expensive, so we just check a few representative files
|
||||
// The fsnotify handler should catch most new files
|
||||
for relPath := range existingPaths {
|
||||
// Check if this file exists in DB
|
||||
_, err := s.db.GetMediaItemByFilePath(ctx, database.GetMediaItemByFilePathParams{
|
||||
FilePath: relPath,
|
||||
LibraryID: libraryID,
|
||||
})
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
// New file found - scan it
|
||||
absPath := folder + "/" + relPath
|
||||
if _, err := os.Stat(absPath); err == nil {
|
||||
fmt.Printf("[POLL-SYNC] New file detected, scanning: %s\n", absPath)
|
||||
if _, err := s.processMediaFile(ctx, absPath); err != nil {
|
||||
fmt.Printf("[POLL-SYNC] Error scanning new file %s: %v\n", absPath, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
fmt.Println("[POLL-SYNC] Filesystem sync completed")
|
||||
return nil
|
||||
}
|
||||
|
||||
// ============================================
|
||||
// SCANNER ENHANCEMENTS
|
||||
// ============================================
|
||||
|
||||
// calculateFileSHA256 calculates SHA-256 hash using streaming to avoid loading entire file into memory
|
||||
func (s *MediaScanner) calculateFileSHA256(filePath string) (string, error) {
|
||||
file, err := os.Open(filePath)
|
||||
if err != nil {
|
||||
|
||||
Reference in New Issue
Block a user