pikasTech-HWLAB/cmd/hwlab-edge-proxy/main.ts

#!/usr/bin/env bun
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 = upstreamHttpUrl(request, upstreamUrl);
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), timeoutMs);
  try {
    return await fetch(target, {
      method: request.method,
      headers: request.headers,
      body: request.method === "GET" || request.method === "HEAD" ? undefined : request.body,
      redirect: "manual",
      signal: controller.signal
    });
  } catch (error) {
    const timedOut = controller.signal.aborted;
    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);
  } finally {
    clearTimeout(timeout);
  }
}

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();
}