siftlode/backend/app/plex/watch_sync.py

"""Plex ↔ Siftlode watch-state sync.

Plex records watch state PER PLEX ACCOUNT; Siftlode records it per Siftlode user in `plex_states`.
A `PlexLink` row maps one Siftlode user to a Plex account. For the owner (MVP) that account is the
server admin (`uses_admin=True`), so the existing admin token already reads/writes the owner's state
— no separate Plex login.

Phase A (this module today): the one-time **"Plex is master" import** — read the owner account's
viewCount / viewOffset / lastViewedAt (present in the same cheap section listing the catalog mirror
already pages) and upsert them into the owner's `plex_states`. Plex wins on overlap; states that
exist only in Siftlode are left untouched (union, Plex-authoritative on the intersection).

Phases B (Siftlode→Plex push via scrobble/timeline) and C (incremental Plex→Siftlode via the watch
history + onDeck) build on this and land in later ships. As of S2 Plex is the source of truth: the
player pushes the resume position back to Plex on the periodic checkpoint (not just the final flush),
and the incremental pull resolves a genuine divergence to Plex's state (see `_pull_apply`) rather than
running a last-write-wins comparison.
"""

# On the PlexError lines in this module: none of them carries a traceback, deliberately. A
# PlexError is httpx's own message — the method, the URL and the status of one call — and the stack
# beneath it is always this `except` plus `PlexClient._get`. The attrs beside the line say which
# user and which phase, which is the part the message cannot know.
from datetime import datetime, timezone

from sqlalchemy import func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from app import logs, sysconfig
from app.db import SessionLocal
from app.models import PlexItem, PlexLink, PlexState
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.sync import (
    _TYPE_EPISODE,
    _TYPE_MOVIE,
    _enabled_section_keys,
    _epoch,
    _paginate,
)

log = logs.get_logger(__name__)
# A resume position below this many seconds isn't worth mirroring — matches the player's own
# _PROGRESS_MIN_S (a stray few-second seek is not "in progress").
_PROGRESS_MIN_S = 5


def _plex_watch_to_state(
    view_count, view_offset_ms, last_viewed_at
) -> tuple[str, int, datetime | None, datetime | None] | None:
    """Translate a Plex leaf's per-account watch fields into a Siftlode state tuple
    ``(status, position_seconds, watched_at, progress_updated_at)``, or None when Plex holds no
    meaningful watch signal for it (so the import skips it and never clears a Siftlode-only state).

    In Siftlode's model (mirrors item_progress/item_state): a *watched* item is ``status="watched"``
    with position 0; an *in-progress* item keeps the default status and carries ``position_seconds``
    plus ``progress_updated_at``."""
    ts = _epoch(last_viewed_at)
    if view_count and int(view_count) > 0:
        return ("watched", 0, ts, ts)
    if view_offset_ms:
        pos = int(view_offset_ms) // 1000
        if pos >= _PROGRESS_MIN_S:
            return ("new", pos, None, ts)
    return None


def _rk_to_id(db: Session) -> dict[str, int]:
    """The rating_key → PlexItem.id lookup used to resolve a Plex leaf back to its mirrored row."""
    return {rk: iid for rk, iid in db.execute(select(PlexItem.rating_key, PlexItem.id)).all()}


def _scan_plex_states(
    db: Session, plex: PlexClient
) -> tuple[dict[int, tuple[str, int, datetime | None, datetime | None]], int]:
    """Scan the enabled movie/show sections and return ``{item_id: (status, pos, watched_at,
    prog_at)}`` for every mirrored leaf Plex holds a watch signal for, plus the scanned-leaf count.
    Shared by the one-time import and the full reconcile (both need Plex's whole current picture)."""
    item_id_by_rk = _rk_to_id(db)
    wanted = _enabled_section_keys(db)
    scanned = 0
    out: dict[int, tuple[str, int, datetime | None, datetime | None]] = {}
    for s in plex.sections():
        stype = s.get("type")
        if stype not in ("movie", "show"):
            continue
        key = str(s.get("key"))
        if wanted is not None and key not in wanted:
            continue
        item_type = _TYPE_MOVIE if stype == "movie" else _TYPE_EPISODE
        for meta in _paginate(plex, key, item_type):
            scanned += 1
            item_id = item_id_by_rk.get(str(meta.get("ratingKey") or ""))
            if item_id is None:
                continue
            res = _plex_watch_to_state(
                meta.get("viewCount"), meta.get("viewOffset"), meta.get("lastViewedAt")
            )
            if res is not None:
                out[item_id] = res
    return out, scanned


