from __future__ import annotations

import asyncio
import json
import os
import threading
import time
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any

from fastapi import FastAPI, Header, HTTPException, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse, Response
from pydantic import BaseModel, ConfigDict, Field, ValidationError
from sse_starlette.sse import EventSourceResponse

from backend import VALID_RATINGS, Backend
from backend.hf_bootstrap import ensure_media_cache, ensure_store_db
from backend.media_remote import remote_media_health_payload, remote_media_url

_app_dir = Path(__file__).resolve().parent
_data_dir = _app_dir / "data"
# App uses the slim DB (local: scripts/build_slim_store.py). Override with KINK_STORE_PATH.
# On HF: point KINK_STORE_PATH at a mounted bucket path; Dockerfile defaults to bundled seed (set KINK_HF_DATASET_* for Hub catalog).
_default_store = Path(os.environ["KINK_STORE_PATH"]) if os.environ.get("KINK_STORE_PATH") else _data_dir / "store_slim.db"
FRONTEND_DIR = _app_dir / "frontend"
FRONTEND_DIST_DIR = FRONTEND_DIR / "dist"
FRONTEND_PATH = FRONTEND_DIST_DIR / "index.html"
MEDIA_ROOT = _app_dir / "data" / "cached_assets"
ADMIN_SECRET = os.environ.get("KINK_ADMIN_SECRET", "")

_MEDIA_CACHE = "public, max-age=86400, immutable"
_MEDIA_FALLBACK_CACHE = "public, max-age=3600"


def _frontend_static_cache_control(target: Path) -> str:
    """Browser e2e and local iteration can disable static caching for built assets."""
    if os.environ.get("KINK_FRONTEND_NO_CACHE") == "1":
        return "no-store"
    if target.suffix in {".js", ".css"} and target.parent.name == "assets":
        return "public, max-age=31536000, immutable"
    return "no-store"


def _require_frontend_build() -> None:
    if not FRONTEND_PATH.is_file():
        raise HTTPException(
            status_code=503,
            detail="Frontend build missing. Run `npm ci && npm run build` before serving the app.",
        )
    assets_dir = FRONTEND_DIST_DIR / "assets"
    if not assets_dir.is_dir() or not any(assets_dir.iterdir()):
        raise HTTPException(
            status_code=503,
            detail="Frontend build incomplete (no bundled assets). Run `npm ci && npm run build` to refresh.",
        )


def _hf_space_published_image() -> bool:
    """True in the Hugging Face Space Docker image (``Dockerfile`` sets ``KINK_HF_SPACE_IMAGE=1``).

    Seed deployments can serve bundled JPEG fallbacks for missing catalog tiles. Full-catalog
    deployments must expose missing bytes as 404s so the UI can use a neutral placeholder instead of
    repeating an unrelated demo image.
    """
    return _env_truthy("KINK_HF_SPACE_IMAGE")


def _media_strict_missing() -> bool:
    """If true, missing ``/media/...`` bytes return 404 instead of a same-origin placeholder image."""
    if _env_is_set("KINK_MEDIA_STRICT"):
        return _env_truthy("KINK_MEDIA_STRICT")
    return _env_truthy("KINK_HF_REQUIRE_FULL_CATALOG")


def _media_fallback_path() -> Path | None:
    """Bundled JPEG used when a catalog ``/media/...`` path has no on-disk mirror (typical on HF)."""
    override = (os.environ.get("KINK_MEDIA_FALLBACK_PATH") or "").strip()
    if override:
        cand = Path(override).expanduser()
        resolved = cand.resolve()
        if resolved.is_file():
            return resolved
        # Invalid override must not disable the bundled default (common HF foot‑gun: stale path in Space vars).
    default = (MEDIA_ROOT / "hf_seed" / "demo_kink_1.jpg").resolve()
    if MEDIA_ROOT.resolve() not in default.parents:
        return None
    return default if default.is_file() else None


def _media_health_payload() -> dict[str, Any]:
    fb = _media_fallback_path()
    payload = {
        "hf_space_image": _hf_space_published_image(),
        "strict_env": _env_truthy("KINK_MEDIA_STRICT"),
        "strict_missing_assets": _media_strict_missing(),
        "fallback_ready": fb is not None,
        "fallback_path": str(fb) if fb is not None else None,
    }
    payload.update(remote_media_health_payload())
    return payload

_backend_impl: Backend | None = None
_backend_lock = threading.Lock()


def _env_truthy(name: str) -> bool:
    return os.environ.get(name, "").strip().lower() in ("1", "true", "yes", "on")


def _env_is_set(name: str) -> bool:
    return os.environ.get(name, "").strip() != ""


def _store_path_is_ephemeral(path: Path) -> bool:
    resolved = path.resolve()
    prefixes = ("/tmp", "/var/tmp", "/dev/shm")
    return any(str(resolved) == prefix or str(resolved).startswith(f"{prefix}/") for prefix in prefixes)


def _store_storage_mode(path: Path) -> str:
    return "ephemeral" if _store_path_is_ephemeral(path) else "persistent"


def _warn_or_fail_ephemeral_store(path: Path) -> None:
    if not _store_path_is_ephemeral(path):
        return
    msg = (
        f"[kink_cli] WARNING: store path {path} is ephemeral; profiles, ratings, and partner state "
        "will be lost on restart unless KINK_STORE_PATH points at persistent storage."
    )
    print(msg, flush=True)
    # Hard-fail only when explicitly requested. Requiring a full catalog download must not imply
    # refusing /tmp on Hugging Face Spaces (default KINK_STORE_PATH): that combination left the
    # backend uninitialized forever (health: starting, all API routes 500).
    if _env_truthy("KINK_FAIL_ON_EPHEMERAL_STORE"):
        raise RuntimeError(msg)


def _boot_phase(name: str) -> "_BootTimer":
    return _BootTimer(name)


class _BootTimer:
    def __init__(self, name: str) -> None:
        self.name = name
        self._t0 = 0.0

    def __enter__(self) -> "_BootTimer":
        self._t0 = time.monotonic()
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        dt = time.monotonic() - self._t0
        status = "err" if exc_type else "ok"
        print(f"[kink_cli boot] phase={self.name} seconds={dt:.2f} status={status}", flush=True)


