vocero-client/vocero-api/vocero_api/api/routes/sessions.py

from __future__ import annotations

from typing import Any

import httpx
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse

from vocero_api.api.responses import validated_json_response
from vocero_api.core import config as settings
from vocero_api.core import logging as app_logging
from vocero_api.models.requests import QueueEndPayload
from vocero_api.models.responses import AllocationResponse, QueueCancellationResponse

router = APIRouter()


@router.post("/api/session", response_model=AllocationResponse)
async def session(request: Request) -> JSONResponse:
    """Relay a session grant or queue ticket, keeping the LB address server-side.

    The grant's `connect_url` contains the compute host and short-lived token the
    browser needs to connect directly to the allocated speech service."""
    client_ip = request.client.host if request.client else "?"
    if not settings.LOAD_BALANCER_URL:
        # No LB configured — this deploy is direct-mode only; the browser should
        # never call this. 404 so it's indistinguishable from a missing route.
        raise HTTPException(status_code=404, detail="No encontrado.")

    app_logging.logger.debug("[EXT][LB] session: client=%s", client_ip)
    url = f"{settings.LOAD_BALANCER_URL.rstrip('/')}/session"
    app_logging.logger.debug("[EXT][LB] session POST %s (handshake)", url)
    try:
        async with httpx.AsyncClient(timeout=15.0) as http:
            lb = await http.post(url, headers={"Content-Type": "application/json"}, content="{}")
    except httpx.RequestError as exc:
        app_logging.logger.warning("Load balancer unreachable: %r", exc)
        raise HTTPException(status_code=502, detail="No se puede conectar con el servicio de voz.")  # noqa: B904
    app_logging.logger.debug(
        "[EXT][LB] session RESP status=%s len=%dB",
        lb.status_code,
        len(lb.content or b""),
    )

    # The queue is full: the LB replies 503 {state:"at_capacity"}. Relay it as-is
    # so the client shows a soft "try again shortly", not a hard error.
    if lb.status_code == 503:
        body = _safe_json(lb)
        if body.get("state") == "at_capacity":
            app_logging.logger.debug("[EXT][LB] session: at_capacity — queue full")
            return validated_json_response({"state": "at_capacity"}, AllocationResponse, status_code=503)

    if lb.status_code != 200:
        # The LB's error body may name the reason (e.g. capacity); it carries no
        # secret, so relay a trimmed copy.
        app_logging.logger.warning("Session handshake failed %s: %s", lb.status_code, lb.text[:300])
        raise HTTPException(status_code=502, detail=f"No se pudo establecer la sesión ({lb.status_code}).")

    data = lb.json()
    state = data.get("state", "?")
    connect_url = (data.get("connect_url") or data.get("connectUrl") or "")[:120]
    session_id = data.get("session_id") or data.get("sessionId")
    app_logging.logger.debug(
        "[EXT][LB] session PARSED state=%s sessionId=%s hasConnectUrl=%s connectUrl=%s",
        state,
        session_id,
        bool(connect_url),
        connect_url,
    )

    # Busy pool: the LB queued us. Relay the ticket for subsequent polling.
    if data.get("state") == "queued":
        app_logging.logger.debug(
            "[EXT][LB] session: queued queueId=%s position=%s",
            data.get("queue_id") or data.get("queueId"),
            data.get("position"),
        )
    return validated_json_response(data, AllocationResponse)


