From b09bca8def3a0d32157bc5c3bdc0a3185dc1a57e Mon Sep 17 00:00:00 2001 From: MythEclipse Date: Thu, 11 Jun 2026 03:56:26 +0700 Subject: [PATCH] feat(core): implement robust fetch utilities and stream management Introduce a centralized `fetch-utils.ts` to handle retry logic with proxy fallback, SSE line buffering to prevent chunk-boundary corruption, and graceful shutdown via active reader tracking. Key changes: - Add `fetchWithRetry` for automatic direct-to-proxy failover. - Implement `SSELineBuffer` to ensure reliable parsing of split SSE chunks. - Add `createStreamBodyLimiter` to enforce payload limits on streaming requests. - Refactor `ProxyPool` to decouple failure marking from rotation. - Standardize CORS handling and environment variable configuration. - Clean up documentation and remove obsolete skill files. --- .claude/skills/debug-issue.md | 27 -- .claude/skills/explore-codebase.md | 28 -- .claude/skills/refactor-safely.md | 28 -- .claude/skills/review-changes.md | 29 -- src/index.ts | 147 +++--- src/lib/ai-proxy.ts | 252 ++++++----- src/lib/anthropic-proxy.ts | 254 ++++++----- src/lib/fetch-utils.ts | 209 +++++++++ src/lib/proxy-pool.ts | 39 +- src/lib/relay-utils.test.ts | 2 +- src/lib/relay-utils.ts | 696 ++++++++++++++++------------- src/middleware/body-limiter.ts | 54 ++- src/worker.ts | 79 ++-- 13 files changed, 1022 insertions(+), 822 deletions(-) delete mode 100644 .claude/skills/debug-issue.md delete mode 100644 .claude/skills/explore-codebase.md delete mode 100644 .claude/skills/refactor-safely.md delete mode 100644 .claude/skills/review-changes.md create mode 100644 src/lib/fetch-utils.ts diff --git a/.claude/skills/debug-issue.md b/.claude/skills/debug-issue.md deleted file mode 100644 index ef1b38a..0000000 --- a/.claude/skills/debug-issue.md +++ /dev/null @@ -1,27 +0,0 @@ ---- -name: Debug Issue -description: Systematically debug issues using graph-powered code navigation ---- - -## Debug Issue - -Use the knowledge graph to systematically trace and debug issues. - -### Steps - -1. Use `semantic_search_nodes` to find code related to the issue. -2. Use `query_graph` with `callers_of` and `callees_of` to trace call chains. -3. Use `get_flow` to see full execution paths through suspected areas. -4. Run `detect_changes` to check if recent changes caused the issue. -5. Use `get_impact_radius` on suspected files to see what else is affected. - -### Tips - -- Check both callers and callees to understand the full context. -- Look at affected flows to find the entry point that triggers the bug. -- Recent changes are the most common source of new issues. - -## Token Efficiency Rules -- ALWAYS start with `get_minimal_context(task="")` before any other graph tool. -- Use `detail_level="minimal"` on all calls. Only escalate to "standard" when minimal is insufficient. -- Target: complete any review/debug/refactor task in ≤5 tool calls and ≤800 total output tokens. diff --git a/.claude/skills/explore-codebase.md b/.claude/skills/explore-codebase.md deleted file mode 100644 index dc7ad10..0000000 --- a/.claude/skills/explore-codebase.md +++ /dev/null @@ -1,28 +0,0 @@ ---- -name: Explore Codebase -description: Navigate and understand codebase structure using the knowledge graph ---- - -## Explore Codebase - -Use the code-review-graph MCP tools to explore and understand the codebase. - -### Steps - -1. Run `list_graph_stats` to see overall codebase metrics. -2. Run `get_architecture_overview` for high-level community structure. -3. Use `list_communities` to find major modules, then `get_community` for details. -4. Use `semantic_search_nodes` to find specific functions or classes. -5. Use `query_graph` with patterns like `callers_of`, `callees_of`, `imports_of` to trace relationships. -6. Use `list_flows` and `get_flow` to understand execution paths. - -### Tips - -- Start broad (stats, architecture) then narrow down to specific areas. -- Use `children_of` on a file to see all its functions and classes. -- Use `find_large_functions` to identify complex code. - -## Token Efficiency Rules -- ALWAYS start with `get_minimal_context(task="")` before any other graph tool. -- Use `detail_level="minimal"` on all calls. Only escalate to "standard" when minimal is insufficient. -- Target: complete any review/debug/refactor task in ≤5 tool calls and ≤800 total output tokens. diff --git a/.claude/skills/refactor-safely.md b/.claude/skills/refactor-safely.md deleted file mode 100644 index cf84420..0000000 --- a/.claude/skills/refactor-safely.md +++ /dev/null @@ -1,28 +0,0 @@ ---- -name: Refactor Safely -description: Plan and execute safe refactoring using dependency analysis ---- - -## Refactor Safely - -Use the knowledge graph to plan and execute refactoring with confidence. - -### Steps - -1. Use `refactor_tool` with mode="suggest" for community-driven refactoring suggestions. -2. Use `refactor_tool` with mode="dead_code" to find unreferenced code. -3. For renames, use `refactor_tool` with mode="rename" to preview all affected locations. -4. Use `apply_refactor_tool` with the refactor_id to apply renames. -5. After changes, run `detect_changes` to verify the refactoring impact. - -### Safety Checks - -- Always preview before applying (rename mode gives you an edit list). -- Check `get_impact_radius` before major refactors. -- Use `get_affected_flows` to ensure no critical paths are broken. -- Run `find_large_functions` to identify decomposition targets. - -## Token Efficiency Rules -- ALWAYS start with `get_minimal_context(task="")` before any other graph tool. -- Use `detail_level="minimal"` on all calls. Only escalate to "standard" when minimal is insufficient. -- Target: complete any review/debug/refactor task in ≤5 tool calls and ≤800 total output tokens. diff --git a/.claude/skills/review-changes.md b/.claude/skills/review-changes.md deleted file mode 100644 index 6bb3558..0000000 --- a/.claude/skills/review-changes.md +++ /dev/null @@ -1,29 +0,0 @@ ---- -name: Review Changes -description: Perform a structured code review using change detection and impact ---- - -## Review Changes - -Perform a thorough, risk-aware code review using the knowledge graph. - -### Steps - -1. Run `detect_changes` to get risk-scored change analysis. -2. Run `get_affected_flows` to find impacted execution paths. -3. For each high-risk function, run `query_graph` with pattern="tests_for" to check test coverage. -4. Run `get_impact_radius` to understand the blast radius. -5. For any untested changes, suggest specific test cases. - -### Output Format - -Provide findings grouped by risk level (high/medium/low) with: -- What changed and why it matters -- Test coverage status -- Suggested improvements -- Overall merge recommendation - -## Token Efficiency Rules -- ALWAYS start with `get_minimal_context(task="")` before any other graph tool. -- Use `detail_level="minimal"` on all calls. Only escalate to "standard" when minimal is insufficient. -- Target: complete any review/debug/refactor task in ≤5 tool calls and ≤800 total output tokens. diff --git a/src/index.ts b/src/index.ts index 4f653e3..f78191a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -5,12 +5,14 @@ * `x-relay-target` request header. Supports WebSocket upgrades * when the target uses `ws://` or `wss://`. * - * ── Environment Variables ─────────────────────────────────────── + * --- 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 */ import { @@ -22,6 +24,7 @@ import { classifyFetchError, createErrorResponse, createCorsPreflightResponse, + getCorsHeaders, } from "./lib/relay-utils"; import { checkBodySize } from "./middleware/body-limiter"; @@ -30,10 +33,11 @@ import { logRelayEvent } from "./middleware/logger"; import { ProxyPool } from "./lib/proxy-pool"; import { handleChatCompletion, listModels } from "./lib/ai-proxy"; import { handleAnthropicMessages } from "./lib/anthropic-proxy"; +import { fetchWithRetry, closeAllActiveReaders, isDevMode } from "./lib/fetch-utils"; import type { Server, ServerWebSocket } from "bun"; -// ─── Configuration ────────────────────────────────────────────────────────────── +// --- Configuration ------------------------------------------------------------ const PORT = Number.parseInt(process.env.PORT ?? "3000", 10); const RELAY_TIMEOUT_MS = Number.parseInt( @@ -43,9 +47,9 @@ const RELAY_TIMEOUT_MS = Number.parseInt( const SERVER_START_TIME = Date.now(); const RELAY_VERSION = "1.0.0"; -// ─── API Key Authentication ───────────────────────────────────────────────────── +// --- API Key Authentication --------------------------------------------------- -const API_KEY = "sk-dummy-key"; +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") ?? ""; @@ -53,11 +57,14 @@ function requireAuth(req: Request): Response | null { 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": "*" } }, + { + status: 401, + headers: { "Content-Type": "application/json", ...getCorsHeaders() }, + }, ); } -// ─── Middleware instances (singletons) ─────────────────────────────────────────── +// --- Middleware instances (singletons) ---------------------------------------- const rateLimiter = createRateLimiter({ maxRequests: Number.parseInt(process.env.RATE_LIMIT_MAX ?? "100", 10), @@ -67,14 +74,14 @@ const rateLimiter = createRateLimiter({ ), }); -// ─── Proxy pool (optional) ─────────────────────────────────────────────────────── +// --- Proxy pool (optional) ---------------------------------------------------- const proxyPool = new ProxyPool(); proxyPool.tryLoad( process.env.PROXY_FILE || process.env.PROXY_LIST || "./proxy.txt", ); -// ─── WebSocket relay data type ────────────────────────────────────────────────── +// --- WebSocket relay data type ----------------------------------------------- interface WSRelayData { target: string; @@ -82,7 +89,7 @@ interface WSRelayData { upstream?: WebSocket; } -// ─── Route handlers ──────────────────────────────────────────────────────────── +// --- Route handlers ---------------------------------------------------------- /** Health check endpoint: returns status, uptime, and version. */ function handleHealth(): Response { @@ -96,7 +103,7 @@ function handleHealth(): Response { status: 200, headers: { "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), }, }, ); @@ -189,7 +196,7 @@ ws.onmessage = (e) => console.log("Got:", e.data); status: 200, headers: { "Content-Type": "text/html; charset=utf-8", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), }, }); } @@ -229,7 +236,7 @@ function handleIndex(): Response { }); } -// ─── HTTP Relay Logic ────────────────────────────────────────────────────────── +// --- HTTP Relay Logic ------------------------------------------------------- /** * Get the client IP address from the request. @@ -271,12 +278,12 @@ async function handleRelay( const clientIP = getClientIP(req, ipGetter); const requestUrl = req.url; - // ── Pre-flight CORS ────────────────────────────────────────────── + // -- Pre-flight CORS -------------------------------------------------------- if (method === "OPTIONS") { return createCorsPreflightResponse(); } - // ── Middleware: Body size check ────────────────────────────────── + // -- Middleware: Body size check ------------------------------------------- const bodyError = checkBodySize(req); if (bodyError) { logRelayEvent({ @@ -289,7 +296,7 @@ async function handleRelay( return bodyError; } - // ── Middleware: Rate limiting ──────────────────────────────────── + // -- Middleware: Rate limiting --------------------------------------------- const rateCheck = rateLimiter.check(clientIP); if (!rateCheck.allowed) { logRelayEvent({ @@ -311,7 +318,7 @@ async function handleRelay( status: 429, headers: { "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), "Retry-After": String( Math.ceil((rateCheck.retryAfterMs ?? 60_000) / 1000), ), @@ -320,11 +327,11 @@ async function handleRelay( ); } - // ── Extract relay parameters from headers ─────────────────────── + // -- 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 ────────────────────── + // -- SSRF: Normalize and validate target URL -------------------------------- const targetUrl = normalizeTargetUrl(target, relayPath); if (!targetUrl) { logRelayEvent({ @@ -358,7 +365,7 @@ async function handleRelay( }); } - // ── Build the upstream request ─────────────────────────────────── + // -- Build the upstream request --------------------------------------------- const filteredHeaders = filterRequestHeaders(req.headers); const fetchOptions = buildRelayRequest( req, @@ -368,55 +375,28 @@ async function handleRelay( const targetUrlString = targetUrl.toString(); - // ── Execute upstream fetch ────────────────────────────────────── - // Strategy: direct first → proxy on failure → rotate on failure - let response: Response | undefined; - let usedProxy = false; + // -- Execute upstream fetch with shared retry ------------------------------- + const result = await fetchWithRetry( + targetUrlString, + fetchOptions, + proxyPool, + "relay", + ); - for (let attempts = 0; attempts < 3; attempts++) { - // Clear proxy on first attempt (direct) - if (attempts === 0) { - delete fetchOptions.proxy; - } else if (attempts === 1 && proxyPool.size > 0) { - // Second attempt: use first proxy - usedProxy = true; - fetchOptions.proxy = proxyPool.getProxyUrl()!; - } else if (attempts === 2 && proxyPool.size > 0) { - // Third attempt: rotate to next proxy - const next = proxyPool.markFailed(); - if (!next) break; - fetchOptions.proxy = proxyPool.getProxyUrl()!; - } else { - break; - } - - try { - response = await fetch(targetUrlString, fetchOptions); - if (usedProxy) proxyPool.markSuccess(); - break; - } catch { - // Fall through to next attempt - } - } - - // All attempts failed — classify the last error - if (!response) { - const lastErr = new Error("All connection attempts failed"); - const classified = classifyFetchError(lastErr); + if (result.errorClassification) { logRelayEvent({ method, url: requestUrl, - status: classified.status, + status: result.errorClassification.status, durationMs: Math.round(performance.now() - startTime), - error: classified.message, + error: result.errorClassification.message, targetUrl: targetUrlString, ip: clientIP, }); - return createErrorResponse(classified); + return createErrorResponse(result.errorClassification); } - // ── Build relay response ───────────────────────────────────────── - const relayedResponse = createRelayResponse(response); + const relayedResponse = createRelayResponse(result.response!); logRelayEvent({ method, @@ -430,7 +410,7 @@ async function handleRelay( return relayedResponse; } -// ─── WebSocket Relay Logic ───────────────────────────────────────────────────── +// --- WebSocket Relay Logic --------------------------------------------------- /** * Upgrade an HTTP request to a WebSocket and relay bidirectionally to the @@ -452,7 +432,6 @@ function handleWebSocketUpgrade( const relayPath = req.headers.get("x-relay-path") ?? "/"; - // Normalize the target URL to verify it's valid const normalized = normalizeTargetUrl(target, relayPath); if (!normalized) return undefined; if (!isAllowedTarget(new URL(normalized.toString()))) return undefined; @@ -467,18 +446,14 @@ function handleWebSocketUpgrade( return new Response("WebSocket upgrade failed", { status: 400 }); } - // Returning undefined signals Bun that the upgrade was handled return undefined; } -// ─── Server ───────────────────────────────────────────────────────────────────── +// --- Server ------------------------------------------------------------------ const server: Server = Bun.serve({ port: PORT, - development: { - hmr: true, - console: true, - }, + development: isDevMode() ? { hmr: true, console: true } : undefined, async fetch(req: Request): Promise { const url = new URL(req.url); @@ -487,9 +462,9 @@ const server: Server = Bun.serve({ if (url.pathname === "/health") return handleHealth(); if (url.pathname === "/docs") return handleDocs(); if (url.pathname === "/" && req.method === "GET" && !req.headers.get("x-relay-target")) - return handleIndex(); + return handleIndex(); - // AI proxy routes — OpenAI-compatible API + // AI proxy routes -- OpenAI-compatible API if (url.pathname === "/v1/chat/completions") { if (req.method === "OPTIONS") { return createCorsPreflightResponse(); @@ -502,15 +477,15 @@ const server: Server = Bun.serve({ try { const body = await req.json(); return handleChatCompletion(body, proxyPool); - } catch (e) { + } catch { 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": "*" } }, + { status: 400, headers: { "Content-Type": "application/json", ...getCorsHeaders() } }, ); } } - // AI proxy routes — Anthropic-compatible API + // AI proxy routes -- Anthropic-compatible API if (url.pathname === "/v1/messages") { if (req.method === "OPTIONS") { return createCorsPreflightResponse(); @@ -523,13 +498,13 @@ const server: Server = Bun.serve({ try { const body = await req.json(); return handleAnthropicMessages(body, proxyPool); - } catch (e) { + } catch { 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": "*" } }, + { status: 400, headers: { "Content-Type": "application/json", ...getCorsHeaders() } }, ); } } @@ -551,22 +526,19 @@ const server: Server = Bun.serve({ status: 200, headers: { "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), }, }, ); } - // WebSocket upgrade check — if the target is ws:// or wss://, - // attempt to upgrade and relay. This must happen before the - // general HTTP relay. + // WebSocket upgrade check if ( req.method === "GET" && req.headers.get("upgrade")?.toLowerCase() === "websocket" ) { const wsResult = handleWebSocketUpgrade(req, server); if (wsResult === undefined) { - // Upgrade was handled by Bun — return undefined return undefined; } return wsResult; @@ -588,11 +560,10 @@ const server: Server = Bun.serve({ targetUrl: target, }); - // Connect to the upstream WebSocket const upstream = new WebSocket(target); upstream.onopen = () => { - // Connection established — ready for bidirectional relay + // Connection established }; upstream.onmessage = (event: MessageEvent) => { @@ -618,7 +589,6 @@ const server: Server = Bun.serve({ ws.close(event.code || 1000, event.reason || "Upstream closed"); }; - // Store the upstream so we can close it on client disconnect ws.data.upstream = upstream; }, @@ -645,21 +615,28 @@ const server: Server = Bun.serve({ }, drain(_ws: ServerWebSocket) { - // Backpressure not implemented in this minimal relay + // Backpressure not implemented }, }, }); -// ─── Startup ─────────────────────────────────────────────────────────────────── +// --- Startup ----------------------------------------------------------------- console.log( `[relay] Edge Proxy Relay v${RELAY_VERSION} listening on http://localhost:${server.port}`, ); +if (isDevMode()) { + console.log("[relay] Development mode: HMR enabled"); +} -// ─── Graceful Shutdown ───────────────────────────────────────────────────────── +// --- Graceful Shutdown ------------------------------------------------------- const shutdownHandler = (signal: string) => { console.log(`\n[relay] Received ${signal}, shutting down gracefully...`); + + // Close active SSE streams so clients get proper stream end events + closeAllActiveReaders(); + server.stop(); process.exit(0); }; @@ -667,7 +644,7 @@ const shutdownHandler = (signal: string) => { process.on("SIGTERM", () => shutdownHandler("SIGTERM")); process.on("SIGINT", () => shutdownHandler("SIGINT")); -// ─── Exports (for testing) ───────────────────────────────────────────────────── +// --- Exports (for testing) --------------------------------------------------- export type { WSRelayData }; export { diff --git a/src/lib/ai-proxy.ts b/src/lib/ai-proxy.ts index 41cc6c3..aa87a52 100644 --- a/src/lib/ai-proxy.ts +++ b/src/lib/ai-proxy.ts @@ -13,8 +13,11 @@ */ import type { ProxyPool } from "./proxy-pool"; +import { fetchWithRetry } from "./fetch-utils"; +import { SSELineBuffer } from "./fetch-utils"; +import { isDevMode } from "./fetch-utils"; -// ─── Types ─────────────────────────────────────────────────────────────────────── +// --- Types ------------------------------------------------------------------- export interface OpenAIRequest { model: string; @@ -44,11 +47,11 @@ export interface BackendConfig { adaptStreamLine?: (line: string, req: OpenAIRequest) => string | null; } -// ─── Model routing table ───────────────────────────────────────────────────────── +// --- Model routing table ------------------------------------------------------- -/** Map of model name → backend configuration. Exported for reuse by anthropic-proxy. */ +/** Map of model name -> backend configuration. */ export const MODEL_ROUTES: Record = { - // ── opencode.ai (OpenAI-compatible — passthrough) ──────────────── + // -- opencode.ai (OpenAI-compatible -- passthrough) -------------------------- "deepseek-v4-flash-free": { provider: "opencode", url: "https://opencode.ai/zen/v1/chat/completions", @@ -57,7 +60,7 @@ export const MODEL_ROUTES: Record = { }, }, - // ── surfsense.com (custom format) ─────────────────────────────── + // -- surfsense.com (custom format) ------------------------------------------- "gpt-5.4-mini-no-login": { provider: "surfsense", url: "https://api.surfsense.com/api/v1/public/anon-chat/stream", @@ -119,7 +122,7 @@ export const MODEL_ROUTES: Record = { }), }, - // ── deep-seek.ai (custom format) ──────────────────────────────── + // -- deep-seek.ai (custom format) -------------------------------------------- "deepseek/deepseek-v4-flash": { provider: "deepseek", url: "https://deep-seek.ai/api/chat", @@ -133,11 +136,8 @@ export const MODEL_ROUTES: Record = { "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/148.0.0.0 Safari/537.36", }, adaptStreamLine: (line) => { - // deep-seek.ai may return plain text chunks or SSE-like data if (!line || line.trim().length === 0) return null; - // If it's already SSE format, try to pass through if (line.startsWith("data: ")) { - // Rewrite the id and object fields try { const parsed = JSON.parse(line.slice(6)); parsed.id = `chatcmpl-${Date.now()}`; @@ -149,7 +149,6 @@ export const MODEL_ROUTES: Record = { return line; } } - // Plain text chunks — wrap in OpenAI SSE format return `data: ${JSON.stringify({ id: `chatcmpl-${Date.now()}`, object: "chat.completion.chunk", @@ -184,7 +183,7 @@ export const MODEL_ROUTES: Record = { }, }; -// ─── Helpers ───────────────────────────────────────────────────────────────────── +// --- Helpers ------------------------------------------------------------------ /** List all available model names. */ export function listModels(): string[] { @@ -196,7 +195,7 @@ export function resolveModel(model: string): BackendConfig | undefined { return MODEL_ROUTES[model]; } -// ─── Request building ──────────────────────────────────────────────────────────── +// --- Request building ---------------------------------------------------------- /** * Build the backend `fetch()` options from an OpenAI-style request. @@ -204,7 +203,6 @@ export function resolveModel(model: string): BackendConfig | undefined { function buildBackendRequest( req: OpenAIRequest, config: BackendConfig, - proxyPool?: ProxyPool, ): { url: string; init: RequestInit & { proxy?: string } } { const body = config.adaptRequest?.(req) ?? { @@ -223,15 +221,10 @@ function buildBackendRequest( body: JSON.stringify(body), }; - // Direct first, proxy as fallback (if pool available) - if (proxyPool && proxyPool.size > 0) { - init.proxy = undefined; // start direct - } - return { url: config.url, init }; } -// ─── Response parsing ─────────────────────────────────────────────────────────── +// --- Response parsing ---------------------------------------------------------- /** * Try to parse a JSON response into OpenAI format. @@ -251,7 +244,7 @@ function parseJSONResponse( } } - // Default fallback — assume raw text is the content + // Default fallback -- assume raw text is the content return { id: `chatcmpl-${Date.now()}`, object: "chat.completion", @@ -268,38 +261,103 @@ function parseJSONResponse( }; } -// ─── Main handler ─────────────────────────────────────────────────────────────── +// --- Input validation --------------------------------------------------------- + +interface ValidationError { + message: string; + type: string; +} + +function validateChatRequest(body: unknown): ValidationError | null { + const req = body as Record; + + if (!req.model || typeof req.model !== "string") { + return { message: "model is required", type: "invalid_request_error" }; + } + + if (!Array.isArray(req.messages) || req.messages.length === 0) { + return { message: "messages must be a non-empty array", type: "invalid_request_error" }; + } + + for (let i = 0; i < req.messages.length; i++) { + const msg = req.messages[i] as Record | undefined; + if (!msg || typeof msg !== "object") { + return { message: `messages[${i}] must be an object`, type: "invalid_request_error" }; + } + if (!msg.role || typeof msg.role !== "string") { + return { message: `messages[${i}].role is required`, type: "invalid_request_error" }; + } + if (msg.content == null) { + return { message: `messages[${i}].content is required`, type: "invalid_request_error" }; + } + } + + return null; +} + +// --- Standardized error helper ------------------------------------------------- + +/** Create a standardized OpenAI-style error response. */ +function openAIError(status: number, message: string, type: string): Response { + return new Response( + JSON.stringify({ error: { message, type } }), + { + status, + headers: { + "Content-Type": "application/json", + "Access-Control-Allow-Origin": "*", + }, + }, + ); +} + +// --- Main handler -------------------------------------------------------------- /** * Handle an OpenAI-compatible chat completions request. - * - * @param body Parsed JSON body (OpenAI format) - * @param proxyPool Optional proxy pool for fallback on failure */ export async function handleChatCompletion( body: unknown, proxyPool?: ProxyPool, ): Promise { - const req = body as OpenAIRequest; - - if (!req.model) { - return new Response( - JSON.stringify({ error: { message: "model is required", type: "invalid_request_error" } }), - { status: 400, headers: { "Content-Type": "application/json" } }, - ); + // -- Input validation ------------------------------------------------------- + const validationError = validateChatRequest(body); + if (validationError) { + return openAIError(400, validationError.message, validationError.type); } + const req = body as OpenAIRequest; + const config = resolveModel(req.model); if (!config) { + return openAIError( + 400, + `Unknown model: ${req.model}. Available: ${listModels().join(", ")}`, + "invalid_request_error", + ); + } + + const wantsStream = req.stream === true; + const { url, init } = buildBackendRequest(req, config); + + // -- Execute (direct -> proxy fallback) with shared retry ------------------- + const result = await fetchWithRetry( + url, + init, + proxyPool, + `openai:${req.model}`, + ); + + if (result.errorClassification) { return new Response( JSON.stringify({ error: { - message: `Unknown model: ${req.model}. Available: ${listModels().join(", ")}`, - type: "invalid_request_error", + message: result.errorClassification.message, + type: "upstream_error", }, }), { - status: 400, + status: result.errorClassification.status, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*", @@ -308,89 +366,30 @@ export async function handleChatCompletion( ); } - const wantsStream = req.stream === true; - const { url, init } = buildBackendRequest(req, config, proxyPool); + const response = result.response!; - // ── Execute (direct → proxy fallback) ───────────────────────── - let response: Response | undefined; - - for (let attempt = 0; attempt < 3; attempt++) { - if (attempt === 0) { - init.proxy = undefined; // direct - } else if (attempt === 1 && proxyPool && proxyPool.size > 0) { - init.proxy = proxyPool.getProxyUrl()!; - } else if (attempt >= 2 && proxyPool && proxyPool.size > 0) { - const next = proxyPool.rotate(); - if (!next) break; - init.proxy = proxyPool.getProxyUrl()!; - } else { - break; - } - - try { - response = await fetch(url, init); - if (response.ok) { - // Success — reset proxy failure if we used one - if (proxyPool && proxyPool.size > 0 && init.proxy && attempt > 0) { - proxyPool.markSuccess(); - } - break; - } - // Non-2xx — mark proxy as failed so next attempt rotates - if (proxyPool && proxyPool.size > 0 && init.proxy) { - proxyPool.markFailed(); - } - } catch { - // Network error — mark proxy as failed, retry - if (proxyPool && proxyPool.size > 0 && init.proxy) { - proxyPool.markFailed(); - } - } - } - - if (!response) { - return new Response( - JSON.stringify({ - error: { message: "Upstream service unreachable after retries", type: "server_error" }, - }), - { status: 502, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } }, - ); - } - - // ── Handle error responses from backend ─────────────────────── + // -- Handle error responses from backend ------------------------------------ if (!response.ok) { - const errBody = await response.text().catch(() => ""); - return new Response( - JSON.stringify({ - error: { - message: `Upstream error ${response.status}: ${errBody.slice(0, 500)}`, - type: "upstream_error", - }, - }), - { - status: response.status, - headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" }, - }, - ); + const status = response.status; + const genericMsg = status >= 500 ? "Upstream server error" : "Upstream rejected request"; + return openAIError(status, genericMsg, "upstream_error"); } - // ── Handle streaming ───────────────────────────────────────── + // -- Handle streaming ------------------------------------------------------- if (wantsStream || isStreamableResponse(response)) { const contentType = response.headers.get("content-type") ?? ""; const isNativeStream = contentType.includes("text/event-stream"); if (isNativeStream && config.provider === "opencode") { // Passthrough for OpenAI-compatible SSE - return new Response(response.body, { - status: 200, - headers: { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - Connection: "keep-alive", - "Access-Control-Allow-Origin": "*", - "X-Accel-Buffering": "no", - }, - }); + const headers: Record = { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + "Access-Control-Allow-Origin": "*", + "X-Accel-Buffering": "no", + }; + return new Response(response.body, { status: 200, headers }); } // Transform the stream @@ -411,7 +410,7 @@ export async function handleChatCompletion( }); } - // ── Handle non-streaming response ───────────────────────────── + // -- Handle non-streaming response ------------------------------------------ const text = await response.text(); const adapted = parseJSONResponse(text, config, req); @@ -424,7 +423,7 @@ export async function handleChatCompletion( }); } -// ─── Stream handling ──────────────────────────────────────────────────────────── +// --- Stream handling ----------------------------------------------------------- function isStreamableResponse(res: Response): boolean { const ct = res.headers.get("content-type") ?? ""; @@ -438,6 +437,7 @@ function isStreamableResponse(res: Response): boolean { /** * Transform a backend ReadableStream into OpenAI SSE format. * Uses the config's `adaptStreamLine` if available. + * Uses SSELineBuffer to handle lines split across chunk boundaries. */ function transformStream( body: ReadableStream, @@ -447,6 +447,7 @@ function transformStream( const reader = body.getReader(); const decoder = new TextDecoder(); const encoder = new TextEncoder(); + const lineBuffer = new SSELineBuffer(); return new ReadableStream({ async pull(controller) { @@ -454,13 +455,25 @@ function transformStream( while (true) { const { done, value } = await reader.read(); if (done) { + // Flush remaining text after stream ends + const remaining = lineBuffer.flush(); + if (remaining.length > 0) { + if (config.adaptStreamLine) { + const adapted = config.adaptStreamLine(remaining, req); + if (adapted) { + controller.enqueue(encoder.encode(adapted + "\n\n")); + } + } else { + controller.enqueue(encoder.encode(remaining + "\n\n")); + } + } controller.enqueue(encoder.encode("data: [DONE]\n\n")); controller.close(); return; } const chunk = decoder.decode(value, { stream: true }); - const lines = chunk.split("\n"); + const lines = lineBuffer.add(chunk); for (const line of lines) { if (config.adaptStreamLine) { @@ -469,17 +482,24 @@ function transformStream( controller.enqueue(encoder.encode(adapted + "\n\n")); } } else { - // Default passthrough controller.enqueue(encoder.encode(line + "\n\n")); } } } } catch (err) { - controller.enqueue( - encoder.encode( - `data: ${JSON.stringify({ error: String(err) })}\n\n`, - ), - ); + if (isDevMode()) { + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ error: String(err) })}\n\n`, + ), + ); + } else { + controller.enqueue( + encoder.encode( + `data: ${JSON.stringify({ error: "Stream error" })}\n\n`, + ), + ); + } controller.close(); } }, diff --git a/src/lib/anthropic-proxy.ts b/src/lib/anthropic-proxy.ts index 458856e..0619181 100644 --- a/src/lib/anthropic-proxy.ts +++ b/src/lib/anthropic-proxy.ts @@ -12,6 +12,9 @@ import type { ProxyPool } from "./proxy-pool"; import { MODEL_ROUTES, type BackendConfig } from "./ai-proxy"; +import { fetchWithRetry } from "./fetch-utils"; +import { SSELineBuffer } from "./fetch-utils"; +import { isDevMode } from "./fetch-utils"; // --- Types ------------------------------------------------------------------- @@ -40,7 +43,7 @@ interface AnthropicResponse { usage: { input_tokens: number; output_tokens: number }; } -// --- Model resolution --------------------------------------------------------- +// --- Model resolution ---------------------------------------------------------- /** Resolve a model name to a backend config (uses MODEL_ROUTES directly). */ function resolveAnthropicModel( @@ -108,7 +111,6 @@ function anthropicToBackend( : anthReq.stop_sequences; } - // If backend has a custom adaptRequest, use it if (config.adaptRequest) { return config.adaptRequest({ model: backendModel, @@ -123,7 +125,7 @@ function anthropicToBackend( return base; } -// --- Translation: Backend -> Anthropic ---------------------------------------- +// --- Translation: Backend -> Anthropic ----------------------------------------- /** * Convert a backend JSON response body into Anthropic Messages format. @@ -150,7 +152,7 @@ function backendToAnthropicResponse( }; } -// --- Streaming: Backend SSE -> Anthropic SSE --------------------------------- +// --- Streaming: Backend SSE -> Anthropic SSE ----------------------------------- /** * Accumulate text from an SSE response body (data: lines) into a single string. @@ -186,7 +188,6 @@ function accumulateSSEText(sseBody: string): string { function extractTextFromSSE(parsed: any): string | null { if (parsed == null) return null; - // Claude Code / Anthropic SSE: type-based events if (typeof parsed === "object") { switch (parsed.type) { case "text-delta": @@ -196,12 +197,10 @@ function extractTextFromSSE(parsed: any): string | null { } } - // OpenAI format: choices[0].delta.content const openai = parsed.choices?.[0]?.delta?.content ?? parsed.choices?.[0]?.text; if (openai) return openai; - // Generic fallbacks if (typeof parsed.content === "string") return parsed.content; if (typeof parsed.text === "string") return parsed.text; if (typeof parsed.delta === "string") return parsed.delta; @@ -211,8 +210,6 @@ function extractTextFromSSE(parsed: any): string | null { /** * Transform a backend SSE line into Anthropic SSE content_block_delta events. - * - * Returns the SSE event string, or null to skip the line. */ function backendLineToAnthropicSSE( line: string, @@ -221,14 +218,12 @@ function backendLineToAnthropicSSE( ): string | null { if (!line || line.trim().length === 0) return null; - // Use the backend's adaptStreamLine if available (for custom backends) if (config.adaptStreamLine) { const adapted = config.adaptStreamLine(line, {} as any); if (!adapted) return null; if (adapted === "data: [DONE]") { - return null; // let the stream transformer handle DONE + return null; } - // Parse the adapted line try { const parsed = JSON.parse(adapted.replace(/^data: /, "")); const text = extractTextFromSSE(parsed); @@ -239,15 +234,13 @@ function backendLineToAnthropicSSE( } } - // All data: lines -- try to parse as JSON in any format if (line.startsWith("data: ")) { const raw = line.slice(6); if (raw === "[DONE]") { - return null; // let the stream transformer handle DONE + return null; } try { const parsed = JSON.parse(raw); - // Skip lifecycle/non-content events if (parsed.type === "start" || parsed.type === "start-step" || parsed.type === "data-thinking-step" || parsed.type === "text-start" || parsed.type === "ping") { @@ -261,7 +254,6 @@ function backendLineToAnthropicSSE( } } - // Plain text chunks (or non-data lines) if (line.length > 0) { return formatContentBlockDelta(line); } @@ -278,7 +270,7 @@ function formatContentBlockDelta(text: string): string { })}`; } -// --- Stream transformer ------------------------------------------------------- +// --- Stream transformer -------------------------------------------------------- function transformAnthropicStream( body: ReadableStream, @@ -288,20 +280,18 @@ function transformAnthropicStream( const reader = body.getReader(); const decoder = new TextDecoder(); const encoder = new TextEncoder(); + const lineBuffer = new SSELineBuffer(); - // State machine for Anthropic SSE protocol let phase: "init" | "block" | "done" = "init"; let messageId = `msg_${Date.now()}`; return new ReadableStream({ async pull(controller) { try { - // --- Phase: emit message_start + content_block_start ------------ if (phase === "init") { phase = "block"; messageId = `msg_${Date.now()}`; - // message_start const startEvent = `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { @@ -317,7 +307,6 @@ function transformAnthropicStream( })}`; controller.enqueue(encoder.encode(startEvent + "\n\n")); - // content_block_start -- must precede any deltas const blockStart = `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, @@ -326,16 +315,22 @@ function transformAnthropicStream( controller.enqueue(encoder.encode(blockStart + "\n\n")); } - // --- Phase: read stream and emit content_block_delta events ----- while (phase === "block") { const { done, value } = await reader.read(); if (done) { + const remaining = lineBuffer.flush(); + if (remaining.length > 0) { + const adapted = backendLineToAnthropicSSE(remaining, model, config); + if (adapted) { + controller.enqueue(encoder.encode(adapted + "\n\n")); + } + } phase = "done"; break; } const chunk = decoder.decode(value, { stream: true }); - const lines = chunk.split("\n"); + const lines = lineBuffer.add(chunk); for (const line of lines) { const adapted = backendLineToAnthropicSSE(line, model, config); @@ -344,22 +339,18 @@ function transformAnthropicStream( } } - // Yield control so we don't block -- let next pull() continue return; } - // --- Phase: emit closing events (content_block_stop, message_delta, message_stop) - if (phase === "done") { - phase = "done"; // prevent re-entry + phase = "done"; - // content_block_stop controller.enqueue( encoder.encode( 'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}\n\n', ), ); - // message_delta -- required before message_stop controller.enqueue( encoder.encode( `event: message_delta\ndata: ${JSON.stringify({ @@ -370,7 +361,6 @@ function transformAnthropicStream( ), ); - // message_stop controller.enqueue( encoder.encode( 'event: message_stop\ndata: {"type":"message_stop"}\n\n', @@ -380,68 +370,104 @@ function transformAnthropicStream( controller.close(); } } catch (err) { - controller.enqueue( - encoder.encode( - `event: error\ndata: ${JSON.stringify({ error: String(err) })}\n\n`, - ), - ); + 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(); } }, }); } -// --- Main handler ------------------------------------------------------------- +// --- Input validation ---------------------------------------------------------- + +interface ValidationError { + message: string; + type: string; +} + +function validateAnthropicRequest(body: unknown): ValidationError | null { + const req = body as Record; + + if (!req.model || typeof req.model !== "string") { + return { message: "model is required", type: "invalid_request_error" }; + } + + if (!req.max_tokens || typeof req.max_tokens !== "number") { + return { message: "max_tokens is required", type: "invalid_request_error" }; + } + + if (!Array.isArray(req.messages) || req.messages.length === 0) { + return { message: "messages must be a non-empty array", type: "invalid_request_error" }; + } + + for (let i = 0; i < req.messages.length; i++) { + const msg = req.messages[i] as Record | undefined; + if (!msg || typeof msg !== "object") { + return { message: `messages[${i}] must be an object`, type: "invalid_request_error" }; + } + if (!msg.role || typeof msg.role !== "string") { + return { message: `messages[${i}].role is required`, type: "invalid_request_error" }; + } + if (msg.content == null) { + return { message: `messages[${i}].content is required`, type: "invalid_request_error" }; + } + } + + return null; +} + +// --- Standardized error helper ------------------------------------------------- + +function anthropicError(status: number, message: string, type: string): Response { + return new Response( + JSON.stringify({ + type: "error", + error: { message, type }, + }), + { + status, + headers: { + "Content-Type": "application/json", + "Access-Control-Allow-Origin": "*", + }, + }, + ); +} + +// --- Main handler -------------------------------------------------------------- /** * Handle an Anthropic-compatible messages request. - * - * @param body Parsed JSON body (Anthropic Messages format) - * @param proxyPool Optional proxy pool for fallback on failure */ export async function handleAnthropicMessages( body: unknown, proxyPool?: ProxyPool, ): Promise { + // -- Input validation ------------------------------------------------------- + const validationError = validateAnthropicRequest(body); + if (validationError) { + return anthropicError(400, validationError.message, validationError.type); + } + const req = body as AnthropicRequest; - if (!req.model) { - return new Response( - JSON.stringify({ - type: "error", - error: { message: "model is required", type: "invalid_request_error" }, - }), - { status: 400, headers: { "Content-Type": "application/json" } }, - ); - } - - if (!req.max_tokens) { - return new Response( - JSON.stringify({ - type: "error", - error: { message: "max_tokens is required", type: "invalid_request_error" }, - }), - { status: 400, headers: { "Content-Type": "application/json" } }, - ); - } - const resolved = resolveAnthropicModel(req.model); if (!resolved) { - return new Response( - JSON.stringify({ - type: "error", - error: { - message: `Unknown model: ${req.model}. Available: ${listAnthropicModels().join(", ")}`, - type: "invalid_request_error", - }, - }), - { - status: 400, - headers: { - "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", - }, - }, + return anthropicError( + 400, + `Unknown model: ${req.model}. Available: ${listAnthropicModels().join(", ")}`, + "invalid_request_error", ); } @@ -457,71 +483,43 @@ export async function handleAnthropicMessages( const url = config.url; - // ---- Execute (direct -> proxy fallback) -------------------------------- - let response: Response | undefined; + // -- Execute (direct -> proxy fallback) with shared retry ------------------- + const result = await fetchWithRetry( + url, + init, + proxyPool, + `anthropic:${req.model}`, + ); - for (let attempt = 0; attempt < 3; attempt++) { - if (attempt === 0) { - init.proxy = undefined; // direct - } else if (attempt === 1 && proxyPool && proxyPool.size > 0) { - init.proxy = proxyPool.getProxyUrl()!; - } else if (attempt >= 2 && proxyPool && proxyPool.size > 0) { - const next = proxyPool.rotate(); - if (!next) break; - init.proxy = proxyPool.getProxyUrl()!; - } else { - break; - } - - try { - response = await fetch(url, init); - if (response.ok) { - // Success -- reset proxy failure if we used one - if (proxyPool && proxyPool.size > 0 && init.proxy && attempt > 0) { - proxyPool.markSuccess(); - } - break; - } - // Non-2xx -- mark proxy as failed so next attempt rotates - if (proxyPool && proxyPool.size > 0 && init.proxy) { - proxyPool.markFailed(); - } - } catch { - // Network error -- mark proxy as failed, retry - if (proxyPool && proxyPool.size > 0 && init.proxy) { - proxyPool.markFailed(); - } - } - } - - if (!response) { - return new Response( - JSON.stringify({ - type: "error", - error: { message: "Upstream service unreachable after retries", type: "server_error" }, - }), - { status: 502, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } }, - ); - } - - if (!response.ok) { - const errBody = await response.text().catch(() => ""); + if (result.errorClassification) { return new Response( JSON.stringify({ type: "error", error: { - message: `Upstream error ${response.status}: ${errBody.slice(0, 500)}`, - type: "upstream_error", + message: result.errorClassification.message, + type: "server_error", }, }), { - status: response.status, - headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" }, + status: result.errorClassification.status, + headers: { + "Content-Type": "application/json", + "Access-Control-Allow-Origin": "*", + }, }, ); } - // ---- Handle streaming ------------------------------------------------- + const response = result.response!; + + // -- Handle error responses from backend ------------------------------------ + if (!response.ok) { + const status = response.status; + const genericMsg = status >= 500 ? "Upstream server error" : "Upstream rejected request"; + return anthropicError(status, genericMsg, "upstream_error"); + } + + // -- Handle streaming ------------------------------------------------------- if (wantsStream) { const transformed = transformAnthropicStream( response.body!, @@ -540,11 +538,9 @@ export async function handleAnthropicMessages( }); } - // ---- Handle non-streaming --------------------------------------------- + // -- Handle non-streaming --------------------------------------------------- const text = await response.text(); - // Backend may return SSE (data: lines) even for non-streaming requests. - // Accumulate all text-delta events to reconstruct the response body. if (text.trimStart().startsWith("data: ")) { const accumulated = accumulateSSEText(text); if (accumulated) { diff --git a/src/lib/fetch-utils.ts b/src/lib/fetch-utils.ts new file mode 100644 index 0000000..0db4b28 --- /dev/null +++ b/src/lib/fetch-utils.ts @@ -0,0 +1,209 @@ +/** + * Shared fetch utilities for the relay proxy. + * + * Consolidates retry logic, proxy rotation, SSE line buffering, + * error sanitization, and graceful shutdown tracking — all in one place. + */ + +import type { ProxyPool } from "./proxy-pool"; + +// ─── Active stream tracking (for graceful shutdown) ─────────────────────── + +/** Set of active ReadableStream readers that should be closed on shutdown. */ +export const ACTIVE_READERS = new Set(); + +/** + * Close all tracked active readers (called during graceful shutdown). + * Each reader's cancellation propagates to the upstream connection. + */ +export function closeAllActiveReaders(): void { + for (const reader of ACTIVE_READERS) { + try { reader.cancel(); } catch { /* already closed */ } + } + ACTIVE_READERS.clear(); +} + +// ─── Dev-mode guard ────────────────────────────────────────────────────── + +/** + * Returns `true` when development features (HMR, verbose console) should + * be enabled. Controlled by the `NODE_ENV` / `BUN_ENV` env var — defaults + * to `true` for convenience during local development. + * + * Set `NODE_ENV=production` or `BUN_ENV=production` to disable. + */ +export function isDevMode(): boolean { + const env = (process.env.NODE_ENV ?? process.env.BUN_ENV ?? "").toLowerCase(); + if (env === "production") return false; + return true; +} + +// ─── SSE line buffer (fixes chunk-boundary corruption) ─────────────────── + +/** + * Accumulates partial lines across stream chunks so that lines split across + * chunk boundaries are correctly reassembled before parsing. + * + * Usage: + * const buf = new SSELineBuffer(); + * for each chunk: const lines = buf.add(chunk); + * after stream: const lastLines = buf.flush(); + */ +export class SSELineBuffer { + private buffer = ""; + + /** + * Feed a chunk of decoded text and return complete lines. + * Lines ending with `\n` are considered complete. + */ + add(chunk: string): string[] { + this.buffer += chunk; + if (!this.buffer.includes("\n")) return []; + + const parts = this.buffer.split("\n"); + // The last element is incomplete if the chunk does not end with '\n' + this.buffer = parts.pop() ?? ""; + return parts; + } + + /** Return any remaining text after the stream ended. */ + flush(): string { + const remaining = this.buffer; + this.buffer = ""; + return remaining; + } +} + +// ─── Error sanitization (prevent leaking upstream details) ────────────── + +/** + * Sanitize an error message for inclusion in a downstream response. + * Generic messages only — no upstream URLs, paths, or stack traces. + */ +export function sanitizeErrorMessage(raw: string): string { + if ( + raw.includes("ENOTFOUND") || + raw.includes("ECONNREFUSED") || + raw.includes("ECONNRESET") || + raw.includes("ECONNABORTED") || + raw.includes("ENETUNREACH") || + raw.includes("ETIMEDOUT") || + raw.includes("DNS") || + raw.includes("dns") || + raw.includes("resolve") + ) { + return "Upstream connection failed"; + } + return "Upstream error"; +} + +// ─── Fetch with retry (direct → proxy fallback) ───────────────────────── + +export interface FetchWithRetryResult { + response?: Response; + errorClassification?: { code: string; status: number; message: string }; +} + +/** + * Execute an upstream `fetch` with automatic retry and proxy fallback. + * + * Strategy: direct first → proxy-1 on failure → rotate proxy on failure. + * Logs every failure to `console.warn` so the operator can diagnose without + * the error body leaking to the downstream client. + */ +export async function fetchWithRetry( + url: string, + init: RequestInit & { proxy?: string }, + proxyPool?: ProxyPool, + context?: string, +): Promise { + let response: Response | undefined; + let lastError: unknown; + let usedProxy = false; + + for (let attempt = 0; attempt < 3; attempt++) { + if (attempt === 0) { + init.proxy = undefined; // direct + } else if (attempt === 1 && proxyPool && proxyPool.size > 0) { + usedProxy = true; + init.proxy = proxyPool.getProxyUrl()!; + } else if (attempt === 2 && proxyPool && proxyPool.size > 0) { + const next = proxyPool.rotate(); + if (!next) break; + usedProxy = true; + init.proxy = proxyPool.getProxyUrl()!; + } else { + break; + } + + try { + response = await fetch(url, init); + if (response.ok) { + if (usedProxy && proxyPool && proxyPool.size > 0) { + proxyPool.markSuccess(); + } + return { response }; + } + + // Non-2xx — mark proxy as failed and retry + lastError = new Error(`Upstream returned ${response.status}`); + if (usedProxy && proxyPool && proxyPool.size > 0 && init.proxy) { + proxyPool.markFailed(); + usedProxy = false; + } + } catch (err) { + lastError = err; + if (usedProxy && proxyPool && proxyPool.size > 0 && init.proxy) { + proxyPool.markFailed(); + usedProxy = false; + } + + // Log the actual error so operators can diagnose + const ctx = context ? `[${context}] ` : ""; + const errMsg = err instanceof Error ? err.message : String(err); + console.warn(`${ctx}fetch attempt ${attempt + 1}/3 failed: ${errMsg}`); + } + } + + // All attempts exhausted — classify the last error + if (!response) { + const err = lastError ?? new Error("All connection attempts failed"); + return { errorClassification: classifyFetchErrorSafe(err) }; + } + + // Non-2xx but we have a response — pass it along (caller handles it) + return { response }; +} + +/** + * Simplified error classification that does NOT expose raw error text + * to downstream clients. + */ +function classifyFetchErrorSafe(error: unknown): { + code: string; + status: number; + message: string; +} { + if ( + error instanceof DOMException && + (error.name === "AbortError" || error.name === "TimeoutError") + ) { + return { code: "TIMEOUT", status: 504, message: "Upstream timed out" }; + } + + if (error instanceof TypeError) { + const msg = error.message.toLowerCase(); + if ( + msg.includes("dns") || msg.includes("resolve") || + msg.includes("hostname") || msg.includes("enotfound") + ) { + return { code: "DNS_FAILURE", status: 502, message: "DNS resolution failed" }; + } + if (msg.includes("refused") || msg.includes("econnrefused")) { + return { code: "CONNECTION_REFUSED", status: 502, message: "Connection refused" }; + } + return { code: "NETWORK_ERROR", status: 502, message: "Network error" }; + } + + return { code: "NETWORK_ERROR", status: 502, message: "Upstream unreachable" }; +} diff --git a/src/lib/proxy-pool.ts b/src/lib/proxy-pool.ts index 225c0a4..ec7d9b4 100644 --- a/src/lib/proxy-pool.ts +++ b/src/lib/proxy-pool.ts @@ -8,7 +8,7 @@ import { readFileSync, existsSync } from "node:fs"; -// ─── Types ─────────────────────────────────────────────────────────────────────── +// --- Types ------------------------------------------------------------------- export interface ProxyEntry { host: string; @@ -17,21 +17,21 @@ export interface ProxyEntry { password: string; } -// ─── ProxyPool ─────────────────────────────────────────────────────────────────── +// --- ProxyPool --------------------------------------------------------------- export class ProxyPool { private proxies: ProxyEntry[] = []; private currentIndex = 0; private failureThreshold = 3; - /** host:port → consecutive failure count */ + /** host:port -> consecutive failure count */ private failures = new Map(); - // ── Load ────────────────────────────────────────────────────────── + // -- Load -------------------------------------------------------------------- /** * Load proxies from a file at `filePath`. * - * Expected format — one proxy per line: + * Expected format -- one proxy per line: * host:port:username:password * * Lines starting with `#` are ignored as comments. @@ -79,7 +79,7 @@ export class ProxyPool { } /** - * Convenience — load from a path or skip. + * Convenience -- load from a path or skip. * Returns `true` if proxies were loaded. */ tryLoad(filePath?: string): boolean { @@ -87,7 +87,7 @@ export class ProxyPool { return this.proxies.length > 0; } - // ── Access ──────────────────────────────────────────────────────── + // -- Access ------------------------------------------------------------------ /** Total number of proxies in the pool. */ get size(): number { @@ -123,7 +123,7 @@ export class ProxyPool { return `http://${auth}${entry.host}:${entry.port}`; } - // ── Rotation ────────────────────────────────────────────────────── + // -- Rotation ---------------------------------------------------------------- /** * Advance to the next proxy (round-robin, wraps around). @@ -136,15 +136,16 @@ export class ProxyPool { } /** - * Mark the **current** proxy as failed. + * Mark the **current** proxy as failed. Increments the failure count; + * when the count reaches the threshold a warning is logged. * - * If the failure count exceeds `threshold`, the proxy is skipped and - * rotation continues until a healthy proxy is found (or we've tried - * all of them). + * NOTE: This no longer calls `rotate()` automatically -- the caller is + * responsible for deciding when to rotate. Previously this was conflated + * and caused double-rotation bugs in retry loops. */ - markFailed(threshold?: number): ProxyEntry | null { + markFailed(threshold?: number): void { const entry = this.getCurrent(); - if (!entry) return null; + if (!entry) return; const key = `${entry.host}:${entry.port}`; const count = (this.failures.get(key) ?? 0) + 1; @@ -153,11 +154,17 @@ export class ProxyPool { const th = threshold ?? this.failureThreshold; if (count >= th) { console.warn( - `[proxy-pool] Proxy ${key} failed ${count}/${th} times — skipping`, + `[proxy-pool] Proxy ${key} failed ${count}/${th} times -- skipping`, ); } + } - return this.rotate(); + /** Check if the current proxy has exceeded the failure threshold. */ + isFailed(threshold?: number): boolean { + const entry = this.getCurrent(); + if (!entry) return true; + const key = `${entry.host}:${entry.port}`; + return (this.failures.get(key) ?? 0) >= (threshold ?? this.failureThreshold); } /** Reset the failure counter for the current proxy. */ diff --git a/src/lib/relay-utils.test.ts b/src/lib/relay-utils.test.ts index 3cb9a1b..5877775 100644 --- a/src/lib/relay-utils.test.ts +++ b/src/lib/relay-utils.test.ts @@ -652,7 +652,7 @@ describe("classifyFetchError", () => { const result = classifyFetchError(error); expect(result.code).toBe("NETWORK_ERROR"); expect(result.status).toBe(502); - expect(result.message).toBe("Unknown upstream error"); + expect(result.message).toBe("Upstream unreachable"); }); test("should handle non-Error thrown values", () => { diff --git a/src/lib/relay-utils.ts b/src/lib/relay-utils.ts index bd11ab1..c16f542 100644 --- a/src/lib/relay-utils.ts +++ b/src/lib/relay-utils.ts @@ -4,29 +4,59 @@ * request building, and error handling. */ -// ─── Types & Classes ─────────────────────────────────────────────────────────── +import { resolve4, resolve6 } from "node:dns/promises"; +import { isIP } from "node:net"; + +// --- Types & Classes ---------------------------------------------------------- export class RelayError extends Error { - public readonly name = 'RelayError' as const; + public readonly name = "RelayError" as const; - constructor( - public readonly code: - | 'TIMEOUT' - | 'DNS_FAILURE' - | 'CONNECTION_REFUSED' - | 'NETWORK_ERROR' - | 'INVALID_TARGET' - | 'SSRF_BLOCKED' - | 'BODY_TOO_LARGE' - | 'UPSTREAM_ERROR', - public readonly status: number, - message: string, - ) { - super(message); - } + constructor( + public readonly code: + | "TIMEOUT" + | "DNS_FAILURE" + | "CONNECTION_REFUSED" + | "NETWORK_ERROR" + | "INVALID_TARGET" + | "SSRF_BLOCKED" + | "BODY_TOO_LARGE" + | "UPSTREAM_ERROR", + public readonly status: number, + message: string, + ) { + super(message); + } } -// ─── URL Handling ────────────────────────────────────────────────────────────── +// --- CORS configuration ------------------------------------------------------- + +/** Return CORS headers. Origin defaults to "*" but can be overridden via env. */ +export function getAllowedOrigin(): string { + const configured = process.env.CORS_ORIGIN?.trim(); + if (configured && configured.length > 0) return configured; + return "*"; +} + +let cachedCorsHeaders: Record | null = null; + +/** Rebuild the CORS headers map (call after changing origin at runtime). */ +export function rebuildCorsHeaders(): Record { + cachedCorsHeaders = { + "Access-Control-Allow-Origin": getAllowedOrigin(), + "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE, PATCH, OPTIONS", + "Access-Control-Allow-Headers": "*", + }; + return cachedCorsHeaders; +} + +/** Return CORS headers with the configured origin (cached after first call). */ +export function getCorsHeaders(): Record { + if (!cachedCorsHeaders) rebuildCorsHeaders(); + return cachedCorsHeaders!; +} + +// --- URL Handling -------------------------------------------------------------- /** * Combines a relay target URL with a path, returning a URL object. @@ -36,185 +66,233 @@ export class RelayError extends Error { * - Returns a `URL` object (call `.toString()` or `.href` for a string). */ export function normalizeTargetUrl( - target: string | null, - relayPath: string, + target: string | null, + relayPath: string, ): URL | null { - if (!target || target.trim().length === 0) return null; + if (!target || target.trim().length === 0) return null; - const normalizedTarget = target.replace(/\/+$/, ''); - const cleanRelayPath = relayPath.startsWith('/') - ? relayPath - : '/' + relayPath; + const normalizedTarget = target.replace(/\/+$/, ""); + const cleanRelayPath = relayPath.startsWith("/") + ? relayPath + : "/" + relayPath; - try { - const baseUrl = new URL(normalizedTarget); - const baseOrigin = baseUrl.origin; - const basePathname = baseUrl.pathname; + try { + const baseUrl = new URL(normalizedTarget); + const baseOrigin = baseUrl.origin; + const basePathname = baseUrl.pathname; - // Strip query string from relayPath before concatenating - const relayPathOnly = cleanRelayPath.includes('?') - ? cleanRelayPath.slice(0, cleanRelayPath.indexOf('?')) - : cleanRelayPath; + // Strip query string from relayPath before concatenating + const relayPathOnly = cleanRelayPath.includes("?") + ? cleanRelayPath.slice(0, cleanRelayPath.indexOf("?")) + : cleanRelayPath; - const combinedPath = - basePathname === '/' - ? relayPathOnly - : basePathname.replace(/\/$/, '') + relayPathOnly; + const combinedPath = + basePathname === "/" + ? relayPathOnly + : basePathname.replace(/\/$/, "") + relayPathOnly; - const combined = new URL(combinedPath, baseOrigin); + const combined = new URL(combinedPath, baseOrigin); - // Preserve query parameters from the target URL - const targetParams = Array.from(baseUrl.searchParams); - for (const [key, value] of targetParams) { - combined.searchParams.set(key, value); - } + // Preserve query parameters from the target URL + const targetParams = Array.from(baseUrl.searchParams); + for (const [key, value] of targetParams) { + combined.searchParams.set(key, value); + } - // Merge query parameters from relayPath - if (cleanRelayPath.includes('?')) { - const relayQueryStr = cleanRelayPath.slice( - cleanRelayPath.indexOf('?') + 1, - ); - if (relayQueryStr.length > 0) { - const relayParams = Array.from(new URLSearchParams(relayQueryStr)); - for (const [key, value] of relayParams) { - combined.searchParams.append(key, value); - } - } - } + // Merge query parameters from relayPath + if (cleanRelayPath.includes("?")) { + const relayQueryStr = cleanRelayPath.slice( + cleanRelayPath.indexOf("?") + 1, + ); + if (relayQueryStr.length > 0) { + const relayParams = Array.from(new URLSearchParams(relayQueryStr)); + for (const [key, value] of relayParams) { + combined.searchParams.append(key, value); + } + } + } - return combined; - } catch { - return null; - } + return combined; + } catch { + return null; + } } -// ─── SSRF Protection ─────────────────────────────────────────────────────────── +// --- SSRF Protection ---------------------------------------------------------- /** Regex patterns for private / loopback / link-local IP ranges. */ const PRIVATE_IP_PATTERNS: RegExp[] = [ - // IPv4 - /^127\./, // loopback - /^10\./, // private class A - /^172\.(?:1[6-9]|2\d|3[01])\./, // private class B - /^192\.168\./, // private class C - /^169\.254\./, // link-local - /^0\./, // current network - /^0\.0\.0\.0$/, // unspecified - // IPv6 - /^::$/, // unspecified - /^::1$/, // loopback - /^fe80:/i, // link-local - /^fd00:/i, // unique local - /^fc00:/i, // unique local + // IPv4 + /^127\./, // loopback + /^10\./, // private class A + /^172\.(?:1[6-9]|2\d|3[01])\./, // private class B + /^192\.168\./, // private class C + /^169\.254\./, // link-local + /^0\./, // current network + /^0\.0\.0\.0$/, // unspecified + // IPv6 + /^::$/, // unspecified + /^::1$/, // loopback + /^fe80:/i, // link-local + /^fd00:/i, // unique local + /^fc00:/i, // unique local ]; const PRIVATE_HOSTNAMES = new Set([ - 'localhost', - 'localhost.localdomain', - 'localhost6', - 'localhost6.localdomain6', - 'metadata.google.internal', - 'metadata.internal', - '169.254.169.254', + "localhost", + "localhost.localdomain", + "localhost6", + "localhost6.localdomain6", + "metadata.google.internal", + "metadata.internal", + "169.254.169.254", ]); -const PRIVATE_HOSTNAME_SUFFIXES = ['.local', '.internal']; +const PRIVATE_HOSTNAME_SUFFIXES = [".local", ".internal"]; /** * Returns `true` when `hostname` is a private / loopback / link-local IP * or a well-known private hostname string. */ export function isPrivateIp(hostname: string): boolean { - const lower = hostname.toLowerCase(); + const lower = hostname.toLowerCase(); - // Check known private hostnames - if (PRIVATE_HOSTNAMES.has(lower)) return true; + // Check known private hostnames + if (PRIVATE_HOSTNAMES.has(lower)) return true; - // Check hostname suffixes (e.g. *.local, *.internal) - for (const suffix of PRIVATE_HOSTNAME_SUFFIXES) { - if (lower.endsWith(suffix)) return true; - } + // Check hostname suffixes (e.g. *.local, *.internal) + for (const suffix of PRIVATE_HOSTNAME_SUFFIXES) { + if (lower.endsWith(suffix)) return true; + } - // Check IP patterns - for (const pattern of PRIVATE_IP_PATTERNS) { - if (pattern.test(lower)) return true; - } + // Check IP patterns + for (const pattern of PRIVATE_IP_PATTERNS) { + if (pattern.test(lower)) return true; + } - return false; + return false; +} + +/** + * Resolve a hostname to its IP addresses and check whether any of them are + * private / loopback / link-local. This protects against DNS rebinding + * attacks where a hostname initially resolves to a public IP (passing the + * name-based check) but later resolves to a private IP. + * + * Returns `true` if the hostname resolves to any private IP, or if the + * resolution itself fails. When the hostname is already an IP literal + * the existing `isPrivateIp` check is used directly. + */ +export async function isPrivateIpAfterResolve(hostname: string): Promise { + const lower = hostname.toLowerCase(); + + // If it's already an IP literal, check directly (no rebinding risk) + if (isIP(lower) !== 0) { + return isPrivateIp(lower); + } + + // Resolve to IPv4 and IPv6 addresses concurrently + try { + const [v4addrs, v6addrs] = await Promise.all([ + resolve4(lower).catch(() => [] as string[]), + resolve6(lower).catch(() => [] as string[]), + ]); + + const allAddrs = [...v4addrs, ...v6addrs]; + if (allAddrs.length === 0) { + // No addresses resolved -- be safe and block + return true; + } + + return allAddrs.some((addr) => isPrivateIp(addr)); + } catch { + // Resolution failure -- block to be safe + return true; + } } /** * Validates a parsed URL is allowed for proxying. * * - Only `http:` and `https:` protocols are permitted. - * - Hostname must not resolve to a private / internal IP (SSRF protection). + * - Hostname must not be a private / internal IP. */ export function isAllowedTarget(url: URL): boolean { - // Protocol check - if (url.protocol !== 'http:' && url.protocol !== 'https:') { - return false; - } + // Protocol check + if (url.protocol !== "http:" && url.protocol !== "https:") { + return false; + } - // SSRF check — block private / internal hosts - if (isPrivateIp(url.hostname)) { - return false; - } + // SSRF check -- block private / internal hosts + if (isPrivateIp(url.hostname)) { + return false; + } - return true; + return true; } -// ─── Header Filtering ────────────────────────────────────────────────────────── +/** + * Full SSRF validation including DNS rebinding protection. + * Resolves the hostname and verifies no resolved IP is private. + */ +export async function isAllowedTargetAsync(url: URL): Promise { + if (!isAllowedTarget(url)) return false; + if (await isPrivateIpAfterResolve(url.hostname)) return false; + return true; +} + +// --- Header Filtering --------------------------------------------------------- /** * Set of exact header names (lower-case) to strip from **outgoing** relay * requests. */ export const BLOCKED_REQUEST_HEADERS = new Set([ - // Relay control headers - 'host', - 'x-relay-target', - 'x-relay-path', - // Hop-by-hop headers (should never be forwarded) - 'connection', - 'keep-alive', - 'proxy-authenticate', - 'proxy-authorization', - 'te', - 'trailers', - 'transfer-encoding', - 'upgrade', - // Security-sensitive — strip by default - 'cookie', - 'set-cookie', - // Vercel platform headers - 'x-vercel-id', - 'x-vercel-deployment-url', - 'x-vercel-oidc-token', - 'x-vercel-signature', - 'x-vercel-edgified', - 'x-vercel-proxy-signature', - 'x-vercel-ip-city', - 'x-vercel-ip-country', - 'x-vercel-ip-country-region', - 'x-vercel-ip-latency', - 'x-vercel-ip-longitude', - 'x-vercel-ip-timezone', - 'x-vercel-forwarded-for', - 'x-vercel-set-bucket', - // Cloudflare platform headers - 'cf-ray', - 'cf-connecting-ip', - 'cf-ipcountry', - 'cf-visitor', - 'cf-worker', - 'cf-edge', - // Forwarded-for metadata (privacy) - 'x-forwarded-for', - 'x-forwarded-host', - 'x-forwarded-proto', - 'x-real-ip', - 'forwarded', - 'via', + // Relay control headers + "host", + "x-relay-target", + "x-relay-path", + // Hop-by-hop headers (should never be forwarded) + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailers", + "transfer-encoding", + "upgrade", + // Security-sensitive -- strip by default + "cookie", + "set-cookie", + // Vercel platform headers + "x-vercel-id", + "x-vercel-deployment-url", + "x-vercel-oidc-token", + "x-vercel-signature", + "x-vercel-edgified", + "x-vercel-proxy-signature", + "x-vercel-ip-city", + "x-vercel-ip-country", + "x-vercel-ip-country-region", + "x-vercel-ip-latency", + "x-vercel-ip-longitude", + "x-vercel-ip-timezone", + "x-vercel-forwarded-for", + "x-vercel-set-bucket", + // Cloudflare platform headers + "cf-ray", + "cf-connecting-ip", + "cf-ipcountry", + "cf-visitor", + "cf-worker", + "cf-edge", + // Forwarded-for metadata (privacy) + "x-forwarded-for", + "x-forwarded-host", + "x-forwarded-proto", + "x-real-ip", + "forwarded", + "via", ]); /** @@ -222,15 +300,15 @@ export const BLOCKED_REQUEST_HEADERS = new Set([ * relay requests. Matching is case-insensitive. */ export const BLOCKED_REQUEST_PREFIXES = [ - 'x-vercel-', - 'cf-', - 'x-forwarded-', - 'x-envoy-', + "x-vercel-", + "cf-", + "x-forwarded-", + "x-envoy-", ]; // Pre-computed lower-case versions for efficient matching const BLOCKED_REQUEST_PREFIXES_LOWER = BLOCKED_REQUEST_PREFIXES.map((p) => - p.toLowerCase(), + p.toLowerCase(), ); /** @@ -241,32 +319,32 @@ const BLOCKED_REQUEST_PREFIXES_LOWER = BLOCKED_REQUEST_PREFIXES.map((p) => * 2. Any header whose lower-case key starts with an entry in * `BLOCKED_REQUEST_PREFIXES`. * - * Returns a **new** `Headers` instance — the original is not mutated. + * Returns a **new** `Headers` instance -- the original is not mutated. */ export function filterRequestHeaders(headers: Headers): Headers { - const filtered = new Headers(); + const filtered = new Headers(); - const headerEntries = Array.from(headers); - for (const [key, value] of headerEntries) { - const lower = key.toLowerCase(); + const headerEntries = Array.from(headers); + for (const [key, value] of headerEntries) { + const lower = key.toLowerCase(); - // Check exact blocked headers - if (BLOCKED_REQUEST_HEADERS.has(lower)) continue; + // Check exact blocked headers + if (BLOCKED_REQUEST_HEADERS.has(lower)) continue; - // Check blocked prefixes - let blockedByPrefix = false; - for (const prefix of BLOCKED_REQUEST_PREFIXES_LOWER) { - if (lower.startsWith(prefix)) { - blockedByPrefix = true; - break; - } - } - if (blockedByPrefix) continue; + // Check blocked prefixes + let blockedByPrefix = false; + for (const prefix of BLOCKED_REQUEST_PREFIXES_LOWER) { + if (lower.startsWith(prefix)) { + blockedByPrefix = true; + break; + } + } + if (blockedByPrefix) continue; - filtered.set(key, value); - } + filtered.set(key, value); + } - return filtered; + return filtered; } /** @@ -274,40 +352,35 @@ export function filterRequestHeaders(headers: Headers): Headers { * the caller. */ export const BLOCKED_RESPONSE_HEADERS = new Set([ - 'set-cookie', - 'transfer-encoding', - 'keep-alive', - 'connection', + "set-cookie", + "transfer-encoding", + "keep-alive", + "connection", ]); -const CORS_HEADERS: Record = { - 'Access-Control-Allow-Origin': '*', - 'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, PATCH, OPTIONS', - 'Access-Control-Allow-Headers': '*', -}; - /** * Strips sensitive headers from a relay **response** and attaches standard * CORS headers. * - * Returns a **new** `Headers` instance — the original is not mutated. + * Returns a **new** `Headers` instance -- the original is not mutated. */ export function filterResponseHeaders(headers: Headers): Headers { - const filtered = new Headers(headers); + const filtered = new Headers(headers); - const blockedKeys = Array.from(BLOCKED_RESPONSE_HEADERS); - for (const key of blockedKeys) { - filtered.delete(key); - } + const blockedKeys = Array.from(BLOCKED_RESPONSE_HEADERS); + for (const key of blockedKeys) { + filtered.delete(key); + } - for (const [key, value] of Object.entries(CORS_HEADERS)) { - filtered.set(key, value); - } + const cors = getCorsHeaders(); + for (const [key, value] of Object.entries(cors)) { + filtered.set(key, value); + } - return filtered; + return filtered; } -// ─── Backward Compatibility ──────────────────────────────────────────────────── +// --- Backward Compatibility --------------------------------------------------- /** * @deprecated Use `filterRequestHeaders` instead. Kept for compatibility @@ -315,7 +388,7 @@ export function filterResponseHeaders(headers: Headers): Headers { */ export const filterHeaders = filterRequestHeaders; -// ─── Request Building ────────────────────────────────────────────────────────── +// --- Request Building --------------------------------------------------------- /** * Returns `true` when the HTTP method typically carries a request body. @@ -324,8 +397,8 @@ export const filterHeaders = filterRequestHeaders; * a body. Everything else (POST, PUT, PATCH, DELETE, OPTIONS, etc.) may. */ export function shouldSendBody(method: string): boolean { - const upper = method.toUpperCase(); - return upper !== 'GET' && upper !== 'HEAD' && upper !== 'CONNECT'; + const upper = method.toUpperCase(); + return upper !== "GET" && upper !== "HEAD" && upper !== "CONNECT"; } /** @@ -337,134 +410,122 @@ export function shouldSendBody(method: string): boolean { * - Attaches an `AbortSignal.timeout()` signal. */ export function buildRelayRequest( - req: Request, - headers: Headers, - timeoutMs?: number, + req: Request, + headers: Headers, + timeoutMs?: number, ): RequestInit { - const timeout = timeoutMs ?? 30_000; - const method = req.method; - const body = shouldSendBody(method) ? req.body : undefined; + const timeout = timeoutMs ?? 30_000; + const method = req.method; + const body = shouldSendBody(method) ? req.body : undefined; - const init: RequestInit & { duplex?: 'half' } = { - method, - headers, - signal: AbortSignal.timeout(timeout), - }; + const init: RequestInit & { duplex?: "half" } = { + method, + headers, + signal: AbortSignal.timeout(timeout), + }; - if (body) { - init.body = body; - init.duplex = 'half'; - } + if (body) { + init.body = body; + init.duplex = "half"; + } - return init; + return init; } -// ─── Response Building ───────────────────────────────────────────────────────── +// --- Response Building -------------------------------------------------------- /** * Creates a relay-friendly `Response` by passing through the upstream status, * status text, and body while sanitising headers via `filterResponseHeaders`. */ export function createRelayResponse(response: Response): Response { - const headers = filterResponseHeaders(response.headers); + const headers = filterResponseHeaders(response.headers); - return new Response(response.body, { - status: response.status, - statusText: response.statusText, - headers, - }); + return new Response(response.body, { + status: response.status, + statusText: response.statusText, + headers, + }); } -// ─── Error Handling ──────────────────────────────────────────────────────────── +// --- Error Handling ----------------------------------------------------------- interface ErrorClassification { - code: string; - status: number; - message: string; + code: string; + status: number; + message: string; } /** * Classifies a caught `unknown` into a structured error with an HTTP status * code and a user-facing message suitable for JSON error responses. + * + * NOTE: The returned `message` is deliberately generic to avoid leaking + * upstream details to downstream clients. */ export function classifyFetchError(error: unknown): ErrorClassification { - // RelayError passes through its own classification - if (error instanceof RelayError) { - return { - code: error.code, - status: error.status, - message: error.message, - }; - } + // RelayError passes through its own classification + if (error instanceof RelayError) { + return { + code: error.code, + status: error.status, + message: error.message, + }; + } - // AbortError from AbortSignal.timeout or controller.abort() - if ( - error instanceof DOMException && - (error.name === 'AbortError' || error.name === 'TimeoutError') - ) { - return { - code: 'TIMEOUT', - status: 504, - message: 'Upstream timed out', - }; - } + // AbortError from AbortSignal.timeout or controller.abort() + if ( + error instanceof DOMException && + (error.name === "AbortError" || error.name === "TimeoutError") + ) { + return { + code: "TIMEOUT", + status: 504, + message: "Upstream timed out", + }; + } - if (error instanceof TypeError) { - const msg = error.message.toLowerCase(); + if (error instanceof TypeError) { + const msg = error.message.toLowerCase(); - if ( - msg.includes('dns') || - msg.includes('resolve') || - msg.includes('hostname') || - msg.includes('enotfound') - ) { - return { - code: 'DNS_FAILURE', - status: 502, - message: 'DNS resolution failed', - }; - } + if ( + msg.includes("dns") || + msg.includes("resolve") || + msg.includes("hostname") || + msg.includes("enotfound") + ) { + return { + code: "DNS_FAILURE", + status: 502, + message: "DNS resolution failed", + }; + } - if ( - msg.includes('refused') || - msg.includes('econnrefused') || - msg.includes('connection refused') - ) { - return { - code: 'CONNECTION_REFUSED', - status: 502, - message: 'Connection refused', - }; - } + if ( + msg.includes("refused") || + msg.includes("econnrefused") || + msg.includes("connection refused") + ) { + return { + code: "CONNECTION_REFUSED", + status: 502, + message: "Connection refused", + }; + } - if ( - msg.includes('fetch failed') || - msg.includes('network') || - msg.includes('econnreset') || - msg.includes('econnaborted') || - msg.includes('enetunreach') - ) { - return { - code: 'NETWORK_ERROR', - status: 502, - message: 'Network error', - }; - } + return { + code: "NETWORK_ERROR", + status: 502, + message: "Network error", + }; + } - // Generic TypeError that doesn't match known patterns - return { - code: 'NETWORK_ERROR', - status: 502, - message: 'Network error', - }; - } - - // Fallback - return { - code: 'NETWORK_ERROR', - status: 502, - message: 'Unknown upstream error', - }; + // Fallback + return { + code: "NETWORK_ERROR", + status: 502, + message: "Upstream unreachable", + }; } /** @@ -474,35 +535,32 @@ export function classifyFetchError(error: unknown): ErrorClassification { * attached so the caller can read the error from a browser. */ export function createErrorResponse(error: ErrorClassification): Response { - const body = JSON.stringify({ - error: true, - code: error.code, - message: error.message, - }); + const body = JSON.stringify({ + error: true, + code: error.code, + message: error.message, + }); - return new Response(body, { - status: error.status, - headers: { - 'Content-Type': 'application/json', - ...CORS_HEADERS, - }, - }); + return new Response(body, { + status: error.status, + headers: { + "Content-Type": "application/json", + ...getCorsHeaders(), + }, + }); } -// ─── CORS Preflight ──────────────────────────────────────────────────────────── +// --- CORS Preflight ----------------------------------------------------------- /** * Returns a 204 No Content response with CORS preflight headers. - * - * This is a convenience so the relay server does not need to manually - * construct CORS OPTIONS responses. */ export function createCorsPreflightResponse(): Response { - return new Response(null, { - status: 204, - headers: { - ...CORS_HEADERS, - 'Access-Control-Max-Age': '86400', - }, - }); + return new Response(null, { + status: 204, + headers: { + ...getCorsHeaders(), + "Access-Control-Max-Age": "86400", + }, + }); } diff --git a/src/middleware/body-limiter.ts b/src/middleware/body-limiter.ts index 26de52e..c0f8f86 100644 --- a/src/middleware/body-limiter.ts +++ b/src/middleware/body-limiter.ts @@ -3,8 +3,9 @@ * * Checks the Content-Length header against a configurable maximum. * Returns a 413 Payload Too Large response when the body exceeds the limit. - * Requests without a Content-Length header are passed through since - * the body size cannot be determined upfront with streaming. + * + * For streaming requests (no Content-Length), a TransformStream-based + * enforcer is available that counts bytes and aborts when the limit is exceeded. */ const DEFAULT_MAX_BODY_SIZE = 1_048_576; // 1 MB @@ -19,16 +20,16 @@ let maxBodySize = DEFAULT_MAX_BODY_SIZE; * cannot be determined (no Content-Length header). */ export function checkBodySize(request: Request): Response | null { - const contentType = request.headers.get("content-length"); - if (contentType === null) { - // Cannot determine size upfront — pass through (streaming body). + const header = request.headers.get("content-length"); + if (header === null) { + // Cannot determine size upfront -- pass through (streaming body). return null; } - const contentLength = Number.parseInt(contentType, 10); + const contentLength = Number.parseInt(header, 10); if (Number.isNaN(contentLength) || contentLength < 0) { - // Malformed Content-Length — pass through and let the server handle it. + // Malformed Content-Length -- pass through and let the server handle it. return null; } @@ -51,6 +52,45 @@ export function checkBodySize(request: Request): Response | null { return null; } +/** + * Wrap a ReadableStream so it enforces a byte limit on the total data + * read. If the limit is exceeded the stream errors with a `BodyTooLarge` + * error and enqueues a 413-style JSON error object. + * + * This catches oversized streaming bodies that have no `Content-Length` + * header and would otherwise bypass the Content-Length check. + */ +export function createStreamBodyLimiter( + stream: ReadableStream, +): ReadableStream { + let totalBytes = 0; + + const transformer = new TransformStream({ + transform(chunk, controller) { + totalBytes += chunk.byteLength; + if (totalBytes > maxBodySize) { + const errBody = JSON.stringify({ + error: "Payload Too Large", + message: `Streaming body exceeded maximum allowed size of ${maxBodySize} bytes`, + maxSizeBytes: maxBodySize, + }); + controller.enqueue( + new TextEncoder().encode( + `data: ${errBody}\n\nevent: error\ndata: {}\n\n`, + ), + ); + controller.error( + new Error(`Body exceeded ${maxBodySize} byte limit`), + ); + return; + } + controller.enqueue(chunk); + }, + }); + + return stream.pipeThrough(transformer); +} + /** * Update the maximum allowed body size. */ diff --git a/src/worker.ts b/src/worker.ts index 230d69c..239e490 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -7,8 +7,7 @@ * Reuses the same relay logic from `src/lib/` and `src/middleware/` as * the standalone Bun.serve() server, but: * - Uses `env` for configuration (Workers don't have process.env) - * - Does NOT support WebSocket upgrades (Workers can proxy WS but - * this handler only handles HTTP relay) + * - Does NOT support WebSocket upgrades * - Rate limiter is per-isolate (resets on cold start) */ @@ -21,6 +20,7 @@ import { classifyFetchError, createErrorResponse, createCorsPreflightResponse, + getCorsHeaders, } from "./lib/relay-utils"; import { checkBodySize } from "./middleware/body-limiter"; @@ -29,7 +29,7 @@ import { logRelayEvent } from "./middleware/logger"; import { handleChatCompletion, listModels } from "./lib/ai-proxy"; import { handleAnthropicMessages } from "./lib/anthropic-proxy"; -// ─── Types ─────────────────────────────────────────────────────────────────────── +// --- Types ------------------------------------------------------------------- export interface Env { /** Upstream fetch timeout in ms (default: 30000) */ @@ -44,7 +44,21 @@ export interface Env { API_KEY?: string; } -// ─── Helpers ───────────────────────────────────────────────────────────────────── +// --- Singletons (per-isolate, survives warm starts) --------------------------- + +let rateLimiter: ReturnType | null = null; + +function getRateLimiter(env: Env) { + if (!rateLimiter) { + rateLimiter = createRateLimiter({ + maxRequests: getNumericEnv(env, "RATE_LIMIT_MAX", 100), + windowMs: getNumericEnv(env, "RATE_LIMIT_WINDOW_MS", 60000), + }); + } + return rateLimiter; +} + +// --- Helpers ------------------------------------------------------------------ function getNumericEnv( env: Env, @@ -68,19 +82,20 @@ function getClientIP(req: Request): string { return "unknown"; } -// ─── Auth Helper ───────────────────────────────────────────────────────────────── +// --- Auth Helper --------------------------------------------------------------- function requireAuth(req: Request): Response | null { + const apiKey = process.env.API_KEY ?? "sk-dummy-key"; const header = req.headers.get("authorization") ?? req.headers.get("x-api-key") ?? ""; const key = header.replace(/^Bearer\s+/i, "").trim(); - if (key === "sk-dummy-key") return null; + if (key === apiKey) return null; return new Response( JSON.stringify({ error: { message: "Unauthorized", type: "auth_error" } }), - { status: 401, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } }, + { status: 401, headers: { "Content-Type": "application/json", ...getCorsHeaders() } }, ); } -// ─── Route Handlers ────────────────────────────────────────────────────────────── +// --- Route Handlers ------------------------------------------------------------ const SERVER_START_TIME = Date.now(); const RELAY_VERSION = "1.0.0"; @@ -96,7 +111,7 @@ function handleHealth(): Response { status: 200, headers: { "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), }, }, ); @@ -183,7 +198,7 @@ function handleDocs(): Response { status: 200, headers: { "Content-Type": "text/html; charset=utf-8", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), }, }); } @@ -222,7 +237,7 @@ function handleIndex(): Response { }); } -// ─── Relay Logic ───────────────────────────────────────────────────────────────── +// --- Relay Logic --------------------------------------------------------------- async function handleRelay(req: Request, env: Env): Promise { const startTime = performance.now(); @@ -232,18 +247,14 @@ async function handleRelay(req: Request, env: Env): Promise { const RELAY_TIMEOUT_MS = getNumericEnv(env, "RELAY_TIMEOUT_MS", 30000); - // Per-isolate rate limiter (recreated on each cold start) - const rateLimiter = createRateLimiter({ - maxRequests: getNumericEnv(env, "RATE_LIMIT_MAX", 100), - windowMs: getNumericEnv(env, "RATE_LIMIT_WINDOW_MS", 60000), - }); + const limiter = getRateLimiter(env); - // ── Pre-flight CORS ────────────────────────────────────────────── + // -- Pre-flight CORS -------------------------------------------------------- if (method === "OPTIONS") { return createCorsPreflightResponse(); } - // ── Middleware: Body size check ────────────────────────────────── + // -- Middleware: Body size check ------------------------------------------- const bodyError = checkBodySize(req); if (bodyError) { logRelayEvent({ @@ -256,8 +267,8 @@ async function handleRelay(req: Request, env: Env): Promise { return bodyError; } - // ── Middleware: Rate limiting ──────────────────────────────────── - const rateCheck = rateLimiter.check(clientIP); + // -- Middleware: Rate limiting --------------------------------------------- + const rateCheck = limiter.check(clientIP); if (!rateCheck.allowed) { logRelayEvent({ method, @@ -278,7 +289,7 @@ async function handleRelay(req: Request, env: Env): Promise { status: 429, headers: { "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), "Retry-After": String( Math.ceil((rateCheck.retryAfterMs ?? 60_000) / 1000), ), @@ -287,11 +298,11 @@ async function handleRelay(req: Request, env: Env): Promise { ); } - // ── Extract relay parameters from headers ─────────────────────── + // -- 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 ───────────────────── + // -- SSRF: Normalize and validate target URL -------------------------------- const targetUrl = normalizeTargetUrl(target, relayPath); if (!targetUrl) { logRelayEvent({ @@ -325,7 +336,7 @@ async function handleRelay(req: Request, env: Env): Promise { }); } - // ── Build the upstream request ────────────────────────────────── + // -- Build the upstream request --------------------------------------------- const filteredHeaders = filterRequestHeaders(req.headers); const fetchOptions = buildRelayRequest( req, @@ -335,7 +346,7 @@ async function handleRelay(req: Request, env: Env): Promise { const targetUrlString = targetUrl.toString(); - // ── Execute upstream fetch ────────────────────────────────────── + // -- Execute upstream fetch ------------------------------------------------- let response: Response; try { response = await fetch(targetUrlString, fetchOptions); @@ -353,7 +364,7 @@ async function handleRelay(req: Request, env: Env): Promise { return createErrorResponse(classified); } - // ── Build relay response ──────────────────────────────────────── + // -- Build relay response --------------------------------------------------- const relayedResponse = createRelayResponse(response); logRelayEvent({ @@ -368,13 +379,12 @@ async function handleRelay(req: Request, env: Env): Promise { return relayedResponse; } -// ─── Exported Worker Handler ───────────────────────────────────────────────────── +// --- Exported Worker Handler --------------------------------------------------- export default { async fetch(req: Request, env: Env): Promise { const url = new URL(req.url); - // Static routes — show index only when no relay target is requested if (url.pathname === "/health") return handleHealth(); if (url.pathname === "/docs") return handleDocs(); if ( @@ -385,7 +395,6 @@ export default { return handleIndex(); } - // WebSocket upgrade — not fully supported in this handler if ( req.method === "GET" && req.headers.get("upgrade")?.toLowerCase() === "websocket" @@ -400,13 +409,12 @@ export default { status: 400, headers: { "Content-Type": "application/json", - "Access-Control-Allow-Origin": "*", + ...getCorsHeaders(), }, }, ); } - // 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 }); @@ -418,12 +426,11 @@ export default { } catch { 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": "*" } }, + { status: 400, headers: { "Content-Type": "application/json", ...getCorsHeaders() } }, ); } } - // 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 }); @@ -435,12 +442,11 @@ export default { } catch { 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": "*" } }, + { status: 400, headers: { "Content-Type": "application/json", ...getCorsHeaders() } }, ); } } - // Models list if (url.pathname === "/v1/models" && req.method === "GET") { const authErr = requireAuth(req); if (authErr) return authErr; @@ -452,11 +458,10 @@ export default { })); return new Response( JSON.stringify({ object: "list", data: models }), - { status: 200, headers: { "Content-Type": "application/json", "Access-Control-Allow-Origin": "*" } }, + { status: 200, headers: { "Content-Type": "application/json", ...getCorsHeaders() } }, ); } - // Generic HTTP relay return handleRelay(req, env); }, };