Added auth and request storage

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/**'",

source/auth/accounts.ts

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

source/auth/auth.ts

@@ -0,0 +1,110 @@
+import type { Database } from '../services/storage/database';
+
+import { hexToBin, instantiateSecp256k1, type Secp256k1, sha256 } from '@bitauth/libauth';
+import { UnauthorizedError } from '../errors/unauthorized-error';
+
+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: 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.
+   */
+
+  /**
+   * @param deps - The dependencies to use
+   * @param options - The options to use
+   */
+  private constructor(private readonly deps: Required<AuthSecp256k1Deps>, private readonly options: AuthSecp256k1Options) {}
+
+  /**
+   * 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> {
+    return true;
+    // 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 = await 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;
+  }
+
+  async verifyUniqueRequest(signature: string): Promise<boolean> {
+    if (!signature) {
+      throw new UnauthorizedError('Signature is required');
+    }
+
+    console.log('Verifying unique request', signature);
+
+    // Check if the signature has been used before
+    const request = await this.deps.database.db.
+      selectFrom('authed_requests')
+      .selectAll()
+      .where('signature', '=', signature)
+      .executeTakeFirst();
+    
+    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 {
+    const age = Math.abs(Date.now() - timestamp);
+    if (age > windowMs) {
+      throw new UnauthorizedError('Timestamp outside allowed window');
+    }
+  }
+}

source/constants.ts

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

source/index.ts

@@ -1,12 +1,16 @@
 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';
 import { WsTransportRouter } from './services/transport/ws-transport.ts';
 import { ServerHost } from './services/server-host.ts';
 import { Logger } from './utils/logger.ts';
+
 import { DataRoute } from './routes/resources.ts';
+import { AccountRoute } from './routes/account.ts';
+import { Accounts } from './auth/accounts.ts';
 
 /** Application composition root. */
 export class App {
@@ -23,18 +27,23 @@ 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 accounts = new Accounts();
         const routes = [
             // DataRoute owns resource read/write/subscribe logic and maps resource
             // ids to broadcaster topics. timestampWindowMs controls write replay protection.
-            new DataRoute(database, broadcaster, config.auth.timestampWindowMs),
+            new DataRoute(database, broadcaster, auth, accounts),
+            // new AccountRoute(database, auth),
         ];
 
         // 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.
@@ -43,7 +52,13 @@ export class App {
         const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes);
         const host = new ServerHost(config, debug, [ http, ws ]);
 
-        return new App(host, database);
+        // Create the app instance
+        const app = new App(host, database);
+        
+        // Start the unique request cleanup interval
+        app.startUniqueRequestCleanup(config.auth.uniqueRequestCleanupIntervalMs, config.auth.timestampWindowMs);
+        
+        return app;
     }
 
     private stopPromise: Promise<void> | undefined;
@@ -69,6 +84,13 @@ 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/account.ts

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

source/routes/resources.ts

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

source/routes/types.ts

@@ -1,5 +1,8 @@
 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. */
@@ -20,7 +23,11 @@ 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;
 

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,7 +1,11 @@
-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 = {
@@ -12,14 +16,32 @@ export type ApplicationRequest = {
     /** 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.
@@ -27,7 +49,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.
@@ -42,7 +64,7 @@ export class ApplicationRouter {
             }
         }
 
-        return new ApplicationRouter(routes);
+        return new ApplicationRouter(deps, routes);
     }
 
     /**
@@ -52,12 +74,36 @@ export class ApplicationRouter {
      * @param connection - Shared connection stream for this transport session.
      */
     async dispatch(request: ApplicationRequest, connection: BaseStream): Promise<void> {
+        const { publicKey, signature, timestamp } = accountSchema.parse(request.headers);
+        // Authenticate the headers on the request. (TODO: Remove the defaults, just here for testing)
+        // const publicKey = request.headers?.['x-public-key'] || 'public-key';
+        // const signature = request.headers?.['x-signature'] || 'signature';
+        // const timestamp = request.headers?.['x-timestamp'] || Date.now();
+
+        // 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/storage/migrations/001-resources.ts

@@ -23,8 +23,19 @@ 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))
+        .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();
 };
 
 /**
@@ -35,4 +46,5 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
 export const down = async (db: Kysely<DatabaseTables>): Promise<void> => {
     await db.schema.dropTable('resource_data').ifExists()
 .execute();
+    await db.schema.dropTable('authed_requests').ifExists().execute();
 };

source/services/storage/tables.ts

@@ -22,9 +22,44 @@ 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;
+}
+
+// export interface PaymentsTable {
+//     /** Unique identifier for the payment. */
+//     payment_id: string;
+
+//     /** Public key of the account in the transaction */
+//     public_key: string;
+
+//     /** Amount of the payment. This can be positive or negative.*/
+//     amount: number;
+
+//     /** Timestamp of the payment. */
+//     timestamp: Timestamp;
+
+//     /** Signature of the payment. */
+//     signature: string;
+
+//     /** Hash of the message that was signed. */
+//     message_hash: string;
+
+//     /** Resource ID of the payment. */
+//     resource_id: string;
+// }
+
 /** Complete Kysely schema mapping for the sync server database. */
 export interface DatabaseTables {
     resource_data: ResourceDataTable;
+    authed_requests: AuthedRequestsTable;
 }

