haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(metadata): prevent stale cross-provider IDs (#480)

Browse Source 
* fix(metadata): prevent stale cross-provider IDs

* fix(metadata): address stale ID review findings

* fix(migrations): build the stale-ID primary key concurrently

ALTER TABLE ... ADD PRIMARY KEY builds the index under ACCESS EXCLUSIVE,
blocking reads and writes on stale_media_ids for the whole build. Create the
wider unique index with CREATE UNIQUE INDEX CONCURRENTLY and attach it with
ADD CONSTRAINT ... PRIMARY KEY USING INDEX instead; all three key columns are
already NOT NULL, so the attach is metadata-only. Same treatment on the
rollback path, plus the repo's INVALID-remnant cleanup so a failed concurrent
build is not silently accepted by IF NOT EXISTS.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Quick <31828688+Quick104@users.noreply.github.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Suspense
2026-07-26 11:19:33 -04:00
committed by GitHub
co-authored by Quick Claude Opus 5
parent 02203d9e40
commit 99d205676f
19 changed files with 1583 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/adminjob/library_refresh.go
+28
View File
@@ -118,11 +118,39 @@ func (l *PGLibraryRefreshItemLister) ListLibraryItems(ctx context.Context, libra
COALESCE(mi.tvdb_id, '') <> ''
OR COALESCE(mi.imdb_id, '') <> ''
)
AND NOT EXISTS (
SELECT 1
FROM stale_media_ids rejected_tmdb
WHERE rejected_tmdb.content_id = mi.content_id
AND LOWER(TRIM(rejected_tmdb.provider)) = 'tmdb'
AND TRIM(rejected_tmdb.provider_id) <> ''
)
)
OR EXISTS (
SELECT 1
FROM stale_media_ids smi
WHERE smi.content_id = mi.content_id
AND LOWER(TRIM(smi.provider)) IN ('tmdb', 'tvdb', 'imdb')
AND TRIM(smi.provider_id) <> ''
AND (
LOWER(TRIM(COALESCE(mi.status, ''))) <> 'matched'
OR (
LOWER(TRIM(smi.provider)) = 'tmdb'
AND TRIM(smi.provider_id) = TRIM(COALESCE(mi.tmdb_id, ''))
)
OR (
LOWER(TRIM(smi.provider)) = 'tvdb'
AND TRIM(smi.provider_id) = TRIM(COALESCE(mi.tvdb_id, ''))
)
OR (
LOWER(TRIM(smi.provider)) = 'imdb'
AND LOWER(TRIM(smi.provider_id)) = LOWER(TRIM(COALESCE(mi.imdb_id, '')))
)
OR LOWER(TRIM(mi.content_id)) =
'movie-' || LOWER(TRIM(smi.provider)) || '-' || LOWER(TRIM(smi.provider_id))
OR LOWER(TRIM(mi.content_id)) =
'series-' || LOWER(TRIM(smi.provider)) || '-' || LOWER(TRIM(smi.provider_id))
)
)
OR (
COALESCE(mi.default_metadata_language, '') <> ''
internal/adminjob/library_refresh_language_test.go
+99
View File
@@ -91,3 +91,102 @@ func TestQuickRefreshListsLanguageMismatchedItems(t *testing.T) {
t.Errorf("quick refresh must not include complete item whose stamped language matches the library language")
}
}
func TestQuickRefreshIgnoresHistoricalSecondaryStaleIDs(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 := fmt.Sprintf("%d", time.Now().UnixNano())
historicalID := "movie-tmdb-" + suffix
actionableProviderID := suffix + "1"
actionableID := "movie-tmdb-" + actionableProviderID
rejectedSiblingID := "series-tvdb-" + suffix
var folderID int
if err := pool.QueryRow(ctx, `
INSERT INTO media_folders (type, name, enabled, metadata_language)
VALUES ('movies', 'Stale Refresh Test', true, 'en')
RETURNING id
`).Scan(&folderID); err != nil {
t.Fatalf("seed folder: %v", err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = ANY($1)`, []string{historicalID, actionableID, rejectedSiblingID})
_, _ = pool.Exec(ctx, `DELETE FROM media_folders WHERE id = $1`, folderID)
})
for _, row := range []struct {
contentID, tmdbID, staleProvider, staleProviderID string
}{
{historicalID, suffix, "tvdb", "999999"},
{actionableID, actionableProviderID, "tmdb", actionableProviderID},
} {
if _, err := pool.Exec(ctx, `
INSERT INTO media_items (
content_id, type, title, status, genres, tmdb_id,
default_metadata_language, overview, poster_path, backdrop_path,
last_refreshed, refresh_failures, episode_metadata_incomplete
) VALUES ($1, 'movie', 'Complete Item', 'matched', '{}'::text[], $2,
'en', 'An overview', '/p.jpg', '/b.jpg', NOW(), 0, FALSE)
`, row.contentID, row.tmdbID); err != nil {
t.Fatalf("seed media item %s: %v", row.contentID, err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO media_item_libraries (content_id, media_folder_id) VALUES ($1, $2)
`, row.contentID, folderID); err != nil {
t.Fatalf("link media item %s: %v", row.contentID, err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO stale_media_ids (content_id, provider, provider_id)
VALUES ($1, $2, $3)
`, row.contentID, row.staleProvider, row.staleProviderID); err != nil {
t.Fatalf("seed stale id for %s: %v", row.contentID, err)
}
}
if _, err := pool.Exec(ctx, `
INSERT INTO media_items (
content_id, type, title, status, genres, tvdb_id,
default_metadata_language, overview, poster_path, backdrop_path,
last_refreshed, refresh_failures, episode_metadata_incomplete
) VALUES ($1, 'series', 'Complete TVDB Item', 'matched', '{}'::text[], $2,
'en', 'An overview', '/p.jpg', '/b.jpg', NOW(), 0, FALSE)
`, rejectedSiblingID, suffix); err != nil {
t.Fatalf("seed TVDB-only media item: %v", err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO media_item_libraries (content_id, media_folder_id) VALUES ($1, $2)
`, rejectedSiblingID, folderID); err != nil {
t.Fatalf("link TVDB-only media item: %v", err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO stale_media_ids (content_id, provider, provider_id)
VALUES ($1, 'tmdb', $2)
`, rejectedSiblingID, suffix+"2"); err != nil {
t.Fatalf("seed rejected TMDB sibling: %v", err)
}
items, err := NewPGLibraryRefreshItemLister(pool).ListLibraryItems(ctx, folderID, LibraryRefreshModeQuick)
if err != nil {
t.Fatalf("ListLibraryItems: %v", err)
}
listed := make(map[string]bool, len(items))
for _, item := range items {
listed[item.ContentID] = true
}
if listed[historicalID] {
t.Fatal("historical secondary stale ID must not keep a complete item in quick refresh")
}
if !listed[actionableID] {
t.Fatal("current canonical stale ID must remain in quick refresh")
}
if listed[rejectedSiblingID] {
t.Fatal("a rejected TMDB sibling must not keep provider-ID enrichment debt active")
}
}
internal/api/handlers/libraries.go
+22 -4
View File
@@ -2340,13 +2340,18 @@ func (h *LibraryHandler) HandleListStaleIDs(w http.ResponseWriter, r *http.Reque
Title string
Year int
ContentType string
Status string
TmdbID string
TvdbID string
ImdbID string
LibraryID int
LibraryName string
}
items := make(map[string]itemInfo, len(contentIDs))
rows, err := h.pool.Query(r.Context(), `
SELECT mi.content_id, mi.title, mi.year, mi.type,
SELECT mi.content_id, mi.title, mi.year, mi.type, COALESCE(mi.status, ''),
COALESCE(mi.tmdb_id, ''), COALESCE(mi.tvdb_id, ''), COALESCE(mi.imdb_id, ''),
COALESCE(mf_lib.folder_id, 0),
COALESCE(mf_lib.folder_name, '')
FROM media_items mi
@@ -2367,18 +2372,31 @@ func (h *LibraryHandler) HandleListStaleIDs(w http.ResponseWriter, r *http.Reque
defer rows.Close()
for rows.Next() {
var cid, title, ctype, libName string
var cid, title, ctype, status, tmdbID, tvdbID, imdbID, libName string
var year, libID int
if err := rows.Scan(&cid, &title, &year, &ctype, &libID, &libName); err != nil {
if err := rows.Scan(&cid, &title, &year, &ctype, &status, &tmdbID, &tvdbID, &imdbID, &libID, &libName); err != nil {
slog.ErrorContext(r.Context(), "scanning item for stale IDs", "component", "api", "error", err)
continue
}
items[cid] = itemInfo{Title: title, Year: year, ContentType: ctype, LibraryID: libID, LibraryName: libName}
items[cid] = itemInfo{
Title: title, Year: year, ContentType: ctype, Status: status,
TmdbID: tmdbID, TvdbID: tvdbID, ImdbID: imdbID,
LibraryID: libID, LibraryName: libName,
}
}
resp := make([]staleMediaIDResponse, 0, len(staleIDs))
for _, s := range staleIDs {
info := items[s.ContentID]
if !metadata.IsActionableStaleProviderID(&models.MediaItem{
ContentID: s.ContentID,
Status: info.Status,
TmdbID: info.TmdbID,
TvdbID: info.TvdbID,
ImdbID: info.ImdbID,
}, s) {
continue
}
resp = append(resp, staleMediaIDResponse{
ContentID: s.ContentID,
LibraryID: info.LibraryID,
internal/contentid/contentid.go
+7
View File
@@ -284,6 +284,13 @@ func IsProviderAnchored(contentID string) bool {
return ok
}
// ProviderAnchor returns the canonical provider and provider ID embedded in a
// provider-derived content ID. It accepts movie, series, season, and episode
// IDs; child IDs return their parent series anchor.
func ProviderAnchor(contentID string) (provider, providerID string, ok bool) {
return parseAnchored(strings.TrimSpace(contentID))
}
// parseAnchored validates a provider-anchored content_id against the exact
// per-kind arity and component shape, returning the canonical (provider,
// seriesId-or-id) anchor when it is well-formed:
internal/contentid/contentid_test.go
+24
View File
@@ -177,3 +177,27 @@ func TestIsProviderAnchored(t *testing.T) {
}
}
}
func TestProviderAnchor(t *testing.T) {
tests := []struct {
contentID string
provider string
providerID string
ok bool
}{
{contentID: "movie-tmdb-603", provider: "tmdb", providerID: "603", ok: true},
{contentID: "series-tvdb-405851", provider: "tvdb", providerID: "405851", ok: true},
{contentID: "episode-tvdb-405851-1-2", provider: "tvdb", providerID: "405851", ok: true},
{contentID: "local-deadbeef"},
{contentID: "series-tmdb-not-a-number"},
}
for _, tt := range tests {
t.Run(tt.contentID, func(t *testing.T) {
provider, providerID, ok := ProviderAnchor(tt.contentID)
if provider != tt.provider || providerID != tt.providerID || ok != tt.ok {
t.Fatalf("ProviderAnchor(%q) = (%q, %q, %v), want (%q, %q, %v)",
tt.contentID, provider, providerID, ok, tt.provider, tt.providerID, tt.ok)
}
})
}
}
internal/metadata/id_consensus_test.go
+8 -8
View File
@@ -32,7 +32,7 @@ func (p *idConsensusPipelineProvider) Search(context.Context, SearchQuery) ([]Se
}
func (p *idConsensusPipelineProvider) GetMetadata(_ context.Context, req MetadataRequest) (*MetadataResult, error) {
if req.ProviderIDs["tvdb"] != "" {
if req.ProviderIDs["tvdb"] != "352440" {
return nil, errUnexpectedQuarantinedProviderID
}
return &MetadataResult{
@@ -40,7 +40,7 @@ func (p *idConsensusPipelineProvider) GetMetadata(_ context.Context, req Metadat
Title: "A Teacher",
Year: 2020,
// A provider detail response may repeat its stale cross-reference. The
// pipeline must not let it undo the search-phase quarantine.
// pipeline must not let it replace the native provider's confirmed ID.
ProviderIDs: map[string]string{
"imdb": "tt10680614", "tmdb": "103992", "tvdb": "473725",
},
@@ -55,7 +55,7 @@ func (*unexpectedQuarantinedProviderIDError) Error() string {
return "quarantined provider ID reached metadata phase"
}
func TestInitialMatchPipelineQuarantinesConflictingConsensusID(t *testing.T) {
func TestInitialMatchPipelineResolvesConflictingConsensusIDFromOwningProvider(t *testing.T) {
t.Parallel()
harness := newTestHarness()
provider := &idConsensusPipelineProvider{}
@@ -77,12 +77,12 @@ func TestInitialMatchPipelineQuarantinesConflictingConsensusID(t *testing.T) {
if item.ImdbID != "tt10680614" || item.TmdbID != "103992" {
t.Fatalf("persisted consensus IDs = imdb:%q tmdb:%q", item.ImdbID, item.TmdbID)
}
if item.TvdbID != "" {
t.Fatalf("conflicting TVDB ID persisted as %q", item.TvdbID)
if item.TvdbID != "352440" {
t.Fatalf("native TVDB ID = %q, want 352440", item.TvdbID)
}
}
func TestRefreshPipelinesQuarantineStoredConflictingConsensusID(t *testing.T) {
func TestRefreshPipelinesReplaceStoredCrossReferenceWithOwningProviderID(t *testing.T) {
for name, mode := range map[string]RefreshMode{"scheduled": ModeScheduledRefresh, "manual": ModeManualRefresh} {
mode := mode
t.Run(name, func(t *testing.T) {
@@ -112,8 +112,8 @@ func TestRefreshPipelinesQuarantineStoredConflictingConsensusID(t *testing.T) {
if err != nil {
t.Fatalf("load refreshed item: %v", err)
}
if item.ImdbID != "tt10680614" || item.TmdbID != "103992" || item.TvdbID != "" {
t.Fatalf("refreshed IDs = imdb:%q tmdb:%q tvdb:%q, want disputed TVDB ID quarantined", item.ImdbID, item.TmdbID, item.TvdbID)
if item.ImdbID != "tt10680614" || item.TmdbID != "103992" || item.TvdbID != "352440" {
t.Fatalf("refreshed IDs = imdb:%q tmdb:%q tvdb:%q, want native TVDB ID 352440", item.ImdbID, item.TmdbID, item.TvdbID)
}
})
}
internal/metadata/identify_stale_id_test.go
+70 -5
View File
@@ -38,8 +38,8 @@ func (p *notFoundMetadataProvider) GetMetadata(_ context.Context, req MetadataRe
// issue #268: an item has a durable tmdb ID already recorded as stale; the
// admin rematches it to a different provider via the Apply Match flow
// (ModeIdentify). The stale tmdb ID must not be re-injected into the identify
// request, and after the successful rematch the item's stale rows must be
// cleared rather than re-recorded with a fresh last_seen.
// request. The rejected value remains a non-actionable negative-cache entry so
// a later detail response cannot resurrect it.
func TestProcess_IdentifyDoesNotResurrectRecordedStaleProviderID(t *testing.T) {
h := newTestHarness()
ctx := context.Background()
@@ -80,7 +80,9 @@ func TestProcess_IdentifyDoesNotResurrectRecordedStaleProviderID(t *testing.T) {
response: &MetadataResult{
HasMetadata: true,
Title: "Formula 1: Drive to Survive",
ProviderIDs: map[string]string{"tvdb": "417585"},
// A detail response may repeat the known-dead cross-reference; it
// must not undo the stale suppression while bootstrapping new IDs.
ProviderIDs: map[string]string{testTVDBProvider: "417585", testTMDBProvider: "324880"},
},
}
@@ -104,13 +106,62 @@ func TestProcess_IdentifyDoesNotResurrectRecordedStaleProviderID(t *testing.T) {
if got := req.ProviderIDs["tvdb"]; got != "417585" {
t.Errorf("identify request tvdb id = %q, want 417585", got)
}
providerRepo.mu.Lock()
persisted := providerRepo.lastReplace["existing-1"]
providerRepo.mu.Unlock()
if persisted["tmdb"] != "" {
t.Errorf("known-dead detail cross-reference was restored: %#v", persisted)
}
stale, err := staleRepo.GetByContentID(ctx, "existing-1")
if err != nil {
t.Fatalf("get stale rows: %v", err)
}
if len(stale) != 0 {
t.Errorf("stale rows after successful rematch = %v, want none", stale)
if len(stale) != 1 || stale[0].Provider != "tmdb" || stale[0].ProviderID != "324880" {
t.Errorf("stale rows after successful rematch = %#v, want retained tmdb=324880 negative cache", stale)
}
}
func TestProcess_IdentifyPreservesProviderAnchoredContentID(t *testing.T) {
const (
contentID = "movie-tmdb-777"
correctedTitle = "Corrected Movie"
)
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: contentID, Type: "movie", Title: correctedTitle, Year: 2020,
Status: "matched", TmdbID: "777",
Studios: []string{}, Networks: []string{}, Countries: []string{}, Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
providerRepo := newFakeProviderIDRepo()
providerRepo.set(contentID, &models.MediaItemProviderID{
ContentID: contentID, ItemType: "movie", Provider: "tmdb", ProviderID: "777",
})
h.service.providerIDRepo = providerRepo
staleRepo := newFakeStaleIDRepo()
staleRepo.set(contentID, &models.StaleMediaID{
ContentID: contentID, Provider: "tmdb", ProviderID: "777",
})
h.service.staleIDRepo = staleRepo
provider := &capturingMetadataProvider{response: &MetadataResult{
HasMetadata: true, Title: correctedTitle,
ProviderIDs: map[string]string{testIMDBProvider: "tt0000777"},
}}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: contentID,
ProviderIDs: map[string]string{testIMDBProvider: "tt0000777"},
Language: "en",
Mode: ModeIdentify,
}, []Provider{provider})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || result.ContentID != contentID {
t.Fatalf("result content id = %#v, want %s preserved", result, contentID)
}
}
@@ -242,4 +293,18 @@ func TestProcess_IdentifyKeepsUserSuppliedIDEvenIfRecordedStale(t *testing.T) {
if got := req.ProviderIDs["tmdb"]; got != "324880" {
t.Errorf("identify request tmdb id = %q, want 324880 (user-supplied id must survive)", got)
}
item, err := h.itemRepo.GetByID(ctx, "existing-1")
if err != nil {
t.Fatalf("load identified item: %v", err)
}
if item.TmdbID != "324880" {
t.Errorf("persisted tmdb id = %q, want explicitly retried 324880", item.TmdbID)
}
staleRows, err := staleRepo.GetByContentID(ctx, "existing-1")
if err != nil {
t.Fatalf("load stale rows: %v", err)
}
if len(staleRows) != 0 {
t.Errorf("stale rows after successful explicit retry = %#v, want none", staleRows)
}
}
internal/metadata/match_candidates.go
+68 -23
View File
@@ -37,10 +37,15 @@ type MatchCandidate struct {
Sources []string `json:"sources"`
AgreementHints []string `json:"agreement_hints"`
DetailScore int `json:"-"`
// ConflictingProviderIDKeys contains canonical provider IDs deliberately
// excluded after two other canonical IDs proved that provider results refer
// to the same work. These keys remain quarantined for the rest of the match
// pipeline so a later detail response cannot silently reintroduce them.
// ConfirmedProviderIDs contains provider IDs returned by their owning
// provider (for example, tmdb from a TMDB result). It arbitrates conflicting
// cross-references without rejecting compatible IDs from aggregators or
// third-party providers.
ConfirmedProviderIDs map[string]string `json:"-"`
// ConflictingProviderIDKeys contains canonical provider IDs excluded during
// candidate selection after two other canonical IDs proved that provider
// results refer to the same work. A value in ConfirmedProviderIDs resolves
// the conflict; otherwise the key remains quarantined through persistence.
ConflictingProviderIDKeys []string `json:"-"`
titleRank int
}
@@ -289,9 +294,11 @@ func NormalizeCandidates(results []SearchResult, contentType string) []MatchCand
// explicitly localized provider title over a provider-marked fallback.
func NormalizeCandidatesForLanguage(results []SearchResult, contentType, language string) []MatchCandidate {
type bucket struct {
candidate MatchCandidate
sources map[string]bool
conflictingIDKeys map[string]bool
candidate MatchCandidate
sources map[string]bool
conflictingIDKeys map[string]bool
confirmedIDValues map[string]string
conflictingConfirmedIDKeys map[string]bool
}
ordered := make([]string, 0)
@@ -319,20 +326,23 @@ func NormalizeCandidatesForLanguage(results []SearchResult, contentType, languag
title, titleLanguage, fallback, titleRank := preferredSearchResultTitle(sr, language)
b = &bucket{
candidate: MatchCandidate{
Title: title,
OriginalTitle: sr.OriginalTitle,
TitleAliases: copyTitleAliases(sr.TitleAliases, sr.Provider),
TitleLanguage: titleLanguage,
TitleIsFallback: fallback,
titleRank: titleRank,
Year: sr.Year,
ContentType: contentType,
ProviderIDs: make(map[string]string),
ImageURL: sr.ImageURL,
Overview: sr.Overview,
Title: title,
OriginalTitle: sr.OriginalTitle,
TitleAliases: copyTitleAliases(sr.TitleAliases, sr.Provider),
TitleLanguage: titleLanguage,
TitleIsFallback: fallback,
titleRank: titleRank,
Year: sr.Year,
ContentType: contentType,
ProviderIDs: make(map[string]string),
ConfirmedProviderIDs: make(map[string]string),
ImageURL: sr.ImageURL,
Overview: sr.Overview,
},
sources: make(map[string]bool),
conflictingIDKeys: make(map[string]bool),
sources: make(map[string]bool),
conflictingIDKeys: make(map[string]bool),
confirmedIDValues: make(map[string]string),
conflictingConfirmedIDKeys: make(map[string]bool),
}
buckets[key] = b
ordered = append(ordered, key)
@@ -355,9 +365,20 @@ func NormalizeCandidatesForLanguage(results []SearchResult, contentType, languag
b.candidate.ProviderIDs[k] = v
}
// Track source providers.
// Track source providers and the ID returned by the provider that owns
// that namespace. Conflicting native results are not authoritative.
if sr.Provider != "" {
b.sources[sr.Provider] = true
provider := strings.ToLower(strings.TrimSpace(sr.Provider))
if slices.Contains(canonicalCandidateIDKeys, provider) && strings.TrimSpace(sr.ProviderIDs[provider]) != "" {
value := strings.TrimSpace(sr.ProviderIDs[provider])
if existing := b.confirmedIDValues[provider]; existing != "" && existing != value {
delete(b.confirmedIDValues, provider)
b.conflictingConfirmedIDKeys[provider] = true
} else if !b.conflictingConfirmedIDKeys[provider] {
b.confirmedIDValues[provider] = value
}
}
}
// Prefer non-empty overview and image.
@@ -382,9 +403,16 @@ func NormalizeCandidatesForLanguage(results []SearchResult, contentType, languag
sort.Strings(sources)
b.candidate.Sources = sources
for _, key := range canonicalCandidateIDKeys {
if value := b.confirmedIDValues[key]; value != "" {
b.candidate.ConfirmedProviderIDs[key] = value
}
if b.conflictingIDKeys[key] {
b.candidate.ConflictingProviderIDKeys = append(b.candidate.ConflictingProviderIDKeys, key)
b.candidate.AgreementHints = append(b.candidate.AgreementHints, "quarantined_"+key+"_id")
if b.candidate.ConfirmedProviderIDs[key] != "" {
b.candidate.AgreementHints = append(b.candidate.AgreementHints, "resolved_"+key+"_id")
} else {
b.candidate.AgreementHints = append(b.candidate.AgreementHints, "quarantined_"+key+"_id")
}
}
}
@@ -631,7 +659,11 @@ func annotateCandidateMatch(candidate *MatchCandidate, hints *MatchHints) {
if len(candidate.ConflictingProviderIDKeys) > 0 {
candidate.MatchReasons = append(candidate.MatchReasons, "provider_id_consensus")
for _, key := range candidate.ConflictingProviderIDKeys {
candidate.MatchReasons = append(candidate.MatchReasons, "quarantined_"+key+"_id")
if strings.TrimSpace(candidate.ConfirmedProviderIDs[key]) != "" {
candidate.MatchReasons = append(candidate.MatchReasons, "resolved_"+key+"_id")
} else {
candidate.MatchReasons = append(candidate.MatchReasons, "quarantined_"+key+"_id")
}
}
}
}
@@ -766,15 +798,28 @@ func anyCandidateHasProviderIDs(group []scoredMatchCandidate) bool {
func adoptGroupProviderIDs(winner *MatchCandidate, group []scoredMatchCandidate) *MatchCandidate {
adopted := *winner
adopted.ProviderIDs = make(map[string]string, len(winner.ProviderIDs))
adopted.ConfirmedProviderIDs = make(map[string]string)
for k, v := range winner.ProviderIDs {
adopted.ProviderIDs[k] = v
}
confirmedConflicts := make(map[string]bool)
for _, c := range group {
for _, key := range canonicalCandidateIDKeys {
if v := strings.TrimSpace(c.candidate.ProviderIDs[key]); v != "" && adopted.ProviderIDs[key] == "" {
adopted.ProviderIDs[key] = v
}
}
for key, value := range c.candidate.ConfirmedProviderIDs {
if confirmedConflicts[key] {
continue
}
if existing := adopted.ConfirmedProviderIDs[key]; existing != "" && existing != value {
delete(adopted.ConfirmedProviderIDs, key)
confirmedConflicts[key] = true
continue
}
adopted.ConfirmedProviderIDs[key] = value
}
}
return &adopted
}
internal/metadata/match_candidates_test.go
+80 -1
View File
@@ -460,9 +460,12 @@ func TestNormalizeCandidates(t *testing.T) {
if len(candidate.ConflictingProviderIDKeys) != 1 || candidate.ConflictingProviderIDKeys[0] != "tvdb" {
t.Fatalf("quarantined keys = %v, want [tvdb]", candidate.ConflictingProviderIDKeys)
}
if candidate.ConfirmedProviderIDs["tvdb"] != "352440" {
t.Fatalf("native TVDB resolution = %q, want 352440", candidate.ConfirmedProviderIDs["tvdb"])
}
annotateCandidateMatch(&candidate, &MatchHints{Title: "A Teacher", Type: "series"})
if !containsString(candidate.MatchReasons, "provider_id_consensus") ||
!containsString(candidate.MatchReasons, "quarantined_tvdb_id") {
!containsString(candidate.MatchReasons, "resolved_tvdb_id") {
t.Fatalf("match reasons = %v", candidate.MatchReasons)
}
},
@@ -1222,3 +1225,79 @@ func TestSelectRefreshMatchCandidate_RejectsConflictingTrustedIDCandidate(t *tes
t.Fatalf("expected conflicting trusted-ID candidate to be rejected")
}
}
func TestApplyCandidateProviderIDConsensusKeepsNonConflictingAggregatorIDs(t *testing.T) {
candidates := NormalizeCandidates([]SearchResult{{
Name: "Example Movie",
Provider: "metadb",
ProviderIDs: map[string]string{
"tmdb": "999",
"imdb": "tt7654321",
},
}}, "movie")
if len(candidates) != 1 {
t.Fatalf("candidate count = %d, want 1", len(candidates))
}
ids := map[string]string{}
applyCandidateProviderIDConsensus(ids, &candidates[0], nil)
if ids["tmdb"] != "999" || ids["imdb"] != "tt7654321" {
t.Fatalf("aggregator provider ids = %#v, want tmdb/imdb retained", ids)
}
}
func TestApplyCandidateProviderIDConsensusPromotesIDsConfirmedByBothProviders(t *testing.T) {
candidates := NormalizeCandidates([]SearchResult{
{
Name: "Example", Provider: "tvdb",
ProviderIDs: map[string]string{"tvdb": "405851", "tmdb": "1234", "imdb": "tt12236904"},
},
{
Name: "Example", Provider: "tmdb",
ProviderIDs: map[string]string{"tvdb": "405851", "tmdb": "1234", "imdb": "tt12236904"},
},
}, "series")
if len(candidates) != 1 {
t.Fatalf("candidate count = %d, want 1", len(candidates))
}
ids := map[string]string{}
applyCandidateProviderIDConsensus(ids, &candidates[0], nil)
if ids["tvdb"] != "405851" || ids["tmdb"] != "1234" || ids["imdb"] != "tt12236904" {
t.Fatalf("confirmed provider ids = %#v, want tmdb/tvdb/imdb", ids)
}
}
func TestApplyCandidateProviderIDConsensusPrefersOwningProviderDuringConflict(t *testing.T) {
for _, reverse := range []bool{false, true} {
name := "foreign cross-reference first"
results := []SearchResult{
{
Name: "Under the Pole", Provider: "tvdb",
ProviderIDs: map[string]string{"tvdb": "405851", "tmdb": "12236904", "imdb": "tt12236904"},
},
{
Name: "Under the Pole", Provider: "tmdb",
ProviderIDs: map[string]string{"tvdb": "405851", "tmdb": "987654", "imdb": "tt12236904"},
},
}
if reverse {
name = "owning provider first"
results[0], results[1] = results[1], results[0]
}
t.Run(name, func(t *testing.T) {
candidates := NormalizeCandidates(results, "series")
if len(candidates) != 1 {
t.Fatalf("candidate count = %d, want 1", len(candidates))
}
ids := map[string]string{}
applyCandidateProviderIDConsensus(ids, &candidates[0], nil)
if ids["tmdb"] != "987654" {
t.Fatalf("resolved tmdb id = %q, want owning-provider value 987654", ids["tmdb"])
}
if ids["tvdb"] != "405851" || ids["imdb"] != "tt12236904" {
t.Fatalf("resolved provider ids = %#v", ids)
}
})
}
}
internal/metadata/provider_404_test.go
+38 -3
View File
@@ -5,7 +5,7 @@ import (
"testing"
)
func TestHandleChildProvider404KeepsProviderID(t *testing.T) {
func TestHandleScopedProvider404KeepsProviderID(t *testing.T) {
t.Parallel()
ids := map[string]string{
@@ -13,8 +13,8 @@ func TestHandleChildProvider404KeepsProviderID(t *testing.T) {
"tvdb": "164951",
}
if !handleChildProvider404("tmdb", ids, errors.New("tmdb: HTTP 404: not found"), "season", 0) {
t.Fatal("handleChildProvider404() = false, want true")
if !handleScopedProvider404("tmdb", ids, errors.New("tmdb: HTTP 404: not found"), "season", 0) {
t.Fatal("handleScopedProvider404() = false, want true")
}
if ids["tmdb"] != "32843" {
t.Fatalf("tmdb id = %q, want preserved", ids["tmdb"])
@@ -42,3 +42,38 @@ func TestHandleProvider404DropsProviderID(t *testing.T) {
t.Fatalf("tvdb id = %q, want preserved", ids["tvdb"])
}
}
func TestProvider404StateKeepsAllRejectedValues(t *testing.T) {
t.Parallel()
state := newProvider404State()
state.record("tmdb", "111")
state.record("tmdb", "222")
if _, ok := state.stale["tmdb"]["111"]; !ok {
t.Fatal("first rejected tmdb value was lost")
}
if _, ok := state.stale["tmdb"]["222"]; !ok {
t.Fatal("second rejected tmdb value was not recorded")
}
}
func TestApplyProvider404sDropsNonDurableIDsWithoutPersistingThem(t *testing.T) {
t.Parallel()
state := newProvider404State()
state.record(testMetaDBProvider, "ephemeral-1")
accumulator := &MetadataResult{ProviderIDs: map[string]string{
testMetaDBProvider: "ephemeral-1",
"tmdb": "603",
}}
applyProvider404sToAccumulator(accumulator, state)
if accumulator.ProviderIDs[testMetaDBProvider] != "" {
t.Fatalf("rejected metadb id was resurrected: %#v", accumulator.ProviderIDs)
}
if accumulator.ProviderIDs["tmdb"] != "603" {
t.Fatalf("unrelated tmdb id was removed: %#v", accumulator.ProviderIDs)
}
if len(accumulator.sameRunStaleProviderIDs) != 0 {
t.Fatalf("ephemeral ID was marked durable stale: %#v", accumulator.sameRunStaleProviderIDs)
}
}
internal/metadata/provider_id_integrity.go
+2 -3
View File
@@ -869,9 +869,8 @@ var mediaItemMergeSteps = []mediaItemMergeStep{
SELECT $2, provider, provider_id, first_seen_at, last_seen_at
FROM stale_media_ids
WHERE content_id = $1
ON CONFLICT (content_id, provider) DO UPDATE
SET provider_id = COALESCE(NULLIF(stale_media_ids.provider_id, ''), EXCLUDED.provider_id),
first_seen_at = LEAST(stale_media_ids.first_seen_at, EXCLUDED.first_seen_at),
ON CONFLICT (content_id, provider, provider_id) DO UPDATE
SET first_seen_at = LEAST(stale_media_ids.first_seen_at, EXCLUDED.first_seen_at),
last_seen_at = GREATEST(stale_media_ids.last_seen_at, EXCLUDED.last_seen_at)`},
{"delete source stale media ids", `DELETE FROM stale_media_ids WHERE content_id = $1`},
{"move watch provider history exports", `UPDATE watch_provider_history_exports SET media_item_id = $2, updated_at = NOW() WHERE media_item_id = $1`},
internal/metadata/provider_id_persistence_test.go
+490 -15
View File
@@ -2,12 +2,20 @@ package metadata
import (
"context"
"errors"
"sync"
"testing"
"github.com/Silo-Server/silo-server/internal/contentid"
"github.com/Silo-Server/silo-server/internal/models"
)
const (
staleRecoveryContentID = "legacy-item-1"
staleRecoveryTitle = "Recovered Movie"
staleRecoveryIMDbID = "tt1234567"
)
type fakeProviderIDRepo struct {
mu sync.Mutex
byContentID map[string][]*models.MediaItemProviderID
@@ -116,24 +124,17 @@ func (r *fakeStaleIDRepo) GetByContentID(_ context.Context, contentID string) ([
func (r *fakeStaleIDRepo) Upsert(_ context.Context, contentID, provider, providerID string) error {
r.mu.Lock()
defer r.mu.Unlock()
updated := false
for i, row := range r.byContentID[contentID] {
if row == nil || row.Provider != provider {
for _, row := range r.byContentID[contentID] {
if row == nil || row.Provider != provider || row.ProviderID != providerID {
continue
}
cp := *row
cp.ProviderID = providerID
r.byContentID[contentID][i] = &cp
updated = true
break
}
if !updated {
r.byContentID[contentID] = append(r.byContentID[contentID], &models.StaleMediaID{
ContentID: contentID,
Provider: provider,
ProviderID: providerID,
})
return nil
}
r.byContentID[contentID] = append(r.byContentID[contentID], &models.StaleMediaID{
ContentID: contentID,
Provider: provider,
ProviderID: providerID,
})
return nil
}
@@ -164,6 +165,90 @@ type capturingMetadataProvider struct {
response *MetadataResult
}
type searchMetadataProvider struct {
slug string
searchResults []SearchResult
metadataResult *MetadataResult
metadataCalls int
}
type image404MetadataProvider struct {
*searchMetadataProvider
}
type searchThen404MetadataProvider struct {
slug string
searchResults []SearchResult
}
type capturingEpisodeBootstrapProvider struct {
providerIDs map[string]string
}
func (p *image404MetadataProvider) GetImages(_ context.Context, _ ImageRequest) ([]RemoteImage, error) {
return nil, errors.New(p.slug + ": HTTP 404: not found")
}
func (p *searchThen404MetadataProvider) Slug() string { return p.slug }
func (p *searchThen404MetadataProvider) Name() string { return p.slug }
func (p *searchThen404MetadataProvider) ForTypes() []string {
return []string{anchoredItemTypeMovie, anchoredItemTypeSeries}
}
func (p *searchThen404MetadataProvider) Search(_ context.Context, _ SearchQuery) ([]SearchResult, error) {
return append([]SearchResult(nil), p.searchResults...), nil
}
func (p *searchThen404MetadataProvider) GetMetadata(_ context.Context, _ MetadataRequest) (*MetadataResult, error) {
return nil, errors.New(p.slug + ": HTTP 404: not found")
}
func (p *capturingEpisodeBootstrapProvider) Slug() string { return contentid.ProviderTVDB }
func (p *capturingEpisodeBootstrapProvider) Name() string { return "TVDB episodes" }
func (p *capturingEpisodeBootstrapProvider) ForTypes() []string {
return []string{anchoredItemTypeSeries}
}
func (p *capturingEpisodeBootstrapProvider) GetSeasons(_ context.Context, req SeasonsRequest) ([]SeasonResult, error) {
p.providerIDs = copyMap(req.ProviderIDs)
return nil, nil
}
func (p *capturingEpisodeBootstrapProvider) GetEpisodes(context.Context, EpisodesRequest) ([]EpisodeResult, error) {
return nil, nil
}
func (p *searchMetadataProvider) Slug() string { return p.slug }
func (p *searchMetadataProvider) Name() string { return p.slug }
func (p *searchMetadataProvider) ForTypes() []string {
return []string{anchoredItemTypeMovie, anchoredItemTypeSeries}
}
func (p *searchMetadataProvider) Search(_ context.Context, _ SearchQuery) ([]SearchResult, error) {
results := make([]SearchResult, len(p.searchResults))
for i := range p.searchResults {
results[i] = p.searchResults[i]
results[i].ProviderIDs = copyMap(p.searchResults[i].ProviderIDs)
}
return results, nil
}
func (p *searchMetadataProvider) GetMetadata(_ context.Context, _ MetadataRequest) (*MetadataResult, error) {
p.metadataCalls++
if p.metadataResult == nil {
return nil, nil
}
result := *p.metadataResult
result.ProviderIDs = copyMap(p.metadataResult.ProviderIDs)
return &result, nil
}
func (p *capturingMetadataProvider) Slug() string { return "capture" }
func (p *capturingMetadataProvider) Name() string { return "capture" }
@@ -188,6 +273,396 @@ func (p *capturingMetadataProvider) lastRequest() MetadataRequest {
return p.lastReq
}
func TestProcess_ScheduledRefreshReplacesRecordedStaleCurrentID(t *testing.T) {
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: staleRecoveryContentID,
Type: anchoredItemTypeMovie,
Title: staleRecoveryTitle,
Year: 2020,
Status: string(MatchOutcomeMatched),
TmdbID: "111",
Studios: []string{},
Networks: []string{},
Countries: []string{},
Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
providerRepo := newFakeProviderIDRepo()
providerRepo.set(staleRecoveryContentID, &models.MediaItemProviderID{
ContentID: staleRecoveryContentID, ItemType: anchoredItemTypeMovie,
Provider: contentid.ProviderTMDB, ProviderID: "111",
})
h.service.providerIDRepo = providerRepo
staleRepo := newFakeStaleIDRepo()
staleRepo.set(staleRecoveryContentID, &models.StaleMediaID{
ContentID: staleRecoveryContentID, Provider: contentid.ProviderTMDB, ProviderID: "111",
})
h.service.staleIDRepo = staleRepo
tmdb := &searchMetadataProvider{
slug: contentid.ProviderTMDB,
searchResults: []SearchResult{{
Name: staleRecoveryTitle, Year: 2020, Provider: contentid.ProviderTMDB,
ProviderIDs: map[string]string{contentid.ProviderTMDB: "222", contentid.ProviderIMDB: staleRecoveryIMDbID},
}},
metadataResult: &MetadataResult{
HasMetadata: true, Title: staleRecoveryTitle, Year: 2020,
ProviderIDs: map[string]string{contentid.ProviderTMDB: "222", contentid.ProviderIMDB: staleRecoveryIMDbID},
},
}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: staleRecoveryContentID,
Language: "en",
Mode: ModeScheduledRefresh,
}, []Provider{tmdb})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || !result.Updated {
t.Fatalf("result = %#v, want Updated=true", result)
}
providerRepo.mu.Lock()
persisted := providerRepo.lastReplace[staleRecoveryContentID]
providerRepo.mu.Unlock()
if persisted[contentid.ProviderTMDB] != "222" {
t.Fatalf("persisted tmdb id = %q, want replacement 222", persisted[contentid.ProviderTMDB])
}
if persisted[contentid.ProviderIMDB] != staleRecoveryIMDbID {
t.Fatalf("persisted imdb id = %q, want %s", persisted[contentid.ProviderIMDB], staleRecoveryIMDbID)
}
if tmdb.metadataCalls != 1 {
t.Fatalf("metadata calls = %d, want 1", tmdb.metadataCalls)
}
}
func TestProcess_InitialMatchPreservesAggregatorProviderIdentity(t *testing.T) {
const (
aggregatorTitle = "Aggregator Movie"
aggregatorIMDbID = "tt7654321"
)
h := newTestHarness()
provider := &searchMetadataProvider{
slug: testMetaDBProvider,
searchResults: []SearchResult{{
Name: aggregatorTitle, Year: 2020, Provider: testMetaDBProvider,
ProviderIDs: map[string]string{
contentid.ProviderTMDB: "999",
contentid.ProviderIMDB: aggregatorIMDbID,
},
}},
metadataResult: &MetadataResult{
HasMetadata: true, Title: aggregatorTitle, Year: 2020,
ProviderIDs: map[string]string{
contentid.ProviderTMDB: "999",
contentid.ProviderIMDB: aggregatorIMDbID,
},
},
}
result, err := h.service.ProcessWithProviders(context.Background(), ProcessRequest{
Hints: &MatchHints{Title: aggregatorTitle, Year: 2020, Type: anchoredItemTypeMovie},
Mode: ModeInitialMatch,
}, []Provider{provider})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || result.ContentID != "movie-tmdb-999" {
t.Fatalf("result = %#v, want movie-tmdb-999", result)
}
item, err := h.itemRepo.GetByID(context.Background(), result.ContentID)
if err != nil {
t.Fatalf("load matched item: %v", err)
}
if item.TmdbID != "999" || item.ImdbID != aggregatorIMDbID {
t.Fatalf("persisted ids = tmdb:%q imdb:%q", item.TmdbID, item.ImdbID)
}
}
func TestProcess_ScheduledRefreshDoesNotReanchorRecordedStaleContentID(t *testing.T) {
const (
contentID = "movie-tmdb-777"
imdbID = "tt0000777"
)
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: contentID, Type: anchoredItemTypeMovie, Title: "Stable Identity",
Status: string(MatchOutcomeMatched), TmdbID: "777", ImdbID: imdbID,
Studios: []string{}, Networks: []string{}, Countries: []string{}, Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
providerRepo := newFakeProviderIDRepo()
providerRepo.set(contentID,
&models.MediaItemProviderID{ContentID: contentID, ItemType: anchoredItemTypeMovie, Provider: contentid.ProviderTMDB, ProviderID: "777"},
&models.MediaItemProviderID{ContentID: contentID, ItemType: anchoredItemTypeMovie, Provider: contentid.ProviderIMDB, ProviderID: imdbID},
)
h.service.providerIDRepo = providerRepo
staleRepo := newFakeStaleIDRepo()
staleRepo.set(contentID, &models.StaleMediaID{
ContentID: contentID, Provider: contentid.ProviderTMDB, ProviderID: "777",
})
h.service.staleIDRepo = staleRepo
provider := &capturingMetadataProvider{response: &MetadataResult{
HasMetadata: true, Title: "Stable Identity",
ProviderIDs: map[string]string{contentid.ProviderIMDB: imdbID},
}}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: contentID, Language: "en", Mode: ModeScheduledRefresh,
}, []Provider{provider})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || result.ContentID != contentID {
t.Fatalf("result content id = %#v, want %s preserved", result, contentID)
}
}
func TestProcess_ScheduledRefreshDoesNotRestoreProviderIDThat404sDuringSameRun(t *testing.T) {
const (
itemID = "legacy-item-new-404"
badTMDB = "333"
goodTVDB = "444"
)
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: itemID,
Type: anchoredItemTypeSeries,
Title: "Still Available",
Status: string(MatchOutcomeMatched),
TmdbID: badTMDB,
TvdbID: goodTVDB,
Studios: []string{}, Networks: []string{}, Countries: []string{}, Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
providerRepo := newFakeProviderIDRepo()
providerRepo.set(itemID,
&models.MediaItemProviderID{ContentID: itemID, ItemType: anchoredItemTypeSeries, Provider: contentid.ProviderTMDB, ProviderID: badTMDB},
&models.MediaItemProviderID{ContentID: itemID, ItemType: anchoredItemTypeSeries, Provider: contentid.ProviderTVDB, ProviderID: goodTVDB},
)
h.service.providerIDRepo = providerRepo
h.service.staleIDRepo = newFakeStaleIDRepo()
tvdb := &searchMetadataProvider{
slug: contentid.ProviderTVDB,
metadataResult: &MetadataResult{
HasMetadata: true,
Title: "Still Available",
ProviderIDs: map[string]string{contentid.ProviderTVDB: goodTVDB},
},
}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: itemID,
Language: "en",
Mode: ModeScheduledRefresh,
}, []Provider{&notFoundMetadataProvider{slug: contentid.ProviderTMDB}, tvdb})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || !result.Updated {
t.Fatalf("result = %#v, want Updated=true", result)
}
providerRepo.mu.Lock()
persisted := providerRepo.lastReplace[itemID]
providerRepo.mu.Unlock()
if persisted[contentid.ProviderTMDB] != "" {
t.Fatalf("same-run 404 tmdb id was restored: %#v", persisted)
}
if persisted[contentid.ProviderTVDB] != goodTVDB {
t.Fatalf("persisted tvdb id = %q, want %s", persisted[contentid.ProviderTVDB], goodTVDB)
}
}
func TestProcess_ScheduledRefreshSuppressesRecordedAndReplacement404Values(t *testing.T) {
const (
itemID = "legacy-item-two-stale-values"
oldTMDBID = "111"
replacementTMDB = "222"
goodTVDBID = "444"
title = "Two Bad IDs"
)
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: itemID, Type: anchoredItemTypeSeries, Title: title, Year: 2020,
Status: string(MatchOutcomeMatched), TmdbID: oldTMDBID,
Studios: []string{}, Networks: []string{}, Countries: []string{}, Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
providerRepo := newFakeProviderIDRepo()
providerRepo.set(itemID, &models.MediaItemProviderID{
ContentID: itemID, ItemType: anchoredItemTypeSeries,
Provider: contentid.ProviderTMDB, ProviderID: oldTMDBID,
})
h.service.providerIDRepo = providerRepo
staleRepo := newFakeStaleIDRepo()
staleRepo.set(itemID, &models.StaleMediaID{
ContentID: itemID, Provider: contentid.ProviderTMDB, ProviderID: oldTMDBID,
})
h.service.staleIDRepo = staleRepo
tmdb := &searchThen404MetadataProvider{
slug: contentid.ProviderTMDB,
searchResults: []SearchResult{{
Name: title, Year: 2020, Provider: contentid.ProviderTMDB,
ProviderIDs: map[string]string{contentid.ProviderTMDB: replacementTMDB},
}},
}
tvdb := &capturingMetadataProvider{response: &MetadataResult{
HasMetadata: true, Title: title,
ProviderIDs: map[string]string{contentid.ProviderTVDB: goodTVDBID},
}}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: itemID, Language: "en", Mode: ModeScheduledRefresh,
}, []Provider{tmdb, tvdb})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || !result.Updated {
t.Fatalf("result = %#v, want Updated=true", result)
}
providerRepo.mu.Lock()
persisted := providerRepo.lastReplace[itemID]
providerRepo.mu.Unlock()
if persisted[contentid.ProviderTMDB] != "" {
t.Fatalf("stale tmdb value was restored: %#v", persisted)
}
if persisted[contentid.ProviderTVDB] != goodTVDBID {
t.Fatalf("persisted tvdb id = %q, want %s", persisted[contentid.ProviderTVDB], goodTVDBID)
}
staleRows, err := staleRepo.GetByContentID(ctx, itemID)
if err != nil {
t.Fatalf("load stale rows: %v", err)
}
staleTMDBIDs := make(map[string]bool)
for _, row := range staleRows {
if row != nil && row.Provider == contentid.ProviderTMDB {
staleTMDBIDs[row.ProviderID] = true
}
}
if !staleTMDBIDs[oldTMDBID] || !staleTMDBIDs[replacementTMDB] {
t.Fatalf("stale tmdb IDs = %#v, want both %s and %s", staleTMDBIDs, oldTMDBID, replacementTMDB)
}
}
func TestProcess_ScheduledRefreshPreservesProviderIDWhenOnlyImages404(t *testing.T) {
const (
itemID = "legacy-item-image-404"
tmdbID = "555"
title = "Metadata Without Images"
)
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: itemID,
Type: anchoredItemTypeMovie,
Title: title,
Status: string(MatchOutcomeMatched),
TmdbID: tmdbID,
Studios: []string{}, Networks: []string{}, Countries: []string{}, Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
providerRepo := newFakeProviderIDRepo()
providerRepo.set(itemID, &models.MediaItemProviderID{
ContentID: itemID, ItemType: anchoredItemTypeMovie,
Provider: contentid.ProviderTMDB, ProviderID: tmdbID,
})
h.service.providerIDRepo = providerRepo
h.service.staleIDRepo = newFakeStaleIDRepo()
provider := &image404MetadataProvider{searchMetadataProvider: &searchMetadataProvider{
slug: contentid.ProviderTMDB,
metadataResult: &MetadataResult{
HasMetadata: true,
Title: title,
ProviderIDs: map[string]string{contentid.ProviderTMDB: tmdbID},
},
}}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: itemID,
Language: "en",
Mode: ModeScheduledRefresh,
}, []Provider{provider})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || !result.Updated {
t.Fatalf("result = %#v, want Updated=true", result)
}
providerRepo.mu.Lock()
persisted := providerRepo.lastReplace[itemID]
providerRepo.mu.Unlock()
if persisted[contentid.ProviderTMDB] != tmdbID {
t.Fatalf("image-phase 404 removed healthy tmdb id: %#v", persisted)
}
stale, err := h.service.staleIDRepo.GetByContentID(ctx, itemID)
if err != nil {
t.Fatalf("get stale ids: %v", err)
}
if len(stale) != 0 {
t.Fatalf("image-phase 404 recorded stale identity: %#v", stale)
}
}
func TestProcess_IdentifyBootstrapsEpisodeProviderFromDetailCrossReference(t *testing.T) {
const (
itemID = "series-bootstrap-detail-id"
tmdbID = "123"
tvdbID = "456"
)
h := newTestHarness()
ctx := context.Background()
if err := h.itemRepo.Upsert(ctx, &models.MediaItem{
ContentID: itemID, Type: anchoredItemTypeSeries, Title: "Bootstrap Show",
Status: string(MatchOutcomeMatched), TmdbID: tmdbID,
Studios: []string{}, Networks: []string{}, Countries: []string{}, Genres: []string{},
}); err != nil {
t.Fatalf("upsert existing item: %v", err)
}
h.service.providerIDRepo = newFakeProviderIDRepo()
h.service.staleIDRepo = newFakeStaleIDRepo()
metadataProvider := &capturingMetadataProvider{response: &MetadataResult{
HasMetadata: true, Title: "Bootstrap Show",
ProviderIDs: map[string]string{
contentid.ProviderTMDB: tmdbID,
contentid.ProviderTVDB: tvdbID,
},
}}
episodeProvider := &capturingEpisodeBootstrapProvider{}
result, err := h.service.ProcessWithProviders(ctx, ProcessRequest{
ContentID: itemID,
ProviderIDs: map[string]string{contentid.ProviderTMDB: tmdbID},
Language: "en",
Mode: ModeIdentify,
}, []Provider{metadataProvider, episodeProvider})
if err != nil {
t.Fatalf("ProcessWithProviders: %v", err)
}
if result == nil || !result.Updated {
t.Fatalf("result = %#v, want Updated=true", result)
}
if episodeProvider.providerIDs[contentid.ProviderTVDB] != tvdbID {
t.Fatalf("season bootstrap ids = %#v, want tvdb=%s", episodeProvider.providerIDs, tvdbID)
}
}
func TestFindExistingByProviderIDsUsesDurableRepository(t *testing.T) {
h := newTestHarness()
ctx := context.Background()
internal/metadata/service.go
+265 -110
View File
@@ -236,8 +236,93 @@ func dropProviderID(providerIDs map[string]string, provider string) {
delete(providerIDs, strings.ToLower(strings.TrimSpace(provider)))
}
type providerIDValueSet map[string]map[string]struct{}
func (s providerIDValueSet) add(provider, providerID string) {
provider = strings.ToLower(strings.TrimSpace(provider))
providerID = normalizeProviderIDComparisonValue(provider, providerID)
if provider == "" || providerID == "" {
return
}
if s[provider] == nil {
s[provider] = make(map[string]struct{})
}
s[provider][providerID] = struct{}{}
}
func (s providerIDValueSet) remove(provider, providerID string) {
provider = strings.ToLower(strings.TrimSpace(provider))
providerID = normalizeProviderIDComparisonValue(provider, providerID)
values := s[provider]
if providerID == "" || len(values) == 0 {
return
}
delete(values, providerID)
if len(values) == 0 {
delete(s, provider)
}
}
func cloneProviderIDValueSet(src providerIDValueSet) providerIDValueSet {
if len(src) == 0 {
return make(providerIDValueSet)
}
dst := make(providerIDValueSet, len(src))
for provider, values := range src {
dst[provider] = maps.Clone(values)
}
return dst
}
type provider404State struct {
// dropped tracks every provider value rejected in this run so an
// accumulator copy cannot resurrect it for later phases.
dropped providerIDValueSet
// stale tracks durable values that must not be persisted as current item
// identity and must remain in the negative cache.
stale providerIDValueSet
}
func newProvider404State() *provider404State {
return &provider404State{
dropped: make(providerIDValueSet),
stale: make(providerIDValueSet),
}
}
func (s *provider404State) record(provider, providerID string) {
if s == nil {
return
}
s.dropped.add(provider, providerID)
provider = strings.ToLower(strings.TrimSpace(provider))
providerID = strings.TrimSpace(providerID)
if isDurableProviderSlug(provider) && providerID != "" {
s.stale.add(provider, providerID)
}
}
func upsertStaleProviderIDValues(
ctx context.Context,
repo metadataStaleIDRepo,
contentID string,
values providerIDValueSet,
) {
if repo == nil || strings.TrimSpace(contentID) == "" {
return
}
for provider, providerIDs := range values {
for providerID := range providerIDs {
if err := repo.Upsert(ctx, contentID, provider, providerID); err != nil {
slog.WarnContext(ctx, "metadata: failed to record stale ID", "component", "metadata",
"content_id", contentID, "provider", provider, "provider_id", providerID, "error", err)
}
}
}
}
func handleProvider404(
provider404s map[string]string,
provider404s *provider404State,
providerIDs map[string]string,
provider string,
err error,
@@ -254,9 +339,7 @@ func handleProvider404(
logAttrs := append([]any{"provider", provider}, attrs...)
if providerID := strings.TrimSpace(providerIDs[provider]); providerID != "" {
if provider404s != nil && isDurableProviderSlug(provider) {
provider404s[provider] = providerID
}
provider404s.record(provider, providerID)
logAttrs = append(logAttrs, "provider_id", providerID)
dropProviderID(providerIDs, provider)
}
@@ -265,7 +348,7 @@ func handleProvider404(
return true
}
func handleChildProvider404(
func handleScopedProvider404(
provider string,
providerIDs map[string]string,
err error,
@@ -283,7 +366,7 @@ func handleChildProvider404(
}
}
logAttrs = append(logAttrs, "error", err)
slog.Info("metadata: provider returned 404 for unavailable child metadata", logAttrs...)
slog.Info("metadata: provider returned 404 for unavailable scoped metadata", logAttrs...)
return true
}
@@ -1039,21 +1122,30 @@ func (s *MetadataService) loadDurableProviderIDs(ctx context.Context, contentID
return providerIDMapFromRows(ids), nil
}
func (s *MetadataService) suppressRecordedStaleProviderIDs(
ctx context.Context,
contentID string,
providerIDs map[string]string,
) error {
if s == nil || s.staleIDRepo == nil || strings.TrimSpace(contentID) == "" || len(providerIDs) == 0 {
return nil
func (s *MetadataService) loadRecordedStaleProviderIDs(ctx context.Context, contentID string) (providerIDValueSet, error) {
staleValues := make(providerIDValueSet)
if s == nil || s.staleIDRepo == nil || strings.TrimSpace(contentID) == "" {
return staleValues, nil
}
staleIDs, err := s.staleIDRepo.GetByContentID(ctx, contentID)
if err != nil {
return fmt.Errorf("loading stale provider ids for %s: %w", contentID, err)
return nil, fmt.Errorf("loading stale provider ids for %s: %w", contentID, err)
}
if len(staleIDs) == 0 {
return nil
for _, staleID := range staleIDs {
if staleID == nil {
continue
}
provider := strings.ToLower(strings.TrimSpace(staleID.Provider))
providerID := strings.TrimSpace(staleID.ProviderID)
staleValues.add(provider, providerID)
}
return staleValues, nil
}
func suppressProviderIDValues(providerIDs map[string]string, staleValues providerIDValueSet) {
if len(providerIDs) == 0 || len(staleValues) == 0 {
return
}
// Index the incoming map by normalized provider key so suppression cannot
// be bypassed by casing or padding differences between the stored stale
@@ -1066,23 +1158,32 @@ func (s *MetadataService) suppressRecordedStaleProviderIDs(
}
keysByProvider[normalized] = append(keysByProvider[normalized], key)
}
for _, staleID := range staleIDs {
if staleID == nil {
continue
}
provider := strings.ToLower(strings.TrimSpace(staleID.Provider))
if provider == "" {
continue
}
staleValue := strings.TrimSpace(staleID.ProviderID)
for provider, staleProviderValues := range staleValues {
provider = strings.ToLower(strings.TrimSpace(provider))
for _, key := range keysByProvider[provider] {
if strings.TrimSpace(providerIDs[key]) != staleValue {
value := normalizeProviderIDComparisonValue(provider, providerIDs[key])
if _, stale := staleProviderValues[value]; !stale {
continue
}
delete(providerIDs, key)
}
}
return nil
}
func applyProvider404sToAccumulator(accumulator *MetadataResult, provider404s *provider404State) {
if accumulator == nil || provider404s == nil {
return
}
accumulator.sameRunStaleProviderIDs = cloneProviderIDValueSet(provider404s.stale)
suppressProviderIDValues(accumulator.ProviderIDs, provider404s.dropped)
}
func shouldReanchorProviderContentID(
contentID string,
isNew bool,
mode RefreshMode,
) bool {
return !isNew && mode == ModeManualRefresh && contentid.IsProviderAnchored(contentID)
}
// ProcessWithProviders runs the pipeline with explicit providers (for testing).
@@ -1101,6 +1202,10 @@ func (s *MetadataService) prepareProcessRequest(ctx context.Context, req Process
if err != nil {
return req, err
}
req.recordedStaleProviderIDs, err = s.loadRecordedStaleProviderIDs(ctx, req.ContentID)
if err != nil {
return req, err
}
if len(durableIDs) == 0 {
return req, nil
}
@@ -1111,9 +1216,7 @@ func (s *MetadataService) prepareProcessRequest(ctx context.Context, req Process
// Only the injected set is filtered: IDs the caller supplied explicitly
// in req.ProviderIDs stay untouched, so an admin deliberately
// re-selecting a previously-stale ID still retries it.
if err := s.suppressRecordedStaleProviderIDs(ctx, req.ContentID, durableIDs); err != nil {
return req, err
}
suppressProviderIDValues(durableIDs, req.recordedStaleProviderIDs)
if len(durableIDs) == 0 {
return req, nil
}
@@ -1154,13 +1257,12 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
// Track provider 404s as stale external IDs. This applies to initial match
// as well so bad embedded folder/file IDs can be recorded and dropped
// without surfacing as generic provider failures.
var provider404s map[string]string
if req.ContentID != "" {
provider404s = make(map[string]string)
}
provider404s := newProvider404State()
recordStaleIDs := req.ContentID != ""
var matchDecision *MatchDecision
var providerMatchErrors []error
quarantinedProviderIDKeys := make(map[string]struct{})
replacedProviderIDKeys := make(map[string]struct{})
switch req.Mode {
case ModeInitialMatch:
@@ -1188,9 +1290,7 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
} else if req.Hints.FilePath != "" {
accumulatedIDs["_filepath"] = req.Hints.FilePath
}
if err := s.suppressRecordedStaleProviderIDs(ctx, req.ContentID, accumulatedIDs); err != nil {
return nil, err
}
suppressProviderIDValues(accumulatedIDs, req.recordedStaleProviderIDs)
// Run search providers and choose a decisive normalized winner instead of
// letting the first non-empty result win.
@@ -1337,7 +1437,8 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
matchDecision.Outcome = MatchOutcomeTrustedIDTypeMismatch
}
if matched && winner != nil {
applyCandidateProviderIDConsensus(accumulatedIDs, winner, quarantinedProviderIDKeys)
maps.Copy(replacedProviderIDKeys,
applyCandidateProviderIDConsensus(accumulatedIDs, winner, quarantinedProviderIDKeys))
}
case ModeIdentify:
@@ -1375,9 +1476,7 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
}
sanitizeCanonicalProviderIDsInPlace(accumulatedIDs)
contentType = existing.Type
if err := s.suppressRecordedStaleProviderIDs(ctx, req.ContentID, accumulatedIDs); err != nil {
return nil, err
}
suppressProviderIDValues(accumulatedIDs, req.recordedStaleProviderIDs)
// Re-resolve itemChain now that contentType is known.
itemLevel = providerChainContentLevel(contentType)
@@ -1440,18 +1539,30 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
}
}
candidates := NormalizeCandidatesForLanguage(allResults, contentType, searchQuery.Language)
if winner, ok := selectRefreshMatchCandidate(existing, wonHints, candidates); ok && winner != nil {
applyCandidateProviderIDConsensus(accumulatedIDs, winner, quarantinedProviderIDKeys)
selectionItem := mediaItemWithProviderIDs(existing, accumulatedIDs)
if winner, ok := selectRefreshMatchCandidate(selectionItem, wonHints, candidates); ok && winner != nil {
maps.Copy(replacedProviderIDKeys,
applyCandidateProviderIDConsensus(accumulatedIDs, winner, quarantinedProviderIDKeys))
}
}
if req.Mode != ModeIdentify {
if err := s.suppressRecordedStaleProviderIDs(ctx, req.ContentID, accumulatedIDs); err != nil {
return nil, err
}
suppressProviderIDValues(accumulatedIDs, req.recordedStaleProviderIDs)
}
// Phase 2: Metadata — all MetadataProviders run, results merge into accumulator.
accumulator := &MetadataResult{ProviderIDs: copyMap(accumulatedIDs)}
recordedStaleIDs := cloneProviderIDValueSet(req.recordedStaleProviderIDs)
if req.Mode == ModeIdentify {
// A caller-provided identify value is an explicit retry. If it succeeds,
// allow it to replace its old stale record; a new 404 will still enter the
// same-run rejection set below.
for provider, providerID := range req.ProviderIDs {
recordedStaleIDs.remove(provider, providerID)
}
}
accumulator := &MetadataResult{
ProviderIDs: copyMap(accumulatedIDs),
recordedStaleProviderIDs: recordedStaleIDs,
}
filePath := ""
representativeFilePath := ""
observedRootPath := ""
@@ -1540,6 +1651,17 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
if len(quarantinedProviderIDKeys) > 0 {
accumulator.quarantinedProviderIDKeys = maps.Clone(quarantinedProviderIDKeys)
}
// Search and item-metadata 404s are identity evidence. Apply them before
// artwork and child phases so a copied accumulator cannot reintroduce a
// rejected ID. Detail responses may also repeat a value recorded stale by an
// earlier run, so suppress that set again after all detail merges. Scoped
// 404s below are deliberately non-destructive.
suppressProviderIDValues(accumulator.ProviderIDs, accumulator.recordedStaleProviderIDs)
applyProvider404sToAccumulator(accumulator, provider404s)
accumulatedIDs = accumulator.ProviderIDs
if len(replacedProviderIDKeys) > 0 {
accumulator.replacedProviderIDKeys = maps.Clone(replacedProviderIDKeys)
}
// Phase 3: Images — all ImageProviders run, collect all available images.
var allImages []RemoteImage
for _, p := range itemChain {
@@ -1562,7 +1684,7 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
PrimarySidecarSearchPaths: primarySidecarSearchPaths,
})
if err != nil {
if handleProvider404(provider404s, accumulatedIDs, p.Slug(), err,
if handleScopedProvider404(p.Slug(), accumulatedIDs, err,
"content_id", req.ContentID,
) {
continue
@@ -1605,10 +1727,9 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
SeasonDirectoryPaths: childCtx.seasonDirectoryPaths,
})
if err != nil {
// Pass nil for provider404s so this refresh can drop the
// provider from the in-memory merge without recording a durable
// stale item ID from the season chain.
if handleProvider404(nil, accumulatedIDs, p.Slug(), err,
// A missing season endpoint is scoped metadata, not proof that the
// parent series identity is stale.
if handleScopedProvider404(p.Slug(), accumulatedIDs, err,
"content_id", req.ContentID,
"season", 0,
) {
@@ -1662,7 +1783,7 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
EpisodeFilePaths: childCtx.episodeFilePaths[seasonNumber],
})
if err != nil {
if handleChildProvider404(p.Slug(), accumulatedIDs, err,
if handleScopedProvider404(p.Slug(), accumulatedIDs, err,
"content_id", req.ContentID,
"season", seasonNumber,
) {
@@ -1678,20 +1799,10 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
allEpisodes = flattenEpisodeResults(episodeResults)
}
}
// Phase 5: Merge & Persist.
if !accumulator.HasMetadata && accumulator.Title == "" {
// Record stale IDs for providers that returned 404.
if s.staleIDRepo != nil && req.ContentID != "" && provider404s != nil {
for slug, providerID := range provider404s {
if providerID != "" {
if err := s.staleIDRepo.Upsert(ctx, req.ContentID, slug, providerID); err != nil {
slog.WarnContext(ctx, "metadata: failed to record stale ID", "component", "metadata",
"content_id", req.ContentID, "provider", slug, "provider_id", providerID, "error", err)
}
}
}
}
upsertStaleProviderIDValues(ctx, s.staleIDRepo, req.ContentID, provider404s.stale)
if matchDecision == nil {
matchDecision = &MatchDecision{Outcome: MatchOutcomeMetadataEmpty, Threshold: automaticMatchAcceptanceFloor}
} else if matchDecision.Outcome == MatchOutcomeMatched {
@@ -1719,35 +1830,31 @@ func (s *MetadataService) processInternal(ctx context.Context, req ProcessReques
}
}
// Refresh stale ID records on successful refresh: clear anything resolved,
// then keep only the providers that still 404ed during this run.
// Refresh stale ID records on successful refresh: clear anything explicitly
// resolved, then retain every earlier or current rejected value.
// Stale-ID follow-up targets the canonical content ID the item was
// persisted/merged into. When mergeAndPersist canonicalizes this item into
// an existing one, req.ContentID is the now-deleted source: clearing or
// recording stale IDs against it would touch nothing (or FK-violate), so
// the still-404ing providers must be recorded on result.ContentID instead.
// provider404s is only allocated when req.ContentID was set (the refresh
// targeted a known item). Guarding on it preserves the original gating so a
// content-id-less refresh that canonicalizes into an existing item does not
// wipe that item's stale rows without re-recording any.
// recordStaleIDs preserves the original gating so a content-id-less refresh
// that canonicalizes into an existing item does not wipe that item's stale
// rows without re-recording any.
followUpContentID := refreshFollowUpContentID(req.ContentID, result)
if s.staleIDRepo != nil && followUpContentID != "" && provider404s != nil {
if s.staleIDRepo != nil && recordStaleIDs && followUpContentID != "" {
if delErr := s.staleIDRepo.DeleteByContentID(ctx, followUpContentID); delErr != nil {
slog.WarnContext(ctx, "metadata: failed to clear stale IDs after refresh", "component", "metadata",
"content_id", followUpContentID, "error", delErr)
}
for slug, providerID := range provider404s {
if providerID == "" {
continue
}
if upsertErr := s.staleIDRepo.Upsert(ctx, followUpContentID, slug, providerID); upsertErr != nil {
slog.WarnContext(ctx, "metadata: failed to persist stale provider ID after partial refresh", "component", "metadata",
"content_id", followUpContentID,
"provider", slug,
"provider_id", providerID,
"error", upsertErr)
// Explicit ModeIdentify retries remove a successful value from
// recordedStaleProviderIDs; current-run 404s add it back through stale.
staleValues := cloneProviderIDValueSet(accumulator.recordedStaleProviderIDs)
for provider, providerIDs := range provider404s.stale {
for providerID := range providerIDs {
staleValues.add(provider, providerID)
}
}
upsertStaleProviderIDValues(ctx, s.staleIDRepo, followUpContentID, staleValues)
}
if result != nil && strings.TrimSpace(result.ContentID) != "" {
if syncErr := s.syncRefreshDebtForItem(ctx, result.ContentID); syncErr != nil {
@@ -2091,12 +2198,11 @@ func (s *MetadataService) mergeAndPersist(
// Re-anchor an already provider-anchored item whose corrected identity now
// derives a different anchor — the recovery path when an admin fixes a wrong
// <uniqueid> in an NFO. Manual refresh only: the accumulator's IDs are seeded
// from the item's stored IDs and only a trusted NFO hint (which wins on
// manual refresh) can change them here, so a scheduled/background refresh
// never flips a stored identity. Reuses the local-promotion machinery under
// the provider-dedup lock; a no-op when the derived anchor is unchanged.
if !isNew && req.Mode == ModeManualRefresh && contentid.IsProviderAnchored(contentID) {
// <uniqueid> in an NFO. Manual refresh only: scheduled jobs and ModeIdentify
// must preserve the client-visible content_id even when an external ID is
// stale. Reuses the local-promotion machinery under the provider-dedup lock;
// a no-op when the derived anchor is unchanged.
if shouldReanchorProviderContentID(contentID, isNew, req.Mode) {
reanchored, err := s.reanchorContentID(
ctx, contentID, providerIDsStruct(accumulator.ProviderIDs), contentType)
if err != nil {
@@ -2116,6 +2222,8 @@ func (s *MetadataService) mergeAndPersist(
}
}
suppressProviderIDValues(durableIDs, accumulator.recordedStaleProviderIDs)
suppressProviderIDValues(durableIDs, accumulator.sameRunStaleProviderIDs)
if len(durableIDs) > 0 {
if accumulator.ProviderIDs == nil {
accumulator.ProviderIDs = make(map[string]string, len(durableIDs))
@@ -2154,6 +2262,11 @@ func (s *MetadataService) mergeAndPersist(
if existingItem != nil {
existingResult := itemToMetadataResult(existingItem)
suppressProviderIDValues(existingResult.ProviderIDs, accumulator.recordedStaleProviderIDs)
suppressProviderIDValues(existingResult.ProviderIDs, accumulator.sameRunStaleProviderIDs)
for key := range accumulator.replacedProviderIDKeys {
delete(existingResult.ProviderIDs, key)
}
if req.Mode == ModeInitialMatch && isSkeletonLikeStatus(existingItem.Status) {
existingResult.Title = ""
existingResult.SortTitle = ""
@@ -2168,6 +2281,9 @@ func (s *MetadataService) mergeAndPersist(
delete(existingResult.ProviderIDs, key)
}
existingResult.quarantinedProviderIDKeys = accumulator.quarantinedProviderIDKeys
existingResult.replacedProviderIDKeys = accumulator.replacedProviderIDKeys
existingResult.recordedStaleProviderIDs = accumulator.recordedStaleProviderIDs
existingResult.sameRunStaleProviderIDs = accumulator.sameRunStaleProviderIDs
accumulator = existingResult
}
@@ -2707,10 +2823,13 @@ func (s *MetadataService) syncRefreshDebtForItem(ctx context.Context, contentID
if itemHasEpisodeMetadataDebt(item) && hasRefreshDebtReason(existingReasonMask, RefreshDebtReasonEpisodeIncomplete) {
reasonMask |= RefreshDebtReasonEpisodeIncomplete
}
if staleReason, err := s.currentStaleRefreshDebtReason(ctx, contentID); err != nil {
staleReason, missingTMDBRejected, err := s.currentStaleRefreshDebtState(ctx, contentID, item)
if err != nil {
return err
} else {
reasonMask |= staleReason
}
reasonMask |= staleReason
if missingTMDBRejected {
reasonMask &^= RefreshDebtReasonProviderIDIncomplete
}
if reasonMask == 0 {
@@ -2840,11 +2959,14 @@ func (s *MetadataService) syncRefreshDebtFailure(ctx context.Context, contentID
if itemHasEpisodeMetadataDebt(item) && hasRefreshDebtReason(existingReasonMask, RefreshDebtReasonEpisodeIncomplete) {
reasonMask |= RefreshDebtReasonEpisodeIncomplete
}
staleReason, err := s.currentStaleRefreshDebtReason(ctx, contentID)
staleReason, missingTMDBRejected, err := s.currentStaleRefreshDebtState(ctx, contentID, item)
if err != nil {
return err
}
reasonMask |= staleReason
if missingTMDBRejected {
reasonMask &^= RefreshDebtReasonProviderIDIncomplete
}
if strings.EqualFold(strings.TrimSpace(item.Status), "matched") &&
!hasRefreshDebtReason(reasonMask, RefreshDebtReasonProviderIDIncomplete) {
// Items missing provider IDs fail for that reason, not because the
@@ -2945,18 +3067,28 @@ func (s *MetadataService) currentRefreshDebtTargetReasonMask(ctx context.Context
return debt.ReasonMask, nil
}
func (s *MetadataService) currentStaleRefreshDebtReason(ctx context.Context, contentID string) (int64, error) {
func (s *MetadataService) currentStaleRefreshDebtState(
ctx context.Context,
contentID string,
item *models.MediaItem,
) (reason int64, missingTMDBRejected bool, err error) {
if s == nil || s.staleIDRepo == nil || strings.TrimSpace(contentID) == "" {
return 0, nil
return 0, false, nil
}
ids, err := s.staleIDRepo.GetByContentID(ctx, contentID)
if err != nil {
return 0, err
return 0, false, err
}
if len(ids) == 0 {
return 0, nil
for _, staleID := range ids {
if staleID != nil && strings.EqualFold(strings.TrimSpace(staleID.Provider), contentid.ProviderTMDB) &&
strings.TrimSpace(staleID.ProviderID) != "" {
missingTMDBRejected = true
}
if IsActionableStaleProviderID(item, staleID) {
reason = RefreshDebtReasonStaleProviderID
}
}
return RefreshDebtReasonStaleProviderID, nil
return reason, missingTMDBRejected, nil
}
func itemHasEpisodeMetadataDebt(item *models.MediaItem) bool {
@@ -3073,9 +3205,11 @@ func (s *MetadataService) resolveSeriesRefreshProviderIDs(ctx context.Context, s
}
maps.Copy(accumulatedIDs, durableIDs)
sanitizeCanonicalProviderIDsInPlace(accumulatedIDs)
if err := s.suppressRecordedStaleProviderIDs(ctx, series.ContentID, accumulatedIDs); err != nil {
recordedStaleIDs, err := s.loadRecordedStaleProviderIDs(ctx, series.ContentID)
if err != nil {
return nil, err
}
suppressProviderIDValues(accumulatedIDs, recordedStaleIDs)
itemChain, err := s.resolveChainCached(ctx, folderID, "series")
if err != nil {
@@ -3097,7 +3231,7 @@ func (s *MetadataService) resolveSeriesRefreshProviderIDs(ctx context.Context, s
searchQuery = suppressTitleYearFallbackForTrustedIDs(searchQuery)
allResults := make([]SearchResult, 0)
provider404s := make(map[string]string)
provider404s := newProvider404State()
for _, p := range itemChain {
sp, ok := p.(SearchProvider)
if !ok {
@@ -3124,12 +3258,12 @@ func (s *MetadataService) resolveSeriesRefreshProviderIDs(ctx context.Context, s
}
}
candidates := NormalizeCandidatesForLanguage(allResults, series.Type, searchQuery.Language)
if winner, ok := selectRefreshMatchCandidate(series, nil, candidates); ok && winner != nil {
selectionItem := mediaItemWithProviderIDs(series, accumulatedIDs)
if winner, ok := selectRefreshMatchCandidate(selectionItem, nil, candidates); ok && winner != nil {
applyCandidateProviderIDConsensus(accumulatedIDs, winner, nil)
}
if err := s.suppressRecordedStaleProviderIDs(ctx, series.ContentID, accumulatedIDs); err != nil {
return nil, err
}
suppressProviderIDValues(accumulatedIDs, recordedStaleIDs)
suppressProviderIDValues(accumulatedIDs, provider404s.dropped)
return accumulatedIDs, nil
}
@@ -3152,7 +3286,7 @@ func (s *MetadataService) fetchTargetSeasonResults(ctx context.Context, provider
SeasonDirectoryPaths: childCtx.seasonDirectoryPaths,
})
if err != nil {
if handleProvider404(nil, providerIDs, p.Slug(), err, "season", seasonNumber) {
if handleScopedProvider404(p.Slug(), providerIDs, err, "season", seasonNumber) {
continue
}
slog.WarnContext(ctx, "metadata: target season provider error", "component", "metadata",
@@ -3188,7 +3322,7 @@ func (s *MetadataService) fetchTargetEpisodeResults(ctx context.Context, provide
EpisodeFilePaths: childCtx.episodeFilePaths[seasonNumber],
})
if err != nil {
if handleChildProvider404(p.Slug(), providerIDs, err, "season", seasonNumber) {
if handleScopedProvider404(p.Slug(), providerIDs, err, "season", seasonNumber) {
continue
}
slog.WarnContext(ctx, "metadata: target episode provider error", "component", "metadata",
@@ -6680,12 +6814,15 @@ func searchResultConflictsWithTrustedIDs(hintedIDs, candidateIDs map[string]stri
}
// applyCandidateProviderIDConsensus replaces accumulated canonical IDs with a
// normalized winner while removing any cross-provider IDs quarantined during
// candidate grouping. The optional quarantine map carries the decision into
// Phase 2 so detail responses cannot reintroduce the disputed ID.
func applyCandidateProviderIDConsensus(accumulatedIDs map[string]string, winner *MatchCandidate, quarantine map[string]struct{}) {
// normalized winner. Compatible IDs retain the historical aggregator/plugin
// bootstrap behavior. When providers conflict, an owning-provider value wins;
// unresolved keys are removed and carried in quarantine through Phase 2 so
// detail responses cannot reintroduce them. The returned keys are deliberate
// replacements that must overwrite stored provider IDs at merge.
func applyCandidateProviderIDConsensus(accumulatedIDs map[string]string, winner *MatchCandidate, quarantine map[string]struct{}) map[string]struct{} {
replaced := make(map[string]struct{})
if winner == nil {
return
return replaced
}
locallyQuarantined := make(map[string]struct{}, len(winner.ConflictingProviderIDKeys))
for _, key := range winner.ConflictingProviderIDKeys {
@@ -6694,6 +6831,12 @@ func applyCandidateProviderIDConsensus(accumulatedIDs map[string]string, winner
continue
}
delete(accumulatedIDs, key)
if replacement := strings.TrimSpace(winner.ConfirmedProviderIDs[key]); replacement != "" {
accumulatedIDs[key] = replacement
delete(quarantine, key)
replaced[key] = struct{}{}
continue
}
locallyQuarantined[key] = struct{}{}
if quarantine != nil {
quarantine[key] = struct{}{}
@@ -6713,6 +6856,18 @@ func applyCandidateProviderIDConsensus(accumulatedIDs map[string]string, winner
}
accumulatedIDs[key] = value
}
return replaced
}
func mediaItemWithProviderIDs(item *models.MediaItem, providerIDs map[string]string) *models.MediaItem {
if item == nil {
return nil
}
selectionItem := *item
selectionItem.TmdbID = strings.TrimSpace(providerIDs[contentid.ProviderTMDB])
selectionItem.TvdbID = strings.TrimSpace(providerIDs[contentid.ProviderTVDB])
selectionItem.ImdbID = strings.TrimSpace(providerIDs[contentid.ProviderIMDB])
return &selectionItem
}
// applyBuiltinIdentityHints consults the chain's IdentityHintProviders (the
internal/metadata/service_test.go
+117 -2
View File
@@ -15,6 +15,15 @@ import (
"github.com/Silo-Server/silo-server/internal/models"
)
const (
testPosterPath = "/poster.jpg"
testBackdropPath = "/backdrop.jpg"
testTMDBProvider = "tmdb"
testTVDBProvider = "tvdb"
testIMDBProvider = "imdb"
testMetaDBProvider = "metadb"
)
// ---------------------------------------------------------------------------
// Fake repositories
// ---------------------------------------------------------------------------
@@ -1008,8 +1017,8 @@ func TestSyncRefreshDebtForItemPreservesRequestedEpisodeDebt(t *testing.T) {
Type: "series",
Status: "matched",
Overview: "Complete",
PosterPath: "/poster.jpg",
BackdropPath: "/backdrop.jpg",
PosterPath: testPosterPath,
BackdropPath: testBackdropPath,
RatingTMDB: &rating,
EpisodeMetadataIncomplete: true,
}
@@ -1043,6 +1052,112 @@ func TestSyncRefreshDebtForItemPreservesRequestedEpisodeDebt(t *testing.T) {
}
}
func TestSyncRefreshDebtForItemClearsDebtAfterSecondaryTMDBRejected(t *testing.T) {
h := newTestHarness()
ctx := context.Background()
debts := newFakeRefreshDebtRepo()
staleIDs := newFakeStaleIDRepo()
h.service.refreshDebtRepo = debts
h.service.staleIDRepo = staleIDs
rating := 8.0
h.itemRepo.items["series-1"] = &models.MediaItem{
ContentID: "series-1",
Type: "series",
Status: "matched",
TvdbID: "405851",
Overview: "Complete",
PosterPath: testPosterPath,
BackdropPath: testBackdropPath,
RatingTMDB: &rating,
}
staleIDs.set("series-1", &models.StaleMediaID{
ContentID: "series-1", Provider: testTMDBProvider, ProviderID: "12236904",
})
debts.debts["series-1"] = &models.MetadataRefreshDebt{
TargetType: RefreshTargetItem, ContentID: "series-1", ReasonMask: RefreshDebtReasonStaleProviderID,
}
if err := h.service.syncRefreshDebtForItem(ctx, "series-1"); err != nil {
t.Fatalf("syncRefreshDebtForItem: %v", err)
}
if _, err := debts.Get(ctx, "series-1"); !errors.Is(err, ErrRefreshDebtNotFound) {
t.Fatalf("Get debt after rejected secondary tmdb = %v, want ErrRefreshDebtNotFound", err)
}
}
func TestSyncRefreshDebtForItemKeepsFailedCurrentProviderID(t *testing.T) {
h := newTestHarness()
ctx := context.Background()
debts := newFakeRefreshDebtRepo()
staleIDs := newFakeStaleIDRepo()
h.service.refreshDebtRepo = debts
h.service.staleIDRepo = staleIDs
rating := 8.0
h.itemRepo.items["series-1"] = &models.MediaItem{
ContentID: "series-1",
Type: "series",
Status: "matched",
TvdbID: "405851",
Overview: "Complete",
PosterPath: testPosterPath,
BackdropPath: testBackdropPath,
RatingTMDB: &rating,
}
staleIDs.set("series-1", &models.StaleMediaID{
ContentID: "series-1", Provider: testTVDBProvider, ProviderID: "405851",
})
if err := h.service.syncRefreshDebtForItem(ctx, "series-1"); err != nil {
t.Fatalf("syncRefreshDebtForItem: %v", err)
}
debt, err := debts.Get(ctx, "series-1")
if err != nil {
t.Fatalf("Get debt: %v", err)
}
if !hasRefreshDebtReason(debt.ReasonMask, RefreshDebtReasonStaleProviderID) {
t.Fatalf("reason mask = %d, want stale provider ID", debt.ReasonMask)
}
}
func TestShouldReanchorProviderContentIDRequiresManualRefresh(t *testing.T) {
const anchoredID = "movie-tmdb-111"
tests := []struct {
name string
contentID string
isNew bool
mode RefreshMode
want bool
}{
{
name: "scheduled refresh",
contentID: anchoredID, mode: ModeScheduledRefresh,
},
{
name: "identify",
contentID: anchoredID, mode: ModeIdentify,
},
{
name: "manual refresh",
contentID: anchoredID, mode: ModeManualRefresh, want: true,
},
{
name: "new item never reanchors",
contentID: anchoredID, isNew: true, mode: ModeManualRefresh,
},
{
name: "local item",
contentID: "local-deadbeef", mode: ModeManualRefresh,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := shouldReanchorProviderContentID(tt.contentID, tt.isNew, tt.mode); got != tt.want {
t.Fatalf("shouldReanchorProviderContentID() = %v, want %v", got, tt.want)
}
})
}
}
func TestRefreshSeriesEpisodeMetadataStateUsesActionableDebt(t *testing.T) {
now := time.Date(2026, 4, 14, 12, 0, 0, 0, time.UTC)
internal/metadata/stale_media_id_repo.go
+65 -5
View File
@@ -4,13 +4,72 @@ import (
"context"
"fmt"
"log/slog"
"strings"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/Silo-Server/silo-server/internal/contentid"
"github.com/Silo-Server/silo-server/internal/models"
)
// IsActionableStaleProviderID reports whether a failed provider ID is still
// part of the item's canonical identity. IDs learned only as secondary
// cross-references are useful negative-cache entries, but they do not mean a
// successfully matched item itself needs repair. Unmatched items remain
// actionable because a rejected path-provided ID may be their only diagnosis.
func IsActionableStaleProviderID(item *models.MediaItem, staleID *models.StaleMediaID) bool {
if item == nil || staleID == nil {
return false
}
providerID := strings.TrimSpace(staleID.ProviderID)
if providerID == "" {
return false
}
provider := strings.ToLower(strings.TrimSpace(staleID.Provider))
if provider != contentid.ProviderTMDB && provider != contentid.ProviderTVDB && provider != contentid.ProviderIMDB {
return false
}
anchorProvider, anchorID, hasAnchor := contentid.ProviderAnchor(item.ContentID)
if hasAnchor && provider == anchorProvider && providerIDValuesEqual(provider, providerID, anchorID) {
return true
}
var currentID string
switch provider {
case contentid.ProviderTMDB:
currentID = item.TmdbID
case contentid.ProviderTVDB:
currentID = item.TvdbID
case contentid.ProviderIMDB:
currentID = item.ImdbID
}
if providerIDValuesEqual(provider, providerID, currentID) {
return true
}
// Unmatched/skeleton items often have only a bad ID parsed from their path;
// it has not yet reached the denormalized provider columns or a provider
// anchor. Those failures are precisely what the admin stale-ID list must
// surface for manual correction.
return !strings.EqualFold(strings.TrimSpace(item.Status), string(MatchOutcomeMatched))
}
func providerIDValuesEqual(provider, left, right string) bool {
left = normalizeProviderIDComparisonValue(provider, left)
right = normalizeProviderIDComparisonValue(provider, right)
if left == "" || right == "" {
return false
}
return left == right
}
func normalizeProviderIDComparisonValue(provider, providerID string) string {
providerID = strings.TrimSpace(providerID)
if strings.EqualFold(strings.TrimSpace(provider), contentid.ProviderIMDB) {
return strings.ToLower(providerID)
}
return providerID
}
// StaleMediaIDRepository persists external IDs that 404 during metadata refresh.
type StaleMediaIDRepository struct {
pool *pgxpool.Pool
@@ -47,14 +106,15 @@ func scanStaleMediaIDs(rows pgx.Rows) ([]*models.StaleMediaID, error) {
return ids, nil
}
// Upsert inserts or updates a stale external ID record.
// Upsert inserts or updates one stale external ID value. A provider can have
// multiple rejected values for the same item; retaining each value prevents a
// later provider response from resurrecting an older known-dead cross-reference.
func (r *StaleMediaIDRepository) Upsert(ctx context.Context, contentID, provider, providerID string) error {
_, err := r.pool.Exec(ctx, `
INSERT INTO stale_media_ids (content_id, provider, provider_id, first_seen_at, last_seen_at)
VALUES ($1, $2, $3, NOW(), NOW())
ON CONFLICT (content_id, provider) DO UPDATE
SET provider_id = EXCLUDED.provider_id,
last_seen_at = NOW()
ON CONFLICT (content_id, provider, provider_id) DO UPDATE
SET last_seen_at = NOW()
`, contentID, provider, providerID)
return resolveStaleUpsertError(err, contentID, provider)
}
@@ -97,7 +157,7 @@ func (r *StaleMediaIDRepository) GetByContentID(ctx context.Context, contentID s
SELECT `+staleMediaIDColumns+`
FROM stale_media_ids
WHERE content_id = $1
ORDER BY provider ASC
ORDER BY provider ASC, provider_id ASC
`, contentID)
if err != nil {
return nil, fmt.Errorf("getting stale media IDs by content_id: %w", err)
internal/metadata/stale_media_id_repo_test.go
+54
View File
@@ -6,8 +6,62 @@ import (
"testing"
"github.com/jackc/pgx/v5/pgconn"
"github.com/Silo-Server/silo-server/internal/contentid"
"github.com/Silo-Server/silo-server/internal/models"
)
func TestIsActionableStaleProviderID(t *testing.T) {
const (
imdbID = "tt0133093"
tmdbID = "603"
tvdbID = "81189"
unknownProvider = "other"
unmatchedStatus = "unmatched"
)
item := &models.MediaItem{Status: string(MatchOutcomeMatched), TmdbID: tmdbID, TvdbID: tvdbID, ImdbID: imdbID}
tests := []struct {
name string
stale *models.StaleMediaID
want bool
}{
{name: "current tmdb", stale: &models.StaleMediaID{Provider: contentid.ProviderTMDB, ProviderID: tmdbID}, want: true},
{name: "foreign tmdb cross reference", stale: &models.StaleMediaID{Provider: contentid.ProviderTMDB, ProviderID: "12236904"}},
{name: "current tvdb", stale: &models.StaleMediaID{Provider: "TVDB", ProviderID: tvdbID}, want: true},
{name: "current imdb normalized case", stale: &models.StaleMediaID{Provider: contentid.ProviderIMDB, ProviderID: "TT0133093"}, want: true},
{name: "unknown provider", stale: &models.StaleMediaID{Provider: unknownProvider, ProviderID: tmdbID}},
{name: "nil stale"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if got := IsActionableStaleProviderID(item, tt.stale); got != tt.want {
t.Fatalf("IsActionableStaleProviderID() = %v, want %v", got, tt.want)
}
})
}
anchorOnly := &models.MediaItem{ContentID: "series-tvdb-" + tvdbID, Status: string(MatchOutcomeMatched)}
if !IsActionableStaleProviderID(anchorOnly, &models.StaleMediaID{
Provider: contentid.ProviderTVDB, ProviderID: tvdbID,
}) {
t.Fatal("provider-anchored content ID must remain actionable when the denormalized ID column is empty")
}
imdbAnchorOnly := &models.MediaItem{ContentID: "movie-imdb-" + imdbID, Status: string(MatchOutcomeMatched)}
if !IsActionableStaleProviderID(imdbAnchorOnly, &models.StaleMediaID{
Provider: contentid.ProviderIMDB, ProviderID: "TT0133093",
}) {
t.Fatal("IMDb anchor comparison must be case-insensitive")
}
unmatchedLocal := &models.MediaItem{ContentID: "local-deadbeef", Status: unmatchedStatus}
if !IsActionableStaleProviderID(unmatchedLocal, &models.StaleMediaID{
Provider: contentid.ProviderTMDB, ProviderID: "999",
}) {
t.Fatal("bad path ID on an unmatched local item must remain actionable")
}
}
func TestIsMissingContentForeignKeyViolation(t *testing.T) {
cases := []struct {
name string
internal/metadata/target_refresh_test.go
+44 -1
View File
@@ -6,21 +6,33 @@ import (
"testing"
"time"
"github.com/Silo-Server/silo-server/internal/contentid"
"github.com/Silo-Server/silo-server/internal/models"
)
type targetRefreshProvider struct {
slug string
searchResults []SearchResult
seasons []SeasonResult
episodesBySeason map[int][]EpisodeResult
episodeRequests []int
}
func (p *targetRefreshProvider) Slug() string { return "target-refresh" }
func (p *targetRefreshProvider) Slug() string {
if p.slug != "" {
return p.slug
}
return "target-refresh"
}
func (p *targetRefreshProvider) Name() string { return "target-refresh" }
func (p *targetRefreshProvider) ForTypes() []string { return []string{"series"} }
func (p *targetRefreshProvider) Search(context.Context, SearchQuery) ([]SearchResult, error) {
return append([]SearchResult(nil), p.searchResults...), nil
}
func (p *targetRefreshProvider) GetSeasons(context.Context, SeasonsRequest) ([]SeasonResult, error) {
return append([]SeasonResult(nil), p.seasons...), nil
}
@@ -80,6 +92,37 @@ func seedTargetRefreshHarness(provider *targetRefreshProvider) (*testHarness, st
return h, seriesID, seasonID, episodeID
}
func TestResolveSeriesRefreshProviderIDsUsesStaleSuppressedSelectionIdentity(t *testing.T) {
const (
staleTMDBID = "12345"
replacementTMDBID = "67890"
)
provider := &targetRefreshProvider{
slug: contentid.ProviderTMDB,
searchResults: []SearchResult{{
Name: "Target Refresh Show", Year: 2020, Provider: contentid.ProviderTMDB,
ProviderIDs: map[string]string{contentid.ProviderTMDB: replacementTMDBID},
}},
}
h, seriesID, _, _ := seedTargetRefreshHarness(provider)
series := h.itemRepo.items[seriesID]
series.Year = 2020
series.TmdbID = staleTMDBID
staleRepo := newFakeStaleIDRepo()
staleRepo.set(seriesID, &models.StaleMediaID{
ContentID: seriesID, Provider: contentid.ProviderTMDB, ProviderID: staleTMDBID,
})
h.service.staleIDRepo = staleRepo
providerIDs, err := h.service.resolveSeriesRefreshProviderIDs(context.Background(), series, 0, "en")
if err != nil {
t.Fatalf("resolveSeriesRefreshProviderIDs: %v", err)
}
if providerIDs[contentid.ProviderTMDB] != replacementTMDBID {
t.Fatalf("resolved tmdb id = %q, want %s", providerIDs[contentid.ProviderTMDB], replacementTMDBID)
}
}
func TestRefreshScheduledTargetEpisodePersistsOnlySelectedEpisode(t *testing.T) {
provider := &targetRefreshProvider{
seasons: []SeasonResult{{
internal/metadata/types.go
+17 -5
View File
@@ -70,7 +70,8 @@ type ProcessRequest struct {
// folder-scoped manual refreshes when the library's configured
// language differs from the item's stamp, so changing a library's
// metadata language actually re-fetches titles/overviews.
AdoptLanguage bool
AdoptLanguage bool
recordedStaleProviderIDs providerIDValueSet
}
// ProcessResult is the output of MetadataService.Process().
@@ -258,11 +259,22 @@ type MetadataResult struct {
// titleAliasProviders records whether each contributing provider declared
// its aliases to be an authoritative snapshot. False means merge-only,
// which is the safe default for legacy plugins and partial responses.
titleAliasProviders map[string]bool
titleAliasProviders map[string]bool
// quarantinedProviderIDKeys prevents unresolved conflicting IDs from being
// reintroduced by later provider details or durable merge.
quarantinedProviderIDKeys map[string]struct{}
ContentRating string
Ratings Ratings
People []models.ItemPerson
// replacedProviderIDKeys forces an owning-provider consensus replacement to
// overwrite the corresponding stored provider-ID column during merge.
replacedProviderIDKeys map[string]struct{}
// recordedStaleProviderIDs contains provider values known dead before this
// refresh and suppresses them when durable state is merged.
recordedStaleProviderIDs providerIDValueSet
// sameRunStaleProviderIDs contains identity values that 404ed during this
// refresh and prevents them from being persisted by the same operation.
sameRunStaleProviderIDs providerIDValueSet
ContentRating string
Ratings Ratings
People []models.ItemPerson
// Images (S3 paths or URLs).
PosterPath string
PosterThumbhash string
migrations/sql/20260726090943_allow_multiple_stale_provider_ids.sql
+85
View File
@@ -0,0 +1,85 @@
-- +goose NO TRANSACTION
-- +goose Up
-- A provider can reject more than one value for the same item, so the negative
-- cache keys on the value as well. Build the wider key concurrently: a plain
-- ADD PRIMARY KEY holds ACCESS EXCLUSIVE for the whole index build, blocking
-- reads and writes on a table the refresh path touches constantly. All three
-- key columns are already NOT NULL, so attaching the finished index is a
-- metadata-only step.
-- Remove INVALID remnants first: IF NOT EXISTS otherwise accepts a failed
-- concurrent build and Goose would record a primary key that cannot be used.
-- +goose StatementBegin
DO $$
BEGIN
IF EXISTS (
SELECT 1
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN pg_index i ON i.indexrelid = c.oid
WHERE n.nspname = 'public'
AND c.relname = 'stale_media_ids_pkey_new'
AND NOT i.indisvalid
) THEN
DROP INDEX public.stale_media_ids_pkey_new;
END IF;
END;
$$;
-- +goose StatementEnd
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS stale_media_ids_pkey_new
ON public.stale_media_ids (content_id, provider, provider_id);
-- Between these two statements the table has no primary key, but
-- stale_media_ids_pkey_new already enforces uniqueness on the new key, so only
-- the newly permitted duplicate (content_id, provider) pairs can appear.
ALTER TABLE stale_media_ids
DROP CONSTRAINT IF EXISTS stale_media_ids_pkey;
ALTER TABLE stale_media_ids
ADD CONSTRAINT stale_media_ids_pkey PRIMARY KEY USING INDEX stale_media_ids_pkey_new;
-- +goose Down
-- Collapse back to one row per (content_id, provider), keeping the most
-- recently rejected value — what the single-value schema would have held.
-- Run this with writers stopped: without a surrounding transaction a concurrent
-- insert between the dedup and the index build fails the build.
WITH ranked AS (
SELECT ctid,
ROW_NUMBER() OVER (
PARTITION BY content_id, provider
ORDER BY last_seen_at DESC, first_seen_at ASC, provider_id ASC
) AS row_number
FROM stale_media_ids
)
DELETE FROM stale_media_ids
USING ranked
WHERE stale_media_ids.ctid = ranked.ctid
AND ranked.row_number > 1;
-- +goose StatementBegin
DO $$
BEGIN
IF EXISTS (
SELECT 1
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN pg_index i ON i.indexrelid = c.oid
WHERE n.nspname = 'public'
AND c.relname = 'stale_media_ids_pkey_old'
AND NOT i.indisvalid
) THEN
DROP INDEX public.stale_media_ids_pkey_old;
END IF;
END;
$$;
-- +goose StatementEnd
CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS stale_media_ids_pkey_old
ON public.stale_media_ids (content_id, provider);
ALTER TABLE stale_media_ids
DROP CONSTRAINT IF EXISTS stale_media_ids_pkey;
ALTER TABLE stale_media_ids
ADD CONSTRAINT stale_media_ids_pkey PRIMARY KEY USING INDEX stale_media_ids_pkey_old;