fix(plex): make cast enrichment insert conflict-safe

A full cold sync crashed and left plex_people empty (cast/people missing in
every detail view). enrich_row_people added PlexPerson rows via the ORM with no
ON CONFLICT, and because SessionLocal is autoflush=False a person shared by two
items in the same enrich batch was queued twice -> the single batch commit
violated the unique index ix_plex_people_name and rolled the whole batch back.
Prod dodged it by building up incrementally; dev hits it on every reseed+sync.

Insert missing people via an immediately-executed pg_insert with
on_conflict_do_nothing (deduped within the cast by name, keeping the first
non-null thumb), mirroring app/sync/videos.py:_insert_stubs. The immediate
execute also lets a later row in the batch see the row via the existing-names
query and skip it; the conflict guard additionally covers the concurrent
info-page enrichment path.

Adds a DB-backed regression test: two items with overlapping cast now enrich
without a unique-violation, one row per name, and a prior-NULL headshot is
back-filled. Mutation-checked (reverting the fix reddens it).

backend/app/plex/people.py

@@ -6,6 +6,7 @@ re-sync (plex.sync), so the two paths never diverge."""
 from urllib.parse import quote
 
 from sqlalchemy import text
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
 from app.models import PlexPerson
@@ -55,19 +56,31 @@ def enrich_row_people(db: Session, row, cast: list[dict]) -> bool:
     # Headshot cache: upsert name -> proxied thumb (a later fetch fills a prior NULL; names without a
     # whitelisted photo still get a row so a card can show a placeholder).
     existing = {p.name: p for p in db.query(PlexPerson).filter(PlexPerson.name.in_(names))}
+    missing: dict[str, str | None] = {}
     for c in cast:
         nm, th = c.get("name"), c.get("thumb")
         if not nm:
             continue
         p = existing.get(nm)
         if p is None:
-            p = PlexPerson(name=nm, photo_thumb=th)
-            db.add(p)
-            existing[nm] = p
-            dirty = True
+            # Not yet cached — queue an insert, deduped by name (keep the first non-null thumb) so a
+            # single INSERT statement can't hit the same conflict target twice.
+            missing[nm] = missing.get(nm) or th
         elif th and not p.photo_thumb:
             p.photo_thumb = th
             dirty = True
+    if missing:
+        # on_conflict_do_nothing keeps a cold sync from crashing when one person appears across two
+        # items in the same batch: SessionLocal is autoflush=False, so the ORM query above can't see a
+        # same-batch pending insert, and this path also races the concurrent info-page enrichment. The
+        # Core execute runs immediately (not deferred to the caller's commit), so a later row in the
+        # batch sees the row and skips it. Mirrors app/sync/videos.py:_insert_stubs.
+        db.execute(
+            pg_insert(PlexPerson)
+            .values([{"name": nm, "photo_thumb": th} for nm, th in missing.items()])
+            .on_conflict_do_nothing(index_elements=[PlexPerson.name])
+        )
+        dirty = True
     return dirty
 
 

backend/tests/test_plex_sync_txn.py

@@ -4,9 +4,9 @@ lane (`db` fixture) with a faked PlexClient — skipped without Postgres.
 """
 from sqlalchemy import select
 
-from app.plex import sync
+from app.plex import people, sync
 from app.plex.client import PlexError
-from app.models import PlexLibrary
+from app.models import PlexItem, PlexLibrary, PlexPerson
 
 
 class _FakePlex:
@@ -42,3 +42,46 @@ def test_late_section_failure_keeps_earlier_sections(db, monkeypatch):
     # Section 1's library survived the section-2 failure (per-section commit); section 2 rolled back.
     keys = db.scalars(select(PlexLibrary.plex_key).order_by(PlexLibrary.plex_key)).all()
     assert keys == ["1"]
+
+
+class _FakeMetaPlex:
+    """Fake PlexClient whose .metadata() returns the per-item rich metadata dict for a rating_key."""
+
+    def __init__(self, by_key):
+        self._by_key = by_key
+
+    def metadata(self, rating_key):
+        return self._by_key.get(rating_key)
+
+
+def test_enrich_people_batch_dedupes_shared_cast(db):
+    """A cold sync where one person appears across two items in the same batch must NOT crash on the
+    unique index — before the on_conflict fix, autoflush=False let both rows queue a duplicate
+    PlexPerson and the batch commit died, leaving plex_people empty. Also covers the prior-NULL photo
+    back-fill (m1 has no thumb, m2 supplies one for the same person)."""
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie")
+    db.add(lib)
+    db.flush()
+    db.add_all(
+        [
+            PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="Movie One"),
+            PlexItem(rating_key="m2", library_id=lib.id, kind="movie", title="Movie Two"),
+        ]
+    )
+    db.commit()
+
+    shared = "Christopher Plummer"
+    thumb = people.PERSON_IMG_HOST + "cp.jpg"
+    fake = _FakeMetaPlex(
+        {
+            "m1": {"Role": [{"tag": shared, "role": "Self"}, {"tag": "Alice Unique"}]},
+            "m2": {"Role": [{"tag": shared, "role": "Self", "thumb": thumb}, {"tag": "Bob Unique"}]},
+        }
+    )
+
+    n = sync._enrich_people_batch(db, fake)
+
+    assert n == 2  # both items enriched — no unique-violation rollback
+    rows = {p.name: p for p in db.query(PlexPerson).all()}
+    assert set(rows) == {shared, "Alice Unique", "Bob Unique"}  # exactly one row per name
+    assert rows[shared].photo_thumb is not None  # m2's photo back-filled m1's NULL