def _get_backend() -> Backend:
    """Open SQLite + warm caches on first use so uvicorn can bind before a multi‑GB Hub download finishes."""
    global _backend_impl
    if _backend_impl is not None:
        return _backend_impl
    with _backend_lock:
        if _backend_impl is not None:
            return _backend_impl
        total_t0 = time.monotonic()
        # ensure_media_cache pulls + extracts a Hub media archive — slow (~50-75s on cpu-basic),
        # and the catalog/SQLite/request paths don't depend on it. Defer to a daemon thread so
        # cold boot returns in seconds; individual media URLs fall through to the bundled
        # placeholder via the strict-missing path until the archive finishes hydrating.
        import threading as _threading
        def _media_warm() -> None:
            t0 = time.monotonic()
            try:
                ensure_media_cache(MEDIA_ROOT)
                print(f"[kink_cli boot] phase=media_cache_bg seconds={time.monotonic()-t0:.2f} status=ok", flush=True)
            except BaseException as exc:  # noqa: BLE001
                print(f"[kink_cli boot] phase=media_cache_bg seconds={time.monotonic()-t0:.2f} status=err exc={exc!r}", flush=True)
        _threading.Thread(target=_media_warm, name="kink-media-warm", daemon=True).start()
        with _boot_phase("ensure_store_db"):
            path = ensure_store_db(_default_store)
        _warn_or_fail_ephemeral_store(path)
        with _boot_phase("user_snapshot_restore"):
            _restore_user_snapshot_on_boot(path)
        with _boot_phase("backend_ctor"):
            b = Backend(path)
        # Catalog must be ready before the first recommendations request. Build it once here:
        # starting a warm thread and then blocking can double-build on slow cpu-basic Spaces.
        with _boot_phase("catalog_build"):
            b._catalog()
        # PPR / full similarity graph warm is RAM-heavy on multi‑GB catalogs; skip only that on
        # small Spaces (see Dockerfile KINK_SKIP_HEAVY_WARM).
        if os.environ.get("KINK_SKIP_HEAVY_WARM", "").strip().lower() not in ("1", "true", "yes", "on"):
            with _boot_phase("ppr_warm"):
                from backend.recsys_graph import warm_ppr_caches

                warm_ppr_caches(b)
        _backend_impl = b
        if os.environ.get("KINK_BACKGROUND_PPR_WARM", "").strip().lower() in ("1", "true", "yes", "on"):
            from backend.recsys_graph import warm_ppr_caches_in_background

            warm_ppr_caches_in_background(b)
        print(f"[kink_cli boot] phase=total seconds={time.monotonic() - total_t0:.2f} status=ok", flush=True)
        return b


class _BackendProxy:
    def __getattr__(self, name: str):
        return getattr(_get_backend(), name)


backend = _BackendProxy()


_LAST_RESTORE: dict[str, Any] = {"status": "unattempted", "applied": 0, "at": None, "detail": None}


def _restore_user_snapshot_on_boot(store_path: Path) -> None:
    """Pull the latest user-state snapshot from the configured Hub dataset when local user tables are empty.

    No-op when KINK_USER_SNAPSHOT_REPO is unset. Failures are logged and swallowed so a Hub outage
    cannot prevent the Space from booting (catalog still serves; periodic push will retry).
    """
    import datetime as _dt

    def _record(status: str, *, applied: int = 0, detail: str | None = None) -> None:
        _LAST_RESTORE["status"] = status
        _LAST_RESTORE["applied"] = applied
        _LAST_RESTORE["at"] = _dt.datetime.now(_dt.timezone.utc).isoformat()
        _LAST_RESTORE["detail"] = detail

    try:
        from backend import user_snapshot, user_snapshot_hub
    except ImportError:
        _record("import_error")
        return
    if not user_snapshot_hub.snapshot_enabled():
        _record("disabled")
        return
    # Always merge the Hub snapshot into the local store on boot. ``restore_user_state`` uses
    # ``INSERT OR REPLACE`` so this is idempotent — the Hub copy wins on conflicting primary
    # keys. We previously gated this on "user tables empty", but HF Spaces sometimes preserves
    # ``/tmp`` across soft restarts; that left stale rows that the empty-check treated as
    # authoritative, silently dropping more recently pushed users (the exact persistence bug
    # this subsystem exists to prevent).
    try:
        tmp = store_path.parent / ".user_state_snapshot.db"
        if not user_snapshot_hub.pull_user_snapshot(tmp):
            _record("pull_returned_false")
            return
        counts = user_snapshot.restore_user_state(store_path, tmp)
        try:
            tmp.unlink()
        except OSError:
            pass
        applied = sum(counts.values())
        users = counts.get("user", 0)
        _record("ok", applied=applied, detail=f"users={users}")
        print(
            f"[kink_cli] restored {applied} user-state rows from snapshot "
            f"({user_snapshot_hub.snapshot_repo()}/{user_snapshot_hub.snapshot_filename()})"
        )
    except Exception as exc:  # noqa: BLE001 — degrade gracefully on any restore failure
        _record("error", detail=f"{type(exc).__name__}: {exc}")
        print(f"[kink_cli] user-state snapshot restore failed: {exc}")


async def _user_snapshot_flusher(store_path: Path, interval_s: float) -> None:
    """Periodic push: dump user tables and upload to Hub when fingerprint changes.

    Initialises ``last_fingerprint`` to a sentinel so the FIRST tick always pushes — otherwise
    a user who signed up between catalog warm and the flusher's baseline-capture would never
    see their state pushed, and the next container restart would silently lose them.
    """
    from backend import user_snapshot, user_snapshot_hub

    last_fingerprint = "<never-pushed>"
    while True:
        try:
            await asyncio.sleep(interval_s)
            fp = await asyncio.to_thread(user_snapshot.compute_user_state_fingerprint, store_path)
            if fp == last_fingerprint:
                continue
            if user_snapshot.user_state_is_empty(store_path):
                # Nothing to push yet; remember the empty fingerprint so we don't keep retrying
                # on every tick when no user has signed up.
                last_fingerprint = fp
                continue
            tmp = store_path.parent / ".user_state_snapshot_out.db"
            await asyncio.to_thread(user_snapshot.dump_user_state, store_path, tmp)
            sent = await asyncio.to_thread(user_snapshot_hub.push_user_snapshot, tmp)
            try:
                tmp.unlink()
            except OSError:
                pass
            if sent:
                last_fingerprint = fp
                print(f"[kink_cli] user-state snapshot pushed (fp={fp[:80]}…)")
        except asyncio.CancelledError:
            raise
        except Exception as exc:  # noqa: BLE001 — never let the flusher die
            print(f"[kink_cli] user-state snapshot flush failed: {exc}")


