Add resource routes and signed writes

docs/demo-app-transport-migration.md

@@ -0,0 +1,267 @@
+# Demo App Transport Migration
+
+This guide describes the client-facing protocol for the route-agnostic transport
+architecture.
+
+## Connection Model
+
+HTTP, SSE, and WebSocket do not authenticate at the transport layer. Do not send
+request-auth headers such as `X-PublicKey`, `X-Signature`, or `X-Timestamp`
+for normal route dispatch.
+
+Authentication is scoped to each resource operation. Every written resource
+carries its own `publicKey`, `timestamp`, and `signature`.
+
+All route input belongs in the request body. Paths are exact route names; there
+are no path parameters, query parameters, or alternate HTTP methods.
+
+## Request Size Limit
+
+The server accepts at most 1048576 encoded bytes per HTTP request body or
+complete WebSocket message by default. Deployments can change this with
+`SERVER_MAX_REQUEST_BODY_BYTES`. The count is over the Extended JSON wire text,
+so hexadecimal `Uint8Array` values use approximately twice their decoded byte
+length.
+
+HTTP limits use Hono's built-in `bodyLimit` middleware and return `413`.
+WebSocket messages do not pass through HTTP middleware; Hono's Node adapter uses
+`ws`, whose native `maxPayload` option rejects oversized frames before the
+application message handler. Response bodies are not limited.
+
+## HTTP Requests
+
+All application HTTP routes use POST.
+
+Read:
+
+```http
+POST /data/get
+Content-Type: application/json
+
+{
+  "resourceId": ["resource-a", "resource-b"]
+}
+```
+
+Write:
+
+```http
+POST /data/write
+Content-Type: application/json
+
+{
+  "resources": [
+    {
+      "id": "resource-a",
+      "publicKey": "...hex...",
+      "timestamp": 1730000000000,
+      "signature": "...der-hex...",
+      "value": "<Uint8Array: deadbeef>"
+    }
+  ]
+}
+```
+
+Write response:
+
+```json
+{
+  "resources": [
+    {
+      "id": "resource-a",
+      "publicKey": "...hex...",
+      "blob": "<Uint8Array: deadbeef>",
+      "timestamp": 1730000001000
+    }
+  ]
+}
+```
+
+If any item has invalid authorization, the whole batch fails.
+
+## SSE Subscriptions
+
+The client explicitly requests SSE using `Accept: text/event-stream`:
+
+```http
+POST /data/subscribe
+Accept: text/event-stream
+Content-Type: application/json
+
+{
+  "resourceId": ["resource-a"]
+}
+```
+
+A subscription without this Accept header receives a `406` JSON error.
+
+The server keeps the SSE request open while its resource topics remain
+subscribed. Subscribing sends no current state; it only enables future
+publications. Request current state separately through `/data/get` when needed.
+
+SSE publication:
+
+```text
+id: 1730000001000
+event: instance-changed
+data: {"resourceId":"resource-a"}
+```
+
+After SSE begins, failures arrive as error events and the server closes that SSE
+connection:
+
+```text
+event: error
+data: {"statusCode":500,"error":"Internal Server Error"}
+```
+
+To change an SSE subscription, abort the existing request and open a new
+`/data/subscribe` request with the complete desired `resourceId` list.
+
+## WebSocket
+
+Connect without a connection-auth message:
+
+```ts
+const ws = new WebSocket("ws://host:port/ws");
+```
+
+Every client message uses this strict envelope. Messages may execute
+concurrently; the server binds `id` and `body` to each individual dispatch so
+responses remain correctly correlated even when they finish out of order:
+
+```ts
+type WsRequest = {
+  id?: string;
+  path: string;
+  body?: unknown;
+};
+```
+
+Old fields such as `event`, `type`, `url`, `method`, `headers`,
+`params`, and `data` are rejected.
+
+Read:
+
+```json
+{
+  "id": "read-1",
+  "path": "/data/get",
+  "body": {
+    "resourceId": ["resource-a", "resource-b"]
+  }
+}
+```
+
+Write:
+
+```json
+{
+  "id": "write-1",
+  "path": "/data/write",
+  "body": {
+    "resources": [
+      {
+        "id": "resource-a",
+        "publicKey": "...hex...",
+        "timestamp": 1730000000000,
+        "signature": "...der-hex...",
+        "value": "<Uint8Array: deadbeef>"
+      }
+    ]
+  }
+}
+```
+
+Subscribe:
+
+```json
+{
+  "id": "sub-1",
+  "path": "/data/subscribe",
+  "body": {
+    "resourceId": ["resource-a"]
+  }
+}
+```
+
+The subscribe request sends no immediate message. Its dispatch remains active
+until those topics are removed or the connection closes. Request current state
+separately through `/data/get` when needed.
+
+The client should consolidate its local subscribers and send only the topic set
+needed by the shared WebSocket.
+
+Unsubscribe selected resources without closing the socket:
+
+```json
+{
+  "id": "unsub-1",
+  "path": "/data/unsubscribe",
+  "body": {
+    "resourceId": ["resource-a"]
+  }
+}
+```
+
+Unsubscribe is idempotent. The response is a normal correlated response:
+
+```json
+{
+  "id": "unsub-1",
+  "type": "response",
+  "statusCode": 200,
+  "body": {}
+}
+```
+
+Normal response:
+
+```json
+{
+  "id": "write-1",
+  "type": "response",
+  "statusCode": 200,
+  "body": {}
+}
+```
+
+Error response:
+
+```json
+{
+  "id": "write-1",
+  "type": "error",
+  "statusCode": 400,
+  "error": "Validation Error",
+  "details": []
+}
+```
+
+Message-level errors do not close the WebSocket. The client decides whether to
+retry, alter its subscription, or reconnect.
+
+Published events retain the shared stream shape:
+
+```json
+{
+  "id": "server-event-id",
+  "type": "instance-changed",
+  "data": {}
+}
+```
+
+## Signing
+
+For each written resource, sign this canonical payload:
+
+```ts
+`${timestamp}${resourceId}${canonicalBody(value)}`;
+```
+
+`value` is the `Uint8Array` payload. `canonicalBody(value)` uses the same
+Extended JSON rules as the server:
+
+```json
+"<Uint8Array: deadbeef>"
+```

