From 69fc23a4c1c68af2b2298a60f9e4d4cb5d5da665 Mon Sep 17 00:00:00 2001 From: Harvmaster Date: Mon, 27 Jul 2026 04:24:20 +0000 Subject: [PATCH] Add HTTP, SSE, and server hosting --- src/app.ts | 22 +- src/services/server-host.ts | 166 ++++++++++++ src/services/stream/hono-sse-stream.ts | 80 ++++++ src/services/stream/http-request-stream.ts | 77 ++++++ src/services/transport/http-transport.ts | 239 +++++++++++++++++ src/services/transport/transport-router.ts | 37 +++ .../transports/http-transport.test.ts | 242 ++++++++++++++++++ 7 files changed, 858 insertions(+), 5 deletions(-) create mode 100644 src/services/server-host.ts create mode 100644 src/services/stream/hono-sse-stream.ts create mode 100644 src/services/stream/http-request-stream.ts create mode 100644 src/services/transport/http-transport.ts create mode 100644 src/services/transport/transport-router.ts create mode 100644 tests/services/transports/http-transport.test.ts diff --git a/src/app.ts b/src/app.ts index 2b2d1df..522af6a 100644 --- a/src/app.ts +++ b/src/app.ts @@ -2,6 +2,8 @@ import { Config } from './services/config.ts'; import { Database, MigrationService } from './services/storage/index.ts'; import { Broadcaster } from './services/broadcaster.ts'; import { ApplicationRouter } from './services/router.ts'; +import { HttpTransportRouter } from './services/transport/http-transport.ts'; +import { ServerHost } from './services/server-host.ts'; import { Logger } from './utils/logger.ts'; /** Application composition root. */ @@ -28,22 +30,32 @@ export class App { // before any client can connect. const router = await ApplicationRouter.create(routes); - return new App(database, broadcaster, router); + const http = new HttpTransportRouter(router, debug); + const host = new ServerHost(config, debug, [http]); + + return new App(host, database); } private stopPromise: Promise | undefined; constructor( + private readonly host: ServerHost, private readonly database: Database, - private readonly broadcaster: Broadcaster, - private readonly router: ApplicationRouter, ) {} - async start(): Promise {} + async start(): Promise { + await this.host.start(); + } /** Stop transports before releasing the database they may still use. */ async stop(): Promise { - this.stopPromise ??= this.database.destroy(); + this.stopPromise ??= (async (): Promise => { + try { + await this.host.stop(); + } finally { + await this.database.destroy(); + } + })(); await this.stopPromise; } } diff --git a/src/services/server-host.ts b/src/services/server-host.ts new file mode 100644 index 0000000..9c72f37 --- /dev/null +++ b/src/services/server-host.ts @@ -0,0 +1,166 @@ +import { Hono, type MiddlewareHandler } from 'hono'; +import { serve } from '@hono/node-server'; +import { cors } from 'hono/cors'; +import { compress } from 'hono/compress'; +import { bodyLimit } from 'hono/body-limit'; + +import type { Logger } from '../utils/logger.ts'; +import type { Config } from './config.ts'; +import type { TransportRouter, UpgradeTransportRouter, AppEnv } from './transport/transport-router.ts'; + +/** + * Owns the Hono server and installs transport adapters around shared middleware. + * + * Application routes are deliberately absent from this layer. + */ +export class ServerHost { + private readonly debug: Logger; + private server: ReturnType | undefined; + private stopPromise: Promise | undefined; + private readonly app: Hono; + + /** + * @param config - Server listen address, CORS, and related settings. + * @param debug - Root logger extended with a server-host namespace. + * @param transports - Protocol adapters registered onto the Hono app. + */ + constructor( + private readonly config: Config, + debug: Logger, + private readonly transports: TransportRouter[], + ) { + this.debug = debug.extend('server-host'); + this.app = new Hono(); + } + + /** Configure middleware, register adapters, and begin listening. */ + async start(): Promise { + const { port, host, cors: corsConfig, maxRequestBodyBytes } = this.config.server; + this.debug(`Starting on http://${host}:${port}`); + + const corsMiddleware = cors({ + origin: corsConfig.origin ?? '*', + allowMethods: corsConfig.methods ?? ['POST', 'OPTIONS'], + allowHeaders: corsConfig.allowedHeaders ?? ['Content-Type', 'Accept'], + }); + + this.app.use('*', corsMiddleware); + + // Reject oversized request bodies before any decoder or route reads them. + // Hono checks Content-Length when it can and counts streamed chunks when it + // cannot, covering both fixed-length and chunked HTTP requests. + this.app.use('*', ServerHost.limitBodySizeMiddleware(maxRequestBodyBytes, this.debug)); + + const compression = compress(); + + // Compression can buffer event streams, so never apply it to requested SSE. + this.app.use('*', async (c, next) => { + if (c.req.header('accept')?.includes('text/event-stream')) { + await next(); + + return; + } + + return compression(c, next); + }); + + this.app.get('/health', (c) => c.json({ status: 'ok' })); + + // Each transport adapter registers its own wire endpoints. + for (const transport of this.transports) { + await transport.register(this.app); + } + + // The Node Hono server accepts one WebSocketServer instance per listener. + const upgradeTransports = this.transports.filter((transport): transport is UpgradeTransportRouter => 'websocketServer' in transport); + if (upgradeTransports.length > 1) { + throw new Error('ServerHost supports only one WebSocket upgrade server'); + } + + const [upgradeTransport] = upgradeTransports; + + this.server = serve({ + fetch: this.app.fetch, + port, + hostname: host, + ...(upgradeTransport ? { websocket: { server: upgradeTransport.websocketServer } } : {}), + }); + + // Wait for the server to start listening or timeout after 10 seconds. + await Promise.race([ + // Wait for the server to start listening. + new Promise((resolve) => { + this.server?.once('listening', resolve); + }), + // Timeout after 10 seconds. + new Promise((_resolve, reject) => { + setTimeout(reject, 10000); + }), + ]); + + this.debug(`Started on http://${host}:${port}`); + } + + /** Stop accepting traffic and close all transport-owned connections. */ + async stop(): Promise { + // If we are already shutting down, return the existing promise + if (this.stopPromise) { + return this.stopPromise; + } + + // If the server hasn't been started yet or already shut down, return early + const server = this.server; + if (!server) { + return; + } + + // Create a promise that resolves when the server is closed + this.debug('stopping server'); + const closeServer = new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + + // Close the transports + 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 = undefined; + }); + + // Return the promise + return this.stopPromise; + } + + /** + * Build the server-wide Hono request-body limit policy. + * + * Keeping this policy in ServerHost ensures it runs before every HTTP adapter + * without making application routes or connection streams aware of HTTP. + * Hono preserves an accepted streamed body for downstream middleware, allowing + * the Extended JSON decoder to read it normally after validation. + * + * @param maxRequestBodyBytes - Maximum encoded HTTP request body size in bytes. + * @param debug - Logger used to record rejected requests and configured limits. + * @returns Hono middleware which returns the shared public 413 error contract. + */ + static limitBodySizeMiddleware(maxRequestBodyBytes: number, debug: Logger): MiddlewareHandler { + return bodyLimit({ + // Hono interprets maxSize as encoded request bytes, not decoded JSON size. + maxSize: maxRequestBodyBytes, + onError: (context) => { + // Do not parse the rejected payload: Hono stopped consuming it at the cap. + debug('request body exceeded configured %d byte limit: %s %s', maxRequestBodyBytes, context.req.method, context.req.path); + + // Limiting occurs before dispatch, so a normal HTTP 413 remains available. + return context.json( + { + statusCode: 413, + error: `Request body exceeds the ${maxRequestBodyBytes} byte limit`, + }, + 413, + ); + }, + }); + } +} diff --git a/src/services/stream/hono-sse-stream.ts b/src/services/stream/hono-sse-stream.ts new file mode 100644 index 0000000..ddbd3a6 --- /dev/null +++ b/src/services/stream/hono-sse-stream.ts @@ -0,0 +1,80 @@ +import { toExtendedJson } from '@xo-cash/utils'; +import { BaseStream, type StreamMessage } from './base-stream.ts'; +import type { SSEStreamingApi } from 'hono/streaming'; + +/** Maps neutral stream messages onto Hono's Server-Sent Events API. */ +export class HonoSSEStream extends BaseStream { + readonly streaming = true; + readonly bidirectional = false; + + /** Observers notified when the SSE connection closes. */ + private readonly closeCallbacks: Array<() => void> = []; + + /** Guards against sends after local closure. */ + private closed = false; + + /** + * @param stream - Hono SSE writer bound to the active HTTP response. + */ + constructor(private readonly stream: SSEStreamingApi) { + super(); + } + + /** + * Encode and write one SSE event frame. + * + * @param message - Neutral response or event envelope to send. + */ + async send(message: StreamMessage): Promise { + if (this.closed) { + throw new Error('Cannot send to a closed SSE stream'); + } + + // SSE carries type and ID as protocol fields; the payload remains Extended JSON. + await this.stream.writeSSE({ + event: message.type, + data: toExtendedJson('body' in message ? message.body : message.data), + ...(message.id === undefined ? {} : { id: message.id }), + }); + } + + /** Close the SSE response and notify lifecycle observers. */ + async close(): Promise { + if (this.closed) { + return; + } + + this.closed = true; + try { + await this.stream.close(); + } catch (error) { + const errorInstance = error instanceof Error ? error : new Error(String(error)); + throw errorInstance; + } + + this.emitClose(); + } + + /** + * Register a callback invoked when the stream closes. + * + * @param callback - Called immediately if the stream is already closed. + */ + onClose(callback: () => void): void { + if (this.closed) { + callback(); + + return; + } + + this.closeCallbacks.push(callback); + } + + /** Drain and invoke all registered close observers. */ + private emitClose(): void { + const callbacks = this.closeCallbacks.splice(0); + for (const callback of callbacks) { + callback(); + } + } +} diff --git a/src/services/stream/http-request-stream.ts b/src/services/stream/http-request-stream.ts new file mode 100644 index 0000000..b85a286 --- /dev/null +++ b/src/services/stream/http-request-stream.ts @@ -0,0 +1,77 @@ +import { BaseStream, type StreamMessage, type StreamResponse } from './base-stream.ts'; + +/** + * One-shot connection stream used to run normal HTTP through the route API. + * + * Sending stores a response until the handler returns; no bytes are committed + * early, so route errors can still replace it with the shared public error. + */ +export class HttpRequestStream extends BaseStream { + readonly streaming = false; + readonly bidirectional = false; + + /** Buffered route response, committed only after dispatch completes. */ + private response: StreamResponse | undefined; + + /** Observers notified when the logical request ends. */ + private readonly closeCallbacks: Array<() => void> = []; + + /** Guards against sends after the request lifecycle ends. */ + private closed = false; + + /** + * Buffer exactly one route response for later HTTP encoding. + * + * @param message - Must be a response envelope, not an event. + */ + async send(message: StreamMessage): Promise { + if (this.closed) { + throw new Error('Cannot send to a closed HTTP request'); + } + + if (!('body' in message)) { + throw new Error('Normal HTTP requests can only send a response'); + } + + if (this.response) { + throw new Error('Normal HTTP requests can only send one response'); + } + + this.response = message; + } + + /** + * Return the buffered response for the transport adapter to encode. + * + * @returns The stored response, or undefined if the handler sent nothing. + */ + getResponse(): StreamResponse | undefined { + return this.response; + } + + /** End the logical request and notify lifecycle observers. */ + close(): void { + if (this.closed) { + return; + } + + this.closed = true; + const callbacks = this.closeCallbacks.splice(0); + callbacks.forEach((callback) => callback()); + } + + /** + * Register a callback invoked when the request closes. + * + * @param callback - Called immediately if the request is already closed. + */ + onClose(callback: () => void): void { + if (this.closed) { + callback(); + + return; + } + + this.closeCallbacks.push(callback); + } +} diff --git a/src/services/transport/http-transport.ts b/src/services/transport/http-transport.ts new file mode 100644 index 0000000..5d7908c --- /dev/null +++ b/src/services/transport/http-transport.ts @@ -0,0 +1,239 @@ +import type { Context as HonoContext, Hono, MiddlewareHandler, ErrorHandler } from 'hono'; +import { streamSSE } from 'hono/streaming'; +import { toExtendedJson, fromExtendedJson } from '@xo-cash/utils'; + +import type { Logger } from '../../utils/logger.ts'; +import { ApplicationError, normalizePublicError } from '../../errors/index.ts'; +import type { ApplicationRequest, ApplicationRouter } from '../router.ts'; +import { HonoSSEStream } from '../stream/hono-sse-stream.ts'; +import { HttpRequestStream } from '../stream/http-request-stream.ts'; +import type { StreamResponse } from '../stream/base-stream.ts'; +import type { AppEnv, TransportRouter } from './transport-router.ts'; + +/** Hono context key where decoded Extended JSON bodies are stored. */ +const PARSED_BODY_KEY = 'parsedBody'; + +/** + * HTTP protocol adapter. + * + * Normal HTTP and SSE both enter the same application router with different + * connection-stream capabilities. + */ +export class HttpTransportRouter implements TransportRouter { + private readonly debug: Logger; + + /** SSE connections retained until their final subscription or peer closes. */ + private readonly activeSseStreams = new Set(); + + /** + * @param router - Shared application router for route dispatch. + * @param debug - Root logger extended with an http-transport namespace. + */ + constructor( + private readonly router: ApplicationRouter, + debug: Logger, + ) { + this.debug = debug.extend('http-transport'); + } + + /** + * Register the single HTTP application entry point. + * + * @param app - Hono instance to attach the POST catch-all handler to. + */ + register(app: Hono): void { + // Create HTTP error handler middleware + app.onError(HttpTransportRouter.createErrorHandler(this.debug)); + + // Create a middleware to decode Extended JSON once at the HTTP boundary before route dispatch. + app.use('*', HttpTransportRouter.createExtJsonMiddleware(this.debug)); + + // Handle HTTP POST requests + app.post('*', async (context) => { + const request = this.createRequest(context); + + // Branch to SSE when the client negotiates an event stream. + if (HttpTransportRouter.acceptsSse(context)) { + return this.openSse(context, request); + } + + return this.handleRequest(request); + }); + } + + /** + * Decode Extended JSON once at the HTTP boundary before route dispatch. + * + * @param debug - Logger used to record request metadata and parse failures. + * @returns Hono middleware that populates parsedBody on the context. + */ + static createExtJsonMiddleware(debug: Logger): MiddlewareHandler { + return async (c: HonoContext, next: () => Promise) => { + debug('request: %s %s', c.req.method, c.req.url); + + // ServerHost's Hono bodyLimit middleware has already accepted this body. + // Hono reconstructs streamed bodies after counting them, so this remains + // the only body read and requires no custom stream-management code. + const rawJsonBody = await c.req.text(); + + // Preserve exact JSON text for any future request-signature middleware. + c.set('rawJsonBody', rawJsonBody); + + // Application routes decode bodies only when the client declares JSON. + const contentType = c.req.header('content-type'); + if (contentType?.includes('application/json') && rawJsonBody.trim().length > 0) { + try { + // Extended JSON revives typed values (Uint8Array blobs, etc.) that + // plain JSON cannot represent. This is the single HTTP decode point. + const parsed = fromExtendedJson(rawJsonBody); + debug('request body: %O', parsed); + c.set(PARSED_BODY_KEY, parsed); + } catch (error) { + debug('invalid Extended JSON request: %O', error); + throw new ApplicationError(400, 'Invalid JSON in request body'); + } + } + + await next(); + }; + } + + /** + * Build a transport-neutral application request from the Hono context. + * + * @param context - Active request context with optional parsed body. + */ + private createRequest(context: HonoContext): ApplicationRequest { + const body = context.get(PARSED_BODY_KEY); + + return { + path: context.req.path, + ...(body === undefined ? {} : { body }), + }; + } + + /** + * Run normal HTTP through a non-streaming connection and close it when the + * handler returns. The connection buffers at most one route response. + * + * @param request - Application request derived from the HTTP context. + */ + private async handleRequest(request: ApplicationRequest): Promise { + const stream = new HttpRequestStream(); + + try { + await this.router.dispatch(request, stream); + + return HttpTransportRouter.toResponse(stream.getResponse()); + } finally { + stream.close(); + } + } + + /** + * Keep SSE open until the dispatched route finishes. + * + * Subscription routes remain active for their subscription lifetime. Once + * streaming starts, failures are events because HTTP status and headers have + * already been committed. + * + * @param context - Active Hono context for the SSE response. + * @param request - Application request derived from the HTTP context. + */ + private openSse(context: HonoContext, request: ApplicationRequest): Response { + return streamSSE(context, async (streamApi) => { + const stream = new HonoSSEStream(streamApi); + this.activeSseStreams.add(stream); + stream.onClose(() => this.activeSseStreams.delete(stream)); + + // Close promptly when the client aborts the underlying HTTP request. + context.req.raw.signal.addEventListener('abort', () => stream.close(), { + once: true, + }); + + try { + // Subscription routes remain pending until their broadcaster + // registration is removed. Ordinary routes return immediately. + await this.router.dispatch(request, stream); + } catch (error) { + this.debug('SSE dispatch failed for %s: %O', request.path, error); + + // SSE is a one-way stream: once streamSSE opens the response, the HTTP + // status (200) and Content-Type (text/event-stream) are already sent. + // Unlike normal HTTP, we cannot replace them with a 4xx/5xx Response. + // The client must learn about failures from an SSE event instead. + try { + await stream.send({ + type: 'error', + // Same PublicError shape as HTTP and WebSocket so clients handle + // validation, auth, and application failures uniformly. + data: normalizePublicError(error), + }); + } catch (sendError) { + // The client may have disconnected before we could deliver the error + // event — there is no further recovery path for one-way SSE. + this.debug('failed to send SSE error event: %O', sendError); + } + } finally { + // Hono closes SSE when this callback returns. Closing here is + // idempotent if the client already ended a subscription. + await stream.close(); + } + }); + } + + /** Close every retained SSE response during application shutdown. */ + async stop(): Promise { + this.debug('closing %d active SSE stream(s)', this.activeSseStreams.size); + + await Promise.all(Array.from(this.activeSseStreams).map((stream) => stream.close())); + } + + /** + * Detect whether the client requested Server-Sent Events. + * + * @param context - Active Hono request context. + */ + private static acceptsSse(context: HonoContext): boolean { + return ( + context.req + .header('accept') + ?.split(',') + .some((value) => value.trim().startsWith('text/event-stream')) ?? false + ); + } + + /** + * Convert a buffered route response into an HTTP Response. + * + * @param response - Buffered response from HttpRequestStream, if any. + */ + private static toResponse(response: StreamResponse | undefined): Response { + if (!response || response.statusCode === 204) { + return new Response(null, { status: 204 }); + } + + return new Response(toExtendedJson(response.body), { + status: response.statusCode, + headers: { 'content-type': 'application/json' }, + }); + } + + /** + * Handle failures which occur before an SSE response has been opened. + * + * @param debug - Logger used to record the underlying failure. + * @returns Hono onError handler returning the public error contract. + */ + static createErrorHandler(debug: Logger): ErrorHandler { + return (error: Error) => { + debug('HTTP dispatch failed: %O', error); + const normalized = normalizePublicError(error); + + return new Response(toExtendedJson(normalized), { + status: normalized.statusCode, + headers: { 'content-type': 'application/json' }, + }); + }; + } +} diff --git a/src/services/transport/transport-router.ts b/src/services/transport/transport-router.ts new file mode 100644 index 0000000..be506cd --- /dev/null +++ b/src/services/transport/transport-router.ts @@ -0,0 +1,37 @@ +import type { WebSocketServerLike } from '@hono/node-server'; +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; + + /** Raw JSON text preserved for signature verification. */ + rawJsonBody?: string; + }; +}; + +/** + * Host-facing adapter contract. + * + * This interface may depend on Hono because it belongs to server composition, + * not application routing. Implementations remain unaware of route modules. + */ +export interface TransportRouter { + /** + * Attach wire endpoints and middleware to the shared Hono application. + * + * @param app - Server host application to register handlers on. + */ + register(app: Hono): Promise | void; + + /** Close transport-owned long-lived connections during server shutdown. */ + stop?(): Promise | void; +} + +/** 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/tests/services/transports/http-transport.test.ts b/tests/services/transports/http-transport.test.ts new file mode 100644 index 0000000..d889a6c --- /dev/null +++ b/tests/services/transports/http-transport.test.ts @@ -0,0 +1,242 @@ +import { Hono } from "hono"; +import { describe, expect, it, vi } from "vitest"; + +import type { RouteDefinition } from "../../../src/routes/types.js"; +import { ApplicationError } from "../../../src/errors/index.js"; +import { ApplicationRouter } from "../../../src/services/router.js"; +import { Broadcaster } from "../../../src/services/broadcaster.js"; +import { HttpTransportRouter } from "../../../src/services/transport/http-transport.js"; +import type { AppEnv } from "../../../src/services/transport/transport-router.js"; +import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; +import { Logger } from "../../../src/utils/logger.js"; +import { ServerHost } from "../../../src/services/server-host.js"; + +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( + [ + { + url: "/echo", + handler: async (stream) => 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", + }); + }); +});