def import_owner_watch_state(db: Session, link: PlexLink) -> dict:
    """One-time "Plex is master" import of the linked user's Plex watch state into their
    `plex_states`. Idempotent (safe to re-run — Plex simply re-wins on the intersection). Only
    touches items Plex has a watch record for; Siftlode-only states are preserved."""
    if not sysconfig.get_bool(db, "plex_enabled"):
        return {"skipped": "disabled"}

    stats = {"watched": 0, "in_progress": 0, "scanned": 0}
    try:
        with PlexClient(db) as plex:
            scanned_states, stats["scanned"] = _scan_plex_states(db, plex)
            # One row per item (keyed by item_id → inherently de-duped) then bulk-UPSERT. An upsert is
            # idempotent AND concurrency-safe: a repeated/overlapping enable (the toggle firing twice
            # while the multi-second import runs) can't raise a duplicate-key error — it re-updates.
            rows = []
            for item_id, (status, pos, watched_at, prog_at) in scanned_states.items():
                rows.append(
                    {
                        "user_id": link.user_id,
                        "item_id": item_id,
                        "status": status,
                        "position_seconds": pos,
                        "watched_at": watched_at,
                        "progress_updated_at": prog_at,
                        # Mirrors Plex now — so Phase B's push doesn't bounce it straight back.
                        "synced_to_plex": True,
                    }
                )
                stats["watched" if status == "watched" else "in_progress"] += 1

            # Bulk upsert in chunks. Plex wins on the intersection; Siftlode-only states (item_ids not
            # scanned) are never referenced, so they're preserved untouched (union).
            for i in range(0, len(rows), 1000):
                stmt = pg_insert(PlexState).values(rows[i : i + 1000])
                stmt = stmt.on_conflict_do_update(
                    index_elements=[PlexState.user_id, PlexState.item_id],
                    set_={
                        "status": stmt.excluded.status,
                        "position_seconds": stmt.excluded.position_seconds,
                        "watched_at": stmt.excluded.watched_at,
                        "progress_updated_at": stmt.excluded.progress_updated_at,
                        "synced_to_plex": stmt.excluded.synced_to_plex,
                        "updated_at": func.now(),
                    },
                )
                db.execute(stmt)
            link.initial_import_done = True
            link.last_watch_sync_at = datetime.now(timezone.utc)
            db.commit()
    # silent-ok: the `{"skipped": …}` return is the report — the admin route answers it and the
    # scheduler logs it on the job line.
    except PlexNotConfigured as e:
        return {"skipped": str(e)}
    except PlexError as e:
        db.rollback()
        log.warning("Plex watch import failed (user %s): %s", link.user_id, e)
        return {"error": str(e)}

    log.info("Plex watch import (user %s): %s", link.user_id, stats)
    return stats


# --- Phase B: Siftlode → Plex push -----------------------------------------------------------------


def link_for_push(db: Session, user_id: int) -> PlexLink | None:
    """The active owner watch-push link for this user, or None when push is off / not the owner.
    Push is gated on `initial_import_done` so we never push a half-known local state over Plex's
    authoritative one before the one-time "Plex is master" import has reconciled them."""
    link = db.execute(select(PlexLink).where(PlexLink.user_id == user_id)).scalars().first()
    if link and link.uses_admin and link.sync_enabled and link.initial_import_done:
        return link
    return None


