pikasTech-HWLAB/internal/dev-entrypoint/http.mjs

import { createServer, request as httpRequest } from "node:http";

import { buildMetadataFromEnv } from "../build-metadata.mjs";
import { DEV_ENDPOINT, ENVIRONMENT_DEV } from "../protocol/index.mjs";

const DEFAULT_PROXY_TIMEOUT_MS = 180000;
export const UPSTREAM_UNAVAILABLE_ERROR_CODE = "upstream_unavailable";
export const UPSTREAM_UNAVAILABLE_MESSAGE = "暂时无法连接上游。";

export function parsePort(value, fallback) {
  const parsed = Number.parseInt(value ?? "", 10);
  if (Number.isInteger(parsed) && parsed > 0 && parsed < 65536) {
    return parsed;
  }
  return fallback;
}

export function resolveHostPort({ listenEnv, hostEnv, portEnv, fallbackPort }) {
  const listen = process.env[listenEnv] ?? "";
  const [listenHost, listenPort] = listen.includes(":") ? listen.split(":") : ["", listen];
  const host = process.env[hostEnv] || listenHost || "0.0.0.0";
  const port = parsePort(process.env[portEnv] || listenPort, fallbackPort);
  return { host, port };
}

export function healthPayload({ serviceId, role, details = {} }) {
  const metadata = buildMetadataFromEnv(process.env, {
    serviceId,
    fallbackImageRepository: `ghcr.io/pikastech/${serviceId}`
  });

  return {
    serviceId,
    environment: ENVIRONMENT_DEV,
    status: "ok",
    revision: metadata.revision,
    service: {
      id: serviceId,
      role,
      healthPath: "/health",
      livePath: "/health/live"
    },
    commit: metadata.commit,
    image: metadata.image,
    build: metadata.build,
    runtime: metadata.runtime,
    endpoint: process.env.HWLAB_PUBLIC_ENDPOINT || DEV_ENDPOINT,
    observedAt: new Date().toISOString(),
    details
  };
}

export function sendJson(response, statusCode, body) {
  const payload = JSON.stringify(body, null, 2);
  response.writeHead(statusCode, {
    "content-type": "application/json; charset=utf-8",
    "content-length": Buffer.byteLength(payload)
  });
  response.end(payload);
}

export function createServiceServer({ serviceId, role, routes = new Map(), details = () => ({}) }) {
  return createServer(async (request, response) => {
    try {
      const url = new URL(request.url || "/", `http://${serviceId}.local`);
      if (request.method === "GET" && (url.pathname === "/health" || url.pathname === "/health/live")) {
        sendJson(response, 200, healthPayload({ serviceId, role, details: details() }));
        return;
      }

      if (request.method === "GET" && url.pathname === "/help") {
        sendJson(response, 200, {
          serviceId,
          commands: ["GET /health", "GET /health/live", ...routes.keys()]
        });
        return;
      }

      const handler = routes.get(`${request.method ?? "GET"} ${url.pathname}`);
      if (!handler) {
        sendJson(response, 404, { error: "not_found", serviceId, path: url.pathname });
        return;
      }

      sendJson(response, 200, await handler({ request, url }));
    } catch (error) {
      sendJson(response, 500, {
        error: "internal_error",
        serviceId,
        message: error instanceof Error ? error.message : String(error)
      });
    }
  });
}

export function listen(server, { serviceId, host, port }) {
  server.listen(port, host, () => {
    process.stdout.write(`${JSON.stringify({ serviceId, status: "listening", host, port })}\n`);
  });

  for (const signal of ["SIGINT", "SIGTERM"]) {
    process.on(signal, () => {
      server.close(() => {
        process.exit(0);
      });
    });
  }
}

export function proxyHttpRequest({ request, response, upstream, timeoutMs = DEFAULT_PROXY_TIMEOUT_MS }) {
  const target = new URL(request.url || "/", upstream);
  const traceId = request.headers["x-trace-id"] ?? null;
  let timedOut = false;
  let settled = false;
  let timeout = null;
  const clearProxyTimeout = () => {
    if (timeout) clearTimeout(timeout);
    timeout = null;
  };
  const sendProxyFailure = (error) => {
    clearProxyTimeout();
    if (settled) return;
    if (response.headersSent) {
      response.destroy(error);
      return;
    }
    settled = true;
    const code = UPSTREAM_UNAVAILABLE_ERROR_CODE;
    const category = timedOut ? "timeout" : "proxy";
    const userMessage = UPSTREAM_UNAVAILABLE_MESSAGE;
    sendJson(response, 502, {
      status: "failed",
      error: {
        code,
        layer: "proxy",
        category,
        retryable: true,
        userMessage,
        message: userMessage,
        timeoutMs: timedOut ? timeoutMs : null,
        traceId,
        route: request.url || null,
        blocker: {
          code,
          layer: "proxy",
          category,
          retryable: true,
          summary: userMessage,
          userMessage,
          traceId,
          route: request.url || null
        }
      },
      upstream,
      traceId,
      message: userMessage
    });
  };
  const proxy = httpRequest(
    target,
    {
      method: request.method,
      headers: {
        ...request.headers,
        host: target.host
      }
    },
    (upstreamResponse) => {
      if (settled) {
        upstreamResponse.resume();
        return;
      }
      settled = true;
      response.writeHead(upstreamResponse.statusCode ?? 502, upstreamResponse.headers);
      upstreamResponse.on("end", clearProxyTimeout);
      upstreamResponse.on("close", clearProxyTimeout);
      upstreamResponse.pipe(response);
    }
  );

  timeout = setTimeout(() => {
    timedOut = true;
    const error = new Error(`upstream timed out after ${timeoutMs}ms`);
    sendProxyFailure(error);
    proxy.destroy(error);
  }, timeoutMs);

  proxy.on("error", sendProxyFailure);

  request.pipe(proxy);
}