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 ""