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
This commit is contained in:
zZebrahz
2026-06-07 21:32:40 -04:00
committed by GitHub
parent 978e1b4954
commit 96bae2f346
12 changed files with 1005 additions and 39 deletions
+1
View File
@@ -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))
}
+13
View File
@@ -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)
}
}
}
+142 -22
View File
@@ -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 {
+200
View File
@@ -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
}
+41
View File
@@ -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)
}
}
}
+98 -14
View File
@@ -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) {
+126 -1
View File
@@ -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.
+8 -2
View File
@@ -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.
@@ -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)
}
@@ -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)
}
}
@@ -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)
}
}
}
@@ -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