diff --git a/internal/api/handlers/admin_split.go b/internal/api/handlers/admin_split.go index 5a6afc51..1c96d232 100644 --- a/internal/api/handlers/admin_split.go +++ b/internal/api/handlers/admin_split.go @@ -18,6 +18,7 @@ import ( "github.com/Silo-Server/silo-server/internal/api/middleware" "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/catalog/filesplit" "github.com/Silo-Server/silo-server/internal/catalog/reattribute" "github.com/Silo-Server/silo-server/internal/contentid" "github.com/Silo-Server/silo-server/internal/metadata" @@ -106,16 +107,9 @@ type mergeItemRequest struct { Into string `json:"into"` } -// splitFile is the slice of media_files the split logic needs. -type splitFile struct { - ID int - MediaFolderID int - FilePath string - ObservedRootPath string - SeasonNumber int - EpisodeNumber int - EpisodeID string -} +// splitFile remains a local alias so the HTTP layer and the automatic repair +// path share the same transactional file-move primitive. +type splitFile = filesplit.File type itemFileResponse struct { ID int `json:"id"` @@ -229,8 +223,6 @@ func (h *AdminSplitHandler) HandleSplitItem(w http.ResponseWriter, r *http.Reque return } - episodePairs := deriveEpisodePairs(sourceItem, moved, target.contentID) - // Everything transactional happens here; a dry run rolls back at the end. tx, err := h.pool.Begin(ctx) if err != nil { @@ -247,7 +239,14 @@ func (h *AdminSplitHandler) HandleSplitItem(w http.ResponseWriter, r *http.Reque return } } - if err := moveFilesToItem(ctx, tx, target.contentID, moved); err != nil { + moveResult, err := filesplit.Move(ctx, tx, filesplit.Options{ + FromContentID: sourceID, + ToContentID: target.contentID, + ItemType: sourceItem.Type, + Files: moved, + HistoryMode: mode, + }) + if err != nil { slog.ErrorContext(ctx, "admin split: moving files", "component", "api", "target", target.contentID, "error", err) writeError(w, http.StatusInternalServerError, "internal_error", "Failed to move files") return @@ -264,18 +263,8 @@ func (h *AdminSplitHandler) HandleSplitItem(w http.ResponseWriter, r *http.Reque } } - report, err := reattribute.Run(ctx, tx, reattribute.Options{ - FromContentID: sourceID, - ToContentID: target.contentID, - MovedFileIDs: fileIDs(moved), - Mode: mode, - EpisodePairs: episodePairs, - }) - if err != nil { - slog.ErrorContext(ctx, "admin split: reattributing user state", "component", "api", "error", err) - writeError(w, http.StatusInternalServerError, "internal_error", "Failed to reattribute user state") - return - } + report := moveResult.Reattribution + episodePairs := moveResult.EpisodePairs if !req.DryRun { if err := tx.Commit(ctx); err != nil { @@ -449,29 +438,17 @@ func (h *AdminSplitHandler) resolveSplitTarget( // possible when the target id is provider-anchored; local targets get no // episode-level reattribution (state stays behind, reported to the operator). func deriveEpisodePairs(sourceItem *models.MediaItem, moved []splitFile, targetContentID string) []reattribute.IDPair { - if sourceItem.Type != "series" || !contentid.IsProviderAnchored(targetContentID) { - return nil - } - seen := map[string]bool{} - var pairs []reattribute.IDPair - for _, f := range moved { - if f.EpisodeID == "" || f.SeasonNumber <= 0 || f.EpisodeNumber <= 0 || seen[f.EpisodeID] { - continue - } - newID, ok := contentid.ForEpisode(targetContentID, f.SeasonNumber, f.EpisodeNumber) - if !ok || newID == f.EpisodeID { - continue - } - seen[f.EpisodeID] = true - pairs = append(pairs, reattribute.IDPair{From: f.EpisodeID, To: newID}) - } - return pairs + return filesplit.DeriveEpisodePairs(sourceItem.Type, moved, targetContentID) } func (h *AdminSplitHandler) loadItemFiles(ctx context.Context, contentID string) ([]splitFile, error) { rows, err := h.pool.Query(ctx, ` - SELECT id, media_folder_id, file_path, + SELECT id, COALESCE(content_id, ''), media_folder_id, file_path, + COALESCE(canonical_root_path, ''), COALESCE(observed_root_path, ''), + COALESCE(group_key_version, 1), + COALESCE(content_group_key, ''), + COALESCE(base_type, ''), COALESCE(season_number, 0), COALESCE(episode_number, 0), COALESCE(episode_id, '') @@ -487,7 +464,20 @@ func (h *AdminSplitHandler) loadItemFiles(ctx context.Context, contentID string) var files []splitFile for rows.Next() { var f splitFile - if err := rows.Scan(&f.ID, &f.MediaFolderID, &f.FilePath, &f.ObservedRootPath, &f.SeasonNumber, &f.EpisodeNumber, &f.EpisodeID); err != nil { + if err := rows.Scan( + &f.ID, + &f.ContentID, + &f.MediaFolderID, + &f.FilePath, + &f.CanonicalRootPath, + &f.ObservedRootPath, + &f.GroupKeyVersion, + &f.ContentGroupKey, + &f.BaseType, + &f.SeasonNumber, + &f.EpisodeNumber, + &f.EpisodeID, + ); err != nil { return nil, err } if f.ObservedRootPath == "" { @@ -528,33 +518,6 @@ func insertSkeletonItem(ctx context.Context, tx pgx.Tx, target splitTarget, sour return nil } -func moveFilesToItem(ctx context.Context, tx pgx.Tx, targetContentID string, moved []splitFile) error { - // episode_id resets: the metadata refresh / scanner relink recreates the - // links against the target series' (deterministic) episode rows. - _, err := tx.Exec(ctx, ` - UPDATE media_files - SET content_id = $1, - episode_id = NULL, - match_attempted_at = NULL, - updated_at = NOW() - WHERE id = ANY($2::int[]) - `, targetContentID, fileIDs(moved)) - if err != nil { - return err - } - // Library membership for the target in every folder that received files. - for _, folderID := range distinctFolderIDs(moved) { - if _, err := tx.Exec(ctx, ` - INSERT INTO media_item_libraries (content_id, media_folder_id) - VALUES ($1, $2) - ON CONFLICT DO NOTHING - `, targetContentID, folderID); err != nil { - return err - } - } - return nil -} - // persistOverrides writes the durable path-scoped identity overrides: one // root-scope override per observed root whose files ALL moved, file-scope // overrides for partially-moved roots. @@ -688,14 +651,6 @@ func (h *AdminSplitHandler) runPostSplitFollowUps(sourceID string, target splitT }() } -func fileIDs(files []splitFile) []int { - ids := make([]int, 0, len(files)) - for _, f := range files { - ids = append(ids, f.ID) - } - return ids -} - func distinctFolderIDs(files []splitFile) []int { seen := map[int]bool{} var ids []int diff --git a/internal/catalog/filesplit/filesplit.go b/internal/catalog/filesplit/filesplit.go new file mode 100644 index 00000000..691ecb92 --- /dev/null +++ b/internal/catalog/filesplit/filesplit.go @@ -0,0 +1,283 @@ +// Package filesplit moves a subset of a catalog item's files to an existing +// item while preserving file-attributable user state. It is shared by the +// admin Split Versions flow and automatic scanner identity repair. +package filesplit + +import ( + "context" + "fmt" + "sort" + + "github.com/jackc/pgx/v5" + + "github.com/Silo-Server/silo-server/internal/catalog" + "github.com/Silo-Server/silo-server/internal/catalog/reattribute" + "github.com/Silo-Server/silo-server/internal/contentid" +) + +const ( + itemTypeMovie = "movie" + itemTypeSeries = "series" +) + +// File is the media-file identity needed for a subset move. +type File struct { + ID int + ContentID string + MediaFolderID int + FilePath string + CanonicalRootPath string + ObservedRootPath string + GroupKeyVersion int + ContentGroupKey string + BaseType string + SeasonNumber int + EpisodeNumber int + EpisodeID string +} + +// Options describes one transactional file-subset move. +type Options struct { + FromContentID string + ToContentID string + ItemType string + Files []File + HistoryMode reattribute.HistoryMode +} + +// Result reports the state reattribution performed by Move. +type Result struct { + EpisodePairs []reattribute.IDPair + Reattribution *reattribute.Report +} + +// Move reassigns files and file-attributable user state inside the caller's +// transaction. The source predicate makes stale/concurrent selections fail +// closed instead of moving an unexpected row. +func Move(ctx context.Context, tx pgx.Tx, opts Options) (*Result, error) { + if opts.FromContentID == "" || opts.ToContentID == "" || opts.FromContentID == opts.ToContentID { + return nil, fmt.Errorf("filesplit: from/to content ids required and must differ") + } + if opts.ItemType != itemTypeMovie && opts.ItemType != itemTypeSeries { + return nil, fmt.Errorf("filesplit: unsupported item type %q", opts.ItemType) + } + if len(opts.Files) == 0 { + return nil, fmt.Errorf("filesplit: at least one file is required") + } + if opts.HistoryMode == "" { + opts.HistoryMode = reattribute.HistoryModeEvidence + } + if !reattribute.ValidHistoryMode(opts.HistoryMode) { + return nil, fmt.Errorf("filesplit: invalid history mode %q", opts.HistoryMode) + } + + fileIDs := distinctFileIDs(opts.Files) + if len(fileIDs) != len(opts.Files) { + return nil, fmt.Errorf("filesplit: file ids must be positive and unique") + } + tag, err := tx.Exec(ctx, ` + UPDATE media_files + SET content_id = $1, + episode_id = NULL, + match_attempted_at = NULL, + updated_at = NOW() + WHERE content_id = $2 + AND id = ANY($3::int[]) + `, opts.ToContentID, opts.FromContentID, fileIDs) + if err != nil { + return nil, fmt.Errorf("filesplit: moving files: %w", err) + } + if tag.RowsAffected() != int64(len(fileIDs)) { + return nil, fmt.Errorf( + "filesplit: moved %d of %d selected files; source ownership changed", + tag.RowsAffected(), + len(fileIDs), + ) + } + + for _, folderID := range distinctFolderIDs(opts.Files) { + if _, err := tx.Exec(ctx, ` + INSERT INTO media_item_libraries (content_id, media_folder_id) + VALUES ($1, $2) + ON CONFLICT DO NOTHING + `, opts.ToContentID, folderID); err != nil { + return nil, fmt.Errorf("filesplit: adding target library membership: %w", err) + } + } + + derivedEpisodePairs := DeriveEpisodePairs(opts.ItemType, opts.Files, opts.ToContentID) + episodePairs, err := keepFullyMovedEpisodePairs(ctx, tx, derivedEpisodePairs) + if err != nil { + return nil, err + } + report, err := reattribute.Run(ctx, tx, reattribute.Options{ + FromContentID: opts.FromContentID, + ToContentID: opts.ToContentID, + MovedFileIDs: fileIDs, + Mode: opts.HistoryMode, + EpisodePairs: episodePairs, + }) + if err != nil { + return nil, fmt.Errorf("filesplit: reattributing user state: %w", err) + } + + // Existing matched series already have deterministic target episode rows. + // Relink immediately when they are available; a later metadata refresh can + // fill any episode that does not exist yet. + if opts.ItemType == itemTypeSeries { + if _, err := tx.Exec(ctx, ` + UPDATE media_files mf + SET episode_id = e.content_id, + updated_at = NOW() + FROM episodes e + WHERE mf.id = ANY($1::int[]) + AND e.series_id = $2 + AND e.season_number = mf.season_number + AND e.episode_number = mf.episode_number + `, fileIDs, opts.ToContentID); err != nil { + return nil, fmt.Errorf("filesplit: relinking target episodes: %w", err) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO episode_libraries (episode_id, media_folder_id, first_seen_at) + SELECT mf.episode_id, mf.media_folder_id, MIN(mf.created_at) + FROM media_files mf + WHERE mf.id = ANY($1::int[]) + AND mf.episode_id IS NOT NULL + GROUP BY mf.episode_id, mf.media_folder_id + ON CONFLICT (episode_id, media_folder_id) DO NOTHING + `, fileIDs); err != nil { + return nil, fmt.Errorf("filesplit: adding target episode library membership: %w", err) + } + if len(derivedEpisodePairs) > 0 { + if _, err := tx.Exec(ctx, ` + DELETE FROM episode_libraries source_membership + WHERE source_membership.episode_id = ANY($1::text[]) + AND NOT EXISTS ( + SELECT 1 + FROM media_files remaining + WHERE remaining.episode_id = source_membership.episode_id + AND remaining.media_folder_id = source_membership.media_folder_id + AND remaining.missing_since IS NULL + ) + `, episodePairSources(derivedEpisodePairs)); err != nil { + return nil, fmt.Errorf("filesplit: removing stale source episode library membership: %w", err) + } + } + if err := catalog.RecomputeSeriesLatestEpisodeAdded( + ctx, + tx, + []string{opts.FromContentID, opts.ToContentID}, + ); err != nil { + return nil, fmt.Errorf("filesplit: %w", err) + } + } + + return &Result{ + EpisodePairs: episodePairs, + Reattribution: report, + }, nil +} + +// DeriveEpisodePairs maps moved files' current episode ids onto deterministic +// episode ids under the target series. +func DeriveEpisodePairs(itemType string, files []File, targetContentID string) []reattribute.IDPair { + if itemType != itemTypeSeries || !contentid.IsProviderAnchored(targetContentID) { + return nil + } + seen := make(map[string]bool) + pairs := make([]reattribute.IDPair, 0) + for _, file := range files { + if file.EpisodeID == "" || file.SeasonNumber <= 0 || file.EpisodeNumber <= 0 || seen[file.EpisodeID] { + continue + } + newID, ok := contentid.ForEpisode(targetContentID, file.SeasonNumber, file.EpisodeNumber) + if !ok || newID == file.EpisodeID { + continue + } + seen[file.EpisodeID] = true + pairs = append(pairs, reattribute.IDPair{From: file.EpisodeID, To: newID}) + } + return pairs +} + +// keepFullyMovedEpisodePairs excludes a source episode when another active +// file still points at it after the selected files were detached. Episode +// progress has no file dimension, so moving a shared episode's state would +// steal progress from the source show. +func keepFullyMovedEpisodePairs( + ctx context.Context, + tx pgx.Tx, + pairs []reattribute.IDPair, +) ([]reattribute.IDPair, error) { + if len(pairs) == 0 { + return nil, nil + } + fromIDs := make([]string, 0, len(pairs)) + toIDs := make([]string, 0, len(pairs)) + for _, pair := range pairs { + fromIDs = append(fromIDs, pair.From) + toIDs = append(toIDs, pair.To) + } + + rows, err := tx.Query(ctx, ` + SELECT pair.from_id, pair.to_id + FROM UNNEST($1::text[], $2::text[]) AS pair(from_id, to_id) + WHERE NOT EXISTS ( + SELECT 1 + FROM media_files remaining + WHERE remaining.episode_id = pair.from_id + AND remaining.missing_since IS NULL + ) + `, fromIDs, toIDs) + if err != nil { + return nil, fmt.Errorf("filesplit: filtering shared episode state: %w", err) + } + defer rows.Close() + + filtered := make([]reattribute.IDPair, 0, len(pairs)) + for rows.Next() { + var pair reattribute.IDPair + if err := rows.Scan(&pair.From, &pair.To); err != nil { + return nil, fmt.Errorf("filesplit: scanning movable episode state: %w", err) + } + filtered = append(filtered, pair) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("filesplit: iterating movable episode state: %w", err) + } + return filtered, nil +} + +func episodePairSources(pairs []reattribute.IDPair) []string { + ids := make([]string, 0, len(pairs)) + for _, pair := range pairs { + ids = append(ids, pair.From) + } + return ids +} + +func distinctFileIDs(files []File) []int { + seen := make(map[int]bool, len(files)) + ids := make([]int, 0, len(files)) + for _, file := range files { + if file.ID > 0 && !seen[file.ID] { + seen[file.ID] = true + ids = append(ids, file.ID) + } + } + sort.Ints(ids) + return ids +} + +func distinctFolderIDs(files []File) []int { + seen := make(map[int]bool, len(files)) + ids := make([]int, 0) + for _, file := range files { + if file.MediaFolderID > 0 && !seen[file.MediaFolderID] { + seen[file.MediaFolderID] = true + ids = append(ids, file.MediaFolderID) + } + } + sort.Ints(ids) + return ids +} diff --git a/internal/metadata/anchored_identity_repair.go b/internal/metadata/anchored_identity_repair.go new file mode 100644 index 00000000..c5875e6a --- /dev/null +++ b/internal/metadata/anchored_identity_repair.go @@ -0,0 +1,891 @@ +package metadata + +import ( + "context" + "errors" + "fmt" + "log/slog" + "path/filepath" + "sort" + "strconv" + "strings" + + "github.com/jackc/pgx/v5" + + "github.com/Silo-Server/silo-server/internal/catalog/filesplit" + "github.com/Silo-Server/silo-server/internal/catalog/reattribute" + "github.com/Silo-Server/silo-server/internal/contentid" + "github.com/Silo-Server/silo-server/internal/pathscope" +) + +const ( + anchoredIdentityRepairBatchSize = 100 + anchoredItemTypeMovie = "movie" + anchoredItemTypeSeries = "series" +) + +type anchoredGroupIdentity struct { + Version int + ItemType string + Provider string + ProviderID string + TargetContent string +} + +type anchoredIdentityRepairCandidate struct { + FolderID int + ObservedRootPath string + SourceContentID string + GroupKeyVersion int + ContentGroupKey string + ItemType string + FileCount int +} + +type anchoredIdentityRepairGroupCursor struct { + SourceContentID string + GroupKeyVersion int + ContentGroupKey string + ItemType string +} + +// repairAnchoredIdentityMismatchesByFolderAndPathPrefix repairs historical +// wrong-version merges after a scan has persisted provider-anchored group +// keys. It deliberately requires a complete root, an existing unique matched +// provider owner, and a proper source subset. Anything less stays manual. +func (s *MetadataService) repairAnchoredIdentityMismatchesByFolderAndPathPrefix( + ctx context.Context, + folderID int, + pathPrefix string, +) (int, error) { + return s.repairAnchoredIdentityMismatchesByFolderAndPathPrefixPageSize( + ctx, + folderID, + pathPrefix, + anchoredIdentityRepairBatchSize, + ) +} + +func (s *MetadataService) repairAnchoredIdentityMismatchesByFolderAndPathPrefixPageSize( + ctx context.Context, + folderID int, + pathPrefix string, + pageSize int, +) (int, error) { + if s == nil || s.dbPool == nil || folderID <= 0 || strings.TrimSpace(pathPrefix) == "" { + return 0, nil + } + if pageSize <= 0 { + pageSize = anchoredIdentityRepairBatchSize + } + + pathPrefix = filepath.Clean(pathPrefix) + var afterGroup anchoredIdentityRepairGroupCursor + repaired := 0 + for { + candidateGroups, err := s.loadAnchoredIdentityRepairCandidateGroups( + ctx, + folderID, + pathPrefix, + afterGroup, + pageSize, + ) + if err != nil { + return repaired, err + } + if len(candidateGroups) == 0 { + if err := s.ensureAnchoredIdentityRepairEpisodeLinks(ctx, folderID, pathPrefix); err != nil { + return repaired, err + } + return repaired, nil + } + + for _, candidates := range candidateGroups { + didRepair, repairErr := s.repairAnchoredIdentityMismatchGroup(ctx, candidates) + if repairErr != nil { + return repaired, repairErr + } + if didRepair { + repaired += len(candidates) + if candidates[0].ItemType == anchoredItemTypeSeries { + anchor, valid := parseAnchoredGroupIdentity( + candidates[0].GroupKeyVersion, + candidates[0].ContentGroupKey, + ) + if valid { + if followUpErr := s.ensureAnchoredIdentityRepairTargetEpisodeLinks( + ctx, + anchor.TargetContent, + ); followUpErr != nil { + return repaired, followUpErr + } + } + } + } + } + lastGroup := candidateGroups[len(candidateGroups)-1] + afterGroup = anchoredIdentityRepairCursor(lastGroup[0]) + } +} + +func (s *MetadataService) ensureAnchoredIdentityRepairEpisodeLinks( + ctx context.Context, + folderID int, + pathPrefix string, +) error { + rows, err := s.dbPool.Query(ctx, ` + SELECT DISTINCT mf.content_id + FROM media_files mf + JOIN media_items item + ON item.content_id = mf.content_id + WHERE mf.media_folder_id = $1 + AND mf.missing_since IS NULL + AND mf.episode_id IS NULL + AND LOWER(TRIM(mf.base_type)) = 'series' + AND SPLIT_PART(mf.content_group_key, '|', 3) = 'anchor' + AND mf.content_id = + LOWER(TRIM(mf.base_type)) || '-' || SPLIT_PART(mf.content_group_key, '|', 4) + AND LOWER(TRIM(item.status)) = 'matched' + AND (mf.file_path = $2 OR mf.file_path LIKE $3 ESCAPE '\') + ORDER BY mf.content_id + `, folderID, pathPrefix, pathscope.PrefixLike(pathPrefix)) + if err != nil { + return fmt.Errorf("querying anchored repair episode follow-ups: %w", err) + } + + contentIDs := make([]string, 0) + for rows.Next() { + var contentID string + if err := rows.Scan(&contentID); err != nil { + rows.Close() + return fmt.Errorf("scanning anchored repair episode follow-up: %w", err) + } + contentIDs = append(contentIDs, contentID) + } + if err := rows.Err(); err != nil { + rows.Close() + return fmt.Errorf("iterating anchored repair episode follow-ups: %w", err) + } + rows.Close() + + for _, contentID := range contentIDs { + if err := s.ensureAnchoredIdentityRepairTargetEpisodeLinks(ctx, contentID); err != nil { + return err + } + } + return nil +} + +func (s *MetadataService) ensureAnchoredIdentityRepairTargetEpisodeLinks( + ctx context.Context, + contentID string, +) error { + if err := s.ensureSeriesEpisodeLinks(ctx, contentID); err != nil { + queueErr := s.RequestStaleMetadataRefresh(ctx, RefreshTargetItem, contentID) + slog.WarnContext(ctx, "metadata: anchored repair episode relink failed", "component", "metadata", + "content_id", contentID, + "error", err, + "refresh_queued", queueErr == nil, + ) + if queueErr != nil { + return errors.Join( + fmt.Errorf("ensuring anchored repair episode links for %s: %w", contentID, err), + fmt.Errorf("queueing anchored repair target refresh for %s: %w", contentID, queueErr), + ) + } + } + return nil +} + +func (s *MetadataService) loadAnchoredIdentityRepairCandidateGroups( + ctx context.Context, + folderID int, + pathPrefix string, + afterGroup anchoredIdentityRepairGroupCursor, + pageSize int, +) ([][]anchoredIdentityRepairCandidate, error) { + rows, err := s.dbPool.Query(ctx, ` + WITH scoped_roots AS ( + SELECT DISTINCT mf.observed_root_path + FROM media_files mf + WHERE mf.media_folder_id = $1 + AND mf.missing_since IS NULL + AND COALESCE(mf.observed_root_path, '') <> '' + AND SPLIT_PART(mf.content_group_key, '|', 3) = 'anchor' + AND (mf.file_path = $2 OR mf.file_path LIKE $3 ESCAPE '\') + ), + root_summaries AS ( + SELECT + mf.media_folder_id, + mf.observed_root_path, + MIN(mf.content_id) AS source_content_id, + MIN(mf.group_key_version) AS group_key_version, + MIN(mf.content_group_key) AS content_group_key, + MIN(LOWER(TRIM(mf.base_type))) AS item_type, + COUNT(*)::int AS file_count + FROM media_files mf + JOIN scoped_roots scoped + ON scoped.observed_root_path = mf.observed_root_path + WHERE mf.media_folder_id = $1 + AND mf.missing_since IS NULL + GROUP BY mf.media_folder_id, mf.observed_root_path + HAVING COUNT(*) FILTER ( + WHERE mf.content_id IS NULL OR TRIM(mf.content_id) = '' + ) = 0 + AND COUNT(DISTINCT mf.content_id) = 1 + AND COUNT(DISTINCT (mf.group_key_version, mf.content_group_key)) = 1 + AND COUNT(DISTINCT LOWER(TRIM(mf.base_type))) = 1 + ), + candidate_summaries AS ( + SELECT + summary.media_folder_id, + summary.observed_root_path, + summary.source_content_id, + summary.group_key_version, + summary.content_group_key, + summary.item_type, + summary.file_count + FROM root_summaries summary + JOIN media_items source + ON source.content_id = summary.source_content_id + WHERE summary.item_type IN ('movie', 'series') + AND source.type = summary.item_type + AND LOWER(TRIM(source.status)) = 'matched' + AND SPLIT_PART(summary.content_group_key, '|', 2) = summary.item_type + AND SPLIT_PART(summary.content_group_key, '|', 3) = 'anchor' + AND summary.source_content_id <> + summary.item_type || '-' || SPLIT_PART(summary.content_group_key, '|', 4) + AND ( + SELECT COUNT(*) + FROM media_files source_file + WHERE source_file.content_id = summary.source_content_id + AND source_file.missing_since IS NULL + ) > summary.file_count + AND NOT EXISTS ( + SELECT 1 + FROM media_identity_overrides identity_override + WHERE identity_override.media_folder_id = summary.media_folder_id + AND ( + (identity_override.scope = 'root' AND identity_override.root_path = summary.observed_root_path) OR + (identity_override.scope = 'file' AND EXISTS ( + SELECT 1 + FROM media_files root_file + WHERE root_file.media_folder_id = summary.media_folder_id + AND root_file.observed_root_path = summary.observed_root_path + AND root_file.file_path = identity_override.file_path + AND root_file.missing_since IS NULL + )) + ) + ) + ), + candidate_groups AS ( + SELECT + source_content_id, + group_key_version, + content_group_key, + item_type + FROM candidate_summaries + WHERE ( + source_content_id, + group_key_version, + content_group_key, + item_type + ) > ($4::text, $5::int, $6::text, $7::text) + GROUP BY + source_content_id, + group_key_version, + content_group_key, + item_type + ORDER BY + source_content_id, + group_key_version, + content_group_key, + item_type + LIMIT $8 + ) + SELECT + summary.media_folder_id, + summary.observed_root_path, + summary.source_content_id, + summary.group_key_version, + summary.content_group_key, + summary.item_type, + summary.file_count + FROM candidate_summaries summary + JOIN candidate_groups candidate_group + USING (source_content_id, group_key_version, content_group_key, item_type) + ORDER BY + summary.source_content_id, + summary.group_key_version, + summary.content_group_key, + summary.item_type, + summary.observed_root_path + `, + folderID, + pathPrefix, + pathscope.PrefixLike(pathPrefix), + afterGroup.SourceContentID, + afterGroup.GroupKeyVersion, + afterGroup.ContentGroupKey, + afterGroup.ItemType, + pageSize, + ) + if err != nil { + return nil, fmt.Errorf("querying anchored identity repairs: %w", err) + } + defer rows.Close() + + candidateGroups := make([][]anchoredIdentityRepairCandidate, 0, pageSize) + for rows.Next() { + var candidate anchoredIdentityRepairCandidate + if err := rows.Scan( + &candidate.FolderID, + &candidate.ObservedRootPath, + &candidate.SourceContentID, + &candidate.GroupKeyVersion, + &candidate.ContentGroupKey, + &candidate.ItemType, + &candidate.FileCount, + ); err != nil { + return nil, fmt.Errorf("scanning anchored identity repair: %w", err) + } + if len(candidateGroups) == 0 || + !sameAnchoredIdentityRepairGroup(candidateGroups[len(candidateGroups)-1][0], candidate) { + candidateGroups = append(candidateGroups, nil) + } + lastIndex := len(candidateGroups) - 1 + candidateGroups[lastIndex] = append(candidateGroups[lastIndex], candidate) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating anchored identity repairs: %w", err) + } + return candidateGroups, nil +} + +func anchoredIdentityRepairCursor( + candidate anchoredIdentityRepairCandidate, +) anchoredIdentityRepairGroupCursor { + return anchoredIdentityRepairGroupCursor{ + SourceContentID: candidate.SourceContentID, + GroupKeyVersion: candidate.GroupKeyVersion, + ContentGroupKey: candidate.ContentGroupKey, + ItemType: candidate.ItemType, + } +} + +func sameAnchoredIdentityRepairGroup( + left anchoredIdentityRepairCandidate, + right anchoredIdentityRepairCandidate, +) bool { + return left.FolderID == right.FolderID && + left.SourceContentID == right.SourceContentID && + left.GroupKeyVersion == right.GroupKeyVersion && + left.ContentGroupKey == right.ContentGroupKey && + left.ItemType == right.ItemType +} + +func (s *MetadataService) repairAnchoredIdentityMismatchGroup( + ctx context.Context, + candidates []anchoredIdentityRepairCandidate, +) (bool, error) { + if len(candidates) == 0 { + return false, nil + } + candidates = append([]anchoredIdentityRepairCandidate(nil), candidates...) + sort.Slice(candidates, func(i, j int) bool { + return candidates[i].ObservedRootPath < candidates[j].ObservedRootPath + }) + candidate := candidates[0] + for _, groupedCandidate := range candidates[1:] { + if !sameAnchoredIdentityRepairGroup(candidate, groupedCandidate) { + return false, nil + } + } + + anchor, ok := parseAnchoredGroupIdentity( + candidate.GroupKeyVersion, + candidate.ContentGroupKey, + ) + if !ok || anchor.ItemType != candidate.ItemType || anchor.TargetContent == candidate.SourceContentID { + return false, nil + } + + tx, err := s.dbPool.Begin(ctx) + if err != nil { + return false, fmt.Errorf("beginning anchored identity repair: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + items, err := lockRepairItems(ctx, tx, candidate.SourceContentID, anchor.TargetContent) + if err != nil { + return false, err + } + if len(items) != 2 || + items[candidate.SourceContentID] != candidate.ItemType || + items[anchor.TargetContent] != candidate.ItemType { + return false, nil + } + + activeSourceFiles, err := lockActiveSourceFiles(ctx, tx, candidate.SourceContentID) + if err != nil { + return false, err + } + + fileCapacity := 0 + for _, groupedCandidate := range candidates { + fileCapacity += groupedCandidate.FileCount + } + files := make([]filesplit.File, 0, fileCapacity) + for _, groupedCandidate := range candidates { + rootFiles, lockErr := lockRepairRootFiles( + ctx, + tx, + groupedCandidate.FolderID, + groupedCandidate.ObservedRootPath, + ) + if lockErr != nil { + return false, lockErr + } + if len(rootFiles) == 0 || len(rootFiles) != groupedCandidate.FileCount { + return false, nil + } + for _, file := range rootFiles { + fileAnchor, valid := parseAnchoredGroupIdentity(file.GroupKeyVersion, file.ContentGroupKey) + if !valid || + file.ContentID != groupedCandidate.SourceContentID || + file.GroupKeyVersion != groupedCandidate.GroupKeyVersion || + file.ContentGroupKey != groupedCandidate.ContentGroupKey || + fileAnchor.TargetContent != anchor.TargetContent || + file.BaseType != groupedCandidate.ItemType || + file.ObservedRootPath != groupedCandidate.ObservedRootPath { + return false, nil + } + } + hasOverride, overrideErr := rootHasIdentityOverride( + ctx, + tx, + groupedCandidate.FolderID, + groupedCandidate.ObservedRootPath, + ) + if overrideErr != nil { + return false, overrideErr + } + if hasOverride { + return false, nil + } + files = append(files, rootFiles...) + } + + if activeSourceFiles <= len(files) { + return false, nil + } + + owners, err := matchedProviderOwners(ctx, tx, anchor) + if err != nil { + return false, err + } + if len(owners) != 1 || owners[0] != anchor.TargetContent { + return false, nil + } + + moveResult, err := filesplit.Move(ctx, tx, filesplit.Options{ + FromContentID: candidate.SourceContentID, + ToContentID: anchor.TargetContent, + ItemType: candidate.ItemType, + Files: files, + HistoryMode: reattribute.HistoryModeEvidence, + }) + if err != nil { + return false, fmt.Errorf( + "repairing %d anchored roots from %s to %s: %w", + len(candidates), + candidate.SourceContentID, + anchor.TargetContent, + err, + ) + } + + var sourceStillHasActiveFiles bool + if err := tx.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 + FROM media_files + WHERE content_id = $1 + AND missing_since IS NULL + ) + `, candidate.SourceContentID).Scan(&sourceStillHasActiveFiles); err != nil { + return false, fmt.Errorf("checking anchored repair source remainder: %w", err) + } + if !sourceStillHasActiveFiles { + // The candidate ceased to be a proper subset while this transaction + // waited for locks. Roll back and let whole-item matching handle it. + return false, nil + } + + claimsReconciled, err := reconcileAnchoredRepairClaims( + ctx, + tx, + candidate, + anchor.TargetContent, + files, + ) + if err != nil { + return false, err + } + if !claimsReconciled { + return false, nil + } + if _, err := tx.Exec(ctx, ` + DELETE FROM media_item_libraries membership + WHERE membership.content_id = $1 + AND membership.media_folder_id = $2 + AND NOT EXISTS ( + SELECT 1 + FROM media_files source_file + WHERE source_file.content_id = membership.content_id + AND source_file.media_folder_id = membership.media_folder_id + AND source_file.missing_since IS NULL + ) + `, candidate.SourceContentID, candidate.FolderID); err != nil { + return false, fmt.Errorf("cleaning anchored repair source membership: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return false, fmt.Errorf("committing anchored identity repair: %w", err) + } + + slog.InfoContext(ctx, "metadata: repaired provider-anchored root ownership", + "component", "metadata", + "folder_id", candidate.FolderID, + "root_count", len(candidates), + "observed_root_path", candidate.ObservedRootPath, + "source_content_id", candidate.SourceContentID, + "target_content_id", anchor.TargetContent, + "files_moved", len(files), + "episode_pairs", len(moveResult.EpisodePairs), + "history_moved", moveResult.Reattribution.HistoryMoved, + "history_ambiguous", moveResult.Reattribution.HistoryAmbiguous, + ) + return true, nil +} + +func parseAnchoredGroupIdentity(groupKeyVersion int, groupKey string) (anchoredGroupIdentity, bool) { + parts := strings.Split(strings.TrimSpace(groupKey), "|") + if len(parts) != 4 || parts[2] != "anchor" { + return anchoredGroupIdentity{}, false + } + versionText, found := strings.CutPrefix(parts[0], "v") + if !found { + return anchoredGroupIdentity{}, false + } + version, err := strconv.Atoi(versionText) + if err != nil || version <= 0 || version != groupKeyVersion { + return anchoredGroupIdentity{}, false + } + itemType := strings.TrimSpace(parts[1]) + provider, providerID, found := strings.Cut(strings.TrimSpace(parts[3]), "-") + if !found || (itemType != anchoredItemTypeMovie && itemType != anchoredItemTypeSeries) { + return anchoredGroupIdentity{}, false + } + + ids := contentid.ProviderIDs{} + switch provider { + case contentid.ProviderTMDB: + ids.Tmdb = providerID + case contentid.ProviderIMDB: + ids.Imdb = providerID + case contentid.ProviderTVDB: + ids.Tvdb = providerID + default: + return anchoredGroupIdentity{}, false + } + + var target string + var ok bool + if itemType == anchoredItemTypeSeries { + target, ok = contentid.ForSeries(ids) + } else { + target, ok = contentid.ForMovie(ids) + } + if !ok || target != itemType+"-"+provider+"-"+strings.ToLower(strings.TrimSpace(providerID)) { + return anchoredGroupIdentity{}, false + } + return anchoredGroupIdentity{ + Version: version, + ItemType: itemType, + Provider: provider, + ProviderID: strings.ToLower(strings.TrimSpace(providerID)), + TargetContent: target, + }, true +} + +func lockActiveSourceFiles( + ctx context.Context, + tx pgx.Tx, + sourceContentID string, +) (int, error) { + rows, err := tx.Query(ctx, ` + SELECT id + FROM media_files + WHERE content_id = $1 + AND missing_since IS NULL + ORDER BY id + FOR UPDATE + `, sourceContentID) + if err != nil { + return 0, fmt.Errorf("locking anchored repair source files: %w", err) + } + defer rows.Close() + + count := 0 + for rows.Next() { + var fileID int + if err := rows.Scan(&fileID); err != nil { + return 0, fmt.Errorf("scanning anchored repair source lock: %w", err) + } + count++ + } + if err := rows.Err(); err != nil { + return 0, fmt.Errorf("iterating anchored repair source locks: %w", err) + } + return count, nil +} + +func lockRepairItems(ctx context.Context, tx pgx.Tx, sourceContentID, targetContentID string) (map[string]string, error) { + contentIDs := []string{sourceContentID, targetContentID} + sort.Strings(contentIDs) + rows, err := tx.Query(ctx, ` + SELECT content_id, type + FROM media_items + WHERE content_id = ANY($1::text[]) + AND LOWER(TRIM(status)) = 'matched' + ORDER BY content_id ASC + FOR UPDATE + `, contentIDs) + if err != nil { + return nil, fmt.Errorf("locking anchored repair items: %w", err) + } + defer rows.Close() + + items := make(map[string]string, 2) + for rows.Next() { + var contentID, itemType string + if err := rows.Scan(&contentID, &itemType); err != nil { + return nil, fmt.Errorf("scanning anchored repair item: %w", err) + } + items[contentID] = strings.ToLower(strings.TrimSpace(itemType)) + } + return items, rows.Err() +} + +func lockRepairRootFiles( + ctx context.Context, + tx pgx.Tx, + folderID int, + observedRootPath string, +) ([]filesplit.File, error) { + rows, err := tx.Query(ctx, ` + SELECT + id, + COALESCE(content_id, ''), + media_folder_id, + file_path, + COALESCE(canonical_root_path, ''), + COALESCE(observed_root_path, ''), + COALESCE(group_key_version, 1), + COALESCE(content_group_key, ''), + LOWER(TRIM(COALESCE(base_type, ''))), + COALESCE(season_number, 0), + COALESCE(episode_number, 0), + COALESCE(episode_id, '') + FROM media_files + WHERE media_folder_id = $1 + AND observed_root_path = $2 + AND missing_since IS NULL + ORDER BY id ASC + FOR UPDATE + `, folderID, observedRootPath) + if err != nil { + return nil, fmt.Errorf("locking anchored repair files: %w", err) + } + defer rows.Close() + + files := make([]filesplit.File, 0) + for rows.Next() { + var file filesplit.File + if err := rows.Scan( + &file.ID, + &file.ContentID, + &file.MediaFolderID, + &file.FilePath, + &file.CanonicalRootPath, + &file.ObservedRootPath, + &file.GroupKeyVersion, + &file.ContentGroupKey, + &file.BaseType, + &file.SeasonNumber, + &file.EpisodeNumber, + &file.EpisodeID, + ); err != nil { + return nil, fmt.Errorf("scanning anchored repair file: %w", err) + } + files = append(files, file) + } + return files, rows.Err() +} + +func rootHasIdentityOverride(ctx context.Context, tx pgx.Tx, folderID int, observedRootPath string) (bool, error) { + var hasOverride bool + if err := tx.QueryRow(ctx, ` + SELECT EXISTS ( + SELECT 1 + FROM media_identity_overrides identity_override + WHERE identity_override.media_folder_id = $1 + AND ( + (identity_override.scope = 'root' AND identity_override.root_path = $2) OR + (identity_override.scope = 'file' AND EXISTS ( + SELECT 1 + FROM media_files root_file + WHERE root_file.media_folder_id = $1 + AND root_file.observed_root_path = $2 + AND root_file.file_path = identity_override.file_path + AND root_file.missing_since IS NULL + )) + ) + ) + `, folderID, observedRootPath).Scan(&hasOverride); err != nil { + return false, fmt.Errorf("checking anchored repair identity overrides: %w", err) + } + return hasOverride, nil +} + +func matchedProviderOwners( + ctx context.Context, + tx pgx.Tx, + anchor anchoredGroupIdentity, +) ([]string, error) { + rows, err := tx.Query(ctx, ` + WITH owners AS ( + SELECT provider_id.content_id + FROM media_item_provider_ids provider_id + JOIN media_items item + ON item.content_id = provider_id.content_id + WHERE provider_id.item_type = $1 + AND provider_id.provider = $2 + AND provider_id.provider_id = $3 + AND LOWER(TRIM(item.status)) = 'matched' + UNION + SELECT item.content_id + FROM media_items item + WHERE item.type = $1 + AND LOWER(TRIM(item.status)) = 'matched' + AND ( + ($2 = 'tmdb' AND item.tmdb_id = $3) OR + ($2 = 'imdb' AND LOWER(item.imdb_id) = $3) OR + ($2 = 'tvdb' AND item.tvdb_id = $3) + ) + ) + SELECT content_id + FROM owners + ORDER BY content_id ASC + `, anchor.ItemType, anchor.Provider, anchor.ProviderID) + if err != nil { + return nil, fmt.Errorf("querying matched provider owners: %w", err) + } + defer rows.Close() + + owners := make([]string, 0, 2) + for rows.Next() { + var contentID string + if err := rows.Scan(&contentID); err != nil { + return nil, fmt.Errorf("scanning matched provider owner: %w", err) + } + owners = append(owners, contentID) + } + return owners, rows.Err() +} + +func reconcileAnchoredRepairClaims( + ctx context.Context, + tx pgx.Tx, + candidate anchoredIdentityRepairCandidate, + targetContentID string, + files []filesplit.File, +) (bool, error) { + canonicalRoots := make([]string, 0) + seenRoots := make(map[string]bool) + for _, file := range files { + root := file.CanonicalRootPath + if strings.TrimSpace(root) != "" && !seenRoots[root] { + seenRoots[root] = true + canonicalRoots = append(canonicalRoots, root) + } + } + sort.Strings(canonicalRoots) + + for _, root := range canonicalRoots { + tag, err := tx.Exec(ctx, ` + INSERT INTO media_item_roots ( + media_folder_id, canonical_root_path, content_id, last_seen_at + ) + SELECT $1, $2, $3, NOW() + WHERE NOT EXISTS ( + SELECT 1 + FROM media_files other_file + WHERE other_file.media_folder_id = $1 + AND other_file.canonical_root_path = $2 + AND other_file.missing_since IS NULL + AND other_file.content_id IS DISTINCT FROM $3 + ) + ON CONFLICT (media_folder_id, canonical_root_path) DO UPDATE + SET content_id = EXCLUDED.content_id, + last_seen_at = NOW() + WHERE NOT EXISTS ( + SELECT 1 + FROM media_files other_file + WHERE other_file.media_folder_id = EXCLUDED.media_folder_id + AND other_file.canonical_root_path = EXCLUDED.canonical_root_path + AND other_file.missing_since IS NULL + AND other_file.content_id IS DISTINCT FROM EXCLUDED.content_id + ) + `, candidate.FolderID, root, targetContentID) + if err != nil { + return false, fmt.Errorf("reconciling anchored repair root claim: %w", err) + } + if tag.RowsAffected() != 1 { + return false, nil + } + } + + tag, err := tx.Exec(ctx, ` + INSERT INTO media_item_groups ( + media_folder_id, group_key_version, content_group_key, content_id, last_seen_at + ) + SELECT $1, $2, $3, $4, NOW() + WHERE NOT EXISTS ( + SELECT 1 + FROM media_files other_file + WHERE other_file.media_folder_id = $1 + AND other_file.group_key_version = $2 + AND other_file.content_group_key = $3 + AND other_file.missing_since IS NULL + AND other_file.content_id IS DISTINCT FROM $4 + ) + ON CONFLICT (media_folder_id, group_key_version, content_group_key) DO UPDATE + SET content_id = EXCLUDED.content_id, + last_seen_at = NOW() + WHERE NOT EXISTS ( + SELECT 1 + FROM media_files other_file + WHERE other_file.media_folder_id = EXCLUDED.media_folder_id + AND other_file.group_key_version = EXCLUDED.group_key_version + AND other_file.content_group_key = EXCLUDED.content_group_key + AND other_file.missing_since IS NULL + AND other_file.content_id IS DISTINCT FROM EXCLUDED.content_id + ) + `, candidate.FolderID, candidate.GroupKeyVersion, candidate.ContentGroupKey, targetContentID) + if err != nil { + return false, fmt.Errorf("reconciling anchored repair group claim: %w", err) + } + return tag.RowsAffected() == 1, nil +} diff --git a/internal/metadata/anchored_identity_repair_test.go b/internal/metadata/anchored_identity_repair_test.go new file mode 100644 index 00000000..8816c4ab --- /dev/null +++ b/internal/metadata/anchored_identity_repair_test.go @@ -0,0 +1,542 @@ +package metadata + +import ( + "context" + "fmt" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/Silo-Server/silo-server/internal/contentid" +) + +func TestParseAnchoredGroupIdentity(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + version int + key string + wantOK bool + wantTarget string + wantProvider string + wantProviderID string + }{ + { + name: "series tvdb", + version: 1, + key: "v1|series|anchor|tvdb-415239", + wantOK: true, + wantTarget: "series-tvdb-415239", + wantProvider: contentid.ProviderTVDB, + wantProviderID: "415239", + }, + { + name: "movie imdb normalizes case", + version: 1, + key: "v1|movie|anchor|imdb-tt33763941", + wantOK: true, + wantTarget: "movie-imdb-tt33763941", + wantProvider: contentid.ProviderIMDB, + wantProviderID: "tt33763941", + }, + {name: "version mismatch", version: 2, key: "v1|series|anchor|tvdb-415239"}, + {name: "title group", version: 1, key: "v1|series|diplomat|2023"}, + {name: "unknown provider", version: 1, key: "v1|series|anchor|anidb-123"}, + {name: "invalid numeric id", version: 1, key: "v1|series|anchor|tvdb-nope"}, + {name: "wrong shape", version: 1, key: "v1|series|anchor"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, ok := parseAnchoredGroupIdentity(tt.version, tt.key) + if ok != tt.wantOK { + t.Fatalf("ok = %v, want %v (identity = %+v)", ok, tt.wantOK, got) + } + if !tt.wantOK { + return + } + if got.TargetContent != tt.wantTarget || + got.Provider != tt.wantProvider || + got.ProviderID != tt.wantProviderID { + t.Fatalf( + "identity = %+v, want target=%q provider=%q id=%q", + got, + tt.wantTarget, + tt.wantProvider, + tt.wantProviderID, + ) + } + }) + } +} + +func TestRepairAnchoredIdentityMismatch_EndToEnd(t *testing.T) { + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("SILO_TEST_DATABASE_URL is not set") + } + + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatalf("connect test database: %v", err) + } + t.Cleanup(pool.Close) + + suffix := time.Now().UnixNano() + targetProviderID := fmt.Sprintf("%d", 800_000_000+suffix%90_000_000) + sourceProviderID := fmt.Sprintf("%d", 900_000_000+suffix%90_000_000) + missingProviderID := fmt.Sprintf("%d", 700_000_000+suffix%90_000_000) + targetContentID := "series-tvdb-" + targetProviderID + sourceContentID := "series-tvdb-" + sourceProviderID + targetEpisodeID := "episode-tvdb-" + targetProviderID + "-1-1" + sourceEpisodeID := "episode-tvdb-" + sourceProviderID + "-1-1" + targetEpisodeID2 := "episode-tvdb-" + targetProviderID + "-1-2" + sourceEpisodeID2 := "episode-tvdb-" + sourceProviderID + "-1-2" + rootBase := fmt.Sprintf("/anchored-repair/%d", suffix) + wrongRoot := rootBase + "/4k/The Diplomat (2023) {tvdb-" + targetProviderID + "}" + wrongRoot2 := rootBase + "/5k/The Diplomat (2023) {tvdb-" + targetProviderID + "} " + stayRoot := rootBase + "/20s/The Diplomat (2023) {tvdb-" + sourceProviderID + "}" + conflictingRoot := rootBase + "/00-ineligible/The Diplomat (2023) {tvdb-" + targetProviderID + "}" + anchoredGroupKey := "v1|series|anchor|tvdb-" + targetProviderID + sourceGroupKey := "v1|series|anchor|tvdb-" + sourceProviderID + unownedGroupKey := "v1|series|anchor|tvdb-" + missingProviderID + + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, year, status, tvdb_id) + VALUES + ($1, 'series', 'The Diplomat', 2023, 'matched', $2), + ($3, 'series', 'The Diplomat', 2023, 'matched', $4) + `, targetContentID, targetProviderID, sourceContentID, sourceProviderID); err != nil { + t.Fatalf("seed items: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1::text[])`, []string{sourceContentID, targetContentID}) + }) + + if _, err := pool.Exec(ctx, ` + INSERT INTO media_item_provider_ids (content_id, item_type, provider, provider_id) + VALUES + ($1, 'series', 'tvdb', $2), + ($3, 'series', 'tvdb', $4) + `, targetContentID, targetProviderID, sourceContentID, sourceProviderID); err != nil { + t.Fatalf("seed provider ids: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO episodes (content_id, series_id, season_number, episode_number, title) + VALUES + ($1, $2, 1, 1, 'Target Episode'), + ($3, $4, 1, 1, 'Source Episode'), + ($5, $4, 1, 2, 'Source Episode 2') + `, targetEpisodeID, targetContentID, sourceEpisodeID, sourceContentID, sourceEpisodeID2); err != nil { + t.Fatalf("seed episodes: %v", err) + } + + var folderID, retainedFolderID int + if err := pool.QueryRow(ctx, ` + INSERT INTO media_folders (type, name, enabled) + VALUES ('series', $1, true) + RETURNING id + `, fmt.Sprintf("anchored-repair-%d", suffix)).Scan(&folderID); err != nil { + t.Fatalf("seed folder: %v", err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO media_folders (type, name, enabled) + VALUES ('series', $1, true) + RETURNING id + `, fmt.Sprintf("anchored-repair-retained-%d", suffix)).Scan(&retainedFolderID); err != nil { + t.Fatalf("seed retained folder: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec( + ctx, + `DELETE FROM media_folders WHERE id = ANY($1::int[])`, + []int{folderID, retainedFolderID}, + ) + }) + + if _, err := pool.Exec(ctx, ` + INSERT INTO media_item_libraries (content_id, media_folder_id) + VALUES ($1, $3), ($2, $3), ($2, $4) + `, targetContentID, sourceContentID, folderID, retainedFolderID); err != nil { + t.Fatalf("seed memberships: %v", err) + } + + var wrongFileID, wrongFileID2, wrongFileID3, stayFileID, conflictingFileID int + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files ( + content_id, episode_id, media_folder_id, file_path, file_size, + canonical_root_path, observed_root_path, group_key_version, + content_group_key, base_type, season_number, episode_number + ) + VALUES ($1, $2, $3, $4, 1000, $5, $5, 1, $6, 'series', 1, 1) + RETURNING id + `, sourceContentID, sourceEpisodeID, folderID, wrongRoot+"/S01E01.mkv", wrongRoot, anchoredGroupKey).Scan(&wrongFileID); err != nil { + t.Fatalf("seed wrong file: %v", err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files ( + content_id, episode_id, media_folder_id, file_path, file_size, + canonical_root_path, observed_root_path, group_key_version, + content_group_key, base_type, season_number, episode_number + ) + VALUES ($1, $2, $3, $4, 1000, $5, $5, 1, $6, 'series', 1, 2) + RETURNING id + `, sourceContentID, sourceEpisodeID2, folderID, wrongRoot+"/S01E02.mkv", wrongRoot, anchoredGroupKey).Scan(&wrongFileID2); err != nil { + t.Fatalf("seed second wrong file: %v", err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files ( + content_id, episode_id, media_folder_id, file_path, file_size, + canonical_root_path, observed_root_path, group_key_version, + content_group_key, base_type, season_number, episode_number + ) + VALUES ($1, $2, $3, $4, 1000, $5, $5, 1, $6, 'series', 1, 1) + RETURNING id + `, sourceContentID, sourceEpisodeID, folderID, wrongRoot2+"/S01E01.mkv", wrongRoot2, anchoredGroupKey).Scan(&wrongFileID3); err != nil { + t.Fatalf("seed third wrong file in repeated anchored group: %v", err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files ( + content_id, episode_id, media_folder_id, file_path, file_size, + canonical_root_path, observed_root_path, group_key_version, + content_group_key, base_type, season_number, episode_number + ) + VALUES ($1, $2, $3, $4, 1000, $5, $5, 1, $6, 'series', 1, 1) + RETURNING id + `, sourceContentID, sourceEpisodeID, retainedFolderID, stayRoot+"/S01E01.mkv", stayRoot, sourceGroupKey).Scan(&stayFileID); err != nil { + t.Fatalf("seed source file: %v", err) + } + if err := pool.QueryRow(ctx, ` + INSERT INTO media_files ( + content_id, episode_id, media_folder_id, file_path, file_size, + canonical_root_path, observed_root_path, group_key_version, + content_group_key, base_type, season_number, episode_number + ) + VALUES ($1, NULL, $2, $3, 1000, $4, $4, 1, $5, 'series', 1, 1) + RETURNING id + `, sourceContentID, folderID, conflictingRoot+"/S01E01.mkv", conflictingRoot, anchoredGroupKey).Scan(&conflictingFileID); err != nil { + t.Fatalf("seed conflicting anchored file: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO episode_libraries (episode_id, media_folder_id) + VALUES ($1, $3), ($1, $4), ($2, $3) + `, sourceEpisodeID, sourceEpisodeID2, folderID, retainedFolderID); err != nil { + t.Fatalf("seed source episode memberships: %v", err) + } + + var userID int + if err := pool.QueryRow(ctx, ` + INSERT INTO users (username, role) + VALUES ($1, 'user') + RETURNING id + `, fmt.Sprintf("anchored-repair-user-%d", suffix)).Scan(&userID); err != nil { + t.Fatalf("seed user: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, userID) + }) + profileID := fmt.Sprintf("00000000-0000-4000-8000-%012d", suffix%1_000_000_000_000) + if _, err := pool.Exec(ctx, ` + INSERT INTO user_watch_progress ( + user_id, profile_id, media_item_id, position_seconds, + duration_seconds, last_file_id + ) + VALUES + ($1, $2, $3, 120, 3600, $4), + ($1, $2, $5, 240, 3600, $6) + `, userID, profileID, sourceEpisodeID, stayFileID, sourceEpisodeID2, wrongFileID2); err != nil { + t.Fatalf("seed source episode progress: %v", err) + } + + if _, err := pool.Exec(ctx, ` + INSERT INTO media_item_roots (media_folder_id, canonical_root_path, content_id) + VALUES ($1, $2, $4), ($1, $3, $4) + `, folderID, wrongRoot, wrongRoot2, sourceContentID); err != nil { + t.Fatalf("seed root claim: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO media_item_groups ( + media_folder_id, group_key_version, content_group_key, content_id + ) + VALUES ($1, 1, $2, $3) + `, folderID, anchoredGroupKey, sourceContentID); err != nil { + t.Fatalf("seed group claim: %v", err) + } + + service := &MetadataService{dbPool: pool} + followUpCalls := 0 + service.hooks.ensureSeriesEpisodeLinks = func(ctx context.Context, seriesID string) error { + followUpCalls++ + if seriesID != targetContentID { + return fmt.Errorf("unexpected episode follow-up target %q", seriesID) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO episodes (content_id, series_id, season_number, episode_number, title) + VALUES ($1, $2, 1, 2, 'Target Episode 2') + ON CONFLICT (content_id) DO NOTHING + `, targetEpisodeID2, targetContentID); err != nil { + return fmt.Errorf("seeding missing target episode in follow-up: %w", err) + } + if _, err := pool.Exec(ctx, ` + UPDATE media_files + SET episode_id = $1, + updated_at = NOW() + WHERE content_id = $2 + AND season_number = 1 + AND episode_number = 2 + AND episode_id IS NULL + `, targetEpisodeID2, targetContentID); err != nil { + return fmt.Errorf("relinking missing target episode in follow-up: %w", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO episode_libraries (episode_id, media_folder_id, first_seen_at) + SELECT episode_id, media_folder_id, MIN(created_at) + FROM media_files + WHERE content_id = $1 + AND episode_id = $2 + GROUP BY episode_id, media_folder_id + ON CONFLICT (episode_id, media_folder_id) DO NOTHING + `, targetContentID, targetEpisodeID2); err != nil { + return fmt.Errorf("adding target episode membership in follow-up: %w", err) + } + return nil + } + var gotContentID, groupOwner string + repaired, err := service.repairAnchoredIdentityMismatchesByFolderAndPathPrefix(ctx, folderID, rootBase+"/4k") + if err != nil { + t.Fatalf("repair with conflicting claim member: %v", err) + } + if repaired != 0 { + t.Fatalf("repair with conflicting claim member repaired %d roots, want 0", repaired) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id + FROM media_files + WHERE id = $1 + `, wrongFileID).Scan(&gotContentID); err != nil { + t.Fatalf("load rolled-back file: %v", err) + } + if gotContentID != sourceContentID { + t.Fatalf("rolled-back file content_id = %q, want %q", gotContentID, sourceContentID) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id + FROM media_item_groups + WHERE media_folder_id = $1 + AND group_key_version = 1 + AND content_group_key = $2 + `, folderID, anchoredGroupKey).Scan(&groupOwner); err != nil { + t.Fatalf("load retained conflicting group claim: %v", err) + } + if groupOwner != sourceContentID { + t.Fatalf("conflicting group claim owner = %q, want %q", groupOwner, sourceContentID) + } + if _, err := pool.Exec(ctx, ` + UPDATE media_files + SET content_group_key = $2 + WHERE id = $1 + `, conflictingFileID, unownedGroupKey); err != nil { + t.Fatalf("make leading candidate safely ineligible: %v", err) + } + + repaired, err = service.repairAnchoredIdentityMismatchesByFolderAndPathPrefixPageSize( + ctx, + folderID, + rootBase, + 1, + ) + if err != nil { + t.Fatalf("repair after skipped first page: %v", err) + } + if repaired != 2 { + t.Fatalf("repaired roots after skipped first page = %d, want 2", repaired) + } + if followUpCalls != 1 { + t.Fatalf("episode repair follow-up calls = %d, want 1", followUpCalls) + } + + var gotEpisodeID string + if err := pool.QueryRow(ctx, ` + SELECT content_id, episode_id + FROM media_files + WHERE id = $1 + `, wrongFileID).Scan(&gotContentID, &gotEpisodeID); err != nil { + t.Fatalf("load repaired file: %v", err) + } + if gotContentID != targetContentID || gotEpisodeID != targetEpisodeID { + t.Fatalf( + "repaired file = (%q, %q), want (%q, %q)", + gotContentID, + gotEpisodeID, + targetContentID, + targetEpisodeID, + ) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id, episode_id + FROM media_files + WHERE id = $1 + `, wrongFileID2).Scan(&gotContentID, &gotEpisodeID); err != nil { + t.Fatalf("load second repaired file: %v", err) + } + if gotContentID != targetContentID || gotEpisodeID != targetEpisodeID2 { + t.Fatalf( + "second repaired file = (%q, %q), want (%q, %q)", + gotContentID, + gotEpisodeID, + targetContentID, + targetEpisodeID2, + ) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id, episode_id + FROM media_files + WHERE id = $1 + `, wrongFileID3).Scan(&gotContentID, &gotEpisodeID); err != nil { + t.Fatalf("load repeated-group repaired file: %v", err) + } + if gotContentID != targetContentID || gotEpisodeID != targetEpisodeID { + t.Fatalf( + "repeated-group repaired file = (%q, %q), want (%q, %q)", + gotContentID, + gotEpisodeID, + targetContentID, + targetEpisodeID, + ) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id + FROM media_files + WHERE id = $1 + `, stayFileID).Scan(&gotContentID); err != nil { + t.Fatalf("load retained file: %v", err) + } + if gotContentID != sourceContentID { + t.Fatalf("retained file content_id = %q, want %q", gotContentID, sourceContentID) + } + rows, err := pool.Query(ctx, ` + SELECT media_item_id + FROM user_watch_progress + WHERE user_id = $1 AND profile_id = $2 + ORDER BY media_item_id + `, userID, profileID) + if err != nil { + t.Fatalf("load source episode progress: %v", err) + } + defer rows.Close() + progressContentIDs := make([]string, 0, 2) + for rows.Next() { + var progressContentID string + if err := rows.Scan(&progressContentID); err != nil { + t.Fatalf("scan source episode progress: %v", err) + } + progressContentIDs = append(progressContentIDs, progressContentID) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate source episode progress: %v", err) + } + wantProgressContentIDs := []string{targetEpisodeID2, sourceEpisodeID} + if len(progressContentIDs) != len(wantProgressContentIDs) || + progressContentIDs[0] != wantProgressContentIDs[0] || + progressContentIDs[1] != wantProgressContentIDs[1] { + t.Fatalf( + "episode progress ids = %v, want %v (shared source stays; fully moved episode follows)", + progressContentIDs, + wantProgressContentIDs, + ) + } + + var rootOwner, rootOwner2 string + if err := pool.QueryRow(ctx, ` + SELECT content_id + FROM media_item_roots + WHERE media_folder_id = $1 AND canonical_root_path = $2 + `, folderID, wrongRoot).Scan(&rootOwner); err != nil { + t.Fatalf("load repaired root claim: %v", err) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id + FROM media_item_roots + WHERE media_folder_id = $1 AND canonical_root_path = $2 + `, folderID, wrongRoot2).Scan(&rootOwner2); err != nil { + t.Fatalf("load repeated-group repaired root claim: %v", err) + } + if err := pool.QueryRow(ctx, ` + SELECT content_id + FROM media_item_groups + WHERE media_folder_id = $1 + AND group_key_version = 1 + AND content_group_key = $2 + `, folderID, anchoredGroupKey).Scan(&groupOwner); err != nil { + t.Fatalf("load repaired group claim: %v", err) + } + if rootOwner != targetContentID || rootOwner2 != targetContentID || groupOwner != targetContentID { + t.Fatalf( + "claim owners = (%q, %q, %q), want %q", + rootOwner, + rootOwner2, + groupOwner, + targetContentID, + ) + } + + rows, err = pool.Query(ctx, ` + SELECT episode_id + FROM episode_libraries + WHERE media_folder_id = $1 + AND episode_id = ANY($2::text[]) + ORDER BY episode_id + `, folderID, []string{sourceEpisodeID, sourceEpisodeID2, targetEpisodeID, targetEpisodeID2}) + if err != nil { + t.Fatalf("load repaired episode memberships: %v", err) + } + defer rows.Close() + gotEpisodeMemberships := make([]string, 0, 2) + for rows.Next() { + var episodeID string + if err := rows.Scan(&episodeID); err != nil { + t.Fatalf("scan repaired episode membership: %v", err) + } + gotEpisodeMemberships = append(gotEpisodeMemberships, episodeID) + } + if err := rows.Err(); err != nil { + t.Fatalf("iterate repaired episode memberships: %v", err) + } + wantEpisodeMemberships := []string{targetEpisodeID, targetEpisodeID2} + if len(gotEpisodeMemberships) != len(wantEpisodeMemberships) { + t.Fatalf("episode memberships = %v, want %v", gotEpisodeMemberships, wantEpisodeMemberships) + } + for index := range wantEpisodeMemberships { + if gotEpisodeMemberships[index] != wantEpisodeMemberships[index] { + t.Fatalf("episode memberships = %v, want %v", gotEpisodeMemberships, wantEpisodeMemberships) + } + } + var retainedSourceMemberships int + if err := pool.QueryRow(ctx, ` + SELECT COUNT(*) + FROM episode_libraries + WHERE media_folder_id = $1 + AND episode_id = $2 + `, retainedFolderID, sourceEpisodeID).Scan(&retainedSourceMemberships); err != nil { + t.Fatalf("load retained source episode membership: %v", err) + } + if retainedSourceMemberships != 1 { + t.Fatalf("retained source episode memberships = %d, want 1", retainedSourceMemberships) + } + + repaired, err = service.repairAnchoredIdentityMismatchesByFolderAndPathPrefix(ctx, folderID, rootBase+"/4k") + if err != nil { + t.Fatalf("repeat anchored identity repair: %v", err) + } + if repaired != 0 { + t.Fatalf("repeat repaired roots = %d, want idempotent 0", repaired) + } + if followUpCalls != 1 { + t.Fatalf("repeat episode repair follow-up calls = %d, want unchanged 1", followUpCalls) + } +} diff --git a/internal/metadata/worker.go b/internal/metadata/worker.go index e308d350..3322cf94 100644 --- a/internal/metadata/worker.go +++ b/internal/metadata/worker.go @@ -1324,6 +1324,19 @@ func (w *MatchWorker) RetryUnmatchedItemsByFolderAndPathPrefix(ctx context.Conte ) } + repaired, repairErr := w.service.repairAnchoredIdentityMismatchesByFolderAndPathPrefix(ctx, folderID, pathPrefix) + retried += repaired + if repairErr != nil { + return retried, stillUnmatched, repairErr + } + if repaired > 0 { + slog.InfoContext(ctx, "metadata: repaired provider-anchored roots in scope", "component", "metadata", + "folder_id", folderID, + "path_prefix", pathPrefix, + "repaired_roots", repaired, + ) + } + return retried, stillUnmatched, nil }