fix(metadata): merge duplicate people on refresh instead of looping on 23505 (#250)
* fix(metadata): merge duplicate people on refresh instead of looping on 23505 The background person-refresh worker re-selected the same people every cycle. When a refresh resolved an external id (tmdb_id/imdb_id) already held by another people row, the UPDATE violated a partial-unique index and raised SQLSTATE 23505; the tx rolled back so updated_at never advanced and FindCandidates re-qualified the row forever. The underlying cause is two people rows for the same human created from credits ingested with disjoint id sets (one tmdb-only, one imdb-only) that BatchFindOrCreate never reconciled. PersonRepository.Update now reconciles the collision instead of failing: - The common no-collision path stays a single plain transaction (no added cost). - On a 23505, updateResolvingConflicts runs the whole reconciliation in ONE transaction, retrying the write via savepoints so it commits atomically: it locks both rows FOR UPDATE in id order, then either merges the partner into the survivor (repoint item_people skipping duplicate credits, fold the partner's ids/fields onto the survivor, delete the partner) or, when the rows are not confidently the same human, drops just the conflicting id. - canMergePeople requires compatible ids AND matching names, so a provider that hands the same id to two genuinely different people cannot trigger a destructive delete; that case falls to the non-destructive drop. - A row merged away concurrently surfaces as pgx.ErrNoRows, which the refresh service maps to ErrPersonNotFound. Existing stuck rows self-heal: they are still re-selected each cycle and now merge (or drop) instead of looping, draining the warning population to zero. Adds unit tests for the merge-decision logic (guard, compatibility, field-folding, constraint mapping). AI-use disclosure: implemented and adversarially reviewed with AI assistance. * fix(metadata): preserve survivor's existing id when declining a person merge The non-mergeable branch of resolveExternalIDConflict blanked the conflicting external-id field before retrying the write. Because execPersonUpdate is a full-row UPDATE, the retry persisted an empty string and returned success, silently dropping a previously-valid provider id (e.g. the admin PATCH path that mutates an existing id into a colliding value) instead of leaving the row unchanged as the 23505 did. Restore the locked survivor's currently-persisted value for the field so the retried write is a no-op on that column: it commits without looping, without deleting a possibly-distinct person, and without blanking an id the survivor already held. Writing a row's own current value back can never violate the unique index, so the field will not re-trigger the conflict. The refresh-worker path is unchanged (survivor value is empty). Convert clearExternalIDField into a general setExternalIDField setter and extend its unit test to cover set-to-value and set-to-empty.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user