Fix package-lock

package.json

@@ -17,7 +17,7 @@
     "scripts": {
         "analyze": "tsdown --clean --no-fixed-extension --sourcemap source/index.ts && esbuild-analyzer dist/",
         "build": "tsdown --clean --sourcemap source/index.ts",
-        "dev": "tsx watch source/app.ts",
+        "dev": "tsx watch source/index.ts",
         "docs": "typedoc --hideGenerator --categorizeByGroup",
         "format": "prettier --write . && eslint --fix",
         "spellcheck": "cspell 'source/**' 'test/**' 'playground/**'",
@@ -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/auth.ts

@@ -0,0 +1,122 @@
+import type { Database } from '../services/storage/database.ts';
+
+import { hexToBin, instantiateSecp256k1, type Secp256k1, sha256 } from '@bitauth/libauth';
+import { UnauthorizedError } from '../errors/unauthorized-error.ts';
+
+export type AuthSecp256k1RequiredDeps = {
+    database: Database;
+};
+
+export type AuthSecp256k1OptionalDeps = {
+    secp256k1: Secp256k1;
+};
+export type AuthSecp256k1Deps = AuthSecp256k1RequiredDeps & Partial<AuthSecp256k1OptionalDeps>;
+
+export type AuthSecp256k1Options = {
+    timestampWindowMs: number;
+};
+
+export class AuthSecp256k1 {
+    /**
+     * Create a new instance of AuthSecp256k1
+     * @returns A new instance of AuthSecp256k1
+     */
+    static async create(inputDeps: AuthSecp256k1Deps, options: AuthSecp256k1Options): Promise<AuthSecp256k1> {
+        const deps = {
+            secp256k1: inputDeps.secp256k1 || await instantiateSecp256k1(),
+            ...inputDeps,
+        };
+
+        return new AuthSecp256k1(deps, options);
+    }
+
+    /**
+     * TODO: Consider adding a Record<string, Mutex> where each key is the signature to guarantee signatures arent being processed concurrently.
+     */
+    readonly #deps: Required<AuthSecp256k1Deps>;
+    readonly #options: AuthSecp256k1Options;
+
+    /**
+     * @param deps - The dependencies to use
+     * @param options - The options to use
+     */
+    private constructor(deps: Required<AuthSecp256k1Deps>, options: AuthSecp256k1Options) {
+        this.#deps = deps;
+        this.#options = options;
+    }
+
+    /**
+     * Verify a signature
+     * @param publicKeyHex - The public key to verify the signature against
+     * @param signatureHex - The signature to verify
+     * @param payload - The payload to verify the signature against
+     * @returns Whether the signature is valid
+     */
+    async verifySignature(publicKeyHex: string, signatureHex: string, payload: string): Promise<boolean> {
+        // Convert the public key and signature to binary
+        const publicKey = hexToBin(publicKeyHex);
+        const signature = hexToBin(signatureHex);
+
+        // Convert the payload to bytes and compute the sha256 hash
+        const payloadBytes = new TextEncoder().encode(payload);
+        const messageHash = sha256.hash(payloadBytes);
+
+        // Verify the signature
+        const verified = this.#deps.secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash);
+
+        // If the signature is not valid, throw an unauthorized error
+        if (!verified) {
+            throw new UnauthorizedError('Invalid signature');
+        }
+
+        // Return the verified signature
+        return verified;
+    }
+
+    /**
+     * Verify a signature is unique and add it to the database if it is
+     *
+     * @param signature - The signature to verify
+     * @returns Whether the signature is unique
+     */
+    async verifyUniqueRequest(signature: string): Promise<boolean> {
+        // If the signature is not provided, throw an unauthorized error
+        if (!signature) {
+            throw new UnauthorizedError('Signature is required');
+        }
+
+        // Check if the signature has been used before
+        const request = await this.#deps.database.db.selectFrom('authed_requests').selectAll()
+.where('signature', '=', signature)
+.executeTakeFirst();
+
+        // If the signature has been used before, throw an unauthorized error
+        if (request) {
+            throw new UnauthorizedError('Request already used');
+        }
+
+        // Add the signature to the requests table
+        await this.#deps.database.db.insertInto('authed_requests').values({ signature })
+.execute();
+
+        // Return true if the signature is valid
+        return true;
+    }
+
+    /**
+     * Assert that a timestamp is within a allowed window
+     * @param timestamp - The timestamp to check
+     * @param windowMs - The window in milliseconds
+     *
+     * @throws An {@link UnauthorizedError} if the timestamp is outside the allowed window
+     */
+    assertTimestampFreshness(timestamp: number, windowMs = this.#options.timestampWindowMs): void {
+        // Subtract the timestamp from the current time to get the age in milliseconds
+        const age = Math.abs(Date.now() - timestamp);
+
+        // If the timestamp is outside the allowed window, throw an unauthorized error
+        if (age > windowMs) {
+            throw new UnauthorizedError('Timestamp outside allowed window');
+        }
+    }
+}

source/index.ts

@@ -1,5 +1,6 @@
 import { Config } from './services/config.ts';
 import { Database, MigrationService } from './services/storage/index.ts';
+import { AuthSecp256k1 } from './auth/auth.ts';
 import { Broadcaster } from './services/broadcaster.ts';
 import { ApplicationRouter } from './services/router.ts';
 import { HttpTransportRouter } from './services/transport/http-transport.ts';
@@ -23,24 +24,34 @@ export class App {
         const migrations = new MigrationService(database, debug);
         await migrations.migrateToLatest();
 
+        // Create an Auth instance that can be passed in for signature validation
+        const auth = await AuthSecp256k1.create({ database }, { timestampWindowMs: config.auth.timestampWindowMs });
+
         // Domain services are shared across all transports and route modules.
         const broadcaster = new Broadcaster(debug);
         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, config.auth.timestampWindowMs),
+            new DataRoute({
+                auth: auth,
+                database: database,
+                broadcaster: broadcaster,
+            }, {
+                timestampWindowMs: config.auth.timestampWindowMs,
+            }),
         ];
 
         // 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);
+        const router = await ApplicationRouter.create({ auth }, routes);
 
         // Both transports share one ApplicationRouter. Routes use the shared
         // Broadcaster directly, while HTTP and WebSocket remain protocol adapters.
         const http = new HttpTransportRouter(router, debug);
+
         const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes);
-        const host = new ServerHost(config, debug, [http, ws]);
+        const host = new ServerHost(config, debug, [ http, ws ]);
 
         return new App(host, database);
     }
@@ -53,6 +64,7 @@ export class App {
     ) {}
 
     async start(): Promise<void> {
+        await this.database.start();
         await this.host.start();
     }
 
@@ -67,6 +79,16 @@ export class App {
         })();
         await this.stopPromise;
     }
+
+    startUniqueRequestCleanup(cleanupIntervalMs: number, timestampWindowMs: number): void {
+        // Every 10 seconds, we will cleanup the requests table
+        setInterval(async () => {
+            await this.database.db
+                .deleteFrom('authed_requests')
+                .where('timestamp', '<', Date.now() - timestampWindowMs)
+                .execute();
+        }, cleanupIntervalMs);
+    }
 }
 
 const app = await App.create();

source/routes/resources.ts

@@ -1,5 +1,3 @@
-import { createHash } from 'node:crypto';
-import { hexToBin, instantiateSecp256k1, type Secp256k1 } from '@bitauth/libauth';
 import { toExtendedJson } from '@xo-cash/utils';
 import { z } from 'zod';
 
