Rename src to source
This commit is contained in:
@@ -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' },
|
||||
});
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -0,0 +1,230 @@
|
||||
import { upgradeWebSocket } from '@hono/node-server';
|
||||
import type { WebSocketServerLike } from '@hono/node-server';
|
||||
import type { Hono } from 'hono';
|
||||
import type { WSContext, WSMessageReceive } from 'hono/ws';
|
||||
import { WebSocketServer } from 'ws';
|
||||
import { toExtendedJson, fromExtendedJson } from '@xo-cash/utils';
|
||||
import { z } from 'zod';
|
||||
|
||||
import type { Logger } from '../../utils/logger.ts';
|
||||
import { ApplicationError, normalizePublicError } from '../../errors/index.ts';
|
||||
import type { ApplicationRequest, ApplicationRouter } from '../router.ts';
|
||||
import { WSStream } from '../stream/ws-stream.ts';
|
||||
import type { AppEnv, UpgradeTransportRouter } from './transport-router.ts';
|
||||
|
||||
/** Default WebSocket upgrade path for application messages. */
|
||||
const WS_ROUTE = '/ws';
|
||||
|
||||
// Strict validation prevents legacy or protocol-specific fields reaching routes.
|
||||
const wsRequestSchema = z
|
||||
.object({
|
||||
id: z.string().min(1).optional(),
|
||||
path: z.string().min(1),
|
||||
body: z.unknown().optional(),
|
||||
})
|
||||
.strict();
|
||||
|
||||
/**
|
||||
* WebSocket protocol adapter.
|
||||
*
|
||||
* One WSStream is shared by every message on a socket, allowing subscribe and
|
||||
* unsubscribe requests to operate on the same broadcaster registration.
|
||||
*/
|
||||
export class WsTransportRouter implements UpgradeTransportRouter {
|
||||
/**
|
||||
* Native Node WebSocket server used by Hono's Node adapter.
|
||||
*
|
||||
* Hono handles the HTTP upgrade and lifecycle callbacks, while the `ws`
|
||||
* server owns frame parsing and therefore enforces the payload limit.
|
||||
*/
|
||||
private readonly wsServer: WebSocketServer;
|
||||
|
||||
/** Upgrade server wired into the Node HTTP listener by ServerHost. */
|
||||
readonly websocketServer: WebSocketServerLike;
|
||||
|
||||
private readonly debug: Logger;
|
||||
|
||||
/**
|
||||
* @param router - Shared application router for route dispatch.
|
||||
* @param debug - Root logger extended with a ws-transport namespace.
|
||||
* @param maxRequestBodyBytes - Maximum encoded WebSocket message size.
|
||||
* @param url - WebSocket upgrade path exposed by this adapter.
|
||||
*/
|
||||
constructor(
|
||||
private readonly router: ApplicationRouter,
|
||||
debug: Logger,
|
||||
private readonly maxRequestBodyBytes: number,
|
||||
private readonly url: string = WS_ROUTE,
|
||||
) {
|
||||
this.debug = debug.extend('ws-transport');
|
||||
|
||||
// Hono's Node WebSocket helper delegates frame handling to `ws`; unlike
|
||||
// HTTP requests, WebSocket messages do not pass through Hono middleware.
|
||||
// maxPayload rejects oversized messages before onMessage receives them.
|
||||
this.wsServer = new WebSocketServer({
|
||||
noServer: true,
|
||||
maxPayload: this.maxRequestBodyBytes,
|
||||
});
|
||||
this.websocketServer = this.wsServer as unknown as WebSocketServerLike;
|
||||
}
|
||||
|
||||
/**
|
||||
* Register the connection endpoint and translate socket lifecycle events.
|
||||
*
|
||||
* @param server - Hono instance to attach the upgrade handler to.
|
||||
*/
|
||||
register(server: Hono<AppEnv>): void {
|
||||
server.get(
|
||||
this.url,
|
||||
upgradeWebSocket(
|
||||
() => {
|
||||
// One WSStream per socket, not per message. Broadcaster topic
|
||||
// subscriptions live on the connection stream so a later
|
||||
// /data/unsubscribe message can remove topics registered by an
|
||||
// earlier /data/subscribe on the same socket.
|
||||
let stream: WSStream | undefined;
|
||||
|
||||
return {
|
||||
onOpen: (_event, ws): void => {
|
||||
stream = new WSStream(ws);
|
||||
this.debug('socket opened: %s', stream.id);
|
||||
},
|
||||
onMessage: (event, ws): void => {
|
||||
// onOpen may not run before the first message on some stacks.
|
||||
stream ??= new WSStream(ws);
|
||||
|
||||
// Each WebSocket frame is an independent application request
|
||||
// (path + body + optional id). Dispatch is fire-and-forget so
|
||||
// multiple in-flight requests can share one connection.
|
||||
this.handleMessage(ws, stream, event.data).catch((error: unknown) => {
|
||||
this.debug('unexpected message failure: %O', error);
|
||||
});
|
||||
},
|
||||
onClose: (): void => {
|
||||
this.debug('socket closed: %s', stream?.id ?? 'unknown');
|
||||
// markClosed, not close — the socket is already gone. This
|
||||
// notifies the broadcaster's onClose hook so all topic
|
||||
// registrations for this connection are torn down.
|
||||
stream?.markClosed();
|
||||
},
|
||||
onError: (event): void => {
|
||||
this.debug('socket error: %O', event);
|
||||
stream?.markClosed();
|
||||
},
|
||||
};
|
||||
},
|
||||
{ onError: (error) => this.debug('upgrade failed: %O', error) },
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/** Stop accepting upgrades and close active sockets during shutdown. */
|
||||
async stop(): Promise<void> {
|
||||
this.debug('closing %d active WebSocket(s)', this.wsServer.clients.size);
|
||||
for (const client of this.wsServer.clients) {
|
||||
client.close(1001, 'Server shutting down');
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
this.wsServer.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispatch one independently correlated message.
|
||||
*
|
||||
* Route failures become error envelopes and intentionally leave the socket open.
|
||||
*
|
||||
* @param ws - Active WebSocket context for error replies.
|
||||
* @param stream - Shared connection stream for this socket.
|
||||
* @param message - Raw WebSocket payload from the client.
|
||||
*/
|
||||
private async handleMessage(ws: WSContext, stream: WSStream, message: WSMessageReceive): Promise<void> {
|
||||
let request: ApplicationRequest;
|
||||
|
||||
try {
|
||||
// Decode the transport envelope { id?, path, body? } and revive any
|
||||
// Extended JSON values before the message reaches application routes.
|
||||
request = await WsTransportRouter.decodeWebSocketRequest(message);
|
||||
} catch (error) {
|
||||
this.debug('invalid message: %O', error);
|
||||
// Malformed envelopes have no request id to echo back.
|
||||
this.sendError(ws, undefined, error);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
// ApplicationRouter wraps the shared connection stream in a per-request
|
||||
// RouteStream (body + requestId). A subscription dispatch intentionally
|
||||
// remains pending until a later request unsubscribes or the socket closes.
|
||||
await this.router.dispatch(request, stream);
|
||||
} catch (error) {
|
||||
this.debug('dispatch failed for %s on socket %s: %O', request.path, stream.id, error);
|
||||
// Reply with a correlated error envelope but keep the socket open.
|
||||
// The client may have other in-flight requests or active subscriptions
|
||||
// on this connection that must survive a single route failure.
|
||||
this.sendError(ws, request.requestId, error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Serialize the shared public error contract without exposing internals.
|
||||
*
|
||||
* @param ws - Active WebSocket context to send the error on.
|
||||
* @param requestId - Optional correlation ID echoed back to the client.
|
||||
* @param error - Failure to normalize into the public error shape.
|
||||
*/
|
||||
private sendError(ws: WSContext, requestId: string | undefined, error: unknown): void {
|
||||
ws.send(
|
||||
toExtendedJson({
|
||||
...(requestId === undefined ? {} : { id: requestId }),
|
||||
type: 'error',
|
||||
...normalizePublicError(error),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode the WebSocket message into an application request.
|
||||
*
|
||||
* @param message - Raw WebSocket payload from the client.
|
||||
* @returns The decoded WebSocket message as an application request.
|
||||
*/
|
||||
static async decodeWebSocketRequest(message: WSMessageReceive): Promise<ApplicationRequest> {
|
||||
const payload = await WsTransportRouter.decodeWebSocketMessage(message);
|
||||
let decoded: unknown;
|
||||
|
||||
try {
|
||||
decoded = fromExtendedJson(payload);
|
||||
} catch {
|
||||
throw new ApplicationError(400, 'Invalid JSON in WebSocket message');
|
||||
}
|
||||
|
||||
const envelope = wsRequestSchema.parse(decoded);
|
||||
|
||||
return {
|
||||
path: envelope.path,
|
||||
...(envelope.id === undefined ? {} : { requestId: envelope.id }),
|
||||
...(envelope.body === undefined ? {} : { body: envelope.body }),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Decode the WebSocket message into a string.
|
||||
*
|
||||
* @param message - Raw WebSocket payload from the client.
|
||||
* @returns The decoded WebSocket message as a string.
|
||||
*/
|
||||
static async decodeWebSocketMessage(message: WSMessageReceive): Promise<string> {
|
||||
if (typeof message === 'string') {
|
||||
return message;
|
||||
}
|
||||
|
||||
if (message instanceof Blob) {
|
||||
return Buffer.from(await message.arrayBuffer()).toString('utf8');
|
||||
}
|
||||
|
||||
return Buffer.from(message).toString('utf8');
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user