"""Download center HTTP API — the per-user surface over the worker/cache/quota layers.

All routes are behind `current_user`; 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, and shares finished files. The library itself — listing, renaming, moving, deleting and
serving a done file, and uploading one — is `routes.files`. Reads come from `downloads.queries`,
writes go through Capture's service (`service`, `links`, `quota`); this module validates, answers in
HTTP and commits.
"""

import re
from typing import TYPE_CHECKING, Literal
from datetime import datetime, timedelta, timezone
from urllib.parse import parse_qs, urlparse

from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.mailtext import lang_for_user
from app import audit, sysconfig
from app.auth import admin_user, current_user
from app.config import settings
from app.db import get_db
from app.downloads import clip as clipmod
from app.downloads import credentials as dlcreds
from app.downloads import edit as editmod
from app.downloads import folders
from app.downloads import links as linksmod
from app.downloads import queries, quota, service, storage
from app.downloads import urlguard
from app.downloads.folders import FolderKind
from app.downloads.status import NOT_REFETCHABLE, JobStatus
from app.models import User, display_text
from app.routes._common import (
    ProgressIn,
    get_or_404,
    kept_position,
    not_found,
)
from app.routes.files import (
    accessible_job,
    own_folder,
    own_job,
    own_job_for_update,
    quota_detail,
    serialize,
    serialize_job,
)
from app.userscope import emails_of, is_messageable_user, messageable_others, user_by_email
from app.refusal import refused

if TYPE_CHECKING:
    from app.downloads.models import DownloadJob, DownloadLink, DownloadProfile, FileFolder

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


# --- 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 refused(400, "link_required")
    if _VIDEO_ID.match(raw):
        return "youtube", raw
    if raw.startswith(("http://", "https://")):
        # `urlparse` raises ValueError on a malformed authority ("http://[::1/x"), and this runs
        # BEFORE the urlguard check that would otherwise turn it into a structured refusal — so
        # unguarded, that typo escapes as a 500. Answered in urlguard's own shape, because it is the
        # same typo class: ":99999" is caught there and localized, and one of the two arriving as a
        # bare English string would be exactly the split the canonical-patterns table forbids.
        try:
            u = urlparse(raw)
        except ValueError as exc:
            raise HTTPException(
                status_code=422,
                detail={"code": "url_blocked", "reason": urlguard.REASON_MALFORMED},
            ) from exc
        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 refused(400, "invalid_link")


# --- 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(current_user), db: Session = Depends(get_db)) -> list[dict]:
    return [_serialize_profile(p) for p in queries.profiles_for(db, user)]


class CreateProfileIn(BaseModel):
    name: str
    spec: dict


