api.py

from __future__ import annotations

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

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

from backend import Backend, VALID_RATINGS
from backend.hf_bootstrap import ensure_store_db

_data_dir = Path(__file__).resolve().parent / "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 sets KINK_HF_DATASET_REPO for full-catalog bootstrap.
_default_store = Path(os.environ["KINK_STORE_PATH"]) if os.environ.get("KINK_STORE_PATH") else _data_dir / "store_slim.db"
FRONTEND_PATH = Path(__file__).resolve().parent / "frontend" / "index.html"
FRONTEND_DIR = Path(__file__).resolve().parent / "frontend"
MEDIA_ROOT = Path(__file__).resolve().parent / "data" / "cached_assets"
ADMIN_SECRET = os.environ.get("KINK_ADMIN_SECRET", "")

_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 _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 _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
        path = ensure_store_db(_default_store)
        _warn_or_fail_ephemeral_store(path)
        b = Backend(path)
        # Optional: set KINK_SKIP_HEAVY_WARM=1 only if you have measured a real memory/startup problem.
        if os.environ.get("KINK_SKIP_HEAVY_WARM", "").strip().lower() not in ("1", "true", "yes", "on"):
            b.warm_up_catalog()
            b._catalog()
            from backend.recsys_graph import warm_ppr_caches

            warm_ppr_caches(b)
        _backend_impl = b
        return b


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


backend = _BackendProxy()


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


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: ES module graph (esm.sh), dynamic import (e.g. web-vitals from unpkg), fetch, beacon, WS.
    connect_src = " ".join(
        sorted(
            {
                "'self'",
                "https://esm.sh",
                "https://unpkg.com",
                *(origin.strip() for origin in cors_origins if origin.strip()),
            }
        )
    )
    csp = (
        "default-src 'self' https://esm.sh https://unpkg.com data:; "
        "base-uri 'self'; frame-ancestors 'none'; form-action 'self'; frame-src 'none'; object-src 'none'; "
        "script-src 'self' https://esm.sh https://unpkg.com; 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")
    return response


class CreateUserRequest(BaseModel):
    pass


class LoginRequest(BaseModel):
    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:
    if not private_token:
        raise HTTPException(status_code=401, detail="Missing private token")
    user = backend.authenticate_user(user_id, private_token)
    if not user:
        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:
    return FileResponse(FRONTEND_PATH, headers={"Cache-Control": "no-store"})


@app.get("/frontend/{filepath:path}")
def frontend_static(filepath: str) -> FileResponse:
    target = (FRONTEND_DIR / filepath).resolve()
    if FRONTEND_DIR.resolve() not in target.parents and target != FRONTEND_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 = "public, max-age=300, stale-while-revalidate=60" if target.suffix in {".js", ".css"} else "no-store"
    return FileResponse(target, headers={"Cache-Control": cache_control})


@app.get("/media/{bucket}/{filename:path}")
def media(bucket: str, filename: str) -> FileResponse:
    target = (MEDIA_ROOT / bucket / filename).resolve()
    if MEDIA_ROOT.resolve() not in target.parents:
        raise HTTPException(status_code=404, detail="Unknown asset")
    if not target.exists() or not target.is_file():
        raise HTTPException(status_code=404, detail="Unknown asset")
    return FileResponse(target, headers={"Cache-Control": "public, max-age=86400, immutable"})


@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,
        }
    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
    )
    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,
        "stats": b.catalog_stats(),
        "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()}


@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_kinks()
    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}