fix(scanner): repair historical provider-anchored merges
This commit is contained in:
Quick104
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user