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 ModuleHttpRouteRegistrar,
  type ModuleId,
  ModuleRealtimeError,
  type ModuleRealtimePublisher,
  type ModuleRegistrationContext,
  type ModuleRouteDefinitionInput,
  type ModuleRuntimeContext,
  type ModuleRuntimeDefinition,
  type ModuleRuntimeServices,
  type ModuleStorage,
  moduleRouteShapeKey,
  moduleRoutesOverlap,
  parseModuleRuntimeDefinition,
  type RegisteredCapability,
  type RegisteredModuleHttpRoute,
  safeParseModuleRouteDefinition,
} from "@ekklesia/contracts";
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_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 `services` is deliberately **one
 * shared object identity every module receives** — a property this runtime has a test for — while
 * configuration is per-module by definition. 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`: `services` is one shared object identity every module receives, while 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>;

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 same frozen services object every started module's `start`/`stop` hooks and (once
   * `apps/api` mounts routes) HTTP handlers receive for this runtime — a required production
   * dependency, never an optional test-only global. See `ModuleRuntimeServices`. */
  readonly services: ModuleRuntimeServices;
}

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

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: ModuleRuntimeServices,
    config: ModuleConfiguration,
    storage: ModuleStorage,
    realtime: ModuleRealtimePublisher,
  ): ModuleRuntimeContext {
    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,
            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}.`,
        ),
      );
    }
    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: ModuleRegistrationContext = Object.freeze({
      capabilities: capabilityStage.capabilities,
      config: configurationFor(configurations, module.discovered.manifest.id),
      http: routeStage.http,
      moduleId: module.discovered.manifest.id,
      realtime: realtimeStage.realtime,
    });
    try {
      await module.definition.register?.(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: ModuleRuntimeServices,
  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;
    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,
              services,
              configurationFor(configurations, moduleId),
              storageFor(storages, moduleId),
              realtimeRegistry.publisherFor(moduleId),
            ),
          ),
        ),
        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: ModuleRuntimeServices,
  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?.(
        registry.createRuntimeContext(
          module.discovered,
          "start",
          services,
          configurationFor(configurations, module.discovered.manifest.id),
          storageFor(storages, module.discovered.manifest.id),
          realtimeRegistry.publisherFor(module.discovered.manifest.id),
        ),
      );
      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: ModuleRuntimeServices,
  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 }),
  });
}