feat(plex): full-library cast backfill via a batched rich re-sync (S1b)

Lazy enrichment only covers viewed titles, so an actor's count/results were
partial. Add a background backfill: each plex_sync rich-fetches a bounded batch
(60) of not-yet-enriched movies/shows and persists their FULL cast, stamping
people_enriched_at so each is fetched once — the whole library is covered over
successive runs (migration 0057 adds the marker). Extracted the enrichment into a
shared app/plex/people.py (cast_from_meta / enrich_row_people / prune_orphan_people
+ the person-image host/proxy) so the info-page lazy path and the batch share ONE
implementation — no duplication; the info page also stamps the marker so the batch
skips it. Verified: an 8-item batch grows those movies' cast_names 3->24 and fills
plex_people (32->224); the info-page cast is intact after the _rich_meta refactor.
(Plex cast-explore S1b)

backend/app/plex/sync.py

@@ -12,11 +12,11 @@ Cast + intro/credit markers are fetched lazily on the item-detail/player path (P
 import logging
 from datetime import datetime, timezone
 
-from sqlalchemy import text
 from sqlalchemy.orm import Session
 
 from app import sysconfig
 from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow
+from app.plex import people
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 
 log = logging.getLogger("siftlode.plex")
@@ -130,6 +130,12 @@ def sync(db: Session) -> dict:
                     _sync_shows(db, plex, lib, stats)
                 _sync_collections(db, plex, lib, stats)
             db.commit()
+            # Rich-fetch a bounded batch of not-yet-enriched titles' FULL cast, so the whole library
+            # becomes filterable over successive runs (the cheap listing above only stored the top ~3).
+            try:
+                stats["people_enriched"] = _enrich_people_batch(db, plex)
+            except Exception as e:  # best-effort — never fail the sync over the backfill
+                log.warning("Plex people enrich batch failed: %s", e)
     except PlexNotConfigured as e:
         return {"skipped": str(e)}
     except PlexError as e:
@@ -139,37 +145,53 @@ def sync(db: Session) -> dict:
     # Prune the headshot cache: a reconcile may have removed titles, orphaning people who no longer
     # appear in ANY item's cast/directors — drop those rows, mirroring how Plex drops orphaned metadata.
     try:
-        stats["people_pruned"] = prune_orphan_people(db)
+        stats["people_pruned"] = people.prune_orphan_people(db)
     except Exception as e:  # best-effort housekeeping — never fail the sync over it
         log.warning("Plex people prune failed: %s", e)
     log.info("Plex sync done: %s", stats)
     return stats
 
 
-# The person names still referenced by ANY movie/show (cast OR directors). plex_people rows outside
-# this set are orphans left by a library removal.
-_REFERENCED_PEOPLE_SQL = """
-    SELECT jsonb_array_elements_text(cast_names) FROM plex_items WHERE jsonb_typeof(cast_names) = 'array'
-    UNION SELECT jsonb_array_elements_text(directors) FROM plex_items WHERE jsonb_typeof(directors) = 'array'
-    UNION SELECT jsonb_array_elements_text(cast_names) FROM plex_shows WHERE jsonb_typeof(cast_names) = 'array'
-    UNION SELECT jsonb_array_elements_text(directors) FROM plex_shows WHERE jsonb_typeof(directors) = 'array'
-"""
+# How many titles to rich-fetch per sync run for the cast backfill — bounds the extra Plex load; the
+# whole library is covered over successive runs (each fetched item is stamped people_enriched_at once).
+_ENRICH_BATCH = 60
 
 
-def prune_orphan_people(db: Session) -> int:
-    """Delete plex_people (headshot cache) rows whose name no longer appears in any movie/show cast or
-    directors — orphaned by a library removal. Returns the number pruned. Commits its own work."""
-    # `WHERE ref IS NOT NULL` matters: a JSON-null array element would surface as a SQL NULL in the
-    # referenced set, and `name NOT IN (…, NULL)` is NULL (never TRUE) for every row — the classic
-    # NOT IN trap that would silently prune nothing. Filtering NULLs out keeps the delete correct.
-    res = db.execute(
-        text(
-            "DELETE FROM plex_people WHERE name NOT IN "
-            f"(SELECT ref FROM ({_REFERENCED_PEOPLE_SQL}) AS refs(ref) WHERE ref IS NOT NULL)"
-        )
+def _enrich_people_batch(db: Session, plex: PlexClient, limit: int = _ENRICH_BATCH) -> int:
+    """Rich-fetch the FULL cast of up to `limit` not-yet-enriched movies/shows and persist it (via the
+    shared enrich_row_people). Oldest-id first, movies before shows. Stamps people_enriched_at only on
+    a SUCCESSFUL fetch, so a transient per-item failure retries next run rather than being lost.
+    Returns how many items were enriched this run."""
+    movies = (
+        db.query(PlexItem)
+        .filter(PlexItem.kind == "movie", PlexItem.people_enriched_at.is_(None))
+        .order_by(PlexItem.id)
+        .limit(limit)
+        .all()
     )
+    remaining = limit - len(movies)
+    shows = (
+        db.query(PlexShow)
+        .filter(PlexShow.people_enriched_at.is_(None))
+        .order_by(PlexShow.id)
+        .limit(remaining)
+        .all()
+        if remaining > 0
+        else []
+    )
+    now = datetime.now(timezone.utc)
+    n = 0
+    for row in [*movies, *shows]:
+        try:
+            meta = plex.metadata(row.rating_key) or {}
+        except Exception as e:
+            log.debug("Plex people enrich fetch failed for %s: %s", row.rating_key, e)
+            continue  # leave people_enriched_at NULL so it's retried on a later run
+        people.enrich_row_people(db, row, people.cast_from_meta(meta))
+        row.people_enriched_at = now
+        n += 1
     db.commit()
-    return res.rowcount or 0
+    return n
 
 
 def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary: