Compare commits

..
12 Commits
Author SHA1 Message Date
peter 909732b3c1 release: 0.108.1 (improvement/e42-backend-units) 2026-10-20 02:12:49 +02:00
peter 215f022fe3 chore(release): 0.108.1 — the backend in four units, each owning its tables; a first Watch later and a first resume point survive a write beside them 2026-10-20 02:04:56 +02:00
peter 6ea750a698 test: the commit check starts every request clean, and the fence docs claim what it holds
The test's own writes are committed before a request, so a handler's Core DML can no longer hide
in a transaction the test had already opened. The entry-point probe answers on a marked line the
API's logging cannot mix into. A listed commit whose function no longer commits fails the fence.
CLAUDE.md and the backend README now name what an AST cannot see; the route-commit tests no longer
claim their read-back holds the commit.
2026-10-20 01:56:31 +02:00
peter c5f121139b fix: every first watch state survives a write beside it, and a folded Watch later keeps its pin
A season marked watched, a pull from Plex and a YouTube resume point made their first row by a read
then an add, and failed on the unique key when a first checkpoint or the watched mark made it in
between. They go through the same insert-then-read as the checkpoint (the resume point is one
upsert); a pull that meets a local change still waiting for its push leaves it.

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

Marking one notice read trimmed before its own flag was flushed, so the trim left one past the cap.
The route test now crosses the cap and reads the result from a session of its own.
2026-10-20 01:56:30 +02:00
peter b04b2cbd62 test: the rewritten write handlers commit, and three more owner filters hold
Each handler that now calls a service and commits is driven through HTTP once.
A second account's watch state, Watch later list and private download profile
are never read or changed. The autotag job's orphan prune is seen committed.
2026-10-20 00:45:22 +02:00
peter 3a5212cd8a fix: a first watch state and a first Watch later survive two requests at once
The first checkpoint of an item and the final flush beside it both read no
state and both inserted; the loser failed on uq_plex_user_item. Two first saves
each made a Watch later list. Both are created by an ON CONFLICT DO NOTHING
insert and then read; a partial unique index holds one Watch later per user, and
the migration folds an existing second list into the first. Plex keeps one
per-item state map (rollup.item_state_map).
2026-10-20 00:45:22 +02:00
peter b376373af5 fix(files): a folder's kind is answered before its new name, as before
A blank name on a source folder answered 400 name_required; it answers 409
not_own again.
2026-10-20 00:45:22 +02:00
peter 8206384c91 test(fence): owner filters in any form, statement writes under any name, core commits by list
The owner-filter check missed an aliased() model, an import alias, .in_() and
filter_by; four live inline filters on aliased VideoState go through owned_by now.
Cross-unit writes are seen under any import name, through sa.update and via
__table__. Core modules may commit only where CORE_COMMITS names the reason;
trim_read and audit.gc flush now and their callers commit. The pin target, the
debug-session header and a link's target fetch through owned_row. The docs
claim what the fence holds, and say what is beyond it.
2026-10-20 00:45:21 +02:00
peter e27656ee7e test: the request commit check sees Core DML and row locks
A db.execute(update/delete) fires no flush and stages nothing, so a handler whose
only write was Core DML could forget its commit and stay green. The check now
asks Postgres whether the request opened a writing transaction. It found one
real case: a repeated issue send returned with the session row still locked.
2026-10-20 00:45:20 +02:00
peter d73cdcee97 fix(capture): every process registers every table before its first query
The models now live in their units' packages, and a core model names a unit's by
string (User.subscriptions -> Subscription). The download child imports neither
catalog nor app.tables before db.get, so mapper configuration failed and every
download, clip, edit and upload job died with child_exit. Each entry point now
imports app.tables, and a test imports each one in a clean interpreter.
2026-10-20 00:45:20 +02:00
peter 5ce292cd28 test(fence): the unit fence has no exemptions, and the owner filter is held in every module
The per-file pin list and the unit-by-unit switch go: every route file is held
to zero database work, every unit's package to its edges, and core code says
whose a row is through userscope too. CLAUDE.md, the backend README and the
gate register describe the units as they now are.
2026-10-19 23:36:01 +02:00
peter ad6ccbac05 refactor(core): the core's owner filters go through userscope, each held by a second account
Tokens, notifications, quota, debug sessions and the reset-link burn filtered
their owner inline. Ten of the fifteen filters had no test that noticed one
going missing; each now has a test where another account's rows must not
count, show or change.
2026-10-19 23:36:01 +02:00
51 changed files with 1511 additions and 239 deletions
+11
View File
@@ -7,6 +7,17 @@ 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.108.1 — 2026-10-20
### New
- Development: the backend is split into four units (downloads, reader, catalog, Plex), each owning its tables. A contract test holds what each one reads and writes, and that every query names whose rows it asks for.
### Fixed
- Watch later: saving the first two videos at once no longer makes two lists. An existing second list is merged into the first, its videos and its comfort-shelf pin with it.
- Playback: the first resume point of a video or a Plex item no longer fails when it is saved together with the watched mark, a whole season marked at once, or a Plex sync.
## 0.108.0 — 2026-10-18
### New
+37 -30
View File
@@ -66,6 +66,13 @@ how it gets built. Concretely:
`downloads/`, `reader/`, `discover/`, `catalog/`, `youtube/`, `issues/` (shared by routes + scheduler + worker) → flat core (`auth`, `apitokens`, `quota`, `jobruns`,
`sysconfig`, `state`, `audit`, `notifications`, `userscope`, `ratelimit`, `bodylimit`, `progress`, `collation`,
`security`, `logs`, `logsink`, `diagnostics`, `debugsession`) → `models/db/config`. **Dependency direction is one-way: core never imports routes.**
The domain is four **units**, each owning its tables: Capture (`downloads/`), Reader (`reader/`, `discover/`, the playlist
sync), Catalog (`catalog/`, the `sync/` ingests) and Plex (`plex/`). A unit's route does no database work: it reads through
the unit's `queries`, writes through its `service` (which flushes) and commits. A unit's code reads another unit's tables
only along a declared edge and writes them only through that unit's service; core reaches a unit's tables through its
`queries`. **Fence-held, with no exemptions** (`backend/tests/test_unit_fences.py`) — as far as an AST sees: what it
cannot (an ORM write on another unit's loaded row, SQL in `text()`, a commit by `begin()`) is review's to catch, listed in
[`backend/app/README.md`](backend/app/README.md) with the conventions.
Frontend: `main.tsx` provider tree → `App.tsx` shell (a shared `/watch` link boots from its own
entry, `watch.html` → `watchMain.tsx`, held app-free by a contract test) → `moduleRegistry` route-chunks → module
components → the **platform layer `components/ui/`** (domain-free primitives: the popover engine and
@@ -82,36 +89,36 @@ CQRS, a repository/DI framework, a router library, a generic CRUD-router factory
## Canonical patterns — when you have a choice, pick THIS
| Area | Use | Not |
| ------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------- |
| ORM | SQLAlchemy 2.0 `select()` | `db.query(...)` |
| Request body | a Pydantic model | `payload: dict` |
| Request SIZE | nothing — `bodylimit` caps every route (SEC-20); need more? add a ceiling to `ROUTE_MAX_BODY_BYTES` | `capped_body` on a route that does not read raw bytes itself |
| Ownership check | `owned_or_404` / `get_or_404` | ad-hoc `_own_playlist`/`_owned`/`_own_job` |
| 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}`), test-held | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok` |
| Transaction commit | the handler commits; helpers `flush` (test-held list) | helpers that commit (only `quota` and `jobruns` own their session) |
| Admin knob storage | `sysconfig` (values) | env / `app_state` / a table of its own (the job intervals had one) |
| Job status word | the `downloads/status.py` StrEnum | loose `"canceled"`/`"cancelled"` literals |
| Current time | inline `datetime.now(timezone.utc)` | new `now_utc`/`_now` helpers |
| FE per-account state | `usePref` | raw `readAccount`/`writeAccount` or bespoke hooks |
| 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 `<body>` |
| FE static label / button-like link | `Badge` (flat, no edge) · `LinkButton` / `IconLink` | a glass pill that reads as a chip; an `<a>` hand-painted like a button |
| FE loading/empty/error | `QueryState` / `StateMessage` | ad-hoc conditionals |
| FE failed write | `meta.failTitle`, else `notifyActionError(err, title)`; a YouTube write action (subscribe, sync, …): `notifyYouTubeActionError` — a 403 offers Connect | a bare `catch {}` or `notify()` with no error attached |
| FE floating or full-screen layer | `ui/Overlay` (portals to `<body>`) or the popover engine (`AnchoredPopover` portals itself); stacking by the z-index scale (`z-menu` … `z-popover`, docs/DESIGN.md "Layers") | one rendered in place inside a virtualized row or a masked/transformed scroller — clipped or painted over; a magic `z-[n]` |
| FE player on a native media element | `lib/mediaSession` (`useMediaSession`) + `lib/mediaKeys` + `lib/resume` (`useResume`), the element held as STATE | a private `navigator.mediaSession` / `keydown` / position-saving effect per player |
| FE diagnostics | `lib/log.ts` (`log.info("event.code", msg, attrs)`); a filter change → `lib/filterLog.ts`, a player → `lib/mediaLog.ts` | `console.*` (lint-enforced); a value only a higher level may carry, logged unguarded |
| FE state that MIRRORS a prop/query | derive it, or hold a draft (`draft ?? server`) | a copy an effect re-syncs — that paints the stale value first |
| FE state reset on a key change | adjust it DURING render against a remembered key | `useEffect(() => setX(init), [key])` |
| FE effect that must not re-subscribe | `useEffectEvent` (React 19.2+); a handler or timer: `useLatest` / `useStableCallback` | a render-phase `ref.current = x` — the React Compiler skips the component |
| FE hook suppression | an inline `eslint-disable-next-line` NAMING the invariant | `eslint-suppressions.json` (deliberately absent — E27 S2) |
| FE popover / scroll-fade ref | destructure it: `const { setReference } = pop.refs`, then `ref={setReference}` | `ref={pop.refs.setReference}` / `ref={fade.ref}` — the compiler then treats the whole object as a ref and skips |
| FE async handler with cleanup | `try { … } catch { … }`, then the cleanup after it (steps that `return` early go in their own function); a conditional, loop or `?.` the `try` needs goes before it or into a module-level helper | `try/finally`, or a `?:` / `??` / loop / `?.` inside `try` — the React Compiler skips the whole component (and `react-hooks/todo` misses most of these) |
| Area | Use | Not |
| ------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| ORM | SQLAlchemy 2.0 `select()` | `db.query(...)` |
| Request body | a Pydantic model | `payload: dict` |
| Request SIZE | nothing — `bodylimit` caps every route (SEC-20); need more? add a ceiling to `ROUTE_MAX_BODY_BYTES` | `capped_body` on a route that does not read raw bytes itself |
| Ownership check | `userscope.owned_by` (a filter) / `owned_row` (a row by id): a unit's `queries.own_x` returns None and its route answers `not_found`; a core route `owned_or_404`. A composite rule (own or shared, built-in or own) says so in its `queries` | an inline owner filter — `Model.user_id == …`, an alias's, `.in_()`, `filter_by(user_id=…)` (fence-held, every module) — or a `db.get` and a compare where `owned_row` fits |
| 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}`), test-held | a bare `{"unsubscribed"}`/`{"cleared": n}`, no `ok` |
| Transaction commit | the handler commits; a unit's `service` and every helper `flush` (fence-held, every module) | a helper that commits — only a listed one (what owns its session, or a core step that must land whatever its caller does next), each with its reason: the fence's `UNIT_COMMITS` / `CORE_COMMITS`, a route's `COMMITTING_HELPERS` (`test_route_conventions`) |
| Admin knob storage | `sysconfig` (values) | env / `app_state` / a table of its own (the job intervals had one) |
| Job status word | the `downloads/status.py` StrEnum | loose `"canceled"`/`"cancelled"` literals |
| Current time | inline `datetime.now(timezone.utc)` | new `now_utc`/`_now` helpers |
| FE per-account state | `usePref` | raw `readAccount`/`writeAccount` or bespoke hooks |
| 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 `<body>` |
| FE static label / button-like link | `Badge` (flat, no edge) · `LinkButton` / `IconLink` | a glass pill that reads as a chip; an `<a>` hand-painted like a button |
| FE loading/empty/error | `QueryState` / `StateMessage` | ad-hoc conditionals |
| FE failed write | `meta.failTitle`, else `notifyActionError(err, title)`; a YouTube write action (subscribe, sync, …): `notifyYouTubeActionError` — a 403 offers Connect | a bare `catch {}` or `notify()` with no error attached |
| FE floating or full-screen layer | `ui/Overlay` (portals to `<body>`) or the popover engine (`AnchoredPopover` portals itself); stacking by the z-index scale (`z-menu` … `z-popover`, docs/DESIGN.md "Layers") | one rendered in place inside a virtualized row or a masked/transformed scroller — clipped or painted over; a magic `z-[n]` |
| FE player on a native media element | `lib/mediaSession` (`useMediaSession`) + `lib/mediaKeys` + `lib/resume` (`useResume`), the element held as STATE | a private `navigator.mediaSession` / `keydown` / position-saving effect per player |
| FE diagnostics | `lib/log.ts` (`log.info("event.code", msg, attrs)`); a filter change → `lib/filterLog.ts`, a player → `lib/mediaLog.ts` | `console.*` (lint-enforced); a value only a higher level may carry, logged unguarded |
| FE state that MIRRORS a prop/query | derive it, or hold a draft (`draft ?? server`) | a copy an effect re-syncs — that paints the stale value first |
| FE state reset on a key change | adjust it DURING render against a remembered key | `useEffect(() => setX(init), [key])` |
| FE effect that must not re-subscribe | `useEffectEvent` (React 19.2+); a handler or timer: `useLatest` / `useStableCallback` | a render-phase `ref.current = x` — the React Compiler skips the component |
| FE hook suppression | an inline `eslint-disable-next-line` NAMING the invariant | `eslint-suppressions.json` (deliberately absent — E27 S2) |
| FE popover / scroll-fade ref | destructure it: `const { setReference } = pop.refs`, then `ref={setReference}` | `ref={pop.refs.setReference}` / `ref={fade.ref}` — the compiler then treats the whole object as a ref and skips |
| FE async handler with cleanup | `try { … } catch { … }`, then the cleanup after it (steps that `return` early go in their own function); a conditional, loop or `?.` the `try` needs goes before it or into a module-level helper | `try/finally`, or a `?:` / `??` / loop / `?.` inside `try` — the React Compiler skips the whole component (and `react-hooks/todo` misses most of these) |
Not repeated here, because a lint rule refuses the wrong one and its message names the right one:
buttons (`ui/Button`, `IconButton`, `Pressable`, `Chip`, `DragHandle`), text fields and native
+1 -1
View File
@@ -1 +1 @@
0.108.0
0.108.1
@@ -0,0 +1,70 @@
"""One Watch later list per user, held by the database
Revision ID: 0112_one_watch_later
Revises: 0111_audit_request_id
Create Date: 2026-10-20
Watch later was made on a user's first save by a read then an insert, so two first saves at once
each made a list, and the reader then picked between them at random. A partial unique index holds
"one per user" now. An instance that already has a second list keeps the oldest: the others' videos
move into it (after its own, in their order; one it already holds is not added twice), a Home pin
of one of them moves to it (unless it is pinned itself; of several, the first on the shelf), then
they go — a pin left on them goes with them, as the kept list already holds one.
"""
import sqlalchemy as sa
from alembic import op
revision: str = "0112_one_watch_later"
down_revision: str | None = "0111_audit_request_id"
branch_labels: str | None = None
depends_on: str | None = None
_EXTRA = """
SELECT p.id, k.keep_id
FROM playlists p
JOIN (
SELECT user_id, min(id) AS keep_id FROM playlists WHERE kind = 'watch_later' GROUP BY user_id
) k ON k.user_id = p.user_id
WHERE p.kind = 'watch_later' AND p.id <> k.keep_id
"""
def upgrade() -> None:
op.execute(
f"""
INSERT INTO playlist_items (playlist_id, video_id, position, added_at)
SELECT e.keep_id, i.video_id,
(SELECT coalesce(max(position), -1) FROM playlist_items WHERE playlist_id = e.keep_id)
+ row_number() OVER (PARTITION BY e.keep_id ORDER BY i.playlist_id, i.position, i.id),
i.added_at
FROM playlist_items i
JOIN ({_EXTRA}) e ON e.id = i.playlist_id
ON CONFLICT ON CONSTRAINT uq_playlist_video DO NOTHING
"""
)
op.execute(
f"""
UPDATE comfort_pins c SET playlist_id = m.keep_id
FROM (
SELECT DISTINCT ON (e.keep_id) p.id AS pin_id, e.keep_id
FROM comfort_pins p
JOIN ({_EXTRA}) e ON e.id = p.playlist_id
WHERE NOT EXISTS (SELECT 1 FROM comfort_pins k WHERE k.playlist_id = e.keep_id)
ORDER BY e.keep_id, p.position, p.id
) m
WHERE c.id = m.pin_id
"""
)
op.execute(f"DELETE FROM playlists WHERE id IN (SELECT id FROM ({_EXTRA}) e)")
op.create_index(
"uq_playlist_watch_later",
"playlists",
["user_id"],
unique=True,
postgresql_where=sa.text("kind = 'watch_later'"),
)
def downgrade() -> None:
op.drop_index("uq_playlist_watch_later", table_name="playlists")
+56 -16
View File
@@ -1,9 +1,8 @@
# Backend conventions
The rules below are already followed across `backend/app` — they lived only in a few module
docstrings (`userscope.py`, `routes/_common.py`) until this file collected them (E3 S3,
B-core-20). When you add code, match these so the next reader doesn't have to guess "which pattern do
I copy".
docstrings (`userscope.py`, `routes/_common.py`) until this file collected them. When you add code,
match these so the next reader doesn't have to guess "which pattern do I copy".
## Module layering
@@ -13,12 +12,13 @@ Three layers, by *who* runs the code — not by feature:
response models, status-code mapping, serialization. It owns logic that runs *only* on a request.
Business logic that a request happens to trigger does **not** live here if anything else runs it too.
2. **`plex/`, `downloads/`, `sync/`, `issues/` — domain logic shared by more than one caller.** Anything the
**scheduler or the worker** runs (not just a request) lives in one of these domain packages, so the
route, the worker, and the scheduler share one definition. Examples from this codebase:
`downloads/service.py` owns enqueue + the job state transitions (routes call it); `plex/sync.py` is
run by both the admin route and the scheduler; `downloads/pipeline.py` is the worker's child pipeline;
`issues/bundle.py` is the one definition of a bug report's files, for the zip export and the tracker.
2. **The units and the other domain packages — logic shared by more than one caller.** A unit's
database work (below), and anything the **scheduler or the worker** runs (not just a request), lives
in a domain package, so the route, the worker, and the scheduler share one definition. Examples from
this codebase: `downloads/service.py` owns enqueue + the job state transitions (routes call it);
`plex/sync.py` is run by both the admin route and the scheduler; `downloads/pipeline.py` is the
worker's child pipeline; `issues/bundle.py` is the one definition of a bug report's files, for the
zip export and the tracker.
3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth` (with `apitokens`, a machine's
folder-scoped way into the Files library), `sysconfig`, `quota`, `audit`, `jobruns` (every
@@ -34,19 +34,59 @@ Three layers, by *who* runs the code — not by feature:
a layer — into the relevant domain package, or a flat core module if it's cross-cutting. Don't
pre-abstract for a single caller; don't leave shared logic duplicated once there are two.
## Units
The domain is four units, each owning its tables (`models.py` in its package):
| Unit | Package | Its routes |
| --- | --- | --- |
| Capture | `downloads/`, the worker | `downloads.py`, `files.py`, `public.py` |
| Reader | `reader/`, `discover/`, `feedquery.py`, the playlist sync and tag suggestions in `sync/` | `feed.py`, `playlists.py`, `saved_views.py`, `discover.py`, `tags.py` |
| Catalog | `catalog/`, the ingests in `sync/` | `channels.py`, `search.py`, `sync.py` |
| Plex | `plex/` | `plex/` |
- **A unit's route does no database work.** It validates, reads through the unit's `queries.py`,
writes through its `service.py`, serializes, and commits. It may name a unit's model as a type
(under `TYPE_CHECKING`), never use one as a value, and it passes a kind (`"movie"`), not a model, to
a query.
- **A unit's code keeps to its edges.** It reads another unit's tables only along a declared edge
(Capture and Reader read Catalog's and Plex's: the feed joins `Video` with `VideoState` in one
query) and writes them only through the owner's service. Core code reaches a unit's tables through
that unit's `queries`.
- **Whose a row is, `userscope` says** — in every module, core included: `owned_by(Model.user_id,
user)` for a filter, `owned_row(db, Model, id, user)` for a row by id (someone else's row is the
same answer as a missing one). A unit's `queries.own_x` returns None and its route answers
`not_found`; a core route calls `owned_or_404`. A rule `owned_row` cannot say — a job its owner
shared, a built-in profile nobody owns, a folder read under `FOR SHARE` — is written out in the
unit's own code, beside the reason.
All of it is held by `tests/test_unit_fences.py`, with no exemptions: the unit map, each unit's
packages and edges (`UNIT_PACKAGES`, `READS`) and what may commit (`UNIT_COMMITS`, `CORE_COMMITS`)
are listed there. It reads the code's AST, so it holds the forms written here: an owner column
compared inline (a model's, an import alias's or an `aliased()` one's, by `==`, `!=`, `.in_()`, or a
`filter_by(user_id=…)`), a model built or an `insert`/`update`/`delete` (under any import name, of a
model or its `__table__`) across units, and a `.commit()` call off the lists all fail it. What an AST
cannot tell is review's to catch: an ORM write on another unit's loaded row (an attribute, a
`db.delete`), SQL in `text()`, a column or model reached another way (bound to a variable, through
`__table__.c` or a module path), and a commit made by a `begin()` block or a `commit` handed in.
## Transaction ownership
This follows CLAUDE.md's canonical rule verbatim — **"the handler/job-wrapper commits; helpers flush
(only `quota` and `jobruns` own their session)"** — so read it that way, not as an exception to it. The commit belongs
to the **operation owner**; anything called *inside* another transaction only stages its change
This follows CLAUDE.md's canonical rule — **the handler commits; a unit's `service` and every helper
flush** — so read it that way, not as an exception to it. The commit belongs to the **operation
owner**; anything called *inside* another transaction only stages its change
(`db.add` / `flush`) and lets the owner commit, so the whole operation is one transaction.
`audit.record()` is the reference: it adds the audit row and the caller commits it together with the
mutation it records, so an audit entry can never exist without its change (or vice-versa).
- The **operation owner** is usually the route handler. A `service.*` operation the handler delegates a
whole transition to (e.g. `service.cancel`) *is* that owner and commits its one transition — it isn't
a mid-operation helper. A helper that owns its OWN short-lived session (the worker's `jobdb` writes,
`jobruns`' run rows — the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes.
- The **operation owner** is the route handler, or a job. A unit's `service` flushes even when it
makes a whole transition (`service.cancel`): the handler commits it, together with whatever else the
request changed. What owns its OWN session commits it: `quota` and `jobruns`' run rows, the
scheduler's jobs, a unit's jobs (the worker's `jobdb` writes, the syncs and ingests that commit per
page or batch so a later failure loses nothing already stored), and a few core steps that must land
whatever their caller does next (a token's last-used stamp, a refreshed YouTube grant) — the fence's
`UNIT_COMMITS` and `CORE_COMMITS` name each with its reason, `test_route_conventions` a route's own
committing helpers. Everything else flushes.
- `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already
loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM
attribute — see `service.decr_ref_count`.
+4 -1
View File
@@ -26,6 +26,7 @@ from fastapi import Depends, Request
from sqlalchemy import delete, or_, select, update
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app import audit, debugsession, logs
from app.audit import AuditAction
from app.auth import current_user
@@ -223,7 +224,9 @@ def revoke_all(db: Session, user_id: int) -> int:
change, "log out everywhere", a revoked untrusted password — ends its tokens too: one minted
from a stolen session must not outlive the owner's answer to that theft. Returns how many."""
gone = (
db.execute(delete(ApiToken).where(ApiToken.user_id == user_id).returning(ApiToken.id))
db.execute(
delete(ApiToken).where(owned_by(ApiToken.user_id, user_id)).returning(ApiToken.id)
)
.scalars()
.all()
)
+1 -2
View File
@@ -110,10 +110,9 @@ def record(
def gc(db: Session, retention_days: int) -> int:
"""Delete audit rows older than `retention_days`. Returns the number removed. A
retention_days <= 0 disables pruning (keep forever)."""
retention_days <= 0 disables pruning (keep forever). Flushes; the job commits."""
if retention_days <= 0:
return 0
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
result = db.execute(delete(AuditLog).where(AuditLog.created_at < cutoff))
db.commit()
return result.rowcount or 0
+2 -1
View File
@@ -4,6 +4,7 @@ from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session
from starlette.requests import HTTPConnection
from app.userscope import owned_by
from app import debugsession
from app import email as email_mod
from app import logs
@@ -230,7 +231,7 @@ def purge_user(db: Session, user: User, background: BackgroundTasks) -> None:
# request's access line, a queued row, a later background task would otherwise put rows back
# under the id after this DELETE (logs.forget_user).
logs.forget_user(user_id)
db.execute(delete(LogEvent).where(LogEvent.user_id == user_id))
db.execute(delete(LogEvent).where(owned_by(LogEvent.user_id, user_id)))
# Every canonically-equal invite goes, not just the literal spelling: is_allowed matches on the
# canonical form, so leaving a sibling row behind would keep granting access to a mailbox whose
# account was just erased — and prod really does carry approved pairs for one mailbox.
+7 -8
View File
@@ -26,6 +26,7 @@ from uuid import uuid4
from sqlalchemy import and_, delete, func, not_, select, update
from sqlalchemy.orm import Session, defer
from app.userscope import owned_by, owned_row
from app import diagnostics
from app.models import DebugScreenshot, DebugSession, LogEvent, User
@@ -76,7 +77,9 @@ def open_session(db: Session, user: User) -> DebugSession | None:
def _unclosed(db: Session, user: User) -> DebugSession | None:
return db.scalar(
select(DebugSession).where(DebugSession.user_id == user.id, DebugSession.ended_at.is_(None))
select(DebugSession).where(
owned_by(DebugSession.user_id, user), DebugSession.ended_at.is_(None)
)
)
@@ -126,12 +129,8 @@ def from_header(db: Session, user: User, value: str | None) -> DebugSession | No
"""The session a request names, if it is the caller's own and still live; else None."""
if not value or len(value) != _ID_LEN or not all(c in "0123456789abcdef" for c in value):
return None
session = db.get(DebugSession, value)
if (
session is None
or session.user_id != user.id
or not is_live(session, datetime.now(timezone.utc))
):
session = owned_row(db, DebugSession, value, user)
if session is None or not is_live(session, datetime.now(timezone.utc)):
return None
return session
@@ -354,7 +353,7 @@ def stored_screenshot_bytes(db: Session, user_id: int) -> int:
db.scalar(
select(func.coalesce(func.sum(DebugScreenshot.size), 0))
.join(DebugSession, DebugSession.id == DebugScreenshot.session_id)
.where(DebugSession.user_id == user_id)
.where(owned_by(DebugSession.user_id, user_id))
)
or 0
)
+2 -2
View File
@@ -72,11 +72,11 @@ def resolve_ref(db: Session, user: User, kind: str, ref: str) -> tuple[str, obje
# such pin target", which is the same answer this function gives for one that does not exist.
except ValueError:
return None
pl = db.get(Playlist, pid)
# A playlist is per-user, unlike the other four (which are shared catalog rows whose PRIVACY
# lives in the state tables). Someone else's list must read as "no such thing", not as a 403 —
# the same reason `owned_or_404` 404s rather than 403s.
return (col, pl.id) if pl is not None and pl.user_id == user.id else None
pl = owned_row(db, Playlist, pid, user)
return (col, pl.id) if pl is not None else None
def reorder(db: Session, user: User, ids: list[int]) -> None:
+3 -3
View File
@@ -93,7 +93,7 @@ def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tup
q = (
select(*feed_columns(user, st), st.progress_updated_at.label("progressed_at"))
.join(Channel, Channel.id == Video.channel_id)
.join(st, and_(st.video_id == Video.id, st.user_id == user.id))
.join(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
.where(
st.position_seconds > 0,
st.status.notin_(("watched", "hidden")),
@@ -136,7 +136,7 @@ def yt_video_row(db: Session, user: User, video_id: str) -> dict | None:
row = db.execute(
select(*feed_columns(user, st))
.join(Channel, Channel.id == Video.channel_id)
.outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
.outerjoin(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
.where(Video.id == video_id)
).first()
return serialize(row) if row is not None else None
@@ -179,7 +179,7 @@ def playlists_summary(
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))
.outerjoin(st, and_(st.video_id == Video.id, owned_by(st.user_id, user)))
.where(PlaylistItem.playlist_id.in_(playlist_ids))
.order_by(PlaylistItem.playlist_id, PlaylistItem.position.asc())
).all()
+6 -1
View File
@@ -328,9 +328,14 @@ def create_own(db: Session, user_id: int, name: str) -> FileFolder:
return folder
def renamable(folder: FileFolder) -> bool:
"""Only a folder of the user's own making takes a name: a source folder's is its origin's."""
return folder.kind == FolderKind.OWN
def rename(db: Session, folder: FileFolder, name: str) -> None:
"""Rename one of the user's own folders (a source folder's name is its origin's)."""
if folder.kind != FolderKind.OWN:
if not renamable(folder):
raise FolderError("file_folder_not_own")
folder.name = name
try:
+3 -5
View File
@@ -296,13 +296,11 @@ def own_link(db: Session, user: User, link_id: int) -> DownloadLink | None:
if link is None:
return None
target = (
db.get(DownloadJob, link.job_id)
owned_row(db, DownloadJob, link.job_id, user)
if link.job_id is not None
else db.get(FileFolder, link.folder_id)
else owned_row(db, FileFolder, link.folder_id, user)
)
if target is None or target.user_id != user.id:
return None
return link
return link if target is not None else None
def links_to(
+1
View File
@@ -16,6 +16,7 @@ import shutil
import subprocess
import sys
import app.tables # noqa: F401 — a fresh interpreter: every table, before the first query
from app import logs
from app import logsink
from app.config import settings
+1 -1
View File
@@ -212,7 +212,7 @@ def filtered_query(
# exactly. 'unwatched'/'all' include row-less 'new' videos, so they keep the outer join.
state_required = show in ("watched", "hidden", "in_progress")
join_state = query.join if state_required else query.outerjoin
query = join_state(state, and_(state.video_id == Video.id, state.user_id == user.id))
query = join_state(state, and_(state.video_id == Video.id, owned_by(state.user_id, user)))
# Hide videos the maintenance job has flagged as currently unplayable (deleted/private/
# abandoned). They're either deleted after a grace period or unhidden if they recover;
+1
View File
@@ -28,6 +28,7 @@ log = logs.get_logger(__name__)
from app import bodylimit, collation, logsink, state, sysconfig # noqa: E402 — logging is configured first
from app.config import settings # noqa: E402 — logging is configured first
from app.db import SessionLocal, engine # noqa: E402 — logging is configured first
import app.tables # noqa: E402 — every table, before the first query
from app.downloads import service as download_service # noqa: E402 — logging is configured first
from app.downloads import storage # noqa: E402 — logging is configured first
from app.plex import stream as plex_stream # noqa: E402 — logging is configured first
+5 -3
View File
@@ -9,6 +9,7 @@ table can't grow without bound for a heavy user.
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app.models import Notification
# Keep at most this many READ notifications per user; older read ones are pruned. Unread
@@ -48,11 +49,13 @@ def create_notification(
def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
"""Delete a user's oldest READ notifications beyond `cap`. Returns the number removed.
Idempotent and cheap; safe to call after marking things read."""
Idempotent and cheap; safe to call after marking things read. Flushes first, so a notice the
caller has just marked read counts; the caller commits."""
db.flush()
ids = (
db.execute(
select(Notification.id)
.where(Notification.user_id == user_id, Notification.read.is_(True))
.where(owned_by(Notification.user_id, user_id), Notification.read.is_(True))
.order_by(Notification.created_at.desc())
.offset(cap)
)
@@ -63,5 +66,4 @@ def trim_read(db: Session, user_id: int, cap: int = READ_SOFT_CAP) -> int:
return 0
# One bulk DELETE rather than a get+delete per row (the ids are already in hand).
db.execute(delete(Notification).where(Notification.id.in_(ids)))
db.commit()
return len(ids)
-14
View File
@@ -108,20 +108,6 @@ def state(db: Session, user: User | int, item_id: int) -> PlexState | None:
)
def states(db: Session, user: User | int, item_ids: list[int]) -> dict[int, PlexState]:
"""The user's state on each of these items that has one, by item id."""
if not item_ids:
return {}
return {
s.item_id: s
for s in db.execute(
select(PlexState).where(
owned_by(PlexState.user_id, user), PlexState.item_id.in_(item_ids)
)
).scalars()
}
# --- a show ----------------------------------------------------------------------------------
+2 -1
View File
@@ -8,6 +8,7 @@ from datetime import datetime, timezone
from sqlalchemy import and_, case, func, or_, select
from sqlalchemy.orm import Session, aliased
from app.models import User
from app.userscope import owned_by
from app.plex.models import PlexItem, PlexState
from app.plex.cards import ShowState
@@ -106,7 +107,7 @@ def state_map(db: Session, user_id: int, show_ids: list[int]) -> dict[int, ShowS
return {r.show_id: show_state(r.total, r.watched, r.inprog) for r in rows}
def item_state_map(db: Session, user_id: int, item_ids: list[int]) -> dict[int, PlexState]:
def item_state_map(db: Session, user_id: User | int, item_ids: list[int]) -> dict[int, PlexState]:
"""A user's PlexState rows for a set of leaf items (movies or episodes), keyed by item_id — the
leaf analogue of `state_map` (which rolls a show's episodes up to a `ShowState`). One batched
fetch, so the caller serializes cards without a per-card query. Empty in → empty dict (no `IN ()`)."""
+36 -11
View File
@@ -19,7 +19,7 @@ from app.plex import people as plex_people
from app.plex import tmdb as plex_tmdb
from app.plex.lookups import by_rating_key
from app.models import User
from app.plex import queries
from app.plex import queries, rollup
from app.plex.models import (
PlexCollection,
PlexItem,
@@ -302,6 +302,36 @@ def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> boo
return False
def state_row(
db: Session,
user: User | int,
item_id: int,
st: PlexState | None = None,
*,
synced: bool = False,
) -> PlexState:
"""The user's state row on this item, made if the read before found none. Made by the insert
itself (ON CONFLICT DO NOTHING) and then read: the first checkpoint of an item and the `final`
flush beside it routinely both read "none", and a plain add let the second one fail on
`uq_plex_user_item` — a 500, and the push it was carrying lost. Every first row goes through
here: a checkpoint, a status, a season marked at once, a pull from Plex. `synced` is what a row
made now says about Plex; a row another writer made first keeps its own."""
if st is not None:
return st
db.execute(
pg_insert(PlexState)
.values(
user_id=user if isinstance(user, int) else user.id,
item_id=item_id,
synced_to_plex=synced,
)
.on_conflict_do_nothing(constraint="uq_plex_user_item")
)
made = queries.state(db, user, item_id)
assert made is not None # inserted now, or committed by the request that won
return made
def clear_progress(db: Session, st: PlexState | None, now: datetime) -> None:
"""A checkpoint too early to keep: a state that held nothing else goes, any other forgets its
position."""
@@ -325,9 +355,7 @@ def checkpoint(
) -> PlexState:
"""Record a resume position. `dirty` marks it not yet mirrored to Plex — set before a push, so
the reconcile re-pushes it if the push fails and the push can settle it once Plex confirms."""
if st is None:
st = PlexState(user_id=user.id, item_id=item_id)
db.add(st)
st = state_row(db, user, item_id, st)
st.position_seconds = position
st.progress_updated_at = now
if dirty:
@@ -345,9 +373,7 @@ def set_status(
db: Session, user: User, item_id: int, st: PlexState | None, status: str
) -> PlexState:
"""Set a watched or hidden status; watched also clears the resume point."""
if st is None:
st = PlexState(user_id=user.id, item_id=item_id)
db.add(st)
st = state_row(db, user, item_id, st)
st.status = status
if status == "watched":
st.watched_at = datetime.now(timezone.utc)
@@ -364,15 +390,14 @@ def bulk_state(
(item id, rating key, action) pushes a linked Plex account needs. Unwatching keeps the user's
private "hidden" marks."""
now = datetime.now(timezone.utc)
existing = queries.states(db, user_id, [e.id for e in eps])
existing = rollup.item_state_map(db, user_id, [e.id for e in eps])
to_push: list[tuple[int, str, str]] = []
for e in eps:
st = existing.get(e.id)
prev = st.status if st is not None else "new"
if watched:
if st is None:
st = PlexState(user_id=user_id, item_id=e.id)
db.add(st)
st = state_row(db, user_id, e.id, st)
prev = st.status # a first checkpoint beside this may have made it meanwhile
st.status = "watched"
st.watched_at = now
st.position_seconds = 0
+7 -20
View File
@@ -32,6 +32,8 @@ from app.db import SessionLocal
from app.userscope import owned_by
from app.plex.models import PlexItem, PlexLibrary, PlexLink, PlexState
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex import queries
from app.plex.service import state_row
from app.plex.sync import (
_TYPE_EPISODE,
_TYPE_MOVIE,
@@ -252,15 +254,7 @@ def push_state_to_plex(
# Flag the row synced — but an "unwatched" via status=new deletes the row, so there may be
# nothing to flag; that's fine (unscrobble still happened). Skip the flag if the row was
# re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile).
st = (
db.execute(
select(PlexState).where(
owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id
)
)
.scalars()
.first()
)
st = queries.state(db, user_id, item_id)
if st is not None and _row_matches(st, expect_ts):
st.synced_to_plex = True
db.commit()
@@ -394,15 +388,7 @@ def _pull_apply(
"""Bring one Plex-side positive signal into Siftlode under last-write-wins. Returns
"watched"/"in_progress" if it changed the row, else None. Never clears a row (un-watch is the
full reconcile's job) — this only pulls in watched/resume."""
st = (
db.execute(
select(PlexState).where(
owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id
)
)
.scalars()
.first()
)
st = queries.state(db, user_id, item_id)
if _same_state(st, status, pos):
if st is not None and not st.synced_to_plex:
st.synced_to_plex = True # already matches Plex → just settle the flag, no churn
@@ -423,8 +409,9 @@ def _pull_apply(
# "almost always already pushed" assumption above excludes.
return None
if st is None:
st = PlexState(user_id=user_id, item_id=item_id)
db.add(st)
st = state_row(db, user_id, item_id, synced=True)
if not st.synced_to_plex: # a local change made it first, beside this pull: the guard above
return None
if status == "watched":
st.status, st.position_seconds, st.watched_at, st.progress_updated_at = (
"watched",
+2 -1
View File
@@ -16,6 +16,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.engine import CursorResult
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app import logs, sysconfig
from app.db import SessionLocal
from app.models import ApiQuotaUsage, QuotaEvent
@@ -173,7 +174,7 @@ def actions_today(db: Session, user_id: int, action: str) -> int:
select(func.count())
.select_from(QuotaEvent)
.where(
QuotaEvent.user_id == user_id,
owned_by(QuotaEvent.user_id, user_id),
QuotaEvent.action == action,
QuotaEvent.created_at >= pacific_day_start_utc(),
)
+11
View File
@@ -11,12 +11,14 @@ from sqlalchemy import (
DateTime,
Float,
ForeignKey,
Index,
Integer,
JSON,
String,
Text,
UniqueConstraint,
func,
text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -93,6 +95,15 @@ class Playlist(Base, TimestampMixin, UpdatedAtMixin):
sync phases; the foundation phase only uses local playlists.)"""
__tablename__ = "playlists"
__table_args__ = (
# One Watch later per user, held by the database: two first saves at once each made one.
Index(
"uq_playlist_watch_later",
"user_id",
unique=True,
postgresql_where=text("kind = 'watch_later'"),
),
)
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+41 -24
View File
@@ -9,7 +9,7 @@ unique key, a row lock taken before a presence check. The comfort shelf's writes
from datetime import datetime, timezone
from sqlalchemy import exists, func, select, update
from sqlalchemy import exists, func, select, text, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session
@@ -239,24 +239,33 @@ def clear_state(db: Session, user: User, video_id: str) -> None:
def set_progress(db: Session, user: User, video_id: str, position: int) -> None:
"""The player's resume point. 0 clears it: a row holding nothing else goes, one with a status
(watched, hidden) keeps it with the position zeroed. The status itself is never changed here."""
row = queries.video_state(db, user, video_id)
(watched, hidden) keeps it with the position zeroed. The status itself is never changed here.
A position is one upsert on (user, video): the first checkpoint and the end-of-playback mark
beside it both read "no row", and a plain add let the second one fail on `uq_user_video`."""
now = datetime.now(timezone.utc)
if not position:
if row is None:
return
if row.status == "new":
db.delete(row)
else:
row.position_seconds = 0
row.progress_updated_at = now
db.flush()
if position:
db.execute(
pg_insert(VideoState)
.values(
user_id=user.id,
video_id=video_id,
position_seconds=position,
progress_updated_at=now,
)
.on_conflict_do_update(
constraint="uq_user_video",
set_={"position_seconds": position, "progress_updated_at": now, "updated_at": now},
)
)
return
row = queries.video_state(db, user, video_id)
if row is None:
row = VideoState(user_id=user.id, video_id=video_id)
db.add(row)
row.position_seconds = position
row.progress_updated_at = now
return
if row.status == "new":
db.delete(row)
else:
row.position_seconds = 0
row.progress_updated_at = now
db.flush()
@@ -319,17 +328,25 @@ def create_playlist(db: Session, user: User, name: str) -> Playlist:
def watch_later(db: Session, user: User) -> Playlist:
"""The user's built-in Watch later playlist, created on first use."""
"""The user's built-in Watch later playlist, created on first use. Created by the insert itself
(ON CONFLICT DO NOTHING on the one-per-user index) and then read: the first two saves at once
both read "none", and each used to make a list of its own."""
pl = queries.watch_later(db, user)
if pl is None:
pl = Playlist(
user_id=user.id,
name="Watch later",
kind="watch_later",
position=_next_playlist_position(db, user.id),
db.execute(
pg_insert(Playlist)
.values(
user_id=user.id,
name="Watch later",
kind="watch_later",
position=_next_playlist_position(db, user.id),
)
.on_conflict_do_nothing(
index_elements=["user_id"], index_where=text("kind = 'watch_later'")
)
)
db.add(pl)
db.flush() # the id, for the item the caller adds next
pl = queries.watch_later(db, user)
assert pl is not None # inserted now, or committed by the request that won
return pl
+3 -2
View File
@@ -89,8 +89,9 @@ def owned_or_404(
) -> M:
"""Fetch a row by primary key that `user` owns, or raise 404 (`lock`: see `get_or_404`). A row owned by someone
else 404s (NOT 403) on purpose — a 403 would confirm the id exists to a stranger.
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."""
A core route calls it directly. A unit's route cannot (it may not name a unit's model as a value):
it asks its unit's `queries.own_x`, which goes through the same `owned_row`, and answers
`not_found` itself."""
obj = owned_row(db, model, id_, user, owner_attr=owner_attr, lock=lock)
if obj is None:
raise not_found(what)
+2 -1
View File
@@ -16,6 +16,7 @@ from sqlalchemy import select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app import apitokens
from app import email as email_mod
from app import logs
@@ -1124,7 +1125,7 @@ def password_reset_confirm(body: TokenPasswordIn, db: Session = Depends(get_db))
# Burn any other outstanding reset tokens for this user.
for row in db.execute(
select(AuthToken).where(
AuthToken.user_id == user.id,
owned_by(AuthToken.user_id, user),
AuthToken.kind.in_(_PASSWORD_TOKEN_KINDS),
AuthToken.used_at.is_(None),
)
+3 -1
View File
@@ -18,6 +18,7 @@ from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app import audit, debugsession, logs, logsink
from app.audit import AuditAction
from app.auth import current_user
@@ -95,7 +96,7 @@ def stop_session(
def list_sessions(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
sessions = db.scalars(
select(DebugSession)
.where(DebugSession.user_id == user.id)
.where(owned_by(DebugSession.user_id, user))
.order_by(DebugSession.started_at.desc())
.limit(LIST_MAX)
).all()
@@ -219,6 +220,7 @@ def send_issue(sid: str, user: User = Depends(current_user), db: Session = Depen
raise refused(409, "issue_sending")
new = session.issue_url is None
if not new and not debugsession.unfinished(session):
db.commit() # nothing to send: the lock ends here, not when the request's session closes
return {"ok": True, "session": _view(db, session)}
key = str(user.id)
if new:
+3
View File
@@ -375,6 +375,9 @@ def update_folder(
folder = own_folder(db, user, folder_id)
sent = body.model_fields_set
if "name" in sent:
# Which folder may be renamed is answered before what the name is, as it always was.
if not folders.renamable(folder):
raise refused(409, "file_folder_not_own")
try:
folders.rename(db, folder, _folder_name(body.name or ""))
except folders.FolderError as e:
+7 -6
View File
@@ -5,6 +5,7 @@ from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, select, update
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app.auth import current_user
from app.db import get_db
from app.models import Notification, User
@@ -36,7 +37,7 @@ def list_notifications(
user: User = Depends(current_user),
db: Session = Depends(get_db),
) -> dict:
where = [Notification.user_id == user.id]
where = [owned_by(Notification.user_id, user)]
if not include_dismissed:
where.append(Notification.dismissed.is_(False))
total = db.scalar(select(func.count()).select_from(Notification).where(*where)) or 0
@@ -61,7 +62,7 @@ def unread_count(user: User = Depends(current_user), db: Session = Depends(get_d
select(func.count())
.select_from(Notification)
.where(
Notification.user_id == user.id,
owned_by(Notification.user_id, user),
Notification.read.is_(False),
Notification.dismissed.is_(False),
)
@@ -77,8 +78,8 @@ def mark_read(
) -> dict:
n = owned_or_404(db, Notification, notification_id, user, what="notification")
n.read = True
db.commit()
trim_read(db, user.id)
db.commit()
return {"ok": True}
@@ -86,11 +87,11 @@ def mark_read(
def mark_all_read(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
db.execute(
update(Notification)
.where(Notification.user_id == user.id, Notification.read.is_(False))
.where(owned_by(Notification.user_id, user), Notification.read.is_(False))
.values(read=True)
)
db.commit()
trim_read(db, user.id)
db.commit()
return {"ok": True}
@@ -113,7 +114,7 @@ def clear_all(user: User = Depends(current_user), db: Session = Depends(get_db))
"""Dismiss every (still-visible) notification for this user."""
db.execute(
update(Notification)
.where(Notification.user_id == user.id, Notification.dismissed.is_(False))
.where(owned_by(Notification.user_id, user), Notification.dismissed.is_(False))
.values(dismissed=True, read=True)
)
db.commit()
+2 -2
View File
@@ -100,7 +100,7 @@ def show_detail(
raise not_found("show")
seasons = queries.seasons(db, sh)
eps = queries.episodes(db, sh)
states = queries.states(db, user, [e.id for e in eps])
states = rollup.item_state_map(db, user, [e.id for e in eps])
by_season: dict[int | None, list] = {}
all_cards: list[dict] = [] # every episode, in season/episode order → show-level rollup
for e in eps:
@@ -247,7 +247,7 @@ def _collection_strips(db: Session, it: "PlexItem", user_id: int, lang: str) ->
for col in ordered:
rk = col.rating_key
members = queries.library_movies_in(db, it.library_id, rk, lang, limit=60)
states = queries.states(db, user_id, [m.id for m in members])
states = rollup.item_state_map(db, user_id, [m.id for m in members])
out.append(
{
"id": rk,
+2 -2
View File
@@ -9,7 +9,7 @@ from sqlalchemy.orm import Session
from app.auth import current_user
from app.db import get_db
from app.models import User
from app.plex import queries, service
from app.plex import queries, rollup, service
from app.plex.cards import display_title, leaf_card, norm_lang
from app.refusal import refused
from app.routes._common import not_found
@@ -99,7 +99,7 @@ def playlist_detail(
lang = norm_lang(lang)
pl = _own_playlist(db, user, pid)
items = queries.playlist_items(db, pl.id)
states = queries.states(db, user, [it.id for it in items])
states = rollup.item_state_map(db, user, [it.id for it in items])
shows = queries.shows_by_ids(db, {it.show_id for it in items if it.show_id})
cards = [
leaf_card(
+2 -1
View File
@@ -10,6 +10,7 @@ from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app import quota
from app.auth import admin_user, current_user
from app.db import get_db
@@ -27,7 +28,7 @@ def _sum(db: Session, *conds) -> int:
@router.get("/my-usage")
def my_usage(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
now = datetime.now(timezone.utc)
mine = QuotaEvent.user_id == user.id
mine = owned_by(QuotaEvent.user_id, user)
by_action = {
action: int(units)
for action, units in db.execute(
+3 -2
View File
@@ -16,6 +16,7 @@ from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.userscope import owned_by
from app import apitokens, audit
from app.apitokens import GRANT_COLUMNS, Right
from app.audit import AuditAction
@@ -86,7 +87,7 @@ def _serialize(db: Session, token: ApiToken) -> dict:
@router.get("")
def list_tokens(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
tokens = (
db.execute(select(ApiToken).where(ApiToken.user_id == user.id).order_by(ApiToken.id))
db.execute(select(ApiToken).where(owned_by(ApiToken.user_id, user)).order_by(ApiToken.id))
.scalars()
.all()
)
@@ -134,7 +135,7 @@ def create_token(
if (epoch or 0) != (user.session_epoch or 0):
raise refused(401, "not_authenticated")
count = db.execute(
select(func.count(ApiToken.id)).where(ApiToken.user_id == user.id)
select(func.count(ApiToken.id)).where(owned_by(ApiToken.user_id, user))
).scalar_one()
if count >= MAX_TOKENS:
raise refused(409, "api_token_limit", limit=MAX_TOKENS)
+4 -3
View File
@@ -1,10 +1,11 @@
"""Whose a row is, and who an account is to the others — one place for both.
**Ownership.** A user-owned table says whose a row is in one column (`user_id`, or `to_user_id` for
something handed to them). A unit's reads and writes name that column through `owned_by` and fetch a
something handed to them). Every read and write names that column through `owned_by` and fetches a
row by id through `owned_row`, never with an inline `Model.user_id == …`: the rule "a row owned by
someone else is answered like a missing one" then lives here, and a contract test
(`test_unit_fences.py`) holds a switched-on unit to it.
someone else is answered like a missing one" then lives here. A contract test
(`test_unit_fences.py`) holds every module, core included, to the filter's form; a rule this module
cannot say (own or shared, built-in or own) is written out where it applies, with its reason.
**Recipients.** "An active, non-suspended account" — the download share-by-email and the recipient
picker filter on it. Kept here so the two forms (SQL clauses for a query, a Python check for a loaded
+1
View File
@@ -31,6 +31,7 @@ from functools import partial
from sqlalchemy import text
from sqlalchemy.orm import Session
import app.tables # noqa: F401 — every table, before the first query
from app import logs
from app import logsink
from app import sysconfig
+15 -4
View File
@@ -105,6 +105,11 @@ def pytest_configure(config):
)
def _write_xact(db) -> str | None:
"""The id of the open transaction if it has written anything, else None (a read never takes one)."""
return db.scalar(text("SELECT pg_current_xact_id_if_assigned()::text"))
def _request_dep(db, unfinished: list[str]):
"""`get_db` for a test request: the test's own session, plus the check a shared session hides.
@@ -114,15 +119,16 @@ def _request_dep(db, unfinished: list[str]):
and the suite stays green over a handler that forgot to commit (deep review, debt burn). So a
request that ends normally with a write it did not commit — flushed, or still only staged — is
recorded, and the test fails on it at teardown. Only the ROOT transaction's end clears it: a
savepoint released inside the handler commits nothing. The test's own staged rows are flushed before the request starts — they are the
test's business, not the handler's. A request that raises is rolled back, as production's
savepoint released inside the handler commits nothing. What the test itself staged or wrote is
committed before the request starts — it is the test's business, not the handler's, and a
request then begins with no write open, so any write left at its end is the handler's. A request that raises is rolled back, as production's
`close()` rolls it back: without that, what a failing handler flushed but never committed stayed
readable to the rest of the test — the one state production cannot be in (an issue created
on the tracker before a failed upload was kept only by an explicit commit, and nothing saw it
missing)."""
def dep(conn: HTTPConnection):
db.flush()
db.commit()
state = {"dirty": False}
def flushed(session, ctx):
@@ -138,7 +144,12 @@ def _request_dep(db, unfinished: list[str]):
try:
yield db
staged = db.new or db.deleted or any(db.is_modified(o) for o in db.dirty)
if state["dirty"] or staged:
# Core DML (`db.execute(update(...))`) fires no flush and stages nothing; the database
# itself says whether a write is open — the request began with none, and a transaction
# that wrote and was not committed holds an id (see `release_read_connection`). A row
# lock (`FOR UPDATE`) takes one too: the handler ends what it locked.
core_dml = _write_xact(db) is not None
if state["dirty"] or staged or core_dml:
unfinished.append(f"{conn.scope.get('method', 'WS')} {conn.url.path}")
except Exception:
db.rollback()
+14
View File
@@ -8,6 +8,8 @@ drops an unknown id, so the view silently showed the whole feed under its old na
from sqlalchemy import select
from app.catalog.models import Channel
from app.db import SessionLocal
from app.sync.autotag import run_autotag_all
from app.reader.models import ChannelTag, SavedView, Tag
from app.reader import service as reader_service
@@ -66,3 +68,15 @@ def test_a_view_kept_tag_list_larger_than_a_bind_parameter_limit_still_cleans_up
db.commit()
assert reader_service.prune_orphan_system_tags(db) == 1
db.commit()
def test_the_job_commits_its_cleanup(db):
"""The prune only flushes; the job ends its own transaction. Read from a session of its own,
the orphan is gone only once the job committed — the job's session would see its own delete
either way."""
orphan = Tag(user_id=None, name="Cooking", category="topic")
db.add(orphan)
db.commit()
assert run_autotag_all(db, only_missing=True)["orphan_tags_removed"] == 1
with SessionLocal() as fresh:
assert fresh.execute(select(Tag).where(Tag.name == "Cooking")).first() is None
+69
View File
@@ -0,0 +1,69 @@
"""Every process the app runs as sees every table before its first query.
The units declare their tables in their own `models` modules, and a core model names a unit's by
string (`User.subscriptions` → `Subscription`). A fresh interpreter that never imported the unit's
module fails to configure its mappers on its first query: the download child once did exactly that,
and every job failed, while this suite stayed green — it imports `app.tables` in-process, which hides
the gap. So each entry point is imported here in a clean interpreter, and must register the whole
schema on its own.
"""
import ast
import json
import subprocess
import sys
from pathlib import Path
import app.tables # noqa: F401
from app.db import Base
BACKEND = Path(__file__).parent.parent
APP = BACKEND / "app"
# Each `python -m` module the deployment starts, and the API's app module.
ENTRY_POINTS = ("app.main", "app.worker", "app.downloads.runner")
# The table list goes out on one marked line: the API's logging writes to the same stdout, and a
# line it logs while importing is not the probe's answer.
_MARK = "entry-point-tables:"
_PROBE = (
"import importlib, json, sys\n"
"importlib.import_module(sys.argv[1])\n"
"from sqlalchemy.orm import configure_mappers\n"
"from app.db import Base\n"
"configure_mappers()\n"
f"print({_MARK!r} + json.dumps(sorted(Base.metadata.tables)))\n"
)
def test_each_entry_point_registers_the_whole_schema_on_its_own():
whole = sorted(Base.metadata.tables)
for module in ENTRY_POINTS:
r = subprocess.run(
[sys.executable, "-c", _PROBE, module],
cwd=BACKEND,
capture_output=True,
text=True,
timeout=120,
check=False,
)
assert r.returncode == 0, (module, r.stderr[-2000:])
answer = [ln for ln in r.stdout.splitlines() if ln.startswith(_MARK)]
assert len(answer) == 1, (module, r.stdout[-2000:])
assert json.loads(answer[0].removeprefix(_MARK)) == whole, module
def test_every_module_that_runs_as_a_process_is_listed():
mains = set()
for path in APP.rglob("*.py"):
tree = ast.parse(path.read_text(encoding="utf-8"))
for node in tree.body:
if (
isinstance(node, ast.If)
and isinstance(node.test, ast.Compare)
and isinstance(node.test.left, ast.Name)
and node.test.left.id == "__name__"
):
rel = path.relative_to(BACKEND).with_suffix("").as_posix()
mains.add(rel.replace("/", "."))
assert mains <= set(ENTRY_POINTS), mains - set(ENTRY_POINTS)
+3
View File
@@ -124,6 +124,9 @@ def test_only_an_own_folder_is_renamed_and_only_by_its_user(db, make_user, as_us
assert clash.json()["detail"]["reason"] == "file_folder_name_taken"
no = c.patch(f"/api/files/folders/{src.id}", json={"name": "YouTube 2"})
assert no.status_code == 409 and no.json()["detail"]["reason"] == "file_folder_not_own"
# Which folder is answered before what the name is: a blank one is still "not yours to name".
blank = c.patch(f"/api/files/folders/{src.id}", json={"name": " "})
assert blank.status_code == 409 and blank.json()["detail"]["reason"] == "file_folder_not_own"
stranger = as_user(make_user()).patch(f"/api/files/folders/{own.id}", json={"name": "X"})
assert stranger.status_code == 404
db.refresh(own)
+90
View File
@@ -0,0 +1,90 @@
"""Migration 0112's data step, DB-lane: a user who already has two Watch later lists keeps the oldest,
with the other's videos moved in after its own (one it already held is not added twice) and a Home
pin of the other moved onto it, and the one-per-user index then refuses a second list.
The schema half is checked by test_migrations.py; a video or a pin lost on the way is invisible there.
"""
from datetime import datetime, timezone
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from app.catalog.models import Channel, Video
from app.db import engine
def _lists(conn, user_id: int, n: int) -> list[int]:
values = ", ".join(f"(:u, 'Watch later', 'watch_later', {i})" for i in range(n))
return list(
conn.execute(
text(
f"INSERT INTO playlists (user_id, name, kind, position) VALUES {values} RETURNING id"
),
{"u": user_id},
).scalars()
)
def _pin(conn, user_id: int, playlist_id: int, position: int) -> None:
conn.execute(
text("INSERT INTO comfort_pins (user_id, playlist_id, position) VALUES (:u, :p, :pos)"),
{"u": user_id, "p": playlist_id, "pos": position},
)
def test_a_second_watch_later_is_folded_into_the_first(db, make_user):
user, other = make_user(), make_user()
db.add(Channel(id="UCm", title="M"))
now = datetime.now(timezone.utc)
for v in ("v1", "v2", "v3"):
db.add(Video(id=v, channel_id="UCm", published_at=now, duration_seconds=60))
db.commit()
cfg = Config("alembic.ini")
command.downgrade(cfg, "0111_audit_request_id")
try:
with engine.begin() as conn:
first, second, third = _lists(conn, user.id, 3)
theirs, their_second = _lists(conn, other.id, 2)
conn.execute(
text(
"INSERT INTO playlist_items (playlist_id, video_id, position) VALUES"
" (:a, 'v1', 0), (:b, 'v2', 0), (:b, 'v1', 1), (:c, 'v3', 0)"
),
{"a": first, "b": second, "c": theirs},
)
# The user pinned two of the copies, the kept one neither: the first on the shelf moves.
_pin(conn, user.id, third, 0)
_pin(conn, user.id, second, 1)
# The other pinned both: the kept one's pin stays, and is the only one.
_pin(conn, other.id, their_second, 0)
_pin(conn, other.id, theirs, 1)
finally:
command.upgrade(cfg, "head")
with engine.connect() as conn:
lists = conn.execute(
text("SELECT id, user_id FROM playlists WHERE kind = 'watch_later' ORDER BY id")
).all()
assert [tuple(r) for r in lists] == [(first, user.id), (theirs, other.id)]
items = conn.execute(
text(
"SELECT playlist_id, video_id, position FROM playlist_items ORDER BY playlist_id, position"
)
).all()
assert [tuple(r) for r in items] == [(first, "v1", 0), (first, "v2", 1), (theirs, "v3", 0)]
pins = conn.execute(
text("SELECT user_id, playlist_id, position FROM comfort_pins ORDER BY user_id")
).all()
assert [tuple(r) for r in pins] == [(user.id, first, 0), (other.id, theirs, 1)]
with pytest.raises(IntegrityError), engine.begin() as conn:
conn.execute(
text(
"INSERT INTO playlists (user_id, name, kind, position)"
" VALUES (:u, 'Watch later', 'watch_later', 2)"
),
{"u": user.id},
)
@@ -82,3 +82,42 @@ def test_a_request_that_raises_leaves_nothing_it_did_not_commit(db):
request.throw(RuntimeError("the handler failed"))
assert db.scalar(select(User).where(User.email == "rolled-back-probe@example.com")) is None
assert db.info["unfinished"] == [] # a failing request is rolled back, not reported
def test_the_commit_check_sees_a_core_dml_write(db, make_user):
"""`db.execute(update(...))` flushes nothing and stages nothing, so the session's own bookkeeping
never sees it: a handler whose only write is Core DML and that forgot its commit passed. The
database's transaction id is what tells — also when the test itself had a write open before the
request, which the handler's write would otherwise share. A committed one must still pass."""
import pytest
from sqlalchemy import update
from starlette.requests import HTTPConnection
from app.models import User
user = make_user()
def drive(commit: bool, *, test_wrote: bool = False) -> None:
if test_wrote: # the test's own write, open and uncommitted, as the request starts
db.execute(update(User).where(User.id == user.id).values(display_name="test"))
conn = HTTPConnection(
{"type": "http", "method": "POST", "path": "/probe", "headers": [], "query_string": b""}
)
request = db.info["request"](conn)
session = next(request)
session.execute(update(User).where(User.id == user.id).values(display_name="dml"))
if commit:
session.commit()
with pytest.raises(StopIteration):
next(request)
drive(commit=True)
assert db.info["unfinished"] == []
drive(commit=False)
assert db.info["unfinished"] == ["POST /probe"]
db.info["unfinished"].clear() # recorded as it should be — not this test's failure
drive(commit=True, test_wrote=True)
assert db.info["unfinished"] == []
drive(commit=False, test_wrote=True)
assert db.info["unfinished"] == ["POST /probe"]
db.info["unfinished"].clear()
+248
View File
@@ -0,0 +1,248 @@
"""The core's owner filters, each held by a second account whose rows must not count, show or change.
Every one of these reads or writes a user-owned table through `userscope.owned_by`; the fence holds
the form (no inline `Model.user_id == …`), these hold that the filter is there at all. Each test gives
a second account the same kind of rows, so a filter that went missing answers with theirs. DB lane.
"""
from datetime import datetime, timedelta, timezone
from sqlalchemy import func, select
from app import debugsession, notifications, quota
from app.catalog.models import Channel, Video
from app.downloads.models import DownloadJob, DownloadProfile
from app.models import DebugScreenshot, DebugSession, Notification, QuotaEvent
from app.reader.models import Playlist, PlaylistItem, VideoState
from app.routes import tokens as tokens_routes
from app.routes.auth import RESET_TTL, _issue_token
def _notify(db, user, n=1, *, read=False):
for i in range(n):
db.add(Notification(user_id=user.id, type="t", title=f"n{i}", read=read))
db.commit()
def _visible(db, user) -> int:
return db.scalar(
select(func.count())
.select_from(Notification)
.where(Notification.user_id == user.id, Notification.dismissed.is_(False))
)
def test_the_notification_list_shows_only_mine(db, make_user, as_user):
me, other = make_user(), make_user()
_notify(db, me)
_notify(db, other, 2)
body = as_user(me).get("/api/me/notifications").json()
assert (body["total"], len(body["items"])) == (1, 1)
def test_the_unread_count_counts_only_mine(db, make_user, as_user):
me, other = make_user(), make_user()
_notify(db, me)
_notify(db, other, 2)
assert as_user(me).get("/api/me/notifications/unread_count").json() == {"count": 1}
def test_reading_all_reads_only_mine(db, make_user, as_user):
me, other = make_user(), make_user()
_notify(db, me)
_notify(db, other, 2)
assert as_user(me).post("/api/me/notifications/read_all").status_code == 200
assert as_user(other).get("/api/me/notifications/unread_count").json() == {"count": 2}
def test_clearing_all_dismisses_only_mine(db, make_user, as_user):
me, other = make_user(), make_user()
_notify(db, me)
_notify(db, other, 2)
assert as_user(me).post("/api/me/notifications/clear").status_code == 200
db.expire_all()
assert (_visible(db, me), _visible(db, other)) == (0, 2)
def test_trimming_read_notifications_trims_only_mine(db, make_user):
me, other = make_user(), make_user()
_notify(db, me, 3, read=True)
_notify(db, other, 3, read=True)
assert notifications.trim_read(db, me.id, cap=1) == 2
db.commit()
assert _visible(db, other) == 3
def test_todays_actions_count_only_mine(db, make_user):
me, other = make_user(), make_user()
quota.log_action(me.id, quota.QuotaAction.VIDEOS_SEARCH)
quota.log_action(other.id, quota.QuotaAction.VIDEOS_SEARCH)
quota.log_action(other.id, quota.QuotaAction.VIDEOS_SEARCH)
assert quota.actions_today(db, me.id, quota.QuotaAction.VIDEOS_SEARCH) == 1
def test_my_usage_sums_only_mine(db, make_user, as_user):
me, other = make_user(), make_user()
db.add_all(
[
QuotaEvent(user_id=me.id, action="videos.list", units=5),
QuotaEvent(user_id=other.id, action="videos.list", units=7),
]
)
db.commit()
body = as_user(me).get("/api/quota/my-usage").json()
assert (body["today"], body["all_time"], body["by_action"]) == (5, 5, {"videos.list": 5})
def test_the_token_ceiling_counts_only_mine(db, make_user, as_user, monkeypatch):
monkeypatch.setattr(tokens_routes, "MAX_TOKENS", 1)
me, other = make_user(), make_user()
first = as_user(other).post("/api/tokens", json={"name": "theirs", "preset": "hub"})
assert first.status_code == 200, first.text
r = as_user(me).post("/api/tokens", json={"name": "mine", "preset": "hub"})
assert r.status_code == 200, r.text
def test_a_reset_burns_only_my_other_links(client, db, make_user):
me, other = make_user(), make_user()
mine = _issue_token(db, me, "reset", RESET_TTL)
theirs = _issue_token(db, other, "reset", RESET_TTL)
db.commit()
r = client.post(
"/auth/password-reset/confirm", json={"token": mine, "password": "a brand new passphrase"}
)
assert r.json() == {"ok": True}
assert client.post("/auth/password-reset/check", json={"token": theirs}).json() == {"ok": True}
def test_screenshot_storage_counts_only_mine(db, make_user):
me, other = make_user(), make_user()
now = datetime.now(timezone.utc)
for i, (user, size) in enumerate([(me, 10), (other, 1000)]):
db.add(
DebugSession(
id=f"s{i}",
user_id=user.id,
level="DEBUG",
started_at=now,
until=now + timedelta(hours=1),
)
)
db.flush()
db.add(
DebugScreenshot(
session_id=f"s{i}", created_at=now, mime="image/png", size=size, data=b"x" * size
)
)
db.commit()
assert debugsession.stored_screenshot_bytes(db, me.id) == 10
# --- the units' owner filters, through their routes --------------------------------------------
# The same rule one layer up: a second account acting on the SAME video, list or profile id must
# reach only its own row. Each test holds the first account's row unchanged afterwards.
def _videos(db, *ids):
db.add(Channel(id="UCo", title="O"))
db.flush()
for vid in ids:
db.add(
Video(
id=vid,
channel_id="UCo",
published_at=datetime.now(timezone.utc),
duration_seconds=3600,
)
)
db.commit()
def _video_state(db, user, vid="v1"):
db.expire_all()
row = db.execute(
select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == vid)
).scalar_one_or_none()
return None if row is None else (row.status, row.position_seconds)
def test_unmarking_a_video_leaves_another_users_state(db, make_user, as_user):
me, other = make_user(), make_user()
_videos(db, "v1")
db.add(VideoState(user_id=other.id, video_id="v1", status="watched"))
db.commit()
r = as_user(me).post("/api/videos/v1/state", json={"status": "new"})
assert r.status_code == 200, r.text
assert _video_state(db, other) == ("watched", 0)
def test_clearing_a_video_leaves_another_users_state(db, make_user, as_user):
me, other = make_user(), make_user()
_videos(db, "v1")
db.add(VideoState(user_id=other.id, video_id="v1", status="hidden", position_seconds=90))
db.commit()
assert as_user(me).delete("/api/videos/v1/state").status_code == 200
assert _video_state(db, other) == ("hidden", 90)
def test_a_resume_point_is_set_and_cleared_on_my_own_row_only(db, make_user, as_user):
me, other = make_user(), make_user()
_videos(db, "v1")
db.add(VideoState(user_id=other.id, video_id="v1", status="hidden", position_seconds=90))
db.commit()
client = as_user(me)
body = {"position_seconds": 600, "duration_seconds": 3600}
assert client.post("/api/videos/v1/progress", json=body).status_code == 200
assert (_video_state(db, me), _video_state(db, other)) == (("new", 600), ("hidden", 90))
body = {"position_seconds": 0, "duration_seconds": 3600}
assert client.post("/api/videos/v1/progress", json=body).status_code == 200
assert (_video_state(db, me), _video_state(db, other)) == (None, ("hidden", 90))
def _watch_later_ids(db, user):
db.expire_all()
return sorted(
db.execute(
select(PlaylistItem.video_id)
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
.where(Playlist.user_id == user.id, Playlist.kind == "watch_later")
).scalars()
)
def test_watch_later_adds_and_removes_use_my_own_list(db, make_user, as_user):
me, other = make_user(), make_user()
_videos(db, "v1", "v2")
theirs = as_user(other).post("/api/playlists/watch-later", json={"video_id": "v1"}).json()
client = as_user(me)
mine = client.post("/api/playlists/watch-later", json={"video_id": "v2"}).json()
assert mine["playlist_id"] != theirs["playlist_id"]
# Their saved video is not on my list, so removing it from mine leaves theirs alone.
assert client.delete("/api/playlists/watch-later/v1").status_code == 200
assert (_watch_later_ids(db, me), _watch_later_ids(db, other)) == (["v2"], ["v1"])
def test_another_users_private_profile_is_refused_like_a_missing_one(db, make_user, as_user):
me, other = make_user(), make_user()
theirs = DownloadProfile(user_id=other.id, name="Theirs", spec={"format": "best"})
builtin = DownloadProfile(user_id=None, name="Best", spec={"format": "best"}, is_builtin=True)
db.add_all([theirs, builtin])
db.commit()
client = as_user(me)
def enqueue(profile_id):
return client.post(
"/api/downloads", json={"source": "dQw4w9WgXcQ", "profile_id": profile_id}
)
missing = enqueue(theirs.id + builtin.id + 1)
refused = enqueue(theirs.id)
assert (refused.status_code, refused.json()) == (404, missing.json())
assert refused.json()["detail"] == {"code": "not_found", "reason": "profile"}
db.expire_all()
assert db.execute(select(DownloadJob)).first() is None
# A built-in is everyone's to use.
r = enqueue(builtin.id)
assert r.status_code == 200, r.text
db.expire_all()
assert db.execute(select(DownloadJob.profile_id)).scalars().all() == [builtin.id]
+33 -4
View File
@@ -12,7 +12,7 @@ from datetime import datetime, timezone
from sqlalchemy import select
from app.plex import library, queries, service
from app.plex import library, queries, rollup, service
from app.plex.models import PlexCollection, PlexItem, PlexLibrary, PlexShow, PlexState
from app.plex.query import LibraryFilters
@@ -71,9 +71,9 @@ def test_a_users_watch_state_is_theirs_alone(db, make_user):
_state(db, owner, it, status="watched")
assert queries.state(db, owner, it.id).status == "watched"
assert queries.state(db, stranger, it.id) is None
assert queries.states(db, stranger, [it.id]) == {}
assert set(queries.states(db, owner, [it.id])) == {it.id}
assert queries.states(db, owner, []) == {}
assert rollup.item_state_map(db, stranger, [it.id]) == {}
assert set(rollup.item_state_map(db, owner, [it.id])) == {it.id}
assert rollup.item_state_map(db, owner, []) == {}
def test_the_grid_shows_the_callers_state_and_hides_only_what_they_hid(db, make_user):
@@ -421,3 +421,32 @@ def test_only_the_owners_own_link_pushes_their_watch_state(db, make_user):
db.commit()
assert watch_sync.link_for_push(db, owner.id).id == link.id
assert watch_sync.link_for_push(db, other.id) is None
def test_a_first_write_that_lost_the_race_updates_the_winners_row(db, make_user):
"""The first checkpoint of an item and the `final` flush beside it both read "no state"; the one
that writes second must land on the row the first made, not fail on `uq_plex_user_item`."""
user = make_user()
it = _movie(db, _lib(db), "m-race")
_state(db, user, it, position_seconds=10) # the request that won, committed
st = service.checkpoint(db, user, it.id, None, 42, NOW, dirty=True) # read "none" before it
assert (st.position_seconds, st.synced_to_plex) == (42, False)
st = service.set_status(db, user, it.id, None, "hidden")
assert st.status == "hidden"
db.commit()
rows = db.scalars(select(PlexState).where(PlexState.item_id == it.id)).all()
assert [(r.position_seconds, r.status) for r in rows] == [(42, "hidden")]
def test_a_season_marked_beside_a_first_checkpoint_updates_its_row(db, make_user, monkeypatch):
"""Marking a season reads every episode's state once; a first checkpoint that made one row since
must be updated, not collided with on `uq_plex_user_item`."""
user = make_user()
lib = _lib(db)
ep = _movie(db, lib, "e-race")
_state(db, user, ep, position_seconds=10) # the checkpoint that won, committed
monkeypatch.setattr(rollup, "item_state_map", lambda *a, **k: {}) # read before it committed
service.bulk_state(db, user.id, [ep], watched=True)
db.commit()
rows = db.scalars(select(PlexState).where(PlexState.item_id == ep.id)).all()
assert [(r.status, r.position_seconds) for r in rows] == [("watched", 0)]
+28
View File
@@ -14,6 +14,7 @@ from types import SimpleNamespace
from sqlalchemy import select
from app.models import User
from app.plex import queries
from app.plex.models import PlexItem, PlexLibrary, PlexState
from app.plex.watch_sync import _pull_apply, _reconcile_inprogress_movies
from app.routes.plex.playback import (
@@ -385,3 +386,30 @@ def test_should_push_resume_periodic_is_throttled():
)
# Once the window elapses, the next tick pushes again.
assert _should_push_resume(*key, now + timedelta(seconds=_PUSH_THROTTLE_S), final=False) is True
def test_a_pull_that_lost_the_race_keeps_the_local_change_it_met(db, monkeypatch):
"""The pull read "no state", and a first local checkpoint made one before it wrote: that row is
still waiting for its push, so the pull leaves it (the dirty guard), and fails on nothing."""
item_id, user_id = _seed_item_and_user(db)
now = datetime.now(timezone.utc)
db.add(PlexState(user_id=user_id, item_id=item_id, position_seconds=300, synced_to_plex=False))
db.commit()
real = queries.state
stale = [True]
monkeypatch.setattr(
queries, "state", lambda d, u, i: None if stale and stale.pop() else real(d, u, i)
)
assert _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now) is None
db.commit()
rows = db.scalars(select(PlexState).where(PlexState.item_id == item_id)).all()
assert [(r.position_seconds, r.synced_to_plex) for r in rows] == [(300, False)]
def test_a_first_pull_makes_a_row_in_step_with_plex(db):
item_id, user_id = _seed_item_and_user(db)
now = datetime.now(timezone.utc)
assert _pull_apply(db, user_id, item_id, "new", 120, plex_ts=now) == "in_progress"
db.commit()
rows = db.scalars(select(PlexState).where(PlexState.item_id == item_id)).all()
assert [(r.position_seconds, r.synced_to_plex) for r in rows] == [(120, True)]
+33
View File
@@ -308,3 +308,36 @@ def test_a_pin_is_one_per_thing_and_only_its_owners(db, make_user):
pins.remove(db, pins.own_pin(db, user, pin.id))
db.commit()
assert db.execute(select(ComfortPin)).first() is None
def test_a_watch_later_save_that_lost_the_race_uses_the_winners_list(db, make_user, monkeypatch):
"""The first two saves at once both read "no Watch later"; the second must land in the list the
first made, not make one of its own."""
user = make_user()
won = service.watch_later(db, user)
db.commit()
real = queries.watch_later
stale = [True] # this request's first read happened before the winner committed
def read(d, u):
return None if stale and stale.pop() else real(d, u)
monkeypatch.setattr(queries, "watch_later", read)
assert service.watch_later(db, user).id == won.id
db.commit()
assert [p.id for p in queries.playlists_for(db, user) if p.kind == "watch_later"] == [won.id]
def test_a_first_resume_point_beside_the_watched_mark_lands_on_its_row(db, make_user, monkeypatch):
"""The first checkpoint of a video and the end-of-playback mark both read "no row"; the
checkpoint that writes second must update the mark's row, not fail on `uq_user_video`."""
user = make_user()
_videos(db, "v1")
service.mark(db, user, "v1", "watched") # the mark that won, committed
db.commit()
monkeypatch.setattr(queries, "video_state", lambda *a, **k: None) # read before it committed
service.set_progress(db, user, "v1", 90)
db.commit()
db.expire_all()
rows = db.execute(select(VideoState)).scalars().all()
assert [(r.status, r.position_seconds) for r in rows] == [("watched", 90)]
+383
View File
@@ -0,0 +1,383 @@
"""Writes driven through HTTP, each read back: the handler called its unit and committed.
A route's helper only flushes; the handler commits. The `db` fixture fails a test whose request left
a write uncommitted, so every test here drives one successful write through its route and then
reads the result back. The read-back holds that the handler called its unit; it runs in the
request's own session, so it sees a write that was never committed — the commit is the fixture's
to hold (a test that must see it committed reads from a session of its own). The services' own
tests hold what each write does; these hold that the route finishes it. Nothing here reaches
YouTube or Plex: the YouTube client is a stub and no user has a Plex link. Needs the DB.
"""
from datetime import datetime, timedelta, timezone
import pytest
from sqlalchemy import select
from app import quota
from app.auth import require_write_scope
from app.catalog.models import Channel, ExploredChannel, SearchFind, Subscription, Video
from app.downloads.models import DownloadJob, DownloadProfile, DownloadShare
from app.plex.models import PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState
from app.db import SessionLocal
from app.models import Notification
from app.notifications import READ_SOFT_CAP
from app.reader.models import SavedView, Tag, VideoState
from app.routes import channels
NOW = datetime.now(timezone.utc)
def _video(db, vid="v1", **kw):
if db.get(Channel, "UCr") is None:
db.add(Channel(id="UCr", title="R"))
db.flush()
db.add(Video(id=vid, channel_id="UCr", published_at=NOW, duration_seconds=60, **kw))
db.commit()
# --- tags ------------------------------------------------------------------------------------
def test_a_tag_made_through_the_route_is_listed_back(db, make_user, as_user):
user = make_user()
client = as_user(user)
r = client.post("/api/tags", json={"name": "Science"})
assert r.status_code == 200, r.text
assert [t["name"] for t in client.get("/api/tags").json()] == ["Science"]
def test_a_tag_renamed_through_the_route_is_listed_back(db, make_user, as_user):
user = make_user()
tag = Tag(user_id=user.id, name="Old")
db.add(tag)
db.commit()
client = as_user(user)
assert client.patch(f"/api/tags/{tag.id}", json={"name": "New"}).status_code == 200
assert [t["name"] for t in client.get("/api/tags").json()] == ["New"]
def test_a_tag_deleted_through_the_route_is_gone(db, make_user, as_user):
user = make_user()
tag = Tag(user_id=user.id, name="Gone")
db.add(tag)
db.commit()
client = as_user(user)
assert client.delete(f"/api/tags/{tag.id}").json() == {"deleted": tag.id}
assert client.get("/api/tags").json() == []
# --- watch state ------------------------------------------------------------------------------
def _state(db, user, vid="v1"):
db.expire_all()
return db.execute(
select(VideoState).where(VideoState.user_id == user.id, VideoState.video_id == vid)
).scalar_one_or_none()
def test_a_video_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
user = make_user()
_video(db)
r = as_user(user).post("/api/videos/v1/state", json={"status": "watched"})
assert r.status_code == 200, r.text
assert _state(db, user).status == "watched"
def test_a_video_unmarked_through_the_route_stays_unmarked(db, make_user, as_user):
# "new" is its own branch of the handler, with its own commit.
user = make_user()
_video(db)
db.add(VideoState(user_id=user.id, video_id="v1", status="watched"))
db.commit()
r = as_user(user).post("/api/videos/v1/state", json={"status": "new"})
assert r.status_code == 200, r.text
assert _state(db, user) is None
def test_a_video_state_cleared_through_the_route_is_gone(db, make_user, as_user):
user = make_user()
_video(db)
db.add(VideoState(user_id=user.id, video_id="v1", status="hidden", position_seconds=90))
db.commit()
assert as_user(user).delete("/api/videos/v1/state").status_code == 200
assert _state(db, user) is None
# --- saved views ------------------------------------------------------------------------------
def test_views_reordered_through_the_route_list_back_in_that_order(db, make_user, as_user):
user = make_user()
a = SavedView(user_id=user.id, source="feed", name="A", filters={}, position=0)
b = SavedView(user_id=user.id, source="feed", name="B", filters={}, position=1)
db.add_all([a, b])
db.commit()
client = as_user(user)
assert client.post("/api/saved-views/reorder", json={"ids": [b.id, a.id]}).status_code == 200
listed = client.get("/api/saved-views").json()
assert [(v["name"], v["position"]) for v in listed] == [("B", 0), ("A", 1)]
# --- download profiles and shares -------------------------------------------------------------
_SPEC = {"format": "best"}
def _profile(db, user, name="Mine"):
p = DownloadProfile(user_id=user.id, name=name, spec=_SPEC, is_builtin=False, sort_order=101)
db.add(p)
db.commit()
return p
def _profile_names(client):
return [p["name"] for p in client.get("/api/downloads/profiles").json()]
def test_a_profile_made_through_the_route_is_listed_back(db, make_user, as_user):
user = make_user()
client = as_user(user)
r = client.post("/api/downloads/profiles", json={"name": "Phone", "spec": _SPEC})
assert r.status_code == 200, r.text
assert _profile_names(client) == ["Phone"]
def test_a_profile_renamed_through_the_route_is_listed_back(db, make_user, as_user):
user = make_user()
p = _profile(db, user)
client = as_user(user)
r = client.patch(f"/api/downloads/profiles/{p.id}", json={"name": "Renamed"})
assert r.status_code == 200, r.text
assert _profile_names(client) == ["Renamed"]
def test_a_profile_deleted_through_the_route_is_gone(db, make_user, as_user):
user = make_user()
p = _profile(db, user)
client = as_user(user)
assert client.delete(f"/api/downloads/profiles/{p.id}").json() == {"deleted": p.id}
assert _profile_names(client) == []
def test_a_share_taken_back_through_the_route_is_gone(db, make_user, as_user):
owner, friend = make_user(), make_user()
job = DownloadJob(
user_id=owner.id, source_kind="youtube", source_ref="dQw4w9WgXcQ", profile_snapshot={}
)
db.add(job)
db.flush()
db.add(DownloadShare(job_id=job.id, from_user_id=owner.id, to_user_id=friend.id))
db.commit()
r = as_user(owner).delete(f"/api/downloads/{job.id}/share/{friend.email}")
assert r.json() == {"ok": True}
db.expire_all()
assert db.execute(select(DownloadShare)).first() is None
# --- channels ---------------------------------------------------------------------------------
class _YouTube:
"""The YouTube client, answering every write the channel routes make."""
def __init__(self, db, user):
pass
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def insert_subscription(self, channel_id):
return "yt-sub-1"
def delete_subscription(self, sub_id):
return None
def get_channels(self, ids):
return []
@pytest.fixture
def youtube(monkeypatch):
monkeypatch.setattr(channels, "YouTubeClient", _YouTube)
monkeypatch.setattr(quota, "backfill_reserve_reached", lambda: False)
def _channel(db, cid="UCc", **kw):
ch = Channel(id=cid, title="C", details_synced_at=NOW, **kw)
db.add(ch)
db.commit()
return ch
def _subscription(db, user, cid="UCc"):
db.expire_all()
return db.execute(
select(Subscription).where(Subscription.user_id == user.id, Subscription.channel_id == cid)
).scalar_one_or_none()
def test_a_subscribe_through_the_route_is_recorded(db, make_user, as_user, override_dep, youtube):
user = make_user()
_channel(db)
client = as_user(user)
override_dep(require_write_scope, lambda: user)
assert client.post("/api/channels/UCc/subscribe").json() == {"ok": True, "subscribed": "UCc"}
assert _subscription(db, user).yt_subscription_id == "yt-sub-1"
def test_an_explore_through_the_route_is_recorded(db, make_user, as_user, youtube, monkeypatch):
user = make_user()
_channel(db)
# The ingest pages YouTube; what the route itself records is the exploration.
monkeypatch.setattr(channels, "details_stale", lambda ch, now: False)
monkeypatch.setattr(
channels,
"explore_ingest_page",
lambda db, yt, ch, token: {"ingested": 0, "next_page_token": None},
)
r = as_user(user).post("/api/channels/UCc/explore")
assert r.status_code == 200, r.text
db.expire_all()
assert db.execute(select(ExploredChannel.user_id)).scalars().all() == [user.id]
assert db.get(Channel, "UCc").from_explore is True
def test_an_unsubscribe_through_the_route_drops_the_subscription(
db, make_user, as_user, override_dep, youtube
):
user = make_user()
_channel(db)
db.add(Subscription(user_id=user.id, channel_id="UCc", yt_subscription_id="yt-sub-1"))
db.commit()
client = as_user(user)
override_dep(require_write_scope, lambda: user)
r = client.delete("/api/channels/UCc/subscription")
assert r.json() == {"ok": True, "unsubscribed": "UCc"}
assert _subscription(db, user) is None
def test_a_backfill_reset_through_the_route_is_kept(db, make_user, as_user, monkeypatch):
admin, reader = make_user(role="admin"), make_user()
_channel(db, backfill_done=True, backfill_cursor="page-9", recent_synced_at=NOW)
db.add(Subscription(user_id=reader.id, channel_id="UCc", deep_requested=False))
db.commit()
# The recent pull afterwards reaches YouTube; the reset is the route's own write.
monkeypatch.setattr(channels, "run_recent_backfill", lambda *a, **kw: None)
r = as_user(admin).post("/api/channels/UCc/reset-backfill")
assert r.status_code == 200, r.text
db.expire_all()
ch = db.get(Channel, "UCc")
assert (ch.backfill_done, ch.backfill_cursor, ch.recent_synced_at) == (False, None, None)
assert _subscription(db, reader).deep_requested is True
# --- live search ------------------------------------------------------------------------------
def test_search_results_cleared_through_the_route_are_gone(db, make_user, as_user):
user = make_user()
_video(db, "s1", via_search=True)
db.add(SearchFind(user_id=user.id, video_id="s1"))
db.commit()
r = as_user(user).post("/api/search/clear", json={"video_ids": ["s1"]})
assert r.json() == {"ok": True, "removed": 1}
db.expire_all()
assert db.get(Video, "s1") is None
assert db.execute(select(SearchFind)).first() is None
# --- Plex show and season marks ---------------------------------------------------------------
def _show(db):
lib = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
db.add(lib)
db.flush()
sh = PlexShow(rating_key="sh1", library_id=lib.id, title="Show")
db.add(sh)
db.flush()
se = PlexSeason(rating_key="se1", show_id=sh.id, season_number=1)
db.add(se)
db.flush()
for n in (1, 2):
db.add(
PlexItem(
rating_key=f"e{n}",
library_id=lib.id,
kind="episode",
title=f"E{n}",
show_id=sh.id,
season_id=se.id,
season_number=1,
episode_number=n,
)
)
db.commit()
def _plex_marks(db, user):
db.expire_all()
return sorted(
db.execute(
select(PlexItem.rating_key, PlexState.status)
.join(PlexState, PlexState.item_id == PlexItem.id)
.where(PlexState.user_id == user.id)
).all()
)
def test_a_show_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
user = make_user()
_show(db)
r = as_user(user).post("/api/plex/show/sh1/state", json={"watched": True})
assert r.json() == {"ok": True, "changed": 2, "watched": True}
assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
def test_a_season_marked_watched_through_the_route_stays_watched(db, make_user, as_user):
user = make_user()
_show(db)
r = as_user(user).post("/api/plex/season/se1/state", json={"watched": True})
assert r.json() == {"ok": True, "changed": 2, "watched": True}
assert _plex_marks(db, user) == [("e1", "watched"), ("e2", "watched")]
def _read_past_the_cap(db, user) -> list[Notification]:
"""READ_SOFT_CAP read notices, oldest first, and one unread notice after them."""
base = datetime(2024, 1, 1, tzinfo=timezone.utc)
rows = [
Notification(
user_id=user.id,
type="t",
title=f"n{i}",
read=i < READ_SOFT_CAP,
created_at=base + timedelta(minutes=i),
)
for i in range(READ_SOFT_CAP + 1)
]
db.add_all(rows)
db.commit()
return rows
def _committed_notices(user) -> tuple[int, bool]:
"""What another session sees: how many of the user's notices are left, and whether the oldest is."""
with SessionLocal() as s:
ids = s.scalars(select(Notification.title).where(Notification.user_id == user.id)).all()
return len(ids), "n0" in ids
@pytest.mark.parametrize("path", ["{id}/read", "read_all"])
def test_reading_a_notice_past_the_cap_commits_the_trim(db, make_user, as_user, path):
"""Marking a notice read makes READ_SOFT_CAP + 1 read ones: the handler commits the read flag and
the trim of the oldest together, and the trim counts the notice just read."""
user = make_user()
rows = _read_past_the_cap(db, user)
r = as_user(user).post("/api/me/notifications/" + path.format(id=rows[-1].id))
assert r.json() == {"ok": True}
assert _committed_notices(user) == (READ_SOFT_CAP, False)
+191 -65
View File
@@ -5,17 +5,14 @@ What it may not do is build SQL, call the session's data methods, change a model
import a unit's tables: when a route did those, ownership, the commit and the conditional `WHERE`
lived in a different place in each handler, and a fix in one never reached its siblings.
The rule is switched on unit by unit. `ALLOWED` pins what each HTTP file still does today, and the
pin is exact: a count that rises fails (new database work in a route), and one that falls fails
until its entry is lowered (the fence only tightens, as `baseline.json` does for basedpyright). An
empty `ALLOWED` is the end state — a file listed there is a debt, not an exemption.
The rule has no exemptions: a single `select()` in a unit's route file fails it.
A core route (identity, admin, ops) is not a unit's HTTP layer and may use the session; it may
still not reach into a unit's tables. A route may NAME a unit's model as a type (under
`TYPE_CHECKING`), never use one as a value.
Below the HTTP layer, a fenced unit's own code keeps to its edges (`FENCED`, `READS`): see the
second half of this file.
Below the HTTP layer, a unit's own code keeps to its edges (`READS`): see the second half of this
file. Whose a row is, every module says through `userscope` — a core one too.
"""
import ast
@@ -107,10 +104,6 @@ UNIT_ROUTES = {
"plex/": "plex",
}
# What each HTTP file still does, by file: the debt a unit's sprint pays off. Exact counts. A core
# route's count is only the unit tables it imports; it is paid by the sprint of the unit it reaches.
ALLOWED: dict[str, int] = {}
# The session methods that read or write data. `commit` and `rollback` stay with the handler.
SESSION_DATA = {
"add",
@@ -257,19 +250,13 @@ def test_every_listed_route_exists():
rels = [rel for rel, _ in _http_files()]
for key in UNIT_ROUTES:
assert any(rel == key or (key.endswith("/") and rel.startswith(key)) for rel in rels), key
assert set(ALLOWED) <= set(rels), set(ALLOWED) - set(rels)
def test_no_http_file_does_database_work_beyond_its_pin():
drift = {}
for rel, src in _http_files():
found = violations(src, _unit_of(rel))
pinned = ALLOWED.get(rel, 0)
if len(found) != pinned:
drift[rel] = (pinned, len(found), found if len(found) > pinned else "lower its pin")
assert drift == {}, (
"an HTTP file's database work moved off its pin (pinned, now, what): move new work into"
" the unit's queries/service; when a count fell, lower ALLOWED to it"
def test_no_http_file_does_database_work():
hits = {rel: found for rel, src in _http_files() if (found := violations(src, _unit_of(rel)))}
assert hits == {}, (
"a route does database work: reads go through the unit's queries, writes through its"
" service; a core route reaches a unit's tables through that unit's queries too"
)
@@ -307,11 +294,11 @@ def test_the_detector_sees_each_kind_of_database_work():
# --- inside a unit: its package --------------------------------------------------------------
# A unit whose sprint has landed is fenced beyond its HTTP layer too. Its own code reads another
# unit's tables only along a declared edge (the feed joins Video with VideoState in one query; a
# boundary that split it would cost the speed the app is for), and writes another unit's tables
# never — that goes through the owner's service. Whose a row is, it says through `userscope`, and
# it commits only where a job owns its session.
# Every unit is fenced beyond its HTTP layer too. Its own code reads another unit's tables only
# along a declared edge (the feed joins Video with VideoState in one query; a boundary that split
# it would cost the speed the app is for), and writes another unit's tables never — that goes
# through the owner's service. It commits only where a job owns its session. Whose a row is, it
# says through `userscope`, and so does core code: the owner filter is checked in every module.
# Each unit's code outside the HTTP layer: its package, and the entry point that runs its jobs.
UNIT_PACKAGES = {
@@ -341,9 +328,7 @@ UNIT_PACKAGES = {
# pages' reads and writes.
"plex": ("plex/",),
}
# The units fenced so far; the rest join with their sprint.
FENCED = {"capture", "reader", "catalog", "plex"}
# Whose tables a unit's own code may READ. Core code reads a fenced unit's tables through its queries.
# Whose tables a unit's own code may READ. Core code reads a unit's tables through its queries.
READS = {
"capture": {"catalog", "plex"},
"reader": {"catalog", "plex"},
@@ -352,7 +337,7 @@ READS = {
}
# The columns that say whose a row is.
OWNER_COLUMNS = {"user_id", "to_user_id", "from_user_id", "created_by"}
# Where a fenced unit's code may commit: the jobs that own their session (the worker and what it
# Where a unit's code may commit: the jobs that own their session (the worker and what it
# runs, the GC, the backfill — none of them inside a request), and the one write that must commit
# where it stands: an upload's rows and the file it placed on disk land together or not at all, so
# a failed commit can remove the file (`upload.ingest_upload`). Everything else flushes; the handler
@@ -423,7 +408,38 @@ UNIT_COMMITS = {
"plex/watch_sync.py:run_plex_watch_reconcile",
},
}
# Where core code may commit, each with why. Everything else in core flushes and leaves the commit
# to its caller, as a unit's code does; a route's own helpers are held by test_route_conventions.
CORE_COMMITS = {
# Quota spend and a background run's row are durable whatever the caller does next: each opens
# a session of its own.
"quota.py",
"jobruns.py",
# The scheduler's jobs own their session.
"scheduler.py",
# The instance's one settings row: created on first read, and the setup steps that finish it.
"state.py",
# A token's last-used stamp lands ahead of the route: a read-only route never commits, and a
# write route's rollback must not take it along. The expiry warning is a scheduled job.
"apitokens.py:_token_caller",
"apitokens.py:warn_expiring",
# An account's erasure is one operation shared by the self-service and the admin delete; the
# Google revoke and the confirmation run after it, in the background.
"auth.py:purge_user",
# The undelivered-email notice opens its own session (its callers are often background tasks).
"email.py:_alert_admins",
# One notice, committed unless its caller batches (`commit=False`).
"notifications.py:create_notification",
# A refreshed (or dead) YouTube grant must persist even when the call that needed it fails and
# its caller rolls back.
"youtube/client.py:_access_token",
"youtube/client.py:_drop_dead_grant",
}
SQL_WRITES = {"insert", "update", "delete", "pg_insert"}
ALL_MODEL_NAMES = set(UNIT_MODEL_NAMES) | CORE_MODELS
# The column methods that compare an owner column as a filter does (`Model.user_id.in_(ids)`).
# `.is_(None)` is not one: it asks for the rows nobody owns (the system's own tags).
OWNER_METHODS = {"in_", "not_in", "notin_", "__eq__", "__ne__"}
def _package_unit(rel: str) -> str | None:
@@ -455,6 +471,64 @@ def _imported_models(tree: ast.AST) -> dict[str, str]:
return out
def _model_aliases(tree: ast.AST) -> dict[str, str]:
"""Every other name a module binds to a model, by that name: an import under another name, and
`x = aliased(Model)` — an alias is the same table, and its owner column the same filter."""
out = {}
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and "models" in (node.module or ""):
for alias in node.names:
if alias.name in ALL_MODEL_NAMES:
out[alias.asname or alias.name] = alias.name
for node in ast.walk(tree):
if (
isinstance(node, ast.Assign)
and isinstance(node.value, ast.Call)
and _called_name(node.value) == "aliased"
and node.value.args
and isinstance(node.value.args[0], ast.Name)
):
source = node.value.args[0].id
model = out.get(source, source if source in ALL_MODEL_NAMES else None)
for target in node.targets:
if model and isinstance(target, ast.Name):
out[target.id] = model
return out
def _called_name(call: ast.Call) -> str | None:
func = call.func
return func.id if isinstance(func, ast.Name) else getattr(func, "attr", None)
def _sql_write_names(tree: ast.AST) -> tuple[set[str], set[str]]:
"""The names a module calls SQLAlchemy's statement writes by — `insert`/`update`/`delete` as
imported, under any alias — and the names it binds SQLAlchemy's modules to (`sa.update(...)`)."""
functions, modules = set(SQL_WRITES), set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and (node.module or "").split(".")[0] == "sqlalchemy":
for alias in node.names:
if alias.name in {"insert", "update", "delete"}:
functions.add(alias.asname or alias.name)
elif alias.name in {"dialects", "postgresql", "sql"}:
modules.add(alias.asname or alias.name)
elif isinstance(node, ast.Import):
for alias in node.names:
if alias.name.split(".")[0] == "sqlalchemy":
modules.add(alias.asname or alias.name.split(".")[0])
return functions, modules
def _written_table(call: ast.Call) -> str | None:
"""The name of what a statement write's first argument names: `Model`, or `Model.__table__`."""
if not call.args:
return None
arg = call.args[0]
if isinstance(arg, ast.Attribute) and arg.attr == "__table__":
arg = arg.value
return arg.id if isinstance(arg, ast.Name) else None
def _enclosing_functions(tree: ast.AST) -> dict[ast.AST, str]:
out = {}
for fn in ast.walk(tree):
@@ -465,64 +539,82 @@ def _enclosing_functions(tree: ast.AST) -> dict[ast.AST, str]:
def package_violations(rel: str, source: str) -> list[str]:
"""What a module outside the HTTP layer does against a fenced unit's boundary, as `line: what`."""
"""What a module outside the HTTP layer does against a unit's boundary, as `line: what`."""
tree = ast.parse(source)
mine = _package_unit(rel)
found = []
models = _imported_models(tree)
for name in models.values():
theirs = UNIT_MODEL_NAMES[name]
if theirs == mine or (mine not in FENCED and theirs not in FENCED):
if theirs == mine:
continue
if theirs not in READS.get(mine or "", set()):
found.append(f"0: {mine or 'core'} imports {theirs}'s {name}")
write_functions, sql_modules = _sql_write_names(tree)
for node in ast.walk(tree):
if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Name):
if not isinstance(node, ast.Call):
continue
func = node.func
written = None
if node.func.id in models:
written = models[node.func.id]
elif node.func.id in SQL_WRITES and node.args and isinstance(node.args[0], ast.Name):
written = models.get(node.args[0].id)
if isinstance(func, ast.Name) and func.id in models:
written = models[func.id]
elif (isinstance(func, ast.Name) and func.id in write_functions) or (
isinstance(func, ast.Attribute)
and func.attr in {"insert", "update", "delete"}
and isinstance(_root(func), ast.Name)
and _root(func).id in sql_modules
):
written = models.get(_written_table(node) or "")
if written is None:
continue
theirs = UNIT_MODEL_NAMES[written]
if theirs != mine and (mine in FENCED or theirs in FENCED):
if theirs != mine:
found.append(f"{node.lineno}: {mine or 'core'} writes {theirs}'s {written}")
if mine in FENCED:
found += _owner_filters(tree)
functions = _enclosing_functions(tree)
allowed = UNIT_COMMITS.get(mine, set())
for node in ast.walk(tree):
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "commit"
and rel not in allowed
and f"{rel}:{functions.get(node)}" not in allowed
):
found.append(f"{node.lineno}: commits in {functions.get(node)}")
found += _owner_filters(tree)
functions = _enclosing_functions(tree)
allowed = UNIT_COMMITS.get(mine, set()) if mine is not None else CORE_COMMITS
for node in ast.walk(tree):
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "commit"
and rel not in allowed
and f"{rel}:{functions.get(node)}" not in allowed
):
found.append(f"{node.lineno}: commits in {functions.get(node)}")
return sorted(found, key=lambda v: int(v.split(":")[0]))
def _owner_filters(tree: ast.AST) -> list[str]:
"""A comparison on a table's owner column written inline (`Model.user_id == …`), not through
`userscope.owned_by`."""
names = {name: name for name in ALL_MODEL_NAMES} | _model_aliases(tree)
def owner_column(node: ast.expr) -> bool:
return (
isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id in names
and node.attr in OWNER_COLUMNS
)
found = []
for node in ast.walk(tree):
if isinstance(node, ast.Compare):
for side in (node.left, *node.comparators):
if (
isinstance(side, ast.Attribute)
and isinstance(side.value, ast.Name)
and side.value.id in set(UNIT_MODEL_NAMES) | CORE_MODELS
and side.attr in OWNER_COLUMNS
):
if owner_column(side):
found.append(f"{node.lineno}: filters {side.value.id}.{side.attr} inline")
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
if node.func.attr in OWNER_METHODS and owner_column(node.func.value):
found.append(f"{node.lineno}: filters an owner column with {node.func.attr}")
elif node.func.attr == "filter_by" and any(
k.arg in OWNER_COLUMNS for k in node.keywords
):
found.append(f"{node.lineno}: filters an owner column with filter_by")
return found
def test_a_fenced_units_code_keeps_to_its_edges():
def test_a_units_code_keeps_to_its_edges():
hits = {rel: found for rel, src in _app_files() if (found := package_violations(rel, src))}
assert hits == {}, (
"read another unit's tables only along READS, write them only through their service, say"
@@ -530,12 +622,8 @@ def test_a_fenced_units_code_keeps_to_its_edges():
)
def test_a_fenced_units_routes_filter_owners_through_userscope():
hits = {
rel: found
for rel, src in _http_files()
if _unit_of(rel) in FENCED and (found := _owner_filters(ast.parse(src)))
}
def test_every_route_filters_owners_through_userscope():
hits = {rel: found for rel, src in _http_files() if (found := _owner_filters(ast.parse(src)))}
assert hits == {}
@@ -543,11 +631,32 @@ def test_every_unit_package_exists():
for prefixes in UNIT_PACKAGES.values():
for p in prefixes:
assert (APP / p).exists(), p
for commits in UNIT_COMMITS.values():
for commits in (*UNIT_COMMITS.values(), CORE_COMMITS):
for entry in commits:
assert (APP / entry.split(":")[0]).exists(), entry
def test_every_listed_commit_is_still_there():
"""An entry names a commit that exists: one left behind by a rename, or by a function that no
longer commits, would go on allowing a commit to whatever takes its name next."""
stale = []
for commits in (*UNIT_COMMITS.values(), CORE_COMMITS):
for entry in commits:
rel, _, name = entry.partition(":")
tree = ast.parse((APP / rel).read_text(encoding="utf-8"))
functions = _enclosing_functions(tree)
committing = {
functions.get(node)
for node in ast.walk(tree)
if isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "commit"
}
if (name not in committing) if name else not committing:
stale.append(entry)
assert stale == []
PACKAGE_CASES = {
# (module path, source): violations
("downloads/x.py", "from app.models import Video"): 0, # capture reads catalog
@@ -562,6 +671,23 @@ PACKAGE_CASES = {
("sync/videos.py", "from app.models import Playlist"): 1, # no catalog→reader edge
("downloads/x.py", "q = DownloadJob.user_id == 1"): 1,
("downloads/x.py", "q = owned_by(DownloadJob.user_id, 1)"): 0,
("quota.py", "q = QuotaEvent.user_id == 1"): 1, # core too
("downloads/x.py", "from app.downloads.models import DownloadJob as J\nq = J.user_id == 1"): 1,
("downloads/x.py", "st = aliased(VideoState)\nq = st.user_id == 1"): 1,
("downloads/x.py", "q = DownloadJob.user_id.in_([1])"): 1,
("downloads/x.py", "q = select(DownloadJob).filter_by(user_id=1)"): 1,
("downloads/x.py", "q = Tag.user_id.is_(None)"): 0, # the rows nobody owns
(
"downloads/x.py",
"from sqlalchemy import update as up\nfrom app.models import Video\nup(Video)",
): 1,
(
"downloads/x.py",
"import sqlalchemy as sa\nfrom app.models import Video\nsa.update(Video)",
): 1,
("downloads/x.py", "from app.models import Video\nupdate(Video.__table__)"): 1,
("notifications.py", "def f(db):\n db.commit()"): 1, # core flushes too
("state.py", "def f(db):\n db.commit()"): 0,
("downloads/x.py", "def f(db):\n db.commit()"): 1,
("downloads/gc.py", "def f(db):\n db.commit()"): 0,
("downloads/upload.py", "def ingest_upload(db):\n db.commit()"): 0,
-1
View File
@@ -113,7 +113,6 @@ project checks needs a row.
| CVE gate | HIGH/CRITICAL with a fix available only | `--ignore-unfixed` + `--severity HIGH,CRITICAL`: a finding with no fix cannot be acted on by an upgrade; it shows in the scan output. A fixable one fails the publish, and the only way past it is `.trivyignore.yaml`. | 0 open (E32 S1) |
| npm audit | `--omit=dev --audit-level=high` | What ships to the browser, at high or above. The dev toolchain's advisories were burned down by hand in E32 S1 (18 → 0). | 0 |
| knip | `entry` — one file added | Not a narrowing: knip's Vite plugin reads entries from index.html only, so the shared link's page (`watch.html` → `src/watchMain.tsx`) is named here, or the lane reports that whole entry unused. It adds files to the lane, never takes any out. | 1 entry |
| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) → 354 in 9 (S3: Catalog 0, fenced; core routes 0) → 178 in 4 (S4a: Plex browse, detail, people, collections 0) → 0 (S4b: Plex 0, fenced — every unit fenced) |
| React Compiler | the functions it skips, pinned | `components/compilerCoverage.contract.test.ts` — the list may only shrink. End state: only VirtualGrid (TanStack Virtual, an IncompatibleLibrary) remains. | 93 (E32 S1) → 29 (E32 S3) → 1, VirtualGrid (E32 S4) |
## Accepted findings
@@ -1,4 +1,15 @@
[
{
"version": "0.108.1",
"date": "2026-10-20",
"new": [
"Development: the backend is split into four units (downloads, reader, catalog, Plex), each owning its tables. A contract test holds what each one reads and writes, and that every query names whose rows it asks for."
],
"fixed": [
"Watch later: saving the first two videos at once no longer makes two lists. An existing second list is merged into the first, its videos and its comfort-shelf pin with it.",
"Playback: the first resume point of a video or a Plex item no longer fails when it is saved together with the watched mark, a whole season marked at once, or a Plex sync."
]
},
{
"version": "0.108.0",
"date": "2026-10-18",
@@ -1,4 +1,15 @@
[
{
"version": "0.108.1",
"date": "2026-10-20",
"new": [
"Fejlesztés: a backend négy egységre bomlik (letöltések, olvasó, katalógus, Plex), mindegyik a saját tábláival. Egy szerződésteszt őrzi, melyik mit olvas és ír, és hogy minden lekérdezés megnevezi, kinek a sorait kéri."
],
"fixed": [
"Megnézem később: ha az első két videót egyszerre mented, már nem lesz két lista. Egy meglévő második lista beolvad az elsőbe, a videóival és a komfort polcos kitűzésével együtt.",
"Lejátszás: egy videó vagy Plex-elem első folytatási pontja már nem hiúsul meg, ha egyszerre ment a megnézett jelöléssel, egy egész évad jelölésével vagy egy Plex-szinkronnal."
]
},
{
"version": "0.108.0",
"date": "2026-10-18",
+1
View File
@@ -24,6 +24,7 @@ sys.path.insert(0, "/app")
from sqlalchemy import delete, func, select
from app.db import SessionLocal
import app.tables # noqa: F401 — every table, before the first query
from app.catalog.models import Channel, Subscription, Video
from app.models import AppState, User
from app.reader.models import Playlist, PlaylistItem, VideoState