siftlode/backend/app/plex/sync.py

"""Mirror a locally-reachable Plex library into Siftlode's own catalog (plex_* tables).

Playback is from the local physical file, so this only mirrors METADATA + the physical-media facts
(codecs/container/path) needed to browse, search, and later play. It reuses Plex's own codec info
(no ffprobe needed) to classify each item's browser playability. Movies + episodes become playable
LEAF rows (plex_items); shows/seasons are mirrored for the drill-down + episode ordering.

The sync is a full reconcile of the enabled sections (upsert by rating_key) AND prunes rows for
titles that disappeared from Plex: each section diffs the rating_keys it just listed against the ones
stored, and deletes the stale ones (their watch-state / playlist links cascade). The prune is guarded
on a NON-empty listing — a section that returns zero items is left untouched, so a transient Plex
glitch that yields an empty page can't wipe a library (the metadata is cheap to re-sync; an accidental
mass-delete is the only real risk). Library-level reconcile (dropping a whole removed section, the
dead `enabled` column) is a separate follow-up. Cast + intro/credit markers are fetched lazily on the
item-detail/player path (P2), not here.
"""
import logging
from datetime import datetime, timezone

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import sysconfig
from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow
from app.plex import detail, people
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.lookups import by_rating_key

log = logging.getLogger("siftlode.plex")

# Codecs/containers a browser can play in a native <video> without any transcoding.
_BROWSER_VIDEO = {"h264"}
_BROWSER_AUDIO = {"aac", "mp3", "opus", "vorbis"}
_BROWSER_CONTAINER = {"mp4", "mov", "m4v"}

# Plex section item types.
_TYPE_MOVIE = 1
_TYPE_SHOW = 2
_TYPE_SEASON = 3
_TYPE_EPISODE = 4

_PAGE = 200


def classify_playable(vcodec: str | None, acodec: str | None, container: str | None) -> str:
    """How the browser can play this file: 'direct' (native <video>), 'remux' (copy video, fix
    container / transcode audio — cheap), or 'transcode' (re-encode video — expensive)."""
    v = (vcodec or "").lower()
    a = (acodec or "").lower()
    c = (container or "").lower()
    if not v or v not in _BROWSER_VIDEO:
        return "transcode"
    if a and a not in _BROWSER_AUDIO:
        return "remux"
    if c and c not in _BROWSER_CONTAINER:
        return "remux"
    return "direct"


def _epoch(v) -> datetime | None:
    try:
        return datetime.fromtimestamp(int(v), tz=timezone.utc) if v else None
    except (TypeError, ValueError, OSError):
        return None


def _date(v):
    """Parse a Plex 'YYYY-MM-DD' string (originallyAvailableAt) → date, or None."""
    try:
        return datetime.strptime(v, "%Y-%m-%d").date() if v else None
    except (TypeError, ValueError):
        return None


def _tags(meta: dict, key: str, limit: int | None = None) -> list | None:
    """The .tag values of a Plex child-tag array (Genre/Director/Role) present in the listing."""
    vals = [t.get("tag") for t in (meta.get(key) or []) if t.get("tag")]
    if limit:
        vals = vals[:limit]
    return vals or None


def _rating(meta: dict) -> float | None:
    try:
        return float(meta.get("audienceRating") or meta.get("rating"))
    except (TypeError, ValueError):
        return None


# Provider preference for the canonical guid (a stable id shared across a title's language twins).
_GUID_PREFERENCE = ("tmdb", "imdb", "tvdb")


def _canonical_guid(meta: dict) -> str | None:
    """The canonical external id used to pair an EN title with its HU-overlay twin (E5 S5). Reads the
    `Guid` array (present only with includeGuids=1 in the listing), preferring tmdb > imdb > tvdb,
    normalised to '<provider>:<id>' (e.g. 'tmdb:12345'). None when no known provider id is present."""
    found: dict[str, str] = {}
    for g in meta.get("Guid") or []:
        raw = str(g.get("id") or "")
        for prov in _GUID_PREFERENCE:
            if raw.startswith(f"{prov}://"):
                found.setdefault(prov, f"{prov}:{raw.split('://', 1)[1]}")
    for prov in _GUID_PREFERENCE:
        if prov in found:
            return found[prov]
    return None


