diff --git a/VERSION b/VERSION index 4d17807..4d8ac4d 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.34.2 \ No newline at end of file +0.35.0 \ No newline at end of file diff --git a/backend/app/config.py b/backend/app/config.py index 0f8defb..c83f0f7 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -202,8 +202,13 @@ class Settings(BaseSettings): plex_client_id: str = "siftlode-server" # How often the catalog sync job mirrors Plex metadata, in minutes. plex_sync_interval_min: int = 360 - # How often the incremental Plex→Siftlode watch-state reconcile runs, in minutes (Phase C). + # How often the incremental Plex→Siftlode watch-state reconcile runs, in minutes (Phase C): + # a cheap history + on-deck pull plus a re-push of any states that failed their immediate push. plex_watch_sync_interval_min: int = 30 + # How often the FULL watch-state reconcile runs, in minutes (Phase C). A whole-section rescan + # that catches what the incremental history feed can't — notably an un-watch done on the Plex + # side — so it runs far less often (daily) than the incremental pass. + plex_watch_reconcile_interval_min: int = 1440 # Max concurrent transcodes for the P3 fallback. Low by default — CPU-only transcode is # expensive; direct-serve (browser-compatible files) has no such limit. plex_max_transcodes: int = 1 diff --git a/backend/app/plex/client.py b/backend/app/plex/client.py index 2dd0af6..034d6d4 100644 --- a/backend/app/plex/client.py +++ b/backend/app/plex/client.py @@ -166,6 +166,73 @@ class PlexClient: items = mc.get("Metadata", []) or [] return items[0] if items else None + # --- Watch-state write-back (P5 Phase B — Siftlode→Plex push; acts on the token's account) --- + _LIBRARY_IDENTIFIER = "com.plexapp.plugins.library" + + def scrobble(self, rating_key: str) -> None: + """Mark the item *watched* on the token's Plex account (viewCount++, resume cleared).""" + self._write( + "GET", + f"/:/scrobble?identifier={self._LIBRARY_IDENTIFIER}" + f"&key={quote(str(rating_key), safe='')}", + ) + + def unscrobble(self, rating_key: str) -> None: + """Mark the item *unwatched* on the token's Plex account (clears viewCount).""" + self._write( + "GET", + f"/:/unscrobble?identifier={self._LIBRARY_IDENTIFIER}" + f"&key={quote(str(rating_key), safe='')}", + ) + + def set_timeline( + self, rating_key: str, time_ms: int, duration_ms: int, state: str = "stopped" + ) -> None: + """Report a resume position (viewOffset) for the item on the token's account. `state` is a + Plex timeline state (playing|paused|stopped); we push the settled position as ``stopped``.""" + key = quote(f"/library/metadata/{rating_key}", safe="") + self._write( + "GET", + f"/:/timeline?identifier={self._LIBRARY_IDENTIFIER}" + f"&ratingKey={quote(str(rating_key), safe='')}&key={key}" + f"&state={state}&time={int(time_ms)}&duration={int(duration_ms)}", + ) + + # --- Watch-state read for incremental sync (P5 Phase C) --- + def accounts(self) -> list[dict]: + """Server accounts (the owner + any managed/home users): id, name. Used to resolve the + owner's accountID so the shared watch-history feed can be filtered to just their views.""" + return self._get("/accounts").get("Account", []) or [] + + def watch_history( + self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500 + ) -> list[dict]: + """The most recent watch-history rows (ratingKey / viewedAt / accountID / type), newest + first — works without Plex Pass. Filtered client-side to `account_id` and to rows at/after + `min_viewed_at` (epoch seconds): the efficient "what changed since T" feed. Sorted desc, so + we stop at the first row older than the cutoff.""" + mc = self._get( + "/status/sessions/history/all", + params={ + "sort": "viewedAt:desc", + "X-Plex-Container-Start": 0, + "X-Plex-Container-Size": size, + }, + ) + out = [] + for r in mc.get("Metadata", []) or []: + if int(r.get("viewedAt") or 0) < min_viewed_at: + break + if account_id is not None and int(r.get("accountID") or -1) != account_id: + continue + out.append(r) + return out + + def on_deck(self) -> list[dict]: + """The token account's Continue-Watching list (ratingKey / viewOffset / lastViewedAt / + type) — the incremental source for resume-position changes made on the Plex side.""" + return self._get("/library/onDeck").get("Metadata", []) or [] + def raw_get(self, path: str) -> bytes: """Raw bytes of an arbitrary Plex resource path (e.g. an external subtitle stream `key` like ``/library/streams/553184``). Keeps the admin token server-side.""" diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 8763785..42a9d37 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -14,13 +14,14 @@ Phases B (Siftlode→Plex push via scrobble/timeline) and C (incremental Plex→ history + onDeck, last-write-wins) build on this and land in later ships. """ import logging -from datetime import datetime, timezone +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 ( @@ -58,6 +59,39 @@ def _plex_watch_to_state( return None +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: dict[str, int] = { + rk: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id) + } + 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 @@ -65,41 +99,17 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: 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) - } stats = {"watched": 0, "in_progress": 0, "scanned": 0} - wanted = _enabled_section_keys(db) - # Accumulate 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 — the conflict just re-updates. - rows_by_item: dict[int, dict] = {} - 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 - rows_by_item[item_id] = { + 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, @@ -109,11 +119,11 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: # 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 + ) + stats["watched" if status == "watched" else "in_progress"] += 1 # Bulk upsert in chunks. Plex wins on the intersection; Siftlode-only states (item_ids not - # in rows_by_item) are never referenced, so they're preserved untouched (union). - rows = list(rows_by_item.values()) + # 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( @@ -140,3 +150,268 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: 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 push_state_to_plex( + user_id: int, + item_id: int, + rating_key: str, + action: str, + time_ms: int = 0, + duration_ms: int = 0, +) -> 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).""" + 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). + st = db.query(PlexState).filter_by(user_id=user_id, item_id=item_id).first() + if st is not None: + st.synced_to_plex = True + 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. The server owner is accountID 1 on their own + server; we confirm + grab their username from /accounts, falling back to 1 if 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.""" + rk_by_id = dict(db.query(PlexItem.id, PlexItem.rating_key)) + 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: + rk = rk_by_id.get(st.item_id) + if rk is None: + continue + try: + if st.status == "watched": + plex.scrobble(rk) + elif st.position_seconds and st.position_seconds >= _PROGRESS_MIN_S: + dur = int((plex.metadata(rk) or {}).get("duration") or 0) + plex.set_timeline(rk, st.position_seconds * 1000, dur) + 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: iid for rk, iid in db.query(PlexItem.rating_key, PlexItem.id)} + + 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 diff --git a/backend/app/routes/plex.py b/backend/app/routes/plex.py index 19d390f..f44b5ec 100644 --- a/backend/app/routes/plex.py +++ b/backend/app/routes/plex.py @@ -16,7 +16,7 @@ from pathlib import Path from urllib.parse import quote import httpx -from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response from fastapi.responses import FileResponse from sqlalchemy import and_, func, or_, text from sqlalchemy.orm import Session, aliased @@ -1486,22 +1486,49 @@ def item_detail( } +def _push_watch( + background: BackgroundTasks, + db: Session, + user_id: int, + item_id: int, + rating_key: str, + action: str, + time_ms: int = 0, + duration_ms: int = 0, +) -> None: + """Schedule a best-effort Siftlode→Plex watch push, but only for a user whose owner link has + push enabled — non-owner users never touch Plex. Guarding here (with the request session) avoids + spinning up a throwaway background session on every state change for the common no-link case.""" + if plex_watch.link_for_push(db, user_id) is None: + return + background.add_task( + plex_watch.push_state_to_plex, user_id, item_id, rating_key, action, time_ms, duration_ms + ) + + @router.post("/item/{rating_key}/progress") def item_progress( rating_key: str, payload: dict, + background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - """Checkpoint the resume position while playing (near the end → mark watched + clear it).""" + """Checkpoint the resume position while playing (near the end → mark watched + clear it). + + `final` (sent on the pause/pagehide/unmount flush, not the periodic 10s checkpoint) is what + debounces the Plex resume push: we mirror the settled position to Plex only on a final flush, so + a single watch doesn't spray `/:/timeline` calls at the server every 10 seconds.""" it = _item_or_404(db, rating_key) try: position = max(0, int(payload.get("position_seconds") or 0)) duration = int(payload.get("duration_seconds") or 0) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="position_seconds must be an integer") + final = bool(payload.get("final")) now = datetime.now(timezone.utc) st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() + was_watched = st is not None and st.status == "watched" near_end = duration > 0 and position > duration - _FINISH_MARGIN_S if near_end: @@ -1512,7 +1539,12 @@ def item_progress( st.watched_at = now st.position_seconds = 0 st.progress_updated_at = now + st.synced_to_plex = False db.commit() + # Immediate on the watched transition; skip a repeat scrobble if it was already watched + # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again). + if not was_watched: + _push_watch(background, db, user.id, it.id, rating_key, "watched") return {"status": "watched", "position_seconds": 0} if position < _PROGRESS_MIN_S: @@ -1529,7 +1561,13 @@ def item_progress( db.add(st) st.position_seconds = position st.progress_updated_at = now + if final: + st.synced_to_plex = False db.commit() + if final: + _push_watch( + background, db, user.id, it.id, rating_key, "resume", position * 1000, duration * 1000 + ) return {"position_seconds": position} @@ -1537,19 +1575,25 @@ def item_progress( def item_state( rating_key: str, payload: dict, + background: BackgroundTasks, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - """Set the per-user watch status (new|watched|hidden).""" + """Set the per-user watch status (new|watched|hidden). watched→scrobble / →new→unscrobble push + to a linked Plex account immediately; `hidden` is a Siftlode-only concept with no Plex analogue, + so it never touches Plex.""" it = _item_or_404(db, rating_key) status = payload.get("status") if status not in ("new", "watched", "hidden"): raise HTTPException(status_code=400, detail="Invalid status") st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first() + prev_status = st.status if st is not None else "new" if status == "new": if st is not None: db.delete(st) db.commit() + if prev_status != "new": # something → unwatched: mirror the un-mark to Plex + _push_watch(background, db, user.id, it.id, rating_key, "unwatched") return {"status": "new"} if st is None: st = PlexState(user_id=user.id, item_id=it.id) @@ -1558,7 +1602,10 @@ def item_state( if status == "watched": st.watched_at = datetime.now(timezone.utc) st.position_seconds = 0 + st.synced_to_plex = False db.commit() + if status == "watched" and prev_status != "watched": + _push_watch(background, db, user.id, it.id, rating_key, "watched") return {"status": status} diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 29d96b3..5e001b6 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -15,6 +15,7 @@ from app.db import SessionLocal from app.downloads.gc import run_download_gc from app.models import SchedulerSetting from app.plex.sync import sync as run_plex_sync +from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync from app.notifications import create_notification from app.state import is_sync_paused from app.sync.autotag import run_autotag_all @@ -63,6 +64,8 @@ JOB_INTERVALS: dict[str, int] = { "explore_cleanup": settings.explore_cleanup_minutes, "download_gc": settings.download_gc_minutes, "plex_sync": settings.plex_sync_interval_min, + "plex_watch_sync": settings.plex_watch_sync_interval_min, + "plex_watch_reconcile": settings.plex_watch_reconcile_interval_min, } # Sane bounds for an admin-set interval (minutes). @@ -292,6 +295,18 @@ def _plex_sync_job() -> None: _job("plex_sync", run_plex_sync) +def _plex_watch_sync_job() -> None: + # Incremental Plex→Siftlode watch-state pull (history + on-deck) + re-push of unsynced local + # states. Cheap; a no-op when Plex is disabled or no owner link has sync enabled. + _job("plex_watch_sync", run_plex_watch_sync) + + +def _plex_watch_reconcile_job() -> None: + # Full watch-state reconcile (whole-section rescan) — catches un-watches the incremental feed + # can't. Heavier, so it runs far less often (daily default). + _job("plex_watch_reconcile", run_plex_watch_reconcile) + + # job_id -> wrapper. The single source of truth for which jobs exist and how to run one, # shared by start_scheduler (recurring registration) and trigger_job (manual "run now"). JOB_FUNCS: dict[str, Callable[[], None]] = { @@ -307,6 +322,8 @@ JOB_FUNCS: dict[str, Callable[[], None]] = { "explore_cleanup": _explore_cleanup_job, "download_gc": _download_gc_job, "plex_sync": _plex_sync_job, + "plex_watch_sync": _plex_watch_sync_job, + "plex_watch_reconcile": _plex_watch_reconcile_job, } diff --git a/frontend/src/components/PlexPlayer.tsx b/frontend/src/components/PlexPlayer.tsx index 792b2dd..4fb7db2 100644 --- a/frontend/src/components/PlexPlayer.tsx +++ b/frontend/src/components/PlexPlayer.tsx @@ -452,23 +452,31 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { const absRef = useRef(0); absRef.current = abs; useEffect(() => { - const flush = () => { + // `final` = a settled checkpoint (pause / page-hide / leaving the player) vs. the periodic tick. + // Only final checkpoints get mirrored to a linked Plex account (backend-side debounce), so Plex + // sees the resume position when the user actually stops — not a timeline call every 10 seconds. + const flush = (final = false) => { if (absRef.current > 0 && durationRef.current > 0) - api.plexProgress(id, Math.floor(absRef.current), durationRef.current).catch(() => {}); + api.plexProgress(id, Math.floor(absRef.current), durationRef.current, final).catch(() => {}); }; const iv = window.setInterval(flush, 10000); + // Pausing is a settled position — push it to Plex now rather than waiting for unmount/unload. + const onPauseFlush = () => flush(true); + const video = videoRef.current; + video?.addEventListener("pause", onPauseFlush); // Save on page unload too (F5/close/navigate): React effect cleanup does NOT run on a full reload, // so without this the resume point lags up to 10s — e.g. a seek right before F5 was lost. Uses a - // keepalive beacon so the POST survives the unload. + // keepalive beacon so the POST survives the unload (a final checkpoint — the user is leaving). const onHide = () => { if (absRef.current > 0 && durationRef.current > 0) - api.plexProgressBeacon(id, Math.floor(absRef.current), durationRef.current); + api.plexProgressBeacon(id, Math.floor(absRef.current), durationRef.current, true); }; window.addEventListener("pagehide", onHide); return () => { window.clearInterval(iv); window.removeEventListener("pagehide", onHide); - flush(); + video?.removeEventListener("pause", onPauseFlush); + flush(true); // Back (unmount) persists the position server-side via flush(), but reopening the same item in // the SAME session read react-query's CACHED detail — whose position_seconds predated this // watch — so the resume landed at the old spot (the progress GET could also race the POST). @@ -558,6 +566,15 @@ export default function PlexPlayer({ itemId, onClose, queue }: Props) { const go = useCallback( (otherId: string | null | undefined) => { if (otherId) { + // Stop the outgoing media BEFORE switching. If the next item fails to load (e.g. no local + // file), loadSession returns without replacing the