import { createServer, type IncomingMessage, type ServerResponse } from 'node:http'; import { IDBFactory } from 'fake-indexeddb'; import { isChatMessageSkeleton, isQuizAttemptSkeleton, needsRuntimeMigration, RUNTIME_DSL_VERSION, runtimeDslVersionOf, validateRuntimeRecord, validateRuntimeSession, } from '@openmaic/dsl'; import type { RuntimeRecordInit, RuntimeSession, RuntimeSessionStatus, ValidationResult, } from '@openmaic/dsl'; import { BrowserRuntimeStore } from '../src/runtime/browser.js'; import { assertJsonValue } from '../src/runtime/json-value.js'; import type { RuntimeAppendOptions, RuntimeSessionInit, RuntimeStore, RuntimeTailOptions, } from '../src/runtime/types.js'; import { RuntimeAppendConflictError } from '../src/runtime/types.js'; export interface HttpConformanceServer { baseUrl: string; fetch: typeof globalThis.fetch; close(): Promise; } export interface HttpConformanceServerOptions { /** Bind a loopback TCP port. Tests can disable this in network-restricted sandboxes. */ listen?: boolean; } interface ErrorBody { error: { code: string; message: string; details?: unknown; }; } class ConformanceHttpError extends Error { constructor( readonly status: number, readonly code: string, message: string, ) { super(message); } } function sendJson(res: ServerResponse, status: number, body: unknown): void { res.writeHead(status, { 'content-type': 'application/json' }); res.end(JSON.stringify(body)); } function sendNoContent(res: ServerResponse): void { res.writeHead(204); res.end(); } async function readJson(req: IncomingMessage): Promise { const chunks: Buffer[] = []; for await (const chunk of req) { chunks.push(typeof chunk === 'string' ? Buffer.from(chunk) : chunk); } if (chunks.length === 0) { throw new ConformanceHttpError(400, 'VALIDATION_FAILED', 'request body must be a JSON object'); } let body: unknown; try { body = JSON.parse(Buffer.concat(chunks).toString('utf8')) as unknown; } catch (error) { const message = error instanceof Error ? error.message : String(error); throw new ConformanceHttpError(400, 'VALIDATION_FAILED', message); } if (typeof body !== 'object' || body === null || Array.isArray(body)) { throw new ConformanceHttpError(400, 'VALIDATION_FAILED', 'request body must be a JSON object'); } return body as T; } function errorResponse(error: unknown): { status: number; body: ErrorBody } { const message = error instanceof Error ? error.message : String(error); if (error instanceof RuntimeAppendConflictError) { return { status: 409, body: { error: { code: 'RUNTIME_APPEND_CONFLICT', message, details: { sessionId: error.sessionId, expectedLastSeq: error.expectedLastSeq, actualLastSeq: error.actualLastSeq, }, }, }, }; } if (error instanceof SyntaxError) { return { status: 400, body: { error: { code: 'VALIDATION_FAILED', message } } }; } if (error instanceof ConformanceHttpError) { return { status: error.status, body: { error: { code: error.code, message } } }; } return { status: 500, body: { error: { code: 'INTERNAL_ERROR', message } } }; } function validationError(result: ValidationResult, label: string): void { if (result.valid) return; const detail = result.errors.map((error) => `${error.path || '/'}: ${error.message}`).join('; '); throw new ConformanceHttpError(400, 'VALIDATION_FAILED', `${label}: ${detail}`); } function assertAddressableSegment(value: string): void { if (value === '.' || value === '..') { throw new ConformanceHttpError( 400, 'VALIDATION_FAILED', `@openmaic/storage: URL path segment must not be ${JSON.stringify(value)}`, ); } } function assertJsonRequestValue(value: unknown, label: string): void { try { assertJsonValue(value, label); } catch (error) { const message = error instanceof Error ? error.message : String(error); throw new ConformanceHttpError(400, 'VALIDATION_FAILED', message); } } function missingSessionError(sessionId: string): ConformanceHttpError { return new ConformanceHttpError( 404, 'SESSION_NOT_FOUND', `@openmaic/storage: no session ${JSON.stringify(sessionId)}`, ); } function assertNotFutureSession(session: RuntimeSession): void { const version = runtimeDslVersionOf(session); if (!needsRuntimeMigration(session) && version !== RUNTIME_DSL_VERSION) { throw new ConformanceHttpError( 409, 'FUTURE_VERSION', `@openmaic/storage: session ${JSON.stringify(session.id)} was written at runtime DSL ` + `version ${JSON.stringify(version)}, newer than this client's ${RUNTIME_DSL_VERSION}`, ); } } function validatePayloadForKind(session: RuntimeSession, payload: unknown): void { if (session.kind === 'chat' && !isChatMessageSkeleton(payload)) { throw new ConformanceHttpError( 400, 'VALIDATION_FAILED', '@openmaic/storage: invalid runtime record: /payload: chat payload must match ' + 'ChatMessageSkeleton (role + content)', ); } if (session.kind === 'quizAttempt' && !isQuizAttemptSkeleton(payload)) { throw new ConformanceHttpError( 400, 'VALIDATION_FAILED', '@openmaic/storage: invalid runtime record: /payload: quizAttempt payload must match ' + 'QuizAttemptSkeleton (phase + answers)', ); } } async function requireSession(store: RuntimeStore, sessionId: string): Promise { const session = await store.getSession(sessionId); if (session === undefined) throw missingSessionError(sessionId); return session; } function pathParts(req: IncomingMessage): { parts: string[]; url: URL } { const url = new URL(req.url ?? '/', 'http://conformance.invalid'); const parts = url.pathname.split('/'); if (parts[0] === '') parts.shift(); return { parts: parts.map((part) => decodeURIComponent(part)), url, }; } async function route( req: IncomingMessage, res: ServerResponse, store: RuntimeStore, ): Promise { const { parts, url } = pathParts(req); const method = req.method ?? 'GET'; if (parts[0] !== 'runtime') { sendJson(res, 404, { error: { code: 'ROUTE_NOT_FOUND', message: 'route not found' } }); return; } if (method === 'POST' && parts.length === 2 && parts[1] === 'sessions') { const init = await readJson(req); assertAddressableSegment(init.id); assertAddressableSegment(init.stageId); assertAddressableSegment(init.learnerKey); validationError( validateRuntimeSession({ ...init, runtimeDslVersion: RUNTIME_DSL_VERSION }), `@openmaic/storage: invalid runtime session ${JSON.stringify(init.id)}`, ); if (await store.getSession(init.id)) { throw new ConformanceHttpError( 409, 'SESSION_ALREADY_EXISTS', `@openmaic/storage: runtime session ${JSON.stringify(init.id)} already exists`, ); } let created: RuntimeSession; try { created = await store.createSession(init); } catch (error) { if (await store.getSession(init.id)) { throw new ConformanceHttpError( 409, 'SESSION_ALREADY_EXISTS', `@openmaic/storage: runtime session ${JSON.stringify(init.id)} already exists`, ); } throw error; } sendJson(res, 201, created); return; } if (parts[1] === 'sessions' && parts.length >= 3) { const sessionId = parts[2]!; if (method === 'GET' && parts.length === 3) { const session = await store.getSession(sessionId); if (session === undefined) { sendJson(res, 404, { error: { code: 'SESSION_NOT_FOUND', message: `@openmaic/storage: no session ${JSON.stringify(sessionId)}`, }, }); } else { sendJson(res, 200, session); } return; } if (method === 'PATCH' && parts.length === 4 && parts[3] === 'status') { const body = await readJson< { status: RuntimeSessionStatus; updatedAt: string } & RuntimeTailOptions >(req); const session = await requireSession(store, sessionId); assertNotFutureSession(session); validationError( validateRuntimeSession({ ...session, status: body.status, updatedAt: body.updatedAt }), `@openmaic/storage: invalid runtime session ${JSON.stringify(sessionId)}`, ); await store.setSessionStatus(sessionId, body.status, body.updatedAt, { ...(body.expectedLastSeq === undefined ? {} : { expectedLastSeq: body.expectedLastSeq }), }); sendNoContent(res); return; } if (method === 'DELETE' && parts.length === 3) { await store.deleteSession(sessionId); sendNoContent(res); return; } if (method === 'POST' && parts.length === 4 && parts[3] === 'records') { const body = await readJson( req, ); const { expectedLastSeq, sessionTransition, ...init } = body; if (init.sessionId !== sessionId) { throw new ConformanceHttpError( 400, 'VALIDATION_FAILED', 'invalid runtime record: body sessionId does not match the request path', ); } validationError( validateRuntimeRecord({ ...init, seq: 0 }), `@openmaic/storage: invalid runtime record ${JSON.stringify(init.id)}`, ); const session = await requireSession(store, sessionId); assertNotFutureSession(session); validatePayloadForKind(session, init.payload); if (session.status !== 'active') { throw new ConformanceHttpError( 400, 'VALIDATION_FAILED', `@openmaic/storage: cannot append to session ${JSON.stringify(sessionId)} with ` + `status '${session.status}' — records may only be appended to an active session`, ); } sendJson( res, 201, await store.appendRecord(init, { ...(expectedLastSeq === undefined ? {} : { expectedLastSeq }), ...(sessionTransition === undefined ? {} : { sessionTransition }), }), ); return; } if (method === 'GET' && parts.length === 4 && parts[3] === 'records') { const sceneId = url.searchParams.get('sceneId'); sendJson( res, 200, await store.listRecords(sessionId, sceneId === null ? undefined : { sceneId }), ); return; } } if ( parts[1] === 'stages' && parts.length === 6 && parts[3] === 'learners' && parts[5] === 'sessions' && method === 'GET' ) { sendJson(res, 200, await store.listSessions(parts[2]!, parts[4]!)); return; } if (parts[1] === 'learners' && parts[2] === 'merge' && parts.length === 3 && method === 'POST') { const body = await readJson<{ fromLearnerKey?: unknown; toLearnerKey?: unknown }>(req); if ( typeof body.fromLearnerKey !== 'string' || body.fromLearnerKey === '' || typeof body.toLearnerKey !== 'string' || body.toLearnerKey === '' ) { throw new ConformanceHttpError( 400, 'VALIDATION_FAILED', '@openmaic/storage: learner keys must be non-empty strings', ); } assertAddressableSegment(body.toLearnerKey); assertJsonRequestValue(body.fromLearnerKey, 'runtime learner merge fromLearnerKey'); assertJsonRequestValue(body.toLearnerKey, 'runtime learner merge toLearnerKey'); assertAddressableSegment(body.toLearnerKey); sendJson(res, 200, { moved: await store.mergeLearner(body.fromLearnerKey, body.toLearnerKey), }); return; } if ( parts[1] === 'stages' && parts.length === 5 && parts[3] === 'learners' && method === 'DELETE' ) { await store.deleteLearnerRuntime(parts[2]!, parts[4]!); sendNoContent(res); return; } if (parts[1] === 'stages' && parts.length === 3 && method === 'DELETE') { await store.deleteStageRuntime(parts[2]!); sendNoContent(res); return; } if (parts.length === 1 && method === 'DELETE') { await store.deleteAllRuntime(); sendNoContent(res); return; } sendJson(res, 404, { error: { code: 'ROUTE_NOT_FOUND', message: 'route not found' } }); } /** * Start a test-only HTTP adapter. Each `x-runtime-store-id` value selects a * fresh BrowserRuntimeStore so factories used by the shared contract remain * isolated without duplicating any persistence logic in this server. */ export async function startHttpConformanceServer( options: HttpConformanceServerOptions = {}, ): Promise { const stores = new Map(); const storeFor = (req: IncomingMessage): RuntimeStore => { const id = req.headers['x-runtime-store-id']; const namespace = typeof id === 'string' && id !== '' ? id : 'default'; let store = stores.get(namespace); if (!store) { store = new BrowserRuntimeStore({ indexedDB: new IDBFactory(), dbName: `http-runtime-${namespace}`, }); stores.set(namespace, store); } return store; }; const server = createServer((req, res) => { void route(req, res, storeFor(req)).catch((error: unknown) => { const mapped = errorResponse(error); sendJson(res, mapped.status, mapped.body); }); }); let baseUrl = 'http://runtime-conformance.invalid'; if (options.listen !== false) { await new Promise((resolve, reject) => { server.once('error', reject); server.listen(0, '127.0.0.1', resolve); }); const address = server.address(); if (address === null || typeof address === 'string') { throw new Error('HTTP conformance server did not bind a TCP port'); } baseUrl = `http://127.0.0.1:${address.port}`; } const injectedFetch: typeof globalThis.fetch = async (input, init) => { const request = new Request(input, init); const url = new URL(request.url); const requestBody = await request.text(); const fakeRequest = { method: request.method, url: `${url.pathname}${url.search}`, headers: Object.fromEntries(request.headers.entries()), async *[Symbol.asyncIterator]() { if (requestBody !== '') yield Buffer.from(requestBody); }, } as unknown as IncomingMessage; let status = 200; let responseHeaders: Record = {}; let responseBody: string | undefined; const fakeResponse = { writeHead(nextStatus: number, headers?: Record) { status = nextStatus; responseHeaders = headers ?? {}; return this; }, end(chunk?: string) { responseBody = chunk; return this; }, } as unknown as ServerResponse; try { await route(fakeRequest, fakeResponse, storeFor(fakeRequest)); } catch (error) { const mapped = errorResponse(error); status = mapped.status; responseHeaders = { 'content-type': 'application/json' }; responseBody = JSON.stringify(mapped.body); } return new Response(status === 204 ? null : responseBody, { status, headers: responseHeaders, }); }; return { baseUrl, fetch: injectedFetch, close: () => server.listening ? new Promise((resolve, reject) => { server.close((error) => (error ? reject(error) : resolve())); }) : Promise.resolve(), }; }