lyon
245 lines
9.7 KiB
TypeScript
245 lines
9.7 KiB
TypeScript
#!/usr/bin/env bun
|
|
import { request as httpRequest } from "node:http";
|
|
import { request as httpsRequest } from "node:https";
|
|
import { Readable } from "node:stream";
|
|
|
|
import {
|
|
healthPayload,
|
|
resolveHostPort,
|
|
} from "../../internal/dev-entrypoint/http.mjs";
|
|
|
|
const serviceId = "hwlab-edge-proxy";
|
|
const upstream = process.env.HWLAB_EDGE_UPSTREAM || "http://hwlab-cloud-api.hwlab-dev.svc.cluster.local:6667";
|
|
const proxyTimeoutMs = parseTimeout(process.env.HWLAB_EDGE_PROXY_TIMEOUT_MS, 1260000, {
|
|
min: 1000,
|
|
max: 2400000
|
|
});
|
|
const idleTimeoutSeconds = parseTimeout(process.env.HWLAB_EDGE_IDLE_TIMEOUT_SECONDS, 120, {
|
|
min: 1,
|
|
max: 255
|
|
});
|
|
const { host, port } = resolveHostPort({
|
|
listenEnv: "HWLAB_EDGE_LISTEN",
|
|
hostEnv: "HWLAB_EDGE_HOST",
|
|
portEnv: "HWLAB_EDGE_PORT",
|
|
fallbackPort: 6667
|
|
});
|
|
|
|
function proxyDetails() {
|
|
return {
|
|
upstream,
|
|
timeoutMs: proxyTimeoutMs,
|
|
idleTimeoutSeconds,
|
|
mode: "dev-edge-proxy"
|
|
};
|
|
}
|
|
|
|
function parseTimeout(value, fallback, { min, max }) {
|
|
const parsed = Number.parseInt(value || "", 10);
|
|
if (!Number.isInteger(parsed)) return fallback;
|
|
return Math.min(Math.max(parsed, min), max);
|
|
}
|
|
|
|
const server = Bun.serve({
|
|
hostname: host,
|
|
port,
|
|
idleTimeout: idleTimeoutSeconds,
|
|
async fetch(request, bunServer) {
|
|
const url = new URL(request.url);
|
|
|
|
if (isWebSocketUpgrade(request)) {
|
|
if (url.pathname === "/v1/internal" || url.pathname.startsWith("/v1/internal/")) return text("not found", 404);
|
|
const upgraded = bunServer.upgrade(request, { data: { upstreamUrl: upstreamWebSocketUrl(request, upstream), upstreamSocket: null, upstreamOpen: false, pendingMessages: [] } });
|
|
return upgraded ? undefined : json({ ok: false, status: "upgrade_required" }, 426);
|
|
}
|
|
|
|
if (request.method === "GET" && url.pathname === "/health/live") return proxyFetch(request, upstream, proxyTimeoutMs);
|
|
|
|
if (request.method === "GET" && (url.pathname === "/health" || url.pathname === "/edge/health")) {
|
|
return json(healthPayload({ serviceId, role: "public-dev-ingress", details: proxyDetails() }));
|
|
}
|
|
|
|
if (request.method === "GET" && (url.pathname === "/status" || url.pathname === "/routes")) {
|
|
return json({
|
|
serviceId,
|
|
environment: "dev",
|
|
status: "ok",
|
|
upstream,
|
|
routes: [{ pathPrefix: "/", upstream, mode: "http-proxy" }]
|
|
});
|
|
}
|
|
|
|
if (request.method === "GET" && url.pathname === "/help") {
|
|
return json({ serviceId, commands: ["GET /health", "GET /edge/health", "GET /health/live", "GET /status", "GET /routes"] });
|
|
}
|
|
|
|
if (url.pathname === "/v1/internal" || url.pathname.startsWith("/v1/internal/")) {
|
|
return json({ error: { code: "not_found", message: "Internal HWLAB routes are not exposed through the public edge proxy." } }, 404);
|
|
}
|
|
|
|
return proxyFetch(request, upstream, proxyTimeoutMs);
|
|
},
|
|
websocket: {
|
|
open(socket: any) {
|
|
const upstreamSocket = new WebSocket(socket.data.upstreamUrl);
|
|
socket.data.upstreamSocket = upstreamSocket;
|
|
upstreamSocket.addEventListener("open", () => {
|
|
logWsBridge("upstream_open", socket.data.upstreamUrl);
|
|
socket.data.upstreamOpen = true;
|
|
for (const message of socket.data.pendingMessages.splice(0)) upstreamSocket.send(message);
|
|
});
|
|
upstreamSocket.addEventListener("message", (event) => {
|
|
logWsBridge("upstream_message", socket.data.upstreamUrl);
|
|
try {
|
|
socket.send(event.data);
|
|
} catch {
|
|
logWsBridge("client_send_failed", socket.data.upstreamUrl);
|
|
upstreamSocket.close();
|
|
}
|
|
});
|
|
upstreamSocket.addEventListener("close", (event) => {
|
|
logWsBridge("upstream_close", socket.data.upstreamUrl, { code: event.code, reason: event.reason || null });
|
|
try {
|
|
socket.close(event.code || 1000, event.reason || undefined);
|
|
} catch {}
|
|
});
|
|
upstreamSocket.addEventListener("error", () => {
|
|
logWsBridge("upstream_error", socket.data.upstreamUrl);
|
|
try {
|
|
socket.close(1011, "upstream websocket error");
|
|
} catch {}
|
|
});
|
|
},
|
|
message(socket: any, message: string | Buffer) {
|
|
const upstreamSocket = socket.data.upstreamSocket;
|
|
if (socket.data.upstreamOpen && upstreamSocket?.readyState === WebSocket.OPEN) upstreamSocket.send(message);
|
|
else socket.data.pendingMessages.push(message);
|
|
},
|
|
close(socket: any) {
|
|
const upstreamSocket = socket.data.upstreamSocket;
|
|
if (upstreamSocket?.readyState === WebSocket.OPEN || upstreamSocket?.readyState === WebSocket.CONNECTING) upstreamSocket.close();
|
|
}
|
|
}
|
|
});
|
|
|
|
process.stdout.write(`${JSON.stringify({ serviceId, status: "listening", host, port: server.port })}\n`);
|
|
|
|
for (const signal of ["SIGINT", "SIGTERM"]) {
|
|
process.on(signal, () => {
|
|
server.stop(true);
|
|
process.exit(0);
|
|
});
|
|
}
|
|
|
|
async function proxyFetch(request: Request, upstreamUrl: string, timeoutMs: number) {
|
|
const target = new URL(upstreamHttpUrl(request, upstreamUrl));
|
|
return await proxyNodeRequest(request, target, upstreamUrl, timeoutMs);
|
|
}
|
|
|
|
function proxyNodeRequest(request: Request, target: URL, upstreamUrl: string, timeoutMs: number): Promise<Response> {
|
|
const method = request.method || "GET";
|
|
let timedOut = false;
|
|
return new Promise((resolve) => {
|
|
let settled = false;
|
|
let timeout: ReturnType<typeof setTimeout> | null = null;
|
|
const fail = () => {
|
|
if (settled) return;
|
|
settled = true;
|
|
if (timeout) clearTimeout(timeout);
|
|
resolve(proxyFailureResponse({ timedOut, timeoutMs, upstreamUrl }));
|
|
};
|
|
const transport = target.protocol === "https:" ? httpsRequest : httpRequest;
|
|
const upstreamRequest = transport(target, { method, headers: nodeRequestHeaders(request.headers, target) }, (upstreamResponse) => {
|
|
if (settled) {
|
|
upstreamResponse.resume();
|
|
return;
|
|
}
|
|
settled = true;
|
|
if (timeout) clearTimeout(timeout);
|
|
const status = upstreamResponse.statusCode ?? 502;
|
|
const body = method === "HEAD" || status === 204 || status === 304 ? null : Readable.toWeb(upstreamResponse) as unknown as ReadableStream<Uint8Array>;
|
|
resolve(new Response(body, { status, headers: responseHeadersFromNode(upstreamResponse.headers) }));
|
|
});
|
|
timeout = setTimeout(() => {
|
|
timedOut = true;
|
|
upstreamRequest.destroy(new Error(`upstream timed out after ${timeoutMs}ms`));
|
|
}, timeoutMs);
|
|
upstreamRequest.on("error", fail);
|
|
if (method === "GET" || method === "HEAD" || !request.body) {
|
|
upstreamRequest.end();
|
|
return;
|
|
}
|
|
const bodyStream = Readable.fromWeb(request.body as unknown as ReadableStream<Uint8Array>);
|
|
bodyStream.on("error", (error) => upstreamRequest.destroy(error));
|
|
bodyStream.pipe(upstreamRequest);
|
|
});
|
|
}
|
|
|
|
function nodeRequestHeaders(headers: Headers, target: URL): Record<string, string> {
|
|
const next: Record<string, string> = {};
|
|
for (const [key, value] of headers.entries()) {
|
|
if (isHopByHopHeader(key)) continue;
|
|
next[key] = value;
|
|
}
|
|
next.host = target.host;
|
|
return next;
|
|
}
|
|
|
|
function responseHeadersFromNode(headers: Record<string, string | string[] | number | undefined>): Headers {
|
|
const next = new Headers();
|
|
for (const [key, value] of Object.entries(headers)) {
|
|
if (value === undefined || isHopByHopHeader(key)) continue;
|
|
if (Array.isArray(value)) {
|
|
for (const item of value) next.append(key, item);
|
|
continue;
|
|
}
|
|
next.set(key, String(value));
|
|
}
|
|
return next;
|
|
}
|
|
|
|
function isHopByHopHeader(key: string): boolean {
|
|
return ["connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade"].includes(key.toLowerCase());
|
|
}
|
|
|
|
function proxyFailureResponse({ timedOut, timeoutMs, upstreamUrl }: { timedOut: boolean; timeoutMs: number; upstreamUrl: string }) {
|
|
const code = timedOut ? "proxy_timeout" : "upstream_unavailable";
|
|
const userMessage = timedOut
|
|
? `Code Agent 代理等待上游超过 ${timeoutMs}ms；输入已保留，可稍后重试。`
|
|
: "Code Agent 代理暂时无法连接上游；输入已保留，可稍后重试。";
|
|
return json({ status: "failed", error: { code, layer: "proxy", category: timedOut ? "timeout" : "proxy", retryable: true, userMessage, message: userMessage, blocker: { code, layer: "proxy", retryable: true, summary: userMessage } }, upstream: upstreamUrl, message: userMessage }, 502);
|
|
}
|
|
|
|
function upstreamHttpUrl(request: Request, upstreamUrl: string) {
|
|
const source = new URL(request.url);
|
|
return new URL(source.pathname + source.search, upstreamUrl).toString();
|
|
}
|
|
|
|
function upstreamWebSocketUrl(request: Request, upstreamUrl: string) {
|
|
const target = new URL(upstreamHttpUrl(request, upstreamUrl));
|
|
target.protocol = target.protocol === "https:" ? "wss:" : "ws:";
|
|
return target.toString();
|
|
}
|
|
|
|
function isWebSocketUpgrade(request: Request) {
|
|
return request.headers.get("upgrade")?.toLowerCase() === "websocket";
|
|
}
|
|
|
|
function json(body: any, status = 200) {
|
|
return new Response(`${JSON.stringify(body, null, 2)}\n`, { status, headers: { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" } });
|
|
}
|
|
|
|
function text(body: string, status = 200) {
|
|
return new Response(body, { status, headers: { "content-type": "text/plain; charset=utf-8", "cache-control": "no-store" } });
|
|
}
|
|
|
|
function logWsBridge(event: string, upstreamUrl: string, extra: Record<string, unknown> = {}) {
|
|
process.stderr.write(`${JSON.stringify({ event: `hwlab-edge-proxy.ws.${event}`, upstreamUrl: redactUrlToken(upstreamUrl), observedAt: new Date().toISOString(), ...extra })}\n`);
|
|
}
|
|
|
|
function redactUrlToken(urlValue: string) {
|
|
const parsed = new URL(urlValue);
|
|
if (parsed.searchParams.has("token")) parsed.searchParams.set("token", "<redacted>");
|
|
return parsed.toString();
|
|
}
|