Ekklesia/apps/api/tests/module-realtime-live.test.ts

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: 7,
        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"]);
  });
});