@@ -9,6 +7,7 @@ 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 { AuthSecp256k1 } from '../auth/auth.ts';
 
 /**
  * Schema to validate a single write resource.
@@ -42,11 +41,21 @@ const resourceIdsSchema = z.object({
     resourceId: z
         .array(z.string().min(1))
         .min(1, 'At least one resourceId is required')
-        .transform((ids) => [...new Set(ids)]),
+        .transform((ids) => [ ...new Set(ids) ]),
 });
 
 type WriteResource = z.infer<typeof writeResource>;
 
+export type DataRouteDeps = {
+    auth: AuthSecp256k1;
+    database: Database;
+    broadcaster: BaseBroadcaster;
+}
+
+export type DataRouteOptions = {
+    timestampWindowMs: number;
+}
+
 /**
  * Resource data domain routes.
  *
@@ -56,16 +65,9 @@ type WriteResource = z.infer<typeof writeResource>;
  * RouteStream API.
  */
 export class DataRoute implements RouteModule {
-    /**
-     * Promise to instantiate the secp256k1 library
-     * This is a bit annoying, but keeping a single instance alive makes more sense than instantiating it for each verification.
-     */
-    private readonly secp256k1Promise: Promise<Secp256k1> = instantiateSecp256k1();
-
     constructor(
-        private readonly database: Database,
+        private readonly deps: DataRouteDeps,
-        private readonly broadcaster: BaseBroadcaster,
+        private readonly options: DataRouteOptions,
-        private readonly timestampWindowMs: number,
     ) {}
 
     /** Declare exact routes; each handler owns its stream behavior. */
@@ -95,7 +97,7 @@ export class DataRoute implements RouteModule {
         const resourceIds = this.getResourceIds(stream);
 
         // Remove duplicates.
-        const uniqueIds = [...new Set(resourceIds)];
+        const uniqueIds = [ ...new Set(resourceIds) ];
 
         // If there are no resource ids, return an empty array.
         if (uniqueIds.length === 0) {
@@ -105,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'])
+            .select([ 'resource_id', 'public_key', 'blob', 'timestamp' ])
             .where('resource_id', 'in', uniqueIds)
             .orderBy('timestamp', 'asc')
             .execute();
@@ -140,26 +142,27 @@ export class DataRoute implements RouteModule {
         const rows = resources.map((resource) => ({
             resourceId: resource.id,
             publicKey: resource.publicKey,
+            signature: resource.signature,
             blob: Buffer.from(resource.value),
         }));
 
         // 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 } of rows) {
+            for (const { resourceId, publicKey, blob, signature } of rows) {
                 await trx
                     .insertInto('resource_data')
                     .values({
                         resource_id: resourceId,
                         public_key: publicKey,
+                        signature: signature,
                         blob,
                         timestamp,
                     })
                     .onConflict((oc) =>
-                        oc.columns(['resource_id', 'public_key']).doUpdateSet({
+                        oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({
                             blob,
                             timestamp,
-                        }),
+                        }))
-                    )
                     .execute();
             }
         });
@@ -177,7 +180,7 @@ export class DataRoute implements RouteModule {
         // Notify subscribers on each changed resource. Topic names are scoped per
         // resource id so clients only receive events for resources they joined.
         for (const { resourceId, instance } of written) {
-            await this.broadcaster.publish(DataRoute.resourceTopic(resourceId), {
+            await this.deps.broadcaster.publish(DataRoute.resourceTopic(resourceId), {
                 type: 'instance-changed',
                 data: { resourceId, ...instance },
             });
@@ -200,7 +203,7 @@ export class DataRoute implements RouteModule {
 
         // 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);
     }
 
     /**
@@ -218,7 +221,7 @@ 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)),
         );
@@ -242,7 +245,7 @@ export class DataRoute implements RouteModule {
     private async verifyWriteResource(resource: WriteResource): Promise<void> {
         this.assertFreshTimestamp(resource.timestamp);
 
-        if (!(await this.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) {
+        if (!(await this.deps.auth.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) {
             throw new UnauthorizedError('Invalid resource signature');
         }
     }
@@ -254,37 +257,11 @@ export class DataRoute implements RouteModule {
      */
     private assertFreshTimestamp(timestamp: number): void {
         const age = Math.abs(Date.now() - timestamp);
-        if (age > this.timestampWindowMs) {
+        if (age > this.options.timestampWindowMs) {
             throw new UnauthorizedError('Timestamp outside allowed window');
         }
     }
 
-    /**
-     * Verify a secp256k1 signature.
-     *
-     * @param publicKeyHex - The public key to verify the signature against.
-     * @param signatureHex - The signature to verify.
-     * @param payload - The payload to verify the signature against.
-     * @returns Whether the signature is valid.
-     */
-    private async verifySignature(publicKeyHex: string, signatureHex: string, payload: string): Promise<boolean> {
-        const secp256k1 = await this.secp256k1Promise;
-
-        try {
-            // Convert the public key and signature to binary.
-            const publicKey = hexToBin(publicKeyHex);
-            const signature = hexToBin(signatureHex);
-
-            // Create a SHA-256 hash of the payload.
-            const messageHash = createHash('sha256').update(payload).digest();
-
-            // Low-S normalization rejects malleable signature encodings.
-            return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash);
-        } catch {
-            return false;
-        }
-    }
-
     /**
      * Build the transport-independent payload authenticated by each signature.
      *
@@ -292,7 +269,7 @@ export class DataRoute implements RouteModule {
      * whether the client sent them over HTTP or WebSocket.
      */
     private static canonicalWritePayload(resource: WriteResource): string {
-        return `${resource.timestamp}${resource.id}${toExtendedJson(resource.value)}`;
+        return `${resource.timestamp}:${resource.id}:${toExtendedJson(resource.value)}`;
     }
 
     /** Broadcaster topic for a single resource's instance-changed events. */

source/routes/types.ts

@@ -1,6 +1,10 @@
 import type { BaseStream } from '../services/stream/base-stream.js';
 
+/** Canonical request headers exposed to transport-neutral route handlers. */
+export type RequestHeaders = Readonly<Record<string, string>>;
+
 export type RouteSendOptions = {
+
     /** Defaults to `response`; any other value sends an application event. */
     type?: string;
 
@@ -15,10 +19,15 @@ export type RouteSendOptions = {
  * 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;
 
+    /** Canonical route path selected for this request. */
+    readonly path: string;
+
     readonly body: unknown;
+    readonly headers: RequestHeaders;
     readonly streaming: boolean;
     readonly bidirectional: boolean;
 
@@ -29,6 +38,7 @@ 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;

source/services/broadcaster.ts

@@ -6,6 +6,7 @@ import { ApplicationError } from '../errors/index.ts';
 
 /** Request-scoped view from which the broadcaster obtains a stable connection. */
 export interface BroadcastStream {
+
     /** Connection identity shared by every request on the same transport session. */
     readonly connection: BaseStream;
 
@@ -15,6 +16,7 @@ export interface BroadcastStream {
 
 /** One pending subscribe call and the topics whose removal will resolve it. */
 interface SubscriptionWaiter {
+
     /** Only topics newly introduced by this particular subscribe call. */
     readonly remainingTopics: Set<string>;
 

source/services/config.ts

@@ -32,7 +32,7 @@ const configSchema = z.object({
             .object({
                 origin: z.string().default('*'),
                 methods: z.array(z.string()).default([ 'GET', 'POST', 'PUT', 'DELETE', 'OPTIONS' ]),
-                allowedHeaders: z.array(z.string()).default([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-PublicKey', 'X-Signature' ]),
+                allowedHeaders: z.array(z.string()).default([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-Public-Key', 'X-Signature' ]),
             })
             .partial()
             .prefault({}),
@@ -48,6 +48,11 @@ const configSchema = z.object({
                 .int()
                 .positive()
                 .default(5 * 60 * 1000),
+            uniqueRequestCleanupIntervalMs: z.coerce
+                .number()
+                .int()
+                .positive()
+                .default(10 * 60 * 1000),
         })
         .prefault({}),
 });
@@ -83,6 +88,7 @@ export class Config {
             },
             auth: {
                 timestampWindowMs: process.env.AUTH_TIMESTAMP_WINDOW_MS ? Number(process.env.AUTH_TIMESTAMP_WINDOW_MS) : undefined,
+                uniqueRequestCleanupIntervalMs: process.env.AUTH_UNIQUE_REQUEST_CLEANUP_INTERVAL_MS ? Number(process.env.AUTH_UNIQUE_REQUEST_CLEANUP_INTERVAL_MS) : undefined,
             },
         });
     }

source/services/route-stream.ts

@@ -1,7 +1,10 @@
-import type { RouteSendOptions, RouteStream } from '../routes/types.ts';
+import type { RequestHeaders, RouteSendOptions, RouteStream } from '../routes/types.ts';
 import type { BaseStream } from './stream/base-stream.ts';
 import { HTTP_STATUS_CODE_NO_CONTENT, HTTP_STATUS_CODE_SUCCESS } from '../constants.ts';
 
+/** Shared immutable value used when a request supplies no headers. */
+const EMPTY_REQUEST_HEADERS: RequestHeaders = Object.freeze({});
+
 /**
  * Binds one application request to a connection-level stream.
  *
@@ -10,16 +13,24 @@ import { HTTP_STATUS_CODE_NO_CONTENT, HTTP_STATUS_CODE_SUCCESS } from '../consta
  * connection-level services such as the broadcaster.
  */
 export class ApplicationRouteStream implements RouteStream {
+    readonly headers: RequestHeaders;
+
     /**
      * @param connection - Shared transport stream backing this request.
      * @param body - Transport-decoded application payload for the route handler.
+     * @param path - Canonical application route selected for this request.
      * @param requestId - Optional correlation ID for multiplexed transports.
+     * @param headers - Transport-normalized request headers for this dispatch.
      */
     constructor(
         readonly connection: BaseStream,
         readonly body: unknown,
+        readonly path: string,
         private readonly requestId?: string,
-    ) {}
+        headers: RequestHeaders = EMPTY_REQUEST_HEADERS,
+    ) {
+        this.headers = headers === EMPTY_REQUEST_HEADERS ? headers : Object.freeze({ ...headers });
+    }
 
     /** Whether the underlying connection can deliver server-pushed events. */
     get streaming(): boolean {

source/services/router.ts

@@ -1,24 +1,53 @@
-import type { RouteDefinition, RouteModule } from '../routes/types.ts';
+import { z } from 'zod';
-import { ApplicationError } from '../errors/index.ts';
+import { toExtendedJson } from '@xo-cash/utils';
+
+import type { RequestHeaders, RouteDefinition, RouteModule } from '../routes/types.ts';
+import { ApplicationError, UnauthorizedError } from '../errors/index.ts';
 import { ApplicationRouteStream } from './route-stream.ts';
 import type { BaseStream } from './stream/base-stream.ts';
+import type { AuthSecp256k1 } from '../auth/auth.ts';
 
 /** Canonical request produced by every transport adapter. */
 export type ApplicationRequest = {
+
     /** Exact application route name. */
     path: string;
 
     /** Transport-decoded application payload. */
     body?: unknown;
 
+    /** Transport-normalized request headers, keyed by lowercase name. */
+    headers?: RequestHeaders;
+
     /** Optional correlation ID supplied by a multiplexed transport. */
     requestId?: string;
 };
 
+export type ApplicationRouterDependencies = {
+
+    /** Authentication service. */
+    auth: AuthSecp256k1;
+};
+
+const accountSchema = z
+    .object({
+        'x-public-key': z.string(),
+        'x-signature': z.string(),
+        'x-timestamp': z.coerce.number(),
+    })
+    .transform((data) => ({
+        publicKey: data['x-public-key'],
+        signature: data['x-signature'],
+        timestamp: data['x-timestamp'],
+    }));
+
 /** 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>) {}
+    private constructor(
+        private readonly deps: ApplicationRouterDependencies,
+        private readonly routes: ReadonlyMap<string, RouteDefinition>,
+    ) {}
 
     /**
      * Load and validate the complete route table before accepting traffic.
@@ -26,7 +55,7 @@ export class ApplicationRouter {
      * @param routeModules - Route modules whose handlers will be registered.
      * @returns A ready-to-dispatch router instance.
      */
-    static async create(routeModules: RouteModule[]): Promise<ApplicationRouter> {
+    static async create(deps: ApplicationRouterDependencies, routeModules: RouteModule[]): Promise<ApplicationRouter> {
         const routes = new Map<string, RouteDefinition>();
 
         // Collect routes from every module and reject duplicates at startup.
@@ -41,7 +70,7 @@ export class ApplicationRouter {
             }
         }
 
-        return new ApplicationRouter(routes);
+        return new ApplicationRouter(deps, routes);
     }
 
     /**
@@ -51,12 +80,33 @@ export class ApplicationRouter {
      * @param connection - Shared connection stream for this transport session.
      */
     async dispatch(request: ApplicationRequest, connection: BaseStream): Promise<void> {
+        // Authenticate the headers on the request.
+        const { publicKey, signature, timestamp } = accountSchema.parse(request.headers);
+
+        // Make sure the request signature is valid and hasnt been used before
+        await this.deps.auth.verifyUniqueRequest(signature);
+
+        // Ensure that the headers are present.
+        if (!publicKey || !signature || !timestamp) {
+            throw new UnauthorizedError('Missing authentication headers');
+        }
+
+        // Compile the signature payload as `Path:Timestamp:Body`
+        const signaturePayload = `${timestamp}:${request.path}:${toExtendedJson(request.body)}`;
+
+        // Verify the signature of the request.
+        const verified = await this.deps.auth.verifySignature(publicKey, signature, signaturePayload);
+        if (!verified) {
+            throw new UnauthorizedError('Invalid signature');
+        }
+
+        // Get the route from the routes map.
         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);
+        const stream = new ApplicationRouteStream(connection, request.body, request.path, request.requestId, request.headers);
         await route.handler(stream);
     }
 

source/services/server-host.ts

@@ -40,8 +40,8 @@ export class ServerHost {
 
         const corsMiddleware = cors({
             origin: corsConfig.origin ?? '*',
-            allowMethods: corsConfig.methods ?? ['POST', 'OPTIONS'],
+            allowMethods: corsConfig.methods ?? [ 'POST', 'OPTIONS' ],
-            allowHeaders: corsConfig.allowedHeaders ?? ['Content-Type', 'Accept'],
+            allowHeaders: corsConfig.allowedHeaders ?? [ 'Content-Type', 'Accept' ],
         });
 
         this.app.use('*', corsMiddleware);
@@ -77,7 +77,7 @@ export class ServerHost {
             throw new Error('ServerHost supports only one WebSocket upgrade server');
         }
 
-        const [upgradeTransport] = upgradeTransports;
+        const [ upgradeTransport ] = upgradeTransports;
 
         this.server = serve({
             fetch: this.app.fetch,
@@ -124,7 +124,7 @@ export class ServerHost {
         const closeTransports = this.transports.map((transport) => Promise.resolve(transport.stop?.()));
 
         // Create a promise that resolves when the server and transports are closed
-        this.stopPromise = Promise.all([closeServer, ...closeTransports]).then(() => {
+        this.stopPromise = Promise.all([ closeServer, ...closeTransports ]).then(() => {
             this.stopPromise = undefined;
         });
 

source/services/storage/database.ts

@@ -6,6 +6,7 @@ import type { Logger } from '../../utils/logger.ts';
 
 /** Options required to open a SQLite database connection. */
 export type DatabaseOptions = {
+
     /** Filesystem path to the SQLite database file. */
     path: string;
 
@@ -39,9 +40,6 @@ export class Database {
         this.kysely = new Kysely<DatabaseTables>({
             dialect: this.dialect,
         });
-
-        // Configure the SQLite pragmas.
-        this.configurePragmas();
     }
 
     /**
@@ -53,6 +51,13 @@ export class Database {
         return this.kysely;
     }
 
+    async start(): Promise<void> {
+        this.debug('starting database connection');
+
+        // Configure the SQLite pragmas.
+        await this.configurePragmas();
+    }
+
     /**
      * Destroys the database connection.
      */
@@ -66,10 +71,10 @@ export class Database {
      *
      * WAL improves write concurrency; foreign keys enforce referential integrity.
      */
-    private configurePragmas(): void {
+    private async configurePragmas(): Promise<void> {
         this.debug('configuring SQLite pragmas');
 
-        this.kysely.executeQuery(CompiledQuery.raw('PRAGMA journal_mode = WAL'));
+        await this.kysely.executeQuery(CompiledQuery.raw('PRAGMA journal_mode = WAL'));
-        this.kysely.executeQuery(CompiledQuery.raw('PRAGMA foreign_keys = ON'));
+        await this.kysely.executeQuery(CompiledQuery.raw('PRAGMA foreign_keys = ON'));
     }
 }

source/services/storage/migrations/001-resources.ts

@@ -23,7 +23,18 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
         .addColumn('public_key', 'text', (col) => col.notNull())
         .addColumn('blob', 'blob', (col) => col.notNull())
         .addColumn('timestamp', 'integer', (col) => col.notNull().defaultTo(millisecondTime))
-        .addPrimaryKeyConstraint('pk_resource_data', ['resource_id', 'public_key'])
+        .addColumn('signature', 'text', (col) => col.notNull())
+        .addPrimaryKeyConstraint('pk_resource_data', [ 'resource_id', 'public_key' ])
+        .execute();
+
+    // Table for authed requests
+    // We will store the signature and the timestamp of the request, and we will clear out rows that are older than our msTimeout for our auth
+    await db.schema
+        .createTable('authed_requests')
+        .ifNotExists()
+        .addColumn('signature', 'text', (col) => col.notNull())
+        .addColumn('timestamp', 'integer', (col) => col.notNull().defaultTo(millisecondTime))
+        .addPrimaryKeyConstraint('pk_authed_requests', [ 'signature' ])
         .execute();
 };
 
@@ -33,5 +44,9 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
  * @param db - Kysely database to apply the rollback against.
  */
 export const down = async (db: Kysely<DatabaseTables>): Promise<void> => {
-    await db.schema.dropTable('resource_data').ifExists().execute();
+    await db.schema.dropTable('resource_data').ifExists()
+.execute();
+
+    await db.schema.dropTable('authed_requests').ifExists()
+.execute();
 };

source/services/storage/tables.ts

@@ -10,6 +10,7 @@ export type BlobColumn = ColumnType<Buffer, Buffer | Uint8Array, Buffer>;
  * One row per (resource_id, public_key). Each publicKey owns a slot within a shared resource.
  */
 export interface ResourceDataTable {
+
     /** Shared resource identifier grouping related instances. */
     resource_id: string;
 
@@ -21,9 +22,22 @@ export interface ResourceDataTable {
 
     /** Millisecond timestamp of the last write. */
     timestamp: Timestamp;
+
+    /** Signature of the write. */
+    signature: string;
+}
+
+export interface AuthedRequestsTable {
+
+    /** Signature of the request. */
+    signature: string;
+
+    /** Millisecond timestamp of the request. */
+    timestamp: Timestamp;
 }
 
 /** Complete Kysely schema mapping for the sync server database. */
 export interface DatabaseTables {
     resource_data: ResourceDataTable;
+    authed_requests: AuthedRequestsTable;
 }

source/services/stream/base-stream.ts

@@ -1,5 +1,6 @@
 /** A normal request/response result before transport encoding. */
 export type StreamResponse = {
+
     /** Optional correlation ID for multiplexed transports. */
     id?: string;
 
@@ -15,6 +16,7 @@ export type StreamResponse = {
 
 /** An application event before a transport applies its wire encoding. */
 export type StreamEvent = {
+
     /** Optional event or correlation ID. */
     id?: string;
 

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,12 +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 { type AppEnv, BaseTransport } from './base-transport.ts';
 
 /** Hono context key where decoded Extended JSON bodies are stored. */
 const PARSED_BODY_KEY = 'parsedBody';
@@ -21,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. */
@@ -35,6 +35,8 @@ export class HttpTransportRouter implements TransportRouter {
         private readonly router: ApplicationRouter,
         debug: Logger,
     ) {
+        super();
+
         this.debug = debug.extend('http-transport');
     }
 
@@ -111,6 +113,7 @@ export class HttpTransportRouter implements TransportRouter {
         return {
             path: context.req.path,
             ...(body === undefined ? {} : { body }),
+            headers: context.req.header(),
         };
     }
 

source/services/transport/request-headers.ts

@@ -0,0 +1,38 @@
+import type { RequestHeaders } from '../../routes/types.ts';
+import { ApplicationError } from '../../errors/index.ts';
+import { HTTP_STATUS_CODE_BAD_REQUEST } from '../../constants.ts';
+
+/** RFC 9110 field-name token grammar. */
+const HEADER_NAME_PATTERN = /^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$/;
+
+/**
+ * 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.
+ */
+export const 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/transport-router.ts

@@ -4,6 +4,7 @@ import type { Hono } from 'hono';
 /** Hono variables populated by transport-boundary middleware. */
 export type AppEnv = {
     Variables: {
+
         /** Decoded Extended JSON request body, when present. */
         parsedBody?: unknown;
 
@@ -19,6 +20,7 @@ export type AppEnv = {
  * not application routing. Implementations remain unaware of route modules.
  */
 export interface TransportRouter {
+
     /**
      * Attach wire endpoints and middleware to the shared Hono application.
      *
@@ -32,6 +34,7 @@ export interface TransportRouter {
 
 /** A transport which also supplies the WebSocket server used during upgrade. */
 export interface UpgradeTransportRouter extends TransportRouter {
+
     /** WebSocket server instance passed to the Node HTTP listener. */
     readonly websocketServer: WebSocketServerLike;
 }

source/services/transport/ws-transport.ts

@@ -8,11 +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 { type AppEnv, BaseTransport } from './base-transport.ts';
 
 /** Default WebSocket upgrade path for application messages. */
 const WS_ROUTE = '/ws';
@@ -20,7 +20,8 @@ const WS_ROUTE = '/ws';
 // Strict validation prevents legacy or protocol-specific fields reaching routes.
 const wsRequestSchema = z
     .object({
-        id: z.string().min(1).optional(),
+        id: z.string().min(1)
+.optional(),
         path: z.string().min(1),
         body: z.unknown().optional(),
     })
@@ -32,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.
      *
@@ -41,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;
 
     /**
@@ -58,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
@@ -67,7 +67,6 @@ export class WsTransportRouter implements UpgradeTransportRouter {
             noServer: true,
             maxPayload: this.maxRequestBodyBytes,
         });
-        this.websocketServer = this.wsServer as unknown as WebSocketServerLike;
     }
 
     /**
@@ -178,13 +177,11 @@ export class WsTransportRouter implements UpgradeTransportRouter {
      * @param error - Failure to normalize into the public error shape.
      */
     private sendError(ws: WSContext, requestId: string | undefined, error: unknown): void {
-        ws.send(
+        ws.send(toExtendedJson({
-            toExtendedJson({
                 ...(requestId === undefined ? {} : { id: requestId }),
                 type: 'error',
                 ...normalizePublicError(error),
-            }),
+            }));
-        );
     }
 
     /**

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/helpers/test-connection.ts

@@ -1,45 +1,43 @@
-import {
+import { BaseStream, type StreamMessage } from '../../source/services/stream/base-stream.ts';
-  BaseStream,
-  type StreamMessage,
-} from "../../source/services/stream/base-stream.js";
 
 /** Minimal observable connection used by application and broadcaster tests. */
 export class TestConnection extends BaseStream {
-  readonly messages: StreamMessage[] = [];
+    readonly messages: StreamMessage[] = [];
-  readonly closeCallbacks: Array<() => void> = [];
+    readonly closeCallbacks: Array<() => void> = [];
-  closed = false;
+    closed = false;
 
-  constructor(
+    constructor(
-    readonly streaming: boolean,
+        readonly streaming: boolean,
-    readonly bidirectional: boolean,
+        readonly bidirectional: boolean,
-  ) {
+    ) {
-    super();
+        super();
-  }
-
-  async send(message: StreamMessage): Promise<void> {
-    if (this.closed) {
-      throw new Error("connection is closed");
     }
 
-    this.messages.push(message);
+    async send(message: StreamMessage): Promise<void> {
-  }
+        if (this.closed) {
+            throw new Error('connection is closed');
+        }
 
-  close(): void {
+        this.messages.push(message);
-    if (this.closed) {
-      return;
     }
 
-    this.closed = true;
+    close(): void {
-    const callbacks = this.closeCallbacks.splice(0);
+        if (this.closed) {
-    callbacks.forEach((callback) => callback());
+            return;
-  }
+        }
 
-  onClose(callback: () => void): void {
+        this.closed = true;
-    if (this.closed) {
+        const callbacks = this.closeCallbacks.splice(0);
-      callback();
+        callbacks.forEach((callback) => callback());
-      return;
     }
 
-    this.closeCallbacks.push(callback);
+    onClose(callback: () => void): void {
-  }
+        if (this.closed) {
+            callback();
+
+            return;
+        }
+
+        this.closeCallbacks.push(callback);
+    }
 }

test/routes/resources.test.ts

@@ -1,140 +1,150 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
 
-import { DataRoute } from "../../source/routes/resources.js";
+import { DataRoute } from '../../source/routes/resources.ts';
-import { UnauthorizedError } from "../../source/errors/index.js";
+import { UnauthorizedError } from '../../source/errors/index.ts';
-import { type BaseBroadcaster } from "../../source/services/broadcaster.js";
+import { type BaseBroadcaster } from '../../source/services/broadcaster.ts';
-import { ApplicationRouteStream } from "../../source/services/route-stream.js";
+import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
-import { Database } from "../../source/services/storage/database.js";
+import type { Database } from '../../source/services/storage/database.ts';
-import { TestConnection } from "../helpers/test-connection.js";
+import { TestConnection } from '../helpers/test-connection.ts';
-import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../source/constants.js";
+import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../../source/constants.ts';
+import { createMockAuth } from '../helpers/misc.ts';
+import { AuthSecp256k1 } from '../../source/auth/auth.ts';
 
-function createBroadcasterStub() {
+const createBroadcasterStub = (): BaseBroadcaster => {
-  return {
+    return {
-    subscribe: vi.fn(),
+        subscribe: vi.fn(),
-    unsubscribe: vi.fn().mockResolvedValue(undefined),
+        unsubscribe: vi.fn().mockResolvedValue(undefined),
-    publish: vi.fn(),
+        publish: vi.fn(),
-    sendEvent: vi.fn(),
+        sendEvent: vi.fn(),
-  } as unknown as BaseBroadcaster;
+    } as unknown as BaseBroadcaster;
-}
+};
 
-describe("DataRoute subscriptions", () => {
+describe('DataRoute subscriptions', (): void => {
-  it("subscribes to future resource changes until removal", async () => {
+    it('subscribes to future resource changes until removal', async (): Promise<void> => {
-    let resolveRemoved: () => void = () => undefined;
+        let resolveRemoved: () => void = () => undefined;
-    const removed = new Promise<void>((resolve) => {
+        const removed = new Promise<void>((resolve) => {
-      resolveRemoved = resolve;
+            resolveRemoved = resolve;
-    });
+        });
-    const storage = {
+        const storage = {
-      db: {
+            db: {
-        transaction: vi.fn(),
+                transaction: vi.fn(),
-      },
-    } as unknown as Database;
-    const broadcaster = createBroadcasterStub();
-    vi.mocked(broadcaster.subscribe).mockReturnValue(removed);
-    const connection = new TestConnection(true, false);
-    const stream = new ApplicationRouteStream(connection, {
-      resourceId: ["a", "b"],
-    });
-    const route = new DataRoute(storage, broadcaster, 0);
-
-    const execution = route.subscribeData(stream);
-
-    expect(broadcaster.subscribe).toHaveBeenCalledWith(
-      stream,
-      ["resource:a", "resource:b"],
-    );
-    expect(storage.db.transaction).not.toHaveBeenCalled();
-    expect(connection.messages).toEqual([]);
-    await expect(
-      Promise.race([
-        execution.then(() => "settled"),
-        Promise.resolve("pending"),
-      ]),
-    ).resolves.toBe("pending");
-
-    resolveRemoved();
-    await execution;
-  });
-
-  it("unsubscribes a bidirectional connection and acknowledges the request", async () => {
-    const storage = {
-      db: {
-        transaction: vi.fn(),
-      },
-    } as unknown as Database;
-    const broadcaster = createBroadcasterStub();
-    const connection = new TestConnection(true, true);
-    const stream = new ApplicationRouteStream(
-      connection,
-      { resourceId: ["a"] },
-      "unsubscribe-1",
-    );
-    const route = new DataRoute(storage, broadcaster, 0);
-
-    await route.unsubscribeData(stream);
-
-    expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [
-      "resource:a",
-    ]);
-    expect(connection.messages).toEqual([
-      {
-        id: "unsubscribe-1",
-        type: "response",
-        statusCode: 200,
-        body: {},
-      },
-    ]);
-  });
-
-  it("rejects selective unsubscribe on a one-way connection", async () => {
-    const storage = {
-      db: {
-        transaction: vi.fn(),
-      },
-    } as unknown as Database;
-    const broadcaster = createBroadcasterStub();
-    const stream = new ApplicationRouteStream(new TestConnection(true, false), {
-      resourceId: ["a"],
-    });
-    const route = new DataRoute(storage, broadcaster, 0);
-
-    await expect(route.unsubscribeData(stream)).rejects.toMatchObject({ statusCode: HTTP_STATUS_CODE_NOT_ACCEPTED });
-    expect(broadcaster.unsubscribe).not.toHaveBeenCalled();
-  });
-});
-
-describe("DataRoute resource write auth", () => {
-  it("rejects an invalid resource signature before writing the batch", async () => {
-    const storage = {
-      db: {
-        transaction: vi.fn(),
-      },
-    } as unknown as Database;
-
-    const broadcaster = createBroadcasterStub()
-    const route = new DataRoute(storage, broadcaster, 0);
-
-    await expect(
-      route.writeData({
-        connection: new TestConnection(true, true),
-        streaming: true,
-        bidirectional: true,
-        send: vi.fn(),
-        body: {
-          resources: [
-            {
-              id: "resource-a",
-              publicKey: "not-a-public-key",
-              timestamp: Date.now(),
-              signature: "not-a-signature",
-              value: new Uint8Array([1, 2, 3]),
             },
-          ],
+        } as unknown as Database;
-        },
+        const broadcaster = createBroadcasterStub();
-      } as unknown as ApplicationRouteStream),
+        vi.mocked(broadcaster.subscribe).mockReturnValue(removed);
-    ).rejects.toBeInstanceOf(UnauthorizedError);
+        const connection = new TestConnection(true, false);
+        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,
+        });
 
-    expect(storage.db.transaction).not.toHaveBeenCalled()
+        const execution = route.subscribeData(stream);
-    expect(broadcaster.publish).not.toHaveBeenCalled();
+
-  });
+        expect(broadcaster.subscribe).toHaveBeenCalledWith(stream, [ 'resource:a', 'resource:b' ]);
+        expect(storage.db.transaction).not.toHaveBeenCalled();
+        expect(connection.messages).toEqual([]);
+        await expect(Promise.race([ execution.then(() => 'settled'), Promise.resolve('pending') ])).resolves.toBe('pending');
+
+        resolveRemoved();
+        await execution;
+    });
+
+    it('unsubscribes a bidirectional connection and acknowledges the request', async (): Promise<void> => {
+        const storage = {
+            db: {
+                transaction: vi.fn(),
+            },
+        } as unknown as Database;
+        const broadcaster = createBroadcasterStub();
+        const connection = new TestConnection(true, true);
+        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);
+
+        expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [ 'resource:a' ]);
+        expect(connection.messages).toEqual([
+            {
+                id: 'unsubscribe-1',
+                type: 'response',
+                statusCode: 200,
+                body: {},
+            },
+        ]);
+    });
+
+    it('rejects selective unsubscribe on a one-way connection', async (): Promise<void> => {
+        const storage = {
+            db: {
+                transaction: vi.fn(),
+            },
+        } as unknown as Database;
+        const broadcaster = createBroadcasterStub();
+        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();
+    });
 });
 
+describe('DataRoute resource write auth', (): void => {
+    it('rejects an invalid resource signature before writing the batch', async (): Promise<void> => {
+        const storage = {
+            db: {
+                transaction: vi.fn(),
+            },
+        } as unknown as Database;
+
+        const broadcaster = createBroadcasterStub();
+        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),
+                streaming: true,
+                bidirectional: true,
+                send: vi.fn(),
+                body: {
+                    resources: [
+                        {
+                            id: 'resource-a',
+                            publicKey: 'not-a-public-key',
+                            timestamp: Date.now(),
+                            signature: 'not-a-signature',
+                            value: new Uint8Array([ 1, 2, 3 ]),
+                        },
+                    ],
+                },
+            } as unknown as ApplicationRouteStream)).rejects.toBeInstanceOf(UnauthorizedError);
+
+        expect(storage.db.transaction).not.toHaveBeenCalled();
+        expect(broadcaster.publish).not.toHaveBeenCalled();
+    });
+});

test/services/broadcaster.test.ts

@@ -1,199 +1,180 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
 
-import { ApplicationError } from "../../source/errors/index.js";
+import { Broadcaster } from '../../source/services/broadcaster.ts';
-import { Broadcaster } from "../../source/services/broadcaster.js";
+import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
-import { ApplicationRouteStream } from "../../source/services/route-stream.js";
+import { Logger } from '../../source/utils/logger.ts';
-import { Logger } from "../../source/utils/logger.js";
+import { TestConnection } from '../helpers/test-connection.ts';
-import { TestConnection } from "../helpers/test-connection.js";
 
-function createBroadcaster(): Broadcaster {
+const createBroadcaster = (): Broadcaster => {
-  return new Broadcaster(new Logger("broadcaster-test"));
+    return new Broadcaster(new Logger('broadcaster-test'));
-}
+};
 
-function routeStream(connection: TestConnection): ApplicationRouteStream {
+const routeStream = (connection: TestConnection): ApplicationRouteStream => {
-  return new ApplicationRouteStream(connection, undefined);
+    return new ApplicationRouteStream(connection, undefined, '/items', 'items-1');
-}
+};
 
-async function expectPending(promise: Promise<void>): Promise<void> {
+const expectPending = async (promise: Promise<void>): Promise<void> => {
-  const settled = vi.fn();
+    const settled = vi.fn();
-  void promise.then(settled);
+    void promise.then(settled);
-  await Promise.resolve();
+    await Promise.resolve();
-  expect(settled).not.toHaveBeenCalled();
+    expect(settled).not.toHaveBeenCalled();
-}
+};
 
-describe("Broadcaster subscriptions", () => {
+describe('Broadcaster subscriptions', () => {
-  it("delivers events and resolves after a later request removes the topic", async () => {
+    it('delivers events and resolves after a later request removes the topic', async (): Promise<void> => {
-    const broadcaster = createBroadcaster();
+        const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
+        const connection = new TestConnection(true, true);
-    const subscribed = broadcaster.subscribe(routeStream(connection), [
+        const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
-      "items",
-    ]);
 
-    await expectPending(subscribed);
+        await expectPending(subscribed);
-    await broadcaster.publish("items", {
+        await broadcaster.publish('items', {
-      type: "item-changed",
+            type: 'item-changed',
-      data: { id: "a" },
+            data: { id: 'a' },
+        });
+
+        expect(connection.messages).toEqual([
+            expect.objectContaining({
+                type: 'item-changed',
+                data: { id: 'a' },
+            }),
+        ]);
+
+        // A different request-scoped facade still resolves the connection's
+        // original subscription.
+        await broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
+        await expect(subscribed).resolves.toBeUndefined();
     });
 
-    expect(connection.messages).toEqual([
+    it('resolves fully duplicate subscriptions immediately', async (): Promise<void> => {
-      expect.objectContaining({
+        const broadcaster = createBroadcaster();
-        type: "item-changed",
+        const connection = new TestConnection(true, true);
-        data: { id: "a" },
+        const first = broadcaster.subscribe(routeStream(connection), [ 'items', 'items' ]);
-      }),
+        const duplicate = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
-    ]);
 
-    // A different request-scoped facade still resolves the connection's
+        await expect(duplicate).resolves.toBeUndefined();
-    // original subscription.
+        await expectPending(first);
-    await broadcaster.unsubscribe(routeStream(connection), ["items"]);
+        expect(connection.closeCallbacks).toHaveLength(1);
-    await expect(subscribed).resolves.toBeUndefined();
-  });
 
-  it("resolves fully duplicate subscriptions immediately", async () => {
+        await broadcaster.unsubscribe(routeStream(connection), [ 'items' ]);
-    const broadcaster = createBroadcaster();
+        await first;
-    const connection = new TestConnection(true, true);
-    const first = broadcaster.subscribe(routeStream(connection), [
-      "items",
-      "items",
-    ]);
-    const duplicate = broadcaster.subscribe(routeStream(connection), ["items"]);
-
-    await expect(duplicate).resolves.toBeUndefined();
-    await expectPending(first);
-    expect(connection.closeCallbacks).toHaveLength(1);
-
-    await broadcaster.unsubscribe(routeStream(connection), ["items"]);
-    await first;
-  });
-
-  it("waits only for topics newly added by a partially overlapping call", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
-    const first = broadcaster.subscribe(routeStream(connection), ["a"]);
-    const second = broadcaster.subscribe(routeStream(connection), ["a", "b"]);
-
-    await broadcaster.unsubscribe(routeStream(connection), ["a"]);
-    await expect(first).resolves.toBeUndefined();
-    await expectPending(second);
-
-    await broadcaster.unsubscribe(routeStream(connection), ["b"]);
-    await expect(second).resolves.toBeUndefined();
-  });
-
-  it("resolves every pending subscription and removes topics on close", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, false);
-    const first = broadcaster.subscribe(routeStream(connection), ["a"]);
-    const second = broadcaster.subscribe(routeStream(connection), ["b"]);
-
-    connection.close();
-    await Promise.all([first, second]);
-    await broadcaster.publish("a", { type: "changed", data: null });
-    await broadcaster.publish("b", { type: "changed", data: null });
-
-    expect(connection.messages).toEqual([]);
-  });
-
-  it("immediately resolves registration against an already-closed connection", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, false);
-    connection.close();
-
-    await expect(
-      broadcaster.subscribe(routeStream(connection), ["items"]),
-    ).resolves.toBeUndefined();
-  });
-
-  it("rejects subscriptions on a non-streaming connection", () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(false, false);
-
-    expect(() =>
-      broadcaster.subscribe(routeStream(connection), ["items"]),
-    ).toThrowError(
-      expect.objectContaining({ statusCode: 406 }),
-    );
-  });
-
-  it("treats an empty subscription and repeated unsubscribe as no-ops", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
-
-    await expect(
-      broadcaster.subscribe(routeStream(connection), []),
-    ).resolves.toBeUndefined();
-    await broadcaster.unsubscribe(routeStream(connection), ["missing"]);
-    await broadcaster.unsubscribe(routeStream(connection));
-
-    expect(connection.closeCallbacks).toHaveLength(0);
-  });
-
-  it("fans out concurrently to independent connections", async () => {
-    const broadcaster = createBroadcaster();
-    const first = new TestConnection(true, false);
-    const second = new TestConnection(true, false);
-    const originalFirstSend = first.send.bind(first);
-    let releaseFirst: () => void = () => undefined;
-    const firstReleased = new Promise<void>((resolve) => {
-      releaseFirst = resolve;
-    });
-    let markSecondSent: () => void = () => undefined;
-    const secondSent = new Promise<void>((resolve) => {
-      markSecondSent = resolve;
     });
 
-    first.send = async (message) => {
+    it('waits only for topics newly added by a partially overlapping call', async (): Promise<void> => {
-      await firstReleased;
+        const broadcaster = createBroadcaster();
-      await originalFirstSend(message);
+        const connection = new TestConnection(true, true);
-    };
+        const first = broadcaster.subscribe(routeStream(connection), [ 'a' ]);
-    second.send = async (message) => {
+        const second = broadcaster.subscribe(routeStream(connection), [ 'a', 'b' ]);
-      await TestConnection.prototype.send.call(second, message);
-      markSecondSent();
-    };
 
-    const firstSubscription = broadcaster.subscribe(routeStream(first), [
+        await broadcaster.unsubscribe(routeStream(connection), [ 'a' ]);
-      "items",
+        await expect(first).resolves.toBeUndefined();
-    ]);
+        await expectPending(second);
-    const secondSubscription = broadcaster.subscribe(routeStream(second), [
+
-      "items",
+        await broadcaster.unsubscribe(routeStream(connection), [ 'b' ]);
-    ]);
+        await expect(second).resolves.toBeUndefined();
-    const publication = broadcaster.publish("items", {
-      type: "item-changed",
-      data: {},
     });
 
-    await secondSent;
+    it('resolves every pending subscription and removes topics on close', async (): Promise<void> => {
-    releaseFirst();
+        const broadcaster = createBroadcaster();
-    await publication;
+        const connection = new TestConnection(true, false);
+        const first = broadcaster.subscribe(routeStream(connection), [ 'a' ]);
+        const second = broadcaster.subscribe(routeStream(connection), [ 'b' ]);
 
-    expect(first.messages).toHaveLength(1);
+        connection.close();
-    expect(second.messages).toHaveLength(1);
+        await Promise.all([ first, second ]);
+        await broadcaster.publish('a', { type: 'changed', data: null });
+        await broadcaster.publish('b', { type: 'changed', data: null });
 
-    first.close();
+        expect(connection.messages).toEqual([]);
-    second.close();
-    await Promise.all([firstSubscription, secondSubscription]);
-  });
-
-  it("closes and removes a connection whose event delivery fails", async () => {
-    const broadcaster = createBroadcaster();
-    const connection = new TestConnection(true, true);
-    connection.send = vi.fn().mockRejectedValue(new Error("socket failed"));
-    const subscribed = broadcaster.subscribe(routeStream(connection), [
-      "items",
-    ]);
-
-    await broadcaster.publish("items", {
-      type: "item-changed",
-      data: {},
     });
-    await subscribed;
 
-    expect(connection.closed).toBe(true);
+    it('immediately resolves registration against an already-closed connection', async (): Promise<void> => {
-    expect(connection.send).toHaveBeenCalledOnce();
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(true, false);
+        connection.close();
 
-    await broadcaster.publish("items", {
+        await expect(broadcaster.subscribe(routeStream(connection), [ 'items' ])).resolves.toBeUndefined();
-      type: "item-changed",
+    });
-      data: {},
+
+    it('rejects subscriptions on a non-streaming connection', (): void => {
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(false, false);
+
+        expect(() => broadcaster.subscribe(routeStream(connection), [ 'items' ])).toThrowError(expect.objectContaining({ statusCode: 406 }));
+    });
+
+    it('treats an empty subscription and repeated unsubscribe as no-ops', async (): Promise<void> => {
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(true, true);
+
+        await expect(broadcaster.subscribe(routeStream(connection), [])).resolves.toBeUndefined();
+        await broadcaster.unsubscribe(routeStream(connection), [ 'missing' ]);
+        await broadcaster.unsubscribe(routeStream(connection));
+
+        expect(connection.closeCallbacks).toHaveLength(0);
+    });
+
+    it('fans out concurrently to independent connections', async (): Promise<void> => {
+        const broadcaster = createBroadcaster();
+        const first = new TestConnection(true, false);
+        const second = new TestConnection(true, false);
+        const originalFirstSend = first.send.bind(first);
+        let releaseFirst: () => void = () => undefined;
+        const firstReleased = new Promise<void>((resolve) => {
+            releaseFirst = resolve;
+        });
+        let markSecondSent: () => void = () => undefined;
+        const secondSent = new Promise<void>((resolve) => {
+            markSecondSent = resolve;
+        });
+
+        first.send = async (message): Promise<void> => {
+            await firstReleased;
+            await originalFirstSend(message);
+        };
+
+        second.send = async (message): Promise<void> => {
+            await TestConnection.prototype.send.call(second, message);
+            markSecondSent();
+        };
+
+        const firstSubscription = broadcaster.subscribe(routeStream(first), [ 'items' ]);
+        const secondSubscription = broadcaster.subscribe(routeStream(second), [ 'items' ]);
+        const publication = broadcaster.publish('items', {
+            type: 'item-changed',
+            data: {},
+        });
+
+        await secondSent;
+        releaseFirst();
+        await publication;
+
+        expect(first.messages).toHaveLength(1);
+        expect(second.messages).toHaveLength(1);
+
+        first.close();
+        second.close();
+        await Promise.all([ firstSubscription, secondSubscription ]);
+    });
+
+    it('closes and removes a connection whose event delivery fails', async (): Promise<void> => {
+        const broadcaster = createBroadcaster();
+        const connection = new TestConnection(true, true);
+        connection.send = vi.fn().mockRejectedValue(new Error('socket failed'));
+        const subscribed = broadcaster.subscribe(routeStream(connection), [ 'items' ]);
+
+        await broadcaster.publish('items', {
+            type: 'item-changed',
+            data: {},
+        });
+        await subscribed;
+
+        expect(connection.closed).toBe(true);
+        expect(connection.send).toHaveBeenCalledOnce();
+
+        await broadcaster.publish('items', {
+            type: 'item-changed',
+            data: {},
+        });
+        expect(connection.send).toHaveBeenCalledOnce();
     });
-    expect(connection.send).toHaveBeenCalledOnce();
-  });
 });

test/services/config.test.ts

@@ -18,7 +18,7 @@ const testConfigDefaultsTo1MiBRequestBodyLimit = (): void => {
     expect(config.server.host).toBe('0.0.0.0');
     expect(config.server.cors.origin).toBe('*');
     expect(config.server.cors.methods).toEqual([ 'GET', 'POST', 'PUT', 'DELETE', 'OPTIONS' ]);
-    expect(config.server.cors.allowedHeaders).toEqual([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-PublicKey', 'X-Signature' ]);
+    expect(config.server.cors.allowedHeaders).toEqual([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-Public-Key', 'X-Signature' ]);
     expect(config.auth.timestampWindowMs).toBe(300000);
 };
 

test/services/router.test.ts

@@ -1,134 +1,149 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it } from 'vitest';
 
-import type { RouteDefinition, RouteModule } from "../../source/routes/types.js";
+// Source
-import { ApplicationError } from "../../source/errors/index.js";
+import { ApplicationRouter } from '../../source/services/router.ts';
-import { ApplicationRouter } from "../../source/services/router.js";
-import { TestConnection } from "../helpers/test-connection.js";
 
-function moduleWith(routes: RouteDefinition[]): RouteModule {
+// Helpers
-  return {
+import { createMockAuth, toRoutes } from '../helpers/misc.ts';
-    async getRoutes() {
+import { TestConnection } from '../helpers/test-connection.ts';
-      return routes;
+import { createControlledRequest } from '../helpers/controlled-request.ts';
-    },
-  };
-}
 
-describe("ApplicationRouter initialization", () => {
+/**
-  it("rejects duplicate exact paths during startup", async () => {
+ * A mock of the AuthSecp256k1 service
-    const route = { url: "/echo", handler: () => undefined };
+ */
+const auth = createMockAuth();
 
-    await expect(
+describe('ApplicationRouter initialization', (): void => {
-      ApplicationRouter.create([moduleWith([route]), moduleWith([route])]),
+    it('rejects duplicate exact paths during startup', async (): Promise<void> => {
-    ).rejects.toThrow("Duplicate application route: /echo");
+        const route = { url: '/echo', handler: (): void => undefined };
-  });
 
-  it.each(["echo", "/", "/items/:id", "/items?active=true", "/items#active"])(
+        await expect(ApplicationRouter.create({ auth }, [ toRoutes([ route ]), toRoutes([ route ]) ])).rejects.toThrow('Duplicate application route: /echo');
-    "rejects the invalid route path %s",
+    });
-    async (url) => {
+
-      await expect(
+    it.each([ 'echo', '/', '/items/:id', '/items?active=true', '/items#active' ])('rejects the invalid route path %s', async (url): Promise<void> => {
-        ApplicationRouter.create([
+        await expect(ApplicationRouter.create({ auth }, [ toRoutes([{ url, handler: (): void => undefined }]) ])).rejects.toThrow('Invalid application route');
-          moduleWith([{ url, handler: () => undefined }]),
+    });
-        ]),
-      ).rejects.toThrow("Invalid application route");
-    },
-  );
 });
 
-describe("ApplicationRouter dispatch", () => {
+describe('ApplicationRouter dispatch', (): void => {
-  it("binds the connection, body, and request ID to one route stream", async () => {
+    it('binds the connection, body, and request ID to one route stream', async (): Promise<void> => {
-    const connection = new TestConnection(false, false);
+        const connection = new TestConnection(false, false);
-    const router = await ApplicationRouter.create([
+        const router = await ApplicationRouter.create({ auth }, [
-      moduleWith([
+            toRoutes([
-        {
+                {
-          url: "/echo",
+                    url: '/echo',
-          handler: async (stream) => {
+                    handler: async (stream): Promise<void> => {
-            expect(stream.connection).toBe(connection);
+                        expect(stream.connection).toBe(connection);
-            await stream.send(stream.body);
+                        expect(stream.path).toBe('/echo');
-          },
+                        expect(stream.headers).toEqual({ 'x-public-key': 'public-key', 'x-signature': 'signature', 'x-timestamp': '1000' });
-        },
+                        await stream.send(stream.body);
-      ]),
+                    },
-    ]);
+                },
+            ]),
+        ]);
 
-    await router.dispatch(
+        await router.dispatch(
-      { path: "/echo", body: { value: 1 }, requestId: "request-1" },
+            {
-      connection,
+                path: '/echo',
-    );
+                body: { value: 1 },
+                requestId: '1',
+                headers: { 'x-public-key': 'public-key', 'x-signature': 'signature', 'x-timestamp': '1000' },
+            },
+            connection,
+        );
 
-    expect(connection.messages).toEqual([
+        expect(connection.messages).toEqual([
-      {
+            {
-        id: "request-1",
+                id: '1',
-        type: "response",
+                type: 'response',
-        statusCode: 200,
+                statusCode: 200,
-        body: { value: 1 },
+                body: { value: 1 },
-      },
+            },
-    ]);
+        ]);
 
-    await expect(
+        await expect(router.dispatch(
-      router.dispatch({ path: "/echo/other", body: {} }, connection),
+                { path: '/echo/other', body: {}, headers: { 'x-public-key': 'public-key', 'x-signature': 'signature', 'x-timestamp': '1000' } },
-    ).rejects.toMatchObject({ statusCode: 404 });
+                connection,
-  });
+            )).rejects.toMatchObject({ statusCode: 404 });
+    });
 
-  it("preserves correlation when concurrent requests finish out of order", async () => {
+    it('preserves correlation when concurrent requests finish out of order', async (): Promise<void> => {
-    const completions = new Map<string, () => void>();
+        const router = await ApplicationRouter.create({ auth }, [
-    const router = await ApplicationRouter.create([
+            toRoutes([
-      moduleWith([
+                {
-        {
+                    url: '/delayed',
-          url: "/delayed",
+                    handler: async (stream): Promise<void> => {
-          handler: async (stream) => {
+                        const { key, signalStarted, released } = stream.body as {
-            const key = (stream.body as { key: string }).key;
+                            key: string;
-            await new Promise<void>((resolve) => completions.set(key, resolve));
+                            signalStarted: () => void;
-            await stream.send({ key });
+                            released: Promise<void>;
-          },
+                        };
-        },
-      ]),
-    ]);
-    const connection = new TestConnection(true, true);
 
-    const first = router.dispatch(
+                        signalStarted();
-      { path: "/delayed", body: { key: "A" }, requestId: "A" },
-      connection,
-    );
-    const second = router.dispatch(
-      { path: "/delayed", body: { key: "B" }, requestId: "B" },
-      connection,
-    );
 
-    completions.get("B")?.();
+                        await released;
-    await second;
+                        await stream.send({ key });
-    completions.get("A")?.();
+                    },
-    await first;
+                },
+            ]),
+        ]);
 
-    expect(connection.messages).toEqual([
+        const connection = new TestConnection(false, false);
-      {
-        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 first = createControlledRequest({ router, connection, path: '/delayed', requestId: 'A', body: { key: 'A' } });
-    const error = new Error("route failed");
+        const second = createControlledRequest({ router, connection, path: '/delayed', requestId: 'B', body: { key: 'B' } });
-    const router = await ApplicationRouter.create([
-      moduleWith([
-        {
-          url: "/failure",
-          handler: () => {
-            throw error;
-          },
-        },
-      ]),
-    ]);
 
-    await expect(
+        expect(connection.messages).toEqual([]);
-      router.dispatch({ path: "/failure" }, new TestConnection(false, false)),
+
-    ).rejects.toBe(error);
+        await Promise.all([ first.started, second.started ]);
-  });
+
+        second.release();
+        await second.request;
+
+        expect(connection.messages).toEqual([
+            {
+                id: 'B',
+                type: 'response',
+                statusCode: 200,
+                body: { key: 'B' },
+            },
+        ]);
+
+        first.release();
+        await first.request;
+
+        expect(connection.messages).toEqual([
+            {
+                id: 'B',
+                type: 'response',
+                statusCode: 200,
+                body: { key: 'B' },
+            },
+            {
+                id: 'A',
+                type: 'response',
+                statusCode: 200,
+                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 }, [
+            toRoutes([
+                {
+                    url: '/failure',
+                    handler: (): void => {
+                        throw error;
+                    },
+                },
+            ]),
+        ]);
+
+        await expect(router.dispatch(
+                { path: '/failure', headers: { 'x-public-key': 'public-key', 'x-signature': 'signature', 'x-timestamp': '1000' } },
+                new TestConnection(false, false),
+            )).rejects.toBe(error);
+    });
 });

test/services/stream/stream.test.ts

@@ -1,62 +1,58 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
 
-import { HonoSSEStream } from "../../../source/services/stream/hono-sse-stream.js";
+import { HonoSSEStream } from '../../../source/services/stream/hono-sse-stream.ts';
-import { HttpRequestStream } from "../../../source/services/stream/http-request-stream.js";
+import { HttpRequestStream } from '../../../source/services/stream/http-request-stream.ts';
-import { WSStream } from "../../../source/services/stream/ws-stream.js";
+import { WSStream } from '../../../source/services/stream/ws-stream.ts';
 
-describe("stream lifecycle observers", () => {
+describe('stream lifecycle observers', (): void => {
-  it("buffers exactly one normal HTTP response", async () => {
+    it('buffers exactly one normal HTTP response', async (): Promise<void> => {
-    const stream = new HttpRequestStream();
+        const stream = new HttpRequestStream();
 
-    await stream.send({
+        await stream.send({
-      type: "response",
+            type: 'response',
-      statusCode: 200,
+            statusCode: 200,
-      body: { ok: true },
+            body: { ok: true },
+        });
+
+        expect(stream.getResponse()).toEqual({
+            type: 'response',
+            statusCode: 200,
+            body: { ok: true },
+        });
+        await expect(stream.send({
+                type: 'response',
+                statusCode: 200,
+                body: { second: true },
+            })).rejects.toThrow('only send one response');
     });
 
-    expect(stream.getResponse()).toEqual({
+    it('notifies WebSocket observers registered after remote closure', (): void => {
-      type: "response",
+        const stream = new WSStream({
-      statusCode: 200,
+            send: vi.fn(),
-      body: { ok: true },
+            close: vi.fn(),
+            readyState: 1,
+        });
+        const onClose = vi.fn();
+
+        stream.markClosed();
+        stream.onClose(onClose);
+
+        expect(onClose).toHaveBeenCalledOnce();
     });
-    await expect(
-      stream.send({
-        type: "response",
-        statusCode: 200,
-        body: { second: true },
-      }),
-    ).rejects.toThrow("only send one response");
-  });
 
-  it("notifies WebSocket observers registered after remote closure", () => {
+    it('notifies SSE observers registered after local closure', async (): Promise<void> => {
-    const stream = new WSStream({
+        const streamApi = {
-      send: vi.fn(),
+            writeSSE: vi.fn(),
-      close: vi.fn(),
+            close: vi.fn(),
-      readyState: 1,
+        };
+        const stream = new HonoSSEStream(streamApi as unknown as ConstructorParameters<typeof HonoSSEStream>[0]);
+        const onClose = vi.fn();
+
+        await stream.close();
+        stream.onClose(onClose);
+        await stream.close();
+
+        expect(streamApi.close).toHaveBeenCalledOnce();
+        expect(onClose).toHaveBeenCalledOnce();
     });
-    const onClose = vi.fn();
-
-    stream.markClosed();
-    stream.onClose(onClose);
-
-    expect(onClose).toHaveBeenCalledOnce();
-  });
-
-  it("notifies SSE observers registered after local closure", () => {
-    const streamApi = {
-      writeSSE: vi.fn(),
-      close: vi.fn(),
-    };
-    const stream = new HonoSSEStream(
-      streamApi as unknown as ConstructorParameters<typeof HonoSSEStream>[0],
-    );
-    const onClose = vi.fn();
-
-    stream.close();
-    stream.onClose(onClose);
-    stream.close();
-
-    expect(streamApi.close).toHaveBeenCalledOnce();
-    expect(onClose).toHaveBeenCalledOnce();
-  });
 });

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

@@ -1,242 +1,257 @@
-import { Hono } from "hono";
+import { Hono } from 'hono';
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
 
-import type { RouteDefinition } from "../../../source/routes/types.js";
+import type { RouteDefinition } from '../../../source/routes/types.ts';
-import { ApplicationError } from "../../../source/errors/index.js";
+import { ApplicationError } from '../../../source/errors/index.ts';
-import { ApplicationRouter } from "../../../source/services/router.js";
+import { ApplicationRouter } from '../../../source/services/router.ts';
-import { Broadcaster } from "../../../source/services/broadcaster.js";
+import { Broadcaster } from '../../../source/services/broadcaster.ts';
-import { HttpTransportRouter } from "../../../source/services/transport/http-transport.js";
+import { HttpTransportRouter } from '../../../source/services/transport/http-transport.ts';
-import type { AppEnv } from "../../../source/services/transport/transport-router.js";
+import type { AppEnv } from '../../../source/services/transport/transport-router.ts';
-import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils";
+import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
-import { Logger } from "../../../source/utils/logger.js";
+import { Logger } from '../../../source/utils/logger.ts';
-import { ServerHost } from "../../../source/services/server-host.js";
+import { ServerHost } from '../../../source/services/server-host.ts';
+import { AuthSecp256k1 } from '../../../source/auth/auth.ts';
 
-async function createApp(
+/**
-  routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
+ * A mock of the AuthSecp256k1 service
-  maxRequestBodyBytes = 1024 * 1024,
+ */
-): Promise<Hono<AppEnv>> {
+const auth = {
-  const debug = new Logger("http-transport-test");
+    verifySignature: vi.fn().mockResolvedValue(true),
-  const broadcaster = new Broadcaster(debug);
+    verifyUniqueRequest: vi.fn().mockResolvedValue(true),
-  const resolvedRoutes =
+    assertTimestampFreshness: vi.fn().mockResolvedValue(true),
-    typeof routes === "function" ? routes(broadcaster) : routes;
+} as unknown as AuthSecp256k1;
-  const router = await ApplicationRouter.create([
-    {
-      async getRoutes() {
-        return resolvedRoutes;
-      },
-    },
-  ]);
-  const transport = new HttpTransportRouter(router, debug);
-  const app = new Hono<AppEnv>();
 
-  app.onError(HttpTransportRouter.createErrorHandler(debug));
+const mockAuthHeaders = {
-  app.use("*", ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
+    'x-public-key': 'public-key',
-  app.use("*", HttpTransportRouter.createExtJsonMiddleware(debug));
+    'x-signature': 'signature',
-  transport.register(app);
+    'x-timestamp': '1000',
-  return app;
+};
-}
 
-describe("HttpTransportRouter", () => {
+const createApp = async (
-  it("runs normal HTTP through a non-streaming route stream", async () => {
+    routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
-    const app = await createApp([
+    maxRequestBodyBytes = 1024 * 1024,
-      {
+): Promise<Hono<AppEnv>> => {
-        url: "/echo",
+    const debug = new Logger('http-transport-test');
-        handler: async (stream) => stream.send(stream.body),
+    const broadcaster = new Broadcaster(debug);
-      },
+    const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes;
-    ]);
+    const router = await ApplicationRouter.create({
-    const value = new Uint8Array([1, 2, 3]);
+        auth,
-
+    }, [
-    const response = await app.request("/echo", {
-      method: "POST",
-      headers: { "content-type": "application/json" },
-      body: toExtendedJson({ value }),
-    });
-
-    expect(response.status).toBe(200);
-    expect(fromExtendedJson(await response.text())).toEqual({ value });
-  });
-
-  it("returns 204 when a normal HTTP route sends nothing", async () => {
-    const app = await createApp([
-      {
-        url: "/nothing",
-        handler: () => undefined,
-      },
-    ]);
-
-    const response = await app.request("/nothing", { method: "POST" });
-
-    expect(response.status).toBe(204);
-    expect(await response.text()).toBe("");
-  });
-
-  it("returns normalized errors for non-streaming requests", async () => {
-    const app = await createApp([]);
-
-    const missing = await app.request("/missing", { method: "POST" });
-    expect(missing.status).toBe(404);
-    expect(await missing.json()).toEqual({
-      statusCode: 404,
-      error: "No route found for /missing",
-    });
-
-    const invalid = await app.request("/missing", {
-      method: "POST",
-      headers: { "content-type": "application/json" },
-      body: "{",
-    });
-    expect(invalid.status).toBe(400);
-    expect(await invalid.json()).toEqual({
-      statusCode: 400,
-      error: "Invalid JSON in request body",
-    });
-  });
-
-  it("rejects subscribe when normal HTTP has no streaming capability", async () => {
-    const app = await createApp((broadcaster) => [
-      {
-        url: "/items/subscribe",
-        handler: async (stream) => {
-          await broadcaster.subscribe(stream, ["items"]);
-        },
-      },
-    ]);
-
-    const response = await app.request("/items/subscribe", { method: "POST" });
-
-    expect(response.status).toBe(406);
-    expect(await response.json()).toMatchObject({ statusCode: 406 });
-  });
-
-  it("sends SSE route errors as events and closes only that stream", async () => {
-    const app = await createApp([
-      {
-        url: "/items/subscribe",
-        handler: () => {
-          throw new Error("private storage failure");
-        },
-      },
-    ]);
-
-    const response = await app.request("/items/subscribe", {
-      method: "POST",
-      headers: { accept: "text/event-stream" },
-    });
-    const events = await response.text();
-
-    expect(response.status).toBe(200);
-    expect(events).toContain("event: error");
-    expect(events).toContain(
-      'data: {"statusCode":500,"error":"Internal Server Error"}',
-    );
-    expect(events).not.toContain("private storage failure");
-  });
-
-  it("sends a normal route as one SSE response event and then closes", async () => {
-    const app = await createApp([
-      {
-        url: "/echo",
-        handler: (stream) => stream.send({ ok: true }),
-      },
-    ]);
-
-    const response = await app.request("/echo", {
-      method: "POST",
-      headers: { accept: "text/event-stream" },
-    });
-    const events = await response.text();
-
-    expect(response.status).toBe(200);
-    expect(events).toContain("event: response");
-    expect(events).toContain('data: {"ok":true}');
-  });
-
-  it("keeps SSE open until the route's subscription promise resolves", async () => {
-    let removeSubscription: () => Promise<void> = async () => undefined;
-    let markSubscribed: () => void = () => undefined;
-    const subscribed = new Promise<void>((resolve) => {
-      markSubscribed = resolve;
-    });
-    const app = await createApp((broadcaster) => [
-      {
-        url: "/items/subscribe",
-        handler: async (stream) => {
-          const topics = ["items"];
-          removeSubscription = () => broadcaster.unsubscribe(stream, topics);
-
-          const removed = broadcaster.subscribe(stream, topics);
-          markSubscribed();
-          await removed;
-        },
-      },
-    ]);
-
-    const response = await app.request("/items/subscribe", {
-      method: "POST",
-      headers: { accept: "text/event-stream" },
-    });
-    const body = response.text();
-    const completed = vi.fn();
-    void body.then(completed);
-
-    await subscribed;
-    await Promise.resolve();
-    expect(completed).not.toHaveBeenCalled();
-
-    await removeSubscription();
-
-    expect(await body).toBe("");
-    expect(completed).toHaveBeenCalledOnce();
-  });
-
-  it("rejects unsubscribe over non-bidirectional SSE", async () => {
-    const app = await createApp((broadcaster) => [
-      {
-        url: "/items/unsubscribe",
-        handler: async (stream) => {
-          if (!stream.bidirectional) {
-            throw new ApplicationError(
-              400,
-              "This route requires an existing bidirectional stream",
-            );
-          }
-
-          await broadcaster.unsubscribe(stream, ["items"]);
-        },
-      },
-    ]);
-
-    const response = await app.request("/items/unsubscribe", {
-      method: "POST",
-      headers: { accept: "text/event-stream" },
-    });
-    const events = await response.text();
-
-    expect(response.status).toBe(200);
-    expect(events).toContain("event: error");
-    expect(events).toContain('"statusCode":400');
-  });
-  it("rejects HTTP bodies larger than the configured byte limit", async () => {
-    const app = await createApp(
-      [
         {
-          url: "/echo",
+            async getRoutes(): Promise<RouteDefinition[]> {
-          handler: async (stream) => stream.send(stream.body),
+                return resolvedRoutes;
+            },
         },
-      ],
+    ]);
-      32,
+    const transport = new HttpTransportRouter(router, debug);
-    );
+    const app = new Hono<AppEnv>();
 
-    const response = await app.request("/echo", {
+    app.onError(HttpTransportRouter.createErrorHandler(debug));
-      method: "POST",
+    app.use('*', ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
-      headers: { "content-type": "application/json" },
+    app.use('*', HttpTransportRouter.createExtJsonMiddleware(debug));
-      body: JSON.stringify({ value: "x".repeat(64) }),
+    transport.register(app);
+
+    return app;
+};
+
+describe('HttpTransportRouter', (): void => {
+    it('runs normal HTTP through a non-streaming route stream', async (): Promise<void> => {
+        const app = await createApp([
+            {
+                url: '/echowtf',
+                handler: async (stream): Promise<void> => stream.send(stream.body),
+            },
+        ]);
+        const value = new Uint8Array([ 1, 2, 3 ]);
+
+        const request = new Request('http://localhost/echowtf', {
+            method: 'POST',
+            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 });
     });
 
-    expect(response.status).toBe(413);
+    it('returns 204 when a normal HTTP route sends nothing', async (): Promise<void> => {
-    expect(await response.json()).toEqual({
+        const app = await createApp([
-      statusCode: 413,
+            {
-      error: "Request body exceeds the 32 byte limit",
+                url: '/nothing',
+                handler: (): void => undefined,
+            },
+        ]);
+
+        const response = await app.request('/nothing', { method: 'POST', headers: { ...mockAuthHeaders } });
+
+        expect(response.status).toBe(204);
+        expect(await response.text()).toBe('');
+    });
+
+    it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
+        const app = await createApp([]);
+
+        const missing = await app.request('/missing', { method: 'POST', headers: { ...mockAuthHeaders } });
+        expect(missing.status).toBe(404);
+        expect(await missing.json()).toEqual({
+            statusCode: 404,
+            error: 'No route found for /missing',
+        });
+
+        const invalid = await app.request('/missing', {
+            method: 'POST',
+            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
+            body: '{',
+        });
+        expect(invalid.status).toBe(400);
+        expect(await invalid.json()).toEqual({
+            statusCode: 400,
+            error: 'Invalid JSON in request body',
+        });
+    });
+
+    it('rejects subscribe when normal HTTP has no streaming capability', async (): Promise<void> => {
+        const app = await createApp((broadcaster) => [
+            {
+                url: '/items/subscribe',
+                handler: async (stream): Promise<void> => {
+                    await broadcaster.subscribe(stream, [ 'items' ]);
+                },
+            },
+        ]);
+
+        const response = await app.request('/items/subscribe', { method: 'POST', headers: { ...mockAuthHeaders } });
+
+        expect(response.status).toBe(406);
+        expect(await response.json()).toMatchObject({ statusCode: 406 });
+    });
+
+    it('sends SSE route errors as events and closes only that stream', async (): Promise<void> => {
+        const app = await createApp([
+            {
+                url: '/items/subscribe',
+                handler: (): void => {
+                    throw new Error('private storage failure');
+                },
+            },
+        ]);
+
+        const response = await app.request('/items/subscribe', {
+            method: 'POST',
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
+        });
+        const events = await response.text();
+
+        expect(response.status).toBe(200);
+        expect(events).toContain('event: error');
+        expect(events).toContain('data: {"statusCode":500,"error":"Internal Server Error"}');
+        expect(events).not.toContain('private storage failure');
+    });
+
+    it('sends a normal route as one SSE response event and then closes', async (): Promise<void> => {
+        const app = await createApp([
+            {
+                url: '/echo',
+                handler: (stream): Promise<void> => stream.send({ ok: true }),
+            },
+        ]);
+
+        const response = await app.request('/echo', {
+            method: 'POST',
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
+        });
+        const events = await response.text();
+
+        expect(response.status).toBe(200);
+        expect(events).toContain('event: response');
+        expect(events).toContain('data: {"ok":true}');
+    });
+
+    it("keeps SSE open until the route's subscription promise resolves", async (): Promise<void> => {
+        let removeSubscription: () => Promise<void> = async () => undefined;
+        let markSubscribed: () => void = () => undefined;
+        const subscribed = new Promise<void>((resolve) => {
+            markSubscribed = resolve;
+        });
+        const app = await createApp((broadcaster) => [
+            {
+                url: '/items/subscribe',
+                handler: async (stream): Promise<void> => {
+                    const topics = [ 'items' ];
+                    removeSubscription = (): Promise<void> => broadcaster.unsubscribe(stream, topics);
+
+                    const removed = broadcaster.subscribe(stream, topics);
+                    markSubscribed();
+                    await removed;
+                },
+            },
+        ]);
+
+        const response = await app.request('/items/subscribe', {
+            method: 'POST',
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
+        });
+        const body = response.text();
+        const completed = vi.fn();
+        void body.then(completed);
+
+        await subscribed;
+        await Promise.resolve();
+        expect(completed).not.toHaveBeenCalled();
+
+        await removeSubscription();
+
+        expect(await body).toBe('');
+        expect(completed).toHaveBeenCalledOnce();
+    });
+
+    it('rejects unsubscribe over non-bidirectional SSE', async (): Promise<void> => {
+        const app = await createApp((broadcaster) => [
+            {
+                url: '/items/unsubscribe',
+                handler: async (stream): Promise<void> => {
+                    if (!stream.bidirectional) {
+                        throw new ApplicationError(400, 'This route requires an existing bidirectional stream');
+                    }
+
+                    await broadcaster.unsubscribe(stream, [ 'items' ]);
+                },
+            },
+        ]);
+
+        const response = await app.request('/items/unsubscribe', {
+            method: 'POST',
+            headers: { accept: 'text/event-stream', ...mockAuthHeaders },
+        });
+        const events = await response.text();
+
+        expect(response.status).toBe(200);
+        expect(events).toContain('event: error');
+        expect(events).toContain('"statusCode":400');
+    });
+    it('rejects HTTP bodies larger than the configured byte limit', async (): Promise<void> => {
+        const app = await createApp(
+            [
+                {
+                    url: '/echo',
+                    handler: async (stream): Promise<void> => stream.send(stream.body),
+                },
+            ],
+            32,
+        );
+
+        const response = await app.request('/echo', {
+            method: 'POST',
+            headers: { 'content-type': 'application/json', ...mockAuthHeaders },
+            body: JSON.stringify({ value: 'x'.repeat(64) }),
+        });
+
+        expect(response.status).toBe(413);
+        expect(await response.json()).toEqual({
+            statusCode: 413,
+            error: 'Request body exceeds the 32 byte limit',
+        });
     });
-  });
 });

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

@@ -1,58 +1,68 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from 'vitest';
-import { z } from "zod";
+import { z } from 'zod';
-import { WebSocketServer } from "ws";
 
-import { ApplicationRouter } from "../../../source/services/router.js";
+import { ApplicationRouter } from '../../../source/services/router.ts';
-import {
+import { WsTransportRouter } from '../../../source/services/transport/ws-transport.ts';
-  WsTransportRouter,
+import { Logger } from '../../../source/utils/logger.ts';
-} from "../../../source/services/transport/ws-transport.js";
+import { toExtendedJson } from '@xo-cash/utils';
-import { Logger } from "../../../source/utils/logger.js";
+import { AuthSecp256k1 } from '../../../source/auth/auth.ts';
-import { toExtendedJson } from "@xo-cash/utils";
+import { toRoutes } from '../../helpers/misc.ts';
 
-describe("WebSocket request decoding", () => {
+/**
-  it("decodes the minimal route-agnostic envelope and Extended JSON body", async () => {
+ * A mock of the AuthSecp256k1 service
-    await expect(
+ */
-      WsTransportRouter.decodeWebSocketRequest(
+const auth = {
-        toExtendedJson({
+    verifySignature: vi.fn().mockResolvedValue(true),
-          id: "request-1",
+    verifyUniqueRequest: vi.fn().mockResolvedValue(true),
-          path: "/data/write",
+    assertTimestampFreshness: vi.fn().mockResolvedValue(true),
-          body: { value: new Uint8Array([1, 2, 3]) },
+} as unknown as AuthSecp256k1;
-        }),
+
-      ),
+describe('WebSocket request decoding', (): void => {
-    ).resolves.toEqual({
+    it('decodes the minimal route-agnostic envelope and Extended JSON body', async (): Promise<void> => {
-      requestId: "request-1",
+        await expect(WsTransportRouter.decodeWebSocketRequest(toExtendedJson({
-      path: "/data/write",
+                    id: 'request-1',
-      body: { value: new Uint8Array([1, 2, 3]) },
+                    path: '/data/write',
+                    body: { value: new Uint8Array([ 1, 2, 3 ]) },
+                }))).resolves.toEqual({
+            requestId: 'request-1',
+            path: '/data/write',
+            body: { value: new Uint8Array([ 1, 2, 3 ]) },
+        });
     });
-  });
 
-  it.each([
+    it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])(
-    "{}",
+        'rejects an invalid envelope: %s',
-    '{"path":42}',
+        async (payload) => {
-    '{"path":"/data/get","id":1}',
+            await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
-    '{"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 () => {
+    it('rejects malformed JSON', async (): Promise<void> => {
-    await expect(WsTransportRouter.decodeWebSocketRequest("{")).rejects.toMatchObject({
+        await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({
-      statusCode: 400,
+            statusCode: 400,
-      message: "Invalid JSON in WebSocket message",
+            message: 'Invalid JSON in WebSocket message',
+        });
     });
-  });
 });
 
-describe("WsTransportRouter payload limits", () => {
+describe('WsTransportRouter payload limits', (): void => {
-  it("configures Hono's ws server with the requested maxPayload", async () => {
+    it("configures Hono's ws server with the requested maxPayload", async () => {
-    const debug = new Logger("ws-transport-test");
+        const debug = new Logger('ws-transport-test');
-    const router = await ApplicationRouter.create([]);
+        const router = await ApplicationRouter.create({
-    const transport = new WsTransportRouter(router, debug, 1024);
+            auth,
-    const wsServer = transport.websocketServer as unknown as WebSocketServer;
+        }, [
+            toRoutes([
+                {
+                    url: '/data/write',
+                    handler: async (stream): Promise<void> => {
+                        await stream.send({});
+                    },
+                },
+            ]),
+        ]);
+        const transport = new WsTransportRouter(router, debug, 1024);
 
-    expect(wsServer.options.maxPayload).toBe(1024);
+        expect(transport['wsServer'].options.maxPayload).toBe(1024);
-    await transport.stop();
+        await transport.stop();
-  });
+    });
 });

test/subscription-flow.test.ts

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

tsconfig.json

@@ -2,8 +2,8 @@
     "compilerOptions": {
         "rootDir": "./source",
         "outDir": "./dist",
-        "module": "es2022",
+        "module": "esnext",
-        "target": "es2022",
+        "target": "esnext",
         "skipLibCheck": true,
         "strict": true,
         "moduleResolution": "bundler",
@@ -15,5 +15,5 @@
         "declarationMap": true,
         "types": ["node"]
     },
-    "exclude": ["node_modules/**/*", "dist/**/*", "test"]
+    "exclude": ["node_modules/**/*", "dist/**/*", "test/**/*"]
 }