From 8066e9d0556ecea97ca94ca086e6cbc5cc9b9016 Mon Sep 17 00:00:00 2001 From: mjalmoudy94 Date: Tue, 13 Oct 2026 13:54:56 +0330 Subject: [PATCH] feat: remove public REST API; route all app calls over WebSocket RPC The platform no longer exposes a public /api/* REST surface. The original HTTP route handlers are preserved verbatim as the single source of truth inside an internal, never-socket-bound FastAPI app (rpc_app), and are now reached only via RPC-over-WebSocket dispatch. - realtime router: anonymous connections allowed only to /api/auth/login and /api/auth/signup; all other RPCs/subscribes refused until token auth. - RPC dispatch re-injects the bearer token and X-Request-ID, then forwards through httpx.ASGITransport against the internal app (no network hop). - Public app now serves only /, /health, /health/ready and /api/ws; docs/openapi disabled. - Frontend: apiRequest routes /api/* to ws.rpc(); /health probes keep fetch. RealtimeClient gains rpc() with id correlation, timeout and AbortSignal. - Tests repointed to get_rpc_app(); anonymous-auth behaviour covered. Result: zero REST on the public surface; 264 backend tests + frontend typecheck pass. --- apps/api/app/interfaces/realtime/manager.py | 7 +- apps/api/app/interfaces/realtime/router.py | 201 +++++++++++++--- apps/api/app/interfaces/realtime/rpc_app.py | 79 ++++++ apps/api/app/main.py | 45 ++-- apps/api/tests/conftest.py | 6 +- apps/api/tests/test_error_log.py | 6 +- apps/api/tests/test_realtime.py | 50 +++- apps/web/src/lib/api.ts | 31 ++- apps/web/src/lib/realtime.tsx | 11 +- apps/web/src/lib/ws.ts | 251 ++++++++++++++++---- apps/web/src/test/api.test.ts | 147 +++++------- packages/contracts/src/index.ts | 19 +- 12 files changed, 631 insertions(+), 222 deletions(-) create mode 100644 apps/api/app/interfaces/realtime/rpc_app.py diff --git a/apps/api/app/interfaces/realtime/manager.py b/apps/api/app/interfaces/realtime/manager.py index fb9fc54..0758943 100644 --- a/apps/api/app/interfaces/realtime/manager.py +++ b/apps/api/app/interfaces/realtime/manager.py @@ -48,11 +48,14 @@ ChannelAuthorizer = Callable[["Connection", str], Awaitable[bool]] #: both its connection set and its channel routing table. @dataclass(slots=True, eq=False) class Connection: - """One authenticated browser socket.""" + """One browser socket. ``user_id`` is ``None`` for an anonymous (pre-login) + connection: it may only call the bootstrap RPCs (login/signup) and may not + subscribe to live channels.""" - user_id: str + user_id: str | None is_platform_owner: bool websocket: WebSocket + token: str | None = None channels: set[str] = field(default_factory=set) _queue: asyncio.Queue[str] = field( default_factory=lambda: asyncio.Queue(maxsize=MAX_QUEUE_DEPTH) diff --git a/apps/api/app/interfaces/realtime/router.py b/apps/api/app/interfaces/realtime/router.py index 6a4d91b..25d9256 100644 --- a/apps/api/app/interfaces/realtime/router.py +++ b/apps/api/app/interfaces/realtime/router.py @@ -1,39 +1,48 @@ -"""The WebSocket endpoint — one socket per browser, every live feed on it. +"""The WebSocket endpoint — one socket per browser, every live feed and every +RPC on it. ``GET /api/ws?token=`` The token travels as a query parameter because the browser WebSocket API cannot -set an ``Authorization`` header. That is a deliberate, contained exception to -the header-only rule: the token is short-lived, the socket is TLS-terminated in -any real deployment, and the alternative (a cookie) would drag CSRF into a -protocol that has no CSRF story. +set an ``Authorization`` header. That is a deliberate, contained exception to the +header-only rule: the token is short-lived, the socket is TLS-terminated in any +real deployment, and the alternative (a cookie) would drag CSRF into a protocol +that has no CSRF story. -Protocol — small, symmetric, JSON in both directions: +Protocol — small, symmetric, JSON in both directions. Server->client frames +share the ``type`` discriminator; client->server frames use ``action``:: -.. code-block:: text - - client -> {"action": "subscribe", "channels": ["org:…", "task:…"]} + client -> {"action": "subscribe", "channels": ["org:…"]} client -> {"action": "unsubscribe", "channels": ["task:…"]} client -> {"action": "ping"} - server -> {"type": "hello", "user_id": "…"} - server -> {"type": "subscribed", "accepted": [...], "rejected": [...]} - server -> {"type": "event", "event": {…envelope…}} + client -> {"action": "rpc", "id": "…", "method": "POST", + "path": "/api/organizations", "body": {"name": "…"}} + server -> {"type": "hello", "user_id": "…"|null} + server -> {"type": "subscribed", "accepted": [...], "rejected": [...]} + server -> {"type": "event", "event": {…envelope…}} + server -> {"type": "rpc_result", "id": "…", "status": 200, "body": …} server -> {"type": "pong"} - server -> {"type": "error", "message": "…"} + server -> {"type": "error", "message": "…"} -A rejected channel is reported, not hidden: "my task page is not live" must be -answerable from the browser console alone. +A connection may be **anonymous** (no token): that is the bootstrap state that +lets a brand-new visitor call ``/api/auth/login`` and ``/api/auth/signup`` over +RPC. Every other RPC and every ``subscribe`` requires an authenticated +connection; the token presented at connect time is re-injected as a bearer +header when the RPC is dispatched into the internal app, so the handlers' own +auth dependencies still run unchanged. -Authorization queries open their own short-lived session: a socket can stay -open for hours, and one SQLAlchemy session per open tab would quietly exhaust -the connection pool. +Authorization queries open their own short-lived session: a socket can stay open +for hours, and one SQLAlchemy session per open tab would quietly exhaust the +connection pool. """ from __future__ import annotations +import asyncio import json import logging +import httpx from fastapi import APIRouter, Query, WebSocket, WebSocketDisconnect from app.application.ports.realtime import dedupe_channels @@ -58,6 +67,27 @@ def realtime_manager() -> ConnectionManager: return _manager +#: Paths an anonymous (not-yet-logged-in) connection is allowed to call. These are +#: the only two calls a visitor can make; everything else needs a session. +_ANONYMOUS_RPC_PATHS = frozenset({"/api/auth/login", "/api/auth/signup"}) + +#: One in-process HTTP client that dispatches RPCs into the internal app. Built +#: lazily so importing this module never pulls the app into existence. +_rpc_client: httpx.AsyncClient | None = None + + +def _get_rpc_client() -> httpx.AsyncClient: + global _rpc_client + if _rpc_client is None: + # The internal app holds every HTTP handler; it is never socket-bound, so + # it is reached here via an in-process ASGI transport instead of the wire. + from app.interfaces.realtime.rpc_app import get_rpc_app + + transport = httpx.ASGITransport(app=get_rpc_app()) + _rpc_client = httpx.AsyncClient(transport=transport, base_url="http://internal") + return _rpc_client + + @router.websocket("/ws") async def realtime_socket( websocket: WebSocket, @@ -69,15 +99,15 @@ async def realtime_socket( await _manager.register(connection) - await _send( - websocket, + _send_frame( + connection, {"type": "hello", "user_id": connection.user_id, "server_time": _now_iso()}, ) try: while True: raw = await websocket.receive_text() - await _handle_message(websocket, connection, raw) + await _handle_message(connection, raw) except WebSocketDisconnect: pass except Exception: # noqa: BLE001 - any read failure ends the socket cleanly @@ -87,8 +117,16 @@ async def realtime_socket( async def _authenticate(websocket: WebSocket, token: str | None) -> Connection | None: - """Accept only a valid, active user. Closes 1008 otherwise.""" + """Accept an anonymous or an authenticated, active user. + + * No token -> an anonymous connection (bootstrap for login/signup). + * A token that decodes to a real, active user -> an authenticated connection + whose raw token is kept for RPC re-injection. + * A presented-but-invalid token -> 1008, so the client can tell "bad token" + from "network dropped" and stop retrying. + """ user = None + raw_token: str | None = None if token: try: payload = decode_access_token(token) @@ -97,35 +135,48 @@ async def _authenticate(websocket: WebSocket, token: str | None) -> Connection | # Own session: the socket must not depend on request scope. async with SessionFactory() as session: user = await UserRepository(session).get(str(user_id)) + if user is not None and user.is_active: + raw_token = token except AuthenticationError: user = None - if user is None or not user.is_active: - # 1008 (policy violation) rather than a bare close: the client's - # reconnect logic can then tell "bad token" from "network dropped". + + if token and user is None: await websocket.close(code=1008, reason="unauthorized") return None + await websocket.accept() return Connection( - user_id=user.id, - is_platform_owner=user.is_platform_owner, + user_id=user.id if user else None, + is_platform_owner=user.is_platform_owner if user else False, + token=raw_token, websocket=websocket, ) -async def _handle_message(websocket: WebSocket, connection: Connection, raw: str) -> None: +async def _handle_message(connection: Connection, raw: str) -> None: try: message = json.loads(raw) except (TypeError, ValueError): - await _send(websocket, {"type": "error", "message": "malformed JSON frame"}) + _send_frame(connection, {"type": "error", "message": "malformed JSON frame"}) return if not isinstance(message, dict): - await _send(websocket, {"type": "error", "message": "frame must be an object"}) + _send_frame(connection, {"type": "error", "message": "frame must be an object"}) return action = message.get("action") if action == "ping": - await _send(websocket, {"type": "pong"}) + _send_frame(connection, {"type": "pong"}) + return + + if action == "rpc": + # Run off the receive loop: a slow handler must not stall pings. + asyncio.create_task(_handle_rpc(connection, message)) + return + + # subscribe / unsubscribe require a logged-in user. + if connection.user_id is None: + _send_frame(connection, {"type": "error", "message": "authentication required"}) return channels = dedupe_channels( @@ -134,17 +185,85 @@ async def _handle_message(websocket: WebSocket, connection: Connection, raw: str if action == "subscribe": accepted, rejected = await _subscribe(connection, channels) - await _send( - websocket, {"type": "subscribed", "accepted": accepted, "rejected": rejected} + _send_frame( + connection, {"type": "subscribed", "accepted": accepted, "rejected": rejected} ) return if action == "unsubscribe": await _manager.unsubscribe(connection, list(channels)) - await _send(websocket, {"type": "unsubscribed", "channels": channels}) + _send_frame(connection, {"type": "unsubscribed", "channels": channels}) return - await _send(websocket, {"type": "error", "message": f"unknown action: {action!r}"}) + _send_frame(connection, {"type": "error", "message": f"unknown action: {action!r}"}) + + +async def _handle_rpc(connection: Connection, message: dict) -> None: + """Dispatch one client RPC into the internal app and return the result.""" + req_id = message.get("id") + if not req_id: + _send_frame(connection, {"type": "error", "message": "rpc frame needs an id"}) + return + + method = str(message.get("method") or "GET").upper() + path = message.get("path") + if not isinstance(path, str) or not path: + _send_rpc_result( + connection, + req_id, + 400, + {"error": {"code": "bad_request", "message": "rpc frame needs a path"}}, + ) + return + + # Gate unauthenticated callers to login/signup only. + if connection.user_id is None and path not in _ANONYMOUS_RPC_PATHS: + _send_rpc_result( + connection, + req_id, + 401, + {"error": {"code": "unauthenticated", "message": "نیاز به ورود دارید."}}, + ) + return + + body = message.get("body") + headers = {"X-Request-ID": str(req_id), "User-Agent": "realtime-rpc"} + if connection.token: + headers["Authorization"] = f"Bearer {connection.token}" + + try: + client = _get_rpc_client() + if isinstance(body, (dict, list)): + response = await client.request(method, path, json=body, headers=headers) + elif body is None: + response = await client.request(method, path, headers=headers) + else: + response = await client.request(method, path, content=str(body), headers=headers) + except Exception: # noqa: BLE001 - reported to the caller as a 500 envelope + logger.exception("realtime: rpc dispatch to internal app failed") + _send_rpc_result( + connection, + req_id, + 500, + {"error": {"code": "internal_error", "message": "خطای داخلی سرور."}}, + ) + return + + status = response.status_code + try: + payload = response.json() + except Exception: + payload = None + _send_rpc_result(connection, req_id, status, payload) + + +def _send_rpc_result( + connection: Connection, req_id: object, status: int, body: object +) -> None: + _send_frame( + connection, + {"type": "rpc_result", "id": req_id, "status": status, "body": body}, + ) async def _subscribe( @@ -157,7 +276,7 @@ async def _subscribe( async with SessionFactory() as session: return await authorize_channel( session, - user_id=conn.user_id, + user_id=conn.user_id or "", is_platform_owner=conn.is_platform_owner, channel=channel, ) @@ -171,5 +290,11 @@ def _now_iso() -> str: return utcnow().isoformat() -async def _send(websocket: WebSocket, payload: dict) -> None: - await websocket.send_text(json.dumps(payload, ensure_ascii=False, default=str)) +def _send_frame(connection: Connection, payload: dict) -> None: + """Queue a frame on the connection's single-writer outbox. + + Every frame — control, event and rpc_result — goes through the same queue so + a slow RPC task can never interleave bytes with the receive loop's own + frames and corrupt the stream. + """ + connection.offer(json.dumps(payload, ensure_ascii=False, default=str)) diff --git a/apps/api/app/interfaces/realtime/rpc_app.py b/apps/api/app/interfaces/realtime/rpc_app.py new file mode 100644 index 0000000..11b8ae2 --- /dev/null +++ b/apps/api/app/interfaces/realtime/rpc_app.py @@ -0,0 +1,79 @@ +"""Internal, never-socket-bound FastAPI app that holds every HTTP handler. + +The public app (``app.main.create_app``) exposes only ``/health`` and the +``/api/ws`` WebSocket. All request/response logic lives here and is reached in +exactly one of two ways: + +* from the WS-RPC proxy (``app.interfaces.realtime.router``), which dispatches + client RPCs into this app over an in-process ASGI transport, and +* from the test-suite's ``AsyncClient``/``TestClient``. + +It is intentionally never bound to a socket, so including the routers here adds no +public REST surface — the only public routes are the ones ``create_app`` +registers. Keeping the handlers in one FastAPI app also means they run through the +real middleware, dependencies, validation and error handling, unchanged. +""" + +from __future__ import annotations + +from fastapi import FastAPI + +from app.core.config import settings +from app.interfaces.http.errors import register_exception_handlers +from app.interfaces.http.request_context import RequestContextMiddleware +from app.interfaces.http.routers import ( + admin, + agy, + auth, + billing, + health, + knowledge, + organizations, + projects, + sandbox, + workforce, +) +from app.interfaces.realtime.router import router as realtime_router +from app.main import lifespan # shared lifespan installs the realtime bus + +#: Module-level singleton so the RPC proxy and the tests share one app instance. +_rpc_app: FastAPI | None = None + + +def build_rpc_app() -> FastAPI: + """Construct the full handler app. Called once per process.""" + application = FastAPI( + title=settings.app_name, + version="0.1.0", + description="AI Workforce Platform — internal RPC dispatch surface.", + lifespan=lifespan, + # This app is reached in-process, never over the network, so the + # interactive docs would only be a liability. + docs_url=None, + openapi_url=None, + ) + application.add_middleware(RequestContextMiddleware) + register_exception_handlers(application) + + # Same mounting as the old public app: health at the root, everything else + # under /api, and the live socket alongside the RPC surface. + application.include_router(health.router) + application.include_router(auth.router, prefix="/api") + application.include_router(organizations.router, prefix="/api") + application.include_router(projects.router, prefix="/api") + application.include_router(workforce.router, prefix="/api") + application.include_router(billing.router, prefix="/api") + application.include_router(knowledge.router, prefix="/api") + application.include_router(sandbox.router, prefix="/api") + application.include_router(admin.router, prefix="/api") + application.include_router(agy.router, prefix="/api") + application.include_router(realtime_router, prefix="/api") + return application + + +def get_rpc_app() -> FastAPI: + """Return the process-wide internal app, building it on first use.""" + global _rpc_app + if _rpc_app is None: + _rpc_app = build_rpc_app() + return _rpc_app diff --git a/apps/api/app/main.py b/apps/api/app/main.py index ba4bc41..91ccade 100644 --- a/apps/api/app/main.py +++ b/apps/api/app/main.py @@ -15,18 +15,7 @@ from app.core.config import settings from app.core.database import Base, engine from app.interfaces.http.errors import register_exception_handlers from app.interfaces.http.request_context import RequestContextMiddleware -from app.interfaces.http.routers import ( - admin, - agy, - auth, - billing, - health, - knowledge, - organizations, - projects, - sandbox, - workforce, -) +from app.interfaces.http.routers import health from app.interfaces.realtime.bus import install as install_realtime_bus from app.interfaces.realtime.bus import start_relay, stop_relay from app.interfaces.realtime.router import realtime_manager @@ -99,13 +88,22 @@ async def _record_startup_failure(exc: BaseException) -> None: def create_app() -> FastAPI: + """Build the *public* app. + + This is the only app uvicorn binds. It deliberately exposes no REST API: the + application surface is reachable only over the ``/api/ws`` WebSocket (live + feeds + RPC). ``/health`` stays as an HTTP liveness probe because deploy + orchestration and the browser's status check both need a plain GET, and a + probe is not an application endpoint. + """ application = FastAPI( title=settings.app_name, version="0.1.0", description="AI Workforce Platform — standalone multi-tenant SaaS.", lifespan=lifespan, - docs_url="/docs", - openapi_url="/openapi.json", + # No REST API on the public surface, so the interactive docs are off. + docs_url=None, + openapi_url=None, ) # Starlette applies middleware outermost-first in reverse registration @@ -128,17 +126,10 @@ def create_app() -> FastAPI: register_exception_handlers(application) + # The only public routes: a liveness probe and the single WebSocket that + # carries every application call (live events + RPC). All HTTP handlers live + # in the internal app reached through that socket. application.include_router(health.router) - application.include_router(auth.router, prefix="/api") - application.include_router(organizations.router, prefix="/api") - application.include_router(projects.router, prefix="/api") - application.include_router(workforce.router, prefix="/api") - application.include_router(billing.router, prefix="/api") - application.include_router(knowledge.router, prefix="/api") - application.include_router(sandbox.router, prefix="/api") - application.include_router(admin.router, prefix="/api") - application.include_router(agy.router, prefix="/api") - # The live feed: one WebSocket per browser, every push on it. application.include_router(realtime_router, prefix="/api") @application.get("/", tags=["meta"]) @@ -146,8 +137,12 @@ def create_app() -> FastAPI: return { "name": settings.app_name, "version": "0.1.0", - "docs": "/docs", "health": "/health", + "websocket": "/api/ws", + "note": ( + "All application calls travel over the WebSocket at /api/ws " + "(live feeds + RPC). There is no REST API on this server." + ), } return application diff --git a/apps/api/tests/conftest.py b/apps/api/tests/conftest.py index 358471f..4df513e 100644 --- a/apps/api/tests/conftest.py +++ b/apps/api/tests/conftest.py @@ -24,7 +24,11 @@ from app.core.config import settings from app.core.database import Base, get_session from app.domain.base import new_id from app.infrastructure.db.models import UserRow -from app.main import app +# The internal "full" app holds every HTTP handler; the public app (app.main.app) +# now serves only /health + /api/ws, so REST tests must target the full app. +from app.interfaces.realtime.rpc_app import get_rpc_app + +app = get_rpc_app() @pytest.fixture(scope="session", autouse=True) diff --git a/apps/api/tests/test_error_log.py b/apps/api/tests/test_error_log.py index 6fab4a1..de0d221 100644 --- a/apps/api/tests/test_error_log.py +++ b/apps/api/tests/test_error_log.py @@ -22,7 +22,11 @@ from app.core.errors import ProviderUnavailableError from app.infrastructure.db.models import ErrorEventRow from app.interfaces.http.deps import OrgContext, get_org_context from app.interfaces.http.request_context import REQUEST_ID_HEADER -from app.main import app +# The internal "full" app holds every HTTP handler; the public app serves only +# /health + /api/ws, so this module's REST exercises must target the full app. +from app.interfaces.realtime.rpc_app import get_rpc_app + +app = get_rpc_app() from tests.conftest import auth_header, signup #: The password in this DSN must never reach the database. diff --git a/apps/api/tests/test_realtime.py b/apps/api/tests/test_realtime.py index aaa19a9..3907a93 100644 --- a/apps/api/tests/test_realtime.py +++ b/apps/api/tests/test_realtime.py @@ -32,7 +32,11 @@ from app.infrastructure.db.models import UserRow from app.interfaces.realtime.authz import authorize_channel from app.interfaces.realtime.bus import get_bus from app.interfaces.realtime.manager import Connection, ConnectionManager -from app.main import app +# The internal "full" app holds every HTTP handler; the public app serves only +# /health + /api/ws. The socket and the RPC surface both live on the full app. +from app.interfaces.realtime.rpc_app import get_rpc_app + +app = get_rpc_app() API_ROOT = Path(__file__).resolve().parents[1] @@ -331,10 +335,46 @@ def _frames_until(websocket, wanted_type: str, *, max_frames: int = 20) -> dict: raise AssertionError(f"no {wanted_type} event arrived") -def test_socket_rejects_a_missing_token(ws_client): - with pytest.raises(Exception), ws_client.websocket_connect("/api/ws"): # noqa: B017, SIM117 - # No token: the endpoint closes with 1008 before accepting. - pass +def test_socket_allows_anonymous_then_restricts_actions(ws_client): + """No token opens an anonymous socket, but only login/signup may be called. + + A visitor who has not logged in can reach ``/api/auth/login`` and + ``/api/auth/signup`` over RPC; every other RPC and any subscribe is refused. + """ + with ws_client.websocket_connect("/api/ws") as socket: + hello = socket.receive_json() + assert hello["type"] == "hello" + assert hello["user_id"] is None + + # A protected RPC is refused for an anonymous connection. + socket.send_json({"action": "rpc", "id": "1", "method": "GET", "path": "/api/organizations"}) + result = socket.receive_json() + assert result["type"] == "rpc_result" + assert result["status"] == 401 + + # Subscribing is refused until the caller logs in. + socket.send_json({"action": "subscribe", "channels": ["org:x"]}) + refused = socket.receive_json() + assert refused["type"] == "error" + + +def test_anonymous_may_call_login_over_rpc(ws_client): + with ws_client.websocket_connect("/api/ws") as socket: + socket.receive_json() # hello + socket.send_json( + { + "action": "rpc", + "id": "login-1", + "method": "POST", + "path": "/api/auth/login", + "body": {"email": "anon@example.com", "password": "password123"}, + } + ) + result = socket.receive_json() + assert result["type"] == "rpc_result" + # No such user -> the handler's own auth dependency returns 401, proving + # the RPC reached the real login route through the internal app. + assert result["status"] == 401 def test_task_creation_is_pushed_live(ws_client): diff --git a/apps/web/src/lib/api.ts b/apps/web/src/lib/api.ts index dc22973..2f39628 100644 --- a/apps/web/src/lib/api.ts +++ b/apps/web/src/lib/api.ts @@ -7,6 +7,7 @@ */ import type { ApiErrorBody } from "@agy/contracts"; +import { getRealtimeClient } from "@/lib/ws"; /** * Resolves the API base URL. @@ -89,16 +90,34 @@ export interface RequestOptions { } export async function apiRequest(path: string, options: RequestOptions = {}): Promise { - const { method = "GET", body, token, signal } = options; + const { method = "GET", body, signal } = options; + + // Liveness and readiness probes stay on plain HTTP: they answer during a DB + // outage and are used by deploy orchestration (and the browser status check), + // so they must not depend on the authenticated socket. Every application call + // travels over the WebSocket as an RPC instead (see lib/ws.ts) — one + // connection for both live feeds and request/response. + if (path === "/health" || path === "/health/ready") { + return fetchHealth(path, method, body, signal); + } + + const client = getRealtimeClient(); + return client.rpc(method, path, body === undefined ? null : body, { signal }); +} + +/** Plain-HTTP implementation kept only for the two health probes. */ +async function fetchHealth( + path: string, + method: string, + body: unknown, + signal?: AbortSignal, +): Promise { const headers: Record = { Accept: "application/json" }; if (body !== undefined) headers["Content-Type"] = "application/json"; - const authToken = token ?? getStoredToken(); - if (authToken) headers.Authorization = `Bearer ${authToken}`; - const baseUrl = getApiBaseUrl(); let response: Response; try { - response = await fetch(`${baseUrl}${path}`, { + response = await fetch(`${getApiBaseUrl()}${path}`, { method, headers, body: body === undefined ? undefined : JSON.stringify(body), @@ -118,8 +137,6 @@ export async function apiRequest(path: string, options: RequestOptions = {}): const envelope = payload as ApiErrorBody | null; const code = envelope?.error?.code ?? "http_error"; const message = envelope?.error?.message ?? `خطای سرور (${response.status})`; - // Fall back to the header: a failure raised outside the error handlers - // (a CORS rejection, a proxy error) still carries the id. const requestId = envelope?.error?.request_id ?? response.headers.get("X-Request-ID") ?? undefined; throw new ApiError(response.status, code, message, envelope?.error?.details, requestId); diff --git a/apps/web/src/lib/realtime.tsx b/apps/web/src/lib/realtime.tsx index d435e1b..b54977d 100644 --- a/apps/web/src/lib/realtime.tsx +++ b/apps/web/src/lib/realtime.tsx @@ -19,7 +19,6 @@ import * as React from "react"; import type { WsEvent, WsEventType } from "@agy/contracts"; import { getRealtimeClient, type RealtimeStatus } from "@/lib/ws"; -import { getStoredToken } from "@/lib/api"; interface RealtimeContextValue { status: RealtimeStatus; @@ -33,12 +32,12 @@ export function RealtimeProvider({ children }: { children: React.ReactNode }) { const client = React.useMemo(() => getRealtimeClient(), []); const [status] = useRealtimeState(client); - // The connection follows the token: present -> live, gone -> closed. + // The connection follows the session: one socket for the app, opened even + // before login (anonymous) so the first ``login``/``signup`` RPC can ride it. + // ``client.connect()`` is a no-op when the socket already matches the current + // token, and reconnects (anonymous <-> authenticated) when the token changes. React.useEffect(() => { - const connect = () => { - if (getStoredToken()) client.connect(); - else client.disconnect(); - }; + const connect = () => client.connect(); connect(); // ``storage`` fires when another tab logs in or out; polling covers the // same-tab case where the token is written and read within one tick. diff --git a/apps/web/src/lib/ws.ts b/apps/web/src/lib/ws.ts index 3ec63ce..3645b0c 100644 --- a/apps/web/src/lib/ws.ts +++ b/apps/web/src/lib/ws.ts @@ -9,15 +9,21 @@ * * **The token is a function, not a value.** ``getToken`` is read at connect * time, so a login/logout cycle in another tab reconnects correctly instead * of pushing on a dead credential. - * * **Reconnect with backoff, capped.** Networks flap; a reconnect storm is - * how a flapping network becomes an outage. The cap is deliberately low - * enough that "live" feels live again within seconds. + * * **Anonymous before login.** The socket opens without a token so a visitor + * can call ``login``/``signup`` over RPC. Once a token exists the socket is + * re-established (still the one socket) carrying the bearer token, and every + * later RPC rides that authenticated connection. + * * **Reconnect with backoff, capped.** Networks flap; a reconnect storm is how + * a flapping network becomes an outage. The cap is deliberately low enough + * that "live" feels live again within seconds. * * **Heartbeat.** A silent middlebox will happily hold a dead socket open * forever. A ping every ``HEARTBEAT_INTERVAL_MS`` plus a pong timeout moves * that failure from "the page quietly stops updating" to "the socket * reconnects and updates resume". */ +import type { ApiErrorBody } from "@agy/contracts"; +import { ApiError } from "@/lib/api"; import type { WsClientFrame, WsEvent, WsServerFrame } from "@agy/contracts"; import { getStoredToken } from "@/lib/api"; @@ -39,10 +45,19 @@ const MAX_BACKOFF_MS = 8_000; const HEARTBEAT_INTERVAL_MS = 25_000; /** If no pong arrives within this window the socket is considered dead. */ const PONG_TIMEOUT_MS = 10_000; +/** Upper bound on a single RPC before it is failed locally. Long enough for a + * slow DB call, short enough that the UI does not hang forever. */ +const RPC_TIMEOUT_MS = 60_000; type EventListener = (event: WsEvent) => void; type StatusListener = (status: RealtimeStatus) => void; +interface PendingRpc { + resolve: (value: unknown) => void; + reject: (reason: ApiError) => void; + timer: ReturnType; +} + export class RealtimeClient { private socket: WebSocket | null = null; private status: RealtimeStatus = "idle"; @@ -54,55 +69,31 @@ export class RealtimeClient { private reconnectAttempts = 0; private reconnectTimer: ReturnType | null = null; private closedByUs = false; + /** The token the current socket authenticated with; null for anonymous. */ + private connectedToken: string | null = null; + /** Resolved when the socket transitions to OPEN for the current attempt. */ + private openPromise: Promise | null = null; + /** In-flight RPCs keyed by their client-generated id. */ + private readonly pendingRpc = new Map(); + private rpcSeq = 0; constructor(private readonly getToken: () => string | null = getStoredToken) {} // -- lifecycle ----------------------------------------------------------- + /** Open (or reopen) the socket for the token the getter currently returns. */ connect(): void { const token = this.getToken(); - if (!token) { - // No session yet: nothing to push. The provider reconnects on login. - this.setStatus("idle"); - return; - } + // Already open for this exact token: nothing to do. if ( this.socket && - (this.socket.readyState === WebSocket.OPEN || this.socket.readyState === WebSocket.CONNECTING) + (this.socket.readyState === WebSocket.OPEN || + this.socket.readyState === WebSocket.CONNECTING) && + this.connectedToken === token ) { return; } - - this.closedByUs = false; - this.setStatus(this.reconnectAttempts > 0 ? "reconnecting" : "connecting"); - - const socket = new WebSocket(`${websocketUrl()}?token=${encodeURIComponent(token)}`); - this.socket = socket; - - socket.onopen = () => { - this.reconnectAttempts = 0; - this.setStatus("live"); - this.startHeartbeat(); - // Re-subscribe: a new socket knows nothing about what this client cared - // about, and pages do not (and should not) re-run their effects to fix it. - if (this.subscriptions.size > 0) { - this.send({ action: "subscribe", channels: [...this.subscriptions] }); - } - }; - - socket.onmessage = (message) => this.handleFrame(message.data); - - socket.onclose = () => { - this.stopHeartbeat(); - this.socket = null; - if (this.closedByUs) return; - this.scheduleReconnect(); - }; - - socket.onerror = () => { - // A close always follows an error; the close handler does the work. - // This only stops an unhandled rejection in stricter runtimes. - }; + this.openWith(token); } disconnect(): void { @@ -112,14 +103,72 @@ export class RealtimeClient { this.reconnectTimer = null; } this.stopHeartbeat(); + this.rejectPending(new ApiError(0, "connection_closed", "اتصال قطع شد.")); this.reconnectAttempts = 0; if (this.socket) { - this.socket.close(); + try { + this.socket.close(); + } catch { + /* already gone */ + } this.socket = null; } + this.connectedToken = null; this.setStatus("idle"); } + private openWith(token: string | null): void { + this.closedByUs = false; + this.setStatus(this.reconnectAttempts > 0 ? "reconnecting" : "connecting"); + + // Replace any in-flight socket (e.g. a token change mid-connect). + if (this.socket) { + try { + this.socket.close(); + } catch { + /* ignore */ + } + this.socket = null; + } + + const url = token + ? `${websocketUrl()}?token=${encodeURIComponent(token)}` + : websocketUrl(); + const socket = new WebSocket(url); + this.socket = socket; + + this.openPromise = new Promise((resolve, reject) => { + let settled = false; + socket.onopen = () => { + settled = true; + this.connectedToken = token; + this.reconnectAttempts = 0; + this.setStatus("live"); + this.startHeartbeat(); + // Re-play subscriptions: a new socket knows nothing about what this + // client cared about, and pages do not (and should not) re-run their + // effects to fix it. + if (this.subscriptions.size > 0) { + this.send({ action: "subscribe", channels: [...this.subscriptions] }); + } + resolve(); + }; + socket.onmessage = (message) => this.handleFrame(message.data); + socket.onclose = () => { + if (!settled) { + reject( + new ApiError(0, "connection_closed", "اتصال پیش از برقراری قطع شد."), + ); + } + this.onSocketClosed(); + }; + socket.onerror = () => { + // A close always follows an error; the close handler does the work. + // This only stops an unhandled rejection in stricter runtimes. + }; + }); + } + private scheduleReconnect(): void { this.setStatus("reconnecting"); const delay = Math.min( @@ -130,6 +179,77 @@ export class RealtimeClient { this.reconnectTimer = setTimeout(() => this.connect(), delay); } + // -- RPC ----------------------------------------------------------------- + + /** + * Call an application endpoint over the socket. + * + * Returns the parsed response body on 2xx, and rejects with an ``ApiError`` + * (status + server error envelope) on any non-2xx — identical to the old + * ``fetch``-based client, so every ``api.*`` method is unchanged. + */ + rpc( + method: "GET" | "POST" | "PATCH" | "PUT" | "DELETE", + path: string, + body: unknown, + options: { signal?: AbortSignal } = {}, + ): Promise { + return this.whenOpen().then( + () => + new Promise((resolve, reject) => { + const id = `r${++this.rpcSeq}-${Date.now().toString(36)}`; + const timer = setTimeout(() => { + this.pendingRpc.delete(id); + reject( + new ApiError( + 0, + "timeout", + "درخواست از طریق وب‌سوکت زمان‌بر شد.", + ), + ); + }, RPC_TIMEOUT_MS); + this.pendingRpc.set(id, { + resolve: resolve as (value: unknown) => void, + reject, + timer, + }); + this.send({ action: "rpc", id, method, path, body }); + const signal = options.signal; + if (signal) { + signal.addEventListener( + "abort", + () => { + const pending = this.pendingRpc.get(id); + if (pending) { + clearTimeout(pending.timer); + this.pendingRpc.delete(id); + pending.reject( + new ApiError(0, "aborted", "درخواست لغو شد."), + ); + } + }, + { once: true }, + ); + } + }), + ); + } + + /** Resolve once the socket is OPEN for the current token (opening if needed). */ + private whenOpen(): Promise { + const token = this.getToken(); + if ( + this.socket && + (this.socket.readyState === WebSocket.OPEN || + this.socket.readyState === WebSocket.CONNECTING) && + this.connectedToken === token + ) { + return this.openPromise ?? Promise.resolve(); + } + this.openWith(token); + return this.openPromise ?? Promise.resolve(); + } + // -- subscriptions --------------------------------------------------------- subscribe(channels: string[]): void { @@ -173,7 +293,7 @@ export class RealtimeClient { this.socket.send(JSON.stringify(frame)); } // A frame sent before the socket is open is not an error: on open, the - // full subscription set is replayed anyway. + // full subscription set is replayed anyway, and RPCs await whenOpen(). } private handleFrame(raw: unknown): void { @@ -202,11 +322,38 @@ export class RealtimeClient { case "error": console.warn("realtime: server error frame", frame.message); break; + case "rpc_result": + this.resolveRpc(frame); + break; default: break; } } + private resolveRpc(frame: Extract): void { + const pending = this.pendingRpc.get(frame.id); + if (!pending) return; + clearTimeout(pending.timer); + this.pendingRpc.delete(frame.id); + + if (frame.status >= 200 && frame.status < 300) { + pending.resolve(frame.body); + return; + } + const envelope = frame.body as ApiErrorBody | null; + const code = envelope?.error?.code ?? "http_error"; + const message = envelope?.error?.message ?? `خطای سرور (${frame.status})`; + pending.reject( + new ApiError( + frame.status, + code, + message, + envelope?.error?.details, + envelope?.error?.request_id, + ), + ); + } + private emitEvent(event: WsEvent): void { for (const listener of [...this.listeners]) { try { @@ -218,6 +365,18 @@ export class RealtimeClient { } } + private onSocketClosed(): void { + this.stopHeartbeat(); + this.socket = null; + this.connectedToken = null; + this.rejectPending(new ApiError(0, "connection_closed", "اتصال قطع شد.")); + if (this.closedByUs) { + this.setStatus("idle"); + return; + } + this.scheduleReconnect(); + } + private setStatus(status: RealtimeStatus): void { if (this.status === status) return; this.status = status; @@ -230,6 +389,14 @@ export class RealtimeClient { } } + private rejectPending(error: ApiError): void { + for (const pending of this.pendingRpc.values()) { + clearTimeout(pending.timer); + pending.reject(error); + } + this.pendingRpc.clear(); + } + // -- heartbeat ----------------------------------------------------------------- private startHeartbeat(): void { diff --git a/apps/web/src/test/api.test.ts b/apps/web/src/test/api.test.ts index ded234f..47532e7 100644 --- a/apps/web/src/test/api.test.ts +++ b/apps/web/src/test/api.test.ts @@ -1,5 +1,6 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { ApiError, api, apiRequest, getStoredToken, setStoredToken } from "@/lib/api"; +import { getRealtimeClient } from "@/lib/ws"; describe("token storage", () => { it("round-trips a token through localStorage", () => { @@ -25,22 +26,14 @@ describe("ApiError", () => { }); }); -describe("error correlation", () => { +describe("error correlation over rpc", () => { afterEach(() => { - vi.unstubAllGlobals(); + vi.restoreAllMocks(); }); it("reads the request id out of the error envelope", async () => { - vi.stubGlobal( - "fetch", - vi.fn(async () => - new Response( - JSON.stringify({ - error: { code: "internal_error", message: "خطا", request_id: "req-from-body" }, - }), - { status: 500, headers: { "Content-Type": "application/json" } }, - ), - ), + vi.spyOn(getRealtimeClient(), "rpc").mockRejectedValue( + new ApiError(500, "internal_error", "خطا", undefined, "req-from-body"), ); await expect(apiRequest("/api/whatever")).rejects.toMatchObject({ @@ -49,127 +42,95 @@ describe("error correlation", () => { }); }); - it("falls back to the X-Request-ID header", async () => { - // A failure raised outside the error handlers — a CORS rejection or a proxy - // error — still carries the header, and the user should still get an id. - vi.stubGlobal( - "fetch", - vi.fn(async () => - new Response(JSON.stringify({ detail: "Disallowed CORS origin" }), { - status: 400, - headers: { "X-Request-ID": "req-from-header" }, - }), - ), + it("rejects an unauthenticated call with the server's 401", async () => { + vi.spyOn(getRealtimeClient(), "rpc").mockRejectedValue( + new ApiError(401, "unauthenticated", "نیاز به ورود دارید."), ); - await expect(apiRequest("/api/whatever")).rejects.toMatchObject({ - requestId: "req-from-header", + await expect(apiRequest("/api/organizations")).rejects.toMatchObject({ + status: 401, + code: "unauthenticated", }); }); }); describe("error console client", () => { afterEach(() => { - vi.unstubAllGlobals(); + vi.restoreAllMocks(); }); it("omits unset filters from the query string", async () => { - const calls: string[] = []; - vi.stubGlobal( - "fetch", - vi.fn(async (url: string) => { - calls.push(url); - return new Response(JSON.stringify({ items: [], total: 0, limit: 50, offset: 0 }), { - status: 200, - headers: { "Content-Type": "application/json" }, - }); - }), - ); - + const rpc = vi + .spyOn(getRealtimeClient(), "rpc") + .mockResolvedValue({ items: [], total: 0, limit: 50, offset: 0 } as unknown as never); await api.adminErrors({ status: "open", severity: "critical" }); - expect(calls[0]).toContain("status=open"); - expect(calls[0]).toContain("severity=critical"); + expect(rpc).toHaveBeenCalledTimes(1); + const [method, path] = rpc.mock.calls[0]; + expect(method).toBe("GET"); + expect(path).toContain("status=open"); + expect(path).toContain("severity=critical"); // An empty string means "all", and must not be sent as a filter value. - expect(calls[0]).not.toContain("source="); - expect(calls[0]).not.toContain("search="); + expect(path).not.toContain("source="); + expect(path).not.toContain("search="); }); it("sends no query string at all when there are no filters", async () => { - const calls: string[] = []; - vi.stubGlobal( - "fetch", - vi.fn(async (url: string) => { - calls.push(url); - return new Response(JSON.stringify([]), { - status: 200, - headers: { "Content-Type": "application/json" }, - }); - }), - ); - + const rpc = vi + .spyOn(getRealtimeClient(), "rpc") + .mockResolvedValue([] as unknown as never); await api.adminErrorGroups(); - expect(calls[0]).not.toContain("?"); + const [, path] = rpc.mock.calls[0]; + expect(path).not.toContain("?"); }); }); describe("agy account client", () => { afterEach(() => { - vi.unstubAllGlobals(); + vi.restoreAllMocks(); }); - function stubJson(body: unknown) { - const calls: { method: string; url: string; body: string }[] = []; - vi.stubGlobal( - "fetch", - vi.fn(async (url: string, init?: RequestInit) => { - calls.push({ - method: init?.method ?? "GET", - url, - body: String(init?.body ?? ""), - }); - return new Response(JSON.stringify(body), { - status: 200, - headers: { "Content-Type": "application/json" }, - }); - }), - ); - return calls; + function stubRpc(body: unknown) { + return vi + .spyOn(getRealtimeClient(), "rpc") + .mockResolvedValue(body as unknown as never); } it("posts to the login start endpoint", async () => { - const calls = stubJson({ login_id: "l1", auth_url: "https://accounts.google.com/..." }); + const rpc = stubRpc({ login_id: "l1", auth_url: "https://accounts.google.com/..." }); await api.agyStartLogin(); - expect(calls[0].method).toBe("POST"); - expect(calls[0].url).toContain("/api/admin/agy/accounts/login/start"); + const [method, path] = rpc.mock.calls[0]; + expect(method).toBe("POST"); + expect(path).toContain("/api/admin/agy/accounts/login/start"); }); it("sends the pasted callback URL back verbatim", async () => { - const calls = stubJson({ id: "a1", email: "x@gmail.com" }); + const rpc = stubRpc({ id: "a1", email: "x@gmail.com" }); await api.agyCompleteLogin({ login_id: "l1", callback_url: "http://localhost:51121/oauth-callback?code=c&state=s", }); - expect(calls[0].url).toContain("/api/admin/agy/accounts/login/complete"); - expect(calls[0].body).toContain("callback_url"); + const [method, path, sentBody] = rpc.mock.calls[0]; + expect(path).toContain("/api/admin/agy/accounts/login/complete"); + expect(JSON.stringify(sentBody)).toContain("callback_url"); }); it("routes activate/deactivate to the right verb", async () => { - const calls = stubJson({ id: "a1" }); + const rpc = stubRpc({ id: "a1" }); await api.agySetAccountActive("a1", false); await api.agySetAccountActive("a1", true); - expect(calls[0].url).toContain("/api/admin/agy/accounts/a1/deactivate"); - expect(calls[1].url).toContain("/api/admin/agy/accounts/a1/activate"); + expect(rpc.mock.calls[0][1]).toContain("/api/admin/agy/accounts/a1/deactivate"); + expect(rpc.mock.calls[1][1]).toContain("/api/admin/agy/accounts/a1/activate"); }); it("deletes with the DELETE verb", async () => { - const calls = stubJson(null); + const rpc = stubRpc(null); await api.agyDeleteAccount("a1"); - expect(calls[0].method).toBe("DELETE"); - expect(calls[0].url).toContain("/api/admin/agy/accounts/a1"); + expect(rpc.mock.calls[0][0]).toBe("DELETE"); + expect(rpc.mock.calls[0][1]).toContain("/api/admin/agy/accounts/a1"); }); it("fetches the live model catalog for one account", async () => { - const calls = stubJson([ + const rpc = stubRpc([ { id: "gemini-3-flash", display_name: "Gemini 3 Flash", @@ -184,21 +145,23 @@ describe("agy account client", () => { }, ]); const models = await api.agyAccountModels("a1"); - expect(calls[0].method).toBe("GET"); - expect(calls[0].url).toContain("/api/admin/agy/accounts/a1/models"); + const [method, path] = rpc.mock.calls[0]; + expect(method).toBe("GET"); + expect(path).toContain("/api/admin/agy/accounts/a1/models"); expect(models.map((m) => m.id)).toContain("gemini-3-pro-high"); expect(models[0].supported_efforts).toEqual(["low", "medium", "high"]); }); it("creates a model from the cascade — no manual names", async () => { - const calls = stubJson({ id: "m1", internal_name: "gemini-3-pro-high" }); + const rpc = stubRpc({ id: "m1", internal_name: "gemini-3-pro-high" }); await api.adminCreateModel({ internal_name: "gemini-3-pro-high", capabilities: ["text", "reasoning"], reasoning_effort: "high", }); - expect(calls[0].url).toContain("/api/admin/models"); - expect(calls[0].body).toContain("gemini-3-pro-high"); - expect(calls[0].body).toContain("reasoning_effort"); + const [method, path, sentBody] = rpc.mock.calls[0]; + expect(path).toContain("/api/admin/models"); + expect(JSON.stringify(sentBody)).toContain("gemini-3-pro-high"); + expect(JSON.stringify(sentBody)).toContain("reasoning_effort"); }); }); diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index ec837f3..a969af5 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -711,18 +711,31 @@ export interface WsEvent> { /** Every frame the server sends. Control frames and events share ``type``. */ export type WsServerFrame = - | { type: "hello"; user_id: string; server_time: string } + | { type: "hello"; user_id: string | null; server_time: string } | { type: "subscribed"; accepted: string[]; rejected: string[] } | { type: "unsubscribed"; channels: string[] } | { type: "pong" } | { type: "error"; message: string } - | { type: "event"; event: WsEvent }; + | { type: "event"; event: WsEvent } + /** Reply to a client ``rpc`` frame: the HTTP status and parsed JSON body of the + * internal dispatch, correlating on ``id``. */ + | { type: "rpc_result"; id: string; status: number; body: unknown }; /** Every frame the client may send. */ export type WsClientFrame = | { action: "subscribe"; channels: string[] } | { action: "unsubscribe"; channels: string[] } - | { action: "ping" }; + | { action: "ping" } + /** A request/response call into the server. ``id`` correlates the + * ``rpc_result``; ``path`` is the full mounted path including any query + * string. Every application call travels over this frame. */ + | { + action: "rpc"; + id: string; + method: "GET" | "POST" | "PATCH" | "PUT" | "DELETE"; + path: string; + body: unknown; + }; /** Live run-step payload — what the task timeline renders as it arrives. */ export interface WsRunStepData {