diff --git a/backend/app/downloads/quota.py b/backend/app/downloads/quota.py
index fb224cf..b60fcb7 100644
--- a/backend/app/downloads/quota.py
+++ b/backend/app/downloads/quota.py
@@ -63,7 +63,7 @@ def footprint(db: Session, user_id: int) -> int:
         )
         .distinct()
     )
-    return (
+    return int(
         db.execute(
             select(func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
                 MediaAsset.id.in_(held_assets), MediaAsset.status == "ready"
diff --git a/backend/app/main.py b/backend/app/main.py
index 7115870..d6329ca 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -31,6 +31,7 @@ from app.routes import (
     admin,
     channels,
     config as config_routes,
+    downloads,
     feed,
     health,
     me,
@@ -122,6 +123,8 @@ app.include_router(messages.router)
 app.include_router(channels.router)
 app.include_router(playlists.router)
 app.include_router(saved_views.router)
+app.include_router(downloads.router)
+app.include_router(downloads.admin_router)
 app.include_router(admin.router)
 app.include_router(config_routes.router)
 app.include_router(scheduler_routes.router)
diff --git a/backend/app/routes/downloads.py b/backend/app/routes/downloads.py
new file mode 100644
index 0000000..1486059
--- /dev/null
+++ b/backend/app/routes/downloads.py
@@ -0,0 +1,619 @@
+"""Download center HTTP API — the per-user surface over the worker/cache/quota layers.
+
+All routes are behind `require_human` (the shared demo account can't spend server disk / needs a
+real identity). Admin routes add `admin_user`. Live job state is polled by the frontend (the
+worker is a separate process); this module never downloads — it enqueues, manages job state,
+serves finished files (range-aware, with the user's custom display name), and shares them.
+"""
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import parse_qs, urlparse
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import FileResponse
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.auth import admin_user, require_human
+from app.config import settings
+from app.db import get_db
+from app.downloads import quota, service, storage
+from app.models import (
+    DownloadJob,
+    DownloadProfile,
+    DownloadQuota,
+    DownloadShare,
+    MediaAsset,
+    User,
+)
+
+router = APIRouter(prefix="/api/downloads", tags=["downloads"])
+admin_router = APIRouter(prefix="/api/admin/downloads", tags=["admin-downloads"])
+
+_VIDEO_ID = re.compile(r"^[A-Za-z0-9_-]{11}$")
+_BASENAME_ILLEGAL = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
+
+
+# --- source + serialization ----------------------------------------------------------------
+
+def resolve_source(raw: str) -> tuple[str, str]:
+    """Map a user-supplied id/URL to (source_kind, source_ref). YouTube is normalized to its
+    11-char id (so the cache dedups across watch/youtu.be/shorts URLs); anything else is kept
+    as a raw URL for yt-dlp's generic extractor."""
+    raw = (raw or "").strip()
+    if not raw:
+        raise HTTPException(status_code=400, detail="A video link or id is required.")
+    if _VIDEO_ID.match(raw):
+        return "youtube", raw
+    if raw.startswith(("http://", "https://")):
+        u = urlparse(raw)
+        host = u.netloc.lower()
+        if "youtube.com" in host:
+            if u.path.startswith(("/shorts/", "/embed/", "/live/")):
+                vid = u.path.split("/")[2] if len(u.path.split("/")) > 2 else ""
+                if _VIDEO_ID.match(vid):
+                    return "youtube", vid
+            qs = parse_qs(u.query).get("v", [])
+            if qs and _VIDEO_ID.match(qs[0]):
+                return "youtube", qs[0]
+        if "youtu.be" in host:
+            vid = u.path.lstrip("/").split("/")[0]
+            if _VIDEO_ID.match(vid):
+                return "youtube", vid
+        return "url", raw
+    raise HTTPException(status_code=400, detail="That doesn't look like a valid video link.")
+
+
+def _serialize(job: DownloadJob, asset: MediaAsset | None) -> dict:
+    ready = asset is not None and asset.status == "ready" and bool(asset.rel_path)
+    return {
+        "id": job.id,
+        "status": job.status,
+        "progress": job.progress,
+        "speed_bps": job.speed_bps,
+        "eta_s": job.eta_s,
+        "phase": job.phase,
+        "error": job.error,
+        "queue_pos": job.queue_pos,
+        "created_at": job.created_at.isoformat() if job.created_at else None,
+        "source_kind": job.source_kind,
+        "source_ref": job.source_ref,
+        "profile_id": job.profile_id,
+        "spec": job.profile_snapshot,
+        "display_name": job.display_name,
+        "can_download": ready and job.status == "done",
+        "expired": job.asset_id is None and job.status == "done",
+        "asset": None
+        if asset is None
+        else {
+            "id": asset.id,
+            "status": asset.status,
+            "title": asset.title,
+            "uploader": asset.uploader,
+            "upload_date": asset.upload_date.isoformat() if asset.upload_date else None,
+            "duration_s": asset.duration_s,
+            "size_bytes": asset.size_bytes,
+            "width": asset.width,
+            "height": asset.height,
+            "container": asset.container,
+            "thumbnail_url": asset.thumbnail_url,
+            "expires_at": asset.expires_at.isoformat() if asset.expires_at else None,
+        },
+    }
+
+
+def _assets_for(db: Session, jobs: list[DownloadJob]) -> dict[int, MediaAsset]:
+    ids = {j.asset_id for j in jobs if j.asset_id}
+    if not ids:
+        return {}
+    rows = db.execute(select(MediaAsset).where(MediaAsset.id.in_(ids))).scalars()
+    return {a.id: a for a in rows}
+
+
+def _own_job(db: Session, user: User, job_id: int) -> DownloadJob:
+    job = db.get(DownloadJob, job_id)
+    if job is None or job.user_id != user.id:
+        raise HTTPException(status_code=404, detail="Unknown download")
+    return job
+
+
+# --- profiles ------------------------------------------------------------------------------
+
+def _serialize_profile(p: DownloadProfile) -> dict:
+    return {
+        "id": p.id,
+        "name": p.name,
+        "spec": p.spec,
+        "is_builtin": p.is_builtin,
+        "sort_order": p.sort_order,
+    }
+
+
+@router.get("/profiles")
+def list_profiles(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> list[dict]:
+    rows = (
+        db.execute(
+            select(DownloadProfile)
+            .where(
+                (DownloadProfile.is_builtin.is_(True))
+                | (DownloadProfile.user_id == user.id)
+            )
+            .order_by(DownloadProfile.is_builtin.desc(), DownloadProfile.sort_order, DownloadProfile.id)
+        )
+        .scalars()
+        .all()
+    )
+    return [_serialize_profile(p) for p in rows]
+
+
+@router.post("/profiles")
+def create_profile(
+    payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    name = (payload.get("name") or "").strip()
+    spec = payload.get("spec")
+    if not name:
+        raise HTTPException(status_code=400, detail="A profile name is required.")
+    if not isinstance(spec, dict):
+        raise HTTPException(status_code=400, detail="spec must be an object")
+    maxpos = db.execute(
+        select(func.coalesce(func.max(DownloadProfile.sort_order), 100)).where(
+            DownloadProfile.user_id == user.id
+        )
+    ).scalar_one()
+    p = DownloadProfile(
+        user_id=user.id, name=name[:80], spec=spec, is_builtin=False, sort_order=maxpos + 1
+    )
+    db.add(p)
+    db.commit()
+    return _serialize_profile(p)
+
+
+@router.patch("/profiles/{profile_id}")
+def update_profile(
+    profile_id: int,
+    payload: dict,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    p = db.get(DownloadProfile, profile_id)
+    if p is None or p.is_builtin or p.user_id != user.id:
+        raise HTTPException(status_code=404, detail="Unknown profile")
+    if "name" in payload:
+        name = (payload.get("name") or "").strip()
+        if not name:
+            raise HTTPException(status_code=400, detail="A profile name is required.")
+        p.name = name[:80]
+    if "spec" in payload:
+        if not isinstance(payload["spec"], dict):
+            raise HTTPException(status_code=400, detail="spec must be an object")
+        p.spec = payload["spec"]
+    db.commit()
+    return _serialize_profile(p)
+
+
+@router.delete("/profiles/{profile_id}")
+def delete_profile(
+    profile_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    p = db.get(DownloadProfile, profile_id)
+    if p is None or p.is_builtin or p.user_id != user.id:
+        raise HTTPException(status_code=404, detail="Unknown profile")
+    db.delete(p)
+    db.commit()
+    return {"deleted": profile_id}
+
+
+def _resolve_spec(db: Session, user: User, payload: dict) -> tuple[dict, int | None]:
+    """Pick the format spec for an enqueue: an explicit profile_id (builtin or the user's own),
+    an inline spec, or the first builtin as a fallback."""
+    pid = payload.get("profile_id")
+    if pid is not None:
+        p = db.get(DownloadProfile, pid)
+        if p is None or (not p.is_builtin and p.user_id != user.id):
+            raise HTTPException(status_code=404, detail="Unknown profile")
+        return p.spec, p.id
+    if isinstance(payload.get("spec"), dict):
+        return payload["spec"], None
+    fallback = db.execute(
+        select(DownloadProfile)
+        .where(DownloadProfile.is_builtin.is_(True))
+        .order_by(DownloadProfile.sort_order)
+        .limit(1)
+    ).scalar_one_or_none()
+    if fallback is None:
+        raise HTTPException(status_code=400, detail="No download profile available.")
+    return fallback.spec, fallback.id
+
+
+# --- enqueue + list + manage ---------------------------------------------------------------
+
+@router.post("")
+def enqueue_download(
+    payload: dict, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    source_kind, source_ref = resolve_source(payload.get("source") or "")
+    spec, profile_id = _resolve_spec(db, user, payload)
+    display_name = (payload.get("display_name") or "").strip() or None
+    try:
+        job = service.enqueue(
+            db, user.id, source_kind, source_ref, spec, profile_id, display_name
+        )
+    except quota.QuotaExceeded as e:
+        raise HTTPException(status_code=422, detail=_quota_message(e))
+    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
+    return _serialize(job, asset)
+
+
+def _quota_message(e: quota.QuotaExceeded) -> str:
+    if e.reason == "max_jobs":
+        return f"You've reached your download limit ({e.limit} items). Remove some first."
+    if e.reason == "max_bytes":
+        gb = e.limit / 1_073_741_824
+        return f"You've used your download storage quota ({gb:.1f} GB). Free up space first."
+    return "Download quota exceeded."
+
+
+@router.get("")
+def list_downloads(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> list[dict]:
+    jobs = (
+        db.execute(
+            select(DownloadJob)
+            .where(DownloadJob.user_id == user.id)
+            .order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc())
+        )
+        .scalars()
+        .all()
+    )
+    assets = _assets_for(db, jobs)
+    return [_serialize(j, assets.get(j.asset_id)) for j in jobs]
+
+
+@router.get("/usage")
+def my_usage(user: User = Depends(require_human), db: Session = Depends(get_db)) -> dict:
+    return quota.usage(db, user.id)
+
+
+@router.get("/shared")
+def shared_with_me(
+    user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> list[dict]:
+    jobs = (
+        db.execute(
+            select(DownloadJob)
+            .join(DownloadShare, DownloadShare.job_id == DownloadJob.id)
+            .where(DownloadShare.to_user_id == user.id)
+            .order_by(DownloadShare.created_at.desc())
+        )
+        .scalars()
+        .all()
+    )
+    assets = _assets_for(db, jobs)
+    out = []
+    for j in jobs:
+        data = _serialize(j, assets.get(j.asset_id))
+        data["shared"] = True
+        out.append(data)
+    return out
+
+
+@router.patch("/{job_id}")
+def rename_download(
+    job_id: int,
+    payload: dict,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    job = _own_job(db, user, job_id)
+    name = (payload.get("display_name") or "").strip()
+    # Display name only — the on-disk file keeps its system-decided, id-bound name.
+    job.display_name = name[:255] or None
+    db.commit()
+    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
+    return _serialize(job, asset)
+
+
+def _set_status(db: Session, job: DownloadJob, status: str) -> None:
+    job.status = status
+    db.commit()
+
+
+@router.post("/{job_id}/pause")
+def pause_download(
+    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    job = _own_job(db, user, job_id)
+    if job.status in ("queued", "running"):
+        _set_status(db, job, "paused")  # a running worker aborts cooperatively via the hook
+    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
+    return _serialize(job, asset)
+
+
+@router.post("/{job_id}/resume")
+def resume_download(
+    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    job = _own_job(db, user, job_id)
+    if job.status in ("paused", "error"):
+        job.progress = 0
+        job.error = None
+        _set_status(db, job, "queued")
+    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
+    return _serialize(job, asset)
+
+
+@router.post("/{job_id}/cancel")
+def cancel_download(
+    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    job = _own_job(db, user, job_id)
+    if job.status not in ("done", "canceled"):
+        job.status = "canceled"
+        _decrement_ref(db, job)
+        db.commit()
+    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
+    return _serialize(job, asset)
+
+
+@router.delete("/{job_id}")
+def delete_download(
+    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
+) -> dict:
+    job = _own_job(db, user, job_id)
+    _decrement_ref(db, job)
+    db.delete(job)
+    db.commit()
+    return {"deleted": job_id}
+
+
+def _decrement_ref(db: Session, job: DownloadJob) -> None:
+    """Drop this job's hold on its asset. The asset itself lingers as cache until its TTL /
+    eviction (so a re-add is still a cache hit); only the reference count changes here."""
+    if job.asset_id and job.status not in ("canceled", "error"):
+        asset = db.get(MediaAsset, job.asset_id)
+        if asset and asset.ref_count > 0:
+            asset.ref_count -= 1
+
+
+# --- file download (range-aware, custom display name) --------------------------------------
+
+def _clean_basename(name: str) -> str:
+    return _BASENAME_ILLEGAL.sub("", name).strip() or "download"
+
+
+def _accessible_job(db: Session, user: User, job_id: int) -> DownloadJob:
+    job = db.get(DownloadJob, job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="Unknown download")
+    if job.user_id == user.id:
+        return job
+    shared = db.execute(
+        select(DownloadShare.id).where(
+            DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id
+        )
+    ).first()
+    if shared is None:
+        raise HTTPException(status_code=404, detail="Unknown download")
+    return job
+
+
+@router.get("/{job_id}/file")
+def download_file(
+    job_id: int,
+    request: Request,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+):
+    job = _accessible_job(db, user, job_id)
+    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
+    if asset is None or asset.status != "ready" or not asset.rel_path:
+        raise HTTPException(status_code=409, detail="This download isn't ready.")
+    path = storage.abs_path(settings.download_root, asset.rel_path).resolve()
+    root = Path(settings.download_root).resolve()
+    if root not in path.parents or not path.exists():
+        raise HTTPException(status_code=404, detail="File is no longer available.")
+
+    ext = asset.container or path.suffix.lstrip(".")
+    base = _clean_basename(job.display_name or asset.title or asset.source_ref)
+    if base.lower().endswith(f".{ext.lower()}"):
+        base = base[: -(len(ext) + 1)]
+    filename = f"{base}.{ext}"
+
+    asset.last_access_at = datetime.now(timezone.utc)
+    db.commit()
+    # Starlette's FileResponse honours the Range header (206 partial content) for seek/resume.
+    return FileResponse(path, filename=filename)
+
+
+# --- sharing -------------------------------------------------------------------------------
+
+@router.post("/{job_id}/share")
+def share_download(
+    job_id: int,
+    payload: dict,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    job = _own_job(db, user, job_id)
+    if job.status != "done" or job.asset_id is None:
+        raise HTTPException(status_code=409, detail="Only a finished download can be shared.")
+    target = (payload.get("email") or "").strip().lower()
+    if not target:
+        raise HTTPException(status_code=400, detail="A recipient email is required.")
+    recipient = db.execute(
+        select(User).where(func.lower(User.email) == target)
+    ).scalar_one_or_none()
+    if recipient is None:
+        raise HTTPException(status_code=404, detail="No user with that email.")
+    if recipient.id == user.id:
+        raise HTTPException(status_code=400, detail="That's already your download.")
+    exists = db.execute(
+        select(DownloadShare.id).where(
+            DownloadShare.job_id == job_id, DownloadShare.to_user_id == recipient.id
+        )
+    ).first()
+    if exists is None:
+        db.add(
+            DownloadShare(job_id=job_id, from_user_id=user.id, to_user_id=recipient.id)
+        )
+        from app.notifications import create_notification
+
+        create_notification(
+            db,
+            user_id=recipient.id,
+            type="download_shared",
+            title=(job.display_name or "A download") + " was shared with you",
+            data={"kind": "download_shared", "from": user.email, "job_id": job_id},
+            commit=False,
+        )
+        db.commit()
+    return {"shared_with": recipient.email}
+
+
+@router.delete("/{job_id}/share/{email}")
+def unshare_download(
+    job_id: int,
+    email: str,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    job = _own_job(db, user, job_id)
+    recipient = db.execute(
+        select(User).where(func.lower(User.email) == email.strip().lower())
+    ).scalar_one_or_none()
+    if recipient is not None:
+        row = db.execute(
+            select(DownloadShare).where(
+                DownloadShare.job_id == job_id,
+                DownloadShare.to_user_id == recipient.id,
+            )
+        ).scalar_one_or_none()
+        if row is not None:
+            db.delete(row)
+            db.commit()
+    return {"ok": True}
+
+
+# --- admin ---------------------------------------------------------------------------------
+
+@admin_router.get("")
+def admin_list_downloads(
+    admin: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> list[dict]:
+    jobs = (
+        db.execute(
+            select(DownloadJob).order_by(DownloadJob.created_at.desc(), DownloadJob.id.desc()).limit(500)
+        )
+        .scalars()
+        .all()
+    )
+    assets = _assets_for(db, jobs)
+    emails = {
+        u.id: u.email
+        for u in db.execute(
+            select(User).where(User.id.in_({j.user_id for j in jobs}))
+        ).scalars()
+    }
+    out = []
+    for j in jobs:
+        data = _serialize(j, assets.get(j.asset_id))
+        data["user_email"] = emails.get(j.user_id)
+        out.append(data)
+    return out
+
+
+@admin_router.get("/storage")
+def admin_storage(
+    admin: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
+    ready = db.execute(
+        select(func.count(), func.coalesce(func.sum(MediaAsset.size_bytes), 0)).where(
+            MediaAsset.status == "ready"
+        )
+    ).one()
+    total_assets = db.execute(select(func.count()).select_from(MediaAsset)).scalar()
+    per_user = db.execute(
+        select(DownloadJob.user_id, func.count(func.distinct(DownloadJob.asset_id)))
+        .where(DownloadJob.asset_id.is_not(None))
+        .group_by(DownloadJob.user_id)
+    ).all()
+    emails = {
+        u.id: u.email
+        for u in db.execute(
+            select(User).where(User.id.in_([uid for uid, _ in per_user]))
+        ).scalars()
+    }
+    return {
+        "ready_files": ready[0],
+        "total_bytes": int(ready[1]),
+        "total_assets": total_assets,
+        "total_cap_bytes": settings.download_total_max_bytes,
+        "per_user": [
+            {"user_id": uid, "email": emails.get(uid), "footprint_bytes": quota.footprint(db, uid)}
+            for uid, _ in per_user
+        ],
+    }
+
+
+@admin_router.get("/quota/{user_id}")
+def admin_get_quota(
+    user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
+    lim = quota.resolve(db, user_id)
+    row = db.get(DownloadQuota, user_id)
+    return {
+        "user_id": user_id,
+        "custom": row is not None,
+        "max_bytes": lim.max_bytes,
+        "max_concurrent": lim.max_concurrent,
+        "max_jobs": lim.max_jobs,
+        "unlimited": lim.unlimited,
+    }
+
+
+@admin_router.put("/quota/{user_id}")
+def admin_set_quota(
+    user_id: int,
+    payload: dict,
+    admin: User = Depends(admin_user),
+    db: Session = Depends(get_db),
+) -> dict:
+    if db.get(User, user_id) is None:
+        raise HTTPException(status_code=404, detail="Unknown user")
+    cur = quota.resolve(db, user_id)
+    row = db.get(DownloadQuota, user_id)
+    if row is None:
+        row = DownloadQuota(
+            user_id=user_id,
+            max_bytes=cur.max_bytes,
+            max_concurrent=cur.max_concurrent,
+            max_jobs=cur.max_jobs,
+            unlimited=cur.unlimited,
+        )
+        db.add(row)
+    if "max_bytes" in payload:
+        row.max_bytes = max(0, int(payload["max_bytes"]))
+    if "max_concurrent" in payload:
+        row.max_concurrent = max(1, int(payload["max_concurrent"]))
+    if "max_jobs" in payload:
+        row.max_jobs = max(1, int(payload["max_jobs"]))
+    if "unlimited" in payload:
+        row.unlimited = bool(payload["unlimited"])
+    db.commit()
+    return admin_get_quota(user_id, admin, db)
+
+
+@admin_router.delete("/quota/{user_id}")
+def admin_reset_quota(
+    user_id: int, admin: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
+    row = db.get(DownloadQuota, user_id)
+    if row is not None:
+        db.delete(row)
+        db.commit()
+    return admin_get_quota(user_id, admin, db)