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>(); closed = false; closeCalls = 0; readyState: number = this.CONNECTING; addEventListener(type: string, listener: EventListener): void { const listeners = this.listeners.get(type) ?? new Set(); 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>)].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>)].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"); }); });