rename src to source
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
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 { Logger } from './utils/logger.ts';
|
||||
|
||||
/** Application composition root. */
|
||||
export class App {
|
||||
/**
|
||||
* Construct infrastructure and validate all routes before returning a runnable app.
|
||||
*/
|
||||
static async create(): Promise<App> {
|
||||
const config = Config.fromEnv();
|
||||
const debug = new Logger('sync-server-v2');
|
||||
debug('config loaded: %O', config);
|
||||
|
||||
// Persistence must be ready before routes accept traffic.
|
||||
const database = new Database({ path: config.database.path, debug });
|
||||
const migrations = new MigrationService(database, debug);
|
||||
await migrations.migrateToLatest();
|
||||
|
||||
// Domain services are shared across all transports and route modules.
|
||||
const broadcaster = new Broadcaster(debug);
|
||||
const routes = [];
|
||||
|
||||
// Route loading is an explicit startup phase, not first-request work.
|
||||
// ApplicationRouter.create validates every path and rejects duplicates
|
||||
// before any client can connect.
|
||||
const router = await ApplicationRouter.create(routes);
|
||||
|
||||
return new App(database, broadcaster, router);
|
||||
}
|
||||
|
||||
private stopPromise: Promise<void> | undefined;
|
||||
|
||||
constructor(
|
||||
private readonly database: Database,
|
||||
private readonly broadcaster: Broadcaster,
|
||||
private readonly router: ApplicationRouter,
|
||||
) {}
|
||||
|
||||
async start(): Promise<void> {}
|
||||
|
||||
/** Stop transports before releasing the database they may still use. */
|
||||
async stop(): Promise<void> {
|
||||
this.stopPromise ??= this.database.destroy();
|
||||
await this.stopPromise;
|
||||
}
|
||||
}
|
||||
|
||||
const app = await App.create();
|
||||
let shuttingDown = false;
|
||||
|
||||
const shutdown = (signal: NodeJS.Signals): void => {
|
||||
if (shuttingDown) {
|
||||
return;
|
||||
}
|
||||
|
||||
shuttingDown = true;
|
||||
void app.stop().catch((error) => {
|
||||
console.error('Graceful shutdown failed after ' + signal, error);
|
||||
process.exitCode = 1;
|
||||
});
|
||||
};
|
||||
|
||||
process.once('SIGINT', (): void => shutdown('SIGINT'));
|
||||
process.once('SIGTERM', (): void => shutdown('SIGTERM'));
|
||||
|
||||
try {
|
||||
await app.start();
|
||||
} catch (error: unknown) {
|
||||
await app.stop();
|
||||
throw error;
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
/**
|
||||
* HTTP status code for "Not Acceptable" error.
|
||||
*/
|
||||
export const HTTP_STATUS_CODE_NOT_ACCEPTED = 406;
|
||||
@@ -0,0 +1,14 @@
|
||||
/** An expected application failure whose message is safe to send to clients. */
|
||||
export class ApplicationError extends Error {
|
||||
/**
|
||||
* @param statusCode - HTTP-equivalent status returned to the client.
|
||||
* @param message - Client-safe error summary.
|
||||
*/
|
||||
constructor(
|
||||
readonly statusCode: number,
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = 'ApplicationError';
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
export * from './application-error.ts';
|
||||
export * from './unauthorized-error.ts';
|
||||
export * from './utils.ts';
|
||||
@@ -0,0 +1,10 @@
|
||||
/** Authentication failure whose message is safe to return to clients. */
|
||||
export class UnauthorizedError extends Error {
|
||||
/**
|
||||
* @param message - Client-safe unauthorized summary.
|
||||
*/
|
||||
constructor(message = 'Unauthorized') {
|
||||
super(message);
|
||||
this.name = 'UnauthorizedError';
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
import { z } from 'zod';
|
||||
|
||||
import { UnauthorizedError } from './unauthorized-error.ts';
|
||||
import { ApplicationError } from './application-error.ts';
|
||||
|
||||
/** Stable error payload shared by HTTP, SSE, and WebSocket. */
|
||||
export type PublicError = {
|
||||
|
||||
/** Protocol-independent status carried by every transport. */
|
||||
statusCode: number;
|
||||
|
||||
/** Client-safe summary which never exposes an unexpected exception. */
|
||||
error: string;
|
||||
|
||||
/** Structured field failures supplied only for validation errors. */
|
||||
details?: Array<{ path: string; message: string }>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Convert application failures into the common public transport contract.
|
||||
*
|
||||
* @param error - Any thrown value from a route or transport boundary.
|
||||
* @returns A sanitized error payload safe to encode on the wire.
|
||||
*/
|
||||
export const normalizePublicError = (error: unknown): PublicError => {
|
||||
if (error instanceof z.ZodError) {
|
||||
return {
|
||||
statusCode: 400,
|
||||
error: 'Validation Error',
|
||||
details: error.issues.map((issue) => ({
|
||||
path: issue.path.join('.'),
|
||||
message: issue.message,
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
if (error instanceof UnauthorizedError) {
|
||||
return { statusCode: 401, error: error.message };
|
||||
}
|
||||
|
||||
if (error instanceof ApplicationError) {
|
||||
return { statusCode: error.statusCode, error: error.message };
|
||||
}
|
||||
|
||||
// Unknown exceptions are logged by adapters, but their messages stay private.
|
||||
return { statusCode: 500, error: 'Internal Server Error' };
|
||||
};
|
||||
@@ -0,0 +1,40 @@
|
||||
import type { BaseStream } from '../services/stream/base-stream.js';
|
||||
|
||||
export type RouteSendOptions = {
|
||||
/** Defaults to `response`; any other value sends an application event. */
|
||||
type?: string;
|
||||
|
||||
/** Applies to normal responses. Defaults to 200, or 204 for undefined data. */
|
||||
statusCode?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* Request-scoped application view over an underlying transport connection.
|
||||
*
|
||||
* The body and request ID are immutable for one dispatch. Subscription state and
|
||||
* connection lifetime are shared with other requests on the same connection.
|
||||
*/
|
||||
export interface RouteStream {
|
||||
/** Connection shared by every request on the same transport session. */
|
||||
readonly connection: BaseStream;
|
||||
|
||||
readonly body: unknown;
|
||||
readonly streaming: boolean;
|
||||
readonly bidirectional: boolean;
|
||||
|
||||
send(data: unknown, options?: RouteSendOptions): Promise<void>;
|
||||
}
|
||||
|
||||
export type RouteHandler = (stream: RouteStream) => void | Promise<void>;
|
||||
|
||||
/** An exact application route with no transport-specific metadata. */
|
||||
export type RouteDefinition = {
|
||||
/** Exact route name. Parameter and wildcard syntax are not supported. */
|
||||
url: string;
|
||||
handler: RouteHandler;
|
||||
};
|
||||
|
||||
/** Supplies and validates routes during application startup. */
|
||||
export interface RouteModule {
|
||||
getRoutes(): Promise<Array<RouteDefinition>>;
|
||||
}
|
||||
@@ -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,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,66 @@
|
||||
import Debug, { type Debugger } from 'debug';
|
||||
|
||||
type LogHandler = {
|
||||
(...args: Parameters<Debugger>): void;
|
||||
extend: (namespace: string) => LogHandler;
|
||||
};
|
||||
|
||||
/**
|
||||
* Declares that Logger instances may also be invoked as functions.
|
||||
*/
|
||||
// eslint-disable-next-line
|
||||
export interface Logger {
|
||||
(...args: unknown[]): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Logger class, similar to 'debug' library but you can call `instanceof Logger` to check if a value is a Logger instance.
|
||||
*/
|
||||
// eslint-disable-next-line
|
||||
export class Logger {
|
||||
public readonly namespace!: string;
|
||||
|
||||
private readonly handler!: LogHandler;
|
||||
|
||||
public constructor(namespace: string, handler: LogHandler = Debug(namespace)) {
|
||||
/**
|
||||
* I'm going to be honest, this file is somewhat an experiment.
|
||||
* The logger from 'debug' is a fancy function with methods on it.
|
||||
* I wanted to extend that functionality to support 'extend' and also determine whether the object is a Logger instance.
|
||||
* This makes it trivial to perform a type check on the logger, since its no longer just a function. But, I wanted to keep the exact same API
|
||||
* as debug, so this uses gross, disgusting, blasphemous prototype methods to assign a function onto this class prototype.
|
||||
*/
|
||||
|
||||
// Make a function that just calls the 'debug' function with the given arguments
|
||||
const logger = ((...args: Parameters<Debugger>): void => {
|
||||
handler(...args);
|
||||
}) as Logger;
|
||||
|
||||
// Mutate the logger function to inherit from this class.
|
||||
// This allows us to use the 'instanceof' operator to check if the object is a Logger instance.
|
||||
Object.setPrototypeOf(logger, new.target.prototype);
|
||||
|
||||
// Add the namespace and the handler to the 'logger' function we defined above
|
||||
// Basically, we are combining this Class with the 'logger' function that we created above.
|
||||
Object.defineProperties(logger, {
|
||||
namespace: {
|
||||
value: namespace,
|
||||
enumerable: true,
|
||||
},
|
||||
handler: {
|
||||
value: handler,
|
||||
},
|
||||
});
|
||||
|
||||
// Instead of returning the class, we return the 'logger' function we created above.
|
||||
return logger;
|
||||
}
|
||||
|
||||
public extend(childNamespace: string): Logger {
|
||||
return new Logger(`${this.namespace}:${childNamespace}`, this.handler.extend(childNamespace));
|
||||
}
|
||||
|
||||
static isLogger(value: unknown): value is Logger {
|
||||
return value instanceof Logger;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user