187 lines
7.3 KiB
TypeScript
187 lines
7.3 KiB
TypeScript
/**
|
||
* HTTP / SSE 入口（架构设计第 2 节）。
|
||
*
|
||
* 只做协议：路由、请求解析、身份解析、响应与错误包装、事件流推送。
|
||
* 业务编排在 app/，控制面在 core/（A4 起）。路由表来自协议里的 ENDPOINT_SPECS，
|
||
* 保证「文档写了什么端点，代码就实现什么端点」。
|
||
*/
|
||
import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http';
|
||
|
||
import { ENDPOINT_SPECS, type CreateSessionRequest, type EndpointSpec, type PlatformEvent, type SendMessageRequest } from '@aps/contracts';
|
||
|
||
import { AppError } from '../app/errors.js';
|
||
import type { ApplicationService } from '../app/service.js';
|
||
import { resolveSubject } from './dev-identity.js';
|
||
|
||
const MAX_BODY_BYTES = 1_000_000;
|
||
const HEARTBEAT_MS = 15_000;
|
||
|
||
interface Route {
|
||
spec: EndpointSpec;
|
||
regex: RegExp;
|
||
names: string[];
|
||
}
|
||
|
||
const ROUTES: Route[] = ENDPOINT_SPECS.map((spec) => {
|
||
const names: string[] = [];
|
||
const pattern = spec.path.replace(/\{(\w+)\}/g, (_match, name: string) => {
|
||
names.push(name);
|
||
return '([^/]+)';
|
||
});
|
||
return { spec, regex: new RegExp(`^${pattern}$`), names };
|
||
});
|
||
|
||
export function createRequestHandler(service: ApplicationService): (req: IncomingMessage, res: ServerResponse) => void {
|
||
return (req, res) => {
|
||
void handleRequest(req, res, service).catch((error: unknown) => respondError(res, error));
|
||
};
|
||
}
|
||
|
||
export function createHttpServer(service: ApplicationService): Server {
|
||
return createServer(createRequestHandler(service));
|
||
}
|
||
|
||
async function handleRequest(req: IncomingMessage, res: ServerResponse, service: ApplicationService): Promise<void> {
|
||
setCorsHeaders(req, res);
|
||
if (req.method === 'OPTIONS') {
|
||
res.writeHead(204).end();
|
||
return;
|
||
}
|
||
|
||
const url = new URL(req.url ?? '/', 'http://localhost');
|
||
const match = matchRoute(req.method ?? 'GET', url.pathname);
|
||
if (!match) {
|
||
throw new AppError('not_found', `没有这个端点：${req.method ?? 'GET'} ${url.pathname}`, { status: 404 });
|
||
}
|
||
|
||
const { route, params } = match;
|
||
const subject = resolveSubject(req.headers);
|
||
const sessionId = params.get('id') ?? '';
|
||
const projectId = url.searchParams.get('projectId') ?? undefined;
|
||
const sessionFilter = url.searchParams.get('sessionId') ?? undefined;
|
||
|
||
switch (`${route.spec.method} ${route.spec.path}`) {
|
||
case 'GET /v1/health':
|
||
return sendJson(res, 200, service.health());
|
||
case 'GET /v1/me':
|
||
return sendJson(res, 200, service.me(subject));
|
||
case 'GET /v1/projects':
|
||
return sendJson(res, 200, service.listProjects());
|
||
case 'GET /v1/sessions':
|
||
return sendJson(res, 200, service.listSessions(projectId));
|
||
case 'POST /v1/sessions':
|
||
return sendJson(res, 200, service.createSession((await readJson(req)) as CreateSessionRequest));
|
||
case 'GET /v1/sessions/{id}':
|
||
return sendJson(res, 200, service.getSession(sessionId));
|
||
case 'GET /v1/sessions/{id}/events':
|
||
return openEventStream(req, res, service, sessionId);
|
||
case 'POST /v1/sessions/{id}/messages':
|
||
return sendJson(res, 202, service.sendMessage(subject, sessionId, (await readJson(req)) as SendMessageRequest));
|
||
case 'POST /v1/sessions/{id}/cancel':
|
||
return sendJson(res, 200, service.cancel(sessionId));
|
||
case 'GET /v1/capabilities':
|
||
return sendJson(res, 200, service.capabilities(subject, sessionFilter));
|
||
case 'GET /v1/executions/{executionId}':
|
||
// 执行记录在 A2（Core.invoke）之后才有
|
||
throw new AppError('not_implemented', '执行记录将在 A2 阶段实现', { status: 501 });
|
||
case 'POST /v1/approvals/{approvalId}':
|
||
// 确认卡在 A4（控制面）之后才有
|
||
throw new AppError('not_implemented', '确认卡将在 A4 阶段实现', { status: 501 });
|
||
default:
|
||
throw new AppError('not_found', `未实现的端点：${route.spec.method} ${route.spec.path}`, { status: 404 });
|
||
}
|
||
}
|
||
|
||
function matchRoute(method: string, pathname: string): { route: Route; params: Map<string, string> } | undefined {
|
||
for (const route of ROUTES) {
|
||
if (route.spec.method !== method) continue;
|
||
const matched = route.regex.exec(pathname);
|
||
if (!matched) continue;
|
||
const params = new Map<string, string>();
|
||
route.names.forEach((name, index) => params.set(name, matched[index + 1] ?? ''));
|
||
return { route, params };
|
||
}
|
||
return undefined;
|
||
}
|
||
|
||
function openEventStream(req: IncomingMessage, res: ServerResponse, service: ApplicationService, sessionId: string): void {
|
||
const url = new URL(req.url ?? '/', 'http://localhost');
|
||
const headerValue = req.headers['last-event-id'];
|
||
const raw = url.searchParams.get('since') ?? (Array.isArray(headerValue) ? headerValue[0] : headerValue) ?? '0';
|
||
const parsed = Number.parseInt(String(raw), 10);
|
||
const since = Number.isFinite(parsed) && parsed > 0 ? parsed : 0;
|
||
|
||
service.lastSeq(sessionId); // 会话不存在时先 404
|
||
|
||
res.writeHead(200, {
|
||
'content-type': 'text/event-stream; charset=utf-8',
|
||
'cache-control': 'no-cache, no-transform',
|
||
connection: 'keep-alive',
|
||
'x-accel-buffering': 'no',
|
||
});
|
||
res.write(': connected\n\n'); // 让客户端立刻知道连接已建立
|
||
|
||
const unsubscribe = service.subscribe(sessionId, since, (event) => writeEvent(res, event));
|
||
const heartbeat = setInterval(() => service.heartbeat(sessionId), HEARTBEAT_MS);
|
||
heartbeat.unref();
|
||
|
||
req.on('close', () => {
|
||
clearInterval(heartbeat);
|
||
unsubscribe();
|
||
res.end();
|
||
});
|
||
}
|
||
|
||
function writeEvent(res: ServerResponse, event: PlatformEvent): void {
|
||
res.write(`id: ${event.seq}\nevent: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`);
|
||
}
|
||
|
||
async function readJson(req: IncomingMessage): Promise<unknown> {
|
||
const chunks: Buffer[] = [];
|
||
let size = 0;
|
||
for await (const chunk of req) {
|
||
const buffer = chunk as Buffer;
|
||
size += buffer.length;
|
||
if (size > MAX_BODY_BYTES) throw new AppError('payload_too_large', '请求体过大', { status: 413 });
|
||
chunks.push(buffer);
|
||
}
|
||
if (size === 0) return {};
|
||
const text = Buffer.concat(chunks).toString('utf8');
|
||
try {
|
||
return JSON.parse(text);
|
||
} catch {
|
||
throw new AppError('invalid_json', '请求体不是合法 JSON', { status: 400 });
|
||
}
|
||
}
|
||
|
||
function sendJson(res: ServerResponse, status: number, body: unknown): void {
|
||
const payload = JSON.stringify(body);
|
||
res.writeHead(status, { 'content-type': 'application/json; charset=utf-8' });
|
||
res.end(payload);
|
||
}
|
||
|
||
function respondError(res: ServerResponse, error: unknown): void {
|
||
if (res.headersSent) {
|
||
res.end();
|
||
return;
|
||
}
|
||
const appError =
|
||
error instanceof AppError
|
||
? error
|
||
: new AppError('internal', error instanceof Error ? error.message : '内部错误', { status: 500, retryable: true });
|
||
sendJson(res, appError.status, {
|
||
error: { code: appError.code, message: appError.message, retryable: appError.retryable },
|
||
});
|
||
}
|
||
|
||
function setCorsHeaders(req: IncomingMessage, res: ServerResponse): void {
|
||
// 开发期回显来源，便于桌面端与浏览器本地联调；A4.1 身份落地后按白名单收紧。
|
||
const origin = req.headers.origin;
|
||
if (origin) {
|
||
res.setHeader('access-control-allow-origin', origin);
|
||
res.setHeader('vary', 'origin');
|
||
res.setHeader('access-control-allow-credentials', 'true');
|
||
}
|
||
res.setHeader('access-control-allow-methods', 'GET,POST,OPTIONS');
|
||
res.setHeader('access-control-allow-headers', 'content-type,last-event-id,x-aps-client,authorization');
|
||
} |