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(withDevErrorDiagnostic(body, statusCode), null, 2); response.writeHead(statusCode, { "content-type": "application/json; charset=utf-8", "content-length": Buffer.byteLength(payload) }); response.end(payload); } function withDevErrorDiagnostic(body, statusCode) { if (!body || typeof body !== "object" || Array.isArray(body) || statusCode < 400 || !("error" in body)) return body; const error = typeof body.error === "object" && body.error !== null && !Array.isArray(body.error) ? { ...body.error } : { code: String(body.error || "unknown_error"), message: String(body.message || body.error || "unknown_error") }; const businessTraceId = typeof error.traceId === "string" && /^trc_[A-Za-z0-9_.:-]+$/u.test(error.traceId) ? error.traceId : typeof body.traceId === "string" && /^trc_[A-Za-z0-9_.:-]+$/u.test(body.traceId) ? body.traceId : null; return { ...body, error: { ...error, userMessage: error.userMessage || error.message || body.message || error.code, diagnostic: { contractVersion: "hwlab-error-diagnostic-v1", traceId: /^[0-9a-f]{32}$/u.test(String(error.traceId ?? body.traceId ?? "")) ? String(error.traceId ?? body.traceId).toLowerCase() : null, businessTraceId, requestId: null, serviceId: body.serviceId || error.serviceId || "hwlab-dev-entrypoint", route: String(error.route || body.path || "/").split("?")[0].slice(0, 180), layer: error.layer || "dev-entrypoint", category: error.category || (statusCode >= 500 ? "server" : "client"), code: error.code || String(body.error || "unknown_error"), httpStatus: statusCode, source: "dev-entrypoint", observedAt: new Date().toISOString(), valuesPrinted: false }, valuesPrinted: error.valuesPrinted === true ? true : false } }; } 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); }