Ekklesia/core/src/modules/runtime.ts

import { pathToFileURL } from "node:url";
import {
  type CapabilityId,
  type CapabilityRegistrar,
  type CapabilityResolver,
  capabilityIdSchema,
  EMPTY_MODULE_CONFIGURATION,
  EMPTY_MODULE_STORAGE,
  type ModuleConfiguration,
  type ModuleDatabaseExecutor,
  type ModuleHttpRouteRegistrar,
  type ModuleId,
  ModuleRealtimeError,
  type ModuleRealtimePublisher,
  type ModuleRegistrationContext,
  type ModuleRouteDefinitionInput,
  type ModuleRouteHandler,
  type ModuleRuntimeContext,
  type ModuleRuntimeDefinition,
  type ModuleStorage,
  moduleRouteShapeKey,
  moduleRoutesOverlap,
  parseModuleRuntimeDefinition,
  type RegisteredCapability,
  type RegisteredModuleHttpRoute,
  safeParseModuleRouteDefinition,
} from "@ekklesia/contracts";
import type { AnyModuleRuntimeServices, ModuleDatabase } from "./database.js";
import {
  type DiscoveredModule,
  discoverModules,
  type ModuleDiscoveryOptions,
} from "./discovery.js";
import {
  ModuleRealtimeRegistrationError,
  type ModuleRealtimeRegistry,
  StagedModuleRealtimeRegistry,
} from "./realtime.js";

export type ModuleRuntimePhase = "load" | "register" | "rollback" | "start" | "stop";

export type ModuleRuntimeIssueCode =
  | "MODULE_CAPABILITY_ACCESS_UNDECLARED"
  | "MODULE_CAPABILITY_DUPLICATE"
  | "MODULE_CAPABILITY_ID_INVALID"
  | "MODULE_CAPABILITY_UNDECLARED"
  | "MODULE_CAPABILITY_UNREGISTERED"
  | "MODULE_REGISTER_FAILED"
  | "MODULE_REALTIME_CHANNEL_DEFINITION_INVALID"
  | "MODULE_REALTIME_CHANNEL_DUPLICATE"
  | "MODULE_REALTIME_CHANNEL_INVALID"
  | "MODULE_REALTIME_CHANNEL_LIMIT"
  | "MODULE_REALTIME_CHANNEL_UNDECLARED"
  | "MODULE_REALTIME_PAYLOAD_INVALID"
  | "MODULE_REALTIME_PAYLOAD_TOO_LARGE"
  | "MODULE_REALTIME_PERMISSION_INVALID"
  | "MODULE_REALTIME_PERMISSION_UNDECLARED"
  | "MODULE_REALTIME_PUBLISH_FAILED"
  | "MODULE_REALTIME_PUBLISHER_STOPPED"
  | "MODULE_REALTIME_SNAPSHOT_MISSING"
  | "MODULE_ROLLBACK_FAILED"
  | "MODULE_ROLLBACK_TIMEOUT"
  | "MODULE_ROUTE_DEFINITION_INVALID"
  | "MODULE_ROUTE_DUPLICATE"
  | "MODULE_ROUTE_BODY_MODE_INVALID"
  | "MODULE_ROUTE_MAX_BODY_BYTES_INVALID"
  | "MODULE_ROUTE_METHOD_INVALID"
  | "MODULE_ROUTE_OVERLAP"
  | "MODULE_ROUTE_PATH_INVALID"
  | "MODULE_ROUTE_PERMISSION_INVALID"
  | "MODULE_ROUTE_PERMISSION_UNDECLARED"
  | "MODULE_RUNTIME_EXPORT_INVALID"
  | "MODULE_RUNTIME_DATABASE_MISMATCH"
  | "MODULE_RUNTIME_ID_MISMATCH"
  | "MODULE_RUNTIME_IMPORT_FAILED"
  | "MODULE_RUNTIME_VERSION_MISMATCH"
  | "MODULE_START_FAILED"
  | "MODULE_STOP_FAILED"
  | "MODULE_STOP_TIMEOUT";

export interface ModuleRuntimeIssue {
  readonly code: ModuleRuntimeIssueCode;
  readonly message: string;
  readonly moduleId: ModuleId;
  readonly phase: ModuleRuntimePhase;
}

export interface CapabilityRegistry extends CapabilityResolver {
  list(): readonly RegisteredCapability[];
}

export interface RunningModule {
  readonly id: ModuleId;
  readonly version: string;
}

