stork/stork/app.py

"""Stork HTTP API + shareable multi-board family UI."""

from __future__ import annotations

import os
import secrets
from pathlib import Path
from typing import Any

from fastapi import Cookie, Depends, FastAPI, File, Header, HTTPException, Request, Response, UploadFile
from fastapi.responses import FileResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field

from stork.audio import ext_for_content_type, sniff_audio
from stork.db import LANGS, MAX_COLUMNS, Store
from stork.rate_limit import limiter

DATA_DIR = Path(os.environ.get("STORK_DATA", "./data"))
INVITE_TOKEN = os.environ.get("STORK_INVITE_TOKEN", "").strip()
ADMIN_TOKEN = os.environ.get("STORK_ADMIN_TOKEN", "").strip()
COOKIE_SECURE = os.environ.get("STORK_COOKIE_SECURE", "false").lower() in {"1", "true", "yes"}
PUBLIC_BASE = os.environ.get("STORK_PUBLIC_URL", "https://stork.levkin.ca").rstrip("/")
READONLY_BOARD_IDS = {
    x.strip()
    for x in os.environ.get("STORK_READONLY_BOARD_IDS", "").split(",")
    if x.strip()
}


def _env_int(name: str, default: int) -> int:
    raw = os.environ.get(name, "").strip()
    if not raw:
        return default
    try:
        return max(1, int(raw))
    except ValueError:
        return default


# Recording upload hard cap (bytes read into memory).
MAX_UPLOAD_BYTES = _env_int("STORK_MAX_UPLOAD_BYTES", 5 * 1024 * 1024)
# Empty boards older than this are deleted opportunistically on create.
ORPHAN_BOARD_HOURS = _env_int("STORK_ORPHAN_BOARD_HOURS", 48)
STATIC = Path(__file__).resolve().parent.parent / "static"

ALLOWED_AUDIO = {
    "audio/webm",
    "audio/ogg",
    "audio/mp4",
    "audio/mpeg",
    "audio/wav",
    "audio/x-m4a",
    "audio/aac",
    "video/webm",
}

app = FastAPI(title="Stork", docs_url=None, redoc_url=None)
store = Store(
    DATA_DIR / "stork.sqlite3",
    bootstrap_invite=INVITE_TOKEN,
    bootstrap_title="Family",
)

if STATIC.is_dir():
    app.mount("/static", StaticFiles(directory=STATIC), name="static")


class LocaleIn(BaseModel):
    pronunciation: str = ""
    origin: str = ""
    meaning: str = ""


class NameIn(BaseModel):
    kind: str = Field(pattern=r"^c[1-4]$")
    spelling: str = Field(min_length=1, max_length=80)
    locales: dict[str, LocaleIn] = Field(default_factory=dict)


class ColumnPatch(BaseModel):
    label: str = Field(default="", max_length=40)


class LocalesPatch(BaseModel):
    locales: dict[str, LocaleIn]


class VoteIn(BaseModel):
    value: int = Field(ge=-1, le=1)


class SessionIn(BaseModel):
    display_name: str = Field(min_length=1, max_length=80)
    board_id: str | None = Field(default=None, max_length=64)
    invite: str | None = Field(default=None, max_length=200)


class BoardCreatePublic(BaseModel):
    title: str = Field(default="Name board", min_length=1, max_length=80)


class BoardCreate(BaseModel):
    title: str = Field(min_length=1, max_length=80)
    invite: str | None = Field(default=None, max_length=200)


class BoardPatch(BaseModel):
    title: str = Field(min_length=1, max_length=80)


def _admin(
    x_stork_admin: str | None = Header(default=None),
) -> None:
    if not ADMIN_TOKEN:
        raise HTTPException(503, "Admin not configured")
    if not x_stork_admin or not secrets.compare_digest(x_stork_admin, ADMIN_TOKEN):
        raise HTTPException(403, "Admin token required")


def _client_ip(request: Request) -> str:
    forwarded = (request.headers.get("x-forwarded-for") or "").split(",")[0].strip()
    if forwarded:
        return forwarded
    if request.client and request.client.host:
        return request.client.host
    return "unknown"


