fix(plex): one scrobble per finish, and no resume push lands after it

Two near-end checkpoints arrive together routinely (the 10 s tick and the
debounced seeked flush); both read "not watched" before writing and both
scrobbled, bumping Plex's viewCount twice. The transition is now decided by
the write: an insert that did not conflict, or an update conditional on the
old status.

The finish also popped the push-ordering entry that _resume_push_superseded
reads, so an older resume push still in flight could land at Plex after the
scrobble and revive a resume point on a watched item. It now stamps the
entry instead; the existing eviction bounds the map.

backend/app/routes/plex/playback.py

@@ -17,7 +17,8 @@ from fastapi import (
 )
 from fastapi.responses import FileResponse
 from pydantic import BaseModel, ConfigDict, Field
-from sqlalchemy import select
+from sqlalchemy import select, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
 from app.auth import current_user, resolved_user_id
@@ -295,6 +296,37 @@ def _push_watch(
     )
 
 
+def _mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool:
+    """Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end
+    checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a
+    read-then-write let both see "not watched" and both scrobble. The transition is decided by the
+    write itself: an insert that did not conflict, or an update conditional on the old status —
+    under READ COMMITTED the second writer re-reads the committed row and matches nothing."""
+    values = {
+        "status": "watched",
+        "watched_at": now,
+        "position_seconds": 0,
+        "progress_updated_at": now,
+        "synced_to_plex": False,
+    }
+    inserted = db.execute(
+        pg_insert(PlexState)
+        .values(user_id=user_id, item_id=item_id, **values)
+        .on_conflict_do_nothing(constraint="uq_plex_user_item")
+        .returning(PlexState.id)
+    ).first()
+    if inserted is not None:
+        return True
+    mine = (PlexState.user_id == user_id) & (PlexState.item_id == item_id)
+    if db.execute(
+        update(PlexState).where(mine, PlexState.status != "watched").values(values)
+    ).rowcount:
+        return True
+    # Already watched: a re-finish still restamps the row, as it always did — just no scrobble.
+    db.execute(update(PlexState).where(mine).values(values))
+    return False
+
+
 class ItemProgressIn(BaseModel):
     # Finite only: JSON parsing here accepts `Infinity`/`NaN`, and the handler's `int()` turned
     # one into a 500.
@@ -335,6 +367,22 @@ def item_progress(
     duration = int(body.duration_seconds)
     final = body.final
     now = datetime.now(timezone.utc)
+    near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
+    if near_end:
+        flipped = _mark_finished(db, user.id, it.id, now)
+        # The finish supersedes every resume push still in flight for this item: stamping it as the
+        # latest push makes `_resume_push_superseded` drop them. Popping the entry instead (to keep
+        # the map small) removed the very guard, and an older resume push could then land at Plex
+        # AFTER the scrobble and revive a resume point on a watched item. Eviction bounds the map.
+        with _push_throttle_lock:
+            _last_push_at[(user.id, it.id)] = now
+        db.commit()
+        # Only the checkpoint that flipped the row scrobbles: a later checkpoint of the same finished
+        # item — or a second one racing this one — would otherwise bump Plex's viewCount again.
+        if flipped:
+            _push_watch(background, db, user.id, it.id, rating_key, "watched")
+        return {"status": "watched", "position_seconds": 0}
+
     st = (
         db.execute(
             select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id)
@@ -342,28 +390,6 @@ def item_progress(
         .scalars()
         .first()
     )
-    was_watched = st is not None and st.status == "watched"
-
-    near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
-    if near_end:
-        if st is None:
-            st = PlexState(user_id=user.id, item_id=it.id)
-            db.add(st)
-        st.status = "watched"
-        st.watched_at = now
-        st.position_seconds = 0
-        st.progress_updated_at = now
-        st.synced_to_plex = False
-        # Terminal state for resume write-back: drop the throttle/ordering entry so the map doesn't
-        # retain a dead key for every finished item.
-        with _push_throttle_lock:
-            _last_push_at.pop((user.id, it.id), None)
-        db.commit()
-        # Immediate on the watched transition; skip a repeat scrobble if it was already watched
-        # (a later checkpoint of the same finished item would otherwise bump Plex's viewCount again).
-        if not was_watched:
-            _push_watch(background, db, user.id, it.id, rating_key, "watched")
-        return {"status": "watched", "position_seconds": 0}
 
     if position < _PROGRESS_MIN_S:
         if st is not None and st.status == "new":

backend/tests/test_plex_finish.py

@@ -0,0 +1,80 @@
+"""The near-end checkpoint that finishes a Plex item.
+
+- Exactly one scrobble per finish, even when two near-end checkpoints race (the 10 s tick and the
+  debounced `seeked` flush arrive together): the transition is decided by a conditional write, not a
+  read taken before it.
+- The finish supersedes every resume push still in flight, so none can land at Plex after the
+  scrobble and revive a resume point on a watched item.
+"""
+
+import threading
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+
+from app.db import SessionLocal
+from app.models import PlexItem, PlexLibrary, PlexState
+from app.routes.plex import playback
+
+
+def _item(db) -> PlexItem:
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    item = PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="A Movie")
+    db.add(item)
+    db.commit()
+    return item
+
+
+def test_two_racing_finishes_flip_the_row_once(db, make_user):
+    user = make_user()
+    item = _item(db)
+    db.add(PlexState(user_id=user.id, item_id=item.id, status="new", position_seconds=5000))
+    db.commit()
+    now = datetime.now(timezone.utc)
+
+    first, second = SessionLocal(), SessionLocal()
+    try:
+        assert playback._mark_finished(first, user.id, item.id, now) is True  # holds the row lock
+        out: list[bool] = []
+        racer = threading.Thread(
+            target=lambda: out.append(playback._mark_finished(second, user.id, item.id, now))
+        )
+        racer.start()
+        racer.join(0.5)
+        assert racer.is_alive(), "the second writer must wait on the first one's row lock"
+        first.commit()
+        racer.join(5)
+        second.commit()
+    finally:
+        first.close()
+        second.close()
+    assert out == [False], "the second checkpoint saw the committed finish: no second scrobble"
+
+
+def test_a_first_ever_finish_inserts_and_counts_once(db, make_user):
+    user = make_user()
+    item = _item(db)
+    now = datetime.now(timezone.utc)
+    assert playback._mark_finished(db, user.id, item.id, now) is True
+    db.commit()
+    assert playback._mark_finished(db, user.id, item.id, now + timedelta(seconds=10)) is False
+    db.commit()
+    st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one()
+    assert (st.status, st.position_seconds) == ("watched", 0)
+
+
+def test_the_finish_supersedes_a_resume_push_still_in_flight(db, as_user, make_user, monkeypatch):
+    user = make_user()
+    item = _item(db)
+    monkeypatch.setattr(playback.plex_stream, "touch_session", lambda key: None)
+    earlier = datetime.now(timezone.utc) - timedelta(seconds=5)
+    # A resume push was scheduled a moment ago and has not reached Plex yet.
+    assert playback._should_push_resume(user.id, item.id, earlier, final=True)
+
+    r = as_user(user).post(
+        "/api/plex/item/m1/progress", json={"position_seconds": 2995, "duration_seconds": 3000}
+    )
+    assert r.json() == {"status": "watched", "position_seconds": 0}
+    assert playback._resume_push_superseded(user.id, item.id, earlier) is True