main.py

#!/usr/bin/env python3
"""
Coomer Viewer - FastAPI web server for browsing and streaming
OnlyFans content from NAS with metadata from TimescaleDB.
"""

import os
import io
import re
import sys
import uuid
import json
import struct
import asyncio
import hashlib
import mimetypes
import time
import ipaddress
import stat as _statmod
import orjson
from cachetools import TTLCache
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional
from dataclasses import dataclass, asdict
from enum import Enum
from urllib.parse import urlparse, quote
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor, as_completed

import anyio
import asyncpg
import aiofiles
from PIL import Image
from smart_thumbs import generate_smart_thumb
from pydantic import BaseModel, Field
from fastapi import FastAPI, HTTPException, Request, Query, Depends
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse, FileResponse, Response
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from compression_middleware import SmartCompressionMiddleware

sys.path.insert(0, str(Path(__file__).parent.parent / "album_scraper"))
from coomer_nas_downloader import CoomerNASDownloader, SUPPORTED_DOMAINS
from ingest_router import detect_route, URLRoute
import ingest_persist

DB_URL = os.environ.get("DB_URL", "postgresql://coomer:coomer_dev@localhost:5433/coomer")
NAS_BASE = os.environ.get("NAS_BASE", "/mnt/nas/drive_2")
EXTERNAL_CONTENT_BASE = os.path.join(NAS_BASE, "onlyfans", "external_content")
CHUNK_SIZE = 1024 * 1024  # 1MB chunks for streaming
THUMB_CACHE_DIR = Path(__file__).parent / ".thumb_cache"
THUMB_CACHE_DIR.mkdir(exist_ok=True)
THUMB_MAX_WIDTH = 400
THUMB_QUALITY = 75

# CORS allow-origins, configurable so the (tailnet-only, single-user) deploy can
# narrow it without a code change and can never lock itself out. Set
# CORS_ORIGINS="http://localhost:8000,http://<tailnet-host>:8000" to scope it;
# left unset it stays "*" (the prior behaviour) which also keeps the browser
# extension — which posts cross-origin from a 100.x tailnet IP — working.
CORS_ORIGINS = [
    o.strip() for o in os.environ.get("CORS_ORIGINS", "*").split(",") if o.strip()
] or ["*"]

# ─────────────────────────────────────────────────────────────────────────────
# Performance layer: NAS-stat caching, DB-row caching, and zero-copy/off-loop
# file delivery. Media/HLS/thumb files are write-once-read-many (immutable per
# id), so caching path existence/size and the media_files row is safe and
# eliminates the CIFS `actimeo=1` stat storm + per-request DB round-trips that
# dominate hot-path latency.
#
# When X_ACCEL_REDIRECT=1 (set by run.sh when behind nginx) file bytes are
# handed to nginx via X-Accel-Redirect (kernel sendfile, off the single Python
# event loop). Otherwise we use Starlette FileResponse, which on
# uvicorn>=0.32 uses os.sendfile via the `http.response.pathsend` ASGI
# extension — still far better than the old hand-rolled 1 MB copy loop.
# ─────────────────────────────────────────────────────────────────────────────
X_ACCEL = os.environ.get("X_ACCEL_REDIRECT", "") == "1"
_NAS_INTERNAL_PREFIX = "/_nas_internal/"
_THUMB_INTERNAL_PREFIX = "/_thumb_internal/"
_THUMB_CACHE_STR = str(THUMB_CACHE_DIR)

_PATH_STAT_TTL = 300.0          # immutable files: 5 min existence/size cache
_PATH_STAT_MAX = 200_000
_path_stat_cache: dict[str, tuple[float, object]] = {}

_MEDIA_ROW_TTL = 120.0
_MEDIA_ROW_MAX = 100_000
_media_row_cache: dict[int, tuple[float, Optional[dict]]] = {}

_M3U8_TTL = 10.0
_m3u8_cache: dict[str, tuple[float, bytes]] = {}

_VIDEO_MIME = {
    ".mp4": "video/mp4", ".m4v": "video/mp4",
    ".webm": "video/webm",
    ".mkv": "video/x-matroska",
    ".mov": "video/quicktime",
    ".avi": "video/x-msvideo",
    ".ts": "video/mp2t",
}


def _cached_stat(path: str):
    """os.stat() with a TTL cache; returns the stat_result or None if the path
    is not a regular file. Files are immutable so a 5-min cache is safe and
    collapses the CIFS actimeo=1 stat amplification (~514x per Agent 05)."""
    now = time.monotonic()
    hit = _path_stat_cache.get(path)
    if hit is not None and hit[0] > now:
        return hit[1]
    try:
        st = os.stat(path)
        if not _statmod.S_ISREG(st.st_mode):
            st = None
    except OSError:
        st = None
    if len(_path_stat_cache) > _PATH_STAT_MAX:
        _path_stat_cache.clear()
    _path_stat_cache[path] = (now + _PATH_STAT_TTL, st)
    return st


def _cached_isfile(path: str) -> bool:
    return _cached_stat(path) is not None


async def _get_media_row(pool, media_id: int) -> Optional[dict]:
    """Cached media_files lookup for the hot file-serving endpoints. The path
    columns never change for a given id, so a short TTL removes a pool
    acquire + DB round-trip from /stream, /hls, /thumb, /preview peak load."""
    now = time.monotonic()
    hit = _media_row_cache.get(media_id)
    if hit is not None and hit[0] > now:
        return hit[1]
    row = await pool.fetchrow(
        "SELECT local_path, hls_path, file_type, original_name "
        "FROM media_files WHERE id = $1",
        media_id,
    )
    data = dict(row) if row else None
    if len(_media_row_cache) > _MEDIA_ROW_MAX:
        _media_row_cache.clear()
    _media_row_cache[media_id] = (now + _MEDIA_ROW_TTL, data)
    return data


def _video_content_type(file_path: str, file_type: str) -> str:
    if file_type == "video":
        return _VIDEO_MIME.get(os.path.splitext(file_path)[1].lower(), "video/mp4")
    return mimetypes.guess_type(file_path)[0] or "application/octet-stream"


def _internal_uri(abs_path: str) -> Optional[str]:
    """Map an absolute on-disk path to an nginx internal redirect URI, or None
    if the path is not under a known internal root."""
    if abs_path.startswith(NAS_BASE + os.sep):
        return _NAS_INTERNAL_PREFIX + quote(abs_path[len(NAS_BASE) + 1:])
    if abs_path.startswith(_THUMB_CACHE_STR + os.sep):
        return _THUMB_INTERNAL_PREFIX + quote(abs_path[len(_THUMB_CACHE_STR) + 1:])
    return None


def _serve_file(abs_path: str, media_type: str, cache_control: str, *,
                 stat_result=None, accept_ranges: bool = True,
                 extra_headers: Optional[dict] = None):
    """Serve a file with the fastest available transport.

    nginx X-Accel-Redirect (kernel sendfile, zero Python bytes) when X_ACCEL
    is on and the path is under an internal root; otherwise FileResponse
    (zero-copy os.sendfile via pathsend on uvicorn>=0.32). FileResponse also
    handles Range/206/416/If-Range/HEAD correctly — replacing the old buggy
    hand-rolled range path."""
    headers = {"Cache-Control": cache_control}
    if extra_headers:
        headers.update(extra_headers)
    if X_ACCEL:
        internal = _internal_uri(abs_path)
        if internal is not None:
            headers["X-Accel-Redirect"] = internal
            headers["Content-Type"] = media_type
            if accept_ranges:
                headers["Accept-Ranges"] = "bytes"
            return Response(status_code=200, headers=headers)
    return FileResponse(abs_path, media_type=media_type,
                        stat_result=stat_result, headers=headers)


_OLD_GVFS_PREFIX = "/run/user/1000/gvfs/smb-share:server=unas-pro-8.local,share=drive_2"

# --- Threadpool / thumbnail concurrency governance -------------------------
# All blocking PIL/ffmpeg work runs via `asyncio.to_thread(...)`. That hands
# off to the event loop's *default* executor — a stdlib ThreadPoolExecutor
# whose max_workers defaults to `min(32, cpu_count + 4)` = **32** on this
# 64-core box. (NOTE: it is NOT anyio's 40-token default limiter — that limiter
# only governs `anyio.to_thread.run_sync`, which this codebase does not use.
# Verified empirically: peak concurrent threads for asyncio.to_thread stayed at
# 32 no matter what we set the anyio limiter to.)
#
# Previously `_THUMB_SEMAPHORE` was `os.cpu_count()` (=64), so the *intended*
# thumb cap was 64 but the *real* wall was the hidden 32-thread stdlib pool —
# and every other `to_thread` caller (ingest_persist, hls/ffprobe, etc.)
# competed for those same 32 threads. Under a cold thumb-grid burst, requests
# 33..N queued behind the threadpool, not the semaphore.
#
# Fix: install a custom default executor sized to _THREADPOOL_TOKENS in
# lifespan (loop.set_default_executor — the only thing that actually governs
# asyncio.to_thread), and reconcile the thumb semaphore to sit below it.
#  * _THREADPOOL_TOKENS = min(cpu_count, 48): 64 cores, but PIL decode/resize
#    and ffmpeg are GIL/CPU-heavy, so unbounded threads thrash the GIL and
#    oversubscribe cores. 48 leaves headroom over the thumb cap for the handful
#    of non-thumb to_thread callers without saturating all 64 cores.
#  * _THUMB_SEMAPHORE = 40 (< threadpool size): the semaphore — not a hidden
#    threadpool wall — is now the real governor for thumbnail generation, and
#    the remaining 8 threads stay available for other blocking work.
_THREADPOOL_TOKENS = min(os.cpu_count() or 4, 48)
_THUMB_CONCURRENCY = 40
# Semaphore to limit concurrent PIL/ffmpeg thumbnail generation. Kept strictly
# below _THREADPOOL_TOKENS so the semaphore is the governor, not the executor.
_THUMB_SEMAPHORE = asyncio.Semaphore(_THUMB_CONCURRENCY)
# Per-host semaphores to avoid hammering a single site (lazily created)
_host_semaphores: dict[str, asyncio.Semaphore] = {}

# Cached index.html content (loaded once on first request)
_INDEX_HTML = None
_INDEX_HTML_MTIME = 0

def get_index_html():
    global _INDEX_HTML, _INDEX_HTML_MTIME
    p = Path(__file__).parent / "static" / "index.html"
    mtime = p.stat().st_mtime
    if _INDEX_HTML is None or mtime > _INDEX_HTML_MTIME:
        _INDEX_HTML = p.read_text()
        _INDEX_HTML_MTIME = mtime
    return _INDEX_HTML

# Allowed on-disk roots for media serving. A DB-supplied absolute path is only
# returned as-is if its realpath (symlinks resolved) is contained under one of
# these roots — otherwise a crafted/corrupt path column could escape the media
# tree and read arbitrary files (e.g. /etc/passwd). EXTERNAL_CONTENT_BASE sits
# under NAS_BASE today but is listed explicitly so the guard survives it moving.
_ALLOWED_MEDIA_ROOTS: tuple[str, ...] = tuple(
    os.path.realpath(p)
    for p in (NAS_BASE, EXTERNAL_CONTENT_BASE, _THUMB_CACHE_STR)
)


def _is_within_allowed_roots(path: str) -> bool:
    """True iff the realpath of `path` is inside one of _ALLOWED_MEDIA_ROOTS.

    Uses os.path.realpath (resolves symlinks/.. ) + os.path.commonpath so a
    path like NAS_BASE + "/../../etc/passwd" or a symlink out of the tree is
    rejected. commonpath raises ValueError for cross-drive / relative mixes,
    which we treat as "not contained".
    """
    try:
        real = os.path.realpath(path)
    except (OSError, ValueError):
        return False
    for root in _ALLOWED_MEDIA_ROOTS:
        try:
            if os.path.commonpath([real, root]) == root:
                return True
        except ValueError:
            continue
    return False


def _resolve_media_path(local_path: str) -> str:
    """Resolve a media file path - handles both absolute and relative paths.

    Existing coomer content uses absolute NAS paths (old GVFS or new CIFS).
    External content (hentairead, nhentai, etc.) uses relative paths
    that need to be resolved against EXTERNAL_CONTENT_BASE.

    Security: an absolute path coming from the DB is only returned when it is
    contained under one of the allowed media roots (realpath check). An escaping
    path returns "" so every caller naturally 404s (request handlers) or skips
    the row (background workers) instead of serving an arbitrary file.
    """
    # Rewrite old GVFS paths to new kernel CIFS mount
    if local_path.startswith(_OLD_GVFS_PREFIX):
        local_path = NAS_BASE + local_path[len(_OLD_GVFS_PREFIX):]

    if os.path.isabs(local_path):
        if _cached_isfile(local_path):
            # Containment guard: only hand back paths that resolve inside the
            # media roots. Anything escaping (symlink or .. traversal) → "".
            if _is_within_allowed_roots(local_path):
                return local_path
            return ""
        # Absolute path doesn't exist — try under NAS_BASE as last resort
        basename = os.path.basename(local_path)
        for base in (EXTERNAL_CONTENT_BASE, NAS_BASE):
            for root, _dirs, files in os.walk(base):
                if basename in files:
                    return os.path.join(root, basename)
                break  # only check top-level, don't walk entire NAS

    # Try resolving relative path against external content base. Containment is
    # re-checked here too: a relative path containing ".." would otherwise
    # os.path.join out of the base and escape the media tree.
    resolved = os.path.join(EXTERNAL_CONTENT_BASE, local_path)
    if _cached_isfile(resolved) and _is_within_allowed_roots(resolved):
        return resolved
    # Fallback: try NAS_BASE directly
    resolved2 = os.path.join(NAS_BASE, local_path)
    if _cached_isfile(resolved2) and _is_within_allowed_roots(resolved2):
        return resolved2
    return ""  # No safe resolution — caller will 404 / skip

# Connection pool (initialized on startup)
_pool: asyncpg.Pool | None = None

# In-memory list/feed cache. Bounded so it can't leak; the per-endpoint
# `time.time() - cached_at < TTL` checks layered on top still set fresher
# windows for each kind (60 s feeds, 300 s creators, 3600 s atlas) — this
# 3600 s outer TTL is just the upper-bound eviction so anything older is
# definitely gone. Keys use prefixes (`recent:`, `posts:`, `tag_posts:`,
# `tag_posts_count:`, `galleries:`, `tags:N`, `creators`, `atlas`) so
# mutations can invalidate only the affected slice (`_cache_invalidate_feeds`)
# instead of the old full `.clear()` thundering herd that re-ran the 9 s /
# 1.6 s queries in parallel on every ingest/like (R2-04 #5).
_cache: TTLCache = TTLCache(maxsize=2000, ttl=3600)
CACHE_TTL = 300  # 5 minutes — per-endpoint freshness ceiling

# Cache-key prefixes affected by content mutations (ingest / delete). `atlas`
# is intentionally NOT in this list: it depends on `post_clusters` which only
# changes via the nightly UMAP refit, so we keep it across ingests — the old
# behaviour re-ran the 7 s atlas build on the very next request after any like.
_FEED_CACHE_PREFIXES = (
    "recent:", "posts:", "tag_posts:", "tag_posts_count:",
    "galleries:", "tags:", "creators", "search:", "explore:",
)


def _cache_invalidate_feeds() -> None:
    """Evict only feed/list cache keys (R2-04 §5). Preserves `atlas` and any
    other namespaced caches that aren't invalidated by content mutations."""
    victims = [k for k in list(_cache.keys())
               if isinstance(k, str) and k.startswith(_FEED_CACHE_PREFIXES)]
    for k in victims:
        _cache.pop(k, None)


class IngestStatus(str, Enum):
    QUEUED = "queued"
    DOWNLOADING = "downloading"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class IngestTask:
    task_id: str
    creator_url: str
    status: IngestStatus = IngestStatus.QUEUED
    creator_name: str | None = None
    total_files: int = 0
    files_downloaded: int = 0
    files_failed: int = 0
    current_file: str | None = None
    queued_at: float = 0.0
    started_at: float | None = None
    completed_at: float | None = None
    error: str | None = None


_ingest_tasks: dict[str, IngestTask] = {}
_ingest_queue: asyncio.Queue | None = None


# === URL Ingest (universal URL handler) ===

@dataclass
class URLIngestTask:
    task_id: str
    url: str
    handler: str
    service: str
    content_id: str | None = None
    status: IngestStatus = IngestStatus.QUEUED
    queued_at: float = 0.0
    started_at: float | None = None
    completed_at: float | None = None
    error: str | None = None
    result: dict | None = None


_url_ingest_tasks: dict[str, URLIngestTask] = {}
_url_ingest_queue: asyncio.Queue | None = None

# State holder for the embedding background worker (readable via /api/embeddings/status)
last_embedding_run: dict = {
    "started_at": None,
    "finished_at": None,
    "pending_at_start": None,
    "exit_code": None,
    "batch_limit": None,
}

_EMBED_DIR = Path(__file__).parent  # cwd for subprocess so batch_embed.log lands here
_EMBED_LOG = _EMBED_DIR / "embedding_pipeline.log"
_EMBED_LOCK = _EMBED_DIR / ".batch_embed.lock"


def _batch_embed_running() -> bool:
    """True if a batch_embed subprocess from any server instance is still alive.

    Guards against the multi-spawn pile-up: a stale server, an overlapping
    tick, or (historically) multiple uvicorn workers each launching their own
    `batch_embed.py --mode all`, which starved the box and leaked the listen fd.
    """
    try:
        pid = int(_EMBED_LOCK.read_text().strip())
    except (FileNotFoundError, ValueError):
        return False
    try:
        os.kill(pid, 0)
    except (ProcessLookupError, PermissionError):
        _EMBED_LOCK.unlink(missing_ok=True)  # stale lock
        return False
    return True


async def _hls_worker():
    """Dedicated single worker: process video-HLS backlog without blocking URL ingest workers."""
    interval = int(os.environ.get("HLS_INTERVAL_SEC", "60"))
    while True:
        try:
            await asyncio.to_thread(_run_hls_for_pending)
        except Exception:
            pass
        await asyncio.sleep(interval)


async def _thumb_worker():
    """Dedicated single worker: pre-generate cover thumbs for new posts."""
    interval = int(os.environ.get("THUMB_INTERVAL_SEC", "30"))
    while True:
        try:
            await asyncio.to_thread(_generate_thumbs_for_pending)
        except Exception:
            pass
        await asyncio.sleep(interval)


async def _refresh_tag_counts():
    """Refresh `tag_post_counts` every TAG_COUNT_REFRESH_SEC (default 300 s).
    This replaces the per-request GROUP BY over 606k post_tags in /api/tags
    (was ~1.5 s — R2-04 §4) with a ~5 ms point read from a pre-aggregated
    table. UPSERT + cleanup keeps writes minimal; concurrent reads see a
    self-consistent snapshot because each statement is atomic."""
    interval = int(os.environ.get("TAG_COUNT_REFRESH_SEC", "300"))
    # Wait one interval before the first refresh — the initial population was
    # done by the migration; this loop just keeps it fresh post-ingest.
    await asyncio.sleep(interval)
    while True:
        try:
            if _pool is not None:
                async with _pool.acquire() as conn:
                    await conn.execute("""
                        INSERT INTO tag_post_counts (tag_id, n)
                        SELECT tag_id, COUNT(*) FROM post_tags GROUP BY tag_id
                        ON CONFLICT (tag_id) DO UPDATE
                          SET n = EXCLUDED.n, updated_at = now()
                    """)
                    await conn.execute("""
                        DELETE FROM tag_post_counts
                         WHERE tag_id NOT IN (SELECT tag_id FROM post_tags GROUP BY tag_id)
                    """)
        except Exception:
            pass
        await asyncio.sleep(interval)


async def _embedding_worker():
    """Periodically run batch_embed.py --mode all if uncovered posts exist."""
    interval = int(os.environ.get("EMBEDDING_INTERVAL_SEC", "600"))
    batch = int(os.environ.get("EMBEDDING_BATCH_LIMIT", "200"))
    while True:
        await asyncio.sleep(interval)
        # Count posts that have media_files but no aggregated embedding row yet.
        # Approximate: joins media_files to detect posts with any media, then checks
        # content_embeddings for a matching aggregation='post' row on that post.
        pending = 0
        try:
            async with _pool.acquire() as conn:
                row = await conn.fetchrow(
                    """SELECT COUNT(DISTINCT mf.post_id) AS n
                       FROM media_files mf
                       WHERE NOT EXISTS (
                           SELECT 1 FROM content_embeddings ce
                           WHERE ce.post_id = mf.post_id
                             AND ce.aggregation = 'post'
                       )"""
                )
                pending = row["n"] if row else 0
        except Exception as exc:
            with open(_EMBED_LOG, "a") as f:
                f.write(f"[{datetime.now(timezone.utc).isoformat()}] ERROR pending-count query failed: {exc}\n")
            continue  # skip this tick, try again next interval

        if pending == 0:
            continue

        if _batch_embed_running():
            with open(_EMBED_LOG, "a") as f:
                f.write(
                    f"[{datetime.now(timezone.utc).isoformat()}] SKIP "
                    f"batch_embed already running (pending={pending})\n"
                )
            continue

        last_embedding_run["pending_at_start"] = pending
        last_embedding_run["batch_limit"] = batch
        last_embedding_run["started_at"] = datetime.now(timezone.utc).isoformat()
        last_embedding_run["finished_at"] = None
        last_embedding_run["exit_code"] = None

        header = (
            f"[{last_embedding_run['started_at']}] START batch_embed --mode all --limit {batch}"
            f" | pending={pending}\n"
        )
        try:
            with open(_EMBED_LOG, "a") as f:
                f.write(header)
            proc = await asyncio.create_subprocess_exec(
                sys.executable, str(_EMBED_DIR / "batch_embed.py"),
                "--mode", "all", "--limit", str(batch),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
                cwd=str(_EMBED_DIR),
                close_fds=True,  # never leak the uvicorn listen socket into the child
            )
            _EMBED_LOCK.write_text(str(proc.pid))
            stdout, _ = await proc.communicate()
            rc = proc.returncode
            last_embedding_run["exit_code"] = rc
            last_embedding_run["finished_at"] = datetime.now(timezone.utc).isoformat()
            with open(_EMBED_LOG, "a") as f:
                if stdout:
                    f.write(stdout.decode(errors="replace"))
                level = "DONE" if rc == 0 else "ERROR"
                f.write(f"[{last_embedding_run['finished_at']}] {level} exit_code={rc}\n")
        except Exception as exc:
            last_embedding_run["finished_at"] = datetime.now(timezone.utc).isoformat()
            last_embedding_run["exit_code"] = -1
            with open(_EMBED_LOG, "a") as f:
                f.write(f"[{last_embedding_run['finished_at']}] ERROR subprocess launch failed: {exc}\n")
        finally:
            _EMBED_LOCK.unlink(missing_ok=True)