export interface StartedModuleRuntime {
  readonly capabilities: CapabilityRegistry;
  readonly modules: readonly RunningModule[];
  /** One immutable startup snapshot: discovered-module order, then registration order within each
   * module. Never mutated after startup — there is no dynamic route addition or removal. */
  readonly routes: readonly RegisteredModuleHttpRoute[];
  /** Opaque, transport-neutral state and subscriptions for the host. */
  readonly realtime: ModuleRealtimeRegistry;
  stop(): Promise<void>;
}

/**
 * Per-module frozen configuration snapshots, keyed by module id (ADR 0029).
 *
 * A lookup rather than a field on `ModuleRuntimeServices`, because configuration is resolved and
 * frozen per module independently of its live service object. A module with no entry receives
 * `EMPTY_MODULE_CONFIGURATION` rather than `undefined`, so `context.config` is always present and
 * module code never branches on whether the platform supplied one.
 */
export type ModuleConfigurationLookup = ReadonlyMap<string, ModuleConfiguration>;

function configurationFor(
  configurations: ModuleConfigurationLookup,
  moduleId: string,
): ModuleConfiguration {
  return configurations.get(moduleId) ?? EMPTY_MODULE_CONFIGURATION;
}

/**
 * Per-module scoped object storage, keyed by module id (ADR 0030).
 *
 * A lookup for the same reason configuration is one, and emphatically **not** a field on
 * `ModuleRuntimeServices`: a storage facade is scoped to exactly the buckets one module declared.
 * Sharing it would hand every
 * module every other module's storage.
 */
export type ModuleStorageLookup = ReadonlyMap<string, ModuleStorage>;

/** One frozen service object per module id. Legacy modules may point at objects wrapping the same
 * facade; ABI-v1 modules each point at their own attributed executor. */
export type ModuleRuntimeServicesLookup = ReadonlyMap<string, AnyModuleRuntimeServices>;
export type ModuleRuntimeServicesSource = ModuleRuntimeServicesLookup | AnyModuleRuntimeServices;

function servicesFor(
  services: ModuleRuntimeServicesSource,
  module: DiscoveredModule,
): AnyModuleRuntimeServices {
  const selected = "get" in services ? services.get(module.manifest.id) : services;
  if (!selected) {
    throw new TypeError(`Module ${module.manifest.id} has no runtime services.`);
  }
  const expectsAbi = module.manifest.database?.abi === 1;
  const isAbi = "abiVersion" in selected.database && selected.database.abiVersion === 1;
  if (expectsAbi !== isAbi) {
    throw new TypeError(`Module ${module.manifest.id} has the wrong database service.`);
  }
  return selected;
}

function storageFor(storages: ModuleStorageLookup, moduleId: string): ModuleStorage {
  return storages.get(moduleId) ?? EMPTY_MODULE_STORAGE;
}

export interface ModuleRuntimeTimeoutOptions {
  /** Test-only narrowing of `MODULE_STOP_HOOK_TIMEOUT_MS`, applied to both normal stop and startup
   * rollback, so a deadline can be proven in milliseconds instead of ten seconds. Production
   * composition never sets this. */
  readonly hookTimeoutMs?: number;
}

export interface ModuleRuntimeCompositionOptions extends ModuleRuntimeTimeoutOptions {
  /**
   * The per-module configuration snapshots this runtime hands to each module's `register`, `start`,
   * `stop`, and HTTP handler contexts (ADR 0029). Production composition supplies exactly the map
   * `resolveModuleEligibility` produced while deciding which modules may run, so the configuration
   * that made a module eligible is the identical object it receives — never revalidated, and never
   * re-read, between those two moments.
   *
   * Omitted by a caller exercising no configuration; every module then receives
   * `EMPTY_MODULE_CONFIGURATION`.
   */
  readonly configurations?: ModuleConfigurationLookup;
  /**
   * The per-module scoped storage facades this runtime hands to each module's `start`, `stop`, and
   * HTTP handler contexts (ADR 0030). Omitted by a caller exercising no storage; every module then
   * receives `EMPTY_MODULE_STORAGE`, which refuses every bucket.
   */
  readonly storages?: ModuleStorageLookup;
}

export interface StartModuleRuntimeOptions
  extends ModuleDiscoveryOptions,
    ModuleRuntimeCompositionOptions {
  /** The manifest-selected service object for each module. One module's `start`/`stop` hooks and
   * handlers receive its same entry; ABI-v1 entries carry distinct attributed executors. */
  readonly services: ModuleRuntimeServicesSource;
}