def _media_facts(meta: dict) -> dict:
    media = meta.get("Media") or []
    if not media:
        return {}
    m = media[0]
    part = (m.get("Part") or [{}])[0]
    return {
        "container": m.get("container") or part.get("container"),
        "codec_video": m.get("videoCodec"),
        "codec_audio": m.get("audioCodec"),
        "size_bytes": part.get("size"),
        "file_path": part.get("file"),
        "part_key": part.get("key"),
    }


def _dur_s(meta: dict) -> int | None:
    d = meta.get("duration")
    return int(d / 1000) if d else None


def _enabled_section_keys(db: Session) -> set[str] | None:
    """The configured allowlist of section keys, or None to mean 'all movie/show sections'."""
    raw = sysconfig.get_str(db, "plex_libraries")
    keys = {k.strip() for k in raw.split(",") if k.strip()}
    return keys or None


def _overlay_map(db: Session) -> dict[str, str]:
    """The configured HU-overlay pairing: {huSectionKey: enSectionKey} from `plex_overlay_map`
    ('huKey:enKey,...'). Each HU section is synced as a non-browsable overlay onto its EN twin.
    Malformed pairs are skipped."""
    raw = sysconfig.get_str(db, "plex_overlay_map")
    out: dict[str, str] = {}
    for pair in raw.split(","):
        hu, sep, en = pair.partition(":")
        hu, en = hu.strip(), en.strip()
        if sep and hu and en:
            out[hu] = en
    return out


def sync(db: Session) -> dict:
    """Full reconcile of the enabled movie/show sections. Returns a small stats dict."""
    if not sysconfig.get_bool(db, "plex_enabled"):
        return {"skipped": "disabled"}
    stats = {
        "libraries": 0, "movies": 0, "shows": 0, "seasons": 0, "episodes": 0,
        "collections": 0, "overlaid": 0, "pruned": 0,
    }
    wanted = _enabled_section_keys(db)
    overlay_map = _overlay_map(db)  # {huKey: enKey}
    overlay_keys = set(overlay_map)
    try:
        with PlexClient(db) as plex:
            sections = {
                str(s.get("key")): s
                for s in plex.sections()
                if s.get("type") in ("movie", "show")
            }
            libs_by_key: dict[str, PlexLibrary] = {}
            # Pass 1 — browsable (non-overlay) sections. HU-overlay sections are handled in pass 2 and
            # are NEVER mirrored as browsable libraries of their own.
            for key, s in sections.items():
                if key in overlay_keys:
                    continue
                if wanted is not None and key not in wanted:
                    continue
                lib = _upsert_library(db, key, s.get("title") or key, s.get("type"))
                libs_by_key[key] = lib
                stats["libraries"] += 1
                if s.get("type") == "movie":
                    _sync_movies(db, plex, lib, stats)
                else:
                    _sync_shows(db, plex, lib, stats)
                _sync_collections(db, plex, lib, stats)
                # Commit each section as it completes: a PlexError on a LATER section then keeps the
                # libraries already mirrored (idempotent — the next run reconciles the rest), instead
                # of the old single end-of-loop commit that a late failure rolled back entirely.
                db.commit()
            # Pass 2 — HU overlays. The EN twins are now synced (with guids), so write each overlay
            # section's Hungarian metadata onto the matched EN rows. Requires the EN section to have been
            # synced in pass 1 (in the allowlist / all) and to share the overlay's kind.
            for hu_key, en_key in overlay_map.items():
                s = sections.get(hu_key)
                en_lib = libs_by_key.get(en_key)
                if s is None or en_lib is None or s.get("type") != en_lib.kind:
                    log.warning(
                        "Plex overlay skipped: hu=%s en=%s (missing section/EN lib or kind mismatch)",
                        hu_key, en_key,
                    )
                    continue
                lib = _upsert_library(db, hu_key, s.get("title") or hu_key, s.get("type"))
                lib.metadata_lang = "hu"
                lib.overlay_of = en_lib.id
                db.flush()
                _sync_overlay(db, plex, lib, en_lib.id, 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:
        db.rollback()
        log.warning("Plex sync failed: %s", e)
        return {"error": str(e)}
    # 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"] = 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


# 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 _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.execute(
        select(PlexItem)
        .where(PlexItem.kind == "movie", PlexItem.people_enriched_at.is_(None))
        .order_by(PlexItem.id)
        .limit(limit)
    ).scalars().all()
    remaining = limit - len(movies)
    shows = (
        db.execute(
            select(PlexShow)
            .where(PlexShow.people_enriched_at.is_(None))
            .order_by(PlexShow.id)
            .limit(remaining)
        ).scalars().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))
        detail.apply_rich_detail(row, meta)  # imdb/tagline/full_cast → served from the row, not live
        row.people_enriched_at = now
        n += 1
    db.commit()
    return n


