579 lines
19 KiB
TypeScript
579 lines
19 KiB
TypeScript
import {
|
|
MODULE_REALTIME_MAX_CHANNELS_PER_MODULE,
|
|
MODULE_REALTIME_MAX_PAYLOAD_BYTES,
|
|
MODULE_REALTIME_MAX_SUBSCRIBERS_PER_CHANNEL,
|
|
type ModuleId,
|
|
type ModuleRealtimeChannel,
|
|
type ModuleRealtimeChannelDefinitionInput,
|
|
type ModuleRealtimeChannelRegistrar,
|
|
ModuleRealtimeError,
|
|
type ModuleRealtimePublisher,
|
|
moduleIdSchema,
|
|
moduleRealtimeChannelSchema,
|
|
type PermissionId,
|
|
safeParseModuleRealtimeChannelDefinition,
|
|
} from "@ekklesia/contracts";
|
|
|
|
export type ModuleRealtimeRegistrationIssueCode =
|
|
| "MODULE_REALTIME_CHANNEL_DEFINITION_INVALID"
|
|
| "MODULE_REALTIME_CHANNEL_INVALID"
|
|
| "MODULE_REALTIME_PERMISSION_INVALID"
|
|
| "MODULE_REALTIME_PERMISSION_UNDECLARED"
|
|
| "MODULE_REALTIME_CHANNEL_DUPLICATE"
|
|
| "MODULE_REALTIME_CHANNEL_LIMIT";
|
|
|
|
/** An internal fixed-code signal. The runtime converts it to its ordinary structured issue shape. */
|
|
export class ModuleRealtimeRegistrationError extends Error {
|
|
readonly code: ModuleRealtimeRegistrationIssueCode;
|
|
|
|
constructor(code: ModuleRealtimeRegistrationIssueCode) {
|
|
super("Module realtime channel registration failed.");
|
|
this.name = "ModuleRealtimeRegistrationError";
|
|
this.code = code;
|
|
this.stack = `${this.name}: ${this.message}`;
|
|
}
|
|
}
|
|
|
|
export interface RegisteredModuleRealtimeChannel {
|
|
readonly moduleId: ModuleId;
|
|
readonly channel: ModuleRealtimeChannel;
|
|
readonly permission: PermissionId;
|
|
}
|
|
|
|
/** A host-facing opaque publication. `serializedPayload` is the exact JSON snapshot serialized once
|
|
* at publish time; hosts forward it without inspecting, mutating, or re-signing values inside it. */
|
|
export interface ModuleRealtimePublication {
|
|
readonly sequence: number;
|
|
readonly serializedPayload: string;
|
|
}
|
|
|
|
/** `snapshot` is captured atomically with installing this subscription. `read` then yields ordered
|
|
* later publications. A slow reader retains only the newest pending publication. */
|
|
export interface ModuleRealtimeSubscription {
|
|
readonly snapshot: ModuleRealtimePublication;
|
|
read(): Promise<ModuleRealtimePublication | undefined>;
|
|
close(): void;
|
|
}
|
|
|
|
export interface ModuleRealtimeRegistry {
|
|
readonly channels: readonly RegisteredModuleRealtimeChannel[];
|
|
/** Refuses new subscriptions and closes existing streams while publishers remain usable for
|
|
* ordinary HTTP requests already draining. Publisher lifetime remains Core runtime-owned. */
|
|
closeSubscriptions(): void;
|
|
publisherFor(moduleId: ModuleId): ModuleRealtimePublisher;
|
|
subscribe(moduleId: ModuleId, channel: string): ModuleRealtimeSubscription;
|
|
}
|
|
|
|
const emptyRealtimePublisher: ModuleRealtimePublisher = Object.freeze({
|
|
publish: (): never => {
|
|
throw new ModuleRealtimeError("REALTIME_CHANNEL_UNDECLARED");
|
|
},
|
|
});
|
|
|
|
/** Explicit no-channel host composition for tests and hosts that start no modules. */
|
|
export const EMPTY_MODULE_REALTIME_REGISTRY: ModuleRealtimeRegistry = Object.freeze({
|
|
channels: Object.freeze([]),
|
|
closeSubscriptions: (): void => undefined,
|
|
publisherFor: (): ModuleRealtimePublisher => emptyRealtimePublisher,
|
|
subscribe: (): never => {
|
|
throw new ModuleRealtimeError("REALTIME_CHANNEL_UNDECLARED");
|
|
},
|
|
});
|
|
|
|
interface ChannelState {
|
|
readonly definition: RegisteredModuleRealtimeChannel;
|
|
readonly subscriptions: Set<BoundedModuleRealtimeSubscription>;
|
|
publication?: ModuleRealtimePublication;
|
|
}
|
|
|
|
interface RealtimeRegistrationStage {
|
|
readonly realtime: ModuleRealtimeChannelRegistrar;
|
|
commit(): void;
|
|
}
|
|
|
|
const textEncoder = new TextEncoder();
|
|
|
|
type JsonWriteResult = "invalid" | "success" | "too-large";
|
|
|
|
/** Accumulates at most the public payload ceiling. Each encoded fragment is itself bounded, so an
|
|
* oversized value cannot create an oversized temporary serialization before being refused. */
|
|
class BoundedJsonWriter {
|
|
readonly #chunks: string[] = [];
|
|
#bytes = 0;
|
|
|
|
get remainingBytes(): number {
|
|
return MODULE_REALTIME_MAX_PAYLOAD_BYTES - this.#bytes;
|
|
}
|
|
|
|
append(fragment: string): boolean {
|
|
const bytes = textEncoder.encode(fragment).byteLength;
|
|
if (bytes > this.remainingBytes) {
|
|
return false;
|
|
}
|
|
this.#chunks.push(fragment);
|
|
this.#bytes += bytes;
|
|
return true;
|
|
}
|
|
|
|
finish(): string {
|
|
return this.#chunks.join("");
|
|
}
|
|
}
|
|
|
|
function appendRawString(
|
|
writer: BoundedJsonWriter,
|
|
value: string,
|
|
start: number,
|
|
end: number,
|
|
): boolean {
|
|
let offset = start;
|
|
while (offset < end) {
|
|
let chunkEnd = Math.min(offset + 1_024, end);
|
|
const finalCodeUnit = value.charCodeAt(chunkEnd - 1);
|
|
if (
|
|
chunkEnd < end &&
|
|
finalCodeUnit >= 0xd800 &&
|
|
finalCodeUnit <= 0xdbff &&
|
|
value.charCodeAt(chunkEnd) >= 0xdc00 &&
|
|
value.charCodeAt(chunkEnd) <= 0xdfff
|
|
) {
|
|
chunkEnd -= 1;
|
|
}
|
|
if (!writer.append(value.slice(offset, chunkEnd))) {
|
|
return false;
|
|
}
|
|
offset = chunkEnd;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
function escapedCodeUnit(codeUnit: number): string | undefined {
|
|
switch (codeUnit) {
|
|
case 0x08:
|
|
return "\\b";
|
|
case 0x09:
|
|
return "\\t";
|
|
case 0x0a:
|
|
return "\\n";
|
|
case 0x0c:
|
|
return "\\f";
|
|
case 0x0d:
|
|
return "\\r";
|
|
case 0x22:
|
|
return '\\"';
|
|
case 0x5c:
|
|
return "\\\\";
|
|
default:
|
|
return codeUnit <= 0x1f ? `\\u${codeUnit.toString(16).padStart(4, "0")}` : undefined;
|
|
}
|
|
}
|
|
|
|
/** Writes the same JSON string representation as `JSON.stringify`, including escaping lone UTF-16
|
|
* surrogates, but stops the moment the output budget cannot be met. */
|
|
function writeJsonString(writer: BoundedJsonWriter, value: string): JsonWriteResult {
|
|
// Every UTF-16 code unit contributes at least one UTF-8 byte to JSON. This rejects an enormous
|
|
// string without first copying or stringifying it.
|
|
if (value.length + 2 > writer.remainingBytes) {
|
|
return "too-large";
|
|
}
|
|
if (!writer.append('"')) {
|
|
return "too-large";
|
|
}
|
|
let rawStart = 0;
|
|
for (let index = 0; index < value.length; index += 1) {
|
|
const codeUnit = value.charCodeAt(index);
|
|
let escaped = escapedCodeUnit(codeUnit);
|
|
if (escaped === undefined && codeUnit >= 0xd800 && codeUnit <= 0xdbff) {
|
|
const following = value.charCodeAt(index + 1);
|
|
if (following >= 0xdc00 && following <= 0xdfff) {
|
|
index += 1;
|
|
continue;
|
|
}
|
|
escaped = `\\u${codeUnit.toString(16)}`;
|
|
} else if (escaped === undefined && codeUnit >= 0xdc00 && codeUnit <= 0xdfff) {
|
|
escaped = `\\u${codeUnit.toString(16)}`;
|
|
}
|
|
if (escaped === undefined) {
|
|
continue;
|
|
}
|
|
if (!appendRawString(writer, value, rawStart, index) || !writer.append(escaped)) {
|
|
return "too-large";
|
|
}
|
|
rawStart = index + 1;
|
|
}
|
|
if (!appendRawString(writer, value, rawStart, value.length) || !writer.append('"')) {
|
|
return "too-large";
|
|
}
|
|
return "success";
|
|
}
|
|
|
|
function writeJsonValue(
|
|
writer: BoundedJsonWriter,
|
|
value: unknown,
|
|
ancestors: Set<object>,
|
|
depth: number,
|
|
): JsonWriteResult {
|
|
if (value === null) {
|
|
return writer.append("null") ? "success" : "too-large";
|
|
}
|
|
if (typeof value === "string") {
|
|
return writeJsonString(writer, value);
|
|
}
|
|
if (typeof value === "boolean") {
|
|
return writer.append(value ? "true" : "false") ? "success" : "too-large";
|
|
}
|
|
if (typeof value === "number") {
|
|
if (!Number.isFinite(value)) {
|
|
return "invalid";
|
|
}
|
|
// A finite JavaScript number's JSON spelling is tiny and cannot be an unbounded allocation.
|
|
const serializedNumber = JSON.stringify(value);
|
|
return writer.append(serializedNumber) ? "success" : "too-large";
|
|
}
|
|
if (typeof value !== "object" || depth > 64 || ancestors.has(value)) {
|
|
return "invalid";
|
|
}
|
|
|
|
ancestors.add(value);
|
|
try {
|
|
// Never execute user-defined serialization hooks. A non-enumerable `toJSON` still changes
|
|
// `JSON.stringify` output, so silently ignoring it would make validation and wire semantics
|
|
// disagree. Reject the hook itself without reading or invoking its value.
|
|
if (Object.hasOwn(value, "toJSON")) {
|
|
return "invalid";
|
|
}
|
|
if (Array.isArray(value)) {
|
|
if (Object.getPrototypeOf(value) !== Array.prototype) {
|
|
return "invalid";
|
|
}
|
|
// Brackets, one single-byte value per element, and separators are the smallest possible
|
|
// representation. If even that cannot fit, do not enumerate a potentially enormous array.
|
|
const minimumBytes = value.length === 0 ? 2 : value.length * 2 + 1;
|
|
if (minimumBytes > writer.remainingBytes) {
|
|
return "too-large";
|
|
}
|
|
for (const key in value) {
|
|
if (!Object.hasOwn(value, key)) {
|
|
continue;
|
|
}
|
|
if (!/^(?:0|[1-9][0-9]*)$/u.test(key) || Number(key) >= value.length) {
|
|
return "invalid";
|
|
}
|
|
}
|
|
if (!writer.append("[")) {
|
|
return "too-large";
|
|
}
|
|
for (let index = 0; index < value.length; index += 1) {
|
|
const descriptor = Object.getOwnPropertyDescriptor(value, String(index));
|
|
if (!descriptor || !("value" in descriptor) || !descriptor.enumerable) {
|
|
return "invalid";
|
|
}
|
|
if (index > 0 && !writer.append(",")) {
|
|
return "too-large";
|
|
}
|
|
const result = writeJsonValue(writer, descriptor.value, ancestors, depth + 1);
|
|
if (result !== "success") {
|
|
return result;
|
|
}
|
|
}
|
|
return writer.append("]") ? "success" : "too-large";
|
|
}
|
|
|
|
const prototype = Object.getPrototypeOf(value);
|
|
if (prototype !== Object.prototype && prototype !== null) {
|
|
return "invalid";
|
|
}
|
|
if (!writer.append("{")) {
|
|
return "too-large";
|
|
}
|
|
let entryCount = 0;
|
|
for (const key in value) {
|
|
if (!Object.hasOwn(value, key)) {
|
|
continue;
|
|
}
|
|
const descriptor = Object.getOwnPropertyDescriptor(value, key);
|
|
if (!descriptor || !("value" in descriptor) || !descriptor.enumerable) {
|
|
return "invalid";
|
|
}
|
|
if (entryCount > 0 && !writer.append(",")) {
|
|
return "too-large";
|
|
}
|
|
const keyResult = writeJsonString(writer, key);
|
|
if (keyResult !== "success") {
|
|
return keyResult;
|
|
}
|
|
if (!writer.append(":")) {
|
|
return "too-large";
|
|
}
|
|
const valueResult = writeJsonValue(writer, descriptor.value, ancestors, depth + 1);
|
|
if (valueResult !== "success") {
|
|
return valueResult;
|
|
}
|
|
entryCount += 1;
|
|
}
|
|
return writer.append("}") ? "success" : "too-large";
|
|
} catch {
|
|
return "invalid";
|
|
} finally {
|
|
ancestors.delete(value);
|
|
}
|
|
}
|
|
|
|
export type ModuleRealtimePayloadSerializationResult =
|
|
| { readonly success: true; readonly serializedPayload: string }
|
|
| { readonly success: false; readonly code: "REALTIME_PAYLOAD_INVALID" }
|
|
| { readonly success: false; readonly code: "REALTIME_PAYLOAD_TOO_LARGE" };
|
|
|
|
/** Serializes once and reports only a fixed code. It never returns or embeds rejected input. */
|
|
export function serializeModuleRealtimePayload(
|
|
payload: unknown,
|
|
): ModuleRealtimePayloadSerializationResult {
|
|
const writer = new BoundedJsonWriter();
|
|
const result = writeJsonValue(writer, payload, new Set<object>(), 0);
|
|
if (result === "invalid") {
|
|
return { success: false, code: "REALTIME_PAYLOAD_INVALID" };
|
|
}
|
|
if (result === "too-large") {
|
|
return { success: false, code: "REALTIME_PAYLOAD_TOO_LARGE" };
|
|
}
|
|
return { success: true, serializedPayload: writer.finish() };
|
|
}
|
|
|
|
class BoundedModuleRealtimeSubscription implements ModuleRealtimeSubscription {
|
|
readonly snapshot: ModuleRealtimePublication;
|
|
#closed = false;
|
|
#pending: ModuleRealtimePublication | undefined;
|
|
#reader: ((publication: ModuleRealtimePublication | undefined) => void) | undefined;
|
|
readonly #onClose: () => void;
|
|
|
|
constructor(snapshot: ModuleRealtimePublication, onClose: () => void) {
|
|
this.snapshot = snapshot;
|
|
this.#onClose = onClose;
|
|
}
|
|
|
|
push(publication: ModuleRealtimePublication): void {
|
|
if (this.#closed) {
|
|
return;
|
|
}
|
|
const reader = this.#reader;
|
|
if (reader) {
|
|
this.#reader = undefined;
|
|
reader(publication);
|
|
return;
|
|
}
|
|
// One slot, overwritten by the newest publication. A publisher never waits for a browser.
|
|
this.#pending = publication;
|
|
}
|
|
|
|
read(): Promise<ModuleRealtimePublication | undefined> {
|
|
if (this.#closed) {
|
|
return Promise.resolve(undefined);
|
|
}
|
|
if (this.#pending) {
|
|
const publication = this.#pending;
|
|
this.#pending = undefined;
|
|
return Promise.resolve(publication);
|
|
}
|
|
if (this.#reader) {
|
|
return Promise.reject(new ModuleRealtimeError("REALTIME_UNAVAILABLE"));
|
|
}
|
|
return new Promise((resolve) => {
|
|
this.#reader = resolve;
|
|
});
|
|
}
|
|
|
|
close(): void {
|
|
if (this.#closed) {
|
|
return;
|
|
}
|
|
this.#closed = true;
|
|
this.#pending = undefined;
|
|
const reader = this.#reader;
|
|
this.#reader = undefined;
|
|
this.#onClose();
|
|
reader?.(undefined);
|
|
}
|
|
}
|
|
|
|
function stateKey(moduleId: string, channel: string): string {
|
|
return `${moduleId}\0${channel}`;
|
|
}
|
|
|
|
/** Registration is staged per module, while state and subscriptions stay private and bounded. */
|
|
export class StagedModuleRealtimeRegistry {
|
|
readonly #channels: RegisteredModuleRealtimeChannel[] = [];
|
|
readonly #publishers = new Map<ModuleId, ModuleRealtimePublisher>();
|
|
readonly #states = new Map<string, ChannelState>();
|
|
readonly #activeModules = new Set<ModuleId>();
|
|
#acceptingSubscriptions = true;
|
|
|
|
createRegistrationStage(
|
|
moduleId: ModuleId,
|
|
ownedPermissionIds: ReadonlySet<PermissionId>,
|
|
): RealtimeRegistrationStage {
|
|
const local: RegisteredModuleRealtimeChannel[] = [];
|
|
const localNames = new Set<ModuleRealtimeChannel>();
|
|
const realtime: ModuleRealtimeChannelRegistrar = Object.freeze({
|
|
channel: (definition: ModuleRealtimeChannelDefinitionInput): void => {
|
|
const parsed = safeParseModuleRealtimeChannelDefinition(definition);
|
|
if (!parsed.success) {
|
|
throw new ModuleRealtimeRegistrationError(parsed.code);
|
|
}
|
|
if (!ownedPermissionIds.has(parsed.data.permission)) {
|
|
throw new ModuleRealtimeRegistrationError("MODULE_REALTIME_PERMISSION_UNDECLARED");
|
|
}
|
|
if (localNames.has(parsed.data.channel)) {
|
|
throw new ModuleRealtimeRegistrationError("MODULE_REALTIME_CHANNEL_DUPLICATE");
|
|
}
|
|
if (local.length >= MODULE_REALTIME_MAX_CHANNELS_PER_MODULE) {
|
|
throw new ModuleRealtimeRegistrationError("MODULE_REALTIME_CHANNEL_LIMIT");
|
|
}
|
|
localNames.add(parsed.data.channel);
|
|
local.push(
|
|
Object.freeze({
|
|
channel: parsed.data.channel,
|
|
moduleId,
|
|
permission: parsed.data.permission,
|
|
}),
|
|
);
|
|
},
|
|
});
|
|
|
|
return {
|
|
realtime,
|
|
commit: (): void => {
|
|
for (const definition of local) {
|
|
this.#channels.push(definition);
|
|
this.#states.set(stateKey(moduleId, definition.channel), {
|
|
definition,
|
|
subscriptions: new Set(),
|
|
});
|
|
}
|
|
this.#activeModules.add(moduleId);
|
|
},
|
|
};
|
|
}
|
|
|
|
publisherFor(moduleId: ModuleId): ModuleRealtimePublisher {
|
|
let publisher = this.#publishers.get(moduleId);
|
|
if (!publisher) {
|
|
publisher = Object.freeze({
|
|
publish: (channel: string, payload: unknown): void => {
|
|
this.#publish(moduleId, channel, payload);
|
|
},
|
|
});
|
|
this.#publishers.set(moduleId, publisher);
|
|
}
|
|
return publisher;
|
|
}
|
|
|
|
missingSnapshots(moduleId: ModuleId): readonly ModuleRealtimeChannel[] {
|
|
return Object.freeze(
|
|
this.#channels
|
|
.filter((definition) => definition.moduleId === moduleId)
|
|
.filter(
|
|
(definition) => !this.#states.get(stateKey(moduleId, definition.channel))?.publication,
|
|
)
|
|
.map((definition) => definition.channel),
|
|
);
|
|
}
|
|
|
|
closeSubscriptions(): void {
|
|
this.#acceptingSubscriptions = false;
|
|
for (const state of this.#states.values()) {
|
|
for (const subscription of [...state.subscriptions]) {
|
|
subscription.close();
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Makes one module's captured publisher inert immediately before that module's stop hook. Other
|
|
* dependency modules remain live until reverse-order shutdown reaches their own hook. */
|
|
deactivateModule(moduleId: ModuleId): void {
|
|
this.#activeModules.delete(moduleId);
|
|
for (const state of this.#states.values()) {
|
|
if (state.definition.moduleId !== moduleId) {
|
|
continue;
|
|
}
|
|
for (const subscription of [...state.subscriptions]) {
|
|
subscription.close();
|
|
}
|
|
}
|
|
}
|
|
|
|
deactivateAll(): void {
|
|
this.closeSubscriptions();
|
|
this.#activeModules.clear();
|
|
}
|
|
|
|
publish(): ModuleRealtimeRegistry {
|
|
const channels = Object.freeze([...this.#channels]);
|
|
return Object.freeze({
|
|
channels,
|
|
closeSubscriptions: (): void => this.closeSubscriptions(),
|
|
publisherFor: (moduleId: ModuleId): ModuleRealtimePublisher => this.publisherFor(moduleId),
|
|
subscribe: (moduleId: ModuleId, channel: string): ModuleRealtimeSubscription =>
|
|
this.subscribe(moduleId, channel),
|
|
});
|
|
}
|
|
|
|
subscribe(moduleId: ModuleId, channel: string): ModuleRealtimeSubscription {
|
|
if (!this.#acceptingSubscriptions) {
|
|
throw new ModuleRealtimeError("REALTIME_UNAVAILABLE");
|
|
}
|
|
if (
|
|
!moduleIdSchema.safeParse(moduleId).success ||
|
|
!moduleRealtimeChannelSchema.safeParse(channel).success
|
|
) {
|
|
throw new ModuleRealtimeError("REALTIME_CHANNEL_INVALID");
|
|
}
|
|
const state = this.#states.get(stateKey(moduleId, channel));
|
|
if (!state) {
|
|
throw new ModuleRealtimeError("REALTIME_CHANNEL_UNDECLARED");
|
|
}
|
|
if (!state.publication) {
|
|
throw new ModuleRealtimeError("REALTIME_SNAPSHOT_UNAVAILABLE");
|
|
}
|
|
if (state.subscriptions.size >= MODULE_REALTIME_MAX_SUBSCRIBERS_PER_CHANNEL) {
|
|
throw new ModuleRealtimeError("REALTIME_SUBSCRIBER_LIMIT");
|
|
}
|
|
const subscription = new BoundedModuleRealtimeSubscription(state.publication, () => {
|
|
state.subscriptions.delete(subscription);
|
|
});
|
|
// JavaScript runs this capture and insertion without interleaving. A later publish therefore
|
|
// either becomes `snapshot` or is observed by `read`, never lost between those two operations.
|
|
state.subscriptions.add(subscription);
|
|
return subscription;
|
|
}
|
|
|
|
#publish(moduleId: ModuleId, channel: string, payload: unknown): void {
|
|
if (!this.#activeModules.has(moduleId)) {
|
|
throw new ModuleRealtimeError("REALTIME_PUBLISHER_STOPPED");
|
|
}
|
|
const parsedChannel = moduleRealtimeChannelSchema.safeParse(channel);
|
|
if (!parsedChannel.success) {
|
|
throw new ModuleRealtimeError("REALTIME_CHANNEL_INVALID");
|
|
}
|
|
const state = this.#states.get(stateKey(moduleId, parsedChannel.data));
|
|
if (!state) {
|
|
throw new ModuleRealtimeError("REALTIME_CHANNEL_UNDECLARED");
|
|
}
|
|
const serialized = serializeModuleRealtimePayload(payload);
|
|
if (!serialized.success) {
|
|
throw new ModuleRealtimeError(serialized.code);
|
|
}
|
|
const previousSequence = state.publication?.sequence ?? 0;
|
|
if (previousSequence >= Number.MAX_SAFE_INTEGER) {
|
|
throw new ModuleRealtimeError("REALTIME_UNAVAILABLE");
|
|
}
|
|
const publication = Object.freeze({
|
|
sequence: previousSequence + 1,
|
|
serializedPayload: serialized.serializedPayload,
|
|
});
|
|
state.publication = publication;
|
|
for (const subscription of state.subscriptions) {
|
|
subscription.push(publication);
|
|
}
|
|
}
|
|
}
|