def _state_marker(st: PlexState | None) -> datetime | None:
    """The freshest local-edit timestamp on a state row (watched_at / progress_updated_at). A push
    captures it at schedule time so it can tell, when it later settles the `synced_to_plex` flag,
    whether the row has been re-edited in the meantime (see `push_state_to_plex`)."""
    if st is None:
        return None
    return max([t for t in (st.watched_at, st.progress_updated_at) if t], default=None)


def push_state_to_plex(
    user_id: int,
    item_id: int,
    rating_key: str,
    action: str,
    time_ms: int = 0,
    duration_ms: int = 0,
    expect_ts: datetime | None = None,
) -> None:
    """Best-effort Siftlode→Plex push, run in a FastAPI BackgroundTask (its OWN DB session — the
    request's session is already closed by the time this runs). Never raises: Plex being down or slow
    must not break the user's action. On success it flags `synced_to_plex` so Phase C's incremental
    pull won't bounce the change straight back; on failure the row stays dirty (synced_to_plex=False)
    for a later retry/reconcile.

    `action` is one of ``watched`` (scrobble), ``unwatched`` (unscrobble), ``resume`` (timeline).
    `expect_ts` is the row's freshest timestamp captured when this push was scheduled: if the row has
    since been re-edited (a newer local change with its own pending push), we must NOT mark it clean —
    doing so would strand that newer value, never pushing it to Plex (lost update)."""
    with SessionLocal() as db:
        if link_for_push(db, user_id) is None:
            return
        try:
            with PlexClient(db) as plex:
                if action == "watched":
                    plex.scrobble(rating_key)
                elif action == "unwatched":
                    plex.unscrobble(rating_key)
                elif action == "resume":
                    plex.set_timeline(rating_key, time_ms, duration_ms)
                else:
                    return
        except PlexError as e:
            log.warning(
                "Plex watch push failed (user %s, item %s, %s): %s", user_id, item_id, action, e
            )
            return
        # Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be
        # nothing to flag; that's fine (unscrobble still happened). Skip the flag if the row was
        # re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile).
        st = (
            db.execute(
                select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
            )
            .scalars()
            .first()
        )
        if st is not None and _row_matches(st, expect_ts):
            st.synced_to_plex = True
            db.commit()


def _row_matches(st: PlexState, expect_ts: datetime | None) -> bool:
    """Whether a state row is still the one a push was scheduled for — true when no marker was given
    (legacy/edge), the row carries no timestamp, or its freshest timestamp is within a second of the
    captured marker (a genuine newer edit is always many seconds later)."""
    if expect_ts is None:
        return True
    cur = _state_marker(st)
    return cur is None or abs((cur - expect_ts).total_seconds()) < 1.0


def push_bulk_state_to_plex(user_id: int, changes: list[tuple[int, str, str]]) -> None:
    """Coalesced Siftlode→Plex push for a whole-show / whole-season mark (see `_bulk_state`). Same
    contract as `push_state_to_plex` (own DB session, best-effort, never raises) but ONE background
    task for the entire batch: it reuses a single DB session and a single keep-alive PlexClient across
    all `changes` instead of scheduling N tasks that each rebuild a session + HTTP client. Plex has no
    batch-scrobble endpoint, so the per-item HTTP calls remain (looped over the shared connection), but
    the O(N) task/session/client setup is gone. `changes` is a list of (item_id, rating_key, action)
    with action ``watched`` (scrobble) or ``unwatched`` (unscrobble)."""
    if not changes:
        return
    with SessionLocal() as db:
        if link_for_push(db, user_id) is None:
            return
        synced_ids: list[int] = []
        try:
            with PlexClient(db) as plex:
                for item_id, rating_key, action in changes:
                    try:
                        if action == "watched":
                            plex.scrobble(rating_key)
                        elif action == "unwatched":
                            plex.unscrobble(rating_key)
                        else:
                            continue
                    except PlexError as e:
                        # One bad item mustn't abort the rest of the batch; leave its row dirty
                        # (synced_to_plex=False) for a later reconcile.
                        log.warning(
                            "Plex bulk push failed (user %s, item %s, %s): %s",
                            user_id,
                            item_id,
                            action,
                            e,
                        )
                        continue
                    synced_ids.append(item_id)
        except PlexError as e:
            # Couldn't even open the client — nothing pushed; everything stays dirty for reconcile.
            log.warning("Plex bulk push could not connect (user %s): %s", user_id, e)
            return
        # Flag every successfully-pushed row that still exists (an "unwatched" deletes the row, so
        # only the "watched" changes have a row to flag).
        if synced_ids:
            db.execute(
                update(PlexState)
                .where(PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids))
                .values(synced_to_plex=True)
            )
            db.commit()