@asynccontextmanager
async def lifespan(app: FastAPI):
    global _pool, _ingest_queue, _url_ingest_queue
    async def _setup_conn(conn):
        # JIT compilation costs ~3s for our parameterized vector queries
        # while execution is <100ms. asyncpg runs DISCARD ALL on release,
        # which resets GUCs — so we have to re-apply on every acquire via
        # `setup` (not `init`).
        await conn.execute("SET jit = off")

    # Single async uvicorn worker shared by request handlers AND the background
    # ingest/embedding/hls/thumb workers; 10 conns was the hard concurrency
    # ceiling and one slow query could starve the request path. PG
    # max_connections=100 has the headroom. (Agent 06 §1)
    _pool = await asyncpg.create_pool(
        DB_URL,
        min_size=8,
        max_size=40,
        max_inactive_connection_lifetime=300,
        command_timeout=30,
        statement_cache_size=512,
        server_settings={
            "application_name": "coomer_viewer",
            # Auto-kill any backend left idle-in-transaction for >30s. A leaked
            # transaction (unreleased conn mid-txn) was the root cause of the
            # recurring lock-exhaustion outage; PG reaps it instead of holding
            # its locks forever. Value is milliseconds.
            "idle_in_transaction_session_timeout": "30000",
        },
        setup=_setup_conn,  # keep: DISCARD ALL resets jit; needed until
                            # jit=off lands in postgresql.conf (infra step)
    )
    # Raise the threadpool concurrency for blocking PIL/ffmpeg work. The hot
    # path uses asyncio.to_thread, which is governed by the event loop's DEFAULT
    # executor (stdlib ThreadPoolExecutor, max_workers=min(32, cpu+4)=32 here),
    # NOT by anyio's limiter. So the real fix is to install a custom default
    # executor sized to _THREADPOOL_TOKENS. We also bump the anyio limiter for
    # completeness (covers any anyio.to_thread.run_sync paths). See the
    # _THREADPOOL_TOKENS / _THUMB_SEMAPHORE notes near the top of this module.
    try:
        _loop = asyncio.get_running_loop()
        _loop.set_default_executor(
            ThreadPoolExecutor(
                max_workers=_THREADPOOL_TOKENS,
                thread_name_prefix="to_thread",
            )
        )
        anyio.to_thread.current_default_thread_limiter().total_tokens = _THREADPOOL_TOKENS
        logging.getLogger("startup").info(
            "to_thread executor max_workers=%d, anyio limiter=%d "
            "(thumb semaphore=%d)",
            _THREADPOOL_TOKENS, _THREADPOOL_TOKENS, _THUMB_CONCURRENCY,
        )
    except Exception as _e:  # noqa: BLE001 — never let tuning block startup
        logging.getLogger("startup").warning(
            "Could not raise to_thread concurrency: %s", _e
        )

    _ingest_queue = asyncio.Queue()
    _url_ingest_queue = asyncio.Queue()

    # Durable queue recovery: re-enqueue everything still pending from before
    # the last restart. Without this every uvicorn restart silently dropped
    # the whole URL-ingest backlog (queue lived only in module state).
    try:
        ingest_persist.init_db()
        _recovered = await asyncio.to_thread(ingest_persist.load_recoverable)
        for _r in _recovered:
            _t = URLIngestTask(
                task_id=_r["task_id"],
                url=_r["url"],
                handler=_r["handler"],
                service=_r["service"],
                content_id=_r["content_id"],
                status=IngestStatus.QUEUED,
                queued_at=_r["queued_at"] or time.time(),
            )
            _url_ingest_tasks[_t.task_id] = _t
            _url_ingest_queue.put_nowait(_t)
        if _recovered:
            _url_ingest_log.info(
                "Recovered %d pending URL-ingest tasks from durable store",
                len(_recovered),
            )
    except Exception as _e:  # noqa: BLE001 — never let recovery block startup
        logging.getLogger("url_ingest").error("Queue recovery failed: %s", _e)

    # COOMER_DISABLE_BG=1 starts the API WITHOUT any background sweep workers
    # (ingest / url-ingest / embedding / HLS / thumb / tag-refresh). This is a
    # light, NAS-safe mode: a backlog sweep (hls_generator / batch_embed) can't
    # saturate the CIFS mount and wedge the box. Request-path thumb generation
    # still works on demand.
    if os.environ.get("COOMER_DISABLE_BG") == "1":
        logging.getLogger("uvicorn").warning(
            "COOMER_DISABLE_BG=1 — background sweep workers DISABLED (light mode)"
        )
    else:
        asyncio.create_task(_ingest_worker())
        _n_url_workers = int(os.environ.get("URL_INGEST_WORKERS", "6"))
        for _ in range(_n_url_workers):
            asyncio.create_task(_url_ingest_worker())
        asyncio.create_task(_url_ingest_watchdog())
        asyncio.create_task(_embedding_worker())  # one GPU-bound embedding worker
        asyncio.create_task(_hls_worker())
        asyncio.create_task(_thumb_worker())
        asyncio.create_task(_refresh_tag_counts())
    yield
    if _pool:
        await _pool.close()


# This FastAPI serializes returned models/dicts to JSON bytes directly (fast,
# no custom class needed — ORJSONResponse is deprecated here). The big
# serialization win was the 5 MB /api/atlas payload, now pre-serialized with
# orjson and cached as raw bytes. compresslevel=6 ≈ 4x faster than gzip-9 for
# ~5% larger output, on every compressed response (R2-09 #5).
app = FastAPI(title="Coomer Viewer", lifespan=lifespan)
# Replaces starlette GZipMiddleware (perf2/transport). Fixes three measured
# issues: (1) API/JSON now negotiates Brotli (~20-30% smaller than gzip) when
# the client sends `Accept-Encoding: br`, gzip otherwise; (2) already-compressed
# content types (woff2/woff/png/jpg/webp/avif/mp4/m4s/ts/zip) are NOT
# re-compressed — woff2 fonts were being gzip-recompressed slightly LARGER;
# (3) responses with a Content-Encoding already set (the precompressed .br/.gz
# static siblings) and 206 range responses pass through untouched. Adds
# `Vary: Accept-Encoding` to every response so caches stay correct.
app.add_middleware(SmartCompressionMiddleware, minimum_size=500,
                   gzip_level=6, brotli_quality=5)
app.add_middleware(CORSMiddleware, allow_origins=CORS_ORIGINS, allow_methods=["*"], allow_headers=["*"])


# Tailscale assigns node addresses out of the 100.64.0.0/10 CGNAT range. The
# app is tailnet-only + loopback; there's no password auth by design, so this
# lightweight guard just rejects anything that isn't loopback or a tailnet peer.
# The browser extension posts from a 100.x tailnet IP, so it passes.
_TRUSTED_NETS = (
    ipaddress.ip_network("127.0.0.0/8"),      # IPv4 loopback
    ipaddress.ip_network("::1/128"),          # IPv6 loopback
    ipaddress.ip_network("100.64.0.0/10"),    # Tailscale CGNAT
)


def require_trusted_client(request: Request) -> None:
    """FastAPI dependency: allow only loopback or Tailscale-CGNAT clients.

    Applied to mutating/ingest endpoints. request.client can be None for some
    ASGI transports (e.g. in-process test client) — treated as untrusted unless
    the host parses to a trusted network.
    """
    client = request.client
    host = client.host if client else None
    if not host:
        raise HTTPException(status_code=403, detail="Forbidden: untrusted client")
    try:
        ip = ipaddress.ip_address(host)
    except ValueError:
        raise HTTPException(status_code=403, detail="Forbidden: untrusted client")
    # IPv4-mapped IPv6 (e.g. ::ffff:100.x.y.z) → compare on the mapped v4 addr.
    if getattr(ip, "ipv4_mapped", None) is not None:
        ip = ip.ipv4_mapped
    if any(ip in net for net in _TRUSTED_NETS):
        return
    raise HTTPException(status_code=403, detail="Forbidden: untrusted client")


def _run_ingest(task: IngestTask):
    """Sync function that runs the download pipeline with progress tracking."""
    downloader = None
    try:
        task.status = IngestStatus.DOWNLOADING
        task.started_at = time.time()

        parsed = urlparse(task.creator_url)
        domain = parsed.netloc.lower()
        if not any(d in domain for d in SUPPORTED_DOMAINS):
            raise ValueError(f"Unsupported domain: {domain}")

        base_site = f"https://{domain}"
        path_parts = [p for p in parsed.path.strip("/").split("/") if p]
        if len(path_parts) < 3 or path_parts[1] != "user":
            raise ValueError(f"Invalid creator URL format: {task.creator_url}")

        service = path_parts[0]
        user_id = path_parts[2]

        downloader = CoomerNASDownloader(
            nas_base=NAS_BASE,
            workers=3,
            rate_limit=2.0,
            db_url=DB_URL,
        )

        username = downloader.fetch_profile(base_site, service, user_id)
        task.creator_name = username

        creator_id = downloader.db.get_or_create_creator(
            username, user_id, service, task.creator_url
        )

        creator_folder = os.path.join(downloader.nas_base, "onlyfans", username)
        os.makedirs(creator_folder, exist_ok=True)

        posts = downloader.fetch_posts(base_site, service, user_id)
        if not posts:
            task.status = IngestStatus.COMPLETED
            task.completed_at = time.time()
            return

        meta_path = os.path.join(creator_folder, "metadata.json")
        with open(meta_path, "w") as f:
            json.dump(
                {
                    "creator": username,
                    "user_id": user_id,
                    "service": service,
                    "platform_url": task.creator_url,
                    "scraped_at": datetime.now(timezone.utc).isoformat(),
                    "total_posts": len(posts),
                    "posts": posts,
                },
                f,
                indent=2,
                default=str,
            )

        downloader.db.upsert_posts(creator_id, posts)

        media = downloader.extract_media(posts, base_site)
        task.total_files = len(media)

        if not media:
            task.status = IngestStatus.COMPLETED
            task.completed_at = time.time()
            return

        def _do_download(item):
            idx, (url, meta) = item
            ext = os.path.splitext(urlparse(url).path)[1] or ".bin"
            post_id = meta.get("id", "unknown")
            orig_name = meta.get("original_name", "")
            if orig_name:
                safe_name = "".join(
                    c if c.isalnum() or c in "._- " else "_" for c in orig_name
                )
                filename = f"{post_id}_{safe_name}"
            else:
                filename = f"{post_id}_{idx}{ext}"

            file_type = downloader._classify_file(ext)
            subfolder = file_type + "s"
            dest = os.path.join(creator_folder, subfolder, filename)
            task.current_file = filename

            if os.path.exists(dest):
                file_size = os.path.getsize(dest)
                downloader.db.upsert_media(
                    creator_id, post_id, url, orig_name, dest,
                    file_type, file_size, meta.get("attachment_type", "file"),
                )
                return True

            ok, file_size = downloader.download_file(url, dest)
            if ok:
                downloader.db.upsert_media(
                    creator_id, post_id, url, orig_name, dest,
                    file_type, file_size, meta.get("attachment_type", "file"),
                )
            return ok

        with ThreadPoolExecutor(max_workers=downloader.workers) as pool:
            futures = {
                pool.submit(_do_download, (i, item)): i
                for i, item in enumerate(media, 1)
            }
            for future in as_completed(futures):
                try:
                    if future.result():
                        task.files_downloaded += 1
                    else:
                        task.files_failed += 1
                except Exception:
                    task.files_failed += 1

        task.status = IngestStatus.COMPLETED
        task.completed_at = time.time()
        task.current_file = None

    except Exception as e:
        task.status = IngestStatus.FAILED
        task.error = str(e)
        task.completed_at = time.time()
    finally:
        if downloader:
            downloader.close()


def _generate_thumbs_for_pending():
    """Generate thumbnails for videos and image-gallery cover frames missing them."""
    import psycopg2 as _pg
    conn = _pg.connect(DB_URL)
    cur = conn.cursor()
    try:
        # Video thumbs (smart-thumb / ffmpeg)
        cur.execute("""
            SELECT id, local_path FROM media_files
            WHERE file_type = 'video' AND local_path IS NOT NULL
            ORDER BY id DESC
            LIMIT 100
        """)
        for media_id, local_path in cur.fetchall():
            thumb_path = THUMB_CACHE_DIR / f"video_{media_id}.jpg"
            if thumb_path.exists():
                continue
            local_path = _resolve_media_path(local_path)
            if not os.path.isfile(local_path):
                continue
            try:
                generate_smart_thumb(local_path, str(thumb_path), THUMB_MAX_WIDTH)
            except Exception:
                pass

        # Image-gallery cover thumbs (PIL resize, one cover per post)
        cur.execute("""
            SELECT (array_agg(mf.id ORDER BY mf.page_order NULLS LAST, mf.id))[1] AS media_id,
                   (array_agg(mf.local_path ORDER BY mf.page_order NULLS LAST, mf.id))[1] AS local_path
            FROM posts p
            JOIN media_files mf ON mf.post_id = p.id
            WHERE (mf.file_type LIKE 'image%' OR mf.file_type = 'photo')
              AND mf.local_path IS NOT NULL
            GROUP BY p.id
            ORDER BY p.id DESC
            LIMIT 200
        """)
        for media_id, local_path in cur.fetchall():
            cache_key = hashlib.md5(f"{media_id}:{THUMB_MAX_WIDTH}:jpg".encode()).hexdigest()
            thumb_path = THUMB_CACHE_DIR / f"{cache_key}.jpg"
            if thumb_path.exists():
                continue
            local_path = _resolve_media_path(local_path)
            if not os.path.isfile(local_path):
                continue
            try:
                with Image.open(local_path) as img:
                    img = img.convert("RGB")
                    ratio = THUMB_MAX_WIDTH / img.width
                    new_h = int(img.height * ratio)
                    img = img.resize((THUMB_MAX_WIDTH, new_h), Image.LANCZOS)
                    img.save(thumb_path, "JPEG", quality=THUMB_QUALITY, optimize=True)
            except Exception:
                pass
    finally:
        conn.close()


def _run_hls_for_pending():
    """Generate HLS for any videos missing it. Called after ingestion."""
    import subprocess as _sp
    import psycopg2 as _pg
    conn = _pg.connect(DB_URL)
    conn.autocommit = False
    cur = conn.cursor()
    try:
        cur.execute("""
            SELECT id, local_path FROM media_files
            WHERE file_type = 'video' AND hls_path IS NULL AND local_path IS NOT NULL
            ORDER BY file_size ASC
            LIMIT 100
        """)
        pending = cur.fetchall()
        if not pending:
            return
        for media_id, local_path in pending:
            # Rewrite old GVFS paths to kernel CIFS mount
            local_path = _resolve_media_path(local_path)
            if not os.path.isfile(local_path):
                continue
            from pathlib import Path as _P
            _hls_base = _P(NAS_BASE) / "drive_2" / "onlyfans" / ".hls_output"
            hls_dir = _hls_base / str(media_id)
            hls_dir.mkdir(parents=True, exist_ok=True)
            playlist = hls_dir / "stream.m3u8"
            if playlist.exists() and playlist.stat().st_size > 0:
                cur.execute("UPDATE media_files SET hls_path = %s WHERE id = %s", (str(hls_dir), media_id))
                conn.commit()
                continue
            result = _sp.run([
                "ffmpeg", "-y", "-i", local_path,
                "-c:v", "copy", "-c:a", "copy",
                "-hls_time", "2", "-hls_list_size", "0",
                "-hls_segment_type", "mpegts",
                "-hls_segment_filename", str(hls_dir / "seg_%04d.ts"),
                "-f", "hls", str(playlist),
            ], capture_output=True, timeout=3600)
            if result.returncode == 0:
                cur.execute("UPDATE media_files SET hls_path = %s WHERE id = %s", (str(hls_dir), media_id))
                conn.commit()
    finally:
        conn.close()


async def _ingest_worker():
    """Async coroutine that processes the ingest queue sequentially."""
    while True:
        task = await _ingest_queue.get()
        try:
            await asyncio.to_thread(_run_ingest, task)
            # Invalidate cache so new creators appear in UI
            _cache_invalidate_feeds()
            # Generate HLS for newly downloaded videos
            await asyncio.to_thread(_run_hls_for_pending)
            # Generate smart thumbnails and previews for new videos
            await asyncio.to_thread(_generate_thumbs_for_pending)
        except Exception as e:
            task.status = IngestStatus.FAILED
            task.error = str(e)
            task.completed_at = time.time()
        finally:
            _ingest_queue.task_done()
            # Trim completed tasks to last 100
            completed = [
                tid for tid, t in _ingest_tasks.items()
                if t.status in (IngestStatus.COMPLETED, IngestStatus.FAILED)
            ]
            if len(completed) > 100:
                for tid in completed[:-100]:
                    del _ingest_tasks[tid]


# === URL Ingest Worker ===

import logging
_url_ingest_log = logging.getLogger("url_ingest")

# Path to download scripts
_SCRIPTS_DIR = Path(__file__).parent / "downloads" / "scripts"

# NAS content base for external downloads
_EXTERNAL_NAS = Path(EXTERNAL_CONTENT_BASE)


def _run_url_ingest(task: URLIngestTask):
    """
    Sync function that dispatches a URL ingest task to the correct handler.
    Each handler is a subprocess call to the matching download script.
    """
    import subprocess as _sp

    task.status = IngestStatus.DOWNLOADING
    task.started_at = time.time()

    handler = task.handler
    url = task.url

    _url_ingest_log.info("Running handler=%s for url=%s", handler, url)

    try:
        if handler == "pornhub_video":
            _run_script_handler(task, "download_pornhub.py", ["--url", url, "--proxy", "socks5h://127.0.0.1:1090"])
        elif handler == "pornhub_creator":
            _run_script_handler(task, "download_pornhub_creator.py", ["--url", url, "--proxy", "socks5h://127.0.0.1:1090"])
        elif handler == "nhentai_gallery":
            gallery_id = task.content_id or ""
            _run_script_handler(task, "download_nhentai_html.py", ["--gallery", gallery_id])
        elif handler == "nhentai_artist":
            artist = task.content_id or ""
            _run_script_handler(task, "download_nhentai_html.py", ["--artist", artist])
        elif handler == "nhentai_group":
            group = task.content_id or ""
            _run_script_handler(task, "download_nhentai_html.py", ["--group", group])
        elif handler == "hentairead_gallery":
            _run_script_handler(task, "download_hentairead.py", ["--url", url])
        elif handler == "hentairead_artist":
            artist = task.content_id or ""
            _run_script_handler(task, "download_creator_galleries.py", ["--site", "hentairead", "--artist", artist])
        elif handler == "beeg_creator":
            creator = task.content_id or ""
            _run_script_handler(task, "download_beeg.py", ["--creator", creator])
        elif handler == "beeg_video":
            _run_generic_ytdlp(task, url, "beeg")
        elif handler == "xvideos_video":
            _run_generic_ytdlp(task, url, "xvideos")
        elif handler == "coomer_creator":
            # Delegate to the existing coomer ingest pipeline
            _run_coomer_ingest(task, url)
        elif handler == "hanime_video":
            _run_script_handler(task, "download_hanime.py", ["--url", url])
        elif handler == "missav_video":
            _run_script_handler(task, "download_missav.py", ["--url", url])
        elif handler == "fpo_video":
            _run_generic_ytdlp(task, url, "fpo")
        elif handler == "rule34_post":
            _run_script_handler(task, "download_rule34.py", ["--url", url])
        elif handler == "generic_ytdlp":
            _run_generic_ytdlp(task, url, "generic")
        else:
            raise ValueError(f"Unknown handler: {handler}")

        task.status = IngestStatus.COMPLETED
        task.completed_at = time.time()
        _url_ingest_log.info("Completed handler=%s url=%s", handler, url)

    except Exception as e:
        task.status = IngestStatus.FAILED
        task.error = str(e)
        task.completed_at = time.time()
        _url_ingest_log.error("Failed handler=%s url=%s error=%s", handler, url, e)


def _assert_nas_healthy() -> None:
    """Refuse to start a download if NAS_BASE isn't a real network mount.

    Without this guard, a missing/unmounted NAS would silently fall back to
    writing into the local filesystem (since os.makedirs would just create
    /mnt/nas/drive_2/... as a regular dir on the root volume).
    """
    target = EXTERNAL_CONTENT_BASE
    try:
        st = os.statvfs(target)
    except OSError as e:
        raise RuntimeError(f"NAS pre-flight failed: cannot stat {target}: {e}") from e
    # smb2/cifs filesystems report a known fstype via /proc/mounts.
    with open("/proc/self/mounts") as f:
        mounts = f.read()
    if not any(
        line.split()[2] in ("cifs", "smb3", "smb2", "nfs", "nfs4")
        and target.startswith(line.split()[1])
        for line in mounts.splitlines()
    ):
        raise RuntimeError(
            f"NAS pre-flight failed: {target} is not on a network mount "
            f"(refusing to download to local disk)"
        )
    if st.f_bavail * st.f_frsize < 1 * 1024**3:
        raise RuntimeError(f"NAS pre-flight failed: <1 GiB free on {target}")


def _run_script_handler(task: URLIngestTask, script_name: str, args: list[str]):
    """Run a download script as a subprocess."""
    import subprocess as _sp

    _assert_nas_healthy()

    script_path = _SCRIPTS_DIR / script_name
    if not script_path.exists():
        raise FileNotFoundError(f"Script not found: {script_path}")

    venv_python = Path(__file__).parent / ".venv" / "bin" / "python"
    python_cmd = str(venv_python) if venv_python.exists() else sys.executable

    cmd = [python_cmd, str(script_path)] + args
    _url_ingest_log.info("Executing: %s", " ".join(cmd))

    result = _sp.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=6 * 3600,  # large creators (50+ videos) overflow shorter caps
        cwd=str(Path(__file__).parent),
    )

    if result.returncode != 0:
        error_msg = result.stderr.strip()[-500:] if result.stderr else "Unknown error"
        raise RuntimeError(f"Script {script_name} failed (rc={result.returncode}): {error_msg}")

    task.result = {"stdout_tail": result.stdout.strip()[-500:] if result.stdout else ""}


def _run_generic_ytdlp(task: URLIngestTask, url: str, service: str):
    """Download a video using yt-dlp directly."""
    import subprocess as _sp
    import hashlib as _hl

    _assert_nas_healthy()

    nas_dir = _EXTERNAL_NAS / service
    nas_dir.mkdir(parents=True, exist_ok=True)

    # Find yt-dlp
    ytdlp_paths = [
        "/home/danny/.local/share/uv/tools/yt-dlp/bin/yt-dlp",
        "/home/danny/.local/bin/yt-dlp",
    ]
    ytdlp = next((p for p in ytdlp_paths if os.path.exists(p)), "yt-dlp")

    output_template = str(nas_dir / "%(title)s [%(id)s].%(ext)s")

    cmd = [
        ytdlp,
        "-f", "bestvideo[height>=720][height<=1080]+bestaudio/best[height>=720][height<=1080]/bestvideo[height>=480][height<=1080]+bestaudio/best[height>=480][height<=1080]/bestvideo[height<=1080]+bestaudio/best[height<=1080]/best",
        "--merge-output-format", "mp4",
        "--fixup", "never",
        "--write-info-json",
        "--no-overwrites",
        "-o", output_template,
        "--socket-timeout", "30",
        "--retries", "5",
        "--proxy", "socks5h://127.0.0.1:1090",
        "-N", "16",
        url,
    ]

    _url_ingest_log.info("yt-dlp command: %s", " ".join(cmd))

    result = _sp.run(cmd, capture_output=True, text=True, timeout=1800)

    if result.returncode != 0:
        error_msg = result.stderr.strip()[-500:] if result.stderr else "Unknown error"
        raise RuntimeError(f"yt-dlp failed (rc={result.returncode}): {error_msg}")

    # Parse output to find downloaded file path
    video_path = None
    for line in result.stdout.split("\n"):
        if "[download] Destination:" in line:
            video_path = line.split("Destination:", 1)[1].strip()
        elif "[Merger] Merging formats into" in line:
            video_path = line.split('"')[1] if '"' in line else None

    # Insert into DB
    if video_path and os.path.exists(video_path):
        _log_low_res_if_needed(task, url, service, video_path)
        _insert_generic_download(url, video_path, service)

    task.result = {"video_path": video_path, "stdout_tail": result.stdout.strip()[-300:]}


def _log_low_res_if_needed(task, url: str, service: str, video_path: str) -> None:
    """Append a TSV row to low_res_downloads.tsv if the file resolves below 480p."""
    try:
        import subprocess, datetime, pathlib
        probe = subprocess.run(
            ["ffprobe", "-v", "error", "-select_streams", "v:0",
             "-show_entries", "stream=height", "-of", "csv=p=0", video_path],
            capture_output=True, text=True, timeout=15,
        )
        height_str = probe.stdout.strip()
        if not height_str:
            return
        height = int(height_str.splitlines()[0])
        if height < 480:
            tsv_path = pathlib.Path(__file__).parent / "queue_progress" / "low_res_downloads.tsv"
            tsv_path.parent.mkdir(parents=True, exist_ok=True)
            ts = datetime.datetime.utcnow().isoformat()
            task_id = getattr(task, "id", "unknown")
            _url_ingest_log.warning("low-resolution download: %dp — %s", height, url)
            with open(tsv_path, "a") as fh:
                fh.write(f"{ts}\t{task_id}\t{height}\t{service}\t{url}\t{video_path}\n")
    except Exception:
        pass


