Ekklesia/apps/api/src/module-realtime-routes.ts

import { ModuleRealtimeError, type ModuleId, type PermissionId } from "@ekklesia/contracts";
import type {
  ModuleRealtimePublication,
  ModuleRealtimeRegistry,
  RegisteredModuleRealtimeChannel,
} from "@ekklesia/core/modules";
import type { Context, Hono } from "hono";
import {
  type AuthorizationMiddlewareOptions,
  requirePermission,
} from "./authorization-middleware.js";

const MODULE_REALTIME_PREFIX = "/api/v1/module-realtime";

/** Host-owned keepalive interval. Comments keep idle HTTP streams alive without entering module code,
 * consuming a sequence, or becoming a browser-visible state update. */
export const MODULE_REALTIME_KEEPALIVE_MS = 15_000;
/** Forces a fresh cookie/permission evaluation periodically rather than letting an authorization
 * decision live for an unbounded stream lifetime. EventSource reconnects after the clean EOF. */
export const MODULE_REALTIME_MAX_CONNECTION_AGE_MS = 5 * 60_000;

export interface ModuleRealtimeTransportTiming {
  /** Test-only narrowing of the production keepalive. Production composition never sets this. */
  readonly keepaliveMs?: number;
  /** Test-only narrowing of the production reauthorization interval. */
  readonly maxConnectionAgeMs?: number;
}

export interface ModuleRealtimeRoutesOptions
  extends AuthorizationMiddlewareOptions,
    ModuleRealtimeTransportTiming {
  readonly realtime: ModuleRealtimeRegistry;
}

/** The server is the sole author of this deployment-private base. Module UI receives it through the
 * host descriptor and never reconstructs the platform prefix or supplies its own module id. */
export function moduleRealtimeBaseUrl(moduleId: string): string {
  return `${MODULE_REALTIME_PREFIX}/${encodeURIComponent(moduleId)}`;
}

function moduleRealtimeChannelUrl(moduleId: ModuleId, channel: string): string {
  return `${moduleRealtimeBaseUrl(moduleId)}/${encodeURIComponent(channel)}`;
}

/** A forged runtime snapshot that could not be permission-gated is a composition failure. Carries no
 * module/channel/permission value and is thrown before any route from that snapshot is usable. */
export class ModuleRealtimeConfigurationError extends Error {
  constructor() {
    super("A published module realtime channel references a permission that is not registered.");
    this.name = "ModuleRealtimeConfigurationError";
    this.stack = `${this.name}: ${this.message}`;
    Object.freeze(this);
  }
}

function realtimeEnvelope(publication: ModuleRealtimePublication): string {
  // `serializedPayload` was validated and serialized exactly once by Core. Concatenating it as the
  // JSON value preserves every opaque string — including signed object-delivery URLs — byte for byte.
  return `{"sequence":${publication.sequence},"payload":${publication.serializedPayload}}`;
}

function realtimeEvent(
  kind: "snapshot" | "update",
  publication: ModuleRealtimePublication,
): string {
  // Core's compact JSON serialization contains no literal CR/LF bytes inside strings, so one `data`
  // line is sufficient. Deliberately no SSE `id`: this transport does not implement Last-Event-ID
  // replay, and a process-local sequence must not be mistaken for a durable cursor.
  return `event: ${kind}\ndata: ${realtimeEnvelope(publication)}\n\n`;
}

function sanitizedRealtimeFailure(c: Context, error: unknown): Response {
  c.header("Cache-Control", "no-store");
  const code =
    error instanceof ModuleRealtimeError && error.code === "REALTIME_SUBSCRIBER_LIMIT"
      ? "REALTIME_SUBSCRIBER_LIMIT"
      : "REALTIME_UNAVAILABLE";
  return c.json({ code }, 503);
}

interface CancelableDelay {
  readonly promise: Promise<"keepalive">;
  cancel(): void;
}

function keepaliveDelay(milliseconds: number): CancelableDelay {
  let timer: ReturnType<typeof setTimeout> | undefined;
  const promise = new Promise<"keepalive">((resolve) => {
    timer = setTimeout(() => resolve("keepalive"), milliseconds);
  });
  return {
    cancel: () => clearTimeout(timer),
    promise,
  };
}

