Merge: Plex watch-state sync Phase A (Plex→Siftlode import)

backend/alembic/versions/0051_plex_link.py

@@ -0,0 +1,50 @@
+"""plex_link: per-user Plex account link for two-way watch-state sync
+
+Phase A of the Plex ↔ Siftlode watch-state sync. One row per Siftlode user who syncs their watch
+state with a Plex account. The owner MVP uses the server admin token (uses_admin=True, no personal
+token); a personal token column is reserved for later friend-account linking (PIN OAuth). See the
+PlexLink model docstring.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0051_plex_link"
+down_revision: Union[str, None] = "0050_asset_poster"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "plex_link",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id",
+            sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("uses_admin", sa.Boolean(), server_default="true", nullable=False),
+        sa.Column("plex_token_enc", sa.Text(), nullable=True),
+        sa.Column("plex_account_id", sa.Integer(), nullable=True),
+        sa.Column("plex_username", sa.String(length=255), nullable=True),
+        sa.Column("sync_enabled", sa.Boolean(), server_default="false", nullable=False),
+        sa.Column("initial_import_done", sa.Boolean(), server_default="false", nullable=False),
+        sa.Column("last_watch_sync_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("linked_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.UniqueConstraint("user_id", name="uq_plex_link_user"),
+    )
+    op.create_index("ix_plex_link_user_id", "plex_link", ["user_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_plex_link_user_id", table_name="plex_link")
+    op.drop_table("plex_link")

backend/app/config.py

@@ -202,6 +202,8 @@ 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).
+    plex_watch_sync_interval_min: int = 30
     # 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

backend/app/models.py

@@ -1123,6 +1123,37 @@ class PlexState(Base, UpdatedAtMixin):
     )
 
 
+class PlexLink(Base, UpdatedAtMixin):
+    """Links a Siftlode user to a Plex account for two-way watch-state sync (one row per user).
+
+    For the owner (the MVP scope) the row uses the server's admin token — `uses_admin=True`,
+    `plex_token_enc=None` — so no separate Plex login is needed; the admin token already reads/writes
+    the owner account's watch state. `plex_account_id` is that Plex account's numeric id, used to
+    filter the shared watch-history feed to this user (Phase C). Reserved for later: a friend can link
+    their OWN Plex account via PIN OAuth, storing a personal `plex_token_enc` (uses_admin=False) — the
+    sync loop then iterates link rows uniformly.
+
+    `initial_import_done` guards the one-time "Plex is master" import; `last_watch_sync_at` is the
+    high-water mark for the incremental Plex→Siftlode pass."""
+
+    __tablename__ = "plex_link"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(
+        ForeignKey("users.id", ondelete="CASCADE"), unique=True, index=True
+    )
+    uses_admin: Mapped[bool] = mapped_column(Boolean, default=True, server_default="true")
+    plex_token_enc: Mapped[str | None] = mapped_column(Text)  # P5b friend tokens (encrypted)
+    plex_account_id: Mapped[int | None] = mapped_column(Integer)
+    plex_username: Mapped[str | None] = mapped_column(String(255))
+    sync_enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
+    initial_import_done: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false"
+    )
+    last_watch_sync_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    linked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+
 class PlexPlaylist(Base, TimestampMixin, UpdatedAtMixin):
     """A per-user ORDERED watch-list of Plex items — Siftlode-native (lives in our own DB, works for
     users WITHOUT a Plex account, like plex_states). Unlike collections (shared, library-level), a

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

backend/app/routes/plex.py