# --- Phase C: incremental + full Plex ↔ Siftlode reconcile (scheduler jobs) ------------------------


def _active_sync_links(db: Session) -> list[PlexLink]:
    """Every link with two-way sync actually running (owner MVP = one row). Iterated uniformly so
    P5b friend links slot in later without touching this loop."""
    return (
        db.execute(
            select(PlexLink).where(
                PlexLink.uses_admin.is_(True),
                PlexLink.sync_enabled.is_(True),
                PlexLink.initial_import_done.is_(True),
            )
        )
        .scalars()
        .all()
    )


def _resolve_account_id(db: Session, plex: PlexClient, link: PlexLink) -> int:
    """The owner's Plex accountID, cached on the link. On a Plex Media Server, accountID 1 is reserved
    for the server owner/admin (managed + shared users get id > 1), and an owner link (`uses_admin`)
    always holds the admin token — so 1 is correct here, not an assumption to second-guess. We still
    confirm + grab the owner's username from /accounts, falling back to 1 if that call is unavailable."""
    if link.plex_account_id:
        return link.plex_account_id
    acct_id, username = 1, None
    try:
        for a in plex.accounts():
            if int(a.get("id") or -1) == 1:
                username = a.get("name")
                break
    except PlexError as exc:
        # The fallback account id 1 is right for the admin-token setup, so sync keeps working —
        # but the username is never stored, and the UI then has no name to show for the link,
        # exactly as if the server had not offered one.
        log.info(
            "the Plex account list could not be read; the link keeps the default account",
            extra=logs.event(
                "plex.watch.accounts_probe_failed",
                user_id=link.user_id,
                reason=type(exc).__name__,
            ),
        )
    link.plex_account_id = acct_id
    if username:
        link.plex_username = username
    db.commit()
    return acct_id


def _same_state(st: PlexState | None, status: str, pos: int) -> bool:
    """Whether a row already matches a target (status, pos) — the core ping-pong guard: when nothing
    would change we skip the write entirely (no row churn, no timestamp bump, no re-pull)."""
    if st is None:
        return False
    if status == "watched":
        return st.status == "watched"
    return st.status != "watched" and abs((st.position_seconds or 0) - pos) <= _PROGRESS_MIN_S


