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:37:22 +00:00
parent 9c0746bb24
commit 8be4467721
4 changed files with 236 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
source/index.ts
+1 -1
View File
@@ -31,7 +31,7 @@ export class App {
const router = await ApplicationRouter.create(routes); const router = await ApplicationRouter.create(routes);
const http = new HttpTransportRouter(router, debug); const http = new HttpTransportRouter(router, debug);
const host = new ServerHost(config, debug, [http]); const host = new ServerHost(config, debug, [ http ]);
return new App(host, database); return new App(host, database);
} }
source/services/server-host.ts
+4 -4
View File
@@ -40,8 +40,8 @@ export class ServerHost {
const corsMiddleware = cors({ const corsMiddleware = cors({
origin: corsConfig.origin ?? '*', origin: corsConfig.origin ?? '*',
allowMethods: corsConfig.methods ?? ['POST', 'OPTIONS'], allowMethods: corsConfig.methods ?? [ 'POST', 'OPTIONS' ],
allowHeaders: corsConfig.allowedHeaders ?? ['Content-Type', 'Accept'], allowHeaders: corsConfig.allowedHeaders ?? [ 'Content-Type', 'Accept' ],
}); });
this.app.use('*', corsMiddleware); this.app.use('*', corsMiddleware);
@@ -77,7 +77,7 @@ export class ServerHost {
throw new Error('ServerHost supports only one WebSocket upgrade server'); throw new Error('ServerHost supports only one WebSocket upgrade server');
} }
const [upgradeTransport] = upgradeTransports; const [ upgradeTransport ] = upgradeTransports;
this.server = serve({ this.server = serve({
fetch: this.app.fetch, fetch: this.app.fetch,
@@ -124,7 +124,7 @@ export class ServerHost {
const closeTransports = this.transports.map((transport) => Promise.resolve(transport.stop?.())); const closeTransports = this.transports.map((transport) => Promise.resolve(transport.stop?.()));
// Create a promise that resolves when the server and transports are closed // Create a promise that resolves when the server and transports are closed
this.stopPromise = Promise.all([closeServer, ...closeTransports]).then(() => { this.stopPromise = Promise.all([ closeServer, ...closeTransports ]).then(() => {
this.stopPromise = undefined; this.stopPromise = undefined;
}); });
source/services/transport/transport-router.ts
+3
View File
@@ -4,6 +4,7 @@ import type { Hono } from 'hono';
/** Hono variables populated by transport-boundary middleware. */ /** Hono variables populated by transport-boundary middleware. */
export type AppEnv = { export type AppEnv = {
Variables: { Variables: {
/** Decoded Extended JSON request body, when present. */ /** Decoded Extended JSON request body, when present. */
parsedBody?: unknown; parsedBody?: unknown;
@@ -19,6 +20,7 @@ export type AppEnv = {
* not application routing. Implementations remain unaware of route modules. * not application routing. Implementations remain unaware of route modules.
*/ */
export interface TransportRouter { export interface TransportRouter {
/** /**
* Attach wire endpoints and middleware to the shared Hono application. * Attach wire endpoints and middleware to the shared Hono application.
* *
@@ -32,6 +34,7 @@ export interface TransportRouter {
/** A transport which also supplies the WebSocket server used during upgrade. */ /** A transport which also supplies the WebSocket server used during upgrade. */
export interface UpgradeTransportRouter extends TransportRouter { export interface UpgradeTransportRouter extends TransportRouter {
/** WebSocket server instance passed to the Node HTTP listener. */ /** WebSocket server instance passed to the Node HTTP listener. */
readonly websocketServer: WebSocketServerLike; readonly websocketServer: WebSocketServerLike;
} }
test/services/transports/http-transport.test.ts
+90 -95
View File
@@ -1,27 +1,26 @@
import { Hono } from "hono"; import { Hono } from 'hono';
import { describe, expect, it, vi } from "vitest"; import { describe, expect, it, vi } from 'vitest';
import type { RouteDefinition } from "../../../source/routes/types.js"; import type { RouteDefinition } from '../../../source/routes/types.ts';
import { ApplicationError } from "../../../source/errors/index.js"; import { ApplicationError } from '../../../source/errors/index.ts';
import { ApplicationRouter } from "../../../source/services/router.js"; import { ApplicationRouter } from '../../../source/services/router.ts';
import { Broadcaster } from "../../../source/services/broadcaster.js"; import { Broadcaster } from '../../../source/services/broadcaster.ts';
import { HttpTransportRouter } from "../../../source/services/transport/http-transport.js"; import { HttpTransportRouter } from '../../../source/services/transport/http-transport.ts';
import type { AppEnv } from "../../../source/services/transport/transport-router.js"; import type { AppEnv } from '../../../source/services/transport/transport-router.ts';
import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils';
import { Logger } from "../../../source/utils/logger.js"; import { Logger } from '../../../source/utils/logger.ts';
import { ServerHost } from "../../../source/services/server-host.js"; import { ServerHost } from '../../../source/services/server-host.ts';
async function createApp( const createApp = async (
routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]), routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]),
maxRequestBodyBytes = 1024 * 1024, maxRequestBodyBytes = 1024 * 1024,
): Promise<Hono<AppEnv>> { ): Promise<Hono<AppEnv>> => {
const debug = new Logger("http-transport-test"); const debug = new Logger('http-transport-test');
const broadcaster = new Broadcaster(debug); const broadcaster = new Broadcaster(debug);
const resolvedRoutes = const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes;
typeof routes === "function" ? routes(broadcaster) : routes;
const router = await ApplicationRouter.create([ const router = await ApplicationRouter.create([
{ {
async getRoutes() { async getRoutes(): Promise<RouteDefinition[]> {
return resolvedRoutes; return resolvedRoutes;
}, },
}, },
@@ -30,25 +29,26 @@ async function createApp(
const app = new Hono<AppEnv>(); const app = new Hono<AppEnv>();
app.onError(HttpTransportRouter.createErrorHandler(debug)); app.onError(HttpTransportRouter.createErrorHandler(debug));
app.use("*", ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug)); app.use('*', ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug));
app.use("*", HttpTransportRouter.createExtJsonMiddleware(debug)); app.use('*', HttpTransportRouter.createExtJsonMiddleware(debug));
transport.register(app); transport.register(app);
return app;
}
describe("HttpTransportRouter", () => { return app;
it("runs normal HTTP through a non-streaming route stream", async () => { };
describe('HttpTransportRouter', (): void => {
it('runs normal HTTP through a non-streaming route stream', async (): Promise<void> => {
const app = await createApp([ const app = await createApp([
{ {
url: "/echo", url: '/echo',
handler: async (stream) => stream.send(stream.body), handler: async (stream): Promise<void> => stream.send(stream.body),
}, },
]); ]);
const value = new Uint8Array([1, 2, 3]); const value = new Uint8Array([ 1, 2, 3 ]);
const response = await app.request("/echo", { const response = await app.request('/echo', {
method: "POST", method: 'POST',
headers: { "content-type": "application/json" }, headers: { 'content-type': 'application/json' },
body: toExtendedJson({ value }), body: toExtendedJson({ value }),
}); });
@@ -56,102 +56,100 @@ describe("HttpTransportRouter", () => {
expect(fromExtendedJson(await response.text())).toEqual({ value }); expect(fromExtendedJson(await response.text())).toEqual({ value });
}); });
it("returns 204 when a normal HTTP route sends nothing", async () => { it('returns 204 when a normal HTTP route sends nothing', async (): Promise<void> => {
const app = await createApp([ const app = await createApp([
{ {
url: "/nothing", url: '/nothing',
handler: () => undefined, handler: (): void => undefined,
}, },
]); ]);
const response = await app.request("/nothing", { method: "POST" }); const response = await app.request('/nothing', { method: 'POST' });
expect(response.status).toBe(204); expect(response.status).toBe(204);
expect(await response.text()).toBe(""); expect(await response.text()).toBe('');
}); });
it("returns normalized errors for non-streaming requests", async () => { it('returns normalized errors for non-streaming requests', async (): Promise<void> => {
const app = await createApp([]); const app = await createApp([]);
const missing = await app.request("/missing", { method: "POST" }); const missing = await app.request('/missing', { method: 'POST' });
expect(missing.status).toBe(404); expect(missing.status).toBe(404);
expect(await missing.json()).toEqual({ expect(await missing.json()).toEqual({
statusCode: 404, statusCode: 404,
error: "No route found for /missing", error: 'No route found for /missing',
}); });
const invalid = await app.request("/missing", { const invalid = await app.request('/missing', {
method: "POST", method: 'POST',
headers: { "content-type": "application/json" }, headers: { 'content-type': 'application/json' },
body: "{", body: '{',
}); });
expect(invalid.status).toBe(400); expect(invalid.status).toBe(400);
expect(await invalid.json()).toEqual({ expect(await invalid.json()).toEqual({
statusCode: 400, statusCode: 400,
error: "Invalid JSON in request body", error: 'Invalid JSON in request body',
}); });
}); });
it("rejects subscribe when normal HTTP has no streaming capability", async () => { it('rejects subscribe when normal HTTP has no streaming capability', async (): Promise<void> => {
const app = await createApp((broadcaster) => [ const app = await createApp((broadcaster) => [
{ {
url: "/items/subscribe", url: '/items/subscribe',
handler: async (stream) => { handler: async (stream): Promise<void> => {
await broadcaster.subscribe(stream, ["items"]); await broadcaster.subscribe(stream, [ 'items' ]);
}, },
}, },
]); ]);
const response = await app.request("/items/subscribe", { method: "POST" }); const response = await app.request('/items/subscribe', { method: 'POST' });
expect(response.status).toBe(406); expect(response.status).toBe(406);
expect(await response.json()).toMatchObject({ statusCode: 406 }); expect(await response.json()).toMatchObject({ statusCode: 406 });
}); });
it("sends SSE route errors as events and closes only that stream", async () => { it('sends SSE route errors as events and closes only that stream', async (): Promise<void> => {
const app = await createApp([ const app = await createApp([
{ {
url: "/items/subscribe", url: '/items/subscribe',
handler: () => { handler: (): void => {
throw new Error("private storage failure"); throw new Error('private storage failure');
}, },
}, },
]); ]);
const response = await app.request("/items/subscribe", { const response = await app.request('/items/subscribe', {
method: "POST", method: 'POST',
headers: { accept: "text/event-stream" }, headers: { accept: 'text/event-stream' },
}); });
const events = await response.text(); const events = await response.text();
expect(response.status).toBe(200); expect(response.status).toBe(200);
expect(events).toContain("event: error"); expect(events).toContain('event: error');
expect(events).toContain( expect(events).toContain('data: {"statusCode":500,"error":"Internal Server Error"}');
'data: {"statusCode":500,"error":"Internal Server Error"}', expect(events).not.toContain('private storage failure');
);
expect(events).not.toContain("private storage failure");
}); });
it("sends a normal route as one SSE response event and then closes", async () => { it('sends a normal route as one SSE response event and then closes', async (): Promise<void> => {
const app = await createApp([ const app = await createApp([
{ {
url: "/echo", url: '/echo',
handler: (stream) => stream.send({ ok: true }), handler: (stream): Promise<void> => stream.send({ ok: true }),
}, },
]); ]);
const response = await app.request("/echo", { const response = await app.request('/echo', {
method: "POST", method: 'POST',
headers: { accept: "text/event-stream" }, headers: { accept: 'text/event-stream' },
}); });
const events = await response.text(); const events = await response.text();
expect(response.status).toBe(200); expect(response.status).toBe(200);
expect(events).toContain("event: response"); expect(events).toContain('event: response');
expect(events).toContain('data: {"ok":true}'); expect(events).toContain('data: {"ok":true}');
}); });
it("keeps SSE open until the route's subscription promise resolves", async () => { it("keeps SSE open until the route's subscription promise resolves", async (): Promise<void> => {
let removeSubscription: () => Promise<void> = async () => undefined; let removeSubscription: () => Promise<void> = async () => undefined;
let markSubscribed: () => void = () => undefined; let markSubscribed: () => void = () => undefined;
const subscribed = new Promise<void>((resolve) => { const subscribed = new Promise<void>((resolve) => {
@@ -159,10 +157,10 @@ describe("HttpTransportRouter", () => {
}); });
const app = await createApp((broadcaster) => [ const app = await createApp((broadcaster) => [
{ {
url: "/items/subscribe", url: '/items/subscribe',
handler: async (stream) => { handler: async (stream): Promise<void> => {
const topics = ["items"]; const topics = [ 'items' ];
removeSubscription = () => broadcaster.unsubscribe(stream, topics); removeSubscription = (): Promise<void> => broadcaster.unsubscribe(stream, topics);
const removed = broadcaster.subscribe(stream, topics); const removed = broadcaster.subscribe(stream, topics);
markSubscribed(); markSubscribed();
@@ -171,9 +169,9 @@ describe("HttpTransportRouter", () => {
}, },
]); ]);
const response = await app.request("/items/subscribe", { const response = await app.request('/items/subscribe', {
method: "POST", method: 'POST',
headers: { accept: "text/event-stream" }, headers: { accept: 'text/event-stream' },
}); });
const body = response.text(); const body = response.text();
const completed = vi.fn(); const completed = vi.fn();
@@ -185,58 +183,55 @@ describe("HttpTransportRouter", () => {
await removeSubscription(); await removeSubscription();
expect(await body).toBe(""); expect(await body).toBe('');
expect(completed).toHaveBeenCalledOnce(); expect(completed).toHaveBeenCalledOnce();
}); });
it("rejects unsubscribe over non-bidirectional SSE", async () => { it('rejects unsubscribe over non-bidirectional SSE', async (): Promise<void> => {
const app = await createApp((broadcaster) => [ const app = await createApp((broadcaster) => [
{ {
url: "/items/unsubscribe", url: '/items/unsubscribe',
handler: async (stream) => { handler: async (stream): Promise<void> => {
if (!stream.bidirectional) { if (!stream.bidirectional) {
throw new ApplicationError( throw new ApplicationError(400, 'This route requires an existing bidirectional stream');
400,
"This route requires an existing bidirectional stream",
);
} }
await broadcaster.unsubscribe(stream, ["items"]); await broadcaster.unsubscribe(stream, [ 'items' ]);
}, },
}, },
]); ]);
const response = await app.request("/items/unsubscribe", { const response = await app.request('/items/unsubscribe', {
method: "POST", method: 'POST',
headers: { accept: "text/event-stream" }, headers: { accept: 'text/event-stream' },
}); });
const events = await response.text(); const events = await response.text();
expect(response.status).toBe(200); expect(response.status).toBe(200);
expect(events).toContain("event: error"); expect(events).toContain('event: error');
expect(events).toContain('"statusCode":400'); expect(events).toContain('"statusCode":400');
}); });
it("rejects HTTP bodies larger than the configured byte limit", async () => { it('rejects HTTP bodies larger than the configured byte limit', async (): Promise<void> => {
const app = await createApp( const app = await createApp(
[ [
{ {
url: "/echo", url: '/echo',
handler: async (stream) => stream.send(stream.body), handler: async (stream): Promise<void> => stream.send(stream.body),
}, },
], ],
32, 32,
); );
const response = await app.request("/echo", { const response = await app.request('/echo', {
method: "POST", method: 'POST',
headers: { "content-type": "application/json" }, headers: { 'content-type': 'application/json' },
body: JSON.stringify({ value: "x".repeat(64) }), body: JSON.stringify({ value: 'x'.repeat(64) }),
}); });
expect(response.status).toBe(413); expect(response.status).toBe(413);
expect(await response.json()).toEqual({ expect(await response.json()).toEqual({
statusCode: 413, statusCode: 413,
error: "Request body exceeds the 32 byte limit", error: 'Request body exceeds the 32 byte limit',
}); });
}); });
}); });