hivecast-sdk/archive/mx-cli-runtime-host-commands/runner-runtime.ts

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<IRunnerRuntimeClosed>;
  shutdown(): Promise<void>;
}

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<IRunnerRuntimeHandle> {
  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<void> {
        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<void> {
        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<void> {
      await runtime.shutdown();
      await transport.disconnect();
      if (!connection.isClosed()) {
        await connection.close();
      }
      await closed;
      await stopChildProcess(child, runnerTransport.pidFile);
    },
  };
}

function watchNatsConnectionClosed(connection: NatsConnection): Promise<IRunnerRuntimeClosed> {
  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<ChildProcessWithoutNullStreams> {
  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<void> {
  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<boolean> {
  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<void> {
  if (child.exitCode == null) {
    await new Promise<void>((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');
}