fix(plex): make watch import concurrency-safe (upsert), guard toggle

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

@@ -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()