From e4ceacade8f7b21ab1cd7886a13508f2097e3d0b Mon Sep 17 00:00:00 2001 From: Harvmaster Date: Mon, 27 Jul 2026 04:24:16 +0000 Subject: [PATCH] Add transport-neutral routing and stream contracts --- src/app.ts | 15 +++- src/routes/types.ts | 40 +++++++++ src/services/route-stream.ts | 61 +++++++++++++ src/services/router.ts | 73 ++++++++++++++++ src/services/stream/base-stream.ts | 60 +++++++++++++ tests/helpers/test-connection.ts | 45 ++++++++++ tests/services/router.test.ts | 134 +++++++++++++++++++++++++++++ 7 files changed, 426 insertions(+), 2 deletions(-) create mode 100644 src/routes/types.ts create mode 100644 src/services/route-stream.ts create mode 100644 src/services/router.ts create mode 100644 src/services/stream/base-stream.ts create mode 100644 tests/helpers/test-connection.ts create mode 100644 tests/services/router.test.ts diff --git a/src/app.ts b/src/app.ts index 08df05e..00e5584 100644 --- a/src/app.ts +++ b/src/app.ts @@ -1,5 +1,6 @@ import { Config } from './services/config.ts'; import { Database, MigrationService } from './services/storage/index.ts'; +import { ApplicationRouter } from './services/router.ts'; import { Logger } from './utils/logger.ts'; /** Application composition root. */ @@ -17,12 +18,22 @@ export class App { const migrations = new MigrationService(database, debug); await migrations.migrateToLatest(); - return new App(database); + const routes = []; + + // Route loading is an explicit startup phase, not first-request work. + // ApplicationRouter.create validates every path and rejects duplicates + // before any client can connect. + const router = await ApplicationRouter.create(routes); + + return new App(database, router); } private stopPromise: Promise | undefined; - constructor(private readonly database: Database) {} + constructor( + private readonly database: Database, + private readonly router: ApplicationRouter, + ) {} async start(): Promise {} diff --git a/src/routes/types.ts b/src/routes/types.ts new file mode 100644 index 0000000..7e39906 --- /dev/null +++ b/src/routes/types.ts @@ -0,0 +1,40 @@ +import type { BaseStream } from '../services/stream/base-stream.js'; + +export type RouteSendOptions = { + /** Defaults to `response`; any other value sends an application event. */ + type?: string; + + /** Applies to normal responses. Defaults to 200, or 204 for undefined data. */ + statusCode?: number; +}; + +/** + * Request-scoped application view over an underlying transport connection. + * + * The body and request ID are immutable for one dispatch. Subscription state and + * connection lifetime are shared with other requests on the same connection. + */ +export interface RouteStream { + /** Connection shared by every request on the same transport session. */ + readonly connection: BaseStream; + + readonly body: unknown; + readonly streaming: boolean; + readonly bidirectional: boolean; + + send(data: unknown, options?: RouteSendOptions): Promise; +} + +export type RouteHandler = (stream: RouteStream) => void | Promise; + +/** An exact application route with no transport-specific metadata. */ +export type RouteDefinition = { + /** Exact route name. Parameter and wildcard syntax are not supported. */ + url: string; + handler: RouteHandler; +}; + +/** Supplies and validates routes during application startup. */ +export interface RouteModule { + getRoutes(): Promise>; +} diff --git a/src/services/route-stream.ts b/src/services/route-stream.ts new file mode 100644 index 0000000..9c41221 --- /dev/null +++ b/src/services/route-stream.ts @@ -0,0 +1,61 @@ +import type { RouteSendOptions, RouteStream } from '../routes/types.ts'; +import type { BaseStream } from './stream/base-stream.ts'; + +/** + * Binds one application request to a connection-level stream. + * + * Request correlation remains immutable even when multiple WebSocket handlers + * execute concurrently. The underlying connection remains available to + * connection-level services such as the broadcaster. + */ +export class ApplicationRouteStream implements RouteStream { + /** + * @param connection - Shared transport stream backing this request. + * @param body - Transport-decoded application payload for the route handler. + * @param requestId - Optional correlation ID for multiplexed transports. + */ + constructor( + readonly connection: BaseStream, + readonly body: unknown, + private readonly requestId?: string, + ) {} + + /** Whether the underlying connection can deliver server-pushed events. */ + get streaming(): boolean { + return this.connection.streaming; + } + + /** Whether the underlying connection supports later unsubscribe requests. */ + get bidirectional(): boolean { + return this.connection.bidirectional; + } + + /** + * Send a response or event envelope through the shared connection. + * + * @param data - Response body or event payload. + * @param options - Controls message type and HTTP status for responses. + */ + async send(data: unknown, options: RouteSendOptions = {}): Promise { + const type = options.type ?? 'response'; + + // Route responses carry an HTTP status and optional correlation ID. + if (type === 'response') { + await this.connection.send({ + ...(this.requestId === undefined ? {} : { id: this.requestId }), + type, + statusCode: options.statusCode ?? (data === undefined ? 204 : 200), + body: data ?? null, + }); + + return; + } + + // Non-response messages are application events with a neutral envelope. + await this.connection.send({ + ...(this.requestId === undefined ? {} : { id: this.requestId }), + type, + data, + }); + } +} diff --git a/src/services/router.ts b/src/services/router.ts new file mode 100644 index 0000000..f4ce543 --- /dev/null +++ b/src/services/router.ts @@ -0,0 +1,73 @@ +import type { RouteDefinition, RouteModule } from '../routes/types.ts'; +import { ApplicationError } from '../errors/index.ts'; +import { ApplicationRouteStream } from './route-stream.ts'; +import type { BaseStream } from './stream/base-stream.ts'; + +/** Canonical request produced by every transport adapter. */ +export type ApplicationRequest = { + /** Exact application route name. */ + path: string; + + /** Transport-decoded application payload. */ + body?: unknown; + + /** Optional correlation ID supplied by a multiplexed transport. */ + requestId?: string; +}; + +/** Exact-match application routing shared by every wire transport. */ +export class ApplicationRouter { + /** @param routes - Validated route table keyed by exact path. */ + private constructor(private readonly routes: ReadonlyMap) {} + + /** + * Load and validate the complete route table before accepting traffic. + * + * @param routeModules - Route modules whose handlers will be registered. + * @returns A ready-to-dispatch router instance. + */ + static async create(routeModules: RouteModule[]): Promise { + const routes = new Map(); + + // Collect routes from every module and reject duplicates at startup. + for (const routeModule of routeModules) { + for (const route of await routeModule.getRoutes()) { + ApplicationRouter.assertValidPath(route.url); + if (routes.has(route.url)) { + throw new Error(`Duplicate application route: ${route.url}`); + } + + routes.set(route.url, route); + } + } + + return new ApplicationRouter(routes); + } + + /** + * Execute one route using a request-scoped facade over the connection. + * + * @param request - Transport-normalized application request. + * @param connection - Shared connection stream for this transport session. + */ + async dispatch(request: ApplicationRequest, connection: BaseStream): Promise { + const route = this.routes.get(request.path); + if (!route) { + throw new ApplicationError(404, `No route found for ${request.path}`); + } + + const stream = new ApplicationRouteStream(connection, request.body, request.requestId); + await route.handler(stream); + } + + /** + * Enforce the deliberately small exact-path routing grammar at startup. + * + * @param path - Candidate route path to validate. + */ + private static assertValidPath(path: string): void { + if (!path.startsWith('/') || path.length === 1 || path.includes(':') || path.includes('?') || path.includes('#')) { + throw new Error(`Invalid application route "${path}": routes must be exact paths beginning with /`); + } + } +} diff --git a/src/services/stream/base-stream.ts b/src/services/stream/base-stream.ts new file mode 100644 index 0000000..b79f824 --- /dev/null +++ b/src/services/stream/base-stream.ts @@ -0,0 +1,60 @@ +/** A normal request/response result before transport encoding. */ +export type StreamResponse = { + /** Optional correlation ID for multiplexed transports. */ + id?: string; + + /** Discriminator marking this message as a route response. */ + type: 'response'; + + /** HTTP-equivalent status code for the response body. */ + statusCode: number; + + /** Serialized response payload. */ + body: unknown; +}; + +/** An application event before a transport applies its wire encoding. */ +export type StreamEvent = { + /** Optional event or correlation ID. */ + id?: string; + + /** Application-defined event type name. */ + type: string; + + /** Serialized event payload. */ + data: unknown; +}; + +/** Union of every message a connection stream can emit. */ +export type StreamMessage = StreamResponse | StreamEvent; + +/** + * Connection-level output channel shared by route requests and the broadcaster. + * + * A WebSocket connection can back many request-scoped RouteStream instances. + * SSE and normal HTTP each create one connection stream per request. + */ +export abstract class BaseStream { + /** Whether this connection can remain open for server events. */ + abstract readonly streaming: boolean; + + /** Whether later requests can modify this connection's subscriptions. */ + abstract readonly bidirectional: boolean; + + /** + * Send one response or event using the transport's wire encoding. + * + * @param message - Neutral response or event envelope to encode. + */ + abstract send(message: StreamMessage): Promise; + + /** Close the underlying connection and notify lifecycle observers. */ + abstract close(): void; + + /** + * Observe closure whether it occurs locally or at the remote peer. + * + * @param callback - Invoked once when the connection closes. + */ + abstract onClose(callback: () => void): void; +} diff --git a/tests/helpers/test-connection.ts b/tests/helpers/test-connection.ts new file mode 100644 index 0000000..98b37a4 --- /dev/null +++ b/tests/helpers/test-connection.ts @@ -0,0 +1,45 @@ +import { + BaseStream, + type StreamMessage, +} from "../../src/services/stream/base-stream.js"; + +/** Minimal observable connection used by application and broadcaster tests. */ +export class TestConnection extends BaseStream { + readonly messages: StreamMessage[] = []; + readonly closeCallbacks: Array<() => void> = []; + closed = false; + + constructor( + readonly streaming: boolean, + readonly bidirectional: boolean, + ) { + super(); + } + + async send(message: StreamMessage): Promise { + if (this.closed) { + throw new Error("connection is closed"); + } + + this.messages.push(message); + } + + close(): void { + if (this.closed) { + return; + } + + this.closed = true; + const callbacks = this.closeCallbacks.splice(0); + callbacks.forEach((callback) => callback()); + } + + onClose(callback: () => void): void { + if (this.closed) { + callback(); + return; + } + + this.closeCallbacks.push(callback); + } +} diff --git a/tests/services/router.test.ts b/tests/services/router.test.ts new file mode 100644 index 0000000..1a3ac29 --- /dev/null +++ b/tests/services/router.test.ts @@ -0,0 +1,134 @@ +import { describe, expect, it } from "vitest"; + +import type { RouteDefinition, RouteModule } from "../../src/routes/types.js"; +import { ApplicationError } from "../../src/errors/index.js"; +import { ApplicationRouter } from "../../src/services/router.js"; +import { TestConnection } from "../helpers/test-connection.js"; + +function moduleWith(routes: RouteDefinition[]): RouteModule { + return { + async getRoutes() { + return routes; + }, + }; +} + +describe("ApplicationRouter initialization", () => { + it("rejects duplicate exact paths during startup", async () => { + const route = { url: "/echo", handler: () => undefined }; + + await expect( + ApplicationRouter.create([moduleWith([route]), moduleWith([route])]), + ).rejects.toThrow("Duplicate application route: /echo"); + }); + + it.each(["echo", "/", "/items/:id", "/items?active=true", "/items#active"])( + "rejects the invalid route path %s", + async (url) => { + await expect( + ApplicationRouter.create([ + moduleWith([{ url, handler: () => undefined }]), + ]), + ).rejects.toThrow("Invalid application route"); + }, + ); +}); + +describe("ApplicationRouter dispatch", () => { + it("binds the connection, body, and request ID to one route stream", async () => { + const connection = new TestConnection(false, false); + const router = await ApplicationRouter.create([ + moduleWith([ + { + url: "/echo", + handler: async (stream) => { + expect(stream.connection).toBe(connection); + await stream.send(stream.body); + }, + }, + ]), + ]); + + await router.dispatch( + { path: "/echo", body: { value: 1 }, requestId: "request-1" }, + connection, + ); + + expect(connection.messages).toEqual([ + { + id: "request-1", + type: "response", + statusCode: 200, + body: { value: 1 }, + }, + ]); + + await expect( + router.dispatch({ path: "/echo/other", body: {} }, connection), + ).rejects.toMatchObject({ statusCode: 404 }); + }); + + it("preserves correlation when concurrent requests finish out of order", async () => { + const completions = new Map void>(); + const router = await ApplicationRouter.create([ + moduleWith([ + { + url: "/delayed", + handler: async (stream) => { + const key = (stream.body as { key: string }).key; + await new Promise((resolve) => completions.set(key, resolve)); + await stream.send({ key }); + }, + }, + ]), + ]); + const connection = new TestConnection(true, true); + + const first = router.dispatch( + { path: "/delayed", body: { key: "A" }, requestId: "A" }, + connection, + ); + const second = router.dispatch( + { path: "/delayed", body: { key: "B" }, requestId: "B" }, + connection, + ); + + completions.get("B")?.(); + await second; + completions.get("A")?.(); + await first; + + expect(connection.messages).toEqual([ + { + id: "B", + type: "response", + statusCode: 200, + body: { key: "B" }, + }, + { + id: "A", + type: "response", + statusCode: 200, + body: { key: "A" }, + }, + ]); + }); + + it("propagates route failures without infrastructure-specific cleanup", async () => { + const error = new Error("route failed"); + const router = await ApplicationRouter.create([ + moduleWith([ + { + url: "/failure", + handler: () => { + throw error; + }, + }, + ]), + ]); + + await expect( + router.dispatch({ path: "/failure" }, new TestConnection(false, false)), + ).rejects.toBe(error); + }); +});