From 96bae2f346aabc0ba5fbd043256b0089b12fc285 Mon Sep 17 00:00:00 2001 From: zZebrahz Date: Sun, 7 Jun 2026 18:32:40 -0700 Subject: [PATCH] fix(library): clean orphaned provisional items (#76) * fix(library): clean orphaned provisional items * test(metadata): use item not found sentinel in fake repo * fix(library): preserve abs references during orphan cleanup * fix(library): avoid dropped abs cleanup tables * fix(library): harden provisional orphan cleanup * fix(library): preserve home dismissal series orphans * fix(library): Delete matched items created after the first orphan sweep --- cmd/silo/main.go | 1 + internal/catalog/folder_delete_test.go | 13 ++ internal/catalog/folder_repo.go | 164 ++++++++++++-- internal/catalog/orphan_cleanup.go | 200 ++++++++++++++++++ internal/catalog/orphan_cleanup_test.go | 41 ++++ internal/metadata/service.go | 112 ++++++++-- internal/metadata/service_test.go | 127 ++++++++++- internal/scanner/file_repo.go | 10 +- .../tasks/cleanup_orphaned_media_items.go | 81 +++++++ .../cleanup_orphaned_media_items_test.go | 112 ++++++++++ ...cleanup_orphaned_provisional_items_test.go | 41 ++++ ...000_cleanup_orphaned_provisional_items.sql | 142 +++++++++++++ 12 files changed, 1005 insertions(+), 39 deletions(-) create mode 100644 internal/catalog/orphan_cleanup.go create mode 100644 internal/catalog/orphan_cleanup_test.go create mode 100644 internal/taskmanager/tasks/cleanup_orphaned_media_items.go create mode 100644 internal/taskmanager/tasks/cleanup_orphaned_media_items_test.go create mode 100644 migrations/cleanup_orphaned_provisional_items_test.go create mode 100644 migrations/sql/20260607220000_cleanup_orphaned_provisional_items.sql diff --git a/cmd/silo/main.go b/cmd/silo/main.go index b955c141..f4cdb480 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1443,6 +1443,7 @@ func main() { if deps.FolderRepo != nil && deps.LibraryScanQueue != nil { taskMgr.Register(tasks.NewScanLibrariesTask(deps.FolderRepo, deps.LibraryScanQueue, deps.EventBus)) } + taskMgr.Register(tasks.NewCleanupOrphanedMediaItemsTask(catalog.NewOrphanedProvisionalCleaner(deps.DB))) if deps.IntroAnalyzer != nil { taskMgr.Register(tasks.NewDetectIntroMarkersTask(deps.IntroAnalyzer, settingsRepo)) } diff --git a/internal/catalog/folder_delete_test.go b/internal/catalog/folder_delete_test.go index 280a3e01..a4000e6e 100644 --- a/internal/catalog/folder_delete_test.go +++ b/internal/catalog/folder_delete_test.go @@ -3,6 +3,7 @@ package catalog import ( "context" "errors" + "strings" "testing" "time" @@ -136,3 +137,15 @@ func TestDeleteInBatchesReturnsError(t *testing.T) { t.Fatalf("expected sentinel, got %v", err) } } + +func TestLateOrphanCleanupUsesProvisionalSafetyPredicate(t *testing.T) { + conditions := normalizePredicateSQL(orphanedProvisionalMediaItemConditions) + for name, query := range map[string]string{ + "collect": collectOrphanedProvisionalIDsByContentIDSQL, + "delete": deleteOrphanedProvisionalIDsByContentIDSQL, + } { + if !strings.Contains(normalizePredicateSQL(query), conditions) { + t.Fatalf("%s late orphan query does not use provisional safety predicate", name) + } + } +} diff --git a/internal/catalog/folder_repo.go b/internal/catalog/folder_repo.go index b1485308..e97579a7 100644 --- a/internal/catalog/folder_repo.go +++ b/internal/catalog/folder_repo.go @@ -563,16 +563,6 @@ func (r *FolderRepository) DeleteWithStats( } } - // Filter accumulated image dirs once, now that orphans are gone, against - // any surviving content. Empty deleting-set means "exclude nothing". - if len(rawDirs) > 0 { - filtered, err := filterUnreferencedImageDirs(ctx, r.pool, dirSetToSlice(rawDirs), []string{}) - if err != nil { - return nil, err - } - stats.OrphanedImageDirs = filtered - } - // Phase 2: delete this folder's media_files (folder-tied, not item-tied). if progress != nil { progress(orphanTotal, orphanTotal, "Deleting media files") @@ -592,23 +582,30 @@ func (r *FolderRepository) DeleteWithStats( } // Phase 3: delete remaining folder memberships (shared items kept; only the - // membership in this folder is removed). + // membership in this folder is removed). Deleting the memberships with + // RETURNING lets us catch skeleton items that a matcher created after the + // phase-1 orphan sweep but before the delete job reached this phase. if progress != nil { progress(orphanTotal, orphanTotal, "Removing library memberships") } - if _, err := deleteInBatches(ctx, folderChildDeleteBatch, func(ctx context.Context) (int64, error) { - tag, e := r.pool.Exec(ctx, ` - DELETE FROM media_item_libraries - WHERE ctid IN ( - SELECT ctid FROM media_item_libraries WHERE media_folder_id = $1 LIMIT $2 - )`, id, folderChildDeleteBatch) - if e != nil { - return 0, e - } - return tag.RowsAffected(), nil - }); err != nil { + lateOrphans, lateImageDirs, err := r.deleteFolderMemberships(ctx, id) + if err != nil { return nil, fmt.Errorf("deleting media item links: %w", err) } + stats.OrphanedItems += lateOrphans + for _, d := range lateImageDirs { + rawDirs[d] = struct{}{} + } + + // Filter accumulated image dirs once, now that all item orphans are gone, + // against any surviving content. Empty deleting-set means "exclude nothing". + if len(rawDirs) > 0 { + filtered, err := filterUnreferencedImageDirs(ctx, r.pool, dirSetToSlice(rawDirs), []string{}) + if err != nil { + return nil, err + } + stats.OrphanedImageDirs = filtered + } // Phase 4: delete the now-lightweight folder row. Tolerate 0 rows so a // resumed run that already removed it still succeeds. @@ -656,6 +653,129 @@ func (r *FolderRepository) collectOrphanBatch(ctx context.Context, folderID, lim return ids, nil } +// deleteFolderMemberships removes this folder's remaining item memberships and +// deletes any content that becomes orphaned as a result. This is intentionally +// separate from the phase-1 orphan sweep because in-flight matchers can create +// new skeleton memberships after phase 1 has already drained. +func (r *FolderRepository) deleteFolderMemberships(ctx context.Context, folderID int) (int, []string, error) { + var orphaned int + var imageDirs []string + for { + var contentIDs []string + if err := retryOnDeadlock(ctx, func() error { + ids, err := r.deleteFolderMembershipBatch(ctx, folderID, folderChildDeleteBatch) + if err != nil { + return err + } + contentIDs = ids + return nil + }); err != nil { + return orphaned, imageDirs, err + } + if len(contentIDs) > 0 { + deleted, dirs, err := r.deleteOrphanedItemsByContentID(ctx, contentIDs) + if err != nil { + return orphaned, imageDirs, err + } + orphaned += deleted + imageDirs = append(imageDirs, dirs...) + } + if len(contentIDs) < folderChildDeleteBatch { + return orphaned, imageDirs, nil + } + } +} + +func (r *FolderRepository) deleteFolderMembershipBatch(ctx context.Context, folderID, limit int) ([]string, error) { + rows, err := r.pool.Query(ctx, ` + DELETE FROM media_item_libraries + WHERE ctid IN ( + SELECT ctid FROM media_item_libraries WHERE media_folder_id = $1 LIMIT $2 + ) + RETURNING content_id`, folderID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var contentIDs []string + for rows.Next() { + var contentID string + if err := rows.Scan(&contentID); err != nil { + return nil, fmt.Errorf("scanning deleted membership content_id: %w", err) + } + contentIDs = append(contentIDs, contentID) + } + if err := rows.Err(); err != nil { + return nil, err + } + return contentIDs, nil +} + +const collectOrphanedProvisionalIDsByContentIDSQL = ` + SELECT mi.content_id + FROM public.media_items mi + WHERE mi.content_id = ANY($1) + AND ` + orphanedMediaItemSafetyConditions + +const deleteOrphanedProvisionalIDsByContentIDSQL = ` + DELETE FROM public.media_items mi + WHERE mi.content_id = ANY($1) + AND ` + orphanedMediaItemSafetyConditions + +func (r *FolderRepository) deleteOrphanedItemsByContentID(ctx context.Context, contentIDs []string) (int, []string, error) { + if len(contentIDs) == 0 { + return 0, nil, nil + } + + orphanIDs, err := r.collectOrphanedProvisionalIDsByContentID(ctx, contentIDs) + if err != nil { + return 0, nil, err + } + if len(orphanIDs) == 0 { + return 0, nil, nil + } + + imageDirs, err := collectImageDirs(ctx, r.pool, orphanIDs) + if err != nil { + return 0, nil, err + } + + var deleted int64 + if err := retryOnDeadlock(ctx, func() error { + tag, err := r.pool.Exec(ctx, deleteOrphanedProvisionalIDsByContentIDSQL, orphanIDs) + if err != nil { + return err + } + deleted = tag.RowsAffected() + return nil + }); err != nil { + return 0, nil, err + } + return int(deleted), imageDirs, nil +} + +func (r *FolderRepository) collectOrphanedProvisionalIDsByContentID(ctx context.Context, contentIDs []string) ([]string, error) { + rows, err := r.pool.Query(ctx, collectOrphanedProvisionalIDsByContentIDSQL, contentIDs) + if err != nil { + return nil, err + } + defer rows.Close() + + var orphanIDs []string + for rows.Next() { + var contentID string + if err := rows.Scan(&contentID); err != nil { + return nil, fmt.Errorf("scanning orphan content_id: %w", err) + } + orphanIDs = append(orphanIDs, contentID) + } + if err := rows.Err(); err != nil { + return nil, err + } + return orphanIDs, nil +} + // dirSetToSlice returns the keys of set as a slice (nil if empty). func dirSetToSlice(set map[string]struct{}) []string { if len(set) == 0 { diff --git a/internal/catalog/orphan_cleanup.go b/internal/catalog/orphan_cleanup.go new file mode 100644 index 00000000..4c661b1b --- /dev/null +++ b/internal/catalog/orphan_cleanup.go @@ -0,0 +1,200 @@ +package catalog + +import ( + "context" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" +) + +const defaultOrphanedProvisionalCleanupBatchSize = 1000 + +const orphanedMediaItemSafetyConditions = `NOT EXISTS ( + SELECT 1 FROM public.media_item_libraries mil + WHERE mil.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.media_files mf + WHERE mf.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.episodes e + WHERE e.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.seasons s + WHERE s.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.library_collection_items lci + WHERE lci.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.abs_bookmarks ab + WHERE ab.library_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.abs_playback_sessions aps + WHERE aps.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.abs_rss_feeds arf + WHERE arf.library_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.downloads d + WHERE d.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.playback_history_admin pha + WHERE pha.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.plex_sync_item_bindings psib + WHERE psib.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.plex_sync_item_state psis + WHERE psis.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.podcast_feeds pf + WHERE pf.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_favorites uf + WHERE uf.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_downloads ud + WHERE ud.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_history_hidden_items uhhi + WHERE uhhi.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_home_item_dismissals uhid + WHERE uhid.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_home_item_dismissals uhid_series + WHERE uhid_series.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_audio_preferences uap + WHERE uap.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_personal_collection_items upci + WHERE upci.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_ratings ur + WHERE ur.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_series_playback_preferences uspp + WHERE uspp.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_subtitle_preferences usp + WHERE usp.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_watch_history uwh + WHERE uwh.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_watch_progress uwp + WHERE uwp.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_watchlist uwl + WHERE uwl.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_provider_favorite_items wpfi + WHERE wpfi.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_provider_history_exports wphe + WHERE wphe.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_provider_scrobble_sessions wpss + WHERE wpss.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_together_rooms wtr + WHERE wtr.selected_content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_together_suggestions wts + WHERE wts.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.webhook_sync_item_state wsis + WHERE wsis.media_item_id = mi.content_id + )` + +const orphanedProvisionalMediaItemConditions = `mi.status IN ('pending', 'unmatched', 'ambiguous') + AND ` + orphanedMediaItemSafetyConditions + +const orphanedProvisionalMediaItemPredicate = ` +WHERE ` + orphanedProvisionalMediaItemConditions + +type OrphanedProvisionalCleanupStats struct { + Candidates int + Deleted int +} + +type OrphanedProvisionalCleaner struct { + pool *pgxpool.Pool +} + +func NewOrphanedProvisionalCleaner(pool *pgxpool.Pool) *OrphanedProvisionalCleaner { + return &OrphanedProvisionalCleaner{pool: pool} +} + +func (c *OrphanedProvisionalCleaner) Cleanup(ctx context.Context, batchSize int) (OrphanedProvisionalCleanupStats, error) { + var stats OrphanedProvisionalCleanupStats + if c == nil || c.pool == nil { + return stats, fmt.Errorf("orphaned provisional cleanup is not configured") + } + if batchSize <= 0 { + batchSize = defaultOrphanedProvisionalCleanupBatchSize + } + + if err := c.pool.QueryRow(ctx, + `SELECT COUNT(*) FROM public.media_items mi `+orphanedProvisionalMediaItemPredicate, + ).Scan(&stats.Candidates); err != nil { + return stats, fmt.Errorf("counting orphaned provisional media items: %w", err) + } + + for { + tag, err := c.pool.Exec(ctx, ` + WITH candidates AS ( + SELECT mi.content_id + FROM public.media_items mi + `+orphanedProvisionalMediaItemPredicate+` + ORDER BY mi.content_id ASC + LIMIT $1 + ) + DELETE FROM public.media_items mi + USING candidates c + WHERE mi.content_id = c.content_id + `, batchSize) + if err != nil { + return stats, fmt.Errorf("deleting orphaned provisional media items: %w", err) + } + + deleted := int(tag.RowsAffected()) + stats.Deleted += deleted + if deleted < batchSize { + break + } + } + + return stats, nil +} diff --git a/internal/catalog/orphan_cleanup_test.go b/internal/catalog/orphan_cleanup_test.go new file mode 100644 index 00000000..fa1734b1 --- /dev/null +++ b/internal/catalog/orphan_cleanup_test.go @@ -0,0 +1,41 @@ +package catalog + +import ( + "strings" + "testing" +) + +func normalizePredicateSQL(s string) string { + return strings.Join(strings.Fields(s), " ") +} + +func TestOrphanedProvisionalPredicatePreservesDurableMediaItemReferences(t *testing.T) { + predicate := normalizePredicateSQL(orphanedProvisionalMediaItemPredicate) + for _, want := range []string{ + "public.abs_bookmarks ab WHERE ab.library_item_id = mi.content_id", + "public.abs_playback_sessions aps WHERE aps.content_id = mi.content_id", + "public.abs_rss_feeds arf WHERE arf.library_item_id = mi.content_id", + "public.episodes e WHERE e.series_id = mi.content_id", + "public.podcast_feeds pf WHERE pf.media_item_id = mi.content_id", + "public.seasons s WHERE s.series_id = mi.content_id", + "public.user_audio_preferences uap WHERE uap.series_id = mi.content_id", + "public.user_home_item_dismissals uhid WHERE uhid.media_item_id = mi.content_id", + "public.user_home_item_dismissals uhid_series WHERE uhid_series.series_id = mi.content_id", + "public.user_personal_collection_items upci WHERE upci.media_item_id = mi.content_id", + "public.user_series_playback_preferences uspp WHERE uspp.series_id = mi.content_id", + "public.user_subtitle_preferences usp WHERE usp.series_id = mi.content_id", + } { + if !strings.Contains(predicate, normalizePredicateSQL(want)) { + t.Fatalf("cleanup predicate missing durable reference guard %q", want) + } + } + for _, droppedTable := range []string{ + "abs_collection_items", + "abs_playlist_items", + "abs_playlists", + } { + if strings.Contains(predicate, droppedTable) { + t.Fatalf("cleanup predicate must not reference dropped legacy table %q", droppedTable) + } + } +} diff --git a/internal/metadata/service.go b/internal/metadata/service.go index 06b43f9f..3cc23d26 100644 --- a/internal/metadata/service.go +++ b/internal/metadata/service.go @@ -70,6 +70,10 @@ type metadataItemRepo interface { ListUnmatchedByFolderAndPathPrefix(ctx context.Context, folderID int, pathPrefix string, limit int) ([]string, error) } +type metadataItemDeleteRepo interface { + Delete(ctx context.Context, contentID string) ([]string, error) +} + type metadataProviderIDRepo interface { GetByContentID(ctx context.Context, contentID string) ([]*models.MediaItemProviderID, error) ReplaceByContentID(ctx context.Context, contentID string, providerIDs map[string]string) error @@ -3741,18 +3745,9 @@ func (s *MetadataService) createOrFindSkeleton(ctx context.Context, file *models return s.hooks.createOrFindSkeleton(ctx, file, folderID) } - libraryType := "" - if s != nil && s.folderRepo != nil && folderID > 0 { - folder, err := s.folderRepo.GetByID(ctx, folderID) - if err != nil { - slog.Warn("metadata: failed to load folder context for path classification", - "folder_id", folderID, - "file_path", file.FilePath, - "error", err, - ) - } else if folder != nil { - libraryType = folder.Type - } + libraryType, err := s.folderTypeForSkeleton(ctx, folderID) + if err != nil { + return nil, err } contentRootPath := filepath.Dir(file.FilePath) @@ -4027,7 +4022,12 @@ func (s *MetadataService) createOrFindSkeleton(ctx context.Context, file *models // pre-confirmation ownership claims. Matching is driven by explicit IDs or // later provider confirmation to avoid cross-root false merges. - // No existing item — create skeleton. + // No existing item — create skeleton. Re-check the folder state while still + // under the dedup lock so a library disabled for deletion cannot receive new + // skeleton items from a matcher that passed an older outer enabled check. + if _, err := s.folderTypeForSkeleton(ctx, folderID); err != nil { + return nil, err + } contentID, err := generateContentID() if err != nil { return nil, fmt.Errorf("generate content id: %w", err) @@ -4054,10 +4054,53 @@ func (s *MetadataService) createOrFindSkeleton(ctx context.Context, file *models // Link file, claim root, and create library membership. if err := s.fileRepo.UpdateContentID(ctx, file.ID, contentID); err != nil { + if cleanupErr := s.deleteCreatedSkeleton(ctx, contentID); cleanupErr != nil { + slog.Warn("metadata: failed to clean up unlinked skeleton", + "content_id", contentID, + "file_id", file.ID, + "folder_id", folderID, + "error", cleanupErr, + ) + } return nil, fmt.Errorf("linking file to skeleton: %w", err) } + if _, err := s.folderTypeForSkeleton(ctx, folderID); err != nil { + if clearErr := s.fileRepo.UpdateContentID(ctx, file.ID, ""); clearErr != nil { + slog.Warn("metadata: failed to clear file link after disabled skeleton folder", + "content_id", contentID, + "file_id", file.ID, + "folder_id", folderID, + "error", clearErr, + ) + } + if cleanupErr := s.deleteCreatedSkeleton(ctx, contentID); cleanupErr != nil { + slog.Warn("metadata: failed to clean up skeleton for disabled folder", + "content_id", contentID, + "file_id", file.ID, + "folder_id", folderID, + "error", cleanupErr, + ) + } + return nil, fmt.Errorf("validating skeleton folder before membership: %w", err) + } if err := s.upsertLibraryMembership(ctx, contentID, folderID); err != nil { - s.logLibraryMembershipError("upserting skeleton membership", contentID, folderID, err) + if clearErr := s.fileRepo.UpdateContentID(ctx, file.ID, ""); clearErr != nil { + slog.Warn("metadata: failed to clear file link after skeleton membership failure", + "content_id", contentID, + "file_id", file.ID, + "folder_id", folderID, + "error", clearErr, + ) + } + if cleanupErr := s.deleteCreatedSkeleton(ctx, contentID); cleanupErr != nil { + slog.Warn("metadata: failed to clean up membershipless skeleton", + "content_id", contentID, + "file_id", file.ID, + "folder_id", folderID, + "error", cleanupErr, + ) + } + return nil, fmt.Errorf("upserting skeleton membership: %w", err) } res.ContentID = contentID @@ -4065,6 +4108,47 @@ func (s *MetadataService) createOrFindSkeleton(ctx context.Context, file *models return res, nil } +func (s *MetadataService) folderTypeForSkeleton(ctx context.Context, folderID int) (string, error) { + if s == nil || s.folderRepo == nil || folderID <= 0 { + return "", nil + } + + folder, err := s.folderRepo.GetByID(ctx, folderID) + if err != nil { + return "", fmt.Errorf("loading folder context for skeleton: %w", err) + } + if folder == nil { + return "", fmt.Errorf("folder %d is unavailable for skeleton creation", folderID) + } + if !folder.Enabled { + return "", fmt.Errorf("folder %d is disabled for skeleton creation", folderID) + } + return folder.Type, nil +} + +func (s *MetadataService) deleteCreatedSkeleton(ctx context.Context, contentID string) error { + if s == nil || strings.TrimSpace(contentID) == "" { + return nil + } + if repo, ok := s.itemRepo.(metadataItemDeleteRepo); ok { + if _, err := repo.Delete(ctx, contentID); err != nil && !errors.Is(err, catalog.ErrItemNotFound) { + return err + } + return nil + } + if s.dbPool == nil { + return nil + } + _, err := s.dbPool.Exec(ctx, ` + DELETE FROM media_items + WHERE content_id = $1 + AND NOT EXISTS ( + SELECT 1 FROM media_item_libraries mil + WHERE mil.content_id = media_items.content_id + )`, contentID) + return err +} + // claimRoot records a canonical root path claim for dedup. Failures are logged // but not fatal — the root claim is an optimization, not a hard requirement. func (s *MetadataService) claimRoot(ctx context.Context, folderID int, rootPath, contentID string) { diff --git a/internal/metadata/service_test.go b/internal/metadata/service_test.go index 71330401..e8a40645 100644 --- a/internal/metadata/service_test.go +++ b/internal/metadata/service_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "github.com/Silo-Server/silo-server/internal/catalog" "github.com/Silo-Server/silo-server/internal/models" ) @@ -81,6 +82,16 @@ func (r *fakeItemRepo) Upsert(_ context.Context, item *models.MediaItem) error { return nil } +func (r *fakeItemRepo) Delete(_ context.Context, contentID string) ([]string, error) { + r.mu.Lock() + defer r.mu.Unlock() + if _, ok := r.items[contentID]; !ok { + return nil, catalog.ErrItemNotFound + } + delete(r.items, contentID) + return nil, nil +} + func (r *fakeItemRepo) IncrementRefreshFailure(_ context.Context, contentID string) error { r.mu.Lock() defer r.mu.Unlock() @@ -259,6 +270,8 @@ type fakeFileRepo struct { groupContent map[string]string // "folderID:version:key" -> contentID groupFiles map[string][]*models.MediaFile matchStamps map[int]time.Time // fileID -> match_attempted_at + updateErrors map[int]error + afterUpdate func(fileID int, contentID string) claimUnmatchedCalls int claimNonSeriesCalls int claimMixedCalls int @@ -273,13 +286,22 @@ func newFakeFileRepo() *fakeFileRepo { groupContent: make(map[string]string), groupFiles: make(map[string][]*models.MediaFile), matchStamps: make(map[int]time.Time), + updateErrors: make(map[int]error), } } func (r *fakeFileRepo) UpdateContentID(_ context.Context, fileID int, contentID string) error { r.mu.Lock() - defer r.mu.Unlock() + if err := r.updateErrors[fileID]; err != nil { + r.mu.Unlock() + return err + } r.contentIDs[fileID] = contentID + afterUpdate := r.afterUpdate + r.mu.Unlock() + if afterUpdate != nil { + afterUpdate(fileID, contentID) + } return nil } @@ -1168,6 +1190,109 @@ func TestCreateOrFindSkeleton_NoFolderIDs(t *testing.T) { } } +func TestCreateOrFindSkeleton_DisabledFolderDoesNotCreateItem(t *testing.T) { + h := newTestHarness() + h.service.folderRepo = &fakeMetadataFolderRepo{ + folders: map[int]*models.MediaFolder{ + 10: {ID: 10, Type: "movies", Enabled: false}, + }, + } + ctx := context.Background() + + file := &models.MediaFile{ + ID: 1, + MediaFolderID: 10, + FilePath: "/media/movies/Inception (2010)/Inception.mkv", + } + + result, err := h.service.createOrFindSkeleton(ctx, file, 10) + if err == nil { + t.Fatal("expected disabled folder error") + } + if result != nil { + t.Fatalf("expected no result, got %+v", result) + } + if !strings.Contains(err.Error(), "disabled") { + t.Fatalf("expected disabled error, got %v", err) + } + if len(h.itemRepo.items) != 0 { + t.Fatalf("expected no items to be created, got %d", len(h.itemRepo.items)) + } + if cid := h.fileRepo.contentIDs[file.ID]; cid != "" { + t.Fatalf("expected file to remain unlinked, got content_id %q", cid) + } +} + +func TestCreateOrFindSkeleton_LinkFailureDeletesSkeleton(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + linkErr := errors.New("media file not found") + h.fileRepo.updateErrors[1] = linkErr + + file := &models.MediaFile{ + ID: 1, + MediaFolderID: 10, + FilePath: "/media/movies/Inception (2010)/Inception.mkv", + } + + result, err := h.service.createOrFindSkeleton(ctx, file, 10) + if err == nil { + t.Fatal("expected link error") + } + if result != nil { + t.Fatalf("expected no result, got %+v", result) + } + if !errors.Is(err, linkErr) { + t.Fatalf("expected wrapped link error, got %v", err) + } + if len(h.itemRepo.items) != 0 { + t.Fatalf("expected skeleton item to be deleted, got %d items", len(h.itemRepo.items)) + } + if len(h.libraryRepo.memberships) != 0 { + t.Fatalf("expected no library memberships, got %d", len(h.libraryRepo.memberships)) + } +} + +func TestCreateOrFindSkeleton_DisabledBeforeMembershipDeletesSkeleton(t *testing.T) { + h := newTestHarness() + ctx := context.Background() + folder := &models.MediaFolder{ID: 10, Type: "movies", Enabled: true} + h.service.folderRepo = &fakeMetadataFolderRepo{ + folders: map[int]*models.MediaFolder{10: folder}, + } + h.fileRepo.afterUpdate = func(_ int, contentID string) { + if contentID != "" { + folder.Enabled = false + } + } + + file := &models.MediaFile{ + ID: 1, + MediaFolderID: 10, + FilePath: "/media/movies/Inception (2010)/Inception.mkv", + } + + result, err := h.service.createOrFindSkeleton(ctx, file, 10) + if err == nil { + t.Fatal("expected disabled folder error") + } + if result != nil { + t.Fatalf("expected no result, got %+v", result) + } + if !strings.Contains(err.Error(), "disabled") { + t.Fatalf("expected disabled error, got %v", err) + } + if len(h.itemRepo.items) != 0 { + t.Fatalf("expected skeleton item to be deleted, got %d items", len(h.itemRepo.items)) + } + if len(h.libraryRepo.memberships) != 0 { + t.Fatalf("expected no library memberships, got %d", len(h.libraryRepo.memberships)) + } + if cid := h.fileRepo.contentIDs[file.ID]; cid != "" { + t.Fatalf("expected file link to be cleared, got content_id %q", cid) + } +} + // TestCreateOrFindSkeleton_PendingItemGetsLibraryMembership verifies that // library membership is created immediately for pending items, not just for // matched items. diff --git a/internal/scanner/file_repo.go b/internal/scanner/file_repo.go index e9000cdf..1a060537 100644 --- a/internal/scanner/file_repo.go +++ b/internal/scanner/file_repo.go @@ -2645,10 +2645,16 @@ func (r *FileRepository) ListByEpisodeIDs(ctx context.Context, episodeIDs []stri // UpdateContentID sets the content_id on a media file, linking it to a matched // media item. This is called by the matcher after a successful resolution. func (r *FileRepository) UpdateContentID(ctx context.Context, fileID int, contentID string) error { - _, err := r.pool.Exec(ctx, + tag, err := r.pool.Exec(ctx, "UPDATE media_files SET content_id = $1, updated_at = NOW() WHERE id = $2", contentID, fileID) - return err + if err != nil { + return err + } + if tag.RowsAffected() == 0 { + return ErrFileNotFound + } + return nil } // ReplaceContentID reassigns all files linked to one content item to another. diff --git a/internal/taskmanager/tasks/cleanup_orphaned_media_items.go b/internal/taskmanager/tasks/cleanup_orphaned_media_items.go new file mode 100644 index 00000000..c45ee321 --- /dev/null +++ b/internal/taskmanager/tasks/cleanup_orphaned_media_items.go @@ -0,0 +1,81 @@ +package tasks + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/taskmanager" +) + +const orphanedMediaCleanupBatchSize = 1000 + +type OrphanedMediaItemCleaner interface { + Cleanup(ctx context.Context, batchSize int) (catalog.OrphanedProvisionalCleanupStats, error) +} + +type CleanupOrphanedMediaItemsTask struct { + cleaner OrphanedMediaItemCleaner +} + +type cleanupOrphanedMediaItemsResult struct { + Candidates int `json:"candidates"` + Deleted int `json:"deleted"` +} + +func NewCleanupOrphanedMediaItemsTask(cleaner OrphanedMediaItemCleaner) *CleanupOrphanedMediaItemsTask { + return &CleanupOrphanedMediaItemsTask{cleaner: cleaner} +} + +func (t *CleanupOrphanedMediaItemsTask) Key() string { return "cleanup_orphaned_media_items" } +func (t *CleanupOrphanedMediaItemsTask) Name() string { + return "Clean Orphaned Media Items" +} +func (t *CleanupOrphanedMediaItemsTask) Description() string { + return "Deletes detached provisional media items that have no files, library memberships, or durable user/sync references." +} +func (t *CleanupOrphanedMediaItemsTask) Category() taskmanager.TaskCategory { + return taskmanager.TaskCategoryLibrary +} +func (t *CleanupOrphanedMediaItemsTask) IsHidden() bool { return false } + +func (t *CleanupOrphanedMediaItemsTask) DefaultTriggers() []taskmanager.TriggerConfig { + return nil +} + +func (t *CleanupOrphanedMediaItemsTask) Execute(ctx context.Context, progress taskmanager.ProgressReporter) error { + if t == nil || t.cleaner == nil { + progress.Report(100, "Orphaned media cleanup is not configured") + return nil + } + + progress.Report(0, "Checking for orphaned provisional media items") + stats, err := t.cleaner.Cleanup(ctx, orphanedMediaCleanupBatchSize) + setCleanupOrphanedMediaItemsResult(progress, cleanupOrphanedMediaItemsResult{ + Candidates: stats.Candidates, + Deleted: stats.Deleted, + }) + if err != nil { + progress.Report(100, fmt.Sprintf("Orphaned media cleanup failed after deleting %d of %d candidates", stats.Deleted, stats.Candidates)) + return err + } + + if stats.Candidates == 0 { + progress.Report(100, "No orphaned provisional media items found") + return nil + } + progress.Report(100, fmt.Sprintf("Deleted %d orphaned provisional media items (%d candidates)", stats.Deleted, stats.Candidates)) + return nil +} + +func setCleanupOrphanedMediaItemsResult(progress taskmanager.ProgressReporter, result cleanupOrphanedMediaItemsResult) { + if progress == nil { + return + } + data, err := json.Marshal(result) + if err != nil { + return + } + progress.SetResultData(data) +} diff --git a/internal/taskmanager/tasks/cleanup_orphaned_media_items_test.go b/internal/taskmanager/tasks/cleanup_orphaned_media_items_test.go new file mode 100644 index 00000000..eca3a39c --- /dev/null +++ b/internal/taskmanager/tasks/cleanup_orphaned_media_items_test.go @@ -0,0 +1,112 @@ +package tasks + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/taskmanager" +) + +type fakeOrphanedMediaCleaner struct { + stats catalog.OrphanedProvisionalCleanupStats + err error + calls int + batchSize int +} + +func (f *fakeOrphanedMediaCleaner) Cleanup(_ context.Context, batchSize int) (catalog.OrphanedProvisionalCleanupStats, error) { + f.calls++ + f.batchSize = batchSize + return f.stats, f.err +} + +type orphanedMediaCleanupProgress struct { + reports []string + result json.RawMessage +} + +func (p *orphanedMediaCleanupProgress) Report(_ float64, message string) { + p.reports = append(p.reports, message) +} + +func (p *orphanedMediaCleanupProgress) SetResultData(data json.RawMessage) { + p.result = append(p.result[:0], data...) +} + +func TestCleanupOrphanedMediaItemsTask(t *testing.T) { + cleaner := &fakeOrphanedMediaCleaner{ + stats: catalog.OrphanedProvisionalCleanupStats{Candidates: 5, Deleted: 5}, + } + task := NewCleanupOrphanedMediaItemsTask(cleaner) + + if task.Key() != "cleanup_orphaned_media_items" { + t.Fatalf("Key() = %q, want cleanup_orphaned_media_items", task.Key()) + } + if task.Category() != taskmanager.TaskCategoryLibrary { + t.Fatalf("Category() = %q, want %q", task.Category(), taskmanager.TaskCategoryLibrary) + } + if task.IsHidden() { + t.Fatal("IsHidden() = true, want false") + } + if triggers := task.DefaultTriggers(); len(triggers) != 0 { + t.Fatalf("DefaultTriggers() = %#v, want none", triggers) + } + + progress := &orphanedMediaCleanupProgress{} + if err := task.Execute(context.Background(), progress); err != nil { + t.Fatalf("Execute returned error: %v", err) + } + if cleaner.calls != 1 { + t.Fatalf("cleaner calls = %d, want 1", cleaner.calls) + } + if cleaner.batchSize != orphanedMediaCleanupBatchSize { + t.Fatalf("batchSize = %d, want %d", cleaner.batchSize, orphanedMediaCleanupBatchSize) + } + last := progress.reports[len(progress.reports)-1] + if !strings.Contains(last, "Deleted 5 orphaned provisional media items") { + t.Fatalf("last progress report = %q", last) + } + var result cleanupOrphanedMediaItemsResult + if err := json.Unmarshal(progress.result, &result); err != nil { + t.Fatalf("unmarshal result data: %v", err) + } + if result.Candidates != 5 || result.Deleted != 5 { + t.Fatalf("result = %#v, want 5 candidates and 5 deleted", result) + } +} + +func TestCleanupOrphanedMediaItemsTaskNoCandidates(t *testing.T) { + task := NewCleanupOrphanedMediaItemsTask(&fakeOrphanedMediaCleaner{}) + progress := &orphanedMediaCleanupProgress{} + + if err := task.Execute(context.Background(), progress); err != nil { + t.Fatalf("Execute returned error: %v", err) + } + last := progress.reports[len(progress.reports)-1] + if last != "No orphaned provisional media items found" { + t.Fatalf("last progress report = %q", last) + } +} + +func TestCleanupOrphanedMediaItemsTaskReturnsCleanupError(t *testing.T) { + wantErr := errors.New("cleanup failed") + cleaner := &fakeOrphanedMediaCleaner{ + stats: catalog.OrphanedProvisionalCleanupStats{Candidates: 5, Deleted: 2}, + err: wantErr, + } + task := NewCleanupOrphanedMediaItemsTask(cleaner) + progress := &orphanedMediaCleanupProgress{} + + err := task.Execute(context.Background(), progress) + if !errors.Is(err, wantErr) { + t.Fatalf("Execute error = %v, want %v", err, wantErr) + } + last := progress.reports[len(progress.reports)-1] + if !strings.Contains(last, "failed after deleting 2 of 5 candidates") { + t.Fatalf("last progress report = %q", last) + } +} diff --git a/migrations/cleanup_orphaned_provisional_items_test.go b/migrations/cleanup_orphaned_provisional_items_test.go new file mode 100644 index 00000000..7002588b --- /dev/null +++ b/migrations/cleanup_orphaned_provisional_items_test.go @@ -0,0 +1,41 @@ +package migrations + +import ( + "strings" + "testing" +) + +func TestCleanupOrphanedProvisionalItemsPreservesDurableReferences(t *testing.T) { + migrationBytes, err := FS.ReadFile("sql/20260607220000_cleanup_orphaned_provisional_items.sql") + if err != nil { + t.Fatalf("read migration: %v", err) + } + migration := normalizeSQL(string(migrationBytes)) + for _, want := range []string{ + "public.abs_bookmarks ab WHERE ab.library_item_id = mi.content_id", + "public.abs_playback_sessions aps WHERE aps.content_id = mi.content_id", + "public.abs_rss_feeds arf WHERE arf.library_item_id = mi.content_id", + "public.episodes e WHERE e.series_id = mi.content_id", + "public.podcast_feeds pf WHERE pf.media_item_id = mi.content_id", + "public.seasons s WHERE s.series_id = mi.content_id", + "public.user_audio_preferences uap WHERE uap.series_id = mi.content_id", + "public.user_home_item_dismissals uhid WHERE uhid.media_item_id = mi.content_id", + "public.user_home_item_dismissals uhid_series WHERE uhid_series.series_id = mi.content_id", + "public.user_personal_collection_items upci WHERE upci.media_item_id = mi.content_id", + "public.user_series_playback_preferences uspp WHERE uspp.series_id = mi.content_id", + "public.user_subtitle_preferences usp WHERE usp.series_id = mi.content_id", + } { + if !strings.Contains(migration, normalizeSQL(want)) { + t.Fatalf("cleanup migration missing durable reference guard %q", want) + } + } + for _, droppedTable := range []string{ + "abs_collection_items", + "abs_playlist_items", + "abs_playlists", + } { + if strings.Contains(migration, droppedTable) { + t.Fatalf("cleanup migration references dropped table %q", droppedTable) + } + } +} diff --git a/migrations/sql/20260607220000_cleanup_orphaned_provisional_items.sql b/migrations/sql/20260607220000_cleanup_orphaned_provisional_items.sql new file mode 100644 index 00000000..f226ed3a --- /dev/null +++ b/migrations/sql/20260607220000_cleanup_orphaned_provisional_items.sql @@ -0,0 +1,142 @@ +-- +goose Up +-- +goose StatementBegin +-- Clean up provisional catalog rows left behind by library-delete/matcher races. +-- Stale external IDs and other item-owned metadata cascade from media_items. +-- Durable user/sync references do not all have media_items foreign keys, so +-- keep any provisional row that is still referenced outside derived metadata. +DELETE FROM public.media_items mi +WHERE mi.status IN ('pending', 'unmatched', 'ambiguous') + AND NOT EXISTS ( + SELECT 1 FROM public.media_item_libraries mil + WHERE mil.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.media_files mf + WHERE mf.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.episodes e + WHERE e.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.seasons s + WHERE s.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.library_collection_items lci + WHERE lci.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.abs_bookmarks ab + WHERE ab.library_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.abs_playback_sessions aps + WHERE aps.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.abs_rss_feeds arf + WHERE arf.library_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.downloads d + WHERE d.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.playback_history_admin pha + WHERE pha.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.plex_sync_item_bindings psib + WHERE psib.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.plex_sync_item_state psis + WHERE psis.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.podcast_feeds pf + WHERE pf.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_favorites uf + WHERE uf.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_downloads ud + WHERE ud.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_history_hidden_items uhhi + WHERE uhhi.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_home_item_dismissals uhid + WHERE uhid.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_home_item_dismissals uhid_series + WHERE uhid_series.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_audio_preferences uap + WHERE uap.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_personal_collection_items upci + WHERE upci.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_ratings ur + WHERE ur.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_series_playback_preferences uspp + WHERE uspp.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_subtitle_preferences usp + WHERE usp.series_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_watch_history uwh + WHERE uwh.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_watch_progress uwp + WHERE uwp.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.user_watchlist uwl + WHERE uwl.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_provider_favorite_items wpfi + WHERE wpfi.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_provider_history_exports wphe + WHERE wphe.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_provider_scrobble_sessions wpss + WHERE wpss.media_item_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_together_rooms wtr + WHERE wtr.selected_content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.watch_together_suggestions wts + WHERE wts.content_id = mi.content_id + ) + AND NOT EXISTS ( + SELECT 1 FROM public.webhook_sync_item_state wsis + WHERE wsis.media_item_id = mi.content_id + ); +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +-- Data cleanup only; deleted orphaned provisional rows cannot be reconstructed. +-- +goose StatementEnd