/** * WmailHub Node —— 纯后端邮件转发节点(docs/03、docs/05 Phase 5)。 * * 职责(仅此而已): * Email Worker --ingress secret--> Node --HMAC 签名--> Console * - 校验 Worker 请求(ingress secret) * - 转发到 Console(有限重试:网络错误/5xx 最多 3 次;401/403 不重试) * - 心跳上报(60s,可配置) * - bootstrap:支持 enrollment token 自动换取身份,secret 落盘 0600 * - 结构化 JSON 日志、优雅退出 * * 配置来源(优先级):环境变量 > config.json(bootstrap 写入) * WMAILHUB_CONSOLE_URL Console 地址 * WMAILHUB_NODE_ID 节点 ID * WMAILHUB_NODE_SECRET node credential secret * WMAILHUB_INGRESS_SECRET Worker -> Node 共享密钥 * WMAILHUB_ENROLLMENT_TOKEN 首次启动用 token 换身份(可选) * WMAILHUB_PORT 监听端口(默认 18788) */ import type { Server } from "bun"; import path from "node:path"; import { signNodeRequest, INGRESS_SECRET_HEADER, NODE_HEARTBEAT_INTERVAL_MS, } from "../../shared/src/index.ts"; import { APP_VERSION } from "./version.ts"; /** # 成员变量: Node 运行配置 **/ interface NodeConfig { consoleUrl: string; nodeId: string; nodeSecret: string; ingressSecret: string; port: number; } /** # 成员变量: config.json 落盘位置 **/ const CONFIG_FILE = path.resolve(import.meta.dir, "../config.json"); /** # 函数 * 结构化 JSON 日志 */ function log(lvl: "info" | "warn" | "error", msg: string, extra: Record = {}) { console.log(JSON.stringify({ t: new Date().toISOString(), lvl, msg, ...extra })); } /** # 函数 * 加载配置:环境变量优先,其次 config.json(enrollment 产物) */ async function loadConfig(): Promise { let file: Partial = {}; try { file = (await Bun.file(CONFIG_FILE).json()) as Partial; } catch { // 无 config.json,纯环境变量模式 } const env = process.env; const consoleUrl = env.WMAILHUB_CONSOLE_URL ?? file.consoleUrl; const ingressSecret = env.WMAILHUB_INGRESS_SECRET ?? file.ingressSecret; let nodeId = env.WMAILHUB_NODE_ID ?? file.nodeId; let nodeSecret = env.WMAILHUB_NODE_SECRET ?? file.nodeSecret; if (!consoleUrl || !ingressSecret) { console.error("missing WMAILHUB_CONSOLE_URL / WMAILHUB_INGRESS_SECRET"); process.exit(1); } // bootstrap enrollment:有 token 而无身份时自动换取并落盘(0600) const enrollmentToken = env.WMAILHUB_ENROLLMENT_TOKEN ?? ""; if ((!nodeId || !nodeSecret) && enrollmentToken) { log("info", "enrolling with console"); const resp = await fetch(new URL("/api/internal/enroll", consoleUrl), { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ enrollment_token: enrollmentToken }), }); if (!resp.ok) { log("error", "enroll failed", { status: resp.status }); process.exit(1); } const data = (await resp.json()) as { node_id: string; credential_id: string; credential_secret: string; ingress_secret: string; }; nodeId = data.node_id; nodeSecret = data.credential_secret; await Bun.write(CONFIG_FILE, JSON.stringify({ consoleUrl, nodeId, nodeSecret, ingressSecret: ingressSecret }), { mode: 0o600, }); log("info", "enrolled, secure config written", { node_id: nodeId, credential_id: data.credential_id }); } if (!nodeId || !nodeSecret) { console.error("missing WMAILHUB_NODE_ID / WMAILHUB_NODE_SECRET (or WMAILHUB_ENROLLMENT_TOKEN)"); process.exit(1); } return { consoleUrl: consoleUrl.replace(/\/$/, ""), nodeId, nodeSecret, ingressSecret, port: Number(env.WMAILHUB_PORT ?? 18788), }; } /** # 函数 * 转发邮件到 Console:构造 HMAC 签名请求;网络错误/5xx/429 有限重试,401/403 立即放弃 */ async function forwardToConsole(cfg: NodeConfig, raw: Uint8Array, mailFrom: string, rcptTo: string): Promise { const url = `${cfg.consoleUrl}/api/internal/nodes/${cfg.nodeId}/emails`; const maxAttempts = 3; let lastError = ""; for (let attempt = 1; attempt <= maxAttempts; attempt++) { const headers = await signNodeRequest(cfg.nodeSecret, cfg.nodeId, "POST", url, raw); headers["Content-Type"] = "message/rfc822"; headers["X-WmailHub-Mail-From"] = mailFrom; headers["X-WmailHub-Rcpt-To"] = rcptTo; try { const resp = await fetch(url, { method: "POST", headers, body: raw.buffer.slice(raw.byteOffset, raw.byteOffset + raw.byteLength) as ArrayBuffer, }); if (resp.ok) return resp; if (resp.status === 401 || resp.status === 403) { // 凭据问题:重试无意义(docs/04:不得无限重试) return resp; } lastError = `console responded ${resp.status}`; } catch (err) { lastError = err instanceof Error ? err.message : String(err); } if (attempt < maxAttempts) { const delayMs = 500 * 2 ** (attempt - 1); log("warn", "forward retry scheduled", { attempt, delayMs, error: lastError }); await new Promise((r) => setTimeout(r, delayMs)); } } return new Response(JSON.stringify({ error: lastError }), { status: 503 }); } /** # 函数 * 发送一次心跳;401/403 只告警不退出 */ async function heartbeat(cfg: NodeConfig): Promise { const url = `${cfg.consoleUrl}/api/internal/nodes/${cfg.nodeId}/heartbeat`; const body = new TextEncoder().encode(JSON.stringify({ version: APP_VERSION })); const headers = await signNodeRequest(cfg.nodeSecret, cfg.nodeId, "POST", url, body); headers["Content-Type"] = "application/json"; try { const resp = await fetch(url, { method: "POST", headers, body }); if (!resp.ok) { log("warn", "heartbeat rejected", { status: resp.status }); } else { log("info", "heartbeat ok"); } } catch (err) { log("warn", "heartbeat failed", { error: err instanceof Error ? err.message : String(err) }); } } /** # 函数 * 主入口:加载配置 -> 起服务 -> 定时心跳 -> 优雅退出 */ async function main() { const cfg = await loadConfig(); let shuttingDown = false; let inflight = 0; const server: Server = Bun.serve({ port: cfg.port, hostname: "0.0.0.0", idleTimeout: 30, async fetch(request) { const url = new URL(request.url); if (url.pathname === "/__health") { return Response.json({ ok: true, node_id: cfg.nodeId, version: APP_VERSION }); } // 邮件 ingress:仅接受 Email Worker 的签名 secret 请求 if (request.method === "POST" && url.pathname === "/ingress/email") { if (request.headers.get(INGRESS_SECRET_HEADER) !== cfg.ingressSecret) { log("warn", "ingress rejected: bad secret", { ip: request.headers.get("x-forwarded-for") ?? "" }); return Response.json({ error: "forbidden" }, { status: 403 }); } const raw = new Uint8Array(await request.arrayBuffer()); if (raw.byteLength === 0) { return Response.json({ error: "empty body" }, { status: 400 }); } const mailFrom = request.headers.get("X-WmailHub-Mail-From") ?? ""; const rcptTo = request.headers.get("X-WmailHub-Rcpt-To") ?? ""; inflight++; try { const resp = await forwardToConsole(cfg, raw, mailFrom, rcptTo); const body = (await resp.json().catch(() => ({}))) as Record; log(resp.ok ? "info" : "error", resp.ok ? "email forwarded" : "forward failed", { bytes: raw.byteLength, status: resp.status, email_id: body.id, }); return Response.json(body, { status: resp.status }); } finally { inflight--; } } return Response.json({ error: "not found" }, { status: 404 }); }, }); log("info", "node started", { node_id: cfg.nodeId, port: cfg.port, console: cfg.consoleUrl, version: APP_VERSION, }); // 心跳(启动即一次) void heartbeat(cfg); const heartbeatTimer = setInterval(() => void heartbeat(cfg), NODE_HEARTBEAT_INTERVAL_MS); /** 优雅退出:停止接新请求,等在途转发完成 */ async function shutdown(signal: string) { if (shuttingDown) return; shuttingDown = true; log("info", "node shutting down", { signal, inflight }); clearInterval(heartbeatTimer); server.stop(false); // 停止接受新连接,保持既有 const deadline = Date.now() + 10_000; while (inflight > 0 && Date.now() < deadline) { await new Promise((r) => setTimeout(r, 100)); } server.stop(true); process.exit(0); } process.on("SIGINT", () => void shutdown("SIGINT")); process.on("SIGTERM", () => void shutdown("SIGTERM")); } main().catch((err) => { log("error", "node crashed", { error: err instanceof Error ? err.stack : String(err) }); process.exit(1); });