def _pull_apply(
    db: Session, user_id: int, item_id: int, status: str, pos: int, plex_ts: datetime | None
) -> str | None:
    """Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
    "watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
    full reconcile's job) — this only pulls in watched/resume."""
    st = (
        db.execute(
            select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id)
        )
        .scalars()
        .first()
    )
    if _same_state(st, status, pos):
        if st is not None and not st.synced_to_plex:
            st.synced_to_plex = True  # already matches Plex → just settle the flag, no churn
        return None
    # Plex-authoritative (S2): on a genuine divergence for an item Plex knows about, Plex wins — no
    # last-write-wins tug-of-war. This is safe because prompt write-back (playback.py pushes the resume
    # position to Plex on the ~10s checkpoint, not just on the final flush) keeps Plex current, so a
    # local edit that's "newer" than Plex has almost always already been pushed and matched above by
    # `_same_state`; a row that reaches here really is out of step with the source of truth. The
    # `_same_state` + `synced_to_plex` short-circuit above still absorbs the freshly-pushed case (Plex
    # echoes the same shape a beat later) so a just-pushed row never bounces.
    if st is not None and not st.synced_to_plex:
        # Durability guard: this row holds a local change that never reached Plex — its immediate push
        # failed, so it's still dirty. Do NOT let the Plex-authoritative pull overwrite it and clear the
        # flag: that would strand the local value forever, since `_repush_dirty` only retries dirty rows.
        # Leave it dirty; `_repush_dirty` (later in this same run) pushes the local value up to Plex. This
        # closes the "Plex flapped mid-session" lost-update — the one case the Plex-wins comment's
        # "almost always already pushed" assumption above excludes.
        return None
    if st is None:
        st = PlexState(user_id=user_id, item_id=item_id)
        db.add(st)
    if status == "watched":
        st.status, st.position_seconds, st.watched_at, st.progress_updated_at = (
            "watched",
            0,
            plex_ts,
            plex_ts,
        )
    else:
        st.status, st.position_seconds, st.progress_updated_at = "new", pos, plex_ts
    st.synced_to_plex = True
    return "watched" if status == "watched" else "in_progress"


def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int:
    """Belt-and-suspenders: push local states that never reached Plex (synced_to_plex=False) — e.g.
    an immediate Phase B push that failed while Plex was down. `hidden` is excluded (Siftlode-only,
    never goes to Plex). Returns the count re-pushed."""
    # id → (rating_key, duration_s): duration comes straight from the mirrored row, so a resume
    # re-push doesn't need an extra per-item plex.metadata() round-trip just to fill set_timeline's
    # duration argument.
    info_by_id: dict[int, tuple[str, int | None]] = {
        iid: (rk, dur)
        for iid, rk, dur in db.execute(
            select(PlexItem.id, PlexItem.rating_key, PlexItem.duration_s)
        ).all()
    }
    dirty = (
        db.execute(
            select(PlexState).where(
                PlexState.user_id == link.user_id,
                PlexState.synced_to_plex.is_(False),
                PlexState.status != "hidden",
            )
        )
        .scalars()
        .all()
    )
    n = 0
    for st in dirty:
        info = info_by_id.get(st.item_id)
        if info is None:
            continue
        rk, dur_s = info
        try:
            if st.status == "watched":
                plex.scrobble(rk)
            elif st.position_seconds and st.position_seconds >= _PROGRESS_MIN_S:
                plex.set_timeline(rk, st.position_seconds * 1000, int(dur_s or 0) * 1000)
            else:
                plex.unscrobble(rk)
            st.synced_to_plex = True
            n += 1
        except PlexError as e:
            log.warning(
                "Plex dirty re-push failed (user %s, item %s): %s", link.user_id, st.item_id, e
            )
    return n


# How many stale in-progress movies we probe per incremental run (one metadata() call each). The set
# is normally tiny (movies started but no longer on Plex's Continue-Watching); the daily full reconcile
# is the backstop for anything beyond the cap. Most-recently-touched are probed first.
_MOVIE_PROBE_CAP = 100


