diff --git a/backend/.basedpyright/baseline.json b/backend/.basedpyright/baseline.json index af17ec70..6ef8921d 100644 --- a/backend/.basedpyright/baseline.json +++ b/backend/.basedpyright/baseline.json @@ -692,48 +692,6 @@ } } ], - "./app/routes/downloads.py": [ - { - "code": "reportOptionalMemberAccess", - "range": { - "startColumn": 17, - "endColumn": 19, - "lineCount": 1 - } - }, - { - "code": "reportOptionalMemberAccess", - "range": { - "startColumn": 82, - "endColumn": 84, - "lineCount": 1 - } - }, - { - "code": "reportOptionalMemberAccess", - "range": { - "startColumn": 87, - "endColumn": 89, - "lineCount": 1 - } - }, - { - "code": "reportOptionalMemberAccess", - "range": { - "startColumn": 30, - "endColumn": 32, - "lineCount": 1 - } - }, - { - "code": "reportOptionalMemberAccess", - "range": { - "startColumn": 49, - "endColumn": 54, - "lineCount": 1 - } - } - ], "./app/routes/playlists.py": [ { "code": "reportOptionalOperand", diff --git a/backend/app/downloads/credentials.py b/backend/app/downloads/credentials.py index 026634e2..587af04f 100644 --- a/backend/app/downloads/credentials.py +++ b/backend/app/downloads/credentials.py @@ -35,6 +35,7 @@ from app.downloads import formats from app.downloads.adapters import _host_is from app.downloads.models import DownloadCredential, MediaAsset from app.security import decrypt_optional, encrypt +from app.userscope import owned_by log = logs.get_logger(__name__) @@ -157,7 +158,7 @@ def salt_for_enqueue(db: Session, user_id: int, url: str | None) -> str | None: def get_credential(db: Session, user_id: int, host_key: str) -> DownloadCredential | None: return db.execute( select(DownloadCredential).where( - DownloadCredential.user_id == user_id, DownloadCredential.host_key == host_key + owned_by(DownloadCredential.user_id, user_id), DownloadCredential.host_key == host_key ) ).scalar_one_or_none() @@ -214,7 +215,7 @@ def status_list(db: Session, user_id: int) -> list[dict]: have = { c.host_key: c for c in db.execute( - select(DownloadCredential).where(DownloadCredential.user_id == user_id) + select(DownloadCredential).where(owned_by(DownloadCredential.user_id, user_id)) ).scalars() } out = [] diff --git a/backend/app/downloads/folders.py b/backend/app/downloads/folders.py index a431a6bc..f829375b 100644 --- a/backend/app/downloads/folders.py +++ b/backend/app/downloads/folders.py @@ -33,6 +33,7 @@ from sqlalchemy.orm import Session from app import sysconfig from app.downloads.status import JobStatus from app.downloads.models import DownloadJob, FileFolder, MediaAsset +from app.userscope import owned_by class FolderKind(StrEnum): @@ -106,7 +107,9 @@ def source_key_for(db: Session, job: DownloadJob) -> str: def source_folder(db: Session, user_id: int, key: str) -> FileFolder: """The user's source folder for `key`, made on first use. A race on the unique (user, key) is settled by re-selecting, inside a SAVEPOINT so the caller's own pending writes survive.""" - stmt = select(FileFolder).where(FileFolder.user_id == user_id, FileFolder.source_key == key) + stmt = select(FileFolder).where( + owned_by(FileFolder.user_id, user_id), FileFolder.source_key == key + ) folder = db.execute(stmt).scalar_one_or_none() if folder is not None: return folder @@ -131,7 +134,7 @@ def hub_folder(db: Session, user_id: int) -> FileFolder: """The user's Hub folder, made on first use; a race on its one-per-user index is settled like `source_folder`'s.""" stmt = select(FileFolder).where( - FileFolder.user_id == user_id, FileFolder.kind == FolderKind.HUB + owned_by(FileFolder.user_id, user_id), FileFolder.kind == FolderKind.HUB ) folder = db.execute(stmt).scalar_one_or_none() if folder is not None: @@ -310,3 +313,28 @@ def delete_own(db: Session, folder: FileFolder) -> int: db.flush() # the files leave before the folder goes: its foreign key RESTRICTs the delete db.delete(folder) return len(jobs) + + +def create_own(db: Session, user_id: int, name: str) -> FileFolder: + """A new folder of the user's own; FolderError("file_folder_name_taken") when they already have + one by that name (the unique (user, lower(name))).""" + folder = FileFolder(user_id=user_id, kind=FolderKind.OWN, name=name) + try: + with db.begin_nested(): + db.add(folder) + db.flush() + except IntegrityError as e: + raise FolderError("file_folder_name_taken") from e + return folder + + +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: + raise FolderError("file_folder_not_own") + folder.name = name + try: + with db.begin_nested(): + db.flush() + except IntegrityError as e: + raise FolderError("file_folder_name_taken") from e diff --git a/backend/app/downloads/links.py b/backend/app/downloads/links.py index 5d46335a..54546761 100644 --- a/backend/app/downloads/links.py +++ b/backend/app/downloads/links.py @@ -15,20 +15,24 @@ grant appended to the file URL, so the raw password never rides in the media URL import hashlib import hmac import secrets -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from sqlalchemy import select +from sqlalchemy import update as sql_update from sqlalchemy.orm import Session -from app import security +from app import audit, security from app.config import settings from app.downloads import storage from app.downloads.status import JobStatus from app.downloads.models import DownloadJob, DownloadLink, FileFolder, MediaAsset +from app.models import User +from app.utils import iso +from app.userscope import owned_by _GRANT_TTL = 6 * 3600 # a watch session's grant is good for 6h, then the viewer re-unlocks -# The most files a folder link lists: as many as the owner's own Files list shows (`routes.files. -# MAX_LIST`; a test holds the two equal — this layer may not import that one). +# The most files a folder link lists: as many as the owner's own Files list shows +# (`queries.MAX_LIST`; a test holds the two equal — `queries` imports this module). FOLDER_MAX = 500 @@ -59,7 +63,7 @@ def in_folder(folder: FileFolder) -> tuple: moved out stops opening through the link at its next request; one moved in starts.""" return ( DownloadJob.folder_id == folder.id, - DownloadJob.user_id == folder.user_id, + owned_by(DownloadJob.user_id, folder.user_id), DownloadJob.status == JobStatus.DONE, ) @@ -170,3 +174,117 @@ def public_meta( "file_url": file_url, "poster_url": poster_url, } + + +# --- the owner's writes -------------------------------------------------------------------- +# Each flushes and records its audit row; the handler commits. + + +def expiry(days: int | None) -> datetime | None: + """None/0/negative → no expiry; otherwise N days out, capped at ~10 years.""" + if not days or days <= 0: + return None + return datetime.now(timezone.utc) + timedelta(days=min(days, 3650)) + + +def audit_view(link: DownloadLink) -> dict: + """A link's settings as the audit records them: whether it has a password, never which.""" + return { + "allow_download": link.allow_download, + "has_password": link.password_hash is not None, + "expires_at": iso(link.expires_at), + } + + +def create( + db: Session, + user: User, + *, + password: str, + allow_download: bool, + expires_days: int | None, + **target: int, +) -> tuple[DownloadLink, str]: + """A link to `target` (`job_id=` or `folder_id=`, already checked to be the user's), and the + token it is opened by — which is in hand only now.""" + token = new_token() + link = DownloadLink( + **target, + created_by=user.id, + allow_download=allow_download, + password_hash=security.hash_password(password) if password else None, + expires_at=expiry(expires_days), + ) + seal(link, token) + db.add(link) + db.flush() + audit.record( + db, + user.id, + audit.AuditAction.DOWNLOAD_LINK_CREATE, + target_type="download_link", + target_id=link.id, + after={**target, **audit_view(link)}, + ) + return link, token + + +def update( + db: Session, + user: User, + link: DownloadLink, + *, + allow_download: bool | None = None, + expires_days: int | None = None, + password: str | None = None, + sent: frozenset[str] = frozenset(), +) -> None: + """Change what the caller sent (`sent` names the fields, so `expires_days=None` can mean "never + expires"). Setting, clearing or changing the password bumps its version, so any grant issued + under the previous one stops working now rather than living out its TTL.""" + before = audit_view(link) + if "allow_download" in sent: + link.allow_download = bool(allow_download) + if "expires_days" in sent: + link.expires_at = expiry(expires_days) + if "password" in sent: + pw = (password or "").strip() + link.password_hash = security.hash_password(pw) if pw else None + link.password_version += 1 + after = audit_view(link) + if after != before or "password" in sent: + audit.record( + db, + user.id, + audit.AuditAction.DOWNLOAD_LINK_UPDATE, + target_type="download_link", + target_id=link.id, + before=before, + after=after, + ) + db.flush() + + +def revoke(db: Session, user: User, link: DownloadLink) -> None: + link_id = link.id + db.delete(link) + audit.record( + db, + user.id, + audit.AuditAction.DOWNLOAD_LINK_REVOKE, + target_type="download_link", + target_id=link_id, + after={"job_id": link.job_id} if link.job_id is not None else {"folder_id": link.folder_id}, + ) + db.flush() + + +def count_view(db: Session, link: DownloadLink) -> None: + """One more opening of the link. In SQL, not `+= 1` on the loaded row: two opens at once would + both write the same count.""" + db.execute( + sql_update(DownloadLink) + .where(DownloadLink.id == link.id) + .values(view_count=DownloadLink.view_count + 1) + .execution_options(synchronize_session=False) + ) diff --git a/backend/app/downloads/queries.py b/backend/app/downloads/queries.py new file mode 100644 index 00000000..edab65c0 --- /dev/null +++ b/backend/app/downloads/queries.py @@ -0,0 +1,406 @@ +"""Capture's reads: what the library, the download tray, the share dialogs and the public watch pages +show. Each returns rows or plain values and writes nothing; whose a row is goes through +`userscope.owned_by` / `owned_row`, so a row of someone else's reads as a missing one. + +The HTTP layer (`routes.downloads`, `routes.files`, `routes.public`) asks here and serializes; the +writes are the service's (`service`, `folders`, `links`, `quota`).""" + +from collections.abc import Sequence +from datetime import datetime, timezone + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from app.apitokens import FilesCaller, Right +from app.collation import collated +from app.downloads import folders, links, quota +from app.downloads.models import ( + DownloadJob, + DownloadLink, + DownloadProfile, + DownloadQuota, + DownloadShare, + FileFolder, + LibraryState, + MediaAsset, +) +from app.downloads.status import FINISHED, JobStatus +from app.models import User +from app.userscope import owned_by, owned_row +from app.utils import like_escape + +# Cap a listing so a heavy library can't return an unbounded result set (the old Download Center's +# own cap; far beyond any real library). +MAX_LIST = 500 + + +# --- jobs and their files ------------------------------------------------------------------ + + +def own_job(db: Session, user: User, job_id: int) -> DownloadJob | None: + return owned_row(db, DownloadJob, job_id, user) + + +def job(db: Session, job_id: int) -> DownloadJob | None: + """A job by id, whoever's — for a public link, which already names its job.""" + return db.get(DownloadJob, job_id) + + +def accessible_job(db: Session, user: User, job_id: int) -> DownloadJob | None: + """A file the caller may read: their own, or one shared with them.""" + row = db.get(DownloadJob, job_id) + if row is None: + return None + if row.user_id == user.id or shared_with(db, job_id, user) is not None: + return row + return None + + +def readable_job(db: Session, caller: FilesCaller, job_id: int) -> DownloadJob | None: + """A file `caller` may read: as `accessible_job` for the user; for a token, one of the user's + own in a folder it may read.""" + if caller.grants is None: + return accessible_job(db, caller.user, job_id) + row = own_job(db, caller.user, job_id) + if row is None or not caller.may(row.folder_id, Right.READ): + return None + return row + + +def asset_of(db: Session, job: DownloadJob | None) -> MediaAsset | None: + """The file under a job, if it has one yet.""" + return db.get(MediaAsset, job.asset_id) if job is not None and job.asset_id else None + + +def assets_for(db: Session, jobs: Sequence[DownloadJob]) -> dict[int, MediaAsset]: + ids = {j.asset_id for j in jobs if j.asset_id} + if not ids: + return {} + rows = db.execute(select(MediaAsset).where(MediaAsset.id.in_(ids))).scalars() + return {a.id: a for a in rows} + + +def positions_for(db: Session, user: User, jobs: Sequence[DownloadJob]) -> dict[int, int]: + """`user`'s resume point in each of `jobs` that has one (the library player's `library_states`).""" + ids = [j.id for j in jobs] + if not ids: + return {} + rows = db.execute( + select(LibraryState.job_id, LibraryState.position_seconds).where( + owned_by(LibraryState.user_id, user), LibraryState.job_id.in_(ids) + ) + ).all() + return {job_id: pos for job_id, pos in rows} + + +def source_keys(db: Session, jobs: Sequence[DownloadJob]) -> dict[int, str]: + """Where each file came from (its source folder's key), for the list's Source column, the + "back to its source folder" move and the download tray's site: a file in a source folder has its + folder's key; one moved into an own folder (or someone else's, shared) is traced back by + `folders.source_key_for`.""" + ids = {j.folder_id for j in jobs if j.folder_id is not None} + by_id = ( + {f.id: f for f in db.execute(select(FileFolder).where(FileFolder.id.in_(ids))).scalars()} + if ids + else {} + ) + out: dict[int, str] = {} + for j in jobs: + folder = by_id.get(j.folder_id) if j.folder_id is not None else None + if folder is not None and folder.kind == folders.FolderKind.SOURCE and folder.source_key: + out[j.id] = folder.source_key + else: + out[j.id] = folders.source_key_for(db, j) + return out + + +def queue(db: Session, user: User, recent: datetime) -> list[DownloadJob]: + """The download tray: what is waiting, running, paused or failed, and what finished (done or + canceled) since `recent`. A done job is timed from when it became done, never from its last + write: a rename, a move or the GC's expiry warning touches an old file's row, and the tray would + announce it as just finished.""" + return list( + db.execute( + select(DownloadJob) + .where( + owned_by(DownloadJob.user_id, user), + DownloadJob.status.not_in(FINISHED) + | ((DownloadJob.status == JobStatus.DONE) & (DownloadJob.ready_at >= recent)) + | ((DownloadJob.status == JobStatus.CANCELED) & (DownloadJob.updated_at >= recent)), + ) + .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()) + .limit(MAX_LIST) + ).scalars() + ) + + +def statuses_by_source(db: Session, user: User, statuses: Sequence[str]) -> list[tuple[str, str]]: + """(source_ref, status) of each of `user`'s jobs in one of `statuses`.""" + return list( + db.execute( + select(DownloadJob.source_ref, DownloadJob.status).where( + owned_by(DownloadJob.user_id, user), DownloadJob.status.in_(tuple(statuses)) + ) + ).tuples() + ) + + +def library( + db: Session, + user: User, + *, + folder_id: int | None, + listable: set[int] | None, + shared: bool, + term: str, + sort: str, + order: str, + lang: str, +) -> list[DownloadJob]: + """The caller's done files (optionally one folder's, or only in the folders a token may list), + or the ones shared with them; searched by name and channel, sorted by when they were added, by + name (in the reader's alphabet) or by size. A file is added to its owner's list when it is done, + not when it was asked for (a download that failed for days and then went through is new); one + shared with the caller, when it was shared.""" + stmt = ( + select(DownloadJob) + .join(MediaAsset, MediaAsset.id == DownloadJob.asset_id) + .where(DownloadJob.status == JobStatus.DONE) + ) + if shared: + stmt = stmt.join(DownloadShare, DownloadShare.job_id == DownloadJob.id).where( + owned_by(DownloadShare.to_user_id, user) + ) + else: + stmt = stmt.where(owned_by(DownloadJob.user_id, user)) + if folder_id is not None: + stmt = stmt.where(DownloadJob.folder_id == folder_id) + elif listable is not None: + stmt = stmt.where(DownloadJob.folder_id.in_(listable)) + name = func.coalesce(DownloadJob.display_name, MediaAsset.title, "") + if term: + like = f"%{like_escape(term)}%" + stmt = stmt.where( + or_( + name.ilike(like), + func.coalesce(DownloadJob.display_uploader, MediaAsset.uploader, "").ilike(like), + ) + ) + key = { + "added": DownloadShare.created_at if shared else DownloadJob.ready_at, + "name": collated(name, lang), + "size": MediaAsset.size_bytes, + }[sort] + first = key.asc() if order == "asc" else key.desc() + tie = DownloadJob.id.asc() if order == "asc" else DownloadJob.id.desc() + return list(db.execute(stmt.order_by(first.nulls_last(), tie).limit(MAX_LIST)).scalars()) + + +def recent_jobs(db: Session) -> list[DownloadJob]: + """Everyone's newest jobs, for the admin's list.""" + return list( + db.execute( + select(DownloadJob) + .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()) + .limit(MAX_LIST) + ).scalars() + ) + + +# --- shares -------------------------------------------------------------------------------- + + +def shared_with(db: Session, job_id: int, user: User | int) -> DownloadShare | None: + """The grant that hands this job to `user`, if there is one.""" + return db.execute( + select(DownloadShare).where( + DownloadShare.job_id == job_id, owned_by(DownloadShare.to_user_id, user) + ) + ).scalar_one_or_none() + + +def shared_at(db: Session, user: User, jobs: Sequence[DownloadJob]) -> dict[int, datetime]: + rows = db.execute( + select(DownloadShare.job_id, DownloadShare.created_at).where( + owned_by(DownloadShare.to_user_id, user), + DownloadShare.job_id.in_([j.id for j in jobs]), + ) + ).all() + return {job_id: at for job_id, at in rows} + + +# --- folders ------------------------------------------------------------------------------- + + +def own_folder(db: Session, user: User, folder_id: int) -> FileFolder | None: + return owned_row(db, FileFolder, folder_id, user) + + +def folders_by_id(db: Session, ids: set[int]) -> dict[int, FileFolder]: + """These folders, by id — for a reader that already knows which ones (a token's grants).""" + if not ids: + return {} + return {f.id: f for f in db.execute(select(FileFolder).where(FileFolder.id.in_(ids))).scalars()} + + +def folders_with_totals( + db: Session, user: User, listable: set[int] | None +) -> list[tuple[FileFolder, int, int]]: + """The user's folders (only `listable` ones, when given) with how many done files each holds and + their size. A source folder its files have all left stays (count 0).""" + done = (DownloadJob.folder_id == FileFolder.id) & (DownloadJob.status == JobStatus.DONE) + stmt = select( + FileFolder, + func.count(DownloadJob.id), + func.coalesce(func.sum(MediaAsset.size_bytes), 0), + ).where(owned_by(FileFolder.user_id, user)) + if listable is not None: + stmt = stmt.where(FileFolder.id.in_(listable)) + rows = db.execute( + stmt.outerjoin(DownloadJob, done) + .outerjoin(MediaAsset, MediaAsset.id == DownloadJob.asset_id) + .group_by(FileFolder.id) + .order_by(FileFolder.kind.desc(), FileFolder.source_key, FileFolder.name, FileFolder.id) + ).tuples() + return [(f, count, int(size or 0)) for f, count, size in rows] + + +def folder_totals(db: Session, folder_id: int) -> tuple[int, int]: + count, size = db.execute( + select(func.count(DownloadJob.id), func.coalesce(func.sum(MediaAsset.size_bytes), 0)) + .select_from(DownloadJob) + .outerjoin(MediaAsset, MediaAsset.id == DownloadJob.asset_id) + .where(DownloadJob.folder_id == folder_id, DownloadJob.status == JobStatus.DONE) + ).one() + return count, int(size) + + +def linked_folders(db: Session, folder_ids: list[int]) -> set[int]: + """Which of these folders a live public link opens (one not past its expiry).""" + now = datetime.now(timezone.utc) + rows = db.execute( + select(DownloadLink.folder_id).where( + DownloadLink.folder_id.in_(folder_ids), + or_(DownloadLink.expires_at.is_(None), DownloadLink.expires_at > now), + ) + ).scalars() + return {folder_id for folder_id in rows if folder_id is not None} + + +# --- links --------------------------------------------------------------------------------- + + +def own_link(db: Session, user: User, link_id: int) -> DownloadLink | None: + """A link to one of the user's files or folders.""" + link = db.get(DownloadLink, link_id) + if link is None: + return None + target = ( + db.get(DownloadJob, link.job_id) + if link.job_id is not None + else db.get(FileFolder, link.folder_id) + ) + if target is None or target.user_id != user.id: + return None + return link + + +def links_to( + db: Session, *, job_id: int | None = None, folder_id: int | None = None +) -> list[DownloadLink]: + """The links to one file or one folder, newest first.""" + target = ( + DownloadLink.job_id == job_id if job_id is not None else DownloadLink.folder_id == folder_id + ) + return list( + db.execute(select(DownloadLink).where(target).order_by(DownloadLink.id.desc())).scalars() + ) + + +def link_folder(db: Session, link: DownloadLink) -> FileFolder | None: + return db.get(FileFolder, link.folder_id) if link.folder_id is not None else None + + +def folder_job(db: Session, folder: FileFolder, job_id: int) -> DownloadJob | None: + """One of the files a link to `folder` reaches (`links.in_folder`), while it is in it.""" + return db.execute( + select(DownloadJob).where(DownloadJob.id == job_id, *links.in_folder(folder)) + ).scalar_one_or_none() + + +def owner_suspended(db: Session, job: DownloadJob) -> bool: + """The job's owner is suspended (or gone): the worker leaves their queued jobs where they are.""" + owner = db.get(User, job.user_id) + return owner is None or owner.is_suspended + + +# --- profiles ------------------------------------------------------------------------------ + + +def profiles_for(db: Session, user: User) -> list[DownloadProfile]: + """The built-in format profiles, then the user's own.""" + return list( + db.execute( + select(DownloadProfile) + .where(DownloadProfile.is_builtin.is_(True) | owned_by(DownloadProfile.user_id, user)) + .order_by( + DownloadProfile.is_builtin.desc(), DownloadProfile.sort_order, DownloadProfile.id + ) + ).scalars() + ) + + +def own_profile(db: Session, user: User, profile_id: int) -> DownloadProfile | None: + """One of the user's own profiles (a built-in is no one's to change).""" + p = owned_row(db, DownloadProfile, profile_id, user) + return None if p is None or p.is_builtin else p + + +def usable_profile(db: Session, user: User, profile_id: int) -> DownloadProfile | None: + """A profile the user may enqueue with: a built-in, or their own.""" + p = db.get(DownloadProfile, profile_id) + if p is None or (not p.is_builtin and p.user_id != user.id): + return None + return p + + +def default_profile(db: Session) -> DownloadProfile | None: + """The first built-in, for an enqueue that named no format.""" + return db.execute( + select(DownloadProfile) + .where(DownloadProfile.is_builtin.is_(True)) + .order_by(DownloadProfile.sort_order) + .limit(1) + ).scalar_one_or_none() + + +# --- storage and quota (admin) ------------------------------------------------------------- + + +def storage_totals(db: Session) -> tuple[int, int, int]: + """(files on disk, their bytes, every asset row). On disk is `quota.ON_DISK`: ready files, and + uploads waiting for or failed in their conversion — the same disk the footprint and the GC cap + count.""" + on_disk = db.execute( + select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK) + ).one() + total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar_one() + return on_disk[0], int(on_disk[1]), total_assets + + +def done_files_per_user(db: Session) -> dict[int, int]: + """How many finished files each user's Files lists.""" + return dict( + db.execute( + select(DownloadJob.user_id, func.count()) + .where(DownloadJob.status == JobStatus.DONE) + .group_by(DownloadJob.user_id) + ) + .tuples() + .all() + ) + + +def custom_quota(db: Session, user_id: int) -> DownloadQuota | None: + """The admin's own limits for this user, if any were set.""" + return db.get(DownloadQuota, user_id) diff --git a/backend/app/downloads/quota.py b/backend/app/downloads/quota.py index ef58665a..aca55ae0 100644 --- a/backend/app/downloads/quota.py +++ b/backend/app/downloads/quota.py @@ -27,6 +27,7 @@ from sqlalchemy.orm import Session from app import sysconfig from app.downloads.status import JobKind, JobStatus from app.downloads.models import DownloadJob, DownloadQuota, MediaAsset +from app.userscope import owned_by # Jobs a user is considered to be "holding" (occupy a job slot / footprint). Terminal failures # (error, canceled) don't count. @@ -85,7 +86,7 @@ def footprint(db: Session, user_id: int) -> int: held_assets = ( select(DownloadJob.asset_id) .where( - DownloadJob.user_id == user_id, + owned_by(DownloadJob.user_id, user_id), or_( DownloadJob.status.in_(_HOLDING), and_(DownloadJob.job_kind == JobKind.UPLOAD, DownloadJob.status == JobStatus.ERROR), @@ -109,7 +110,7 @@ def _count(db: Session, user_id: int, states) -> int: db.execute( select(func.count()) .select_from(DownloadJob) - .where(DownloadJob.user_id == user_id, DownloadJob.status.in_(states)) + .where(owned_by(DownloadJob.user_id, user_id), DownloadJob.status.in_(states)) ).scalar() or 0 ) @@ -201,3 +202,46 @@ def usage(db: Session, user_id: int) -> dict: "max_jobs": lim.max_jobs, "unlimited": lim.unlimited, } + + +def set_custom( + db: Session, + user_id: int, + *, + max_bytes: int | None = None, + max_concurrent: int | None = None, + max_jobs: int | None = None, + unlimited: bool | None = None, +) -> None: + """The admin's own limits for a user: starts from what applies now, then each value given (a + size floored at 0, a count at 1). The handler commits.""" + row = db.get(DownloadQuota, user_id) + if row is None: + cur = resolve(db, user_id) + row = DownloadQuota( + user_id=user_id, + max_bytes=cur.max_bytes, + max_concurrent=cur.max_concurrent, + max_jobs=cur.max_jobs, + unlimited=cur.unlimited, + ) + db.add(row) + if max_bytes is not None: + row.max_bytes = max(0, max_bytes) + if max_concurrent is not None: + row.max_concurrent = max(1, max_concurrent) + if max_jobs is not None: + row.max_jobs = max(1, max_jobs) + if unlimited is not None: + row.unlimited = unlimited + db.flush() + + +def reset_custom(db: Session, user_id: int) -> bool: + """Back to the defaults; False when the user had no limits of their own.""" + row = db.get(DownloadQuota, user_id) + if row is None: + return False + db.delete(row) + db.flush() + return True diff --git a/backend/app/downloads/service.py b/backend/app/downloads/service.py index 54169d71..f1b85e0e 100644 --- a/backend/app/downloads/service.py +++ b/backend/app/downloads/service.py @@ -1,16 +1,19 @@ -"""Enqueue + cache resolution for the download center (the per-user layer over MediaAsset). +"""Capture's writes: enqueue + cache resolution, the job transitions, and what the library's +dialogs change (profiles, resume points, shares, a file's display fields). `enqueue` is the single entry point used by the API: it resolves (or creates) the shared MediaAsset for the requested source+format, links a per-user DownloadJob to it, and — on a cache hit (asset already ready) — marks the job done instantly with no re-download. The worker -handles everything else. State transitions used by the routes (pause/resume/cancel/delete) and -the worker live in `worker.py` / the routes; this module owns enqueue + asset bookkeeping. +handles everything else. Every function here flushes and leaves the commit to its caller (the +handler); the folder, link and quota writes are `folders`, `links` and `quota`. """ import uuid from datetime import datetime, timezone +from sqlalchemy import delete as delete_rows from sqlalchemy import func, select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session @@ -21,10 +24,17 @@ from app.downloads import credentials from app.downloads import edit as editmod from app.downloads import folders, formats, mediakind, quota, storage from app.downloads.status import HOLDING, JobKind, JobStatus -from app.models import DISPLAY_TEXT_MAX, Channel, PlexItem, PlexShow, Video -from app.downloads.models import DownloadJob, MediaAsset +from app.models import DISPLAY_TEXT_MAX, Channel, PlexItem, PlexShow, User, Video +from app.downloads.models import ( + DownloadJob, + DownloadProfile, + DownloadShare, + LibraryState, + MediaAsset, +) from app.plex import lookups as plex_lookups from app.plex import paths as plex_paths +from app.userscope import owned_by, owned_row class EditError(Exception): @@ -223,8 +233,7 @@ def enqueue( folders.stamp_done(db, job) asset.last_access_at = datetime.now(timezone.utc) - db.commit() - db.refresh(job) + db.flush() return job @@ -303,8 +312,7 @@ def enqueue_edit( folders.stamp_done(db, job) asset.last_access_at = datetime.now(timezone.utc) - db.commit() - db.refresh(job) + db.flush() return job @@ -392,8 +400,7 @@ def enqueue_clip( folders.stamp_done(db, job) asset.last_access_at = datetime.now(timezone.utc) - db.commit() - db.refresh(job) + db.flush() return job @@ -501,8 +508,8 @@ def release_asset(db: Session, job: DownloadJob) -> None: # --- Job state transitions --------------------------------------------------------------------- # The single home for the pause/resume/cancel/delete/redownload transitions + their asset # bookkeeping (release, failed-asset reset, fresh-asset rebuild), so the routes are thin callers and -# the worker/GC could share the same definitions (B-sub-8). Each commits its own transition; the -# route keeps only auth + HTTP mapping + serialization. +# the worker/GC could share the same definitions (B-sub-8). Each flushes its transition and the caller +# commits it; the route keeps only auth + HTTP mapping + serialization. def pause(db: Session, job: DownloadJob) -> None: @@ -510,7 +517,7 @@ def pause(db: Session, job: DownloadJob) -> None: in any other state.""" if job.status in (JobStatus.QUEUED, JobStatus.RUNNING): job.status = JobStatus.PAUSED - db.commit() + db.flush() def resume(db: Session, job: DownloadJob) -> None: @@ -534,7 +541,7 @@ def resume(db: Session, job: DownloadJob) -> None: asset.status = "pending" asset.error = None job.status = JobStatus.QUEUED - db.commit() + db.flush() def _follow_the_cookie(db: Session, job: DownloadJob, asset: MediaAsset) -> MediaAsset: @@ -569,14 +576,14 @@ def cancel(db: Session, job: DownloadJob) -> None: if job.status not in (JobStatus.DONE, JobStatus.CANCELED): release_asset(db, job) # release while the job still counts as holding job.status = JobStatus.CANCELED - db.commit() + db.flush() def delete(db: Session, job: DownloadJob) -> None: """Delete a job, releasing its asset hold first (which frees the file iff this was the last holder).""" release_asset(db, job) db.delete(job) - db.commit() + db.flush() def redownload(db: Session, job: DownloadJob) -> None: @@ -625,4 +632,181 @@ def redownload(db: Session, job: DownloadJob) -> None: job.queue_pos = next_queue_pos(db) # tail of the queue, not its stale original slot job.expires_at = None # stamped again when the rebuild is done job.expiry_warned = False - db.commit() + db.flush() + + +# --- the library's writes ------------------------------------------------------------------ +# What the HTTP layer changes for a user: each flushes, the handler commits. + + +def lock_own_job(db: Session, user: User, job_id: int) -> DownloadJob | None: + """One of the user's jobs, re-read under a row lock, for a transition that moves its asset's ref + count (resume, cancel, delete, redownload) or writes the status those decide on (pause). Two + overlapping requests for one job run one after the other and the second decides on the first's + committed result — never both releasing the same hold, and never a pause turning a job a cancel + just released back into a holder; either can free a file another job still holds. None: not + theirs, or gone. + + The asset is locked BEFORE the job: the order a finishing worker child takes them in + (`finalize_if_running`), so the two can never deadlock.""" + while True: + owned = owned_row(db, DownloadJob, job_id, user) + if owned is None: + return None + asset_id = owned.asset_id + if asset_id is not None: + db.execute(select(MediaAsset.id).where(MediaAsset.id == asset_id).with_for_update()) + job = db.execute( + select(DownloadJob) + .where(DownloadJob.id == job_id) + .with_for_update() + .execution_options(populate_existing=True) + ).scalar_one_or_none() + if job is None: + return None + if job.asset_id == asset_id: + return job + db.rollback() # another request moved the job to another asset meanwhile: lock that one + + +def hold_for_share(db: Session, user: User, job_id: int) -> DownloadJob | None: + """One of the user's files, held (FOR SHARE) for the rest of the transaction: a delete of it at + the same moment (the user's, or the retention sweep's) waits and takes the new link with it, + rather than failing the link's foreign key — or, done first, leaves None.""" + if owned_row(db, DownloadJob, job_id, user) is None: + return None + return db.execute( + select(DownloadJob) + .where(DownloadJob.id == job_id) + .with_for_update(read=True) + .execution_options(populate_existing=True) + ).scalar_one_or_none() + + +def create_profile(db: Session, user: User, name: str, spec: dict) -> DownloadProfile: + """A format profile of the user's own, after their others.""" + maxpos = db.execute( + select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where( + owned_by(DownloadProfile.user_id, user) + ) + ).scalar_one() + p = DownloadProfile( + user_id=user.id, name=name[:80], spec=spec, is_builtin=False, sort_order=maxpos + 1 + ) + db.add(p) + db.flush() + return p + + +def update_profile(profile: DownloadProfile, *, name: str | None, spec: dict | None) -> None: + """Rename a profile and/or replace its spec; None leaves that one as it is.""" + if name is not None: + profile.name = name[:80] + if spec is not None: + profile.spec = spec + + +def delete_profile(db: Session, profile: DownloadProfile) -> None: + db.delete(profile) + db.flush() + + +def set_position(db: Session, user: User, job_id: int, position: int) -> None: + """The library player's resume point in a file the user can play; 0 clears it. + + A clear is a statement, not a loaded row's delete: the periodic checkpoint and the pagehide beacon + of one stop may both clear it, and the second then deletes nothing rather than failing. A set is + an upsert, one row per (user, job) even when two checkpoints race, not a read-then-insert that + the second would fail on the unique key.""" + if not position: + db.execute( + delete_rows(LibraryState).where( + owned_by(LibraryState.user_id, user), LibraryState.job_id == job_id + ) + ) + return + now = datetime.now(timezone.utc) + db.execute( + pg_insert(LibraryState) + .values(user_id=user.id, job_id=job_id, position_seconds=position, progress_updated_at=now) + .on_conflict_do_update( + constraint="uq_library_user_job", + set_={"position_seconds": position, "progress_updated_at": now}, + ) + ) + + +def share(db: Session, job: DownloadJob, sender: User, recipient: User) -> bool: + """Hand a file to another user, with a notice for them; False when it was already theirs.""" + from app.notifications import create_notification + + exists = db.execute( + select(DownloadShare.id).where( + DownloadShare.job_id == job.id, owned_by(DownloadShare.to_user_id, recipient) + ) + ).first() + if exists is not None: + return False + db.add(DownloadShare(job_id=job.id, from_user_id=sender.id, to_user_id=recipient.id)) + create_notification( + db, + user_id=recipient.id, + type="download_shared", + title=(job.display_name or "A download") + " was shared with you", + # The client names the notice in the reader's language from `data`; the title is the + # English fallback only. + data={ + "kind": "download_shared", + "from": sender.email, + "job_id": job.id, + "title": job.display_name, + }, + commit=False, + ) + db.flush() + return True + + +def unshare(db: Session, job_id: int, recipient: User) -> bool: + """Take a file back from one user (its owner's act, or the recipient's own dismissal); False + when it was not theirs.""" + row = db.execute( + select(DownloadShare).where( + DownloadShare.job_id == job_id, owned_by(DownloadShare.to_user_id, recipient) + ) + ).scalar_one_or_none() + if row is None: + return False + db.delete(row) + db.flush() + return True + + +# The display fields a user may set on their file; the file on disk keeps its own name. +DISPLAY_FIELDS = frozenset( + {"display_name", "display_uploader", "display_uploader_url", "extra_links"} +) + + +def set_display(job: DownloadJob, **fields: object) -> None: + """Set a file's display fields (each already cleaned by the caller); a field not given is left.""" + unknown = set(fields) - DISPLAY_FIELDS + if unknown: + raise ValueError(f"not a display field: {sorted(unknown)}") + for name, value in fields.items(): + setattr(job, name, value) + + +def touch(asset: MediaAsset) -> None: + """The file was just served: the cache's idle clock starts again.""" + asset.last_access_at = datetime.now(timezone.utc) + + +def store_sha256(db: Session, asset_id: int, digest: str) -> None: + """Keep a file's hash once it is known — only where none is, so a concurrent hash of the same + file never overwrites the first.""" + db.execute( + update(MediaAsset) + .where(MediaAsset.id == asset_id, MediaAsset.sha256.is_(None)) + .values(sha256=digest) + ) diff --git a/backend/app/routes/_common.py b/backend/app/routes/_common.py index 740f3f2d..567a2713 100644 --- a/backend/app/routes/_common.py +++ b/backend/app/routes/_common.py @@ -16,6 +16,7 @@ from app.db import Base from app.models import POSITION_MAX_S, User from app.issues import gitea from app.refusal import refused +from app.userscope import owned_row from app.youtube.client import YouTubeError, refusal_reason M = TypeVar("M", bound=Base) @@ -90,8 +91,8 @@ def owned_or_404( 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.""" - obj = get_or_404(db, model, id_, what=what, lock=lock) - if getattr(obj, owner_attr) != user.id: + obj = owned_row(db, model, id_, user, owner_attr=owner_attr, lock=lock) + if obj is None: raise not_found(what) return obj diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py index 31d68988..b872c0e7 100644 --- a/backend/app/routes/downloads.py +++ b/backend/app/routes/downloads.py @@ -3,23 +3,22 @@ All routes are behind `current_user`; admin routes add `admin_user`. Live job state is polled by the frontend (the worker is a separate process); this module never downloads — it enqueues, manages job state, and shares finished files. The library itself — listing, renaming, moving, deleting and -serving a done file, and uploading one — is `routes.files`. +serving a done file, and uploading one — is `routes.files`. Reads come from `downloads.queries`, +writes go through Capture's service (`service`, `links`, `quota`); this module validates, answers in +HTTP and commits. """ import re -from typing import Literal +from typing import TYPE_CHECKING, Literal from datetime import datetime, timedelta, timezone from urllib.parse import parse_qs, urlparse from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel, Field -from sqlalchemy import delete, func, select -from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app.mailtext import lang_for_user -from app.collation import collated from app import audit, sysconfig from app.auth import admin_user, current_user from app.config import settings @@ -29,44 +28,32 @@ from app.downloads import credentials as dlcreds from app.downloads import edit as editmod from app.downloads import folders from app.downloads import links as linksmod -from app.downloads import quota, service, storage +from app.downloads import queries, quota, service, storage from app.downloads import urlguard from app.downloads.folders import FolderKind -from app.downloads.status import FINISHED, NOT_REFETCHABLE, JobStatus +from app.downloads.status import NOT_REFETCHABLE, JobStatus from app.models import User, display_text -from app.downloads.models import ( - DownloadJob, - DownloadLink, - FileFolder, - DownloadProfile, - DownloadQuota, - DownloadShare, - LibraryState, - MediaAsset, -) from app.routes._common import ( ProgressIn, get_or_404, kept_position, not_found, - owned_or_404, ) from app.routes.files import ( - MAX_LIST, accessible_job, - assets_for, + own_folder, + own_job, own_job_for_update, - positions_for, quota_detail, serialize, serialize_job, - source_keys, ) -from app.security import hash_password -from app.userscope import is_messageable_user, messageable_clauses -from app.utils import iso +from app.userscope import emails_of, is_messageable_user, messageable_others, user_by_email from app.refusal import refused +if TYPE_CHECKING: + from app.downloads.models import DownloadJob, DownloadLink, DownloadProfile, FileFolder + router = APIRouter(prefix="/api/downloads", tags=["downloads"]) admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"]) @@ -118,7 +105,7 @@ def resolve_source(raw: str) -> tuple[str, str]: # --- profiles ------------------------------------------------------------------------------ -def _serialize_profile(p: DownloadProfile) -> dict: +def _serialize_profile(p: "DownloadProfile") -> dict: return { "id": p.id, "name": p.name, @@ -130,18 +117,7 @@ def _serialize_profile(p: DownloadProfile) -> dict: @router.get("/profiles") def list_profiles(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]: - rows = ( - db.execute( - select(DownloadProfile) - .where((DownloadProfile.is_builtin.is_(True)) | (DownloadProfile.user_id == user.id)) - .order_by( - DownloadProfile.is_builtin.desc(), DownloadProfile.sort_order, DownloadProfile.id - ) - ) - .scalars() - .all() - ) - return [_serialize_profile(p) for p in rows] + return [_serialize_profile(p) for p in queries.profiles_for(db, user)] class CreateProfileIn(BaseModel): @@ -157,15 +133,7 @@ def create_profile( spec = body.spec if not name: raise refused(400, "profile_name_required") - maxpos = db.execute( - select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where( - DownloadProfile.user_id == user.id - ) - ).scalar_one() - p = DownloadProfile( - user_id=user.id, name=name[:80], spec=spec, is_builtin=False, sort_order=maxpos + 1 - ) - db.add(p) + p = service.create_profile(db, user, name, spec) db.commit() return _serialize_profile(p) @@ -182,31 +150,32 @@ def update_profile( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - p = db.get(DownloadProfile, profile_id) - if p is None or p.is_builtin or p.user_id != user.id: - raise not_found("profile") + p = _own_profile(db, user, profile_id) sent = body.model_fields_set + name = None if "name" in sent: name = (body.name or "").strip() if not name: raise refused(400, "profile_name_required") - p.name = name[:80] - if "spec" in sent: - if body.spec is None: - raise refused(400, "spec_not_object") - p.spec = body.spec + if "spec" in sent and body.spec is None: + raise refused(400, "spec_not_object") + service.update_profile(p, name=name, spec=body.spec if "spec" in sent else None) db.commit() return _serialize_profile(p) +def _own_profile(db: Session, user: User, profile_id: int) -> "DownloadProfile": + p = queries.own_profile(db, user, profile_id) + if p is None: + raise not_found("profile") + return p + + @router.delete("/profiles/{profile_id}") def delete_profile( profile_id: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: - p = db.get(DownloadProfile, profile_id) - if p is None or p.is_builtin or p.user_id != user.id: - raise not_found("profile") - db.delete(p) + service.delete_profile(db, _own_profile(db, user, profile_id)) db.commit() return {"deleted": profile_id} @@ -276,18 +245,13 @@ def _resolve_spec( """Pick the format spec for an enqueue: an explicit profile_id (builtin or the user's own), an inline spec, or the first builtin as a fallback.""" if profile_id is not None: - p = db.get(DownloadProfile, profile_id) - if p is None or (not p.is_builtin and p.user_id != user.id): + p = queries.usable_profile(db, user, profile_id) + if p is None: raise not_found("profile") return p.spec, p.id if spec is not None: return spec, None - fallback = db.execute( - select(DownloadProfile) - .where(DownloadProfile.is_builtin.is_(True)) - .order_by(DownloadProfile.sort_order) - .limit(1) - ).scalar_one_or_none() + fallback = queries.default_profile(db) if fallback is None: raise refused(400, "no_profile") return fallback.spec, fallback.id @@ -325,6 +289,7 @@ def enqueue_download( job = service.enqueue(db, user.id, source_kind, source_ref, spec, profile_id, display_name) except quota.QuotaExceeded as e: raise HTTPException(status_code=422, detail=quota_detail(e)) from e + db.commit() return serialize_job(db, job) @@ -354,6 +319,7 @@ def enqueue_edit( raise HTTPException(status_code=422, detail=quota_detail(e)) from e except service.EditError as e: raise HTTPException(status_code=400, detail={"code": "edit", "reason": e.reason}) from e + db.commit() return serialize_job(db, job) @@ -397,38 +363,22 @@ def enqueue_clip( raise HTTPException(status_code=status, detail=detail) from e except quota.QuotaExceeded as e: raise HTTPException(status_code=422, detail=quota_detail(e)) from e + db.commit() return serialize_job(db, job) # How long a finished job stays in the queue: the tray polls it from every page and must see a job # turn done (its "Done · Open" row and the done notice) — the finished file itself is the library's. -# A done job is timed from when it became done, never from its last write: a rename, a move or the -# GC's expiry warning touches an old file's row, and the tray would announce it as just finished. QUEUE_RECENT = timedelta(minutes=10) @router.get("") def list_downloads(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]: """The queue: what is waiting, running, paused or failed, and what finished a moment ago.""" - recent = datetime.now(timezone.utc) - QUEUE_RECENT - jobs = ( - db.execute( - select(DownloadJob) - .where( - DownloadJob.user_id == user.id, - DownloadJob.status.not_in(FINISHED) - | ((DownloadJob.status == JobStatus.DONE) & (DownloadJob.ready_at >= recent)) - | ((DownloadJob.status == JobStatus.CANCELED) & (DownloadJob.updated_at >= recent)), - ) - .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()) - .limit(MAX_LIST) - ) - .scalars() - .all() - ) - assets = assets_for(db, jobs) - positions = positions_for(db, user, jobs) - sources = source_keys(db, jobs) + jobs = queries.queue(db, user, datetime.now(timezone.utc) - QUEUE_RECENT) + assets = queries.assets_for(db, jobs) + positions = queries.positions_for(db, user, jobs) + sources = queries.source_keys(db, jobs) return [ { **serialize(j, assets.get(j.asset_id) if j.asset_id is not None else None), @@ -452,14 +402,8 @@ _INDEX_RANK: dict[str, int] = { def my_download_index(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """source_ref -> best active status, for the feed's per-card "downloaded / queued" badge. Small + cheap so the feed can poll it without loading the full job list.""" - rows = db.execute( - select(DownloadJob.source_ref, DownloadJob.status).where( - DownloadJob.user_id == user.id, - DownloadJob.status.in_(tuple(_INDEX_RANK)), - ) - ).all() out: dict[str, str] = {} - for ref, status in rows: + for ref, status in queries.statuses_by_source(db, user, tuple(_INDEX_RANK)): if _INDEX_RANK.get(status, 0) > _INDEX_RANK.get(out.get(ref, ""), 0): out[ref] = status return out @@ -471,6 +415,7 @@ def pause_download( ) -> dict: job = own_job_for_update(db, user, job_id) service.pause(db, job) + db.commit() return serialize_job(db, job) @@ -480,6 +425,7 @@ def resume_download( ) -> dict: job = own_job_for_update(db, user, job_id) service.resume(db, job) + db.commit() return serialize_job(db, job) @@ -507,6 +453,7 @@ def redownload_download( if job.status in (JobStatus.QUEUED, JobStatus.RUNNING, JobStatus.PAUSED): raise refused(409, "still_downloading") service.redownload(db, job) + db.commit() return serialize_job(db, job) @@ -516,6 +463,7 @@ def cancel_download( ) -> dict: job = own_job_for_update(db, user, job_id) service.cancel(db, job) + db.commit() return serialize_job(db, job) @@ -533,24 +481,7 @@ def set_library_progress( one shared with them). A position too early or too close to the end clears it instead.""" accessible_job(db, user, job_id) position = kept_position(body) - mine = (LibraryState.user_id == user.id) & (LibraryState.job_id == job_id) - if not position: - # A statement, not a loaded row's delete: the periodic checkpoint and the pagehide beacon of - # one stop may both clear it, and the second then deletes nothing rather than failing. - db.execute(delete(LibraryState).where(mine)) - db.commit() - return {"ok": True, "job_id": job_id, "position_seconds": 0} - now = datetime.now(timezone.utc) - # One row per (user, job) even when two checkpoints race (the same pair): an upsert, not a - # read-then-insert that the second would fail on the unique key. - db.execute( - pg_insert(LibraryState) - .values(user_id=user.id, job_id=job_id, position_seconds=position, progress_updated_at=now) - .on_conflict_do_update( - constraint="uq_library_user_job", - set_={"position_seconds": position, "progress_updated_at": now}, - ) - ) + service.set_position(db, user, job_id, position) db.commit() return {"ok": True, "job_id": job_id, "position_seconds": position} @@ -563,8 +494,7 @@ def storyboard_meta( Best-effort: for a very long/high-res source the sprite may not generate in time — then `available` is False and the editor shows a plain timeline.""" - job = accessible_job(db, user, job_id) - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + asset = queries.asset_of(db, accessible_job(db, user, job_id)) if asset is None or asset.status != "ready" or not asset.rel_path: raise refused(409, "download_not_ready") meta = editmod.ensure_storyboard(asset, settings.files_root) @@ -584,8 +514,7 @@ def storyboard_meta( @router.get("/{job_id}/poster.jpg") def poster_image(job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)): """The locally-generated poster frame for a download whose source had no thumbnail.""" - job = accessible_job(db, user, job_id) - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + asset = queries.asset_of(db, accessible_job(db, user, job_id)) if asset is None or not asset.poster_path: raise not_found("poster") path = storage.safe_abs_path(settings.files_root, asset.poster_path) @@ -598,8 +527,7 @@ def poster_image(job_id: int, user: User = Depends(current_user), db: Session = def storyboard_image( job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db) ): - job = accessible_job(db, user, job_id) - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None + asset = queries.asset_of(db, accessible_job(db, user, job_id)) if asset is None or not asset.storyboard_path: raise not_found("filmstrip") path = storage.safe_abs_path(settings.files_root, asset.storyboard_path) @@ -622,45 +550,20 @@ def share_download( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - job = owned_or_404(db, DownloadJob, job_id, user, what="download") + job = own_job(db, user, job_id) if job.status != JobStatus.DONE or job.asset_id is None: raise refused(409, "share_unfinished") target = body.email.strip().lower() if not target: raise refused(400, "recipient_required") - recipient = db.execute( - select(User).where(func.lower(User.email) == target) - ).scalar_one_or_none() + recipient = user_by_email(db, target) # Uniform "no user" for a missing OR non-messageable (suspended/deactivated) target, so # sharing can't be used to probe which addresses exist or to reach a suspended account. - if not is_messageable_user(recipient): + if recipient is None or not is_messageable_user(recipient): raise not_found("recipient") if recipient.id == user.id: raise refused(400, "share_own") - exists = db.execute( - select(DownloadShare.id).where( - DownloadShare.job_id == job_id, DownloadShare.to_user_id == recipient.id - ) - ).first() - if exists is None: - db.add(DownloadShare(job_id=job_id, from_user_id=user.id, to_user_id=recipient.id)) - from app.notifications import create_notification - - create_notification( - db, - user_id=recipient.id, - type="download_shared", - title=(job.display_name or "A download") + " was shared with you", - # The client names the notice in the reader's language from `data`; the title is the - # English fallback only. - data={ - "kind": "download_shared", - "from": user.email, - "job_id": job_id, - "title": job.display_name, - }, - commit=False, - ) + if service.share(db, job, user, recipient): db.commit() return {"ok": True, "shared_with": recipient.email} @@ -672,22 +575,10 @@ def unshare_download( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - owned_or_404( - db, DownloadJob, job_id, user, what="download" - ) # ownership guard (404s if not the caller's job) - recipient = db.execute( - select(User).where(func.lower(User.email) == email.strip().lower()) - ).scalar_one_or_none() - if recipient is not None: - row = db.execute( - select(DownloadShare).where( - DownloadShare.job_id == job_id, - DownloadShare.to_user_id == recipient.id, - ) - ).scalar_one_or_none() - if row is not None: - db.delete(row) - db.commit() + own_job(db, user, job_id) # ownership guard (404s if not the caller's job) + recipient = user_by_email(db, email) + if recipient is not None and service.unshare(db, job_id, recipient): + db.commit() return {"ok": True} @@ -700,54 +591,26 @@ def share_recipients( ) -> list[dict]: """Registered users this user can share a download with (excludes self + any non-messageable account — suspended or deactivated — so the picker never leaks those addresses).""" - rows = ( - db.execute( - select(User) - .where(User.id != user.id, *messageable_clauses()) - .order_by(collated(User.email, lang_for_user(user))) - ) - .scalars() - .all() - ) + rows = messageable_others(db, user, lang_for_user(user)) return [{"id": u.id, "email": u.email, "display_name": u.display_name} for u in rows] # --- public watch links (share by link) ---------------------------------------------------- -def _own_link(db: Session, user: User, link_id: int) -> DownloadLink: - link = db.get(DownloadLink, link_id) +def _own_link(db: Session, user: User, link_id: int) -> "DownloadLink": + link = queries.own_link(db, user, link_id) if link is None: raise not_found("link") - target = ( - db.get(DownloadJob, link.job_id) - if link.job_id is not None - else db.get(FileFolder, link.folder_id) - ) - if target is None or target.user_id != user.id: - raise not_found("link") return link -def _link_expiry(days: int | None) -> datetime | None: - # None/0/negative → no expiry; otherwise N days out, capped at ~10 years. The Pydantic model - # already coerced/validated the type, so a non-number is a 422 before we get here. - if not days or days <= 0: - return None - return datetime.now(timezone.utc) + timedelta(days=min(days, 3650)) - - -def _links_of(db: Session, *where) -> list[dict]: - rows = db.execute(select(DownloadLink).where(*where).order_by(DownloadLink.id.desc())) - return [linksmod.owner_view(row) for row in rows.scalars()] - - @router.get("/{job_id}/links") def list_links( job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> list[dict]: - owned_or_404(db, DownloadJob, job_id, user, what="download") - return _links_of(db, DownloadLink.job_id == job_id) + own_job(db, user, job_id) + return [linksmod.owner_view(link) for link in queries.links_to(db, job_id=job_id)] class CreateLinkIn(BaseModel): @@ -763,30 +626,18 @@ class CreateLinkIn(BaseModel): def _new_link( db: Session, user: User, body: CreateLinkIn, **target: int -) -> tuple[DownloadLink, str]: - """Make a link to `target` (`job_id=` or `folder_id=`, already checked to be the user's), and - the token it is opened by — which is in hand only now.""" - pw = body.password.strip() - token = linksmod.new_token() - link = DownloadLink( - **target, - created_by=user.id, - allow_download=body.allow_download, - password_hash=hash_password(pw) if pw else None, - expires_at=_link_expiry(body.expires_days), - ) - linksmod.seal(link, token) - db.add(link) - db.flush() - audit.record( +) -> tuple["DownloadLink", str]: + """A link to `target` (`job_id=` or `folder_id=`, already checked to be the user's) and its + token (`links.create`). The Pydantic model already coerced the expiry, so a non-number is a 422 + before this runs.""" + return linksmod.create( db, - user.id, - audit.AuditAction.DOWNLOAD_LINK_CREATE, - target_type="download_link", - target_id=link.id, - after={**target, **_link_audit(link)}, + user, + password=body.password.strip(), + allow_download=body.allow_download, + expires_days=body.expires_days, + **target, ) - return link, token @router.post("/{job_id}/links") @@ -804,28 +655,20 @@ def create_link( return linksmod.owner_view(link, token=token) -def _shareable_job(db: Session, user: User, job_id: int) -> DownloadJob: - """One of the user's files, held (FOR SHARE) for the rest of the transaction: a delete of it at - the same moment (the user's, or the retention sweep's) waits and takes the new link with it, - rather than failing the link's foreign key — or, done first, leaves a 404.""" - owned_or_404(db, DownloadJob, job_id, user, what="download") - job = db.execute( - select(DownloadJob) - .where(DownloadJob.id == job_id) - .with_for_update(read=True) - .execution_options(populate_existing=True) - ).scalar_one_or_none() +def _shareable_job(db: Session, user: User, job_id: int) -> "DownloadJob": + """One of the user's files, held for the link's transaction (`service.hold_for_share`).""" + job = service.hold_for_share(db, user, job_id) if job is None: raise not_found("download") return job -def _shareable_folder(db: Session, user: User, folder_id: int) -> FileFolder: +def _shareable_folder(db: Session, user: User, folder_id: int) -> "FileFolder": """One of the user's own folders. A source folder lets its files go on its own clock, and the Hub is the user's channel to a machine of theirs: neither is a shelf to hand out. Held (FOR SHARE) for the rest of the transaction, so a delete of it at the same moment waits and takes the new link with it, rather than failing the link's foreign key.""" - owned_or_404(db, FileFolder, folder_id, user, what="folder") + own_folder(db, user, folder_id) try: folder = folders.own_folder(db, user.id, folder_id) except folders.FolderError as e: @@ -839,8 +682,8 @@ def _shareable_folder(db: Session, user: User, folder_id: int) -> FileFolder: def list_folder_links( folder_id: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> list[dict]: - owned_or_404(db, FileFolder, folder_id, user, what="folder") - return _links_of(db, DownloadLink.folder_id == folder_id) + own_folder(db, user, folder_id) + return [linksmod.owner_view(link) for link in queries.links_to(db, folder_id=folder_id)] @router.post("/folders/{folder_id}/links") @@ -856,15 +699,6 @@ def create_folder_link( return linksmod.owner_view(link, token=token) -def _link_audit(link: DownloadLink) -> dict: - """A link's settings as the audit records them: whether it has a password, never which.""" - return { - "allow_download": link.allow_download, - "has_password": link.password_hash is not None, - "expires_at": iso(link.expires_at), - } - - class UpdateLinkIn(BaseModel): allow_download: bool | None = None expires_days: int | None = None @@ -879,29 +713,15 @@ def update_link( db: Session = Depends(get_db), ) -> dict: link = _own_link(db, user, link_id) - before = _link_audit(link) - sent = body.model_fields_set - if "allow_download" in sent: - link.allow_download = bool(body.allow_download) - if "expires_days" in sent: - link.expires_at = _link_expiry(body.expires_days) - if "password" in sent: - pw = (body.password or "").strip() - link.password_hash = hash_password(pw) if pw else None - # Bump the version so any grant issued under the previous password stops working now, rather - # than living out its 6h TTL (setting/clearing/changing the password all count as a rotation). - link.password_version += 1 - after = _link_audit(link) - if after != before or "password" in sent: - audit.record( - db, - user.id, - audit.AuditAction.DOWNLOAD_LINK_UPDATE, - target_type="download_link", - target_id=link_id, - before=before, - after=after, - ) + linksmod.update( + db, + user, + link, + allow_download=body.allow_download, + expires_days=body.expires_days, + password=body.password, + sent=frozenset(body.model_fields_set), + ) db.commit() return linksmod.owner_view(link) @@ -910,16 +730,7 @@ def update_link( def revoke_link( link_id: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: - link = _own_link(db, user, link_id) - db.delete(link) - audit.record( - db, - user.id, - audit.AuditAction.DOWNLOAD_LINK_REVOKE, - target_type="download_link", - target_id=link_id, - after={"job_id": link.job_id} if link.job_id is not None else {"folder_id": link.folder_id}, - ) + linksmod.revoke(db, user, _own_link(db, user, link_id)) db.commit() return {"ok": True, "deleted": link_id} @@ -931,20 +742,9 @@ def revoke_link( def admin_list_downloads( admin: User = Depends(admin_user), db: Session = Depends(get_db) ) -> list[dict]: - jobs = ( - db.execute( - select(DownloadJob) - .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()) - .limit(MAX_LIST) - ) - .scalars() - .all() - ) - assets = assets_for(db, jobs) - emails = { - u.id: u.email - for u in db.execute(select(User).where(User.id.in_({j.user_id for j in jobs}))).scalars() - } + jobs = queries.recent_jobs(db) + assets = queries.assets_for(db, jobs) + emails = emails_of(db, {j.user_id for j in jobs}) out = [] for j in jobs: data = serialize(j, assets.get(j.asset_id) if j.asset_id is not None else None) @@ -956,27 +756,11 @@ def admin_list_downloads( @admin_router.get("/storage") def admin_storage(admin: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: - # Every file on disk (`quota.ON_DISK`): ready ones, and uploads waiting for or failed in their - # conversion — the same disk the footprint and the GC cap count. - on_disk = db.execute( - select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(quota.ON_DISK) - ).one() - total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar() + files_on_disk, total_bytes, total_assets = queries.storage_totals(db) # EVERY user, not only those holding a file: a quota is set before anyone downloads. - emails: dict[int, str] = { - uid: email - for uid, email in db.execute(select(User.id, User.email).order_by(User.email)).tuples() - } + emails = emails_of(db) uids = list(emails) - # What each one's Files lists: their finished files. - files: dict[int, int] = { - uid: n - for uid, n in db.execute( - select(DownloadJob.user_id, func.count()) - .where(DownloadJob.status == JobStatus.DONE) - .group_by(DownloadJob.user_id) - ).tuples() - } + files = queries.done_files_per_user(db) def row(uid: int) -> dict: lim = quota.resolve(db, uid) @@ -990,8 +774,8 @@ def admin_storage(admin: User = Depends(admin_user), db: Session = Depends(get_d } return { - "files_on_disk": on_disk[0], - "total_bytes": int(on_disk[1]), + "files_on_disk": files_on_disk, + "total_bytes": total_bytes, "total_assets": total_assets, "total_cap_bytes": sysconfig.get_int(db, "download_total_max_bytes"), "per_user": [row(uid) for uid in uids], @@ -1003,10 +787,9 @@ def admin_get_quota( user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: lim = quota.resolve(db, user_id) - row = db.get(DownloadQuota, user_id) return { "user_id": user_id, - "custom": row is not None, + "custom": queries.custom_quota(db, user_id) is not None, "max_bytes": lim.max_bytes, "max_concurrent": lim.max_concurrent, "max_jobs": lim.max_jobs, @@ -1031,25 +814,14 @@ def admin_set_quota( db: Session = Depends(get_db), ) -> dict: get_or_404(db, User, user_id, what="user") - cur = quota.resolve(db, user_id) - row = db.get(DownloadQuota, user_id) - if row is None: - row = DownloadQuota( - user_id=user_id, - max_bytes=cur.max_bytes, - max_concurrent=cur.max_concurrent, - max_jobs=cur.max_jobs, - unlimited=cur.unlimited, - ) - db.add(row) - if body.max_bytes is not None: - row.max_bytes = max(0, body.max_bytes) - if body.max_concurrent is not None: - row.max_concurrent = max(1, body.max_concurrent) - if body.max_jobs is not None: - row.max_jobs = max(1, body.max_jobs) - if body.unlimited is not None: - row.unlimited = body.unlimited + quota.set_custom( + db, + user_id, + max_bytes=body.max_bytes, + max_concurrent=body.max_concurrent, + max_jobs=body.max_jobs, + unlimited=body.unlimited, + ) db.commit() return admin_get_quota(user_id, admin, db) @@ -1058,8 +830,6 @@ def admin_set_quota( def admin_reset_quota( user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db) ) -> dict: - row = db.get(DownloadQuota, user_id) - if row is not None: - db.delete(row) + if quota.reset_custom(db, user_id): db.commit() return admin_get_quota(user_id, admin, db) diff --git a/backend/app/routes/files.py b/backend/app/routes/files.py index fcf385d8..dd16f0ea 100644 --- a/backend/app/routes/files.py +++ b/backend/app/routes/files.py @@ -5,8 +5,10 @@ the per-user layer (name, folder, kept, expiry, shares and links all hang off it bytes. Folders are rows (`downloads.folders`), so a rename or a move never touches the disk. In-flight jobs are not here: they are the download tray's (`routes.downloads`). The helpers above -the routes are shared with that module — serialization, the ownership locks, the upload spool — -because a download and a library file are one row at two moments. +the routes are shared with that module — serialization, the ownership checks, the upload spool — +because a download and a library file are one row at two moments. What is read comes from +`downloads.queries`, what is written goes through Capture's service (`service`, `folders`); this +module validates, answers in HTTP and commits. A machine with an API token (`app.apitokens`) may use the routes that take `files_caller`: list the folders and files, fetch a file and its hash, upload, delete — each only in a folder its token was @@ -19,49 +21,37 @@ import hashlib import os import tempfile import threading -from collections.abc import AsyncIterator, Sequence +from collections.abc import AsyncIterator from dataclasses import dataclass -from datetime import datetime, timezone from pathlib import Path -from typing import BinaryIO, Literal +from typing import TYPE_CHECKING, BinaryIO, Literal from urllib.parse import unquote, urlparse from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import FileResponse, Response from pydantic import BaseModel, Field -from sqlalchemy import func, or_, select, update -from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app import audit, sysconfig from app.apitokens import FilesCaller, Right, files_caller from app.audit import AuditAction from app.auth import current_user -from app.collation import collated from app.config import settings from app.db import get_db, release_read_connection -from app.downloads import folders, preview, quota, service, storage +from app.downloads import folders, preview, queries, quota, service, storage from app.downloads import upload as uploadmod from app.downloads.status import JobStatus from app.mailtext import lang_for_user from app.models import DISPLAY_TEXT_MAX, User, display_text -from app.downloads.models import ( - DownloadJob, - DownloadLink, - DownloadShare, - FileFolder, - LibraryState, - MediaAsset, -) from app.refusal import refused -from app.routes._common import capped_stream_to, not_found, owned_or_404 -from app.utils import iso, like_escape +from app.routes._common import capped_stream_to, not_found +from app.utils import iso + +if TYPE_CHECKING: + from app.downloads.models import DownloadJob, FileFolder, MediaAsset router = APIRouter(prefix="/api/files", tags=["files"]) -# Cap the listing so a heavy library can't return an unbounded result set (the old Download Center's -# own cap; far beyond any real library). -MAX_LIST = 500 # The longest a folder may keep a file, in days — ten years is "forever" in all but name. _MAX_RETENTION_DAYS = 3650 _HASH_CHUNK = 1 << 20 @@ -73,7 +63,7 @@ _HASHING = threading.BoundedSemaphore(_HASHERS) # --- shared with routes.downloads ------------------------------------------------------- -def serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: +def serialize(job: "DownloadJob", asset: "MediaAsset | None") -> dict: ready = asset is not None and asset.status == "ready" and bool(asset.rel_path) return { "id": job.id, @@ -133,58 +123,18 @@ def serialize(job: DownloadJob, asset: MediaAsset | None) -> dict: } -def assets_for(db: Session, jobs: Sequence[DownloadJob]) -> dict[int, MediaAsset]: - ids = {j.asset_id for j in jobs if j.asset_id} - if not ids: - return {} - rows = db.execute(select(MediaAsset).where(MediaAsset.id.in_(ids))).scalars() - return {a.id: a for a in rows} - - -def positions_for(db: Session, user: User, jobs: Sequence[DownloadJob]) -> dict[int, int]: - """`user`'s resume point in each of `jobs` that has one (the library player's `library_states`).""" - ids = [j.id for j in jobs] - if not ids: - return {} - rows = db.execute( - select(LibraryState.job_id, LibraryState.position_seconds).where( - LibraryState.user_id == user.id, LibraryState.job_id.in_(ids) - ) - ).all() - return {job_id: pos for job_id, pos in rows} - - -def serialize_job(db: Session, job: DownloadJob) -> dict: +def serialize_job(db: Session, job: "DownloadJob") -> dict: """Resolve a job's cache asset (if any) and serialize the pair — the resolve-then-serialize step every single-job handler shares.""" - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None - return serialize(job, asset) + return serialize(job, queries.asset_of(db, job)) -def own_job_for_update(db: Session, user: User, job_id: int) -> DownloadJob: - """The ownership check (`owned_or_404`) for a transition that moves its asset's ref count (resume, cancel, delete, redownload) - or writes the status those decide on (pause). The job is re-read under a row lock, so two - overlapping requests for one job run one after the other and the second decides on the first's - committed result — never both releasing the same hold, and never a pause turning a job a cancel - just released back into a holder; either can free a file another job still holds. - - The asset is locked BEFORE the job: the order a finishing worker child takes them in - (`service.finalize_if_running`), so the two can never deadlock.""" - while True: - asset_id = owned_or_404(db, DownloadJob, job_id, user, what="download").asset_id - if asset_id is not None: - db.execute(select(MediaAsset.id).where(MediaAsset.id == asset_id).with_for_update()) - job = db.execute( - select(DownloadJob) - .where(DownloadJob.id == job_id) - .with_for_update() - .execution_options(populate_existing=True) - ).scalar_one_or_none() - if job is None: - raise not_found("download") - if job.asset_id == asset_id: - return job - db.rollback() # another request moved the job to another asset meanwhile: lock that one +def own_job_for_update(db: Session, user: User, job_id: int) -> "DownloadJob": + """One of the user's jobs under its row lock (`service.lock_own_job`), or a 404.""" + job = service.lock_own_job(db, user, job_id) + if job is None: + raise not_found("download") + return job def quota_detail(e: quota.QuotaExceeded) -> dict: @@ -320,30 +270,32 @@ async def spool_upload( tmp_path.unlink(missing_ok=True) -def accessible_job(db: Session, user: User, job_id: int) -> DownloadJob: - """A file the caller may read: their own, or one shared with them.""" - job = db.get(DownloadJob, job_id) +def accessible_job(db: Session, user: User, job_id: int) -> "DownloadJob": + """A file the caller may read (`queries.accessible_job`: their own, or shared with them).""" + job = queries.accessible_job(db, user, job_id) if job is None: raise not_found("download") - if job.user_id == user.id: - return job - shared = db.execute( - select(DownloadShare.id).where( - DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id - ) - ).first() - if shared is None: - raise not_found("download") return job -def readable_job(db: Session, caller: FilesCaller, job_id: int) -> DownloadJob: - """A file `caller` may read: as `accessible_job` for the user; for a token, one of the user's - own in a folder it may read.""" - if caller.grants is None: - return accessible_job(db, caller.user, job_id) - job = db.get(DownloadJob, job_id) - if job is None or job.user_id != caller.user.id or not caller.may(job.folder_id, Right.READ): +def own_job(db: Session, user: User, job_id: int) -> "DownloadJob": + job = queries.own_job(db, user, job_id) + if job is None: + raise not_found("download") + return job + + +def own_folder(db: Session, user: User, folder_id: int) -> "FileFolder": + folder = queries.own_folder(db, user, folder_id) + if folder is None: + raise not_found("folder") + return folder + + +def readable_job(db: Session, caller: FilesCaller, job_id: int) -> "DownloadJob": + """A file `caller` may read (`queries.readable_job`), or a 404.""" + job = queries.readable_job(db, caller, job_id) + if job is None: raise not_found("download") return job @@ -351,7 +303,7 @@ def readable_job(db: Session, caller: FilesCaller, job_id: int) -> DownloadJob: # --- folders ------------------------------------------------------------------------------- -def _serialize_folder(folder: FileFolder, count: int, size: int, linked: bool = False) -> dict: +def _serialize_folder(folder: "FileFolder", count: int, size: int, linked: bool = False) -> dict: return { "id": folder.id, "kind": folder.kind, @@ -367,18 +319,6 @@ def _serialize_folder(folder: FileFolder, count: int, size: int, linked: bool = } -def _linked(db: Session, folder_ids: list[int]) -> set[int]: - """Which of these folders a live public link opens (one not past its expiry).""" - now = datetime.now(timezone.utc) - rows = db.execute( - select(DownloadLink.folder_id).where( - DownloadLink.folder_id.in_(folder_ids), - or_(DownloadLink.expires_at.is_(None), DownloadLink.expires_at > now), - ) - ).scalars() - return {folder_id for folder_id in rows if folder_id is not None} - - @router.get("/folders") def list_folders( caller: FilesCaller = Depends(files_caller), db: Session = Depends(get_db) @@ -386,23 +326,9 @@ def list_folders( """The caller's folders with how many files each holds and their size (a token: the ones it may list). A source folder its files have all left stays (count 0): the client decides whether to show it.""" - done = (DownloadJob.folder_id == FileFolder.id) & (DownloadJob.status == JobStatus.DONE) - stmt = select( - FileFolder, - func.count(DownloadJob.id), - func.coalesce(func.sum(MediaAsset.size_bytes), 0), - ).where(FileFolder.user_id == caller.user.id) - listable = caller.folders_with(Right.LIST) - if listable is not None: - stmt = stmt.where(FileFolder.id.in_(listable)) - rows = db.execute( - stmt.outerjoin(DownloadJob, done) - .outerjoin(MediaAsset, MediaAsset.id == DownloadJob.asset_id) - .group_by(FileFolder.id) - .order_by(FileFolder.kind.desc(), FileFolder.source_key, FileFolder.name, FileFolder.id) - ).all() - linked = _linked(db, [f.id for f, _, _ in rows]) - return [_serialize_folder(f, count, int(size), f.id in linked) for f, count, size in rows] + rows = queries.folders_with_totals(db, caller.user, caller.folders_with(Right.LIST)) + linked = queries.linked_folders(db, [f.id for f, _, _ in rows]) + return [_serialize_folder(f, count, size, f.id in linked) for f, count, size in rows] class FolderIn(BaseModel): @@ -420,13 +346,10 @@ def _folder_name(raw: str) -> str: def create_folder( body: FolderIn, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: - folder = FileFolder(user_id=user.id, kind=folders.FolderKind.OWN, name=_folder_name(body.name)) try: - with db.begin_nested(): - db.add(folder) - db.flush() - except IntegrityError as e: # the unique (user, lower(name)): one folder per name - raise refused(409, "file_folder_name_taken") from e + folder = folders.create_own(db, user.id, _folder_name(body.name)) + except folders.FolderError as e: + raise refused(409, e.reason) from e audit.record( db, user.id, AuditAction.FILE_FOLDER_CREATE, target_type="file_folder", target_id=folder.id ) @@ -449,17 +372,13 @@ def update_folder( ) -> dict: """Rename one of the caller's own folders (a source folder's name is its origin's), and/or set how long any of their folders keeps a file (`folders.set_retention`).""" - folder = owned_or_404(db, FileFolder, folder_id, user, what="folder") + folder = own_folder(db, user, folder_id) sent = body.model_fields_set if "name" in sent: - if folder.kind != folders.FolderKind.OWN: - raise refused(409, "file_folder_not_own") - folder.name = _folder_name(body.name or "") try: - with db.begin_nested(): - db.flush() - except IntegrityError as e: - raise refused(409, "file_folder_name_taken") from e + folders.rename(db, folder, _folder_name(body.name or "")) + except folders.FolderError as e: + raise refused(409, e.reason) from e audit.record( db, user.id, @@ -484,18 +403,9 @@ def update_folder( after={"retention_days": body.retention_days}, ) db.commit() - count, size = _folder_totals(db, folder.id) - return _serialize_folder(folder, count, size, folder.id in _linked(db, [folder.id])) - - -def _folder_totals(db: Session, folder_id: int) -> tuple[int, int]: - count, size = db.execute( - select(func.count(DownloadJob.id), func.coalesce(func.sum(MediaAsset.size_bytes), 0)) - .select_from(DownloadJob) - .outerjoin(MediaAsset, MediaAsset.id == DownloadJob.asset_id) - .where(DownloadJob.folder_id == folder_id, DownloadJob.status == JobStatus.DONE) - ).one() - return count, int(size) + count, size = queries.folder_totals(db, folder.id) + linked = folder.id in queries.linked_folders(db, [folder.id]) + return _serialize_folder(folder, count, size, linked) @router.delete("/folders/{folder_id}") @@ -503,7 +413,7 @@ def delete_folder( folder_id: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: """Delete one of the caller's own folders; its files go back to their source folders.""" - folder = owned_or_404(db, FileFolder, folder_id, user, what="folder") + folder = own_folder(db, user, folder_id) try: moved = folders.delete_own(db, folder) except folders.FolderError as e: @@ -545,13 +455,7 @@ def remove_shared_with_me( """Remove a file that was shared WITH this user from their 'Shared with me' list. Deletes only the recipient's share grant — the owner's job and the physical file are untouched (this is a per-user dismissal, not a delete).""" - row = db.execute( - select(DownloadShare).where( - DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id - ) - ).scalar_one_or_none() - if row is not None: - db.delete(row) + if service.unshare(db, job_id, user): db.commit() return {"ok": True, "removed": job_id} @@ -580,46 +484,22 @@ def list_files( raise refused(403, "api_token_scope") if folder_id is not None and folder_id not in listable: raise not_found("folder") - stmt = ( - select(DownloadJob) - .join(MediaAsset, MediaAsset.id == DownloadJob.asset_id) - .where(DownloadJob.status == JobStatus.DONE) - ) shared = scope == "shared" - if shared: - stmt = stmt.join(DownloadShare, DownloadShare.job_id == DownloadJob.id).where( - DownloadShare.to_user_id == user.id - ) - else: - stmt = stmt.where(DownloadJob.user_id == user.id) - if folder_id is not None: - stmt = stmt.where(DownloadJob.folder_id == folder_id) - elif listable is not None: - stmt = stmt.where(DownloadJob.folder_id.in_(listable)) - name = func.coalesce(DownloadJob.display_name, MediaAsset.title, "") - term = q.strip() - if term: - like = f"%{like_escape(term)}%" - stmt = stmt.where( - or_( - name.ilike(like), - func.coalesce(DownloadJob.display_uploader, MediaAsset.uploader, "").ilike(like), - ) - ) - # A file is added to its owner's list when it is done, not when it was asked for (a download that - # failed for days and then went through is new); one shared with the caller, when it was shared. - key = { - "added": DownloadShare.created_at if shared else DownloadJob.ready_at, - "name": collated(name, lang_for_user(user)), - "size": MediaAsset.size_bytes, - }[sort] - first = key.asc() if order == "asc" else key.desc() - tie = DownloadJob.id.asc() if order == "asc" else DownloadJob.id.desc() - jobs = db.execute(stmt.order_by(first.nulls_last(), tie).limit(MAX_LIST)).scalars().all() - assets = assets_for(db, jobs) - positions = positions_for(db, user, jobs) - shared_at = _shared_at(db, user, jobs) if shared else {} - sources = source_keys(db, jobs) + jobs = queries.library( + db, + user, + folder_id=folder_id, + listable=listable, + shared=shared, + term=q.strip(), + sort=sort, + order=order, + lang=lang_for_user(user), + ) + assets = queries.assets_for(db, jobs) + positions = queries.positions_for(db, user, jobs) + shared_at = queries.shared_at(db, user, jobs) if shared else {} + sources = queries.source_keys(db, jobs) return [ { **serialize(j, assets.get(j.asset_id) if j.asset_id is not None else None), @@ -632,36 +512,6 @@ def list_files( ] -def source_keys(db: Session, jobs: Sequence[DownloadJob]) -> dict[int, str]: - """Where each file came from (its source folder's key), for the list's Source column, the - "back to its source folder" move and the download tray's site: a file in a source folder has its folder's key; one moved into - an own folder (or someone else's, shared) is traced back by `folders.source_key_for`.""" - ids = {j.folder_id for j in jobs if j.folder_id is not None} - by_id = ( - {f.id: f for f in db.execute(select(FileFolder).where(FileFolder.id.in_(ids))).scalars()} - if ids - else {} - ) - out: dict[int, str] = {} - for j in jobs: - folder = by_id.get(j.folder_id) if j.folder_id is not None else None - if folder is not None and folder.kind == folders.FolderKind.SOURCE and folder.source_key: - out[j.id] = folder.source_key - else: - out[j.id] = folders.source_key_for(db, j) - return out - - -def _shared_at(db: Session, user: User, jobs: Sequence[DownloadJob]) -> dict[int, datetime]: - rows = db.execute( - select(DownloadShare.job_id, DownloadShare.created_at).where( - DownloadShare.to_user_id == user.id, - DownloadShare.job_id.in_([j.id for j in jobs]), - ) - ).all() - return {job_id: at for job_id, at in rows} - - _MAX_EXTRA_LINKS = 8 @@ -709,18 +559,20 @@ def update_file( """Rename a file and its channel, set its reference links (all display-only: the file on disk keeps its own name), move it to another folder (`folders.move`), or keep it until deleted / hand it back to its folder's retention.""" - job = owned_or_404(db, DownloadJob, job_id, user, what="download") + job = own_job(db, user, job_id) sent = body.model_fields_set + display: dict[str, object] = {} if "name" in sent: - job.display_name = display_text(body.name) + display["display_name"] = display_text(body.name) if "uploader" in sent: - job.display_uploader = display_text(body.uploader) + display["display_uploader"] = display_text(body.uploader) if "uploader_url" in sent: - job.display_uploader_url = _clean_url(body.uploader_url) + display["display_uploader_url"] = _clean_url(body.uploader_url) if "extra_links" in sent: raw = body.extra_links or [] cleaned = [u for u in (_clean_url(x) for x in raw) if u][:_MAX_EXTRA_LINKS] - job.extra_links = cleaned or None + display["extra_links"] = cleaned or None + service.set_display(job, **display) if "folder_id" in sent: if body.folder_id is None: raise refused(400, "file_folder_gone") @@ -766,7 +618,8 @@ def delete_file( target_id=job_id, summary=caller.via, ) - service.delete(db, job) # commits, the audit row with it + service.delete(db, job) + db.commit() # the audit row with it return {"ok": True, "deleted": job_id} @@ -785,7 +638,7 @@ def upload_file( return serialize_job(db, ingest_spooled(db, upload)) -def ingest_spooled(db: Session, upload: SpooledUpload) -> DownloadJob: +def ingest_spooled(db: Session, upload: SpooledUpload) -> "DownloadJob": """The spooled upload into the library (`upload.ingest_upload`, which commits its own rows), its refusals answered as HTTP.""" if upload.size == 0: @@ -809,8 +662,8 @@ def ingest_spooled(db: Session, upload: SpooledUpload) -> DownloadJob: return job -def _ready_file(db: Session, job: DownloadJob) -> tuple[MediaAsset, Path]: - asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None +def _ready_file(db: Session, job: "DownloadJob") -> tuple["MediaAsset", Path]: + asset = queries.asset_of(db, job) if asset is None or asset.status != "ready" or not asset.rel_path: raise refused(409, "download_not_ready") path = storage.safe_abs_path(settings.files_root, asset.rel_path) @@ -819,7 +672,7 @@ def _ready_file(db: Session, job: DownloadJob) -> tuple[MediaAsset, Path]: return asset, path -def serve_file(db: Session, job: DownloadJob) -> FileResponse: +def serve_file(db: Session, job: "DownloadJob") -> FileResponse: """The file of `job` (the caller may read it), range-aware, named by the user's display name. The served Content-Type comes from the ffprobe-verified media_kind + a container allow-list @@ -837,7 +690,7 @@ def serve_file(db: Session, job: DownloadJob) -> FileResponse: if asset.sha256: digest = base64.b64encode(bytes.fromhex(asset.sha256)).decode() headers["Repr-Digest"] = f"sha-256=:{digest}:" - asset.last_access_at = datetime.now(timezone.utc) # the handler commits it + service.touch(asset) # the handler commits it media_type = storage.served_media_type(asset.media_kind, asset.container) if media_type is None: return FileResponse( @@ -892,13 +745,13 @@ def preview_file( text = preview.text_utf8(head) if text is None: raise refused(415, "file_no_preview") - asset.last_access_at = datetime.now(timezone.utc) + service.touch(asset) db.commit() return Response(text, media_type=preview.media_type(kind, asset.container), headers=headers) filename = storage.download_filename( job.display_name or asset.title or asset.source_ref, asset.container, path ) - asset.last_access_at = datetime.now(timezone.utc) + service.touch(asset) db.commit() return FileResponse( path, @@ -934,10 +787,6 @@ def get_sha256( finally: _HASHING.release() digest = h.hexdigest() - db.execute( - update(MediaAsset) - .where(MediaAsset.id == asset_id, MediaAsset.sha256.is_(None)) - .values(sha256=digest) - ) + service.store_sha256(db, asset_id, digest) db.commit() return {"sha256": digest} diff --git a/backend/app/routes/public.py b/backend/app/routes/public.py index 1fc6949c..f0312c15 100644 --- a/backend/app/routes/public.py +++ b/backend/app/routes/public.py @@ -6,24 +6,26 @@ the app or any account data. Password-protected links exchange the password once a short-lived signed grant that the media URL carries (a `