@asynccontextmanager
async def lifespan(_app: FastAPI):
    """Begin Hub download + DB open in a worker thread so the server binds to PORT immediately (HF Spaces timeout)."""
    loop = asyncio.get_running_loop()
    loop.run_in_executor(None, _get_backend)

    flusher_task: asyncio.Task | None = None
    try:
        from backend import user_snapshot_hub
    except ImportError:
        user_snapshot_hub = None  # type: ignore[assignment]
    if user_snapshot_hub is not None and user_snapshot_hub.snapshot_enabled():
        interval_s = float(os.environ.get("KINK_USER_SNAPSHOT_INTERVAL_S", "60") or "60")
        print(
            f"[kink_cli] user-state snapshot ENABLED — repo={user_snapshot_hub.snapshot_repo()} "
            f"file={user_snapshot_hub.snapshot_filename()} interval={interval_s:.0f}s"
        )

        async def _start_when_ready() -> None:
            while _backend_impl is None:
                await asyncio.sleep(1.0)
            await _user_snapshot_flusher(_backend_impl.path, interval_s)

        flusher_task = asyncio.create_task(_start_when_ready())
    else:
        repo = user_snapshot_hub.snapshot_repo() if user_snapshot_hub is not None else "<no module>"
        has_token = bool((os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "").strip())
        print(
            f"[kink_cli] user-state snapshot DISABLED — repo={repo!r} HF_TOKEN={'set' if has_token else 'missing'}; "
            "users will NOT persist across container restarts. Set HF_TOKEN (and KINK_USER_SNAPSHOT_REPO if needed) to enable."
        )

    try:
        yield
    finally:
        if flusher_task is not None:
            flusher_task.cancel()
            try:
                await flusher_task
            except (asyncio.CancelledError, Exception):  # noqa: BLE001
                pass
            if _backend_impl is not None:
                try:
                    from backend import user_snapshot, user_snapshot_hub as _hub
                    tmp = _backend_impl.path.parent / ".user_state_snapshot_shutdown.db"
                    user_snapshot.dump_user_state(_backend_impl.path, tmp)
                    _hub.push_user_snapshot(tmp, commit_message="user-state snapshot (shutdown)")
                    try:
                        tmp.unlink()
                    except OSError:
                        pass
                except Exception as exc:  # noqa: BLE001
                    print(f"[kink_cli] final user-state snapshot push failed: {exc}")


app = FastAPI(title="Kink Discovery API", version="0.1.0", lifespan=lifespan)

cors_origins = os.environ.get("KINK_CORS_ORIGINS", "http://localhost:8011,http://127.0.0.1:8011,http://localhost:8012,http://127.0.0.1:8012").split(",")
app.add_middleware(
    CORSMiddleware,
    allow_origins=[o.strip() for o in cors_origins],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    # connect-src covers fetch/beacon/WS. Frontend dependencies are bundled by Vite and served same-origin.
    connect_src = " ".join(
        sorted(
            {
                "'self'",
                *(origin.strip() for origin in cors_origins if origin.strip()),
            }
        )
    )
    csp = (
        "default-src 'self' data:; "
        "base-uri 'self'; frame-ancestors 'none'; form-action 'self'; frame-src 'none'; object-src 'none'; "
        "script-src 'self'; script-src-attr 'none'; "
        "style-src 'self' 'unsafe-inline'; style-src-attr 'unsafe-inline'; "
        "img-src 'self' data: https:; media-src 'self' data: https:; font-src 'self' data: https:; "
        f"connect-src {connect_src}; "
        "manifest-src 'self'; worker-src 'self' blob:"
    )
    response.headers.setdefault("Content-Security-Policy", csp)
    response.headers.setdefault("X-Frame-Options", "DENY")
    response.headers.setdefault("X-Content-Type-Options", "nosniff")
    response.headers.setdefault("Referrer-Policy", "same-origin")
    response.headers.setdefault("Permissions-Policy", "accelerometer=(), autoplay=(), camera=(), geolocation=(), gyroscope=(), magnetometer=(), microphone=(), payment=(), usb=()")
    # credentialless: cross-origin images (e.g. CDN asset URLs in catalog) load without CORP on the image host;
    # require-corp blocked many <img> sources under COEP. Assign (not setdefault): HF README custom_headers / inner
    # layers may set require-corp first; we must win. Keep README custom_headers in sync (see repo README frontmatter).
    response.headers["Cross-Origin-Embedder-Policy"] = "credentialless"
    response.headers.setdefault("Cross-Origin-Opener-Policy", "same-origin")
    response.headers.setdefault("Cross-Origin-Resource-Policy", "same-origin")
    # Personalized GETs must not be cached by intermediaries or the browser (Playwright e2e saw stale ``/users/…`` payloads).
    if request.method == "GET" and request.url.path.startswith("/users/"):
        response.headers["Cache-Control"] = "private, no-store"
    return response


class CreateUserRequest(BaseModel):
    pass


class LoginRequest(BaseModel):
    """Strip accidental whitespace from pasted credentials (common on mobile)."""

    model_config = ConfigDict(str_strip_whitespace=True)

    user_id: str
    private_token: str


class PlaySaveRequest(BaseModel):
    kink_id: str
    interest_state: str = Field(pattern="^(love|like|curious|not_interested|hard_no)$")
    directions: list[str] = Field(default_factory=list)


class ScenarioPreferenceRequest(BaseModel):
    parent_kink_id: str
    scenario_kink_id: str
    interest_state: str = Field(pattern="^(love|like|curious|not_interested|hard_no)$")
    directions: list[str] = Field(default_factory=list)


class LinkPartnerRequest(BaseModel):
    partner_id: str


class PartnerAcceptBody(BaseModel):
    from_user_id: str


class PartnerDeclineBody(BaseModel):
    from_user_id: str


class PartnerCancelBody(BaseModel):
    to_user_id: str


class PartnerGroupCreateRequest(BaseModel):
    name: str
    member_ids: list[str] = Field(default_factory=list)


class PartnerGroupMembersRequest(BaseModel):
    member_ids: list[str] = Field(default_factory=list)


class ShareToggleRequest(BaseModel):
    share: bool


class RoleSaveRequest(BaseModel):
    kink_id: str
    selected: bool = True


class PreferenceRequest(BaseModel):
    show_images: bool


class KinkBatchRequest(BaseModel):
    ids: list[str]


class PromptRespondRequest(BaseModel):
    interest_state: str | None = Field(default=None, pattern="^(love|like|curious|not_interested|hard_no)$")
    directions: list[str] = Field(default_factory=list)
    dismiss: bool = False


def require_user(user_id: str) -> dict:
    user = backend.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="Unknown user")
    return user


def sanitize_user(user: dict) -> dict:
    return {
        "id": user["id"],
        "plays": user.get("plays", {}),
        "scenario_preferences": user.get("scenario_preferences", {}),
        "roles": user.get("roles", []),
        "partners": user["partners"],
        "incoming_partner_requests": user.get("incoming_partner_requests", []),
        "outgoing_partner_requests": user.get("outgoing_partner_requests", []),
        "partner_groups": user.get("partner_groups", []),
        "preferences": user.get("preferences", {"show_images": False}),
    }


def require_user_auth(user_id: str, private_token: str | None) -> dict:
    user_id = (user_id or "").strip()
    private_token = private_token.strip() if private_token else None
    if not private_token:
        raise HTTPException(status_code=401, detail="Missing private token")
    user = backend.get_user(user_id)
    if not user:
        raise HTTPException(
            status_code=404,
            detail=(
                "Unknown profile — this user id is not on this server. "
                "If you are on a Hugging Face Space, the database may have reset (ephemeral storage); create a new profile."
            ),
        )
    if user["private_token"] != private_token:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return user


