refactor(api): scope uv to s2s-client and validate compatible responses

2026-10-06 18:23:08 -03:00
parent 92c53dc7ca
commit f8d5b7857a
39 changed files with 316 additions and 67 deletions
@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import JSONResponse
+
+from s2s_client.api.responses import validated_json_response
+from s2s_client.core import config as settings
+from s2s_client.core import logging as app_logging
+from s2s_client.models.requests import QueueEndPayload
+from s2s_client.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 ""