runa/vscode/integration/server.ts

import { ChildProcess, spawn } from "node:child_process";
import { once } from "node:events";
import { resolve } from "node:path";
import { createInterface } from "node:readline";
import { json } from "../test/fixtures";
import WebSocket from "ws";
import { decodeNotification, WEBSOCKET_SUBPROTOCOL } from "../src/protocol";

export async function startImage(): Promise<{ address: string; stop: () => Promise<void> }> {
  const executable = resolve(
    __dirname,
    "../../../target/debug",
    process.platform === "win32" ? "runa.exe" : "runa",
  );
  const child = spawn(executable, ["--listen", "127.0.0.1:0"], {
    stdio: ["ignore", "ignore", "pipe"],
  });
  const stop = () => stopImage(child);
  let address: string;
  try {
    address = await new Promise<string>((resolve, reject) => {
      const lines = createInterface({ input: child.stderr });
      let diagnostics = "";
      const cleanup = () => {
        clearTimeout(timer);
        child.off("error", fail);
        child.off("exit", exited);
        lines.close();
        child.stderr.resume();
      };
      const fail = (error: Error) => {
        cleanup();
        reject(error);
      };
      const exited = () => fail(new Error(`Image exited during startup: ${diagnostics}`));
      const timer = setTimeout(
        () => fail(new Error(`Image startup timed out: ${diagnostics}`)),
        10000,
      );
      child.once("error", fail);
      child.once("exit", exited);
      lines.on("line", (line) => {
        diagnostics = (diagnostics + line + "\n").slice(-8192);
        const address = line.match(/^runa: listening on (.+)$/)?.[1];
        if (address) {
          cleanup();
          resolve(address);
        }
      });
    });
  } catch (error) {
    await stop();
    throw error;
  }
  return { address, stop };
}

async function stopImage(child: ChildProcess): Promise<void> {
  if (child.exitCode !== null || child.signalCode !== null || child.pid === undefined) return;
  const exited = once(child, "exit");
  child.kill();
  await exited;
}

export async function control(address: string, command: unknown): Promise<unknown> {
  const socket = new WebSocket(`ws://${address}/`, WEBSOCKET_SUBPROTOCOL);
  const timer = setTimeout(() => socket.terminate(), 5000);
  try {
    await once(socket, "open");
    const response = new Promise<unknown>((resolve, reject) => {
      socket.on("message", (data, binary) => {
        try {
          if (binary) throw new Error("Unexpected binary control reply");
          const text = data.toString();
          if (decodeNotification(text) !== undefined) return;
          const decoded: unknown = JSON.parse(text);
          resolve(decoded);
        } catch (error) {
          reject(error);
        }
      });
      socket.once("close", () => reject(new Error("Control connection closed")));
      socket.once("error", reject);
    });
    socket.send(json({ version: 4, request_id: "integration-control", command }));
    return await response;
  } finally {
    clearTimeout(timer);
    socket.close();
  }
}

export async function spawnCounter(address: string): Promise<string> {
  const reply = await control(address, {
    operation: "spawn",
    parameters: {
      path: "actors/counter",
      state: { type: "integer", data: 9223372036854775807n },
      mailbox_capacity: 4,
    },
  });
  const ref = json(reply).match(/"actor_ref":"([A-Za-z0-9_-]{21})"/)?.[1];
  if (!ref) throw new Error(`Spawn failed: ${json(reply)}`);
  return ref;
}