Fix package-lock

package.json

@@ -63,7 +63,7 @@
         "@types/debug": "^4.1.13",
         "@types/node": "^25.9.3",
         "@types/ws": "^8.18.1",
-        "@vitest/coverage-v8": "^4.1.9",
+        "@vitest/coverage-v8": "^5.0.0",
         "@viz-kit/esbuild-analyzer": "^1.0.0",
         "@xo-cash/eslint-config": "1.0.2",
         "cspell": "^10.0.1",
@@ -74,6 +74,6 @@
         "typedoc-plugin-coverage": "^4.0.2",
         "typescript": "^6.0.3",
         "typescript-eslint": "^8.65.0",
-        "vitest": "^4.1.9"
+        "vitest": "^5.0.0"
     }
 }

source/auth/accounts.ts

@@ -1,49 +0,0 @@
-/**
- * This is a shim for payments that isnt really going to be too reflective of the real world payment system.
- * 
- * The reason im doing it in such an overly simplified way is because an actual payment system is COMPLEX.
- * 
- * Double entry accounting, transaction objects, idempotent requests, etc... They are a LOT to implement.
- * 
- * Im certain that we can implement this more complex system into this project, its just going to be a lot of work on the actual payment handling.
- * The route side can actually remain pretty simple because it can be wrapped into a function call.
- * 
- * This implementation will just offer a `getBalance(publicKey: string): Promise<number>` and `setBalance(publicKey: string, amount: number): Promise<void>`
- * 
- * Routes will just do a setBalance(pulicKey, await getBalance(publicKey) - amount) where amount is the size of the data being written.
- */
-export class Accounts {
-  private accounts: Map<string, number> = new Map();
-
-  async getBalance(publicKey: string): Promise<number> {
-    if (!this.accounts.has(publicKey)) {
-      this.accounts.set(publicKey, 1_000_000_000_000);
-    }
-    
-    return this.accounts.get(publicKey)!;
-  }
-
-  async setBalance(publicKey: string, amount: number): Promise<number> {
-    this.accounts.set(publicKey, amount);
-    
-    return amount;
-  }
-
-  async deductBalance(publicKey: string, amount: number): Promise<number> {
-    const balance = await this.getBalance(publicKey);
-
-    if (balance < amount) {
-      throw new Error('Insufficient balance');
-    }
-
-    const balanceAfterDeduction = balance - amount;
-
-    return await this.setBalance(publicKey, balanceAfterDeduction);
-  }
-
-  async hasSufficientBalance(publicKey: string, amount: number): Promise<boolean> {
-    return true;
-    const balance = await this.getBalance(publicKey);
-    return balance >= amount;
-  }
-}

source/auth/auth.ts