def _rate_limit(request: Request, action: str) -> None:
    allowed, retry_after = limiter.check(action, _client_ip(request))
    if not allowed:
        raise HTTPException(
            429,
            "Too many requests — try again later",
            headers={"Retry-After": str(retry_after)},
        )


def _is_readonly(board_id: str) -> bool:
    return board_id in READONLY_BOARD_IDS


def _assert_writable(board_id: str) -> None:
    if _is_readonly(board_id):
        raise HTTPException(403, "This board is read-only")


def _board_public(board: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": board["id"],
        "title": board["title"],
        "url": _board_url(board["id"]),
        "readonly": _is_readonly(board["id"]),
    }


def _resolve_board(
    board_id: str | None = None,
    invite: str | None = None,
) -> dict[str, Any]:
    board_id = (board_id or "").strip()
    invite = (invite or "").strip()
    if board_id:
        board = store.find_board(board_id)
        if board:
            return board
    if invite:
        board = store.find_board_by_invite(invite)
        if board:
            return board
    raise HTTPException(404, "Board not found")


def _session(
    stork_board: str | None = Cookie(default=None),
    stork_invite: str | None = Cookie(default=None),
    stork_voter: str | None = Cookie(default=None),
    stork_name: str | None = Cookie(default=None),
    x_stork_board: str | None = Header(default=None),
    x_stork_invite: str | None = Header(default=None),
    x_stork_display_name: str | None = Header(default=None),
    x_stork_voter: str | None = Header(default=None),
) -> dict[str, str]:
    board = _resolve_board(
        board_id=x_stork_board or stork_board,
        invite=x_stork_invite or stork_invite,
    )
    voter = (x_stork_voter or stork_voter or "").strip()
    name = (x_stork_display_name or stork_name or "").strip()
    if not voter or not name:
        raise HTTPException(401, "Session required — enter your name on this board")
    return {
        "voter_key": voter,
        "display_name": name,
        "board_id": board["id"],
        "board_title": board["title"],
    }


def _board_url(board_id: str) -> str:
    return f"{PUBLIC_BASE}/b/{board_id}"


def _cookie_kwargs() -> dict[str, Any]:
    return {
        "httponly": True,
        "samesite": "lax",
        "secure": COOKIE_SECURE,
        "max_age": 60 * 60 * 24 * 400,
        "path": "/",
    }


@app.get("/api/health")
def health() -> dict[str, Any]:
    return {"ok": True, "service": "stork"}


@app.get("/api/meta")
def meta() -> dict[str, Any]:
    return {
        "langs": list(LANGS),
        "max_columns": MAX_COLUMNS,
        "multi_board": True,
    }


def _protected_board_ids() -> set[str]:
    protect = set(READONLY_BOARD_IDS)
    if INVITE_TOKEN:
        family = store.find_board_by_invite(INVITE_TOKEN)
        if family:
            protect.add(family["id"])
    return protect


def _gc_orphan_boards() -> None:
    store.gc_empty_boards(
        older_than_seconds=float(ORPHAN_BOARD_HOURS) * 3600.0,
        protect_ids=_protected_board_ids(),
    )


@app.post("/api/boards")
def create_board_public(body: BoardCreatePublic, request: Request) -> dict[str, Any]:
    """Start a new board — no password. Share the returned URL with family."""
    _rate_limit(request, "create_board")
    _gc_orphan_boards()
    board = store.create_board(body.title.strip() or "Name board")
    return _board_public(board)


@app.get("/api/boards/{board_id}")
def get_board_public(board_id: str) -> dict[str, Any]:
    board = store.find_board(board_id)
    if not board:
        raise HTTPException(404, "Board not found")
    return _board_public(board)


@app.get("/api/resolve")
def resolve_invite(invite: str = "") -> dict[str, Any]:
    """Map a legacy invite token to a board URL."""
    board = store.find_board_by_invite(invite)
    if not board:
        raise HTTPException(404, "Board not found")
    return _board_public(board)


@app.post("/api/session")
def create_session(body: SessionIn, response: Response, request: Request) -> dict[str, Any]:
    _rate_limit(request, "session")
    board = _resolve_board(board_id=body.board_id, invite=body.invite)
    voter = secrets.token_urlsafe(16)
    display = body.display_name.strip()
    kwargs = _cookie_kwargs()
    response.set_cookie("stork_board", board["id"], **kwargs)
    response.set_cookie("stork_voter", voter, **kwargs)
    response.set_cookie("stork_name", display, **{**kwargs, "httponly": False})
    response.delete_cookie("stork_invite", path="/")
    return {
        "ok": True,
        "display_name": display,
        "board": _board_public(board),
    }


