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

import {
  ANONYMOUS_PRINCIPAL,
  createAuthenticatedPrincipal,
  ModuleRealtimeError,
  moduleIdSchema,
  parseEkklesiaUserId,
  parsePermissionId,
} from "@ekklesia/contracts";
import { createPermissionRegistry, type PermissionRegistry } from "@ekklesia/core/authorization";
import type { EkklesiaIdentityRuntime } from "@ekklesia/core/identity";
import { type ModuleRealtimeRegistry, StagedModuleRealtimeRegistry } from "@ekklesia/core/modules";
import { Hono } from "hono";
import { describe, expect, it, vi } from "vitest";
import {
  MODULE_REALTIME_KEEPALIVE_MS,
  MODULE_REALTIME_MAX_CONNECTION_AGE_MS,
  ModuleRealtimeConfigurationError,
  moduleRealtimeBaseUrl,
  mountModuleRealtimeRoutes,
} from "../src/module-realtime-routes.js";

const moduleId = moduleIdSchema.parse("announcements");
const permission = parsePermissionId("announcements.slideshow.view");
const userId = parseEkklesiaUserId("88888888-8888-4888-8888-888888888888");
const endpoint = "/api/v1/module-realtime/announcements/state";

function permissionRegistry(): PermissionRegistry {
  return createPermissionRegistry([
    {
      baseline: "restricted",
      description: "View the announcement slideshow.",
      id: permission,
    },
  ]);
}

function realtimeFixture(initialPayload: unknown = { mode: "on" }): {
  readonly publisher: ReturnType<StagedModuleRealtimeRegistry["publisherFor"]>;
  readonly realtime: ModuleRealtimeRegistry;
} {
  const staged = new StagedModuleRealtimeRegistry();
  const registration = staged.createRegistrationStage(moduleId, new Set([permission]));
  registration.realtime.channel({ channel: "state", permission });
  registration.commit();
  const publisher = staged.publisherFor(moduleId);
  publisher.publish("state", initialPayload as never);
  return { publisher, realtime: staged.publish() };
}

function appFor(
  realtime: ModuleRealtimeRegistry,
  options: {
    readonly allowed?: boolean;
    readonly keepaliveMs?: number;
    readonly maxConnectionAgeMs?: number;
    readonly signedIn?: boolean;
  } = {},
): Hono {
  const app = new Hono();
  const runtime: EkklesiaIdentityRuntime = {
    handleAuthRequest: vi.fn(),
    resolvePrincipal: vi
      .fn()
      .mockResolvedValue(
        options.signedIn === false ? ANONYMOUS_PRINCIPAL : createAuthenticatedPrincipal(userId),
      ),
  };
  mountModuleRealtimeRoutes(app, {
    evaluator: {
      authorize: vi
        .fn()
        .mockResolvedValue({ allowed: options.signedIn !== false && options.allowed !== false }),
    },
    ...(options.keepaliveMs === undefined ? {} : { keepaliveMs: options.keepaliveMs }),
    ...(options.maxConnectionAgeMs === undefined
      ? {}
      : { maxConnectionAgeMs: options.maxConnectionAgeMs }),
    realtime,
    registry: permissionRegistry(),
    runtime,
  });
  return app;
}

async function readChunk(reader: ReadableStreamDefaultReader<Uint8Array>): Promise<string> {
  let timer: ReturnType<typeof setTimeout> | undefined;
  try {
    const result = await Promise.race([
      reader.read(),
      new Promise<never>((_, reject) => {
        timer = setTimeout(() => reject(new Error("Realtime test read timed out.")), 1_000);
      }),
    ]);
    return result.done ? "" : new TextDecoder().decode(result.value);
  } finally {
    clearTimeout(timer);
  }
}

