feat: add module runtime lifecycle
This commit is contained in:
ekklesia
@@ -0,0 +1,441 @@
|
||||
import { pathToFileURL } from "node:url";
|
||||
import {
|
||||
type CapabilityId,
|
||||
type CapabilityRegistrar,
|
||||
type CapabilityResolver,
|
||||
capabilityIdSchema,
|
||||
type ModuleId,
|
||||
type ModuleRegistrationContext,
|
||||
type ModuleRuntimeContext,
|
||||
type ModuleRuntimeDefinition,
|
||||
parseModuleRuntimeDefinition,
|
||||
type RegisteredCapability,
|
||||
} from "@ekklesia/contracts";
|
||||
import {
|
||||
type DiscoveredModule,
|
||||
discoverModules,
|
||||
type ModuleDiscoveryOptions,
|
||||
} from "./discovery.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_ROLLBACK_FAILED"
|
||||
| "MODULE_RUNTIME_EXPORT_INVALID"
|
||||
| "MODULE_RUNTIME_ID_MISMATCH"
|
||||
| "MODULE_RUNTIME_IMPORT_FAILED"
|
||||
| "MODULE_RUNTIME_VERSION_MISMATCH"
|
||||
| "MODULE_START_FAILED"
|
||||
| "MODULE_STOP_FAILED";
|
||||
|
||||
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[];
|
||||
stop(): Promise<void>;
|
||||
}
|
||||
|
||||
export interface StartModuleRuntimeOptions extends ModuleDiscoveryOptions {}
|
||||
|
||||
interface LoadedModule {
|
||||
readonly definition: ModuleRuntimeDefinition;
|
||||
readonly discovered: DiscoveredModule;
|
||||
}
|
||||
|
||||
interface RegistrationStage {
|
||||
readonly context: ModuleRegistrationContext;
|
||||
commit(): readonly ModuleRuntimeIssue[];
|
||||
}
|
||||
|
||||
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 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): RegistrationStage {
|
||||
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,
|
||||
});
|
||||
const context: ModuleRegistrationContext = Object.freeze({
|
||||
capabilities,
|
||||
moduleId: module.manifest.id,
|
||||
});
|
||||
|
||||
return {
|
||||
context,
|
||||
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",
|
||||
): 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, moduleId: module.manifest.id });
|
||||
}
|
||||
|
||||
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.`,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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[],
|
||||
registry: StagedCapabilityRegistry,
|
||||
): Promise<void> {
|
||||
for (const module of modules) {
|
||||
const stage = registry.createRegistrationStage(module.discovered);
|
||||
try {
|
||||
await module.definition.register?.(stage.context);
|
||||
} catch (error) {
|
||||
if (error instanceof ModuleRuntimeSignal) {
|
||||
throw new ModuleRuntimeError([error.issue]);
|
||||
}
|
||||
throw new ModuleRuntimeError([
|
||||
runtimeIssue(
|
||||
"MODULE_REGISTER_FAILED",
|
||||
"register",
|
||||
module.discovered.manifest.id,
|
||||
`Module ${module.discovered.manifest.id} registration failed.`,
|
||||
),
|
||||
]);
|
||||
}
|
||||
const issues = stage.commit();
|
||||
if (issues.length > 0) {
|
||||
throw new ModuleRuntimeError(issues);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function stopModules(
|
||||
modules: readonly LoadedModule[],
|
||||
registry: StagedCapabilityRegistry,
|
||||
phase: "rollback" | "stop",
|
||||
): Promise<readonly ModuleRuntimeIssue[]> {
|
||||
const issues: ModuleRuntimeIssue[] = [];
|
||||
for (const module of [...modules].reverse()) {
|
||||
try {
|
||||
await module.definition.stop?.(registry.createRuntimeContext(module.discovered, phase));
|
||||
} catch (error) {
|
||||
if (error instanceof ModuleRuntimeSignal) {
|
||||
issues.push(error.issue);
|
||||
} else {
|
||||
issues.push(
|
||||
runtimeIssue(
|
||||
phase === "rollback" ? "MODULE_ROLLBACK_FAILED" : "MODULE_STOP_FAILED",
|
||||
phase,
|
||||
module.discovered.manifest.id,
|
||||
`Module ${module.discovered.manifest.id} ${phase} failed.`,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
return issues;
|
||||
}
|
||||
|
||||
async function startModules(
|
||||
modules: readonly LoadedModule[],
|
||||
registry: StagedCapabilityRegistry,
|
||||
): Promise<void> {
|
||||
const started: LoadedModule[] = [];
|
||||
for (const module of modules) {
|
||||
try {
|
||||
await module.definition.start?.(registry.createRuntimeContext(module.discovered, "start"));
|
||||
started.push(module);
|
||||
} catch (error) {
|
||||
const startIssue =
|
||||
error instanceof ModuleRuntimeSignal
|
||||
? error.issue
|
||||
: runtimeIssue(
|
||||
"MODULE_START_FAILED",
|
||||
"start",
|
||||
module.discovered.manifest.id,
|
||||
`Module ${module.discovered.manifest.id} startup failed.`,
|
||||
);
|
||||
const rollbackIssues = await stopModules(started, registry, "rollback");
|
||||
throw new ModuleRuntimeError([startIssue, ...rollbackIssues]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function startModuleRuntime(
|
||||
options: StartModuleRuntimeOptions,
|
||||
): Promise<StartedModuleRuntime> {
|
||||
const discovered = await discoverModules(options);
|
||||
const modules = await loadRuntimeDefinitions(discovered);
|
||||
const stagedRegistry = new StagedCapabilityRegistry();
|
||||
await registerModules(modules, stagedRegistry);
|
||||
await startModules(modules, stagedRegistry);
|
||||
|
||||
const capabilities = stagedRegistry.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,
|
||||
stop: (): Promise<void> => {
|
||||
stopPromise ??= (async () => {
|
||||
const issues = await stopModules(modules, stagedRegistry, "stop");
|
||||
if (issues.length > 0) {
|
||||
throw new ModuleRuntimeError(issues);
|
||||
}
|
||||
})();
|
||||
return stopPromise;
|
||||
},
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user