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 (