describe("module realtime HTTP transport", () => {
  it("builds one server-owned root and rejects an unenforceable channel snapshot", () => {
    expect(moduleRealtimeBaseUrl("announcements")).toBe("/api/v1/module-realtime/announcements");
    expect(MODULE_REALTIME_KEEPALIVE_MS).toBe(15_000);
    expect(MODULE_REALTIME_MAX_CONNECTION_AGE_MS).toBe(300_000);

    const { realtime } = realtimeFixture();
    expect(() =>
      mountModuleRealtimeRoutes(new Hono(), {
        evaluator: { authorize: vi.fn() },
        realtime,
        registry: createPermissionRegistry([]),
        runtime: { handleAuthRequest: vi.fn(), resolvePrincipal: vi.fn() },
      }),
    ).toThrow(ModuleRealtimeConfigurationError);
  });

  it("denies anonymous and unauthorized callers before allocating a subscription", async () => {
    const { realtime } = realtimeFixture();
    const subscribe = vi.fn(realtime.subscribe.bind(realtime));
    const observed: ModuleRealtimeRegistry = {
      channels: realtime.channels,
      closeSubscriptions: () => realtime.closeSubscriptions(),
      publisherFor: (owner) => realtime.publisherFor(owner),
      subscribe,
    };

    const anonymous = await appFor(observed, { signedIn: false }).request(endpoint);
    expect(anonymous.status).toBe(403);
    expect(await anonymous.json()).toEqual({ code: "AUTHORIZATION_DENIED" });
    expect(subscribe).not.toHaveBeenCalled();

    const unauthorized = await appFor(observed, { allowed: false }).request(endpoint);
    expect(unauthorized.status).toBe(403);
    expect(await unauthorized.json()).toEqual({ code: "AUTHORIZATION_DENIED" });
    expect(subscribe).not.toHaveBeenCalled();
  });

  it("refuses Hono's implicit HEAD handling without allocating a stream subscription", async () => {
    const { realtime } = realtimeFixture();
    const subscribe = vi.fn(realtime.subscribe.bind(realtime));
    const observed: ModuleRealtimeRegistry = {
      channels: realtime.channels,
      closeSubscriptions: () => realtime.closeSubscriptions(),
      publisherFor: (owner) => realtime.publisherFor(owner),
      subscribe,
    };

    const response = await appFor(observed).request(endpoint, { method: "HEAD" });

    expect(response.status).toBe(405);
    expect(response.headers.get("allow")).toBe("GET");
    expect(response.headers.get("cache-control")).toBe("no-store");
    expect(subscribe).not.toHaveBeenCalled();
  });

  it("opens an authorized snapshot-first stream with proxy-safe headers and unchanged URLs", async () => {
    const deliveryUrl =
      "https://objects.example.test/slides/welcome.png?X-Amz-Signature=byte-identical";
    const { realtime } = realtimeFixture({
      currentSlide: 0,
      mode: "on",
      previousSlide: -1,
      slides: [deliveryUrl],
    });
    const response = await appFor(realtime).request(endpoint);

    expect(response.status).toBe(200);
    expect(response.headers.get("content-type")).toBe("text/event-stream; charset=utf-8");
    expect(response.headers.get("cache-control")).toBe("no-store");
    expect(response.headers.get("x-accel-buffering")).toBe("no");
    expect(response.headers.get("vary")).toBe("Origin, Cookie");

    const reader = response.body?.getReader();
    expect(reader).toBeDefined();
    const snapshot = await readChunk(reader as ReadableStreamDefaultReader<Uint8Array>);
    expect(snapshot).toContain("event: snapshot\n");
    expect(snapshot).toContain('"sequence":1');
    expect(snapshot).toContain(deliveryUrl);
    expect(snapshot).not.toContain("\nid:");
    await reader?.cancel();
  });

  it("streams ordered updates, emits transport-only keepalives, and closes on shutdown", async () => {
    const { publisher, realtime } = realtimeFixture({ currentSlide: 0 });
    const response = await appFor(realtime, { keepaliveMs: 5 }).request(endpoint);
    const reader = response.body?.getReader() as ReadableStreamDefaultReader<Uint8Array>;

    expect(await readChunk(reader)).toContain("event: snapshot");
    publisher.publish("state", { currentSlide: 1 });
    const update = await readChunk(reader);
    expect(update).toContain("event: update");
    expect(update).toContain('"sequence":2');
    expect(update).toContain('"currentSlide":1');
    expect(await readChunk(reader)).toBe(": keepalive\n\n");

    const closed = reader.read();
    realtime.closeSubscriptions();
    await expect(closed).resolves.toEqual({ done: true, value: undefined });
  });

  it("closes a stream at the host-owned maximum age so reconnect reauthorizes", async () => {
    const { realtime } = realtimeFixture({ currentSlide: 0 });
    let activeSubscriptions = 0;
    const observed: ModuleRealtimeRegistry = {
      channels: realtime.channels,
      closeSubscriptions: () => realtime.closeSubscriptions(),
      publisherFor: (owner) => realtime.publisherFor(owner),
      subscribe: (owner, channel) => {
        const subscription = realtime.subscribe(owner, channel);
        activeSubscriptions += 1;
        let closed = false;
        return Object.freeze({
          close: () => {
            if (!closed) {
              closed = true;
              activeSubscriptions -= 1;
            }
            subscription.close();
          },
          read: () => subscription.read(),
          snapshot: subscription.snapshot,
        });
      },
    };
    const response = await appFor(observed, {
      keepaliveMs: 1_000,
      maxConnectionAgeMs: 25,
    }).request(endpoint);
    const reader = response.body?.getReader() as ReadableStreamDefaultReader<Uint8Array>;

    expect(await readChunk(reader)).toContain("event: snapshot");
    expect(activeSubscriptions).toBe(1);
    expect(await readChunk(reader)).toBe("");
    expect(activeSubscriptions).toBe(0);
    const replacement = realtime.subscribe(moduleId, "state");
    replacement.close();
  });

  it("does not expose unknown module/channel combinations", async () => {
    const { realtime } = realtimeFixture();
    const app = appFor(realtime);
    expect((await app.request("/api/v1/module-realtime/calendar/state")).status).toBe(404);
    expect((await app.request("/api/v1/module-realtime/announcements/other")).status).toBe(404);
  });

  it("maps subscriber-capacity and unexpected allocation failures to fixed non-secret responses", async () => {
    const { realtime } = realtimeFixture();
    const observed = (failure: unknown): ModuleRealtimeRegistry => ({
      channels: realtime.channels,
      closeSubscriptions: () => realtime.closeSubscriptions(),
      publisherFor: (owner) => realtime.publisherFor(owner),
      subscribe: () => {
        throw failure;
      },
    });

    const atCapacity = await appFor(
      observed(new ModuleRealtimeError("REALTIME_SUBSCRIBER_LIMIT")),
    ).request(endpoint);
    expect(atCapacity.status).toBe(503);
    expect(await atCapacity.json()).toEqual({ code: "REALTIME_SUBSCRIBER_LIMIT" });

    const detail = "secret-token-and-machine-path";
    const unavailable = await appFor(observed(new Error(detail))).request(endpoint);
    expect(unavailable.status).toBe(503);
    const body = await unavailable.text();
    expect(body).toBe('{"code":"REALTIME_UNAVAILABLE"}');
    expect(body).not.toContain(detail);
  });
});