diff --git a/internal/catalog/person_merge_test.go b/internal/catalog/person_merge_test.go new file mode 100644 index 00000000..418b3c00 --- /dev/null +++ b/internal/catalog/person_merge_test.go @@ -0,0 +1,230 @@ +package catalog + +import ( + "testing" + "time" + + "github.com/Silo-Server/silo-server/internal/models" +) + +func TestConflictingExternalID(t *testing.T) { + p := models.Person{TmdbID: "111", ImdbID: "nm222"} + + tests := []struct { + name string + constraint string + wantField string + wantValue string + wantOK bool + }{ + {"tmdb", "idx_people_tmdb_id", "tmdb_id", "111", true}, + {"imdb", "idx_people_imdb_id", "imdb_id", "nm222", true}, + {"unknown constraint", "idx_people_name", "", "", false}, + {"empty constraint", "", "", "", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + field, value, ok := conflictingExternalID(tt.constraint, p) + if field != tt.wantField || value != tt.wantValue || ok != tt.wantOK { + t.Fatalf("conflictingExternalID(%q) = (%q,%q,%v), want (%q,%q,%v)", + tt.constraint, field, value, ok, tt.wantField, tt.wantValue, tt.wantOK) + } + }) + } + + // A blank id must never be reported as the conflict source. + if _, _, ok := conflictingExternalID("idx_people_tmdb_id", models.Person{}); ok { + t.Fatal("expected ok=false when the conflicting field is empty") + } +} + +func TestExternalIDsCompatible(t *testing.T) { + tests := []struct { + name string + a, b models.Person + want bool + }{ + { + name: "tmdb-only vs imdb-only (the core dedup case)", + a: models.Person{TmdbID: "111", ImdbID: "nm9"}, + b: models.Person{ImdbID: "nm9"}, + want: true, + }, + { + name: "identical ids", + a: models.Person{TmdbID: "111", ImdbID: "nm9"}, + b: models.Person{TmdbID: "111", ImdbID: "nm9"}, + want: true, + }, + { + name: "conflicting tmdb ids", + a: models.Person{TmdbID: "111", ImdbID: "nm9"}, + b: models.Person{TmdbID: "999", ImdbID: "nm9"}, + want: false, + }, + { + name: "conflicting tvdb ids", + a: models.Person{ImdbID: "nm9", TvdbID: "5"}, + b: models.Person{ImdbID: "nm9", TvdbID: "6"}, + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := externalIDsCompatible(tt.a, tt.b); got != tt.want { + t.Fatalf("externalIDsCompatible = %v, want %v", got, tt.want) + } + }) + } +} + +func TestCanMergePeople(t *testing.T) { + tests := []struct { + name string + a, b models.Person + want bool + }{ + { + name: "same name, compatible ids -> merge", + a: models.Person{Name: "Jane Doe", TmdbID: "111"}, + b: models.Person{Name: "Jane Doe", ImdbID: "nm9"}, + want: true, + }, + { + name: "case/space-insensitive name match -> merge", + a: models.Person{Name: " jane doe ", TmdbID: "111"}, + b: models.Person{Name: "Jane Doe", ImdbID: "nm9"}, + want: true, + }, + { + name: "different names, compatible ids -> do not merge", + a: models.Person{Name: "Jane Doe", TmdbID: "111"}, + b: models.Person{Name: "John Smith", ImdbID: "nm9"}, + want: false, + }, + { + name: "same name but contradictory ids -> do not merge", + a: models.Person{Name: "Jane Doe", TmdbID: "111"}, + b: models.Person{Name: "Jane Doe", TmdbID: "999"}, + want: false, + }, + { + name: "empty name on one side -> do not merge", + a: models.Person{Name: "", TmdbID: "111"}, + b: models.Person{Name: "Jane Doe", ImdbID: "nm9"}, + want: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := canMergePeople(tt.a, tt.b); got != tt.want { + t.Fatalf("canMergePeople = %v, want %v", got, tt.want) + } + }) + } +} + +func TestMergePersonFields(t *testing.T) { + birth := time.Date(1980, 1, 2, 0, 0, 0, 0, time.UTC) + + t.Run("backfills only empty fields, survivor wins", func(t *testing.T) { + dst := models.Person{ + Name: "Jane Doe", + TmdbID: "111", + Bio: "survivor bio", + } + src := models.Person{ + Name: "Janey", + TmdbID: "999", // must NOT overwrite a populated survivor id + ImdbID: "nm9", // fills the gap + Bio: "partner bio", + Birthplace: "Springfield", + BirthDate: &birth, + PhotoPath: "/p.jpg", + } + mergePersonFields(&dst, src) + + if dst.Name != "Jane Doe" { + t.Errorf("Name overwritten: %q", dst.Name) + } + if dst.TmdbID != "111" { + t.Errorf("populated TmdbID overwritten: %q", dst.TmdbID) + } + if dst.ImdbID != "nm9" { + t.Errorf("empty ImdbID not backfilled: %q", dst.ImdbID) + } + if dst.Bio != "survivor bio" { + t.Errorf("populated Bio overwritten: %q", dst.Bio) + } + if dst.Birthplace != "Springfield" { + t.Errorf("empty Birthplace not backfilled: %q", dst.Birthplace) + } + if dst.BirthDate == nil || !dst.BirthDate.Equal(birth) { + t.Errorf("nil BirthDate not backfilled: %v", dst.BirthDate) + } + if dst.PhotoPath != "/p.jpg" { + t.Errorf("empty PhotoPath not backfilled: %q", dst.PhotoPath) + } + }) + + t.Run("backfills name when survivor is blank", func(t *testing.T) { + dst := models.Person{TmdbID: "111"} + src := models.Person{Name: "Jane Doe"} + mergePersonFields(&dst, src) + if dst.Name != "Jane Doe" { + t.Errorf("blank survivor Name not backfilled: %q", dst.Name) + } + }) + + t.Run("photo trio is folded together", func(t *testing.T) { + dst := models.Person{} // no photo at all + src := models.Person{ + PhotoPath: "/p.jpg", + PhotoSourcePath: "tmdb://p", + PhotoThumbhash: "hash", + } + mergePersonFields(&dst, src) + if dst.PhotoPath != "/p.jpg" || dst.PhotoSourcePath != "tmdb://p" || dst.PhotoThumbhash != "hash" { + t.Errorf("photo trio not folded together: %+v", dst) + } + + // A survivor that already has a photo keeps its own trio intact. + keep := models.Person{PhotoPath: "/keep.jpg"} + mergePersonFields(&keep, src) + if keep.PhotoPath != "/keep.jpg" || keep.PhotoSourcePath != "" || keep.PhotoThumbhash != "" { + t.Errorf("survivor photo trio was disturbed: %+v", keep) + } + }) +} + +func TestExternalIDValueAndSet(t *testing.T) { + p := models.Person{TmdbID: "111", ImdbID: "nm9", TvdbID: "5", PlexGUID: "g"} + + for field, want := range map[string]string{ + "tmdb_id": "111", "imdb_id": "nm9", "tvdb_id": "5", "plex_guid": "g", "nope": "", + } { + if got := externalIDValue(p, field); got != want { + t.Errorf("externalIDValue(%q) = %q, want %q", field, got, want) + } + } + + // Setting a field to a new value replaces only that column. + updated := p + setExternalIDField(&updated, "imdb_id", "nm12345") + if updated.ImdbID != "nm12345" { + t.Errorf("setExternalIDField did not set imdb_id: %q", updated.ImdbID) + } + if updated.TmdbID != "111" { + t.Errorf("setExternalIDField touched the wrong field: %q", updated.TmdbID) + } + + // Setting a field to "" blanks it, matching the old clear behavior. + cleared := p + setExternalIDField(&cleared, "imdb_id", "") + if cleared.ImdbID != "" { + t.Errorf("setExternalIDField did not blank imdb_id: %q", cleared.ImdbID) + } + if cleared.TmdbID != "111" { + t.Errorf("setExternalIDField touched the wrong field: %q", cleared.TmdbID) + } +} diff --git a/internal/catalog/person_repo.go b/internal/catalog/person_repo.go index 73538785..dbd46ece 100644 --- a/internal/catalog/person_repo.go +++ b/internal/catalog/person_repo.go @@ -2,7 +2,9 @@ package catalog import ( "context" + "errors" "fmt" + "log/slog" "strconv" "strings" "time" @@ -528,15 +530,114 @@ func (r *PersonRepository) Search(ctx context.Context, query string, limit int) return people, rows.Err() } +// maxPersonConflictResolutions bounds how many external-id collisions a single +// Update is allowed to resolve before giving up. One per unique external-id +// field is the natural ceiling (today only tmdb_id and imdb_id are unique). +const maxPersonConflictResolutions = 4 + // Update updates all non-key fields on a person. +// +// A person's resolved external ids (tmdb_id / imdb_id) can collide with another +// row that turns out to be the same human discovered through a different id — +// e.g. one credit ingested with only a tmdb_id and another with only an imdb_id, +// later reconciled when a refresh fills in the missing cross-id. Rather than +// failing that write (SQLSTATE 23505) and letting the refresh worker retry the +// same row forever, Update detects the collision and resolves it: it merges the +// two people when they are the same human, or drops just the conflicting id when +// they are not, then retries the write so updated_at advances and the row leaves +// the stale set. +// +// The common, no-collision path runs as a single plain transaction. Only when a +// 23505 surfaces does Update fall back to updateResolvingConflicts, which does +// the whole reconciliation — merge or drop, plus the field write and reindex — +// inside one transaction so it commits atomically (or not at all). Returns +// pgx.ErrNoRows if the row no longer exists (e.g. merged away concurrently). func (r *PersonRepository) Update(ctx context.Context, p models.Person) error { + err := r.applyUpdate(ctx, p) + if err == nil || !isDuplicateKeyError(err) { + return err + } + return r.updateResolvingConflicts(ctx, p) +} + +// applyUpdate writes all non-key fields on a person and enqueues a search +// reindex for the items they appear in, all in one transaction. +func (r *PersonRepository) applyUpdate(ctx context.Context, p models.Person) error { tx, err := r.pool.Begin(ctx) if err != nil { return fmt.Errorf("begin person update tx: %w", err) } defer func() { _ = tx.Rollback(ctx) }() - _, err = tx.Exec(ctx, ` + affected, err := execPersonUpdate(ctx, tx, p) + if err != nil { + return err + } + if affected == 0 { + // The row was deleted out from under us (e.g. merged away concurrently). + return pgx.ErrNoRows + } + if err := reindexPersonItems(ctx, tx, p.ID); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit person update tx: %w", err) + } + return nil +} + +// updateResolvingConflicts performs the person write inside a single transaction, +// reconciling external-id collisions (merge or drop) as they arise. The write is +// retried after each resolution via a savepoint so a failed attempt does not +// poison the transaction; the whole thing commits atomically once the write +// lands. Bounded by maxPersonConflictResolutions so it cannot spin. +func (r *PersonRepository) updateResolvingConflicts(ctx context.Context, p models.Person) error { + tx, err := r.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin person merge tx: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + var affected int64 + for attempt := 0; ; attempt++ { + affected, err = tryPersonUpdate(ctx, tx, p) + if err == nil { + break + } + if !isDuplicateKeyError(err) || attempt >= maxPersonConflictResolutions { + return err + } + field, value, ok := conflictingExternalID(extractConstraint(err), p) + if !ok { + // Unique violation on a constraint we do not know how to reconcile. + return err + } + resolved, resolveErr := r.resolveExternalIDConflict(ctx, tx, &p, field, value) + if resolveErr != nil { + return resolveErr + } + if !resolved { + return err + } + } + + if affected == 0 { + // Survivor was deleted concurrently (e.g. merged into another row); there + // is nothing left to persist for this id. + return pgx.ErrNoRows + } + if err := reindexPersonItems(ctx, tx, p.ID); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit person merge tx: %w", err) + } + return nil +} + +// execPersonUpdate writes all non-key person fields and returns the rows affected. +func execPersonUpdate(ctx context.Context, tx pgx.Tx, p models.Person) (int64, error) { + tag, err := tx.Exec(ctx, ` UPDATE people SET name=$2, sort_name=$3, bio=$4, birth_date=$5, death_date=$6, birthplace=$7, homepage=$8, photo_path=$9, photo_source_path=$10, photo_thumbhash=$11, tmdb_id=$12, imdb_id=$13, tvdb_id=$14, plex_guid=$15, updated_at=now() @@ -546,30 +647,299 @@ func (r *PersonRepository) Update(ctx context.Context, p models.Person) error { p.TmdbID, p.ImdbID, p.TvdbID, p.PlexGUID, ) if err != nil { - return err + return 0, err } + return tag.RowsAffected(), nil +} +// tryPersonUpdate runs execPersonUpdate inside a savepoint so a unique violation +// rolls back only the attempt, leaving the surrounding transaction usable. +func tryPersonUpdate(ctx context.Context, tx pgx.Tx, p models.Person) (int64, error) { + sp, err := tx.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("begin person update savepoint: %w", err) + } + affected, err := execPersonUpdate(ctx, sp, p) + if err != nil { + _ = sp.Rollback(ctx) + return 0, err + } + if err := sp.Commit(ctx); err != nil { + return 0, fmt.Errorf("release person update savepoint: %w", err) + } + return affected, nil +} + +// reindexPersonItems enqueues a catalog-search reindex for every item the person +// appears in. After a merge the survivor's link set already includes the merged +// person's items, so this covers all items whose denormalized people text changed. +func reindexPersonItems(ctx context.Context, tx pgx.Tx, personID int64) error { rows, err := tx.Query(ctx, ` SELECT DISTINCT content_id FROM item_people WHERE person_id = $1 - `, p.ID) + `, personID) if err != nil { - return fmt.Errorf("listing items linked to person %d: %w", p.ID, err) + return fmt.Errorf("listing items linked to person %d: %w", personID, err) } contentIDs, err := pgx.CollectRows(rows, pgx.RowTo[string]) if err != nil { - return fmt.Errorf("collecting items linked to person %d: %w", p.ID, err) + return fmt.Errorf("collecting items linked to person %d: %w", personID, err) } if err := EnqueueSearchIndexUpserts(ctx, tx, contentIDs); err != nil { - return fmt.Errorf("enqueueing catalog search person update for person %d: %w", p.ID, err) - } - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("commit person update tx: %w", err) + return fmt.Errorf("enqueueing catalog search person update for person %d: %w", personID, err) } return nil } +// resolveExternalIDConflict reconciles a unique-violation on (field=value) raised +// while updating *p, operating within the caller's transaction tx. It mutates *p +// in place (folding in the partner's data, or dropping the conflicting id) and +// reports whether the caller should retry the write. resolved=false means the +// conflict could not be reconciled and the original error should be surfaced. +func (r *PersonRepository) resolveExternalIDConflict( + ctx context.Context, + tx pgx.Tx, + p *models.Person, + field string, + value string, +) (resolved bool, err error) { + if value == "" { + return false, nil + } + + // Identify the row that currently owns the colliding id. + var partnerID int64 + lookup := fmt.Sprintf("SELECT id FROM people WHERE %s = $1", field) + switch scanErr := tx.QueryRow(ctx, lookup, value).Scan(&partnerID); { + case errors.Is(scanErr, pgx.ErrNoRows): + // The conflicting owner vanished (concurrent merge); retry the plain write. + return true, nil + case scanErr != nil: + return false, fmt.Errorf("locate person owning %s=%q: %w", field, value, scanErr) + } + if partnerID == p.ID { + // Self-conflict is impossible under the unique index; bail rather than loop. + return false, nil + } + + // Lock both rows in a deterministic order so concurrent merges cannot deadlock. + people, err := scanPeopleByIDsForUpdate(ctx, tx, p.ID, partnerID) + if err != nil { + return false, err + } + if _, ok := people[p.ID]; !ok { + // The survivor was deleted under us (concurrent merge); retry the plain + // write, which will affect zero rows and surface as pgx.ErrNoRows. + return true, nil + } + partner, ok := people[partnerID] + if !ok { + // Partner already gone; retry the plain write. + return true, nil + } + // Re-confirm the partner still holds the colliding value after locking. + if externalIDValue(partner, field) != value { + return true, nil + } + + if !canMergePeople(*p, partner) { + // Not confidently the same human (contradictory ids, or names disagree): + // restore the survivor's currently-persisted value for this field so the + // retried write becomes a no-op on that column. This lets the write commit + // instead of looping forever, without destructively deleting a + // possibly-distinct person and without blanking an id the survivor already + // held (blanking would silently drop a previously-valid provider id, e.g. + // on the admin PATCH path that mutates an existing id into a colliding one). + // Writing the row's own current value back can never violate the unique + // index, so this field will not re-trigger the conflict. + setExternalIDField(p, field, externalIDValue(people[p.ID], field)) + slog.Warn("person merge: declining to merge, preserving survivor's existing id", + "person_id", p.ID, "partner_id", partnerID, "field", field, "value", value, + "person_name", p.Name, "partner_name", partner.Name) + return true, nil + } + + // Fold the partner's ids and any metadata this write lacks onto the survivor, + // so the retried write persists data only the partner had. + mergePersonFields(p, partner) + + // Repoint the partner's credits onto the survivor, skipping links that would + // duplicate an existing (content_id, kind, character) credit on the survivor. + if _, err := tx.Exec(ctx, ` + UPDATE item_people ip SET person_id = $1 + WHERE ip.person_id = $2 + AND NOT EXISTS ( + SELECT 1 FROM item_people e + WHERE e.person_id = $1 + AND e.content_id = ip.content_id + AND e.kind = ip.kind + AND e.character = ip.character + )`, p.ID, partnerID); err != nil { + return false, fmt.Errorf("repoint item_people from %d to %d: %w", partnerID, p.ID, err) + } + + // Delete the partner; any leftover duplicate credits cascade away. + if _, err := tx.Exec(ctx, `DELETE FROM people WHERE id = $1`, partnerID); err != nil { + return false, fmt.Errorf("delete merged person %d: %w", partnerID, err) + } + + slog.Info("person merge: folded duplicate person", + "survivor_id", p.ID, "merged_id", partnerID, "field", field, "value", value) + return true, nil +} + +// scanPeopleByIDsForUpdate loads the given people, locking them FOR UPDATE in +// ascending id order so concurrent merges acquire locks consistently. +func scanPeopleByIDsForUpdate(ctx context.Context, tx pgx.Tx, ids ...int64) (map[int64]models.Person, error) { + rows, err := tx.Query(ctx, ` + SELECT id, name, sort_name, bio, birth_date, death_date, birthplace, homepage, + photo_path, photo_source_path, photo_thumbhash, tmdb_id, imdb_id, tvdb_id, plex_guid, created_at, updated_at + FROM people + WHERE id = ANY($1) + ORDER BY id + FOR UPDATE`, ids) + if err != nil { + return nil, fmt.Errorf("lock people for merge: %w", err) + } + defer rows.Close() + + out := make(map[int64]models.Person, len(ids)) + for rows.Next() { + var p models.Person + if err := rows.Scan(&p.ID, &p.Name, &p.SortName, &p.Bio, &p.BirthDate, &p.DeathDate, &p.Birthplace, &p.Homepage, + &p.PhotoPath, &p.PhotoSourcePath, &p.PhotoThumbhash, &p.TmdbID, &p.ImdbID, &p.TvdbID, &p.PlexGUID, &p.CreatedAt, &p.UpdatedAt, + ); err != nil { + return nil, fmt.Errorf("scan locked person: %w", err) + } + out[p.ID] = p + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate locked people: %w", err) + } + return out, nil +} + +// conflictingExternalID maps a violated unique constraint to the person field +// and value that caused it. ok is false for constraints Update cannot reconcile. +func conflictingExternalID(constraint string, p models.Person) (field, value string, ok bool) { + switch constraint { + case "idx_people_tmdb_id": + return "tmdb_id", p.TmdbID, p.TmdbID != "" + case "idx_people_imdb_id": + return "imdb_id", p.ImdbID, p.ImdbID != "" + default: + return "", "", false + } +} + +// externalIDValue returns the value of the named external-id column on p. +func externalIDValue(p models.Person, field string) string { + switch field { + case "tmdb_id": + return p.TmdbID + case "imdb_id": + return p.ImdbID + case "tvdb_id": + return p.TvdbID + case "plex_guid": + return p.PlexGUID + default: + return "" + } +} + +// setExternalIDField sets the named external-id column on p to value. +func setExternalIDField(p *models.Person, field, value string) { + switch field { + case "tmdb_id": + p.TmdbID = value + case "imdb_id": + p.ImdbID = value + case "tvdb_id": + p.TvdbID = value + case "plex_guid": + p.PlexGUID = value + } +} + +// externalIDsCompatible reports whether a and b can be the same person: no +// external-id field where both hold a different non-empty value. A field where +// one side is empty (or both match) is compatible. +func externalIDsCompatible(a, b models.Person) bool { + return idsCompatible(a.TmdbID, b.TmdbID) && + idsCompatible(a.ImdbID, b.ImdbID) && + idsCompatible(a.TvdbID, b.TvdbID) && + idsCompatible(a.PlexGUID, b.PlexGUID) +} + +func idsCompatible(x, y string) bool { + return x == "" || y == "" || x == y +} + +// canMergePeople reports whether two rows that collided on a shared external id +// are confidently the same human and may be folded together. A shared id is a +// strong signal, but providers occasionally hand the same id to two distinct +// people; requiring the names to agree as well guards against destructively +// deleting a genuinely different person. When this returns false the caller +// drops the conflicting id instead (non-destructive). +func canMergePeople(a, b models.Person) bool { + return externalIDsCompatible(a, b) && personNamesMatch(a.Name, b.Name) +} + +func personNamesMatch(a, b string) bool { + a = strings.TrimSpace(a) + b = strings.TrimSpace(b) + if a == "" || b == "" { + return false + } + return strings.EqualFold(a, b) +} + +// mergePersonFields fills empty external ids and metadata on dst from src, +// preserving dst's existing values. dst is the refreshed survivor, so its +// populated fields win; src (the row being merged away) only backfills gaps. +func mergePersonFields(dst *models.Person, src models.Person) { + if dst.Name == "" { + dst.Name = src.Name + } + if dst.TmdbID == "" { + dst.TmdbID = src.TmdbID + } + if dst.ImdbID == "" { + dst.ImdbID = src.ImdbID + } + if dst.TvdbID == "" { + dst.TvdbID = src.TvdbID + } + if dst.PlexGUID == "" { + dst.PlexGUID = src.PlexGUID + } + if dst.Bio == "" { + dst.Bio = src.Bio + } + if dst.Birthplace == "" { + dst.Birthplace = src.Birthplace + } + if dst.Homepage == "" { + dst.Homepage = src.Homepage + } + if dst.SortName == "" { + dst.SortName = src.SortName + } + if dst.BirthDate == nil { + dst.BirthDate = src.BirthDate + } + if dst.DeathDate == nil { + dst.DeathDate = src.DeathDate + } + if dst.PhotoPath == "" { + dst.PhotoPath = src.PhotoPath + dst.PhotoSourcePath = src.PhotoSourcePath + dst.PhotoThumbhash = src.PhotoThumbhash + } +} + func (r *PersonRepository) UpdatePhotoIfSourceMatches(ctx context.Context, personID int64, sourcePath, cachedPath, thumbhash string) (bool, error) { if r == nil || r.pool == nil { return false, pgx.ErrNoRows diff --git a/internal/metadata/person_refresh.go b/internal/metadata/person_refresh.go index 813a8a24..b17b0cf7 100644 --- a/internal/metadata/person_refresh.go +++ b/internal/metadata/person_refresh.go @@ -167,6 +167,11 @@ func (s *PersonRefreshService) refreshPersonWithProviders( } if err := s.repo.Update(ctx, refreshed); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + // The row was merged into another person concurrently; there is no + // longer anything to refresh under this id. + return nil, ErrPersonNotFound + } return nil, fmt.Errorf("update person %d: %w", id, err) } s.enqueuePersonPhoto(ctx, refreshed, accumulator.ProviderIDs, photoProviderID)