实验目标与边界
前两节分别讨论了网关的职责与对外契约。这一节把它们落到可运行的代码：一个只依赖 node:http 的模型网关，提供 POST /v1/chat，同时支持非流式 JSON 与 SSE 流式响应，内部用 3.1 的阶段组合器串联认证、校验与执行，提供方使用 3.2 的确定性 mock。
实验刻意不实现配额、缓存与审计。它们在 3.1 里已经作为职责讨论过，实现方式是在 pipeline 上追加阶段，不改变 HTTP 契约。把范围收窄，才能看清“协议转换”与“策略执行”各自的位置。
HTTP 接口
| 项目 | 约定 |
|---|---|
| 方法与路径 | POST /v1/chat |
| 认证 | authorization: Bearer <api-key> |
| 请求体 | ChatRequest JSON |
| 非流式响应 | 200 + ChatResponse JSON |
| 流式响应 | 200 + text/event-stream |
| 参数错误 | 400，错误类别 invalid-request |
| 认证失败 | 401，错误类别 auth |
| 上游失败 | 502 / 503 / 504 |
请求体在进入 pipeline 之前先做一次形状检查。缺少 requestId、messages 或预算字段时直接返回 400，不进入认证与执行。JSON 解析失败同样在 pipeline 之前被拒绝，避免把解析异常误记成提供方故障。非流式响应直接返回 ChatResponse。流式响应先发 SSE 头，再按 start、若干 delta、usage、done 的顺序写帧；如果中途失败，用 error 帧替代 done，HTTP 状态仍然是 200，因为响应头已经发出。
认证与租户隔离
TenantPolicy 把 API key 映射到租户。网关在认证阶段完成映射，后续阶段只使用租户对象。请求体里的 tenantId 必须与 key 解析出的租户一致，否则按认证失败处理。这不是重复校验，而是防止调用方拿着自己的 key 冒充另一个租户。构造网关时检查 key 是否重复。两个租户共用一个 key，配额、审计与吊销都会失去意义。真实系统还应该使用常量时间比较、支持多 key 与轮换，这些不改变本实验的结构。
服务端实现
createGatewayServer 只负责装配，不监听端口。测试可以先创建实例，再选择临时端口，避免固定端口冲突。请求处理函数把 HTTP 细节翻译成上下文，剩下的顺序与错误归一都交给 pipeline。
// file: src/gateway/server.ts
// tier: L1
// verify: node --test test/gateway/server.test.ts
import { createServer } from "node:http";
import type { IncomingMessage, Server, ServerResponse } from "node:http";
import { AiError, normalizeError } from "../core/errors.ts";
import type { AiErrorKind } from "../core/errors.ts";
import { systemClock } from "../core/types.ts";
import type { ChatRequest, ChatResponse, ModelProfile, Usage } from "../core/types.ts";
import { validateRequest } from "../protocol/chat.ts";
import type { ChatProvider } from "../protocol/chat.ts";
import { encodeStreamEvent } from "../protocol/stream.ts";
import type { StreamEvent } from "../protocol/stream.ts";
import { createPipeline, proceed } from "./pipeline.ts";
import type { Stage } from "./pipeline.ts";
export interface StreamChunk {
text: string;
done?: boolean;
usage?: Usage;
finishReason?: ChatResponse["finishReason"];
}
export interface StreamingChatProvider extends ChatProvider {
stream(request: ChatRequest, model: ModelProfile): AsyncIterable<StreamChunk>;
}
export interface TenantPolicy { tenantId: string; apiKey: string }
export interface GatewayOptions {
provider: ChatProvider;
model: ModelProfile;
tenants: TenantPolicy[];
}
interface Context {
res: ServerResponse;
requestId: string;
authorization: string;
startedAt: number;
request: ChatRequest;
tenant?: TenantPolicy;
response?: ChatResponse;
streamStarted: boolean;
}
function statusOf(kind: AiErrorKind): number {
if (kind === "auth") return 401;
if (kind === "quota" || kind === "rate-limit") return 429;
if (kind === "timeout") return 504;
if (kind === "unavailable") return 503;
if (kind === "invalid-request") return 400;
return 502;
}
/** 对外只给稳定类别与安全描述；提供方原文与 cause 留在服务端日志里。 */
function publicMessage(error: AiError): string {
if (error.kind === "provider" || error.kind === "unavailable") return "上游模型服务暂时不可用";
if (error.kind === "auth") return "认证失败";
return error.message;
}
function isChatRequest(value: unknown): value is ChatRequest {
if (typeof value !== "object" || value === null) return false;
const request = value as ChatRequest;
if (!Array.isArray(request.messages)) return false;
const messagesOk = request.messages.every(
(message) => typeof message.role === "string" && typeof message.content === "string",
);
return (
messagesOk &&
typeof request.requestId === "string" &&
typeof request.capability === "string" &&
typeof request.tenantId === "string" &&
typeof request.maxOutputTokens === "number" &&
typeof request.deadlineMs === "number" &&
(request.stream === undefined || typeof request.stream === "boolean")
);
}
function supportsStream(provider: ChatProvider): provider is StreamingChatProvider {
return typeof (provider as Partial<StreamingChatProvider>).stream === "function";
}
function writeJson(res: ServerResponse, status: number, payload: unknown): void {
if (res.headersSent || res.writableEnded) return;
res.writeHead(status, { "content-type": "application/json; charset=utf-8" });
res.end(JSON.stringify(payload));
}
function writeError(res: ServerResponse, error: AiError, requestId: string): void {
writeJson(res, statusOf(error.kind), {
error: {
kind: error.kind,
message: publicMessage(error),
retryable: error.retryable,
requestId,
},
});
}
function writeEvent(res: ServerResponse, event: StreamEvent): void {
if (!res.writableEnded && !res.destroyed) res.write(encodeStreamEvent(event));
}
async function readJson(req: IncomingMessage): Promise<unknown> {
const chunks: Buffer[] = [];
for await (const chunk of req) chunks.push(chunk as Buffer);
if (chunks.length === 0) throw new AiError("invalid-request", "请求体为空");
try {
return JSON.parse(Buffer.concat(chunks).toString("utf8"));
} catch {
throw new AiError("invalid-request", "请求体不是合法 JSON");
}
}
function authStage(options: GatewayOptions): Stage<Context> {
return {
name: "auth",
run(context) {
const token = context.authorization.startsWith("Bearer ")
? context.authorization.slice(7).trim()
: "";
const tenant = options.tenants.find((item) => item.apiKey === token);
if (token === "" || !tenant) throw new AiError("auth", "无效的 API key");
if (context.request.tenantId !== tenant.tenantId) {
throw new AiError("auth", "租户与 API key 不匹配");
}
context.tenant = tenant;
return proceed(context);
},
};
}
function validateStage(): Stage<Context> {
return {
name: "validate",
run(context) {
validateRequest(context.request);
return proceed(context);
},
};
}
async function streamChat(context: Context, options: GatewayOptions): Promise<void> {
const { res, request } = context;
res.writeHead(200, {
"content-type": "text/event-stream; charset=utf-8",
"cache-control": "no-cache, no-transform",
"x-accel-buffering": "no",
});
context.streamStarted = true;
writeEvent(res, { type: "start", requestId: request.requestId, modelId: options.model.id });
let usage: Usage | undefined;
let finishReason: ChatResponse["finishReason"] = "stop";
for await (const chunk of options.provider.stream(request, options.model)) {
if (res.destroyed) throw new AiError("timeout", "客户端在流式响应中断开");
if (chunk.text !== "") writeEvent(res, { type: "delta", text: chunk.text });
if (chunk.usage) usage = chunk.usage;
if (chunk.finishReason) finishReason = chunk.finishReason;
}
writeEvent(res, { type: "usage", usage: usage ?? { inputTokens: 0, outputTokens: 0, costMicros: 0 } });
writeEvent(res, { type: "done", finishReason });
res.end();
}
function executeStage(options: GatewayOptions): Stage<Context> {
return {
name: "execute",
async run(context) {
if (!context.request.stream) {
context.response = await options.provider.complete(context.request, options.model);
} else if (supportsStream(options.provider)) {
await streamChat(context, options);
} else {
throw new AiError("invalid-request", "provider 不支持流式响应");
}
return proceed(context);
},
};
}
// 零依赖模型网关：HTTP 只做协议转换，认证、校验、执行由 pipeline 串起来。
export function createGatewayServer(options: GatewayOptions) {
if (new Set(options.tenants.map((item) => item.apiKey)).size !== options.tenants.length) {
throw new Error("gateway: apiKey 必须在租户之间隔离");
}
const clock = systemClock;
const server = createServer((req, res) => {
void route(req, res).catch((error: unknown) => {
const failure = normalizeError(error);
if (!res.headersSent) writeError(res, failure, "unknown");
else if (!res.writableEnded) res.end();
});
});
async function route(req: IncomingMessage, res: ServerResponse): Promise<void> {
const path = (req.url ?? "/").split("?")[0];
if (path !== "/v1/chat" || req.method !== "POST") {
writeJson(res, 404, { error: { kind: "invalid-request", message: "not found" } });
return;
}
let payload: unknown;
try {
payload = await readJson(req);
} catch (error) {
writeError(res, normalizeError(error), "unknown");
return;
}
if (!isChatRequest(payload)) {
writeError(res, new AiError("invalid-request", "请求体缺少必要字段"), "unknown");
return;
}
const context: Context = {
res,
requestId: payload.requestId,
authorization: req.headers.authorization ?? "",
startedAt: clock.now(),
request: payload,
streamStarted: false,
};
const pipeline = createPipeline<Context>("chat-gateway", {
clock,
remainingMs: (current) => current.startedAt + current.request.deadlineMs - clock.now(),
})
.use(authStage(options))
.use(validateStage())
.use(executeStage(options));
try {
await pipeline.run(context);
if (!context.request.stream && context.response && !res.writableEnded) {
writeJson(res, 200, context.response);
}
} catch (error) {
const failure = normalizeError(error);
if (!context.streamStarted) {
writeError(res, failure, context.requestId);
return;
}
writeEvent(res, {
type: "error",
kind: failure.kind,
message: publicMessage(failure),
retryable: failure.retryable,
});
if (!res.writableEnded) res.end();
}
}
return {
server,
listen(port = 0, host = "127.0.0.1") {
return new Promise((resolve, reject) => {
const onError = (error: Error): void => reject(error);
server.once("error", onError);
server.listen(port, host, () => {
server.off("error", onError);
const address = server.address();
const actualPort = typeof address === "object" && address ? address.port : port;
resolve({ port: actualPort, url: `http://${host}:${actualPort}` });
});
});
},
async close() {
if (!server.listening) return;
server.closeAllConnections();
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
});
},
};
}
三个细节对应前两节的结论。第一，statusOf 把错误类别映射成 HTTP 状态，业务方不需要解析提供方状态码。第二，流式分支在写响应头之前不会设置 streamStarted，因此认证与校验失败仍然能返回带正确状态的 JSON，而不是一条已经开始的 SSE。第三，publicMessage 对 provider 与 unavailable 返回统一话术，提供方原文只留在服务端。
一次 SSE 请求长什么样
下面是一次流式请求的帧序列，文本与用量为案例示例，实际值由 mock 的确定性算法给出：
data: {"type":"start","requestId":"req-sse","modelId":"mock-echo-1"}
data: {"type":"delta","text":"【回声 "}
data: {"type":"usage","usage":{"inputTokens":14,"outputTokens":18,"costMicros":8}}
data: {"type":"done","finishReason":"stop"}
每个 data: 行都是单行 JSON，帧之间用空行分隔。客户端按事件类型处理，不依赖 delta 的切分位置；切分大小是网关的实现细节，可以随提供方变化。
测试：真实端口与真实请求
测试覆盖非流式契约、缺少与伪造 key、租户不匹配、结构错误、SSE 帧顺序、流中途失败与非法 JSON。在实验工程根目录执行：
node --test test/gateway/server.test.ts
本机一次运行的输出末尾如下，耗时随机器变化：
ℹ tests 7
ℹ pass 7
ℹ fail 0
ℹ duration_ms 288.151708
非流式用例先调用 provider.complete 算出期望值，再发真实请求，逐字段比较 output、usage 与 finishReason。SSE 用例把响应体按空行拆帧，用 decodeStreamEvent 解码，断言第一帧是 start、最后一帧是 done、所有 delta 拼接后等于期望输出。流中途失败用例通过 metadata.mockFailAfter 让 mock 在第一个 chunk 后抛错，断言收到 error 且没有 done。
从实验到生产还差什么
- 认证：当前是线性查找与普通字符串比较，生产环境需要常量时间比较、多 key、过期与轮换。
- 配额：实验没有实现租户配额与限流，需要按 3.1 的阶段顺序补上，并区分配额与限流。
- 请求体上限：
readJson没有限制大小，生产环境必须设置上限并在超限时断开连接。 - 取消传播：客户端断开后，当前实现只停止写帧，没有把取消信号传给提供方。
- 背压：
res.write的返回值没有处理，慢客户端会让内存持续增长。 - 持久化：审计与用量只在内存里，进程重启即丢失，生产环境需要落盘或发送到消息队列。
小结
零依赖网关的结构可以很小：HTTP 层只做解析与写回，认证、校验、执行由 pipeline 串联，提供方通过统一契约接入。非流式与流式共享同一套请求、错误与计量，区别只在响应如何写出。SSE 的关键不是分块传输，而是事件语义：start 定模型，delta 增量文本，usage 给用量，done 给结束原因，error 表示中途失败。一旦发出过 delta，就不能重试整条请求。
实验省略的配额、缓存、审计与取消传播，都是生产网关必须补上的部分；它们应该作为新阶段加入 pipeline，而不是散落在 HTTP 处理函数里。
阅读导航：上一节：3.2 统一契约与流式协议 · 下一节：4.1 路由的五个维度 。
常见错误与边界
- 先写 SSE 响应头再认证。认证失败时状态码已经固定为 200，客户端只能从错误帧推断失败。
- 用固定端口写测试。并行运行或端口占用会让测试随机失败，应该监听
127.0.0.1:0。 - 把提供方状态码直接写回。业务方会被迫理解上游语义，换提供方就要改代码。
- 流式失败后补发
done。客户端会把半截输出当成完整结果，应该发error并结束。 - 在 HTTP 处理函数里堆积配额与审计。职责会重新散落，pipeline 也失去复用价值。
- 忽略
res.write的背压。慢客户端会让缓冲区无限增长，最终拖垮进程。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。