"""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. 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 from app.config import settings from app.downloads import adapters from app.downloads import clip as clipmod 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, PlexItem, PlexShow, User from app.catalog.models import Channel, 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): """Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key.""" def __init__(self, reason: str): self.reason = reason super().__init__(reason) def source_url(source_kind: str, source_ref: str) -> str: """The URL yt-dlp should fetch. YouTube stores the 11-char id (the adapter reconstructs the watch URL); ad-hoc sources store the URL verbatim.""" return adapters.resolve_adapter(source_kind, source_ref).source_url(source_ref) def reference_url(job, asset) -> str | None: """The clean "downloaded from" URL for a job — shared by the Files library and the public watch page so both surfaces always agree. None for edit derivatives (a clip's source is the user's own earlier download, not a web page) and for a Plex clip (the library is not a public page); a YouTube clip links its video at the clip's start; else the canonical page URL yt-dlp recorded; else one derived from source_kind+source_ref (covers queued/older rows before the worker fills it).""" if job is None or job.job_kind == JobKind.EDIT: return None if job.job_kind == JobKind.CLIP: return clipmod.reference_url(job.edit_spec or {}) if asset is not None and asset.source_webpage_url: return asset.source_webpage_url if job.source_kind in ("youtube", "url"): return source_url(job.source_kind, job.source_ref) return None def library_left_behind(db: Session, files_root: str) -> bool: """True when the library names files and none of the newest of them is under `files_root` — a fresh, empty volume mounted in the library's place (a compose file changed under an install, its old volume left behind). The mount check cannot see that: the new volume is a real mount.""" rels = ( db.execute( select(MediaAsset.rel_path) .where(MediaAsset.status == "ready", MediaAsset.rel_path.is_not(None)) .order_by(MediaAsset.id.desc()) .limit(20) ) .scalars() .all() ) return bool(rels) and not any(storage.abs_path(files_root, r).exists() for r in rels if r) def get_or_create_asset( db: Session, source_kind: str, source_ref: str, format_sig: str ) -> MediaAsset: """Fetch the cache row for this (source, format) or create a fresh `pending` one. Races on the unique cache key are resolved by re-selecting after an IntegrityError, so two simultaneous enqueues of the same video+format converge on one shared asset.""" stmt = select(MediaAsset).where( MediaAsset.source_kind == source_kind, MediaAsset.source_ref == source_ref, MediaAsset.format_sig == format_sig, ) # Locked to the caller's commit: a release or the GC freeing this asset meanwhile (they lock it # too) either waits for the new hold, or has deleted the row by the time this read returns — and # the caller then gets a fresh one, never a job pointing at a row that is gone. asset = db.execute( stmt.with_for_update().execution_options(populate_existing=True) ).scalar_one_or_none() if asset is not None: # A previously-FAILED shared asset must be re-attempted for the new job about to hold it — # reset it to pending (+clear the error) so the worker actually re-downloads instead of # short-circuiting the new job with the asset's stale error. (Mirrors resume_download; without # this a fresh enqueue of a once-failed (source,format) pair is permanently poisoned, since # errored assets carry no expires_at and GC never clears them.) if asset.status == "error": asset.status = "pending" asset.error = None return asset asset = MediaAsset( source_kind=source_kind, source_ref=source_ref, format_sig=format_sig, status="pending", ref_count=0, ) try: # A SAVEPOINT: losing the race undoes only this insert, never the caller's own pending writes # (Retry clears the job's error before it gets here; a whole-session rollback threw that away). with db.begin_nested(): db.add(asset) db.flush() # silent-ok: the expected outcome of the race this savepoint exists for — two jobs wanting the # same asset. Losing it means the other insert won, and re-selecting is the whole handling. except IntegrityError: asset = db.execute(stmt).scalar_one() return asset def populate_from_catalog(db: Session, asset: MediaAsset) -> bool: """Fill an asset's display metadata (title/uploader/thumbnail/date/duration) from our own catalog when the source is a known YouTube video — so a queued/downloading row shows the real title + thumbnail immediately instead of the bare id. 0 network cost. Returns True if it found a catalog match.""" if asset.source_kind != "youtube" or asset.title: return False video = db.get(Video, asset.source_ref) if video is None: return False asset.title = video.title asset.thumbnail_url = video.thumbnail_url asset.duration_s = video.duration_seconds if video.published_at: asset.upload_date = video.published_at.date() channel = db.get(Channel, video.channel_id) if channel is not None: asset.uploader = channel.title return True def next_queue_pos(db: Session) -> int: return (db.execute(select(func.coalesce(func.max(DownloadJob.queue_pos), 0))).scalar() or 0) + 1 def incr_ref_count(db: Session, asset_id: int) -> None: """Atomically +1 an asset's hold count AT THE DB. A Python read-modify-write (`asset.ref_count = asset.ref_count + 1`) loses an update when two enqueues of the same (source,format) run in parallel — both read the same value and write the same +1 — leaving the count one short, so the first release can free a file another job still holds (B-sub-5).""" db.execute( update(MediaAsset) .where(MediaAsset.id == asset_id) .values(ref_count=MediaAsset.ref_count + 1) .execution_options(synchronize_session=False) ) def decr_ref_count(db: Session, asset_id: int) -> int: """Atomically -1 an asset's hold count (clamped at 0) and RETURN the new value. The caller decides whether to free the asset from the returned count while still holding the row's write lock, so two concurrent releases can't both observe the same pre-decrement value and double-free — Postgres serialises the two UPDATEs on the row lock (B-sub-5). synchronize_session=False: the ORM copy is intentionally left stale; callers read the returned value, not `asset.ref_count`.""" return db.execute( update(MediaAsset) .where(MediaAsset.id == asset_id) .values(ref_count=func.greatest(MediaAsset.ref_count - 1, 0)) .returning(MediaAsset.ref_count) .execution_options(synchronize_session=False) ).scalar_one() def enqueue( db: Session, user_id: int, source_kind: str, source_ref: str, spec: dict, profile_id: int | None = None, display_name: str | None = None, ) -> DownloadJob: """Queue a download for `user_id`. Returns the DownloadJob (already `done` on a cache hit). Raises quota.QuotaExceeded if the user is at their job-count or footprint cap.""" quota.check_enqueue(db, user_id) spec = formats.normalize(spec) # If this user has a stored cookie for the download's host, salt the cache signature with their # id so the cookie-authed result is a PRIVATE asset — never dedup-shared with, or cache-served # to, another user (the load-bearing privacy fix; the worker attaches the cookie to exactly this # salted identity). A normal download (no matching cookie) gets salt=None and shares as before. salt = credentials.salt_for_enqueue(db, user_id, source_url(source_kind, source_ref)) sig = formats.format_sig(spec, salt=salt) asset = get_or_create_asset(db, source_kind, source_ref, sig) # Show a real title + thumbnail the moment it's queued (feed downloads are catalog videos). populate_from_catalog(db, asset) job = DownloadJob( user_id=user_id, asset_id=asset.id, source_kind=source_kind, source_ref=source_ref, profile_id=profile_id, profile_snapshot=spec, display_name=display_name or (asset.title[:DISPLAY_TEXT_MAX] if asset.status == "ready" and asset.title else None), status=JobStatus.QUEUED, queue_pos=next_queue_pos(db), ) db.add(job) folders.place(db, job) incr_ref_count(db, asset.id) # Cache hit: the file already exists — no worker round-trip needed. if asset.status == "ready": job.status = JobStatus.DONE job.progress = 100 folders.stamp_done(db, job) asset.last_access_at = datetime.now(timezone.utc) db.flush() return job def clip_title(source_title: str | None, display_name: str | None) -> str: # Clipped to DISPLAY_TEXT_MAX, not to a literal: this feeds `DownloadJob.display_name` through # `enqueue_edit`, the same column `display_text` guards. A machine-made title, so it is clipped # rather than run through `display_text` — blank here means "no title", not None. if display_name: return display_name[:DISPLAY_TEXT_MAX] base = source_title or "Clip" return f"{base} (clip)"[:DISPLAY_TEXT_MAX] def enqueue_edit( db: Session, user_id: int, source_job: DownloadJob, edit_spec: dict, display_name: str | None = None, ) -> DownloadJob: """Queue a phase-2 editor derivative (trim/crop) of `source_job` for `user_id`. The derivative is per-user (never shared-cached): its asset uses `source_kind='edit'` with a cache key that embeds the user id, so an identical re-edit by the same user is a cache hit but it never dedups across users. Reuses the same worker/quota/GC path as a download. Raises quota.QuotaExceeded (holdings cap) or EditError (empty spec / source not ready).""" src_asset = db.get(MediaAsset, source_job.asset_id) if source_job.asset_id else None if src_asset is None or src_asset.status != "ready" or not src_asset.rel_path: raise EditError("source_not_ready") # A generic uploaded file (media_kind='file') has no timeline to trim/crop — the editor path is # media-only. Reject here so an upload's file never reaches the ffmpeg edit worker. if src_asset.media_kind == mediakind.FILE: raise EditError("not_media") spec = editmod.normalize_edit_spec(edit_spec) if not spec: raise EditError("empty_edit") quota.check_enqueue(db, user_id) sig = editmod.edit_sig(spec) ref = f"{user_id}:{src_asset.id}"[:512] asset = get_or_create_asset(db, "edit", ref, sig) if not asset.title: asset.title = clip_title(src_asset.title, display_name) asset.uploader = src_asset.uploader asset.thumbnail_url = src_asset.thumbnail_url asset.upload_date = src_asset.upload_date asset.duration_s = editmod.clip_duration(spec, src_asset.duration_s) # A clip inherits its source's kind (a video edit stays video, an audio trim stays audio); # the ffmpeg edit worker doesn't re-probe codecs, so set it here where the source is known. asset.media_kind = src_asset.media_kind or mediakind.VIDEO job = DownloadJob( user_id=user_id, asset_id=asset.id, job_kind=JobKind.EDIT, source_kind="edit", source_ref=ref, source_job_id=source_job.id, source_asset_id=src_asset.id, edit_spec=spec, profile_snapshot={}, display_name=display_name or asset.title[:DISPLAY_TEXT_MAX], status=JobStatus.QUEUED, queue_pos=next_queue_pos(db), ) db.add(job) folders.place(db, job) incr_ref_count(db, asset.id) if asset.status == "ready": # identical clip already produced for this user → instant job.status = JobStatus.DONE job.progress = 100 folders.stamp_done(db, job) asset.last_access_at = datetime.now(timezone.utc) db.flush() return job def _clip_source_facts(db: Session, spec: dict) -> dict: """Title, uploader, thumbnail and length of what a clip is cut from, read from our own catalog. Raises ClipError when a Plex item is unknown or its file cannot be read here — said at enqueue, rather than as a job that fails in the queue a minute later.""" if spec["source"] == "youtube": video = db.get(Video, spec["ref"]) if video is None: # not in the catalog: titled "Clip" until yt-dlp names it (worker) return {"title": None, "uploader": None, "thumbnail_url": None, "length_s": None} channel = db.get(Channel, video.channel_id) return { "title": video.title, "uploader": channel.title if channel else None, "thumbnail_url": video.thumbnail_url, "length_s": video.duration_seconds, } item = plex_lookups.by_rating_key(db, PlexItem, spec["ref"]) if item is None: raise clipmod.ClipError("not_found") if plex_paths.local_media_path(db, item.file_path) is None: raise clipmod.ClipError("media_missing") title = item.title show = db.get(PlexShow, item.show_id) if item.show_id else None if show is not None: episode = ( f"S{item.season_number}·E{item.episode_number}" if item.season_number is not None and item.episode_number is not None else None ) title = " — ".join(p for p in (show.title, episode, item.title) if p) return { "title": title, "uploader": None, # Plex art is served through our own authenticated routes, never a public URL; the worker # cuts the clip's poster from the clip itself. "thumbnail_url": None, "length_s": item.duration_s, } def enqueue_clip( db: Session, user_id: int, spec: dict, display_name: str | None = None ) -> DownloadJob: """Queue an ad-hoc clip (`clipmod.normalize`d spec) for `user_id`: a NEW per-user asset cut straight from the source — a YouTube range or a Plex library file (see downloads/clip). Raises ClipError (unknown or unreachable source, a range past its end) or quota.QuotaExceeded.""" facts = _clip_source_facts(db, spec) # One second of grace: a mark set on the last frame may round past the catalog's whole seconds. if facts["length_s"] and spec["end_s"] > facts["length_s"] + 1: raise clipmod.ClipError("past_end") quota.check_enqueue(db, user_id) ref = clipmod.asset_ref(user_id, spec)[:512] asset = get_or_create_asset(db, "clip", ref, clipmod.sig(spec)) if not asset.title: asset.title = clip_title(facts["title"], display_name) asset.uploader = facts["uploader"] asset.thumbnail_url = facts["thumbnail_url"] asset.duration_s = round(clipmod.duration(spec)) asset.media_kind = mediakind.VIDEO job = DownloadJob( user_id=user_id, asset_id=asset.id, job_kind=JobKind.CLIP, source_kind="clip", source_ref=ref, edit_spec=spec, profile_snapshot={}, display_name=display_name or asset.title[:DISPLAY_TEXT_MAX], status=JobStatus.QUEUED, queue_pos=next_queue_pos(db), ) db.add(job) folders.place(db, job) incr_ref_count(db, asset.id) if asset.status == "ready": # the same range was already cut for this user → instant job.status = JobStatus.DONE job.progress = 100 folders.stamp_done(db, job) asset.last_access_at = datetime.now(timezone.utc) db.flush() return job def finalize_if_running(db: Session, job_id: int, **fields) -> bool: """Write terminal `fields` to a job in ONE atomic ``UPDATE ... WHERE status='running'`` and report whether it fired. This is the single owner of the invariant "a finishing worker child must never resurrect a job a route just stopped": the atomic WHERE closes the read-then-write race an ``if job.status == 'running'`` guard leaves open, so a concurrent cancel/pause can no longer be overwritten with done/error (which would re-open the double-release → file-delete-under-live-holder class, since `done`/`error` hold the asset but `canceled` does not — see release_asset). Runs in the CALLER's session and does NOT commit — the caller commits it together with its asset writes in one transaction. It flushes first so any pending asset write takes its row lock BEFORE this takes the job's, matching the cancel route's asset-then-job order (no deadlock). `fields` are mapped-column kwargs (never user input). Returns True when the row was still running and got updated, False when a route had already moved it out of 'running'.""" if not fields: return False db.flush() # asset-then-job lock order: push pending asset writes before we lock the job row if fields.get("status") == JobStatus.DONE: return _finalize_done(db, job_id, {**fields, "ready_at": datetime.now(timezone.utc)}) # A Core UPDATE with the guard as a WHERE — same atomic "only while still running" write as the # rest of the codebase's 2.0 statements, no hand-built SQL string. synchronize_session=False: this # is a terminal write, so we don't pay to sync the (about-to-be-discarded) ORM identity map. Bound # StrEnum values bind as their plain str literal (StrEnum is a str subclass). result = db.execute( update(DownloadJob) .where(DownloadJob.id == job_id, DownloadJob.status == JobStatus.RUNNING.value) .values(**fields) .execution_options(synchronize_session=False) ) return result.rowcount > 0 def _finalize_done(db: Session, job_id: int, fields: dict) -> bool: """`finalize_if_running` for a job that finished: done is when a file's retention starts, its folder's unless the user kept it. The expiry is written only while the job is still where it was read from — a move, a keep, or its folder's delete (which sends its files home) landing while the folder's retention was read sends the read round again, never stamping a folder's retention on a file now in another one.""" while True: held = db.execute( select(DownloadJob.folder_id, DownloadJob.kept, DownloadJob.status).where( DownloadJob.id == job_id ) ).one_or_none() if held is None or held.status != JobStatus.RUNNING: return False stamped = { **fields, "expires_at": folders.done_expiry(db, held.folder_id, held.kept), "expiry_warned": False, } hit = db.execute( update(DownloadJob) .where( DownloadJob.id == job_id, DownloadJob.status == JobStatus.RUNNING.value, DownloadJob.folder_id.is_not_distinct_from(held.folder_id), DownloadJob.kept == held.kept, ) .values(**stamped) .returning(DownloadJob.id) .execution_options(synchronize_session=False) ).scalar_one_or_none() if hit is not None: return True def release_asset(db: Session, job: DownloadJob) -> None: """Drop this job's hold on its asset when it leaves a holding state. Once NO job holds the asset anymore, delete the file + row immediately — a deleted download should free its disk, and the shared cache only needs to span *overlapping* holders (a later re-add just downloads again). `error` counts as holding: enqueue always +1'd ref_count and the worker never decrements on failure (ref_count is the route's job), so an errored job releases here on delete — else its +1 leaks and a later resume→ready keeps the file pinned past its last holder. Idempotent for the cancel path: cancel already released while the job was holding, then set it `canceled` (∉ the set below), so delete-after-cancel is a no-op and never double-releases. We DON'T delete an errored asset row here even at ref==0: another request may be concurrently re-enqueuing the same (source,format) — get_or_create_asset would reset that row to `pending` and +1 it, and deleting it under that would FK-null the new job's asset (a lost update on ref_count). A ready asset is still freed at ref==0 (its file is the point); an orphaned errored row is cheap (no file) and gets reused+reset by the next enqueue, or reclaimed by a later GC pass. An UPLOAD is the exception: it is never shared or re-enqueued (its ref is unique), and while it waits for or failed its phone-safe conversion the original file already sits at `rel_path` — so it is freed at ref==0 whatever its state, except mid-conversion (`downloading`: the worker child still reads it; the GC upload pass takes it once the child has stopped).""" if not job.asset_id or job.status not in HOLDING: return asset = db.get(MediaAsset, job.asset_id) if asset is None: return new_count = decr_ref_count(db, asset.id) freeable = asset.status == "ready" or ( asset.source_kind == "upload" and asset.status != "downloading" ) if new_count == 0 and freeable: if asset.rel_path: storage.delete_asset_files(settings.files_root, asset.rel_path) db.delete(asset) # --- 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 flushes its transition and the caller # commits it; the route keeps only auth + HTTP mapping + serialization. def pause(db: Session, job: DownloadJob) -> None: """Pause a queued/running job (a running worker aborts cooperatively via the cancel poll); a no-op in any other state.""" if job.status in (JobStatus.QUEUED, JobStatus.RUNNING): job.status = JobStatus.PAUSED db.flush() def resume(db: Session, job: DownloadJob) -> None: """Requeue a paused/errored job for a fresh attempt, clearing its error. If the SHARED asset itself failed, reset it to pending so the worker actually re-downloads — otherwise the requeued job would instantly re-inherit the asset's stale error (the worker short-circuits an errored asset). A cookie the user added AFTER the job was queued moves the job onto their private asset first (`_follow_the_cookie`): the worker only ever attaches a cookie to that identity, so without the move the Retry a login-wall message asks for would fail on the same wall.""" if job.status in (JobStatus.PAUSED, JobStatus.ERROR): job.progress = 0 job.error = None job.error_code = None job.error_detail = None if job.asset_id: asset = db.get(MediaAsset, job.asset_id) if asset is not None: asset = _follow_the_cookie(db, job, asset) if asset and asset.status == "error": # MediaAsset.status (no enum), not a JobStatus asset.status = "pending" asset.error = None job.status = JobStatus.QUEUED db.flush() def _follow_the_cookie(db: Session, job: DownloadJob, asset: MediaAsset) -> MediaAsset: """Re-point a requeued job at the asset enqueue would pick NOW, when the user has a cookie for the link's host but the job still holds an asset the worker won't attach it to (queued before the cookie existed). Returns the asset the job holds afterwards. Only a `pending` / `error` asset moves — neither has a file nor a download in flight. A `ready` one already has the file, and a `downloading` one may be this job's own child still being stopped by the pause; the next Retry after it settles moves the job then. Removing a cookie never moves a job back: the worker attaches nothing without a credential, so the private asset is as good as any.""" if asset.status not in ("pending", "error"): return asset url = source_url(job.source_kind, job.source_ref) salt = credentials.salt_for_enqueue(db, job.user_id, url) spec = formats.normalize(job.profile_snapshot) if salt is None or credentials.is_authed_asset(asset, spec, job.user_id, url): return asset target = get_or_create_asset( db, job.source_kind, job.source_ref, formats.format_sig(spec, salt=salt) ) populate_from_catalog(db, target) release_asset(db, job) # drop the old hold BEFORE re-pointing — it keys off job.asset_id incr_ref_count(db, target.id) job.asset_id = target.id return target def cancel(db: Session, job: DownloadJob) -> None: """Cancel a job unless it's already done/canceled, releasing its asset hold WHILE it still counts as holding — `canceled` ∉ HOLDING, so a later delete won't double-release.""" 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.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.flush() def redownload(db: Session, job: DownloadJob) -> None: """Rebuild a finished/failed job's file into a FRESH, private asset (the cache sig is salted with the new asset's own id, so it never reuses the dedup pool) and requeue the SAME job row — so any public share link (DownloadLink -> job_id) keeps resolving to the newly-downloaded file. The old hold is dropped first (freeing the old file iff this job was its last holder). The caller has already rejected edit clips + still-in-progress jobs (HTTP 409). When the user has a cookie for the link's host, the salt carries it (`credentials.rebuild_salt`), so the worker recognises the rebuild as their private identity and attaches the cookie — a cookie download, or one the user added after a failure, rebuilds with the cookie.""" spec = formats.normalize(job.profile_snapshot) cookie = credentials.salt_for_enqueue( db, job.user_id, source_url(job.source_kind, job.source_ref) ) # The salt names the row's own id, which exists only once the row does: insert it under a unique # placeholder, then write the real signature in the same transaction (nobody sees the placeholder). asset = MediaAsset( source_kind=job.source_kind, source_ref=job.source_ref, format_sig=f"rebuild:{uuid.uuid4().hex}", status="pending", ref_count=0, ) db.add(asset) db.flush() asset.format_sig = formats.format_sig(spec, salt=credentials.rebuild_salt(asset.id, cookie)) db.flush() populate_from_catalog(db, asset) # Drop the old hold BEFORE re-pointing the job — release_asset keys off the job's current asset. release_asset(db, job) incr_ref_count(db, asset.id) job.asset_id = asset.id job.status = JobStatus.QUEUED job.progress = 0 job.phase = None job.error = None job.error_code = None # a rebuild that fails again gets its own diagnosis, never the old one job.error_detail = None job.speed_bps = None job.eta_s = None # Re-derive the display name from the freshly cleaned title on completion (else an older, un-cleaned # name would survive the rebuild). job.display_name = 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.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) )