interface LoadedModule {
  readonly definition: ModuleRuntimeDefinition;
  readonly discovered: DiscoveredModule;
}

type AnyRegistrationHook = (
  context: ModuleRegistrationContext<ModuleDatabase | ModuleDatabaseExecutor>,
) => unknown;
type AnyRuntimeHook = (
  context: ModuleRuntimeContext<ModuleDatabase | ModuleDatabaseExecutor>,
) => unknown;

interface CapabilityRegistrationStage {
  readonly capabilities: CapabilityRegistrar;
  commit(): readonly ModuleRuntimeIssue[];
}

interface HttpRouteRegistrationStage {
  readonly http: ModuleHttpRouteRegistrar;
  commit(): void;
}

class ModuleRuntimeSignal extends Error {
  readonly issue: ModuleRuntimeIssue;

  constructor(issue: ModuleRuntimeIssue) {
    super(issue.message);
    this.name = "ModuleRuntimeSignal";
    this.issue = issue;
  }
}

export class ModuleRuntimeError extends Error {
  readonly issues: readonly ModuleRuntimeIssue[];

  constructor(issues: readonly ModuleRuntimeIssue[]) {
    super(`Module runtime failed with ${issues.length} issue(s).`);
    this.name = "ModuleRuntimeError";
    this.stack = `${this.name}: ${this.message}`;
    this.issues = Object.freeze(issues.map((issue) => Object.freeze({ ...issue })));
  }
}

function runtimeIssue(
  code: ModuleRuntimeIssueCode,
  phase: ModuleRuntimePhase,
  moduleId: ModuleId,
  message: string,
): ModuleRuntimeIssue {
  return { code, message, moduleId, phase };
}