def _insert_generic_download(url: str, video_path: str, service: str):
    """Insert a generic yt-dlp download into the database."""
    import psycopg2

    conn = psycopg2.connect(DB_URL)
    try:
        with conn.cursor() as cur:
            # Get or create creator
            cur.execute(
                """INSERT INTO creators (username, user_id, service, platform_url)
                   VALUES (%s, %s, %s, %s)
                   ON CONFLICT (user_id, service) DO UPDATE SET last_scraped_at = NOW()
                   RETURNING id""",
                (f"{service}_collection", f"{service}_collection", service, url),
            )
            creator_id = cur.fetchone()[0]
            conn.commit()

            # Generate post ID from URL hash
            import hashlib
            post_id = int(hashlib.sha256(url.encode()).hexdigest()[:15], 16)

            cur.execute(
                """INSERT INTO posts (id, creator_id, title, content, published_at)
                   VALUES (%s, %s, %s, %s, NOW())
                   ON CONFLICT (id, published_at) DO NOTHING""",
                (post_id, creator_id, os.path.basename(video_path), url),
            )
            conn.commit()

            file_size = os.path.getsize(video_path) if os.path.exists(video_path) else 0
            cur.execute(
                """INSERT INTO media_files
                       (post_id, creator_id, download_url, original_name,
                        local_path, file_type, file_size, attachment_type)
                   VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
                   ON CONFLICT (download_url) DO UPDATE SET
                       local_path = EXCLUDED.local_path, file_size = EXCLUDED.file_size""",
                (post_id, creator_id, url, os.path.basename(video_path),
                 video_path, "video", file_size, "file"),
            )
            conn.commit()
    finally:
        conn.close()


def _run_coomer_ingest(task: URLIngestTask, url: str):
    """Delegate to the existing coomer/kemono ingest pipeline."""
    downloader = None
    try:
        downloader = CoomerNASDownloader(
            nas_base=NAS_BASE,
            workers=3,
            rate_limit=2.0,
            db_url=DB_URL,
        )
        parsed = urlparse(url)
        domain = parsed.netloc.lower()
        base_site = f"https://{domain}"
        path_parts = [p for p in parsed.path.strip("/").split("/") if p]

        if len(path_parts) < 3 or path_parts[1] != "user":
            raise ValueError(f"Invalid coomer/kemono URL format: {url}")

        service = path_parts[0]
        user_id = path_parts[2]

        username = downloader.fetch_profile(base_site, service, user_id)
        creator_id = downloader.db.get_or_create_creator(username, user_id, service, url)

        creator_folder = os.path.join(downloader.nas_base, "onlyfans", username)
        os.makedirs(creator_folder, exist_ok=True)

        posts = downloader.fetch_posts(base_site, service, user_id)
        if not posts:
            task.result = {"message": "No posts found"}
            return

        downloaded = 0
        for post_data in posts:
            try:
                downloader.process_post(post_data, creator_id, creator_folder)
                downloaded += 1
            except Exception as e:
                _url_ingest_log.warning("Failed to process post: %s", e)

        task.result = {"creator": username, "posts_processed": downloaded}
    finally:
        if downloader:
            try:
                downloader.close()
            except Exception:
                pass


async def _url_ingest_watchdog():
    """Reset rows stuck at status='downloading' so a single hung subprocess
    can't permanently consume a worker slot in the persistent store.

    The worker's finally clause already handles graceful interruption; this
    handles the pathological case where the worker thread itself never
    returns from `subprocess.run` (or the row gets out of sync some other
    way). Sweep interval is short; the age threshold is conservatively
    larger than the worst-case legitimate run (6h for full creator scrapes).
    """
    interval = float(os.environ.get("URL_INGEST_WATCHDOG_INTERVAL", "300"))
    max_age = float(os.environ.get("URL_INGEST_WATCHDOG_MAX_AGE", "25200"))  # 7h
    while True:
        try:
            await asyncio.sleep(interval)
            reset = await asyncio.to_thread(
                ingest_persist.reset_stale_downloading, max_age
            )
            if reset:
                _url_ingest_log.warning(
                    "watchdog reset %d stale 'downloading' row(s) older than %ds",
                    len(reset), int(max_age),
                )
                # Pull them back into the live queue immediately so we don't
                # wait for the next restart.
                for r in reset:
                    if not r["url"]:
                        continue
                    if r["task_id"] in _url_ingest_tasks:
                        t = _url_ingest_tasks[r["task_id"]]
                        t.status = IngestStatus.QUEUED
                        t.started_at = None
                    else:
                        t = URLIngestTask(
                            task_id=r["task_id"],
                            url=r["url"],
                            handler=r["handler"],
                            service=r["service"],
                            content_id=r["content_id"],
                            status=IngestStatus.QUEUED,
                            queued_at=r["queued_at"] or time.time(),
                        )
                        _url_ingest_tasks[t.task_id] = t
                    if _url_ingest_queue is not None:
                        _url_ingest_queue.put_nowait(t)
        except asyncio.CancelledError:
            raise
        except Exception as e:  # noqa: BLE001 — watchdog must never die
            _url_ingest_log.error("watchdog tick failed: %s", e)


async def _url_ingest_worker():
    """Async coroutine that processes the URL ingest queue (parallel workers share the queue)."""
    _per_host_cap = int(os.environ.get("URL_INGEST_PER_HOST", "2"))
    while True:
        task = await _url_ingest_queue.get()
        host = urlparse(task.url).netloc.lower().lstrip("www.")
        if host not in _host_semaphores:
            _host_semaphores[host] = asyncio.Semaphore(_per_host_cap)
        sem = _host_semaphores[host]
        # Skip past saturated hosts so workers don't block holding a task they can't run.
        if sem._value <= 0:
            await _url_ingest_queue.put(task)
            _url_ingest_queue.task_done()
            await asyncio.sleep(0.5)
            continue
        try:
            task.status = IngestStatus.DOWNLOADING
            task.started_at = time.time()
            await asyncio.to_thread(ingest_persist.record, task)
            async with sem:
                await asyncio.to_thread(_run_url_ingest, task)
            _cache_invalidate_feeds()
            # HLS + thumb generation run in dedicated background coroutines, not here
        except asyncio.CancelledError:
            # Worker cancelled mid-task (shutdown / signal). The subprocess
            # thread keeps running but this coroutine is going away — reset
            # to queued so the next boot's lifespan recovery re-enqueues
            # cleanly. Without this the row stays at status='downloading'
            # forever (the bug that wedged 6 worker slots on 2026-05-22).
            task.status = IngestStatus.QUEUED
            task.started_at = None
            task.error = "interrupted by shutdown — re-queued"
            raise
        except Exception as e:
            task.status = IngestStatus.FAILED
            task.error = str(e)
            task.completed_at = time.time()
        finally:
            # Safety net: anything still flagged DOWNLOADING here is an
            # orphan (subprocess returned but the worker path didn't flip
            # status, or an unexpected control flow). Park back as queued.
            if task.status == IngestStatus.DOWNLOADING:
                task.status = IngestStatus.QUEUED
                task.started_at = None
                task.error = "orphan recovery — re-queued"
            # Durable write-through of the terminal (or interrupted) state so
            # the queue survives the next restart.
            try:
                await asyncio.to_thread(ingest_persist.record, task)
            except Exception as _pe:  # noqa: BLE001
                _url_ingest_log.error("persist on finish failed: %s", _pe)
            _url_ingest_queue.task_done()
            completed = [
                tid for tid, t in _url_ingest_tasks.items()
                if t.status in (IngestStatus.COMPLETED, IngestStatus.FAILED)
            ]
            if len(completed) > 200:
                for tid in completed[:-200]:
                    del _url_ingest_tasks[tid]


async def get_db():
    if _pool is None:
        raise HTTPException(500, "Database pool not initialized")
    return _pool


@app.get(
    "/api/health",
    summary="Health check",
    description="Returns server status and database connectivity. Returns HTTP 503 if the DB is unreachable.",
)
async def health_check():
    """Simple health check with DB connectivity test."""
    try:
        pool = await get_db()
        await pool.fetchval("SELECT 1")
        return {"status": "ok", "db": "connected"}
    except Exception:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "db": "disconnected"},
        )


@app.get(
    "/api/creators",
    summary="List creators",
    description="Returns all creators with post/video/image counts. Results are cached in-memory for 15 minutes.",
)
async def list_creators():
    # Check in-memory cache
    cache_key = "creators"
    if cache_key in _cache:
        cached_at, data = _cache[cache_key]
        if time.time() - cached_at < CACHE_TTL:
            return JSONResponse(
                content=data,
                headers={"Cache-Control": "public, max-age=900", "X-Cache": "HIT"},
            )

    pool = await get_db()
    rows = await pool.fetch("""
        SELECT c.id, c.username, c.service, c.platform_url,
               COALESCE(pc.post_count, 0) as post_count,
               COALESCE(mc.video_count, 0) as video_count,
               COALESCE(mc.image_count, 0) as image_count,
               pc.last_post_at
        FROM creators c
        LEFT JOIN (
            SELECT creator_id, COUNT(*) as post_count,
                   MAX(COALESCE(published_at, scraped_at)) as last_post_at
            FROM posts
            GROUP BY creator_id
        ) pc ON pc.creator_id = c.id
        LEFT JOIN (
            SELECT creator_id,
                   COUNT(*) FILTER (WHERE file_type = 'video') as video_count,
                   COUNT(*) FILTER (WHERE file_type = 'image') as image_count
            FROM media_files
            GROUP BY creator_id
        ) mc ON mc.creator_id = c.id
        ORDER BY c.username
    """)
    result = [dict(r) for r in rows]
    for r in result:
        if r.get("last_post_at") is not None and hasattr(r["last_post_at"], "isoformat"):
            r["last_post_at"] = r["last_post_at"].isoformat()
    _cache[cache_key] = (time.time(), result)
    return JSONResponse(
        content=result,
        headers={"Cache-Control": "public, max-age=900", "X-Cache": "MISS"},
    )


