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, last-write-wins) build on this and land in later ships.
"""
import logging
from datetime import datetime, timedelta, timezone

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

from app import 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 = logging.getLogger("siftlode.plex.watch")

# 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.query(PlexItem.rating_key, PlexItem.id)}


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()
    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.query(PlexLink).filter_by(user_id=user_id).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.query(PlexState).filter_by(user_id=user_id, item_id=item_id).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.query(PlexState).filter(
                PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids)
            ).update({PlexState.synced_to_plex: True}, synchronize_session=False)
            db.commit()


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

# Clock-skew / round-trip slack when comparing a Plex timestamp to a Siftlode one. A state we just
# pushed (Plex stamps viewedAt a beat after our watched_at) must not read as "newer on Plex" and get
# pulled back with a different shape. Genuine Plex-side edits are minutes from our writes — far
# outside this — so the window only absorbs skew, never real changes.
_LWW_TOLERANCE_S = 90


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.query(PlexLink)
        .filter_by(uses_admin=True, sync_enabled=True, initial_import_done=True)
        .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:
        pass
    link.plex_account_id = acct_id
    if username:
        link.plex_username = username
    db.commit()
    return acct_id


def _sift_ts(st: PlexState) -> datetime | None:
    """The freshest Siftlode-side timestamp on a row, for last-write-wins."""
    return max([t for t in (st.watched_at, st.progress_updated_at) if t], default=None)


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.query(PlexState).filter_by(user_id=user_id, item_id=item_id).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
    # Conflict: keep Siftlode when it's newer than Plex (plus skew slack).
    sts = _sift_ts(st) if st else None
    if sts and plex_ts and plex_ts <= sts + timedelta(seconds=_LWW_TOLERANCE_S):
        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.query(PlexItem.id, PlexItem.rating_key, PlexItem.duration_s)
    }
    dirty = (
        db.query(PlexState)
        .filter(
            PlexState.user_id == link.user_id,
            PlexState.synced_to_plex.is_(False),
            PlexState.status != "hidden",
        )
        .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


def run_plex_watch_sync(db: Session) -> dict:
    """Scheduler job (default 30 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 under
    last-write-wins, 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}
    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

                for row in plex.on_deck():
                    iid = rk_to_id.get(str(row.get("ratingKey") or ""))
                    if iid is None:
                        continue
                    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

                total["repushed"] += _repush_dirty(db, plex, link)
                link.last_watch_sync_at = now
                db.commit()
        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 → last-write-wins (Plex usually the newer here).
      - 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.query(PlexState).filter_by(user_id=link.user_id).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()
        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