529 lines
17 KiB
TypeScript
529 lines
17 KiB
TypeScript
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import path from "node:path";
|
|
import {
|
|
ANONYMOUS_PRINCIPAL,
|
|
createAuthenticatedPrincipal,
|
|
type ModuleDatabase,
|
|
type ModuleRealtimePublisher,
|
|
type ModuleRuntimeServices,
|
|
parseEkklesiaUserId,
|
|
} from "@ekklesia/contracts";
|
|
import {
|
|
createAuthorizationEvaluator,
|
|
type AuthorizationEvaluator,
|
|
type PermissionRegistry,
|
|
} from "@ekklesia/core/authorization";
|
|
import type { EkklesiaIdentityRuntime } from "@ekklesia/core/identity";
|
|
import {
|
|
composePlatformPermissionRegistry,
|
|
discoverModules,
|
|
type ModuleRealtimeRegistry,
|
|
type ModuleRealtimeSubscription,
|
|
startDiscoveredModuleRuntime,
|
|
type StartedModuleRuntime,
|
|
} from "@ekklesia/core/modules";
|
|
import { type ServerType, serve } from "@hono/node-server";
|
|
import { Hono } from "hono";
|
|
import { describe, expect, it, vi } from "vitest";
|
|
import { mountModuleRealtimeRoutes } from "../src/module-realtime-routes.js";
|
|
|
|
const allowedUserId = parseEkklesiaUserId("99999999-9999-4999-8999-999999999999");
|
|
const deniedUserId = parseEkklesiaUserId("77777777-7777-4777-8777-777777777777");
|
|
const stableDeliveryUrl =
|
|
"https://objects.example.test/announcements/slides/welcome.png?X-Amz-Signature=stable";
|
|
const TRANSPORT_DEADLINE_MS = 2_000;
|
|
|
|
async function withinTransportDeadline<Value>(
|
|
operation: Promise<Value>,
|
|
message: string,
|
|
onTimeout: () => void = () => undefined,
|
|
): Promise<Value> {
|
|
let timer: ReturnType<typeof setTimeout> | undefined;
|
|
try {
|
|
return await Promise.race([
|
|
operation,
|
|
new Promise<never>((_, reject) => {
|
|
timer = setTimeout(() => {
|
|
onTimeout();
|
|
reject(new Error(message));
|
|
}, TRANSPORT_DEADLINE_MS);
|
|
}),
|
|
]);
|
|
} finally {
|
|
clearTimeout(timer);
|
|
}
|
|
}
|
|
|
|
async function waitForTransportCondition(predicate: () => boolean, message: string): Promise<void> {
|
|
let cancelled = false;
|
|
await withinTransportDeadline(
|
|
new Promise<void>((resolve) => {
|
|
const inspect = (): void => {
|
|
if (cancelled) {
|
|
return;
|
|
}
|
|
if (predicate()) {
|
|
resolve();
|
|
return;
|
|
}
|
|
setTimeout(inspect, 5);
|
|
};
|
|
inspect();
|
|
}),
|
|
message,
|
|
() => {
|
|
cancelled = true;
|
|
},
|
|
);
|
|
}
|
|
|
|
interface FixtureControl {
|
|
readonly events: string[];
|
|
publisher?: ModuleRealtimePublisher;
|
|
}
|
|
|
|
interface LiveEvent {
|
|
readonly event: "snapshot" | "update";
|
|
readonly payload: unknown;
|
|
readonly sequence: number;
|
|
}
|
|
|
|
class LiveSseConnection {
|
|
readonly #abort: AbortController;
|
|
readonly #reader: ReadableStreamDefaultReader<Uint8Array>;
|
|
#buffer = "";
|
|
|
|
constructor(response: Response, abort: AbortController) {
|
|
if (!response.body) {
|
|
throw new Error("Realtime integration response had no body.");
|
|
}
|
|
this.#abort = abort;
|
|
this.#reader = response.body.getReader();
|
|
}
|
|
|
|
async readEvent(): Promise<LiveEvent> {
|
|
for (;;) {
|
|
const boundary = this.#buffer.indexOf("\n\n");
|
|
if (boundary >= 0) {
|
|
const block = this.#buffer.slice(0, boundary);
|
|
this.#buffer = this.#buffer.slice(boundary + 2);
|
|
if (block.startsWith(":")) {
|
|
continue;
|
|
}
|
|
const event = block
|
|
.split("\n")
|
|
.find((line) => line.startsWith("event: "))
|
|
?.slice("event: ".length);
|
|
const data = block
|
|
.split("\n")
|
|
.find((line) => line.startsWith("data: "))
|
|
?.slice("data: ".length);
|
|
if ((event !== "snapshot" && event !== "update") || data === undefined) {
|
|
throw new Error("Realtime integration received an invalid event.");
|
|
}
|
|
const envelope = JSON.parse(data) as { payload?: unknown; sequence?: unknown };
|
|
if (!Number.isSafeInteger(envelope.sequence) || (envelope.sequence as number) <= 0) {
|
|
throw new Error("Realtime integration received an invalid sequence.");
|
|
}
|
|
return { event, payload: envelope.payload, sequence: envelope.sequence as number };
|
|
}
|
|
|
|
const result = await this.#readWithinDeadline();
|
|
if (result.done) {
|
|
throw new Error("Realtime integration stream closed before the expected event.");
|
|
}
|
|
this.#buffer += new TextDecoder().decode(result.value, { stream: true });
|
|
}
|
|
}
|
|
|
|
async expectClosed(): Promise<void> {
|
|
const result = await this.#readWithinDeadline();
|
|
expect(result.done).toBe(true);
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
this.#abort.abort();
|
|
await withinTransportDeadline(
|
|
// Abort and cancel race by design; a rejection means one of them already released the
|
|
// transport. A deadline still propagates and fails the test.
|
|
this.#reader.cancel().catch(() => undefined),
|
|
"Realtime integration client cancellation timed out.",
|
|
);
|
|
}
|
|
|
|
async #readWithinDeadline() {
|
|
return withinTransportDeadline(
|
|
this.#reader.read(),
|
|
"Realtime integration read timed out.",
|
|
() => this.#abort.abort(),
|
|
);
|
|
}
|
|
}
|
|
|
|
async function createFixtureModule(control: FixtureControl): Promise<{
|
|
readonly directory: string;
|
|
readonly symbol: symbol;
|
|
}> {
|
|
const directory = await mkdtemp(path.join(tmpdir(), "ekklesia-native-realtime-"));
|
|
const moduleDirectory = path.join(directory, "announcements");
|
|
await mkdir(path.join(moduleDirectory, "dist/server"), { recursive: true });
|
|
const symbol = Symbol.for(`ekklesia.native-realtime.${Date.now()}.${Math.random()}`);
|
|
(globalThis as unknown as Record<PropertyKey, unknown>)[symbol] = control;
|
|
const expression = `globalThis[Symbol.for(${JSON.stringify(Symbol.keyFor(symbol))})]`;
|
|
|
|
await Promise.all([
|
|
writeFile(
|
|
path.join(moduleDirectory, "package.json"),
|
|
JSON.stringify({
|
|
ekklesia: { layer: "module" },
|
|
name: "@test/module-announcements",
|
|
type: "module",
|
|
version: "0.1.0",
|
|
}),
|
|
),
|
|
writeFile(
|
|
path.join(moduleDirectory, "ekklesia.module.json"),
|
|
JSON.stringify({
|
|
compatibility: { moduleSdk: "^0.1.0", platform: "^0.1.0" },
|
|
entrypoints: { server: "./dist/server/index.js" },
|
|
id: "announcements",
|
|
manifestVersion: 6,
|
|
name: "Announcements fixture",
|
|
permissions: [
|
|
{
|
|
baseline: "restricted",
|
|
description: "View the announcement slideshow.",
|
|
id: "announcements.slideshow.view",
|
|
},
|
|
],
|
|
version: "0.1.0",
|
|
}),
|
|
),
|
|
writeFile(
|
|
path.join(moduleDirectory, "dist/server/index.js"),
|
|
`export default {
|
|
id: "announcements",
|
|
version: "0.1.0",
|
|
register({ realtime }) {
|
|
realtime.channel({ channel: "state", permission: "announcements.slideshow.view" });
|
|
${expression}.events.push("register");
|
|
},
|
|
start({ realtime }) {
|
|
${expression}.publisher = realtime;
|
|
realtime.publish("state", {
|
|
mode: "on",
|
|
currentSlide: 0,
|
|
previousSlide: -1,
|
|
slides: [${JSON.stringify(stableDeliveryUrl)}]
|
|
});
|
|
${expression}.events.push("start");
|
|
},
|
|
stop() { ${expression}.events.push("stop"); }
|
|
};\n`,
|
|
),
|
|
]);
|
|
return { directory, symbol };
|
|
}
|
|
|
|
function services(): ModuleRuntimeServices {
|
|
return Object.freeze({ database: Object.freeze({}) as ModuleDatabase });
|
|
}
|
|
|
|
function identityRuntime(): EkklesiaIdentityRuntime {
|
|
return {
|
|
handleAuthRequest: vi.fn(),
|
|
resolvePrincipal: async (headers) => {
|
|
const cookie = headers.get("cookie");
|
|
if (cookie === "ekklesia-session=allowed") {
|
|
return createAuthenticatedPrincipal(allowedUserId);
|
|
}
|
|
if (cookie === "ekklesia-session=denied") {
|
|
return createAuthenticatedPrincipal(deniedUserId);
|
|
}
|
|
return ANONYMOUS_PRINCIPAL;
|
|
},
|
|
};
|
|
}
|
|
|
|
function authorization(): AuthorizationEvaluator {
|
|
return createAuthorizationEvaluator({
|
|
grants: { hasActiveGrant: async ({ userId }) => userId === allowedUserId },
|
|
platformOwners: { isPlatformOwner: async () => false },
|
|
});
|
|
}
|
|
|
|
async function bindLiveServer(
|
|
runtime: StartedModuleRuntime,
|
|
registry: PermissionRegistry,
|
|
): Promise<{
|
|
readonly activeSubscriptions: () => number;
|
|
readonly origin: string;
|
|
readonly server: ServerType;
|
|
}> {
|
|
const app = new Hono();
|
|
let activeSubscriptions = 0;
|
|
const observedRealtime: ModuleRealtimeRegistry = {
|
|
channels: runtime.realtime.channels,
|
|
closeSubscriptions: () => runtime.realtime.closeSubscriptions(),
|
|
publisherFor: (moduleId) => runtime.realtime.publisherFor(moduleId),
|
|
subscribe: (moduleId, channel): ModuleRealtimeSubscription => {
|
|
const subscription = runtime.realtime.subscribe(moduleId, channel);
|
|
activeSubscriptions += 1;
|
|
let closed = false;
|
|
return Object.freeze({
|
|
close: (): void => {
|
|
if (!closed) {
|
|
closed = true;
|
|
activeSubscriptions -= 1;
|
|
}
|
|
subscription.close();
|
|
},
|
|
read: () => subscription.read(),
|
|
snapshot: subscription.snapshot,
|
|
});
|
|
},
|
|
};
|
|
mountModuleRealtimeRoutes(app, {
|
|
evaluator: authorization(),
|
|
realtime: observedRealtime,
|
|
registry,
|
|
runtime: identityRuntime(),
|
|
});
|
|
app.notFound((c) => c.json({ code: "NOT_FOUND" }, 404));
|
|
|
|
let pendingServer: ServerType | undefined;
|
|
const server = await withinTransportDeadline(
|
|
new Promise<ServerType>((resolve, reject) => {
|
|
pendingServer = serve({ fetch: app.fetch, hostname: "127.0.0.1", port: 0 }, () =>
|
|
resolve(pendingServer as ServerType),
|
|
);
|
|
pendingServer.once("error", reject);
|
|
}),
|
|
"Realtime integration server bind timed out.",
|
|
() => pendingServer?.close(),
|
|
);
|
|
const address = server.address();
|
|
if (!address || typeof address === "string") {
|
|
throw new Error("Realtime integration server did not bind a TCP port.");
|
|
}
|
|
return {
|
|
activeSubscriptions: () => activeSubscriptions,
|
|
origin: `http://127.0.0.1:${address.port}`,
|
|
server,
|
|
};
|
|
}
|
|
|
|
async function connect(
|
|
origin: string,
|
|
cookie: string,
|
|
): Promise<{
|
|
readonly connection: LiveSseConnection;
|
|
readonly response: Response;
|
|
}> {
|
|
const abort = new AbortController();
|
|
const response = await withinTransportDeadline(
|
|
fetch(`${origin}/api/v1/module-realtime/announcements/state`, {
|
|
headers: { cookie },
|
|
signal: abort.signal,
|
|
}),
|
|
"Realtime integration connection timed out.",
|
|
() => abort.abort(),
|
|
);
|
|
return { connection: new LiveSseConnection(response, abort), response };
|
|
}
|
|
|
|
async function closeServer(server: ServerType): Promise<void> {
|
|
await withinTransportDeadline(
|
|
new Promise<void>((resolve) => server.close(() => resolve())),
|
|
"Realtime integration server close timed out.",
|
|
() =>
|
|
(
|
|
server as ServerType & {
|
|
closeAllConnections?: () => void;
|
|
}
|
|
).closeAllConnections?.(),
|
|
);
|
|
}
|
|
|
|
async function requestJson(
|
|
url: string,
|
|
init?: RequestInit,
|
|
): Promise<{ readonly body: unknown; readonly response: Response }> {
|
|
const abort = new AbortController();
|
|
const operation = (async (): Promise<{ readonly body: unknown; readonly response: Response }> => {
|
|
const response = await fetch(url, { ...init, signal: abort.signal });
|
|
const body = (await response.json()) as unknown;
|
|
return { body, response };
|
|
})();
|
|
return withinTransportDeadline(operation, "Realtime integration finite request timed out.", () =>
|
|
abort.abort(),
|
|
);
|
|
}
|
|
|
|
describe("native realtime over a live Node HTTP server", () => {
|
|
it("loads a real module and proves auth, snapshot, fanout, update, reconnect, and shutdown", async () => {
|
|
const control: FixtureControl = { events: [] };
|
|
const fixture = await createFixtureModule(control);
|
|
let runtime: StartedModuleRuntime | undefined;
|
|
let server: ServerType | undefined;
|
|
const connections: LiveSseConnection[] = [];
|
|
const failures: unknown[] = [];
|
|
try {
|
|
const modules = await discoverModules({
|
|
moduleSdkVersion: "0.1.0",
|
|
modulesDirectory: fixture.directory,
|
|
platformVersion: "0.1.0",
|
|
});
|
|
runtime = await startDiscoveredModuleRuntime(modules, services());
|
|
expect(control.events).toEqual(["register", "start"]);
|
|
const live = await bindLiveServer(runtime, composePlatformPermissionRegistry(modules));
|
|
server = live.server;
|
|
|
|
const anonymous = await requestJson(
|
|
`${live.origin}/api/v1/module-realtime/announcements/state`,
|
|
);
|
|
expect(anonymous.response.status).toBe(403);
|
|
expect(anonymous.body).toEqual({ code: "AUTHORIZATION_DENIED" });
|
|
const missingPermission = await requestJson(
|
|
`${live.origin}/api/v1/module-realtime/announcements/state`,
|
|
{ headers: { cookie: "ekklesia-session=denied" } },
|
|
);
|
|
expect(missingPermission.response.status).toBe(403);
|
|
|
|
const first = await connect(live.origin, "ekklesia-session=allowed");
|
|
const second = await connect(live.origin, "ekklesia-session=allowed");
|
|
connections.push(first.connection, second.connection);
|
|
expect(first.response.status).toBe(200);
|
|
expect(first.response.headers.get("content-type")).toBe("text/event-stream; charset=utf-8");
|
|
|
|
const expectedInitial = {
|
|
currentSlide: 0,
|
|
mode: "on",
|
|
previousSlide: -1,
|
|
slides: [stableDeliveryUrl],
|
|
};
|
|
expect(await first.connection.readEvent()).toEqual({
|
|
event: "snapshot",
|
|
payload: expectedInitial,
|
|
sequence: 1,
|
|
});
|
|
expect(await second.connection.readEvent()).toEqual({
|
|
event: "snapshot",
|
|
payload: expectedInitial,
|
|
sequence: 1,
|
|
});
|
|
expect(live.activeSubscriptions()).toBe(2);
|
|
|
|
control.publisher?.publish("state", {
|
|
currentSlide: 1,
|
|
mode: "on",
|
|
previousSlide: 0,
|
|
slides: [stableDeliveryUrl],
|
|
});
|
|
const expectedUpdate = {
|
|
event: "update",
|
|
payload: {
|
|
currentSlide: 1,
|
|
mode: "on",
|
|
previousSlide: 0,
|
|
slides: [stableDeliveryUrl],
|
|
},
|
|
sequence: 2,
|
|
};
|
|
expect(await first.connection.readEvent()).toEqual(expectedUpdate);
|
|
expect(await second.connection.readEvent()).toEqual(expectedUpdate);
|
|
|
|
await first.connection.close();
|
|
await waitForTransportCondition(
|
|
() => live.activeSubscriptions() === 1,
|
|
"Realtime integration subscriber cleanup timed out.",
|
|
);
|
|
control.publisher?.publish("state", {
|
|
currentSlide: 2,
|
|
mode: "off",
|
|
previousSlide: 1,
|
|
slides: [stableDeliveryUrl],
|
|
});
|
|
expect(await second.connection.readEvent()).toMatchObject({ event: "update", sequence: 3 });
|
|
|
|
const reconnected = await connect(live.origin, "ekklesia-session=allowed");
|
|
connections.push(reconnected.connection);
|
|
expect(await reconnected.connection.readEvent()).toEqual({
|
|
event: "snapshot",
|
|
payload: {
|
|
currentSlide: 2,
|
|
mode: "off",
|
|
previousSlide: 1,
|
|
slides: [stableDeliveryUrl],
|
|
},
|
|
sequence: 3,
|
|
});
|
|
expect(live.activeSubscriptions()).toBe(2);
|
|
|
|
const secondClosed = second.connection.expectClosed();
|
|
const reconnectedClosed = reconnected.connection.expectClosed();
|
|
runtime.realtime.closeSubscriptions();
|
|
await Promise.all([secondClosed, reconnectedClosed]);
|
|
await waitForTransportCondition(
|
|
() => live.activeSubscriptions() === 0,
|
|
"Realtime integration shutdown cleanup timed out.",
|
|
);
|
|
// Stream shutdown happens before ordinary HTTP drain in production. A publisher must remain
|
|
// usable until that drain completes so an already-running command can publish its result.
|
|
expect(() => control.publisher?.publish("state", null)).not.toThrow();
|
|
|
|
const afterShutdown = await requestJson(
|
|
`${live.origin}/api/v1/module-realtime/announcements/state`,
|
|
{ headers: { cookie: "ekklesia-session=allowed" } },
|
|
);
|
|
expect(afterShutdown.response.status).toBe(503);
|
|
expect(afterShutdown.body).toEqual({ code: "REALTIME_UNAVAILABLE" });
|
|
await runtime.stop();
|
|
expect(() => control.publisher?.publish("state", null)).toThrowError(
|
|
expect.objectContaining({ code: "REALTIME_PUBLISHER_STOPPED" }),
|
|
);
|
|
} catch (error) {
|
|
failures.push(error);
|
|
} finally {
|
|
try {
|
|
runtime?.realtime.closeSubscriptions();
|
|
} catch (error) {
|
|
failures.push(error);
|
|
}
|
|
const connectionCleanup = await Promise.allSettled(
|
|
connections.map((connection) => connection.close()),
|
|
);
|
|
for (const result of connectionCleanup) {
|
|
if (result.status === "rejected") {
|
|
failures.push(result.reason);
|
|
}
|
|
}
|
|
if (server) {
|
|
try {
|
|
await closeServer(server);
|
|
} catch (error) {
|
|
failures.push(error);
|
|
}
|
|
}
|
|
if (runtime) {
|
|
try {
|
|
await runtime.stop();
|
|
} catch (error) {
|
|
failures.push(error);
|
|
}
|
|
}
|
|
delete (globalThis as unknown as Record<PropertyKey, unknown>)[fixture.symbol];
|
|
try {
|
|
await rm(fixture.directory, { force: true, recursive: true });
|
|
} catch (error) {
|
|
failures.push(error);
|
|
}
|
|
}
|
|
if (failures.length > 0) {
|
|
throw failures[0];
|
|
}
|
|
expect(control.events).toEqual(["register", "start", "stop"]);
|
|
});
|
|
});
|