/** * 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'; import { clientDistDir, createStaticHandler } from './static.js'; const MAX_BODY_BYTES = 1_000_000; const HEARTBEAT_MS = 15_000; interface Route { spec: EndpointSpec; regex: RegExp; names: string[]; } const serveStatic = createStaticHandler(clientDistDir()); 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 { setCorsHeaders(req, res); if (req.method === 'OPTIONS') { res.writeHead(204).end(); return; } const url = new URL(req.url ?? '/', 'http://localhost'); // 非 API 路径先尝试托管客户端产物(A1.5;未构建时返回 false,继续走协议错误处理) if ((req.method === 'GET' || req.method === 'HEAD') && !url.pathname.startsWith('/v1')) { if (await serveStatic(url.pathname, res)) return; } 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 } | 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(); 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 { 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'); }