refactor(downloads): service owns the job state transitions (E3 S2, B-sub-8)

pause/resume/cancel/delete/redownload each hand-rolled their asset bookkeeping
inline in the route (release, failed-asset reset, the 20-line fresh-asset
rebuild), duplicating logic the worker/GC also touch. Move each transition into a
service.* method (service.pause/resume/cancel/delete/redownload) that owns the
mutation + its commit; the routes keep only auth, the HTTP 409 guards, and
serialization. Drops the now-dead _set_status route helper and two stale imports.

Pure behaviour-preserving code motion (routes -56 lines). Green under the E3 S2
transition net + the full downloads suite (363 pass), ruff clean.

backend/app/downloads/service.py

@@ -306,3 +306,78 @@ def release_asset(db: Session, job: DownloadJob) -> None:
         if asset.rel_path:
             storage.delete_asset_files(settings.download_root, asset.rel_path)
         db.delete(asset)
+
+
+# --- Job state transitions ---------------------------------------------------------------------
+# The single home for the pause/resume/cancel/delete/redownload transitions + their asset
+# bookkeeping (release, failed-asset reset, fresh-asset rebuild), so the routes are thin callers and
+# the worker/GC could share the same definitions (B-sub-8). Each commits its own transition; the
+# route keeps only auth + HTTP mapping + serialization.
+
+
+def pause(db: Session, job: DownloadJob) -> None:
+    """Pause a queued/running job (a running worker aborts cooperatively via the cancel poll); a no-op
+    in any other state."""
+    if job.status in ("queued", "running"):
+        job.status = "paused"
+        db.commit()
+
+
+def resume(db: Session, job: DownloadJob) -> None:
+    """Requeue a paused/errored job for a fresh attempt, clearing its error. If the SHARED asset itself
+    failed, reset it to pending so the worker actually re-downloads — otherwise the requeued job would
+    instantly re-inherit the asset's stale error (the worker short-circuits an errored asset)."""
+    if job.status in ("paused", "error"):
+        job.progress = 0
+        job.error = None
+        job.error_code = None
+        if job.asset_id:
+            asset = db.get(MediaAsset, job.asset_id)
+            if asset and asset.status == "error":
+                asset.status = "pending"
+                asset.error = None
+        job.status = "queued"
+        db.commit()
+
+
+def cancel(db: Session, job: DownloadJob) -> None:
+    """Cancel a job unless it's already done/canceled, releasing its asset hold WHILE it still counts
+    as holding — `canceled` ∉ HOLDING, so a later delete won't double-release."""
+    if job.status not in (JobStatus.DONE, JobStatus.CANCELED):
+        release_asset(db, job)  # release while the job still counts as holding
+        job.status = JobStatus.CANCELED
+        db.commit()
+
+
+def delete(db: Session, job: DownloadJob) -> None:
+    """Delete a job, releasing its asset hold first (which frees the file iff this was the last holder)."""
+    release_asset(db, job)
+    db.delete(job)
+    db.commit()
+
+
+def redownload(db: Session, job: DownloadJob) -> None:
+    """Rebuild a finished/failed job's file into a FRESH, private asset (the cache sig is salted with
+    the job id + current asset id, so it never reuses the dedup pool) and requeue the SAME job row —
+    so any public share link (DownloadLink -> job_id) keeps resolving to the newly-downloaded file.
+    The old hold is dropped first (freeing the old file iff this job was its last holder). The caller
+    has already rejected edit clips + still-in-progress jobs (HTTP 409)."""
+    spec = formats.normalize(job.profile_snapshot)
+    sig = formats.format_sig(spec, salt=f"{job.id}:{job.asset_id}")
+    asset = get_or_create_asset(db, job.source_kind, job.source_ref, sig)
+    populate_from_catalog(db, asset)
+    # Drop the old hold BEFORE re-pointing the job — release_asset keys off the job's current asset.
+    release_asset(db, job)
+    incr_ref_count(db, asset.id)
+    job.asset_id = asset.id
+    job.status = "queued"
+    job.progress = 0
+    job.phase = None
+    job.error = None
+    job.speed_bps = None
+    job.eta_s = None
+    # Re-derive the display name from the freshly cleaned title on completion (else an older, un-cleaned
+    # name would survive the rebuild).
+    job.display_name = None
+    job.queue_pos = next_queue_pos(db)  # tail of the queue, not its stale original slot
+    db.commit()