function realtimeRuntimeIssueCode(error: ModuleRealtimeError): ModuleRuntimeIssueCode {
  switch (error.code) {
    case "REALTIME_CHANNEL_INVALID":
      return "MODULE_REALTIME_CHANNEL_INVALID";
    case "REALTIME_CHANNEL_UNDECLARED":
      return "MODULE_REALTIME_CHANNEL_UNDECLARED";
    case "REALTIME_PAYLOAD_INVALID":
      return "MODULE_REALTIME_PAYLOAD_INVALID";
    case "REALTIME_PAYLOAD_TOO_LARGE":
      return "MODULE_REALTIME_PAYLOAD_TOO_LARGE";
    case "REALTIME_PUBLISHER_STOPPED":
      return "MODULE_REALTIME_PUBLISHER_STOPPED";
    default:
      return "MODULE_REALTIME_PUBLISH_FAILED";
  }
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function defaultExport(moduleNamespace: unknown): unknown {
  return isRecord(moduleNamespace) ? moduleNamespace.default : undefined;
}

class StagedCapabilityRegistry {
  readonly #registered = new Map<CapabilityId, RegisteredCapability>();

  createRegistrationStage(module: DiscoveredModule): CapabilityRegistrationStage {
    const local = new Map<CapabilityId, RegisteredCapability>();
    const declarations = new Map(
      module.manifest.capabilities.map((declaration) => [declaration.id, declaration] as const),
    );
    const resolve = (capabilityId: CapabilityId): RegisteredCapability | undefined => {
      this.#assertAccess(module, capabilityId, "register");
      return local.get(capabilityId) ?? this.#registered.get(capabilityId);
    };
    const capabilities: CapabilityRegistrar = Object.freeze({
      provide: (capabilityId: CapabilityId, implementation: unknown): void => {
        this.#assertValidCapabilityId(module, capabilityId, "register");
        const declaration = declarations.get(capabilityId);
        if (!declaration) {
          throw new ModuleRuntimeSignal(
            runtimeIssue(
              "MODULE_CAPABILITY_UNDECLARED",
              "register",
              module.manifest.id,
              `Module ${module.manifest.id} attempted to provide undeclared capability ${capabilityId}.`,
            ),
          );
        }
        if (local.has(capabilityId) || this.#registered.has(capabilityId)) {
          throw new ModuleRuntimeSignal(
            runtimeIssue(
              "MODULE_CAPABILITY_DUPLICATE",
              "register",
              module.manifest.id,
              `Capability ${capabilityId} was provided more than once.`,
            ),
          );
        }
        local.set(
          capabilityId,
          Object.freeze({
            id: capabilityId,
            implementation,
            moduleId: module.manifest.id,
            version: declaration.version,
          }),
        );
      },
      resolve,
    });

    return {
      capabilities,
      commit: (): readonly ModuleRuntimeIssue[] => {
        const missing = [...declarations.keys()]
          .filter((capabilityId) => !local.has(capabilityId))
          .sort();
        if (missing.length > 0) {
          return missing.map((capabilityId) =>
            runtimeIssue(
              "MODULE_CAPABILITY_UNREGISTERED",
              "register",
              module.manifest.id,
              `Module ${module.manifest.id} did not provide declared capability ${capabilityId}.`,
            ),
          );
        }
        for (const [capabilityId, capability] of local) {
          this.#registered.set(capabilityId, capability);
        }
        return [];
      },
    };
  }

  createRuntimeContext(
    module: DiscoveredModule,
    phase: "rollback" | "start" | "stop",
    services: AnyModuleRuntimeServices,
    config: ModuleConfiguration,
    storage: ModuleStorage,
    realtime: ModuleRealtimePublisher,
  ): ModuleRuntimeContext<ModuleDatabase | ModuleDatabaseExecutor> {
    const capabilities: CapabilityResolver = Object.freeze({
      resolve: (capabilityId: CapabilityId): RegisteredCapability | undefined => {
        this.#assertAccess(module, capabilityId, phase);
        return this.#registered.get(capabilityId);
      },
    });
    return Object.freeze({
      capabilities,
      config,
      moduleId: module.manifest.id,
      realtime,
      services,
      storage,
    });
  }

  publish(): CapabilityRegistry {
    const capabilities = Object.freeze(
      [...this.#registered.values()].sort((left, right) =>
        left.id < right.id ? -1 : left.id > right.id ? 1 : 0,
      ),
    );
    const byId = new Map(capabilities.map((capability) => [capability.id, capability] as const));
    return Object.freeze({
      list: (): readonly RegisteredCapability[] => capabilities,
      resolve: (capabilityId: CapabilityId): RegisteredCapability | undefined =>
        byId.get(capabilityId),
    });
  }

  #assertAccess(
    module: DiscoveredModule,
    capabilityId: CapabilityId,
    phase: "register" | "rollback" | "start" | "stop",
  ): void {
    this.#assertValidCapabilityId(module, capabilityId, phase);
    const ownsCapability = module.manifest.capabilities.some(
      (declaration) => declaration.id === capabilityId,
    );
    const declaresDependency =
      Object.hasOwn(module.manifest.dependencies.capabilities.required, capabilityId) ||
      Object.hasOwn(module.manifest.dependencies.capabilities.optional, capabilityId);
    if (!ownsCapability && !declaresDependency) {
      throw new ModuleRuntimeSignal(
        runtimeIssue(
          "MODULE_CAPABILITY_ACCESS_UNDECLARED",
          phase,
          module.manifest.id,
          `Module ${module.manifest.id} attempted to resolve undeclared capability ${capabilityId}.`,
        ),
      );
    }
  }

  #assertValidCapabilityId(
    module: DiscoveredModule,
    capabilityId: unknown,
    phase: "register" | "rollback" | "start" | "stop",
  ): asserts capabilityId is CapabilityId {
    if (!capabilityIdSchema.safeParse(capabilityId).success) {
      throw new ModuleRuntimeSignal(
        runtimeIssue(
          "MODULE_CAPABILITY_ID_INVALID",
          phase,
          module.manifest.id,
          `Module ${module.manifest.id} used an invalid capability identifier.`,
        ),
      );
    }
  }
}

/**
 * Stages a module's HTTP route registrations during `register`, exactly like
 * `StagedCapabilityRegistry` stages capability provision: routes accumulate in a per-module local
 * list, validated and rejected immediately (`ModuleRuntimeSignal`) at the `http.route(...)` call
 * site, and are only merged into the shared, eventually-published list by `commit()` — called only
 * after the module's `register` hook returns without throwing. Unlike capabilities, there is no
 * "declared but never provided" concept for routes (a manifest never declares routes), so `commit`
 * has nothing to validate; it only publishes what already passed validation.
 */
class StagedModuleHttpRoutes {
  readonly #routes: RegisteredModuleHttpRoute[] = [];

