Ekklesia/surfaces/shell/tests/module-realtime-context.test.ts

import type { ModuleUiRealtimeError, ModuleUiRealtimeStatus } from "@ekklesia/module-ui-sdk";
import { describe, expect, it, vi } from "vitest";
import type {
  HostEventSourceFactory,
  HostEventSourceOptions,
  HostModuleUiDescriptor,
} from "../src/host-protocol.js";
import { createModuleUiContext } from "../src/module-context.js";

const apiBaseUrl = "https://ekklesia.example.test";

const exampleDescriptor: HostModuleUiDescriptor = {
  contributions: [],
  apiBaseUrl: "/api/v1/modules/example",
  id: "example",
  permissions: ["example.state.read"],
  realtimeBaseUrl: "/api/v1/module-realtime/example",
  uiUrl: "/api/v1/module-ui/example/0.1.0",
  version: "0.1.0",
};

class FakeEventSource {
  readonly CLOSED = 2 as const;
  readonly CONNECTING = 0 as const;
  readonly OPEN = 1 as const;
  readonly listeners = new Map<string, Set<EventListener>>();
  closed = false;
  closeCalls = 0;
  readyState: number = this.CONNECTING;

  addEventListener(type: string, listener: EventListener): void {
    const listeners = this.listeners.get(type) ?? new Set<EventListener>();
    listeners.add(listener);
    this.listeners.set(type, listeners);
  }

  removeEventListener(type: string, listener: EventListener): void {
    this.listeners.get(type)?.delete(listener);
  }

  close(): void {
    this.closed = true;
    this.closeCalls += 1;
    this.readyState = this.CLOSED;
  }

  emit(type: string, data?: string): void {
    if (type === "open") {
      this.readyState = this.OPEN;
    }
    const event = Object.freeze({ data, type }) as unknown as Event;
    for (const listener of this.listeners.get(type) ?? []) {
      listener(event);
    }
  }
}

interface EventSourceHarness {
  readonly calls: Array<{
    readonly options: HostEventSourceOptions;
    readonly source: FakeEventSource;
    readonly url: string;
  }>;
  readonly factory: HostEventSourceFactory;
}

function eventSourceHarness(): EventSourceHarness {
  const calls: EventSourceHarness["calls"] = [];
  const factory = ((url: string, options: HostEventSourceOptions): EventSource => {
    const source = new FakeEventSource();
    calls.push({ options, source, url });
    return source as unknown as EventSource;
  }) as HostEventSourceFactory;
  return { calls, factory };
}

function context(factory: HostEventSourceFactory, descriptor = exampleDescriptor) {
  return createModuleUiContext({
    config: { apiBaseUrl, eventSourceFactory: factory },
    descriptor,
    principal: { kind: "authenticated", userId: "user-1" },
  });
}

function envelope(sequence: number, payload: unknown): string {
  return JSON.stringify({ payload, sequence });
}