def _upsert_library(db: Session, key: str, title: str, kind: str) -> PlexLibrary:
    lib = db.execute(
        select(PlexLibrary).where(PlexLibrary.plex_key == key)
    ).scalars().first()
    if lib is None:
        lib = PlexLibrary(plex_key=key)
        db.add(lib)
    lib.title = title
    lib.kind = kind
    lib.synced_at = datetime.now(timezone.utc)
    db.flush()
    return lib


def _paginate(plex: PlexClient, key: str, item_type: int, progress: dict | None = None):
    """Yield a section's items page by page. If `progress` is given, set `progress["complete"]` True
    ONLY when the walk reached the server's reported `total` — a truncated fetch (an empty/short page
    mid-sequence, e.g. a transient Plex hiccup, before `total` is reached) leaves it False so the
    caller can skip pruning. Pruning on a partial listing would delete rows the listing merely failed
    to mention (the metadata is re-syncable, but that still cascades watch-state)."""
    start = 0
    total = None
    while True:
        items, total = plex.section_items(key, item_type, start, _PAGE)
        if not items:
            break
        for it in items:
            yield it
        start += len(items)
        if total and start >= total:
            break
        if len(items) < _PAGE:
            break
    if progress is not None:
        # A truthy `total` we reached = a trustworthy complete listing. `total` unknown/zero-with-rows
        # (a Plex that doesn't report totalSize) is treated as NOT complete, so prune stays conservative.
        progress["complete"] = bool(total) and start >= total


# The section listing exposes at most the top ~20 cast (Plex's own cap); a stored cast longer than this
# can only have come from the rich enrich, never the listing.
_LISTING_CAST_CAP = 20


def _apply_facet_fields(row, meta: dict) -> None:
    """Set the filterable/orderable metadata columns shared by plex_items + plex_shows (identical
    column names) from the cheap section listing: rating, content_rating, studio, release date,
    genres/directors/cast, and the searchable people blob folded into search_vector at sync time."""
    row.rating = _rating(meta)
    row.content_rating = (meta.get("contentRating") or None)
    row.studio = (meta.get("studio") or None) and str(meta.get("studio"))[:255]
    row.originally_available_at = _date(meta.get("originallyAvailableAt"))
    row.genres = _tags(meta, "Genre")
    row.directors = _tags(meta, "Director")
    # The listing carries only the top ~20 cast, so re-applying it would DOWNGRADE an already-enriched
    # full cast (up to 24, grown by the rich backfill/info-page enrich), silently breaking the
    # person-page cast filter for names past the top ~20. The watermark gate below (`people_enriched_at
    # is None`) normally protects this, but it is NOT enough on its own: a migration that resets the
    # watermark re-opens the regression (0065 nulled it, which is exactly how the S5b fix got undone in
    # prod). So guard STRUCTURALLY too, by LENGTH against the listing cap: a stored cast LONGER than the
    # cap can only be an enriched fuller list, so keep it (robust to any watermark reset). A stored cast
    # at or below the cap is itself listing-sourced, so take the fresh listing — which tracks Plex cast
    # EDITS in EITHER direction (comparing against this listing's length instead would silently keep a
    # stale longer cast when Plex TRIMS one).
    if row.people_enriched_at is None:
        listing_cast = _tags(meta, "Role", limit=_LISTING_CAST_CAP)
        if len(row.cast_names or []) <= _LISTING_CAST_CAP:
            row.cast_names = listing_cast
    people = list(dict.fromkeys((row.directors or []) + (row.cast_names or [])))
    row.people_text = " ".join(people) or None


