fix: every first watch state survives a write beside it, and a folded Watch later keeps its pin

A season marked watched, a pull from Plex and a YouTube resume point made their first row by a read
then an add, and failed on the unique key when a first checkpoint or the watched mark made it in
between. They go through the same insert-then-read as the checkpoint (the resume point is one
upsert); a pull that meets a local change still waiting for its push leaves it.

Folding duplicate Watch later lists moved their videos but not a Home pin, which the cascade then
deleted; the first pin of a dropped copy moves onto the kept list.

Marking one notice read trimmed before its own flag was flushed, so the trim left one past the cap.
The route test now crosses the cap and reads the result from a session of its own.

backend/app/notifications.py

@@ -49,7 +49,9 @@ def create_notification(
 
 def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
     """Delete a user's oldest READ notifications beyond `cap`. Returns the number removed.
-    Idempotent and cheap; safe to call after marking things read. Flushes; the caller commits."""
+    Idempotent and cheap; safe to call after marking things read. Flushes first, so a notice the
+    caller has just marked read counts; the caller commits."""
+    db.flush()
     ids = (
         db.execute(
             select(Notification.id)

backend/app/plex/service.py

@@ -302,16 +302,29 @@ 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:
+def state_row(
+    db: Session,
+    user: User | int,
+    item_id: int,
+    st: PlexState | None = None,
+    *,
+    synced: bool = False,
+) -> 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."""
+    `uq_plex_user_item` — a 500, and the push it was carrying lost. Every first row goes through
+    here: a checkpoint, a status, a season marked at once, a pull from Plex. `synced` is what a row
+    made now says about Plex; a row another writer made first keeps its own."""
     if st is not None:
         return st
     db.execute(
         pg_insert(PlexState)
-        .values(user_id=user.id, item_id=item_id)
+        .values(
+            user_id=user if isinstance(user, int) else user.id,
+            item_id=item_id,
+            synced_to_plex=synced,
+        )
         .on_conflict_do_nothing(constraint="uq_plex_user_item")
     )
     made = queries.state(db, user, item_id)
@@ -342,7 +355,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."""
-    st = _state_row(db, user, item_id, st)
+    st = state_row(db, user, item_id, st)
     st.position_seconds = position
     st.progress_updated_at = now
     if dirty:
@@ -360,7 +373,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."""
-    st = _state_row(db, user, item_id, st)
+    st = state_row(db, user, item_id, st)
     st.status = status
     if status == "watched":
         st.watched_at = datetime.now(timezone.utc)
@@ -383,9 +396,8 @@ def bulk_state(
         st = existing.get(e.id)
         prev = st.status if st is not None else "new"
         if watched:
-            if st is None:
-                st = PlexState(user_id=user_id, item_id=e.id)
-                db.add(st)
+            st = state_row(db, user_id, e.id, st)
+            prev = st.status  # a first checkpoint beside this may have made it meanwhile
             st.status = "watched"
             st.watched_at = now
             st.position_seconds = 0

backend/app/plex/watch_sync.py

@@ -32,6 +32,8 @@ from app.db import SessionLocal
 from app.userscope import owned_by
 from app.plex.models import PlexItem, PlexLibrary, PlexLink, PlexState
 from app.plex.client import PlexClient, PlexError, PlexNotConfigured
+from app.plex import queries
+from app.plex.service import state_row
 from app.plex.sync import (
     _TYPE_EPISODE,
     _TYPE_MOVIE,
@@ -252,15 +254,7 @@ def push_state_to_plex(
         # Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be
         # nothing to flag; that's fine (unscrobble still happened). Skip the flag if the row was
         # re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile).
-        st = (
-            db.execute(
-                select(PlexState).where(
-                    owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id
-                )
-            )
-            .scalars()
-            .first()
-        )
+        st = queries.state(db, user_id, item_id)
         if st is not None and _row_matches(st, expect_ts):
             st.synced_to_plex = True
             db.commit()
@@ -394,15 +388,7 @@ def _pull_apply(
     """Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
     "watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
     full reconcile's job) — this only pulls in watched/resume."""
-    st = (
-        db.execute(
-            select(PlexState).where(
-                owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id
-            )
-        )
-        .scalars()
-        .first()
-    )
+    st = queries.state(db, user_id, item_id)
     if _same_state(st, status, pos):
         if st is not None and not st.synced_to_plex:
             st.synced_to_plex = True  # already matches Plex → just settle the flag, no churn
@@ -423,8 +409,9 @@ def _pull_apply(
         # "almost always already pushed" assumption above excludes.
         return None
     if st is None:
-        st = PlexState(user_id=user_id, item_id=item_id)
-        db.add(st)
+        st = state_row(db, user_id, item_id, synced=True)
+        if not st.synced_to_plex:  # a local change made it first, beside this pull: the guard above
+            return None
     if status == "watched":
         st.status, st.position_seconds, st.watched_at, st.progress_updated_at = (
             "watched",

backend/app/reader/service.py

@@ -239,24 +239,33 @@ def clear_state(db: Session, user: User, video_id: str) -> None:
 
 def set_progress(db: Session, user: User, video_id: str, position: int) -> None:
     """The player's resume point. 0 clears it: a row holding nothing else goes, one with a status
-    (watched, hidden) keeps it with the position zeroed. The status itself is never changed here."""
-    row = queries.video_state(db, user, video_id)
+    (watched, hidden) keeps it with the position zeroed. The status itself is never changed here.
+    A position is one upsert on (user, video): the first checkpoint and the end-of-playback mark
+    beside it both read "no row", and a plain add let the second one fail on `uq_user_video`."""
     now = datetime.now(timezone.utc)
-    if not position:
-        if row is None:
-            return
-        if row.status == "new":
-            db.delete(row)
-        else:
-            row.position_seconds = 0
-            row.progress_updated_at = now
-        db.flush()
+    if position:
+        db.execute(
+            pg_insert(VideoState)
+            .values(
+                user_id=user.id,
+                video_id=video_id,
+                position_seconds=position,
+                progress_updated_at=now,
+            )
+            .on_conflict_do_update(
+                constraint="uq_user_video",
+                set_={"position_seconds": position, "progress_updated_at": now, "updated_at": now},
+            )
+        )
         return
+    row = queries.video_state(db, user, video_id)
     if row is None:
-        row = VideoState(user_id=user.id, video_id=video_id)
-        db.add(row)
-    row.position_seconds = position
-    row.progress_updated_at = now
+        return
+    if row.status == "new":
+        db.delete(row)
+    else:
+        row.position_seconds = 0
+        row.progress_updated_at = now
     db.flush()