siftlode/backend/app/routes/downloads.py

"""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, timedelta, timezone
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 import sysconfig
from app.auth import admin_user, require_human
from app.config import settings
from app.db import get_db
from app.downloads import edit as editmod
from app.downloads import links as linksmod
from app.downloads import formats, quota, service, storage
from app.models import (
    DownloadJob,
    DownloadLink,
    DownloadProfile,
    DownloadQuota,
    DownloadShare,
    MediaAsset,
    User,
)
from app.security import hash_password
from app.userscope import is_messageable_user, messageable_clauses
from app.utils import iso

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 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": iso(job.created_at),
        "source_kind": job.source_kind,
        "source_ref": job.source_ref,
        "source_url": service.reference_url(job, asset),
        # A locally-generated poster to fall back to when the source had no thumbnail.
        "poster_url": f"/api/downloads/{job.id}/poster.jpg"
        if (asset is not None and asset.poster_path)
        else None,
        "profile_id": job.profile_id,
        "spec": job.profile_snapshot,
        "display_name": job.display_name,
        "display_uploader": job.display_uploader,
        "display_uploader_url": job.display_uploader_url,
        "extra_links": job.extra_links or [],
        "job_kind": job.job_kind,
        "source_job_id": job.source_job_id,
        "edit_spec": job.edit_spec,
        "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": iso(asset.upload_date),
            "duration_s": asset.duration_s,
            "size_bytes": asset.size_bytes,
            "width": asset.width,
            "height": asset.height,
            "container": asset.container,
            "uploader_url": asset.uploader_url,
            "thumbnail_url": asset.thumbnail_url,
            "expires_at": iso(asset.expires_at),
        },
    }


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 _serialize_job(db: Session, job: DownloadJob) -> dict:
    """Resolve a job's cache asset (if any) and serialize the pair — the resolve-then-serialize step
    every single-job handler shares."""
    asset = db.get(MediaAsset, job.asset_id) if job.asset_id else None
    return _serialize(job, asset)


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_detail(e))
    return _serialize_job(db, job)


@router.post("/edit")
def enqueue_edit(
    payload: dict, user: User = Depends(require_human), 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, int(payload.get("source_job_id") or 0))
    if source_job.status != "done" or source_job.asset_id is None:
        raise HTTPException(status_code=409, detail="You can only edit a finished download.")
    spec = payload.get("edit_spec")
    if not isinstance(spec, dict):
        raise HTTPException(status_code=400, detail="edit_spec must be an object")
    display_name = (payload.get("display_name") or "").strip() or None
    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))
    except service.EditError as e:
        raise HTTPException(status_code=400, detail={"code": "edit", "reason": e.reason})
    return _serialize_job(db, job)


def _quota_detail(e: quota.QuotaExceeded) -> dict:
    """Structured error the trilingual client localizes (reason key + numbers), instead of a
    pre-baked English string with a dot-decimal GB value. See lib/api.ts localizeDetail()."""
    return {"code": "quota", "reason": e.reason, "limit": e.limit, "current": e.current}


@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)


# Priority when a source has several jobs (e.g. a done one plus a fresh queued one).
_INDEX_RANK = {"done": 4, "running": 3, "paused": 2, "queued": 1}


@router.get("/index")
def my_download_index(
    user: User = Depends(require_human), 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."""
    rows = db.execute(
        select(DownloadJob.source_ref, DownloadJob.status).where(
            DownloadJob.user_id == user.id,
            DownloadJob.status.in_(("queued", "running", "paused", "done")),
        )
    ).all()
    out: dict[str, str] = {}
    for ref, status in rows:
        if _INDEX_RANK.get(status, 0) > _INDEX_RANK.get(out.get(ref, ""), 0):
            out[ref] = status
    return out


@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


_MAX_EXTRA_LINKS = 8


def _clean_url(raw) -> str | None:
    """Accept only a plausible http(s) URL (these get rendered as clickable links + embedded in a
    public watch page); anything else is dropped rather than trusted."""
    if not isinstance(raw, str):
        return None
    url = raw.strip()
    if not url:
        return None
    if not url.startswith(("http://", "https://")):
        return None
    parsed = urlparse(url)
    if not parsed.netloc:
        return None
    return url[:2048]