def require_kink(kink_id: str) -> dict:
    kink = backend.get_kink(kink_id)
    if not kink:
        raise HTTPException(status_code=404, detail="Unknown kink")
    return kink


def require_partner_group_access(user_id: str, group_id: str) -> None:
    if not backend.can_access_partner_group(user_id, group_id):
        raise HTTPException(status_code=403, detail="Unknown or inaccessible partner group")


def require_admin(x_admin_secret: str | None) -> None:
    if not ADMIN_SECRET:
        raise HTTPException(status_code=403, detail="Admin access disabled (KINK_ADMIN_SECRET not configured)")
    if not x_admin_secret or x_admin_secret != ADMIN_SECRET:
        raise HTTPException(status_code=403, detail="Invalid admin secret")


async def parse_payload(request: Request, model: type[BaseModel]) -> BaseModel:
    raw = await request.body()
    if not raw:
        raw = b"{}"
    try:
        return model.model_validate_json(raw)
    except ValidationError as exc:
        raise HTTPException(status_code=422, detail=exc.errors()) from exc
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail="Invalid JSON body") from exc


async def run_blocking(func, /, *args, **kwargs):
    """Run sync backend / SQLite work away from the asyncio event loop."""
    return await asyncio.to_thread(func, *args, **kwargs)


@app.get("/", response_class=HTMLResponse, responses={200: {"content": {"text/html": {}}}})
def frontend() -> FileResponse:
    _require_frontend_build()
    return FileResponse(FRONTEND_PATH, headers={"Cache-Control": "no-store"})


@app.get("/frontend/{filepath:path}")
def frontend_static(filepath: str) -> FileResponse:
    _require_frontend_build()
    target = (FRONTEND_DIST_DIR / filepath).resolve()
    if FRONTEND_DIST_DIR.resolve() not in target.parents and target != FRONTEND_DIST_DIR.resolve():
        raise HTTPException(status_code=404, detail="Not found")
    if not target.exists() or not target.is_file():
        raise HTTPException(status_code=404, detail="Not found")
    cache_control = _frontend_static_cache_control(target)
    return FileResponse(target, headers={"Cache-Control": cache_control})


@app.get("/favicon.ico")
def favicon() -> FileResponse:
    """Browsers request this by default; serve a small bundled tile so the console stays clean on HF."""
    fb = _media_fallback_path()
    if fb is not None:
        return FileResponse(
            fb,
            media_type="image/jpeg",
            headers={"Cache-Control": _MEDIA_FALLBACK_CACHE},
        )
    raise HTTPException(status_code=404, detail="Not found")


@app.get("/media/{bucket}/{filename:path}", response_model=None)
def media(bucket: str, filename: str) -> Any:
    target = (MEDIA_ROOT / bucket / filename).resolve()
    if MEDIA_ROOT.resolve() not in target.parents:
        raise HTTPException(status_code=404, detail="Unknown asset")
    if target.is_file():
        return FileResponse(target, headers={"Cache-Control": _MEDIA_CACHE})
    remote_url = remote_media_url(f"{bucket}/{filename}")
    if remote_url:
        return RedirectResponse(
            remote_url,
            status_code=307,
            headers={"Cache-Control": "private, max-age=300"},
        )
    if not _media_strict_missing():
        fb = _media_fallback_path()
        if fb is not None:
            return FileResponse(
                fb,
                media_type="image/jpeg",
                headers={"Cache-Control": _MEDIA_FALLBACK_CACHE, "X-Kink-Media-Fallback": "1"},
            )
    raise HTTPException(status_code=404, detail="Unknown asset")


@app.get("/health")
def health() -> dict:
    store_path = _default_store.resolve()
    warnings: list[str] = []
    if _store_path_is_ephemeral(store_path):
        warnings.append("Store path is ephemeral; user data will not survive restarts.")
    if _backend_impl is None:
        return {
            "ok": False,
            "starting": True,
            "store_path": str(store_path),
            "storage_mode": _store_storage_mode(store_path),
            "store_path_persistent": not _store_path_is_ephemeral(store_path),
            "warnings": warnings,
            "media": _media_health_payload(),
        }
    b = _backend_impl
    store_path = b.path.resolve()
    warnings = []
    if _store_path_is_ephemeral(store_path):
        warnings.append("Store path is ephemeral; user data will not survive restarts.")
    cand = b.recsys_settings.candidates_path
    if cand is None:
        cand = b.path.parent / "recsys" / "candidates.json"
    settings_path = b.path.parent / "recsys" / "settings.json"
    cache_exists = cand.is_file()
    cache_age_s = int(max(0, time.time() - cand.stat().st_mtime)) if cache_exists else None
    cache_payload = b._load_ots_candidate_cache() if cache_exists else None
    cache_user_count = (
        len(cache_payload.get("users", {}))
        if isinstance(cache_payload, dict) and isinstance(cache_payload.get("users", {}), dict)
        else None
    )
    snapshot_payload: dict[str, Any] = {"enabled": False}
    try:
        from backend import user_snapshot_hub as _hub
        snapshot_payload = {
            "enabled": _hub.snapshot_enabled(),
            "repo": _hub.snapshot_repo() or None,
            "filename": _hub.snapshot_filename(),
            "has_hf_token": bool((os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "").strip()),
            "last_restore": dict(_LAST_RESTORE),
        }
    except Exception:  # noqa: BLE001
        pass
    return {
        "ok": True,
        "store_path": str(store_path),
        "storage_mode": _store_storage_mode(store_path),
        "store_path_persistent": not _store_path_is_ephemeral(store_path),
        "warnings": warnings,
        "media": _media_health_payload(),
        "stats": b.catalog_stats(),
        "user_snapshot": snapshot_payload,
        "recsys": {
            "source": b.recsys_settings.source,
            "settings_path": str(settings_path),
            "settings_file_exists": settings_path.is_file(),
            "candidates_path": str(cand),
            "candidates_cache_exists": cache_exists,
            "recsys_candidate_cache_exists": cache_exists,
            "recsys_candidate_cache_age_s": cache_age_s,
            "recsys_candidate_user_count": cache_user_count,
        },
    }


@app.get("/health/catalog-sample")
def health_catalog_sample() -> dict:
    """Cheap proof the catalog table is readable (one ``LIMIT 1`` id; no full list)."""
    if _backend_impl is None:
        return {"ok": False, "starting": True}
    kid = _backend_impl.peek_catalog_kink_id()
    return {"ok": True, "sample_kink_id": kid}


@app.get("/stats")
def stats() -> dict:
    return backend.catalog_stats()


