From 5592f60e854b74e156fe81083f69700e4a429a97 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 9 Jul 2026 15:18:19 +0200 Subject: [PATCH 1/2] fix(plex): make watch import concurrency-safe (upsert), guard toggle MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Plex watch import inserted plex_states via per-row ORM add against a pre-read 'existing' set, which is not safe when the enable call fires more than once while the multi-second import runs (the toggle wasn't disabled during it): overlapping imports raced to INSERT the same (user_id, item_id) rows and all but the first hit a UniqueViolation -> 500 (seen on prod). Rewrite the write path as a chunked PostgreSQL UPSERT (on_conflict_do_update, rows de-duped by item_id) — idempotent and concurrency-safe, Plex still wins on the intersection and Siftlode-only states are untouched. Also disable the Settings toggle (and ignore its onChange) while a sync mutation is pending, so it can't be fired repeatedly; adds a reusable 'disabled' prop to the Switch primitive. --- backend/app/plex/watch_sync.py | 50 +++++++++++++++-------- frontend/src/components/SettingsPanel.tsx | 6 ++- frontend/src/components/ui/form.tsx | 5 ++- 3 files changed, 43 insertions(+), 18 deletions(-) diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 0b5fae2..8763785 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -16,6 +16,8 @@ history + onDeck, last-write-wins) build on this and land in later ships. import logging from datetime import datetime, timezone +from sqlalchemy import func +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app import sysconfig @@ -67,13 +69,12 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: 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) + # 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: @@ -98,18 +99,35 @@ def import_owner_watch_state(db: Session, link: PlexLink) -> dict: 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 + rows_by_item[item_id] = { + "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 + # in rows_by_item) are never referenced, so they're preserved untouched (union). + rows = list(rows_by_item.values()) + 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() diff --git a/frontend/src/components/SettingsPanel.tsx b/frontend/src/components/SettingsPanel.tsx index f98421d..5556a0a 100644 --- a/frontend/src/components/SettingsPanel.tsx +++ b/frontend/src/components/SettingsPanel.tsx @@ -469,7 +469,11 @@ function PlexWatchSync() {

{t("settings.plexSync.intro")}

- toggle.mutate(v)} /> + !toggle.isPending && toggle.mutate(v)} + /> {enabled && (
diff --git a/frontend/src/components/ui/form.tsx b/frontend/src/components/ui/form.tsx index 8171d24..9c6dc36 100644 --- a/frontend/src/components/ui/form.tsx +++ b/frontend/src/components/ui/form.tsx @@ -10,10 +10,12 @@ export function Switch({ checked, onChange, label, + disabled, }: { checked: boolean; onChange: (v: boolean) => void; label?: string; + disabled?: boolean; }) { return (