function createRealtimeStream(
  definition: RegisteredModuleRealtimeChannel,
  registry: ModuleRealtimeRegistry,
  keepaliveMs: number,
  maxConnectionAgeMs: number,
): ReadableStream<Uint8Array> {
  const encoder = new TextEncoder();
  const subscription = registry.subscribe(definition.moduleId, definition.channel);
  let snapshotPending = true;
  let closed = false;
  let pendingRead: Promise<ModuleRealtimePublication | undefined> | undefined;
  let activeDelay: CancelableDelay | undefined;
  let expiryTimer: ReturnType<typeof setTimeout> | undefined;

  const close = (): void => {
    if (closed) {
      return;
    }
    closed = true;
    activeDelay?.cancel();
    activeDelay = undefined;
    clearTimeout(expiryTimer);
    expiryTimer = undefined;
    subscription.close();
  };

  return new ReadableStream<Uint8Array>({
    start(controller) {
      expiryTimer = setTimeout(() => {
        close();
        try {
          controller.close();
        } catch {
          // A simultaneous browser cancellation or completed pull already closed the controller.
        }
      }, maxConnectionAgeMs);
    },
    cancel: () => close(),
    async pull(controller) {
      if (closed) {
        return;
      }
      try {
        if (snapshotPending) {
          snapshotPending = false;
          controller.enqueue(encoder.encode(realtimeEvent("snapshot", subscription.snapshot)));
          return;
        }

        // Only one Core read is ever outstanding. If a keepalive wins the race, the same read is
        // retained for the next pull instead of allocating a second listener. Core itself holds only
        // one newest pending publication, and this stream asks for work only when the browser-facing
        // ReadableStream has demand, so a slow client cannot create an array at either layer.
        pendingRead ??= subscription.read();
        activeDelay = keepaliveDelay(keepaliveMs);
        const result = await Promise.race([pendingRead, activeDelay.promise]);
        activeDelay.cancel();
        activeDelay = undefined;

        if (result === "keepalive") {
          controller.enqueue(encoder.encode(": keepalive\n\n"));
          return;
        }

        pendingRead = undefined;
        if (result === undefined) {
          close();
          controller.close();
          return;
        }
        controller.enqueue(encoder.encode(realtimeEvent("update", result)));
      } catch {
        close();
        try {
          controller.close();
        } catch {
          // The browser may have canceled between the awaited read/timer and this close. Cancellation
          // already performed the same idempotent subscription cleanup.
        }
      }
    },
  });
}

function mountChannel(
  app: Hono,
  definition: RegisteredModuleRealtimeChannel,
  options: ModuleRealtimeRoutesOptions,
  keepaliveMs: number,
  maxConnectionAgeMs: number,
): void {
  const requireChannelPermission = requirePermission(
    options,
    definition.permission as PermissionId,
  );
  app.get(
    moduleRealtimeChannelUrl(definition.moduleId, definition.channel),
    requireChannelPermission,
    (c) => {
      // Hono deliberately lets a registered GET handler answer HEAD. That is useful for finite
      // resources, but unsafe for an infinite stream: the adapter discards the response body without
      // necessarily canceling it, which could retain a Core subscriber forever. Refuse HEAD before
      // allocating anything. Authentication/authorization has still run through the same middleware.
      if (c.req.method !== "GET") {
        return new Response(null, {
          headers: { Allow: "GET", "Cache-Control": "no-store" },
          status: 405,
        });
      }
      let stream: ReadableStream<Uint8Array>;
      try {
        stream = createRealtimeStream(
          definition,
          options.realtime,
          keepaliveMs,
          maxConnectionAgeMs,
        );
      } catch (error) {
        return sanitizedRealtimeFailure(c, error);
      }
      return new Response(stream, {
        headers: {
          "Cache-Control": "no-store",
          Connection: "keep-alive",
          "Content-Type": "text/event-stream; charset=utf-8",
          Vary: "Origin, Cookie",
          // Honored by nginx and harmless elsewhere. Other proxies must likewise disable buffering for
          // this path; no proxy-specific behavior enters the module-facing contract.
          "X-Accel-Buffering": "no",
        },
        status: 200,
      });
    },
  );
}

/** Mounts one exact, immutable, permission-gated GET route per successfully started module channel.
 * There is no dynamic module/channel lookup route and no request-controlled permission selection. */
export function mountModuleRealtimeRoutes(app: Hono, options: ModuleRealtimeRoutesOptions): void {
  const keepaliveMs = options.keepaliveMs ?? MODULE_REALTIME_KEEPALIVE_MS;
  const maxConnectionAgeMs = options.maxConnectionAgeMs ?? MODULE_REALTIME_MAX_CONNECTION_AGE_MS;
  if (!Number.isSafeInteger(keepaliveMs) || keepaliveMs <= 0) {
    throw new TypeError("The module realtime keepalive interval is invalid.");
  }
  if (!Number.isSafeInteger(maxConnectionAgeMs) || maxConnectionAgeMs <= 0) {
    throw new TypeError("The module realtime maximum connection age is invalid.");
  }
  for (const definition of options.realtime.channels) {
    if (!options.registry.has(definition.permission)) {
      throw new ModuleRealtimeConfigurationError();
    }
  }
  for (const definition of options.realtime.channels) {
    mountChannel(app, definition, options, keepaliveMs, maxConnectionAgeMs);
  }
}