refactor(downloads): split worker.py parent/child into pipeline.py + jobdb.py (E3 S2, B-sub-9)

worker.py (~1090 L) mixed the PARENT supervisor (claim + supervise + the process
loop) with the CHILD pipeline (the yt-dlp/ffmpeg download+edit that runs in the
spawned subprocess). Extract them:

  worker.py  1090 -> 475   parent: claim/supervise/loop
  downloads/pipeline.py  (new 592)  the child pipeline (_download/_process_edit +
                                    hooks/progress/ffmpeg helpers + _Aborted)
  downloads/jobdb.py     (new  73)  the short-lived-session helpers both sides
                                    share (_set_job/_job_status/_touch_job/
                                    _requeue_if_running/_reset_asset_pending +
                                    the staging-dir paths)

The graph is one-directional (runner -> pipeline -> jobdb; worker -> jobdb; no
worker<->pipeline edge) — the parent never calls the child directly, it spawns
runner.py as a subprocess. Consumers rewired: runner import, and two tests follow
the moved symbols (_media_info/_produced_file -> pipeline; the ffmpeg-callback
patches target pipeline where the names are looked up).

Pure code motion, verbatim bodies. ruff clean, full lane 363 pass (unchanged).

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}"

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 <base>.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)

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.
+    # real work, and to keep the import graph obvious: runner is a thin entry over the pipeline.
-    from app.worker import _download, _process_edit
+    from app.downloads.pipeline import _download, _process_edit
 
     if job_kind == "edit":
         _process_edit(job_id, asset_id)

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 datetime import datetime, timezone
-from pathlib import Path
 
-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 quota, service
-from app.downloads import adapters, formats, quota, service, storage
+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 <base>.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:

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
 
 

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))
+        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
-        worker._ffmpeg_progress_writer(7, "processing")(12.4)
+        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))
+        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
-        write = worker._ffmpeg_progress_writer(7, "processing")
+        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))
+        monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
-        write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
+        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(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
-        monkeypatch.setattr(worker, "_touch_job", lambda job_id: None)
+        monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
-        write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
+        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(pipeline, "_set_job", lambda job_id, **f: writes.append(f))
-        monkeypatch.setattr(worker, "_touch_job", lambda job_id: touches.append(job_id))
+        monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: touches.append(job_id))
-        write = worker._ffmpeg_progress_writer(7, "processing", now=lambda: clock["t"])
+        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")
+        monkeypatch.setattr(pipeline, "_job_status", lambda job_id: calls.append(job_id) or "running")
-        cancelled = worker._cancel_poll(7)
+        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)
+        monkeypatch.setattr(pipeline, "_job_status", status)
-        cancelled = worker._cancel_poll(7, now=lambda: clock["t"])
+        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"])
+        monkeypatch.setattr(pipeline, "_job_status", lambda job_id: state["status"])
-        cancelled = worker._cancel_poll(7, now=lambda: clock["t"])
+        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