def _reconcile_inprogress_movies(
    db: Session, plex: PlexClient, link: PlexLink, on_deck_ids: set[int]
) -> dict[str, int]:
    """Close the movie "stuck in progress" gap that the history + onDeck feeds structurally miss.

    When a movie is finished on Plex (often by scrubbing through the end credits) Plex sets
    ``viewCount>0``, clears ``viewOffset`` and DROPS the movie from Continue-Watching. So onDeck no
    longer lists it (and onDeck can only ever ADD an in-progress row, never resolve one to watched —
    see `_pull_apply`), and a scrub-to-end frequently emits no `/status/sessions/history` session row,
    so the history feed misses it too. The stale local ``status='new', position>0`` row then survives
    until the *daily* full reconcile. Movies are hit hardest: a movie is a single leaf with no
    episode-rollup to mask the stuck row, and once watched it leaves onDeck entirely.

    Fix: for each local in-progress MOVIE that Plex's onDeck no longer lists, read the item's current
    per-(token-)account ``viewCount``/``viewOffset`` with one cheap `metadata()` call and apply it via
    the shared `_pull_apply` — flipping to watched when Plex says so. Returns ``{"watched", "in_progress",
    "probed"}`` counts. Episodes are intentionally left to onDeck/history (they promote the next episode
    and aren't affected). Note: like the rest of this job, `metadata()` reads the TOKEN account (the
    owner) — a HTPC playing under a managed Plex user (accountID>1) is out of scope here as elsewhere."""
    conds = [
        PlexState.user_id == link.user_id,
        PlexState.status.notin_(("watched", "hidden")),
        PlexState.position_seconds > _PROGRESS_MIN_S,
        PlexItem.kind == "movie",
    ]
    if on_deck_ids:
        conds.append(PlexState.item_id.notin_(on_deck_ids))
    rows = db.execute(
        select(PlexState, PlexItem.rating_key)
        .join(PlexItem, PlexItem.id == PlexState.item_id)
        .where(*conds)
        .order_by(PlexState.progress_updated_at.desc().nullslast())
        .limit(_MOVIE_PROBE_CAP + 1)
    ).all()
    if len(rows) > _MOVIE_PROBE_CAP:
        rows = rows[:_MOVIE_PROBE_CAP]
        log.warning(
            "in-progress movie probe hit the %d cap (user %s); reconcile catches the tail",
            _MOVIE_PROBE_CAP,
            link.user_id,
        )
    out = {"watched": 0, "in_progress": 0, "probed": 0}
    for st, rk in rows:
        try:
            meta = plex.metadata(rk)
        except PlexError as e:
            log.warning(
                "movie watch-probe failed (user %s, item %s): %s", link.user_id, st.item_id, e
            )
            continue
        out["probed"] += 1
        if meta is None:
            continue
        res = _plex_watch_to_state(
            meta.get("viewCount"), meta.get("viewOffset"), meta.get("lastViewedAt")
        )
        if res is None:
            # Plex holds no watch signal now (viewCount 0, no offset) — don't clear it here; an un-watch
            # is the daily reconcile's job, and a Siftlode-only in-progress row must be preserved.
            continue
        status, pos, watched_at, prog_at = res
        ch = _pull_apply(db, link.user_id, st.item_id, status, pos, watched_at or prog_at)
        if ch:
            out[ch] += 1
    return out


