Add transport-neutral routing and stream contracts

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<void> | undefined;
 
-    constructor(private readonly database: Database) {}
+    constructor(
+        private readonly database: Database,
+        private readonly router: ApplicationRouter,
+    ) {}
 
     async start(): Promise<void> {}
 

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<void>;
+}
+
+export type RouteHandler = (stream: RouteStream) => void | Promise<void>;
+
+/** 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<Array<RouteDefinition>>;
+}

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<void> {
+        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,
+        });
+    }
+}

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<string, RouteDefinition>) {}
+
+    /**
+     * 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<ApplicationRouter> {
+        const routes = new Map<string, RouteDefinition>();
+
+        // 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<void> {
+        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 /`);
+        }
+    }
+}

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<void>;
+
+    /** 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;
+}

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<void> {
+    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);
+  }
+}

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<string, () => void>();
+    const router = await ApplicationRouter.create([
+      moduleWith([
+        {
+          url: "/delayed",
+          handler: async (stream) => {
+            const key = (stream.body as { key: string }).key;
+            await new Promise<void>((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);
+  });
+});