def _apply_item(row: PlexItem, lib_id: int, kind: str, meta: dict) -> None:
    row.library_id = lib_id
    row.kind = kind
    # EN<->HU pairing key (movies; harmless/unused on episodes). Keep the prior guid if this listing
    # carries none (a server ignoring includeGuids mustn't wipe the overlay's only match key).
    row.guid = _canonical_guid(meta) or row.guid
    row.title = (meta.get("title") or "").strip() or "Untitled"
    row.summary = meta.get("summary")
    row.year = meta.get("year")
    row.duration_s = _dur_s(meta)
    row.thumb_key = meta.get("thumb")
    row.art_key = meta.get("art") or meta.get("grandparentArt")
    row.added_at = _epoch(meta.get("addedAt"))
    _apply_facet_fields(row, meta)  # filterable/orderable extras (shared with shows)
    mf = _media_facts(meta)
    row.file_path = mf.get("file_path")
    row.part_key = mf.get("part_key")
    row.container = mf.get("container")
    row.codec_video = mf.get("codec_video")
    row.codec_audio = mf.get("codec_audio")
    row.size_bytes = mf.get("size_bytes")
    row.playable = classify_playable(mf.get("codec_video"), mf.get("codec_audio"), mf.get("container"))


def _prune_stale(
    db: Session, existing: dict, seen: set[str], stats: dict, *, complete: bool
) -> None:
    """Delete stored rows whose rating_key is no longer in the freshly-listed `seen` set — the titles
    that disappeared from Plex (their watch-state / playlist links cascade via the FKs). GUARDED on a
    COMPLETE, non-empty listing: a truncated fetch (`complete` False — a mid-sequence empty/short page
    before the server's total) or a zero-item listing prunes NOTHING, so a transient Plex hiccup can
    never wipe a library over rows it merely failed to list. Callers pass the rating_key -> row map
    they already built for the upsert."""
    if not complete or not seen:
        return
    for rk, row in existing.items():
        if rk not in seen:
            db.delete(row)
            stats["pruned"] = stats.get("pruned", 0) + 1
    db.flush()


def _sync_movies(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
    existing = {
        r.rating_key: r
        for r in db.execute(
            select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
        ).scalars()
    }
    seen: set[str] = set()
    progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_MOVIE, progress):
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        row = existing.get(rk) or PlexItem(rating_key=rk)
        if row.id is None:
            db.add(row)
        _apply_item(row, lib.id, "movie", meta)
        seen.add(rk)
        stats["movies"] += 1
    db.flush()
    _prune_stale(db, existing, seen, stats, complete=progress.get("complete", False))