source/services/transport/http-transport.ts

@@ -11,6 +11,7 @@ import { ApplicationError, normalizePublicError } from '../../errors/index.ts';
 import { HTTP_STATUS_CODE_BAD_REQUEST, HTTP_STATUS_CODE_NO_CONTENT } from '../../constants.ts';
 import { HonoSSEStream } from '../stream/hono-sse-stream.ts';
 import { HttpRequestStream } from '../stream/http-request-stream.ts';
+import { normalizeRequestHeaders } from './request-headers.ts';
 
 /** Hono context key where decoded Extended JSON bodies are stored. */
 const PARSED_BODY_KEY = 'parsedBody';
@@ -110,6 +111,7 @@ export class HttpTransportRouter implements TransportRouter {
 
         return {
             path: context.req.path,
+            headers: normalizeRequestHeaders(context.req.header()),
             ...(body === undefined ? {} : { body }),
         };
     }

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/ws-transport.ts

@@ -13,6 +13,7 @@ import type { AppEnv, UpgradeTransportRouter } from './transport-router.ts';
 import { ApplicationError, normalizePublicError } from '../../errors/index.ts';
 import { HTTP_STATUS_CODE_BAD_REQUEST } from '../../constants.ts';
 import { WSStream } from '../stream/ws-stream.ts';
+import { normalizeRequestHeaders } from './request-headers.ts';
 
 /** Default WebSocket upgrade path for application messages. */
 const WS_ROUTE = '/ws';
@@ -24,6 +25,7 @@ const wsRequestSchema = z
 .optional(),
         path: z.string().min(1),
         body: z.unknown().optional(),
+        headers: z.record(z.string(), z.string()).optional(),
     })
     .strict();
 
@@ -208,6 +210,7 @@ export class WsTransportRouter implements UpgradeTransportRouter {
             path: envelope.path,
             ...(envelope.id === undefined ? {} : { requestId: envelope.id }),
             ...(envelope.body === undefined ? {} : { body: envelope.body }),
+            ...(envelope.headers === undefined ? {} : { headers: normalizeRequestHeaders(envelope.headers) }),
         };
     }
 

test/routes/resources.test.ts

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

test/services/broadcaster.test.ts

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

test/services/router.test.ts

@@ -33,13 +33,15 @@ describe('ApplicationRouter dispatch', (): void => {
                     url: '/echo',
                     handler: async (stream): Promise<void> => {
                         expect(stream.connection).toBe(connection);
+                        expect(stream.path).toBe('/echo');
+                        expect(stream.headers).toEqual({ 'x-request-token': 'route-1' });
                         await stream.send(stream.body);
                     },
                 },
             ]),
         ]);
 
-        await router.dispatch({ path: '/echo', body: { value: 1 }, requestId: 'request-1' }, connection);
+        await router.dispatch({ path: '/echo', body: { value: 1 }, requestId: 'request-1', headers: { 'x-request-token': 'route-1' } }, connection);
 
         expect(connection.messages).toEqual([
             {
@@ -61,16 +63,23 @@ describe('ApplicationRouter dispatch', (): void => {
                     url: '/delayed',
                     handler: async (stream): Promise<void> => {
                         const key = (stream.body as { key: string }).key;
+                        const token = stream.headers['x-request-token'];
                         await new Promise<void>((resolve) => completions.set(key, resolve));
-                        await stream.send({ key });
+                        await stream.send({ key, token });
                     },
                 },
             ]),
         ]);
         const connection = new TestConnection(true, true);
 