@app.get("/api/session")
def get_session(
    stork_board: str | None = Cookie(default=None),
    stork_invite: str | None = Cookie(default=None),
    stork_voter: str | None = Cookie(default=None),
    stork_name: str | None = Cookie(default=None),
    x_stork_board: str | None = Header(default=None),
    x_stork_invite: str | None = Header(default=None),
) -> dict[str, Any]:
    board = None
    try:
        board = _resolve_board(
            board_id=x_stork_board or stork_board,
            invite=x_stork_invite or stork_invite,
        )
    except HTTPException:
        board = None
    authed = bool(board) and bool(stork_voter) and bool(stork_name)
    return {
        "authenticated": authed,
        "display_name": stork_name or "",
        "board": (_board_public(board) if board else None),
    }


@app.delete("/api/session")
def clear_session(response: Response) -> dict[str, Any]:
    response.delete_cookie("stork_board", path="/")
    response.delete_cookie("stork_voter", path="/")
    response.delete_cookie("stork_name", path="/")
    response.delete_cookie("stork_invite", path="/")
    return {"ok": True}


@app.get("/api/columns")
def list_columns(session: dict[str, str] = Depends(_session)) -> dict[str, Any]:
    return {"items": store.list_columns(session["board_id"]), "max": MAX_COLUMNS}


@app.post("/api/columns")
def add_column(session: dict[str, str] = Depends(_session)) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    try:
        return store.add_column(session["board_id"], "")
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc


@app.patch("/api/columns/{column_id}")
def rename_column(
    column_id: str,
    body: ColumnPatch,
    session: dict[str, str] = Depends(_session),
) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    updated = store.rename_column(session["board_id"], column_id, body.label)
    if not updated:
        raise HTTPException(404, "Column not found")
    return updated


@app.delete("/api/columns/{column_id}")
def delete_column(
    column_id: str,
    session: dict[str, str] = Depends(_session),
) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    try:
        if not store.delete_column(session["board_id"], column_id):
            raise HTTPException(404, "Column not found")
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    return {"ok": True, "items": store.list_columns(session["board_id"])}


@app.get("/api/names")
def list_names(kind: str = "c1", session: dict[str, str] = Depends(_session)) -> dict[str, Any]:
    try:
        return {
            "items": store.list_names(
                session["board_id"], kind, voter_key=session["voter_key"]
            )
        }
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc


