huskies/frontend/src/api/rpc.ts

/**
 * Lightweight read-RPC client over the `/ws` WebSocket.
 *
 * Each `rpcCall` opens a short-lived WebSocket, sends an `rpc_request` frame,
 * waits for the matching `rpc_response`, then closes the connection.
 *
 * On a transient connection failure the call is retried once before rejecting,
 * which lets a freshly-started backend race finish before the user sees an
 * error.  Failures surface as `Error` instances whose `.message` is intended
 * to be visible (toast / banner) — callers must not swallow them silently.
 */

let correlationCounter = 0;

function nextCorrelationId(): string {
	return `rpc-${Date.now()}-${++correlationCounter}`;
}

/**
 * Build the WebSocket URL for the `/ws` endpoint, deriving the protocol
 * (ws/wss) and host from the current page location.
 */
function buildWsUrl(): string {
	const proto = window.location.protocol === "https:" ? "wss:" : "ws:";
	return `${proto}//${window.location.host}/ws`;
}

export interface RpcResponse<T = unknown> {
	ok: boolean;
	result?: T;
	error?: string;
	code?: string;
}

/** Error subclass for RPC failures so callers can recognise them. */
export class RpcError extends Error {
	constructor(
		message: string,
		public readonly code?: string,
		public readonly method?: string,
	) {
		super(message);
		this.name = "RpcError";
	}
}

/** Maximum number of automatic retries on transient WebSocket failure. */
const MAX_RETRIES = 1;

/** Delay between retry attempts (ms). */
const RETRY_DELAY_MS = 250;

/**
 * Internal: a single one-shot RPC attempt. Resolves with the result or
 * rejects with an `RpcError`.
 */
function rpcAttempt<T>(
	method: string,
	params: Record<string, unknown>,
	timeoutMs: number,
): Promise<T> {
	return new Promise<T>((resolve, reject) => {
		const correlationId = nextCorrelationId();
		let ws: WebSocket;
		try {
			ws = new WebSocket(buildWsUrl());
		} catch (err) {
			reject(
				new RpcError(
					`Failed to open WebSocket for ${method}: ${(err as Error).message}`,
					"CONNECT_FAILED",
					method,
				),
			);
			return;
		}
		let settled = false;

		const timer = setTimeout(() => {
			if (!settled) {
				settled = true;
				try {
					ws.close();
				} catch {
					/* ignore */
				}
				reject(new RpcError(`RPC timeout for ${method}`, "TIMEOUT", method));
			}
		}, timeoutMs);

		ws.onopen = () => {
			ws.send(
				JSON.stringify({
					kind: "rpc_request",
					version: 1,
					correlation_id: correlationId,
					ttl_ms: timeoutMs,
					method,
					params,
				}),
			);
		};

		ws.onmessage = (event) => {
			try {
				const data = JSON.parse(event.data);
				if (
					data.kind === "rpc_response" &&
					data.correlation_id === correlationId
				) {
					settled = true;
					clearTimeout(timer);
					try {
						ws.close();
					} catch {
						/* ignore */
					}
					if (data.ok) {
						resolve(data.result as T);
					} else {
						reject(
							new RpcError(
								data.error || `RPC error: ${data.code || "UNKNOWN"}`,
								data.code,
								method,
							),
						);
					}
				}
				// Ignore other frames (pipeline_state, onboarding_status, etc.)
			} catch {
				/* ignore non-JSON / malformed frames */
			}
		};

		ws.onerror = () => {
			if (!settled) {
				settled = true;
				clearTimeout(timer);
				reject(
					new RpcError(
						`WebSocket error during RPC call to ${method}`,
						"CONNECT_FAILED",
						method,
					),
				);
			}
		};

		ws.onclose = () => {
			if (!settled) {
				settled = true;
				clearTimeout(timer);
				reject(
					new RpcError(
						`WebSocket closed before RPC response for ${method}`,
						"CONNECT_FAILED",
						method,
					),
				);
			}
		};
	});
}

/** Return true if the error is one we should retry (connection-level). */
function isRetryable(err: unknown): boolean {
	return (
		err instanceof RpcError &&
		(err.code === "CONNECT_FAILED" || err.code === "TIMEOUT")
	);
}

function sleep(ms: number): Promise<void> {
	return new Promise((r) => setTimeout(r, ms));
}

/**
 * Send a read-RPC request over a temporary WebSocket connection and return
 * the result.  On transient connection failure the call is retried once
 * before rejecting.  Rejects with [`RpcError`] on server-side errors,
 * timeouts, or persistent connection failures.
 */
export async function rpcCall<T = unknown>(
	method: string,
	params: Record<string, unknown> = {},
	timeoutMs = 5000,
): Promise<T> {
	let lastErr: unknown;
	for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
		try {
			return await rpcAttempt<T>(method, params, timeoutMs);
		} catch (err) {
			lastErr = err;
			if (attempt < MAX_RETRIES && isRetryable(err)) {
				await sleep(RETRY_DELAY_MS);
				continue;
			}
			throw err;
		}
	}
	// Unreachable but TypeScript can't prove it.
	throw lastErr;
}