diff --git a/backend/app/downloads/jobdb.py b/backend/app/downloads/jobdb.py
new file mode 100644
index 00000000..ff10eac9
--- /dev/null
+++ b/backend/app/downloads/jobdb.py
@@ -0,0 +1,73 @@
+"""Short-lived-session DB helpers + staging paths shared by the worker parent and the child pipeline."""
+from pathlib import Path
+
+from sqlalchemy import text
+
+from app.config import settings
+from app.db import SessionLocal
+from app.models import DownloadJob, MediaAsset
+
+
+# --- small DB helpers (short-lived sessions; the download itself holds no transaction) -------
+
+def _set_job(job_id: int, **fields) -> None:
+ with SessionLocal() as db:
+ job = db.get(DownloadJob, job_id)
+ if job is None:
+ return
+ for k, v in fields.items():
+ setattr(job, k, v)
+ db.commit()
+
+
+def _job_status(job_id: int) -> str | None:
+ with SessionLocal() as db:
+ return db.execute(
+ text("SELECT status FROM download_jobs WHERE id = :id"), {"id": job_id}
+ ).scalar()
+
+
+def _touch_job(job_id: int) -> None:
+ """Bump a running job's `updated_at` without changing any value — a stall-watchdog heartbeat for
+ phases that legitimately write no progress. Extraction writes nothing, and a multi-client-set
+ fallback re-writes identical values (progress 0 / phase 'downloading'), so ORM `onupdate` would
+ NOT fire and two slow extraction attempts could otherwise cross the stall timeout and be killed
+ mid-extraction though they'd have succeeded."""
+ with SessionLocal() as db:
+ db.execute(
+ text("UPDATE download_jobs SET updated_at=now() WHERE id=:id AND status='running'"),
+ {"id": job_id},
+ )
+ db.commit()
+
+
+def _requeue_if_running(job_id: int, phase: str) -> None:
+ """Put the job back to 'queued' for a later retry, but ONLY while it is still 'running'. The
+ route flips status to paused/cancelled to stop a job; a blind `status='queued'` here would
+ clobber that and silently revive a job the user just cancelled. The conditional WHERE makes the
+ requeue a no-op unless we still own a running job."""
+ with SessionLocal() as db:
+ db.execute(
+ text(
+ "UPDATE download_jobs SET status='queued', phase=:phase, updated_at=now() "
+ "WHERE id=:id AND status='running'"
+ ),
+ {"id": job_id, "phase": phase},
+ )
+ db.commit()
+
+
+def _reset_asset_pending(asset_id: int) -> None:
+ with SessionLocal() as db:
+ asset = db.get(MediaAsset, asset_id)
+ if asset and asset.status == "downloading":
+ asset.status = "pending"
+ db.commit()
+
+
+def _staging_dir(asset_id: int) -> Path:
+ return Path(settings.download_root) / ".staging" / f"asset-{asset_id}"
+
+
+def _edit_staging_dir(asset_id: int) -> Path:
+ return Path(settings.download_root) / ".staging" / f"edit-{asset_id}"
diff --git a/backend/app/downloads/pipeline.py b/backend/app/downloads/pipeline.py
new file mode 100644
index 00000000..2b0e0e40
--- /dev/null
+++ b/backend/app/downloads/pipeline.py
@@ -0,0 +1,592 @@
+"""Child-side download/edit pipeline — the yt-dlp download + ffmpeg edit that runs in the spawned subprocess."""
+import logging
+import re
+import shutil
+import time
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+
+import yt_dlp
+
+from app import sysconfig
+from app.config import settings
+from app.db import SessionLocal
+from app.downloads import edit as editmod
+from app.downloads import errors as dlerrors
+from app.downloads import adapters, formats, service, storage
+from app.downloads.jobdb import (
+ _edit_staging_dir,
+ _job_status,
+ _reset_asset_pending,
+ _set_job,
+ _staging_dir,
+ _touch_job,
+)
+from app.downloads.status import JobStatus
+from app.models import DownloadJob, MediaAsset
+from app.notifications import create_notification
+from app.titles import normalize_title
+
+log = logging.getLogger("siftlode.worker")
+
+
+class _Aborted(Exception):
+ """Raised inside the progress hook when the job was paused/canceled mid-download."""
+
+
+# Both progress sources call back FAR faster than the database should be touched: yt-dlp fires its
+# hook per chunk, and ffmpeg's `-progress pipe:1` emits a block of ~12 lines every ~0.5s while
+# run_ffmpeg calls its callbacks once per LINE. Every callback here is a round-trip (`_job_status`
+# opens a session and SELECTs; `_set_job` opens one and UPDATE+COMMITs), so unthrottled they cost
+# ~10^5 queries on a long job. ONE interval, ONE implementation, used by both paths — three
+# hand-rolled copies is how the ffmpeg side ended up unthrottled while the download side was fine.
+_DB_WRITE_INTERVAL_S = 1.0
+_CANCEL_POLL_INTERVAL_S = 1.0
+
+_THUMB_SUFFIXES = (".jpg", ".jpeg", ".png", ".webp")
+# yt-dlp's working files in the staging dir. None of them is ever the finished media, but several
+# CAN be the biggest file there (a video-only per-format stream, an abandoned fragment run), so the
+# staging fallback must exclude them by name — picking one would store e.g. a silent video-only
+# stream as a ready asset.
+_YTDLP_TEMP_RE = re.compile(
+ r"\.part(-Frag\d+)?$" # in-flight download / fragment
+ r"|\.ytdl$" # resume state
+ r"|\.temp\.[^.]+$" # merge/postprocess temp
+ r"|\.f\d+\.[^.]+$", # per-format stream kept before the merge
+ re.IGNORECASE,
+)
+
+
+def _notify_download_done(
+ db, user_id: int | None, title: str | None, job_id, *, kind: str
+) -> None:
+ """Post a best-effort 'ready' notice to the owner's inbox — downloads/edits used to finish
+ silently while scheduler jobs and playlist-push already notified (R11 feedback symmetry). A
+ notification failure must never mask the completed job, so it is isolated + rolled back (mirrors
+ scheduler._notify_done). `kind` ("download"/"edit") lets the inbox pick accurate copy; the
+ English title here is only a fallback for an unknown type. `user_id`/`title` are captured by the
+ caller BEFORE its commit, so reading them here needs no post-commit refresh of an expired row."""
+ if user_id is None:
+ return
+ try:
+ create_notification(
+ db,
+ user_id=user_id,
+ type="download_done",
+ title=f"Ready to watch: {title}" if title else "Ready to watch",
+ data={"title": title, "kind": kind},
+ )
+ except Exception:
+ db.rollback()
+ log.exception("failed to write download-done notification for job %s", job_id)
+
+
+def _parse_upload_date(raw) -> date | None:
+ if not raw:
+ return None
+ try:
+ return datetime.strptime(str(raw), "%Y%m%d").date()
+ except ValueError:
+ return None
+
+
+def _fill_asset_meta_early(asset_id: int, info: dict) -> None:
+ """Populate an asset's display metadata from yt-dlp's info_dict the first time it's seen
+ (only if not already filled at enqueue from our catalog) — so an ad-hoc URL's row shows a
+ real title + thumbnail while it downloads, not just the URL. Zero extra network cost."""
+ with SessionLocal() as db:
+ asset = db.get(MediaAsset, asset_id)
+ if asset is None or asset.title:
+ return
+ # `or info.get("id")` avoids a blank card while downloading when the cleaned title is empty
+ # (e.g. a title that was only a social engagement prefix); the completion path later sets
+ # the final title with its own source_ref fallback.
+ asset.title = normalize_title(info.get("title")) or info.get("id")
+ asset.uploader = info.get("uploader") or info.get("channel")
+ asset.uploader_url = adapters.resolve_adapter(None, None, info).uploader_url(info)
+ asset.thumbnail_url = info.get("thumbnail")
+ asset.duration_s = int(info["duration"]) if info.get("duration") else None
+ asset.upload_date = _parse_upload_date(info.get("upload_date"))
+ asset.source_webpage_url = info.get("webpage_url")
+ db.commit()
+
+
+def _make_progress_hook(job_id: int, asset_id: int):
+ state = {"meta_done": False}
+ cancelled = _cancel_poll(job_id) # same policy as the ffmpeg path
+ allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S)
+
+ def hook(d: dict) -> None:
+ # Cooperative pause/cancel: if the job is no longer 'running', abort the download.
+ if cancelled():
+ raise _Aborted
+ if not state["meta_done"] and d.get("info_dict"):
+ state["meta_done"] = True
+ try:
+ _fill_asset_meta_early(asset_id, d["info_dict"])
+ except Exception: # noqa: BLE001 — metadata is best-effort, never fail a download
+ pass
+ if d.get("status") != "downloading":
+ return
+ if not allow_write():
+ return
+ # yt-dlp downloads the video and audio streams SEPARATELY (each 0→100%), then merges.
+ # Label the phase so the user understands the bar resetting between streams instead of
+ # seeing a mysterious 95%→5% jump.
+ info = d.get("info_dict") or {}
+ vcodec = info.get("vcodec") or "none"
+ acodec = info.get("acodec") or "none"
+ if vcodec != "none" and acodec == "none":
+ phase = "video"
+ elif acodec != "none" and vcodec == "none":
+ phase = "audio"
+ else:
+ phase = "media"
+ total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
+ done = d.get("downloaded_bytes") or 0
+ pct = int(done * 100 / total) if total else 0
+ _set_job(
+ job_id,
+ progress=min(pct, 99),
+ speed_bps=int(d["speed"]) if d.get("speed") else None,
+ eta_s=int(d["eta"]) if d.get("eta") else None,
+ phase=phase,
+ )
+
+ return hook
+
+
+def _rate_limiter(interval_s: float, *, now=time.monotonic):
+ """`should_run()` → True at most once per `interval_s` (the first call always passes).
+ `now` is injectable so tests drive a clock instead of patching the global one."""
+ # `None`, not 0.0: "never ran" has to be a VALUE, not a falsy timestamp. A truthiness test here
+ # reads a legitimate `now() == 0.0` as "never ran" and lets a second call straight through —
+ # unreachable with real `time.monotonic()`, but `now` exists to be injected, and a test clock
+ # starting at 0 is the obvious choice. That would let a throttle test encode the wrong
+ # behaviour and still pass.
+ state: dict[str, float | None] = {"at": None}
+
+ def should_run() -> bool:
+ t = now()
+ last = state["at"]
+ if last is not None and t - last < interval_s:
+ return False
+ state["at"] = t
+ return True
+
+ return should_run
+
+
+def _cancel_poll(job_id: int, interval_s: float = _CANCEL_POLL_INTERVAL_S, *, now=time.monotonic):
+ """Cooperative "has this job been paused/cancelled?" poll, rate-limited and sticky: once it
+ says yes it never queries again. Shared by the yt-dlp hook and run_ffmpeg so cancel latency is
+ one policy, not one per call site."""
+ allow = _rate_limiter(interval_s, now=now)
+ state = {"cancelled": False}
+
+ def cancelled() -> bool:
+ if not state["cancelled"] and allow():
+ state["cancelled"] = _job_status(job_id) not in ("running", None)
+ return state["cancelled"]
+
+ return cancelled
+
+
+def _ffmpeg_progress_writer(job_id: int, phase: str, *, now=time.monotonic):
+ """`on_progress` for run_ffmpeg: persist at most one row write per interval, and only when the
+ whole-percent value actually changed (the bar shows integers)."""
+ allow = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now)
+ state = {"pct": -1}
+
+ def write(pct: float) -> None:
+ if not allow():
+ return
+ whole = int(pct)
+ if whole == state["pct"]:
+ # Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264
+ # or a crop on the CPU-only host) can sit on one percent for minutes. Heartbeat the row's
+ # updated_at so the stall watchdog doesn't kill a live job as "stalled" (B-sub-4), without
+ # rewriting the unchanged progress bar.
+ _touch_job(job_id)
+ return
+ state["pct"] = whole
+ _set_job(job_id, progress=whole, phase=phase)
+
+ return write
+
+
+def _pp_phase(pp: str) -> str:
+ """Map a yt-dlp postprocessor class name to a user-facing phase label. ffmpeg post-steps
+ aren't byte-progress operations (no %), so the UI shows the step name + an indeterminate
+ pulse — this makes the long merge/finalize legible instead of a silent 'Processing'."""
+ if "Merg" in pp:
+ return "merging"
+ if "ExtractAudio" in pp:
+ return "audio_extract"
+ if "Thumbnail" in pp:
+ return "thumbnail"
+ if "SponsorBlock" in pp or "ModifyChapters" in pp:
+ return "sponsorblock"
+ if "Metadata" in pp:
+ return "metadata"
+ return "processing"
+
+
+def _make_pp_hook(job_id: int):
+ """Postprocessor hook: surface the current ffmpeg step so the bar shows a concrete label
+ (Merging / Embedding thumbnail / …) instead of freezing at 100% after the streams finish."""
+
+ def hook(d: dict) -> None:
+ if d.get("status") in ("started", "processing"):
+ _set_job(job_id, phase=_pp_phase(d.get("postprocessor") or ""), speed_bps=None, eta_s=None)
+
+ return hook
+
+
+def _find_thumbnail(staging: Path, media: Path) -> Path | None:
+ for p in staging.iterdir():
+ if p != media and p.suffix.lower() in _THUMB_SUFFIXES:
+ return p
+ return None
+
+
+def _produced_file(info: dict, staging: Path) -> Path:
+ """The media file yt-dlp actually wrote.
+
+ `requested_downloads[0].filepath` is the authoritative answer on YouTube, but an arbitrary-URL
+ extractor may not fill it in at all — and `Path("")` is `Path(".")`, which *passes* `.exists()`,
+ so a missing value used to sail past the fallback and blow up two steps later (`with_name` on an
+ empty name: the BBC-news failure). Anything that isn't a real file counts as absent; then fall
+ back to the largest file left in staging that is neither a thumbnail nor one of yt-dlp's own
+ working files (see `_YTDLP_TEMP_RE`)."""
+ req = (info.get("requested_downloads") or [{}])[0]
+ fp = req.get("filepath") or req.get("_filename") or info.get("filepath") or info.get("_filename")
+ if fp:
+ produced = Path(str(fp))
+ if produced.is_file():
+ return produced
+ cands = [
+ p
+ for p in staging.iterdir()
+ if p.is_file()
+ and p.suffix.lower() not in _THUMB_SUFFIXES
+ and not _YTDLP_TEMP_RE.search(p.name)
+ ]
+ if not cands:
+ raise RuntimeError("yt-dlp produced no output file")
+ return max(cands, key=lambda p: p.stat().st_size)
+
+
+def _media_info(info: dict) -> dict:
+ """Unwrap a playlist/multi_video extraction to the entry that carries the download.
+
+ A single non-YouTube page (a news article's video, a multi-rendition embed) can extract as a
+ wrapper whose own dict has no `requested_downloads` — the media lives on an entry. Pick the
+ FIRST entry that actually downloaded something: on a page with a skipped teaser in front of the
+ real video, entries[0] carries no file, and taking it blindly would label the produced file
+ with the wrong id/title/duration."""
+ entries = [e for e in (info.get("entries") or []) if e]
+ if not entries or info.get("requested_downloads"):
+ return info
+ chosen = next((e for e in entries if e.get("requested_downloads")), entries[0])
+ return {**info, **chosen}
+
+
+def _extract_with_fallback(job_id, asset_id, url, spec, staging, adapter):
+ """Download via yt-dlp using the source `adapter`'s client sets + options. For YouTube the adapter
+ supplies the primary player-client set then a POT-backed fallback (retried on a bot/DRM/format
+ error); other sources make a single attempt. Returns the info dict of the successful attempt."""
+ client_sets = adapter.client_sets()
+ outtmpl = str(staging / "%(id)s.%(ext)s")
+ last_exc: Exception | None = None
+ for i, clients in enumerate(client_sets):
+ opts = formats.build_ydl_opts(
+ spec, outtmpl, _make_progress_hook(job_id, asset_id),
+ extra=adapter.ydl_opts_extra(clients),
+ )
+ opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
+ _touch_job(job_id) # stall-watchdog heartbeat: extraction writes no progress, so a slow /
+ # multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at.
+ # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still
+ # killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.)
+ try:
+ with yt_dlp.YoutubeDL(opts) as ydl:
+ return ydl.extract_info(url, download=True)
+ except yt_dlp.utils.DownloadError as exc:
+ last_exc = exc
+ if i < len(client_sets) - 1 and adapter.retriable(str(exc)):
+ log.info("job %s: clients %s failed (%s); retrying with fallback", job_id, clients, str(exc)[:70])
+ for p in list(staging.iterdir()): # drop partials before the retry
+ try:
+ p.unlink()
+ except OSError:
+ pass
+ _set_job(job_id, progress=0, phase="downloading")
+ continue
+ raise
+ raise last_exc # unreachable (loop either returns or raises)
+
+
+def _download_settings() -> tuple[str, int]:
+ """Layout + retention days from the DB config (admin-overridable on the Configuration page),
+ falling back to the env defaults. Read here, not from `settings`, so an admin edit takes effect."""
+ with SessionLocal() as db:
+ return (
+ sysconfig.get_str(db, "download_layout"),
+ sysconfig.get_int(db, "download_retention_days"),
+ )
+
+
+def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict) -> Path:
+ """Make an mp4-compat-profile download play in every browser: H.264 + AAC/MP3 in mp4. If the
+ download (whatever the source shipped) isn't already that, re-encode the offending stream(s) —
+ or just remux when the codecs are fine but the container isn't. **Scope:** only the mp4 compat
+ profile is normalized; an explicit `vcodec` or a non-mp4 custom profile is the user opting out
+ of the guarantee and is left as-is (so a `webm` profile can still ship VP9/Opus). Returns the
+ file to store (converted, or the original) and corrects `info`'s codec fields to match disk."""
+ if spec.get("container") != "mp4" or spec.get("vcodec"):
+ return produced
+ vcodec, acodec = editmod.probe_codecs(produced)
+ is_mp4 = produced.suffix.lower() == ".mp4"
+ needs, reenc_v, reenc_a = editmod.browser_transcode_flags(vcodec, acodec, is_mp4)
+ if not needs:
+ return produced
+
+ dest = produced.with_name(produced.stem + ".browser.mp4")
+ action = "re-encode" if (reenc_v or reenc_a) else "remux"
+ log.info("job %s: %s (v=%s a=%s mp4=%s) -> browser-safe mp4", job_id, action, vcodec, acodec, is_mp4)
+ _set_job(job_id, phase="processing", progress=0, speed_bps=None, eta_s=None)
+ # Prefer the info duration; probe the file when the extractor didn't provide one, so a long
+ # re-encode still shows real progress instead of a frozen 0%.
+ total_s = int(info["duration"]) if info.get("duration") else editmod.probe_duration(produced)
+ editmod.run_ffmpeg(
+ editmod.build_browser_safe_cmd(produced, dest, reenc_v, reenc_a),
+ total_s=total_s,
+ on_progress=_ffmpeg_progress_writer(job_id, "processing"),
+ should_cancel=_cancel_poll(job_id),
+ )
+ produced.unlink(missing_ok=True)
+ # Keep the stored asset metadata truthful about what's now on disk.
+ if reenc_v:
+ info["vcodec"] = "h264"
+ if reenc_a:
+ info["acodec"] = "aac"
+ return dest
+
+
+def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spec: dict) -> None:
+ staging = _staging_dir(asset_id)
+ try:
+ if staging.exists():
+ shutil.rmtree(staging, ignore_errors=True)
+ staging.mkdir(parents=True, exist_ok=True)
+
+ _set_job(job_id, phase="downloading", progress=0)
+ adapter = adapters.resolve_adapter(source_kind, source_ref)
+ url = service.source_url(source_kind, source_ref)
+ info = _media_info(_extract_with_fallback(job_id, asset_id, url, spec, staging, adapter))
+ # Re-resolve now that yt-dlp's info reveals the true source (a bare/redirected url only
+ # identifies post-extraction) so the .nfo uniqueid type + uploader_url agree on one source.
+ adapter = adapters.resolve_adapter(source_kind, source_ref, info)
+
+ produced = _produced_file(info, staging)
+ produced = _ensure_browser_playable(job_id, produced, spec, info)
+ ext = produced.suffix.lstrip(".").lower()
+ meta = storage.MediaMeta(
+ video_id=info.get("id") or source_ref,
+ title=normalize_title(info.get("title")) or source_ref,
+ uploader=info.get("uploader") or info.get("channel") or "Unknown Channel",
+ upload_date=_parse_upload_date(info.get("upload_date")),
+ uniqueid_type=adapter.nfo_uniqueid_type(),
+ duration_s=int(info["duration"]) if info.get("duration") else None,
+ description=info.get("description"),
+ thumbnail_url=info.get("thumbnail"),
+ )
+
+ layout, retention_days = _download_settings()
+ rel = storage.rel_path(meta, ext, layout, asset_id=asset_id)
+ thumb = _find_thumbnail(staging, produced)
+ # Convert the raw thumbnail (webp/png/jpeg) to a real JPEG ourselves — robustly — for the
+ # poster + og:image sidecar, now that yt-dlp's fatal thumbnail convertor is gone. On failure
+ # this is None and write_sidecars writes no poster, so ensure_poster cuts a frame instead.
+ poster_jpg = editmod.thumbnail_to_jpg(thumb, staging / "poster.jpg")
+ storage.place_file(produced, settings.download_root, rel)
+ nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, poster_jpg)
+
+ final = storage.abs_path(settings.download_root, rel)
+ size = final.stat().st_size if final.exists() else None
+ expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
+
+ with SessionLocal() as db:
+ asset = db.get(MediaAsset, asset_id)
+ if asset:
+ asset.status = "ready"
+ asset.rel_path = rel
+ asset.size_bytes = size
+ asset.duration_s = meta.duration_s
+ asset.width = int(info["width"]) if info.get("width") else None
+ asset.height = int(info["height"]) if info.get("height") else None
+ asset.container = ext
+ asset.vcodec = info.get("vcodec") if info.get("vcodec") not in (None, "none") else None
+ asset.acodec = info.get("acodec") if info.get("acodec") not in (None, "none") else None
+ asset.title = meta.title
+ asset.uploader = meta.uploader
+ asset.uploader_url = adapter.uploader_url(info)
+ asset.upload_date = meta.upload_date
+ asset.thumbnail_url = meta.thumbnail_url
+ # Record a self-hosted poster for every download: it's the reliable link-preview
+ # (og:image) source — a remote thumbnail, esp. Facebook's signed CDN URL, expires
+ # and can't be relied on cross-origin. write_sidecars already wrote .jpg from
+ # the thumbnail; ensure_poster returns that, or cuts a frame if there was none.
+ asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
+ asset.source_webpage_url = info.get("webpage_url")
+ asset.nfo_written = nfo_ok
+ asset.error = None
+ asset.last_access_at = datetime.now(timezone.utc)
+ asset.expires_at = expires
+ job = db.get(DownloadJob, job_id)
+ # Capture the owner before commit expires the row (avoids a refresh SELECT just to notify).
+ owner_id = job.user_id if job else None
+ fields = {"status": JobStatus.DONE, "progress": 100, "phase": None, "speed_bps": None,
+ "eta_s": None}
+ if job and not job.display_name and meta.title:
+ # display_name is VARCHAR(255) — clip a long title (see _fill_asset_meta_early).
+ fields["display_name"] = meta.title[:255]
+ # Atomic terminal write: finalizes ONLY while the row is still running, so a child that
+ # finishes just after a route cancel/pause can't resurrect it (see finalize_if_running).
+ finalized = service.finalize_if_running(db, job_id, **fields)
+ db.commit()
+ if finalized:
+ _notify_download_done(db, owner_id, meta.title, job_id, kind="download")
+ log.info("job %s done: %s", job_id, rel)
+
+ except (_Aborted, editmod.EditAborted):
+ # Paused/canceled mid-download or mid-transcode: free the asset for a later retry; keep the
+ # job's status (the route set it to paused/canceled). ref_count is adjusted by the route.
+ _reset_asset_pending(asset_id)
+ log.info("job %s aborted (pause/cancel)", job_id)
+ except Exception as exc: # noqa: BLE001 — surface any failure to the user, keep worker alive
+ msg = str(exc)[:500]
+ with SessionLocal() as db:
+ asset = db.get(MediaAsset, asset_id)
+ if asset:
+ asset.status = "error"
+ asset.error = msg
+ # Atomic fail: only while still running, so a route-set cancel/pause is never overwritten
+ # with `error` (error also holds the asset → would re-open the double-release class).
+ service.finalize_if_running(
+ db, job_id, status=JobStatus.ERROR, error=msg,
+ error_code=dlerrors.classify(msg), phase=None,
+ )
+ db.commit()
+ log.warning("job %s failed: %s", job_id, msg)
+ finally:
+ shutil.rmtree(staging, ignore_errors=True)
+
+
+def _fail_edit(job_id: int, asset_id: int, msg: str) -> None:
+ with SessionLocal() as db:
+ asset = db.get(MediaAsset, asset_id)
+ if asset:
+ asset.status = "error"
+ asset.error = msg
+ # Atomic fail: only while still running (see the download error path).
+ service.finalize_if_running(
+ db, job_id, status=JobStatus.ERROR, error=msg,
+ error_code=dlerrors.classify(msg), phase=None,
+ )
+ db.commit()
+ log.warning("edit job %s failed: %s", job_id, msg)
+
+
+def _process_edit(job_id: int, asset_id: int) -> None:
+ """Produce a per-user trim/crop clip from the source asset's downloaded file via ffmpeg."""
+ with SessionLocal() as db:
+ job = db.get(DownloadJob, job_id)
+ src = db.get(MediaAsset, job.source_asset_id) if job and job.source_asset_id else None
+ edit_spec = dict(job.edit_spec or {}) if job else {}
+ user_id = job.user_id if job else None
+ src_ready = src is not None and src.status == "ready" and bool(src.rel_path)
+ src_rel = src.rel_path if src else None
+ src_w, src_h, src_dur = (src.width, src.height, src.duration_s) if src else (None, None, None)
+
+ if not src_ready or not src_rel:
+ _fail_edit(job_id, asset_id, "The source download is no longer available.")
+ return
+ src_path = storage.abs_path(settings.download_root, src_rel)
+ if not src_path.exists():
+ _fail_edit(job_id, asset_id, "The source file is missing.")
+ return
+
+ src_ext = src_path.suffix.lstrip(".").lower() or "mp4"
+ crop = editmod.has_crop(edit_spec)
+ out_ext = "mp4" if editmod.needs_reencode(edit_spec) else src_ext
+ total_s = editmod.clip_duration(edit_spec, src_dur) or src_dur or 0
+ staging = _edit_staging_dir(asset_id)
+
+ try:
+ if staging.exists():
+ shutil.rmtree(staging, ignore_errors=True)
+ staging.mkdir(parents=True, exist_ok=True)
+ dest_staging = staging / f"out.{out_ext}"
+
+ _set_job(job_id, phase="editing", progress=0, speed_bps=None, eta_s=None)
+ if edit_spec.get("segments"):
+ cmd, prep = editmod.build_concat_plan(src_path, dest_staging, edit_spec, out_ext, staging)
+ for p, content in prep.items():
+ p.write_text(content, encoding="utf-8")
+ else:
+ cmd = editmod.build_edit_cmd(src_path, dest_staging, edit_spec, out_ext)
+ editmod.run_ffmpeg(
+ cmd,
+ total_s,
+ on_progress=_ffmpeg_progress_writer(job_id, "editing"),
+ should_cancel=_cancel_poll(job_id),
+ )
+ if not dest_staging.exists():
+ raise RuntimeError("ffmpeg produced no output file")
+
+ rel = storage.edit_rel_path(user_id, asset_id, out_ext)
+ storage.place_file(dest_staging, settings.download_root, rel)
+ final = storage.abs_path(settings.download_root, rel)
+ size = final.stat().st_size if final.exists() else None
+ _, retention_days = _download_settings()
+ expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
+
+ with SessionLocal() as db:
+ asset = db.get(MediaAsset, asset_id)
+ if asset:
+ asset.status = "ready"
+ asset.rel_path = rel
+ asset.size_bytes = size
+ asset.container = out_ext
+ asset.duration_s = total_s or asset.duration_s
+ if crop:
+ asset.width = edit_spec["crop"]["w"]
+ asset.height = edit_spec["crop"]["h"]
+ else:
+ asset.width, asset.height = src_w, src_h
+ asset.error = None
+ asset.last_access_at = datetime.now(timezone.utc)
+ asset.expires_at = expires
+ job = db.get(DownloadJob, job_id)
+ # Capture owner + title before commit expires the rows (avoids a refresh SELECT).
+ owner_id = job.user_id if job else None
+ dl_title = asset.title if asset else None
+ # Atomic terminal write (see the download path): don't resurrect a route-stopped job.
+ finalized = service.finalize_if_running(
+ db, job_id, status=JobStatus.DONE, progress=100, phase=None
+ )
+ db.commit()
+ if finalized:
+ _notify_download_done(db, owner_id, dl_title, job_id, kind="edit")
+ log.info("edit job %s done: %s", job_id, rel)
+
+ except editmod.EditAborted:
+ # Paused/canceled mid-encode: free the asset for a later retry (ref_count is the route's job).
+ _reset_asset_pending(asset_id)
+ log.info("edit job %s aborted (pause/cancel)", job_id)
+ except Exception as exc: # noqa: BLE001 — surface the failure, keep the worker alive
+ _fail_edit(job_id, asset_id, str(exc)[:500])
+ finally:
+ shutil.rmtree(staging, ignore_errors=True)
diff --git a/backend/app/downloads/runner.py b/backend/app/downloads/runner.py
index 44e9f64d..78515670 100644
--- a/backend/app/downloads/runner.py
+++ b/backend/app/downloads/runner.py
@@ -38,8 +38,8 @@ def main() -> None:
return
# Imported here (not at module top) so the child pays the yt-dlp import cost only when it has
- # real work, and to keep the import graph obvious: runner is a thin entry over worker's helpers.
- from app.worker import _download, _process_edit
+ # real work, and to keep the import graph obvious: runner is a thin entry over the pipeline.
+ from app.downloads.pipeline import _download, _process_edit
if job_kind == "edit":
_process_edit(job_id, asset_id)
diff --git a/backend/app/worker.py b/backend/app/worker.py
index 7739cb90..085633d5 100644
--- a/backend/app/worker.py
+++ b/backend/app/worker.py
@@ -17,125 +17,40 @@ Design:
"""
import logging
import os
-import re
import shutil
import signal
import subprocess
import sys
import threading
import time
-from datetime import date, datetime, timedelta, timezone
-from pathlib import Path
+from datetime import datetime, timezone
-import yt_dlp
from sqlalchemy import text
from app import sysconfig
from app.config import settings
from app.db import SessionLocal
from app.downloads import edit as editmod
-from app.downloads import errors as dlerrors
-from app.downloads import adapters, formats, quota, service, storage
+from app.downloads import quota, service
+from app.downloads.jobdb import (
+ _edit_staging_dir,
+ _requeue_if_running,
+ _reset_asset_pending,
+ _set_job,
+ _staging_dir,
+)
from app.downloads.status import STOPPED_BY_ROUTE, JobStatus
from app.models import DownloadJob, MediaAsset
-from app.notifications import create_notification
-from app.titles import normalize_title
log = logging.getLogger("siftlode.worker")
-
-def _notify_download_done(
- db, user_id: int | None, title: str | None, job_id, *, kind: str
-) -> None:
- """Post a best-effort 'ready' notice to the owner's inbox — downloads/edits used to finish
- silently while scheduler jobs and playlist-push already notified (R11 feedback symmetry). A
- notification failure must never mask the completed job, so it is isolated + rolled back (mirrors
- scheduler._notify_done). `kind` ("download"/"edit") lets the inbox pick accurate copy; the
- English title here is only a fallback for an unknown type. `user_id`/`title` are captured by the
- caller BEFORE its commit, so reading them here needs no post-commit refresh of an expired row."""
- if user_id is None:
- return
- try:
- create_notification(
- db,
- user_id=user_id,
- type="download_done",
- title=f"Ready to watch: {title}" if title else "Ready to watch",
- data={"title": title, "kind": kind},
- )
- except Exception:
- db.rollback()
- log.exception("failed to write download-done notification for job %s", job_id)
-
_stop = threading.Event()
-class _Aborted(Exception):
- """Raised inside the progress hook when the job was paused/canceled mid-download."""
-
-
def _handle_signal(signum, frame):
_stop.set()
-# --- small DB helpers (short-lived sessions; the download itself holds no transaction) -------
-
-def _set_job(job_id: int, **fields) -> None:
- with SessionLocal() as db:
- job = db.get(DownloadJob, job_id)
- if job is None:
- return
- for k, v in fields.items():
- setattr(job, k, v)
- db.commit()
-
-
-def _job_status(job_id: int) -> str | None:
- with SessionLocal() as db:
- return db.execute(
- text("SELECT status FROM download_jobs WHERE id = :id"), {"id": job_id}
- ).scalar()
-
-
-def _touch_job(job_id: int) -> None:
- """Bump a running job's `updated_at` without changing any value — a stall-watchdog heartbeat for
- phases that legitimately write no progress. Extraction writes nothing, and a multi-client-set
- fallback re-writes identical values (progress 0 / phase 'downloading'), so ORM `onupdate` would
- NOT fire and two slow extraction attempts could otherwise cross the stall timeout and be killed
- mid-extraction though they'd have succeeded."""
- with SessionLocal() as db:
- db.execute(
- text("UPDATE download_jobs SET updated_at=now() WHERE id=:id AND status='running'"),
- {"id": job_id},
- )
- db.commit()
-
-
-def _requeue_if_running(job_id: int, phase: str) -> None:
- """Put the job back to 'queued' for a later retry, but ONLY while it is still 'running'. The
- route flips status to paused/cancelled to stop a job; a blind `status='queued'` here would
- clobber that and silently revive a job the user just cancelled. The conditional WHERE makes the
- requeue a no-op unless we still own a running job."""
- with SessionLocal() as db:
- db.execute(
- text(
- "UPDATE download_jobs SET status='queued', phase=:phase, updated_at=now() "
- "WHERE id=:id AND status='running'"
- ),
- {"id": job_id, "phase": phase},
- )
- db.commit()
-
-
-def _parse_upload_date(raw) -> date | None:
- if not raw:
- return None
- try:
- return datetime.strptime(str(raw), "%Y%m%d").date()
- except ValueError:
- return None
-
-
# --- claim + recovery -----------------------------------------------------------------------
def _wait_for_schema(timeout: float = 300.0) -> bool:
@@ -506,550 +421,6 @@ def _finish_from_cache(job_id: int, asset_id: int) -> None:
db.commit()
-def _fill_asset_meta_early(asset_id: int, info: dict) -> None:
- """Populate an asset's display metadata from yt-dlp's info_dict the first time it's seen
- (only if not already filled at enqueue from our catalog) — so an ad-hoc URL's row shows a
- real title + thumbnail while it downloads, not just the URL. Zero extra network cost."""
- with SessionLocal() as db:
- asset = db.get(MediaAsset, asset_id)
- if asset is None or asset.title:
- return
- # `or info.get("id")` avoids a blank card while downloading when the cleaned title is empty
- # (e.g. a title that was only a social engagement prefix); the completion path later sets
- # the final title with its own source_ref fallback.
- asset.title = normalize_title(info.get("title")) or info.get("id")
- asset.uploader = info.get("uploader") or info.get("channel")
- asset.uploader_url = adapters.resolve_adapter(None, None, info).uploader_url(info)
- asset.thumbnail_url = info.get("thumbnail")
- asset.duration_s = int(info["duration"]) if info.get("duration") else None
- asset.upload_date = _parse_upload_date(info.get("upload_date"))
- asset.source_webpage_url = info.get("webpage_url")
- db.commit()
-
-
-def _make_progress_hook(job_id: int, asset_id: int):
- state = {"meta_done": False}
- cancelled = _cancel_poll(job_id) # same policy as the ffmpeg path
- allow_write = _rate_limiter(_DB_WRITE_INTERVAL_S)
-
- def hook(d: dict) -> None:
- # Cooperative pause/cancel: if the job is no longer 'running', abort the download.
- if cancelled():
- raise _Aborted
- if not state["meta_done"] and d.get("info_dict"):
- state["meta_done"] = True
- try:
- _fill_asset_meta_early(asset_id, d["info_dict"])
- except Exception: # noqa: BLE001 — metadata is best-effort, never fail a download
- pass
- if d.get("status") != "downloading":
- return
- if not allow_write():
- return
- # yt-dlp downloads the video and audio streams SEPARATELY (each 0→100%), then merges.
- # Label the phase so the user understands the bar resetting between streams instead of
- # seeing a mysterious 95%→5% jump.
- info = d.get("info_dict") or {}
- vcodec = info.get("vcodec") or "none"
- acodec = info.get("acodec") or "none"
- if vcodec != "none" and acodec == "none":
- phase = "video"
- elif acodec != "none" and vcodec == "none":
- phase = "audio"
- else:
- phase = "media"
- total = d.get("total_bytes") or d.get("total_bytes_estimate") or 0
- done = d.get("downloaded_bytes") or 0
- pct = int(done * 100 / total) if total else 0
- _set_job(
- job_id,
- progress=min(pct, 99),
- speed_bps=int(d["speed"]) if d.get("speed") else None,
- eta_s=int(d["eta"]) if d.get("eta") else None,
- phase=phase,
- )
-
- return hook
-
-
-# Both progress sources call back FAR faster than the database should be touched: yt-dlp fires its
-# hook per chunk, and ffmpeg's `-progress pipe:1` emits a block of ~12 lines every ~0.5s while
-# run_ffmpeg calls its callbacks once per LINE. Every callback here is a round-trip (`_job_status`
-# opens a session and SELECTs; `_set_job` opens one and UPDATE+COMMITs), so unthrottled they cost
-# ~10^5 queries on a long job. ONE interval, ONE implementation, used by both paths — three
-# hand-rolled copies is how the ffmpeg side ended up unthrottled while the download side was fine.
-_DB_WRITE_INTERVAL_S = 1.0
-_CANCEL_POLL_INTERVAL_S = 1.0
-
-
-def _rate_limiter(interval_s: float, *, now=time.monotonic):
- """`should_run()` → True at most once per `interval_s` (the first call always passes).
- `now` is injectable so tests drive a clock instead of patching the global one."""
- # `None`, not 0.0: "never ran" has to be a VALUE, not a falsy timestamp. A truthiness test here
- # reads a legitimate `now() == 0.0` as "never ran" and lets a second call straight through —
- # unreachable with real `time.monotonic()`, but `now` exists to be injected, and a test clock
- # starting at 0 is the obvious choice. That would let a throttle test encode the wrong
- # behaviour and still pass.
- state: dict[str, float | None] = {"at": None}
-
- def should_run() -> bool:
- t = now()
- last = state["at"]
- if last is not None and t - last < interval_s:
- return False
- state["at"] = t
- return True
-
- return should_run
-
-
-def _cancel_poll(job_id: int, interval_s: float = _CANCEL_POLL_INTERVAL_S, *, now=time.monotonic):
- """Cooperative "has this job been paused/cancelled?" poll, rate-limited and sticky: once it
- says yes it never queries again. Shared by the yt-dlp hook and run_ffmpeg so cancel latency is
- one policy, not one per call site."""
- allow = _rate_limiter(interval_s, now=now)
- state = {"cancelled": False}
-
- def cancelled() -> bool:
- if not state["cancelled"] and allow():
- state["cancelled"] = _job_status(job_id) not in ("running", None)
- return state["cancelled"]
-
- return cancelled
-
-
-def _ffmpeg_progress_writer(job_id: int, phase: str, *, now=time.monotonic):
- """`on_progress` for run_ffmpeg: persist at most one row write per interval, and only when the
- whole-percent value actually changed (the bar shows integers)."""
- allow = _rate_limiter(_DB_WRITE_INTERVAL_S, now=now)
- state = {"pct": -1}
-
- def write(pct: float) -> None:
- if not allow():
- return
- whole = int(pct)
- if whole == state["pct"]:
- # Same integer percent, but ffmpeg IS still working — a slow re-encode (AV1/VP9 -> H.264
- # or a crop on the CPU-only host) can sit on one percent for minutes. Heartbeat the row's
- # updated_at so the stall watchdog doesn't kill a live job as "stalled" (B-sub-4), without
- # rewriting the unchanged progress bar.
- _touch_job(job_id)
- return
- state["pct"] = whole
- _set_job(job_id, progress=whole, phase=phase)
-
- return write
-
-
-def _pp_phase(pp: str) -> str:
- """Map a yt-dlp postprocessor class name to a user-facing phase label. ffmpeg post-steps
- aren't byte-progress operations (no %), so the UI shows the step name + an indeterminate
- pulse — this makes the long merge/finalize legible instead of a silent 'Processing'."""
- if "Merg" in pp:
- return "merging"
- if "ExtractAudio" in pp:
- return "audio_extract"
- if "Thumbnail" in pp:
- return "thumbnail"
- if "SponsorBlock" in pp or "ModifyChapters" in pp:
- return "sponsorblock"
- if "Metadata" in pp:
- return "metadata"
- return "processing"
-
-
-def _make_pp_hook(job_id: int):
- """Postprocessor hook: surface the current ffmpeg step so the bar shows a concrete label
- (Merging / Embedding thumbnail / …) instead of freezing at 100% after the streams finish."""
-
- def hook(d: dict) -> None:
- if d.get("status") in ("started", "processing"):
- _set_job(job_id, phase=_pp_phase(d.get("postprocessor") or ""), speed_bps=None, eta_s=None)
-
- return hook
-
-
-def _staging_dir(asset_id: int) -> Path:
- return Path(settings.download_root) / ".staging" / f"asset-{asset_id}"
-
-
-_THUMB_SUFFIXES = (".jpg", ".jpeg", ".png", ".webp")
-# yt-dlp's working files in the staging dir. None of them is ever the finished media, but several
-# CAN be the biggest file there (a video-only per-format stream, an abandoned fragment run), so the
-# staging fallback must exclude them by name — picking one would store e.g. a silent video-only
-# stream as a ready asset.
-_YTDLP_TEMP_RE = re.compile(
- r"\.part(-Frag\d+)?$" # in-flight download / fragment
- r"|\.ytdl$" # resume state
- r"|\.temp\.[^.]+$" # merge/postprocess temp
- r"|\.f\d+\.[^.]+$", # per-format stream kept before the merge
- re.IGNORECASE,
-)
-
-
-def _find_thumbnail(staging: Path, media: Path) -> Path | None:
- for p in staging.iterdir():
- if p != media and p.suffix.lower() in _THUMB_SUFFIXES:
- return p
- return None
-
-
-def _produced_file(info: dict, staging: Path) -> Path:
- """The media file yt-dlp actually wrote.
-
- `requested_downloads[0].filepath` is the authoritative answer on YouTube, but an arbitrary-URL
- extractor may not fill it in at all — and `Path("")` is `Path(".")`, which *passes* `.exists()`,
- so a missing value used to sail past the fallback and blow up two steps later (`with_name` on an
- empty name: the BBC-news failure). Anything that isn't a real file counts as absent; then fall
- back to the largest file left in staging that is neither a thumbnail nor one of yt-dlp's own
- working files (see `_YTDLP_TEMP_RE`)."""
- req = (info.get("requested_downloads") or [{}])[0]
- fp = req.get("filepath") or req.get("_filename") or info.get("filepath") or info.get("_filename")
- if fp:
- produced = Path(str(fp))
- if produced.is_file():
- return produced
- cands = [
- p
- for p in staging.iterdir()
- if p.is_file()
- and p.suffix.lower() not in _THUMB_SUFFIXES
- and not _YTDLP_TEMP_RE.search(p.name)
- ]
- if not cands:
- raise RuntimeError("yt-dlp produced no output file")
- return max(cands, key=lambda p: p.stat().st_size)
-
-
-def _media_info(info: dict) -> dict:
- """Unwrap a playlist/multi_video extraction to the entry that carries the download.
-
- A single non-YouTube page (a news article's video, a multi-rendition embed) can extract as a
- wrapper whose own dict has no `requested_downloads` — the media lives on an entry. Pick the
- FIRST entry that actually downloaded something: on a page with a skipped teaser in front of the
- real video, entries[0] carries no file, and taking it blindly would label the produced file
- with the wrong id/title/duration."""
- entries = [e for e in (info.get("entries") or []) if e]
- if not entries or info.get("requested_downloads"):
- return info
- chosen = next((e for e in entries if e.get("requested_downloads")), entries[0])
- return {**info, **chosen}
-
-
-def _extract_with_fallback(job_id, asset_id, url, spec, staging, adapter):
- """Download via yt-dlp using the source `adapter`'s client sets + options. For YouTube the adapter
- supplies the primary player-client set then a POT-backed fallback (retried on a bot/DRM/format
- error); other sources make a single attempt. Returns the info dict of the successful attempt."""
- client_sets = adapter.client_sets()
- outtmpl = str(staging / "%(id)s.%(ext)s")
- last_exc: Exception | None = None
- for i, clients in enumerate(client_sets):
- opts = formats.build_ydl_opts(
- spec, outtmpl, _make_progress_hook(job_id, asset_id),
- extra=adapter.ydl_opts_extra(clients),
- )
- opts["postprocessor_hooks"] = [_make_pp_hook(job_id)]
- _touch_job(job_id) # stall-watchdog heartbeat: extraction writes no progress, so a slow /
- # multi-client-set fallback would otherwise cross the stall timeout with a stale updated_at.
- # (Per-ATTEMPT only; a single extract_info that itself runs past the stall timeout is still
- # killed — acceptable, as socket_timeout + extractor_retries bound one attempt well under it.)
- try:
- with yt_dlp.YoutubeDL(opts) as ydl:
- return ydl.extract_info(url, download=True)
- except yt_dlp.utils.DownloadError as exc:
- last_exc = exc
- if i < len(client_sets) - 1 and adapter.retriable(str(exc)):
- log.info("job %s: clients %s failed (%s); retrying with fallback", job_id, clients, str(exc)[:70])
- for p in list(staging.iterdir()): # drop partials before the retry
- try:
- p.unlink()
- except OSError:
- pass
- _set_job(job_id, progress=0, phase="downloading")
- continue
- raise
- raise last_exc # unreachable (loop either returns or raises)
-
-
-def _download_settings() -> tuple[str, int]:
- """Layout + retention days from the DB config (admin-overridable on the Configuration page),
- falling back to the env defaults. Read here, not from `settings`, so an admin edit takes effect."""
- with SessionLocal() as db:
- return (
- sysconfig.get_str(db, "download_layout"),
- sysconfig.get_int(db, "download_retention_days"),
- )
-
-
-def _ensure_browser_playable(job_id: int, produced: Path, spec: dict, info: dict) -> Path:
- """Make an mp4-compat-profile download play in every browser: H.264 + AAC/MP3 in mp4. If the
- download (whatever the source shipped) isn't already that, re-encode the offending stream(s) —
- or just remux when the codecs are fine but the container isn't. **Scope:** only the mp4 compat
- profile is normalized; an explicit `vcodec` or a non-mp4 custom profile is the user opting out
- of the guarantee and is left as-is (so a `webm` profile can still ship VP9/Opus). Returns the
- file to store (converted, or the original) and corrects `info`'s codec fields to match disk."""
- if spec.get("container") != "mp4" or spec.get("vcodec"):
- return produced
- vcodec, acodec = editmod.probe_codecs(produced)
- is_mp4 = produced.suffix.lower() == ".mp4"
- needs, reenc_v, reenc_a = editmod.browser_transcode_flags(vcodec, acodec, is_mp4)
- if not needs:
- return produced
-
- dest = produced.with_name(produced.stem + ".browser.mp4")
- action = "re-encode" if (reenc_v or reenc_a) else "remux"
- log.info("job %s: %s (v=%s a=%s mp4=%s) -> browser-safe mp4", job_id, action, vcodec, acodec, is_mp4)
- _set_job(job_id, phase="processing", progress=0, speed_bps=None, eta_s=None)
- # Prefer the info duration; probe the file when the extractor didn't provide one, so a long
- # re-encode still shows real progress instead of a frozen 0%.
- total_s = int(info["duration"]) if info.get("duration") else editmod.probe_duration(produced)
- editmod.run_ffmpeg(
- editmod.build_browser_safe_cmd(produced, dest, reenc_v, reenc_a),
- total_s=total_s,
- on_progress=_ffmpeg_progress_writer(job_id, "processing"),
- should_cancel=_cancel_poll(job_id),
- )
- produced.unlink(missing_ok=True)
- # Keep the stored asset metadata truthful about what's now on disk.
- if reenc_v:
- info["vcodec"] = "h264"
- if reenc_a:
- info["acodec"] = "aac"
- return dest
-
-
-def _download(job_id: int, asset_id: int, source_kind: str, source_ref: str, spec: dict) -> None:
- staging = _staging_dir(asset_id)
- try:
- if staging.exists():
- shutil.rmtree(staging, ignore_errors=True)
- staging.mkdir(parents=True, exist_ok=True)
-
- _set_job(job_id, phase="downloading", progress=0)
- adapter = adapters.resolve_adapter(source_kind, source_ref)
- url = service.source_url(source_kind, source_ref)
- info = _media_info(_extract_with_fallback(job_id, asset_id, url, spec, staging, adapter))
- # Re-resolve now that yt-dlp's info reveals the true source (a bare/redirected url only
- # identifies post-extraction) so the .nfo uniqueid type + uploader_url agree on one source.
- adapter = adapters.resolve_adapter(source_kind, source_ref, info)
-
- produced = _produced_file(info, staging)
- produced = _ensure_browser_playable(job_id, produced, spec, info)
- ext = produced.suffix.lstrip(".").lower()
- meta = storage.MediaMeta(
- video_id=info.get("id") or source_ref,
- title=normalize_title(info.get("title")) or source_ref,
- uploader=info.get("uploader") or info.get("channel") or "Unknown Channel",
- upload_date=_parse_upload_date(info.get("upload_date")),
- uniqueid_type=adapter.nfo_uniqueid_type(),
- duration_s=int(info["duration"]) if info.get("duration") else None,
- description=info.get("description"),
- thumbnail_url=info.get("thumbnail"),
- )
-
- layout, retention_days = _download_settings()
- rel = storage.rel_path(meta, ext, layout, asset_id=asset_id)
- thumb = _find_thumbnail(staging, produced)
- # Convert the raw thumbnail (webp/png/jpeg) to a real JPEG ourselves — robustly — for the
- # poster + og:image sidecar, now that yt-dlp's fatal thumbnail convertor is gone. On failure
- # this is None and write_sidecars writes no poster, so ensure_poster cuts a frame instead.
- poster_jpg = editmod.thumbnail_to_jpg(thumb, staging / "poster.jpg")
- storage.place_file(produced, settings.download_root, rel)
- nfo_ok = storage.write_sidecars(settings.download_root, rel, meta, poster_jpg)
-
- final = storage.abs_path(settings.download_root, rel)
- size = final.stat().st_size if final.exists() else None
- expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
-
- with SessionLocal() as db:
- asset = db.get(MediaAsset, asset_id)
- if asset:
- asset.status = "ready"
- asset.rel_path = rel
- asset.size_bytes = size
- asset.duration_s = meta.duration_s
- asset.width = int(info["width"]) if info.get("width") else None
- asset.height = int(info["height"]) if info.get("height") else None
- asset.container = ext
- asset.vcodec = info.get("vcodec") if info.get("vcodec") not in (None, "none") else None
- asset.acodec = info.get("acodec") if info.get("acodec") not in (None, "none") else None
- asset.title = meta.title
- asset.uploader = meta.uploader
- asset.uploader_url = adapter.uploader_url(info)
- asset.upload_date = meta.upload_date
- asset.thumbnail_url = meta.thumbnail_url
- # Record a self-hosted poster for every download: it's the reliable link-preview
- # (og:image) source — a remote thumbnail, esp. Facebook's signed CDN URL, expires
- # and can't be relied on cross-origin. write_sidecars already wrote .jpg from
- # the thumbnail; ensure_poster returns that, or cuts a frame if there was none.
- asset.poster_path = editmod.ensure_poster(asset, settings.download_root)
- asset.source_webpage_url = info.get("webpage_url")
- asset.nfo_written = nfo_ok
- asset.error = None
- asset.last_access_at = datetime.now(timezone.utc)
- asset.expires_at = expires
- job = db.get(DownloadJob, job_id)
- # Capture the owner before commit expires the row (avoids a refresh SELECT just to notify).
- owner_id = job.user_id if job else None
- fields = {"status": JobStatus.DONE, "progress": 100, "phase": None, "speed_bps": None,
- "eta_s": None}
- if job and not job.display_name and meta.title:
- # display_name is VARCHAR(255) — clip a long title (see _fill_asset_meta_early).
- fields["display_name"] = meta.title[:255]
- # Atomic terminal write: finalizes ONLY while the row is still running, so a child that
- # finishes just after a route cancel/pause can't resurrect it (see finalize_if_running).
- finalized = service.finalize_if_running(db, job_id, **fields)
- db.commit()
- if finalized:
- _notify_download_done(db, owner_id, meta.title, job_id, kind="download")
- log.info("job %s done: %s", job_id, rel)
-
- except (_Aborted, editmod.EditAborted):
- # Paused/canceled mid-download or mid-transcode: free the asset for a later retry; keep the
- # job's status (the route set it to paused/canceled). ref_count is adjusted by the route.
- _reset_asset_pending(asset_id)
- log.info("job %s aborted (pause/cancel)", job_id)
- except Exception as exc: # noqa: BLE001 — surface any failure to the user, keep worker alive
- msg = str(exc)[:500]
- with SessionLocal() as db:
- asset = db.get(MediaAsset, asset_id)
- if asset:
- asset.status = "error"
- asset.error = msg
- # Atomic fail: only while still running, so a route-set cancel/pause is never overwritten
- # with `error` (error also holds the asset → would re-open the double-release class).
- service.finalize_if_running(
- db, job_id, status=JobStatus.ERROR, error=msg,
- error_code=dlerrors.classify(msg), phase=None,
- )
- db.commit()
- log.warning("job %s failed: %s", job_id, msg)
- finally:
- shutil.rmtree(staging, ignore_errors=True)
-
-
-def _reset_asset_pending(asset_id: int) -> None:
- with SessionLocal() as db:
- asset = db.get(MediaAsset, asset_id)
- if asset and asset.status == "downloading":
- asset.status = "pending"
- db.commit()
-
-
-# --- editor derivatives (phase 2: trim/crop via ffmpeg) -------------------------------------
-
-def _edit_staging_dir(asset_id: int) -> Path:
- return Path(settings.download_root) / ".staging" / f"edit-{asset_id}"
-
-
-def _fail_edit(job_id: int, asset_id: int, msg: str) -> None:
- with SessionLocal() as db:
- asset = db.get(MediaAsset, asset_id)
- if asset:
- asset.status = "error"
- asset.error = msg
- # Atomic fail: only while still running (see the download error path).
- service.finalize_if_running(
- db, job_id, status=JobStatus.ERROR, error=msg,
- error_code=dlerrors.classify(msg), phase=None,
- )
- db.commit()
- log.warning("edit job %s failed: %s", job_id, msg)
-
-
-def _process_edit(job_id: int, asset_id: int) -> None:
- """Produce a per-user trim/crop clip from the source asset's downloaded file via ffmpeg."""
- with SessionLocal() as db:
- job = db.get(DownloadJob, job_id)
- src = db.get(MediaAsset, job.source_asset_id) if job and job.source_asset_id else None
- edit_spec = dict(job.edit_spec or {}) if job else {}
- user_id = job.user_id if job else None
- src_ready = src is not None and src.status == "ready" and bool(src.rel_path)
- src_rel = src.rel_path if src else None
- src_w, src_h, src_dur = (src.width, src.height, src.duration_s) if src else (None, None, None)
-
- if not src_ready or not src_rel:
- _fail_edit(job_id, asset_id, "The source download is no longer available.")
- return
- src_path = storage.abs_path(settings.download_root, src_rel)
- if not src_path.exists():
- _fail_edit(job_id, asset_id, "The source file is missing.")
- return
-
- src_ext = src_path.suffix.lstrip(".").lower() or "mp4"
- crop = editmod.has_crop(edit_spec)
- out_ext = "mp4" if editmod.needs_reencode(edit_spec) else src_ext
- total_s = editmod.clip_duration(edit_spec, src_dur) or src_dur or 0
- staging = _edit_staging_dir(asset_id)
-
- try:
- if staging.exists():
- shutil.rmtree(staging, ignore_errors=True)
- staging.mkdir(parents=True, exist_ok=True)
- dest_staging = staging / f"out.{out_ext}"
-
- _set_job(job_id, phase="editing", progress=0, speed_bps=None, eta_s=None)
- if edit_spec.get("segments"):
- cmd, prep = editmod.build_concat_plan(src_path, dest_staging, edit_spec, out_ext, staging)
- for p, content in prep.items():
- p.write_text(content, encoding="utf-8")
- else:
- cmd = editmod.build_edit_cmd(src_path, dest_staging, edit_spec, out_ext)
- editmod.run_ffmpeg(
- cmd,
- total_s,
- on_progress=_ffmpeg_progress_writer(job_id, "editing"),
- should_cancel=_cancel_poll(job_id),
- )
- if not dest_staging.exists():
- raise RuntimeError("ffmpeg produced no output file")
-
- rel = storage.edit_rel_path(user_id, asset_id, out_ext)
- storage.place_file(dest_staging, settings.download_root, rel)
- final = storage.abs_path(settings.download_root, rel)
- size = final.stat().st_size if final.exists() else None
- _, retention_days = _download_settings()
- expires = datetime.now(timezone.utc) + timedelta(days=retention_days)
-
- with SessionLocal() as db:
- asset = db.get(MediaAsset, asset_id)
- if asset:
- asset.status = "ready"
- asset.rel_path = rel
- asset.size_bytes = size
- asset.container = out_ext
- asset.duration_s = total_s or asset.duration_s
- if crop:
- asset.width = edit_spec["crop"]["w"]
- asset.height = edit_spec["crop"]["h"]
- else:
- asset.width, asset.height = src_w, src_h
- asset.error = None
- asset.last_access_at = datetime.now(timezone.utc)
- asset.expires_at = expires
- job = db.get(DownloadJob, job_id)
- # Capture owner + title before commit expires the rows (avoids a refresh SELECT).
- owner_id = job.user_id if job else None
- dl_title = asset.title if asset else None
- # Atomic terminal write (see the download path): don't resurrect a route-stopped job.
- finalized = service.finalize_if_running(
- db, job_id, status=JobStatus.DONE, progress=100, phase=None
- )
- db.commit()
- if finalized:
- _notify_download_done(db, owner_id, dl_title, job_id, kind="edit")
- log.info("edit job %s done: %s", job_id, rel)
-
- except editmod.EditAborted:
- # Paused/canceled mid-encode: free the asset for a later retry (ref_count is the route's job).
- _reset_asset_pending(asset_id)
- log.info("edit job %s aborted (pause/cancel)", job_id)
- except Exception as exc: # noqa: BLE001 — surface the failure, keep the worker alive
- _fail_edit(job_id, asset_id, str(exc)[:500])
- finally:
- shutil.rmtree(staging, ignore_errors=True)
-
-
# --- loop -----------------------------------------------------------------------------------
def _worker_loop(idx: int) -> None:
diff --git a/backend/tests/test_download_output.py b/backend/tests/test_download_output.py
index 3f9e51c8..c37b6c1d 100644
--- a/backend/tests/test_download_output.py
+++ b/backend/tests/test_download_output.py
@@ -6,7 +6,7 @@ with "PosixPath('.') has an empty name" instead of downloading."""
import pytest
from app.downloads.edit import _concat_quote, parse_probe_streams
-from app.worker import _media_info, _produced_file
+from app.downloads.pipeline import _media_info, _produced_file
from pathlib import Path, PurePosixPath
diff --git a/backend/tests/test_ffmpeg_callbacks.py b/backend/tests/test_ffmpeg_callbacks.py
index 9405ef75..b351f03c 100644
--- a/backend/tests/test_ffmpeg_callbacks.py
+++ b/backend/tests/test_ffmpeg_callbacks.py
@@ -4,20 +4,20 @@ run_ffmpeg calls `on_progress`/`should_cancel` once per ffmpeg output LINE (`-pr
emits a block of ~12 every half second), and both callbacks are database round-trips: `_set_job`
opens a session and UPDATE+COMMITs, `_job_status` opens one and SELECTs. Unthrottled, a long
re-encode is ~10^5 of each."""
-from app import worker
+import app.downloads.pipeline as pipeline
class TestProgressWriter:
def test_writes_the_first_value_immediately(self, monkeypatch):
writes = []
- monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
- worker._ffmpeg_progress_writer(7, "processing")(12.4)
+ monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
+ pipeline._ffmpeg_progress_writer(7, "processing")(12.4)
assert writes == [{"progress": 12, "phase": "processing"}]
def test_a_burst_of_lines_costs_one_write(self, monkeypatch):
writes = []
- monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
- write = worker._ffmpeg_progress_writer(7, "processing")
+ monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
+ write = pipeline._ffmpeg_progress_writer(7, "processing")
for pct in range(0, 40): # what one second of ffmpeg output looks like
write(pct)
assert len(writes) == 1
@@ -25,8 +25,8 @@ class TestProgressWriter:
def test_the_next_second_writes_again(self, monkeypatch):
writes = []
clock = {"t": 1000.0}
- monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
- write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
+ monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
+ write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
write(10)
write(11) # same second → dropped
clock["t"] += 1.5
@@ -36,9 +36,9 @@ class TestProgressWriter:
def test_an_unchanged_whole_percent_is_never_rewritten(self, monkeypatch):
writes = []
clock = {"t": 1000.0}
- monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
- monkeypatch.setattr(worker, "_touch_job", lambda job_id: None)
- write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
+ monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
+ monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
+ write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
write(10.1)
clock["t"] += 5
write(10.9) # still 10% — the bar shows integers
@@ -50,9 +50,9 @@ class TestProgressWriter:
progress, so the stall watchdog keeps seeing a fresh updated_at."""
writes, touches = [], []
clock = {"t": 1000.0}
- monkeypatch.setattr(worker, "_set_job", lambda job_id, **f: writes.append(f))
- monkeypatch.setattr(worker, "_touch_job", lambda job_id: touches.append(job_id))
- write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
+ monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
+ monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: touches.append(job_id))
+ write = pipeline._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
write(10.1) # first real write → progress 10
for _ in range(3): # three more intervals stuck on 10%
clock["t"] += 1.5
@@ -64,8 +64,8 @@ class TestProgressWriter:
class TestCancelPoll:
def test_asks_once_then_reuses_the_answer(self, monkeypatch):
calls = []
- monkeypatch.setattr(worker, "_job_status", lambda job_id: calls.append(job_id) or "running")
- cancelled = worker._cancel_poll(7)
+ monkeypatch.setattr(pipeline, "_job_status", lambda job_id: calls.append(job_id) or "running")
+ cancelled = pipeline._cancel_poll(7)
for _ in range(40):
assert cancelled() is False
assert len(calls) == 1
@@ -78,8 +78,8 @@ class TestCancelPoll:
calls.append(job_id)
return "cancelled"
- monkeypatch.setattr(worker, "_job_status", status)
- cancelled = worker._cancel_poll(7, now=lambda: clock["t"])
+ monkeypatch.setattr(pipeline, "_job_status", status)
+ cancelled = pipeline._cancel_poll(7, now=lambda: clock["t"])
assert cancelled() is True
clock["t"] += 60 # cancellation is sticky: no re-query, and no un-cancelling
assert cancelled() is True
@@ -88,8 +88,8 @@ class TestCancelPoll:
def test_a_cancel_arriving_later_is_seen_on_the_next_tick(self, monkeypatch):
clock = {"t": 1000.0}
state = {"status": "running"}
- monkeypatch.setattr(worker, "_job_status", lambda job_id: state["status"])
- cancelled = worker._cancel_poll(7, now=lambda: clock["t"])
+ monkeypatch.setattr(pipeline, "_job_status", lambda job_id: state["status"])
+ cancelled = pipeline._cancel_poll(7, now=lambda: clock["t"])
assert cancelled() is False
state["status"] = "paused"
clock["t"] += 1.0 # the throttle must not delay a cancel by more than its interval