@router.patch("/{job_id}")
def update_download_meta(
    job_id: int,
    payload: dict,
    user: User = Depends(require_human),
    db: Session = Depends(get_db),
) -> dict:
    """Update a download's display metadata: title, channel name + link, and extra reference URLs.
    All are display-only overrides — the on-disk file keeps its system-decided, id-bound name."""
    job = _own_job(db, user, job_id)
    if "display_name" in payload:
        job.display_name = ((payload.get("display_name") or "").strip())[:255] or None
    if "display_uploader" in payload:
        job.display_uploader = ((payload.get("display_uploader") or "").strip())[:255] or None
    if "display_uploader_url" in payload:
        job.display_uploader_url = _clean_url(payload.get("display_uploader_url"))
    if "extra_links" in payload:
        raw = payload.get("extra_links") or []
        if not isinstance(raw, list):
            raise HTTPException(status_code=400, detail="extra_links must be a list")
        cleaned = [u for u in (_clean_url(x) for x in raw) if u][:_MAX_EXTRA_LINKS]
        job.extra_links = cleaned or None
    db.commit()
    return _serialize_job(db, job)


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
    return _serialize_job(db, job)


@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
        # 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 a job whose asset is already errored).
        if job.asset_id:
            asset = db.get(MediaAsset, job.asset_id)
            if asset and asset.status == "error":
                asset.status = "pending"
                asset.error = None
        _set_status(db, job, "queued")
    return _serialize_job(db, job)


@router.post("/{job_id}/redownload")
def redownload_download(
    job_id: int, user: User = Depends(require_human), 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 job
    id + current asset id) 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(db, user, job_id)
    if job.job_kind == "edit":
        raise HTTPException(status_code=409, detail="Editor clips can't be re-downloaded.")
    if job.status in ("queued", "running", "paused"):
        raise HTTPException(status_code=409, detail="This download is still in progress.")

    # Resolve the fresh private asset FIRST (before touching any file). Salting with the current
    # asset id makes it distinct from both the dedup pool and this job's current asset, so the
    # insert never collides — no IntegrityError/rollback can strand us after a file was deleted.
    spec = formats.normalize(job.profile_snapshot)
    sig = formats.format_sig(spec, salt=f"{job.id}:{job.asset_id}")
    asset = service.get_or_create_asset(db, job.source_kind, job.source_ref, sig)
    service.populate_from_catalog(db, asset)

    # Now drop the old hold — `_release_asset` deletes the old file + row only if this job was its
    # last holder (a co-held asset is left intact for its other holders).
    _release_asset(db, job)

    asset.ref_count = (asset.ref_count or 0) + 1
    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 (card title + Content-Disposition filename) from the freshly
    # cleaned title — otherwise a name auto-set from an older, un-cleaned title (e.g. still carrying
    # a "1.6M views · … |" social prefix) would survive the rebuild. The worker fills it on completion.
    job.display_name = None
    job.queue_pos = service.next_queue_pos(db)  # tail of the queue, not its stale original slot
    db.commit()
    return _serialize_job(db, job)


@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"):
        _release_asset(db, job)  # release while the job still counts as holding
        job.status = "canceled"
        db.commit()
    return _serialize_job(db, job)


@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)
    _release_asset(db, job)
    db.delete(job)
    db.commit()
    return {"deleted": job_id}


def _release_asset(db: Session, job: DownloadJob) -> None:
    """Drop this job's hold on its asset when it leaves a holding state. Once NO job holds the
    asset anymore, delete the file + row immediately — a deleted download should free its disk,
    and the shared cache only needs to span *overlapping* holders (a later re-add just downloads
    again). `error` counts as holding: enqueue always +1'd ref_count and the worker never
    decrements on failure (ref_count is the route's job), so an errored job releases here on
    delete — else its +1 leaks and a later resume→ready keeps the file pinned past its last holder.
    Idempotent for the cancel path: cancel already released while the job was holding, then set it
    `canceled` (∉ the set below), so delete-after-cancel is a no-op and never double-releases.

    We DON'T delete an errored asset row here even at ref==0: another request may be concurrently
    re-enqueuing the same (source,format) — get_or_create_asset would reset that row to `pending` and
    +1 it, and deleting it under that would FK-null the new job's asset (a lost update on ref_count).
    A ready asset is still freed at ref==0 (its file is the point); an orphaned errored row is cheap
    (no file) and gets reused+reset by the next enqueue, or reclaimed by a later GC pass."""
    if not job.asset_id or job.status not in ("queued", "running", "paused", "done", "error"):
        return
    asset = db.get(MediaAsset, job.asset_id)
    if asset is None:
        return
    asset.ref_count = max(0, (asset.ref_count or 0) - 1)
    if asset.ref_count == 0 and asset.status == "ready":
        if asset.rel_path:
            storage.delete_asset_files(settings.download_root, asset.rel_path)
        db.delete(asset)


# --- file download (range-aware, custom display name) --------------------------------------

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.safe_abs_path(settings.download_root, asset.rel_path)
    if path is None:
        raise HTTPException(status_code=404, detail="File is no longer available.")

    filename = storage.download_filename(
        job.display_name or asset.title or asset.source_ref, asset.container, path
    )

    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)


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

