Ekklesia/core/tests/module-realtime.test.ts

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.");
  });
});