import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; import * as fs from 'node:fs'; import * as net from 'node:net'; import * as path from 'node:path'; import { connect as natsConnect, type NatsConnection } from 'nats'; import { DEFAULT_SECURITY_POLICY, DEFAULT_SUPERVISION_POLICY, SecurityRealm, } from '@open-matrix/core'; import { createPolicyEngine } from '@open-matrix/core/core/security/PolicyPresets'; import { MatrixRuntime } from '@open-matrix/core/runtime/MatrixRuntime.js'; import { InMemoryBroker } from '@open-matrix/core/transport/InMemoryBroker.js'; import { InMemoryTransport } from '@open-matrix/core/transport/InMemoryTransport.js'; import { NatsTransport } from '@open-matrix/core/transport/NatsTransport.js'; import { authenticatorFromCredentialsRef } from '@open-matrix/nats-auth'; import { SqliteRecordStoreProvider } from '@open-matrix/system-sqlite'; import type { IResolvedRunnerTransportPlan } from './runner-transport.js'; export interface IRunnerRuntimeHandle { readonly runtime: MatrixRuntime; readonly transportMode: 'in-memory' | 'external' | 'embedded'; readonly root?: string; readonly runnerTransport?: IResolvedRunnerTransportPlan; readonly closed?: Promise; shutdown(): Promise; } export interface IRunnerRuntimeClosed { readonly error?: string; } export function formatRunnerRuntimeClosed(closed: IRunnerRuntimeClosed): string { const detail = closed.error?.trim(); return detail ? `Runner transport closed: ${detail}` : 'Runner transport closed'; } export function createRunnerSecurityRealm(): SecurityRealm { return new SecurityRealm( DEFAULT_SUPERVISION_POLICY, DEFAULT_SECURITY_POLICY, undefined, undefined, createPolicyEngine({ policyEngine: 'declarative', preset: 'standard' }), ); } export async function createRunnerRuntimeHandle( root: string | undefined, runnerTransport?: IResolvedRunnerTransportPlan, ): Promise { if (!runnerTransport) { const transport = new InMemoryTransport(new InMemoryBroker(), { name: 'mx-service-runner', ...(root ? { root } : {}), }); const runtime = new MatrixRuntime({ transport, logging: false, allowDynamicCompilation: false, securityRealm: createRunnerSecurityRealm(), recordStoreProvider: new SqliteRecordStoreProvider(), }); return { runtime, transportMode: 'in-memory', ...(root ? { root } : {}), async shutdown(): Promise { await runtime.shutdown(); await transport.disconnect(); }, }; } if (runnerTransport.mode === 'external') { const authenticator = await authenticatorFromCredentialsRef(runnerTransport.credentialsRef); const connection = await natsConnect({ servers: [runnerTransport.url], name: `mx-runner-${process.pid}`, timeout: 5_000, ...(authenticator ? { authenticator } : {}), }); const closed = watchNatsConnectionClosed(connection); const transport = new NatsTransport(connection, { root: runnerTransport.root }); const runtime = new MatrixRuntime({ transport, logging: false, allowDynamicCompilation: false, securityRealm: createRunnerSecurityRealm(), recordStoreProvider: new SqliteRecordStoreProvider(), }); return { runtime, transportMode: 'external', root: runnerTransport.root, runnerTransport, closed, async shutdown(): Promise { await runtime.shutdown(); await transport.disconnect(); if (!connection.isClosed()) { await connection.close(); } await closed; }, }; } const child = await startEmbeddedNatsServer(runnerTransport); const connection = await natsConnect({ servers: [runnerTransport.url], name: `mx-runner-embedded-${process.pid}`, timeout: 5_000, }); const closed = watchNatsConnectionClosed(connection); const transport = new NatsTransport(connection, { root: runnerTransport.root }); writePidFile(runnerTransport.pidFile, child.pid); const runtime = new MatrixRuntime({ transport, logging: false, allowDynamicCompilation: false, securityRealm: createRunnerSecurityRealm(), }); return { runtime, transportMode: 'embedded', root: runnerTransport.root, runnerTransport, closed, async shutdown(): Promise { await runtime.shutdown(); await transport.disconnect(); if (!connection.isClosed()) { await connection.close(); } await closed; await stopChildProcess(child, runnerTransport.pidFile); }, }; } function watchNatsConnectionClosed(connection: NatsConnection): Promise { return connection.closed().then( (reason): IRunnerRuntimeClosed => { if (reason instanceof Error) { return { error: reason.message }; } return {}; }, (err: unknown): IRunnerRuntimeClosed => ({ error: err instanceof Error ? err.message : String(err), }), ); } async function startEmbeddedNatsServer( runnerTransport: IResolvedRunnerTransportPlan, ): Promise { if (!runnerTransport.binaryPath) { throw new Error('Embedded runner transport requires binaryPath'); } if (!runnerTransport.port) { throw new Error('Embedded runner transport requires port'); } if (!runnerTransport.wsPort) { throw new Error('Embedded runner transport requires wsPort'); } if (!runnerTransport.dataDir) { throw new Error('Embedded runner transport requires dataDir'); } fs.mkdirSync(runnerTransport.dataDir, { recursive: true }); const configPath = path.join(runnerTransport.dataDir, 'nats-server.conf'); const escapedStoreDir = runnerTransport.dataDir.replace(/\\/g, '\\\\').replace(/"/g, '\\"'); fs.writeFileSync(configPath, [ 'host: "127.0.0.1"', `port: ${runnerTransport.port}`, 'jetstream {', ` store_dir: "${escapedStoreDir}"`, '}', 'websocket {', ' host: "127.0.0.1"', ` port: ${runnerTransport.wsPort}`, ' no_tls: true', '}', '', ].join('\n')); const child = spawn(runnerTransport.binaryPath, ['-c', configPath], { stdio: 'pipe', env: process.env, windowsHide: true, }); let stderr = ''; let spawnError: Error | null = null; child.once('error', (error: Error) => { spawnError = error; }); child.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString('utf8'); }); await waitForPorts( [runnerTransport.port, runnerTransport.wsPort], 10_000, child, () => { if (spawnError) { return `${spawnError.message}${stderr ? ` | ${stderr.trim()}` : ''}`; } return stderr.trim(); }, () => spawnError, ); return child; } async function waitForPorts( ports: number[], timeoutMs: number, child: ChildProcessWithoutNullStreams, getLogs: () => string, getSpawnError?: () => Error | null, ): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { const spawnError = getSpawnError?.(); if (spawnError) { throw new Error(`Failed to spawn nats-server: ${spawnError.message}`); } if (child.exitCode != null) { throw new Error(`nats-server exited early (${child.exitCode}): ${getLogs()}`); } const allListening = await Promise.all(ports.map((port) => isPortListening(port))); if (allListening.every(Boolean)) { return; } await new Promise((resolve) => setTimeout(resolve, 100)); } throw new Error(`Timed out waiting for nats-server on ports ${ports.join(', ')}: ${getLogs()}`); } async function isPortListening(port: number): Promise { return await new Promise((resolve) => { const socket = net.connect({ host: '127.0.0.1', port }); socket.once('connect', () => { socket.destroy(); resolve(true); }); socket.once('error', () => { socket.destroy(); resolve(false); }); socket.setTimeout(500, () => { socket.destroy(); resolve(false); }); }); } async function stopChildProcess( child: ChildProcessWithoutNullStreams, pidFile?: string, ): Promise { if (child.exitCode == null) { await new Promise((resolve) => { const timeout = setTimeout(() => { if (child.exitCode == null) { child.kill('SIGKILL'); } }, 2_000); child.once('exit', () => { clearTimeout(timeout); resolve(); }); child.kill('SIGTERM'); }); } if (pidFile) { fs.rmSync(pidFile, { force: true }); } } function writePidFile(pidFile: string | undefined, pid: number | undefined): void { if (!pidFile || !pid) { return; } fs.mkdirSync(path.dirname(pidFile), { recursive: true }); fs.writeFileSync(pidFile, `${pid}\n`, 'utf8'); }