"""Enqueue + cache resolution for the download center (the per-user layer over MediaAsset). `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. """ from datetime import datetime, timezone from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.config import settings from app.downloads import adapters from app.downloads import edit as editmod from app.downloads import formats, quota, storage from app.downloads.status import HOLDING from app.models import Channel, DownloadJob, MediaAsset, Video 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 private Downloads page 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); 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 == "edit": return None 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 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, ) asset = db.execute(stmt).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, ) db.add(asset) try: db.flush() except IntegrityError: db.rollback() 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 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) sig = formats.format_sig(spec) 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[:255] if asset.status == "ready" and asset.title else None), status="queued", queue_pos=next_queue_pos(db), ) db.add(job) asset.ref_count = (asset.ref_count or 0) + 1 # Cache hit: the file already exists — no worker round-trip needed. if asset.status == "ready": job.status = "done" job.progress = 100 asset.last_access_at = datetime.now(timezone.utc) db.commit() db.refresh(job) return job def _clip_title(source_title: str | None, display_name: str | None) -> str: if display_name: return display_name[:255] base = source_title or "Clip" return f"{base} (clip)"[:255] 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") 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) job = DownloadJob( user_id=user_id, asset_id=asset.id, job_kind="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, status="queued", queue_pos=next_queue_pos(db), ) db.add(job) asset.ref_count = (asset.ref_count or 0) + 1 if asset.status == "ready": # identical clip already produced for this user → instant job.status = "done" job.progress = 100 asset.last_access_at = datetime.now(timezone.utc) db.commit() db.refresh(job) return job 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.""" if not job.asset_id or job.status not in HOLDING: return asset = db.get(MediaAsset, job.asset_id) if asset is None: return asset.ref_count = max(0, (asset.ref_count or 0) - 1) if asset.ref_count == 0 and asset.status == "ready": if asset.rel_path: storage.delete_asset_files(settings.download_root, asset.rel_path) db.delete(asset)