@@ -29,6 +29,7 @@ from app.models import (
     PlexCollection,
     PlexItem,
     PlexLibrary,
+    PlexLink,
     PlexPlaylist,
     PlexPlaylistItem,
     PlexSeason,
@@ -39,6 +40,7 @@ from app.models import (
 from app.plex import paths as plex_paths
 from app.plex import stream as plex_stream
 from app.plex import sync as plex_sync
+from app.plex import watch_sync as plex_watch
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
 from app.routes.admin import admin_user
 from app.routes.feed import _to_tsquery_str
@@ -122,6 +124,73 @@ def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -
     return plex_sync.sync(db)
 
 
+# --- Watch-state sync (Plex ↔ Siftlode) -------------------------------------------------------
+# Phase A: the owner links their Siftlode account to the Plex admin account and does a one-time
+# "Plex is master" import. Two-way push/pull lands in later phases. Admin-only for now because it
+# rides the server admin token (which reads/writes the OWNER Plex account's state).
+
+
+def _link_dict(link: PlexLink | None) -> dict:
+    return {
+        "linked": link is not None,
+        "sync_enabled": bool(link and link.sync_enabled),
+        "uses_admin": bool(link and link.uses_admin),
+        "initial_import_done": bool(link and link.initial_import_done),
+        "last_watch_sync_at": (
+            link.last_watch_sync_at.isoformat()
+            if link and link.last_watch_sync_at
+            else None
+        ),
+    }
+
+
+@router.get("/watch/link")
+def watch_link_status(
+    user: User = Depends(current_user), db: Session = Depends(get_db)
+) -> dict:
+    """This user's Plex watch-sync link status (drives the Settings toggle)."""
+    link = db.query(PlexLink).filter_by(user_id=user.id).first()
+    return _link_dict(link)
+
+
+@router.post("/watch/link")
+def watch_link_set(
+    payload: dict,
+    user: User = Depends(admin_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    """Admin: turn two-way Plex watch-sync on/off for THIS (owner) account. The link uses the server
+    admin token, so no separate Plex login. Enabling it for the first time runs the one-time
+    "Plex is master" import immediately so the owner's existing Plex history shows up in Siftlode."""
+    if not sysconfig.get_bool(db, "plex_enabled"):
+        raise HTTPException(status_code=409, detail="The Plex module is disabled.")
+    enabled = bool(payload.get("enabled"))
+    link = db.query(PlexLink).filter_by(user_id=user.id).first()
+    if link is None:
+        link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc))
+        db.add(link)
+    link.sync_enabled = enabled
+    db.commit()
+
+    result: dict = {}
+    if enabled and not link.initial_import_done:
+        result = plex_watch.import_owner_watch_state(db, link)
+    return {**_link_dict(link), "import": result or None}
+
+
+@router.post("/watch/import")
+def watch_import_now(
+    user: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
+    """Admin: (re-)run the "Plex is master" import for this account — Plex re-wins on the
+    intersection; Siftlode-only states are preserved. Idempotent."""
+    link = db.query(PlexLink).filter_by(user_id=user.id).first()
+    if link is None or not link.sync_enabled:
+        raise HTTPException(status_code=409, detail="Enable Plex watch-sync first.")
+    result = plex_watch.import_owner_watch_state(db, link)
+    return {**_link_dict(link), "import": result}
+
+
 # --- Read: libraries / browse / show / image --------------------------------------------------
 
 

backend/app/sysconfig.py

@@ -84,6 +84,8 @@ SPECS: tuple[ConfigSpec, ...] = (
     ConfigSpec("plex_libraries", "str", "plex", "plex_libraries"),
     # (plex_sync_interval_min is tuned from the Scheduler dashboard, not here — it's a scheduler job.)
     ConfigSpec("plex_max_transcodes", "int", "plex", "plex_max_transcodes", min=0, max=16),
+    # NOTE: plex_watch_sync_interval_min (config.py default) is surfaced here in Phase C, when the
+    # incremental watch-sync scheduler job that reads it lands.
 )
 
 _BY_KEY: dict[str, ConfigSpec] = {s.key: s for s in SPECS}

frontend/src/components/SettingsPanel.tsx

@@ -1,7 +1,7 @@
 import { useState } from "react";
 import { useTranslation } from "react-i18next";
-import { useQueryClient } from "@tanstack/react-query";
+import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
-import { Bell, Monitor, Trash2, User } from "lucide-react";
+import { Bell, Monitor, RefreshCw, Trash2, User } from "lucide-react";
 import { SCHEMES, type Scheme, type ThemePrefs } from "../lib/theme";
 import { api, type Me } from "../lib/api";
 import { LS, useAccountPersistedState } from "../lib/storage";
@@ -426,6 +426,70 @@ function SignInMethods({ me }: { me: Me }) {
   );
 }
 
+// Two-way Plex watch-state sync (Phase A): the owner links their account to the Plex admin account
+// and does a one-time "Plex is master" import. Admin-only (rides the server admin token) and only
+// shown when the Plex module is enabled. Two-way push/pull arrive in later phases.
+function PlexWatchSync() {
+  const { t, i18n } = useTranslation();
+  const qc = useQueryClient();
+  const link = useQuery({ queryKey: ["plex-watch-link"], queryFn: () => api.plexWatchLink() });
+  const enabled = link.data?.sync_enabled ?? false;
+
+  const announce = (imp: { watched: number; in_progress: number } | null | undefined) => {
+    if (imp) {
+      notify({
+        level: "success",
+        message: t("settings.plexSync.imported", { watched: imp.watched, in_progress: imp.in_progress }),
+      });
+      // The browse grid's watch badges read plex_states — refresh them.
+      qc.invalidateQueries({ queryKey: ["plex"] });
+    }
+  };
+
+  const toggle = useMutation({
+    mutationFn: (next: boolean) => api.plexWatchSetLink(next),
+    onSuccess: (res) => {
+      qc.setQueryData(["plex-watch-link"], res);
+      announce(res.import);
+    },
+  });
+  const reimport = useMutation({
+    mutationFn: () => api.plexWatchImport(),
+    onSuccess: (res) => {
+      qc.setQueryData(["plex-watch-link"], res);
+      announce(res.import);
+    },
+  });
+
+  const last = link.data?.last_watch_sync_at
+    ? new Date(link.data.last_watch_sync_at).toLocaleString(i18n.language)
+    : null;
+
+  return (
+    <Section title={t("settings.plexSync.title")}>
+      <p className="text-xs text-muted leading-relaxed mb-2">{t("settings.plexSync.intro")}</p>
+      <SettingRow label={t("settings.plexSync.toggle")} hint={t("settings.plexSync.toggleHint")}>
+        <Switch checked={enabled} onChange={(v) => toggle.mutate(v)} />
+      </SettingRow>
+      {enabled && (
+        <div className="mt-2 flex items-center justify-between gap-3">
+          <p className="text-xs text-muted">
+            {last ? t("settings.plexSync.lastSync", { when: last }) : t("settings.plexSync.never")}
+          </p>
+          <button
+            onClick={() => reimport.mutate()}
+            disabled={reimport.isPending}
+            className="inline-flex items-center gap-1.5 px-3 py-2 rounded-xl text-sm border border-border hover:bg-card/60 disabled:opacity-50 transition shrink-0"
+          >
+            <RefreshCw className={`w-4 h-4 ${reimport.isPending ? "animate-spin" : ""}`} />
+            {t("settings.plexSync.importNow")}
+          </button>
+        </div>
+      )}
+    </Section>
+  );
+}
+
 function Account({ me, onOpenWizard }: { me: Me; onOpenWizard: () => void }) {
   const { t } = useTranslation();
   const confirm = useConfirm();
@@ -464,6 +528,8 @@ function Account({ me, onOpenWizard }: { me: Me; onOpenWizard: () => void }) {
 
     {!me.is_demo && <SignInMethods me={me} />}
 
+    {me.role === "admin" && me.plex_enabled && <PlexWatchSync />}
+
     {me.is_demo ? (
       <Section title={t("settings.account.youtubeAccess")}>
         <p className="text-xs text-muted leading-relaxed">

frontend/src/i18n/locales/de/settings.json

@@ -5,6 +5,16 @@
     "notifications": "Benachrichtigungen",
     "account": "Konto"
   },
+  "plexSync": {
+    "title": "Plex-Wiedergabesync",
+    "intro": "Halte den Status „gesehen / wird gerade angesehen“ zwischen Plex und Siftlode synchron. Beim Einschalten wird dein vorhandener Plex-Verlauf einmalig importiert (Plex gewinnt); die laufende Zwei-Wege-Synchronisierung folgt.",
+    "toggle": "Zwei-Wege-Sync mit Plex",
+    "toggleHint": "Nutzt das Plex-Konto dieses Servers. Beim Aktivieren wird dein Plex-Wiedergabeverlauf einmalig importiert.",
+    "importNow": "Jetzt aus Plex importieren",
+    "imported": "Aus Plex importiert: {{watched}} gesehen, {{in_progress}} laufend.",
+    "lastSync": "Letzter Import: {{when}}",
+    "never": "Noch nicht importiert"
+  },
   "save": {
     "unsaved": "Nicht gespeicherte Änderungen",
     "saving": "Speichern…",

frontend/src/i18n/locales/en/settings.json

@@ -5,6 +5,16 @@
     "notifications": "Notifications",
     "account": "Account"
   },
+  "plexSync": {
+    "title": "Plex watch sync",
+    "intro": "Sync your watched / in-progress status between Plex and Siftlode. Turning this on imports your existing Plex history now (Plex wins); ongoing two-way sync follows.",
+    "toggle": "Two-way sync with Plex",
+    "toggleHint": "Uses this server's Plex account. Turning it on imports your Plex watch history once.",
+    "importNow": "Import from Plex now",
+    "imported": "Imported from Plex: {{watched}} watched, {{in_progress}} in progress.",
+    "lastSync": "Last import: {{when}}",
+    "never": "Not imported yet"
+  },
   "save": {
     "unsaved": "Unsaved changes",
     "saving": "Saving…",

frontend/src/i18n/locales/hu/settings.json

@@ -5,6 +5,16 @@
     "notifications": "Értesítések",
     "account": "Fiók"
   },
+  "plexSync": {
+    "title": "Plex nézettség-szinkron",
+    "intro": "Tartsd szinkronban a megnézett / folyamatban lévő státuszt a Plex és a Siftlode között. A bekapcsolás most beimportálja a meglévő Plex-előzményedet (a Plex az irányadó); a folyamatos kétirányú szinkron ezután jön.",
+    "toggle": "Kétirányú szinkron a Plexszel",
+    "toggleHint": "A szerver Plex-fiókját használja. Bekapcsoláskor egyszer beimportálja a Plex nézettségi előzményedet.",
+    "importNow": "Importálás a Plexből most",
+    "imported": "Importálva a Plexből: {{watched}} megnézve, {{in_progress}} folyamatban.",
+    "lastSync": "Utolsó import: {{when}}",
+    "never": "Még nincs importálva"
+  },
   "save": {
     "unsaved": "Nem mentett változások",
     "saving": "Mentés…",

frontend/src/lib/api.ts

@@ -810,6 +810,20 @@ export interface PlexPlaySession {
   start_s: number;
 }
 
+// Two-way Plex watch-state sync link status (owner account; Phase A).
+export interface PlexWatchLink {
+  linked: boolean;
+  sync_enabled: boolean;
+  uses_admin: boolean;
+  initial_import_done: boolean;
+  last_watch_sync_at: string | null;
+}
+export interface PlexWatchImport {
+  watched: number;
+  in_progress: number;
+  scanned: number;
+}
+
 // Admin Users & roles page.
 export interface AdminUserRow {
   id: number;
@@ -1261,6 +1275,12 @@ export const api = {
       method: "POST",
       body: JSON.stringify({ status }),
     }),
+  // Two-way Plex watch-sync (Phase A: owner link + one-time "Plex is master" import).
+  plexWatchLink: (): Promise<PlexWatchLink> => req("/api/plex/watch/link"),
+  plexWatchSetLink: (enabled: boolean): Promise<PlexWatchLink & { import: PlexWatchImport | null }> =>
+    req("/api/plex/watch/link", { method: "POST", body: JSON.stringify({ enabled }) }),
+  plexWatchImport: (): Promise<PlexWatchLink & { import: PlexWatchImport }> =>
+    req("/api/plex/watch/import", { method: "POST" }),
   plexStreamFileUrl: (id: string): string => `/api/plex/stream/${encodeURIComponent(id)}/file`,
   plexDownloadUrl: (id: string): string =>
     `/api/plex/stream/${encodeURIComponent(id)}/file?download=1`,