@router.get("/api/queue/{queue_id}", response_model=AllocationResponse)
async def queue_status(queue_id: str, request: Request) -> JSONResponse:
    """Relay a waiting ticket's position or its newly claimed session grant."""
    client_ip = request.client.host if request.client else "?"
    if not settings.LOAD_BALANCER_URL:
        raise HTTPException(status_code=404, detail="No encontrado.")

    app_logging.logger.debug(
        "[EXT][LB] queue poll: client=%s queueId=%s",
        client_ip,
        queue_id[:16],
    )

    url = f"{settings.LOAD_BALANCER_URL.rstrip('/')}/queue/{queue_id}"
    app_logging.logger.debug("[EXT][LB] queue GET %s", url)
    try:
        async with httpx.AsyncClient(timeout=15.0) as http:
            lb = await http.get(url)
    except httpx.RequestError as exc:
        app_logging.logger.warning("Load balancer unreachable: %r", exc)
        raise HTTPException(status_code=502, detail="No se puede conectar con el servicio de voz.")  # noqa: B904
    app_logging.logger.debug(
        "[EXT][LB] queue RESP status=%s len=%dB",
        lb.status_code,
        len(lb.content or b""),
    )

    if lb.status_code == 404:
        # Ticket unknown/expired (reaped after we stopped polling). Tell the client
        # to start over rather than spin.
        app_logging.logger.debug("[EXT][LB] queue: queueId=%s EXPIRED (404)", queue_id[:16])
        return validated_json_response({"state": "expired"}, AllocationResponse, status_code=404)

    if lb.status_code != 200:
        app_logging.logger.warning("Queue poll failed %s: %s", lb.status_code, lb.text[:300])
        raise HTTPException(status_code=502, detail=f"No se pudo consultar la cola de espera ({lb.status_code}).")

    data = lb.json()
    state = data.get("state", "?")
    position = data.get("position")
    app_logging.logger.debug(
        "[EXT][LB] queue PARSED queueId=%s state=%s position=%s sessionId=%s",
        queue_id[:16],
        state,
        position,
        data.get("session_id") or data.get("sessionId"),
    )

    return validated_json_response(data, AllocationResponse)


@router.delete("/api/queue/{queue_id}", response_model=QueueCancellationResponse)
async def queue_leave(queue_id: str) -> JSONResponse:
    """Leave the queue from the explicit 'Leave queue' button (a real fetch)."""
    if not settings.LOAD_BALANCER_URL:
        raise HTTPException(status_code=404, detail="No encontrado.")
    app_logging.logger.debug("[EXT][LB] queue LEAVE (button) queueId=%s", queue_id[:16])
    await _lb_leave(queue_id)
    return validated_json_response({"ok": True}, QueueCancellationResponse)


@router.post("/api/queue/end", response_model=QueueCancellationResponse)
async def queue_end(request: Request) -> JSONResponse:
    """Leave the queue on teardown/tab-close (navigator.sendBeacon, which can only
    POST). Body: { queueId }. Best-effort; the LB reaps the ticket on TTL anyway."""
    if not settings.LOAD_BALANCER_URL:
        raise HTTPException(status_code=404, detail="No encontrado.")
    qid = await _queue_id(request)
    if qid:
        app_logging.logger.debug("[EXT][LB] queue LEAVE (teardown) queueId=%s", qid[:16])
        await _lb_leave(qid)
    return validated_json_response({"ok": True}, QueueCancellationResponse)


async def _lb_leave(queue_id: str) -> None:
    """Best-effort: tell the LB to drop a waiting ticket."""
    url = f"{settings.LOAD_BALANCER_URL.rstrip('/')}/queue/{queue_id}"
    app_logging.logger.debug("[EXT][LB] queue DELETE %s", url)
    try:
        async with httpx.AsyncClient(timeout=5.0) as http:
            resp = await http.delete(url)
        app_logging.logger.debug(
            "[EXT][LB] queue DELETE RESP queueId=%s status=%s",
            queue_id[:16],
            resp.status_code,
        )
    except httpx.RequestError as exc:
        app_logging.logger.warning("Queue leave failed: %r", exc)


def _safe_json(response: httpx.Response) -> dict[str, Any]:
    try:
        body = response.json()
    except Exception:
        return {}
    return body if isinstance(body, dict) else {}


async def _queue_id(request: Request) -> Any:
    """Pull `queueId` from a JSON body, tolerating sendBeacon's blob posts."""
    try:
        data = await request.json()
    except Exception:
        return ""
    return QueueEndPayload.model_validate(data).queueId if isinstance(data, dict) else ""