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.
This commit is contained in:
mjalmoudy94
2026-10-13 13:54:56 +03:30
parent f1e5df29ce
commit 8066e9d055
12 changed files with 631 additions and 222 deletions
+24 -7
View File
@@ -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<T>(path: string, options: RequestOptions = {}): Promise<T> {
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<T>(path, method, body, signal);
}
const client = getRealtimeClient();
return client.rpc<T>(method, path, body === undefined ? null : body, { signal });
}
/** Plain-HTTP implementation kept only for the two health probes. */
async function fetchHealth<T>(
path: string,
method: string,
body: unknown,
signal?: AbortSignal,
): Promise<T> {
const headers: Record<string, string> = { 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<T>(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);
+5 -6
View File
@@ -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.
+209 -42
View File
@@ -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<typeof setTimeout>;
}
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<typeof setTimeout> | 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<void> | null = null;
/** In-flight RPCs keyed by their client-generated id. */
private readonly pendingRpc = new Map<string, PendingRpc>();
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<void>((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<T>(
method: "GET" | "POST" | "PATCH" | "PUT" | "DELETE",
path: string,
body: unknown,
options: { signal?: AbortSignal } = {},
): Promise<T> {
return this.whenOpen().then(
() =>
new Promise<T>((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<void> {
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<WsServerFrame, { type: "rpc_result" }>): 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 {
+55 -92
View File
@@ -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");
});
});