diff --git a/api/relay.ts b/api/relay.ts
index 3b6d5bb..7363c40 100644
--- a/api/relay.ts
+++ b/api/relay.ts
@@ -1,296 +1,36 @@
/**
* Vercel-compatible relay handler (Bun runtime).
*
- * This file is the entry point for Vercel serverless function deployments.
- * It exports `{ fetch }` — the contract Vercel's Bun runtime expects for
- * serverless functions.
- *
- * It reuses the same relay logic from `src/lib/` and `src/middleware/` as
- * the standalone Bun.serve() server, but:
- * - Does NOT call Bun.serve() (Vercel manages the server)
- * - Does NOT support WebSocket upgrades (not available in Vercel Functions)
- * - Uses a simplified IP detection (no server.requestIP())
- * - Rate limiter resets on cold starts (per-instance memory)
+ * Thin wrapper around the shared router in src/lib/router.ts.
+ * Does NOT call Bun.serve() (Vercel manages the server).
+ * Does NOT support WebSocket upgrades.
*/
import {
- normalizeTargetUrl,
- isAllowedTarget,
- filterRequestHeaders,
- buildRelayRequest,
- createRelayResponse,
- classifyFetchError,
- createErrorResponse,
- createCorsPreflightResponse,
-} from "../src/lib/relay-utils";
+ handleRelayPlain,
+ handleRequest,
+ getClientIP,
+} from "../src/lib/router";
+import type { RouterEnv } from "../src/lib/router";
-import { checkBodySize } from "../src/middleware/body-limiter";
-import { createRateLimiter } from "../src/middleware/rate-limiter";
-import { logRelayEvent } from "../src/middleware/logger";
-import { handleChatCompletion, listModels } from "../src/lib/ai-proxy";
-import { handleAnthropicMessages } from "../src/lib/anthropic-proxy";
import { getTestPageHtml } from "../src/lib/test-page";
-// ─── Configuration ──────────────────────────────────────────────────────────────
+// --- Singletons (survives warm invocations) ----------------------------------
-const RELAY_TIMEOUT_MS = Number.parseInt(
- process.env.RELAY_TIMEOUT_MS ?? "30000",
- 10,
-);
-const SERVER_START_TIME = Date.now();
-const RELAY_VERSION = "1.0.0";
-
-// ─── API Key Authentication ─────────────────────────────────────────────────────
-
-const API_KEY = process.env.API_KEY ?? "sk-dummy-key";
-
-function requireAuth(req: Request): Response | null {
- const header = req.headers.get("authorization") ?? req.headers.get("x-api-key") ?? "";
- const key = header.replace(/^Bearer\s+/i, "").trim();
- if (key === API_KEY) return null;
- return new Response(
- JSON.stringify({ error: { message: "Unauthorized", type: "auth_error" } }),
- { status: 401, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } },
- );
-}
-
-// ─── Middleware instances (singletons — persist across warm invocations) ─────────
-
-const rateLimiter = createRateLimiter({
- maxRequests: Number.parseInt(process.env.RATE_LIMIT_MAX ?? "100", 10),
- windowMs: Number.parseInt(
- process.env.RATE_LIMIT_WINDOW_MS ?? "60000",
- 10,
- ),
-});
-
-// ─── Helpers ─────────────────────────────────────────────────────────────────────
-
-/**
- * Get the client IP from the request headers.
- *
- * Vercel populates `x-forwarded-for` and/or `cf-connecting-ip` automatically.
- * Unlike the standalone server, we do NOT call `server.requestIP()` since
- * that Bun API is not available in Vercel Functions.
- */
-function getClientIP(req: Request): string {
- const forwarded = req.headers.get("x-forwarded-for");
- if (forwarded) {
- const first = forwarded.split(",")[0]?.trim();
- if (first) return first;
- }
-
- const cfIp = req.headers.get("cf-connecting-ip");
- if (cfIp) return cfIp;
-
- return "unknown";
-}
-
-// ─── Route Handlers ─────────────────────────────────────────────────────────────
-
-function handleHealth(): Response {
- return new Response(
- JSON.stringify({
- status: "ok",
- uptime: Date.now() - SERVER_START_TIME,
- version: RELAY_VERSION,
- }),
- {
- status: 200,
- headers: {
- "Content-Type": "application/json",
- "Access-Control-Allow-Origin": "*",
- },
- },
- );
-}
-
-function handleIndex(): Response {
- const html = `
-
-
-
-
- Edge Proxy Relay
-
-
-
-
- Edge Proxy Relay
- Server is running
- /health
-
-
-`;
-
- return new Response(html, {
- status: 200,
- headers: {
- "Content-Type": "text/html; charset=utf-8",
- },
- });
-}
-
-// ─── Relay Logic ────────────────────────────────────────────────────────────────
-
-async function handleRelay(req: Request): Promise {
- const startTime = performance.now();
- const method = req.method;
- const clientIP = getClientIP(req);
- const requestUrl = req.url;
-
- // ── Pre-flight CORS ──────────────────────────────────────────────
- if (method === "OPTIONS") {
- return createCorsPreflightResponse();
- }
-
- // ── Middleware: Body size check ──────────────────────────────────
- const bodyError = checkBodySize(req);
- if (bodyError) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: bodyError.status,
- durationMs: Math.round(performance.now() - startTime),
- ip: clientIP,
- });
- return bodyError;
- }
-
- // ── Middleware: Rate limiting ────────────────────────────────────
- const rateCheck = await rateLimiter.checkAsync(clientIP);
- if (!rateCheck.allowed) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: 429,
- durationMs: Math.round(performance.now() - startTime),
- error: "rate_limit_exceeded",
- ip: clientIP,
- });
- return new Response(
- JSON.stringify({
- error: true,
- code: "RATE_LIMITED",
- message: "Too many requests",
- retryAfterMs: rateCheck.retryAfterMs,
- }),
- {
- status: 429,
- headers: {
- "Content-Type": "application/json",
- "Access-Control-Allow-Origin": "*",
- "Retry-After": String(
- Math.ceil((rateCheck.retryAfterMs ?? 60_000) / 1000),
- ),
- },
- },
- );
- }
-
- // ── Extract relay parameters from headers ───────────────────────
- const target = req.headers.get("x-relay-target");
- const relayPath = req.headers.get("x-relay-path") ?? "/";
-
- // ── SSRF: Normalize and validate target URL ─────────────────────
- const targetUrl = normalizeTargetUrl(target, relayPath);
- if (!targetUrl) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: 400,
- durationMs: Math.round(performance.now() - startTime),
- error: "missing_target_header",
- ip: clientIP,
- });
- return createErrorResponse({
- code: "INVALID_TARGET",
- status: 400,
- message: "Missing or invalid x-relay-target header",
- });
- }
-
- if (!isAllowedTarget(targetUrl)) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: 403,
- durationMs: Math.round(performance.now() - startTime),
- error: "target_not_allowed",
- ip: clientIP,
- });
- return createErrorResponse({
- code: "SSRF_BLOCKED",
- status: 403,
- message: "Target domain not allowed",
- });
- }
-
- // ── Build the upstream request ──────────────────────────────────
- const filteredHeaders = filterRequestHeaders(req.headers);
- const fetchOptions = buildRelayRequest(
- req,
- filteredHeaders,
- RELAY_TIMEOUT_MS,
- );
-
- const targetUrlString = targetUrl.toString();
-
- // ── Execute upstream fetch ──────────────────────────────────────
- let response: Response;
- try {
- response = await fetch(targetUrlString, fetchOptions);
- } catch (err) {
- const classified = classifyFetchError(err);
- logRelayEvent({
- method,
- url: requestUrl,
- status: classified.status,
- durationMs: Math.round(performance.now() - startTime),
- error: classified.message,
- targetUrl: targetUrlString,
- ip: clientIP,
- });
- return createErrorResponse(classified);
- }
-
- // ── Build relay response ────────────────────────────────────────
- const relayedResponse = createRelayResponse(response);
-
- logRelayEvent({
- method,
- url: requestUrl,
- status: relayedResponse.status,
- durationMs: Math.round(performance.now() - startTime),
- targetUrl: targetUrlString,
- ip: clientIP,
- });
-
- return relayedResponse;
-}
-
-// ─── Exported Vercel Function Handler ───────────────────────────────────────────
+const routerEnv: RouterEnv = {};
/**
* Hybrid handler for Vercel/Node web-api and Bun/Workers runtimes.
- *
- * Exports both:
- * 1. Default function: parsed by Vercel Node runtime.
- * 2. Default.fetch(): parsed by Cloudflare / Bun runtimes.
*/
async function fetchHandler(req: Request): Promise {
const url = new URL(req.url);
+ const clientIP = getClientIP(req);
- // Static routes — show index only when no relay target is requested
- if (url.pathname === "/health") return handleHealth();
+ // Static routes
+ if (url.pathname === "/health") {
+ const { handleHealth } = await import("../src/lib/router");
+ return handleHealth();
+ }
if (url.pathname === "/docs" || url.pathname === "/test") {
return new Response(getTestPageHtml(), {
headers: { "Content-Type": "text/html; charset=utf-8" },
@@ -301,102 +41,19 @@ async function fetchHandler(req: Request): Promise {
req.method === "GET" &&
!req.headers.get("x-relay-target")
) {
+ const { handleIndex } = await import("../src/lib/router");
return handleIndex();
}
- // WebSocket upgrade — not supported in Vercel Functions
- if (
- req.method === "GET" &&
- req.headers.get("upgrade")?.toLowerCase() === "websocket"
- ) {
- return new Response(
- JSON.stringify({
- error: true,
- code: "UNSUPPORTED",
- message: "WebSocket relay is not supported on this deployment",
- }),
- {
- status: 400,
- headers: {
- "Content-Type": "application/json",
- "Access-Control-Allow-Origin": "*",
- },
- },
- );
- }
+ // Delegate to shared router (handles AI proxy, model list, relay)
+ const result = await handleRequest(req, routerEnv, clientIP, {
+ isWebSocketSupported: false,
+ skipProxyPool: true,
+ });
+ if (result !== undefined) return result;
- // AI proxy routes — OpenAI-compatible
- if (url.pathname === "/v1/chat/completions") {
- if (req.method === "OPTIONS") return createCorsPreflightResponse();
- if (req.method !== "POST") return new Response("Method Not Allowed", { status: 405 });
- const authErr = requireAuth(req);
- if (authErr) return authErr;
- try {
- const body = await req.json();
- return await handleChatCompletion(body);
- } catch (err) {
- const isJsonError = err instanceof Error && (err.name === "SyntaxError" || err.message.includes("JSON"));
- if (isJsonError) {
- return new Response(
- JSON.stringify({ error: { message: "Invalid JSON body", type: "invalid_request_error" } }),
- { status: 400, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } },
- );
- }
- return new Response(
- JSON.stringify({ error: { message: err instanceof Error ? err.message : "Internal Server Error", type: "server_error" } }),
- { status: 500, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } },
- );
- }
- }
-
- // AI proxy routes — Anthropic-compatible
- if (url.pathname === "/v1/messages") {
- if (req.method === "OPTIONS") return createCorsPreflightResponse();
- if (req.method !== "POST") return new Response("Method Not Allowed", { status: 405 });
- const authErr = requireAuth(req);
- if (authErr) return authErr;
- try {
- const body = await req.json();
- return await handleAnthropicMessages(body);
- } catch (err) {
- const isJsonError = err instanceof Error && (err.name === "SyntaxError" || err.message.includes("JSON"));
- if (isJsonError) {
- return new Response(
- JSON.stringify({
- type: "error",
- error: { message: "Invalid JSON body", type: "invalid_request_error" },
- }),
- { status: 400, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } },
- );
- }
- return new Response(
- JSON.stringify({
- type: "error",
- error: { message: err instanceof Error ? err.message : "Internal Server Error", type: "server_error" },
- }),
- { status: 500, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } },
- );
- }
- }
-
- // Models list
- if (url.pathname === "/v1/models" && req.method === "GET") {
- const authErr = requireAuth(req);
- if (authErr) return authErr;
- const models = listModels().map((id) => ({
- id,
- object: "model",
- created: Math.floor(Date.now() / 1000),
- owned_by: "proxy",
- }));
- return new Response(
- JSON.stringify({ object: "list", data: models }),
- { status: 200, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } },
- );
- }
-
- // Generic HTTP relay
- return handleRelay(req);
+ // Fallback (shouldn't reach here for relay)
+ return handleRelayPlain(req, routerEnv, clientIP);
}
export default Object.assign(fetchHandler, {
diff --git a/src/index.test.ts b/src/index.test.ts
index 8ba3fc2..ed209c9 100644
--- a/src/index.test.ts
+++ b/src/index.test.ts
@@ -5,15 +5,15 @@
* bypassing the HTTP server layer.
*/
-import { test, expect, describe, beforeAll, afterAll } from "bun:test";
+import { test, expect, describe } from "bun:test";
// Import handler functions directly from index.ts
// Note: this will also start the Bun.serve() instance, which we allow.
import {
handleHealth,
handleIndex,
- getClientIP,
-} from "./index";
+ getClientIPFromServer as getClientIP,
+} from "./lib/router";
describe("handleHealth", () => {
test("should return 200 with JSON body", async () => {
diff --git a/src/index.ts b/src/index.ts
index 35a2f47..3435993 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -1,94 +1,48 @@
/**
- * Edge Proxy Relay — Pure Bun HTTP + WebSocket relay server.
+ * Edge Proxy Relay — Standalone Bun.serve() entry point.
*
- * Forwards requests/responses to a target URL specified via the
- * `x-relay-target` request header. Supports WebSocket upgrades
- * when the target uses `ws://` or `wss://`.
- *
- * --- Environment Variables ----------------------------------------------------
- * PORT — Server listen port (default: 3000)
- * RELAY_TIMEOUT_MS — Upstream fetch timeout (default: 30_000)
- * BODY_MAX_BYTES — Maximum accepted request body (default: 1_048_576)
- * RATE_LIMIT_MAX — Max requests per sliding window (default: 100)
- * RATE_LIMIT_WINDOW_MS— Sliding window duration (default: 60_000)
- * CORS_ORIGIN — Allowed CORS origin (default: *)
- * NODE_ENV — Set to "production" to disable dev features
+ * Thin wrapper around the shared router in src/lib/router.ts.
+ * Adds:
+ * - Bun.serve() bindings
+ * - WebSocket relay support
+ * - IPv6 source rotation
+ * - Proxy pool file loading
+ * - Graceful shutdown
*/
import {
normalizeTargetUrl,
isAllowedTarget,
- filterRequestHeaders,
- buildRelayRequest,
- createRelayResponse,
createErrorResponse,
- createCorsPreflightResponse,
- getCorsHeaders,
setSsrfDnsCheck,
} from "./lib/relay-utils";
-import { checkBodySize } from "./middleware/body-limiter";
-import { createRateLimiter } from "./middleware/rate-limiter";
-import { logRelayEvent } from "./middleware/logger";
-import { ProxyPool, SessionProxyPool } from "./lib/proxy-pool";
+import {
+ handleHealth,
+ handleIndex,
+ getClientIPFromServer as getClientIP,
+ getSharedProxyPool,
+} from "./lib/router";
+import type { RouterEnv } from "./lib/router";
+
import { IPv6SourcePool } from "./lib/ipv6-pool";
-import { handleChatCompletion, listModels } from "./lib/ai-proxy";
-import { handleAnthropicMessages } from "./lib/anthropic-proxy";
-import { fetchWithRetry, closeAllActiveReaders, isDevMode } from "./lib/fetch-utils";
+import { closeAllActiveReaders, isDevMode } from "./lib/fetch-utils";
import type { Server, ServerWebSocket } from "bun";
// --- Configuration ------------------------------------------------------------
+const RELAY_VERSION = "1.0.0";
const PORT = Number.parseInt(process.env.PORT ?? "3000", 10);
const HOST = process.env.HOST ?? "::";
-const RELAY_TIMEOUT_MS = Number.parseInt(
- process.env.RELAY_TIMEOUT_MS ?? "30000",
- 10,
-);
-const SERVER_START_TIME = Date.now();
-const RELAY_VERSION = "1.0.0";
-
-// --- API Key Authentication ---------------------------------------------------
-
-const API_KEY = process.env.API_KEY ?? "sk-dummy-key";
-
-function requireAuth(req: Request): Response | null {
- const header = req.headers.get("authorization") ?? req.headers.get("x-api-key") ?? "";
- const key = header.replace(/^Bearer\s+/i, "").trim();
- if (key === API_KEY) return null;
- return new Response(
- JSON.stringify({ error: { message: "Unauthorized", type: "auth_error" } }),
- {
- status: 401,
- headers: { "Content-Type": "application/json", ...getCorsHeaders() },
- },
- );
-}
-
-// --- Middleware instances (singletons) ----------------------------------------
-
-const rateLimiter = createRateLimiter({
- maxRequests: Number.parseInt(process.env.RATE_LIMIT_MAX ?? "100", 10),
- windowMs: Number.parseInt(
- process.env.RATE_LIMIT_WINDOW_MS ?? "60000",
- 10,
- ),
-});
// --- Proxy pool (optional) ----------------------------------------------------
-const proxyPool = new ProxyPool();
+const proxyPool = getSharedProxyPool();
proxyPool.tryLoad(
process.env.PROXY_FILE || process.env.PROXY_LIST || "./proxy.txt",
);
-// Session-aware proxy pool wrapping the base pool.
-// Sessions are per-request and short-lived — each streaming request gets
-// a random session ID so the same proxy is reused for the entire stream.
-const sessionPool = new SessionProxyPool(proxyPool);
-sessionPool.setFailureThreshold(3);
-
// --- IPv6 source pool (optional) ----------------------------------------------
const ipv6Pool = new IPv6SourcePool();
@@ -104,264 +58,21 @@ if (process.env.SSRF_DNS_CHECK === "true") {
console.log("[relay] SSRF DNS rebinding protection enabled");
}
+// --- Env bag for router (falls through to process.env) ------------------------
+
+const routerEnv: RouterEnv = {};
+
// --- WebSocket relay data type -----------------------------------------------
interface WSRelayData {
target: string;
relayPath: string;
upstream?: WebSocket;
- /** Set when client buffer exceeds threshold — stops forwarding upstream data */
paused?: boolean;
}
-// --- Route handlers ----------------------------------------------------------
+// --- WebSocket relay ---------------------------------------------------------
-/** Health check endpoint: returns status, uptime, and version. */
-function handleHealth(): Response {
- return new Response(
- JSON.stringify({
- status: "ok",
- uptime: Date.now() - SERVER_START_TIME,
- version: RELAY_VERSION,
- }),
- {
- status: 200,
- headers: {
- "Content-Type": "application/json",
- ...getCorsHeaders(),
- },
- },
- );
-}
-
-/** Status page shown at the root `/`. */
-function handleIndex(): Response {
- const html = `
-
-
-
-
- Edge Proxy Relay
-
-
-
-
- Edge Proxy Relay
- Server is running
-
-
-
-`;
-
- return new Response(html, {
- status: 200,
- headers: {
- "Content-Type": "text/html; charset=utf-8",
- },
- });
-}
-
-// --- HTTP Relay Logic -------------------------------------------------------
-
-/**
- * Get the client IP address from the request.
- * Tries `x-forwarded-for` first, then `cf-connecting-ip`, then falls back
- * to the direct connection address from `server.requestIP()`.
- */
-function getClientIP(
- req: Request,
- ipGetter: { requestIP(req: Request): { address: string } | null },
-): string {
- const forwarded = req.headers.get("x-forwarded-for");
- if (forwarded) {
- const first = forwarded.split(",")[0]?.trim();
- if (first) return first;
- }
-
- const cfIp = req.headers.get("cf-connecting-ip");
- if (cfIp) return cfIp;
-
- const remote = ipGetter.requestIP(req);
- if (remote) return remote.address;
-
- return "unknown";
-}
-
-/**
- * Core HTTP relay handler.
- *
- * Expects `x-relay-target` header to determine the upstream URL.
- * Applies middleware (body size check, rate limiting, logging) and
- * proxies the request while filtering sensitive headers.
- */
-async function handleRelay(
- req: Request,
- ipGetter: { requestIP(req: Request): { address: string } | null },
-): Promise {
- const startTime = performance.now();
- const method = req.method;
- const clientIP = getClientIP(req, ipGetter);
- const requestUrl = req.url;
-
- // -- Pre-flight CORS --------------------------------------------------------
- if (method === "OPTIONS") {
- return createCorsPreflightResponse();
- }
-
- // -- Middleware: Body size check -------------------------------------------
- const bodyError = checkBodySize(req);
- if (bodyError) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: bodyError.status,
- durationMs: Math.round(performance.now() - startTime),
- ip: clientIP,
- });
- return bodyError;
- }
-
- // -- Middleware: Rate limiting ---------------------------------------------
- const rateCheck = await rateLimiter.checkAsync(clientIP);
- if (!rateCheck.allowed) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: 429,
- durationMs: Math.round(performance.now() - startTime),
- error: "rate_limit_exceeded",
- ip: clientIP,
- });
- return new Response(
- JSON.stringify({
- error: true,
- code: "RATE_LIMITED",
- message: "Too many requests",
- retryAfterMs: rateCheck.retryAfterMs,
- }),
- {
- status: 429,
- headers: {
- "Content-Type": "application/json",
- ...getCorsHeaders(),
- "Retry-After": String(
- Math.ceil((rateCheck.retryAfterMs ?? 60_000) / 1000),
- ),
- },
- },
- );
- }
-
- // -- Extract relay parameters from headers ---------------------------------
- const target = req.headers.get("x-relay-target");
- const relayPath = req.headers.get("x-relay-path") ?? "/";
-
- // -- SSRF: Normalize and validate target URL --------------------------------
- const targetUrl = normalizeTargetUrl(target, relayPath);
- if (!targetUrl) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: 400,
- durationMs: Math.round(performance.now() - startTime),
- error: "missing_target_header",
- ip: clientIP,
- });
- return createErrorResponse({
- code: "INVALID_TARGET",
- status: 400,
- message: "Missing or invalid x-relay-target header",
- });
- }
-
- if (!isAllowedTarget(targetUrl)) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: 403,
- durationMs: Math.round(performance.now() - startTime),
- error: "target_not_allowed",
- ip: clientIP,
- });
- return createErrorResponse({
- code: "SSRF_BLOCKED",
- status: 403,
- message: "Target domain not allowed",
- });
- }
-
- // -- Build the upstream request ---------------------------------------------
- const filteredHeaders = filterRequestHeaders(req.headers);
- const fetchOptions = buildRelayRequest(
- req,
- filteredHeaders,
- RELAY_TIMEOUT_MS,
- ) as RequestInit & { proxy?: string };
-
- const targetUrlString = targetUrl.toString();
-
- // -- Get IPv6 source for outbound binding ----------------------------------
- const ipv6Source = ipv6Pool.getNext() ?? undefined;
-
- // -- Execute upstream fetch with shared retry -------------------------------
- const result = await fetchWithRetry(
- targetUrlString,
- fetchOptions,
- proxyPool,
- "relay",
- ipv6Source,
- );
-
- if (result.errorClassification) {
- logRelayEvent({
- method,
- url: requestUrl,
- status: result.errorClassification.status,
- durationMs: Math.round(performance.now() - startTime),
- error: result.errorClassification.message,
- targetUrl: targetUrlString,
- ip: clientIP,
- });
- return createErrorResponse(result.errorClassification);
- }
-
- const relayedResponse = createRelayResponse(result.response!);
-
- logRelayEvent({
- method,
- url: requestUrl,
- status: relayedResponse.status,
- durationMs: Math.round(performance.now() - startTime),
- targetUrl: targetUrlString,
- ip: clientIP,
- });
-
- return relayedResponse;
-}
-
-// --- WebSocket Relay Logic ---------------------------------------------------
-
-/**
- * Upgrade an HTTP request to a WebSocket and relay bidirectionally to the
- * target URL specified in the `x-relay-target` header.
- *
- * Returns `undefined` when the upgrade has been accepted (Bun takes over),
- * or a Response when the upgrade failed or the target is invalid.
- */
function handleWebSocketUpgrade(
req: Request,
srv: Server,
@@ -369,8 +80,7 @@ function handleWebSocketUpgrade(
const target = req.headers.get("x-relay-target");
if (!target) return undefined;
- const isWS =
- target.startsWith("ws://") || target.startsWith("wss://");
+ const isWS = target.startsWith("ws://") || target.startsWith("wss://");
if (!isWS) return undefined;
const relayPath = req.headers.get("x-relay-path") ?? "/";
@@ -391,10 +101,8 @@ function handleWebSocketUpgrade(
});
}
- const targetUrl = normalized.toString();
-
const upgraded = srv.upgrade(req, {
- data: { target: targetUrl, relayPath },
+ data: { target: normalized.toString(), relayPath },
});
if (!upgraded) {
@@ -414,6 +122,7 @@ const server: Server = Bun.serve({
async fetch(req: Request): Promise {
const url = new URL(req.url);
+ const ipv6Source = ipv6Pool.getNext() ?? undefined;
// Static routes
if (url.pathname === "/health") return handleHealth();
@@ -425,106 +134,46 @@ const server: Server = Bun.serve({
headers: { "Content-Type": "text/html; charset=utf-8" },
});
}
- if (url.pathname === "/" && req.method === "GET" && !req.headers.get("x-relay-target"))
+ if (
+ url.pathname === "/" &&
+ req.method === "GET" &&
+ !req.headers.get("x-relay-target")
+ )
return handleIndex();
- // AI proxy routes -- OpenAI-compatible API
- if (url.pathname === "/v1/chat/completions") {
- if (req.method === "OPTIONS") {
- return createCorsPreflightResponse();
- }
- if (req.method !== "POST") {
- return new Response("Method Not Allowed", { status: 405 });
- }
- const authErr = requireAuth(req);
- if (authErr) return authErr;
- try {
- const body = await req.json();
- const sessionId = crypto.randomUUID();
- const ipv6Source = ipv6Pool.getNext() ?? undefined;
- console.log(`[index] POST /v1/chat/completions session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions} ipv6=${ipv6Source ?? "none"}`);
- return handleChatCompletion(body, proxyPool, sessionPool, sessionId, ipv6Source);
- } catch {
- return new Response(
- JSON.stringify({ error: { message: "Invalid JSON body", type: "invalid_request_error" } }),
- { status: 400, headers: { "Content-Type": "application/json", ...getCorsHeaders() } },
- );
- }
- }
-
- // AI proxy routes -- Anthropic-compatible API
- if (url.pathname === "/v1/messages") {
- if (req.method === "OPTIONS") {
- return createCorsPreflightResponse();
- }
- if (req.method !== "POST") {
- return new Response("Method Not Allowed", { status: 405 });
- }
- const authErr = requireAuth(req);
- if (authErr) return authErr;
- try {
- const body = await req.json();
- const sessionId = crypto.randomUUID();
- const ipv6Source = ipv6Pool.getNext() ?? undefined;
- const anthropicVersion = req.headers.get("anthropic-version") ?? undefined;
- console.log(`[index] POST /v1/messages session=${sessionId.slice(0, 8)} model=${(body as any).model} poolSize=${proxyPool.size} sessionPool.active=${sessionPool.activeSessions} ipv6=${ipv6Source ?? "none"} anthropic-version=${anthropicVersion ?? "none"}`);
- return handleAnthropicMessages(body, proxyPool, sessionPool, sessionId, ipv6Source, anthropicVersion);
- } catch {
- return new Response(
- JSON.stringify({
- type: "error",
- error: { message: "Invalid JSON body", type: "invalid_request_error" },
- }),
- { status: 400, headers: { "Content-Type": "application/json", ...getCorsHeaders() } },
- );
- }
- }
-
- if (url.pathname === "/v1/models" && req.method === "GET") {
- const authErr = requireAuth(req);
- if (authErr) return authErr;
- const models = listModels().map((id) => ({
- id,
- object: "model",
- created: Math.floor(Date.now() / 1000),
- owned_by: "edge-proxy",
- features: ["prompt_caching"],
- }));
- return new Response(
- JSON.stringify({
- object: "list",
- data: models,
- }),
- {
- status: 200,
- headers: {
- "Content-Type": "application/json",
- ...getCorsHeaders(),
- },
- },
- );
- }
-
// WebSocket upgrade check
if (
req.method === "GET" &&
req.headers.get("upgrade")?.toLowerCase() === "websocket"
) {
const wsResult = handleWebSocketUpgrade(req, server);
- if (wsResult === undefined) {
- return undefined;
- }
+ if (wsResult === undefined) return undefined;
return wsResult;
}
- // Generic HTTP relay
- return handleRelay(req, server);
+ // Delegate all other routing (including AI proxy routes) to the shared router
+ const clientIP = getClientIP(req, server);
+ const result = await (await import("./lib/router")).handleRequest(
+ req,
+ routerEnv,
+ clientIP,
+ {
+ isWebSocketSupported: true,
+ ipv6Source,
+ },
+ );
+ if (result !== undefined) return result;
+
+ return undefined;
},
websocket: {
open(ws: ServerWebSocket) {
const { target } = ws.data;
+ const {
+ logRelayEvent,
+ } = require("./middleware/logger");
logRelayEvent({
method: "WS",
url: target,
@@ -535,14 +184,10 @@ const server: Server = Bun.serve({
const upstream = new WebSocket(target);
- upstream.onopen = () => {
- // Connection established
- };
+ upstream.onopen = () => {};
upstream.onmessage = (event: MessageEvent) => {
- // Backpressure: drop messages when client buffer is full
if (ws.data.paused) return;
-
const data = event.data;
if (typeof data === "string") {
ws.sendText(data);
@@ -550,9 +195,7 @@ const server: Server = Bun.serve({
ws.sendBinary(new Uint8Array(data));
} else if (data instanceof Blob) {
data.arrayBuffer().then((buf) => {
- if (!ws.data.paused) {
- ws.sendBinary(new Uint8Array(buf));
- }
+ if (!ws.data.paused) ws.sendBinary(new Uint8Array(buf));
});
} else {
ws.sendBinary(data as unknown as Uint8Array);
@@ -570,14 +213,13 @@ const server: Server = Bun.serve({
ws.data.upstream = upstream;
},
- message(ws: ServerWebSocket, message: string | Buffer) {
+ message(
+ ws: ServerWebSocket,
+ message: string | Buffer,
+ ) {
const upstream = ws.data.upstream;
if (upstream && upstream.readyState === WebSocket.OPEN) {
- if (typeof message === "string") {
- upstream.send(message);
- } else {
- upstream.send(message);
- }
+ upstream.send(message);
}
},
@@ -593,22 +235,25 @@ const server: Server = Bun.serve({
},
drain(ws: ServerWebSocket) {
- // Backpressure: pause forwarding when client buffer is large
const upstream = ws.data.upstream;
if (!upstream || upstream.readyState !== WebSocket.OPEN) return;
const buffered = (ws as any).bufferAmount ?? 0;
const BACKPRESSURE_THRESHOLD = 512 * 1024; // 512KB
- const RESUME_THRESHOLD = 64 * 1024; // 64KB
+ const RESUME_THRESHOLD = 64 * 1024; // 64KB
if (buffered > BACKPRESSURE_THRESHOLD) {
if (!ws.data.paused) {
ws.data.paused = true;
- console.warn(`[ws] Client backpressure: pausing upstream forwarding (${buffered} bytes buffered)`);
+ console.warn(
+ `[ws] Client backpressure: pausing upstream forwarding (${buffered} bytes buffered)`,
+ );
}
} else if (ws.data.paused && buffered < RESUME_THRESHOLD) {
ws.data.paused = false;
- console.log(`[ws] Client backpressure cleared: resuming upstream forwarding (${buffered} bytes buffered)`);
+ console.log(
+ `[ws] Client backpressure cleared: resuming upstream forwarding (${buffered} bytes buffered)`,
+ );
}
},
},
@@ -627,10 +272,7 @@ if (isDevMode()) {
const shutdownHandler = async (signal: string) => {
console.log(`\n[relay] Received ${signal}, shutting down gracefully...`);
-
- // Close active SSE streams so clients get proper stream end events
await closeAllActiveReaders();
-
server.stop();
process.exit(0);
};
@@ -641,10 +283,4 @@ process.on("SIGINT", () => shutdownHandler("SIGINT"));
// --- Exports (for testing) ---------------------------------------------------
export type { WSRelayData };
-export {
- server,
- handleHealth,
- handleIndex,
- handleRelay,
- getClientIP,
-};
+export { server, handleHealth, handleIndex, getClientIP };
diff --git a/src/lib/anthropic-proxy.ts b/src/lib/anthropic-proxy.ts
index 30e1e41..722364b 100644
--- a/src/lib/anthropic-proxy.ts
+++ b/src/lib/anthropic-proxy.ts
@@ -41,7 +41,6 @@ export interface AnthropicRequest {
temperature?: number;
top_p?: number;
top_k?: number;
- stream?: boolean;
stop_sequences?: string[];
system?: string | AnthropicSystemBlock[];
metadata?: Record;
@@ -429,6 +428,108 @@ interface OutputCounter {
chars: number;
}
+// --- Stream state machine helpers ---------------------------------------------
+
+/**
+ * Emit message_start and content_block_start events.
+ */
+function emitInitEvents(
+ controller: ReadableStreamDefaultController,
+ encoder: TextEncoder,
+ model: string,
+ messageId: string,
+ usage: AnthropicResponse["usage"],
+): void {
+ controller.enqueue(encoder.encode(
+ `event: message_start\ndata: ${JSON.stringify({
+ type: "message_start",
+ message: { id: messageId, type: "message", role: "assistant", content: [], model, stop_reason: null, stop_sequence: null, usage },
+ })}\n\n`,
+ ));
+ controller.enqueue(encoder.encode(
+ 'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}\n\n',
+ ));
+}
+
+/**
+ * Emit content_block_stop + message_delta + message_stop and close.
+ * Estimates output tokens from char count as fallback.
+ */
+function emitDoneEvents(
+ controller: ReadableStreamDefaultController,
+ encoder: TextEncoder,
+ usage: AnthropicResponse["usage"],
+ outputCounter: OutputCounter,
+): void {
+ if (usage.output_tokens === 0 && outputCounter.chars > 0) {
+ usage.output_tokens = Math.max(1, Math.round(outputCounter.chars / 4));
+ }
+
+ controller.enqueue(encoder.encode('event: content_block_stop\ndata: {"type":"content_block_stop","index":0}\n\n'));
+ controller.enqueue(encoder.encode(
+ `event: message_delta\ndata: ${JSON.stringify({
+ type: "message_delta",
+ delta: { stop_reason: "end_turn", stop_sequence: null },
+ usage: { output_tokens: usage.output_tokens },
+ })}\n\n`,
+ ));
+ controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n'));
+ controller.close();
+}
+
+/**
+ * Emit error event and close — dev mode includes the error detail.
+ */
+function emitErrorEvent(
+ controller: ReadableStreamDefaultController,
+ encoder: TextEncoder,
+ err: unknown,
+): void {
+ if (isDevMode()) {
+ controller.enqueue(encoder.encode(`event: error\ndata: ${JSON.stringify({ error: String(err) })}\n\n`));
+ } else {
+ controller.enqueue(encoder.encode('event: error\ndata: {"error":"Stream error"}\n\n'));
+ }
+ controller.close();
+}
+
+/**
+ * Process one chunk from the upstream reader through the SSE line buffer
+ * and emit adapted Anthropic SSE events for each complete line.
+ * Returns true if the stream is done (reader returned done=true).
+ */
+async function processStreamChunk(
+ reader: ReadableStreamDefaultReader,
+ lineBuffer: SSELineBuffer,
+ decoder: TextDecoder,
+ controller: ReadableStreamDefaultController,
+ encoder: TextEncoder,
+ model: string,
+ config: BackendConfig,
+ usage: AnthropicResponse["usage"],
+ outputCounter: OutputCounter,
+): Promise {
+ const { done, value } = await reader.read();
+ if (done) {
+ // Flush remaining buffered data
+ const remaining = lineBuffer.flush();
+ if (remaining.length > 0) {
+ const adapted = backendLineToAnthropicSSE(remaining, model, config, usage, outputCounter);
+ if (adapted) controller.enqueue(encoder.encode(adapted + "\n\n"));
+ }
+ return true;
+ }
+
+ const chunk = decoder.decode(value, { stream: true });
+ const lines = lineBuffer.add(chunk);
+
+ for (const line of lines) {
+ const adapted = backendLineToAnthropicSSE(line, model, config, usage, outputCounter);
+ if (adapted) controller.enqueue(encoder.encode(adapted + "\n\n"));
+ }
+ return false;
+}
+
// --- Stream transformer --------------------------------------------------------
function transformAnthropicStream(
@@ -437,41 +538,26 @@ function transformAnthropicStream(
config: BackendConfig,
): ReadableStream {
const reader = trackReader(body.getReader() as any as ReadableStreamDefaultReader);
- const decoder = new TextDecoder();
const encoder = new TextEncoder();
+ const decoder = new TextDecoder();
const lineBuffer = new SSELineBuffer();
let phase: "init" | "block" | "done" = "init";
- let messageId = `msg_${Date.now()}`;
const outputCounter: OutputCounter = { chars: 0 };
+ const usage: AnthropicResponse["usage"] = { input_tokens: 0, output_tokens: 0 };
- // Accumulate usage from backend SSE lines (some backends include usage
- // in their SSE data). Start with zeros, update as we parse lines.
- const usage: AnthropicResponse["usage"] = {
- input_tokens: 0,
- output_tokens: 0,
- };
-
- // SSE keepalive: send a comment every 15s to prevent LB/proxy timeout
let keepaliveTimer: ReturnType | null = null;
const KEEPALIVE_INTERVAL_MS = 15_000;
function startKeepalive(controller: ReadableStreamDefaultController) {
if (keepaliveTimer) return;
keepaliveTimer = setInterval(() => {
- try {
- controller.enqueue(encoder.encode(": keepalive\n\n"));
- } catch {
- if (keepaliveTimer) clearInterval(keepaliveTimer);
- }
+ try { controller.enqueue(encoder.encode(": keepalive\n\n")); } catch { stopKeepalive(); }
}, KEEPALIVE_INTERVAL_MS);
}
function stopKeepalive() {
- if (keepaliveTimer) {
- clearInterval(keepaliveTimer);
- keepaliveTimer = null;
- }
+ if (keepaliveTimer) { clearInterval(keepaliveTimer); keepaliveTimer = null; }
}
return new ReadableStream({
@@ -481,120 +567,35 @@ function transformAnthropicStream(
if (phase === "init") {
phase = "block";
- messageId = `msg_${Date.now()}`;
-
- const startEvent = `event: message_start\ndata: ${JSON.stringify({
- type: "message_start",
- message: {
- id: messageId,
- type: "message",
- role: "assistant",
- content: [],
- model,
- stop_reason: null,
- stop_sequence: null,
- usage,
- },
- })}`;
- controller.enqueue(encoder.encode(startEvent + "\n\n"));
-
- const blockStart = `event: content_block_start\ndata: ${JSON.stringify({
- type: "content_block_start",
- index: 0,
- content_block: { type: "text", text: "" },
- })}`;
- controller.enqueue(encoder.encode(blockStart + "\n\n"));
+ emitInitEvents(controller, encoder, model, `msg_${Date.now()}`, usage);
}
- while (phase === "block") {
- const BATCH_SIZE = 8;
- let chunksProcessed = 0;
+ const BATCH_SIZE = 8;
+ let chunksProcessed = 0;
- while (phase === "block") {
- const { done, value } = await reader.read();
- if (done) {
- stopKeepalive();
- releaseReader(reader);
- const remaining = lineBuffer.flush();
- if (remaining.length > 0) {
- const adapted = backendLineToAnthropicSSE(remaining, model, config, usage, outputCounter);
- if (adapted) {
- controller.enqueue(encoder.encode(adapted + "\n\n"));
- }
- }
- phase = "done";
- break;
- }
-
- const chunk = decoder.decode(value, { stream: true });
- const lines = lineBuffer.add(chunk);
-
- for (const line of lines) {
- const adapted = backendLineToAnthropicSSE(line, model, config, usage, outputCounter);
- if (adapted) {
- controller.enqueue(encoder.encode(adapted + "\n\n"));
- }
- }
-
- chunksProcessed++;
- if (chunksProcessed >= BATCH_SIZE) {
- chunksProcessed = 0;
- await new Promise((r) => setTimeout(r, 0));
- return;
- }
+ while (phase === "block" && chunksProcessed < BATCH_SIZE) {
+ const isDone = await processStreamChunk(reader, lineBuffer, decoder, controller, encoder, model, config, usage, outputCounter);
+ if (isDone) {
+ stopKeepalive();
+ releaseReader(reader);
+ phase = "done";
+ break;
}
+ chunksProcessed++;
+ }
+
+ if (chunksProcessed >= BATCH_SIZE) {
+ await new Promise((r) => setTimeout(r, 0));
+ return;
}
if (phase === "done") {
- phase = "done";
-
- // Estimate output tokens from char count as fallback
- // when the backend doesn't report usage in SSE lines.
- if (usage.output_tokens === 0 && outputCounter.chars > 0) {
- usage.output_tokens = Math.max(1, Math.round(outputCounter.chars / 4));
- }
-
- controller.enqueue(
- encoder.encode(
- 'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}\n\n',
- ),
- );
-
- controller.enqueue(
- encoder.encode(
- `event: message_delta\ndata: ${JSON.stringify({
- type: "message_delta",
- delta: { stop_reason: "end_turn", stop_sequence: null },
- usage: { output_tokens: usage.output_tokens },
- })}\n\n`,
- ),
- );
-
- controller.enqueue(
- encoder.encode(
- 'event: message_stop\ndata: {"type":"message_stop"}\n\n',
- ),
- );
-
- controller.close();
+ emitDoneEvents(controller, encoder, usage, outputCounter);
}
} catch (err) {
stopKeepalive();
releaseReader(reader);
- if (isDevMode()) {
- controller.enqueue(
- encoder.encode(
- `event: error\ndata: ${JSON.stringify({ error: String(err) })}\n\n`,
- ),
- );
- } else {
- controller.enqueue(
- encoder.encode(
- 'event: error\ndata: {"error":"Stream error"}\n\n',
- ),
- );
- }
- controller.close();
+ emitErrorEvent(controller, encoder, err);
}
},
cancel() {
@@ -687,6 +688,73 @@ function anthropicError(status: number, message: string, type: string): Response
);
}
+/**
+ * Handle a backend error response: parse upstream body, release session, return Anthropic error.
+ */
+async function handleUpstreamError(
+ response: Response,
+ sessionPool?: SessionProxyPool,
+ sessionId?: string,
+): Promise {
+ const status = response.status;
+ let upstreamMsg = status >= 500 ? "Upstream server error" : "Upstream rejected request";
+ try {
+ const errBody = await response.text();
+ if (errBody) {
+ const errJson = JSON.parse(errBody);
+ if (errJson?.error?.message) upstreamMsg = errJson.error.message;
+ else if (errJson?.type === "error" && errJson?.error?.message) upstreamMsg = errJson.error.message;
+ else if (errJson?.message) upstreamMsg = errJson.message;
+ else if (typeof errBody === "string" && errBody.length < 500) upstreamMsg = errBody;
+ }
+ } catch { /* keep default */ }
+ if (sessionPool && sessionId) sessionPool.release(sessionId);
+ return anthropicError(status, upstreamMsg, "upstream_error");
+}
+
+/** Build streaming response headers with CORS and cache forwarding. */
+function buildStreamHeaders(response: Response): Record {
+ const h: Record = {
+ "Content-Type": "text/event-stream", "Cache-Control": "no-cache",
+ Connection: "keep-alive", "Access-Control-Allow-Origin": "*", "X-Accel-Buffering": "no",
+ };
+ forwardCacheHeaders(response, h);
+ return h;
+}
+
+/** Build JSON response headers with CORS and cache forwarding. */
+function buildJsonHeaders(response: Response): Record {
+ const h: Record = {
+ "Content-Type": "application/json", "Access-Control-Allow-Origin": "*",
+ };
+ forwardCacheHeaders(response, h);
+ return h;
+}
+
+/**
+ * Handle a non-streaming backend response that came back in SSE format
+ * (it happens when the backend only supports SSE but we asked for non-stream).
+ * Accumulates the text deltas into a single Anthropic response.
+ */
+function handleBackendSSEExtract(
+ text: string,
+ model: string,
+): AnthropicResponse | null {
+ if (!text.trimStart().startsWith("data: ")) return null;
+ const accumulated = accumulateSSEText(text);
+ if (!accumulated) return null;
+ return {
+ id: `msg_${Date.now()}`,
+ type: "message",
+ role: "assistant",
+ content: [{ type: "text", text: accumulated }],
+ model,
+ stop_reason: "end_turn",
+ stop_sequence: null,
+ usage: { input_tokens: 0, output_tokens: 0 },
+ };
+}
+
// --- Main handler --------------------------------------------------------------
/**
@@ -758,11 +826,9 @@ export async function handleAnthropicMessages(
const wantsStream = req.stream === true;
// Default anthropic-version to 2023-06-01 (required for prompt caching).
- // The client can override via the anthropic-version header.
const version = anthropicVersion || "2023-06-01";
- // Translate Anthropic -> backend, preserving cache_control directives
- // and forwarding anthropic-version when available
+ // Translate Anthropic -> backend
const { body: backendBody, headers: extraHeaders } = anthropicToBackend(
req, config, backendModel, version,
);
@@ -782,23 +848,15 @@ export async function handleAnthropicMessages(
: await fetchWithRetry(url, init, proxyPool, `anthropic:${req.model}`, ipv6Source);
if (result.errorClassification) {
- if (sessionPool && sessionId) {
- sessionPool.release(sessionId);
- }
+ if (sessionPool && sessionId) sessionPool.release(sessionId);
return new Response(
JSON.stringify({
type: "error",
- error: {
- message: result.errorClassification.message,
- type: "server_error",
- },
+ error: { message: result.errorClassification.message, type: "server_error" },
}),
{
status: result.errorClassification.status,
- headers: {
- "Content-Type": "application/json",
- "Access-Control-Allow-Origin": "*",
- },
+ headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" },
},
);
}
@@ -807,124 +865,43 @@ export async function handleAnthropicMessages(
// -- Handle error responses from backend ------------------------------------
if (!response.ok) {
- const status = response.status;
- // Read the upstream error body so clients see the actual rejection reason,
- // not just a generic "Upstream rejected request" message.
- let upstreamMsg = status >= 500 ? "Upstream server error" : "Upstream rejected request";
- try {
- const errBody = await response.text();
- if (errBody) {
- const errJson = JSON.parse(errBody);
- // OpenAI-style: { error: { message, type } }
- if (errJson?.error?.message) {
- upstreamMsg = errJson.error.message;
- }
- // Anthropic-style: { type: "error", error: { message, type } }
- else if (errJson?.type === "error" && errJson?.error?.message) {
- upstreamMsg = errJson.error.message;
- }
- // Plain JSON with message field
- else if (errJson?.message) {
- upstreamMsg = errJson.message;
- }
- // Raw text error body
- else if (typeof errBody === "string" && errBody.length < 500) {
- upstreamMsg = errBody;
- }
- }
- } catch {
- // Could not read error body — keep generic message
- }
- if (sessionPool && sessionId) {
- sessionPool.release(sessionId);
- }
- return anthropicError(status, upstreamMsg, "upstream_error");
+ return handleUpstreamError(response, sessionPool, sessionId);
}
// -- For native Anthropic passthrough, relay the raw backend response -------
if (config.anthropicPassthrough) {
if (wantsStream) {
- const headers: Record = {
- "Content-Type": "text/event-stream",
- "Cache-Control": "no-cache",
- Connection: "keep-alive",
- "Access-Control-Allow-Origin": "*",
- "X-Accel-Buffering": "no",
- };
- forwardCacheHeaders(response, headers);
return new Response(wrapAnthropicStreamMaybe(response.body!, sessionPool, sessionId), {
status: 200,
- headers,
+ headers: buildStreamHeaders(response),
});
}
const rawBody = await response.text();
- if (sessionPool && sessionId) {
- sessionPool.release(sessionId);
- }
- const headers: Record = {
- "Content-Type": "application/json",
- "Access-Control-Allow-Origin": "*",
- };
- forwardCacheHeaders(response, headers);
- return new Response(rawBody, { status: 200, headers });
+ if (sessionPool && sessionId) sessionPool.release(sessionId);
+ return new Response(rawBody, { status: 200, headers: buildJsonHeaders(response) });
}
// -- Handle streaming (OpenAI-compatible backend) ---------------------------
if (wantsStream) {
- let transformed = transformAnthropicStream(
- response.body!,
- req.model,
- config,
- );
+ let transformed = transformAnthropicStream(response.body!, req.model, config);
transformed = wrapAnthropicStreamMaybe(transformed, sessionPool, sessionId);
- const headers: Record = {
- "Content-Type": "text/event-stream",
- "Cache-Control": "no-cache",
- Connection: "keep-alive",
- "Access-Control-Allow-Origin": "*",
- "X-Accel-Buffering": "no",
- };
- forwardCacheHeaders(response, headers);
- return new Response(transformed, { status: 200, headers });
+ return new Response(transformed, { status: 200, headers: buildStreamHeaders(response) });
}
// -- Handle non-streaming (OpenAI-compatible backend) -----------------------
const text = await response.text();
- if (sessionPool && sessionId) {
- sessionPool.release(sessionId);
- }
+ if (sessionPool && sessionId) sessionPool.release(sessionId);
- const buildBaseHeaders = (): Record => {
- const h: Record = {
- "Content-Type": "application/json",
- "Access-Control-Allow-Origin": "*",
- };
- forwardCacheHeaders(response, h);
- return h;
- };
-
- if (text.trimStart().startsWith("data: ")) {
- const accumulated = accumulateSSEText(text);
- if (accumulated) {
- return new Response(
- JSON.stringify({
- id: `msg_${Date.now()}`,
- type: "message",
- role: "assistant",
- content: [{ type: "text", text: accumulated }],
- model: req.model,
- stop_reason: "end_turn",
- stop_sequence: null,
- usage: { input_tokens: 0, output_tokens: 0 },
- }),
- {
- status: 200,
- headers: buildBaseHeaders(),
- },
- );
- }
+ // Backend returned SSE even though we didn't ask for stream
+ const sseAdapted = handleBackendSSEExtract(text, req.model);
+ if (sseAdapted) {
+ return new Response(JSON.stringify(sseAdapted), {
+ status: 200,
+ headers: buildJsonHeaders(response),
+ });
}
+ // Parse JSON and adapt
let parsed: any;
try {
parsed = JSON.parse(text);
@@ -933,10 +910,9 @@ export async function handleAnthropicMessages(
}
const adapted = backendToAnthropicResponse(parsed, req.model);
-
return new Response(JSON.stringify(adapted), {
status: 200,
- headers: buildBaseHeaders(),
+ headers: buildJsonHeaders(response),
});
}
diff --git a/src/lib/router.ts b/src/lib/router.ts
index 3a75a16..9b4908d 100644
--- a/src/lib/router.ts
+++ b/src/lib/router.ts
@@ -1,3 +1,20 @@
+/**
+ * Consolidated router — single source of truth for all route handlers.
+ *
+ * Exports every shared handler that the 3 entry points (standalone Bun,
+ * Vercel, Cloudflare Workers) need.
+ *
+ * Handles:
+ * - Health check (GET /health)
+ * - Index page (GET /)
+ * - API docs (GET /docs)
+ * - Auth (API_KEY check)
+ * - AI proxy (OpenAI /v1/chat/completions, Anthropic /v1/messages, /v1/models)
+ * - Generic relay (x-relay-target)
+ * - WebSocket (upgrade check)
+ * - SSRF DNS rebinding protection
+ */
+
import {
normalizeTargetUrl,
isAllowedTarget,
@@ -6,6 +23,7 @@ import {
filterRequestHeaders,
buildRelayRequest,
createRelayResponse,
+ classifyFetchError,
createErrorResponse,
createCorsPreflightResponse,
getCorsHeaders,
@@ -30,15 +48,14 @@ export interface RouterEnv {
CORS_ORIGIN?: string;
NODE_ENV?: string;
API_KEY?: string;
- PROXY_LIST?: string; // Comma-separated list of proxies for serverless
- // Optional KV binding for rate limiter
+ PROXY_LIST?: string;
KV?: {
get(key: string): Promise;
put(key: string, value: any, options?: { expirationTtl?: number }): Promise;
};
}
-// --- Global singletons (survives warm starts) --------------------------------
+// --- Globals (module-scoped singletons) --------------------------------------
let rateLimiter: ReturnType | null = null;
let proxyPool: ProxyPool | null = null;
@@ -49,29 +66,56 @@ const RELAY_VERSION = "1.0.0";
// --- Helpers -----------------------------------------------------------------
-function getNumericEnv(env: RouterEnv, key: keyof RouterEnv, fallback: number): number {
+function getNumericEnv(
+ env: RouterEnv,
+ key: keyof RouterEnv,
+ fallback: number,
+): number {
const raw = env[key];
- const val = typeof raw === "string" ? raw : (typeof process !== "undefined" ? process.env[key as string] : undefined);
+ const val =
+ typeof raw === "string"
+ ? raw
+ : typeof process !== "undefined"
+ ? process.env[key as string]
+ : undefined;
return Number.parseInt(val ?? String(fallback), 10);
}
-function getEnv(env: RouterEnv, key: keyof RouterEnv, fallback: string): string {
+function getEnv(
+ env: RouterEnv,
+ key: keyof RouterEnv,
+ fallback: string,
+): string {
const raw = env[key];
- const fromEnv = typeof raw === "string" ? raw : (typeof process !== "undefined" ? process.env[key as string] : undefined);
+ const fromEnv =
+ typeof raw === "string"
+ ? raw
+ : typeof process !== "undefined"
+ ? process.env[key as string]
+ : undefined;
return fromEnv ?? fallback;
}
+/** Determine if an error is a JSON parse failure. */
+function isJsonParseError(err: unknown): boolean {
+ return err instanceof Error && (err.name === "SyntaxError" || err.message.includes("JSON"));
+}
+
+// --- Init --------------------------------------------------------------------
+
function initGlobals(env: RouterEnv) {
if (!rateLimiter) {
- const kvAdapter = env.KV ? {
- get: async (k: string) => {
- const val = await env.KV!.get(k);
- return val ? JSON.parse(val) : null;
- },
- set: async (k: string, v: number[], ttl?: number) => {
- await env.KV!.put(k, JSON.stringify(v), { expirationTtl: ttl });
- }
- } : undefined;
+ const kvAdapter = env.KV
+ ? {
+ get: async (k: string) => {
+ const val = await env.KV!.get(k);
+ return val ? JSON.parse(val) : null;
+ },
+ set: async (k: string, v: number[], ttl?: number) => {
+ await env.KV!.put(k, JSON.stringify(v), { expirationTtl: ttl });
+ },
+ }
+ : undefined;
rateLimiter = createRateLimiter({
maxRequests: getNumericEnv(env, "RATE_LIMIT_MAX", 100),
@@ -82,7 +126,6 @@ function initGlobals(env: RouterEnv) {
if (!proxyPool) {
proxyPool = new ProxyPool();
- // For Bun (process.env.PROXY_FILE) it's loaded in index.ts, but for serverless we can load from env
const proxies = getEnv(env, "PROXY_LIST", "");
if (proxies) {
for (const p of proxies.split(",")) {
@@ -95,11 +138,21 @@ function initGlobals(env: RouterEnv) {
}
}
-function requireAuth(req: Request, env: RouterEnv): Response | null {
- const API_KEY = getEnv(env, "API_KEY", "sk-dummy-key");
- const header = req.headers.get("authorization") ?? req.headers.get("x-api-key") ?? "";
+// --- Auth --------------------------------------------------------------------
+
+function requireAuth(req: Request, env?: RouterEnv): Response | null {
+ const API_KEY = env
+ ? getEnv(env, "API_KEY", "")
+ : (process.env.API_KEY ?? "");
+ if (!API_KEY) return null; // auth disabled
+
+ const header =
+ req.headers.get("authorization") ??
+ req.headers.get("x-api-key") ??
+ "";
const key = header.replace(/^Bearer\s+/i, "").trim();
if (key === API_KEY) return null;
+
return new Response(
JSON.stringify({ error: { message: "Unauthorized", type: "auth_error" } }),
{
@@ -109,6 +162,35 @@ function requireAuth(req: Request, env: RouterEnv): Response | null {
);
}
+// --- IP Detection ------------------------------------------------------------
+
+function getClientIP(req: Request): string {
+ const forwarded = req.headers.get("x-forwarded-for");
+ if (forwarded) {
+ const first = forwarded.split(",")[0]?.trim();
+ if (first) return first;
+ }
+ const cfIp = req.headers.get("cf-connecting-ip");
+ if (cfIp) return cfIp;
+ return "unknown";
+}
+
+function getClientIPFromServer(
+ req: Request,
+ ipGetter: { requestIP(req: Request): { address: string } | null },
+): string {
+ const forwarded = req.headers.get("x-forwarded-for");
+ if (forwarded) {
+ const first = forwarded.split(",")[0]?.trim();
+ if (first) return first;
+ }
+ const cfIp = req.headers.get("cf-connecting-ip");
+ if (cfIp) return cfIp;
+ const remote = ipGetter.requestIP(req);
+ if (remote) return remote.address;
+ return "unknown";
+}
+
// --- Static Handlers ---------------------------------------------------------
function handleHealth(): Response {
@@ -120,45 +202,40 @@ function handleHealth(): Response {
}),
{
status: 200,
- headers: {
- "Content-Type": "application/json",
- ...getCorsHeaders(),
- },
+ headers: { "Content-Type": "application/json", ...getCorsHeaders() },
},
);
}
function handleIndex(): Response {
const html = `
-
-
-
-
- Edge Proxy Relay
-
-
-
-
- Edge Proxy Relay
- Server is running
- /health · /docs
-
-
- `;
+
+
+
+
+ Edge Proxy Relay
+
+
+
+
+ Edge Proxy Relay
+ Server is running
+ /health · /docs
+
+
+`;
return new Response(html, {
status: 200,
- headers: {
- "Content-Type": "text/html; charset=utf-8",
- },
+ headers: { "Content-Type": "text/html; charset=utf-8" },
});
}
@@ -214,9 +291,9 @@ function handleDocs(_isWebSocketSupported: boolean): Response {
▶
@@ -307,7 +384,8 @@ function handleDocs(_isWebSocketSupported: boolean): Response {