@app.post("/api/names")
def add_name(body: NameIn, session: dict[str, str] = Depends(_session)) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    locales = {k: v.model_dump() for k, v in body.locales.items()}
    try:
        return store.add_name(
            board_id=session["board_id"],
            kind=body.kind,
            spelling=body.spelling,
            created_by=session["display_name"],
            locales=locales,
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc


@app.patch("/api/names/{name_id}")
def patch_name(
    name_id: int,
    body: LocalesPatch,
    session: dict[str, str] = Depends(_session),
) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    locales = {k: v.model_dump() for k, v in body.locales.items()}
    updated = store.update_locales(name_id, locales, board_id=session["board_id"])
    if not updated:
        raise HTTPException(404, "Name not found")
    return updated


@app.post("/api/names/{name_id}/vote")
def vote_name(
    name_id: int,
    body: VoteIn,
    session: dict[str, str] = Depends(_session),
) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    try:
        updated = store.vote(
            name_id=name_id,
            voter_key=session["voter_key"],
            voter_name=session["display_name"],
            value=body.value,
            board_id=session["board_id"],
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    if not updated:
        raise HTTPException(404, "Name not found")
    return updated


@app.post("/api/names/{name_id}/recording/{lang}")
async def upload_recording(
    name_id: int,
    lang: str,
    request: Request,
    session: dict[str, str] = Depends(_session),
    file: UploadFile = File(...),
) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    _rate_limit(request, "upload")
    if lang not in LANGS:
        raise HTTPException(400, "lang must be en, ru, or he")
    raw = await file.read(MAX_UPLOAD_BYTES + 1)
    if len(raw) > MAX_UPLOAD_BYTES:
        raise HTTPException(413, f"recording too large (max {MAX_UPLOAD_BYTES} bytes)")
    sniffed = sniff_audio(raw)
    if not sniffed:
        raise HTTPException(400, "unrecognized audio format")
    claimed = (file.content_type or sniffed).split(";")[0].strip().lower()
    if claimed not in ALLOWED_AUDIO and claimed != sniffed:
        raise HTTPException(400, f"unsupported audio type: {claimed}")
    content_type = sniffed
    ext = ext_for_content_type(content_type)
    try:
        updated = store.save_recording(
            name_id=name_id,
            lang=lang,
            data=raw,
            content_type=content_type,
            recorded_by=session["display_name"],
            ext=ext,
            board_id=session["board_id"],
        )
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    if not updated:
        raise HTTPException(404, "Name not found")
    return updated


@app.get("/api/names/{name_id}/recording/{lang}")
def get_recording(
    name_id: int,
    lang: str,
    session: dict[str, str] = Depends(_session),
) -> FileResponse:
    if lang not in LANGS:
        raise HTTPException(400, "lang must be en, ru, or he")
    found = store.get_recording(name_id, lang, board_id=session["board_id"])
    if not found:
        raise HTTPException(404, "Recording not found")
    path, content_type = found
    return FileResponse(path, media_type=content_type, filename=path.name)


@app.delete("/api/names/{name_id}/recording/{lang}")
def delete_recording(
    name_id: int,
    lang: str,
    session: dict[str, str] = Depends(_session),
) -> dict[str, Any]:
    _assert_writable(session["board_id"])
    try:
        updated = store.delete_recording(name_id, lang, board_id=session["board_id"])
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    if not updated:
        raise HTTPException(404, "Name not found")
    return updated


@app.delete("/api/names/{name_id}")
def delete_name(name_id: int, _: None = Depends(_admin)) -> dict[str, Any]:
    if not store.delete_name(name_id):
        raise HTTPException(404, "Name not found")
    return {"ok": True}


@app.get("/api/admin/boards")
def admin_list_boards(_: None = Depends(_admin)) -> dict[str, Any]:
    items = []
    for b in store.list_boards(include_invite=True):
        items.append(
            {
                **_board_public(b),
                "created_at": b["created_at"],
            }
        )
    return {"items": items}


@app.post("/api/admin/boards")
def admin_create_board(body: BoardCreate, _: None = Depends(_admin)) -> dict[str, Any]:
    try:
        board = store.create_board(body.title, body.invite)
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    return {**_board_public(board), "created_at": board["created_at"]}


@app.patch("/api/admin/boards/{board_id}")
def admin_rename_board(
    board_id: str,
    body: BoardPatch,
    _: None = Depends(_admin),
) -> dict[str, Any]:
    try:
        updated = store.rename_board(board_id, body.title)
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    if not updated:
        raise HTTPException(404, "Board not found")
    return {**_board_public(updated), "created_at": updated["created_at"]}


@app.delete("/api/admin/boards/{board_id}")
def admin_delete_board(board_id: str, _: None = Depends(_admin)) -> dict[str, Any]:
    try:
        if not store.delete_board(board_id):
            raise HTTPException(404, "Board not found")
    except ValueError as exc:
        raise HTTPException(400, str(exc)) from exc
    return {"ok": True}


def _ui_page() -> FileResponse:
    page = STATIC / "v2.html"
    if not page.is_file():
        raise HTTPException(404, "UI missing")
    return FileResponse(page)


@app.get("/")
def home() -> FileResponse:
    return _ui_page()


@app.get("/b/{board_id}")
def board_page(board_id: str) -> FileResponse:
    if not store.find_board(board_id):
        raise HTTPException(404, "Board not found")
    return _ui_page()


@app.get("/v1")
def v1() -> RedirectResponse:
    return RedirectResponse(url="/", status_code=302)


@app.get("/v2")
def v2() -> RedirectResponse:
    return RedirectResponse(url="/", status_code=302)


@app.get("/logos")
def logos() -> FileResponse:
    page = STATIC / "logos.html"
    if not page.is_file():
        raise HTTPException(404, "Logo gallery missing")
    return FileResponse(page)