Harvmaster/sync-server-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Formatting

Browse Source
This commit is contained in:
Harvmaster
2026-08-03 03:40:57 +00:00
parent 97d4422b8f
commit c9845d0f28
6 changed files with 214 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/index.ts
+1 -1
View File
@@ -41,7 +41,7 @@ export class App {
const http = new HttpTransportRouter(router, debug); const http = new HttpTransportRouter(router, debug);
const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes); 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); return new App(host, database);
} }
source/routes/resources.ts
+7 -7
View File
@@ -42,7 +42,7 @@ const resourceIdsSchema = z.object({
resourceId: z resourceId: z
.array(z.string().min(1)) .array(z.string().min(1))
.min(1, 'At least one resourceId is required') .min(1, 'At least one resourceId is required')
.transform((ids) => [...new Set(ids)]), .transform((ids) => [ ...new Set(ids) ]),
}); });
type WriteResource = z.infer<typeof writeResource>; type WriteResource = z.infer<typeof writeResource>;
@@ -95,7 +95,7 @@ export class DataRoute implements RouteModule {
const resourceIds = this.getResourceIds(stream); const resourceIds = this.getResourceIds(stream);
// Remove duplicates. // Remove duplicates.
const uniqueIds = [...new Set(resourceIds)]; const uniqueIds = [ ...new Set(resourceIds) ];
// If there are no resource ids, return an empty array. // If there are no resource ids, return an empty array.
if (uniqueIds.length === 0) { if (uniqueIds.length === 0) {
@@ -107,7 +107,7 @@ export class DataRoute implements RouteModule {
// Read the data from the database. // Read the data from the database.
const rows = await this.database.db const rows = await this.database.db
.selectFrom('resource_data') .selectFrom('resource_data')
.select(['resource_id', 'public_key', 'blob', 'timestamp']) .select([ 'resource_id', 'public_key', 'blob', 'timestamp' ])
.where('resource_id', 'in', uniqueIds) .where('resource_id', 'in', uniqueIds)
.orderBy('timestamp', 'asc') .orderBy('timestamp', 'asc')
.execute(); .execute();
@@ -155,11 +155,10 @@ export class DataRoute implements RouteModule {
timestamp, timestamp,
}) })
.onConflict((oc) => .onConflict((oc) =>
oc.columns(['resource_id', 'public_key']).doUpdateSet({ oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({
blob, blob,
timestamp, timestamp,
}), }))
)
.execute(); .execute();
} }
}); });
@@ -276,7 +275,8 @@ export class DataRoute implements RouteModule {
const signature = hexToBin(signatureHex); const signature = hexToBin(signatureHex);
// Create a SHA-256 hash of the payload. // 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. // Low-S normalization rejects malleable signature encodings.
return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash); return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash);
source/services/transport/ws-transport.ts
+4 -5
View File
@@ -20,7 +20,8 @@ const WS_ROUTE = '/ws';
// Strict validation prevents legacy or protocol-specific fields reaching routes. // Strict validation prevents legacy or protocol-specific fields reaching routes.
const wsRequestSchema = z const wsRequestSchema = z
.object({ .object({
id: z.string().min(1).optional(), id: z.string().min(1)
.optional(),
path: z.string().min(1), path: z.string().min(1),
body: z.unknown().optional(), body: z.unknown().optional(),
}) })
@@ -178,13 +179,11 @@ export class WsTransportRouter implements UpgradeTransportRouter {
* @param error - Failure to normalize into the public error shape. * @param error - Failure to normalize into the public error shape.
*/ */
private sendError(ws: WSContext, requestId: string | undefined, error: unknown): void { private sendError(ws: WSContext, requestId: string | undefined, error: unknown): void {
ws.send( ws.send(toExtendedJson({
toExtendedJson({
...(requestId === undefined ? {} : { id: requestId }), ...(requestId === undefined ? {} : { id: requestId }),
type: 'error', type: 'error',
...normalizePublicError(error), ...normalizePublicError(error),
}), }));
);
} }
/** /**
test/routes/resources.test.ts
+32 -49
View File
@@ -1,24 +1,24 @@
import { describe, expect, it, vi } from "vitest"; import { describe, expect, it, vi } from 'vitest';
import { DataRoute } from "../../source/routes/resources.js"; import { DataRoute } from '../../source/routes/resources.ts';
import { UnauthorizedError } from "../../source/errors/index.js"; import { UnauthorizedError } from '../../source/errors/index.ts';
import { type BaseBroadcaster } from "../../source/services/broadcaster.js"; import { type BaseBroadcaster } from '../../source/services/broadcaster.ts';
import { ApplicationRouteStream } from "../../source/services/route-stream.js"; import { ApplicationRouteStream } from '../../source/services/route-stream.ts';
import { Database } from "../../source/services/storage/database.js"; import type { Database } from '../../source/services/storage/database.ts';
import { TestConnection } from "../helpers/test-connection.js"; import { TestConnection } from '../helpers/test-connection.ts';
import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../source/constants.js"; import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../../source/constants.ts';
function createBroadcasterStub() { const createBroadcasterStub = (): BaseBroadcaster => {
return { return {
subscribe: vi.fn(), subscribe: vi.fn(),
unsubscribe: vi.fn().mockResolvedValue(undefined), unsubscribe: vi.fn().mockResolvedValue(undefined),
publish: vi.fn(), publish: vi.fn(),
sendEvent: vi.fn(), sendEvent: vi.fn(),
} as unknown as BaseBroadcaster; } as unknown as BaseBroadcaster;
} };
describe("DataRoute subscriptions", () => { describe('DataRoute subscriptions', (): void => {
it("subscribes to future resource changes until removal", async () => { it('subscribes to future resource changes until removal', async (): Promise<void> => {
let resolveRemoved: () => void = () => undefined; let resolveRemoved: () => void = () => undefined;
const removed = new Promise<void>((resolve) => { const removed = new Promise<void>((resolve) => {
resolveRemoved = resolve; resolveRemoved = resolve;
@@ -32,30 +32,22 @@ describe("DataRoute subscriptions", () => {
vi.mocked(broadcaster.subscribe).mockReturnValue(removed); vi.mocked(broadcaster.subscribe).mockReturnValue(removed);
const connection = new TestConnection(true, false); const connection = new TestConnection(true, false);
const stream = new ApplicationRouteStream(connection, { const stream = new ApplicationRouteStream(connection, {
resourceId: ["a", "b"], resourceId: [ 'a', 'b' ],
}); });
const route = new DataRoute(storage, broadcaster, 0); const route = new DataRoute(storage, broadcaster, 0);
const execution = route.subscribeData(stream); const execution = route.subscribeData(stream);
expect(broadcaster.subscribe).toHaveBeenCalledWith( expect(broadcaster.subscribe).toHaveBeenCalledWith(stream, [ 'resource:a', 'resource:b' ]);
stream,
["resource:a", "resource:b"],
);
expect(storage.db.transaction).not.toHaveBeenCalled(); expect(storage.db.transaction).not.toHaveBeenCalled();
expect(connection.messages).toEqual([]); expect(connection.messages).toEqual([]);
await expect( await expect(Promise.race([ execution.then(() => 'settled'), Promise.resolve('pending') ])).resolves.toBe('pending');
Promise.race([
execution.then(() => "settled"),
Promise.resolve("pending"),
]),
).resolves.toBe("pending");
resolveRemoved(); resolveRemoved();
await execution; await execution;
}); });
it("unsubscribes a bidirectional connection and acknowledges the request", async () => { it('unsubscribes a bidirectional connection and acknowledges the request', async (): Promise<void> => {
const storage = { const storage = {
db: { db: {
transaction: vi.fn(), transaction: vi.fn(),
@@ -63,29 +55,23 @@ describe("DataRoute subscriptions", () => {
} as unknown as Database; } as unknown as Database;
const broadcaster = createBroadcasterStub(); const broadcaster = createBroadcasterStub();
const connection = new TestConnection(true, true); const connection = new TestConnection(true, true);
const stream = new ApplicationRouteStream( const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ] }, 'unsubscribe-1');
connection,
{ resourceId: ["a"] },
"unsubscribe-1",
);
const route = new DataRoute(storage, broadcaster, 0); const route = new DataRoute(storage, broadcaster, 0);
await route.unsubscribeData(stream); await route.unsubscribeData(stream);
expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [ expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [ 'resource:a' ]);
"resource:a",
]);
expect(connection.messages).toEqual([ expect(connection.messages).toEqual([
{ {
id: "unsubscribe-1", id: 'unsubscribe-1',
type: "response", type: 'response',
statusCode: 200, statusCode: 200,
body: {}, body: {},
}, },
]); ]);
}); });
it("rejects selective unsubscribe on a one-way connection", async () => { it('rejects selective unsubscribe on a one-way connection', async (): Promise<void> => {
const storage = { const storage = {
db: { db: {
transaction: vi.fn(), transaction: vi.fn(),
@@ -93,7 +79,7 @@ describe("DataRoute subscriptions", () => {
} as unknown as Database; } as unknown as Database;
const broadcaster = createBroadcasterStub(); const broadcaster = createBroadcasterStub();
const stream = new ApplicationRouteStream(new TestConnection(true, false), { const stream = new ApplicationRouteStream(new TestConnection(true, false), {
resourceId: ["a"], resourceId: [ 'a' ],
}); });
const route = new DataRoute(storage, broadcaster, 0); const route = new DataRoute(storage, broadcaster, 0);
@@ -102,19 +88,18 @@ describe("DataRoute subscriptions", () => {
}); });
}); });
describe("DataRoute resource write auth", () => { describe('DataRoute resource write auth', (): void => {
it("rejects an invalid resource signature before writing the batch", async () => { it('rejects an invalid resource signature before writing the batch', async (): Promise<void> => {
const storage = { const storage = {
db: { db: {
transaction: vi.fn(), transaction: vi.fn(),
}, },
} as unknown as Database; } as unknown as Database;
const broadcaster = createBroadcasterStub() const broadcaster = createBroadcasterStub();
const route = new DataRoute(storage, broadcaster, 0); const route = new DataRoute(storage, broadcaster, 0);
await expect( await expect(route.writeData({
route.writeData({
connection: new TestConnection(true, true), connection: new TestConnection(true, true),
streaming: true, streaming: true,
bidirectional: true, bidirectional: true,
@@ -122,19 +107,17 @@ describe("DataRoute resource write auth", () => {
body: { body: {
resources: [ resources: [
{ {
id: "resource-a", id: 'resource-a',
publicKey: "not-a-public-key", publicKey: 'not-a-public-key',
timestamp: Date.now(), timestamp: Date.now(),
signature: "not-a-signature", signature: 'not-a-signature',
value: new Uint8Array([1, 2, 3]), value: new Uint8Array([ 1, 2, 3 ]),
}, },
], ],
}, },
} as unknown as ApplicationRouteStream), } as unknown as ApplicationRouteStream)).rejects.toBeInstanceOf(UnauthorizedError);
).rejects.toBeInstanceOf(UnauthorizedError);
expect(storage.db.transaction).not.toHaveBeenCalled() expect(storage.db.transaction).not.toHaveBeenCalled();
expect(broadcaster.publish).not.toHaveBeenCalled(); expect(broadcaster.publish).not.toHaveBeenCalled();
}); });
}); });
test/services/stream/stream.test.ts
+16 -20
View File
@@ -1,34 +1,32 @@
import { describe, expect, it, vi } from "vitest"; import { describe, expect, it, vi } from 'vitest';
import { HonoSSEStream } from "../../../source/services/stream/hono-sse-stream.js"; import { HonoSSEStream } from '../../../source/services/stream/hono-sse-stream.ts';
import { HttpRequestStream } from "../../../source/services/stream/http-request-stream.js"; import { HttpRequestStream } from '../../../source/services/stream/http-request-stream.ts';
import { WSStream } from "../../../source/services/stream/ws-stream.js"; import { WSStream } from '../../../source/services/stream/ws-stream.ts';
describe("stream lifecycle observers", () => { describe('stream lifecycle observers', (): void => {
it("buffers exactly one normal HTTP response", async () => { it('buffers exactly one normal HTTP response', async (): Promise<void> => {
const stream = new HttpRequestStream(); const stream = new HttpRequestStream();
await stream.send({ await stream.send({
type: "response", type: 'response',
statusCode: 200, statusCode: 200,
body: { ok: true }, body: { ok: true },
}); });
expect(stream.getResponse()).toEqual({ expect(stream.getResponse()).toEqual({
type: "response", type: 'response',
statusCode: 200, statusCode: 200,
body: { ok: true }, body: { ok: true },
}); });
await expect( await expect(stream.send({
stream.send({ type: 'response',
type: "response",
statusCode: 200, statusCode: 200,
body: { second: true }, body: { second: true },
}), })).rejects.toThrow('only send one response');
).rejects.toThrow("only send one response");
}); });
it("notifies WebSocket observers registered after remote closure", () => { it('notifies WebSocket observers registered after remote closure', (): void => {
const stream = new WSStream({ const stream = new WSStream({
send: vi.fn(), send: vi.fn(),
close: vi.fn(), close: vi.fn(),
@@ -42,19 +40,17 @@ describe("stream lifecycle observers", () => {
expect(onClose).toHaveBeenCalledOnce(); expect(onClose).toHaveBeenCalledOnce();
}); });
it("notifies SSE observers registered after local closure", () => { it('notifies SSE observers registered after local closure', async (): Promise<void> => {
const streamApi = { const streamApi = {
writeSSE: vi.fn(), writeSSE: vi.fn(),
close: vi.fn(), close: vi.fn(),
}; };
const stream = new HonoSSEStream( const stream = new HonoSSEStream(streamApi as unknown as ConstructorParameters<typeof HonoSSEStream>[0]);
streamApi as unknown as ConstructorParameters<typeof HonoSSEStream>[0],
);
const onClose = vi.fn(); const onClose = vi.fn();
stream.close(); await stream.close();
stream.onClose(onClose); stream.onClose(onClose);
stream.close(); await stream.close();
expect(streamApi.close).toHaveBeenCalledOnce(); expect(streamApi.close).toHaveBeenCalledOnce();
expect(onClose).toHaveBeenCalledOnce(); expect(onClose).toHaveBeenCalledOnce();
test/services/transports/ws-transport.test.ts
+27 -37
View File
@@ -1,53 +1,43 @@
import { describe, expect, it } from "vitest"; import { describe, expect, it } from 'vitest';
import { z } from "zod"; import { z } from 'zod';
import { WebSocketServer } from "ws"; import type { WebSocketServer } from 'ws';
import { ApplicationRouter } from "../../../source/services/router.js"; import { ApplicationRouter } from '../../../source/services/router.ts';
import { import { WsTransportRouter } from '../../../source/services/transport/ws-transport.ts';
WsTransportRouter, import { Logger } from '../../../source/utils/logger.ts';
} from "../../../source/services/transport/ws-transport.js"; import { toExtendedJson } from '@xo-cash/utils';
import { Logger } from "../../../source/utils/logger.js";
import { toExtendedJson } from "@xo-cash/utils";
describe("WebSocket request decoding", () => { describe('WebSocket request decoding', (): void => {
it("decodes the minimal route-agnostic envelope and Extended JSON body", async () => { it('decodes the minimal route-agnostic envelope and Extended JSON body', async (): Promise<void> => {
await expect( await expect(WsTransportRouter.decodeWebSocketRequest(toExtendedJson({
WsTransportRouter.decodeWebSocketRequest( id: 'request-1',
toExtendedJson({ path: '/data/write',
id: "request-1", body: { value: new Uint8Array([ 1, 2, 3 ]) },
path: "/data/write", }))).resolves.toEqual({
body: { value: new Uint8Array([1, 2, 3]) }, requestId: '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([ it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])(
"{}", 'rejects an invalid envelope: %s',
'{"path":42}', async (payload) => {
'{"path":"/data/get","id":1}', await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError);
'{"path":"/data/get","method":"POST"}', },
])("rejects an invalid envelope: %s", async (payload) => {
await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(
z.ZodError,
); );
});
it("rejects malformed JSON", async () => { it('rejects malformed JSON', async (): Promise<void> => {
await expect(WsTransportRouter.decodeWebSocketRequest("{")).rejects.toMatchObject({ await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({
statusCode: 400, statusCode: 400,
message: "Invalid JSON in WebSocket message", message: 'Invalid JSON in WebSocket message',
}); });
}); });
}); });
describe("WsTransportRouter payload limits", () => { describe('WsTransportRouter payload limits', (): void => {
it("configures Hono's ws server with the requested maxPayload", async () => { it("configures Hono's ws server with the requested maxPayload", async () => {
const debug = new Logger("ws-transport-test"); const debug = new Logger('ws-transport-test');
const router = await ApplicationRouter.create([]); const router = await ApplicationRouter.create([]);
const transport = new WsTransportRouter(router, debug, 1024); const transport = new WsTransportRouter(router, debug, 1024);
const wsServer = transport.websocketServer as unknown as WebSocketServer; const wsServer = transport.websocketServer as unknown as WebSocketServer;