cnc_wams/wasm-port/runtime/sdk/src/tool-db-node-runtime-adapter.js

import { createWriteStream } from "node:fs";
import { mkdir, rm } from "node:fs/promises";
import { dirname, join } from "node:path";
import { fileURLToPath } from "node:url";
import { spawn } from "node:child_process";

function requireString(value, label) {
  if (typeof value !== "string" || value.length === 0) {
    throw new Error(`${label} must be a non-empty string.`);
  }
  return value;
}

function defaultRepoRoot() {
  return fileURLToPath(new URL("../../../..", import.meta.url));
}

function appendPathEnv(current, next) {
  return current ? `${next}:${current}` : next;
}

export function createLinuxCncToolDbNodeRuntimeAdapter({
  repoRoot = defaultRepoRoot(),
  linuxCncRoot = join(repoRoot, "linuxcnc"),
  dbProgramPath = join(linuxCncRoot, "configs/sim/axis/db_demo/db_nonran.py"),
  dbProgramCwd = dirname(dbProgramPath),
  dbSavefile = join(repoRoot, "wasm-port/build/native/tool-db-runtime/sdk-node-db-nonran"),
  stdoutLog = join(repoRoot, "wasm-port/build/native/tool-db-runtime/sdk-node-db-program.stdout.log"),
  stderrLog = join(repoRoot, "wasm-port/build/native/tool-db-runtime/sdk-node-db-program.stderr.log"),
  periodMinutes = 999,
} = {}) {
  requireString(repoRoot, "repoRoot");
  requireString(linuxCncRoot, "linuxCncRoot");
  requireString(dbProgramPath, "dbProgramPath");
  requireString(dbProgramCwd, "dbProgramCwd");
  requireString(dbSavefile, "dbSavefile");

  let proc = null;
  let stdoutFile = null;
  let stderrFile = null;
  let stdoutBuffer = "";
  const lineQueue = [];
  const waiters = [];
  let closed = false;

  function flushWaiters() {
    while (lineQueue.length > 0 && waiters.length > 0) {
      const waiter = waiters.shift();
      waiter.resolve(lineQueue.shift());
    }
  }

  function failWaiters(error) {
    while (waiters.length > 0) {
      const waiter = waiters.shift();
      waiter.reject(error);
    }
  }

  function pushStdout(chunk) {
    stdoutBuffer += chunk.toString("utf8");
    let newlineIndex = stdoutBuffer.indexOf("\n");
    while (newlineIndex !== -1) {
      const line = stdoutBuffer.slice(0, newlineIndex).replace(/\r$/, "");
      stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1);
      lineQueue.push(line);
      newlineIndex = stdoutBuffer.indexOf("\n");
    }
    flushWaiters();
  }

  return {
    runtimeMode: "node-linuxcnc-db-program",
    dbSavefile,
    stdoutLog,
    stderrLog,

    async start() {
      await mkdir(dirname(dbSavefile), { recursive: true });
      await rm(dbSavefile, { force: true });
      stdoutFile = createWriteStream(stdoutLog, { encoding: "utf8" });
      stderrFile = createWriteStream(stderrLog, { encoding: "utf8" });

      const env = {
        ...process.env,
        PYTHONPATH: appendPathEnv(process.env.PYTHONPATH, join(linuxCncRoot, "lib/python")),
        LD_LIBRARY_PATH: appendPathEnv(process.env.LD_LIBRARY_PATH, join(linuxCncRoot, "lib")),
      };

      proc = spawn(dbProgramPath, [`--period_minutes=${periodMinutes}`, dbSavefile], {
        cwd: dbProgramCwd,
        env,
        stdio: ["pipe", "pipe", "pipe"],
      });
      proc.stdout.on("data", (chunk) => {
        stdoutFile.write(chunk);
        pushStdout(chunk);
      });
      proc.stderr.on("data", (chunk) => stderrFile.write(chunk));
      proc.on("error", (error) => failWaiters(error));
      proc.on("exit", (code) => {
        if (!closed && code !== 0) {
          failWaiters(new Error(`DB_PROGRAM exited with code ${code}.`));
        }
      });
    },

    async writeLine(line) {
      if (!proc?.stdin || proc.stdin.destroyed) {
        throw new Error("DB_PROGRAM stdin is not writable.");
      }
      proc.stdin.write(`${line}\n`);
    },

    async readLine({ timeoutMs = 5000 } = {}) {
      if (lineQueue.length > 0) {
        return lineQueue.shift();
      }
      return new Promise((resolve, reject) => {
        let waiter;
        const timer = setTimeout(() => {
          const index = waiters.indexOf(waiter);
          if (index !== -1) {
            waiters.splice(index, 1);
          }
          reject(new Error("timeout waiting for DB_PROGRAM output."));
        }, timeoutMs);
        waiter = {
          resolve: (line) => {
            clearTimeout(timer);
            resolve(line);
          },
          reject: (error) => {
            clearTimeout(timer);
            reject(error);
          },
        };
        waiters.push(waiter);
      });
    },

    async close() {
      closed = true;
      try {
        proc?.stdin?.end();
      } catch {
        // ignore close races
      }
      if (proc && proc.exitCode === null && proc.signalCode === null) {
        await new Promise((resolve) => {
          proc.once("exit", resolve);
          proc.kill("SIGTERM");
          setTimeout(resolve, 1000);
        });
      }
      stdoutFile?.end();
      stderrFile?.end();
    },
  };
}