@app.get(
    "/api/creators/{creator_id}/posts",
    summary="List creator posts",
    description="Returns paginated posts for a creator with associated media. Supports filtering by media type and sorting by date, likes, or scrape time.",
)
async def list_posts(
    creator_id: int,
    page: int = Query(1, ge=1, description="Page number (1-based)"),
    per_page: int = Query(50, ge=1, le=200, description="Results per page"),
    media_type: Optional[str] = Query(None, pattern="^(video|image)$", description="Filter by media type: video or image"),
    sort: Optional[str] = Query(None, pattern="^(date|likes|scraped_desc)$", description="Sort order: date (default), likes, or scraped_desc"),
):
    offset = (page - 1) * per_page

    # Check in-memory cache
    cache_key = f"posts:{creator_id}:{page}:{per_page}:{media_type}:{sort}"
    cached = _cache.get(cache_key)
    if cached and time.time() - cached[0] < 60:
        return JSONResponse(content=cached[1], headers={"Cache-Control": "public, max-age=60", "X-Cache": "HIT"})

    pool = await get_db()

    # Paginate POSTS first (indexed, cheap), THEN fan out to media_files via a
    # correlated subquery. The old query JOINed posts×media_files and only then
    # GROUP/aggregated/sorted/LIMITed, forcing Postgres to materialize the full
    # post×media cross-product (~1.3M rows for big creators → measured 31s)
    # before discarding all but `per_page` rows. Now only `per_page` posts ever
    # touch media_files. (Agent 06 §2.1)  Soft-hides nhentai posts tagged
    # 'chinese' without 'english'/'japanese' (user policy).
    nh_filter = """NOT (
            c.service = 'nhentai'
            AND EXISTS (SELECT 1 FROM post_tags pt JOIN tags t ON t.id=pt.tag_id
                        WHERE pt.post_id=p.id AND t.category='language' AND t.name='chinese')
            AND NOT EXISTS (SELECT 1 FROM post_tags pt JOIN tags t ON t.id=pt.tag_id
                            WHERE pt.post_id=p.id AND t.category='language'
                            AND t.name IN ('english','japanese'))
          )"""

    params = [creator_id]
    media_exists = ""
    media_filter = ""
    if media_type:
        params.append(media_type)
        mt = len(params)
        media_exists = (f" AND EXISTS (SELECT 1 FROM media_files m "
                        f"WHERE m.post_id=p.id AND m.file_type=${mt})")
        media_filter = f" AND m.file_type = ${mt}"

    if sort == "likes":
        order_cte = "liked DESC, published_at DESC NULLS LAST"
        order_fin = "pg.liked DESC, pg.published_at DESC NULLS LAST"
    elif sort == "scraped_desc":
        order_cte = "scraped_at DESC NULLS LAST, id DESC"
        order_fin = "pg.scraped_at DESC NULLS LAST, pg.id DESC"
    else:
        order_cte = "published_at DESC NULLS LAST"
        order_fin = "pg.published_at DESC NULLS LAST"

    params.append(per_page)
    lim = len(params)
    params.append(offset)
    off = len(params)

    sql = f"""
        WITH page AS (
            SELECT p.id, p.title, p.content, p.published_at, p.scraped_at,
                   (pl.post_id IS NOT NULL) AS liked
            FROM posts p
            JOIN creators c ON c.id = p.creator_id
            LEFT JOIN post_likes pl ON pl.post_id = p.id
            WHERE p.creator_id = $1
              AND {nh_filter}{media_exists}
            ORDER BY {order_cte}
            LIMIT ${lim} OFFSET ${off}
        ),
        cnt AS (
            SELECT count(*) AS total
            FROM posts p
            JOIN creators c ON c.id = p.creator_id
            WHERE p.creator_id = $1
              AND {nh_filter}{media_exists}
        )
        SELECT pg.id, pg.title, pg.content, pg.published_at, pg.scraped_at,
               pg.liked,
               (SELECT json_agg(json_build_object(
                       'id', t.id,
                       'file_type', t.file_type,
                       'original_name', t.original_name,
                       'file_size', t.file_size
                   ) ORDER BY t.is_image DESC, t.id)
                  FROM (
                      SELECT m.id, m.file_type, m.original_name, m.file_size,
                             (m.file_type = 'image' OR m.file_type LIKE 'image%') AS is_image
                      FROM media_files m
                      WHERE m.post_id = pg.id{media_filter}
                      ORDER BY (m.file_type = 'image' OR m.file_type LIKE 'image%') DESC, m.id
                      LIMIT 6
                  ) t) AS media,
               (SELECT count(*) FROM media_files m
                  WHERE m.post_id = pg.id{media_filter}) AS media_count,
               (SELECT total FROM cnt) AS total_count
        FROM page pg
        ORDER BY {order_fin}
    """
    rows = await pool.fetch(sql, *params)
    total = rows[0]["total_count"] if rows else 0
    posts = [dict(r) for r in rows]

    # asyncpg returns json_agg as a string, parse it
    for post in posts:
        del post["total_count"]
        post.pop("scraped_at", None)  # internal sort key, not part of the API
        if isinstance(post.get("media"), str):
            post["media"] = json.loads(post["media"])
        # Convert large post IDs to strings for JS precision safety
        if post.get("id") is not None:
            post["id"] = str(post["id"])
        # Convert datetime to ISO string for JSON serialization
        if post.get("published_at") is not None and hasattr(post["published_at"], "isoformat"):
            post["published_at"] = post["published_at"].isoformat()

    result = {
        "posts": posts,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page if total > 0 else 0,
    }
    _cache[cache_key] = (time.time(), result)
    return JSONResponse(content=result, headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"})


@app.get(
    "/api/recent",
    summary="Recent posts",
    description="Returns the most recently scraped posts across all creators, with media and creator info. Results are cached for 60 seconds.",
)
async def recent_posts(limit: int = Query(60, ge=1, le=500, description="Maximum number of posts to return")):
    cache_key = f"recent:{limit}"
    cached = _cache.get(cache_key)
    if cached and time.time() - cached[0] < 60:
        return JSONResponse(content=cached[1], headers={"Cache-Control": "public, max-age=60", "X-Cache": "HIT"})

    pool = await get_db()
    # The frontend only renders 1 cover + up to 4 strip thumbs + a count label,
    # so json_agg-ing EVERY media row per post wasted ~85-90% of the payload.
    # We now cap the embedded media array at 6 rows per post via a LATERAL
    # (image-first so the gallery cover survives, then by id) and expose the
    # true total separately as `media_count`. The old inner JOIN media_files +
    # GROUP BY also acted as a "posts must have media" filter — preserved here
    # with an EXISTS so the result set is identical. (Agent 01 §media-overfetch)
    rows = await pool.fetch("""
        SELECT p.id, p.title, p.content, p.published_at, p.scraped_at,
               c.id as creator_id, c.username, c.service,
               (pl.post_id IS NOT NULL) as liked,
               mm.media,
               (SELECT count(*) FROM media_files m WHERE m.post_id = p.id) AS media_count
        FROM posts p
        JOIN creators c ON c.id = p.creator_id
        LEFT JOIN post_likes pl ON pl.post_id = p.id
        CROSS JOIN LATERAL (
            SELECT json_agg(json_build_object(
                       'id', t.id,
                       'file_type', t.file_type,
                       'original_name', t.original_name,
                       'file_size', t.file_size
                   ) ORDER BY t.is_image DESC, t.id) AS media
            FROM (
                SELECT m.id, m.file_type, m.original_name, m.file_size,
                       (m.file_type = 'image' OR m.file_type LIKE 'image%') AS is_image
                FROM media_files m
                WHERE m.post_id = p.id
                ORDER BY (m.file_type = 'image' OR m.file_type LIKE 'image%') DESC, m.id
                LIMIT 6
            ) t
        ) mm
        WHERE EXISTS (SELECT 1 FROM media_files m WHERE m.post_id = p.id)
        AND NOT (
            c.service = 'nhentai'
            AND EXISTS (SELECT 1 FROM post_tags pt JOIN tags t ON t.id=pt.tag_id
                        WHERE pt.post_id=p.id AND t.category='language' AND t.name='chinese')
            AND NOT EXISTS (SELECT 1 FROM post_tags pt JOIN tags t ON t.id=pt.tag_id
                            WHERE pt.post_id=p.id AND t.category='language'
                            AND t.name IN ('english','japanese'))
        )
        ORDER BY p.scraped_at DESC NULLS LAST, p.id DESC
        LIMIT $1
    """, limit)

    posts = [dict(r) for r in rows]
    for post in posts:
        if isinstance(post.get("media"), str):
            post["media"] = json.loads(post["media"])
        if post.get("id") is not None:
            post["id"] = str(post["id"])
        if post.get("published_at") is not None and hasattr(post["published_at"], "isoformat"):
            post["published_at"] = post["published_at"].isoformat()
        if post.get("scraped_at") is not None and hasattr(post["scraped_at"], "isoformat"):
            post["scraped_at"] = post["scraped_at"].isoformat()
        if post.get("creator_id") is not None:
            post["creator_id"] = str(post["creator_id"])
        # Home page only renders the title; trim long body content to slash payload.
        if post.get("content") and len(post["content"]) > 200:
            post["content"] = post["content"][:200].rstrip() + "…"

    _cache[cache_key] = (time.time(), posts)
    return JSONResponse(content=posts, headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"})


@app.get(
    "/api/media/{media_id}/info",
    summary="Media file info",
    description="Returns metadata for a single media file including creator, post title, and HLS availability flag.",
)
async def media_info(media_id: int):
    pool = await get_db()
    row = await pool.fetchrow("""
        SELECT m.*, c.username, p.title, p.content, p.published_at
        FROM media_files m
        JOIN creators c ON c.id = m.creator_id
        JOIN posts p ON p.id = m.post_id
        WHERE m.id = $1
    """, media_id)
    if not row:
        raise HTTPException(404, "Media not found")
    result = dict(row)
    result["hls_available"] = bool(result.get("hls_path"))
    # Convert large integers to strings to avoid JS precision loss
    if result.get("post_id") is not None:
        result["post_id"] = str(result["post_id"])
    return result


@app.get(
    "/api/media/{media_id}/context",
    summary="Media viewer context",
    description="Combined fetch: media info, all sibling media in the post, and like status in a single request. Used by the viewer page to avoid N+1 round trips.",
)
async def media_context(media_id: int):
    """Combined endpoint: media info + post media + like status in one request."""
    pool = await get_db()
    row = await pool.fetchrow("""
        SELECT m.*, c.username, p.title, p.content, p.published_at, p.id as post_id
        FROM media_files m JOIN creators c ON c.id = m.creator_id
        JOIN posts p ON p.id = m.post_id WHERE m.id = $1
    """, media_id)
    if not row:
        raise HTTPException(404, "Media not found")

    post_id = row["post_id"]
    media_rows, like_row = await asyncio.gather(
        pool.fetch("""
            SELECT id, file_type, original_name, file_size, page_order, download_url, hls_path
            FROM media_files WHERE post_id = $1
            ORDER BY COALESCE(page_order, 999999), id
        """, post_id),
        pool.fetchrow("SELECT 1 FROM post_likes WHERE post_id = $1", post_id),
    )

    info = dict(row)
    info["hls_available"] = bool(info.get("hls_path"))
    # Check if HEVC variants exist in the HLS directory.
    # hls_path may still hold the legacy GVFS prefix; rewrite to the CIFS
    # mount (same rewrite serve_hls / _resolve_media_path apply) so this
    # check works for the ~11.6k gvfs-path videos too.
    info["hevc_available"] = False
    if info.get("hls_path"):
        hls_dir = info["hls_path"]
        if hls_dir.startswith(_OLD_GVFS_PREFIX):
            hls_dir = NAS_BASE + hls_dir[len(_OLD_GVFS_PREFIX):]
        master = os.path.join(hls_dir, "stream.m3u8")
        try:
            if os.path.isfile(master):
                with open(master) as f:
                    info["hevc_available"] = "hvc1" in f.read()
        except Exception:
            pass
    if info.get("post_id") is not None:
        info["post_id"] = str(info["post_id"])

    # Resolve and probe video resolution (cache in DB)
    if info.get("file_type") == "video" and not info.get("height"):
        file_path = _resolve_media_path(info.get("local_path", ""))
        if os.path.isfile(file_path):
            try:
                import subprocess as _sp
                probe = await asyncio.to_thread(
                    lambda: _sp.run(
                        ["ffprobe", "-v", "quiet", "-select_streams", "v:0",
                         "-show_entries", "stream=width,height", "-of", "csv=p=0", file_path],
                        capture_output=True, text=True, timeout=10
                    )
                )
                if probe.returncode == 0 and probe.stdout.strip():
                    parts = probe.stdout.strip().split(",")
                    if len(parts) == 2:
                        w, h = int(parts[0]), int(parts[1])
                        info["width"] = w
                        info["height"] = h
                        # Cache in DB
                        await pool.execute(
                            "UPDATE media_files SET width=$1, height=$2 WHERE id=$3",
                            w, h, media_id
                        )
            except Exception:
                pass

    # Convert datetime fields to ISO strings for JSON serialization
    for key in ("published_at", "downloaded_at", "scraped_at", "first_scraped_at", "last_scraped_at"):
        if key in info and info[key] is not None:
            info[key] = info[key].isoformat() if hasattr(info[key], 'isoformat') else str(info[key])

    # Convert any remaining non-serializable types (memoryview, etc.)
    for key, val in info.items():
        if isinstance(val, memoryview):
            info[key] = bytes(val).hex()

    result_data = {
        "info": info,
        "media": [
            {
                "id": r["id"],
                "file_type": r["file_type"],
                "original_name": r["original_name"],
                "file_size": r["file_size"],
                "page_order": r["page_order"],
                "download_url": r["download_url"],
                "hls_available": bool(r["hls_path"]),
            }
            for r in media_rows
        ],
        "liked": like_row is not None,
    }

    # Use default=str to handle any remaining non-serializable types
    content = json.loads(json.dumps(result_data, default=str))
    return JSONResponse(content=content, headers={"Cache-Control": "public, max-age=60"})


@app.get(
    "/api/media/{media_id}/hls/{filename:path}",
    summary="Serve HLS segment or playlist",
    description="Serves an HLS playlist (.m3u8) or transport stream segment (.ts) for the given media. Playlists are served from an in-memory cache; segments are immutable and served with a 1-year max-age.",
)
async def serve_hls(media_id: int, filename: str):
    """Serve HLS playlist (.m3u8) or segment (.ts) files."""
    if ".." in filename or filename.startswith("/"):
        raise HTTPException(400, "Invalid filename")

    if not any(filename.endswith(ext) for ext in ('.m3u8', '.ts')):
        raise HTTPException(400, "Invalid file type")

    pool = await get_db()
    row = await _get_media_row(pool, media_id)
    if not row or not row["hls_path"]:
        raise HTTPException(404, "HLS not available for this media")

    hls_dir = row["hls_path"]
    # hls_path may still hold the legacy GVFS prefix; rewrite to the CIFS mount
    # (same rewrite _resolve_media_path applies to media files).
    if hls_dir.startswith(_OLD_GVFS_PREFIX):
        hls_dir = NAS_BASE + hls_dir[len(_OLD_GVFS_PREFIX):]

    file_path = os.path.join(hls_dir, filename)

    if filename.endswith('.m3u8'):
        # Playlists are tiny and effectively static for VOD HLS. Serve from an
        # in-memory TTL cache so the player's repeated manifest polling never
        # hits the CIFS NAS (collapses the 1.47s master-TTFB p95, Agent 04).
        now = time.monotonic()
        hit = _m3u8_cache.get(file_path)
        if hit is not None and hit[0] > now:
            body = hit[1]
        else:
            try:
                with open(file_path, "rb") as fh:
                    body = fh.read()
            except OSError:
                raise HTTPException(404, "HLS file not found")
            _m3u8_cache[file_path] = (now + _M3U8_TTL, body)
        return Response(
            content=body,
            media_type="application/vnd.apple.mpegurl",
            headers={
                "Cache-Control": "public, max-age=60",
                "Access-Control-Allow-Origin": "*",
            },
        )

    st = _cached_stat(file_path)
    if st is None:
        raise HTTPException(404, "HLS file not found")
    # .ts segments are content-addressed and never change once written.
    return _serve_file(
        file_path, "video/mp2t",
        "public, max-age=31536000, immutable",
        stat_result=st, accept_ranges=True,
        extra_headers={"Access-Control-Allow-Origin": "*"},
    )


@app.get(
    "/api/hls/stats",
    summary="HLS generation stats",
    description="Returns counts of total videos, videos with HLS ready, and videos pending HLS transcoding.",
)
async def hls_stats():
    """Return HLS generation progress."""
    pool = await get_db()
    stats = await pool.fetchrow("""
        SELECT
            COUNT(*) FILTER (WHERE file_type = 'video') as total_videos,
            COUNT(*) FILTER (WHERE file_type = 'video' AND hls_path IS NOT NULL) as hls_ready,
            COUNT(*) FILTER (WHERE file_type = 'video' AND hls_path IS NULL) as hls_pending
        FROM media_files
    """)
    return dict(stats)


def _parse_range(range_header: str, file_size: int):
    """Parse Range header, return (start, end)."""
    try:
        unit, ranges = range_header.split("=", 1)
        if unit.strip() != "bytes":
            return 0, file_size - 1
        start_str, end_str = ranges.split("-", 1)
        start = int(start_str) if start_str else 0
        end = int(end_str) if end_str else file_size - 1
        end = min(end, file_size - 1)
        return start, end
    except (ValueError, IndexError):
        return 0, file_size - 1


async def _stream_file(path: str, start: int, end: int):
    """Async generator that yields file chunks."""
    async with aiofiles.open(path, "rb") as f:
        await f.seek(start)
        remaining = end - start + 1
        while remaining > 0:
            chunk = await f.read(min(CHUNK_SIZE, remaining))
            if not chunk:
                break
            remaining -= len(chunk)
            yield chunk


@app.head("/api/media/{media_id}/stream")
@app.get(
    "/api/media/{media_id}/stream",
    summary="Stream media file",
    description="Streams a raw video or image file with Range request support. Returns 206 Partial Content for ranged requests. Uses os.sendfile/X-Accel-Redirect when available.",
)
async def stream_media(media_id: int, request: Request):
    pool = await get_db()
    row = await _get_media_row(pool, media_id)
    if not row:
        raise HTTPException(404, "Media not found")

    file_path = _resolve_media_path(row["local_path"])
    st = _cached_stat(file_path)
    if st is None:
        raise HTTPException(404, "File not found on disk")

    content_type = _video_content_type(file_path, row["file_type"])
    # FileResponse handles Range/206/416/If-Range/HEAD natively and uses
    # os.sendfile (pathsend, uvicorn>=0.32). When behind nginx, _serve_file
    # hands the bytes to nginx via X-Accel-Redirect entirely off the worker.
    return _serve_file(
        file_path, content_type, "public, max-age=3600",
        stat_result=st, accept_ranges=True,
    )


def _etag_response(cache_path: Path, media_type: str, request: Request):
    """Serve a cached thumbnail. The cache key is md5(media_id:w:ext) (or
    video_{id}/preview_{id}) so the file is content-addressed and immutable —
    serve it `immutable` and skip the ETag/If-None-Match revalidation round
    trip entirely (saved a full Tailscale RTT + a CIFS-class stat per revisit;
    on a 50-thumb grid that is ~50 RTTs — Agents 03/09). Also routes through
    _serve_file so the bytes go out via nginx X-Accel / os.sendfile instead of
    the Python copy loop (the dominant grid-page bottleneck — Agents 01/10).
    `Vary: Accept` is required for correctness once any intermediary cache
    sits in front of these (nginx X-Accel, SW, future CDN) since the same URL
    can serve JPEG/WebP/AVIF depending on the client's Accept (R2-07 #5).
    `request` kept for signature compatibility with existing call sites."""
    return _serve_file(
        str(cache_path), media_type,
        "public, max-age=31536000, immutable",
        accept_ranges=False,
        extra_headers={"Vary": "Accept"},
    )


@app.get(
    "/api/media/{media_id}/thumb",
    summary="Media thumbnail",
    description="Returns a resized thumbnail for an image or video. Format is negotiated via Accept header (AVIF/WebP/JPEG). Thumbnails are disk-cached; subsequent requests are served immutably.",
)
async def thumb_media(media_id: int, request: Request, w: int = Query(THUMB_MAX_WIDTH, ge=50, le=800, description="Thumbnail width in pixels")):
    """Serve a resized JPEG thumbnail with disk caching.

    Handles both image and video file types. For videos, delegates to
    video thumbnail generation (smart thumb or ffmpeg fallback).
    """
    # Format negotiation by Accept header. AVIF is opt-in (THUMB_AVIF=1) until
    # the warmup pipeline (R2-01) has produced AVIFs at the served widths — on
    # a cold miss with on-demand AVIF encode, the single worker would block
    # 1–3 s. Once warmed, flipping the flag is the win (~−41 % bytes vs WebP).
    accept = request.headers.get("accept", "")
    use_avif = os.environ.get("THUMB_AVIF", "") == "1" and "image/avif" in accept
    use_webp = (not use_avif) and "image/webp" in accept
    if use_avif:
        ext = "avif"; media_type = "image/avif"
    elif use_webp:
        ext = "webp"; media_type = "image/webp"
    else:
        ext = "jpg"; media_type = "image/jpeg"
    cache_key = hashlib.md5(f"{media_id}:{w}:{ext}".encode()).hexdigest()
    cache_path = THUMB_CACHE_DIR / f"{cache_key}.{ext}"

    # Check disk cache BEFORE hitting the database
    if cache_path.exists():
        return _etag_response(cache_path, media_type, request)

    # Cache miss - need DB lookup for source file
    pool = await get_db()
    row = await _get_media_row(pool, media_id)
    if not row:
        raise HTTPException(404, "Media not found")

    # ── Video files: delegate to video thumbnail logic ──
    if row["file_type"] == "video":
        video_cache = THUMB_CACHE_DIR / f"video_{media_id}.jpg"
        if video_cache.exists():
            return _etag_response(video_cache, "image/jpeg", request)

        file_path = _resolve_media_path(row["local_path"])
        if not os.path.isfile(file_path):
            raise HTTPException(404, "File not found on disk")

        # Try smart thumbnail generation first
        async with _THUMB_SEMAPHORE:
            try:
                success = await asyncio.to_thread(generate_smart_thumb, file_path, str(video_cache), w)
            except Exception:
                success = False

        if not success:
            # Fallback: extract first frame with ffmpeg
            async with _THUMB_SEMAPHORE:
                try:
                    proc = await asyncio.create_subprocess_exec(
                        "ffmpeg", "-y",
                        "-i", file_path,
                        "-vframes", "1",
                        "-vf", f"scale={w}:-1",
                        "-f", "image2",
                        str(video_cache),
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE,
                    )
                    try:
                        await asyncio.wait_for(proc.communicate(), timeout=30)
                    except asyncio.TimeoutError:
                        proc.kill()
                        await proc.wait()
                        raise HTTPException(500, "Video thumbnail extraction timed out")

                    if proc.returncode != 0 or not video_cache.exists():
                        raise HTTPException(500, "Failed to extract video thumbnail")
                except FileNotFoundError:
                    raise HTTPException(500, "ffmpeg not found on server")

        return _etag_response(video_cache, "image/jpeg", request)

    # ── Image files: resize with PIL ──
    ft = (row["file_type"] or "").lower()
    if ft != "image" and not ft.startswith("image"):
        raise HTTPException(400, "Thumbnails only for images and videos")

    file_path = _resolve_media_path(row["local_path"])
    if not os.path.isfile(file_path):
        raise HTTPException(404, "File not found on disk")

    # Generate thumbnail (PIL is CPU-bound, run in thread pool)
    def _generate_thumb():
        with Image.open(file_path) as img:
            img = img.convert("RGB")
            ratio = w / img.width
            new_h = int(img.height * ratio)
            img = img.resize((w, new_h), Image.LANCZOS)
            if use_avif:
                # Pillow 12+ has native AVIF. speed=9 ≈ webp method=4 in encode
                # time; quality 50 ≈ webp quality 75 in visual quality but
                # ~−41 % bytes (R2-07 #3). Opt-in via THUMB_AVIF=1.
                img.save(cache_path, "AVIF", quality=50, speed=9)
            elif use_webp:
                img.save(cache_path, "WEBP", quality=THUMB_QUALITY, method=4)
            else:
                img.save(cache_path, "JPEG", quality=THUMB_QUALITY, optimize=True)

    try:
        async with _THUMB_SEMAPHORE:
            await asyncio.to_thread(_generate_thumb)
    except Exception:
        return _serve_file(file_path, "image/jpeg", "public, max-age=3600")

    return _etag_response(cache_path, media_type, request)


@app.get(
    "/api/media/{media_id}/video-thumb",
    summary="Video thumbnail",
    description="Extracts the best frame from a video using smart thumbnail generation (or ffmpeg fallback) and returns it as a JPEG. Result is disk-cached and served immutably on subsequent requests.",
)
async def video_thumb_media(media_id: int, request: Request):
    """Extract first frame of a video and serve as JPEG thumbnail with disk caching."""
    cache_path = THUMB_CACHE_DIR / f"video_{media_id}.jpg"

    # Check disk cache BEFORE hitting the database
    if cache_path.exists():
        return _etag_response(cache_path, "image/jpeg", request)

    # Cache miss - need DB lookup for source file
    pool = await get_db()
    row = await _get_media_row(pool, media_id)
    if not row:
        raise HTTPException(404, "Media not found")

    if row["file_type"] != "video":
        raise HTTPException(400, "Video thumbnails only for video files")

    file_path = _resolve_media_path(row["local_path"])
    if not os.path.isfile(file_path):
        raise HTTPException(404, "File not found on disk")

    # Try smart thumbnail generation first (samples multiple frames, picks best)
    async with _THUMB_SEMAPHORE:
        try:
            success = await asyncio.to_thread(generate_smart_thumb, file_path, str(cache_path), THUMB_MAX_WIDTH)
        except Exception:
            success = False

    if not success:
        # Fallback: extract first frame with ffmpeg
        async with _THUMB_SEMAPHORE:
            try:
                proc = await asyncio.create_subprocess_exec(
                    "ffmpeg", "-y",
                    "-i", file_path,
                    "-vframes", "1",
                    "-vf", f"scale={THUMB_MAX_WIDTH}:-1",
                    "-f", "image2",
                    str(cache_path),
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
                try:
                    await asyncio.wait_for(proc.communicate(), timeout=30)
                except asyncio.TimeoutError:
                    proc.kill()
                    await proc.wait()
                    raise HTTPException(500, "Video thumbnail extraction timed out")

                if proc.returncode != 0 or not cache_path.exists():
                    raise HTTPException(500, "Failed to extract video thumbnail")
            except FileNotFoundError:
                raise HTTPException(500, "ffmpeg not found on server")

    return _etag_response(cache_path, "image/jpeg", request)


@app.get(
    "/api/media/{media_id}/preview",
    summary="Animated video preview",
    description="Returns an animated WebP preview (3-second clip at 10 fps, starting at 25% of duration) for a video file. Result is disk-cached and served immutably.",
)
async def video_preview(media_id: int):
    """Serve an animated WebP preview of a video (3-second clip at 10fps)."""
    cache_path = THUMB_CACHE_DIR / f"preview_{media_id}.webp"

    # Check disk cache BEFORE hitting the database
    if cache_path.exists():
        return _serve_file(str(cache_path), "image/webp",
                            "public, max-age=31536000, immutable",
                            accept_ranges=False)

    pool = await get_db()
    row = await _get_media_row(pool, media_id)
    if not row:
        raise HTTPException(404, "Media not found")
    if row["file_type"] != "video":
        raise HTTPException(400, "Previews only for video files")

    file_path = _resolve_media_path(row["local_path"])
    if not os.path.isfile(file_path):
        raise HTTPException(404, "File not found on disk")

    async def _generate_preview():
        # Get duration via ffprobe
        probe = await asyncio.create_subprocess_exec(
            "ffprobe", "-v", "error", "-show_entries", "format=duration",
            "-of", "json", file_path,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )
        stdout, _ = await asyncio.wait_for(probe.communicate(), timeout=15)
        info = json.loads(stdout)
        duration = float(info.get("format", {}).get("duration", 0))
        if duration < 2:
            return False

        start_time = duration * 0.25

        proc = await asyncio.create_subprocess_exec(
            "ffmpeg", "-y",
            "-ss", str(start_time),
            "-i", file_path,
            "-t", "3",
            "-vf", "scale=320:-1:flags=lanczos,fps=10",
            "-loop", "0",
            "-quality", "50",
            "-compression_level", "4",
            "-f", "webp",
            str(cache_path),
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )
        try:
            await asyncio.wait_for(proc.communicate(), timeout=60)
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            return False

        return proc.returncode == 0 and cache_path.exists()

    try:
        success = await _generate_preview()
    except Exception:
        success = False

    if not success:
        raise HTTPException(500, "Failed to generate preview")

    return _serve_file(str(cache_path), "image/webp",
                        "public, max-age=31536000, immutable",
                        accept_ranges=False)


# ── Storyboard sprites (scrub-bar preview thumbnails) ─────────────────────
#
# The player asks for /storyboard.vtt; the cues inside it point at regions of
# /storyboard.jpg via #xywh media fragments. That is the standard WebVTT
# thumbnail format, so the artefacts stay useful if the player changes.
#
# Both files are generated together, on demand, and cached in THUMB_CACHE_DIR
# next to the existing preview_/video_ derivatives — which means they are
# already covered by the nginx /_thumb_internal/ block and the immutable
# caching path, with no infra changes.

_storyboard_log = logging.getLogger("storyboard")

STORYBOARD_TILE_W = 160          # px; height is derived from the source aspect
STORYBOARD_MAX_TILES = 100       # 10x10 grid keeps the sprite under ~200 KB
STORYBOARD_COLS = 10
STORYBOARD_MIN_INTERVAL = 2.0    # seconds; denser than this is not useful

# One generation per media id per process. A storyboard decodes a whole video,
# so letting two concurrent requests both start ffmpeg would double the most
# expensive operation this server performs.
_storyboard_locks: dict[int, asyncio.Lock] = {}


def _storyboard_paths(media_id: int):
    return (THUMB_CACHE_DIR / f"storyboard_{media_id}.jpg",
            THUMB_CACHE_DIR / f"storyboard_{media_id}.vtt")


def _storyboard_decode_source(hls_path: Optional[str], fallback: str) -> str:
    """Pick the cheapest thing to decode for the storyboard.

    Tiling a video forces a full decode. When HLS variants exist we decode the
    *smallest* one instead of the original: for a 1080p source that is roughly
    an order of magnitude less work for images that end up 160px wide anyway.
    """
    if not hls_path:
        return fallback
    hls_dir = hls_path
    if hls_dir.startswith(_OLD_GVFS_PREFIX):
        hls_dir = NAS_BASE + hls_dir[len(_OLD_GVFS_PREFIX):]
    master = os.path.join(hls_dir, "stream.m3u8")
    if not os.path.isfile(master):
        return fallback
    try:
        with open(master) as f:
            text = f.read()
        # Single-bitrate output writes a media playlist here, not a master.
        if "#EXT-X-STREAM-INF" not in text:
            return master
        lines = text.splitlines()
        best, best_h = None, None
        for i, line in enumerate(lines):
            if not line.startswith("#EXT-X-STREAM-INF"):
                continue
            uri = lines[i + 1].strip() if i + 1 < len(lines) else ""
            if not uri or uri.startswith("#"):
                continue
            m = re.search(r"RESOLUTION=\d+x(\d+)", line)
            height = int(m.group(1)) if m else 1 << 30
            if best_h is None or height < best_h:
                best_h, best = height, os.path.join(hls_dir, uri)
        if best and os.path.isfile(best):
            return best
    except Exception:
        pass
    return fallback


def _format_vtt_timestamp(seconds: float) -> str:
    if seconds < 0:
        seconds = 0.0
    h, rem = divmod(int(seconds), 3600)
    m, s = divmod(rem, 60)
    ms = int(round((seconds - int(seconds)) * 1000))
    if ms == 1000:      # rounding carried into the next second
        ms, s = 0, s + 1
    return f"{h:02d}:{m:02d}:{s:02d}.{ms:03d}"


async def _probe_video_geometry(path: str):
    """Return (duration_seconds, width, height) or None."""
    proc = await asyncio.create_subprocess_exec(
        "ffprobe", "-v", "error",
        "-select_streams", "v:0",
        "-show_entries", "format=duration:stream=width,height",
        "-of", "json", path,
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    try:
        stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        return None
    try:
        info = json.loads(stdout)
        duration = float(info.get("format", {}).get("duration", 0) or 0)
        stream = (info.get("streams") or [{}])[0]
        width = int(stream.get("width") or 0)
        height = int(stream.get("height") or 0)
    except Exception:
        return None
    if duration <= 0 or width <= 0 or height <= 0:
        return None
    return duration, width, height


async def _generate_storyboard(media_id: int, src: str) -> bool:
    sprite_path, vtt_path = _storyboard_paths(media_id)

    geom = await _probe_video_geometry(src)
    if not geom:
        return False
    duration, width, height = geom

    # Uniform tiles, derived from the source aspect and rounded to even pixels
    # (odd dimensions break several yuv420p filter paths).
    tile_w = STORYBOARD_TILE_W
    tile_h = max(2, int(round(tile_w * height / width / 2)) * 2)

    interval = max(STORYBOARD_MIN_INTERVAL, duration / STORYBOARD_MAX_TILES)
    tiles = max(1, min(STORYBOARD_MAX_TILES, int(duration // interval)))
    cols = min(STORYBOARD_COLS, tiles)
    rows = max(1, (tiles + cols - 1) // cols)

    vf = (
        f"fps=1/{interval:.6f},"
        f"scale={tile_w}:{tile_h}:force_original_aspect_ratio=decrease,"
        f"pad={tile_w}:{tile_h}:(ow-iw)/2:(oh-ih)/2,"
        f"tile={cols}x{rows}"
    )
    tmp_sprite = sprite_path.with_suffix(".jpg.tmp")
    proc = await asyncio.create_subprocess_exec(
        "ffmpeg", "-y",
        "-loglevel", "error",
        "-i", src,
        "-an", "-sn",
        "-vf", vf,
        "-frames:v", "1",
        "-q:v", "6",
        # The output name ends in .tmp, so ffmpeg cannot infer a muxer from the
        # extension — state it explicitly rather than relying on the filename.
        "-f", "image2",
        str(tmp_sprite),
        stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
    )
    try:
        _, stderr = await asyncio.wait_for(proc.communicate(), timeout=180)
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        tmp_sprite.unlink(missing_ok=True)
        _storyboard_log.warning("storyboard %s: ffmpeg timed out", media_id)
        return False
    if proc.returncode != 0 or not tmp_sprite.exists() or tmp_sprite.stat().st_size == 0:
        tmp_sprite.unlink(missing_ok=True)
        _storyboard_log.warning("storyboard %s: ffmpeg failed (rc=%s): %s",
                    media_id, proc.returncode,
                    (stderr or b"").decode("utf-8", "replace")[:300])
        return False

    # ffmpeg pads the final row with the last frame when the tile grid is not
    # exactly filled, so only emit cues for tiles we actually asked for.
    cues = []
    for i in range(tiles):
        start = i * interval
        end = min(duration, start + interval)
        if end <= start:
            break
        x = (i % cols) * tile_w
        y = (i // cols) * tile_h
        cues.append(
            f"{_format_vtt_timestamp(start)} --> {_format_vtt_timestamp(end)}\n"
            f"/api/media/{media_id}/storyboard.jpg#xywh={x},{y},{tile_w},{tile_h}"
        )
    vtt = "WEBVTT\n\n" + "\n\n".join(cues) + "\n"

    tmp_vtt = vtt_path.with_suffix(".vtt.tmp")
    tmp_vtt.write_text(vtt)
    # Publish both atomically-ish: the sprite first, so a reader that sees the
    # .vtt can always resolve the image it points at.
    os.replace(tmp_sprite, sprite_path)
    os.replace(tmp_vtt, vtt_path)
    return True


async def _ensure_storyboard(media_id: int) -> bool:
    sprite_path, vtt_path = _storyboard_paths(media_id)
    if sprite_path.exists() and vtt_path.exists():
        return True

    lock = _storyboard_locks.setdefault(media_id, asyncio.Lock())
    async with lock:
        # Re-check: another request may have generated it while we waited.
        if sprite_path.exists() and vtt_path.exists():
            return True

        pool = await get_db()
        row = await _get_media_row(pool, media_id)
        if not row or row["file_type"] != "video":
            return False
        file_path = _resolve_media_path(row["local_path"])
        if not os.path.isfile(file_path):
            return False
        src = _storyboard_decode_source(row["hls_path"], file_path)
        try:
            return await _generate_storyboard(media_id, src)
        except Exception:
            return False
        finally:
            _storyboard_locks.pop(media_id, None)


@app.get(
    "/api/media/{media_id}/storyboard.vtt",
    summary="Storyboard thumbnail track",
    description="WebVTT track of scrub-bar preview thumbnails. Cues reference regions of the matching storyboard.jpg sprite via #xywh fragments. Generated on first request and disk-cached.",
)
async def storyboard_vtt(media_id: int):
    """Serve the WebVTT thumbnail track for a video."""
    _, vtt_path = _storyboard_paths(media_id)
    if not vtt_path.exists():
        if not await _ensure_storyboard(media_id):
            # The player treats this as "no previews" and still shows a
            # timestamp bubble, so a 404 is a normal, non-fatal outcome.
            raise HTTPException(404, "Storyboard unavailable")
    return _serve_file(str(vtt_path), "text/vtt",
                       "public, max-age=31536000, immutable",
                       accept_ranges=False)


@app.get(
    "/api/media/{media_id}/storyboard.jpg",
    summary="Storyboard sprite sheet",
    description="Sprite sheet of preview frames referenced by the matching storyboard.vtt track.",
)
async def storyboard_sprite(media_id: int):
    """Serve the storyboard sprite sheet for a video."""
    sprite_path, _ = _storyboard_paths(media_id)
    if not sprite_path.exists():
        if not await _ensure_storyboard(media_id):
            raise HTTPException(404, "Storyboard unavailable")
    return _serve_file(str(sprite_path), "image/jpeg",
                       "public, max-age=31536000, immutable",
                       accept_ranges=False)


@app.get(
    "/api/performers",
    summary="List performers",
    description="Returns all performers with their total video counts, ordered by video count descending. Results are cached for 2 minutes.",
)
async def list_performers():
    """List all performers with video counts."""
    pool = await get_db()
    rows = await pool.fetch("""
        SELECT p.id, p.name, COUNT(mp.media_id) as video_count
        FROM performers p
        JOIN media_performers mp ON mp.performer_id = p.id
        GROUP BY p.id, p.name
        ORDER BY video_count DESC
    """)
    return JSONResponse(
        content=[dict(r) for r in rows],
        headers={"Cache-Control": "public, max-age=120"},
    )


@app.get(
    "/api/performers/{performer_id}/videos",
    summary="Performer videos",
    description="Returns paginated videos for a performer across all creators and collections, ordered by media ID descending.",
)
async def performer_videos(
    performer_id: int,
    page: int = Query(1, ge=1, description="Page number (1-based)"),
    per_page: int = Query(48, ge=1, le=200, description="Results per page"),
):
    """List all videos for a performer across all collections."""
    pool = await get_db()
    offset = (page - 1) * per_page

    # Get performer name
    performer = await pool.fetchrow(
        "SELECT id, name FROM performers WHERE id = $1", performer_id
    )
    if not performer:
        raise HTTPException(404, "Performer not found")

    rows = await pool.fetch("""
        SELECT m.id, m.file_type, m.original_name, m.file_size, m.width, m.height,
               p.title, p.published_at, c.username, c.service, c.id as creator_id,
               m.hls_path IS NOT NULL as hls_available,
               COUNT(*) OVER() as total_count
        FROM media_performers mp
        JOIN media_files m ON m.id = mp.media_id
        JOIN posts p ON p.id = m.post_id
        JOIN creators c ON c.id = m.creator_id
        WHERE mp.performer_id = $1 AND m.file_type = 'video'
        ORDER BY m.id DESC
        LIMIT $2 OFFSET $3
    """, performer_id, per_page, offset)

    total = rows[0]["total_count"] if rows else 0

    videos = []
    for r in rows:
        d = dict(r)
        del d["total_count"]
        if d.get("published_at") is not None and hasattr(d["published_at"], "isoformat"):
            d["published_at"] = d["published_at"].isoformat()
        videos.append(d)

    return {
        "performer": dict(performer),
        "videos": videos,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page if total > 0 else 0,
    }


@app.get(
    "/api/search",
    summary="Full-text search posts",
    description="Search posts by title/content with optional creator and media-type filters.",
)
async def search_posts(
    q: str = Query(..., min_length=1, description="Search query (matched against title and content)"),
    page: int = Query(1, ge=1, description="Page number (1-based)"),
    per_page: int = Query(50, ge=1, le=200, description="Results per page"),
    creator_id: Optional[int] = Query(None, description="Restrict results to a specific creator ID"),
    media_type: Optional[str] = Query(None, pattern="^(video|image|comic)$", description="Filter by media type: video, image (single), or comic (gallery)"),
):
    cache_key = f"search:{q}:{page}:{per_page}:{creator_id}:{media_type}"
    cached = _cache.get(cache_key)
    if cached and time.time() - cached[0] < 30:
        return JSONResponse(content=cached[1],
                            headers={"Cache-Control": "public, max-age=30", "X-Cache": "HIT"})

    pool = await get_db()
    offset = (page - 1) * per_page

    where_parts = ["(p.title ILIKE $1 OR p.content ILIKE $1)"]
    params: list = [f"%{q}%"]
    if creator_id is not None:
        params.append(creator_id)
        where_parts.append(f"p.creator_id = ${len(params)}")
    where_sql = " AND ".join(where_parts)

    # media_type filters via HAVING on per-post media counts:
    #   video: post has at least one video file
    #   image: post has exactly one image and no videos (single-image post)
    #   comic: post has 2+ images and no videos (gallery)
    having_sql = ""
    if media_type == "video":
        having_sql = "HAVING COUNT(m.id) FILTER (WHERE m.file_type = 'video') > 0"
    elif media_type == "image":
        having_sql = (
            "HAVING COUNT(m.id) FILTER (WHERE m.file_type = 'video') = 0 "
            "AND COUNT(m.id) FILTER (WHERE m.file_type = 'image') = 1"
        )
    elif media_type == "comic":
        having_sql = (
            "HAVING COUNT(m.id) FILTER (WHERE m.file_type = 'video') = 0 "
            "AND COUNT(m.id) FILTER (WHERE m.file_type = 'image') >= 2"
        )

    # Two query shapes, both driven by the GIN trigram indexes on
    # posts.title/content (idx_posts_title_trgm / idx_posts_content_trgm) so the
    # ILIKE '%q%' is a bitmap index scan, never a hypertable seq-scan.
    #
    # The previous single query LEFT JOIN'd media_files and GROUP BY'd BEFORE the
    # LIMIT, so the planner hash-joined ALL 2.8M media rows for every search
    # (~1.4s). We now restrict the media aggregation to only the posts we keep:
    #   - no media_type filter: LIMIT the matched posts FIRST (MATERIALIZED CTE),
    #     then aggregate media for just that page (~50 posts via idx_media_post).
    #   - media_type filter: composition is needed to filter, so aggregate over
    #     the trigram-matched posts (MATERIALIZED so the small post set is built
    #     before the join), apply HAVING, then window-count + paginate.
    if having_sql:
        params.extend([per_page, offset])
        sql = f"""
            WITH matched AS MATERIALIZED (
                SELECT p.id, p.title, p.content, p.published_at, p.creator_id
                FROM posts p
                WHERE {where_sql}
            ),
            agg AS (
                SELECT mt.id, mt.title, mt.content, mt.published_at, mt.creator_id,
                       COUNT(m.id) FILTER (WHERE m.file_type = 'video') as video_count,
                       COUNT(m.id) FILTER (WHERE m.file_type = 'image') as image_count,
                       MIN(m.id) FILTER (WHERE m.file_type = 'image') as first_image_id,
                       MIN(m.id) FILTER (WHERE m.file_type = 'video') as first_video_id,
                       MIN(m.id) as first_media_id
                FROM matched mt
                LEFT JOIN media_files m ON m.post_id = mt.id
                GROUP BY mt.id, mt.title, mt.content, mt.published_at, mt.creator_id
                {having_sql}
            )
            SELECT a.id, a.title, a.content, a.published_at,
                   c.username, c.id as creator_id,
                   a.video_count, a.image_count,
                   a.first_image_id, a.first_video_id, a.first_media_id,
                   COUNT(*) OVER() as total_count
            FROM agg a
            JOIN creators c ON c.id = a.creator_id
            ORDER BY a.published_at DESC NULLS LAST
            LIMIT ${len(params) - 1} OFFSET ${len(params)}
        """
        rows = await pool.fetch(sql, *params)
        total = rows[0]["total_count"] if rows else 0
    else:
        # Page of matched posts first (trigram-indexed), then aggregate media for
        # just that page. total_count comes from a separate cheap indexed COUNT.
        page_params = params + [per_page, offset]
        sql = f"""
            WITH matched AS MATERIALIZED (
                SELECT p.id, p.title, p.content, p.published_at, p.creator_id
                FROM posts p
                WHERE {where_sql}
                ORDER BY p.published_at DESC NULLS LAST
                LIMIT ${len(page_params) - 1} OFFSET ${len(page_params)}
            )
            SELECT mt.id, mt.title, mt.content, mt.published_at,
                   c.username, c.id as creator_id,
                   COUNT(m.id) FILTER (WHERE m.file_type = 'video') as video_count,
                   COUNT(m.id) FILTER (WHERE m.file_type = 'image') as image_count,
                   MIN(m.id) FILTER (WHERE m.file_type = 'image') as first_image_id,
                   MIN(m.id) FILTER (WHERE m.file_type = 'video') as first_video_id,
                   MIN(m.id) as first_media_id
            FROM matched mt
            JOIN creators c ON c.id = mt.creator_id
            LEFT JOIN media_files m ON m.post_id = mt.id
            GROUP BY mt.id, mt.title, mt.content, mt.published_at, c.username, c.id
            ORDER BY mt.published_at DESC NULLS LAST
        """
        count_sql = f"SELECT COUNT(*) AS total FROM posts p WHERE {where_sql}"
        rows, total_row = await asyncio.gather(
            pool.fetch(sql, *page_params),
            pool.fetchrow(count_sql, *params),
        )
        total = total_row["total"] if total_row else 0
    results = []
    for r in rows:
        d = dict(r)
        d.pop("total_count", None)  # only present in the media_type branch
        if d.get("id") is not None:
            d["id"] = str(d["id"])
        if d.get("published_at") is not None and hasattr(d["published_at"], "isoformat"):
            d["published_at"] = d["published_at"].isoformat()
        results.append(d)
    payload = {
        "posts": results,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page if total > 0 else 0,
    }
    _cache[cache_key] = (time.time(), payload)
    return JSONResponse(
        content=payload,
        headers={"Cache-Control": "public, max-age=30", "X-Cache": "MISS"},
    )


@app.post(
    "/api/posts/{post_id}/like",
    summary="Toggle like on a post",
    description="Like a post if not yet liked, or unlike it if already liked.",
)
async def toggle_like(post_id: int):
    """Toggle like on a post. If already liked, unlike it; otherwise like it."""
    pool = await get_db()
    # Try to delete (unlike). If a row was deleted, it was previously liked.
    result = await pool.execute("DELETE FROM post_likes WHERE post_id = $1", post_id)
    if result == "DELETE 1":
        return {"liked": False, "post_id": str(post_id)}
    # Not previously liked - insert (like)
    await pool.execute("INSERT INTO post_likes (post_id) VALUES ($1)", post_id)
    return {"liked": True, "post_id": str(post_id)}


@app.get(
    "/api/posts/{post_id}/like",
    summary="Check like status",
    description="Return whether the given post is currently liked.",
)
async def check_like(post_id: int):
    """Check if a post is liked."""
    pool = await get_db()
    row = await pool.fetchrow("SELECT 1 FROM post_likes WHERE post_id = $1", post_id)
    return {"liked": row is not None, "post_id": str(post_id)}


# === Deletion endpoints ===

@app.delete(
    "/api/posts/{post_id}",
    summary="Delete a post",
    description="Delete a post and its dependencies (media, likes, tag-links). Files on disk are NOT removed — only DB rows.",
    dependencies=[Depends(require_trusted_client)],
)
async def delete_post(post_id: int):
    """Delete a post and its dependencies (media, likes, tag-links).
    Files on disk are NOT removed — only DB rows."""
    pool = await get_db()
    async with pool.acquire() as conn:
        async with conn.transaction():
            row = await conn.fetchrow("SELECT id FROM posts WHERE id = $1", post_id)
            if not row:
                raise HTTPException(status_code=404, detail="post not found")
            await conn.execute("DELETE FROM media_files WHERE post_id = $1", post_id)
            await conn.execute("DELETE FROM post_likes WHERE post_id = $1", post_id)
            try:
                await conn.execute("DELETE FROM post_tags WHERE post_id = $1", post_id)
            except Exception:
                pass  # post_tags table may not exist in some installs
            await conn.execute("DELETE FROM posts WHERE id = $1", post_id)
    _cache_invalidate_feeds()
    return {"deleted": True, "post_id": str(post_id)}


@app.delete(
    "/api/creators/{creator_id}",
    summary="Delete a creator",
    description="Delete a creator and ALL their posts, media, likes, and tag-links. Files on disk are NOT removed — only DB rows.",
    dependencies=[Depends(require_trusted_client)],
)
async def delete_creator(creator_id: int):
    """Delete a creator and ALL their posts, media, likes, and tag-links.
    Files on disk are NOT removed — only DB rows."""
    pool = await get_db()
    async with pool.acquire() as conn:
        async with conn.transaction():
            row = await conn.fetchrow("SELECT id, username FROM creators WHERE id = $1", creator_id)
            if not row:
                raise HTTPException(status_code=404, detail="creator not found")
            # Collect post IDs first so we can clean up post-scoped tables
            post_ids = [r["id"] for r in await conn.fetch(
                "SELECT id FROM posts WHERE creator_id = $1", creator_id
            )]
            await conn.execute("DELETE FROM media_files WHERE creator_id = $1", creator_id)
            if post_ids:
                await conn.execute("DELETE FROM post_likes WHERE post_id = ANY($1::bigint[])", post_ids)
                try:
                    await conn.execute("DELETE FROM post_tags WHERE post_id = ANY($1::bigint[])", post_ids)
                except Exception:
                    pass
            await conn.execute("DELETE FROM posts WHERE creator_id = $1", creator_id)
            await conn.execute("DELETE FROM creators WHERE id = $1", creator_id)
    _cache_invalidate_feeds()
    return {"deleted": True, "creator_id": creator_id, "post_count": len(post_ids)}


@app.delete(
    "/api/media/{media_id}",
    summary="Delete a media file record",
    description="Delete a single media_files row. The file on disk is NOT removed.",
    dependencies=[Depends(require_trusted_client)],
)
async def delete_media(media_id: int):
    """Delete a single media file row (the file on disk is not removed)."""
    pool = await get_db()
    result = await pool.execute("DELETE FROM media_files WHERE id = $1", media_id)
    if result != "DELETE 1":
        raise HTTPException(status_code=404, detail="media not found")
    _cache_invalidate_feeds()
    return {"deleted": True, "media_id": media_id}


# === External Content API ===

class ExternalMediaItem(BaseModel):
    local_path: str
    file_type: str = "image"
    original_name: str = ""
    page_order: Optional[int] = None
    download_url: Optional[str] = None


class ExternalIngestRequest(BaseModel):
    source_url: str = Field(..., min_length=1)
    source_site: str = Field(..., min_length=1)
    title: str = ""
    content: str = ""
    media_files: list[ExternalMediaItem] = []
    username: Optional[str] = None


@app.get(
    "/api/posts/{post_id}/media",
    summary="List media for a post",
    description="Return all media files for a post, along with post and creator metadata. post_id accepted as string to avoid JS precision loss.",
)
async def get_post_media(post_id: str):
    """Return all media files for a specific post with post and creator info."""
    post_id = int(post_id)  # Accept string to avoid JS precision loss
    pool = await get_db()

    # Get post with creator info
    post_row = await pool.fetchrow("""
        SELECT p.id, p.title, p.content, p.published_at, p.source_url,
               p.creator_id, c.username
        FROM posts p
        JOIN creators c ON c.id = p.creator_id
        WHERE p.id = $1
    """, post_id)

    if not post_row:
        raise HTTPException(404, "Post not found")

    # Get media files ordered by page_order (if set) then by id
    media_rows = await pool.fetch("""
        SELECT id, post_id, creator_id, download_url, original_name,
               local_path, file_type, file_size, attachment_type,
               downloaded_at, hls_path, page_order
        FROM media_files
        WHERE post_id = $1
        ORDER BY COALESCE(page_order, 999999), id
    """, post_id)

    return {
        "post": {
            "id": post_row["id"],
            "title": post_row["title"],
            "content": post_row["content"],
            "published_at": post_row["published_at"].isoformat() if post_row["published_at"] else None,
            "source_url": post_row["source_url"],
        },
        "creator": {
            "creator_id": post_row["creator_id"],
            "username": post_row["username"],
        },
        "media": [
            {
                "id": r["id"],
                "file_type": r["file_type"],
                "original_name": r["original_name"],
                "local_path": r["local_path"],
                "file_size": r["file_size"],
                "page_order": r["page_order"],
                "download_url": r["download_url"],
                "hls_available": bool(r["hls_path"]),
            }
            for r in media_rows
        ],
    }


@app.post(
    "/api/ingest/external",
    summary="Ingest external content",
    description="Ingest content from an external site. Creates creator (if needed), post, and media_file records. Returns 409 on duplicate source_url.",
    dependencies=[Depends(require_trusted_client)],
)
async def ingest_external(body: ExternalIngestRequest):
    """
    Ingest content from an external site.
    Creates creator (if needed), post, and media_file records.
    """
    pool = await get_db()

    # Check for duplicate source_url
    existing = await pool.fetchval(
        "SELECT 1 FROM posts WHERE source_url = $1 LIMIT 1",
        body.source_url,
    )
    if existing:
        raise HTTPException(409, f"Post with source_url already exists: {body.source_url}")

    # Determine username from body or extract from source_url
    username = body.username or body.source_site
    service = body.source_site

    # Get or create creator
    creator_row = await pool.fetchrow(
        "SELECT id FROM creators WHERE user_id = $1 AND service = $2",
        username, service,
    )
    if creator_row:
        creator_id = creator_row["id"]
        # Update source_site if needed
        await pool.execute(
            """UPDATE creators SET source_site = $1, last_scraped_at = NOW()
               WHERE id = $2 AND (source_site IS NULL OR source_site != $1)""",
            body.source_site, creator_id,
        )
    else:
        creator_id = await pool.fetchval(
            """INSERT INTO creators (username, user_id, service, platform_url, source_site,
                                    first_scraped_at, last_scraped_at)
               VALUES ($1, $1, $2, $3, $4, NOW(), NOW())
               RETURNING id""",
            username, service, body.source_url, body.source_site,
        )

    # Generate a unique post id
    import time as _time
    import random as _random
    post_id = int(_time.time() * 1_000_000 + _random.randint(0, 999)) % (2**53)
    published_at = datetime.now(timezone.utc)

    await pool.execute(
        """INSERT INTO posts (id, creator_id, title, content, published_at,
                             added_at, scraped_at, source_url)
           VALUES ($1, $2, $3, $4, $5, NOW(), NOW(), $6)""",
        post_id, creator_id, body.title, body.content, published_at, body.source_url,
    )

    # Insert media files
    media_ids = []
    for i, mf in enumerate(body.media_files, 1):
        download_url = mf.download_url or f"local://{mf.local_path}"
        page_order = mf.page_order if mf.page_order is not None else i
        file_size = None
        if os.path.isfile(mf.local_path):
            file_size = os.path.getsize(mf.local_path)

        mid = await pool.fetchval(
            """INSERT INTO media_files
                   (post_id, creator_id, download_url, original_name,
                    local_path, file_type, file_size, attachment_type,
                    downloaded_at, page_order)
               VALUES ($1, $2, $3, $4, $5, $6, $7, 'file', NOW(), $8)
               ON CONFLICT (download_url) DO UPDATE SET
                   local_path = EXCLUDED.local_path,
                   page_order = EXCLUDED.page_order,
                   file_size = EXCLUDED.file_size
               RETURNING id""",
            post_id, creator_id, download_url, mf.original_name,
            mf.local_path, mf.file_type, file_size, page_order,
        )
        media_ids.append(mid)

    _cache_invalidate_feeds()

    return {
        "post_id": str(post_id),
        "creator_id": creator_id,
        "media_ids": media_ids,
    }


@app.get(
    "/api/galleries",
    summary="List galleries / comics",
    description="Return paginated posts that contain 2 or more image media files (galleries/comics), with optional service filter.",
)
async def list_galleries(
    page: int = Query(1, ge=1, description="Page number (1-based)"),
    per_page: int = Query(50, ge=1, le=200, description="Results per page"),
    service: Optional[str] = Query(None, description="Filter by source service (e.g. onlyfans, fansly)"),
):
    """List posts that have 2+ image media files (galleries/comics)."""
    pool = await get_db()

    # Build WHERE clause for service filter
    where_clause = ""
    params = []
    param_idx = 1
    if service:
        where_clause = f"WHERE (c.source_site = ${param_idx} OR c.service = ${param_idx})"
        params = [service]
        param_idx += 1

    offset = (page - 1) * per_page

    # Paginate qualifying posts FIRST, then compute image_count/first_media_id
    # for just the page. The old query JOINed all 458K image rows × posts,
    # GROUP/HAVING/COUNT(*) OVER() then LIMITed — materializing the full
    # cross-product (measured 9.1 s, R2-04 #1). `EXISTS … OFFSET 1` short
    # circuits the ">=2 images" test at the 2nd row instead of counting all.
    has_2_images = ("EXISTS (SELECT 1 FROM media_files m "
                    "WHERE m.post_id = p.id AND m.file_type='image' "
                    "OFFSET 1 LIMIT 1)")
    params.extend([per_page, offset])
    lim, off = param_idx, param_idx + 1
    sql = f"""
        WITH page AS (
            SELECT p.id AS post_id, p.title, p.content, p.published_at,
                   p.source_url, c.id AS creator_id, c.username,
                   c.service, c.source_site
            FROM posts p
            JOIN creators c ON c.id = p.creator_id
            {where_clause}{' AND ' if where_clause else 'WHERE '}{has_2_images}
            ORDER BY p.published_at DESC NULLS LAST
            LIMIT ${lim} OFFSET ${off}
        ),
        cnt AS (
            SELECT count(*) AS total
            FROM posts p
            JOIN creators c ON c.id = p.creator_id
            {where_clause}{' AND ' if where_clause else 'WHERE '}{has_2_images}
        )
        SELECT pg.*,
               (SELECT count(*) FROM media_files m
                  WHERE m.post_id = pg.post_id AND m.file_type='image')
                  AS image_count,
               (SELECT min(m.id) FROM media_files m
                  WHERE m.post_id = pg.post_id AND m.file_type='image')
                  AS first_media_id,
               (SELECT total FROM cnt) AS total_count
        FROM page pg
        ORDER BY pg.published_at DESC NULLS LAST
    """
    rows = await pool.fetch(sql, *params)

    total = rows[0]["total_count"] if rows else 0

    galleries = []
    for r in rows:
        galleries.append({
            "post_id": str(r["post_id"]),
            "title": r["title"],
            "content": r["content"],
            "published_at": r["published_at"].isoformat() if r["published_at"] else None,
            "source_url": r["source_url"],
            "creator_id": r["creator_id"],
            "username": r["username"],
            "service": r["service"],
            "source_site": r["source_site"],
            "image_count": r["image_count"],
            "first_media_id": r["first_media_id"],
        })

    return {
        "galleries": galleries,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page if total > 0 else 0,
    }


# === Ingestion API ===

@app.get(
    "/api/ingest/status",
    summary="List all ingest tasks",
    description="Return all ingest tasks (newest first) with status, progress, and elapsed time.",
)
async def ingest_list():
    """List all ingest tasks, newest first."""
    tasks = sorted(_ingest_tasks.values(), key=lambda t: t.queued_at, reverse=True)
    result = []
    for t in tasks:
        d = asdict(t)
        d["status"] = t.status.value
        elapsed = None
        if t.started_at:
            end = t.completed_at or time.time()
            elapsed = round(end - t.started_at, 1)
        d["elapsed_seconds"] = elapsed
        d["progress_pct"] = (
            round((t.files_downloaded + t.files_failed) / t.total_files * 100, 1)
            if t.total_files > 0 else 0.0
        )
        result.append(d)
    return result


@app.get(
    "/api/ingest/{task_id}/status",
    summary="Get ingest task status",
    description="Return status, progress percentage, and elapsed time for a single ingest task. 404 if task_id unknown.",
)
async def ingest_status(task_id: str):
    """Get status of a single ingest task."""
    task = _ingest_tasks.get(task_id)
    if not task:
        raise HTTPException(404, "Task not found")
    d = asdict(task)
    d["status"] = task.status.value
    elapsed = None
    if task.started_at:
        end = task.completed_at or time.time()
        elapsed = round(end - task.started_at, 1)
    d["elapsed_seconds"] = elapsed
    d["progress_pct"] = (
        round((task.files_downloaded + task.files_failed) / task.total_files * 100, 1)
        if task.total_files > 0 else 0.0
    )
    return d


class IngestBody(BaseModel):
    url: str = Field(..., min_length=1, description="Full creator URL, e.g. https://coomer.su/onlyfans/user/{user_id}")


@app.post(
    "/api/ingest",
    summary="Queue creator for download",
    description="Queue a coomer.su / kemono.su creator URL for background downloading. Returns a task_id to poll for progress.",
    dependencies=[Depends(require_trusted_client)],
)
async def ingest_creator(body: IngestBody):
    """Queue a creator URL for downloading."""
    url = body.url.strip()
    if not url:
        raise HTTPException(400, "Missing 'url' field")

    # Validate domain
    parsed = urlparse(url)
    domain = parsed.netloc.lower()
    if not any(d in domain for d in SUPPORTED_DOMAINS):
        raise HTTPException(400, f"Unsupported domain: {domain}. Supported: {SUPPORTED_DOMAINS}")

    # Validate path format
    path_parts = [p for p in parsed.path.strip("/").split("/") if p]
    if len(path_parts) < 3 or path_parts[1] != "user":
        raise HTTPException(400, "Invalid URL format. Expected: https://coomer.su/{service}/user/{user_id}")

    # Duplicate detection: if same URL is already queued/downloading, return existing task
    for existing in _ingest_tasks.values():
        if existing.creator_url == url and existing.status in (
            IngestStatus.QUEUED, IngestStatus.DOWNLOADING
        ):
            return {"task_id": existing.task_id, "status": existing.status.value, "duplicate": True}

    task = IngestTask(
        task_id=str(uuid.uuid4()),
        creator_url=url,
        queued_at=time.time(),
    )
    _ingest_tasks[task.task_id] = task
    await _ingest_queue.put(task)

    return {"task_id": task.task_id, "status": task.status.value}


# Service worker must be served from the origin root with
# `Service-Worker-Allowed: /` so its fetch-intercept scope covers /api/media/*
# (thumb/HLS cache-first) and the SPA — not just /static/. Without this the SW
# is scoped to /static/ and can't accelerate repeat media browsing.
@app.get(
    "/sw.js",
    summary="Serve service worker",
    description="Serve sw.js from the origin root with Service-Worker-Allowed: / so its scope covers all API and SPA routes.",
)
async def service_worker():
    sw = Path(__file__).parent / "static" / "sw.js"
    if not sw.is_file():
        raise HTTPException(404, "sw.js not found")
    return FileResponse(
        sw,
        media_type="application/javascript",
        headers={
            "Service-Worker-Allowed": "/",
            "Cache-Control": "no-cache",
        },
    )


# Serve frontend static assets with precompressed (.br/.gz) negotiation and
# long-lived immutable caching for content-hashed bundles. Replaces the vanilla
# StaticFiles mount, which ignored the prebuilt .br/.gz siblings (GZipMiddleware
# re-gzipped on the fly and never served Brotli) and set no Cache-Control.
_STATIC_DIR = (Path(__file__).parent / "static").resolve()


def _is_immutable_asset(rel_path: str) -> bool:
    """True for content-hashed bundles (app.<hash>.js/css), vendored libs, and
    the self-hosted fonts — their content is stable for a given URL, so they
    can cache forever.

    Fonts (static/fonts/*.woff2|woff) have stable filenames whose bytes never
    change without a deploy that also bumps the referencing CSS, so serving them
    `immutable, max-age=31536000` (instead of the old `max-age=3600`) is safe
    and removes a revalidation round-trip on every repeat visit (perf2/transport
    item 3)."""
    if rel_path.startswith("vendor/"):
        return True
    # Everything the esbuild build emits under static/dist/ is content-hashed
    # (entry bundles and code-split chunks alike). Keying on the directory
    # rather than on a hash format keeps this correct regardless of the
    # bundler's hash alphabet — esbuild uses base32, the old pipeline hex.
    if rel_path.startswith("dist/"):
        return True
    name = rel_path.rsplit("/", 1)[-1]
    if name.endswith((".woff2", ".woff")):
        return True
    parts = name.split(".")
    if len(parts) >= 3 and parts[-1] in ("js", "css"):
        h = parts[-2]
        return len(h) >= 8 and all(c in "0123456789abcdef" for c in h)
    return False


@app.get("/static/{path:path}", name="static")
async def serve_static(path: str, request: Request):
    full = (_STATIC_DIR / path).resolve()
    # Containment check — block path traversal out of the static dir.
    if not str(full).startswith(str(_STATIC_DIR) + os.sep) or not full.is_file():
        raise HTTPException(404, "Not found")

    media_type = mimetypes.guess_type(full.name)[0] or "application/octet-stream"
    if full.suffix == ".mjs":
        media_type = "text/javascript"  # required for ESM `import()` in browsers

    cache_control = ("public, max-age=31536000, immutable"
                     if _is_immutable_asset(path) else "public, max-age=3600")
    headers = {"Cache-Control": cache_control, "Vary": "Accept-Encoding"}

    # Serve the prebuilt Brotli (best) or gzip sibling when the client accepts it.
    # Setting Content-Encoding makes SmartCompressionMiddleware skip the response
    # (it never recompresses anything already carrying a Content-Encoding), so no
    # double compression. Brotli for JS/CSS is ~16% smaller than gzip, ~78% vs raw.
    accepted = {t.strip().split(";")[0] for t in
                request.headers.get("accept-encoding", "").split(",")}
    for enc, ext in (("br", ".br"), ("gzip", ".gz")):
        if enc in accepted:
            pre = full.with_name(full.name + ext)
            if pre.is_file():
                return FileResponse(pre, media_type=media_type,
                                    headers={**headers, "Content-Encoding": enc})

    return FileResponse(full, media_type=media_type, headers=headers)


@app.get(
    "/api/similar/{media_id}",
    summary="Find similar media (back-compat)",
    description="Find content similar to a given media item. Back-compat endpoint — resolves media_id to its post then delegates to the post-level vector fusion query.",
)
async def similar_media(
    media_id: int,
    limit: int = Query(12, ge=1, le=48, description="Max results to return"),
    content_type: Optional[str] = Query(None, pattern="^(video|image)$", description="Filter results by media type"),
    exclude_creator: bool = Query(False, description="Exclude results from the same creator"),
):
    """Find content similar to a given media item.

    Back-compat endpoint. Internally upgraded post-migration 009 to use
    post-level vectors via _post_neighbors() — same response shape, better
    neighbors (esp. for video/comic). Looks up the post for media_id, then
    delegates to the canonical post-level fusion query.
    """
    pool = await get_db()

    # media_files.id is int32; a stray post-id (bigint) here would make asyncpg
    # raise DataError -> 500. Out-of-range simply means "no such media".
    if media_id > 2**31 - 1 or media_id < -(2**31):
        return JSONResponse(
            content={"source_id": media_id, "items": []},
            headers={"Cache-Control": "public, max-age=60"},
        )

    async with pool.acquire() as conn:
        seed_post_row = await conn.fetchrow(
            "SELECT post_id FROM media_files WHERE id = $1",
            media_id,
        )
        if not seed_post_row:
            return JSONResponse(
                content={"source_id": media_id, "items": []},
                headers={"Cache-Control": "public, max-age=60"},
            )

        # Map the legacy content_type filter onto the new post_kind axis.
        types_set: Optional[set[str]] = None
        if content_type == "image":
            types_set = {"image", "comic"}
        elif content_type == "video":
            types_set = {"video"}

        items = await _post_neighbors(
            conn,
            seed_post_id=seed_post_row["post_id"],
            visual_weight=0.7,  # legacy callers don't tune fusion
            limit=limit,
            exclude_creator=exclude_creator,
            content_types=types_set,
        )

    if items is None:
        return JSONResponse(
            content={"source_id": media_id, "items": []},
            headers={"Cache-Control": "public, max-age=60"},
        )

    # Reshape to the legacy schema. Adds post_kind / media_count as additive
    # fields so type-aware rail rendering (Phase 4a) doesn't need a second call.
    legacy = []
    for it in items:
        legacy.append({
            "media_id": it["thumbnail_media_id"],
            "post_id": it["post_id"],
            "file_type": (
                "video" if it["post_kind"] == "video"
                else "image"  # comic posts surface as image cards in the old UI
            ),
            "creator_id": it["creator_id"],
            "username": it["username"],
            "title": it["title"],
            "similarity": it["score"],
            # New post-level metadata for the type-aware similar rail.
            "post_kind": it["post_kind"],
            "media_count": it["media_count"],
        })

    return JSONResponse(
        content={"source_id": media_id, "items": legacy},
        headers={"Cache-Control": "public, max-age=60"},
    )


# ---------------------------------------------------------------------------
# Similarity v2 (post-level fusion + explore + cluster)
# ---------------------------------------------------------------------------

# Hard-coded model versions per pre-mortem R3: bumping these is a deliberate
# cross-cutting change, not a silent config drift.
MODEL_VERSION_VISUAL = "ViT-L/14"
MODEL_VERSION_TAGS = "bge-m3"

# Embedding-type constants mirror batch_embed.py.
_VISUAL_POST_TYPES = ("clip_image_post", "clip_video_post", "clip_comic_post")
# Comic dual-axis: clip_comic_post = mean of all pages (story-feel),
# clip_comic_cover = first-page only (art-style). When the seed is a comic
# the user can toggle which axis drives the query AND which type the comic
# candidate set uses.
_VISUAL_POST_TYPES_COVER = ("clip_image_post", "clip_video_post", "clip_comic_cover")
_TAG_TYPE = "bge_tags"

# Feature flag — when 0, the new explore/cluster endpoints 404. The legacy
# /api/similar/{media_id} always runs (it just gets a smarter implementation).
ENABLE_SIMILARITY_V2 = os.environ.get("ENABLE_SIMILARITY_V2", "1") == "1"

# Slider clamp for the visual/tag fusion weight (R6 mitigation).
_FUSION_WEIGHT_MIN = 0.2
_FUSION_WEIGHT_MAX = 0.95


def _clamp_visual_weight(w: float) -> float:
    """Clamp the slider value to [_FUSION_WEIGHT_MIN, _FUSION_WEIGHT_MAX]."""
    if w is None:
        return 0.7
    return max(_FUSION_WEIGHT_MIN, min(_FUSION_WEIGHT_MAX, float(w)))


def _parse_vec(s: str) -> list[float]:
    """asyncpg returns vector columns as the pgvector text repr '[v1,v2,...]'."""
    return [float(x) for x in s.strip("[]").split(",")] if isinstance(s, str) else list(s)


def _vec_mean(vecs: list[list[float]]) -> Optional[list[float]]:
    """Plain mean-pool of vectors. Returns None for empty input."""
    if not vecs:
        return None
    n = len(vecs)
    dim = len(vecs[0])
    summed = [0.0] * dim
    for v in vecs:
        for i in range(dim):
            summed[i] += v[i]
    return [x / n for x in summed]


def _l2_normalize(v: list[float]) -> list[float]:
    norm = sum(x * x for x in v) ** 0.5
    return [x / norm for x in v] if norm > 0 else v


def _format_vec(v: list[float]) -> str:
    return "[" + ",".join(repr(x) for x in v) + "]"


def _avg_normalize(vecs: list[list[float]]) -> Optional[str]:
    """Mean-pool L2-normalized vectors; renormalize; return pgvector text repr."""
    mean = _vec_mean(vecs)
    if mean is None:
        return None
    return _format_vec(_l2_normalize(mean))


def _triplet_query(pos_vecs: list[list[float]], neg_vecs: list[list[float]]) -> Optional[str]:
    """Qdrant Discover triplet: query = 2*avg(pos) - avg(neg), then L2-normalize.
    With no negs, falls back to avg(pos)."""
    pos = _vec_mean(pos_vecs)
    if pos is None:
        return None
    neg = _vec_mean(neg_vecs)
    if neg is None:
        return _format_vec(_l2_normalize(pos))
    combined = [2 * p - n for p, n in zip(pos, neg)]
    return _format_vec(_l2_normalize(combined))


async def _post_neighbors(
    conn,
    seed_post_id=None,  # int or list[int] (Qdrant average_vector pattern)
    visual_weight: float = 0.7,
    limit: int = 24,
    candidate_pool: int = 200,
    exclude_creator: bool = False,
    content_types: Optional[set[str]] = None,
    exclude_post_ids: Optional[set[int]] = None,
    neg_post_ids: Optional[list[int]] = None,
    comic_axis: str = "content",  # "content" (mean of pages) | "cover" (first page only)
):
    """Run the post-level fusion query and return [{post_id, score, ...}] dicts.

    seed_post_id can be a single int OR a list[int]. With multiple seeds the
    visual + tag query vectors become the L2-normalized mean of each seed set
    (Qdrant Recommend API `average_vector` strategy).

    neg_post_ids (Qdrant Discover triplet): when provided, the query vector
    becomes 2*avg(pos) - avg(neg), pushing away from the negative centroid.
    All seeds (pos and neg) are excluded from results.

    comic_axis: "content" uses clip_comic_post (mean of pages → story-feel);
    "cover" uses clip_comic_cover (first-page only → art-style). The candidate
    set's comic type also follows this axis so the comparison is consistent.
    """
    seed_ids = [seed_post_id] if isinstance(seed_post_id, int) else list(seed_post_id or [])
    neg_ids = list(neg_post_ids or [])
    if not seed_ids:
        return None

    # Defensive whitelist — comic_candidate_type is interpolated into SQL below.
    if comic_axis not in ("content", "cover"):
        comic_axis = "content"
    visual_types = _VISUAL_POST_TYPES_COVER if comic_axis == "cover" else _VISUAL_POST_TYPES
    comic_candidate_type = "clip_comic_cover" if comic_axis == "cover" else "clip_comic_post"
    assert comic_candidate_type in ("clip_comic_post", "clip_comic_cover"), "comic_candidate_type whitelist"

    # Pull visual embeddings for all positive seeds.
    visual_rows = await conn.fetch(
        """SELECT post_id, embedding, creator_id
             FROM content_embeddings
            WHERE post_id = ANY($1::bigint[])
              AND aggregation = 'post'
              AND embedding_type = ANY($2::text[])""",
        seed_ids,
        list(visual_types),
    )
    if not visual_rows:
        return None  # none of the seeds have visual vectors

    pos_v_vecs = [_parse_vec(r["embedding"]) for r in visual_rows]
    seed_creator_id = visual_rows[0]["creator_id"]  # only used when exclude_creator

    # Negatives — visual + tag — fetched with one query each.
    neg_v_vecs: list[list[float]] = []
    neg_t_vecs: list[list[float]] = []
    if neg_ids:
        neg_visual_rows = await conn.fetch(
            """SELECT embedding FROM content_embeddings
                WHERE post_id = ANY($1::bigint[])
                  AND aggregation = 'post'
                  AND embedding_type = ANY($2::text[])""",
            neg_ids, list(visual_types),
        )
        neg_v_vecs = [_parse_vec(r["embedding"]) for r in neg_visual_rows]

    seed_v_emb = _triplet_query(pos_v_vecs, neg_v_vecs)

    tag_rows = await conn.fetch(
        """SELECT embedding
             FROM content_embeddings
            WHERE post_id = ANY($1::bigint[])
              AND aggregation = 'post'
              AND embedding_type = $2""",
        seed_ids,
        _TAG_TYPE,
    )
    pos_t_vecs = [_parse_vec(r["embedding"]) for r in tag_rows] if tag_rows else []

    if neg_ids:
        neg_tag_rows = await conn.fetch(
            """SELECT embedding FROM content_embeddings
                WHERE post_id = ANY($1::bigint[])
                  AND aggregation = 'post'
                  AND embedding_type = $2""",
            neg_ids, _TAG_TYPE,
        )
        neg_t_vecs = [_parse_vec(r["embedding"]) for r in neg_tag_rows]

    seed_t_emb = _triplet_query(pos_t_vecs, neg_t_vecs) if pos_t_vecs else None
    # Exclude both positive and negative seeds from candidate set.
    excluded_ids = list({*seed_ids, *neg_ids})

    # Per-type top-N via literal type filter. The monolithic DiskANN partial
    # index has catastrophic cross-manifold recall (a comic seed never reaches
    # the comic cluster through the graph), so we bypass it: each per-type
    # subquery uses idx_ce_post_type_post_agg (btree on (post_id, embedding_type)
    # WHERE aggregation='post') as a bitmap scan + top-N heapsort, which is
    # exact KNN over ~5-15k rows per type at ~50ms each. Total ~150ms.
    per_type_n = max(candidate_pool // 2, 80)

    # The comic candidate type swaps based on comic_axis.
    sql = f"""
        WITH visual_top AS (
            (SELECT ce.post_id, ce.creator_id,
                    1 - (ce.embedding <=> $1::vector) AS s_visual
               FROM content_embeddings ce
              WHERE ce.aggregation = 'post'
                AND ce.embedding_type = '{comic_candidate_type}'
                AND ce.post_id <> ALL($2::bigint[])
              ORDER BY ce.embedding <=> $1::vector
              LIMIT $3)
            UNION ALL
            (SELECT ce.post_id, ce.creator_id,
                    1 - (ce.embedding <=> $1::vector) AS s_visual
               FROM content_embeddings ce
              WHERE ce.aggregation = 'post'
                AND ce.embedding_type = 'clip_image_post'
                AND ce.post_id <> ALL($2::bigint[])
              ORDER BY ce.embedding <=> $1::vector
              LIMIT $3)
            UNION ALL
            (SELECT ce.post_id, ce.creator_id,
                    1 - (ce.embedding <=> $1::vector) AS s_visual
               FROM content_embeddings ce
              WHERE ce.aggregation = 'post'
                AND ce.embedding_type = 'clip_video_post'
                AND ce.post_id <> ALL($2::bigint[])
              ORDER BY ce.embedding <=> $1::vector
              LIMIT $3)
        ),
        tag_dist AS (
            SELECT vt.post_id, vt.creator_id, vt.s_visual,
                   CASE WHEN $4::vector IS NULL THEN NULL
                        ELSE 1 - (ce.embedding <=> $4::vector)
                   END AS s_tag
              FROM visual_top vt
              LEFT JOIN content_embeddings ce
                ON ce.post_id = vt.post_id
               AND ce.embedding_type = $5
               AND ce.aggregation  = 'post'
        )
        SELECT td.post_id, td.creator_id, td.s_visual, td.s_tag,
               $6::real * td.s_visual
                 + (1 - $6::real) * COALESCE(td.s_tag, td.s_visual) AS score
          FROM tag_dist td
         ORDER BY score DESC
    """

    rows = await conn.fetch(
        sql,
        seed_v_emb,
        excluded_ids,
        per_type_n,
        seed_t_emb,
        _TAG_TYPE,
        visual_weight,
    )

    # Filter and decorate. We do this in Python rather than SQL because the
    # post_kind rollup (image/video/comic) is a simple aggregate but cleaner
    # to compute alongside username/title in one trip.
    keep_post_ids = []
    raw = []
    for r in rows:
        if exclude_creator and r["creator_id"] == seed_creator_id:
            continue
        if exclude_post_ids and r["post_id"] in exclude_post_ids:
            continue
        raw.append(r)
        keep_post_ids.append(r["post_id"])
        if len(raw) >= max(limit * 4, candidate_pool):
            break  # plenty of headroom; we filter further below.

    if not raw:
        return []

    # Per-post media aggregate: a single PK lookup into the materialized
    # post_media_rollup instead of a LEFT JOIN LATERAL over the 2.77M-row
    # media_files table (same pattern as /api/atlas and /api/evoc/parcoords).
    # Output column names/shape match the old lateral so callers are unaffected.
    enrich_rows = await conn.fetch(
        """SELECT p.id AS post_id, p.title, p.creator_id, p.published_at,
                  c.username, c.source_site,
                  r.media_count, r.has_video, r.has_image, r.thumbnail_media_id
             FROM posts p
             LEFT JOIN creators c ON c.id = p.creator_id
             LEFT JOIN post_media_rollup r ON r.post_id = p.id
            WHERE p.id = ANY($1::bigint[])""",
        keep_post_ids,
    )
    enrich_by_id = {r["post_id"]: r for r in enrich_rows}

    items = []
    for r in raw:
        e = enrich_by_id.get(r["post_id"])
        if not e:
            continue
        # Posts whose media was deleted (or whose rollup row is missing) have
        # no thumbnail_media_id: they render as dead title-only cards linking
        # to /video/None. Embeddings for them may still exist — skip here so
        # every consumer (similar rail, explore, end-of-video recs) is clean.
        if e["thumbnail_media_id"] is None:
            continue

        # Derive post_kind: comic if multi-media + has_image, else video, else image.
        if e["media_count"] and e["media_count"] > 1 and e["has_image"]:
            post_kind = "comic"
        elif e["has_video"]:
            post_kind = "video"
        else:
            post_kind = "image"

        if content_types is not None and post_kind not in content_types:
            continue

        items.append({
            "post_id": str(r["post_id"]),
            "media_count": int(e["media_count"] or 0),
            "post_kind": post_kind,
            "thumbnail_media_id": e["thumbnail_media_id"],
            "creator_id": r["creator_id"],
            "username": e["username"],
            "title": e["title"],
            "source_site": e["source_site"],
            "published_at": e["published_at"],
            "score": round(float(r["score"]), 4),
            "score_visual": round(float(r["s_visual"]), 4),
            "score_tag": round(float(r["s_tag"]), 4) if r["s_tag"] is not None else None,
        })

        # Note: don't slice to `limit` here — dedup may collapse items below.

    # ── Same-creator near-dup clustering (ehentai/iqdb pattern) ─────────
    # Items already ordered by score DESC. Group by (creator_id, 7-day bucket).
    # Keep the top-scoring item per cluster; attach the rest as dedup_siblings.
    # Avoids 'one creator's set floods the grid' failure mode.
    SEVEN_DAYS = 7 * 86400
    clusters: dict[tuple, list[dict]] = {}
    cluster_order: list[tuple] = []
    for it in items:
        cid = it["creator_id"]
        pa = it["published_at"]
        bucket = int(pa.timestamp() // SEVEN_DAYS) if pa is not None else None
        key = (cid, bucket) if cid is not None and bucket is not None else ("__solo__", id(it))
        if key not in clusters:
            clusters[key] = []
            cluster_order.append(key)
        clusters[key].append(it)

    deduped: list[dict] = []
    for key in cluster_order:
        group = clusters[key]
        rep = group[0]  # highest-scoring (input was sorted)
        siblings = group[1:]
        if siblings:
            rep["dedup_count"] = len(siblings)
            rep["dedup_siblings"] = [
                {"post_id": s["post_id"],
                 "thumbnail_media_id": s["thumbnail_media_id"],
                 "title": s["title"],
                 "score": s["score"]}
                for s in siblings
            ]
        # Drop transient fields before returning.
        rep.pop("published_at", None)
        deduped.append(rep)
        if len(deduped) >= limit:
            break

    return deduped


@app.get(
    "/api/similar/post/{post_id}",
    summary="Similar posts by visual+tag fusion",
    description=(
        "Return the nearest-neighbor posts to the given post using a weighted "
        "fusion of visual (CLIP) and tag embeddings. The canonical per-post "
        "similarity endpoint used by the viewer rail."
    ),
)
async def similar_post(
    post_id: int,
    limit: int = Query(24, ge=1, le=100, description="Number of results to return"),
    visual_weight: float = Query(0.7, ge=0.0, le=1.0, description="Weight for visual embedding vs tag embedding (0=tags only, 1=visual only)"),
    content_types: Optional[str] = Query(None, description="Comma-separated content type filter: image,video,comic"),
    exclude_creator: bool = Query(False, description="Exclude posts by the same creator as the seed"),
):
    """Post-level neighbors using visual+tag fusion. The canonical /api/similar/{post_id}."""
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    weight = _clamp_visual_weight(visual_weight)
    types_set: Optional[set[str]] = None
    if content_types:
        types_set = {t.strip() for t in content_types.split(",") if t.strip()}

    sim_ck = f"simpost:{post_id}:{limit}:{round(weight, 3)}:{content_types}:{exclude_creator}"
    _sc = _cache.get(sim_ck)
    if _sc and time.time() - _sc[0] < 300:
        return JSONResponse(content=_sc[1],
                            headers={"Cache-Control": "public, max-age=60", "X-Cache": "HIT"})

    pool = await get_db()
    async with pool.acquire() as conn:
        items = await _post_neighbors(
            conn,
            seed_post_id=post_id,
            visual_weight=weight,
            limit=limit,
            exclude_creator=exclude_creator,
            content_types=types_set,
        )

    if items is None:
        return JSONResponse(
            content={
                "seed_post_id": str(post_id),
                "fusion": {"visual_weight": weight, "tag_weight": 1 - weight, "fusion_used": "none"},
                "items": [],
                "reason": "no_visual_embedding_for_seed",
            },
            headers={"Cache-Control": "public, max-age=60"},
        )

    fusion_used = "visual+tags"
    if items and all(it.get("score_tag") is None for it in items):
        fusion_used = "visual_only"

    payload = {
        "seed_post_id": str(post_id),
        "fusion": {
            "visual_weight": weight,
            "tag_weight": round(1 - weight, 4),
            "fusion_used": fusion_used,
        },
        "items": items,
    }
    _cache[sim_ck] = (time.time(), payload)
    return JSONResponse(
        content=payload,
        headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"},
    )


def _mmr_rerank(items: list[dict], lambda_: float, k: int) -> list[dict]:
    """Maximal Marginal Relevance over already-scored candidates.

    items: dicts with at least {post_id, score (relevance to seed)}.
    We have NO candidate-to-candidate similarity here, so we use a cheap proxy:
    diversify on (creator_id, post_kind, source_site) tuple. Items sharing all
    three with an already-selected one take a relevance penalty.
    """
    if not items:
        return []
    selected: list[dict] = []
    pool = list(items)
    while pool and len(selected) < k:
        best_i, best_score = -1, -1e9
        for i, c in enumerate(pool):
            relevance = c["score"]
            if not selected:
                penalty = 0.0
            else:
                # Penalty: max overlap with anything already picked (0 / 0.5 / 1).
                penalty = max(
                    (
                        (0.5 if s.get("creator_id") == c.get("creator_id") else 0)
                        + (0.25 if s.get("post_kind") == c.get("post_kind") else 0)
                        + (0.25 if s.get("source_site") == c.get("source_site") else 0)
                    )
                    for s in selected
                )
            mmr = lambda_ * relevance - (1 - lambda_) * penalty
            if mmr > best_score:
                best_score = mmr
                best_i = i
        selected.append(pool.pop(best_i))
    return selected


@app.get(
    "/api/explore",
    summary="Lateral-pivot explore",
    description=(
        "Pivot from a seed post or set of posts using similar/diverse/bridge strategies. "
        "Supports multi-seed centroid, negative-set repulsion, and MMR diversification. "
        "With no seed returns a curated starter grid sampled across cluster centroids."
    ),
)
async def explore(
    seed_post_id: Optional[int] = Query(None),
    seed_post_ids: Optional[str] = Query(None, description="Comma-separated post ids for multi-seed centroid (Qdrant average_vector)"),
    neg_post_ids: Optional[str] = Query(None, description="Comma-separated post ids to push away from (Qdrant Discover triplet)"),
    seed_post_id_b: Optional[int] = Query(None, description="Second seed for strategy=bridge"),
    strategy: str = Query("similar", pattern="^(similar|diverse|bridge)$", description="similar=nearest, diverse=MMR-reranked, bridge=intersection of two seeds"),
    limit: int = Query(24, ge=1, le=100, description="Number of results to return"),
    mmr_lambda: float = Query(0.6, ge=0.0, le=1.0, description="MMR diversity tradeoff (1=pure relevance, 0=max diversity); only used with strategy=diverse"),
    visual_weight: float = Query(0.7, ge=0.0, le=1.0, description="Weight for visual embedding vs tag embedding"),
    exclude_creator: bool = Query(False, description="Exclude posts by the same creator as any seed"),
    comic_axis: str = Query("content", pattern="^(content|cover)$",
        description="content=story-feel (clip_comic_post), cover=art-style (clip_comic_cover)"),
):
    """Lateral-pivot explore endpoint. Powers the /explore UI."""
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    weight = _clamp_visual_weight(visual_weight)
    pool = await get_db()

    # Resolve seed set: explicit seed_post_ids wins over single seed.
    multi_seeds: Optional[list[int]] = None
    if seed_post_ids:
        try:
            multi_seeds = [int(x) for x in seed_post_ids.split(",") if x.strip()]
        except ValueError:
            raise HTTPException(400, "seed_post_ids must be comma-separated integers")
    seed_set = multi_seeds if multi_seeds else ([seed_post_id] if seed_post_id is not None else None)

    neg_set: Optional[list[int]] = None
    if neg_post_ids:
        try:
            neg_set = [int(x) for x in neg_post_ids.split(",") if x.strip()]
        except ValueError:
            raise HTTPException(400, "neg_post_ids must be comma-separated integers")

    # Cache deterministic seeded results (slider tweaks re-hit the same params).
    # The no-seed starter grid below uses random() and must NOT be cached.
    explore_ck = None
    if seed_set is not None:
        explore_ck = (f"explore:{','.join(map(str, seed_set))}:{seed_post_id_b}:"
                      f"{','.join(map(str, neg_set)) if neg_set else ''}:{strategy}:"
                      f"{limit}:{round(weight, 3)}:{round(mmr_lambda, 3)}:"
                      f"{exclude_creator}:{comic_axis}")
        _ec = _cache.get(explore_ck)
        if _ec and time.time() - _ec[0] < 300:
            return JSONResponse(content=_ec[1],
                                headers={"Cache-Control": "public, max-age=60", "X-Cache": "HIT"})

    # No seed: return a curated starter grid sampled across cluster_ids.
    if seed_set is None:
        async with pool.acquire() as conn:
            rows = await conn.fetch(
                """SELECT DISTINCT ON (cluster_id) post_id, cluster_id
                     FROM post_clusters
                    WHERE cluster_id IS NOT NULL AND cluster_id <> -1
                    ORDER BY cluster_id, random()
                    LIMIT $1""",
                limit,
            )
            if not rows:
                # No clusters yet (UMAP hasn't run). Just return random recent posts.
                rows = await conn.fetch(
                    """SELECT post_id FROM content_embeddings
                       WHERE aggregation='post'
                       GROUP BY post_id
                       ORDER BY random()
                       LIMIT $1""",
                    limit,
                )
            return JSONResponse(
                content={
                    "strategy": "discover",
                    "items": [{"post_id": str(r["post_id"])} for r in rows],
                },
                headers={"Cache-Control": "public, max-age=60"},
            )

    async with pool.acquire() as conn:
        if strategy == "bridge" and seed_post_id_b is not None:
            # Get neighbors of A and B independently, then re-rank by joint score.
            # Run the two neighbor queries concurrently (matching search_posts'
            # asyncio.gather fan-out). asyncpg forbids concurrent ops on a single
            # connection, so B runs on a second pooled connection while A uses the
            # one already held. gather preserves argument order → items_a is A.
            async def _neighbors_on_new_conn(seed):
                async with pool.acquire() as conn_b:
                    return await _post_neighbors(conn_b, seed, weight, limit * 4)

            items_a, items_b = await asyncio.gather(
                _post_neighbors(conn, seed_post_id, weight, limit * 4),
                _neighbors_on_new_conn(seed_post_id_b),
            )
            if items_a is None or items_b is None:
                raise HTTPException(404, "One or both seeds lack a visual embedding")
            sa = {it["post_id"]: it["score"] for it in items_a}
            sb = {it["post_id"]: it["score"] for it in items_b}
            common = set(sa) & set(sb)
            ranked = sorted(
                (
                    {"post_id": pid,
                     "score": sa[pid] * sb[pid] - 0.3 * abs(sa[pid] - sb[pid])}
                    for pid in common
                ),
                key=lambda r: r["score"],
                reverse=True,
            )[:limit]
            payload = {"strategy": "bridge", "seed_a": str(seed_post_id),
                       "seed_b": str(seed_post_id_b), "items": ranked}
            if explore_ck:
                _cache[explore_ck] = (time.time(), payload)
            return JSONResponse(
                content=payload,
                headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"},
            )

        # similar / diverse — pass the seed_set list (single or multi) +
        # optional neg_set for triplet "less like this" pushing +
        # comic_axis selecting cover (art-style) vs content (story-feel).
        candidate_limit = limit if strategy == "similar" else max(limit * 4, 100)
        items = await _post_neighbors(
            conn,
            seed_post_id=seed_set if len(seed_set) > 1 else seed_set[0],
            visual_weight=weight,
            limit=candidate_limit,
            exclude_creator=exclude_creator,
            neg_post_ids=neg_set,
            comic_axis=comic_axis,
        )

        # Quick check: is the (single) seed a comic? Drives UI toggle.
        seed_is_comic = False
        if len(seed_set) == 1:
            row = await conn.fetchrow(
                """SELECT 1 FROM content_embeddings
                    WHERE post_id = $1 AND aggregation = 'post'
                      AND embedding_type = 'clip_comic_post' LIMIT 1""",
                seed_set[0],
            )
            seed_is_comic = row is not None

    if items is None:
        return JSONResponse(
            content={"strategy": strategy, "items": [], "reason": "no_visual_embedding_for_seed"},
            headers={"Cache-Control": "public, max-age=60"},
        )

    if strategy == "diverse":
        items = _mmr_rerank(items, lambda_=mmr_lambda, k=limit)

    payload = {
        "strategy": strategy,
        "seed_post_id": str(seed_set[0]) if len(seed_set) == 1 else None,
        "seed_post_ids": [str(s) for s in seed_set] if len(seed_set) > 1 else None,
        "seed_is_comic": seed_is_comic,
        "comic_axis": comic_axis if seed_is_comic else None,
        "fusion": {"visual_weight": weight, "mmr_lambda": mmr_lambda if strategy == "diverse" else None},
        "items": items[:limit],
    }
    if explore_ck:
        _cache[explore_ck] = (time.time(), payload)
    return JSONResponse(
        content=payload,
        headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"},
    )


@app.get(
    "/api/cluster/{post_id}",
    summary="UMAP cluster neighborhood for a post",
    description=(
        "Return the seed post's UMAP (x, y) coordinates and those of its top-N "
        "nearest neighbors from the post_clusters table. Used by the /explore "
        "cluster-scatter view. Degrades gracefully when the seed has no cached layout."
    ),
)
async def cluster(post_id: int, neighborhood: int = Query(200, ge=10, le=2000, description="Number of neighborhood points to return around the seed")):
    """UMAP cluster view backing for /explore?view=cluster.

    Returns the seed's cached (x, y) plus the (x, y) of the seed's top-N
    neighborhood. If the seed isn't in post_clusters yet (newly embedded since
    the last nightly UMAP run), returns cluster_status='fallback_no_layout'
    so the UI degrades gracefully to the rail.
    """
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    pool = await get_db()
    async with pool.acquire() as conn:
        seed = await conn.fetchrow(
            "SELECT post_id, umap_x, umap_y, cluster_id, neighbors, computed_at, model_version "
            "FROM post_clusters WHERE post_id = $1",
            post_id,
        )
        if not seed:
            return JSONResponse(
                content={
                    "seed_post_id": str(post_id),
                    "cluster_status": "fallback_no_layout",
                    "umap": [],
                    "neighbors": [],
                },
                headers={"Cache-Control": "public, max-age=60"},
            )

        neighbor_ids = list(seed["neighbors"][:neighborhood]) if seed["neighbors"] else []
        rows = await conn.fetch(
            """SELECT pc.post_id, pc.umap_x, pc.umap_y, pc.cluster_id,
                      (array_agg(mf.id ORDER BY mf.page_order NULLS LAST, mf.id))[1] AS thumbnail_media_id
                 FROM post_clusters pc
                 LEFT JOIN media_files mf ON mf.post_id = pc.post_id
                WHERE pc.post_id = ANY($1::bigint[])
                GROUP BY pc.post_id, pc.umap_x, pc.umap_y, pc.cluster_id""",
            neighbor_ids + [post_id],
        )

    return JSONResponse(
        content={
            "seed_post_id": str(post_id),
            "cluster_status": "cached",
            "cluster_id": seed["cluster_id"],
            "computed_at": seed["computed_at"].isoformat(),
            "model_version": seed["model_version"],
            "umap": [
                {
                    "post_id": str(r["post_id"]),
                    "x": float(r["umap_x"]),
                    "y": float(r["umap_y"]),
                    "cluster_id": r["cluster_id"],
                    "thumbnail_media_id": r["thumbnail_media_id"],
                }
                for r in rows
            ],
            "neighbors": [str(n) for n in neighbor_ids],
        },
        headers={"Cache-Control": "public, max-age=60"},
    )


@app.get(
    "/api/atlas",
    summary="Full-corpus embedding atlas",
    description=(
        "Return every clustered post's UMAP coordinates, cluster membership, "
        "post kind, and thumbnail for the /atlas scatter-map. ~30k rows; "
        "response is gzip-compressed and cached in-process for 1 hour."
    ),
)
async def atlas():
    """Full-corpus embedding map for /atlas. Returns one row per clustered post:
    (post_id, x, y, cluster_id, post_kind, thumbnail_media_id).

    ~30k rows × ~50 bytes ≈ 1.5 MB; gzip middleware compresses it ~5×.
    Cached aggressively since this only changes on the nightly UMAP refit:
    in-process for 1h, browser for 5 min.
    """
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    cache_key = "atlas"
    cached = _cache.get(cache_key)
    if cached is not None:
        cached_at, body = cached
        if time.time() - cached_at < 3600:  # 1h in-process; nightly data
            # body is pre-serialized orjson bytes — skip the ~108 ms re-encode
            # of the 5 MB payload that froze the single worker on every HIT.
            return Response(
                content=body, media_type="application/json",
                headers={"Cache-Control": "public, max-age=300", "X-Cache": "HIT"},
            )

    pool = await get_db()
    async with pool.acquire() as conn:
        # Cluster sizes computed in SQL (was a Python loop over ~30k rows).
        size_rows = await conn.fetch(
            "SELECT COALESCE(cluster_id, -1) AS cluster_id, COUNT(*) AS n "
            "FROM post_clusters GROUP BY cluster_id ORDER BY n DESC"
        )
        # The per-post media aggregate (formerly a LEFT JOIN LATERAL over the
        # 2.77M-row media_files table) is now a single PK lookup into the
        # materialized post_media_rollup, which also lets us drop the `posts`
        # hypertable join for creator_id (the rollup is keyed by post_id and the
        # creator_id still comes from posts only when present). See
        # migrations/012_post_media_rollup.sql.
        rows = await conn.fetch(
            """SELECT pc.post_id, pc.umap_x, pc.umap_y, pc.cluster_id,
                      p.creator_id, c.username,
                      r.thumbnail_media_id, r.media_count, r.has_video, r.has_image
                 FROM post_clusters pc
                 JOIN posts p ON p.id = pc.post_id
                 LEFT JOIN creators c ON c.id = p.creator_id
                 LEFT JOIN post_media_rollup r ON r.post_id = pc.post_id"""
        )

    points = []
    for r in rows:
        if r["media_count"] and r["media_count"] > 1 and r["has_image"]:
            kind = "comic"
        elif r["has_video"]:
            kind = "video"
        else:
            kind = "image"
        points.append({
            "post_id": str(r["post_id"]),
            "x": float(r["umap_x"]),
            "y": float(r["umap_y"]),
            "cluster_id": int(r["cluster_id"]) if r["cluster_id"] is not None else -1,
            "post_kind": kind,
            "thumbnail_media_id": r["thumbnail_media_id"],
            "creator_id": r["creator_id"],
            "username": r["username"],
        })

    payload = {
        "n": len(points),
        "clusters": [{"id": int(r["cluster_id"]), "size": int(r["n"])} for r in size_rows],
        "points": points,
    }
    body = orjson.dumps(payload)
    _cache[cache_key] = (time.time(), body)
    return Response(
        content=body, media_type="application/json",
        headers={"Cache-Control": "public, max-age=300", "X-Cache": "MISS"},
    )


# ---------------------------------------------------------------------------
# Binary / typed-array atlas payload
# ---------------------------------------------------------------------------
# The legacy /api/atlas response was the single heaviest payload in the app:
# ~6.77 MB JSON (1.37 MB gzip) for ~39k points. Every row carried 8 fields,
# but the scatter renderer (ss() in the SPA bundle) only needs x/y/cluster_id
# at DRAW time, plus post_id by index on SELECT. The four hover-only fields
# (thumbnail_media_id, post_kind, creator_id, username) are read one point at
# a time and so don't belong in the bulk payload at all.
#
# New scheme — three endpoints:
#   GET /api/atlas/meta   -> small JSON {n, clusters:[{id,size}], post_ids:[...]}
#                            post_ids are strings (max post_id > 2^53, unsafe as
#                            JS Number) and only consumed by index on select.
#   GET /api/atlas/coords -> application/octet-stream, little-endian:
#                              magic   : 4 bytes  b"ATL1"  (read big-endian: 0x41544C31)
#                              n        : uint32 (little-endian)
#                              x        : float32[n]   (raw UMAP x, full precision)
#                              y        : float32[n]   (raw UMAP y)
#                              cluster  : int32[n]      (cluster_id, -1 = noise)
#                            8-byte header keeps every typed array 4-byte aligned
#                            (required by Float32Array/Int32Array views). Total =
#                            8 + 12*n bytes (≈ 463 KB for 39k points).
#                            Normalisation to clip space happens client-side in
#                            a single pass (no 39k-arg Math.min/max spreads).
#   GET /api/atlas/point   -> small JSON for a single post, fetched lazily on
#                            hover/click: {post_kind, thumbnail_media_id,
#                            username, cluster_id}.
#
# float32 gives ~7 significant digits — far beyond what a screen-space scatter
# plot can resolve — so coordinate fidelity is not a concern.

_ATLAS_COORDS_MAGIC = b"ATL1"


def _atlas_kind(media_count, has_image, has_video) -> str:
    """Derive a post's display kind from its media aggregate (shared logic)."""
    if media_count and media_count > 1 and has_image:
        return "comic"
    if has_video:
        return "video"
    return "image"


async def _atlas_fetch_rows(conn):
    """Fetch atlas rows + cluster sizes once; reused by meta/coords builders.

    Ordering of `rows` is the canonical point index used by BOTH the binary
    coords blob and the meta post_ids array, so the two stay aligned.
    """
    size_rows = await conn.fetch(
        "SELECT COALESCE(cluster_id, -1) AS cluster_id, COUNT(*) AS n "
        "FROM post_clusters GROUP BY cluster_id ORDER BY n DESC"
    )
    # INNER JOIN posts so the binary/meta point set matches the legacy
    # /api/atlas exactly (excludes ~hundreds of orphan cluster rows whose
    # post has been deleted) — guarantees every plotted point also has a
    # working /api/atlas/point hover lookup.
    rows = await conn.fetch(
        """SELECT pc.post_id, pc.umap_x, pc.umap_y, pc.cluster_id
             FROM post_clusters pc
             JOIN posts p ON p.id = pc.post_id
            ORDER BY pc.post_id"""
    )
    return size_rows, rows


@app.get(
    "/api/atlas/meta",
    summary="Atlas metadata (counts, clusters, post_ids)",
    description=(
        "Small JSON companion to /api/atlas/coords. Returns the point count, "
        "cluster sizes for the legend, and the per-point post_ids (as strings, "
        "in the same index order as the binary coords blob). Cached 1h."
    ),
)
async def atlas_meta():
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    cache_key = "atlas_meta"
    cached = _cache.get(cache_key)
    if cached is not None:
        cached_at, body = cached
        if time.time() - cached_at < 3600:
            return Response(
                content=body, media_type="application/json",
                headers={"Cache-Control": "public, max-age=300", "X-Cache": "HIT"},
            )

    pool = await get_db()
    async with pool.acquire() as conn:
        size_rows, rows = await _atlas_fetch_rows(conn)

    payload = {
        "n": len(rows),
        "clusters": [{"id": int(r["cluster_id"]), "size": int(r["n"])} for r in size_rows],
        "post_ids": [str(r["post_id"]) for r in rows],
    }
    body = orjson.dumps(payload)
    _cache[cache_key] = (time.time(), body)
    return Response(
        content=body, media_type="application/json",
        headers={"Cache-Control": "public, max-age=300", "X-Cache": "MISS"},
    )


@app.get(
    "/api/atlas/coords",
    summary="Atlas coordinates as a binary typed-array blob",
    description=(
        "application/octet-stream: magic 'ATL1', uint32 n, float32 x[n], "
        "float32 y[n], int32 cluster_id[n] (little-endian). ~12 bytes/point, "
        "an order of magnitude smaller than the JSON /api/atlas payload. "
        "Pair with /api/atlas/meta for post_ids and cluster sizes."
    ),
)
async def atlas_coords():
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    cache_key = "atlas_coords"
    cached = _cache.get(cache_key)
    if cached is not None:
        cached_at, body = cached
        if time.time() - cached_at < 3600:
            return Response(
                content=body, media_type="application/octet-stream",
                headers={"Cache-Control": "public, max-age=300", "X-Cache": "HIT"},
            )

    pool = await get_db()
    async with pool.acquire() as conn:
        _size_rows, rows = await _atlas_fetch_rows(conn)

    n = len(rows)
    xs = bytearray(4 * n)
    ys = bytearray(4 * n)
    cs = bytearray(4 * n)
    pack_f = struct.Struct("<f").pack_into
    pack_i = struct.Struct("<i").pack_into
    off = 0
    for r in rows:
        pack_f(xs, off, float(r["umap_x"]))
        pack_f(ys, off, float(r["umap_y"]))
        cid = r["cluster_id"]
        pack_i(cs, off, int(cid) if cid is not None else -1)
        off += 4

    header = _ATLAS_COORDS_MAGIC + struct.pack("<I", n)
    body = bytes(header) + bytes(xs) + bytes(ys) + bytes(cs)
    _cache[cache_key] = (time.time(), body)
    return Response(
        content=body, media_type="application/octet-stream",
        headers={"Cache-Control": "public, max-age=300", "X-Cache": "MISS"},
    )


@app.get(
    "/api/atlas/point",
    summary="Per-point atlas detail (lazy hover/click lookup)",
    description=(
        "Returns the hover-only fields for a single atlas point: post_kind, "
        "thumbnail_media_id, username, cluster_id. Fetched one point at a time "
        "by the SPA on pointover, so these never need to ride in the bulk blob."
    ),
)
async def atlas_point(post_id: int = Query(..., description="post_clusters.post_id")):
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    pool = await get_db()
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """SELECT pc.cluster_id, c.username,
                      agg.thumbnail_media_id, agg.media_count,
                      agg.has_video, agg.has_image
                 FROM post_clusters pc
                 JOIN posts p ON p.id = pc.post_id
                 LEFT JOIN creators c ON c.id = p.creator_id
                 LEFT JOIN LATERAL (
                      SELECT count(*) AS media_count,
                             bool_or(file_type = 'video') AS has_video,
                             bool_or(file_type LIKE 'image%' OR file_type='photo') AS has_image,
                             (array_agg(id ORDER BY page_order NULLS LAST, id))[1] AS thumbnail_media_id
                        FROM media_files mf
                       WHERE mf.post_id = p.id
                 ) agg ON TRUE
                WHERE pc.post_id = $1""",
            post_id,
        )
    if row is None:
        raise HTTPException(404, "Point not found")

    payload = {
        "post_id": str(post_id),
        "post_kind": _atlas_kind(row["media_count"], row["has_image"], row["has_video"]),
        "thumbnail_media_id": row["thumbnail_media_id"],
        "username": row["username"],
        "cluster_id": int(row["cluster_id"]) if row["cluster_id"] is not None else -1,
    }
    return Response(
        content=orjson.dumps(payload), media_type="application/json",
        headers={"Cache-Control": "public, max-age=300"},
    )


@app.get(
    "/api/evoc/tree",
    summary="Hierarchical EVOC cluster tree",
    description=(
        "Return a nested d3.hierarchy-compatible tree of EVOC clusters for the "
        "/evoc-tree treemap. Each node carries representative thumb IDs. "
        "Noise clusters (cluster_id=-1) are excluded. Defaults to the latest model_version."
    ),
)
async def evoc_tree(model_version: Optional[str] = Query(None, description="EVOC model version; defaults to the most recent run")):
    """Hierarchical EVOC cluster tree for the /evoc-tree treemap.

    Returns a nested {name, value, children, ...} structure suitable for
    d3.hierarchy(). Each node carries representative_thumb_ids so the UI can
    pick a representative cover image for the rectangle / preview pane.
    Excludes noise clusters (cluster_id = -1) from the tree to keep the
    treemap clean — noise members surface only in the per-post lookup.
    """
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    pool = await get_db()
    async with pool.acquire() as conn:
        if model_version is None:
            row = await conn.fetchrow(
                "SELECT model_version FROM evoc_clusters "
                "ORDER BY computed_at DESC LIMIT 1"
            )
            if row is None:
                return JSONResponse({"model_version": None, "tree": None, "n_clusters": 0})
            model_version = row["model_version"]

        rows = await conn.fetch(
            """SELECT id, layer_idx, cluster_id, parent_id, n_members,
                      representative_post_ids, representative_thumb_ids, label
                 FROM evoc_clusters
                WHERE model_version = $1
                  AND cluster_id >= 0
                ORDER BY layer_idx DESC, n_members DESC""",
            model_version,
        )

    if not rows:
        return JSONResponse({"model_version": model_version, "tree": None, "n_clusters": 0})

    by_id: dict[int, dict] = {}
    children: dict[int, list[int]] = {}
    for r in rows:
        node_id = int(r["id"])
        by_id[node_id] = {
            "id": node_id,
            "name": f"L{r['layer_idx']}.C{r['cluster_id']}",
            "label": r["label"] or "",
            "layer_idx": int(r["layer_idx"]),
            "cluster_id": int(r["cluster_id"]),
            "value": int(r["n_members"]),
            "thumb_ids": [int(t) for t in (r["representative_thumb_ids"] or [])][:6],
            "post_ids": [str(p) for p in (r["representative_post_ids"] or [])][:24],
            "children": [],
        }
        pid = r["parent_id"]
        if pid is not None:
            children.setdefault(int(pid), []).append(node_id)

    for parent_id, kids in children.items():
        if parent_id in by_id:
            by_id[parent_id]["children"] = [by_id[k] for k in kids if k in by_id]

    roots = [n for n in by_id.values() if not any(n["id"] in kids for kids in children.values())]
    if len(roots) == 1:
        tree = roots[0]
    else:
        tree = {
            "id": 0,
            "name": "root",
            "layer_idx": -1,
            "cluster_id": -1,
            "value": sum(r["value"] for r in roots),
            "thumb_ids": [],
            "post_ids": [],
            "children": roots,
        }

    return JSONResponse(
        content={
            "model_version": model_version,
            "n_clusters": len(by_id),
            "tree": tree,
        },
        headers={"Cache-Control": "public, max-age=60"},
    )


@app.get(
    "/api/evoc/parcoords",
    summary="Per-post EVOC layer assignments for parallel-coordinates plot",
    description=(
        "Return one row per post with its cluster assignment at every EVOC layer "
        "(coarsest to finest), plus a thumbnail. The frontend brushes this to filter "
        "the treemap. ~32k rows; cached in-process 10 min."
    ),
)
async def evoc_parcoords(model_version: Optional[str] = Query(None, description="EVOC model version; defaults to the most recent run")):
    """Per-post layer assignments for the parallel-coordinates plot.

    Returns one row per post with its cluster_id at every EVOC layer
    (coarsest → finest), plus a thumbnail. The frontend uses this to draw a
    32k-line polyline canvas; brushing on any axis filters down which clusters
    appear in the treemap below.

    Payload size: ~32k rows × (1 post_id + 9 layers + 1 thumb) × 4 bytes
    = ~1.3 MB; ~400 KB gzipped. Cached 60 s during active dev.

    Server-side memoized for 10 min per (model_version) so the 400-700ms
    DB+aggregation doesn't run on every page load.
    """
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    cache_key = f"parcoords:{model_version or 'latest'}"
    cached = _cache.get(cache_key)
    if cached is not None:
        cached_at, payload = cached
        if time.time() - cached_at < 600:  # 10 min in-process; shorter than atlas (active dev)
            return JSONResponse(
                content=payload,
                headers={"Cache-Control": "public, max-age=60", "X-Cache": "HIT"},
            )

    pool = await get_db()
    async with pool.acquire() as conn:
        if model_version is None:
            row = await conn.fetchrow(
                "SELECT model_version FROM evoc_clusters "
                "ORDER BY computed_at DESC LIMIT 1"
            )
            if row is None:
                return JSONResponse({"model_version": None, "n_layers": 0,
                                     "n_posts": 0, "posts": []})
            model_version = row["model_version"]

        # Determine layer count for this model.
        max_layer_row = await conn.fetchrow(
            "SELECT MAX(layer_idx) AS m FROM evoc_clusters WHERE model_version = $1",
            model_version,
        )
        n_layers = (max_layer_row["m"] or 0) + 1

        # Per-post layer assignments + thumbnail. The per-post thumbnail/has_video
        # aggregate (formerly a LEFT JOIN LATERAL over the 2.77M-row media_files
        # table for each of ~32k posts) is now a single PK lookup into the
        # materialized post_media_rollup. See migrations/012_post_media_rollup.sql.
        rows = await conn.fetch(
            """SELECT pem.post_id,
                      array_agg(ec.cluster_id ORDER BY pem.layer_idx DESC)
                          AS clusters_high_to_low,
                      r.thumbnail_media_id,
                      r.has_video
                 FROM post_evoc_membership pem
                 JOIN evoc_clusters ec ON ec.id = pem.cluster_db_id
                 LEFT JOIN post_media_rollup r ON r.post_id = pem.post_id
                WHERE pem.model_version = $1
                GROUP BY pem.post_id, r.thumbnail_media_id, r.has_video""",
            model_version,
        )

    posts = [
        {
            "id": str(r["post_id"]),
            "l": [int(c) for c in r["clusters_high_to_low"]],
            "t": int(r["thumbnail_media_id"]) if r["thumbnail_media_id"] else None,
            "v": bool(r["has_video"]),
        }
        for r in rows
    ]

    payload = {
        "model_version": model_version,
        "n_layers": n_layers,
        "n_posts": len(posts),
        "posts": posts,
    }
    _cache[cache_key] = (time.time(), payload)
    return JSONResponse(
        content=payload,
        headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"},
    )


async def _prewarm_thumb(media_id: int, is_video: bool) -> None:
    """Background task: ensure thumbnail cache file exists for *media_id*.

    Mirrors the logic of thumb_media / video_thumb_media without issuing an
    HTTP response. Errors are swallowed so a missing/slow NAS file never
    blocks the cluster-members response.
    """
    try:
        # Image thumb (JPEG, w=200 to match the frontend request)
        w = 200
        cache_key = hashlib.md5(f"{media_id}:{w}:jpg".encode()).hexdigest()
        cache_path = THUMB_CACHE_DIR / f"{cache_key}.jpg"
        if cache_path.exists():
            return  # already cached

        pool = await get_db()
        row = await _get_media_row(pool, media_id)
        if not row:
            return

        file_path = _resolve_media_path(row["local_path"])
        if not os.path.isfile(file_path):
            return

        ft = (row["file_type"] or "").lower()
        if ft == "video" or is_video:
            video_cache = THUMB_CACHE_DIR / f"video_{media_id}.jpg"
            if video_cache.exists():
                return
            async with _THUMB_SEMAPHORE:
                try:
                    await asyncio.to_thread(generate_smart_thumb, file_path, str(video_cache), w)
                except Exception:
                    pass
        else:
            def _do_resize() -> None:
                with Image.open(file_path) as img:
                    img = img.convert("RGB")
                    ratio = w / img.width
                    new_h = int(img.height * ratio)
                    img = img.resize((w, new_h), Image.LANCZOS)
                    img.save(cache_path, "JPEG", quality=THUMB_QUALITY, optimize=True)
            async with _THUMB_SEMAPHORE:
                try:
                    await asyncio.to_thread(_do_resize)
                except Exception:
                    pass
    except Exception:
        pass  # Never let pre-warming crash the caller


@app.get(
    "/api/evoc/cluster/{cluster_db_id}",
    summary="Representative posts for an EVOC cluster",
    description=(
        "Return enriched post cards for the given EVOC cluster's representative posts, "
        "ordered by membership strength. Called by the /evoc-tree bottom preview pane "
        "when a cluster rectangle is selected."
    ),
)
async def evoc_cluster_members(
    cluster_db_id: int,
    limit: int = Query(60, ge=1, le=240, description="Max number of representative posts to return"),
):
    """Representative posts for a single EVOC cluster (any layer).

    Returns enriched cards (post_kind, thumbnail_media_id, creator) for the
    cluster's representative_post_ids, in the order they were stored
    (top-by-membership-strength). The bottom preview pane on /evoc-tree calls
    this when a cluster is selected.
    """
    if not ENABLE_SIMILARITY_V2:
        raise HTTPException(404, "Similarity v2 disabled")

    pool = await get_db()
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            """SELECT id, layer_idx, cluster_id, n_members, representative_post_ids
                 FROM evoc_clusters WHERE id = $1""",
            cluster_db_id,
        )
        if row is None:
            raise HTTPException(404, "cluster not found")

        rep_ids = [int(p) for p in (row["representative_post_ids"] or [])][:limit]
        if not rep_ids:
            members = await conn.fetch(
                """SELECT post_id
                     FROM post_evoc_membership
                    WHERE cluster_db_id = $1
                    ORDER BY membership_strength DESC NULLS LAST, post_id
                    LIMIT $2""",
                cluster_db_id, limit,
            )
            rep_ids = [int(m["post_id"]) for m in members]

        if not rep_ids:
            return JSONResponse({
                "cluster_db_id": cluster_db_id,
                "layer_idx": int(row["layer_idx"]),
                "cluster_id": int(row["cluster_id"]),
                "n_members": int(row["n_members"]),
                "items": [],
            })

        cards = await conn.fetch(
            """SELECT p.id AS post_id, p.title, p.creator_id, c.username,
                      agg.thumbnail_media_id, agg.media_count, agg.has_video, agg.has_image
                 FROM posts p
                 LEFT JOIN creators c ON c.id = p.creator_id
                 LEFT JOIN LATERAL (
                      SELECT count(*) AS media_count,
                             bool_or(file_type = 'video') AS has_video,
                             bool_or(file_type LIKE 'image%' OR file_type='photo') AS has_image,
                             (array_agg(id ORDER BY page_order NULLS LAST, id))[1] AS thumbnail_media_id
                        FROM media_files mf
                       WHERE mf.post_id = p.id
                 ) agg ON TRUE
                WHERE p.id = ANY($1::bigint[])""",
            rep_ids,
        )
    by_id = {int(r["post_id"]): r for r in cards}

    items = []
    for pid in rep_ids:
        r = by_id.get(pid)
        if r is None:
            continue
        if r["media_count"] and r["media_count"] > 1 and r["has_image"]:
            kind = "comic"
        elif r["has_video"]:
            kind = "video"
        else:
            kind = "image"
        items.append({
            "post_id": str(pid),
            "title": r["title"] or "",
            "creator_id": r["creator_id"],
            "username": r["username"],
            "post_kind": kind,
            "thumbnail_media_id": r["thumbnail_media_id"],
            "media_count": r["media_count"] or 0,
        })

    # Fire-and-forget: warm the thumbnail cache for the first 12 representative
    # items so that when the frontend subsequently requests /api/media/.../thumb
    # the cache file already exists and the response is instant.
    _PREWARM_N = 12
    for it in items[:_PREWARM_N]:
        mid = it.get("thumbnail_media_id")
        if mid:
            asyncio.ensure_future(
                _prewarm_thumb(int(mid), it.get("post_kind") == "video")
            )

    return JSONResponse({
        "cluster_db_id": cluster_db_id,
        "layer_idx": int(row["layer_idx"]),
        "cluster_id": int(row["cluster_id"]),
        "n_members": int(row["n_members"]),
        "items": items,
    })


@app.get(
    "/api/tags",
    summary="List tags with post counts",
    description=(
        "Return tags sorted by post count descending. Optionally filter by category "
        "(tag, artist, parody, character, group, language) and/or a case-insensitive "
        "substring search."
    ),
)
async def list_tags(
    category: Optional[str] = Query(None, pattern="^(tag|artist|parody|character|group|language)$", description="Filter by tag category"),
    search: Optional[str] = Query(None, description="Case-insensitive substring match on tag name"),
    limit: int = Query(100, ge=1, le=1000, description="Maximum tags to return"),
):
    """List all tags with post counts, sorted by count."""
    pool = await get_db()

    # Pre-aggregated counts in `tag_post_counts` (refreshed by
    # `_refresh_tag_counts` every 5 min). Replaces the old
    # `JOIN post_tags … GROUP BY … ORDER BY count(*) DESC` over 606k rows
    # that measured 1.5 s; the join+sort is now 11k rows from a table sorted
    # by n DESC via the supporting index → ~5 ms (R2-04 §4). INNER JOIN keeps
    # the previous behaviour of hiding tags with zero posts.
    sql = """
        SELECT t.id, t.name, t.category, tpc.n AS post_count
        FROM tags t
        JOIN tag_post_counts tpc ON tpc.tag_id = t.id
    """
    conditions = []
    params = []
    param_idx = 1

    if category:
        conditions.append(f"t.category = ${param_idx}")
        params.append(category)
        param_idx += 1

    if search:
        conditions.append(f"t.name ILIKE ${param_idx}")
        params.append(f"%{search}%")
        param_idx += 1

    if conditions:
        sql += " WHERE " + " AND ".join(conditions)

    sql += f"""
        ORDER BY post_count DESC
        LIMIT ${param_idx}
    """
    params.append(limit)

    rows = await pool.fetch(sql, *params)
    return [dict(r) for r in rows]


@app.get(
    "/api/tags/{tag_name}/posts",
    summary="Posts for a specific tag",
    description=(
        "Paginated posts that carry the given tag, with optional sorting "
        "(by date or popularity) and media-type filtering."
    ),
)
async def tag_posts(
    tag_name: str,
    page: int = Query(1, ge=1, description="1-based page number"),
    per_page: int = Query(48, ge=1, le=200, description="Posts per page"),
    sort: Optional[str] = Query("date", pattern="^(date|popularity)$", description="Sort order: date (newest first) or popularity (liked first)"),
    media_type: Optional[str] = Query(None, pattern="^(video|image)$", description="Restrict to video or image posts only"),
):
    """Get all posts that have a specific tag, with sorting and pagination."""
    # In-process memo (per worker). Tag posts don't change minute-to-minute so
    # a 60s cache cuts repeat 1+ s queries to <2 ms. With 4 workers each builds
    # its own cache — first hits per worker still slow, then warm.
    cache_key = f"tag_posts:{tag_name}:{page}:{per_page}:{sort or 'date'}:{media_type or ''}"
    cached = _cache.get(cache_key)
    if cached is not None:
        cached_at, payload = cached
        if time.time() - cached_at < 60:
            return JSONResponse(
                content=payload,
                headers={"Cache-Control": "public, max-age=60", "X-Cache": "HIT"},
            )

    pool = await get_db()
    offset = (page - 1) * per_page

    # Paginate posts-with-this-tag FIRST, then fan out to media_files via a
    # correlated subquery (same join-then-paginate fix as /creators/{id}/posts;
    # was measured 5.3s — Agent 06 §2.2). EXISTS on post_tags dedupes posts that
    # match the tag-name through more than one tag id.
    tag_exists = ("EXISTS (SELECT 1 FROM post_tags pt JOIN tags t ON t.id=pt.tag_id "
                  "WHERE pt.post_id=p.id AND t.name=$1)")
    params = [tag_name]
    media_exists = ""
    media_filter = ""
    if media_type:
        params.append(media_type)
        mt = len(params)
        media_exists = (f" AND EXISTS (SELECT 1 FROM media_files m "
                        f"WHERE m.post_id=p.id AND m.file_type=${mt})")
        media_filter = f" AND m.file_type = ${mt}"

    if sort == "popularity":
        order_cte = "liked DESC, published_at DESC NULLS LAST"
        order_fin = "pg.liked DESC, pg.published_at DESC NULLS LAST"
    else:
        order_cte = "published_at DESC NULLS LAST"
        order_fin = "pg.published_at DESC NULLS LAST"

    params.append(per_page)
    lim = len(params)
    params.append(offset)
    off = len(params)

    sql = f"""
        WITH page AS (
            SELECT p.id, p.title, p.content, p.published_at,
                   p.creator_id, c.username,
                   (pl.post_id IS NOT NULL) AS liked
            FROM posts p
            JOIN creators c ON c.id = p.creator_id
            LEFT JOIN post_likes pl ON pl.post_id = p.id
            WHERE {tag_exists}{media_exists}
            ORDER BY {order_cte}
            LIMIT ${lim} OFFSET ${off}
        ),
        cnt AS (
            SELECT count(*) AS total
            FROM posts p
            WHERE {tag_exists}{media_exists}
        )
        SELECT pg.id, pg.title, pg.content, pg.published_at,
               pg.creator_id, pg.username, pg.liked,
               (SELECT json_agg(json_build_object(
                       'id', m.id,
                       'file_type', m.file_type,
                       'original_name', m.original_name,
                       'file_size', m.file_size
                   ) ORDER BY m.id)
                  FROM media_files m
                  WHERE m.post_id = pg.id{media_filter}) AS media,
               (SELECT total FROM cnt) AS total_count
        FROM page pg
        ORDER BY {order_fin}
    """

    rows = await pool.fetch(sql, *params)
    total = rows[0]["total_count"] if rows else 0
    posts = [dict(r) for r in rows]

    # Post-process like list_posts does
    for post in posts:
        del post["total_count"]
        if isinstance(post.get("media"), str):
            post["media"] = json.loads(post["media"])
        if post.get("id") is not None:
            post["id"] = str(post["id"])
        if post.get("published_at") is not None and hasattr(post["published_at"], "isoformat"):
            post["published_at"] = post["published_at"].isoformat()

    payload = {
        "posts": posts,
        "total": total,
        "page": page,
        "per_page": per_page,
        "pages": (total + per_page - 1) // per_page if total > 0 else 0,
    }
    _cache[cache_key] = (time.time(), payload)
    return JSONResponse(
        content=payload,
        headers={"Cache-Control": "public, max-age=60", "X-Cache": "MISS"},
    )


@app.get(
    "/api/posts/{post_id}/tags",
    summary="Tags for a specific post",
    description="Return all tags attached to the given post, ordered by category then name.",
)
async def post_tags(post_id: int):
    """Return tags for a specific post, grouped by category."""
    post_id_int = post_id
    pool = await get_db()
    rows = await pool.fetch("""
        SELECT t.id, t.name, t.category
        FROM tags t
        JOIN post_tags pt ON pt.tag_id = t.id
        WHERE pt.post_id = $1
        ORDER BY t.category, t.name
    """, post_id_int)
    return [dict(r) for r in rows]


# === Universal URL Ingest API ===


class IngestUrlBody(BaseModel):
    """Request body for POST /api/ingest/url."""
    url: str = Field(
        ...,
        description=(
            "Any HTTP(S) URL to queue for download. Routed by ingest_router: "
            "nhentai galleries/artists/groups, hentairead, pornhub video/creator, "
            "coomer/kemono creators, xvideos, beeg, hanime, missav, fpo, "
            "rule34, or generic_ytdlp fallback for any other site."
        ),
        examples=["https://www.pornhub.com/pornstar/leana-lovings"],
    )


class IngestUrlResponse(BaseModel):
    """Response from POST /api/ingest/url."""
    task_id: str
    status: str = Field(description="queued / downloading / completed / failed")
    handler: str = Field(description="Resolved handler from ingest_router.detect_route")
    url: str
    duplicate: bool | None = Field(
        default=None,
        description="True if a prior task for this URL already exists in any state",
    )
    persisted: bool | None = Field(
        default=None,
        description="True if duplicate was found via the durable store (vs in-memory)",
    )


@app.post(
    "/api/ingest/url",
    response_model=IngestUrlResponse,
    summary="Queue any URL for download",
    description=(
        "Queue any HTTP(S) URL. The URL is matched against the route table "
        "(see ingest_router.ROUTE_TABLE) to pick a handler; unmatched URLs "
        "fall back to yt-dlp. Persistent dedup against the SQLite ingest "
        "queue store: re-POSTing a URL already queued / downloading / "
        "completed / failed returns the existing task_id with `duplicate=true` "
        "instead of re-queueing. Set INGEST_RETRY_FAILED=1 in the server env "
        "to permit re-queuing of previously-failed URLs."
    ),
    dependencies=[Depends(require_trusted_client)],
)
async def ingest_url(body: IngestUrlBody) -> dict:
    """Queue any URL for download. Detects URL type and routes to the correct handler.
    Returns a task_id for status tracking."""
    url = body.url.strip()
    if not url:
        raise HTTPException(400, "Missing or empty 'url' field")

    route = detect_route(url)
    if route is None:
        raise HTTPException(400, f"Could not parse URL: {url}")

    # Content-aware dedup: match by (handler, content_id) so /g/123/ and /g/123/57/
    # collapse to one task. Falls back to URL-string match when no content_id.
    for existing in _url_ingest_tasks.values():
        if existing.status not in (IngestStatus.QUEUED, IngestStatus.DOWNLOADING):
            continue
        if route.content_id and existing.content_id:
            same = (existing.handler == route.handler
                    and existing.content_id == route.content_id)
        else:
            same = (existing.url == url)
        if same:
            return {
                "task_id": existing.task_id,
                "status": existing.status.value,
                "handler": existing.handler,
                "url": existing.url,
                "duplicate": True,
            }

    # Durable dedup: catches tasks that already completed/failed (and were
    # trimmed from the in-memory table) or survived a restart. Honors the
    # "don't re-download what we have / don't retry what failed" contract.
    prior = await asyncio.to_thread(
        ingest_persist.find_duplicate, route.handler, route.content_id, url
    )
    if prior:
        return {
            "task_id": prior["task_id"],
            "status": prior["status"],
            "handler": prior["handler"],
            "url": prior["url"],
            "duplicate": True,
            "persisted": True,
        }

    task = URLIngestTask(
        task_id=str(uuid.uuid4()),
        url=url,
        handler=route.handler,
        service=route.service,
        content_id=route.content_id,
        queued_at=time.time(),
    )
    _url_ingest_tasks[task.task_id] = task
    await asyncio.to_thread(ingest_persist.record, task)
    if _url_ingest_queue is not None:
        await _url_ingest_queue.put(task)

    return {
        "task_id": task.task_id,
        "status": task.status.value,
        "handler": task.handler,
        "url": task.url,
    }


@app.get(
    "/api/ingest/url/tasks",
    summary="List all URL ingest tasks",
    description="Return all in-memory URL ingest tasks, sorted newest-first, with timing and status details.",
)
async def ingest_url_list():
    """List all URL ingest tasks, newest first."""
    tasks = sorted(_url_ingest_tasks.values(), key=lambda t: t.queued_at, reverse=True)
    result = []
    for t in tasks:
        elapsed = None
        if t.started_at:
            end = t.completed_at or time.time()
            elapsed = round(end - t.started_at, 1)
        result.append({
            "task_id": t.task_id,
            "url": t.url,
            "handler": t.handler,
            "service": t.service,
            "content_id": t.content_id,
            "status": t.status.value,
            "queued_at": t.queued_at,
            "started_at": t.started_at,
            "completed_at": t.completed_at,
            "elapsed_seconds": elapsed,
            "error": t.error,
        })
    return result


@app.get(
    "/api/ingest/url/{task_id}",
    summary="Status of a single URL ingest task",
    description="Return the full status record for a task by its UUID task_id, including elapsed time, handler, and result.",
)
async def ingest_url_status(task_id: str):
    """Get status of a single URL ingest task."""
    task = _url_ingest_tasks.get(task_id)
    if not task:
        raise HTTPException(404, "Task not found")
    elapsed = None
    if task.started_at:
        end = task.completed_at or time.time()
        elapsed = round(end - task.started_at, 1)
    return {
        "task_id": task.task_id,
        "url": task.url,
        "handler": task.handler,
        "service": task.service,
        "content_id": task.content_id,
        "status": task.status.value,
        "queued_at": task.queued_at,
        "started_at": task.started_at,
        "completed_at": task.completed_at,
        "elapsed_seconds": elapsed,
        "error": task.error,
        "result": task.result,
    }


@app.get(
    "/api/embeddings/status",
    summary="Embedding worker status",
    description="Read-only snapshot of the most recent (or currently in-progress) embedding pipeline run, including counts and timing.",
)
async def embeddings_status():
    """Read-only snapshot of the last (or in-progress) embedding worker run."""
    return last_embedding_run


@app.get(
    "/",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Home page",
    description="Serves the SPA shell for the home / feed view.",
)
@app.get(
    "/creator/{creator_id}",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Creator page",
    description="Serves the SPA shell for a specific creator's media feed.",
)
@app.get(
    "/video/{media_id}",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Video viewer page",
    description="Serves the SPA shell for the full-screen video/image viewer.",
)
@app.get(
    "/performers",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Performers list page",
    description="Serves the SPA shell for the performers directory.",
)
@app.get(
    "/performer/{performer_id}",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Performer detail page",
    description="Serves the SPA shell for a specific performer's profile.",
)
@app.get(
    "/tags",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Tags list page",
    description="Serves the SPA shell for the tags directory.",
)
@app.get(
    "/tag/{tag_name}",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Tag detail page",
    description="Serves the SPA shell for media filtered by a specific tag.",
)
@app.get(
    "/ingest",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Ingest page",
    description="Serves the SPA shell for the URL ingest / import tool.",
)
@app.get(
    "/collections",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Collections list page",
    description="Serves the SPA shell for the collections directory.",
)
@app.get(
    "/collection/{collection_id}",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Collection detail page",
    description="Serves the SPA shell for a specific collection.",
)
@app.get(
    "/explore",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Explore page",
    description="Serves the SPA shell for the semantic-search / explore view.",
)
@app.get(
    "/explore/{post_id}",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Explore post detail page",
    description="Serves the SPA shell for a specific post in the explore view.",
)
@app.get(
    "/atlas",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Atlas page",
    description="Serves the SPA shell for the embedding-space atlas visualisation.",
)
@app.get(
    "/evoc-tree",
    response_class=HTMLResponse,
    tags=["pages"],
    summary="Evoc-tree page",
    description="Serves the SPA shell for the evocative-cluster tree view.",
)
async def serve_spa(
    request: Request,
    creator_id: int = None,
    media_id: int = None,
    performer_id: int = None,
    tag_name: str = None,
    collection_id: int = None,
    post_id: int = None,
):
    # When behind nginx the SPA shell is served directly from a precompressed
    # .br/.gz file and this handler is never reached. Standalone, add a cheap
    # ETag so repeat visits revalidate to a 304 instead of re-downloading the
    # ~370 KB shell (and re-running the 30 ms gzip) every navigation.
    html = get_index_html()
    etag = f'"{hashlib.md5(html.encode()).hexdigest()}"'
    if request.headers.get("if-none-match") == etag:
        return Response(status_code=304,
                        headers={"ETag": etag, "Cache-Control": "no-cache"})
    return HTMLResponse(html, headers={"ETag": etag, "Cache-Control": "no-cache"})