def run_plex_watch_sync(db: Session) -> dict:
    """Scheduler job (default 15 min) — the cheap incremental pass. Pulls Plex-side changes since the
    high-water mark (watch history + on-deck, filtered to the owner account) into Siftlode, resolving a
    divergence to Plex (the source of truth — see `_pull_apply`), reconciles stale in-progress movies
    Plex has since finished (see `_reconcile_inprogress_movies`), then re-pushes any still-unsynced
    local states. Does NOT propagate un-watches (the history feed only carries positive views) — that's
    `run_plex_watch_reconcile`'s job."""
    if not sysconfig.get_bool(db, "plex_enabled"):
        return {"skipped": "disabled"}
    total = {"links": 0, "watched": 0, "in_progress": 0, "repushed": 0, "movie_probes": 0}
    for link in _active_sync_links(db):
        total["links"] += 1
        try:
            with PlexClient(db) as plex:
                acct = _resolve_account_id(db, plex, link)
                since = int(link.last_watch_sync_at.timestamp()) if link.last_watch_sync_at else 0
                now = datetime.now(timezone.utc)
                rk_to_id = _rk_to_id(db)

                for row in plex.watch_history(min_viewed_at=since, account_id=acct):
                    iid = rk_to_id.get(str(row.get("ratingKey") or ""))
                    if iid is None:
                        continue
                    ch = _pull_apply(
                        db, link.user_id, iid, "watched", 0, _epoch(row.get("viewedAt"))
                    )
                    if ch:
                        total[ch] += 1

                # onDeck resolves resume-position changes AND tells us which items Plex still considers
                # in-progress (needed by the movie reconcile below to know which local rows are stale).
                on_deck_ids: set[int] = set()
                for row in plex.on_deck():
                    iid = rk_to_id.get(str(row.get("ratingKey") or ""))
                    if iid is None:
                        continue
                    on_deck_ids.add(iid)
                    off = int(row.get("viewOffset") or 0) // 1000
                    if off < _PROGRESS_MIN_S:
                        continue
                    ch = _pull_apply(
                        db, link.user_id, iid, "new", off, _epoch(row.get("lastViewedAt"))
                    )
                    if ch:
                        total[ch] += 1

                mv = _reconcile_inprogress_movies(db, plex, link, on_deck_ids)
                total["watched"] += mv["watched"]
                total["in_progress"] += mv["in_progress"]
                total["movie_probes"] += mv["probed"]

                total["repushed"] += _repush_dirty(db, plex, link)
                link.last_watch_sync_at = now
                db.commit()
        # silent-ok: the `{"skipped": …}` return is the report, as above.
        except PlexNotConfigured as e:
            return {"skipped": str(e)}
        except PlexError as e:
            db.rollback()
            log.warning("Plex watch sync failed (user %s): %s", link.user_id, e)
    log.info("Plex watch sync: %s", total)
    return total


def run_plex_watch_reconcile(db: Session) -> dict:
    """Scheduler job (default daily) — the authoritative full reconcile. Rescans every section and,
    per item, uses `synced_to_plex` to settle what the incremental feed can't:
      - Plex has a state, Siftlode differs → Plex wins (the source of truth — see `_pull_apply`).
      - Siftlode has a state, Plex has NO record:
          · synced_to_plex=True  → Plex un-watched it since our last sync → clear it (un-watch).
          · synced_to_plex=False → a local change that never reached Plex → re-push it (via
            `_repush_dirty`), never dropped.
      - `hidden` rows are Siftlode-only and never cleared.
    Union-preserving: a never-synced Siftlode-only state is pushed up, not lost."""
    if not sysconfig.get_bool(db, "plex_enabled"):
        return {"skipped": "disabled"}
    total = {"links": 0, "applied": 0, "cleared": 0, "repushed": 0, "scanned": 0}
    for link in _active_sync_links(db):
        total["links"] += 1
        try:
            with PlexClient(db) as plex:
                _resolve_account_id(db, plex, link)
                plex_states, scanned = _scan_plex_states(db, plex)
                total["scanned"] += scanned

                for item_id, (status, pos, watched_at, prog_at) in plex_states.items():
                    ch = _pull_apply(db, link.user_id, item_id, status, pos, watched_at or prog_at)
                    if ch:
                        total["applied"] += 1

                # Siftlode rows Plex no longer has a record for: clear the ones that were mirrored
                # (Plex un-watched them); leave hidden + dirty (dirty gets re-pushed just below).
                for st in (
                    db.execute(select(PlexState).where(PlexState.user_id == link.user_id))
                    .scalars()
                    .all()
                ):
                    if st.item_id in plex_states or st.status == "hidden":
                        continue
                    if st.synced_to_plex:
                        db.delete(st)
                        total["cleared"] += 1

                total["repushed"] += _repush_dirty(db, plex, link)
                link.last_watch_sync_at = datetime.now(timezone.utc)
                db.commit()
        # silent-ok: the `{"skipped": …}` return is the report, as above.
        except PlexNotConfigured as e:
            return {"skipped": str(e)}
        except PlexError as e:
            db.rollback()
            log.warning("Plex watch reconcile failed (user %s): %s", link.user_id, e)
    log.info("Plex watch reconcile: %s", total)
    return total