@app.get("/admin/fetlife/jobs")
def fetlife_jobs(limit: int = Query(default=100, ge=1, le=500), state: str | None = None, x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return {"items": backend.fetlife_jobs(limit=limit, state=state), "stats": backend.fetlife_job_stats()}


@app.get("/admin/fetlife/coverage")
def fetlife_coverage(limit: int = Query(default=50, ge=1, le=200), x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return {
        "items": backend.fetlife_source_coverage(limit=limit),
        "debt": backend.fetlife_coverage_debt(limit=min(20, limit)),
    }


@app.get("/admin/fetlife/data-health")
def fetlife_data_health(limit: int = Query(default=20, ge=1, le=100), x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return backend.fetlife_data_health(limit=limit)


@app.get("/admin/fetlife/supervisor-status")
def fetlife_supervisor_status(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return backend.fetlife_supervisor_status()


@app.post("/admin/fetlife/queue-priority-debt")
def queue_fetlife_priority_debt(limit: int = Query(default=20, ge=1, le=100), x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return backend.queue_fetlife_priority_debt(limit=limit)


@app.get("/admin/recommendation-debug/{user_id}")
def recommendation_debug(
    user_id: str,
    limit: int = Query(default=10, ge=1, le=50),
    group_id: str | None = None,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
    x_admin_secret: str | None = Header(default=None, alias="x-admin-secret"),
) -> dict:
    require_admin(x_admin_secret)
    require_user_auth(user_id, x_private_token)
    return backend.recommendation_debug(user_id, limit=limit, group_id=group_id)


@app.get("/admin/types/coverage")
def type_coverage(limit: int = Query(default=20, ge=1, le=200), x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return backend.content_type_coverage(limit=limit)


@app.get("/admin/fetlife/samples")
def fetlife_samples(limit: int = Query(default=25, ge=1, le=200), x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    return backend.fetlife_sample_coverage(limit=limit)


@app.post("/admin/reload")
def admin_reload(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    require_admin(x_admin_secret)
    backend.refresh_cache()
    return {"ok": True, "stats": backend.catalog_stats()}


# ─────────────────────────────────────────────────────────────────────────────
# Admin: user-state surface for ops + safe deploy workflow.
# All endpoints below require ``x-admin-secret`` matching ``KINK_ADMIN_SECRET``.
# Designed so ``scripts/admin_deploy.py`` can:
#   • back up every profile to a local SQLite before a risky deploy
#   • restore it after if the snapshot subsystem hiccups
#   • fix individual user issues (lost token, stuck state, bad rating) without redeploying
# Coverage: tests/test_admin_endpoints.py exercises every route in this block including
# auth gate sweeps, INSERT-OR-REPLACE roundtrip on /admin/import, and Hub mocks for push/pull.
# ─────────────────────────────────────────────────────────────────────────────


@app.get("/admin/users")
def admin_list_users(
    limit: int = Query(default=200, ge=1, le=2000),
    offset: int = Query(default=0, ge=0),
    x_admin_secret: str | None = Header(default=None, alias="x-admin-secret"),
) -> dict:
    """List every user with summary counts. Sorted by id; paginate via limit/offset."""
    require_admin(x_admin_secret)
    from sqlalchemy import text as _sql_text
    with backend.engine.connect() as conn:
        users = [r[0] for r in conn.execute(_sql_text('SELECT id FROM "user" ORDER BY id'))]
        plays_by_user: dict[str, int] = dict(
            conn.execute(_sql_text('SELECT user_id, COUNT(*) FROM playpreference GROUP BY user_id')).all()
        )
        scen_by_user: dict[str, int] = dict(
            conn.execute(_sql_text('SELECT user_id, COUNT(*) FROM scenariopreference GROUP BY user_id')).all()
        )
        roles_by_user: dict[str, int] = dict(
            conn.execute(_sql_text('SELECT user_id, COUNT(*) FROM rolepreference GROUP BY user_id')).all()
        )
        groups_by_user: dict[str, int] = dict(
            conn.execute(_sql_text('SELECT member_user_id, COUNT(*) FROM partnergroupmember GROUP BY member_user_id')).all()
        )
        links_rows = conn.execute(_sql_text('SELECT left_user_id, right_user_id FROM partnerlink')).all()

    links_by_user: dict[str, int] = {}
    for left, right in links_rows:
        links_by_user[left] = links_by_user.get(left, 0) + 1
        links_by_user[right] = links_by_user.get(right, 0) + 1

    items = [
        {
            "user_id": uid,
            "plays": int(plays_by_user.get(uid, 0)),
            "scenarios": int(scen_by_user.get(uid, 0)),
            "roles": int(roles_by_user.get(uid, 0)),
            "partner_links": int(links_by_user.get(uid, 0)),
            "group_memberships": int(groups_by_user.get(uid, 0)),
        }
        for uid in users
    ]
    return {"total": len(items), "limit": limit, "offset": offset, "items": items[offset : offset + limit]}


@app.get("/admin/users/{user_id}")
def admin_get_user(user_id: str, x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Full user state INCLUDING ``private_token`` (don't ship to clients). For helping users
    who lost their pass or for debugging stuck profiles."""
    require_admin(x_admin_secret)
    user = backend.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail=f"Unknown user {user_id!r}")
    return user


@app.post("/admin/users/{user_id}/regenerate-token")
def admin_regenerate_token(user_id: str, x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Issue a new ``private_token`` for a user who has lost theirs. Old token stops working."""
    require_admin(x_admin_secret)
    from sqlmodel import Session
    from models import User
    new_token = backend._make_memorable_pass()
    with Session(backend.engine) as session:
        user = session.get(User, user_id)
        if not user:
            raise HTTPException(status_code=404, detail=f"Unknown user {user_id!r}")
        user.private_token = new_token
        session.add(user)
        session.commit()
    return {"user_id": user_id, "private_token": new_token}


@app.delete("/admin/users/{user_id}")
async def admin_delete_user(user_id: str, x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Hard-delete a profile + all its plays / scenarios / partner state. Idempotent."""
    require_admin(x_admin_secret)
    backend.delete_user(user_id)
    return {"ok": True, "user_id": user_id}


@app.post("/admin/users/{user_id}/scrub-plays")
async def admin_scrub_plays(user_id: str, x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Wipe every PlayPreference for one user (keeps profile, scenarios, roles)."""
    require_admin(x_admin_secret)
    from sqlmodel import Session, delete as sql_delete
    from models import PlayPreference
    with Session(backend.engine) as session:
        result = session.exec(sql_delete(PlayPreference).where(PlayPreference.user_id == user_id))
        deleted = getattr(result, "rowcount", 0) or 0
        session.commit()
    return {"ok": True, "user_id": user_id, "deleted": int(deleted)}


@app.delete("/admin/users/{user_id}/plays/{kink_id}")
def admin_delete_play(
    user_id: str,
    kink_id: str,
    x_admin_secret: str | None = Header(default=None, alias="x-admin-secret"),
) -> dict:
    """Remove ONE play preference for a user (no token required)."""
    require_admin(x_admin_secret)
    backend.remove_play_preference(user_id, kink_id)
    return {"ok": True, "user_id": user_id, "kink_id": kink_id}


@app.get("/admin/export")
def admin_export(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> Response:
    """Stream the user-state SQLite (USER_TABLES only — no catalog). Suitable as a pre-deploy
    backup. Identical schema/format to what the snapshot flusher pushes to Hub.
    """
    require_admin(x_admin_secret)
    import tempfile
    from backend import user_snapshot
    store_path = _default_store.resolve()
    tmp = Path(tempfile.mkdtemp()) / "user_state.db"
    user_snapshot.dump_user_state(store_path, tmp)
    data = tmp.read_bytes()
    try:
        tmp.unlink()
        tmp.parent.rmdir()
    except OSError:
        pass
    headers = {"Content-Disposition": 'attachment; filename="user_state_export.db"'}
    return Response(content=data, media_type="application/octet-stream", headers=headers)


@app.post("/admin/import")
async def admin_import(
    request: Request,
    x_admin_secret: str | None = Header(default=None, alias="x-admin-secret"),
) -> dict:
    """Upload a user-state SQLite (raw body, octet-stream) and apply via INSERT OR REPLACE.

    Existing rows with conflicting primary keys are overwritten by the snapshot's values; rows
    only in the live store are kept. Catalog tables are untouched.
    """
    require_admin(x_admin_secret)
    import tempfile
    from backend import user_snapshot
    body = await request.body()
    if not body:
        raise HTTPException(status_code=400, detail="empty body — POST raw user_state.db bytes")
    tmp = Path(tempfile.mkdtemp()) / "import.db"
    tmp.write_bytes(body)
    try:
        counts = user_snapshot.restore_user_state(_default_store.resolve(), tmp)
    finally:
        try:
            tmp.unlink()
            tmp.parent.rmdir()
        except OSError:
            pass
    backend.refresh_cache()
    return {"ok": True, "applied": counts, "total_rows": sum(counts.values())}


@app.post("/admin/snapshot/push")
async def admin_snapshot_push(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Force an immediate user-state push to the Hub dataset (bypasses the periodic flusher)."""
    require_admin(x_admin_secret)
    from backend import user_snapshot, user_snapshot_hub
    if not user_snapshot_hub.snapshot_enabled():
        raise HTTPException(status_code=409, detail="snapshot subsystem not enabled (HF_TOKEN or repo missing)")
    store_path = _default_store.resolve()
    if user_snapshot.user_state_is_empty(store_path):
        return {"ok": False, "skipped": "store has no user rows; refusing to overwrite Hub with empty snapshot"}
    import tempfile
    tmp = Path(tempfile.mkdtemp()) / "out.db"
    user_snapshot.dump_user_state(store_path, tmp)
    try:
        sent = user_snapshot_hub.push_user_snapshot(tmp, commit_message="admin: manual push")
    finally:
        try:
            tmp.unlink()
            tmp.parent.rmdir()
        except OSError:
            pass
    return {"ok": bool(sent), "fingerprint": user_snapshot.compute_user_state_fingerprint(store_path)}


@app.post("/admin/snapshot/pull")
async def admin_snapshot_pull(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Force pull the Hub snapshot and merge into the live store (INSERT OR REPLACE).

    Unlike boot-restore, this runs even when the store already has rows — so use it to
    recover a user who's missing from /tmp but present on Hub.
    """
    require_admin(x_admin_secret)
    from backend import user_snapshot, user_snapshot_hub
    if not user_snapshot_hub.snapshot_enabled():
        raise HTTPException(status_code=409, detail="snapshot subsystem not enabled")
    store_path = _default_store.resolve()
    tmp = store_path.parent / ".admin_pull_snapshot.db"
    if not user_snapshot_hub.pull_user_snapshot(tmp):
        raise HTTPException(status_code=502, detail="Hub pull returned no file (snapshot not present yet?)")
    counts = user_snapshot.restore_user_state(store_path, tmp)
    try:
        tmp.unlink()
    except OSError:
        pass
    backend.refresh_cache()
    return {"ok": True, "applied": counts, "total_rows": sum(counts.values())}


@app.get("/admin/stats")
def admin_stats(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Aggregate counts for an at-a-glance ops dashboard."""
    require_admin(x_admin_secret)
    import sqlite3
    store_path = _default_store.resolve()
    conn = sqlite3.connect(f"file:{store_path}?mode=ro", uri=True)
    try:
        result = {}
        for table in ("user", "userpreference", "playpreference", "rolepreference",
                      "scenariopreference", "promptdismissal", "partnerlink",
                      "partnerlinkrequest", "partnergroup", "partnergroupmember"):
            try:
                result[table] = int(conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0])
            except sqlite3.OperationalError:
                result[table] = None
        # Top-rated kinks across all users
        top = conn.execute(
            "SELECT kink_id, COUNT(*) AS n FROM playpreference GROUP BY kink_id ORDER BY n DESC LIMIT 20"
        ).fetchall()
        top_kinks = [{"kink_id": k, "save_count": int(n)} for k, n in top]
    finally:
        conn.close()
    return {"row_counts": result, "top_kinks": top_kinks, "store_path": str(store_path)}


@app.get("/admin/health")
def admin_health(x_admin_secret: str | None = Header(default=None, alias="x-admin-secret")) -> dict:
    """Detailed health for ops — includes the boot-restore log and snapshot subsystem state."""
    require_admin(x_admin_secret)
    from backend import user_snapshot_hub
    return {
        "ok": _backend_impl is not None,
        "store_path": str(_default_store.resolve()),
        "store_path_persistent": not _store_path_is_ephemeral(_default_store),
        "snapshot": {
            "enabled": user_snapshot_hub.snapshot_enabled(),
            "repo": user_snapshot_hub.snapshot_repo() or None,
            "filename": user_snapshot_hub.snapshot_filename(),
            "has_hf_token": bool((os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or "").strip()),
            "last_restore": dict(_LAST_RESTORE),
        },
    }


@app.post("/admin/users/{user_id}/partners/{partner_id}/force-link")
def admin_force_partner_link(
    user_id: str,
    partner_id: str,
    x_admin_secret: str | None = Header(default=None, alias="x-admin-secret"),
) -> dict:
    """Force-create a bidirectional partner link without going through the request/accept flow.

    Useful when one side's UI is stuck on a pending invite and the other side has already
    confirmed verbally. Idempotent — duplicate calls are a no-op.
    """
    require_admin(x_admin_secret)
    if user_id == partner_id:
        raise HTTPException(status_code=400, detail="cannot link a user to themselves")
    if not backend.get_user(user_id):
        raise HTTPException(status_code=404, detail=f"Unknown user {user_id!r}")
    if not backend.get_user(partner_id):
        raise HTTPException(status_code=404, detail=f"Unknown user {partner_id!r}")
    from sqlmodel import Session, select
    from models import PartnerLink, PartnerLinkRequest
    a, b = sorted([user_id, partner_id])
    with Session(backend.engine) as session:
        existing = session.exec(
            select(PartnerLink).where(PartnerLink.left_user_id == a, PartnerLink.right_user_id == b)
        ).first()
        if existing:
            return {"ok": True, "user_id": user_id, "partner_id": partner_id, "created": False}
        session.add(PartnerLink(left_user_id=a, right_user_id=b))
        # Drop any stale pending requests in either direction.
        for req in session.exec(
            select(PartnerLinkRequest).where(
                ((PartnerLinkRequest.from_user_id == user_id) & (PartnerLinkRequest.to_user_id == partner_id))
                | ((PartnerLinkRequest.from_user_id == partner_id) & (PartnerLinkRequest.to_user_id == user_id))
            )
        ).all():
            session.delete(req)
        session.commit()
    return {"ok": True, "user_id": user_id, "partner_id": partner_id, "created": True}


@app.delete("/admin/partners/{user_a}/{user_b}")
def admin_unlink_partners(
    user_a: str,
    user_b: str,
    x_admin_secret: str | None = Header(default=None, alias="x-admin-secret"),
) -> dict:
    """Drop a partner link between two users."""
    require_admin(x_admin_secret)
    a, b = sorted([user_a, user_b])
    from sqlmodel import Session, select
    from models import PartnerLink
    with Session(backend.engine) as session:
        link = session.exec(
            select(PartnerLink).where(PartnerLink.left_user_id == a, PartnerLink.right_user_id == b)
        ).first()
        if not link:
            return {"ok": True, "removed": False}
        session.delete(link)
        session.commit()
    return {"ok": True, "removed": True}


@app.get("/kinks")
def list_kinks(limit: int = Query(default=100, ge=1, le=500), offset: int = Query(default=0, ge=0)) -> dict:
    items = backend.list_kink_summaries()
    sliced = items[offset : offset + limit]
    return {"items": sliced, "total": len(items), "limit": limit, "offset": offset}


@app.get("/kinks/{kink_id}")
def get_kink(kink_id: str) -> dict:
    return require_kink(kink_id)


@app.get("/kinks/{kink_id}/similar")
def similar_kinks(kink_id: str, limit: int = Query(default=8, ge=1, le=50)) -> dict:
    require_kink(kink_id)
    return {"items": backend.similar_kinks(kink_id, limit=limit)}


@app.get("/kinks/{kink_id}/scenarios")
def kink_scenarios(kink_id: str, limit: int = Query(default=24, ge=1, le=100)) -> dict:
    require_kink(kink_id)
    return {"items": backend.list_scenarios_for_parent(kink_id, limit=limit)}


@app.get("/search")
def search(q: str = Query(min_length=1), limit: int = Query(default=10, ge=1, le=100)) -> dict:
    return {"items": backend.search_kinks(q, limit=limit)}


@app.get("/roles/search")
def search_roles(q: str = Query(min_length=1), limit: int = Query(default=10, ge=1, le=100)) -> dict:
    return {"items": backend.search_roles(q, limit=limit)}


@app.post("/kinks/batch", responses={422: {"description": "Validation Error", "content": {"application/json": {}}}})
async def kink_batch(request: Request) -> dict:
    payload = await parse_payload(request, KinkBatchRequest)
    return await run_blocking(lambda: {"items": backend.get_kinks_batch(payload.ids)})


@app.post("/users")
async def create_user(request: Request) -> dict:
    await parse_payload(request, CreateUserRequest)
    user = await run_blocking(backend.create_user)
    return {
        "id": user.id,
        "private_token": user.private_token,
    }


@app.post("/auth/login")
async def login(request: Request) -> dict:
    payload = await parse_payload(request, LoginRequest)
    user = await run_blocking(require_user_auth, payload.user_id, payload.private_token)
    return {
        "user": sanitize_user(user),
        "private_token": payload.private_token,
    }


@app.get("/users/{user_id}")
def get_user(user_id: str, x_private_token: str | None = Header(default=None, alias="x-private-token")) -> dict:
    return sanitize_user(require_user_auth(user_id, x_private_token))


@app.get("/users/{user_id}/play-board")
def play_board(user_id: str, x_private_token: str | None = Header(default=None, alias="x-private-token")) -> dict:
    require_user_auth(user_id, x_private_token)
    return backend.play_board(user_id)


@app.get("/users/{user_id}/scenario-parents")
def scenario_parents(
    user_id: str,
    limit: int = Query(default=100, ge=1, le=500),
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    return {"items": backend.scenario_parents_for_user(user_id, limit=limit)}


@app.post("/users/{user_id}/preferences")
async def save_preferences(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PreferenceRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        user = backend.save_preferences(user_id, show_images=payload.show_images)
        if not user:
            raise HTTPException(status_code=404, detail="Unknown user")
        return sanitize_user(user)
    return await run_blocking(work)


@app.delete("/users/{user_id}/plays/{kink_id}")
def delete_play(
    user_id: str,
    kink_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    backend.remove_play_preference(user_id, kink_id)
    return {"ok": True}


@app.post("/users/{user_id}/plays")
async def save_play(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PlaySaveRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        require_kink(payload.kink_id)
        if payload.interest_state not in VALID_RATINGS:
            raise HTTPException(status_code=400, detail="Invalid interest state")
        backend.save_play_preference(user_id, payload.kink_id, payload.interest_state, payload.directions)
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.post("/users/{user_id}/scenario-preferences")
async def save_scenario_preference(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, ScenarioPreferenceRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        require_kink(payload.parent_kink_id)
        require_kink(payload.scenario_kink_id)
        if payload.interest_state not in VALID_RATINGS:
            raise HTTPException(status_code=400, detail="Invalid interest state")
        ok = backend.save_scenario_preference(
            user_id,
            payload.parent_kink_id,
            payload.scenario_kink_id,
            payload.interest_state,
            payload.directions,
        )
        if not ok:
            raise HTTPException(status_code=400, detail="Scenario is not linked to that parent kink")
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.delete("/users/{user_id}/scenario-preferences/{scenario_kink_id}")
async def delete_scenario_preference(
    user_id: str,
    scenario_kink_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        backend.remove_scenario_preference(user_id, scenario_kink_id)
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.get("/users/{user_id}/roles")
def get_roles(user_id: str, x_private_token: str | None = Header(default=None, alias="x-private-token")) -> dict:
    require_user_auth(user_id, x_private_token)
    return {"items": backend.role_cards(user_id)}


@app.post("/users/{user_id}/roles")
async def save_role(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, RoleSaveRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        require_kink(payload.kink_id)
        backend.save_role_preference(user_id, payload.kink_id, payload.selected)
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.get("/users/{user_id}/recommendations")
def recommendations(
    user_id: str,
    limit: int = Query(default=10, ge=1, le=100),
    group_id: str | None = None,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    if group_id:
        require_partner_group_access(user_id, group_id)
    return {"items": backend.recommend(user_id, limit=limit, group_id=group_id)}


@app.get("/users/{user_id}/prompts")
def prompts(
    user_id: str,
    limit: int = Query(default=8, ge=1, le=50),
    group_id: str | None = None,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    if group_id:
        require_partner_group_access(user_id, group_id)
    return {"items": backend.prompt_candidates(user_id, limit=limit, group_id=group_id)}


@app.post("/users/{user_id}/prompts/{kink_id}")
async def respond_prompt(
    user_id: str,
    kink_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PromptRespondRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        require_kink(kink_id)
        if payload.dismiss and payload.interest_state:
            raise HTTPException(status_code=400, detail="Dismiss or rate a prompt, not both")
        if payload.dismiss:
            backend.dismiss_prompt(user_id, kink_id)
            return {"ok": True}
        if not payload.interest_state:
            raise HTTPException(status_code=400, detail="Missing interest state")
        backend.save_play_preference(user_id, kink_id, payload.interest_state, payload.directions)
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.post("/users/{user_id}/partners")
async def request_partner_link_endpoint(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    """Send a link request; the other user must accept before you appear as partners."""
    payload = await parse_payload(request, LinkPartnerRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        require_user(payload.partner_id)
        ok, msg = backend.request_partner_link(user_id, payload.partner_id)
        if not ok:
            raise HTTPException(status_code=400, detail=msg)
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.post("/users/{user_id}/partner-requests/accept")
async def accept_partner_request_endpoint(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PartnerAcceptBody)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        if not backend.accept_partner_request(user_id, payload.from_user_id):
            raise HTTPException(status_code=400, detail="No pending request from that user")
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.post("/users/{user_id}/partner-requests/decline")
async def decline_partner_request_endpoint(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PartnerDeclineBody)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        if not backend.decline_partner_request(user_id, payload.from_user_id):
            raise HTTPException(status_code=400, detail="No pending request from that user")
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.post("/users/{user_id}/partner-requests/cancel")
async def cancel_partner_request_endpoint(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PartnerCancelBody)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        if not backend.cancel_partner_request(user_id, payload.to_user_id):
            raise HTTPException(status_code=400, detail="No outgoing request to that user")
        return sanitize_user(require_user(user_id))
    return await run_blocking(work)


@app.get("/users/{user_id}/partner-groups")
def list_partner_groups(
    user_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    return {"items": backend.list_partner_groups(user_id)}


@app.post("/users/{user_id}/partner-groups")
async def create_partner_group(
    user_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PartnerGroupCreateRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        group = backend.create_partner_group(user_id, payload.name, payload.member_ids)
        if not group:
            raise HTTPException(status_code=400, detail="Could not create partner group")
        return group
    return await run_blocking(work)


@app.post("/users/{user_id}/partner-groups/{group_id}/members")
async def add_partner_group_members(
    user_id: str,
    group_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, PartnerGroupMembersRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        group = backend.add_partner_group_members(user_id, group_id, payload.member_ids)
        if not group:
            raise HTTPException(status_code=400, detail="Could not update group members")
        return group
    return await run_blocking(work)


@app.delete("/users/{user_id}/partner-groups/{group_id}/members/{member_id}")
def remove_partner_group_member(
    user_id: str,
    group_id: str,
    member_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    if not backend.remove_partner_group_member(user_id, group_id, member_id):
        raise HTTPException(status_code=404, detail="Unknown group member")
    return {"ok": True}


@app.get("/users/{user_id}/partner-groups/{group_id}/overlap")
def overlap_group(
    user_id: str,
    group_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    require_partner_group_access(user_id, group_id)
    return backend.shared_play_list_for_group(user_id, group_id)


@app.get("/users/{user_id}/partner-groups/{group_id}/prompts")
def prompts_group(
    user_id: str,
    group_id: str,
    limit: int = Query(default=8, ge=1, le=50),
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    require_partner_group_access(user_id, group_id)
    return {"items": backend.group_prompt_candidates(user_id, group_id, limit=limit)}


@app.get("/users/{user_id}/overlap/{partner_id}")
def overlap(
    user_id: str,
    partner_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    user = require_user_auth(user_id, x_private_token)
    require_user(partner_id)
    if partner_id not in user.get("partners", []):
        raise HTTPException(status_code=403, detail="Users are not linked partners")
    return backend.shared_play_list(user_id, partner_id)


@app.delete("/users/{user_id}")
async def delete_user(
    user_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        backend.delete_user(user_id)
        return {"ok": True}
    return await run_blocking(work)


@app.get("/users/{user_id}/export")
def export_user(
    user_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    return backend.export_user_data(user_id)


@app.get("/users/{user_id}/events")
async def user_events(
    user_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> EventSourceResponse:
    await run_blocking(require_user_auth, user_id, x_private_token)

    async def event_stream():
        last_hash = ""
        while True:
            await asyncio.sleep(5)
            current = await run_blocking(backend.partner_activity_hash, user_id)
            if current != last_hash:
                last_hash = current
                yield {"event": "partner-update", "data": json.dumps({"user_id": user_id, "hash": current})}

    return EventSourceResponse(event_stream())


@app.post("/users/{user_id}/partner-groups/{group_id}/share-toggle")
async def toggle_share(
    user_id: str,
    group_id: str,
    request: Request,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    payload = await parse_payload(request, ShareToggleRequest)
    def work() -> dict:
        require_user_auth(user_id, x_private_token)
        if not backend.toggle_share_full_list(user_id, group_id, payload.share):
            raise HTTPException(status_code=400, detail="Could not update sharing preference")
        return {"ok": True, "share": payload.share}
    return await run_blocking(work)


@app.get("/users/{user_id}/partner-groups/{group_id}/partner-board/{partner_id}")
def partner_board(
    user_id: str,
    group_id: str,
    partner_id: str,
    x_private_token: str | None = Header(default=None, alias="x-private-token"),
) -> dict:
    require_user_auth(user_id, x_private_token)
    board = backend.partner_full_board(user_id, group_id, partner_id)
    if board is None:
        raise HTTPException(status_code=403, detail="Partner has not enabled sharing in this group")
    return board


TELEMETRY_FILE = Path(__file__).resolve().parent / "data" / "telemetry.jsonl"


@app.post("/telemetry")
async def receive_telemetry(request: Request) -> dict:
    import time
    body = await request.body()
    try:
        payload = json.loads(body) if body else {}
    except json.JSONDecodeError:
        return {"ok": False}
    entry = {"ts": time.time(), "payload": payload}
    def write_entry() -> None:
        TELEMETRY_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(TELEMETRY_FILE, "a") as f:
            f.write(json.dumps(entry) + "\n")
    await run_blocking(write_entry)
    return {"ok": True}