diff --git a/backend/alembic/versions/0112_one_watch_later.py b/backend/alembic/versions/0112_one_watch_later.py index 6294b9ea..513e3c11 100644 --- a/backend/alembic/versions/0112_one_watch_later.py +++ b/backend/alembic/versions/0112_one_watch_later.py @@ -7,7 +7,9 @@ Create Date: 2026-10-20 Watch later was made on a user's first save by a read then an insert, so two first saves at once each made a list, and the reader then picked between them at random. A partial unique index holds "one per user" now. An instance that already has a second list keeps the oldest: the others' videos -move into it (after its own, in their order; one it already holds is not added twice), then they go. +move into it (after its own, in their order; one it already holds is not added twice), a Home pin +of one of them moves to it (unless it is pinned itself; of several, the first on the shelf), then +they go — a pin left on them goes with them, as the kept list already holds one. """ import sqlalchemy as sa @@ -41,6 +43,19 @@ def upgrade() -> None: ON CONFLICT ON CONSTRAINT uq_playlist_video DO NOTHING """ ) + op.execute( + f""" + UPDATE comfort_pins c SET playlist_id = m.keep_id + FROM ( + SELECT DISTINCT ON (e.keep_id) p.id AS pin_id, e.keep_id + FROM comfort_pins p + JOIN ({_EXTRA}) e ON e.id = p.playlist_id + WHERE NOT EXISTS (SELECT 1 FROM comfort_pins k WHERE k.playlist_id = e.keep_id) + ORDER BY e.keep_id, p.position, p.id + ) m + WHERE c.id = m.pin_id + """ + ) op.execute(f"DELETE FROM playlists WHERE id IN (SELECT id FROM ({_EXTRA}) e)") op.create_index( "uq_playlist_watch_later", diff --git a/backend/app/notifications.py b/backend/app/notifications.py index 26f05d2c..75212f9c 100644 --- a/backend/app/notifications.py +++ b/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) diff --git a/backend/app/plex/service.py b/backend/app/plex/service.py index af2bef5c..215303e7 100644 --- a/backend/app/plex/service.py +++ b/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 diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index 5f6487ef..90515374 100644 --- a/backend/app/plex/watch_sync.py +++ b/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", diff --git a/backend/app/reader/service.py b/backend/app/reader/service.py index 2be0136f..ca16ab49 100644 --- a/backend/app/reader/service.py +++ b/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() diff --git a/backend/tests/test_migration_0112.py b/backend/tests/test_migration_0112.py index 9cd4fcf9..72683acb 100644 --- a/backend/tests/test_migration_0112.py +++ b/backend/tests/test_migration_0112.py @@ -1,8 +1,8 @@ """Migration 0112's data step, DB-lane: a user who already has two Watch later lists keeps the oldest, -with the other's videos moved in after its own (one it already held is not added twice), and the -one-per-user index then refuses a second list. +with the other's videos moved in after its own (one it already held is not added twice) and a Home +pin of the other moved onto it, and the one-per-user index then refuses a second list. -The schema half is checked by test_migrations.py; a video lost on the way is invisible there. +The schema half is checked by test_migrations.py; a video or a pin lost on the way is invisible there. """ from datetime import datetime, timezone @@ -17,6 +17,25 @@ from app.catalog.models import Channel, Video from app.db import engine +def _lists(conn, user_id: int, n: int) -> list[int]: + values = ", ".join(f"(:u, 'Watch later', 'watch_later', {i})" for i in range(n)) + return list( + conn.execute( + text( + f"INSERT INTO playlists (user_id, name, kind, position) VALUES {values} RETURNING id" + ), + {"u": user_id}, + ).scalars() + ) + + +def _pin(conn, user_id: int, playlist_id: int, position: int) -> None: + conn.execute( + text("INSERT INTO comfort_pins (user_id, playlist_id, position) VALUES (:u, :p, :pos)"), + {"u": user_id, "p": playlist_id, "pos": position}, + ) + + def test_a_second_watch_later_is_folded_into_the_first(db, make_user): user, other = make_user(), make_user() db.add(Channel(id="UCm", title="M")) @@ -28,15 +47,8 @@ def test_a_second_watch_later_is_folded_into_the_first(db, make_user): command.downgrade(cfg, "0111_audit_request_id") try: with engine.begin() as conn: - ids = conn.execute( - text( - "INSERT INTO playlists (user_id, name, kind, position) VALUES" - " (:u, 'Watch later', 'watch_later', 0), (:u, 'Watch later', 'watch_later', 1)," - " (:o, 'Watch later', 'watch_later', 0) RETURNING id" - ), - {"u": user.id, "o": other.id}, - ).scalars() - first, second, theirs = list(ids) + first, second, third = _lists(conn, user.id, 3) + theirs, their_second = _lists(conn, other.id, 2) conn.execute( text( "INSERT INTO playlist_items (playlist_id, video_id, position) VALUES" @@ -44,6 +56,12 @@ def test_a_second_watch_later_is_folded_into_the_first(db, make_user): ), {"a": first, "b": second, "c": theirs}, ) + # The user pinned two of the copies, the kept one neither: the first on the shelf moves. + _pin(conn, user.id, third, 0) + _pin(conn, user.id, second, 1) + # The other pinned both: the kept one's pin stays, and is the only one. + _pin(conn, other.id, their_second, 0) + _pin(conn, other.id, theirs, 1) finally: command.upgrade(cfg, "head") @@ -58,6 +76,10 @@ def test_a_second_watch_later_is_folded_into_the_first(db, make_user): ) ).all() assert [tuple(r) for r in items] == [(first, "v1", 0), (first, "v2", 1), (theirs, "v3", 0)] + pins = conn.execute( + text("SELECT user_id, playlist_id, position FROM comfort_pins ORDER BY user_id") + ).all() + assert [tuple(r) for r in pins] == [(user.id, first, 0), (other.id, theirs, 1)] with pytest.raises(IntegrityError), engine.begin() as conn: conn.execute( text( diff --git a/backend/tests/test_plex_service.py b/backend/tests/test_plex_service.py index 6dad5e30..3858db86 100644 --- a/backend/tests/test_plex_service.py +++ b/backend/tests/test_plex_service.py @@ -436,3 +436,17 @@ def test_a_first_write_that_lost_the_race_updates_the_winners_row(db, make_user) db.commit() rows = db.scalars(select(PlexState).where(PlexState.item_id == it.id)).all() assert [(r.position_seconds, r.status) for r in rows] == [(42, "hidden")] + + +def test_a_season_marked_beside_a_first_checkpoint_updates_its_row(db, make_user, monkeypatch): + """Marking a season reads every episode's state once; a first checkpoint that made one row since + must be updated, not collided with on `uq_plex_user_item`.""" + user = make_user() + lib = _lib(db) + ep = _movie(db, lib, "e-race") + _state(db, user, ep, position_seconds=10) # the checkpoint that won, committed + monkeypatch.setattr(rollup, "item_state_map", lambda *a, **k: {}) # read before it committed + service.bulk_state(db, user.id, [ep], watched=True) + db.commit() + rows = db.scalars(select(PlexState).where(PlexState.item_id == ep.id)).all() + assert [(r.status, r.position_seconds) for r in rows] == [("watched", 0)] diff --git a/backend/tests/test_plex_watch_sync.py b/backend/tests/test_plex_watch_sync.py index eae986c7..038e7ff0 100644 --- a/backend/tests/test_plex_watch_sync.py +++ b/backend/tests/test_plex_watch_sync.py @@ -14,6 +14,7 @@ from types import SimpleNamespace from sqlalchemy import select from app.models import User +from app.plex import queries from app.plex.models import PlexItem, PlexLibrary, PlexState from app.plex.watch_sync import _pull_apply, _reconcile_inprogress_movies from app.routes.plex.playback import ( @@ -385,3 +386,30 @@ def test_should_push_resume_periodic_is_throttled(): ) # Once the window elapses, the next tick pushes again. assert _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S), final=False) is True + + +def test_a_pull_that_lost_the_race_keeps_the_local_change_it_met(db, monkeypatch): + """The pull read "no state", and a first local checkpoint made one before it wrote: that row is + still waiting for its push, so the pull leaves it (the dirty guard), and fails on nothing.""" + item_id, user_id = _seed_item_and_user(db) + now = datetime.now(timezone.utc) + db.add(PlexState(user_id=user_id, item_id=item_id, position_seconds=300, synced_to_plex=False)) + db.commit() + real = queries.state + stale = [True] + monkeypatch.setattr( + queries, "state", lambda d, u, i: None if stale and stale.pop() else real(d, u, i) + ) + assert _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now) is None + db.commit() + rows = db.scalars(select(PlexState).where(PlexState.item_id == item_id)).all() + assert [(r.position_seconds, r.synced_to_plex) for r in rows] == [(300, False)] + + +def test_a_first_pull_makes_a_row_in_step_with_plex(db): + item_id, user_id = _seed_item_and_user(db) + now = datetime.now(timezone.utc) + assert _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now) == "in_progress" + db.commit() + rows = db.scalars(select(PlexState).where(PlexState.item_id == item_id)).all() + assert [(r.position_seconds, r.synced_to_plex) for r in rows] == [(120, True)] diff --git a/backend/tests/test_reader_service.py b/backend/tests/test_reader_service.py index 6230912c..97522c3d 100644 --- a/backend/tests/test_reader_service.py +++ b/backend/tests/test_reader_service.py @@ -326,3 +326,18 @@ def test_a_watch_later_save_that_lost_the_race_uses_the_winners_list(db, make_us assert service.watch_later(db, user).id == won.id db.commit() assert [p.id for p in queries.playlists_for(db, user) if p.kind == "watch_later"] == [won.id] + + +def test_a_first_resume_point_beside_the_watched_mark_lands_on_its_row(db, make_user, monkeypatch): + """The first checkpoint of a video and the end-of-playback mark both read "no row"; the + checkpoint that writes second must update the mark's row, not fail on `uq_user_video`.""" + user = make_user() + _videos(db, "v1") + service.mark(db, user, "v1", "watched") # the mark that won, committed + db.commit() + monkeypatch.setattr(queries, "video_state", lambda *a, **k: None) # read before it committed + service.set_progress(db, user, "v1", 90) + db.commit() + db.expire_all() + rows = db.execute(select(VideoState)).scalars().all() + assert [(r.status, r.position_seconds) for r in rows] == [("watched", 90)] diff --git a/backend/tests/test_route_commits.py b/backend/tests/test_route_commits.py index 726ca399..5a6fc0bb 100644 --- a/backend/tests/test_route_commits.py +++ b/backend/tests/test_route_commits.py @@ -2,12 +2,14 @@ A route's helper only flushes; the handler commits. The `db` fixture fails a test whose request left a write uncommitted, so every test here drives one successful write through its route and then -reads the result back — a handler that lost its commit fails on both counts. The services' own +reads the result back. The read-back holds that the handler called its unit; it runs in the +request's own session, so it sees a write that was never committed — the commit is the fixture's +to hold (a test that must see it committed reads from a session of its own). The services' own tests hold what each write does; these hold that the route finishes it. Nothing here reaches YouTube or Plex: the YouTube client is a stub and no user has a Plex link. Needs the DB. """ -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone import pytest from sqlalchemy import select @@ -17,7 +19,9 @@ from app.auth import require_write_scope from app.catalog.models import Channel, ExploredChannel, SearchFind, Subscription, Video from app.downloads.models import DownloadJob, DownloadProfile, DownloadShare from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState +from app.db import SessionLocal from app.models import Notification +from app.notifications import READ_SOFT_CAP from app.reader.models import SavedView, Tag, VideoState from app.routes import channels @@ -343,14 +347,37 @@ def test_a_season_marked_watched_through_the_route_stays_watched(db, make_user, assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")] -def test_reading_one_notice_is_committed_with_its_trim(db, make_user, as_user): - """Marking a notice read and trimming the read ones past the cap are one write the handler - commits — the trim no longer commits on its own.""" - user = make_user() - n = Notification(user_id=user.id, type="t", title="x") - db.add(n) +def _read_past_the_cap(db, user) -> list[Notification]: + """READ_SOFT_CAP read notices, oldest first, and one unread notice after them.""" + base = datetime(2024, 1, 1, tzinfo=timezone.utc) + rows = [ + Notification( + user_id=user.id, + type="t", + title=f"n{i}", + read=i < READ_SOFT_CAP, + created_at=base + timedelta(minutes=i), + ) + for i in range(READ_SOFT_CAP + 1) + ] + db.add_all(rows) db.commit() - r = as_user(user).post(f"/api/me/notifications/{n.id}/read") + return rows + + +def _committed_notices(user) -> tuple[int, bool]: + """What another session sees: how many of the user's notices are left, and whether the oldest is.""" + with SessionLocal() as s: + ids = s.scalars(select(Notification.title).where(Notification.user_id == user.id)).all() + return len(ids), "n0" in ids + + +@pytest.mark.parametrize("path", ["{id}/read", "read_all"]) +def test_reading_a_notice_past_the_cap_commits_the_trim(db, make_user, as_user, path): + """Marking a notice read makes READ_SOFT_CAP + 1 read ones: the handler commits the read flag and + the trim of the oldest together, and the trim counts the notice just read.""" + user = make_user() + rows = _read_past_the_cap(db, user) + r = as_user(user).post("/api/me/notifications/" + path.format(id=rows[-1].id)) assert r.json() == {"ok": True} - db.expire_all() - assert db.get(Notification, n.id).read is True + assert _committed_notices(user) == (READ_SOFT_CAP, False)