diff --git a/source/index.ts b/source/index.ts index 2559873..2ff7ba8 100644 --- a/source/index.ts +++ b/source/index.ts @@ -41,7 +41,7 @@ export class App { 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); } diff --git a/source/routes/resources.ts b/source/routes/resources.ts index eaf833d..ac7a250 100644 --- a/source/routes/resources.ts +++ b/source/routes/resources.ts @@ -42,7 +42,7 @@ 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; @@ -95,7 +95,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) { @@ -107,7 +107,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' ]) .where('resource_id', 'in', uniqueIds) .orderBy('timestamp', 'asc') .execute(); @@ -155,11 +155,10 @@ export class DataRoute implements RouteModule { timestamp, }) .onConflict((oc) => - oc.columns(['resource_id', 'public_key']).doUpdateSet({ + oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({ blob, timestamp, - }), - ) + })) .execute(); } }); @@ -276,7 +275,8 @@ export class DataRoute implements RouteModule { const signature = hexToBin(signatureHex); // Create a SHA-256 hash of the payload. - const messageHash = createHash('sha256').update(payload).digest(); + const messageHash = createHash('sha256').update(payload) +.digest(); // Low-S normalization rejects malleable signature encodings. return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash); diff --git a/source/services/transport/ws-transport.ts b/source/services/transport/ws-transport.ts index 0f34bdd..544d467 100644 --- a/source/services/transport/ws-transport.ts +++ b/source/services/transport/ws-transport.ts @@ -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(), }) @@ -178,13 +179,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( - toExtendedJson({ + ws.send(toExtendedJson({ ...(requestId === undefined ? {} : { id: requestId }), type: 'error', ...normalizePublicError(error), - }), - ); + })); } /** diff --git a/test/routes/resources.test.ts b/test/routes/resources.test.ts index ff6980a..65a47b1 100644 --- a/test/routes/resources.test.ts +++ b/test/routes/resources.test.ts @@ -1,140 +1,123 @@ -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it, vi } from 'vitest'; -import { DataRoute } from "../../source/routes/resources.js"; -import { UnauthorizedError } from "../../source/errors/index.js"; -import { type BaseBroadcaster } from "../../source/services/broadcaster.js"; -import { ApplicationRouteStream } from "../../source/services/route-stream.js"; -import { Database } from "../../source/services/storage/database.js"; -import { TestConnection } from "../helpers/test-connection.js"; -import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../source/constants.js"; +import { DataRoute } from '../../source/routes/resources.ts'; +import { UnauthorizedError } from '../../source/errors/index.ts'; +import { type BaseBroadcaster } from '../../source/services/broadcaster.ts'; +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'; -function createBroadcasterStub() { - return { - subscribe: vi.fn(), - unsubscribe: vi.fn().mockResolvedValue(undefined), - publish: vi.fn(), - sendEvent: vi.fn(), - } as unknown as BaseBroadcaster; -} +const createBroadcasterStub = (): BaseBroadcaster => { + 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((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]), +describe('DataRoute subscriptions', (): void => { + it('subscribes to future resource changes until removal', async (): Promise => { + let resolveRemoved: () => void = () => undefined; + const removed = new Promise((resolve) => { + resolveRemoved = resolve; + }); + const storage = { + db: { + transaction: vi.fn(), }, - ], - }, - } as unknown as ApplicationRouteStream), - ).rejects.toBeInstanceOf(UnauthorizedError); + } 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); - expect(storage.db.transaction).not.toHaveBeenCalled() - expect(broadcaster.publish).not.toHaveBeenCalled(); - }); + 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 (): Promise => { + 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 (): Promise => { + 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', (): void => { + it('rejects an invalid resource signature before writing the batch', async (): Promise => { + 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(); + }); +}); diff --git a/test/services/stream/stream.test.ts b/test/services/stream/stream.test.ts index 5b6a055..74121ef 100644 --- a/test/services/stream/stream.test.ts +++ b/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 { HttpRequestStream } from "../../../source/services/stream/http-request-stream.js"; -import { WSStream } from "../../../source/services/stream/ws-stream.js"; +import { HonoSSEStream } from '../../../source/services/stream/hono-sse-stream.ts'; +import { HttpRequestStream } from '../../../source/services/stream/http-request-stream.ts'; +import { WSStream } from '../../../source/services/stream/ws-stream.ts'; -describe("stream lifecycle observers", () => { - it("buffers exactly one normal HTTP response", async () => { - const stream = new HttpRequestStream(); +describe('stream lifecycle observers', (): void => { + it('buffers exactly one normal HTTP response', async (): Promise => { + const stream = new HttpRequestStream(); - await stream.send({ - type: "response", - statusCode: 200, - body: { ok: true }, + await stream.send({ + type: 'response', + statusCode: 200, + 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({ - type: "response", - statusCode: 200, - body: { ok: true }, + it('notifies WebSocket observers registered after remote closure', (): void => { + const stream = new WSStream({ + send: vi.fn(), + 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", () => { - const stream = new WSStream({ - send: vi.fn(), - close: vi.fn(), - readyState: 1, + it('notifies SSE observers registered after local closure', async (): Promise => { + const streamApi = { + writeSSE: vi.fn(), + close: vi.fn(), + }; + const stream = new HonoSSEStream(streamApi as unknown as ConstructorParameters[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[0], - ); - const onClose = vi.fn(); - - stream.close(); - stream.onClose(onClose); - stream.close(); - - expect(streamApi.close).toHaveBeenCalledOnce(); - expect(onClose).toHaveBeenCalledOnce(); - }); }); diff --git a/test/services/transports/ws-transport.test.ts b/test/services/transports/ws-transport.test.ts index a4fb98a..662316f 100644 --- a/test/services/transports/ws-transport.test.ts +++ b/test/services/transports/ws-transport.test.ts @@ -1,58 +1,48 @@ -import { describe, expect, it } from "vitest"; -import { z } from "zod"; -import { WebSocketServer } from "ws"; +import { describe, expect, it } from 'vitest'; +import { z } from 'zod'; +import type { WebSocketServer } from 'ws'; -import { ApplicationRouter } from "../../../source/services/router.js"; -import { - WsTransportRouter, -} from "../../../source/services/transport/ws-transport.js"; -import { Logger } from "../../../source/utils/logger.js"; -import { toExtendedJson } from "@xo-cash/utils"; +import { ApplicationRouter } from '../../../source/services/router.ts'; +import { WsTransportRouter } from '../../../source/services/transport/ws-transport.ts'; +import { Logger } from '../../../source/utils/logger.ts'; +import { toExtendedJson } from '@xo-cash/utils'; -describe("WebSocket request decoding", () => { - it("decodes the minimal route-agnostic envelope and Extended JSON body", async () => { - await expect( - WsTransportRouter.decodeWebSocketRequest( - toExtendedJson({ - id: "request-1", - 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]) }, +describe('WebSocket request decoding', (): void => { + it('decodes the minimal route-agnostic envelope and Extended JSON body', async (): Promise => { + await expect(WsTransportRouter.decodeWebSocketRequest(toExtendedJson({ + id: 'request-1', + 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([ - "{}", - '{"path":42}', - '{"path":"/data/get","id":1}', - '{"path":"/data/get","method":"POST"}', - ])("rejects an invalid envelope: %s", async (payload) => { - await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf( - z.ZodError, + it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])( + 'rejects an invalid envelope: %s', + async (payload) => { + await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError); + }, ); - }); - it("rejects malformed JSON", async () => { - await expect(WsTransportRouter.decodeWebSocketRequest("{")).rejects.toMatchObject({ - statusCode: 400, - message: "Invalid JSON in WebSocket message", + it('rejects malformed JSON', async (): Promise => { + await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({ + statusCode: 400, + message: 'Invalid JSON in WebSocket message', + }); }); - }); }); -describe("WsTransportRouter payload limits", () => { - it("configures Hono's ws server with the requested maxPayload", async () => { - const debug = new Logger("ws-transport-test"); - const router = await ApplicationRouter.create([]); - const transport = new WsTransportRouter(router, debug, 1024); - const wsServer = transport.websocketServer as unknown as WebSocketServer; +describe('WsTransportRouter payload limits', (): void => { + it("configures Hono's ws server with the requested maxPayload", async () => { + const debug = new Logger('ws-transport-test'); + const router = await ApplicationRouter.create([]); + const transport = new WsTransportRouter(router, debug, 1024); + const wsServer = transport.websocketServer as unknown as WebSocketServer; - expect(wsServer.options.maxPayload).toBe(1024); - await transport.stop(); - }); + expect(wsServer.options.maxPayload).toBe(1024); + await transport.stop(); + }); });