feat: add native module realtime transport

2026-08-23 18:47:18 -05:00
parent fcc531a39c
commit fc6c8045c4
53 changed files with 3959 additions and 108 deletions
@@ -0,0 +1,528 @@
+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: 5,
+        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"]);
+  });
+});