diff --git a/backend/alembic/versions/0112_one_watch_later.py b/backend/alembic/versions/0112_one_watch_later.py new file mode 100644 index 00000000..6294b9ea --- /dev/null +++ b/backend/alembic/versions/0112_one_watch_later.py @@ -0,0 +1,55 @@ +"""One Watch later list per user, held by the database + +Revision ID: 0112_one_watch_later +Revises: 0111_audit_request_id +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. +""" + +import sqlalchemy as sa +from alembic import op + +revision: str = "0112_one_watch_later" +down_revision: str | None = "0111_audit_request_id" +branch_labels: str | None = None +depends_on: str | None = None + +_EXTRA = """ + SELECT p.id, k.keep_id + FROM playlists p + JOIN ( + SELECT user_id, min(id) AS keep_id FROM playlists WHERE kind = 'watch_later' GROUP BY user_id + ) k ON k.user_id = p.user_id + WHERE p.kind = 'watch_later' AND p.id <> k.keep_id +""" + + +def upgrade() -> None: + op.execute( + f""" + INSERT INTO playlist_items (playlist_id, video_id, position, added_at) + SELECT e.keep_id, i.video_id, + (SELECT coalesce(max(position), -1) FROM playlist_items WHERE playlist_id = e.keep_id) + + row_number() OVER (PARTITION BY e.keep_id ORDER BY i.playlist_id, i.position, i.id), + i.added_at + FROM playlist_items i + JOIN ({_EXTRA}) e ON e.id = i.playlist_id + ON CONFLICT ON CONSTRAINT uq_playlist_video DO NOTHING + """ + ) + op.execute(f"DELETE FROM playlists WHERE id IN (SELECT id FROM ({_EXTRA}) e)") + op.create_index( + "uq_playlist_watch_later", + "playlists", + ["user_id"], + unique=True, + postgresql_where=sa.text("kind = 'watch_later'"), + ) + + +def downgrade() -> None: + op.drop_index("uq_playlist_watch_later", table_name="playlists") diff --git a/backend/app/plex/queries.py b/backend/app/plex/queries.py index 9d1f7cb3..eb145ad8 100644 --- a/backend/app/plex/queries.py +++ b/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 ---------------------------------------------------------------------------------- diff --git a/backend/app/plex/rollup.py b/backend/app/plex/rollup.py index 78144fac..2bb6071e 100644 --- a/backend/app/plex/rollup.py +++ b/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 ()`).""" diff --git a/backend/app/plex/service.py b/backend/app/plex/service.py index 0ca9ae70..af2bef5c 100644 --- a/backend/app/plex/service.py +++ b/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) diff --git a/backend/app/reader/models.py b/backend/app/reader/models.py index 8cad2eaa..f55bd23a 100644 --- a/backend/app/reader/models.py +++ b/backend/app/reader/models.py @@ -11,12 +11,14 @@ from sqlalchemy import ( DateTime, Float, ForeignKey, + Index, Integer, JSON, String, Text, UniqueConstraint, func, + text, ) from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -93,6 +95,15 @@ class Playlist(Base, TimestampMixin, UpdatedAtMixin): sync phases; the foundation phase only uses local playlists.)""" __tablename__ = "playlists" + __table_args__ = ( + # One Watch later per user, held by the database: two first saves at once each made one. + Index( + "uq_playlist_watch_later", + "user_id", + unique=True, + postgresql_where=text("kind = 'watch_later'"), + ), + ) id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) diff --git a/backend/app/reader/service.py b/backend/app/reader/service.py index 676e1862..2be0136f 100644 --- a/backend/app/reader/service.py +++ b/backend/app/reader/service.py @@ -9,7 +9,7 @@ unique key, a row lock taken before a presence check. The comfort shelf's writes from datetime import datetime, timezone -from sqlalchemy import exists, func, select, update +from sqlalchemy import exists, func, select, text, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session @@ -319,17 +319,25 @@ def create_playlist(db: Session, user: User, name: str) -> Playlist: def watch_later(db: Session, user: User) -> Playlist: - """The user's built-in Watch later playlist, created on first use.""" + """The user's built-in Watch later playlist, created on first use. Created by the insert itself + (ON CONFLICT DO NOTHING on the one-per-user index) and then read: the first two saves at once + both read "none", and each used to make a list of its own.""" pl = queries.watch_later(db, user) if pl is None: - pl = Playlist( - user_id=user.id, - name="Watch later", - kind="watch_later", - position=_next_playlist_position(db, user.id), + db.execute( + pg_insert(Playlist) + .values( + user_id=user.id, + name="Watch later", + kind="watch_later", + position=_next_playlist_position(db, user.id), + ) + .on_conflict_do_nothing( + index_elements=["user_id"], index_where=text("kind = 'watch_later'") + ) ) - db.add(pl) - db.flush() # the id, for the item the caller adds next + pl = queries.watch_later(db, user) + assert pl is not None # inserted now, or committed by the request that won return pl diff --git a/backend/app/routes/plex/detail.py b/backend/app/routes/plex/detail.py index 10eaf028..ac981005 100644 --- a/backend/app/routes/plex/detail.py +++ b/backend/app/routes/plex/detail.py @@ -100,7 +100,7 @@ def show_detail( raise not_found("show") seasons = queries.seasons(db, sh) eps = queries.episodes(db, sh) - states = queries.states(db, user, [e.id for e in eps]) + states = rollup.item_state_map(db, user, [e.id for e in eps]) by_season: dict[int | None, list] = {} all_cards: list[dict] = [] # every episode, in season/episode order → show-level rollup for e in eps: @@ -247,7 +247,7 @@ def _collection_strips(db: Session, it: "PlexItem", user_id: int, lang: str) -> for col in ordered: rk = col.rating_key members = queries.library_movies_in(db, it.library_id, rk, lang, limit=60) - states = queries.states(db, user_id, [m.id for m in members]) + states = rollup.item_state_map(db, user_id, [m.id for m in members]) out.append( { "id": rk, diff --git a/backend/app/routes/plex/playlists.py b/backend/app/routes/plex/playlists.py index bb6ddfc5..71b5d453 100644 --- a/backend/app/routes/plex/playlists.py +++ b/backend/app/routes/plex/playlists.py @@ -9,7 +9,7 @@ from sqlalchemy.orm import Session from app.auth import current_user from app.db import get_db from app.models import User -from app.plex import queries, service +from app.plex import queries, rollup, service from app.plex.cards import display_title, leaf_card, norm_lang from app.refusal import refused from app.routes._common import not_found @@ -99,7 +99,7 @@ def playlist_detail( lang = norm_lang(lang) pl = _own_playlist(db, user, pid) items = queries.playlist_items(db, pl.id) - states = queries.states(db, user, [it.id for it in items]) + states = rollup.item_state_map(db, user, [it.id for it in items]) shows = queries.shows_by_ids(db, {it.show_id for it in items if it.show_id}) cards = [ leaf_card( diff --git a/backend/tests/test_migration_0112.py b/backend/tests/test_migration_0112.py new file mode 100644 index 00000000..9cd4fcf9 --- /dev/null +++ b/backend/tests/test_migration_0112.py @@ -0,0 +1,68 @@ +"""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. + +The schema half is checked by test_migrations.py; a video lost on the way is invisible there. +""" + +from datetime import datetime, timezone + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import text +from sqlalchemy.exc import IntegrityError + +from app.catalog.models import Channel, Video +from app.db import engine + + +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")) + now = datetime.now(timezone.utc) + for v in ("v1", "v2", "v3"): + db.add(Video(id=v, channel_id="UCm", published_at=now, duration_seconds=60)) + db.commit() + cfg = Config("alembic.ini") + 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) + conn.execute( + text( + "INSERT INTO playlist_items (playlist_id, video_id, position) VALUES" + " (:a, 'v1', 0), (:b, 'v2', 0), (:b, 'v1', 1), (:c, 'v3', 0)" + ), + {"a": first, "b": second, "c": theirs}, + ) + finally: + command.upgrade(cfg, "head") + + with engine.connect() as conn: + lists = conn.execute( + text("SELECT id, user_id FROM playlists WHERE kind = 'watch_later' ORDER BY id") + ).all() + assert [tuple(r) for r in lists] == [(first, user.id), (theirs, other.id)] + items = conn.execute( + text( + "SELECT playlist_id, video_id, position FROM playlist_items ORDER BY playlist_id, position" + ) + ).all() + assert [tuple(r) for r in items] == [(first, "v1", 0), (first, "v2", 1), (theirs, "v3", 0)] + with pytest.raises(IntegrityError), engine.begin() as conn: + conn.execute( + text( + "INSERT INTO playlists (user_id, name, kind, position)" + " VALUES (:u, 'Watch later', 'watch_later', 2)" + ), + {"u": user.id}, + ) diff --git a/backend/tests/test_plex_service.py b/backend/tests/test_plex_service.py index a15f6fd8..6dad5e30 100644 --- a/backend/tests/test_plex_service.py +++ b/backend/tests/test_plex_service.py @@ -12,7 +12,7 @@ from datetime import datetime, timezone from sqlalchemy import select -from app.plex import library, queries, service +from app.plex import library, queries, rollup, service from app.plex.models import PlexCollection, PlexItem, PlexLibrary, PlexShow, PlexState from app.plex.query import LibraryFilters @@ -71,9 +71,9 @@ def test_a_users_watch_state_is_theirs_alone(db, make_user): _state(db, owner, it, status="watched") assert queries.state(db, owner, it.id).status == "watched" assert queries.state(db, stranger, it.id) is None - assert queries.states(db, stranger, [it.id]) == {} - assert set(queries.states(db, owner, [it.id])) == {it.id} - assert queries.states(db, owner, []) == {} + assert rollup.item_state_map(db, stranger, [it.id]) == {} + assert set(rollup.item_state_map(db, owner, [it.id])) == {it.id} + assert rollup.item_state_map(db, owner, []) == {} def test_the_grid_shows_the_callers_state_and_hides_only_what_they_hid(db, make_user): @@ -421,3 +421,18 @@ def test_only_the_owners_own_link_pushes_their_watch_state(db, make_user): db.commit() assert watch_sync.link_for_push(db, owner.id).id == link.id assert watch_sync.link_for_push(db, other.id) is None + + +def test_a_first_write_that_lost_the_race_updates_the_winners_row(db, make_user): + """The first checkpoint of an item and the `final` flush beside it both read "no state"; the one + that writes second must land on the row the first made, not fail on `uq_plex_user_item`.""" + user = make_user() + it = _movie(db, _lib(db), "m-race") + _state(db, user, it, position_seconds=10) # the request that won, committed + st = service.checkpoint(db, user, it.id, None, 42, NOW, dirty=True) # read "none" before it + assert (st.position_seconds, st.synced_to_plex) == (42, False) + st = service.set_status(db, user, it.id, None, "hidden") + assert st.status == "hidden" + 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")] diff --git a/backend/tests/test_reader_service.py b/backend/tests/test_reader_service.py index f62d3d01..6230912c 100644 --- a/backend/tests/test_reader_service.py +++ b/backend/tests/test_reader_service.py @@ -308,3 +308,21 @@ def test_a_pin_is_one_per_thing_and_only_its_owners(db, make_user): pins.remove(db, pins.own_pin(db, user, pin.id)) db.commit() assert db.execute(select(ComfortPin)).first() is None + + +def test_a_watch_later_save_that_lost_the_race_uses_the_winners_list(db, make_user, monkeypatch): + """The first two saves at once both read "no Watch later"; the second must land in the list the + first made, not make one of its own.""" + user = make_user() + won = service.watch_later(db, user) + db.commit() + real = queries.watch_later + stale = [True] # this request's first read happened before the winner committed + + def read(d, u): + return None if stale and stale.pop() else real(d, u) + + monkeypatch.setattr(queries, "watch_later", read) + 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]