  createRegistrationStage(module: DiscoveredModule): HttpRouteRegistrationStage {
    const local: RegisteredModuleHttpRoute[] = [];
    const shapes = new Set<string>();
    const ownedPermissionIds = new Set(
      module.manifest.permissions.map((declaration) => declaration.id),
    );

    const http: ModuleHttpRouteRegistrar = Object.freeze({
      route: (definition: ModuleRouteDefinitionInput): void => {
        const parsed = safeParseModuleRouteDefinition(definition);
        if (!parsed.success) {
          throw new ModuleRuntimeSignal(
            runtimeIssue(
              parsed.code,
              "register",
              module.manifest.id,
              `Module ${module.manifest.id} registered an invalid HTTP route.`,
            ),
          );
        }
        // Ownership is checked against this module's own manifest-declared permissions only, never
        // the composed platform registry — a module can therefore never gate a route with Core's or
        // another module's permission, structurally, regardless of what that other permission is.
        if (!ownedPermissionIds.has(parsed.data.permission)) {
          throw new ModuleRuntimeSignal(
            runtimeIssue(
              "MODULE_ROUTE_PERMISSION_UNDECLARED",
              "register",
              module.manifest.id,
              `Module ${module.manifest.id} registered an HTTP route with a permission its own manifest does not declare.`,
            ),
          );
        }
        const shapeKey = moduleRouteShapeKey(parsed.data.method, parsed.data.path);
        if (shapes.has(shapeKey)) {
          throw new ModuleRuntimeSignal(
            runtimeIssue(
              "MODULE_ROUTE_DUPLICATE",
              "register",
              module.manifest.id,
              `Module ${module.manifest.id} registered a duplicate HTTP route.`,
            ),
          );
        }
        // Distinct from a duplicate: these are *different* routes that can both match at least one
        // request path. With an order-sensitive matcher, whichever registered first would decide
        // which permission governs that path — a permission-policy decision no routing order may
        // make. Core rejects the ambiguity here instead of ranking by specificity, so the module
        // publishes none of its staged routes and no request can reach an ambiguous match.
        const overlapping = local.find((existing) => moduleRoutesOverlap(existing, parsed.data));
        if (overlapping) {
          throw new ModuleRuntimeSignal(
            runtimeIssue(
              "MODULE_ROUTE_OVERLAP",
              "register",
              module.manifest.id,
              `Module ${module.manifest.id} registered HTTP routes that can both match the same request path.`,
            ),
          );
        }
        shapes.add(shapeKey);
        local.push(
          Object.freeze({
            bodyMode: parsed.data.bodyMode,
            handler: parsed.data.handler as ModuleRouteHandler,
            maxBodyBytes: parsed.data.maxBodyBytes,
            method: parsed.data.method,
            moduleId: module.manifest.id,
            path: parsed.data.path,
            permission: parsed.data.permission,
          }),
        );
      },
    });

    return {
      http,
      commit: (): void => {
        this.#routes.push(...local);
      },
    };
  }

