From 8be4467721ca5e9717ae0e9d292e5c9ac1dfc106 Mon Sep 17 00:00:00 2001 From: Harvmaster Date: Mon, 3 Aug 2026 03:37:22 +0000 Subject: [PATCH] Formatting --- source/index.ts | 2 +- source/services/server-host.ts | 8 +- source/services/transport/transport-router.ts | 3 + .../transports/http-transport.test.ts | 461 +++++++++--------- 4 files changed, 236 insertions(+), 238 deletions(-) diff --git a/source/index.ts b/source/index.ts index c5ca463..1e00cac 100644 --- a/source/index.ts +++ b/source/index.ts @@ -31,7 +31,7 @@ export class App { const router = await ApplicationRouter.create(routes); 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); } diff --git a/source/services/server-host.ts b/source/services/server-host.ts index 9c72f37..e4c8442 100644 --- a/source/services/server-host.ts +++ b/source/services/server-host.ts @@ -40,8 +40,8 @@ export class ServerHost { const corsMiddleware = cors({ origin: corsConfig.origin ?? '*', - allowMethods: corsConfig.methods ?? ['POST', 'OPTIONS'], - allowHeaders: corsConfig.allowedHeaders ?? ['Content-Type', 'Accept'], + allowMethods: corsConfig.methods ?? [ 'POST', 'OPTIONS' ], + allowHeaders: corsConfig.allowedHeaders ?? [ 'Content-Type', 'Accept' ], }); this.app.use('*', corsMiddleware); @@ -77,7 +77,7 @@ export class ServerHost { throw new Error('ServerHost supports only one WebSocket upgrade server'); } - const [upgradeTransport] = upgradeTransports; + const [ upgradeTransport ] = upgradeTransports; this.server = serve({ fetch: this.app.fetch, @@ -124,7 +124,7 @@ export class ServerHost { const closeTransports = this.transports.map((transport) => Promise.resolve(transport.stop?.())); // 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; }); diff --git a/source/services/transport/transport-router.ts b/source/services/transport/transport-router.ts index be506cd..4e05e42 100644 --- a/source/services/transport/transport-router.ts +++ b/source/services/transport/transport-router.ts @@ -4,6 +4,7 @@ import type { Hono } from 'hono'; /** Hono variables populated by transport-boundary middleware. */ export type AppEnv = { Variables: { + /** Decoded Extended JSON request body, when present. */ parsedBody?: unknown; @@ -19,6 +20,7 @@ export type AppEnv = { * not application routing. Implementations remain unaware of route modules. */ export interface TransportRouter { + /** * 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. */ export interface UpgradeTransportRouter extends TransportRouter { + /** WebSocket server instance passed to the Node HTTP listener. */ readonly websocketServer: WebSocketServerLike; } diff --git a/test/services/transports/http-transport.test.ts b/test/services/transports/http-transport.test.ts index fbf213f..095d266 100644 --- a/test/services/transports/http-transport.test.ts +++ b/test/services/transports/http-transport.test.ts @@ -1,242 +1,237 @@ -import { Hono } from "hono"; -import { describe, expect, it, vi } from "vitest"; +import { Hono } from 'hono'; +import { describe, expect, it, vi } from 'vitest'; -import type { RouteDefinition } from "../../../source/routes/types.js"; -import { ApplicationError } from "../../../source/errors/index.js"; -import { ApplicationRouter } from "../../../source/services/router.js"; -import { Broadcaster } from "../../../source/services/broadcaster.js"; -import { HttpTransportRouter } from "../../../source/services/transport/http-transport.js"; -import type { AppEnv } from "../../../source/services/transport/transport-router.js"; -import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; -import { Logger } from "../../../source/utils/logger.js"; -import { ServerHost } from "../../../source/services/server-host.js"; +import type { RouteDefinition } from '../../../source/routes/types.ts'; +import { ApplicationError } from '../../../source/errors/index.ts'; +import { ApplicationRouter } from '../../../source/services/router.ts'; +import { Broadcaster } from '../../../source/services/broadcaster.ts'; +import { HttpTransportRouter } from '../../../source/services/transport/http-transport.ts'; +import type { AppEnv } from '../../../source/services/transport/transport-router.ts'; +import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils'; +import { Logger } from '../../../source/utils/logger.ts'; +import { ServerHost } from '../../../source/services/server-host.ts'; -async function createApp( - routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]), - maxRequestBodyBytes = 1024 * 1024, -): Promise> { - const debug = new Logger("http-transport-test"); - const broadcaster = new Broadcaster(debug); - const resolvedRoutes = - typeof routes === "function" ? routes(broadcaster) : routes; - const router = await ApplicationRouter.create([ - { - async getRoutes() { - return resolvedRoutes; - }, - }, - ]); - const transport = new HttpTransportRouter(router, debug); - const app = new Hono(); - - app.onError(HttpTransportRouter.createErrorHandler(debug)); - app.use("*", ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug)); - app.use("*", HttpTransportRouter.createExtJsonMiddleware(debug)); - transport.register(app); - return app; -} - -describe("HttpTransportRouter", () => { - it("runs normal HTTP through a non-streaming route stream", async () => { - const app = await createApp([ - { - url: "/echo", - handler: async (stream) => stream.send(stream.body), - }, - ]); - const value = new Uint8Array([1, 2, 3]); - - const response = await app.request("/echo", { - method: "POST", - headers: { "content-type": "application/json" }, - body: toExtendedJson({ value }), - }); - - expect(response.status).toBe(200); - expect(fromExtendedJson(await response.text())).toEqual({ value }); - }); - - it("returns 204 when a normal HTTP route sends nothing", async () => { - const app = await createApp([ - { - url: "/nothing", - handler: () => undefined, - }, - ]); - - const response = await app.request("/nothing", { method: "POST" }); - - expect(response.status).toBe(204); - expect(await response.text()).toBe(""); - }); - - it("returns normalized errors for non-streaming requests", async () => { - const app = await createApp([]); - - const missing = await app.request("/missing", { method: "POST" }); - expect(missing.status).toBe(404); - expect(await missing.json()).toEqual({ - statusCode: 404, - error: "No route found for /missing", - }); - - const invalid = await app.request("/missing", { - method: "POST", - headers: { "content-type": "application/json" }, - body: "{", - }); - expect(invalid.status).toBe(400); - expect(await invalid.json()).toEqual({ - statusCode: 400, - error: "Invalid JSON in request body", - }); - }); - - it("rejects subscribe when normal HTTP has no streaming capability", async () => { - const app = await createApp((broadcaster) => [ - { - url: "/items/subscribe", - handler: async (stream) => { - await broadcaster.subscribe(stream, ["items"]); - }, - }, - ]); - - const response = await app.request("/items/subscribe", { method: "POST" }); - - expect(response.status).toBe(406); - expect(await response.json()).toMatchObject({ statusCode: 406 }); - }); - - it("sends SSE route errors as events and closes only that stream", async () => { - const app = await createApp([ - { - url: "/items/subscribe", - handler: () => { - throw new Error("private storage failure"); - }, - }, - ]); - - const response = await app.request("/items/subscribe", { - method: "POST", - headers: { accept: "text/event-stream" }, - }); - const events = await response.text(); - - expect(response.status).toBe(200); - expect(events).toContain("event: error"); - expect(events).toContain( - 'data: {"statusCode":500,"error":"Internal Server Error"}', - ); - expect(events).not.toContain("private storage failure"); - }); - - it("sends a normal route as one SSE response event and then closes", async () => { - const app = await createApp([ - { - url: "/echo", - handler: (stream) => stream.send({ ok: true }), - }, - ]); - - const response = await app.request("/echo", { - method: "POST", - headers: { accept: "text/event-stream" }, - }); - const events = await response.text(); - - expect(response.status).toBe(200); - expect(events).toContain("event: response"); - expect(events).toContain('data: {"ok":true}'); - }); - - it("keeps SSE open until the route's subscription promise resolves", async () => { - let removeSubscription: () => Promise = async () => undefined; - let markSubscribed: () => void = () => undefined; - const subscribed = new Promise((resolve) => { - markSubscribed = resolve; - }); - const app = await createApp((broadcaster) => [ - { - url: "/items/subscribe", - handler: async (stream) => { - const topics = ["items"]; - removeSubscription = () => broadcaster.unsubscribe(stream, topics); - - const removed = broadcaster.subscribe(stream, topics); - markSubscribed(); - await removed; - }, - }, - ]); - - const response = await app.request("/items/subscribe", { - method: "POST", - headers: { accept: "text/event-stream" }, - }); - const body = response.text(); - const completed = vi.fn(); - void body.then(completed); - - await subscribed; - await Promise.resolve(); - expect(completed).not.toHaveBeenCalled(); - - await removeSubscription(); - - expect(await body).toBe(""); - expect(completed).toHaveBeenCalledOnce(); - }); - - it("rejects unsubscribe over non-bidirectional SSE", async () => { - const app = await createApp((broadcaster) => [ - { - url: "/items/unsubscribe", - handler: async (stream) => { - if (!stream.bidirectional) { - throw new ApplicationError( - 400, - "This route requires an existing bidirectional stream", - ); - } - - await broadcaster.unsubscribe(stream, ["items"]); - }, - }, - ]); - - const response = await app.request("/items/unsubscribe", { - method: "POST", - headers: { accept: "text/event-stream" }, - }); - const events = await response.text(); - - expect(response.status).toBe(200); - expect(events).toContain("event: error"); - expect(events).toContain('"statusCode":400'); - }); - it("rejects HTTP bodies larger than the configured byte limit", async () => { - const app = await createApp( - [ +const createApp = async ( + routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]), + maxRequestBodyBytes = 1024 * 1024, +): Promise> => { + const debug = new Logger('http-transport-test'); + const broadcaster = new Broadcaster(debug); + const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes; + const router = await ApplicationRouter.create([ { - url: "/echo", - handler: async (stream) => stream.send(stream.body), + async getRoutes(): Promise { + return resolvedRoutes; + }, }, - ], - 32, - ); + ]); + const transport = new HttpTransportRouter(router, debug); + const app = new Hono(); - const response = await app.request("/echo", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ value: "x".repeat(64) }), + app.onError(HttpTransportRouter.createErrorHandler(debug)); + app.use('*', ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, debug)); + app.use('*', HttpTransportRouter.createExtJsonMiddleware(debug)); + transport.register(app); + + return app; +}; + +describe('HttpTransportRouter', (): void => { + it('runs normal HTTP through a non-streaming route stream', async (): Promise => { + const app = await createApp([ + { + url: '/echo', + handler: async (stream): Promise => stream.send(stream.body), + }, + ]); + const value = new Uint8Array([ 1, 2, 3 ]); + + const response = await app.request('/echo', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: toExtendedJson({ value }), + }); + + expect(response.status).toBe(200); + expect(fromExtendedJson(await response.text())).toEqual({ value }); }); - expect(response.status).toBe(413); - expect(await response.json()).toEqual({ - statusCode: 413, - error: "Request body exceeds the 32 byte limit", + it('returns 204 when a normal HTTP route sends nothing', async (): Promise => { + const app = await createApp([ + { + url: '/nothing', + handler: (): void => undefined, + }, + ]); + + const response = await app.request('/nothing', { method: 'POST' }); + + expect(response.status).toBe(204); + expect(await response.text()).toBe(''); + }); + + it('returns normalized errors for non-streaming requests', async (): Promise => { + const app = await createApp([]); + + const missing = await app.request('/missing', { method: 'POST' }); + expect(missing.status).toBe(404); + expect(await missing.json()).toEqual({ + statusCode: 404, + error: 'No route found for /missing', + }); + + const invalid = await app.request('/missing', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: '{', + }); + expect(invalid.status).toBe(400); + expect(await invalid.json()).toEqual({ + statusCode: 400, + error: 'Invalid JSON in request body', + }); + }); + + it('rejects subscribe when normal HTTP has no streaming capability', async (): Promise => { + const app = await createApp((broadcaster) => [ + { + url: '/items/subscribe', + handler: async (stream): Promise => { + await broadcaster.subscribe(stream, [ 'items' ]); + }, + }, + ]); + + const response = await app.request('/items/subscribe', { method: 'POST' }); + + expect(response.status).toBe(406); + expect(await response.json()).toMatchObject({ statusCode: 406 }); + }); + + it('sends SSE route errors as events and closes only that stream', async (): Promise => { + const app = await createApp([ + { + url: '/items/subscribe', + handler: (): void => { + throw new Error('private storage failure'); + }, + }, + ]); + + const response = await app.request('/items/subscribe', { + method: 'POST', + headers: { accept: 'text/event-stream' }, + }); + const events = await response.text(); + + expect(response.status).toBe(200); + expect(events).toContain('event: error'); + expect(events).toContain('data: {"statusCode":500,"error":"Internal Server Error"}'); + expect(events).not.toContain('private storage failure'); + }); + + it('sends a normal route as one SSE response event and then closes', async (): Promise => { + const app = await createApp([ + { + url: '/echo', + handler: (stream): Promise => stream.send({ ok: true }), + }, + ]); + + const response = await app.request('/echo', { + method: 'POST', + headers: { accept: 'text/event-stream' }, + }); + const events = await response.text(); + + expect(response.status).toBe(200); + expect(events).toContain('event: response'); + expect(events).toContain('data: {"ok":true}'); + }); + + it("keeps SSE open until the route's subscription promise resolves", async (): Promise => { + let removeSubscription: () => Promise = async () => undefined; + let markSubscribed: () => void = () => undefined; + const subscribed = new Promise((resolve) => { + markSubscribed = resolve; + }); + const app = await createApp((broadcaster) => [ + { + url: '/items/subscribe', + handler: async (stream): Promise => { + const topics = [ 'items' ]; + removeSubscription = (): Promise => broadcaster.unsubscribe(stream, topics); + + const removed = broadcaster.subscribe(stream, topics); + markSubscribed(); + await removed; + }, + }, + ]); + + const response = await app.request('/items/subscribe', { + method: 'POST', + headers: { accept: 'text/event-stream' }, + }); + const body = response.text(); + const completed = vi.fn(); + void body.then(completed); + + await subscribed; + await Promise.resolve(); + expect(completed).not.toHaveBeenCalled(); + + await removeSubscription(); + + expect(await body).toBe(''); + expect(completed).toHaveBeenCalledOnce(); + }); + + it('rejects unsubscribe over non-bidirectional SSE', async (): Promise => { + const app = await createApp((broadcaster) => [ + { + url: '/items/unsubscribe', + handler: async (stream): Promise => { + if (!stream.bidirectional) { + throw new ApplicationError(400, 'This route requires an existing bidirectional stream'); + } + + await broadcaster.unsubscribe(stream, [ 'items' ]); + }, + }, + ]); + + const response = await app.request('/items/unsubscribe', { + method: 'POST', + headers: { accept: 'text/event-stream' }, + }); + const events = await response.text(); + + expect(response.status).toBe(200); + expect(events).toContain('event: error'); + expect(events).toContain('"statusCode":400'); + }); + it('rejects HTTP bodies larger than the configured byte limit', async (): Promise => { + const app = await createApp( + [ + { + url: '/echo', + handler: async (stream): Promise => stream.send(stream.body), + }, + ], + 32, + ); + + const response = await app.request('/echo', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ value: 'x'.repeat(64) }), + }); + + expect(response.status).toBe(413); + expect(await response.json()).toEqual({ + statusCode: 413, + error: 'Request body exceeds the 32 byte limit', + }); }); - }); });