def _sync_shows(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
    # 1) shows
    shows = {
        r.rating_key: r
        for r in db.execute(
            select(PlexShow).where(PlexShow.library_id == lib.id)
        ).scalars()
    }
    shows_seen: set[str] = set()
    shows_progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_SHOW, shows_progress):
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        shows_seen.add(rk)
        sh = shows.get(rk) or PlexShow(rating_key=rk)
        if sh.id is None:
            db.add(sh)
            shows[rk] = sh
        sh.library_id = lib.id
        sh.guid = _canonical_guid(meta) or sh.guid  # EN<->HU pairing key; keep prior if listing has none
        sh.title = (meta.get("title") or "").strip() or "Untitled"
        sh.summary = meta.get("summary")
        sh.year = meta.get("year")
        sh.thumb_key = meta.get("thumb")
        sh.art_key = meta.get("art")
        sh.child_count = meta.get("childCount")
        sh.added_at = _epoch(meta.get("addedAt"))
        _apply_facet_fields(sh, meta)  # same filterable/orderable tags as movies
        stats["shows"] += 1
    db.flush()
    show_id = {rk: sh.id for rk, sh in shows.items()}

    # 2) seasons
    seasons = {
        r.rating_key: r
        for r in db.execute(
            select(PlexSeason).join(PlexShow).where(PlexShow.library_id == lib.id)
        ).scalars()
    }
    seasons_seen: set[str] = set()
    seasons_progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_SEASON, seasons_progress):
        rk = str(meta.get("ratingKey") or "")
        sid = show_id.get(str(meta.get("parentRatingKey") or ""))
        if not rk or sid is None:
            continue
        seasons_seen.add(rk)
        se = seasons.get(rk) or PlexSeason(rating_key=rk)
        if se.id is None:
            db.add(se)
            seasons[rk] = se
        se.show_id = sid
        se.title = meta.get("title")
        se.season_number = meta.get("index")
        se.thumb_key = meta.get("thumb")
        stats["seasons"] += 1
    db.flush()
    season_id = {rk: se.id for rk, se in seasons.items()}

    # 3) episodes (playable leaves)
    episodes = {
        r.rating_key: r
        for r in db.execute(
            select(PlexItem).where(PlexItem.library_id == lib.id, PlexItem.kind == "episode")
        ).scalars()
    }
    episodes_seen: set[str] = set()
    episodes_progress: dict = {}
    for meta in _paginate(plex, lib.plex_key, _TYPE_EPISODE, episodes_progress):
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        episodes_seen.add(rk)
        row = episodes.get(rk) or PlexItem(rating_key=rk)
        if row.id is None:
            db.add(row)
        _apply_item(row, lib.id, "episode", meta)
        row.show_id = show_id.get(str(meta.get("grandparentRatingKey") or ""))
        row.season_id = season_id.get(str(meta.get("parentRatingKey") or ""))
        row.season_number = meta.get("parentIndex")
        row.episode_number = meta.get("index")
        stats["episodes"] += 1
    db.flush()

    # Prune leaf-first (episodes -> seasons -> shows) so a stale row is deleted directly, before its
    # parent's cascade would reach it — avoids deleting the same ORM object twice. Each level is
    # independently guarded on a non-empty listing by _prune_stale.
    _prune_stale(db, episodes, episodes_seen, stats, complete=episodes_progress.get("complete", False))
    _prune_stale(db, seasons, seasons_seen, stats, complete=seasons_progress.get("complete", False))
    _prune_stale(db, shows, shows_seen, stats, complete=shows_progress.get("complete", False))


def _sync_overlay(
    db: Session, plex: PlexClient, lib: PlexLibrary, en_lib_id: int, stats: dict
) -> None:
    """Write an overlay section's Hungarian metadata onto its EN twin's rows, matched by guid. Creates
    NO browsable rows — an overlay library is never listed on its own. Only title/summary/genres differ
    by language; the language-neutral people_text feeds both search vectors.

    This is a full RECONCILE: every guid-bearing EN row's HU fields are cleared first, then the HU
    section repopulates the matches — so an EN row whose HU twin vanished (or was re-guid'd) falls back
    to EN instead of keeping stale Hungarian metadata. HU items with no EN match are skipped."""
    if lib.kind == "movie":
        rows = db.execute(
            select(PlexItem).where(
                PlexItem.library_id == en_lib_id,
                PlexItem.kind == "movie",
                PlexItem.guid.is_not(None),
            )
        ).scalars()
        item_type = _TYPE_MOVIE
    else:
        rows = db.execute(
            select(PlexShow).where(
                PlexShow.library_id == en_lib_id, PlexShow.guid.is_not(None)
            )
        ).scalars()
        item_type = _TYPE_SHOW
    # guid -> the EN rows carrying it (a list: two editions of one film can share a guid; both overlay).
    by_guid: dict[str, list] = {}
    for r in rows:
        by_guid.setdefault(r.guid, []).append(r)
    # Reconcile step: clear HU fields on all guid-bearing rows. Only rows that actually held HU data
    # become dirty (SQLAlchemy skips a no-net-change UPDATE), so this is cheap; matches are re-set below.
    for group in by_guid.values():
        for row in group:
            row.title_hu = row.summary_hu = row.genres_hu = None
    seen: set[str] = set()  # first HU entry for a guid wins; a dup HU item can't overwrite or double-count
    for meta in _paginate(plex, lib.plex_key, item_type):
        g = _canonical_guid(meta)
        targets = by_guid.get(g) if g else None
        if not targets or g in seen:
            continue
        seen.add(g)
        title_hu = (meta.get("title") or "").strip() or None
        summary_hu = meta.get("summary")
        genres_hu = _tags(meta, "Genre")
        for row in targets:
            row.title_hu = title_hu
            row.summary_hu = summary_hu
            row.genres_hu = genres_hu
        stats["overlaid"] = stats.get("overlaid", 0) + 1
    db.flush()


