Rename src to source
This commit is contained in:
@@ -0,0 +1,313 @@
|
||||
import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts';
|
||||
|
||||
import type { BaseStream, StreamEvent } from './stream/base-stream.ts';
|
||||
import type { Logger } from '../utils/logger.ts';
|
||||
import { ApplicationError } from '../errors/index.ts';
|
||||
|
||||
/** Request-scoped view from which the broadcaster obtains a stable connection. */
|
||||
export interface BroadcastStream {
|
||||
/** Connection identity shared by every request on the same transport session. */
|
||||
readonly connection: BaseStream;
|
||||
|
||||
/** Whether the connection can remain open to receive published events. */
|
||||
readonly streaming: boolean;
|
||||
}
|
||||
|
||||
/** One pending subscribe call and the topics whose removal will resolve it. */
|
||||
interface SubscriptionWaiter {
|
||||
/** Only topics newly introduced by this particular subscribe call. */
|
||||
readonly remainingTopics: Set<string>;
|
||||
|
||||
/** Completes the promise returned to the subscribing route. */
|
||||
readonly resolve: () => void;
|
||||
}
|
||||
|
||||
/** Topic delivery contract consumed by domain routes. */
|
||||
export abstract class BaseBroadcaster {
|
||||
/**
|
||||
* Subscribe a connection and wait until the topics added by this call are removed.
|
||||
*
|
||||
* Fully duplicate subscriptions resolve immediately.
|
||||
*
|
||||
* @param stream - The stream to subscribe.
|
||||
* @param topics - The topics to subscribe to.
|
||||
*/
|
||||
abstract subscribe(stream: BroadcastStream, topics: string[]): Promise<void>;
|
||||
|
||||
/**
|
||||
* Unsubscribes a stream from a list of topics.
|
||||
* @param stream - The stream to unsubscribe.
|
||||
* @param topics - The topics to unsubscribe from.
|
||||
*/
|
||||
abstract unsubscribe(stream: BroadcastStream, topics?: string[]): Promise<void>;
|
||||
|
||||
/**
|
||||
* Publishes an event to a topic.
|
||||
* @param topic - The topic to publish to.
|
||||
* @param event - The event to publish.
|
||||
* @returns The published event.
|
||||
*/
|
||||
abstract publish(topic: string, event: Omit<StreamEvent, 'id'>): Promise<StreamEvent>;
|
||||
|
||||
/**
|
||||
* Sends an event to a stream.
|
||||
* @param stream - The stream to send the event to.
|
||||
* @param event - The event to send.
|
||||
*/
|
||||
abstract sendEvent(stream: BaseStream, event: StreamEvent): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory topic index with reverse lookup for deterministic stream cleanup.
|
||||
*
|
||||
* A stream is stored strongly only while it has topics. The WeakSet records that
|
||||
* its close observer has already been installed without extending its lifetime.
|
||||
*/
|
||||
export class Broadcaster extends BaseBroadcaster {
|
||||
/** Namespaced diagnostic logger for subscription and publication activity. */
|
||||
private readonly debug: Logger;
|
||||
|
||||
/** Forward index: topic name to the connections receiving that topic. */
|
||||
private readonly topicStreams = new Map<string, Set<BaseStream>>();
|
||||
|
||||
/** Reverse index: connection to all topics currently attached to it. */
|
||||
private readonly streamTopics = new Map<BaseStream, Set<string>>();
|
||||
|
||||
/** Pending subscribe calls grouped by their stable connection identity. */
|
||||
private readonly subscriptionWaiters = new Map<BaseStream, Set<SubscriptionWaiter>>();
|
||||
|
||||
/** Connections which already have the single required close observer. */
|
||||
private readonly observedStreams = new WeakSet<BaseStream>();
|
||||
|
||||
/**
|
||||
* Creates a new Broadcaster.
|
||||
* @param debug - The debug logger.
|
||||
*/
|
||||
constructor(debug: Logger) {
|
||||
super();
|
||||
|
||||
// Extend the debug logger to include the broadcaster namespace.
|
||||
this.debug = debug.extend('broadcaster');
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe a connection and return a promise for this call's additions.
|
||||
*
|
||||
* Registration is synchronous. The returned promise resolves after all
|
||||
* topics newly added by this call are removed, or when the connection closes.
|
||||
* If every requested topic already exists, it resolves immediately.
|
||||
*
|
||||
* @param stream - The stream to subscribe.
|
||||
* @param topics - The topics to subscribe to.
|
||||
*/
|
||||
subscribe(stream: BroadcastStream, topics: string[]): Promise<void> {
|
||||
// Normal HTTP cannot receive later publications, so fail before
|
||||
// mutating either subscription index.
|
||||
if (!stream.streaming) {
|
||||
throw new ApplicationError(HTTP_STATUS_CODE_NOT_ACCEPTED, 'This route requires a stream-capable connection');
|
||||
}
|
||||
|
||||
// ApplicationRouteStream is request-scoped, but subscriptions must survive
|
||||
// across requests. Always index by the shared underlying connection.
|
||||
const connection = stream.connection;
|
||||
|
||||
// Reuse the connection's reverse-index entry when it already has topics.
|
||||
// A new Set is not stored until this call actually introduces a topic.
|
||||
const trackedTopics = this.streamTopics.get(connection) ?? new Set<string>();
|
||||
|
||||
// Deduplicate the request itself
|
||||
const deduplicatedTopics = Array.from(new Set(topics));
|
||||
|
||||
// Filter topics that are already subscribed to by the connection.
|
||||
const topicsToAdd = deduplicatedTopics.filter((topic) => !trackedTopics.has(topic));
|
||||
|
||||
// A fully duplicate (or empty) subscription adds no lifetime to track.
|
||||
if (topicsToAdd.length === 0) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
// Store the reverse index before installing the close observer. An
|
||||
// already-closed connection invokes onClose immediately and must be able
|
||||
// to remove the topics registered by this call.
|
||||
this.streamTopics.set(connection, trackedTopics);
|
||||
|
||||
for (const topic of topicsToAdd) {
|
||||
// Find or create the forward-index set for this topic.
|
||||
let streams = this.topicStreams.get(topic);
|
||||
|
||||
if (!streams) {
|
||||
streams = new Set();
|
||||
this.topicStreams.set(topic, streams);
|
||||
}
|
||||
|
||||
// Update both indexes together: publication uses the forward index,
|
||||
// while unsubscribe and connection cleanup use the reverse index.
|
||||
streams.add(connection);
|
||||
trackedTopics.add(topic);
|
||||
}
|
||||
|
||||
// Create the lifecycle promise returned to the route. Its waiter owns only
|
||||
// the topics added above, not duplicate topics owned by earlier calls.
|
||||
const removed = new Promise<void>((resolve) => {
|
||||
// Several non-overlapping subscription requests can remain active on
|
||||
// one WebSocket connection, so each connection stores a set of waiters.
|
||||
const waiters = this.subscriptionWaiters.get(connection) ?? new Set<SubscriptionWaiter>();
|
||||
|
||||
waiters.add({
|
||||
remainingTopics: new Set(topicsToAdd),
|
||||
resolve,
|
||||
});
|
||||
|
||||
this.subscriptionWaiters.set(connection, waiters);
|
||||
});
|
||||
|
||||
// Register the waiter before observing closure: onClose invokes its
|
||||
// callback immediately when registration races with an already-closed stream.
|
||||
if (!this.observedStreams.has(connection)) {
|
||||
// WeakSet prevents repeated subscribe requests from adding duplicate
|
||||
// close callbacks without retaining an otherwise unused connection.
|
||||
this.observedStreams.add(connection);
|
||||
|
||||
// Remote disconnect, local close, and shutdown all use the same cleanup
|
||||
// path, which also resolves every affected subscription promise.
|
||||
connection.onClose(() => this.removeSubscriptions(connection));
|
||||
}
|
||||
|
||||
// Log only the topics introduced by this call; duplicates were no-ops.
|
||||
this.debug('subscribed stream to topics %o', topicsToAdd);
|
||||
|
||||
// Keep the route dispatch pending for exactly this subscription's
|
||||
// lifetime. Duplicate calls return an already-resolved promise above.
|
||||
return removed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Unsubscribes a stream from a list of topics.
|
||||
* @param stream - The stream to unsubscribe.
|
||||
* @param topics - The topics to unsubscribe from.
|
||||
*/
|
||||
async unsubscribe(stream: BroadcastStream, topics?: string[]): Promise<void> {
|
||||
// Resolve request-scoped facades to the same stable connection key used
|
||||
// during subscribe, allowing a later WebSocket request to unsubscribe.
|
||||
const topicsToRemove = this.removeSubscriptions(stream.connection, topics);
|
||||
|
||||
// Logging remains useful even for idempotent removal of missing topics.
|
||||
this.debug('unsubscribed stream from topics %o', topicsToRemove);
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove topics from a connection and resolve affected subscription calls.
|
||||
*
|
||||
* @param connection - Stable connection stored in the topic index.
|
||||
* @param topics - Specific topics to remove, or all current topics.
|
||||
* @returns The topics considered for removal.
|
||||
*/
|
||||
private removeSubscriptions(connection: BaseStream, topics?: string[]): string[] {
|
||||
// Missing connections are valid: unsubscribe is deliberately idempotent.
|
||||
const trackedTopics = this.streamTopics.get(connection);
|
||||
|
||||
// Omitting topics means connection cleanup, so remove every tracked topic.
|
||||
const topicsToRemove = topics ?? Array.from(trackedTopics ?? []);
|
||||
|
||||
// Waiters should advance only for topics that were genuinely active.
|
||||
const removedTopics = new Set<string>();
|
||||
|
||||
for (const topic of topicsToRemove) {
|
||||
// Deleting from the reverse index reports whether this call actually
|
||||
// removed an active connection/topic relationship.
|
||||
if (trackedTopics?.delete(topic)) {
|
||||
removedTopics.add(topic);
|
||||
}
|
||||
|
||||
// Remove the same relationship from the publication index.
|
||||
this.topicStreams.get(topic)?.delete(connection);
|
||||
|
||||
// Empty topic sets have no value and would unnecessarily retain maps.
|
||||
if (this.topicStreams.get(topic)?.size === 0) {
|
||||
this.topicStreams.delete(topic);
|
||||
}
|
||||
}
|
||||
|
||||
// Stop strongly retaining connections after their final topic is removed.
|
||||
if (!trackedTopics || trackedTopics.size === 0) {
|
||||
this.streamTopics.delete(connection);
|
||||
}
|
||||
|
||||
// Resolve subscribe calls whose newly added topics have all disappeared.
|
||||
const waiters = this.subscriptionWaiters.get(connection);
|
||||
|
||||
if (waiters) {
|
||||
for (const waiter of waiters) {
|
||||
// Partial unsubscribe removes only the affected portion of each
|
||||
// waiter's outstanding topic set.
|
||||
for (const topic of removedTopics) {
|
||||
waiter.remainingTopics.delete(topic);
|
||||
}
|
||||
|
||||
// The route completes once every topic introduced by its call has
|
||||
// been removed, even if the connection still has other topics.
|
||||
if (waiter.remainingTopics.size === 0) {
|
||||
waiters.delete(waiter);
|
||||
waiter.resolve();
|
||||
}
|
||||
}
|
||||
|
||||
// Avoid retaining an empty waiter collection after all routes settle.
|
||||
if (waiters.size === 0) {
|
||||
this.subscriptionWaiters.delete(connection);
|
||||
}
|
||||
}
|
||||
|
||||
// Return the requested removal list for consistent unsubscribe logging.
|
||||
return topicsToRemove;
|
||||
}
|
||||
|
||||
/**
|
||||
* Publishes an event to a topic.
|
||||
* @param topic - The topic to publish to.
|
||||
* @param event - The event to publish.
|
||||
* @returns The published event.
|
||||
*/
|
||||
async publish(topic: string, event: Omit<StreamEvent, 'id'>): Promise<StreamEvent> {
|
||||
// Get the current timestamp.
|
||||
const timestamp = Date.now();
|
||||
|
||||
// Add an ID to the event.
|
||||
const eventWithId: StreamEvent = {
|
||||
...event,
|
||||
id: String(timestamp),
|
||||
};
|
||||
|
||||
// Copy the current subscriber set and start every send immediately.
|
||||
// Promise.all provides concurrent fan-out while still allowing publish to
|
||||
// wait until every local delivery attempt has settled.
|
||||
await Promise.all(Array.from(this.topicStreams.get(topic) ?? [], (stream) => this.sendEvent(stream, eventWithId)));
|
||||
|
||||
// Log the published event.
|
||||
this.debug('published %s to topic %s', event.type, topic);
|
||||
|
||||
return eventWithId;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends an event to a stream.
|
||||
* @param stream - The stream to send the event to.
|
||||
* @param event - The event to send.
|
||||
*/
|
||||
async sendEvent(stream: BaseStream, event: StreamEvent): Promise<void> {
|
||||
try {
|
||||
// Broadcaster messages bypass the request facade because pushed events
|
||||
// are connection-level and must not inherit a request correlation ID.
|
||||
await stream.send(event);
|
||||
} catch (error) {
|
||||
// Log the error.
|
||||
this.debug('failed to send event to stream: %O', error);
|
||||
|
||||
// A failed connection cannot receive future publications. Closing it
|
||||
// triggers the normal observer cleanup; the explicit removal also
|
||||
// makes this path safe for unusual stream implementations.
|
||||
stream.close();
|
||||
this.removeSubscriptions(stream);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
import 'dotenv/config';
|
||||
import { z } from 'zod';
|
||||
|
||||
/**
|
||||
* The configuration schema for the server.
|
||||
*/
|
||||
const configSchema = z.object({
|
||||
|
||||
/**
|
||||
* The database configuration.
|
||||
*/
|
||||
database: z.object({
|
||||
path: z.string().default('data.db'),
|
||||
}),
|
||||
|
||||
/**
|
||||
* The server configuration.
|
||||
*/
|
||||
server: z.object({
|
||||
port: z.coerce.number().int()
|
||||
.positive()
|
||||
.default(3000),
|
||||
host: z.string().default('0.0.0.0'),
|
||||
|
||||
/** Maximum encoded HTTP body or WebSocket message size in bytes. */
|
||||
maxRequestBodyBytes: z.coerce
|
||||
.number()
|
||||
.int()
|
||||
.positive()
|
||||
.default(1024 * 1024),
|
||||
cors: z
|
||||
.object({
|
||||
origin: z.string().default('*'),
|
||||
methods: z.array(z.string()).default([ 'GET', 'POST', 'PUT', 'DELETE', 'OPTIONS' ]),
|
||||
allowedHeaders: z.array(z.string()).default([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-PublicKey', 'X-Signature' ]),
|
||||
})
|
||||
.partial()
|
||||
.prefault({}),
|
||||
}),
|
||||
|
||||
/**
|
||||
* The authentication configuration.
|
||||
*/
|
||||
auth: z
|
||||
.object({
|
||||
timestampWindowMs: z.coerce
|
||||
.number()
|
||||
.int()
|
||||
.positive()
|
||||
.default(5 * 60 * 1000),
|
||||
})
|
||||
.prefault({}),
|
||||
});
|
||||
|
||||
/** Raw configuration object accepted before Zod parsing. */
|
||||
type ConfigInput = z.input<typeof configSchema>;
|
||||
|
||||
/** Fully parsed and defaulted configuration shape. */
|
||||
type ConfigSchema = z.output<typeof configSchema>;
|
||||
|
||||
/**
|
||||
* Typed, validated server configuration loaded from environment or objects.
|
||||
*/
|
||||
export class Config {
|
||||
/**
|
||||
* Creates a new Config from the environment variables.
|
||||
* @returns The Config.
|
||||
*/
|
||||
static fromEnv(): Config {
|
||||
return this.from({
|
||||
database: {
|
||||
path: process.env.DATABASE_PATH,
|
||||
},
|
||||
server: {
|
||||
port: process.env.SERVER_PORT,
|
||||
maxRequestBodyBytes: process.env.SERVER_MAX_REQUEST_BODY_BYTES,
|
||||
host: process.env.SERVER_HOST,
|
||||
cors: {
|
||||
origin: process.env.CORS_ORIGIN,
|
||||
methods: process.env.CORS_METHODS?.split(','),
|
||||
allowedHeaders: process.env.CORS_ALLOWED_HEADERS?.split(','),
|
||||
},
|
||||
},
|
||||
auth: {
|
||||
timestampWindowMs: process.env.AUTH_TIMESTAMP_WINDOW_MS ? Number(process.env.AUTH_TIMESTAMP_WINDOW_MS) : undefined,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new Config from a configuration object.
|
||||
* @param config - The configuration object.
|
||||
* @returns The Config.
|
||||
*/
|
||||
static from(config: ConfigInput): Config {
|
||||
return new Config(configSchema.parse(config));
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the database configuration.
|
||||
* @returns The database configuration.
|
||||
*/
|
||||
public get database(): Readonly<ConfigSchema['database']> {
|
||||
return this.config.database;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the server configuration.
|
||||
* @returns The server configuration.
|
||||
*/
|
||||
public get server(): Readonly<ConfigSchema['server']> {
|
||||
return this.config.server;
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the authentication configuration.
|
||||
* @returns The authentication configuration.
|
||||
*/
|
||||
public get auth(): Readonly<ConfigSchema['auth']> {
|
||||
return this.config.auth;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param config - Parsed configuration produced by the Zod schema.
|
||||
*/
|
||||
private constructor(private readonly config: ConfigSchema) {}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
import type { RouteSendOptions, RouteStream } from '../routes/types.ts';
|
||||
import type { BaseStream } from './stream/base-stream.ts';
|
||||
|
||||
/**
|
||||
* Binds one application request to a connection-level stream.
|
||||
*
|
||||
* Request correlation remains immutable even when multiple WebSocket handlers
|
||||
* execute concurrently. The underlying connection remains available to
|
||||
* connection-level services such as the broadcaster.
|
||||
*/
|
||||
export class ApplicationRouteStream implements RouteStream {
|
||||
/**
|
||||
* @param connection - Shared transport stream backing this request.
|
||||
* @param body - Transport-decoded application payload for the route handler.
|
||||
* @param requestId - Optional correlation ID for multiplexed transports.
|
||||
*/
|
||||
constructor(
|
||||
readonly connection: BaseStream,
|
||||
readonly body: unknown,
|
||||
private readonly requestId?: string,
|
||||
) {}
|
||||
|
||||
/** Whether the underlying connection can deliver server-pushed events. */
|
||||
get streaming(): boolean {
|
||||
return this.connection.streaming;
|
||||
}
|
||||
|
||||
/** Whether the underlying connection supports later unsubscribe requests. */
|
||||
get bidirectional(): boolean {
|
||||
return this.connection.bidirectional;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a response or event envelope through the shared connection.
|
||||
*
|
||||
* @param data - Response body or event payload.
|
||||
* @param options - Controls message type and HTTP status for responses.
|
||||
*/
|
||||
async send(data: unknown, options: RouteSendOptions = {}): Promise<void> {
|
||||
const type = options.type ?? 'response';
|
||||
|
||||
// Route responses carry an HTTP status and optional correlation ID.
|
||||
if (type === 'response') {
|
||||
await this.connection.send({
|
||||
...(this.requestId === undefined ? {} : { id: this.requestId }),
|
||||
type,
|
||||
statusCode: options.statusCode ?? (data === undefined ? 204 : 200),
|
||||
body: data ?? null,
|
||||
});
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Non-response messages are application events with a neutral envelope.
|
||||
await this.connection.send({
|
||||
...(this.requestId === undefined ? {} : { id: this.requestId }),
|
||||
type,
|
||||
data,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
import type { RouteDefinition, RouteModule } from '../routes/types.ts';
|
||||
import { ApplicationError } from '../errors/index.ts';
|
||||
import { ApplicationRouteStream } from './route-stream.ts';
|
||||
import type { BaseStream } from './stream/base-stream.ts';
|
||||
|
||||
/** Canonical request produced by every transport adapter. */
|
||||
export type ApplicationRequest = {
|
||||
/** Exact application route name. */
|
||||
path: string;
|
||||
|
||||
/** Transport-decoded application payload. */
|
||||
body?: unknown;
|
||||
|
||||
/** Optional correlation ID supplied by a multiplexed transport. */
|
||||
requestId?: string;
|
||||
};
|
||||
|
||||
/** Exact-match application routing shared by every wire transport. */
|
||||
export class ApplicationRouter {
|
||||
/** @param routes - Validated route table keyed by exact path. */
|
||||
private constructor(private readonly routes: ReadonlyMap<string, RouteDefinition>) {}
|
||||
|
||||
/**
|
||||
* Load and validate the complete route table before accepting traffic.
|
||||
*
|
||||
* @param routeModules - Route modules whose handlers will be registered.
|
||||
* @returns A ready-to-dispatch router instance.
|
||||
*/
|
||||
static async create(routeModules: RouteModule[]): Promise<ApplicationRouter> {
|
||||
const routes = new Map<string, RouteDefinition>();
|
||||
|
||||
// Collect routes from every module and reject duplicates at startup.
|
||||
for (const routeModule of routeModules) {
|
||||
for (const route of await routeModule.getRoutes()) {
|
||||
ApplicationRouter.assertValidPath(route.url);
|
||||
if (routes.has(route.url)) {
|
||||
throw new Error(`Duplicate application route: ${route.url}`);
|
||||
}
|
||||
|
||||
routes.set(route.url, route);
|
||||
}
|
||||
}
|
||||
|
||||
return new ApplicationRouter(routes);
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute one route using a request-scoped facade over the connection.
|
||||
*
|
||||
* @param request - Transport-normalized application request.
|
||||
* @param connection - Shared connection stream for this transport session.
|
||||
*/
|
||||
async dispatch(request: ApplicationRequest, connection: BaseStream): Promise<void> {
|
||||
const route = this.routes.get(request.path);
|
||||
if (!route) {
|
||||
throw new ApplicationError(404, `No route found for ${request.path}`);
|
||||
}
|
||||
|
||||
const stream = new ApplicationRouteStream(connection, request.body, request.requestId);
|
||||
await route.handler(stream);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enforce the deliberately small exact-path routing grammar at startup.
|
||||
*
|
||||
* @param path - Candidate route path to validate.
|
||||
*/
|
||||
private static assertValidPath(path: string): void {
|
||||
if (!path.startsWith('/') || path.length === 1 || path.includes(':') || path.includes('?') || path.includes('#')) {
|
||||
throw new Error(`Invalid application route "${path}": routes must be exact paths beginning with /`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
);
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
import { CompiledQuery, Kysely } from 'kysely';
|
||||
import { NodeNativeSqliteDialect } from 'kysely-node-native-sqlite';
|
||||
|
||||
import type { DatabaseTables } from './tables.ts';
|
||||
import type { Logger } from '../../utils/logger.ts';
|
||||
|
||||
/** Options required to open a SQLite database connection. */
|
||||
export type DatabaseOptions = {
|
||||
/** Filesystem path to the SQLite database file. */
|
||||
path: string;
|
||||
|
||||
/** Logger extended with a database namespace for diagnostics. */
|
||||
debug: Logger;
|
||||
};
|
||||
|
||||
/**
|
||||
* Thin wrapper around Kysely and NodeNativeSqliteDialect (which uses node:sqlite).
|
||||
*
|
||||
* Owns connection setup, pragma configuration, and graceful teardown.
|
||||
*/
|
||||
export class Database {
|
||||
private readonly debug: Logger;
|
||||
private readonly dialect: NodeNativeSqliteDialect;
|
||||
private readonly kysely: Kysely<DatabaseTables>;
|
||||
|
||||
/**
|
||||
* Open a SQLite database and configure it for concurrent writes.
|
||||
*
|
||||
* @param options - Database file path and debug logger.
|
||||
*/
|
||||
constructor(options: DatabaseOptions) {
|
||||
// Extend the debug logger to include the database namespace.
|
||||
this.debug = options.debug.extend('database');
|
||||
|
||||
// Create the SQLite database.
|
||||
this.dialect = new NodeNativeSqliteDialect(options.path);
|
||||
|
||||
// Create the Kysely database.
|
||||
this.kysely = new Kysely<DatabaseTables>({
|
||||
dialect: this.dialect,
|
||||
});
|
||||
|
||||
// Configure the SQLite pragmas.
|
||||
this.configurePragmas();
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets the Kysely database.
|
||||
*
|
||||
* @returns The typed Kysely query builder for DatabaseTables.
|
||||
*/
|
||||
get db(): Kysely<DatabaseTables> {
|
||||
return this.kysely;
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroys the database connection.
|
||||
*/
|
||||
async destroy(): Promise<void> {
|
||||
this.debug('destroying database connection');
|
||||
await this.kysely.destroy();
|
||||
}
|
||||
|
||||
/**
|
||||
* Configures the SQLite pragmas.
|
||||
*
|
||||
* WAL improves write concurrency; foreign keys enforce referential integrity.
|
||||
*/
|
||||
private configurePragmas(): void {
|
||||
this.debug('configuring SQLite pragmas');
|
||||
|
||||
this.kysely.executeQuery(CompiledQuery.raw('PRAGMA journal_mode = WAL'));
|
||||
this.kysely.executeQuery(CompiledQuery.raw('PRAGMA foreign_keys = ON'));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
/** Public storage module surface re-exported for application wiring. */
|
||||
export { Database } from './database.ts';
|
||||
export { MigrationService } from './migrate.ts';
|
||||
@@ -0,0 +1,56 @@
|
||||
import { promises as fs } from 'node:fs';
|
||||
import path from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
import { FileMigrationProvider, Migrator } from 'kysely/migration';
|
||||
import type { Database } from './database.ts';
|
||||
import type { Logger } from '../../utils/logger.ts';
|
||||
|
||||
/**
|
||||
* Applies versioned schema migrations from the on-disk migrations folder.
|
||||
*/
|
||||
export class MigrationService {
|
||||
private readonly debug: Logger;
|
||||
private readonly migrator: Migrator;
|
||||
|
||||
/**
|
||||
* @param database - Open database whose schema will be migrated.
|
||||
* @param debug - Root logger extended with a migrations namespace.
|
||||
*/
|
||||
constructor(database: Database, debug: Logger) {
|
||||
// Extend the debug logger to include the migrations namespace.
|
||||
this.debug = debug.extend('migrations');
|
||||
|
||||
// Resolve the migrations directory relative to this module file.
|
||||
const currentFilePath = fileURLToPath(import.meta.url);
|
||||
const currentDirectory = path.dirname(currentFilePath);
|
||||
const migrationsPath = path.join(currentDirectory, 'migrations');
|
||||
|
||||
// Create the migrator backed by filesystem migration files.
|
||||
this.migrator = new Migrator({
|
||||
db: database.db,
|
||||
provider: new FileMigrationProvider({
|
||||
fs,
|
||||
path,
|
||||
migrationFolder: migrationsPath,
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Migrates the database to the latest version.
|
||||
*
|
||||
* Throws when any pending migration fails so startup can abort cleanly.
|
||||
*/
|
||||
async migrateToLatest(): Promise<void> {
|
||||
this.debug('migrating database to latest');
|
||||
const { error } = await this.migrator.migrateToLatest();
|
||||
|
||||
if (error) {
|
||||
const errorInstance = error instanceof Error ? error : new Error(String(error));
|
||||
this.debug('migration failed: %O', errorInstance);
|
||||
throw errorInstance;
|
||||
}
|
||||
|
||||
this.debug('database migrations complete');
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import type { Kysely } from 'kysely';
|
||||
import { sql } from 'kysely';
|
||||
import type { DatabaseTables } from '../tables.ts';
|
||||
|
||||
/**
|
||||
* Helper for converting the current time to a millisecond timestamp.
|
||||
*
|
||||
* @returns SQLite expression producing the current time in milliseconds.
|
||||
*/
|
||||
const millisecondTime = sql`(CAST(unixepoch('subsec') * 1000 AS INTEGER))`;
|
||||
|
||||
/**
|
||||
* Creates the resource_data table.
|
||||
*
|
||||
* @param db - Kysely database to apply the migration against.
|
||||
*/
|
||||
export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
|
||||
// Composite primary key enforces one blob slot per (resource, public key).
|
||||
await db.schema
|
||||
.createTable('resource_data')
|
||||
.ifNotExists()
|
||||
.addColumn('resource_id', 'text', (col) => col.notNull())
|
||||
.addColumn('public_key', 'text', (col) => col.notNull())
|
||||
.addColumn('blob', 'blob', (col) => col.notNull())
|
||||
.addColumn('timestamp', 'integer', (col) => col.notNull().defaultTo(millisecondTime))
|
||||
.addPrimaryKeyConstraint('pk_resource_data', ['resource_id', 'public_key'])
|
||||
.execute();
|
||||
};
|
||||
|
||||
/**
|
||||
* Drops the resource_data table.
|
||||
*
|
||||
* @param db - Kysely database to apply the rollback against.
|
||||
*/
|
||||
export const down = async (db: Kysely<DatabaseTables>): Promise<void> => {
|
||||
await db.schema.dropTable('resource_data').ifExists().execute();
|
||||
};
|
||||
@@ -0,0 +1,29 @@
|
||||
import type { ColumnType } from 'kysely';
|
||||
|
||||
/** Kysely column type for millisecond epoch timestamps stored as integers. */
|
||||
export type Timestamp = ColumnType<number, number | undefined, number | undefined>;
|
||||
|
||||
/** Kysely column type for binary blobs accepting Buffer or Uint8Array on insert. */
|
||||
export type BlobColumn = ColumnType<Buffer, Buffer | Uint8Array, Buffer>;
|
||||
|
||||
/**
|
||||
* One row per (resource_id, public_key). Each publicKey owns a slot within a shared resource.
|
||||
*/
|
||||
export interface ResourceDataTable {
|
||||
/** Shared resource identifier grouping related instances. */
|
||||
resource_id: string;
|
||||
|
||||
/** Owner identity for this instance slot within the resource. */
|
||||
public_key: string;
|
||||
|
||||
/** Opaque serialized resource payload. */
|
||||
blob: BlobColumn;
|
||||
|
||||
/** Millisecond timestamp of the last write. */
|
||||
timestamp: Timestamp;
|
||||
}
|
||||
|
||||
/** Complete Kysely schema mapping for the sync server database. */
|
||||
export interface DatabaseTables {
|
||||
resource_data: ResourceDataTable;
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
/** A normal request/response result before transport encoding. */
|
||||
export type StreamResponse = {
|
||||
/** Optional correlation ID for multiplexed transports. */
|
||||
id?: string;
|
||||
|
||||
/** Discriminator marking this message as a route response. */
|
||||
type: 'response';
|
||||
|
||||
/** HTTP-equivalent status code for the response body. */
|
||||
statusCode: number;
|
||||
|
||||
/** Serialized response payload. */
|
||||
body: unknown;
|
||||
};
|
||||
|
||||
/** An application event before a transport applies its wire encoding. */
|
||||
export type StreamEvent = {
|
||||
/** Optional event or correlation ID. */
|
||||
id?: string;
|
||||
|
||||
/** Application-defined event type name. */
|
||||
type: string;
|
||||
|
||||
/** Serialized event payload. */
|
||||
data: unknown;
|
||||
};
|
||||
|
||||
/** Union of every message a connection stream can emit. */
|
||||
export type StreamMessage = StreamResponse | StreamEvent;
|
||||
|
||||
/**
|
||||
* Connection-level output channel shared by route requests and the broadcaster.
|
||||
*
|
||||
* A WebSocket connection can back many request-scoped RouteStream instances.
|
||||
* SSE and normal HTTP each create one connection stream per request.
|
||||
*/
|
||||
export abstract class BaseStream {
|
||||
/** Whether this connection can remain open for server events. */
|
||||
abstract readonly streaming: boolean;
|
||||
|
||||
/** Whether later requests can modify this connection's subscriptions. */
|
||||
abstract readonly bidirectional: boolean;
|
||||
|
||||
/**
|
||||
* Send one response or event using the transport's wire encoding.
|
||||
*
|
||||
* @param message - Neutral response or event envelope to encode.
|
||||
*/
|
||||
abstract send(message: StreamMessage): Promise<void>;
|
||||
|
||||
/** Close the underlying connection and notify lifecycle observers. */
|
||||
abstract close(): void;
|
||||
|
||||
/**
|
||||
* Observe closure whether it occurs locally or at the remote peer.
|
||||
*
|
||||
* @param callback - Invoked once when the connection closes.
|
||||
*/
|
||||
abstract onClose(callback: () => void): void;
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import type { WSContext } from 'hono/ws';
|
||||
import { toExtendedJson } from '@xo-cash/utils';
|
||||
|
||||
import { BaseStream, type StreamMessage } from './base-stream.ts';
|
||||
|
||||
/** WebSocket readyState value indicating an open socket. */
|
||||
const OPEN = 1;
|
||||
|
||||
/** Adapts one WebSocket connection to the shared connection-stream contract. */
|
||||
export class WSStream extends BaseStream {
|
||||
readonly streaming = true;
|
||||
readonly bidirectional = true;
|
||||
|
||||
/** Stable server-side identity used only for diagnostics. */
|
||||
readonly id: string;
|
||||
|
||||
/** Observers notified when the socket closes. */
|
||||
private closeCallbacks: Array<() => void> = [];
|
||||
|
||||
/** Guards against sends after local or remote closure. */
|
||||
private closed = false;
|
||||
|
||||
/**
|
||||
* @param ws - Minimal WebSocket surface required for send/close operations.
|
||||
*/
|
||||
constructor(private readonly ws: Pick<WSContext, 'send' | 'close' | 'readyState'>) {
|
||||
super();
|
||||
this.id = randomUUID();
|
||||
}
|
||||
|
||||
/**
|
||||
* Encode and send one Extended JSON message frame.
|
||||
*
|
||||
* @param message - Neutral response or event envelope to send.
|
||||
*/
|
||||
async send(message: StreamMessage): Promise<void> {
|
||||
if (this.closed || this.ws.readyState !== OPEN) {
|
||||
throw new Error('Cannot send to a closed WebSocket stream');
|
||||
}
|
||||
|
||||
// WebSocket carries the complete neutral response or event envelope.
|
||||
this.ws.send(toExtendedJson(message));
|
||||
}
|
||||
|
||||
/** Close the socket from the server side and notify observers. */
|
||||
close(): void {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.closed = true;
|
||||
this.ws.close();
|
||||
this.emitClose();
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a callback invoked when the socket closes.
|
||||
*
|
||||
* @param callback - Called immediately if the stream is already closed.
|
||||
*/
|
||||
onClose(callback: () => void): void {
|
||||
if (this.closed) {
|
||||
callback();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
this.closeCallbacks.push(callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* Record closure reported by WebSocket callbacks without closing the socket
|
||||
* again. This shares the same observer notification path as local closure.
|
||||
*/
|
||||
markClosed(): void {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.closed = true;
|
||||
this.emitClose();
|
||||
}
|
||||
|
||||
/** Drain and invoke all registered close observers. */
|
||||
private emitClose(): void {
|
||||
const callbacks = this.closeCallbacks;
|
||||
this.closeCallbacks = [];
|
||||
|
||||
for (const callback of callbacks) {
|
||||
callback();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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