feat(plex): Phase A — one-time Plex→Siftlode watch-state import

Plex records watch state per Plex account; Siftlode per user in plex_states. A new plex_link table
maps a Siftlode user to a Plex account; for the owner (MVP) the row uses the server admin token
(uses_admin=True), so no separate Plex login. app/plex/watch_sync.py reads the owner account's
viewCount/viewOffset/lastViewedAt (already present in the catalog mirror's section listing) and
upserts them into the owner's plex_states — 'Plex is master' on this first import, but only where
Plex has a watch record (Siftlode-only states are preserved; union on the intersection). Idempotent.

Admin routes: GET/POST /api/plex/watch/link (status + enable/disable; first enable runs the import)
and POST /api/plex/watch/import (re-run). Migration 0051_plex_link. Two-way push (Phase B) and the
incremental Plex→Siftlode reconcile (Phase C) build on this in later ships.

backend/app/plex/watch_sync.py

@@ -0,0 +1,124 @@
+"""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, timezone
+
+from sqlalchemy.orm import Session
+
+from app import sysconfig
+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 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"}
+
+    # rating_key -> plex_items.id for every mirrored playable leaf (movie/episode).
+    item_id_by_rk: dict[str, int] = {
+        rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)
+    }
+    # This user's existing states, by item — so we upsert rather than duplicate.
+    existing: dict[int, PlexState] = {
+        st.item_id: st
+        for st in db.query(PlexState).filter_by(user_id=link.user_id)
+    }
+    stats = {"watched": 0, "in_progress": 0, "scanned": 0}
+    wanted = _enabled_section_keys(db)
+
+    try:
+        with PlexClient(db) as plex:
+            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):
+                    stats["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 None:
+                        continue
+                    status, pos, watched_at, prog_at = res
+                    st = existing.get(item_id)
+                    if st is None:
+                        st = PlexState(user_id=link.user_id, item_id=item_id)
+                        db.add(st)
+                        existing[item_id] = st
+                    st.status = status
+                    st.position_seconds = pos
+                    st.watched_at = watched_at
+                    st.progress_updated_at = prog_at
+                    # This state now mirrors Plex — mark it so Phase B's push doesn't bounce it back.
+                    st.synced_to_plex = True
+                    stats["watched" if status == "watched" else "in_progress"] += 1
+            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