From fc69a0e93cd3cfb3b4c43c85cea265d3b98fd3f7 Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Wed, 24 Jun 2026 23:48:16 +0700 Subject: [PATCH] refactor: konsolidasi handler ke router.ts + complexity + test coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit === DRY Konsolidasi === - handleRelay, requireAuth, handleHealth, handleIndex, getClientIP → 1x di src/lib/router.ts - api/relay.ts, src/worker.ts, src/index.ts jadi thin wrapper === Dead exports === - 35 dead exports dibersihkan (dari router.ts, worker.ts, relay.ts) - Duplikasi simbol dari 12 ke 3 (wajar) === Keamanan === - Default API key sk-dummy-key dihapus — requireAuth return null saat key kosong - Semua CORS konsisten via getCorsHeaders() === Kompleksitas === - handleRequest → pecah ke validateRelayTarget, createJsonErrorResponse, dll - transformAnthropicStream (170 baris) → 4 sub-fungsi (emitInitEvents, emitDoneEvents, etc) - handleAnthropicMessages → handleUpstreamError, buildStreamHeaders, buildJsonHeaders - Fix duplicate stream? field di AnthropicRequest interface === Test Coverage === - Test naik dari 171 ke 248 (+77 test) - File baru: src/router.test.ts (requireAuth, getClientIP, CORS, health, index) - File baru: src/relay-integration.test.ts (filterResponseHeaders, shouldSendBody, buildRelayRequest, classifyFetchError, createRelayResponse, normalizeTargetUrl, SSRF, isPrivateIp) Co-Authored-By: Claude Opus 4.8 (1M context) --- api/relay.ts | 393 ++-------------------- src/index.test.ts | 6 +- src/index.ts | 494 ++++----------------------- src/lib/anthropic-proxy.ts | 446 ++++++++++++------------ src/lib/router.ts | 617 +++++++++++++++++++--------------- src/relay-integration.test.ts | 403 ++++++++++++++++++++++ src/router.test.ts | 261 ++++++++++++++ src/worker.ts | 390 ++------------------- 8 files changed, 1343 insertions(+), 1667 deletions(-) create mode 100644 src/relay-integration.test.ts create mode 100644 src/router.test.ts 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

-
- Interactive Test Page - Health Check -
-
- -`; - - 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 {