def _upsert_collection(
    db: Session, existing: PlexCollection | None, rk: str, lib_id: int, meta: dict
) -> PlexCollection:
    """Get-or-create a PlexCollection row and apply its metadata from a Plex `meta` dict. `editable`
    is user-managed and preserved (never set here). Shared by the full sync + the targeted re-sync."""
    col = existing or PlexCollection(rating_key=rk)
    if col.id is None:
        db.add(col)
    col.library_id = lib_id
    col.title = (meta.get("title") or "").strip() or "Untitled"
    col.summary = meta.get("summary")
    col.thumb_key = meta.get("thumb")
    col.child_count = meta.get("childCount")
    col.smart = str(meta.get("smart")) == "1"
    col.synced_at = datetime.now(timezone.utc)
    return col


def resync_collection(db: Session, plex: PlexClient, lib: PlexLibrary, rk: str) -> PlexCollection:
    """Targeted re-sync of ONE collection after a write-back (create/add/remove/rename) — avoids the
    full ~4-min library sync. Upserts the row (preserving `editable`) and reconciles ONLY this
    collection's membership on the affected member rows."""
    meta = plex.metadata(rk) or {}
    col = _upsert_collection(
        db, by_rating_key(db, PlexCollection, rk), rk, lib.id, meta
    )
    Model = PlexItem if lib.kind == "movie" else PlexShow
    new_members = {str(c.get("ratingKey")) for c in plex.collection_children(rk) if c.get("ratingKey")}
    current = db.execute(
        select(Model).where(Model.collection_keys.contains([rk]))
    ).scalars().all()
    for row in current:  # drop rk from rows that are no longer members
        if row.rating_key not in new_members:
            row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
    to_add = new_members - {r.rating_key for r in current}
    if to_add:
        for row in db.execute(
            select(Model).where(Model.rating_key.in_(to_add))
        ).scalars():
            row.collection_keys = list(dict.fromkeys([*(row.collection_keys or []), rk]))
    db.flush()
    return col


def delete_collection_local(db: Session, lib: PlexLibrary, rk: str) -> None:
    """Local cleanup after a collection is deleted from Plex: strip rk from members + drop the row."""
    Model = PlexItem if lib.kind == "movie" else PlexShow
    for row in db.execute(
        select(Model).where(Model.collection_keys.contains([rk]))
    ).scalars():
        row.collection_keys = [k for k in (row.collection_keys or []) if k != rk] or None
    col = by_rating_key(db, PlexCollection, rk)
    if col is not None:
        db.delete(col)
    db.flush()


def _sync_collections(db: Session, plex: PlexClient, lib: PlexLibrary, stats: dict) -> None:
    """Mirror a library's collections + membership. Reads never touch Plex afterwards. Membership is
    written as `collection_keys` on member movies (plex_items) / shows (plex_shows). A full reconcile:
    collections gone from Plex are pruned, and each member row's keys are rebuilt from scratch."""
    cols = plex.collections(lib.plex_key)
    existing = {
        c.rating_key: c
        for c in db.execute(
            select(PlexCollection).where(PlexCollection.library_id == lib.id)
        ).scalars()
    }
    seen: set[str] = set()
    membership: dict[str, list[str]] = {}  # member rating_key -> [collection rating_key, ...]
    for meta in cols:
        rk = str(meta.get("ratingKey") or "")
        if not rk:
            continue
        _upsert_collection(db, existing.get(rk), rk, lib.id, meta)
        seen.add(rk)
        for child in plex.collection_children(rk):
            crk = str(child.get("ratingKey") or "")
            if crk:
                membership.setdefault(crk, []).append(rk)
        stats["collections"] += 1
    db.flush()
    for rk, col in existing.items():
        if rk not in seen:
            db.delete(col)
    # Rebuild collection_keys on every member row of this library (clear + set).
    for row in db.execute(select(PlexItem).where(PlexItem.library_id == lib.id)).scalars():
        row.collection_keys = membership.get(row.rating_key) or None
    for row in db.execute(select(PlexShow).where(PlexShow.library_id == lib.id)).scalars():
        row.collection_keys = membership.get(row.rating_key) or None
    db.flush()