-        const first = router.dispatch({ path: '/delayed', body: { key: 'A' }, requestId: 'A' }, connection);
+        const first = router.dispatch(
-        const second = router.dispatch({ path: '/delayed', body: { key: 'B' }, requestId: 'B' }, connection);
+            { path: '/delayed', body: { key: 'A' }, requestId: 'A', headers: { 'x-request-token': 'token-a' } },
+            connection,
+        );
+        const second = router.dispatch(
+            { path: '/delayed', body: { key: 'B' }, requestId: 'B', headers: { 'x-request-token': 'token-b' } },
+            connection,
+        );
 
         completions.get('B')?.();
         await second;
@@ -82,13 +91,13 @@ describe('ApplicationRouter dispatch', (): void => {
                 id: 'B',
                 type: 'response',
                 statusCode: 200,
-                body: { key: 'B' },
+                body: { key: 'B', token: 'token-b' },
             },
             {
                 id: 'A',
                 type: 'response',
                 statusCode: 200,
-                body: { key: 'A' },
+                body: { key: 'A', token: 'token-a' },
             },
         ]);
     });

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

@@ -70,6 +70,23 @@ describe('HttpTransportRouter', (): void => {
         expect(await response.text()).toBe('');
     });
 
+    it('passes normalized HTTP request headers to the route stream', async (): Promise<void> => {
+        const app = await createApp([
+            {
+                url: '/headers',
+                handler: async (stream): Promise<void> => stream.send({ path: stream.path, token: stream.headers['x-request-token'] }),
+            },
+        ]);
+
+        const response = await app.request('/headers', {
+            method: 'POST',
+            headers: { 'X-Request-Token': 'http-token' },
+        });
+
+        expect(response.status).toBe(200);
+        expect(await response.json()).toEqual({ path: '/headers', token: 'http-token' });
+    });
+
     it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
         const app = await createApp([]);
 
@@ -149,6 +166,27 @@ describe('HttpTransportRouter', (): void => {
         expect(events).toContain('data: {"ok":true}');
     });
 
+    it('passes normalized HTTP request headers to an SSE route stream', async (): Promise<void> => {
+        const app = await createApp([
+            {
+                url: '/headers',
+                handler: (stream): Promise<void> => stream.send({ token: stream.headers['x-request-token'] }),
+            },
+        ]);
+
+        const response = await app.request('/headers', {
+            method: 'POST',
+            headers: {
+                accept: 'text/event-stream',
+                'X-Request-Token': 'sse-token',
+            },
+        });
+        const events = await response.text();
+
+        expect(response.status).toBe(200);
+        expect(events).toContain('data: {"token":"sse-token"}');
+    });
+
     it("keeps SSE open until the route's subscription promise resolves", async (): Promise<void> => {
         let removeSubscription: () => Promise<void> = async () => undefined;
         let markSubscribed: () => void = () => undefined;

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

@@ -13,19 +13,36 @@ describe('WebSocket request decoding', (): void => {
                     id: 'request-1',
                     path: '/data/write',
                     body: { value: new Uint8Array([ 1, 2, 3 ]) },
+                    headers: { 'X-Request-Token': 'ws-token' },
                 }))).resolves.toEqual({
             requestId: 'request-1',
             path: '/data/write',
             body: { value: new Uint8Array([ 1, 2, 3 ]) },
+            headers: { 'x-request-token': 'ws-token' },
         });
     });
 
-    it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])(
+    it('allows the optional headers object to be omitted', async (): Promise<void> => {
-        'rejects an invalid envelope: %s',
+        await expect(WsTransportRouter.decodeWebSocketRequest('{"path":"/data/get"}')).resolves.toEqual({ path: '/data/get' });
-        async (payload) => {
+    });
-            await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
+
-        },
+    it.each([
-    );
+        '{}',
+        '{"path":42}',
+        '{"path":"/data/get","id":1}',
+        '{"path":"/data/get","method":"POST"}',
+        '{"path":"/data/get","headers":{"x-request-token":1}}',
+    ])('rejects an invalid envelope: %s', async (payload) => {
+        await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
+    });
+
+    it.each([
+        '{"path":"/data/get","headers":{"bad header":"value"}}',
+        '{"path":"/data/get","headers":{"x-request-token":"first","X-Request-Token":"second"}}',
+        '{"path":"/data/get","headers":{"x-request-token":"first\\r\\nsecond"}}',
+    ])('rejects malformed request headers: %s', async (payload): Promise<void> => {
+        await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toMatchObject({ statusCode: 400 });
+    });
 
     it('rejects malformed JSON', async (): Promise<void> => {
         await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({