describe("module UI realtime host adapter", () => {
  it("opens the server-provided module channel URL with host-owned credentials", () => {
    const harness = eventSourceHarness();

    context(harness.factory).realtime.subscribe("slideshow-state", {});

    expect(harness.calls).toHaveLength(1);
    expect(harness.calls[0]).toMatchObject({
      options: { withCredentials: true },
      url: `${apiBaseUrl}/api/v1/module-realtime/example/slideshow-state`,
    });
  });

  it("scopes the same local channel to each descriptor without accepting a module id", () => {
    const harness = eventSourceHarness();
    const calendar: HostModuleUiDescriptor = {
      ...exampleDescriptor,
      contributions: [],
      apiBaseUrl: "/api/v1/modules/calendar",
      id: "calendar",
      realtimeBaseUrl: "/api/v1/module-realtime/calendar",
      uiUrl: "/api/v1/module-ui/calendar/1.0.0",
      version: "1.0.0",
    };

    context(harness.factory).realtime.subscribe("state", {});
    context(harness.factory, calendar).realtime.subscribe("state", {});

    expect(harness.calls.map((call) => call.url)).toEqual([
      `${apiBaseUrl}/api/v1/module-realtime/example/state`,
      `${apiBaseUrl}/api/v1/module-realtime/calendar/state`,
    ]);
  });

  it("delivers snapshot first, ordered updates, and fresh state after automatic reconnect", () => {
    const harness = eventSourceHarness();
    const statuses: ModuleUiRealtimeStatus[] = [];
    const snapshots: unknown[] = [];
    const updates: unknown[] = [];
    const errors: ModuleUiRealtimeError[] = [];

    context(harness.factory).realtime.subscribe<{ readonly mode: string }>("state", {
      onError: (error) => errors.push(error),
      onSnapshot: (state, sequence) => snapshots.push({ sequence, state }),
      onStatus: (status) => statuses.push(status),
      onUpdate: (state, sequence) => updates.push({ sequence, state }),
    });
    const source = harness.calls[0]?.source;
    expect(source).toBeDefined();

    source?.emit("open");
    source?.emit("snapshot", envelope(1, { mode: "on" }));
    source?.emit("update", envelope(2, { mode: "off" }));
    // Native EventSource owns reconnect. The host reports it, then treats the next snapshot as a
    // fresh authoritative base even when its sequence is the same as the last state already seen.
    if (source) {
      source.readyState = source.CONNECTING;
      source.emit("error");
    }
    source?.emit("open");
    source?.emit("snapshot", envelope(2, { mode: "off" }));
    source?.emit("update", envelope(3, { mode: "on" }));

    expect(statuses).toEqual(["connecting", "connected", "reconnecting", "connected"]);
    expect(snapshots).toEqual([
      { sequence: 1, state: { mode: "on" } },
      { sequence: 2, state: { mode: "off" } },
    ]);
    expect(updates).toEqual([
      { sequence: 2, state: { mode: "off" } },
      { sequence: 3, state: { mode: "on" } },
    ]);
    expect(errors).toEqual([
      {
        code: "REALTIME_CONNECTION_FAILED",
        message: "The live connection could not be maintained.",
      },
    ]);
  });

  it("rejects malformed envelopes and out-of-order deliveries with one fixed safe error", () => {
    const harness = eventSourceHarness();
    const updates: unknown[] = [];
    const errors: ModuleUiRealtimeError[] = [];
    context(harness.factory).realtime.subscribe("state", {
      onError: (error) => errors.push(error),
      onUpdate: (state, sequence) => updates.push({ sequence, state }),
    });
    const source = harness.calls[0]?.source;
    source?.emit("open");

    source?.emit("update", envelope(1, { secret: "not delivered" }));
    source?.emit("snapshot", "not json: /private/server/path");
    source?.emit("snapshot", JSON.stringify({ extra: true, payload: {}, sequence: 1 }));
    source?.emit("snapshot", envelope(0, {}));
    source?.emit("snapshot", envelope(2, { mode: "on" }));
    source?.emit("update", envelope(2, { mode: "stale" }));

    expect(updates).toEqual([]);
    expect(errors).toHaveLength(5);
    expect(new Set(errors.map((error) => JSON.stringify(error)))).toEqual(
      new Set([
        JSON.stringify({
          code: "REALTIME_MESSAGE_INVALID",
          message: "The host received an invalid live update.",
        }),
      ]),
    );
    expect(JSON.stringify(errors)).not.toContain("private");
    expect(JSON.stringify(errors)).not.toContain("secret");
  });

  it("refuses invalid channel names before opening a connection", () => {
    const harness = eventSourceHarness();
    const realtime = context(harness.factory).realtime;

    for (const channel of [
      "",
      "State",
      "slide_state",
      "slide/state",
      "state.*",
      "-state",
      "state-",
      "a".repeat(49),
    ]) {
      expect(() => realtime.subscribe(channel, {}), channel).toThrowError(
        expect.objectContaining({ code: "MODULE_REALTIME_CHANNEL_INVALID" }),
      );
    }
    expect(harness.calls).toEqual([]);
  });

  it("closes idempotently, removes every listener, and delivers nothing afterward", () => {
    const harness = eventSourceHarness();
    const statuses: ModuleUiRealtimeStatus[] = [];
    const snapshots = vi.fn();
    const subscription = context(harness.factory).realtime.subscribe("state", {
      onSnapshot: snapshots,
      onStatus: (status) => statuses.push(status),
    });
    const source = harness.calls[0]?.source;
    source?.emit("open");

    subscription.close();
    subscription.close();
    source?.emit("snapshot", envelope(1, { mode: "on" }));
    source?.emit("open");

    expect(source?.closed).toBe(true);
    expect(source?.closeCalls).toBe(1);
    expect(
      [...((source?.listeners.values() ?? []) as Iterable<Set<EventListener>>)].every(
        (listeners) => listeners.size === 0,
      ),
    ).toBe(true);
    expect(snapshots).not.toHaveBeenCalled();
    expect(statuses).toEqual(["connecting", "connected", "closed"]);
  });

  it("treats a terminal native CLOSED error as closed and eagerly removes all listeners", () => {
    const harness = eventSourceHarness();
    const statuses: ModuleUiRealtimeStatus[] = [];
    const errors: ModuleUiRealtimeError[] = [];
    const snapshots = vi.fn();
    const subscription = context(harness.factory).realtime.subscribe("state", {
      onError: (error) => errors.push(error),
      onSnapshot: snapshots,
      onStatus: (status) => statuses.push(status),
    });
    const source = harness.calls[0]?.source;
    source?.emit("open");
    source?.emit("snapshot", envelope(1, { mode: "on" }));

    if (source) {
      // Fatal HTTP/protocol failures leave native EventSource CLOSED; unlike CONNECTING, the browser
      // will not retry them, so the host must not promise `reconnecting`.
      source.readyState = source.CLOSED;
      source.emit("error");
      source.emit("error");
      source.emit("open");
      source.emit("snapshot", envelope(2, { mode: "off" }));
    }
    subscription.close();

    expect(source?.closed).toBe(true);
    expect(source?.closeCalls).toBe(1);
    expect(
      [...((source?.listeners.values() ?? []) as Iterable<Set<EventListener>>)].every(
        (listeners) => listeners.size === 0,
      ),
    ).toBe(true);
    expect(statuses).toEqual(["connecting", "connected", "closed"]);
    expect(errors).toEqual([
      {
        code: "REALTIME_CONNECTION_FAILED",
        message: "The live connection could not be maintained.",
      },
    ]);
    expect(snapshots).toHaveBeenCalledOnce();
  });

  it("sanitizes constructor failures and closes without exposing the caught value", () => {
    const errors: ModuleUiRealtimeError[] = [];
    const statuses: ModuleUiRealtimeStatus[] = [];
    const factory = vi.fn(() => {
      throw new Error("token=secret at /home/server/realtime");
    }) as unknown as HostEventSourceFactory;

    const subscription = context(factory).realtime.subscribe("state", {
      onError: (error) => errors.push(error),
      onStatus: (status) => statuses.push(status),
    });
    subscription.close();

    expect(statuses).toEqual(["connecting", "closed"]);
    expect(errors).toEqual([
      {
        code: "REALTIME_CONNECTION_FAILED",
        message: "The live connection could not be maintained.",
      },
    ]);
    expect(JSON.stringify(errors)).not.toContain("secret");
    expect(JSON.stringify(errors)).not.toContain("home");
  });
});