src/app.ts

@@ -6,6 +6,7 @@ import { HttpTransportRouter } from './services/transport/http-transport.ts';
 import { WsTransportRouter } from './services/transport/ws-transport.ts';
 import { ServerHost } from './services/server-host.ts';
 import { Logger } from './utils/logger.ts';
+import { DataRoute } from './routes/resources.ts';
 
 /** Application composition root. */
 export class App {
@@ -24,13 +25,19 @@ export class App {
 
         // Domain services are shared across all transports and route modules.
         const broadcaster = new Broadcaster(debug);
-        const routes = [];
+        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),
+        ];
 
         // 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);
 
+        // 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]);

src/routes/resources.ts

@@ -0,0 +1,302 @@
+import { createHash } from 'node:crypto';
+import { hexToBin, instantiateSecp256k1, type Secp256k1 } from '@bitauth/libauth';
+import { toExtendedJson } from '@xo-cash/utils';
+import { z } from 'zod';
+
+import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.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';
+
+/**
+ * Schema to validate a single write resource.
+ *
+ * Each write targets one resource slot identified by (id, publicKey).
+ * The publicKey owner signs the payload to prove write authority.
+ */
+const writeResource = z.object({
+    id: z.string().min(1),
+    publicKey: z.string().min(1),
+    timestamp: z.number().positive(),
+    signature: z.string().min(1),
+    value: z.instanceof(Uint8Array),
+});
+
+/**
+ * Schema to validate the write body.
+ *
+ * The body contains an array of write resources.
+ */
+const writeBody = z.object({
+    resources: z.array(writeResource).min(1),
+});
+
+/**
+ * Schema to validate and deduplicate resource IDs supplied in an application body.
+ *
+ * The body contains an array of resource IDs.
+ */
+const resourceIdsSchema = z.object({
+    resourceId: z
+        .array(z.string().min(1))
+        .min(1, 'At least one resourceId is required')
+        .transform((ids) => [...new Set(ids)]),
+});
+
+type WriteResource = z.infer<typeof writeResource>;
+
+/**
+ * Resource data domain routes.
+ *
+ * A "resource" is a shared document identified by id. Multiple instances
+ * (one per publicKey) can coexist within the same resource. Clients read,
+ * write, subscribe to changes, and unsubscribe over the transport-neutral
+ * 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,
+    ) {}
+
+    /** Declare exact routes; each handler owns its stream behavior. */
+    async getRoutes(): Promise<Array<RouteDefinition>> {
+        return [
+            {
+                url: '/data/get',
+                handler: this.getData.bind(this),
+            },
+            {
+                url: '/data/write',
+                handler: this.writeData.bind(this),
+            },
+            {
+                url: '/data/subscribe',
+                handler: this.subscribeData.bind(this),
+            },
+            {
+                url: '/data/unsubscribe',
+                handler: this.unsubscribeData.bind(this),
+            },
+        ];
+    }
+
+    /** Reads the requested resources from the database, returning all instances for each resource */
+    async getData(stream: RouteStream): Promise<void> {
+        const resourceIds = this.getResourceIds(stream);
+
+        // Remove duplicates.
+        const uniqueIds = [...new Set(resourceIds)];
+
+        // If there are no resource ids, return an empty array.
+        if (uniqueIds.length === 0) {
+            await stream.send([]);
+
+            return;
+        }
+
+        // Read the data from the database.
+        const rows = await this.database.db
+            .selectFrom('resource_data')
+            .select(['resource_id', 'public_key', 'blob', 'timestamp'])
+            .where('resource_id', 'in', uniqueIds)
+            .orderBy('timestamp', 'asc')
+            .execute();
+
+        // Format the rows into the read resource responses.
+        const formattedRows = rows.map((row) => ({
+            resourceId: row.resource_id,
+            publicKey: row.public_key,
+            blob: new Uint8Array(row.blob),
+            timestamp: row.timestamp,
+        }));
+
+        await stream.send(formattedRows);
+    }
+
+    /**
+     * Authenticated batch write.
+     *
+     * Each resource in the batch is verified independently, then persisted and
+     * broadcast to any subscribers listening on that resource's topic.
+     */
+    async writeData(stream: RouteStream): Promise<void> {
+        const { resources } = writeBody.parse(stream.body);
+
+        // 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)));
+
+        const timestamp = Date.now();
+        const rows = resources.map((resource) => ({
+            resourceId: resource.id,
+            publicKey: resource.publicKey,
+            blob: Buffer.from(resource.value),
+        }));
+
+        // 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) {
+                await trx
+                    .insertInto('resource_data')
+                    .values({
+                        resource_id: resourceId,
+                        public_key: publicKey,
+                        blob,
+                        timestamp,
+                    })
+                    .onConflict((oc) =>
+                        oc.columns(['resource_id', 'public_key']).doUpdateSet({
+                            blob,
+                            timestamp,
+                        }),
+                    )
+                    .execute();
+            }
+        });
+
+        // Format the rows into the written resource responses.
+        const written = rows.map(({ resourceId, publicKey, blob }) => ({
+            resourceId,
+            instance: {
+                publicKey,
+                blob: new Uint8Array(blob),
+                timestamp,
+            },
+        }));
+
+        // 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), {
+                type: 'instance-changed',
+                data: { resourceId, ...instance },
+            });
+        }
+
+        // Return the persisted instances so the writer can confirm what was stored.
+        await stream.send({
+            resources: written.map(({ resourceId, instance }) => ({
+                id: resourceId,
+                ...instance,
+            })),
+        });
+    }
+
+    /** Subscribe this connection to future changes for the requested resources. */
+    async subscribeData(stream: RouteStream): Promise<void> {
+        const resourceIds = this.getResourceIds(stream);
+
+        const topics = resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId));
+
+        // Subscribe registers the topics synchronously, then keeps this route
+        // active until those topics are removed or the connection closes.
+        await this.broadcaster.subscribe(stream, topics);
+    }
+
+    /**
+     * Leave resource topics without closing the connection.
+     *
+     * Only available on bidirectional transports (WebSocket). SSE clients
+     * unsubscribe implicitly by aborting the HTTP request.
+     */
+    async unsubscribeData(stream: RouteStream): Promise<void> {
+        const resourceIds = this.getResourceIds(stream);
+
+        // If the stream is not bidirectional, throw an error.
+        if (!stream.bidirectional) {
+            throw new ApplicationError(HTTP_STATUS_CODE_NOT_ACCEPTED, 'This route requires an existing bidirectional stream');
+        }
+
+        // Unsubscribe from the resource topics.
+        await this.broadcaster.unsubscribe(
+            stream,
+            resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)),
+        );
+        await stream.send({});
+    }
+
+    /** Extract resource id list from the decoded request body. */
+    private getResourceIds(stream: RouteStream): string[] {
+        const body =
+            typeof stream.body === 'object' && stream.body !== null && !Array.isArray(stream.body) ? (stream.body as Record<string, unknown>) : {};
+
+        return resourceIdsSchema.parse(body).resourceId;
+    }
+
+    /**
+     * Verify one write's timestamp freshness and secp256k1 signature.
+     *
+     * The client must sign the canonical payload built from timestamp, resource
+     * id, and value — not the raw HTTP/WebSocket envelope.
+     */
+    private async verifyWriteResource(resource: WriteResource): Promise<void> {
+        this.assertFreshTimestamp(resource.timestamp);
+
+        if (!(await this.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) {
+            throw new UnauthorizedError('Invalid resource signature');
+        }
+    }
+
+    /**
+     * Reject writes with stale timestamps to limit replay window.
+     *
+     * Both past and future timestamps outside the window are rejected.
+     */
+    private assertFreshTimestamp(timestamp: number): void {
+        const age = Math.abs(Date.now() - timestamp);
+        if (age > this.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}`;
+    }
+}

tests/routes/resources.test.ts

@@ -0,0 +1,140 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { DataRoute } from "../../src/routes/resources.js";
+import { UnauthorizedError } from "../../src/errors/index.js";
+import { type BaseBroadcaster } from "../../src/services/broadcaster.js";
+import { ApplicationRouteStream } from "../../src/services/route-stream.js";
+import { Database } from "../../src/services/storage/database.js";
+import { TestConnection } from "../helpers/test-connection.js";
+import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../src/constants.js";
+
+function createBroadcasterStub() {
+  return {
+    subscribe: vi.fn(),
+    unsubscribe: vi.fn().mockResolvedValue(undefined),
+    publish: vi.fn(),
+    sendEvent: vi.fn(),
+  } as unknown as BaseBroadcaster;
+}
+
+describe("DataRoute subscriptions", () => {
+  it("subscribes to future resource changes until removal", async () => {
+    let resolveRemoved: () => void = () => undefined;
+    const removed = new Promise<void>((resolve) => {
+      resolveRemoved = resolve;
+    });
+    const storage = {
+      db: {
+        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 ApplicationRouteStream),
+    ).rejects.toBeInstanceOf(UnauthorizedError);
+
+    expect(storage.db.transaction).not.toHaveBeenCalled()
+    expect(broadcaster.publish).not.toHaveBeenCalled();
+  });
+});
+