diff --git a/CHANGELOG.md b/CHANGELOG.md index eec53d1c..ad1f016b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,35 @@ edit the release notes and re-run the generator. Upgrading, pinning a version and rolling back: [docs/self-hosting.md](docs/self-hosting.md#upgrading). +## 0.92.0 — 2026-09-19 + +A debt burn: every item the roadmap had deferred is done or dropped — Plex playback per viewer, refusals said in your language, controls that keep the keyboard focus, and a screen reader that hears Hungarian. + +### Features + +- The command palette has a visible way in: a Ctrl K key cap in the header's search field, and a “Jump to…” pill on pages without one. +- The Playlists rail says, under “Missing a list? Why?”, which lists YouTube's API does not hand out (Watch later, History, lists saved from other channels). +- The last native drop-downs are the app's own controls: the live search's result count, rows per page, a download format's content, audio format and quality, and the download dialog's format. +- Plex plays one session per viewer, so two accounts can watch the same film at once; an administrator sets the ceiling on concurrent sessions (Configuration → Plex). + +### Fixes + +- A YouTube action YouTube refuses says what to do — wait for tomorrow's quota, reconnect the account, or try again later — instead of a server error; a passing Google hiccup no longer asks you to reconnect. +- Every refusal the server gives is said in your language, and every label a screen reader or tooltip reads is translated. +- Dragging to reorder a list is announced in your language and with the right position, and a Plex playlist keeps a drag within the list it started in. +- A button that turns itself off, or a row that removes itself, leaves the keyboard focus where it was; Back from a Plex title returns the focus to the tile you opened, even after using a button there. +- With Plex turned off, the Home shelf shows your YouTube pins only, and reordering it leaves the hidden Plex pins in their places; two reorders at once can no longer put two pins on one slot. +- The Home hero from a pinned show offers to unpin the show, and “Surprise me” from the command palette really asks for a surprise. +- A finished Plex title scrobbles once and gets no resume point pushed after it; a subtitle pass that stalled resumes instead of starting over. +- Tables keep their j/k keys after the window is resized across the phone breakpoint, and boxes grow with the text-size setting. +- Whitelisting an address and activating its account happen in one step, and the audit log names the entry. + +### Internal + +- Every mutation answers `ok`, and route helpers leave the commit to the request — both held by tests; a request that leaves a write uncommitted now fails its test. +- New lint rules and contracts: no native select, no hover that ignores aria-disabled, no untranslated accessible text, no unused translation key. +- Two adversarial review rounds at the epic's end: 23 findings, all fixed, each regression test proven to fail with its own fix removed. + ## 0.91.0 — 2026-09-19 What a test run on the live instance turned up, fixed: new videos keep arriving while YouTube's feeds are down, an open tab survives a release, and Back returns to what you left. diff --git a/CLAUDE.md b/CLAUDE.md index 70bc60c6..cb86892c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -82,8 +82,8 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory | Admin gate import | `app.auth` | the `app.routes.admin` re-export | | Upstream failure status | `422` with JSON detail | `502` (reserve 502 for raw byte-proxies) | | User-facing error | structured `detail` (`{code, reason}`) | a bare English string under a bilingual UI | -| Mutation response | `{"ok": true, …}` (`{"deleted": id}`) | ad-hoc `{"unsubscribed"}`/`{"cleared": n}`/echo | -| Transaction commit | the handler/job-wrapper commits; helpers `flush` | helpers that commit (only `quota` owns its session) | +| Mutation response | `{"ok": true, …}` (`{"deleted": id}`), test-held | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok` | +| Transaction commit | the handler commits; helpers `flush` (test-held list) | helpers that commit (only `quota` owns its session) | | Admin knob storage | `sysconfig` (values) | env / `app_state` / `scheduler_settings` scatter | | Job status word | the `downloads/status.py` StrEnum | loose `"canceled"`/`"cancelled"` literals | | Current time | inline `datetime.now(timezone.utc)` | new `now_utc`/`_now` helpers | @@ -92,7 +92,7 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory | FE keyboard shortcut | a row in `lib/keymap.ts` + `dispatch()` | a private `keydown` listener with its own key literals | | FE list/grid keys | `NavGrid`/`keyboardNav` + `navItemAttrs` + `data-nav-open` | a private arrow-key / roving-tabindex handler | | FE view push (history) | `rememberNavPosition()` right before `history.pushState` | a push that forgets: Back strands focus on `` | -| FE control | the chip/segmented primitives (`Select` only for >6 options) | a native `` (lint-enforced) | | FE loading/empty/error | `QueryState` / `StateMessage` | ad-hoc conditionals | | FE failed write | `meta.failTitle`, else `notifyActionError(err, title)` | a bare `catch {}` or `notify()` with no error attached | | FE network call | `req`/`uploadReq`/`beacon`, or `lib/http.ts` `send` (public pages) | a bare `fetch`/`XMLHttpRequest`/`WebSocket` (lint-enforced) | diff --git a/VERSION b/VERSION index 8f63f4f9..36545ad3 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.91.0 +0.92.0 diff --git a/backend/app/auth.py b/backend/app/auth.py index b8044a9c..763954cf 100644 --- a/backend/app/auth.py +++ b/backend/app/auth.py @@ -14,6 +14,7 @@ from app.db import get_db from app.models import Invite, LogEvent, User from app.security import decrypt_optional from app.utils import canonical_email +from app.refusal import refused PASSWORD_MIN_LEN = 10 @@ -341,19 +342,19 @@ def current_user(request: Request, db: Session = Depends(get_db)) -> User: def _session_user(request: Request, db: Session, user_id: int | None, is_default: bool) -> User: if not user_id: - raise HTTPException(status_code=401, detail="Not authenticated") + raise refused(401, "not_authenticated") user = db.get(User, user_id) if user is None or not user.is_active or user.is_suspended: # Suspended/deactivated mid-session → end it so the block takes effect at once. _end_account_session(request, user_id, is_default) - raise HTTPException(status_code=401, detail="Not authenticated") + raise refused(401, "not_authenticated") # SA4: reject a cookie whose recorded epoch is behind the account's current one (revoked by a # password reset/change or "log out everywhere"). A cookie with no recorded epoch (pre-SA4, or an # account signed in before this shipped) is treated as epoch 0 — so the FIRST bump revokes it too. epochs = request.session.get("epochs") or {} if int(epochs.get(str(user_id), 0)) != (user.session_epoch or 0): _end_account_session(request, user_id, is_default) - raise HTTPException(status_code=401, detail="Not authenticated") + raise refused(401, "not_authenticated") # Always keep the session default in the switchable list — covers sessions created before # multi-session existed (their account_ids was never seeded), so the switcher isn't empty. default_id = request.session.get("user_id") @@ -418,10 +419,7 @@ def require_write_scope(user: User = Depends(require_human)) -> User: checks that had drifted into three different 403 messages across channels/playlists. Layers on `require_human` because writing to YouTube needs a real, non-demo identity.""" if not has_write_scope(user): - raise HTTPException( - status_code=403, - detail="Enable YouTube editing in Settings first.", - ) + raise refused(403, "youtube_editing_off") return user @@ -429,7 +427,7 @@ def admin_user(user: User = Depends(current_user)) -> User: """Dependency: require the signed-in user to be an admin. Single source of truth for the admin gate — every admin route/action depends on this rather than re-checking the role.""" if user.role != "admin": - raise HTTPException(status_code=403, detail="Admin only") + raise refused(403, "admin_only") return user diff --git a/backend/app/config.py b/backend/app/config.py index 2490b51a..8933fd62 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -315,6 +315,10 @@ class Settings(BaseSettings): # re-encode is CPU-heavy (the VAAPI hardware path is far cheaper and could sustain more); direct-serve # and remux (stream-copy) are not capped by this. plex_max_transcodes: int = 4 + # Max concurrent playback sessions of ANY kind — remux (stream-copy) included. Cheap on the CPU but + # each one is a whole segmented copy of the film on the scratch volume, written faster than + # realtime. One account holds at most two regardless (plex/stream.py `_MAX_SESSIONS_PER_USER`). + plex_max_sessions: int = 8 # --- Transcode (playable="transcode") re-encode tuning (E6 S1) --- # Hardware-encoder policy: "auto" = VAAPI (h264_vaapi) when the iGPU render node is present AND the # source codec is hardware-decodable, else libx264 software; "on" = force VAAPI wherever the render diff --git a/backend/app/discover/hero.py b/backend/app/discover/hero.py index df7a3320..e75a2e88 100644 --- a/backend/app/discover/hero.py +++ b/backend/app/discover/hero.py @@ -81,7 +81,11 @@ def pick( return None def _comfort() -> dict | None: - for card in pins.playables(pins.shelf(db, user) if shelf is None else shelf): + if shelf is None: + own = pins.shelf(db, user, include_plex=rails.plex_available(db)) + else: + own = shelf + for card in pins.playables(own): if cards.key(card) not in skip and fits(card["remaining_seconds"], budget): return card return None diff --git a/backend/app/discover/pins.py b/backend/app/discover/pins.py index 7a49ca47..318b68ec 100644 --- a/backend/app/discover/pins.py +++ b/backend/app/discover/pins.py @@ -8,14 +8,16 @@ second one. Every shelf card carries a resolved `next`: the thing that actually plays when you click the tile — a show's in-progress-or-next episode, a channel's newest unwatched video, a playlist's first -unwatched item. It costs a query per non-leaf pin, which is affordable precisely because this shelf is -short and hand-curated, and it buys the property the whole surface exists for: one click, no -intermediate page, no second decision. +unwatched item. That buys the property the whole surface exists for: one click, no intermediate page, +no second decision. Channels and playlists are resolved for the whole shelf at once (`_Batch`), so the +cost does not grow a query or two per pin. Domain module, so nothing here raises HTTP. `resolve_ref` returns None for "no such thing, or not yours" and the route turns that into its 404 — core never imports routes. """ +from dataclasses import dataclass, field + from sqlalchemy import func, select from sqlalchemy.orm import Session @@ -73,6 +75,44 @@ def resolve_ref(db: Session, user: User, kind: str, ref: str) -> tuple[str, obje return (col, pl.id) if pl is not None and pl.user_id == user.id else None +def reorder(db: Session, user: User, ids: list[int]) -> None: + """Put the listed pins in the listed order, WITHIN the slots they already hold. + + The client sends the order it can SEE, which is not always the whole shelf: with Plex off the Home + rail hides the Plex pins, and a stale tab may not know a pin added elsewhere. Numbering only the + listed ids 0..n-1 collided with every pin left out — two tiles on one position, and the hidden + ones jumped about when they came back. So the pins left out keep their slots, and the listed ones + are dealt into the slots the listed set occupies. Ids that are not this user's, or repeated, are + ignored rather than failing a stale replay wholesale. Flushes; the route commits.""" + # Locked: the rows it leaves alone are part of the answer too. Two overlapping reorders each + # skipped the rows that already matched in their own snapshot and interleaved into two pins on + # one slot; the second now waits and deals from the first one's result. Locked in id order and + # sorted HERE: with `ORDER BY position … FOR UPDATE` Postgres sorts before it locks, so the one + # that waited got the fresh rows in its stale order and dealt the pins it left out back to their + # old slots — and one global lock order means two reorders cannot deadlock. + locked = db.execute( + select(ComfortPin) + .where(ComfortPin.user_id == user.id) + .order_by(ComfortPin.id) + .with_for_update() + .execution_options(populate_existing=True) + ).scalars() + current = sorted(locked, key=lambda p: (p.position, p.id)) + owned = {p.id for p in current} + listed: list[int] = [] + for i in ids: + if i in owned and i not in listed: + listed.append(i) + chosen = set(listed) + queue = iter(listed) + order = [next(queue) if p.id in chosen else p.id for p in current] + by_id = {p.id: p for p in current} + for position, pin_id in enumerate(order): + if by_id[pin_id].position != position: + by_id[pin_id].position = position + db.flush() + + def list_pins(db: Session, user: User) -> list[ComfortPin]: return list( db.execute( @@ -103,7 +143,34 @@ def next_position(db: Session, user: User) -> int: ) -def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: +@dataclass +class _Batch: + """What the shelf resolves for every pinned channel and playlist in one go.""" + + channels: dict[str, Channel] = field(default_factory=dict) + channel_summary: dict[str, tuple[int, dict | None]] = field(default_factory=dict) + playlists: dict[int, Playlist] = field(default_factory=dict) + playlist_summary: dict[int, tuple[int, dict | None]] = field(default_factory=dict) + + @classmethod + def load(cls, db: Session, user: User, pins: list[ComfortPin]) -> "_Batch": + ch_ids = [p.channel_id for p in pins if p.channel_id is not None] + pl_ids = [p.playlist_id for p in pins if p.playlist_id is not None] + b = cls() + if ch_ids: + b.channels = { + c.id: c for c in db.scalars(select(Channel).where(Channel.id.in_(ch_ids))) + } + b.channel_summary = queries.channels_summary(db, user, ch_ids) + if pl_ids: + b.playlists = { + p.id: p for p in db.scalars(select(Playlist).where(Playlist.id.in_(pl_ids))) + } + b.playlist_summary = queries.playlists_summary(db, user, pl_ids) + return b + + +def _shelf_card(db: Session, user: User, pin: ComfortPin, batch: _Batch) -> dict | None: """One pin as a unified card with its resolved `next`. None when the pinned row has gone (the FK cascade removes the pin itself, so this is only the window between a delete and a refetch).""" card: dict | None = None @@ -117,11 +184,11 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: nxt = dict(card) # a video plays itself elif pin.channel_id is not None: - ch = db.get(Channel, pin.channel_id) + ch = batch.channels.get(pin.channel_id) if ch is None: return None - card = cards.from_channel(ch, queries.channel_unwatched_count(db, user, ch.id)) - row = queries.channel_next_video(db, user, ch.id) + count, row = batch.channel_summary.get(ch.id, (0, None)) + card = cards.from_channel(ch, count) nxt = cards.from_video(row) if row else None elif pin.plex_item_id is not None: @@ -142,10 +209,10 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: nxt = cards.from_plex(ep) if ep else None elif pin.playlist_id is not None: - pl = db.get(Playlist, pin.playlist_id) + pl = batch.playlists.get(pin.playlist_id) if pl is None: return None - count, row = queries.playlist_summary(db, user, pl.id) + count, row = batch.playlist_summary.get(pl.id, (0, None)) card = cards.from_playlist(pl, count, row.get("thumbnail_url") if row else None) nxt = cards.from_video(row) if row else None @@ -159,11 +226,24 @@ def _shelf_card(db: Session, user: User, pin: ComfortPin) -> dict | None: return card -def shelf(db: Session, user: User) -> list[dict]: - """The whole shelf, in the user's own order.""" +_PLEX_COLUMNS = ("plex_item_id", "plex_show_id") + + +def shelf(db: Session, user: User, *, include_plex: bool = True) -> list[dict]: + """The whole shelf, in the user's own order. + + `include_plex=False` is the Home surface with the deployment's Plex toggle off (the owner's call, + 2026-09-19): the rails beside it go YouTube-only, and a Plex tile there would offer a Play that + lands on a gated module. The pins themselves are kept — the pin menus still read the full shelf + (`GET /pins`), and switching Plex back on brings the tiles back in their old slots + (`reorder` keeps a hidden pin's place).""" out = [] - for pin in list_pins(db, user): - card = _shelf_card(db, user, pin) + pins = list_pins(db, user) + if not include_plex: + pins = [p for p in pins if all(getattr(p, c) is None for c in _PLEX_COLUMNS)] + batch = _Batch.load(db, user, pins) + for pin in pins: + card = _shelf_card(db, user, pin, batch) if card is not None: out.append(card) return out @@ -171,5 +251,17 @@ def shelf(db: Session, user: User) -> list[dict]: def playables(shelf_cards: list[dict]) -> list[dict]: """The `next` of every shelf entry that has one — what the hero may draw from. A pinned channel - with nothing unwatched left, or an empty playlist, simply is not an answer right now.""" - return [c["next"] for c in shelf_cards if c.get("next")] + with nothing unwatched left, or an empty playlist, simply is not an answer right now. + + Each carries `pinned_from`: the pin it was resolved from. A show's next EPISODE is not itself + pinned, so without it the hero's menu looked the episode up, missed, and offered "Pin" under a + kicker that says "From your shelf" — and taking it inserted a second pin. With it, the hero's + pin row acts on the shelf entry the card came from (the owner's call, 2026-09-19: "Unpin").""" + return [ + { + **c["next"], + "pinned_from": {k: c[k] for k in ("source", "kind", "id", "title")}, + } + for c in shelf_cards + if c.get("next") + ] diff --git a/backend/app/discover/queries.py b/backend/app/discover/queries.py index 403bd8c9..8b6e82f7 100644 --- a/backend/app/discover/queries.py +++ b/backend/app/discover/queries.py @@ -43,8 +43,9 @@ from app.plex import rollup from app.plex.cards import leaf_card, movie_card, show_card from app.plex.query import apply_show_state_filter, content_libs -# How many rows a single rail may carry. Matches `routes/plex/discover.py`'s RAIL_LIMIT on purpose — -# the two surfaces should feel like the same app, and a rail is a rail. +# How many rows a single rail may carry — Home's rails and the Plex module's Discover rails alike +# (`routes/plex/discover.py` imports it): the two surfaces should feel like the same app, and a rail +# is a rail. One definition, because a comment was all that held two copies in step. RAIL_LIMIT = 20 # The feed parameters a Home rail runs with. Spelled out rather than borrowed from the route's @@ -129,18 +130,6 @@ def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) -> return [r for r in out if not r["position_seconds"]][:limit] -def channel_next_video(db: Session, user: User, channel_id: str) -> dict | None: - """The newest unwatched video of a pinned channel — what "put on that channel" resolves to.""" - # `library_source="all"`: a pinned channel means the whole channel, including videos that only - # ever arrived through a search. The feed's "organic" default is a feed-scoping choice, not a - # statement about what belongs to a channel. - params = dict(_FEED_BASE, channel_ids=[channel_id], scope="all", library_source="all") - query, rank_expr = filtered_query(db, user, **params) - keys = sort_keys("newest", 0, rank_expr) - row = db.execute(query.order_by(*order_by(keys)).limit(1)).first() - return serialize(row) if row is not None else None - - def yt_video_row(db: Session, user: User, video_id: str) -> dict | None: """One video as a feed row, with this user's watch state. @@ -158,31 +147,55 @@ def yt_video_row(db: Session, user: User, video_id: str) -> dict | None: return serialize(row) if row is not None else None -def channel_unwatched_count(db: Session, user: User, channel_id: str) -> int: - """How many unwatched videos a pinned channel is holding — the tile's only honest promise, since a - channel has no runtime of its own.""" - params = dict(_FEED_BASE, channel_ids=[channel_id], scope="all", library_source="all") - query, _ = filtered_query(db, user, **params) - return int(db.execute(select(func.count()).select_from(query.subquery())).scalar() or 0) +def channels_summary( + db: Session, user: User, channel_ids: list[str] +) -> dict[str, tuple[int, dict | None]]: + """For each pinned channel: (unwatched count, its newest unwatched video) — the tile's only honest + promise, since a channel has no runtime of its own, and what "put on that channel" resolves to. + + ONE query for the whole shelf, not two per channel: the count and the first row per channel are + window functions over the same filtered set. `library_source="all"`: a pinned channel means the + whole channel, including videos that only ever arrived through a search — the feed's "organic" + default is a feed-scoping choice, not a statement about what belongs to a channel.""" + if not channel_ids: + return {} + params = dict(_FEED_BASE, channel_ids=list(channel_ids), scope="all", library_source="all") + query, rank_expr = filtered_query(db, user, **params) + newest = order_by(sort_keys("newest", 0, rank_expr)) + per_channel = {"partition_by": Video.channel_id} + sub = query.add_columns( + func.row_number().over(**per_channel, order_by=newest).label("_rn"), + func.count().over(**per_channel).label("_total"), + ).subquery() + rows = db.execute(select(sub).where(sub.c._rn == 1)).all() + return {r.channel_id: (int(r._total), serialize(r)) for r in rows} -def playlist_summary(db: Session, user: User, playlist_id: int) -> tuple[int, dict | None]: - """(item count, the row to play next) for a pinned playlist — its first not-yet-watched item, or - its first item once the whole list is watched. The same row doubles as the tile's artwork, so a - shelf of playlists is scannable instead of a row of blank boxes.""" +def playlists_summary( + db: Session, user: User, playlist_ids: list[int] +) -> dict[int, tuple[int, dict | None]]: + """For each pinned playlist: (item count, the row to play next) — its first not-yet-watched item, + or its first item once the whole list is watched. The same row doubles as the tile's artwork, so + a shelf of playlists is scannable instead of a row of blank boxes. One query for the whole shelf.""" + if not playlist_ids: + return {} st = aliased(VideoState) rows = db.execute( - select(*feed_columns(user, st), PlaylistItem.position) + select(*feed_columns(user, st), PlaylistItem.playlist_id.label("_playlist")) .join(PlaylistItem, PlaylistItem.video_id == Video.id) .join(Channel, Channel.id == Video.channel_id) .outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id)) - .where(PlaylistItem.playlist_id == playlist_id) - .order_by(PlaylistItem.position.asc()) + .where(PlaylistItem.playlist_id.in_(playlist_ids)) + .order_by(PlaylistItem.playlist_id, PlaylistItem.position.asc()) ).all() - if not rows: - return 0, None - nxt = next((r for r in rows if (r.status or "new") not in ("watched", "hidden")), rows[0]) - return len(rows), serialize(nxt) + grouped: dict[int, list] = {} + for r in rows: + grouped.setdefault(r._playlist, []).append(r) + out: dict[int, tuple[int, dict | None]] = {} + for pid, items in grouped.items(): + nxt = next((r for r in items if (r.status or "new") not in ("watched", "hidden")), items[0]) + out[pid] = (len(items), serialize(nxt)) + return out # ── Plex ────────────────────────────────────────────────────────────────────────────────────────── diff --git a/backend/app/plex/rollup.py b/backend/app/plex/rollup.py index 53211250..7358338b 100644 --- a/backend/app/plex/rollup.py +++ b/backend/app/plex/rollup.py @@ -43,7 +43,9 @@ def rollup(cards: list[dict]) -> dict: ] resume = None if inprog: - resume = inprog[-1] # continue the latest-started episode + # The LAST in-progress episode in broadcast order — no timestamp is consulted. Home's + # `discover.queries.next_episode` picks the same one, so both Resume buttons agree. + resume = inprog[-1] else: resume = next((c for c in cards if c.get("status") != "watched"), None) return { diff --git a/backend/app/plex/stream.py b/backend/app/plex/stream.py index 9975e1be..656dabd9 100644 --- a/backend/app/plex/stream.py +++ b/backend/app/plex/stream.py @@ -60,8 +60,10 @@ _WARMUP_SEG_TIMEOUT_S = 8.0 # its index for that one. See `segment_name` for why this is a constant and not a comment. _VIDEO_VARIANT = "0" _SESSION_IDLE_S = 600 # reap a session with no access for this long -# A session directory is named `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}` (see start_session); -# before multi-audio it was just `{rating_key}_{int(start_s)}`. `sweep_orphans` deletes ONLY names +# A session directory is named `{rating_key}_u{user_id}_{int(start_s)}_{tag}_{aoff:+.2f}` (see +# start_session); before per-viewer sessions it was `{rating_key}_{int(start_s)}_{tag}_{aoff:+.2f}`, +# and before multi-audio just `{rating_key}_{int(start_s)}` — the older shapes stay sweepable, since a +# restart leaves them behind like any other. `sweep_orphans` deletes ONLY names # matching one of those two shapes: `_HLS_ROOT` is admin-configurable (PLEX_HLS_DIR), so it may # legitimately point at a shared scratch path we do not own, and a blanket "delete every directory # under the root" would take the neighbours with it. @@ -72,7 +74,9 @@ _SESSION_IDLE_S = 600 # reap a session with no access for this long # non-numeric rating_key would merely leave its directory unswept, which is the safe direction to # fail. Covered by test_plex_stream.py — this regex is the only thing standing between the sweep # and someone else's data. -_SESSION_DIR_RE = re.compile(r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$") +_SESSION_DIR_RE = re.compile( + r"^\d+_\d+(_[^_]+_[+-]\d+\.\d{2})?$|^\d+_u\d+_\d+_[^_]+_[+-]\d+\.\d{2}$" +) # When this process started. `sweep_orphans` only deletes directories older than this: everything # newer belongs to a session THIS run created, and since `start_session` reuses a directory path # for the same (rating_key, offset, tag), "stale name" is not the same as "stale directory". @@ -88,7 +92,16 @@ _PROCESS_START = time.time() _MTIME_GRACE_S = 2.0 _lock = threading.Lock() -_sessions: dict[str, "HlsSession"] = {} +# Keyed by VIEWER and item, (user_id, rating_key). Keyed by the item alone, a second account opening +# the same film `_kill`ed the first one's session: the displaced viewer got a 404, reclaimed at its +# playhead, and so evicted the other in turn — two people could not watch one film at once. (One +# account on two devices still shares a session; that case ends in the player's bounded reclaim.) +_sessions: dict[tuple[int, str], "HlsSession"] = {} +# How many live sessions one account may hold at once (two devices, or a next item starting while the +# last one winds down). A new one past it replaces that account's least recently used — whose waiting +# session request then returns at once, since `wait_for` watches the killed ffmpeg — so one account +# can neither pile up film-sized scratch copies nor park more than this many request workers. +_MAX_SESSIONS_PER_USER = 2 class HlsSession: @@ -102,16 +115,17 @@ class HlsSession: transcode: bool = False, hw: bool = False, tonemap: bool = False, + user_id: int = 0, ): - # Unguessable per-session id, carried in the URLs the client is handed. Sessions are keyed - # by rating_key alone, so a seek-restart or a SECOND viewer of the same item replaces the - # registered one — and the old viewer's next request for `seg_12.ts`, resolved by rating_key, - # was answered 200 with the NEW session's twelfth segment, produced from a different offset + # Unguessable per-session id, carried in the URLs the client is handed. A seek-restart + # replaces the viewer's registered session for the item — and the old player's next request + # for `seg_12.ts`, resolved by (viewer, rating_key) alone, was answered 200 with the NEW session's twelfth segment, produced from a different offset # under `-copyts`. Wrong bytes, not an error, so the client's reclaim never fired on it # (deep-review round 5). A request naming a session that is no longer registered is a 404, # which the player recovers from by restarting at the playhead. self.id = secrets.token_urlsafe(9) - self.key = key + self.key = key # the item's rating_key (what the logs name) + self.user_id = user_id # whose session it is: the registry key is (user_id, key) self.dir = directory self.proc = proc self.transcode = transcode # video re-encoded (non-h264 source), not stream-copied @@ -637,29 +651,47 @@ def _kill(s: HlsSession) -> None: shutil.rmtree(s.dir, ignore_errors=True) -def _enforce_cap(cap: int) -> None: - # Caller holds _lock. Only TRANSCODE sessions count toward plex_max_transcodes — remux - # (stream-copy) and direct-serve are cheap and deliberately uncapped (see config.py: - # plex_max_transcodes, and the user-facing config hint). Drop the least-recently-accessed - # transcode sessions over the cap, leaving any concurrent remux streams untouched. - tx = [(key, s) for key, s in _sessions.items() if s.transcode] - if len(tx) <= cap: - return - for key, s in sorted(tx, key=lambda kv: kv[1].last_access)[: len(tx) - cap]: - # Somebody's playback stops here. The route turns the missing session into a retryable - # 404 and the player restarts at the playhead, so from the server nothing failed — and - # "my film keeps rebuffering" has no other explanation anywhere. - log.info( - "a transcode session was dropped to stay under the cap", - extra=logs.event("plex.stream.session_culled", key=key, cap=cap, sessions=len(tx)), - ) - _kill(s) - _sessions.pop(key, None) +def _enforce_caps(user_id: int, max_transcodes: int, max_sessions: int) -> None: + """Caller holds _lock. Three bounds, applied after a session is registered, each dropping the + least recently accessed sessions over it: + + * this account's own sessions (`_MAX_SESSIONS_PER_USER`) — first, so a single account filling the + box evicts itself before anyone else; + * TRANSCODE sessions (plex_max_transcodes): the CPU/GPU-heavy ones; + * EVERY session (plex_max_sessions). A remux is cheap on the CPU but not on the disk: each one is a + whole segmented copy of the film under `_HLS_ROOT`, written faster than realtime. Counted only + by the transcode cap, N distinct items were N ffmpegs and N film-sized scratch directories. + + Somebody's playback stops at each drop. The route turns the missing session into a retryable 404 + and the player restarts at the playhead, so from the server nothing failed — and "my film keeps + rebuffering" has no other explanation anywhere, hence the line.""" + + def cull(pool: list[tuple[tuple[int, str], HlsSession]], cap: int, bound: str) -> None: + if len(pool) <= cap: + return + for reg, sess in sorted(pool, key=lambda kv: kv[1].last_access)[: len(pool) - cap]: + log.info( + "a session was dropped to stay under a cap", + extra=logs.event( + "plex.stream.session_culled", + key=sess.key, + bound=bound, + cap=cap, + sessions=len(pool), + ), + ) + _kill(sess) + _sessions.pop(reg, None) + + cull([kv for kv in _sessions.items() if kv[0][0] == user_id], _MAX_SESSIONS_PER_USER, "user") + cull([kv for kv in _sessions.items() if kv[1].transcode], max_transcodes, "transcode") + cull(list(_sessions.items()), max_sessions, "total") def start_session( db: DbSession, item: PlexItem, + user_id: int, start_s: float, audio_ord: int | None = None, aoff: float = 0.0, @@ -679,6 +711,7 @@ def start_session( # session respawn (the HW->SW fallback) must not pick up a mid-playback config change. loudnorm = sysconfig.get_bool(db, "plex_audio_loudnorm") key = item.rating_key + reg = (user_id, key) # the registry key: one session per viewer per item start_s = max(0.0, float(start_s)) audio_langs = _probe_audio_langs(src) if multi else [] is_multi = len(audio_langs) > 1 # only worth a master playlist when there really are ≥2 tracks @@ -691,7 +724,7 @@ def start_session( # the tag with a DASH — `_SESSION_DIR_RE` treats the tag as one underscore-free field, and that # regex is the only thing keeping the orphan sweep off a shared scratch path (see its comment). tag = ("multi" if is_multi else str(audio_ord)) + ("-ln" if loudnorm else "") - directory = _HLS_ROOT / f"{key}_{int(start_s)}_{tag}_{aoff:+.2f}" + directory = _HLS_ROOT / f"{key}_u{user_id}_{int(start_s)}_{tag}_{aoff:+.2f}" def _seg(i: int) -> Path: return directory / segment_name(i, _VIDEO_VARIANT if variants else None) @@ -717,9 +750,9 @@ def start_session( # platform without fork (the API always runs in a Linux container in prod) it's simply omitted. preexec = _sw_child_preexec if (transcode and not hw and os.name == "posix") else None with _lock: - if replace_only is not None and _sessions.get(key) is not replace_only: + if replace_only is not None and _sessions.get(reg) is not replace_only: return None # superseded/culled while we waited out of lock — don't clobber the newer one - old = _sessions.pop(key, None) + old = _sessions.pop(reg, None) if old is not None: _kill(old) shutil.rmtree(directory, ignore_errors=True) @@ -753,8 +786,9 @@ def start_session( transcode=transcode, hw=hw, tonemap=tonemap, + user_id=user_id, ) - _sessions[key] = s + _sessions[reg] = s # The one line that says which pipeline a session actually got. Everything downstream # — the cap, the HW→SW fallback, the subtitle offset — is about this choice, and until # now it was only inferable from the ffmpeg command nobody keeps. @@ -770,9 +804,13 @@ def start_session( start_s=round(start_s, 1), ), ) - # Admin-configurable concurrency cap (Configuration → Plex → Max concurrent transcodes); - # at least 1 so playback can't be capped to zero. - _enforce_cap(max(1, sysconfig.get_int(db, "plex_max_transcodes"))) + # Admin-configurable caps (Configuration → Plex); at least 1 each so playback can't be + # capped to zero. + _enforce_caps( + user_id, + max(1, sysconfig.get_int(db, "plex_max_transcodes")), + max(1, sysconfig.get_int(db, "plex_max_sessions")), + ) return s s = _spawn(*_select_encode(db, item, transcode, src)) @@ -798,7 +836,7 @@ def start_session( # turns a None return into a "Media file not found on disk" 404 — dishonest for a file that # exists. A (dead) session instead degrades to a retryable 404 ("No active playback session" / # "Not available yet") the client re-requests. - return current_session(key) or s + return current_session(user_id, key) or s s = respawned ready = wait_for(_seg(0), timeout=25.0, proc=s.proc) if ready: @@ -829,9 +867,9 @@ def start_session( return s -def current_session(key: str) -> HlsSession | None: +def current_session(user_id: int, key: str) -> HlsSession | None: with _lock: - s = _sessions.get(key) + s = _sessions.get((user_id, key)) if s is not None: s.last_access = time.time() return s @@ -1060,21 +1098,21 @@ def reap_idle(finished_idle_s: float = _FINISHED_IDLE_S) -> int: now = time.time() dropped = 0 with _lock: - for key, s in list(_sessions.items()): + for reg, s in list(_sessions.items()): done = s.proc.poll() is not None idle = now - s.last_access if idle > _SESSION_IDLE_S or (done and idle > finished_idle_s): _kill(s) - _sessions.pop(key, None) + _sessions.pop(reg, None) dropped += 1 return dropped -def touch_session(key: str) -> None: - """Say a viewer is still there for `key`, without fetching anything. `last_access` used to move +def touch_session(user_id: int, key: str) -> None: + """Say this viewer is still there for `key`, without fetching anything. `last_access` used to move only on an HLS file fetch, which a paused player does not do — so the only way to keep a paused viewer's session alive was to keep EVERY abandoned one just as long.""" with _lock: - s = _sessions.get(key) + s = _sessions.get((user_id, key)) if s is not None: s.last_access = time.time() diff --git a/backend/app/plex/subtitles.py b/backend/app/plex/subtitles.py index c09e97ec..374a731e 100644 --- a/backend/app/plex/subtitles.py +++ b/backend/app/plex/subtitles.py @@ -461,11 +461,18 @@ _FULL_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="plex-subs-ful # which time something has finished. _MAX_WINDOWS_INFLIGHT = 8 _MAX_FULL_INFLIGHT = 6 +# …of which one account may hold this many. The full-pass slots are minutes each, and a hostile (or +# merely curious) account opening subtitles on film after film could otherwise hold all of them while +# everyone else's tracks never completed. A viewer over their share still gets windows — the part +# that makes a track watchable now — just not another whole-file read until one of theirs ends. +_MAX_FULL_PER_USER = 3 _MAX_TRACK_WINDOWS = 2 _LOCK = threading.Lock() # Queued or running jobs → the absolute span a window job reads (None for a full pass). _INFLIGHT: dict[tuple, tuple[float, float] | None] = {} +# rating_key → the account whose poll queued that item's full pass (see _MAX_FULL_PER_USER). +_FULL_OWNERS: dict[str, int] = {} # Track-level failures: a track that cannot be served at all (no item, a bitmap track, Plex down…). _FAILURES: dict[str, tuple[str, float]] = {} # rating_key → when its full pass last failed (see FULL_RETRY_S). @@ -515,6 +522,7 @@ def reset_state() -> None: global _STOPPING with _LOCK: _INFLIGHT.clear() + _FULL_OWNERS.clear() _FAILURES.clear() _FULL_FAILED.clear() _WIN_FAILED.clear() @@ -598,7 +606,7 @@ def _window_failed(key: str, at: float) -> bool: return any(a <= at <= b for a, b, _ in _live_window_failures(key)) -def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: +def ensure(rating_key: str, sub_ord: int, at: float = 0.0, viewer: int = 0) -> dict: """State of one subtitle track, scheduling whatever extraction is still missing. NEVER blocks. ``ready`` = fully extracted · ``partial`` = usable, covers ``at``, still growing · ``preparing`` @@ -641,8 +649,23 @@ def ensure(rating_key: str, sub_ord: int, at: float = 0.0) -> dict: if edge - at <= LOOKAHEAD_S and edge < _FULL_END_S: probe = edge + 1.0 want_window = not _window_failed(key, probe) - _schedule(rating_key, sub_ord, probe, want_window=want_window, want_full=not full_failed) - return {"state": "partial" if covered else "preparing", "code": None, "gen": gen} + _schedule( + rating_key, + sub_ord, + probe, + want_window=want_window, + want_full=not full_failed, + viewer=viewer, + ) + # `growing`: whether the track will fill in on its own. False once the full pass has failed and + # is backed off — a partial track then stays partial until a window is asked for, so the player + # may poll slowly (only to read ahead of an edge) instead of every two seconds for the whole film. + return { + "state": "partial" if covered else "preparing", + "code": None, + "gen": gen, + "growing": not full_failed, + } def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None]: @@ -659,7 +682,12 @@ def cached_track(rating_key: str, sub_ord: int) -> tuple[str | None, str | None] def _schedule( - rating_key: str, sub_ord: int, at: float, want_window: bool, want_full: bool = True + rating_key: str, + sub_ord: int, + at: float, + want_window: bool, + want_full: bool = True, + viewer: int = 0, ) -> None: """Kick off the jobs this track still needs: the window that makes it watchable now, and the one full pass that makes it complete. A 1 s status poll (or a second viewer) joins a job already @@ -692,8 +720,15 @@ def _schedule( ident = ("win", rating_key, sub_ord, w_start) _INFLIGHT[ident] = (w_start, w_start + WINDOW_PRE_S + WINDOW_LEN_S) jobs.append((ident, w_start)) - if want_full and n_full < _MAX_FULL_INFLIGHT and ("full", rating_key) not in _INFLIGHT: + mine = sum(1 for owner in _FULL_OWNERS.values() if owner == viewer) + if ( + want_full + and n_full < _MAX_FULL_INFLIGHT + and ("full", rating_key) not in _INFLIGHT + and mine < _MAX_FULL_PER_USER + ): _INFLIGHT[("full", rating_key)] = None + _FULL_OWNERS[rating_key] = viewer jobs.append((("full", rating_key), None)) for ident, window_start in jobs: pool = _FULL_POOL if window_start is None else _WIN_POOL @@ -757,6 +792,8 @@ def _run( finally: with _LOCK: _INFLIGHT.pop(ident, None) + if ident[0] == "full": + _FULL_OWNERS.pop(rating_key, None) def _pass_failed( @@ -822,10 +859,13 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No for i, s in enumerate(subs) if s.get("index") is not None and not _is_bitmap(s) } - start = window_start or 0.0 + # A full pass starts where every track's leading coverage ends: a pass that ran out of time + # (`EXTRACT_TIMEOUT_S`) kept what it reached, and restarting from 0 each `FULL_RETRY_S` meant a file + # too big to read in one go never completed. Windows handle an offset start already; so does this. + start = window_start if window_start is not None else _full_pass_start(rating_key, targets) length = (WINDOW_LEN_S + WINDOW_PRE_S) if window_start is not None else None end = _FULL_END_S if length is None else start + length - texts, clean = _ffmpeg_extract(str(src), sorted(targets), window_start, length) + texts, clean = _ffmpeg_extract(str(src), sorted(targets), start or None, length) for emb_ord, vtt in texts.items(): key = cache_key(rating_key, targets[emb_ord]) # A pass that exited non-zero holds only what ffmpeg reached before it died. Claim no more @@ -846,6 +886,24 @@ def _extract_and_publish(rating_key: str, sub_ord: int, window_start: float | No _pass_failed(rating_key, [key], window_start, (covered_end, end)) +def _full_pass_start(rating_key: str, targets: dict[int, int]) -> float: + """Where the full pass may begin: the end of the span from 0 that EVERY target track already has + (one read feeds them all, so the least-covered track decides), backed off by `WINDOW_PRE_S` so a + cue running across the join is read whole — `merge_vtt` folds the overlap. 0 when any track has + nothing from the start.""" + ends = [] + for ord_ in targets.values(): + key = cache_key(rating_key, ord_) + ranges, _ = read_meta(key) if _exists(_vtt_path(key)) else ([], False) + lead = [b for a, b in ranges if a <= 0.0] + if not lead: + return 0.0 + ends.append(max(lead)) + if not ends: + return 0.0 + return max(0.0, float(f"{min(ends) - WINDOW_PRE_S:.3f}")) + + def _sidecar_vtt(db: Session, stream: dict) -> str: """An external subtitle file, fetched through Plex and converted to WebVTT.""" with PlexClient(db) as plex: @@ -884,9 +942,10 @@ def _ffmpeg_extract( cmd += ["-map", f"0:s:{emb_ord}"] if length is not None: cmd += ["-t", f"{length:.3f}"] + if start: # The input seek lands on the keyframe at or BEFORE `start`, so a cue already running # at `start` comes out at a negative time — and by default the muxer then shifts the - # WHOLE output so that cue sits at 0: every cue of the window late by the distance to + # WHOLE output so that cue sits at 0: every cue of the read late by the distance to # that keyframe (1–5 s measured on a real library file, a whole GOP on a synthetic # one), and doubled for good once the full pass merged in beside it. Kept negative, the # times are exact on the zero-based clock (`_vtt_ts_to_s` reads the sign). diff --git a/backend/app/refusal.py b/backend/app/refusal.py new file mode 100644 index 00000000..fc70f7b6 --- /dev/null +++ b/backend/app/refusal.py @@ -0,0 +1,37 @@ +"""A request the app refuses, said in a way the UI can put in the reader's language. + +Every refusal the API answers carries a structured `detail` — `{code, reason, …params}` — and the client +translates it (`lib/api/core.ts` `localizeDetail`). A bare English string reached the Hungarian UI +verbatim: the blocking error dialog, and every inline surface that printed `HttpError.detail`, showed +whatever sentence the route happened to hand-write. `refused` is the generic family +(`errors.refused.`, the params interpolated into the sentence); the older specific codes +(`not_found`, `youtube`, `playlist`, the admin refusals, …) stay, since each is already translated. + +A test (`test_http_details.py`) reads every `HTTPException` in the app and fails on a string detail, +and the frontend's contract test requires every `refused(…, "")` to be named in both +languages — so the rule is enforced, not described. +""" + +from fastapi import HTTPException + + +def refused(status: int, reason: str, **params: object) -> HTTPException: + """The refusal to raise: `raise refused(409, "view_name_taken", name=name)`. `params` travel in + the detail for the sentence to interpolate (`{{name}}`); keep them to values the reader may see.""" + return HTTPException(status_code=status, detail={"code": "refused", "reason": reason, **params}) + + +# `sysconfig.InvalidSetting.reason` → the refusal it answers as. A dict of literals, so the frontend's +# contract test can read every reason this can produce. +_SETTING_REASONS = { + "not_a_number": "setting_not_a_number", + "too_low": "setting_too_low", + "too_high": "setting_too_high", + "not_a_choice": "setting_not_a_choice", + "no_encryption_key": "no_encryption_key", +} + + +def setting_refused(exc) -> HTTPException: + """A value the admin entered that `sysconfig.set_value` refused (`exc`: its `InvalidSetting`).""" + return refused(400, _SETTING_REASONS[exc.reason], key=exc.key, **exc.params) diff --git a/backend/app/routes/_common.py b/backend/app/routes/_common.py index 7223ce93..6689f6fa 100644 --- a/backend/app/routes/_common.py +++ b/backend/app/routes/_common.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session from app.db import Base from app.models import User +from app.youtube.client import YouTubeError, refusal_reason M = TypeVar("M", bound=Base) @@ -45,9 +46,20 @@ def owned_or_404( ) -> M: """Fetch a row by primary key that `user` owns, or raise 404. A row owned by someone else 404s (NOT 403) on purpose — a 403 would confirm the id exists to a stranger. - Generalizes the near-identical per-route ownership helpers (e.g. plex's - `_own_playlist_or_404`, the saved-views owner check).""" + It replaced the near-identical per-route ownership helpers, and is called directly — a thin + `_own_x` wrapper around it is the ad-hoc shape CLAUDE.md's patterns table rules out.""" obj = get_or_404(db, model, id_, what=what) if getattr(obj, owner_attr) != user.id: raise not_found(what) return obj + + +def youtube_refusal(exc: Exception) -> HTTPException: + """The answer to an action YouTube refused: a 422 whose reason the UI translates + (`errors.youtube.`). One definition, because the routes had grown three shapes of it — + a bare English string quoting the upstream message, a playlist-scoped reason, and nothing at + all (a raw 500 from the manual syncs). The upstream failure is already logged where it + happened (`youtube.api.error`). A failure that is not the Data API's (the search scraper's) is + YouTube being unavailable to us — the retryable answer.""" + reason = refusal_reason(exc) if isinstance(exc, YouTubeError) else "unavailable" + return HTTPException(status_code=422, detail={"code": "youtube", "reason": reason}) diff --git a/backend/app/routes/admin.py b/backend/app/routes/admin.py index 63e070da..b58ff724 100644 --- a/backend/app/routes/admin.py +++ b/backend/app/routes/admin.py @@ -39,6 +39,7 @@ from app.routes.auth import ( send_verification_link, ) from app.utils import canonical_email, iso, valid_email +from app.refusal import refused router = APIRouter(prefix="/api/admin", tags=["admin"]) @@ -169,7 +170,7 @@ def _activate_user_by_email(db: Session, email: str) -> None: user = next((u for u in candidates if u.email == email), None) or next(iter(candidates), None) if user is not None and not user.is_active: user.is_active = True - db.commit() + db.flush() # the handler commits (CLAUDE.md: helpers flush) def _decide(db: Session, invite_id: int, admin: User, approved: bool) -> Invite: @@ -179,7 +180,7 @@ def _decide(db: Session, invite_id: int, admin: User, approved: bool) -> Invite: # sending a duplicate decision mail, even flipping an already-approved invite to "denied" and # mailing a rejection to an already-active account. if inv.status != "pending": - raise HTTPException(status_code=409, detail="Invite already decided") + raise refused(409, "invite_decided") inv.status = "approved" if approved else "denied" inv.decided_at = datetime.now(timezone.utc) inv.decided_by = admin.email @@ -191,7 +192,7 @@ def _decide(db: Session, invite_id: int, admin: User, approved: bool) -> Invite: target_id=inv.id, summary=inv.email, ) - db.commit() + db.flush() # the handler commits, with the activation that goes with it return inv @@ -210,6 +211,9 @@ def approve_invite( link, has_account, hours = _welcome_mail_parts(db, inv.email) lang = mailtext.lang_for_email(db, inv.email) background.add_task(email_mod.send_access_approved, inv.email, lang, link, has_account, hours) + # Before the response, so the mail (a background task, after it) never describes a decision + # the database does not hold yet. + db.commit() return _serialize(inv, has_account=bool(_accounts_for(db, [inv.email]))) @@ -226,6 +230,7 @@ def deny_invite( background.add_task( email_mod.send_access_denied, inv.email, mailtext.lang_for_email(db, inv.email) ) + db.commit() return _serialize(inv, has_account=bool(_accounts_for(db, [inv.email]))) @@ -242,7 +247,7 @@ def add_invite( """Manually whitelist an email (approved straight away). Convenience for the admin.""" email = body.email.strip().lower() if not valid_email(email): - raise HTTPException(status_code=400, detail="Enter a valid email address.") + raise refused(400, "invalid_email") # Reuse a canonically-equal row instead of adding a second one for the same mailbox. Prod carries # exactly that pair for one person from 2026-06-16 (hartmann.peter1986@ / hartmannpeter1986@), # which is how an admin ends up approving the same human twice and still seeing them bounce. @@ -259,11 +264,13 @@ def add_invite( inv.status = "approved" inv.decided_at = datetime.now(timezone.utc) inv.decided_by = admin.email + db.flush() # a new invite's id, for the audit row — before it, the row named no target audit.record( db, admin.id, AuditAction.INVITE_ADD, target_type="invite", target_id=inv.id, summary=email ) - db.commit() + # One transaction: the whitelisting and the activation it implies land together or not at all. _activate_user_by_email(db, email) + db.commit() return _serialize(inv, has_account=bool(_accounts_for(db, [email]))) @@ -366,10 +373,7 @@ def delete_invite( promoted to admin on its next Google sign-in). The admin has to hear that, not guess it.""" inv = get_or_404(db, Invite, invite_id, what="invite") if _accounts_for(db, [inv.email]): - raise HTTPException( - status_code=409, - detail="That address already has an account — delete the account instead.", - ) + raise refused(409, "address_has_account") still_allowed_by_env = _env_allowed(inv.email) audit.record( db, @@ -435,19 +439,19 @@ def set_user_role( account, and can't remove the last remaining admin. Emails the user when it actually changes.""" role = body.role if role not in ("user", "admin"): - raise HTTPException(status_code=400, detail="role must be 'user' or 'admin'") + raise refused(400, "invalid_role") target = get_or_404(db, User, user_id, what="user") if target.is_demo: - raise HTTPException(status_code=400, detail="The demo account's role can't be changed.") + raise refused(400, "demo_role") if target.id == admin.id: - raise HTTPException(status_code=400, detail="You can't change your own role.") + raise refused(400, "own_role") if ( target.role == "admin" and role == "user" and not target.is_suspended and count_admins(db, active_only=True) <= 1 ): - raise HTTPException(status_code=400, detail="Can't remove the last admin.") + raise refused(400, "last_admin") changed = target.role != role if changed: audit.record( @@ -493,16 +497,16 @@ def set_user_suspended( suspended = body.suspended target = get_or_404(db, User, user_id, what="user") if target.is_demo: - raise HTTPException(status_code=400, detail="The demo account can't be suspended.") + raise refused(400, "demo_suspend") if target.id == admin.id: - raise HTTPException(status_code=400, detail="You can't suspend your own account.") + raise refused(400, "own_suspend") if ( suspended and target.role == "admin" and not target.is_suspended and count_admins(db, active_only=True) <= 1 ): - raise HTTPException(status_code=400, detail="Can't suspend the last active admin.") + raise refused(400, "last_active_admin") was_suspended = target.is_suspended if was_suspended != suspended: audit.record( @@ -537,17 +541,15 @@ def admin_delete_user( not the demo account, not yourself (use Settings → Account), and not the last admin.""" target = get_or_404(db, User, user_id, what="user") if target.is_demo: - raise HTTPException(status_code=400, detail="The demo account can't be deleted.") + raise refused(400, "demo_delete") if target.id == admin.id: - raise HTTPException( - status_code=400, detail="Delete your own account from Settings → Account." - ) + raise refused(400, "own_delete") if ( target.role == "admin" and not target.is_suspended and count_admins(db, active_only=True) <= 1 ): - raise HTTPException(status_code=400, detail="Can't delete the last admin.") + raise refused(400, "last_admin") # Record BEFORE the purge — the target row (and its email) is gone afterwards. audit.record( db, @@ -691,7 +693,7 @@ def activate_user( `_activate_user_by_email`). What was missing was any way to know they had to.""" target = get_or_404(db, User, user_id, what="user") if target.is_demo: - raise HTTPException(status_code=400, detail="The demo account can't be activated.") + raise refused(400, "demo_activate") if target.is_active: return _serialize_user(target) # idempotent: a double click is not an error if not is_allowed(db, target.email): @@ -788,7 +790,7 @@ def add_demo_whitelist( sign-in). Idempotent — re-adding an existing email just returns it.""" email = body.email.strip().lower() if not valid_email(email): - raise HTTPException(status_code=400, detail="Enter a valid email address.") + raise refused(400, "invalid_email") row = db.execute(select(DemoWhitelist).where(DemoWhitelist.email == email)).scalar_one_or_none() if row is None: row = DemoWhitelist( @@ -835,7 +837,7 @@ def reset_demo(admin: User = Depends(admin_user), db: Session = Depends(get_db)) db, admin.id, AuditAction.DEMO_RESET, summary=f"{seeded} sample playlists re-seeded" ) db.commit() - return {"reset": True, "playlists_seeded": seeded} + return {"ok": True, "reset": True, "playlists_seeded": seeded} @router.post("/purge-discovery") @@ -854,4 +856,4 @@ def purge_discovery(admin: User = Depends(admin_user), db: Session = Depends(get after=result, ) db.commit() - return result + return {"ok": True, **result} diff --git a/backend/app/routes/audit.py b/backend/app/routes/audit.py index 6562e3cf..d9a907b9 100644 --- a/backend/app/routes/audit.py +++ b/backend/app/routes/audit.py @@ -68,4 +68,4 @@ def clear_audit(admin: User = Depends(admin_user), db: Session = Depends(get_db) n = db.execute(delete(AuditLog)).rowcount or 0 audit.record(db, admin.id, AuditAction.AUDIT_CLEAR, summary=f"Cleared {n} audit entries") db.commit() - return {"cleared": n} + return {"ok": True, "cleared": n} diff --git a/backend/app/routes/auth.py b/backend/app/routes/auth.py index 6d135751..405acb02 100644 --- a/backend/app/routes/auth.py +++ b/backend/app/routes/auth.py @@ -43,6 +43,7 @@ from app.models import AuthToken, DemoWhitelist, Invite, OAuthToken, User from app.ratelimit import RateLimiter, client_key from app.security import encrypt, hash_password, hash_token, verify_password from app.utils import canonical_email, valid_email +from app.refusal import refused VERIFY_TTL = timedelta(hours=24) # A confirmation link that rides an APPROVAL or an invitation lives longer than one the person is @@ -378,7 +379,7 @@ async def callback( userinfo = token.get("userinfo") if not userinfo or not userinfo.get("sub"): - raise HTTPException(status_code=400, detail="No user info returned by Google") + raise refused(400, "google_no_userinfo") sub = userinfo["sub"] email = (userinfo.get("email") or "").lower() @@ -653,15 +654,15 @@ async def request_access( endpoint can't be probed for throttling either.""" email = (body.email or "").strip().lower() if not valid_email(email): - raise HTTPException(status_code=400, detail="Enter a valid email address.") + raise refused(400, "invalid_email") if not _request_access_limiter.allow(client_key(request)): - return {"status": "pending"} + return {"ok": True, "status": "pending"} if is_allowed(db, email): - return {"status": "pending"} # already allowed — don't reveal it + return {"ok": True, "status": "pending"} # already allowed — don't reveal it req = upsert_pending_invite(db, email, body.lang) if req.is_new: notify_admins_of_request(db, background, email) - return {"status": "pending"} + return {"ok": True, "status": "pending"} @router.post("/demo") @@ -672,7 +673,7 @@ def demo_login(body: EmailIn, request: Request, db: Session = Depends(get_db)) - exactly like a non-match (authenticated=false) so the field can't be hammered as an enumeration oracle. On a match it sets the session and the client reloads into the app.""" if not _demo_limiter.allow(client_key(request)): - return {"authenticated": False} + return {"ok": True, "authenticated": False} email = (body.email or "").strip().lower() if valid_email(email) and is_demo_allowed(db, email): demo = get_or_create_demo_user(db) @@ -683,8 +684,8 @@ def demo_login(body: EmailIn, request: Request, db: Session = Depends(get_db)) - _remember_epoch(request, demo) # The demo key IS an address, so it is not in the line — see the denied branch above. log.info("signed in", extra=logs.event("auth.login.ok", method="demo", uid=demo.id)) - return {"authenticated": True} - return {"authenticated": False} + return {"ok": True, "authenticated": True} + return {"ok": True, "authenticated": False} def _establish_session(request: Request, user: User) -> None: @@ -815,23 +816,20 @@ def register( password = body.password or "" # Input validation is safe to reveal (independent of whether the email exists). if not valid_email(email): - raise HTTPException(status_code=400, detail="Enter a valid email address.") + raise refused(400, "invalid_email") if len(password) < PASSWORD_MIN_LEN: - raise HTTPException( - status_code=400, - detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.", - ) + raise refused(400, "weak_password", min=PASSWORD_MIN_LEN) if not sysconfig.get_bool(db, "allow_registration"): - raise HTTPException(status_code=403, detail="Registration is currently closed.") + raise refused(403, "registration_closed") if not _register_limiter.allow(client_key(request)): - return {"status": "ok"} # silently throttle; uniform response + return {"ok": True, "status": "ok"} # silently throttle; uniform response # Do the existence check + account creation OFF the response path (a background task with its own # session). The create path hashes the password + writes several rows + schedules emails; doing it # inline would make a NEW email respond measurably slower than an already-registered one (which # skips all that) — a timing oracle for enumeration. Off-path, any valid email responds the same. background.add_task(_register_account, email, password, body.lang) - return {"status": "ok"} + return {"ok": True, "status": "ok"} def _register_account(email: str, password: str, lang: str | None = None) -> None: @@ -997,11 +995,11 @@ def verify_email_resend( mail had no way back. Uniform response + timing regardless of whether the email has a pending account (the lookup runs off the response path), so it can't probe for registered emails.""" if not _verify_resend_limiter.allow(client_key(request)): - return {"status": "ok"} # silently throttle; uniform response + return {"ok": True, "status": "ok"} # silently throttle; uniform response email = (body.email or "").strip().lower() if valid_email(email): background.add_task(_resend_verify_if_eligible, email) - return {"status": "ok"} + return {"ok": True, "status": "ok"} def _resend_verify_if_eligible(email: str) -> None: @@ -1091,14 +1089,14 @@ def password_reset_request( """Request a password-reset link. Uniform response + timing regardless of whether the email has a password account (the lookup runs off the response path), so it can't probe for registered emails.""" if not _reset_limiter.allow(client_key(request)): - return {"status": "ok"} + return {"ok": True, "status": "ok"} email = (body.email or "").strip().lower() # Do the account lookup + token issue + email OFF the response path (a background task with its # own session), so the endpoint takes the same time for any valid email whether or not it has a # password account — no timing-based enumeration. Any valid email schedules the same task. if valid_email(email): background.add_task(_send_reset_if_eligible, email) - return {"status": "ok"} + return {"ok": True, "status": "ok"} def _send_reset_if_eligible(email: str) -> None: @@ -1293,13 +1291,10 @@ def set_password( raise HTTPException(status_code=403, detail={"code": "demo"}) new_password = body.password or "" if len(new_password) < PASSWORD_MIN_LEN: - raise HTTPException( - status_code=400, - detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.", - ) + raise refused(400, "weak_password", min=PASSWORD_MIN_LEN) if user.password_hash: if not verify_password(body.current_password or "", user.password_hash): - raise HTTPException(status_code=403, detail="Current password is incorrect.") + raise refused(403, "wrong_password") user.password_hash = hash_password(new_password) user.password_revoked_at = None # same as the reset path: a password exists again # SA4: a password change logs out every OTHER session; re-stamp this cookie at the new epoch so diff --git a/backend/app/routes/channels.py b/backend/app/routes/channels.py index 604bfaf2..5fdd68e8 100644 --- a/backend/app/routes/channels.py +++ b/backend/app/routes/channels.py @@ -4,7 +4,7 @@ hidden, tags) and per-channel sync state. Local reads/writes plus YouTube-side w from datetime import datetime, timezone -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException +from fastapi import APIRouter, BackgroundTasks, Depends from pydantic import BaseModel from sqlalchemy import func, select, update from sqlalchemy.orm import Session @@ -25,12 +25,13 @@ from app.models import ( User, Video, ) -from app.routes._common import get_or_404, not_found, owned_or_404 +from app.routes._common import get_or_404, not_found, owned_or_404, youtube_refusal from app.sync.explore import explore_ingest_page from app.sync.runner import backfill_channel_now, run_recent_backfill from app.sync.subscriptions import apply_channel_details from app.utils import iso from app.youtube.client import YouTubeClient, YouTubeError +from app.refusal import refused router = APIRouter(prefix="/api/channels", tags=["channels"]) @@ -383,12 +384,9 @@ def explore_channel( backfill quota reserve; blocked for the demo account (it spends shared quota).""" channel = get_or_404(db, Channel, channel_id, what="channel") if _is_blocked(db, user, channel_id): - raise HTTPException(status_code=403, detail="You've blocked this channel.") + raise refused(403, "channel_blocked") if quota.backfill_reserve_reached(): - raise HTTPException( - status_code=429, - detail="The shared daily quota reserve is reached — try exploring later.", - ) + raise refused(429, "quota_reserve") # Record/refresh the exploration (DB only) before ingesting, so the explorer can see the # videos as soon as they land. Flag the channel ephemeral unless it's organically followed. @@ -424,8 +422,9 @@ def explore_channel( channel = db.get(Channel, channel_id) result = explore_ingest_page(db, yt, channel, page_token) except YouTubeError as exc: - raise HTTPException(status_code=422, detail=f"YouTube couldn't load this channel: {exc}") + raise youtube_refusal(exc) return { + "ok": True, "channel_id": channel_id, "ingested": result["ingested"], "next_page_token": result["next_page_token"], @@ -507,6 +506,7 @@ def update_channel( background.add_task(backfill_channel_now, channel_id, user.id) return { + "ok": True, "id": channel_id, "priority": sub.priority, "hidden": sub.hidden, @@ -541,7 +541,7 @@ def reset_backfill( db.commit() with quota.attribute(user.id, quota.QuotaAction.VIDEOS_BACKFILL_RECENT): run_recent_backfill(db, [channel], max_channels=1) - return {"id": channel_id, "reset": True} + return {"ok": True, "id": channel_id, "reset": True} @router.post("/{channel_id}/subscribe") @@ -562,7 +562,7 @@ def subscribe( ) ).scalar_one_or_none() if existing is not None: - raise HTTPException(status_code=409, detail="Already subscribed to this channel.") + raise refused(409, "already_subscribed") with ( quota.attribute(user.id, quota.QuotaAction.CHANNELS_SUBSCRIBE), YouTubeClient(db, user) as yt, @@ -583,9 +583,7 @@ def subscribe( if "subscriptionDuplicate" in exc.reason.split(","): yt_sub_id = "" else: - raise HTTPException( - status_code=422, detail=f"YouTube couldn't subscribe you: {exc}" - ) + raise youtube_refusal(exc) # Enrich a stub channel (title/thumbnail/subscriber count) so it shows up properly right # away, on BOTH the normal and the desync path — a resurfaced stub still needs metadata to # render. Keep the client open (we're still inside the `with`); a metadata failure here @@ -622,7 +620,7 @@ def subscribe( .values(via_explore=False) ) db.commit() - return {"subscribed": channel_id} + return {"ok": True, "subscribed": channel_id} @router.delete("/{channel_id}/subscription") @@ -636,10 +634,7 @@ def unsubscribe( should Hide instead.""" sub = _user_subscription(db, user, channel_id) if not sub.yt_subscription_id: - raise HTTPException( - status_code=400, - detail="No YouTube subscription id on record — sync subscriptions first.", - ) + raise refused(400, "no_subscription_id") try: with ( quota.attribute(user.id, quota.QuotaAction.CHANNELS_UNSUBSCRIBE), @@ -649,10 +644,10 @@ def unsubscribe( except YouTubeError as exc: # 422 (a real, explainable error → speaking modal), not 502 which the client treats as # a transient "connection lost" gateway blip. - raise HTTPException(status_code=422, detail=f"YouTube couldn't unsubscribe you: {exc}") + raise youtube_refusal(exc) db.delete(sub) db.commit() - return {"unsubscribed": channel_id} + return {"ok": True, "unsubscribed": channel_id} @router.get("/blocked/list") @@ -693,7 +688,7 @@ def block_channel( ) ) db.commit() - return {"blocked": channel_id} + return {"ok": True, "blocked": channel_id} @router.delete("/{channel_id}/block") @@ -708,7 +703,7 @@ def unblock_channel( ) ) db.commit() - return {"unblocked": channel_id} + return {"ok": True, "unblocked": channel_id} class AttachTagIn(BaseModel): @@ -743,7 +738,7 @@ def attach_tag( ) ) db.commit() - return {"channel_id": channel_id, "tag_id": body.tag_id} + return {"ok": True, "channel_id": channel_id, "tag_id": body.tag_id} @router.delete("/{channel_id}/tags/{tag_id}") @@ -763,4 +758,4 @@ def detach_tag( if row is not None: db.delete(row) db.commit() - return {"channel_id": channel_id, "tag_id": tag_id} + return {"ok": True, "channel_id": channel_id, "tag_id": tag_id} diff --git a/backend/app/routes/config.py b/backend/app/routes/config.py index b850e6be..fbbc0e24 100644 --- a/backend/app/routes/config.py +++ b/backend/app/routes/config.py @@ -4,7 +4,7 @@ app.sysconfig. Secret values (e.g. SMTP password) are write-only — never retur import re from typing import Any -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy.orm import Session @@ -17,6 +17,9 @@ from app.audit import AuditAction from app.auth import admin_user from app.db import get_db from app.models import User +from app.refusal import refused +from app.routes._common import not_found +from app.refusal import setting_refused router = APIRouter(prefix="/api/admin/config", tags=["admin"]) @@ -59,19 +62,16 @@ def set_config( ) -> dict: s = sysconfig.spec(key) if s is None: - raise HTTPException(status_code=404, detail="Unknown config key") + raise not_found("config_key") if "value" not in body.model_fields_set: - raise HTTPException(status_code=400, detail="Missing 'value'") + raise refused(400, "value_missing") if s.secret and not sysconfig.secrets_manageable(): - raise HTTPException( - status_code=400, - detail="Set TOKEN_ENCRYPTION_KEY to store secrets in the database.", - ) + raise refused(400, "no_encryption_key") old = None if s.secret else sysconfig.get(db, key) try: sysconfig.set_value(db, key, body.value) - except (TypeError, ValueError) as exc: - raise HTTPException(status_code=400, detail=str(exc) or "Invalid value") + except sysconfig.InvalidSetting as exc: + raise setting_refused(exc) from None # Never log a secret's value — only that it was changed, by whom. Non-secret values are logged # but with URL credentials redacted (a proxy/URL setting can embed user:pass). Skip the row when # a non-secret value didn't actually change (a no-op re-save shouldn't add trail noise); secrets @@ -101,7 +101,7 @@ def reset_config( ) -> dict: s = sysconfig.spec(key) if s is None: - raise HTTPException(status_code=404, detail="Unknown config key") + raise not_found("config_key") old = None if s.secret else sysconfig.get(db, key) sysconfig.reset(db, key) audit.record( @@ -126,8 +126,8 @@ def send_test_email( """Send a test email to the requesting admin using the current (DB or env) SMTP config — lets the admin verify the settings actually work.""" if not email_mod.email_enabled(): - raise HTTPException(status_code=400, detail="SMTP is not configured.") + raise refused(400, "smtp_off") sent = email_mod.send_test(user.email, mailtext.lang_for_user(user)) if not sent: - raise HTTPException(status_code=422, detail="SMTP send failed — check the settings.") - return {"sent": True, "to": user.email} + raise refused(422, "smtp_failed") + return {"ok": True, "sent": True, "to": user.email} diff --git a/backend/app/routes/discover.py b/backend/app/routes/discover.py index e7447c09..a1f7ae23 100644 --- a/backend/app/routes/discover.py +++ b/backend/app/routes/discover.py @@ -11,9 +11,9 @@ thing it exists to remove. from typing import Literal -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from pydantic import BaseModel -from sqlalchemy import select, update +from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -23,7 +23,7 @@ from app.discover import hero as hero_mod from app.discover import pins, rails from app.discover.budget import Budget, DEFAULT_BUDGET from app.models import ComfortPin, User -from app.routes._common import owned_or_404 +from app.routes._common import owned_or_404, not_found router = APIRouter(prefix="/api/discover", tags=["discover"]) @@ -53,7 +53,9 @@ def get_discover( `resume` and `shelf` are built once and handed to the hero, so the two rails the hero draws from are not queried twice for a single page load.""" resume = rails.continue_watching(db, user) - shelf = pins.shelf(db, user) + # Plex off → the shelf rail (and so the hero's comfort tier) goes YouTube-only like the rails + # beside it; the pins are kept and `GET /pins` still lists them (`pins.shelf`). + shelf = pins.shelf(db, user, include_plex=rails.plex_available(db)) return { "budget": budget, "hero": hero_mod.pick(db, user, budget, seed, resume=resume, shelf=shelf), @@ -125,7 +127,7 @@ def add_pin( constraint is the mechanism, not the message.""" resolved = pins.resolve_ref(db, user, body.kind, body.ref) if resolved is None: - raise HTTPException(status_code=404, detail="Unknown item") + raise not_found("item") column, value = resolved existing = db.execute( @@ -157,7 +159,7 @@ def add_pin( # between `resolve_ref` confirming it and this INSERT (a Plex sync dropping a removed # item), which trips the FK. Semantically that is the same "no such thing" the pre-check # returns above — re-raising handed the user a 500 they could do nothing with. - raise HTTPException(status_code=404, detail="Unknown item") + raise not_found("item") return {"ok": True, "id": winner.id, "created": False} return {"ok": True, "id": pin.id, "created": True} @@ -180,18 +182,11 @@ def reorder_pins( user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: - """Rewrite the shelf order from a full list of pin ids. + """Rewrite the shelf order from the list of pin ids the client shows. Whole-list rather than a move-one delta: the shelf is short, and a full rewrite cannot leave two - tiles claiming the same slot the way an interrupted sequence of single moves can. Ids that are not - this user's are ignored rather than 404'd — a stale tab replaying an order that includes a pin - deleted meanwhile should reorder what still exists, not fail wholesale.""" - owned = {p.id for p in pins.list_pins(db, user)} - for position, pin_id in enumerate(i for i in body.ids if i in owned): - db.execute( - update(ComfortPin) - .where(ComfortPin.id == pin_id, ComfortPin.user_id == user.id) - .values(position=position) - ) + tiles claiming the same slot the way an interrupted sequence of single moves can. The list may be + PARTIAL (hidden Plex pins, a stale tab) — `pins.reorder` keeps what it leaves out in place.""" + pins.reorder(db, user, body.ids) db.commit() return {"ok": True} diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index 3077f192..90e36197 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -38,10 +38,11 @@ from app.models import ( MediaAsset, User, ) -from app.routes._common import get_or_404, not_found +from app.routes._common import get_or_404, not_found, owned_or_404 from app.security import hash_password from app.userscope import is_messageable_user, messageable_clauses from app.utils import iso +from app.refusal import refused router = APIRouter(prefix="/api/downloads", tags=["downloads"]) admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"]) @@ -63,7 +64,7 @@ def resolve_source(raw: str) -> tuple[str, str]: as a raw URL for yt-dlp's generic extractor.""" raw = (raw or "").strip() if not raw: - raise HTTPException(status_code=400, detail="A video link or id is required.") + raise refused(400, "link_required") if _VIDEO_ID.match(raw): return "youtube", raw if raw.startswith(("http://", "https://")): @@ -93,7 +94,7 @@ def resolve_source(raw: str) -> tuple[str, str]: if _VIDEO_ID.match(vid): return "youtube", vid return "url", raw - raise HTTPException(status_code=400, detail="That doesn't look like a valid video link.") + raise refused(400, "invalid_link") def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: @@ -166,15 +167,8 @@ def _serialize_job(db: Session, job: DownloadJob) -> dict: return _serialize(job, asset) -def _own_job(db: Session, user: User, job_id: int) -> DownloadJob: - job = db.get(DownloadJob, job_id) - if job is None or job.user_id != user.id: - raise HTTPException(status_code=404, detail="Unknown download") - return job - - def _own_job_for_update(db: Session, user: User, job_id: int) -> DownloadJob: - """`_own_job` for a transition that moves its asset's ref count (resume, cancel, delete, redownload) + """The ownership check (`owned_or_404`) for a transition that moves its asset's ref count (resume, cancel, delete, redownload) or writes the status those decide on (pause). The job is re-read under a row lock, so two overlapping requests for one job run one after the other and the second decides on the first's committed result — never both releasing the same hold, and never a pause turning a job a cancel @@ -183,7 +177,7 @@ def _own_job_for_update(db: Session, user: User, job_id: int) -> DownloadJob: The asset is locked BEFORE the job: the order a finishing worker child takes them in (`service.finalize_if_running`), so the two can never deadlock.""" while True: - asset_id = _own_job(db, user, job_id).asset_id + asset_id = owned_or_404(db, DownloadJob, job_id, user, what="download").asset_id if asset_id is not None: db.execute(select(MediaAsset.id).where(MediaAsset.id == asset_id).with_for_update()) job = db.execute( @@ -193,7 +187,7 @@ def _own_job_for_update(db: Session, user: User, job_id: int) -> DownloadJob: .execution_options(populate_existing=True) ).scalar_one_or_none() if job is None: - raise HTTPException(status_code=404, detail="Unknown download") + raise not_found("download") if job.asset_id == asset_id: return job db.rollback() # another request moved the job to another asset meanwhile: lock that one @@ -240,7 +234,7 @@ def create_profile( name = body.name.strip() spec = body.spec if not name: - raise HTTPException(status_code=400, detail="A profile name is required.") + raise refused(400, "profile_name_required") maxpos = db.execute( select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where( DownloadProfile.user_id == user.id @@ -268,16 +262,16 @@ def update_profile( ) -> dict: p = db.get(DownloadProfile, profile_id) if p is None or p.is_builtin or p.user_id != user.id: - raise HTTPException(status_code=404, detail="Unknown profile") + raise not_found("profile") sent = body.model_fields_set if "name" in sent: name = (body.name or "").strip() if not name: - raise HTTPException(status_code=400, detail="A profile name is required.") + raise refused(400, "profile_name_required") p.name = name[:80] if "spec" in sent: if body.spec is None: - raise HTTPException(status_code=400, detail="spec must be an object") + raise refused(400, "spec_not_object") p.spec = body.spec db.commit() return _serialize_profile(p) @@ -289,7 +283,7 @@ def delete_profile( ) -> dict: p = db.get(DownloadProfile, profile_id) if p is None or p.is_builtin or p.user_id != user.id: - raise HTTPException(status_code=404, detail="Unknown profile") + raise not_found("profile") db.delete(p) db.commit() return {"deleted": profile_id} @@ -319,7 +313,7 @@ def set_credential( """Store (encrypted) this user's cookies.txt for one supported host. Write-only: the value is never echoed back, logged, or audited — only the fact that it changed.""" if not dlcreds.is_supported_host(host_key): - raise HTTPException(status_code=404, detail="Unknown host") + raise not_found("host") try: cleaned = dlcreds.validate_cookies(body.cookies) except dlcreds.InvalidCookies as e: @@ -341,7 +335,7 @@ def delete_credential( host_key: str, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> dict: if not dlcreds.is_supported_host(host_key): - raise HTTPException(status_code=404, detail="Unknown host") + raise not_found("host") if dlcreds.delete_credential(db, user.id, host_key): audit.record( db, @@ -362,7 +356,7 @@ def _resolve_spec( if profile_id is not None: p = db.get(DownloadProfile, profile_id) if p is None or (not p.is_builtin and p.user_id != user.id): - raise HTTPException(status_code=404, detail="Unknown profile") + raise not_found("profile") return p.spec, p.id if spec is not None: return spec, None @@ -373,7 +367,7 @@ def _resolve_spec( .limit(1) ).scalar_one_or_none() if fallback is None: - raise HTTPException(status_code=400, detail="No download profile available.") + raise refused(400, "no_profile") return fallback.spec, fallback.id @@ -484,7 +478,7 @@ def enqueue_edit( source file is only read, never modified, so editing a shared video is safe.""" source_job = _accessible_job(db, user, body.source_job_id) if source_job.status != "done" or source_job.asset_id is None: - raise HTTPException(status_code=409, detail="You can only edit a finished download.") + raise refused(409, "edit_unfinished") spec = body.edit_spec display_name = (body.display_name or "").strip() or None try: @@ -614,7 +608,7 @@ def update_download_meta( ) -> dict: """Update a download's display metadata: title, channel name + link, and extra reference URLs. All are display-only overrides — the on-disk file keeps its system-decided, id-bound name.""" - job = _own_job(db, user, job_id) + job = owned_or_404(db, DownloadJob, job_id, user, what="download") sent = body.model_fields_set if "display_name" in sent: job.display_name = ((body.display_name or "").strip())[:255] or None @@ -668,9 +662,9 @@ def redownload_download( if job.job_kind in ("edit", "upload"): # Neither has a re-fetchable source: an editor clip is derived locally, an upload IS the # source. Re-running yt-dlp against source_kind='upload'/'edit' would just fail. - raise HTTPException(status_code=409, detail="This item can't be re-downloaded.") + raise refused(409, "not_redownloadable") if job.status in ("queued", "running", "paused"): - raise HTTPException(status_code=409, detail="This download is still in progress.") + raise refused(409, "still_downloading") service.redownload(db, job) return _serialize_job(db, job) @@ -699,7 +693,7 @@ def delete_download( def _accessible_job(db: Session, user: User, job_id: int) -> DownloadJob: job = db.get(DownloadJob, job_id) if job is None: - raise HTTPException(status_code=404, detail="Unknown download") + raise not_found("download") if job.user_id == user.id: return job shared = db.execute( @@ -708,7 +702,7 @@ def _accessible_job(db: Session, user: User, job_id: int) -> DownloadJob: ) ).first() if shared is None: - raise HTTPException(status_code=404, detail="Unknown download") + raise not_found("download") return job @@ -722,10 +716,10 @@ def download_file( job = _accessible_job(db, user, job_id) asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or asset.status != "ready" or not asset.rel_path: - raise HTTPException(status_code=409, detail="This download isn't ready.") + raise refused(409, "download_not_ready") path = storage.safe_abs_path(settings.download_root, asset.rel_path) if path is None: - raise HTTPException(status_code=404, detail="File is no longer available.") + raise not_found("file") filename = storage.download_filename( job.display_name or asset.title or asset.source_ref, asset.container, path @@ -764,7 +758,7 @@ def storyboard_meta( job = _accessible_job(db, user, job_id) asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or asset.status != "ready" or not asset.rel_path: - raise HTTPException(status_code=409, detail="This download isn't ready.") + raise refused(409, "download_not_ready") meta = editmod.ensure_storyboard(asset, settings.download_root) if meta is None: return {"available": False} @@ -785,10 +779,10 @@ def poster_image(job_id: int, user: User = Depends(require_human), db: Session = job = _accessible_job(db, user, job_id) asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or not asset.poster_path: - raise HTTPException(status_code=404, detail="No poster.") + raise not_found("poster") path = storage.safe_abs_path(settings.download_root, asset.poster_path) if path is None: - raise HTTPException(status_code=404, detail="No poster.") + raise not_found("poster") return FileResponse(path, media_type="image/jpeg") @@ -799,10 +793,10 @@ def storyboard_image( job = _accessible_job(db, user, job_id) asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None if asset is None or not asset.storyboard_path: - raise HTTPException(status_code=404, detail="No filmstrip.") + raise not_found("filmstrip") path = storage.safe_abs_path(settings.download_root, asset.storyboard_path) if path is None: - raise HTTPException(status_code=404, detail="No filmstrip.") + raise not_found("filmstrip") return FileResponse(path, media_type="image/jpeg") @@ -820,12 +814,12 @@ def share_download( user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: - job = _own_job(db, user, job_id) + job = owned_or_404(db, DownloadJob, job_id, user, what="download") if job.status != "done" or job.asset_id is None: - raise HTTPException(status_code=409, detail="Only a finished download can be shared.") + raise refused(409, "share_unfinished") target = body.email.strip().lower() if not target: - raise HTTPException(status_code=400, detail="A recipient email is required.") + raise refused(400, "recipient_required") recipient = db.execute( select(User).where(func.lower(User.email) == target) ).scalar_one_or_none() @@ -834,7 +828,7 @@ def share_download( if not is_messageable_user(recipient): raise not_found("recipient") if recipient.id == user.id: - raise HTTPException(status_code=400, detail="That's already your download.") + raise refused(400, "share_own") exists = db.execute( select(DownloadShare.id).where( DownloadShare.job_id == job_id, DownloadShare.to_user_id == recipient.id @@ -853,7 +847,7 @@ def share_download( commit=False, ) db.commit() - return {"shared_with": recipient.email} + return {"ok": True, "shared_with": recipient.email} @router.delete("/{job_id}/share/{email}") @@ -863,7 +857,9 @@ def unshare_download( user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: - _own_job(db, user, job_id) # ownership guard (404s if not the caller's job) + owned_or_404( + db, DownloadJob, job_id, user, what="download" + ) # ownership guard (404s if not the caller's job) recipient = db.execute( select(User).where(func.lower(User.email) == email.strip().lower()) ).scalar_one_or_none() @@ -895,7 +891,7 @@ def remove_shared_with_me( if row is not None: db.delete(row) db.commit() - return {"removed": job_id} + return {"ok": True, "removed": job_id} # --- recipients (for the internal "share with a user" picker) ------------------------------ @@ -925,10 +921,10 @@ def share_recipients( def _own_link(db: Session, user: User, link_id: int) -> DownloadLink: link = db.get(DownloadLink, link_id) if link is None: - raise HTTPException(status_code=404, detail="Unknown link") + raise not_found("link") job = db.get(DownloadJob, link.job_id) if job is None or job.user_id != user.id: - raise HTTPException(status_code=404, detail="Unknown link") + raise not_found("link") return link @@ -944,7 +940,7 @@ def _link_expiry(days: int | None) -> datetime | None: def list_links( job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db) ) -> list[dict]: - _own_job(db, user, job_id) + owned_or_404(db, DownloadJob, job_id, user, what="download") rows = ( db.execute( select(DownloadLink) @@ -975,9 +971,9 @@ def create_link( user: User = Depends(require_human), db: Session = Depends(get_db), ) -> dict: - job = _own_job(db, user, job_id) + job = owned_or_404(db, DownloadJob, job_id, user, what="download") if job.status != "done" or job.asset_id is None: - raise HTTPException(status_code=409, detail="Only a finished download can be shared.") + raise refused(409, "share_unfinished") pw = body.password.strip() link = DownloadLink( job_id=job_id, diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 9c2dac8d..fb5edcb3 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -1,6 +1,6 @@ from datetime import date, datetime, timezone -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, Depends, Query from pydantic import BaseModel, ConfigDict, Field from sqlalchemy import and_, func, or_, select from sqlalchemy.dialects.postgresql import insert as pg_insert @@ -29,10 +29,11 @@ from app.models import ( Video, VideoState, ) -from app.routes._common import get_or_404 +from app.routes._common import get_or_404, youtube_refusal, not_found from app.sync.videos import parse_iso8601_duration from app.utils import iso from app.youtube.client import YouTubeClient, YouTubeError +from app.refusal import refused router = APIRouter(prefix="/api", tags=["feed"]) @@ -120,7 +121,7 @@ def get_feed( vals, cursor_id = decode_cursor(keys, cursor) except ValueError: # `from None`: the ValueError is an expected bad-input signal, not an error to chain. - raise HTTPException(status_code=400, detail="Invalid feed cursor") from None + raise refused(400, "invalid_cursor") from None for k, v in zip(keys, vals): k["value"] = v query = query.where(keyset_predicate(keys, cursor_id)) @@ -235,7 +236,7 @@ def set_video_state( ) -> dict: status = body.status if status not in VALID_STATES: - raise HTTPException(status_code=400, detail=f"status must be one of {VALID_STATES}") + raise refused(400, "invalid_status") get_or_404(db, Video, video_id, what="video") now = datetime.now(timezone.utc) @@ -260,7 +261,7 @@ def set_video_state( # or moved this row. The state the caller asked for is the state that holds either way. except StaleDataError: db.rollback() - return {"video_id": video_id, "status": "new"} + return {"ok": True, "video_id": video_id, "status": "new"} # watched / hidden: atomic upsert. The player marks "watched" at end-of-playback at # the same instant a progress checkpoint may DELETE the in-progress row; a plain @@ -281,7 +282,7 @@ def set_video_state( ) db.execute(stmt) db.commit() - return {"video_id": video_id, "status": status} + return {"ok": True, "video_id": video_id, "status": status} @router.delete("/videos/{video_id}/state") @@ -303,7 +304,7 @@ def clear_video_state( # silent-ok: the row went under the delete, which is the outcome the delete wanted. except StaleDataError: db.rollback() - return {"video_id": video_id, "status": "new", "position_seconds": 0} + return {"ok": True, "video_id": video_id, "status": "new", "position_seconds": 0} class VideoProgressIn(BaseModel): @@ -361,7 +362,7 @@ def set_video_progress( # silent-ok: exactly the race the comment above describes; the response stays true. except StaleDataError: db.rollback() - return {"video_id": video_id, "position_seconds": 0} + return {"ok": True, "video_id": video_id, "position_seconds": 0} if row is None: row = VideoState(user_id=user.id, video_id=video_id) @@ -369,7 +370,7 @@ def set_video_progress( row.position_seconds = position row.progress_updated_at = datetime.now(timezone.utc) db.commit() - return {"video_id": video_id, "position_seconds": position} + return {"ok": True, "video_id": video_id, "position_seconds": position} @router.get("/videos/{video_id}") @@ -402,7 +403,7 @@ def get_video_detail( # The DB path above is free; the off-catalog YouTube lookup below spends a quota unit, so the # demo account can't reach it — it just sees "unknown" for a video we don't already store. if user.is_demo: - raise HTTPException(status_code=404, detail="Unknown video") + raise not_found("video") try: with ( @@ -411,9 +412,9 @@ def get_video_detail( ): items = yt.get_videos([video_id]) except YouTubeError as exc: - raise HTTPException(status_code=422, detail=f"YouTube lookup failed: {exc}") + raise youtube_refusal(exc) if not items: - raise HTTPException(status_code=404, detail="Unknown video") + raise not_found("video") snippet = items[0].get("snippet", {}) stats = items[0].get("statistics", {}) diff --git a/backend/app/routes/me.py b/backend/app/routes/me.py index 74b15360..6f1784cf 100644 --- a/backend/app/routes/me.py +++ b/backend/app/routes/me.py @@ -24,6 +24,8 @@ from app import sysconfig from app.db import get_db from app.models import Invite, User from app.utils import iso +from app.refusal import refused +from app.routes._common import not_found router = APIRouter(prefix="/api/me", tags=["me"]) @@ -70,7 +72,7 @@ def switch_account( here) are allowed, and access is re-checked in case the invite was revoked since.""" target = body.user_id if target not in (request.session.get("account_ids") or []): - raise HTTPException(status_code=403, detail="Not an account you've signed into here.") + raise refused(403, "not_your_account") # An account `_session_user` would refuse — gone, deactivated, suspended, or its session revoked — is # refused here too, and dropped from the switch list the way `_session_user` drops it. Made the session # default instead, the next request's current_user would refuse it as the DEFAULT and clear the whole @@ -78,22 +80,20 @@ def switch_account( u = db.get(User, target) if u is None: drop_from_wallet(request.session, target) - raise HTTPException(status_code=404, detail="That account no longer exists.") + raise not_found("account") if not is_allowed(db, u.email): - raise HTTPException(status_code=403, detail="That account no longer has access.") + raise refused(403, "account_no_access") if not u.is_active: drop_from_wallet(request.session, target) - raise HTTPException(status_code=403, detail="That account no longer has access.") + raise refused(403, "account_no_access") if u.is_suspended: drop_from_wallet(request.session, target) - raise HTTPException(status_code=403, detail="That account is suspended.") + raise refused(403, "account_suspended") # Revoked: a password reset/change or "log out everywhere" bumped the account's epoch. epochs = request.session.get("epochs") or {} if int(epochs.get(str(target), 0)) != (u.session_epoch or 0): drop_from_wallet(request.session, target) - raise HTTPException( - status_code=403, detail="That account was signed out elsewhere — sign in again." - ) + raise refused(403, "account_signed_out") request.session["user_id"] = target return {"ok": True, "user_id": target} @@ -181,10 +181,7 @@ def delete_my_account( if user.is_demo: raise HTTPException(status_code=403, detail={"code": "demo", "reason": "delete_account"}) if user.role == "admin" and count_admins(db) <= 1: - raise HTTPException( - status_code=400, - detail="You're the only admin — promote another admin before deleting your account.", - ) + raise refused(400, "only_admin_delete") user_id = user.id # Full erasure (cascades) + access-request cleanup + Google-grant revocation; shared with the # admin delete path. Capture the id first — `user` is detached once purge_user deletes the row. @@ -217,7 +214,7 @@ def update_preferences( user.preferences = merged db.add(user) db.commit() - return {"preferences": merged} + return {"ok": True, "preferences": merged} class DiagnosticsIn(BaseModel): diff --git a/backend/app/routes/messages.py b/backend/app/routes/messages.py index e9f705c6..1cbe96fd 100644 --- a/backend/app/routes/messages.py +++ b/backend/app/routes/messages.py @@ -16,7 +16,7 @@ routing and unread counts. from datetime import datetime, timezone -from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect +from fastapi import APIRouter, Depends, WebSocket, WebSocketDisconnect from pydantic import BaseModel from sqlalchemy import and_, case, delete, func, or_, select from sqlalchemy.orm import Session @@ -29,6 +29,8 @@ from app.realtime import manager from app.security import verify_password from app.userscope import is_messageable_user, messageable_clauses from app.utils import iso +from app.refusal import refused +from app.routes._common import not_found router = APIRouter(prefix="/api/messages", tags=["messages"]) @@ -152,7 +154,7 @@ def setup_keys( """Store this user's public key + passphrase-wrapped private key. Set-once: changing the key would orphan every existing conversation key, so re-setup is rejected here.""" if db.get(MessageKey, user.id) is not None: - raise HTTPException(status_code=409, detail="Secure messaging is already set up.") + raise refused(409, "messaging_set_up") db.add( MessageKey( user_id=user.id, @@ -164,7 +166,7 @@ def setup_keys( ) ) db.commit() - return {"configured": True} + return {"ok": True, "configured": True} @router.post("/keys/reset") @@ -185,11 +187,11 @@ def reset_keys( less (Google-only) account has nothing to re-verify, so it relies on its live session plus the client-side danger confirmation.""" if not _reset_key_limiter.allow(str(user.id)): - raise HTTPException(status_code=429, detail="Too many attempts. Try again shortly.") + raise refused(429, "too_many_attempts") if user.password_hash and not verify_password( payload.current_password or "", user.password_hash ): - raise HTTPException(status_code=403, detail="Current password is incorrect.") + raise refused(403, "wrong_password") # Delete the now-dead ciphertext first (both directions), then the wrapped keypair itself so a # fresh setup is allowed. Order doesn't matter (no FK between them), but do it in one commit. db.execute( @@ -202,7 +204,7 @@ def reset_keys( if key is not None: db.delete(key) db.commit() - return {"configured": False} + return {"ok": True, "configured": False} @router.get("/keys/{user_id}") @@ -223,10 +225,10 @@ def public_key( p = db.get(User, user_id) has_history = db.scalar(select(Message.id).where(cond).limit(1)) if p is None or (not is_messageable_user(p) and not has_history): - raise HTTPException(status_code=404, detail="User not found") + raise not_found("user") k = db.get(MessageKey, user_id) if k is None: - raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.") + raise refused(404, "no_messaging") return {"user_id": user_id, "public_key": k.public_key} @@ -344,7 +346,7 @@ def get_thread( # still opens). Otherwise the same 404 as a non-existent id → no enumeration oracle. has_history = db.scalar(select(Message.id).where(cond).limit(1)) if p is None or (not is_messageable_user(p) and not has_history): - raise HTTPException(status_code=404, detail="User not found") + raise not_found("user") partner = _serialize_user(p) msgs = ( db.execute(select(Message).where(cond).order_by(Message.created_at.asc())).scalars().all() @@ -374,20 +376,20 @@ def send_message( """Send an end-to-end-encrypted message. The server stores only the ciphertext the client produced, then pushes it live to the recipient (and the sender's other devices).""" if not _send_limiter.allow(f"msg:{user.id}"): - raise HTTPException(status_code=429, detail="Too many messages — slow down a moment.") + raise refused(429, "message_rate") ct = (payload.ciphertext or "").strip() iv = (payload.iv or "").strip() if not ct or not iv: - raise HTTPException(status_code=400, detail="Message is empty.") + raise refused(400, "message_empty") if len(ct) > MAX_CIPHERTEXT: - raise HTTPException(status_code=400, detail="Message is too long.") + raise refused(400, "message_too_long") if payload.recipient_id == user.id: - raise HTTPException(status_code=400, detail="You can't message yourself.") + raise refused(400, "message_self") recipient = db.get(User, payload.recipient_id) if not is_messageable_user(recipient): - raise HTTPException(status_code=404, detail="Recipient not available.") + raise not_found("recipient") if db.get(MessageKey, recipient.id) is None: - raise HTTPException(status_code=404, detail="That member hasn't set up messaging yet.") + raise refused(404, "no_messaging") m = Message(kind="user", sender_id=user.id, recipient_id=recipient.id, ciphertext=ct, iv=iv) db.add(m) db.commit() diff --git a/backend/app/routes/notifications.py b/backend/app/routes/notifications.py index f5d43aca..33fb61d4 100644 --- a/backend/app/routes/notifications.py +++ b/backend/app/routes/notifications.py @@ -5,7 +5,7 @@ All endpoints use `current_user` (not `require_human`): notifications and their state are personal UI state with no quota or YouTube involvement, so the shared demo account manages its own inbox normally.""" -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, Depends, Query from sqlalchemy import func, select, update from sqlalchemy.orm import Session @@ -14,6 +14,7 @@ from app.db import get_db from app.models import Notification, User from app.notifications import trim_read from app.utils import iso +from app.routes._common import owned_or_404 router = APIRouter(prefix="/api/me/notifications", tags=["notifications"]) @@ -72,20 +73,13 @@ def unread_count(user: User = Depends(current_user), db: Session = Depends(get_d return {"count": count or 0} -def _owned(db: Session, notification_id: int, user: User) -> Notification: - n = db.get(Notification, notification_id) - if n is None or n.user_id != user.id: - raise HTTPException(status_code=404, detail="Notification not found") - return n - - @router.post("/{notification_id}/read") def mark_read( notification_id: int, user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - n = _owned(db, notification_id, user) + n = owned_or_404(db, Notification, notification_id, user, what="notification") n.read = True db.commit() trim_read(db, user.id) @@ -110,7 +104,7 @@ def dismiss( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - n = _owned(db, notification_id, user) + n = owned_or_404(db, Notification, notification_id, user, what="notification") # Dismiss implies read (it no longer counts toward the unread badge). n.dismissed = True n.read = True diff --git a/backend/app/routes/playlists.py b/backend/app/routes/playlists.py index c7617c50..f4a09fb0 100644 --- a/backend/app/routes/playlists.py +++ b/backend/app/routes/playlists.py @@ -3,11 +3,10 @@ CRUD (create / rename / delete, add / remove / reorder items) plus YouTube two-w mirror the user's own YouTube playlists in (`/sync-youtube`) and push/revert local edits back out (`/{id}/push-plan`, `/{id}/revert-youtube`), behind the optional write scope.""" -from datetime import datetime, timezone - from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy import func, select, and_ +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, aliased from app import quota @@ -23,7 +22,7 @@ from app.sync.playlists import ( sync_user_playlists, ) from app.ratelimit import RateLimiter -from app.routes._common import get_or_404, owned_or_404 +from app.routes._common import get_or_404, owned_or_404, youtube_refusal from app.youtube.client import YouTubeClient, YouTubeError router = APIRouter(prefix="/api/playlists", tags=["playlists"]) @@ -48,10 +47,6 @@ def _check_pull_cooldown(user: User) -> None: raise HTTPException(status_code=429, detail={"code": "sync_cooldown"}) -def _own_playlist(db: Session, user: User, playlist_id: int) -> Playlist: - return owned_or_404(db, Playlist, playlist_id, user, what="playlist") - - def _refused(status: int, reason: str) -> HTTPException: """A playlist action the server refuses, as a reason the UI translates (`playlists.errors.`) — a bare English detail reached the Hungarian UI verbatim.""" @@ -64,8 +59,11 @@ def _touch(pl: Playlist) -> None: `UpdatedAtMixin`'s `onupdate` fires on an UPDATE of the PLAYLIST row, and add / remove / reorder / move write the item rows instead — so without this the rail's "recently changed" sort would only ever see renames. The assignment is what makes the column mean "last time - this list changed", which is what the rail promises the reader.""" - pl.updated_at = datetime.now(timezone.utc) + this list changed", which is what the rail promises the reader. + + The DATABASE's clock, like the mixin's own `onupdate`: one sort key written from two clocks + (this process's and Postgres's) could order a rename and an item edit the wrong way round.""" + pl.updated_at = func.now() def _next_playlist_position(db: Session, user_id: int) -> int: @@ -108,7 +106,7 @@ def _add_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) -> if mark_dirty: recompute_dirty(db, pl) _touch(pl) - db.commit() + db.flush() # the handler commits (CLAUDE.md: helpers flush) return True @@ -126,7 +124,7 @@ def _remove_item(db: Session, pl: Playlist, video_id: str, *, mark_dirty: bool) if mark_dirty: recompute_dirty(db, pl) _touch(pl) - db.commit() + db.flush() # the handler commits (CLAUDE.md: helpers flush) return True @@ -280,7 +278,7 @@ def _watch_later(db: Session, user: User) -> Playlist: position=_next_playlist_position(db, user.id), ) db.add(pl) - db.commit() + db.flush() # the id, for the item the caller adds next; the handler commits return pl @@ -295,7 +293,8 @@ def add_watch_later( get_or_404(db, Video, video_id, what="video") pl = _watch_later(db, user) _add_item(db, pl, video_id, mark_dirty=False) - return {"saved": True, "playlist_id": pl.id} + db.commit() + return {"ok": True, "saved": True, "playlist_id": pl.id} @router.delete("/watch-later/{video_id}") @@ -309,7 +308,8 @@ def remove_watch_later( ) if pl is not None: _remove_item(db, pl, video_id, mark_dirty=False) - return {"saved": False} + db.commit() + return {"ok": True, "saved": False} @router.post("/sync-youtube") @@ -322,8 +322,12 @@ def sync_youtube( if not has_read_scope(user): raise _refused(403, "youtube_read_scope") _check_pull_cooldown(user) - with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): - return sync_user_playlists(db, user) + try: + with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): + return {"ok": True, **sync_user_playlists(db, user)} + except YouTubeError as exc: + db.rollback() + raise youtube_refusal(exc) @router.post("/{playlist_id}/revert-youtube") @@ -335,7 +339,7 @@ def revert_youtube( """Reset one YouTube-linked playlist to its current YouTube state, discarding local edits (and clearing the dirty flag). The user-facing 'undo all my changes' for a list once the in-session undo history is gone.""" - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") if not pl.yt_playlist_id: raise _refused(400, "not_on_youtube") if not has_read_scope(user): @@ -343,16 +347,16 @@ def revert_youtube( _check_pull_cooldown(user) try: with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PULL): - return repull_playlist(db, user, pl) - except YouTubeError: + return {"ok": True, **repull_playlist(db, user, pl)} + except YouTubeError as exc: db.rollback() - raise _refused(422, "youtube_failed") + raise youtube_refusal(exc) def _pushable(db: Session, user: User, playlist_id: int) -> Playlist: """A local playlist the user may push to YouTube (create or update). The built-in Watch later and read-only YouTube mirrors are never pushable.""" - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") if pl.kind == "watch_later": raise _refused(400, "watch_later_not_pushable") return pl @@ -371,8 +375,8 @@ def push_plan( try: with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_PUSH): plan = plan_merge(db, user, pl) - except YouTubeError: - raise _refused(422, "youtube_failed") + except YouTubeError as exc: + raise youtube_refusal(exc) plan["remaining_today"] = quota.remaining_today(db) plan["affordable"] = plan["units_estimate"] <= plan["remaining_today"] return plan @@ -403,10 +407,10 @@ def push( plan = plan_merge(db, user, pl, live=live) if plan["units_estimate"] > quota.remaining_today(db): raise _refused(429, "quota_exceeded") - return merge_playlist(db, user, pl, live=live) - except YouTubeError: + return {"ok": True, **merge_playlist(db, user, pl, live=live)} + except YouTubeError as exc: db.rollback() - raise _refused(422, "youtube_failed") + raise youtube_refusal(exc) @router.get("/{playlist_id}") @@ -415,7 +419,7 @@ def get_playlist( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") state = aliased(VideoState) rows = db.execute( select(*feed_columns(user, state)) @@ -447,7 +451,7 @@ def rename_playlist( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") if "name" in body.model_fields_set: name = (body.name or "").strip() if not name: @@ -484,7 +488,7 @@ def delete_playlist( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") if pl.kind == "watch_later": raise _refused(400, "watch_later_undeletable") if on_youtube: @@ -496,8 +500,8 @@ def delete_playlist( with quota.attribute(user.id, quota.QuotaAction.PLAYLISTS_DELETE): with YouTubeClient(db, user) as yt: yt.delete_playlist(pl.yt_playlist_id) - except YouTubeError: - raise _refused(422, "youtube_failed") + except YouTubeError as exc: + raise youtube_refusal(exc) db.delete(pl) # items cascade db.commit() return {"deleted": playlist_id} @@ -510,11 +514,12 @@ def add_item( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") video_id = body.video_id get_or_404(db, Video, video_id, what="video") _add_item(db, pl, video_id, mark_dirty=True) - return {"playlist_id": pl.id, "video_id": video_id} + db.commit() + return {"ok": True, "playlist_id": pl.id, "video_id": video_id} @router.delete("/{playlist_id}/items/{video_id}") @@ -524,9 +529,10 @@ def remove_item( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") _remove_item(db, pl, video_id, mark_dirty=True) - return {"playlist_id": pl.id, "video_id": video_id} + db.commit() + return {"ok": True, "playlist_id": pl.id, "video_id": video_id} class MoveItemIn(BaseModel): @@ -548,7 +554,7 @@ def move_item( exactly the kind of half-applied state this codebase treats as a bug rather than an edge case. Idempotent: if the video is already in the target, the source row is simply dropped (moving into a list that already has it is a removal, not a duplicate).""" - src = _own_playlist(db, user, playlist_id) + src = owned_or_404(db, Playlist, playlist_id, user, what="playlist") # Its own reason: "this playlist doesn't exist any more" about a TARGET that is gone was read as # the list on screen — which is still there. dst = owned_or_404(db, Playlist, body.target_playlist_id, user, what="move_target") @@ -571,19 +577,33 @@ def move_item( PlaylistItem.playlist_id == dst.id, PlaylistItem.video_id == video_id ) ) - if already is not None: - db.delete(item) # the uq_playlist_video constraint would reject a second row anyway - else: + dst_changed = False + if already is None: # Re-parent the existing row rather than delete+insert: `added_at` ("when I saved this") - # survives the move, and there is no window where the video belongs to neither list. - item.playlist_id = dst.id - item.position = _next_item_position(db, dst.id) - # Both sides changed, so both get the dirty recompute and the timestamp — the asymmetric - # version of this (touching only the source) is the recurring smell CLAUDE.md warns about. + # survives the move, and there is no window where the video belongs to neither list. The + # check above is not a lock: a second tab's add or the scheduler's YouTube merge can insert + # the same (playlist, video) in between, and uq_playlist_video then refuses the re-parent. + # That is the already-present outcome arriving late, not a failure — answer it the same way + # instead of a raw 500. + try: + with db.begin_nested(): + item.playlist_id = dst.id + item.position = _next_item_position(db, dst.id) + db.flush() + dst_changed = True + # silent-ok: the conflict IS the answer — the target already holds the video, which is the + # idempotent outcome; the savepoint put the row back in the source list, dropped below. + except IntegrityError: + pass + if not dst_changed: + db.delete(item) + # The source always changed; the target only when the row landed in it. `_touch` on a target + # that already had the video promoted an untouched list in the rail's "recently changed" sort. recompute_dirty(db, src) - recompute_dirty(db, dst) _touch(src) - _touch(dst) + if dst_changed: + recompute_dirty(db, dst) + _touch(dst) db.commit() return {"ok": True, "from_playlist_id": src.id, "to_playlist_id": dst.id, "video_id": video_id} @@ -599,7 +619,7 @@ def reorder_items( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist(db, user, playlist_id) + pl = owned_or_404(db, Playlist, playlist_id, user, what="playlist") order = body.video_ids items = { it.video_id: it @@ -625,4 +645,4 @@ def reorder_items( recompute_dirty(db, pl) _touch(pl) db.commit() - return {"playlist_id": pl.id, "count": pos} + return {"ok": True, "playlist_id": pl.id, "count": pos} diff --git a/backend/app/routes/plex/_shared.py b/backend/app/routes/plex/_shared.py index 02ec2fc3..78319f9c 100644 --- a/backend/app/routes/plex/_shared.py +++ b/backend/app/routes/plex/_shared.py @@ -1,11 +1,11 @@ """Shared helpers, serializers and constants for the plex route package.""" -from fastapi import HTTPException from sqlalchemy.orm import Session from app import logs from app.models import PlexItem from app.plex.lookups import by_rating_key +from app.routes._common import not_found log = logs.get_logger(__name__) @@ -13,7 +13,7 @@ log = logs.get_logger(__name__) def item_or_404(db: Session, rating_key: str) -> PlexItem: it = by_rating_key(db, PlexItem, rating_key) if it is None: - raise HTTPException(status_code=404, detail="Unknown Plex item") + raise not_found("plex_item") return it diff --git a/backend/app/routes/plex/admin.py b/backend/app/routes/plex/admin.py index 23600b0b..82642f9e 100644 --- a/backend/app/routes/plex/admin.py +++ b/backend/app/routes/plex/admin.py @@ -2,12 +2,12 @@ from datetime import datetime, timezone -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.orm import Session -from app import sysconfig +from app import logs, sysconfig from app.auth import admin_user, current_user from app.db import get_db from app.models import ( @@ -20,6 +20,9 @@ from app.plex import sync as plex_sync from app.plex import watch_sync as plex_watch from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.utils import iso +from app.refusal import refused + +log = logs.get_logger(__name__) router = APIRouter(prefix="/api/plex", tags=["plex"]) @@ -35,10 +38,15 @@ def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db) info = plex.server_info() sections = plex.sections() media_check = _media_check(db, plex, sections) - except PlexNotConfigured as e: - raise HTTPException(status_code=400, detail=str(e)) + except PlexNotConfigured: + raise refused(400, "plex_not_configured") from None except PlexError as e: - raise HTTPException(status_code=422, detail=f"Plex connection failed: {e}") + # What the upstream said is for the log, not the toast: it can quote the server URL back. + log.warning( + "the Plex connection test failed", + extra=logs.event("plex.admin.connection_failed", reason=type(e).__name__, error=str(e)), + ) + raise refused(422, "plex_unreachable") from None return { "ok": True, "server_name": info.get("friendlyName") or info.get("title1") or "Plex", @@ -95,7 +103,7 @@ def _media_check(db: Session, plex: PlexClient, sections: list[dict]) -> dict: def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a while on a large library); a background/scheduled sync also runs periodically.""" - return plex_sync.sync(db) + return {"ok": True, **plex_sync.sync(db)} def _link_dict(link: PlexLink | None) -> dict: @@ -129,7 +137,7 @@ def watch_link_set( admin token, so no separate Plex login. Enabling it for the first time runs the one-time "Plex is master" import immediately so the owner's existing Plex history shows up in Siftlode.""" if not sysconfig.get_bool(db, "plex_enabled"): - raise HTTPException(status_code=409, detail="The Plex module is disabled.") + raise refused(409, "plex_disabled") enabled = body.enabled link = db.execute(select(PlexLink).where(PlexLink.user_id == user.id)).scalars().first() if link is None: @@ -141,7 +149,7 @@ def watch_link_set( result: dict = {} if enabled and not link.initial_import_done: result = plex_watch.import_owner_watch_state(db, link) - return {**_link_dict(link), "import": result or None} + return {"ok": True, **_link_dict(link), "import": result or None} @router.post("/watch/import") @@ -150,6 +158,6 @@ def watch_import_now(user: User = Depends(admin_user), db: Session = Depends(get intersection; Siftlode-only states are preserved. Idempotent.""" link = db.execute(select(PlexLink).where(PlexLink.user_id == user.id)).scalars().first() if link is None or not link.sync_enabled: - raise HTTPException(status_code=409, detail="Enable Plex watch-sync first.") + raise refused(409, "watch_sync_off") result = plex_watch.import_owner_watch_state(db, link) - return {**_link_dict(link), "import": result} + return {"ok": True, **_link_dict(link), "import": result} diff --git a/backend/app/routes/plex/collections.py b/backend/app/routes/plex/collections.py index ada9eb92..f22c9288 100644 --- a/backend/app/routes/plex/collections.py +++ b/backend/app/routes/plex/collections.py @@ -9,6 +9,7 @@ from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.orm import Session +from app import logs from app.auth import admin_user, current_user from app.db import get_db from app.models import ( @@ -26,6 +27,10 @@ from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.plex.lookups import by_rating_key from app.plex.query import content_libs, runtime_by_key, top_art_keys from app.utils import like_escape +from app.refusal import refused +from app.routes._common import not_found + +log = logs.get_logger(__name__) # The most-restrictive → least ordering used to pick a collection's effective content rating from its # members (Plex shows one cert on the collection; we don't store it, so derive it). @@ -234,7 +239,7 @@ def collection_detail( endpoints below.""" col = by_rating_key(db, PlexCollection, rating_key) if col is None: - raise HTTPException(status_code=404, detail="Unknown collection") + raise not_found("collection") movies = ( db.execute( select(PlexItem) @@ -325,9 +330,9 @@ def collection_detail( def _editable_collection_or_403(db: Session, rating_key: str) -> PlexCollection: col = by_rating_key(db, PlexCollection, rating_key) if col is None: - raise HTTPException(status_code=404, detail="Unknown collection") + raise not_found("collection") if not _collection_editable(col): - raise HTTPException(status_code=403, detail="This collection is read-only") + raise refused(403, "collection_read_only") return col @@ -362,14 +367,13 @@ def _write_or_pending_error(wrote: bool, e: Exception) -> HTTPException: Both use 422, NOT a 5xx: the frontend treats 502/503/504 as "gateway unreachable" and raises a sticky offline banner (see api/core.ts RETRIABLE_GATEWAY). A 422 surfaces the detail as an ordinary action-error toast, which is what both of these are — the app answered, it didn't go unreachable.""" - if wrote: - detail = ( - "Collection updated in Plex, but refreshing the local copy failed; " - f"it will reconcile on the next sync ({e})" - ) - else: - detail = f"Plex write failed: {e}" - return HTTPException(status_code=422, detail=detail) + log.warning( + "a Plex collection write did not settle", + extra=logs.event( + "plex.collections.write_failed", wrote=wrote, reason=type(e).__name__, error=str(e) + ), + ) + return refused(422, "collection_pending_sync") if wrote else refused(422, "plex_write_failed") class CreateCollectionIn(BaseModel): @@ -389,7 +393,7 @@ def create_collection( title = body.title.strip() seed = body.item_rating_key if not title or not seed: - raise HTTPException(status_code=400, detail="title and item_rating_key are required") + raise refused(400, "collection_fields_required") if library: lib = ( db.execute(select(PlexLibrary).where(PlexLibrary.plex_key == library)).scalars().first() @@ -399,7 +403,7 @@ def create_collection( seed_item = by_rating_key(db, PlexItem, seed) or by_rating_key(db, PlexShow, seed) lib = db.get(PlexLibrary, seed_item.library_id) if seed_item else None if lib is None: - raise HTTPException(status_code=404, detail="Unknown library") + raise not_found("library") _reject_duplicate_name(db, lib.id, title) wrote = False try: @@ -473,7 +477,7 @@ def collection_rename( col = _editable_collection_or_403(db, rating_key) title = body.title.strip() if not title: - raise HTTPException(status_code=400, detail="title is required") + raise refused(400, "title_required") _reject_duplicate_name(db, col.library_id, title, exclude_rk=rating_key) lib = db.get(PlexLibrary, col.library_id) wrote = False @@ -502,7 +506,7 @@ def collection_delete( db.commit() except (PlexError, PlexNotConfigured) as e: db.rollback() - raise HTTPException(status_code=422, detail=f"Plex write failed: {e}") + raise _write_or_pending_error(False, e) from None return {"deleted": rating_key} @@ -521,13 +525,10 @@ def collection_set_editable( (IMDb/TMDb/…) can never be made editable — they're regenerated outside Plex.""" col = by_rating_key(db, PlexCollection, rating_key) if col is None: - raise HTTPException(status_code=404, detail="Unknown collection") + raise not_found("collection") want = body.editable if want and (col.smart or _collection_source(col) != "collection"): - raise HTTPException( - status_code=400, - detail="Smart / auto-list collections can't be made editable", - ) + raise refused(400, "collection_smart") col.editable = want db.commit() return _collection_card(col) diff --git a/backend/app/routes/plex/detail.py b/backend/app/routes/plex/detail.py index 8a52df0e..f621ba6b 100644 --- a/backend/app/routes/plex/detail.py +++ b/backend/app/routes/plex/detail.py @@ -5,7 +5,7 @@ import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from sqlalchemy import func, select from sqlalchemy.orm import Session @@ -30,6 +30,7 @@ from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.plex.lookups import by_rating_key from app.routes.plex._shared import _norm_lang, item_or_404 from app.routes.plex.collections import _collection_source +from app.routes._common import not_found log = logs.get_logger(__name__) @@ -109,7 +110,7 @@ def show_detail( hu = _norm_lang(lang) == "hu" sh = by_rating_key(db, PlexShow, rating_key) if sh is None: - raise HTTPException(status_code=404, detail="Unknown Plex show") + raise not_found("show") seasons = ( db.execute( select(PlexSeason) diff --git a/backend/app/routes/plex/discover.py b/backend/app/routes/plex/discover.py index 3bf13054..6a6283a6 100644 --- a/backend/app/routes/plex/discover.py +++ b/backend/app/routes/plex/discover.py @@ -32,12 +32,12 @@ from app.plex.query import ( meta_filter_conds, top_art_keys, ) +from app.discover.queries import RAIL_LIMIT from app.plex.tmdb import resolve_recs from app.routes.plex._shared import _norm_lang router = APIRouter(prefix="/api/plex", tags=["plex"]) -RAIL_LIMIT = 20 GENRE_TILES = 12 SURPRISE_PER_KIND = 3 # Surprise-me: at most this many movies + this many shows (6 total) diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index d08b81ca..635c2b47 100644 --- a/backend/app/routes/plex/playback.py +++ b/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 @@ -38,6 +39,8 @@ from app.plex import stream as plex_stream from app.plex import watch_sync as plex_watch from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.routes.plex._shared import item_or_404 +from app.refusal import refused +from app.routes._common import not_found router = APIRouter(prefix="/api/plex", tags=["plex"]) @@ -48,7 +51,7 @@ def _require_session(request: Request) -> int: low-sensitivity; a valid session is enough.""" uid, _ = resolved_user_id(request) if not uid: - raise HTTPException(status_code=401, detail="Not authenticated") + raise refused(401, "not_authenticated") return uid @@ -77,15 +80,17 @@ def image(rating_key: str, request: Request, variant: str = "thumb") -> Response with SessionLocal() as db: key = imgcache.key_for(db, str(rating_key), v) if not key: - raise HTTPException(status_code=404, detail="No image") + raise not_found("image") plex = PlexClient(db) # reads base URL + token now; the fetch below needs no DB - except PlexNotConfigured as e: - raise HTTPException(status_code=400, detail=str(e)) + except PlexNotConfigured: + raise refused(400, "plex_not_configured") from None try: with plex: data, ct = plex.image_bytes(key, w, h) - except PlexError as e: - raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}") + # silent-ok: a byte proxy for an ; the browser shows the placeholder and no reader sees the + # detail — the 502 is the whole answer, as for the person-image proxy below. + except PlexError: + raise refused(502, "image_fetch_failed") from None imgcache.write(cache_key, data, ct) return imgcache.response(data, ct) @@ -120,7 +125,7 @@ def subtitle_status( `at` is the viewer's absolute position: the extractor windows around it, so what gets read first is the part actually about to be watched.""" - return subtitles.ensure(rating_key, sub_ord, at) + return subtitles.ensure(rating_key, sub_ord, at, viewer=user.id) @router.get("/subtitle/{rating_key}/{sub_ord}") @@ -213,7 +218,7 @@ def person_image(u: str, request: Request) -> Response: Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch.""" _require_session(request) if not u.startswith(plex_people.PERSON_IMG_HOST): - raise HTTPException(status_code=400, detail="Unsupported image host") + raise refused(400, "image_host") cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest() hit = imgcache.read(cache_key) if hit: @@ -223,7 +228,7 @@ def person_image(u: str, request: Request) -> Response: r = c.get(u) r.raise_for_status() except httpx.HTTPError: - raise HTTPException(status_code=502, detail="Image fetch failed") + raise refused(502, "image_fetch_failed") ct = r.headers.get("content-type", "image/jpeg") imgcache.write(cache_key, r.content, ct) return imgcache.response(r.content, ct) @@ -295,6 +300,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. @@ -330,11 +366,27 @@ def item_progress( # finish latch is up, or before the first position is known. Those are the states in which # reaping is the right answer anyway, and a viewer who scrubs back into a reaped session gets # the same 404 → reclaim the player already handles. - plex_stream.touch_session(str(rating_key)) + plex_stream.touch_session(user.id, str(rating_key)) position = max(0, int(body.position_seconds)) 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 {"ok": True, "status": "watched", "position_seconds": 0} + st = ( db.execute( select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) @@ -342,28 +394,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": @@ -372,7 +402,7 @@ def item_progress( st.position_seconds = 0 st.progress_updated_at = now db.commit() - return {"position_seconds": 0} + return {"ok": True, "position_seconds": 0} if st is None: st = PlexState(user_id=user.id, item_id=it.id) @@ -397,7 +427,7 @@ def item_progress( duration * 1000, intent=now, ) - return {"position_seconds": position} + return {"ok": True, "position_seconds": position} class ItemStateIn(BaseModel): @@ -418,7 +448,7 @@ def item_state( it = item_or_404(db, rating_key) status = body.status if status not in ("new", "watched", "hidden"): - raise HTTPException(status_code=400, detail="Invalid status") + raise refused(400, "invalid_status") st = ( db.execute( select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) @@ -443,7 +473,7 @@ def item_state( db.commit() if mirrored_on_plex: _push_watch(background, db, user.id, it.id, rating_key, "unwatched") - return {"status": "new"} + return {"ok": True, "status": "new"} if st is None: st = PlexState(user_id=user.id, item_id=it.id) db.add(st) @@ -455,7 +485,7 @@ def item_state( db.commit() if status == "watched" and prev_status != "watched": _push_watch(background, db, user.id, it.id, rating_key, "watched") - return {"status": status} + return {"ok": True, "status": status} def _bulk_state( @@ -497,7 +527,9 @@ def _bulk_state( db.delete(st) if prev != "new": to_push.append((e.id, e.rating_key, "unwatched")) - db.commit() + # Flush, not commit: the handler commits, and it does so before the response — which is when + # the background push below runs, so the push always reads the committed marks. + db.flush() if pushable and to_push: # One coalesced background task for the whole batch (reuses a single session + Plex client), # not one task per episode — a 200-ep show mark is 1 task, not 200. @@ -520,7 +552,7 @@ def show_state( """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex.""" sh = by_rating_key(db, PlexShow, rating_key) if sh is None: - raise HTTPException(status_code=404, detail="Unknown Plex show") + raise not_found("show") watched = body.watched eps = ( db.execute(select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode")) @@ -528,7 +560,8 @@ def show_state( .all() ) changed = _bulk_state(background, db, user.id, eps, watched) - return {"changed": changed, "watched": watched} + db.commit() + return {"ok": True, "changed": changed, "watched": watched} @router.post("/season/{rating_key}/state") @@ -542,7 +575,7 @@ def season_state( """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex.""" se = by_rating_key(db, PlexSeason, rating_key) if se is None: - raise HTTPException(status_code=404, detail="Unknown Plex season") + raise not_found("season") watched = body.watched eps = ( db.execute(select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode")) @@ -550,7 +583,8 @@ def season_state( .all() ) changed = _bulk_state(background, db, user.id, eps, watched) - return {"changed": changed, "watched": watched} + db.commit() + return {"ok": True, "changed": changed, "watched": watched} @router.post("/stream/{rating_key}/session") @@ -575,21 +609,23 @@ def stream_session( # the client can switch audio without a restart — even for otherwise direct-playable files. if it.playable == "direct" and audio is None and not aoff and not multi: return { + "ok": True, "mode": "direct", "url": f"/api/plex/stream/{it.rating_key}/file", "start_s": 0.0, } if plex_paths.local_media_path(db, it.file_path) is None: - raise HTTPException(status_code=404, detail="Media file not found on disk") - s = plex_stream.start_session(db, it, start, audio, aoff, multi, transcode=transcode) + raise refused(404, "media_missing") + s = plex_stream.start_session(db, it, user.id, start, audio, aoff, multi, transcode=transcode) if s is None: - raise HTTPException(status_code=404, detail="Media file not found on disk") + raise refused(404, "media_missing") # Report the REAL media start (the keyframe ffmpeg landed on, which hls.js zero-bases to), not the # requested offset — the client's absolute clock and the subtitle cue-shift both key off this. # `transcode`/`hw`/`tonemap` let the player show an honest cold-start ("Preparing stream…") and # surface the active mode + encoder in the stats overlay (`tonemap` = the HW HDR→SDR path); both # segment streams are otherwise identical. return { + "ok": True, "mode": "hls", "start_s": s.media_start_s, # The session id is part of the path (not a query param) so the playlist's RELATIVE segment @@ -614,7 +650,7 @@ def stream_file( it = item_or_404(db, rating_key) src = plex_paths.local_media_path(db, it.file_path) if src is None: - raise HTTPException(status_code=404, detail="Media file not found on disk") + raise refused(404, "media_missing") if download: # A clean, safe download name from the title + the real file extension. ext = src.suffix or (f".{it.container}" if it.container else "") @@ -651,9 +687,9 @@ def stream_hls_file( references its segments/subtitles by relative name, all resolving back here. The session id in the path is what makes "back here" mean the same session. Sessions are keyed by - rating_key alone and a seek-restart or a second viewer of the same item replaces the registered - one, while the DIRECTORY is per (key, offset, tag) — so resolving by rating_key handed the old - viewer the new session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong + (viewer, rating_key) and a seek-restart replaces the viewer's registered one, while the DIRECTORY + is per (key, viewer, offset, tag) — so resolving by the key alone handed the old player the new + session's `seg_12.ts`, produced from a different offset under `-copyts`: wrong bytes with a 200, which the player has no way to notice (deep-review round 5). A request for a session that is no longer the registered one is a 404, and the player reclaims at the playhead. @@ -668,10 +704,10 @@ def stream_hls_file( or "\\" in filename or ".." in filename ): - raise HTTPException(status_code=404, detail="Not found") - s = plex_stream.current_session(str(rating_key)) + raise not_found("file") + s = plex_stream.current_session(user.id, str(rating_key)) if s is None or s.id != session_id: - raise HTTPException(status_code=404, detail="No active playback session") + raise refused(404, "session_gone") path = s.dir / filename # Only a name THIS session can actually produce is worth waiting for: its own playlists, and a # segment of one of its own renditions whose predecessor is already on disk (ffmpeg writes them @@ -681,14 +717,14 @@ def stream_hls_file( # past the produced region — or a deliberate flood of hopeless names — from parking threadpool # workers and their DB connections (see `artifact_may_arrive`). if not plex_stream.artifact_may_arrive(s, filename): - raise HTTPException(status_code=404, detail="Not available yet") + raise refused(404, "not_ready") timeout = 30 if ext == ".m3u8" else 20 # `proc` so a session whose ffmpeg has already exited answers at once instead of polling out the # whole timeout: this is a SYNC route on the ~40-slot threadpool, holding a pooled DB connection # with it, so every such wait is a worker (and a connection) parked for 20 s — an ordinary seek # past the produced region was enough to do it (round 5). if not plex_stream.wait_for(path, timeout, proc=s.proc): - raise HTTPException(status_code=404, detail="Not available yet") + raise refused(404, "not_ready") try: content = path.read_bytes() except OSError: @@ -699,7 +735,7 @@ def stream_hls_file( # restarting at the playhead. Ordinary trigger: `loadSession` awaits the new session — which # kills and rmtrees the old one — before destroying the old hls instance, so the outgoing # instance's in-flight segment request races the rmtree on every seek-restart. - raise HTTPException(status_code=404, detail="No active playback session") from None + raise refused(404, "session_gone") from None return Response( content=content, media_type=_HLS_CT[ext], diff --git a/backend/app/routes/plex/playlists.py b/backend/app/routes/plex/playlists.py index 022af3ae..8786584d 100644 --- a/backend/app/routes/plex/playlists.py +++ b/backend/app/routes/plex/playlists.py @@ -2,7 +2,7 @@ from datetime import datetime, timezone -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy import delete, func, select, update from sqlalchemy.orm import Session @@ -19,15 +19,12 @@ from app.models import ( ) from app.plex.cards import leaf_card from app.plex.lookups import by_rating_key -from app.routes._common import owned_or_404 +from app.routes._common import owned_or_404, not_found +from app.refusal import refused router = APIRouter(prefix="/api/plex", tags=["plex"]) -def _own_playlist_or_404(db: Session, pid: int, user: User) -> PlexPlaylist: - return owned_or_404(db, PlexPlaylist, pid, user, what="playlist") - - def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict: return { "id": pl.id, @@ -120,7 +117,7 @@ def create_playlist( ) -> dict: title = body.title.strip() if not title: - raise HTTPException(status_code=400, detail="title is required") + raise refused(400, "title_required") pl = PlexPlaylist(user_id=user.id, title=title) db.add(pl) db.flush() @@ -140,7 +137,7 @@ def create_playlist( def playlist_detail( pid: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") rows = db.execute(_playlist_item_query(pid)).all() items = [it for (_pi, it) in rows] states = { @@ -183,10 +180,10 @@ def rename_playlist( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") title = body.title.strip() if not title: - raise HTTPException(status_code=400, detail="title is required") + raise refused(400, "title_required") pl.title = title db.commit() count = ( @@ -203,7 +200,7 @@ def rename_playlist( def delete_playlist( pid: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") db.delete(pl) db.commit() return {"deleted": pid} @@ -220,10 +217,10 @@ def playlist_add_item( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") it = by_rating_key(db, PlexItem, body.item_rating_key) if it is None: - raise HTTPException(status_code=404, detail="Unknown item") + raise not_found("item") exists = ( db.execute( select(PlexPlaylistItem).where( @@ -252,7 +249,7 @@ def playlist_remove_item( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") it = by_rating_key(db, PlexItem, item_rating_key) if it is not None: db.execute( @@ -297,7 +294,7 @@ def playlist_add_bulk( ) -> dict: """Append many items at once (a whole season/show). Already-present items are skipped; the input order is preserved for the newly-added ones. Returns how many were added + the new total.""" - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") items = _items_by_rks(db, body.item_rating_keys) existing = set( db.execute( @@ -328,7 +325,7 @@ def playlist_add_bulk( ) or 0 ) - return {"added": added, "item_count": total} + return {"ok": True, "added": added, "item_count": total} @router.post("/playlists/{pid}/items/remove-bulk") @@ -340,7 +337,7 @@ def playlist_remove_bulk( ) -> dict: """Remove many items at once (a whole season/show, or a multi-select). Returns how many were removed + the new total.""" - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") ids = [it.id for it in _items_by_rks(db, body.item_rating_keys)] removed = 0 if ids: @@ -359,7 +356,7 @@ def playlist_remove_bulk( ) or 0 ) - return {"removed": removed, "item_count": total} + return {"ok": True, "removed": removed, "item_count": total} @router.put("/playlists/{pid}/order") @@ -370,7 +367,7 @@ def reorder_playlist( db: Session = Depends(get_db), ) -> dict: """Set the playlist order from an explicit list of item rating_keys (0-based positions).""" - pl = _own_playlist_or_404(db, pid, user) + pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") order = body.item_rating_keys id_by_rk = { it.rating_key: it.id diff --git a/backend/app/routes/public.py b/backend/app/routes/public.py index e580a77c..fa2a2107 100644 --- a/backend/app/routes/public.py +++ b/backend/app/routes/public.py @@ -6,7 +6,7 @@ the app or any account data. Password-protected links exchange the password once a short-lived signed grant that the media URL carries (a `