  publish(): readonly RegisteredModuleHttpRoute[] {
    return Object.freeze([...this.#routes]);
  }
}

async function loadRuntimeDefinitions(
  discoveredModules: readonly DiscoveredModule[],
): Promise<readonly LoadedModule[]> {
  const issues: ModuleRuntimeIssue[] = [];
  const loaded: LoadedModule[] = [];

  for (const discovered of discoveredModules) {
    let moduleNamespace: unknown;
    try {
      moduleNamespace = (await import(
        pathToFileURL(discovered.entrypoints.server).href
      )) as unknown;
    } catch {
      issues.push(
        runtimeIssue(
          "MODULE_RUNTIME_IMPORT_FAILED",
          "load",
          discovered.manifest.id,
          `Module ${discovered.manifest.id} server entrypoint could not be imported.`,
        ),
      );
      continue;
    }

    let definition: ModuleRuntimeDefinition;
    try {
      definition = parseModuleRuntimeDefinition(defaultExport(moduleNamespace));
    } catch {
      issues.push(
        runtimeIssue(
          "MODULE_RUNTIME_EXPORT_INVALID",
          "load",
          discovered.manifest.id,
          `Module ${discovered.manifest.id} server entrypoint has an invalid default export.`,
        ),
      );
      continue;
    }

    let matchesManifest = true;
    if (definition.id !== discovered.manifest.id) {
      matchesManifest = false;
      issues.push(
        runtimeIssue(
          "MODULE_RUNTIME_ID_MISMATCH",
          "load",
          discovered.manifest.id,
          `Module runtime ID ${definition.id} does not match manifest ID ${discovered.manifest.id}.`,
        ),
      );
    }
    if (definition.version !== discovered.manifest.version) {
      matchesManifest = false;
      issues.push(
        runtimeIssue(
          "MODULE_RUNTIME_VERSION_MISMATCH",
          "load",
          discovered.manifest.id,
          `Module runtime version ${definition.version} does not match manifest version ${discovered.manifest.version}.`,
        ),
      );
    }
    const runtimeDeclaresAbi = definition.database?.abi === 1;
    const manifestDeclaresAbi = discovered.manifest.database?.abi === 1;
    if (runtimeDeclaresAbi !== manifestDeclaresAbi) {
      matchesManifest = false;
      issues.push(
        runtimeIssue(
          "MODULE_RUNTIME_DATABASE_MISMATCH",
          "load",
          discovered.manifest.id,
          `Module runtime database declaration does not match manifest ${discovered.manifest.id}.`,
        ),
      );
    }
    if (matchesManifest) {
      loaded.push({ definition, discovered });
    }
  }

  if (issues.length > 0) {
    throw new ModuleRuntimeError(issues);
  }
  return Object.freeze(loaded);
}

async function registerModules(
  modules: readonly LoadedModule[],
  capabilityRegistry: StagedCapabilityRegistry,
  httpRoutes: StagedModuleHttpRoutes,
  realtimeRegistry: StagedModuleRealtimeRegistry,
  configurations: ModuleConfigurationLookup,
): Promise<void> {
  for (const module of modules) {
    const capabilityStage = capabilityRegistry.createRegistrationStage(module.discovered);
    const routeStage = httpRoutes.createRegistrationStage(module.discovered);
    const realtimeStage = realtimeRegistry.createRegistrationStage(
      module.discovered.manifest.id,
      new Set(module.discovered.manifest.permissions.map((declaration) => declaration.id)),
    );
    const context = Object.freeze({
      capabilities: capabilityStage.capabilities,
      config: configurationFor(configurations, module.discovered.manifest.id),
      http: routeStage.http,
      moduleId: module.discovered.manifest.id,
      realtime: realtimeStage.realtime,
    }) as ModuleRegistrationContext<ModuleDatabase | ModuleDatabaseExecutor>;
    try {
      await (module.definition.register as AnyRegistrationHook | undefined)?.(context);
    } catch (error) {
      if (error instanceof ModuleRuntimeSignal) {
        throw new ModuleRuntimeError([error.issue]);
      }
      if (error instanceof ModuleRealtimeRegistrationError) {
        throw new ModuleRuntimeError([
          runtimeIssue(
            error.code,
            "register",
            module.discovered.manifest.id,
            `Module ${module.discovered.manifest.id} registered an invalid realtime channel.`,
          ),
        ]);
      }
      throw new ModuleRuntimeError([
        runtimeIssue(
          "MODULE_REGISTER_FAILED",
          "register",
          module.discovered.manifest.id,
          `Module ${module.discovered.manifest.id} registration failed.`,
        ),
      ]);
    }
    const issues = capabilityStage.commit();
    if (issues.length > 0) {
      throw new ModuleRuntimeError(issues);
    }
    // Routes are only merged into the shared published list once this module's registration has
    // fully succeeded (register hook did not throw, and every declared capability was provided) —
    // a failed registration therefore publishes no routes for this module, matching the same
    // all-or-nothing guarantee capabilities already have.
    routeStage.commit();
    realtimeStage.commit();
  }
}

/**
 * The platform's fixed wait for one module's `stop` hook, during either a normal stop or a startup
 * rollback. Not configurable in this slice — no environment variable, manifest field, or runtime
 * service — so every deployment bounds cleanup identically.
 *
 * This bounds *waiting*, not execution. An in-process promise cannot be canceled: a hook that
 * exceeds this deadline keeps running and may settle later, but it no longer delays the rest of
 * shutdown, and the pool closes after the bounded sequence, so its late database work fails. The
 * platform does not sandbox, terminate, or undo a trusted module's JavaScript.
 */
export const MODULE_STOP_HOOK_TIMEOUT_MS = 10_000;

interface StopModulesOptions {
  /** Test-only narrowing of `MODULE_STOP_HOOK_TIMEOUT_MS` so a deadline can be proven without
   * waiting out the production value. Production composition never sets this. */
  readonly hookTimeoutMs?: number;
}

/**
 * Awaits one hook under a deadline. The losing branch is always handled: the timer is cleared when
 * the hook wins, and a hook that loses gets a permanent no-op rejection handler attached so a late
 * throw can never surface as an unhandled rejection. A timed-out hook is never awaited again.
 */
async function awaitHookWithinDeadline(
  hook: Promise<void>,
  timeoutMs: number,
): Promise<"completed" | "timed-out"> {
  let timer: ReturnType<typeof setTimeout> | undefined;
  // Attached before the race so the rejection is considered handled even if the timeout wins.
  const observed = hook.then(
    () => "completed" as const,
    (error: unknown) => {
      throw error;
    },
  );
  observed.catch(() => undefined);
  try {
    return await Promise.race([
      observed,
      new Promise<"timed-out">((resolve) => {
        timer = setTimeout(() => resolve("timed-out"), timeoutMs);
      }),
    ]);
  } finally {
    clearTimeout(timer);
  }
}

/**
 * Stops modules in reverse order, giving each hook a bounded slice of time and continuing to the
 * next module whatever the previous one did. Every applicable hook is invoked at most once.
 *
 * The whole sequence is finite because the module count is finite and fixed at discovery: its
 * worst case is `startedModuleCount * MODULE_STOP_HOOK_TIMEOUT_MS`, not one global ten-second
 * bound.
 */
async function stopModules(
  modules: readonly LoadedModule[],
  registry: StagedCapabilityRegistry,
  phase: "rollback" | "stop",
  services: ModuleRuntimeServicesSource,
  configurations: ModuleConfigurationLookup,
  storages: ModuleStorageLookup,
  realtimeRegistry: StagedModuleRealtimeRegistry,
  options: StopModulesOptions = {},
): Promise<readonly ModuleRuntimeIssue[]> {
  const timeoutMs = options.hookTimeoutMs ?? MODULE_STOP_HOOK_TIMEOUT_MS;
  const issues: ModuleRuntimeIssue[] = [];
  for (const module of [...modules].reverse()) {
    const moduleId = module.discovered.manifest.id;
    // A dependency may keep doing ordinary work while its dependents stop. Deactivate only the
    // module whose stop hook is about to run, not every publisher at the beginning of the sequence.
    realtimeRegistry.deactivateModule(moduleId);
    const stop = module.definition.stop as AnyRuntimeHook | undefined;
    if (!stop) {
      // No synthetic callback and no timer for a module that declares no stop hook.
      continue;
    }
    try {
      const outcome = await awaitHookWithinDeadline(
        Promise.resolve(
          stop(
            registry.createRuntimeContext(
              module.discovered,
              phase,
              servicesFor(services, module.discovered),
              configurationFor(configurations, moduleId),
              storageFor(storages, moduleId),
              realtimeRegistry.publisherFor(moduleId),
            ) as ModuleRuntimeContext<ModuleDatabase | ModuleDatabaseExecutor>,
          ),
        ).then(() => undefined),
        timeoutMs,
      );
      if (outcome === "timed-out") {
        issues.push(
          runtimeIssue(
            phase === "rollback" ? "MODULE_ROLLBACK_TIMEOUT" : "MODULE_STOP_TIMEOUT",
            phase,
            moduleId,
            `Module ${moduleId} ${phase} did not complete within the platform deadline.`,
          ),
        );
      }
    } catch (error) {
      if (error instanceof ModuleRuntimeSignal) {
        issues.push(error.issue);
      } else {
        issues.push(
          runtimeIssue(
            phase === "rollback" ? "MODULE_ROLLBACK_FAILED" : "MODULE_STOP_FAILED",
            phase,
            moduleId,
            `Module ${moduleId} ${phase} failed.`,
          ),
        );
      }
    }
  }
  return issues;
}

async function startModules(
  modules: readonly LoadedModule[],
  registry: StagedCapabilityRegistry,
  services: ModuleRuntimeServicesSource,
  configurations: ModuleConfigurationLookup,
  storages: ModuleStorageLookup,
  realtimeRegistry: StagedModuleRealtimeRegistry,
  options: StopModulesOptions = {},
): Promise<void> {
  // A module joins the rollback set the moment its `start` hook is *invoked*, not when it
  // completes. A hook that acquired a connection, registered a listener, or recorded partial
  // initialization before throwing still has cleanup to do, and previously only the modules that
  // had already finished were rolled back — leaving the failing module's own resources behind. Its
  // `stop` hook therefore runs first, ahead of the earlier modules in reverse order, which is why
  // module authors must make `stop` safe after a partially completed `start`.
  const rollbackCandidates: LoadedModule[] = [];
  for (const module of modules) {
    rollbackCandidates.push(module);
    try {
      await (module.definition.start as AnyRuntimeHook | undefined)?.(
        registry.createRuntimeContext(
          module.discovered,
          "start",
          servicesFor(services, module.discovered),
          configurationFor(configurations, module.discovered.manifest.id),
          storageFor(storages, module.discovered.manifest.id),
          realtimeRegistry.publisherFor(module.discovered.manifest.id),
        ) as ModuleRuntimeContext<ModuleDatabase | ModuleDatabaseExecutor>,
      );
      const missingSnapshots = realtimeRegistry.missingSnapshots(module.discovered.manifest.id);
      if (missingSnapshots.length > 0) {
        throw new ModuleRuntimeSignal(
          runtimeIssue(
            "MODULE_REALTIME_SNAPSHOT_MISSING",
            "start",
            module.discovered.manifest.id,
            `Module ${module.discovered.manifest.id} did not publish an initial snapshot for every declared realtime channel.`,
          ),
        );
      }
    } catch (error) {
      const startIssue =
        error instanceof ModuleRuntimeSignal
          ? error.issue
          : error instanceof ModuleRealtimeError
            ? runtimeIssue(
                realtimeRuntimeIssueCode(error),
                "start",
                module.discovered.manifest.id,
                `Module ${module.discovered.manifest.id} realtime publication failed during startup.`,
              )
            : runtimeIssue(
                "MODULE_START_FAILED",
                "start",
                module.discovered.manifest.id,
                `Module ${module.discovered.manifest.id} startup failed.`,
              );
      realtimeRegistry.closeSubscriptions();
      const rollbackIssues = await stopModules(
        rollbackCandidates,
        registry,
        "rollback",
        services,
        configurations,
        storages,
        realtimeRegistry,
        options,
      );
      realtimeRegistry.deactivateAll();
      throw new ModuleRuntimeError([startIssue, ...rollbackIssues]);
    }
  }
}

/**
 * Starts the module runtime from a module list the caller already discovered — performs no
 * module-directory scan or manifest read of its own, and preserves the given list's order exactly.
 * `startModuleRuntime` below is the convenient discovery-plus-start wrapper most callers want;
 * `apps/api`'s production startup calls this form directly (`prepareStartupModuleInventory`,
 * `apps/api/src/platform-permissions.ts`) so permission composition and module HTTP/runtime
 * startup share the exact same one-time discovery rather than discovering modules twice.
 */
export async function startDiscoveredModuleRuntime(
  discoveredModules: readonly DiscoveredModule[],
  services: ModuleRuntimeServicesSource,
  timeouts: ModuleRuntimeCompositionOptions = {},
): Promise<StartedModuleRuntime> {
  const modules = await loadRuntimeDefinitions(discoveredModules);
  const capabilityRegistry = new StagedCapabilityRegistry();
  const httpRoutes = new StagedModuleHttpRoutes();
  const realtimeRegistry = new StagedModuleRealtimeRegistry();
  const configurations = timeouts.configurations ?? new Map<string, ModuleConfiguration>();
  const storages = timeouts.storages ?? new Map<string, ModuleStorage>();
  await registerModules(modules, capabilityRegistry, httpRoutes, realtimeRegistry, configurations);
  await startModules(
    modules,
    capabilityRegistry,
    services,
    configurations,
    storages,
    realtimeRegistry,
    timeouts,
  );

  const capabilities = capabilityRegistry.publish();
  const routes = httpRoutes.publish();
  const realtime = realtimeRegistry.publish();
  const runningModules = Object.freeze(
    modules.map((module) =>
      Object.freeze({
        id: module.discovered.manifest.id,
        version: module.discovered.manifest.version,
      }),
    ),
  );
  let stopPromise: Promise<void> | undefined;

  return Object.freeze({
    capabilities,
    modules: runningModules,
    realtime,
    routes,
    stop: (): Promise<void> => {
      stopPromise ??= (async () => {
        realtime.closeSubscriptions();
        const issues = await stopModules(
          modules,
          capabilityRegistry,
          "stop",
          services,
          configurations,
          storages,
          realtimeRegistry,
          timeouts,
        );
        realtimeRegistry.deactivateAll();
        if (issues.length > 0) {
          throw new ModuleRuntimeError(issues);
        }
      })();
      return stopPromise;
    },
  });
}

export async function startModuleRuntime(
  options: StartModuleRuntimeOptions,
): Promise<StartedModuleRuntime> {
  const discovered = await discoverModules(options);
  return startDiscoveredModuleRuntime(discovered, options.services, {
    ...(options.configurations === undefined ? {} : { configurations: options.configurations }),
    ...(options.storages === undefined ? {} : { storages: options.storages }),
    ...(options.hookTimeoutMs === undefined ? {} : { hookTimeoutMs: options.hookTimeoutMs }),
  });
}