fix: a first watch state and a first Watch later survive two requests at once

The first checkpoint of an item and the final flush beside it both read no
state and both inserted; the loser failed on uq_plex_user_item. Two first saves
each made a Watch later list. Both are created by an ON CONFLICT DO NOTHING
insert and then read; a partial unique index holds one Watch later per user, and
the migration folds an existing second list into the first. Plex keeps one
per-item state map (rollup.item_state_map).

backend/app/plex/queries.py

@@ -108,20 +108,6 @@ def state(db: Session, user: User | int, item_id: int) -> PlexState | None:
     )
 
 
-def states(db: Session, user: User | int, item_ids: list[int]) -> dict[int, PlexState]:
-    """The user's state on each of these items that has one, by item id."""
-    if not item_ids:
-        return {}
-    return {
-        s.item_id: s
-        for s in db.execute(
-            select(PlexState).where(
-                owned_by(PlexState.user_id, user), PlexState.item_id.in_(item_ids)
-            )
-        ).scalars()
-    }
-
-
 # --- a show ----------------------------------------------------------------------------------
 
 

backend/app/plex/rollup.py

@@ -8,6 +8,7 @@ from datetime import datetime, timezone
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.orm import Session, aliased
 
+from app.models import User
 from app.userscope import owned_by
 from app.plex.models import PlexItem, PlexState
 from app.plex.cards import ShowState
@@ -106,7 +107,7 @@ def state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, ShowS
     return {r.show_id: show_state(r.total, r.watched, r.inprog) for r in rows}
 
 
-def item_state_map(db: Session, user_id: int, item_ids: list[int]) -> dict[int, PlexState]:
+def item_state_map(db: Session, user_id: User | int, item_ids: list[int]) -> dict[int, PlexState]:
     """A user's PlexState rows for a set of leaf items (movies or episodes), keyed by item_id — the
     leaf analogue of `state_map` (which rolls a show's episodes up to a `ShowState`). One batched
     fetch, so the caller serializes cards without a per-card query. Empty in → empty dict (no `IN ()`)."""

backend/app/plex/service.py

@@ -19,7 +19,7 @@ from app.plex import people as plex_people
 from app.plex import tmdb as plex_tmdb
 from app.plex.lookups import by_rating_key
 from app.models import User
-from app.plex import queries
+from app.plex import queries, rollup
 from app.plex.models import (
     PlexCollection,
     PlexItem,
@@ -302,6 +302,23 @@ def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> boo
     return False
 
 
+def _state_row(db: Session, user: User, item_id: int, st: PlexState | None) -> PlexState:
+    """The user's state row on this item, made if the read before found none. Made by the insert
+    itself (ON CONFLICT DO NOTHING) and then read: the first checkpoint of an item and the `final`
+    flush beside it routinely both read "none", and a plain add let the second one fail on
+    `uq_plex_user_item` — a 500, and the push it was carrying lost."""
+    if st is not None:
+        return st
+    db.execute(
+        pg_insert(PlexState)
+        .values(user_id=user.id, item_id=item_id)
+        .on_conflict_do_nothing(constraint="uq_plex_user_item")
+    )
+    made = queries.state(db, user, item_id)
+    assert made is not None  # inserted now, or committed by the request that won
+    return made
+
+
 def clear_progress(db: Session, st: PlexState | None, now: datetime) -> None:
     """A checkpoint too early to keep: a state that held nothing else goes, any other forgets its
     position."""
@@ -325,9 +342,7 @@ def checkpoint(
 ) -> PlexState:
     """Record a resume position. `dirty` marks it not yet mirrored to Plex — set before a push, so
     the reconcile re-pushes it if the push fails and the push can settle it once Plex confirms."""
-    if st is None:
-        st = PlexState(user_id=user.id, item_id=item_id)
-        db.add(st)
+    st = _state_row(db, user, item_id, st)
     st.position_seconds = position
     st.progress_updated_at = now
     if dirty:
@@ -345,9 +360,7 @@ def set_status(
     db: Session, user: User, item_id: int, st: PlexState | None, status: str
 ) -> PlexState:
     """Set a watched or hidden status; watched also clears the resume point."""
-    if st is None:
-        st = PlexState(user_id=user.id, item_id=item_id)
-        db.add(st)
+    st = _state_row(db, user, item_id, st)
     st.status = status
     if status == "watched":
         st.watched_at = datetime.now(timezone.utc)
@@ -364,7 +377,7 @@ def bulk_state(
     (item id, rating key, action) pushes a linked Plex account needs. Unwatching keeps the user's
     private "hidden" marks."""
     now = datetime.now(timezone.utc)
-    existing = queries.states(db, user_id, [e.id for e in eps])
+    existing = rollup.item_state_map(db, user_id, [e.id for e in eps])
     to_push: list[tuple[int, str, str]] = []
     for e in eps:
         st = existing.get(e.id)