@router.get("/{job_id}/storyboard")
def storyboard_meta(
    job_id: int, user: User = Depends(require_human), 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."""
    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.")
    meta = editmod.ensure_storyboard(asset, settings.download_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(require_human), db: Session = Depends(get_db)
):
    """The locally-generated poster frame for a download whose source had no thumbnail."""
    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 not asset.poster_path:
        raise HTTPException(status_code=404, detail="No poster.")
    path = storage.safe_abs_path(settings.download_root, asset.poster_path)
    if path is None:
        raise HTTPException(status_code=404, detail="No poster.")
    return FileResponse(path, media_type="image/jpeg")


@router.get("/{job_id}/storyboard.jpg")
def storyboard_image(
    job_id: int, 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 not asset.storyboard_path:
        raise HTTPException(status_code=404, detail="No filmstrip.")
    path = storage.safe_abs_path(settings.download_root, asset.storyboard_path)
    if path is None:
        raise HTTPException(status_code=404, detail="No filmstrip.")
    return FileResponse(path, media_type="image/jpeg")


# --- 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()
    # Uniform "no user" for a missing OR non-messageable (demo/suspended/deactivated) target, so
    # sharing can't be used to probe which addresses exist or to reach a suspended account.
    if not is_messageable_user(recipient):
        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:
    _own_job(db, user, job_id)  # ownership guard (404s if not the caller's job)
    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}


@router.delete("/shared/{job_id}")
def remove_shared_with_me(
    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
) -> dict:
    """Remove a download that was shared WITH this user from their 'Shared with me' list. Deletes
    only the recipient's share grant — the owner's job and the physical file are untouched (this is
    a per-user dismissal, not a delete)."""
    row = db.execute(
        select(DownloadShare).where(
            DownloadShare.job_id == job_id, DownloadShare.to_user_id == user.id
        )
    ).scalar_one_or_none()
    if row is not None:
        db.delete(row)
        db.commit()
    return {"removed": job_id}


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

@router.get("/recipients")
def share_recipients(
    user: User = Depends(require_human), db: Session = Depends(get_db)
) -> list[dict]:
    """Registered users this user can share a download with (excludes self + any non-messageable
    account — demo, suspended, or deactivated — so the picker never leaks those addresses)."""
    rows = (
        db.execute(
            select(User)
            .where(User.id != user.id, *messageable_clauses())
            .order_by(func.lower(User.email))
        )
        .scalars()
        .all()
    )
    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 = db.get(DownloadLink, link_id)
    if link is None:
        raise HTTPException(status_code=404, detail="Unknown link")
    job = db.get(DownloadJob, link.job_id)
    if job is None or job.user_id != user.id:
        raise HTTPException(status_code=404, detail="Unknown link")
    return link


def _link_expiry(payload: dict) -> datetime | None:
    days = payload.get("expires_days")
    if days in (None, "", 0, "0"):
        return None
    try:
        d = int(days)
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="expires_days must be a number")
    if d <= 0:
        return None
    return datetime.now(timezone.utc) + timedelta(days=min(d, 3650))


@router.get("/{job_id}/links")
def list_links(
    job_id: int, user: User = Depends(require_human), db: Session = Depends(get_db)
) -> list[dict]:
    _own_job(db, user, job_id)
    rows = (
        db.execute(
            select(DownloadLink).where(DownloadLink.job_id == job_id).order_by(DownloadLink.id.desc())
        )
        .scalars()
        .all()
    )
    return [linksmod.owner_view(row) for row in rows]


@router.post("/{job_id}/links")
def create_link(
    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.")
    pw = (payload.get("password") or "").strip()
    link = DownloadLink(
        job_id=job_id,
        token=linksmod.new_token(),
        created_by=user.id,
        allow_download=bool(payload.get("allow_download")),
        password_hash=hash_password(pw) if pw else None,
        expires_at=_link_expiry(payload),
    )
    db.add(link)
    db.commit()
    db.refresh(link)
    return linksmod.owner_view(link)


@router.patch("/links/{link_id}")
def update_link(
    link_id: int,
    payload: dict,
    user: User = Depends(require_human),
    db: Session = Depends(get_db),
) -> dict:
    link = _own_link(db, user, link_id)
    if "allow_download" in payload:
        link.allow_download = bool(payload["allow_download"])
    if "expires_days" in payload:
        link.expires_at = _link_expiry(payload)
    if "password" in payload:
        pw = (payload.get("password") or "").strip()
        link.password_hash = hash_password(pw) if pw else None
        # Bump the version so any grant issued under the previous password stops working now, rather
        # than living out its 6h TTL (setting/clearing/changing the password all count as a rotation).
        link.password_version += 1
    db.commit()
    return linksmod.owner_view(link)


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


# --- 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": sysconfig.get_int(db, "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)