Add HTTP, SSE, and server hosting

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<void> | undefined;
 
     constructor(
+        private readonly host: ServerHost,
         private readonly database: Database,
-        private readonly broadcaster: Broadcaster,
-        private readonly router: ApplicationRouter,
     ) {}
 
-    async start(): Promise<void> {}
+    async start(): Promise<void> {
+        await this.host.start();
+    }
 
     /** Stop transports before releasing the database they may still use. */
     async stop(): Promise<void> {
-        this.stopPromise ??= this.database.destroy();
+        this.stopPromise ??= (async (): Promise<void> => {
+            try {
+                await this.host.stop();
+            } finally {
+                await this.database.destroy();
+            }
+        })();
         await this.stopPromise;
     }
 }

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<typeof serve> | undefined;
+    private stopPromise: Promise<void> | undefined;
+    private readonly app: Hono<AppEnv>;
+
+    /**
+     * @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<AppEnv>();
+    }
+
+    /** Configure middleware, register adapters, and begin listening. */
+    async start(): Promise<void> {
+        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<void> {
+        // 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<void>((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<AppEnv> {
+        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,
+                );
+            },
+        });
+    }
+}

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<void> {
+        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<void> {
+        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();
+        }
+    }
+}

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<void> {
+        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);
+    }
+}

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<HonoSSEStream>();
+
+    /**
+     * @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<AppEnv>): 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<AppEnv> {
+        return async (c: HonoContext<AppEnv>, next: () => Promise<void>) => {
+            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<AppEnv>): 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<Response> {
+        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<AppEnv>, 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<void> {
+        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<AppEnv>): 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<AppEnv> {
+        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' },
+            });
+        };
+    }
+}

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<AppEnv>): Promise<void> | void;
+
+    /** Close transport-owned long-lived connections during server shutdown. */
+    stop?(): Promise<void> | 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;
+}

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<Hono<AppEnv>> {
+  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<AppEnv>();
+
+  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<void> = async () => undefined;
+    let markSubscribed: () => void = () => undefined;
+    const subscribed = new Promise<void>((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",
+    });
+  });
+});