363 lines
14 KiB
TypeScript
363 lines
14 KiB
TypeScript
import {
|
|
MODULE_REALTIME_MAX_CHANNELS_PER_MODULE,
|
|
MODULE_REALTIME_MAX_PAYLOAD_BYTES,
|
|
MODULE_REALTIME_MAX_SUBSCRIBERS_PER_CHANNEL,
|
|
ModuleRealtimeError,
|
|
moduleIdSchema,
|
|
parsePermissionId,
|
|
} from "@ekklesia/contracts";
|
|
import { describe, expect, it, vi } from "vitest";
|
|
import {
|
|
ModuleRealtimeRegistrationError,
|
|
StagedModuleRealtimeRegistry,
|
|
serializeModuleRealtimePayload,
|
|
} from "../src/modules/realtime.js";
|
|
|
|
const permission = parsePermissionId("announcements.slideshow.view");
|
|
|
|
function registeredRegistry(
|
|
modules: Readonly<Record<string, readonly string[]>>,
|
|
): StagedModuleRealtimeRegistry {
|
|
const registry = new StagedModuleRealtimeRegistry();
|
|
for (const [rawModuleId, channels] of Object.entries(modules)) {
|
|
const moduleId = moduleIdSchema.parse(rawModuleId);
|
|
const stage = registry.createRegistrationStage(moduleId, new Set([permission]));
|
|
for (const channel of channels) {
|
|
stage.realtime.channel({ channel, permission });
|
|
}
|
|
stage.commit();
|
|
}
|
|
return registry;
|
|
}
|
|
|
|
function expectRealtimeCode(callback: () => unknown, code: string): void {
|
|
try {
|
|
callback();
|
|
} catch (error) {
|
|
expect(error).toBeInstanceOf(ModuleRealtimeError);
|
|
expect(error).toMatchObject({ code });
|
|
return;
|
|
}
|
|
throw new Error(`Expected realtime error ${code}.`);
|
|
}
|
|
|
|
describe("module realtime registry", () => {
|
|
it("publishes immutable module-scoped channel inventory", () => {
|
|
const staged = registeredRegistry({ announcements: ["state"], calendar: ["schedule"] });
|
|
const registry = staged.publish();
|
|
|
|
expect(registry.channels).toEqual([
|
|
{ moduleId: "announcements", channel: "state", permission },
|
|
{ moduleId: "calendar", channel: "schedule", permission },
|
|
]);
|
|
expect(Object.isFrozen(registry.channels)).toBe(true);
|
|
expect(Object.isFrozen(registry.channels[0])).toBe(true);
|
|
|
|
staged.publisherFor(moduleIdSchema.parse("announcements")).publish("state", { mode: "on" });
|
|
staged.publisherFor(moduleIdSchema.parse("calendar")).publish("schedule", { mode: "off" });
|
|
expect(
|
|
JSON.parse(
|
|
registry.subscribe(moduleIdSchema.parse("announcements"), "state").snapshot
|
|
.serializedPayload,
|
|
),
|
|
).toEqual({ mode: "on" });
|
|
expect(
|
|
JSON.parse(
|
|
registry.subscribe(moduleIdSchema.parse("calendar"), "schedule").snapshot.serializedPayload,
|
|
),
|
|
).toEqual({ mode: "off" });
|
|
expectRealtimeCode(
|
|
() =>
|
|
staged
|
|
.publisherFor(moduleIdSchema.parse("announcements"))
|
|
.publish("schedule", { mode: "off" }),
|
|
"REALTIME_CHANNEL_UNDECLARED",
|
|
);
|
|
expectRealtimeCode(
|
|
() => staged.publisherFor(moduleIdSchema.parse("calendar")).publish("state", { mode: "on" }),
|
|
"REALTIME_CHANNEL_UNDECLARED",
|
|
);
|
|
});
|
|
|
|
it("captures an immediate snapshot, preserves ordered live updates, and reconnects at latest", async () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
const publisher = staged.publisherFor(moduleId);
|
|
const registry = staged.publish();
|
|
publisher.publish("state", { currentSlide: 0 });
|
|
|
|
const first = registry.subscribe(moduleId, "state");
|
|
expect(first.snapshot).toEqual({ sequence: 1, serializedPayload: '{"currentSlide":0}' });
|
|
|
|
const secondPublication = first.read();
|
|
publisher.publish("state", { currentSlide: 1 });
|
|
await expect(secondPublication).resolves.toEqual({
|
|
sequence: 2,
|
|
serializedPayload: '{"currentSlide":1}',
|
|
});
|
|
const thirdPublication = first.read();
|
|
publisher.publish("state", { currentSlide: 2 });
|
|
await expect(thirdPublication).resolves.toEqual({
|
|
sequence: 3,
|
|
serializedPayload: '{"currentSlide":2}',
|
|
});
|
|
|
|
const reconnected = registry.subscribe(moduleId, "state");
|
|
expect(reconnected.snapshot).toEqual({
|
|
sequence: 3,
|
|
serializedPayload: '{"currentSlide":2}',
|
|
});
|
|
first.close();
|
|
reconnected.close();
|
|
});
|
|
|
|
it("coalesces a slow subscriber to its newest single pending update", async () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
const publisher = staged.publisherFor(moduleId);
|
|
publisher.publish("state", { value: 0 });
|
|
const subscription = staged.publish().subscribe(moduleId, "state");
|
|
|
|
for (let value = 1; value <= 1_000; value += 1) {
|
|
publisher.publish("state", { value });
|
|
}
|
|
await expect(subscription.read()).resolves.toEqual({
|
|
sequence: 1_001,
|
|
serializedPayload: '{"value":1000}',
|
|
});
|
|
subscription.close();
|
|
});
|
|
|
|
it("bounds subscribers and releases every slot on close", () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
staged.publisherFor(moduleId).publish("state", null);
|
|
const registry = staged.publish();
|
|
const subscriptions = Array.from({ length: MODULE_REALTIME_MAX_SUBSCRIBERS_PER_CHANNEL }, () =>
|
|
registry.subscribe(moduleId, "state"),
|
|
);
|
|
expectRealtimeCode(() => registry.subscribe(moduleId, "state"), "REALTIME_SUBSCRIBER_LIMIT");
|
|
subscriptions[0]?.close();
|
|
expect(() => registry.subscribe(moduleId, "state")).not.toThrow();
|
|
for (const subscription of subscriptions) {
|
|
subscription.close();
|
|
}
|
|
});
|
|
|
|
it("closes pending readers and permanently deactivates publishers during shutdown", async () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
const publisher = staged.publisherFor(moduleId);
|
|
publisher.publish("state", { mode: "on" });
|
|
const registry = staged.publish();
|
|
const subscription = registry.subscribe(moduleId, "state");
|
|
const pending = subscription.read();
|
|
|
|
staged.deactivateAll();
|
|
await expect(pending).resolves.toBeUndefined();
|
|
expectRealtimeCode(
|
|
() => publisher.publish("state", { mode: "off" }),
|
|
"REALTIME_PUBLISHER_STOPPED",
|
|
);
|
|
expectRealtimeCode(() => registry.subscribe(moduleId, "state"), "REALTIME_UNAVAILABLE");
|
|
});
|
|
|
|
it("can close transport subscriptions while an in-flight HTTP publisher finishes", async () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
const publisher = staged.publisherFor(moduleId);
|
|
publisher.publish("state", { mode: "on" });
|
|
const registry = staged.publish();
|
|
const pending = registry.subscribe(moduleId, "state").read();
|
|
|
|
registry.closeSubscriptions();
|
|
|
|
await expect(pending).resolves.toBeUndefined();
|
|
expect(() => publisher.publish("state", { mode: "off" })).not.toThrow();
|
|
expectRealtimeCode(() => registry.subscribe(moduleId, "state"), "REALTIME_UNAVAILABLE");
|
|
|
|
staged.deactivateAll();
|
|
expectRealtimeCode(
|
|
() => publisher.publish("state", { mode: "on" }),
|
|
"REALTIME_PUBLISHER_STOPPED",
|
|
);
|
|
});
|
|
|
|
it("deactivates only the module whose reverse-order stop is beginning", () => {
|
|
const staged = registeredRegistry({ announcements: ["state"], calendar: ["schedule"] });
|
|
const announcementsId = moduleIdSchema.parse("announcements");
|
|
const calendarId = moduleIdSchema.parse("calendar");
|
|
const announcements = staged.publisherFor(announcementsId);
|
|
const calendar = staged.publisherFor(calendarId);
|
|
announcements.publish("state", { mode: "on" });
|
|
calendar.publish("schedule", { date: "2026-08-23" });
|
|
|
|
staged.deactivateModule(announcementsId);
|
|
|
|
expectRealtimeCode(
|
|
() => announcements.publish("state", { mode: "off" }),
|
|
"REALTIME_PUBLISHER_STOPPED",
|
|
);
|
|
expect(() => calendar.publish("schedule", { date: "2026-08-24" })).not.toThrow();
|
|
});
|
|
|
|
it("serializes opaque state once so later source mutation cannot change it", () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
const slides = ["https://delivery.invalid/object?signature=stable"];
|
|
staged.publisherFor(moduleId).publish("state", { slides });
|
|
slides[0] = "https://delivery.invalid/object?signature=changed";
|
|
|
|
const snapshot = staged.publish().subscribe(moduleId, "state").snapshot;
|
|
expect(snapshot.serializedPayload).toContain("signature=stable");
|
|
expect(snapshot.serializedPayload).not.toContain("signature=changed");
|
|
});
|
|
|
|
it("accepts a named module state interface while runtime validation remains authoritative", () => {
|
|
interface SlideshowState {
|
|
readonly currentSlide: number;
|
|
readonly mode: "on" | "off";
|
|
readonly slides: readonly string[];
|
|
}
|
|
const state: SlideshowState = {
|
|
currentSlide: 4,
|
|
mode: "on",
|
|
slides: ["https://delivery.invalid/welcome.png"],
|
|
};
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
|
|
staged.publisherFor(moduleId).publish("state", state);
|
|
|
|
expect(
|
|
JSON.parse(staged.publish().subscribe(moduleId, "state").snapshot.serializedPayload),
|
|
).toEqual(state);
|
|
});
|
|
|
|
it("rejects non-JSON values and payloads above the UTF-8 limit with fixed codes", () => {
|
|
const staged = registeredRegistry({ announcements: ["state"] });
|
|
const publisher = staged.publisherFor(moduleIdSchema.parse("announcements"));
|
|
const cyclic: Record<string, unknown> = {};
|
|
cyclic.self = cyclic;
|
|
const sparse = Array.from({ length: 2 }) as unknown[];
|
|
sparse[0] = "present";
|
|
const toJson = vi.fn(() => ({ mode: "rewritten" }));
|
|
const hooked = { mode: "original" };
|
|
Object.defineProperty(hooked, "toJSON", { value: toJson });
|
|
|
|
for (const payload of [
|
|
undefined,
|
|
Number.NaN,
|
|
Number.POSITIVE_INFINITY,
|
|
1n,
|
|
new Date(),
|
|
cyclic,
|
|
sparse,
|
|
hooked,
|
|
]) {
|
|
expectRealtimeCode(() => publisher.publish("state", payload), "REALTIME_PAYLOAD_INVALID");
|
|
}
|
|
expect(toJson).not.toHaveBeenCalled();
|
|
expectRealtimeCode(
|
|
() => publisher.publish("state", "x".repeat(MODULE_REALTIME_MAX_PAYLOAD_BYTES)),
|
|
"REALTIME_PAYLOAD_TOO_LARGE",
|
|
);
|
|
});
|
|
|
|
it("enforces the serialized UTF-8 boundary including JSON escaping", () => {
|
|
const exactlyAtLimit = serializeModuleRealtimePayload(
|
|
"x".repeat(MODULE_REALTIME_MAX_PAYLOAD_BYTES - 2),
|
|
);
|
|
expect(exactlyAtLimit).toMatchObject({ success: true });
|
|
if (exactlyAtLimit.success) {
|
|
expect(new TextEncoder().encode(exactlyAtLimit.serializedPayload)).toHaveLength(
|
|
MODULE_REALTIME_MAX_PAYLOAD_BYTES,
|
|
);
|
|
}
|
|
expect(
|
|
serializeModuleRealtimePayload("x".repeat(MODULE_REALTIME_MAX_PAYLOAD_BYTES - 1)),
|
|
).toEqual({ success: false, code: "REALTIME_PAYLOAD_TOO_LARGE" });
|
|
expect(serializeModuleRealtimePayload("\0".repeat(11_000))).toEqual({
|
|
success: false,
|
|
code: "REALTIME_PAYLOAD_TOO_LARGE",
|
|
});
|
|
});
|
|
|
|
it("rejects enormous strings and containers before attempting an unbounded stringify", () => {
|
|
const hugeString = "x".repeat(1_000_000);
|
|
const hugeArray = new Array<unknown>(1_000_000).fill(0);
|
|
const stringify = vi.spyOn(JSON, "stringify");
|
|
const stringResult = serializeModuleRealtimePayload(hugeString);
|
|
const arrayResult = serializeModuleRealtimePayload(hugeArray);
|
|
const calls = [...stringify.mock.calls];
|
|
stringify.mockRestore();
|
|
|
|
expect(stringResult).toEqual({ success: false, code: "REALTIME_PAYLOAD_TOO_LARGE" });
|
|
expect(arrayResult).toEqual({ success: false, code: "REALTIME_PAYLOAD_TOO_LARGE" });
|
|
expect(calls).toEqual([]);
|
|
});
|
|
|
|
it("matches canonical JSON escaping without converting invalid values", () => {
|
|
const payload = {
|
|
controls: '\b\t\n\f\r\0"\\',
|
|
loneSurrogate: "\ud800",
|
|
unicode: "Welcome 👋 — καλημέρα",
|
|
};
|
|
expect(serializeModuleRealtimePayload(payload)).toEqual({
|
|
success: true,
|
|
serializedPayload: JSON.stringify(payload),
|
|
});
|
|
expect(serializeModuleRealtimePayload({ invalid: undefined })).toEqual({
|
|
success: false,
|
|
code: "REALTIME_PAYLOAD_INVALID",
|
|
});
|
|
});
|
|
|
|
it("validates duplicate, undeclared-permission, and finite channel registration", () => {
|
|
const moduleId = moduleIdSchema.parse("announcements");
|
|
const duplicateRegistry = new StagedModuleRealtimeRegistry();
|
|
const duplicate = duplicateRegistry.createRegistrationStage(moduleId, new Set([permission]));
|
|
duplicate.realtime.channel({ channel: "state", permission });
|
|
expect(() => duplicate.realtime.channel({ channel: "state", permission })).toThrowError(
|
|
expect.objectContaining({ code: "MODULE_REALTIME_CHANNEL_DUPLICATE" }),
|
|
);
|
|
|
|
const foreign = new StagedModuleRealtimeRegistry().createRegistrationStage(moduleId, new Set());
|
|
expect(() => foreign.realtime.channel({ channel: "state", permission })).toThrowError(
|
|
expect.objectContaining({ code: "MODULE_REALTIME_PERMISSION_UNDECLARED" }),
|
|
);
|
|
|
|
const limited = new StagedModuleRealtimeRegistry().createRegistrationStage(
|
|
moduleId,
|
|
new Set([permission]),
|
|
);
|
|
for (let index = 0; index < MODULE_REALTIME_MAX_CHANNELS_PER_MODULE; index += 1) {
|
|
limited.realtime.channel({ channel: `channel-${index}`, permission });
|
|
}
|
|
expect(() => limited.realtime.channel({ channel: "one-too-many", permission })).toThrowError(
|
|
expect.objectContaining({ code: "MODULE_REALTIME_CHANNEL_LIMIT" }),
|
|
);
|
|
});
|
|
|
|
it("keeps modules without realtime declarations fully compatible", () => {
|
|
const staged = registeredRegistry({ calendar: [] });
|
|
const moduleId = moduleIdSchema.parse("calendar");
|
|
expect(staged.publish().channels).toEqual([]);
|
|
expectRealtimeCode(
|
|
() => staged.publisherFor(moduleId).publish("state", null),
|
|
"REALTIME_CHANNEL_UNDECLARED",
|
|
);
|
|
expect(staged.missingSnapshots(moduleId)).toEqual([]);
|
|
});
|
|
|
|
it("reports payload validation without reflecting contents", () => {
|
|
const secret = "/machine/private/secret?token=do-not-leak";
|
|
const result = serializeModuleRealtimePayload({ secret, invalid: undefined });
|
|
expect(result).toEqual({ success: false, code: "REALTIME_PAYLOAD_INVALID" });
|
|
expect(JSON.stringify(result)).not.toContain(secret);
|
|
expect(() => {
|
|
throw new ModuleRealtimeRegistrationError("MODULE_REALTIME_CHANNEL_INVALID");
|
|
}).toThrow("Module realtime channel registration failed.");
|
|
});
|
|
});
|