feat(downloads): editor backend — trim/crop derivatives (phase 2)

Per-user trim/crop clips derived from a finished download via ffmpeg, riding the same
MediaAsset/DownloadJob/worker/quota/GC machinery (asset source_kind='edit', per-user cache key):
- migration 0041: download_jobs job_kind/source_job_id/source_asset_id/edit_spec
- app/downloads/edit.py: spec normalize+sig, ffmpeg trim/crop cmd (fast stream-copy vs frame-
  accurate re-encode, per-edit 'accurate' toggle), progress+cooperative-cancel runner, filmstrip
- service.enqueue_edit; worker _process_edit branch; storage.edit_rel_path (.edits/{user} tree)
- routes: POST /edit, GET /{id}/storyboard(.jpg)

backend/app/downloads/service.py

@@ -12,10 +12,19 @@ from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
+from app.downloads import edit as editmod
 from app.downloads import formats, quota
 from app.models import Channel, DownloadJob, MediaAsset, Video
 
 
+class EditError(Exception):
+    """Bad edit request (empty spec / source not ready); `reason` is an i18n-mappable key."""
+
+    def __init__(self, reason: str):
+        self.reason = reason
+        super().__init__(reason)
+
+
 def source_url(source_kind: str, source_ref: str) -> str:
     """The URL yt-dlp should fetch. YouTube stores the 11-char id; ad-hoc stores the URL."""
     if source_kind == "youtube":
@@ -121,3 +130,71 @@ def enqueue(
     db.commit()
     db.refresh(job)
     return job
+
+
+def _clip_title(source_title: str | None, display_name: str | None) -> str:
+    if display_name:
+        return display_name[:255]
+    base = source_title or "Clip"
+    return f"{base} (clip)"[:255]
+
+
+def enqueue_edit(
+    db: Session,
+    user_id: int,
+    source_job: DownloadJob,
+    edit_spec: dict,
+    display_name: str | None = None,
+) -> DownloadJob:
+    """Queue a phase-2 editor derivative (trim/crop) of `source_job` for `user_id`.
+
+    The derivative is per-user (never shared-cached): its asset uses `source_kind='edit'` with a
+    cache key that embeds the user id, so an identical re-edit by the same user is a cache hit but
+    it never dedups across users. Reuses the same worker/quota/GC path as a download.
+
+    Raises quota.QuotaExceeded (holdings cap) or EditError (empty spec / source not ready)."""
+    src_asset = db.get(MediaAsset, source_job.asset_id) if source_job.asset_id else None
+    if src_asset is None or src_asset.status != "ready" or not src_asset.rel_path:
+        raise EditError("source_not_ready")
+
+    spec = editmod.normalize_edit_spec(edit_spec)
+    if not spec:
+        raise EditError("empty_edit")
+
+    quota.check_enqueue(db, user_id)
+
+    sig = editmod.edit_sig(spec)
+    ref = f"{user_id}:{src_asset.id}"[:512]
+    asset = get_or_create_asset(db, "edit", ref, sig)
+    if not asset.title:
+        asset.title = _clip_title(src_asset.title, display_name)
+        asset.uploader = src_asset.uploader
+        asset.thumbnail_url = src_asset.thumbnail_url
+        asset.upload_date = src_asset.upload_date
+        asset.duration_s = editmod.clip_duration(spec, src_asset.duration_s)
+
+    job = DownloadJob(
+        user_id=user_id,
+        asset_id=asset.id,
+        job_kind="edit",
+        source_kind="edit",
+        source_ref=ref,
+        source_job_id=source_job.id,
+        source_asset_id=src_asset.id,
+        edit_spec=spec,
+        profile_snapshot={},
+        display_name=display_name or asset.title,
+        status="queued",
+        queue_pos=_next_queue_pos(db),
+    )
+    db.add(job)
+    asset.ref_count = (asset.ref_count or 0) + 1
+
+    if asset.status == "ready":  # identical clip already produced for this user → instant
+        job.status = "done"
+        job.progress = 100
+        asset.last_access_at = datetime.now(timezone.utc)
+
+    db.commit()
+    db.refresh(job)
+    return job