@@ -23,7 +23,7 @@ export class AuthSecp256k1 {
      */
     static async create(inputDeps: AuthSecp256k1Deps, options: AuthSecp256k1Options): Promise<AuthSecp256k1> {
         const deps = {
-            secp256k1: await instantiateSecp256k1(),
+            secp256k1: inputDeps.secp256k1 || await instantiateSecp256k1(),
             ...inputDeps,
         };
 

source/constants.ts

@@ -22,8 +22,3 @@ export const HTTP_STATUS_CODE_BAD_REQUEST = 400;
  * HTTP status code for "Not Acceptable" error.
  */
 export const HTTP_STATUS_CODE_NOT_ACCEPTED = 406;
-
-/**
- * Not implemented response status code.
- */
-export const HTTP_STATUS_CODE_NOT_IMPLEMENTED = 501;

source/index.ts

@@ -7,10 +7,7 @@ import { HttpTransportRouter } from './services/transport/http-transport.ts';
 import { WsTransportRouter } from './services/transport/ws-transport.ts';
 import { ServerHost } from './services/server-host.ts';
 import { Logger } from './utils/logger.ts';
-
 import { DataRoute } from './routes/resources.ts';
-import { AccountRoute } from './routes/account.ts';
-import { Accounts } from './auth/accounts.ts';
 
 /** Application composition root. */
 export class App {
@@ -32,12 +29,16 @@ export class App {
 
         // Domain services are shared across all transports and route modules.
         const broadcaster = new Broadcaster(debug);
-        const accounts = new Accounts();
         const routes = [
             // DataRoute owns resource read/write/subscribe logic and maps resource
             // ids to broadcaster topics. timestampWindowMs controls write replay protection.
-            new DataRoute(database, broadcaster, auth, accounts),
-            // new AccountRoute(database, auth),
+            new DataRoute({
+                auth: auth,
+                database: database,
+                broadcaster: broadcaster,
+            }, {
+                timestampWindowMs: config.auth.timestampWindowMs,
+            }),
         ];
 
         // Route loading is an explicit startup phase, not first-request work.
@@ -52,13 +53,7 @@ export class App {
         const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes);
         const host = new ServerHost(config, debug, [ http, ws ]);
 
-        // Create the app instance
-        const app = new App(host, database);
-        
-        // Start the unique request cleanup interval
-        app.startUniqueRequestCleanup(config.auth.uniqueRequestCleanupIntervalMs, config.auth.timestampWindowMs);
-        
-        return app;
+        return new App(host, database);
     }
 
     private stopPromise: Promise<void> | undefined;

source/routes/account.ts

@@ -1,105 +0,0 @@
-import { z } from "zod";
-
-import type { AuthSecp256k1 } from '../auth/auth.ts';
-import type { Database } from "../services/storage/database.ts";
-import type { RouteDefinition, RouteStream } from './types.ts';
-import type { Accounts } from '../auth/accounts.ts';
-
-import { HTTP_STATUS_CODE_NOT_IMPLEMENTED, HTTP_STATUS_CODE_SUCCESS } from "../constants.ts";
-import { UnauthorizedError } from '../errors/index.ts';
-
-const accountSchema = z.object({
-  publicKey: z.string(),
-  signature: z.string(),
-  timestamp: z.coerce.number(),
-});
-
-const setBalanceSchema = z.object({
-  amount: z.number(),
-});
-
-export class AccountRoute {
-  
-  constructor(private readonly database: Database, private readonly auth: AuthSecp256k1, private readonly accounts: Accounts) {}
-
-  async getRoutes(): Promise<Array<RouteDefinition>> {
-    return [
-      {
-        url: '/account',
-        handler: this.getAccount.bind(this),
-      },
-      {
-        url: '/account/balance',
-        handler: this.getAccount.bind(this),
-      },
-      {
-        url: '/account/setbalance',
-        handler: this.setBalance.bind(this),
-      },
-      // This one may not make sense. It could be a large overhead for something that most wont use?
-      // Maybe make this an optional endpoint or one that we don't support, but the client can try to hit or something?
-      {
-        url: '/account/ledger',
-        handler: this.getLedger.bind(this),
-      }
-    ];
-  }
-
-  async getAccount(stream: RouteStream): Promise<void> {
-    // Get the public key, signature and timestamp from the headers
-    const { publicKey, signature, timestamp } = accountSchema.parse(stream.headers);
-
-    // Create the canonical payload for the signature verification
-    const payload = `${stream.path}:${timestamp}`;
-
-    // Verify the signature
-    const verified = await this.auth.verifySignature(publicKey, signature, payload);
-
-    if (!verified) {
-      throw new UnauthorizedError('Invalid signature');
-    }
-
-    // Get the account's balance from the database
-    const balance = await this.accounts.getBalance(publicKey);
-
-    // Send the balance to the client
-    stream.send({
-      statusCode: HTTP_STATUS_CODE_SUCCESS,
-      body: balance,
-    });
-  }
-
-  async setBalance(stream: RouteStream): Promise<void> {
-    const { publicKey, signature, timestamp } = accountSchema.parse(stream.headers);
-    const { amount } = setBalanceSchema.parse(stream.body);
-
-    // Create the canonical payload for the signature verification
-    const payload = `${stream.path}:${timestamp}`;
-
-    // Verify the signature
-    const verified = await this.auth.verifySignature(publicKey, signature, payload);
-
-    if (!verified) {
-      throw new UnauthorizedError('Invalid signature');
-    }
-
-    // Get the account's balance from the database
-    const balance = await this.accounts.getBalance(publicKey);
-
-    // Set the balance to the database
-    await this.accounts.setBalance(publicKey, amount);
-
-    // Send the balance to the client
-    stream.send({
-      statusCode: HTTP_STATUS_CODE_SUCCESS,
-      body: balance,
-    });
-  }
-
-  async getLedger(stream: RouteStream): Promise<void> {
-    stream.send({
-      statusCode: HTTP_STATUS_CODE_NOT_IMPLEMENTED,
-      body: 'Not implemented',
-    });
-  }
-}

source/routes/resources.ts

@@ -1,16 +1,13 @@
-
-// NOTE: Replace this with libauth sha256
 import { toExtendedJson } from '@xo-cash/utils';
 import { z } from 'zod';
 
 import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts';
 
-import type{ AuthSecp256k1 } from '../auth/auth.ts';
 import type { BaseBroadcaster } from '../services/broadcaster.ts';
 import { ApplicationError, UnauthorizedError } from '../errors/index.ts';
 import type { Database } from '../services/storage/database.ts';
 import type { RouteDefinition, RouteModule, RouteStream } from './types.ts';
-import type { Accounts } from '../auth/accounts.ts';
+import { AuthSecp256k1 } from '../auth/auth.ts';
 
 /**
  * Schema to validate a single write resource.
@@ -49,6 +46,16 @@ const resourceIdsSchema = z.object({
 
 type WriteResource = z.infer<typeof writeResource>;
 
+export type DataRouteDeps = {
+    auth: AuthSecp256k1;
+    database: Database;
+    broadcaster: BaseBroadcaster;
+}
+
+export type DataRouteOptions = {
+    timestampWindowMs: number;
+}
+
 /**
  * Resource data domain routes.
  *
@@ -59,10 +66,8 @@ type WriteResource = z.infer<typeof writeResource>;
  */
 export class DataRoute implements RouteModule {
     constructor(
-        private readonly database: Database,
-        private readonly broadcaster: BaseBroadcaster,
-        private readonly auth: AuthSecp256k1,
-        private readonly accounts: Accounts,
+        private readonly deps: DataRouteDeps,
+        private readonly options: DataRouteOptions,
     ) {}
 
     /** Declare exact routes; each handler owns its stream behavior. */
@@ -89,7 +94,7 @@ export class DataRoute implements RouteModule {
 
     /** Reads the requested resources from the database, returning all instances for each resource */
     async getData(stream: RouteStream): Promise<void> {
-        const resourceIds = resourceIdsSchema.parse(stream.body).resourceId;
+        const resourceIds = this.getResourceIds(stream);
 
         // Remove duplicates.
         const uniqueIds = [ ...new Set(resourceIds) ];
@@ -102,9 +107,9 @@ export class DataRoute implements RouteModule {
         }
 
         // Read the data from the database.
-        const rows = await this.database.db
+        const rows = await this.deps.database.db
             .selectFrom('resource_data')
-            .select([ 'resource_id', 'public_key', 'blob', 'timestamp', 'signature' ])
+            .select([ 'resource_id', 'public_key', 'blob', 'timestamp' ])
             .where('resource_id', 'in', uniqueIds)
             .orderBy('timestamp', 'asc')
             .execute();
@@ -115,7 +120,6 @@ export class DataRoute implements RouteModule {
             publicKey: row.public_key,
             blob: new Uint8Array(row.blob),
             timestamp: row.timestamp,
-            signature: row.signature,
         }));
 
         await stream.send(formattedRows);
@@ -130,10 +134,6 @@ export class DataRoute implements RouteModule {
     async writeData(stream: RouteStream): Promise<void> {
         const { resources } = writeBody.parse(stream.body);
 
-        // temporarily disable the signature verification for testing
-        // const publicKey = stream.headers?.['x-public-key']
-        const publicKey = 'public-key';
-
         // Authenticate the whole batch before producing any storage side effects.
         // A single bad signature rejects the entire write — no partial commits.
         await Promise.all(resources.map((resource) => this.verifyWriteResource(resource)));
@@ -142,31 +142,21 @@ export class DataRoute implements RouteModule {
         const rows = resources.map((resource) => ({
             resourceId: resource.id,
             publicKey: resource.publicKey,
-            blob: Buffer.from(resource.value),
             signature: resource.signature,
+            blob: Buffer.from(resource.value),
         }));
 
-        // Get the total size of the bytes being written
-        const totalSize = rows.reduce((acc, row) => acc + row.blob.length, 0);
-
-        if (!await this.accounts.hasSufficientBalance(publicKey, totalSize)) {
-            throw new ApplicationError(203, 'Insufficient balance');
-        }
-
-        // Set the balance of the public key
-        await this.accounts.deductBalance(publicKey, totalSize);
-
         // Upsert every row atomically; conflicts update blob and timestamp only.
-        await this.database.db.transaction().execute(async (trx) => {
+        await this.deps.database.db.transaction().execute(async (trx) => {
             for (const { resourceId, publicKey, blob, signature } of rows) {
                 await trx
                     .insertInto('resource_data')
                     .values({
                         resource_id: resourceId,
                         public_key: publicKey,
+                        signature: signature,
                         blob,
                         timestamp,
-                        signature,
                     })
                     .onConflict((oc) =>
                         oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({
@@ -178,44 +168,42 @@ export class DataRoute implements RouteModule {
         });
 
         // Format the rows into the written resource responses.
-        const written = rows.map(({ resourceId, publicKey, blob, signature }) => ({
+        const written = rows.map(({ resourceId, publicKey, blob }) => ({
+            resourceId,
             instance: {
-                resourceId,
                 publicKey,
                 blob: new Uint8Array(blob),
                 timestamp,
-                signature,
             },
         }));
 
         // Notify subscribers on each changed resource. Topic names are scoped per
         // resource id so clients only receive events for resources they joined.
-        for (const { instance } of written) {
-            await this.broadcaster.publish(DataRoute.resourceTopic(instance.resourceId), {
+        for (const { resourceId, instance } of written) {
+            await this.deps.broadcaster.publish(DataRoute.resourceTopic(resourceId), {
                 type: 'instance-changed',
-                data: instance,
+                data: { resourceId, ...instance },
             });
         }
 
         // Return the persisted instances so the writer can confirm what was stored.
         await stream.send({
-            resources: written.map(({ instance }) => ({
-                id: instance.resourceId,
+            resources: written.map(({ resourceId, instance }) => ({
+                id: resourceId,
                 ...instance,
             })),
-            balance: await this.accounts.getBalance(publicKey),
         });
     }
 
     /** Subscribe this connection to future changes for the requested resources. */
     async subscribeData(stream: RouteStream): Promise<void> {
-        const resourceIds = resourceIdsSchema.parse(stream.body).resourceId;
+        const resourceIds = this.getResourceIds(stream);
 
         const topics = resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId));
 
         // Subscribe registers the topics synchronously, then keeps this route
         // active until those topics are removed or the connection closes.
-        await this.broadcaster.subscribe(stream, topics);
+        await this.deps.broadcaster.subscribe(stream, topics);
     }
 
     /**
@@ -225,7 +213,7 @@ export class DataRoute implements RouteModule {
      * unsubscribe implicitly by aborting the HTTP request.
      */
     async unsubscribeData(stream: RouteStream): Promise<void> {
-        const resourceIds = resourceIdsSchema.parse(stream.body).resourceId;
+        const resourceIds = this.getResourceIds(stream);
 
         // If the stream is not bidirectional, throw an error.
         if (!stream.bidirectional) {
@@ -233,13 +221,21 @@ export class DataRoute implements RouteModule {
         }
 
         // Unsubscribe from the resource topics.
-        await this.broadcaster.unsubscribe(
+        await this.deps.broadcaster.unsubscribe(
             stream,
             resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)),
         );
         await stream.send({});
     }
 
+    /** Extract resource id list from the decoded request body. */
+    private getResourceIds(stream: RouteStream): string[] {
+        const body =
+            typeof stream.body === 'object' && stream.body !== null && !Array.isArray(stream.body) ? (stream.body as Record<string, unknown>) : {};
+
+        return resourceIdsSchema.parse(body).resourceId;
+    }
+
     /**
      * Verify one write's timestamp freshness and secp256k1 signature.
      *
@@ -247,16 +243,35 @@ export class DataRoute implements RouteModule {
      * id, and value — not the raw HTTP/WebSocket envelope.
      */
     private async verifyWriteResource(resource: WriteResource): Promise<void> {
-        this.auth.assertTimestampFreshness(resource.timestamp);
+        this.assertFreshTimestamp(resource.timestamp);
 
-        // Compile the signature payload as `Timestamp:ID:Value`
-        const signaturePayload = `${resource.timestamp}:${resource.id}:${toExtendedJson(resource.value)}`;
-
-        if (!(await this.auth.verifySignature(resource.publicKey, resource.signature, signaturePayload))) {
+        if (!(await this.deps.auth.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) {
             throw new UnauthorizedError('Invalid resource signature');
         }
     }
 
+    /**
+     * Reject writes with stale timestamps to limit replay window.
+     *
+     * Both past and future timestamps outside the window are rejected.
+     */
+    private assertFreshTimestamp(timestamp: number): void {
+        const age = Math.abs(Date.now() - timestamp);
+        if (age > this.options.timestampWindowMs) {
+            throw new UnauthorizedError('Timestamp outside allowed window');
+        }
+    }
+
+    /**
+     * Build the transport-independent payload authenticated by each signature.
+     *
+     * canonicalBody ensures Uint8Array values hash consistently regardless of
+     * whether the client sent them over HTTP or WebSocket.
+     */
+    private static canonicalWritePayload(resource: WriteResource): string {
+        return `${resource.timestamp}:${resource.id}:${toExtendedJson(resource.value)}`;
+    }
+
     /** Broadcaster topic for a single resource's instance-changed events. */
     private static resourceTopic(resourceId: string): string {
         return `resource:${resourceId}`;

source/services/transport/base-transport.ts

@@ -0,0 +1,59 @@
+import { Hono } from "hono";
+import { RequestHeaders } from "../../routes/types";
+import { ApplicationError } from "../../errors";
+import { HTTP_STATUS_CODE_BAD_REQUEST } from "../../constants";
+
+/** RFC 9110 field-name token grammar. */
+const HEADER_NAME_PATTERN = /^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$/;
+
+/** Hono variables populated by transport-boundary middleware. */
+export type AppEnv = {
+  Variables: {
+
+      /** Decoded Extended JSON request body, when present. */
+      parsedBody?: unknown;
+
+      /** Raw JSON text preserved for signature verification. */
+      rawJsonBody?: string;
+  };
+};
+
+export abstract class BaseTransport {
+    /** Attach wire endpoints and middleware to the shared Hono application. */
+    abstract register(app: Hono<AppEnv>): void
+
+    /** Close transport-owned long-lived connections during server shutdown. */
+    abstract stop(): Promise<void>
+    
+    /**
+     * Validate and normalize request headers at a transport boundary.
+     *
+     * Lowercase names give HTTP and WebSocket routes identical lookup semantics.
+     * Case-insensitive duplicates are rejected instead of selecting an ambiguous
+     * authentication value. Header values may not contain line breaks.
+     */
+    public static normalizeRequestHeaders(headers: Readonly<Record<string, string>>): RequestHeaders {
+        const normalizedEntries: Array<[string, string]> = [];
+        const names = new Set<string>();
+    
+        for (const [ name, value ] of Object.entries(headers)) {
+            if (!HEADER_NAME_PATTERN.test(name)) {
+                throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Invalid request header name');
+            }
+    
+            if (value.includes('\r') || value.includes('\n')) {
+                throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Invalid request header value');
+            }
+    
+            const normalizedName = name.toLowerCase();
+            if (names.has(normalizedName)) {
+                throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Duplicate request header name');
+            }
+    
+            names.add(normalizedName);
+            normalizedEntries.push([ normalizedName, value ]);
+        }
+    
+        return Object.freeze(Object.fromEntries(normalizedEntries));
+    };
+}

source/services/transport/http-transport.ts

@@ -5,13 +5,12 @@ import { toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
 import type { Logger } from '../../utils/logger.ts';
 import type { ApplicationRequest, ApplicationRouter } from '../router.ts';
 import type { StreamResponse } from '../stream/base-stream.ts';
-import type { AppEnv, TransportRouter } from './transport-router.ts';
 
 import { ApplicationError, normalizePublicError } from '../../errors/index.ts';
 import { HTTP_STATUS_CODE_BAD_REQUEST, HTTP_STATUS_CODE_NO_CONTENT } from '../../constants.ts';
 import { HonoSSEStream } from '../stream/hono-sse-stream.ts';
 import { HttpRequestStream } from '../stream/http-request-stream.ts';
-import { normalizeRequestHeaders } from './request-headers.ts';
+import { type AppEnv, BaseTransport } from './base-transport.ts';
 
 /** Hono context key where decoded Extended JSON bodies are stored. */
 const PARSED_BODY_KEY = 'parsedBody';
@@ -22,7 +21,7 @@ const PARSED_BODY_KEY = 'parsedBody';
  * Normal HTTP and SSE both enter the same application router with different
  * connection-stream capabilities.
  */
-export class HttpTransportRouter implements TransportRouter {
+export class HttpTransportRouter extends BaseTransport {
     private readonly debug: Logger;
 
     /** SSE connections retained until their final subscription or peer closes. */
@@ -36,6 +35,8 @@ export class HttpTransportRouter implements TransportRouter {
         private readonly router: ApplicationRouter,
         debug: Logger,
     ) {
+        super();
+
         this.debug = debug.extend('http-transport');
     }
 
@@ -111,8 +112,8 @@ export class HttpTransportRouter implements TransportRouter {
 
         return {
             path: context.req.path,
-            headers: normalizeRequestHeaders(context.req.header()),
             ...(body === undefined ? {} : { body }),
+            headers: context.req.header(),
         };
     }
 

source/services/transport/ws-transport.ts

@@ -8,12 +8,11 @@ import { z } from 'zod';
 
 import type { Logger } from '../../utils/logger.ts';
 import type { ApplicationRequest, ApplicationRouter } from '../router.ts';
-import type { AppEnv, UpgradeTransportRouter } from './transport-router.ts';
 
 import { ApplicationError, normalizePublicError } from '../../errors/index.ts';
 import { HTTP_STATUS_CODE_BAD_REQUEST } from '../../constants.ts';
 import { WSStream } from '../stream/ws-stream.ts';
-import { normalizeRequestHeaders } from './request-headers.ts';
+import { type AppEnv, BaseTransport } from './base-transport.ts';
 
 /** Default WebSocket upgrade path for application messages. */
 const WS_ROUTE = '/ws';
@@ -25,7 +24,6 @@ const wsRequestSchema = z
 .optional(),
         path: z.string().min(1),
         body: z.unknown().optional(),
-        headers: z.record(z.string(), z.string()).optional(),
     })
     .strict();
 
@@ -35,7 +33,7 @@ const wsRequestSchema = z
  * One WSStream is shared by every message on a socket, allowing subscribe and
  * unsubscribe requests to operate on the same broadcaster registration.
  */
-export class WsTransportRouter implements UpgradeTransportRouter {
+export class WsTransportRouter extends BaseTransport {
     /**
      * Native Node WebSocket server used by Hono's Node adapter.
      *
@@ -44,9 +42,6 @@ export class WsTransportRouter implements UpgradeTransportRouter {
      */
     private readonly wsServer: WebSocketServer;
 
-    /** Upgrade server wired into the Node HTTP listener by ServerHost. */
-    readonly websocketServer: WebSocketServerLike;
-
     private readonly debug: Logger;
 
     /**
@@ -61,6 +56,8 @@ export class WsTransportRouter implements UpgradeTransportRouter {
         private readonly maxRequestBodyBytes: number,
         private readonly url: string = WS_ROUTE,
     ) {
+        super();
+
         this.debug = debug.extend('ws-transport');
 
         // Hono's Node WebSocket helper delegates frame handling to `ws`; unlike
@@ -70,7 +67,6 @@ export class WsTransportRouter implements UpgradeTransportRouter {
             noServer: true,
             maxPayload: this.maxRequestBodyBytes,
         });
-        this.websocketServer = this.wsServer as unknown as WebSocketServerLike;
     }
 
     /**
@@ -210,7 +206,6 @@ export class WsTransportRouter implements UpgradeTransportRouter {
             path: envelope.path,
             ...(envelope.id === undefined ? {} : { requestId: envelope.id }),
             ...(envelope.body === undefined ? {} : { body: envelope.body }),
-            ...(envelope.headers === undefined ? {} : { headers: normalizeRequestHeaders(envelope.headers) }),
         };
     }
 

test/helpers/controlled-request.ts

@@ -0,0 +1,68 @@
+import { ApplicationRouter } from "../../source/services/router";
+import { BaseStream } from "../../source/services/stream/base-stream";
+
+export type ControlledRequestDeps = {
+  router: ApplicationRouter;
+  connection: BaseStream;
+}
+
+export type ControlledRequestOptions = {
+  path: string;
+  requestId: string;
+  body?: unknown;
+  headers?: Record<string, string>;
+  includeAuthHeaders?: boolean;
+  timeout?: number;
+}
+
+export type ControlledRequestParams = ControlledRequestOptions & ControlledRequestDeps;
+
+export type ControlledRequest = {
+  request: Promise<void>;
+  started: Promise<void>;
+  release: () => void;
+}
+
+export const createControlledRequest = ({ router, connection, ...requestOptions }: ControlledRequestParams): ControlledRequest => {
+  // Create promises with resolvers to signal the request has started and released
+  const { promise: started, resolve: signalStarted } = Promise.withResolvers<void>();
+  const { promise: released, resolve: release } = Promise.withResolvers<void>();
+
+  // The request options (sets the includeAuthHeaders to true by default)
+  const options = {
+    includeAuthHeaders: true,
+    ...requestOptions
+  }
+
+  // The test auth headers
+  const authHeaders = {
+    'x-public-key':  'public-key',
+    'x-signature': 'signature',
+    'x-timestamp': '1000',
+  }
+
+  // If the request should include the test auth headers (default is true)
+  if (options.includeAuthHeaders) {
+    options.headers = {
+      ...options.headers,
+      ...authHeaders,
+    };
+  }
+
+  // Add the signalStarted and released promises to the request body
+  options.body = {
+    ...(options.body ?? {}),
+    signalStarted,
+    released,
+  }
+
+  // Dispatch the request
+  const request = router.dispatch(options, connection);
+  
+  // Return the request, started, and release promises
+  return {
+    request,
+    started,
+    release,
+  };
+};

test/helpers/misc.ts

@@ -0,0 +1,29 @@
+import { vi } from "vitest";
+import type { AuthSecp256k1 } from "../../source/auth/auth";
+import type { RouteDefinition, RouteModule } from "../../source/routes/types";
+
+/**
+ * Convert an array of route definitions to a route module by returning an object with a getRoutes method that returns the routes.
+ * @param routes - The route definitions to convert.
+ * @returns A route module.
+ */
+export const toRoutes = (routes: RouteDefinition[]): RouteModule => {
+  return {
+    getRoutes: async () => routes,
+  };
+};
+
+
+export const mockAuth = {
+  verifySignature: vi.fn().mockResolvedValue(true),
+  verifyUniqueRequest: vi.fn().mockResolvedValue(true),
+  assertTimestampFreshness: vi.fn().mockResolvedValue(true),
+} as unknown as AuthSecp256k1;
+
+/**
+ * Create a mock of the AuthSecp256k1 service
+ * @returns A mock of the AuthSecp256k1 service
+ */
+export const createMockAuth = (): AuthSecp256k1 => {
+  return mockAuth;
+};

test/routes/resources.test.ts

@@ -7,8 +7,8 @@ import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
 import type { Database } from '../../source/services/storage/database.ts';
 import { TestConnection } from '../helpers/test-connection.ts';
 import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../../source/constants.ts';
-import type { AuthSecp256k1 } from '../../source/auth/auth.ts';
-import { Accounts } from '../../source/auth/accounts.ts';
+import { createMockAuth } from '../helpers/misc.ts';
+import { AuthSecp256k1 } from '../../source/auth/auth.ts';
 
 const createBroadcasterStub = (): BaseBroadcaster => {
     return {
@@ -19,31 +19,6 @@ const createBroadcasterStub = (): BaseBroadcaster => {
     } as unknown as BaseBroadcaster;
 };
 
-const createAuthStub = (): AuthSecp256k1 => {
-    return {
-        verifySignature: vi.fn().mockImplementation(() => {
-            return true;
-        }),
-    } as unknown as AuthSecp256k1;
-};
-
-const createAccountsStub = (): Accounts => {
-    return {
-        getBalance: vi.fn().mockImplementation(() => {
-            return 0;
-        }),
-        setBalance: vi.fn().mockImplementation(() => {
-            return;
-        }),
-        deductBalance: vi.fn().mockImplementation(() => {
-            return;
-        }),
-        hasSufficientBalance: vi.fn().mockImplementation(() => {
-            return true;
-        }),
-    } as unknown as Accounts;
-};
-
 describe('DataRoute subscriptions', (): void => {
     it('subscribes to future resource changes until removal', async (): Promise<void> => {
         let resolveRemoved: () => void = () => undefined;
@@ -58,14 +33,16 @@ describe('DataRoute subscriptions', (): void => {
         const broadcaster = createBroadcasterStub();
         vi.mocked(broadcaster.subscribe).mockReturnValue(removed);
         const connection = new TestConnection(true, false);
-        const stream = new ApplicationRouteStream(
-            connection,
-            {
-                resourceId: [ 'a', 'b' ],
-            },
-            '/data/subscribe',
-        );
-        const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0);
+        const stream = new ApplicationRouteStream(connection, {
+            resourceId: [ 'a', 'b' ],
+        }, 'subscribe-1');
+        const route = new DataRoute({
+            auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }),
+            database: storage,
+            broadcaster: broadcaster,
+        }, {
+            timestampWindowMs: 0,
+        });
 
         const execution = route.subscribeData(stream);
 
@@ -86,8 +63,14 @@ describe('DataRoute subscriptions', (): void => {
         } as unknown as Database;
         const broadcaster = createBroadcasterStub();
         const connection = new TestConnection(true, true);
-        const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ] }, '/data/unsubscribe', 'unsubscribe-1');
-        const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0);
+        const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ] }, '/unsubscribe', 'unsubscribe-1');
+        const route = new DataRoute({
+            auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }),
+            database: storage,
+            broadcaster: broadcaster,
+        }, {
+            timestampWindowMs: 0,
+        });
 
         await route.unsubscribeData(stream);
 
@@ -109,14 +92,17 @@ describe('DataRoute subscriptions', (): void => {
             },
         } as unknown as Database;
         const broadcaster = createBroadcasterStub();
-        const stream = new ApplicationRouteStream(
-            new TestConnection(true, false),
-            {
-                resourceId: [ 'a' ],
-            },
-            '/data/unsubscribe',
-        );
-        const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0);
+        const connection = new TestConnection(true, false);
+        const stream = new ApplicationRouteStream(connection, {
+            resourceId: [ 'a' ],
+        }, '/unsubscribe', 'unsubscribe-1');
+        const route = new DataRoute({
+            auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }),
+            database: storage,
+            broadcaster: broadcaster,
+        }, {
+            timestampWindowMs: 0,
+        });
 
         await expect(route.unsubscribeData(stream)).rejects.toMatchObject({ statusCode: HTTP_STATUS_CODE_NOT_ACCEPTED });
         expect(broadcaster.unsubscribe).not.toHaveBeenCalled();
@@ -132,7 +118,13 @@ describe('DataRoute resource write auth', (): void => {
         } as unknown as Database;
 
         const broadcaster = createBroadcasterStub();
-        const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0);
+        const route = new DataRoute({
+            auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }),
+            database: storage,
+            broadcaster: broadcaster,
+        }, {
+            timestampWindowMs: 0,
+        });
 
         await expect(route.writeData({
                 connection: new TestConnection(true, true),
@@ -150,11 +142,6 @@ describe('DataRoute resource write auth', (): void => {
                         },
                     ],
                 },
-                headers: {
-                    publicKey: 'public-key',
-                    signature: 'signature',
-                    timestamp: Date.now(),
-                },
             } as unknown as ApplicationRouteStream)).rejects.toBeInstanceOf(UnauthorizedError);
 
         expect(storage.db.transaction).not.toHaveBeenCalled();

test/services/broadcaster.test.ts

@@ -10,7 +10,7 @@ const createBroadcaster = (): Broadcaster => {
 };
 
 const routeStream = (connection: TestConnection): ApplicationRouteStream => {
-    return new ApplicationRouteStream(connection, undefined, '/test');
+    return new ApplicationRouteStream(connection, undefined, '/items', 'items-1');
 };
 
 const expectPending = async (promise: Promise<void>): Promise<void> => {

test/services/router.test.ts

@@ -1,52 +1,27 @@
-import { describe, expect, it, vi } from 'vitest';
+import { describe, expect, it } from 'vitest';
 
-import type { RouteDefinition, RouteModule } from '../../source/routes/types.ts';
+// Source
 import { ApplicationRouter } from '../../source/services/router.ts';
+
+// Helpers
+import { createMockAuth, toRoutes } from '../helpers/misc.ts';
 import { TestConnection } from '../helpers/test-connection.ts';
-
-import type { AuthSecp256k1 } from '../../source/auth/auth.ts';
-
-/**
- * A controlled request is a request that is controlled by the test.
- * It is used to control the request flow and ensure that the request is completed in the correct order.
- */
-type ControlledRequest = {
-    request: Promise<void>;
-    started: Promise<void>;
-    release: () => void;
-};
+import { createControlledRequest } from '../helpers/controlled-request.ts';
 
 /**
  * A mock of the AuthSecp256k1 service
  */
-const auth = {
-    verifySignature: vi.fn().mockResolvedValue(true),
-    verifyUniqueRequest: vi.fn().mockResolvedValue(true),
-    assertTimestampFreshness: vi.fn().mockResolvedValue(true),
-} as unknown as AuthSecp256k1;
-
-/**
- * A helper function to create a route module with the given routes
- * @param routes - The routes to create the module with
- * @returns The created route module
- */
-const moduleWith = (routes: RouteDefinition[]): RouteModule => {
-    return {
-        async getRoutes(): Promise<RouteDefinition[]> {
-            return routes;
-        },
-    };
-};
+const auth = createMockAuth();
 
 describe('ApplicationRouter initialization', (): void => {
     it('rejects duplicate exact paths during startup', async (): Promise<void> => {
         const route = { url: '/echo', handler: (): void => undefined };
 
-        await expect(ApplicationRouter.create({ auth }, [ moduleWith([ route ]), moduleWith([ route ]) ])).rejects.toThrow('Duplicate application route: /echo');
+        await expect(ApplicationRouter.create({ auth }, [ toRoutes([ route ]), toRoutes([ 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): Promise<void> => {
-        await expect(ApplicationRouter.create({ auth }, [ moduleWith([{ url, handler: (): void => undefined }]) ])).rejects.toThrow('Invalid application route');
+        await expect(ApplicationRouter.create({ auth }, [ toRoutes([{ url, handler: (): void => undefined }]) ])).rejects.toThrow('Invalid application route');
     });
 });
 
@@ -54,7 +29,7 @@ describe('ApplicationRouter dispatch', (): void => {
     it('binds the connection, body, and request ID to one route stream', async (): Promise<void> => {
         const connection = new TestConnection(false, false);
         const router = await ApplicationRouter.create({ auth }, [
-            moduleWith([
+            toRoutes([
                 {
                     url: '/echo',
                     handler: async (stream): Promise<void> => {
@@ -94,7 +69,7 @@ describe('ApplicationRouter dispatch', (): void => {
 
     it('preserves correlation when concurrent requests finish out of order', async (): Promise<void> => {
         const router = await ApplicationRouter.create({ auth }, [
-            moduleWith([
+            toRoutes([
                 {
                     url: '/delayed',
                     handler: async (stream): Promise<void> => {
@@ -115,38 +90,8 @@ describe('ApplicationRouter dispatch', (): void => {
 
         const connection = new TestConnection(false, false);
 
-        const createControlledRequest = (key: string): ControlledRequest => {
-            const { promise: started, resolve: signalStarted } = Promise.withResolvers<void>();
-
-            const { promise: released, resolve: release } = Promise.withResolvers<void>();
-
-            const request = router.dispatch(
-                {
-                    path: '/delayed',
-                    body: {
-                        key,
-                        signalStarted,
-                        released,
-                    },
-                    requestId: key,
-                    headers: {
-                        'x-public-key': 'public-key',
-                        'x-signature': 'signature',
-                        'x-timestamp': '1000',
-                    },
-                },
-                connection,
-            );
-
-            return {
-                request,
-                started,
-                release,
-            };
-        };
-
-        const first = createControlledRequest('A');
-        const second = createControlledRequest('B');
+        const first = createControlledRequest({ router, connection, path: '/delayed', requestId: 'A', body: { key: 'A' } });
+        const second = createControlledRequest({ router, connection, path: '/delayed', requestId: 'B', body: { key: 'B' } });
 
         expect(connection.messages).toEqual([]);
 
@@ -172,21 +117,21 @@ describe('ApplicationRouter dispatch', (): void => {
                 id: 'B',
                 type: 'response',
                 statusCode: 200,
-                body: { key: 'B', token: 'token-b' },
+                body: { key: 'B' },
             },
             {
                 id: 'A',
                 type: 'response',
                 statusCode: 200,
-                body: { key: 'A', token: 'token-a' },
+                body: { key: 'A' },
             },
         ]);
-    });
+    }, 1000);
 
     it('propagates route failures without infrastructure-specific cleanup', async (): Promise<void> => {
         const error = new Error('route failed');
         const router = await ApplicationRouter.create({ auth }, [
-            moduleWith([
+            toRoutes([
                 {
                     url: '/failure',
                     handler: (): void => {

test/services/transports/http-transport.test.ts

@@ -10,6 +10,22 @@ import type { AppEnv } from '../../../source/services/transport/transport-router
 import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
 import { Logger } from '../../../source/utils/logger.ts';
 import { ServerHost } from '../../../source/services/server-host.ts';
+import { AuthSecp256k1 } from '../../../source/auth/auth.ts';
+
+/**
+ * A mock of the AuthSecp256k1 service
+ */
+const auth = {
+    verifySignature: vi.fn().mockResolvedValue(true),
+    verifyUniqueRequest: vi.fn().mockResolvedValue(true),
+    assertTimestampFreshness: vi.fn().mockResolvedValue(true),
+} as unknown as AuthSecp256k1;
+
+const mockAuthHeaders = {
+    'x-public-key': 'public-key',
+    'x-signature': 'signature',
+    'x-timestamp': '1000',
+};
 
 const createApp = async (
     routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
@@ -18,7 +34,9 @@ const createApp = async (
     const debug = new Logger('http-transport-test');
     const broadcaster = new Broadcaster(debug);
     const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes;
-    const router = await ApplicationRouter.create([
+    const router = await ApplicationRouter.create({
+        auth,
+    }, [
         {
             async getRoutes(): Promise<RouteDefinition[]> {
                 return resolvedRoutes;
@@ -40,18 +58,20 @@ describe('HttpTransportRouter', (): void => {
     it('runs normal HTTP through a non-streaming route stream', async (): Promise<void> => {
         const app = await createApp([
             {
-                url: '/echo',
+                url: '/echowtf',
                 handler: async (stream): Promise<void> => stream.send(stream.body),
             },
         ]);
         const value = new Uint8Array([ 1, 2, 3 ]);
 
-        const response = await app.request('/echo', {
+        const request = new Request('http://localhost/echowtf', {
             method: 'POST',
-            headers: { 'content-type': 'application/json' },
+            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
             body: toExtendedJson({ value }),
         });
 
+        const response = await app.request(request);
+
         expect(response.status).toBe(200);
         expect(fromExtendedJson(await response.text())).toEqual({ value });
     });
@@ -64,33 +84,16 @@ describe('HttpTransportRouter', (): void => {
             },
         ]);
 
-        const response = await app.request('/nothing', { method: 'POST' });
+        const response = await app.request('/nothing', { method: 'POST', headers: { ...mockAuthHeaders } });
 
         expect(response.status).toBe(204);
         expect(await response.text()).toBe('');
     });
 
-    it('passes normalized HTTP request headers to the route stream', async (): Promise<void> => {
-        const app = await createApp([
-            {
-                url: '/headers',
-                handler: async (stream): Promise<void> => stream.send({ path: stream.path, token: stream.headers['x-request-token'] }),
-            },
-        ]);
-
-        const response = await app.request('/headers', {
-            method: 'POST',
-            headers: { 'X-Request-Token': 'http-token' },
-        });
-
-        expect(response.status).toBe(200);
-        expect(await response.json()).toEqual({ path: '/headers', token: 'http-token' });
-    });
-
     it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
         const app = await createApp([]);
 
-        const missing = await app.request('/missing', { method: 'POST' });
+        const missing = await app.request('/missing', { method: 'POST', headers: { ...mockAuthHeaders } });
         expect(missing.status).toBe(404);
         expect(await missing.json()).toEqual({
             statusCode: 404,
@@ -99,7 +102,7 @@ describe('HttpTransportRouter', (): void => {
 
         const invalid = await app.request('/missing', {
             method: 'POST',
-            headers: { 'content-type': 'application/json' },
+            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
             body: '{',
         });
         expect(invalid.status).toBe(400);
@@ -119,7 +122,7 @@ describe('HttpTransportRouter', (): void => {
             },
         ]);
 
-        const response = await app.request('/items/subscribe', { method: 'POST' });
+        const response = await app.request('/items/subscribe', { method: 'POST', headers: { ...mockAuthHeaders } });
 
         expect(response.status).toBe(406);
         expect(await response.json()).toMatchObject({ statusCode: 406 });
@@ -137,7 +140,7 @@ describe('HttpTransportRouter', (): void => {
 
         const response = await app.request('/items/subscribe', {
             method: 'POST',
-            headers: { accept: 'text/event-stream' },
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
         });
         const events = await response.text();
 
@@ -157,7 +160,7 @@ describe('HttpTransportRouter', (): void => {
 
         const response = await app.request('/echo', {
             method: 'POST',
-            headers: { accept: 'text/event-stream' },
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
         });
         const events = await response.text();
 
@@ -166,27 +169,6 @@ describe('HttpTransportRouter', (): void => {
         expect(events).toContain('data: {"ok":true}');
     });
 
-    it('passes normalized HTTP request headers to an SSE route stream', async (): Promise<void> => {
-        const app = await createApp([
-            {
-                url: '/headers',
-                handler: (stream): Promise<void> => stream.send({ token: stream.headers['x-request-token'] }),
-            },
-        ]);
-
-        const response = await app.request('/headers', {
-            method: 'POST',
-            headers: {
-                accept: 'text/event-stream',
-                'X-Request-Token': 'sse-token',
-            },
-        });
-        const events = await response.text();
-
-        expect(response.status).toBe(200);
-        expect(events).toContain('data: {"token":"sse-token"}');
-    });
-
     it("keeps SSE open until the route's subscription promise resolves", async (): Promise<void> => {
         let removeSubscription: () => Promise<void> = async () => undefined;
         let markSubscribed: () => void = () => undefined;
@@ -209,7 +191,7 @@ describe('HttpTransportRouter', (): void => {
 
         const response = await app.request('/items/subscribe', {
             method: 'POST',
-            headers: { accept: 'text/event-stream' },
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
         });
         const body = response.text();
         const completed = vi.fn();
@@ -241,7 +223,7 @@ describe('HttpTransportRouter', (): void => {
 
         const response = await app.request('/items/unsubscribe', {
             method: 'POST',
-            headers: { accept: 'text/event-stream' },
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
         });
         const events = await response.text();
 
@@ -262,7 +244,7 @@ describe('HttpTransportRouter', (): void => {
 
         const response = await app.request('/echo', {
             method: 'POST',
-            headers: { 'content-type': 'application/json' },
+            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
             body: JSON.stringify({ value: 'x'.repeat(64) }),
         });
 

test/services/transports/ws-transport.test.ts

@@ -1,11 +1,21 @@
-import { describe, expect, it } from 'vitest';
+import { describe, expect, it, vi } from 'vitest';
 import { z } from 'zod';
-import type { WebSocketServer } from 'ws';
 
 import { ApplicationRouter } from '../../../source/services/router.ts';
 import { WsTransportRouter } from '../../../source/services/transport/ws-transport.ts';
 import { Logger } from '../../../source/utils/logger.ts';
 import { toExtendedJson } from '@xo-cash/utils';
+import { AuthSecp256k1 } from '../../../source/auth/auth.ts';
+import { toRoutes } from '../../helpers/misc.ts';
+
+/**
+ * A mock of the AuthSecp256k1 service
+ */
+const auth = {
+    verifySignature: vi.fn().mockResolvedValue(true),
+    verifyUniqueRequest: vi.fn().mockResolvedValue(true),
+    assertTimestampFreshness: vi.fn().mockResolvedValue(true),
+} as unknown as AuthSecp256k1;
 
 describe('WebSocket request decoding', (): void => {
     it('decodes the minimal route-agnostic envelope and Extended JSON body', async (): Promise<void> => {
@@ -13,36 +23,19 @@ describe('WebSocket request decoding', (): void => {
                     id: 'request-1',
                     path: '/data/write',
                     body: { value: new Uint8Array([ 1, 2, 3 ]) },
-                    headers: { 'X-Request-Token': 'ws-token' },
                 }))).resolves.toEqual({
             requestId: 'request-1',
             path: '/data/write',
             body: { value: new Uint8Array([ 1, 2, 3 ]) },
-            headers: { 'x-request-token': 'ws-token' },
         });
     });
 
-    it('allows the optional headers object to be omitted', async (): Promise<void> => {
-        await expect(WsTransportRouter.decodeWebSocketRequest('{"path":"/data/get"}')).resolves.toEqual({ path: '/data/get' });
-    });
-
-    it.each([
-        '{}',
-        '{"path":42}',
-        '{"path":"/data/get","id":1}',
-        '{"path":"/data/get","method":"POST"}',
-        '{"path":"/data/get","headers":{"x-request-token":1}}',
-    ])('rejects an invalid envelope: %s', async (payload) => {
-        await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
-    });
-
-    it.each([
-        '{"path":"/data/get","headers":{"bad header":"value"}}',
-        '{"path":"/data/get","headers":{"x-request-token":"first","X-Request-Token":"second"}}',
-        '{"path":"/data/get","headers":{"x-request-token":"first\\r\\nsecond"}}',
-    ])('rejects malformed request headers: %s', async (payload): Promise<void> => {
-        await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toMatchObject({ statusCode: 400 });
-    });
+    it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])(
+        'rejects an invalid envelope: %s',
+        async (payload) => {
+            await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
+        },
+    );
 
     it('rejects malformed JSON', async (): Promise<void> => {
         await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({
@@ -55,11 +48,21 @@ describe('WebSocket request decoding', (): void => {
 describe('WsTransportRouter payload limits', (): void => {
     it("configures Hono's ws server with the requested maxPayload", async () => {
         const debug = new Logger('ws-transport-test');
-        const router = await ApplicationRouter.create([]);
+        const router = await ApplicationRouter.create({
+            auth,
+        }, [
+            toRoutes([
+                {
+                    url: '/data/write',
+                    handler: async (stream): Promise<void> => {
+                        await stream.send({});
+                    },
+                },
+            ]),
+        ]);
         const transport = new WsTransportRouter(router, debug, 1024);
-        const wsServer = transport.websocketServer as unknown as WebSocketServer;
 
-        expect(wsServer.options.maxPayload).toBe(1024);
+        expect(transport['wsServer'].options.maxPayload).toBe(1024);
         await transport.stop();
     });
 });

test/subscription-flow.test.ts

@@ -1,35 +1,38 @@
 import { describe, expect, it, vi } from 'vitest';
 
-import type { RouteDefinition, RouteModule } from '../source/routes/types.ts';
 import { ApplicationRouter } from '../source/services/router.ts';
 import { Broadcaster } from '../source/services/broadcaster.ts';
 import { Logger } from '../source/utils/logger.ts';
 import { TestConnection } from './helpers/test-connection.ts';
 
-const moduleWith = (routes: RouteDefinition[]): RouteModule => {
-    return {
-        async getRoutes(): Promise<RouteDefinition[]> {
-            return routes;
-        },
-    };
-};
+import { createControlledRequest } from './helpers/controlled-request.ts';
+import { createMockAuth, toRoutes } from './helpers/misc.ts';
 
-const expectPending = async (promise: Promise<void>): Promise<void> => {
-    const settled = vi.fn();
-    void promise.then(settled);
-    await Promise.resolve();
-    expect(settled).not.toHaveBeenCalled();
-};
+/**
+ * A mock of the AuthSecp256k1 service
+ */
+const auth = createMockAuth();
 
 describe('long-lived subscription dispatch', (): void => {
     it('allows duplicate subscribe and unsubscribe requests while the original request waits', async (): Promise<void> => {
         const broadcaster = new Broadcaster(new Logger('subscription-flow-test'));
-        const router = await ApplicationRouter.create([
-            moduleWith([
+        const router = await ApplicationRouter.create({
+                auth,
+            },
+            [
+            toRoutes([
                 {
                     url: '/items/subscribe',
                     handler: async (stream): Promise<void> => {
+                        const { signalStarted, released } = stream.body as {
+                            signalStarted: () => void;
+                            released: Promise<void>;
+                        };
+                        
+                        signalStarted();
+                        
                         await broadcaster.subscribe(stream, [ 'items' ]);
+                        await released;
                     },
                 },
                 {
@@ -43,17 +46,24 @@ describe('long-lived subscription dispatch', (): void => {
         ]);
         const connection = new TestConnection(true, true);
 
-        const original = router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-1' }, connection);
+        const original = createControlledRequest({ router, connection, path: '/items/subscribe', requestId: 'subscribe-1' });
         await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1));
-        await expectPending(original);
+        await original.started;
 
         // This request uses a different ApplicationRouteStream over the same
         // connection. Since the topic already exists, its dispatch completes.
-        await router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-2' }, connection);
-        await expectPending(original);
+        const second = createControlledRequest({ router, connection, path: '/items/subscribe', requestId: 'subscribe-2' });
+        second.release();
+        await second.request;
 
-        await router.dispatch({ path: '/items/unsubscribe', requestId: 'unsubscribe-1' }, connection);
-        await original;
+        // Unsubscribe the original request stream
+        const third = createControlledRequest({ router, connection, path: '/items/unsubscribe', requestId: 'unsubscribe-1' });
+        third.release();
+        await third.request;
+
+        // Release the original request stream
+        original.release();
+        await original.request;
 
         expect(connection.messages).toEqual([
             {