@router.post("/profiles")
def create_profile(
    body: CreateProfileIn, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    name = body.name.strip()
    spec = body.spec
    if not name:
        raise refused(400, "profile_name_required")
    p = service.create_profile(db, user, name, spec)
    db.commit()
    return _serialize_profile(p)


class UpdateProfileIn(BaseModel):
    name: str | None = None
    spec: dict | None = None


@router.patch("/profiles/{profile_id}")
def update_profile(
    profile_id: int,
    body: UpdateProfileIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    p = _own_profile(db, user, profile_id)
    sent = body.model_fields_set
    name = None
    if "name" in sent:
        name = (body.name or "").strip()
        if not name:
            raise refused(400, "profile_name_required")
    if "spec" in sent and body.spec is None:
        raise refused(400, "spec_not_object")
    service.update_profile(p, name=name, spec=body.spec if "spec" in sent else None)
    db.commit()
    return _serialize_profile(p)


def _own_profile(db: Session, user: User, profile_id: int) -> "DownloadProfile":
    p = queries.own_profile(db, user, profile_id)
    if p is None:
        raise not_found("profile")
    return p


@router.delete("/profiles/{profile_id}")
def delete_profile(
    profile_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    service.delete_profile(db, _own_profile(db, user, profile_id))
    db.commit()
    return {"deleted": profile_id}


# --- per-user download auth (cookies) ------------------------------------------------------
# Write-only surface: the stored blob is NEVER returned — only the supported hosts + whether each
# is configured + when. See downloads.credentials for the host-scope + cache-privacy guarantees.


@router.get("/credentials")
def list_credentials(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
    return {"hosts": dlcreds.status_list(db, user.id)}


class SetCredentialIn(BaseModel):
    cookies: str


@router.put("/credentials/{host_key}")
def set_credential(
    host_key: str,
    body: SetCredentialIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Store (encrypted) this user's cookies.txt for one supported host. Write-only: the value is
    never echoed back, logged, or audited — only the fact that it changed."""
    if not dlcreds.is_supported_host(host_key):
        raise not_found("host")
    try:
        cleaned = dlcreds.validate_cookies(body.cookies)
    except dlcreds.InvalidCookies as e:
        raise HTTPException(status_code=400, detail={"code": "cookies", "reason": str(e)}) from e
    dlcreds.upsert_credential(db, user.id, host_key, cleaned)
    audit.record(
        db,
        user.id,
        audit.AuditAction.DOWNLOAD_CREDENTIAL_SET,
        target_type="download_credential",
        target_id=host_key,
    )
    db.commit()
    return {"ok": True, "hosts": dlcreds.status_list(db, user.id)}


@router.delete("/credentials/{host_key}")
def delete_credential(
    host_key: str, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    if not dlcreds.is_supported_host(host_key):
        raise not_found("host")
    if dlcreds.delete_credential(db, user.id, host_key):
        audit.record(
            db,
            user.id,
            audit.AuditAction.DOWNLOAD_CREDENTIAL_DELETE,
            target_type="download_credential",
            target_id=host_key,
        )
    db.commit()
    return {"ok": True, "hosts": dlcreds.status_list(db, user.id)}


def _resolve_spec(
    db: Session, user: User, profile_id: int | None, spec: dict | None
) -> 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."""
    if profile_id is not None:
        p = queries.usable_profile(db, user, profile_id)
        if p is None:
            raise not_found("profile")
        return p.spec, p.id
    if spec is not None:
        return spec, None
    fallback = queries.default_profile(db)
    if fallback is None:
        raise refused(400, "no_profile")
    return fallback.spec, fallback.id


# --- enqueue + list + manage ---------------------------------------------------------------


class EnqueueDownloadIn(BaseModel):
    source: str = ""
    profile_id: int | None = None
    spec: dict | None = None
    display_name: str | None = None


@router.post("")
def enqueue_download(
    body: EnqueueDownloadIn, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    source_kind, source_ref = resolve_source(body.source)
    # SEC-12: a non-YouTube source is a raw URL the worker will fetch server-side, so refuse the
    # ones pointing back into our own network BEFORE a job exists. The worker re-checks (DNS can
    # move between here and the download); see downloads/urlguard.py for what that does and does
    # not cover.
    if source_kind == "url":
        try:
            urlguard.check_url(source_ref)
        except urlguard.BlockedUrl as e:
            raise HTTPException(
                status_code=422, detail={"code": "url_blocked", "reason": e.reason}
            ) from e
    spec, profile_id = _resolve_spec(db, user, body.profile_id, body.spec)
    display_name = display_text(body.display_name)
    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_detail(e)) from e
    db.commit()
    return serialize_job(db, job)


class EnqueueEditIn(BaseModel):
    source_job_id: int = 0
    edit_spec: dict
    display_name: str | None = None


@router.post("/edit")
def enqueue_edit(
    body: EnqueueEditIn, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """Queue a phase-2 editor derivative (trim/crop) of a download this user can access.

    The source may be the user's own download OR one shared with them — either way the resulting
    clip is a NEW per-user derivative in the editor's own library (counts against their quota); the
    source file is only read, never modified, so editing a shared video is safe."""
    source_job = accessible_job(db, user, body.source_job_id)
    if source_job.status != JobStatus.DONE or source_job.asset_id is None:
        raise refused(409, "edit_unfinished")
    spec = body.edit_spec
    display_name = display_text(body.display_name)
    try:
        job = service.enqueue_edit(db, user.id, source_job, spec, display_name)
    except quota.QuotaExceeded as e:
        raise HTTPException(status_code=422, detail=quota_detail(e)) from e
    except service.EditError as e:
        raise HTTPException(status_code=400, detail={"code": "edit", "reason": e.reason}) from e
    db.commit()
    return serialize_job(db, job)


class ClipCropIn(BaseModel):
    x: float
    y: float
    w: float
    h: float
    aspect: float


class EnqueueClipIn(BaseModel):
    source: Literal["youtube", "plex"]
    # The YouTube video id or the Plex rating key.
    ref: str = Field(max_length=32)
    start_s: float
    end_s: float
    # Plex only: which of the file's audio tracks (0-based, in file order) — the one being heard.
    audio_track: int | None = None
    # A crop frame (downloads/clip): fractions of the box it was drawn in, and that box's aspect.
    crop: ClipCropIn | None = None
    display_name: str | None = None


@router.post("/clip")
def enqueue_clip(
    body: EnqueueClipIn, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """Queue an ad-hoc clip: the marked range of a YouTube video or a Plex item, cut straight from
    the source into a new asset in this user's library (downloads/clip)."""
    try:
        crop = body.crop.model_dump() if body.crop else None
        spec = clipmod.normalize(
            body.source, body.ref, body.start_s, body.end_s, body.audio_track, crop
        )
        display_name = display_text(body.display_name)
        job = service.enqueue_clip(db, user.id, spec, display_name=display_name)
    except clipmod.ClipError as e:
        status = 404 if e.reason == "not_found" else 422
        detail = {"code": "clip", "reason": e.reason, **e.params}
        raise HTTPException(status_code=status, detail=detail) from e
    except quota.QuotaExceeded as e:
        raise HTTPException(status_code=422, detail=quota_detail(e)) from e
    db.commit()
    return serialize_job(db, job)


# How long a finished job stays in the queue: the tray polls it from every page and must see a job
# turn done (its "Done · Open" row and the done notice) — the finished file itself is the library's.
QUEUE_RECENT = timedelta(minutes=10)


@router.get("")
def list_downloads(user: User = Depends(current_user), db: Session = Depends(get_db)) -> list[dict]:
    """The queue: what is waiting, running, paused or failed, and what finished a moment ago."""
    jobs = queries.queue(db, user, datetime.now(timezone.utc) - QUEUE_RECENT)
    assets = queries.assets_for(db, jobs)
    positions = queries.positions_for(db, user, jobs)
    sources = queries.source_keys(db, jobs)
    return [
        {
            **serialize(j, assets.get(j.asset_id) if j.asset_id is not None else None),
            "position_seconds": positions.get(j.id, 0),
            "source_key": sources[j.id],
        }
        for j in jobs
    ]


# Priority when a source has several jobs (e.g. a done one plus a fresh queued one).
_INDEX_RANK: dict[str, int] = {
    JobStatus.DONE: 4,
    JobStatus.RUNNING: 3,
    JobStatus.PAUSED: 2,
    JobStatus.QUEUED: 1,
}


@router.get("/index")
def my_download_index(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict:
    """source_ref -> best active status, for the feed's per-card "downloaded / queued" badge.
    Small + cheap so the feed can poll it without loading the full job list."""
    out: dict[str, str] = {}
    for ref, status in queries.statuses_by_source(db, user, tuple(_INDEX_RANK)):
        if _INDEX_RANK.get(status, 0) > _INDEX_RANK.get(out.get(ref, ""), 0):
            out[ref] = status
    return out


@router.post("/{job_id}/pause")
def pause_download(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    job = own_job_for_update(db, user, job_id)
    service.pause(db, job)
    db.commit()
    return serialize_job(db, job)


@router.post("/{job_id}/resume")
def resume_download(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    job = own_job_for_update(db, user, job_id)
    service.resume(db, job)
    db.commit()
    return serialize_job(db, job)


@router.post("/{job_id}/redownload")
def redownload_download(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """Force a fresh re-download of a finished (or failed) job, rebuilding the file + Plex tree.

    The SAME job row is kept, so any public share link (DownloadLink → job_id) keeps working and
    resolves to the newly-downloaded file — a re-download never breaks a link you already sent out
    (only a *failed* re-download leaves the job in `error`, which is the intended broken-link case).

    The rebuild always re-fetches into a FRESH, private asset (the cache sig is salted with the new
    asset's own id, and the user's cookie salt when one applies — `credentials.rebuild_salt`) rather
    than the dedup pool. That means it (a) always genuinely rebuilds
    under the current format/naming rules — never silently reuses a shared cached file — and (b)
    never deletes a file another job/user still shares: the old asset is only removed if THIS job
    was its last holder."""
    job = own_job_for_update(db, user, job_id)
    if job.job_kind in NOT_REFETCHABLE:
        # None has a re-fetchable source: an editor clip is derived locally, an upload IS the
        # source, an ad-hoc clip is a range of a source the clip itself does not re-fetch.
        raise refused(409, "not_redownloadable")
    if job.status in (JobStatus.QUEUED, JobStatus.RUNNING, JobStatus.PAUSED):
        raise refused(409, "still_downloading")
    service.redownload(db, job)
    db.commit()
    return serialize_job(db, job)


@router.post("/{job_id}/cancel")
def cancel_download(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    job = own_job_for_update(db, user, job_id)
    service.cancel(db, job)
    db.commit()
    return serialize_job(db, job)


# --- editor filmstrip (scrub track) --------------------------------------------------------


@router.post("/{job_id}/progress")
def set_library_progress(
    job_id: int,
    body: ProgressIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Checkpoint the library player's resume point in a download the caller can play (their own, or
    one shared with them). A position too early or too close to the end clears it instead."""
    accessible_job(db, user, job_id)
    position = kept_position(body)
    service.set_position(db, user, job_id, position)
    db.commit()
    return {"ok": True, "job_id": job_id, "position_seconds": position}


@router.get("/{job_id}/storyboard")
def storyboard_meta(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """Filmstrip geometry for the editor's scrub track, generating the sprite on first use.

    Best-effort: for a very long/high-res source the sprite may not generate in time — then
    `available` is False and the editor shows a plain timeline."""
    asset = queries.asset_of(db, accessible_job(db, user, job_id))
    if asset is None or asset.status != "ready" or not asset.rel_path:
        raise refused(409, "download_not_ready")
    meta = editmod.ensure_storyboard(asset, settings.files_root)
    if meta is None:
        return {"available": False}
    db.commit()  # persist storyboard_path if it was just generated
    return {
        "available": True,
        "cols": meta["cols"],
        "rows": meta["rows"],
        "count": meta["count"],
        "interval_s": meta["interval_s"],
        "url": f"/api/downloads/{job_id}/storyboard.jpg",
    }


@router.get("/{job_id}/poster.jpg")
def poster_image(job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)):
    """The locally-generated poster frame for a download whose source had no thumbnail."""
    asset = queries.asset_of(db, accessible_job(db, user, job_id))
    if asset is None or not asset.poster_path:
        raise not_found("poster")
    path = storage.safe_abs_path(settings.files_root, asset.poster_path)
    if path is None:
        raise not_found("poster")
    return FileResponse(path, media_type="image/jpeg")


@router.get("/{job_id}/storyboard.jpg")
def storyboard_image(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
):
    asset = queries.asset_of(db, accessible_job(db, user, job_id))
    if asset is None or not asset.storyboard_path:
        raise not_found("filmstrip")
    path = storage.safe_abs_path(settings.files_root, asset.storyboard_path)
    if path is None:
        raise not_found("filmstrip")
    return FileResponse(path, media_type="image/jpeg")


# --- sharing -------------------------------------------------------------------------------


class ShareDownloadIn(BaseModel):
    email: str = ""


@router.post("/{job_id}/share")
def share_download(
    job_id: int,
    body: ShareDownloadIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    job = own_job(db, user, job_id)
    if job.status != JobStatus.DONE or job.asset_id is None:
        raise refused(409, "share_unfinished")
    target = body.email.strip().lower()
    if not target:
        raise refused(400, "recipient_required")
    recipient = user_by_email(db, target)
    # Uniform "no user" for a missing OR non-messageable (suspended/deactivated) target, so
    # sharing can't be used to probe which addresses exist or to reach a suspended account.
    if recipient is None or not is_messageable_user(recipient):
        raise not_found("recipient")
    if recipient.id == user.id:
        raise refused(400, "share_own")
    if service.share(db, job, user, recipient):
        db.commit()
    return {"ok": True, "shared_with": recipient.email}


@router.delete("/{job_id}/share/{email}")
def unshare_download(
    job_id: int,
    email: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    own_job(db, user, job_id)  # ownership guard (404s if not the caller's job)
    recipient = user_by_email(db, email)
    if recipient is not None and service.unshare(db, job_id, recipient):
        db.commit()
    return {"ok": True}


# --- recipients (for the internal "share with a user" picker) ------------------------------


@router.get("/recipients")
def share_recipients(
    user: User = Depends(current_user), db: Session = Depends(get_db)
) -> list[dict]:
    """Registered users this user can share a download with (excludes self + any non-messageable
    account — suspended or deactivated — so the picker never leaks those addresses)."""
    rows = messageable_others(db, user, lang_for_user(user))
    return [{"id": u.id, "email": u.email, "display_name": u.display_name} for u in rows]


# --- public watch links (share by link) ----------------------------------------------------


def _own_link(db: Session, user: User, link_id: int) -> "DownloadLink":
    link = queries.own_link(db, user, link_id)
    if link is None:
        raise not_found("link")
    return link


@router.get("/{job_id}/links")
def list_links(
    job_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> list[dict]:
    own_job(db, user, job_id)
    return [linksmod.owner_view(link) for link in queries.links_to(db, job_id=job_id)]


class CreateLinkIn(BaseModel):
    password: str = ""
    # A newly shared download is downloadable by default — the common intent is "let the
    # recipient have the file", not stream-only. Forward-only: this only sets the default for links
    # minted from now on; existing links keep whatever `allow_download` they were created with (the
    # column default stays False, so no back-catalogue link silently flips). The owner can still
    # uncheck it in the Share dialog, and toggle any link stream-only afterwards.
    allow_download: bool = True
    expires_days: int | None = None


def _new_link(
    db: Session, user: User, body: CreateLinkIn, **target: int
) -> tuple["DownloadLink", str]:
    """A link to `target` (`job_id=` or `folder_id=`, already checked to be the user's) and its
    token (`links.create`). The Pydantic model already coerced the expiry, so a non-number is a 422
    before this runs."""
    return linksmod.create(
        db,
        user,
        password=body.password.strip(),
        allow_download=body.allow_download,
        expires_days=body.expires_days,
        **target,
    )


@router.post("/{job_id}/links")
def create_link(
    job_id: int,
    body: CreateLinkIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    job = _shareable_job(db, user, job_id)
    if job.status != JobStatus.DONE or job.asset_id is None:
        raise refused(409, "share_unfinished")
    link, token = _new_link(db, user, body, job_id=job_id)
    db.commit()
    return linksmod.owner_view(link, token=token)


def _shareable_job(db: Session, user: User, job_id: int) -> "DownloadJob":
    """One of the user's files, held for the link's transaction (`service.hold_for_share`)."""
    job = service.hold_for_share(db, user, job_id)
    if job is None:
        raise not_found("download")
    return job


def _shareable_folder(db: Session, user: User, folder_id: int) -> "FileFolder":
    """One of the user's own folders. A source folder lets its files go on its own clock, and the Hub
    is the user's channel to a machine of theirs: neither is a shelf to hand out. Held (FOR SHARE)
    for the rest of the transaction, so a delete of it at the same moment waits and takes the new
    link with it, rather than failing the link's foreign key."""
    own_folder(db, user, folder_id)
    try:
        folder = folders.own_folder(db, user.id, folder_id)
    except folders.FolderError as e:
        raise refused(409, e.reason) from e
    if folder.kind != FolderKind.OWN:
        raise refused(409, "share_folder_kind")
    return folder


@router.get("/folders/{folder_id}/links")
def list_folder_links(
    folder_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> list[dict]:
    own_folder(db, user, folder_id)
    return [linksmod.owner_view(link) for link in queries.links_to(db, folder_id=folder_id)]


@router.post("/folders/{folder_id}/links")
def create_folder_link(
    folder_id: int,
    body: CreateLinkIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    _shareable_folder(db, user, folder_id)
    link, token = _new_link(db, user, body, folder_id=folder_id)
    db.commit()
    return linksmod.owner_view(link, token=token)


class UpdateLinkIn(BaseModel):
    allow_download: bool | None = None
    expires_days: int | None = None
    password: str | None = None


@router.patch("/links/{link_id}")
def update_link(
    link_id: int,
    body: UpdateLinkIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    link = _own_link(db, user, link_id)
    linksmod.update(
        db,
        user,
        link,
        allow_download=body.allow_download,
        expires_days=body.expires_days,
        password=body.password,
        sent=frozenset(body.model_fields_set),
    )
    db.commit()
    return linksmod.owner_view(link)


@router.delete("/links/{link_id}")
def revoke_link(
    link_id: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    linksmod.revoke(db, user, _own_link(db, user, link_id))
    db.commit()
    return {"ok": True, "deleted": link_id}


# --- admin ---------------------------------------------------------------------------------


@admin_router.get("")
def admin_list_downloads(
    admin: User = Depends(admin_user), db: Session = Depends(get_db)
) -> list[dict]:
    jobs = queries.recent_jobs(db)
    assets = queries.assets_for(db, jobs)
    emails = emails_of(db, {j.user_id for j in jobs})
    out = []
    for j in jobs:
        data = serialize(j, assets.get(j.asset_id) if j.asset_id is not None else None)
        data["user_id"] = j.user_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:
    files_on_disk, total_bytes, total_assets = queries.storage_totals(db)
    # EVERY user, not only those holding a file: a quota is set before anyone downloads.
    emails = emails_of(db)
    uids = list(emails)
    files = queries.done_files_per_user(db)

    def row(uid: int) -> dict:
        lim = quota.resolve(db, uid)
        return {
            "user_id": uid,
            "email": emails[uid],
            "footprint_bytes": quota.footprint(db, uid),
            "files": files.get(uid, 0),
            "max_bytes": lim.max_bytes,
            "unlimited": lim.unlimited,
        }

    return {
        "files_on_disk": files_on_disk,
        "total_bytes": total_bytes,
        "total_assets": total_assets,
        "total_cap_bytes": sysconfig.get_int(db, "download_total_max_bytes"),
        "per_user": [row(uid) for uid in uids],
    }


@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)
    return {
        "user_id": user_id,
        "custom": queries.custom_quota(db, user_id) is not None,
        "max_bytes": lim.max_bytes,
        "max_concurrent": lim.max_concurrent,
        "max_jobs": lim.max_jobs,
        "unlimited": lim.unlimited,
    }


class AdminSetQuotaIn(BaseModel):
    # Each field is set only when provided as a real value (is not None). A non-numeric limit
    # used to reach int() and 500; it's now a clean 422.
    max_bytes: int | None = None
    max_concurrent: int | None = None
    max_jobs: int | None = None
    unlimited: bool | None = None


@admin_router.put("/quota/{user_id}")
def admin_set_quota(
    user_id: int,
    body: AdminSetQuotaIn,
    admin: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    get_or_404(db, User, user_id, what="user")
    quota.set_custom(
        db,
        user_id,
        max_bytes=body.max_bytes,
        max_concurrent=body.max_concurrent,
        max_jobs=body.max_jobs,
        unlimited=body.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:
    if quota.reset_custom(db, user_id):
        db.commit()
    return admin_get_quota(user_id, admin, db)