pikasTech-HWLAB/scripts/gateway-outbound-demo-smoke.mjs

#!/usr/bin/env bun
import { spawn } from "node:child_process";
import { createServer } from "node:net";
import { existsSync } from "node:fs";
import http from "node:http";
import https from "node:https";
import path from "node:path";
import { setTimeout as delay } from "node:timers/promises";
import assert from "node:assert/strict";

const cwd = new URL("..", import.meta.url).pathname.replace(/^\/([A-Za-z]:)/u, "$1");
const useEdgeProxy = process.argv.includes("--edge");
const cloudPort = await freePort();
const gatewayPort = await freePort();
const edgePort = useEdgeProxy ? await freePort() : null;
const noProxy = mergeNoProxy(process.env.NO_PROXY, process.env.no_proxy, [
  "localhost",
  "127.0.0.1",
  "::1"
]);
const commonEnv = {
  ...process.env,
  NO_PROXY: noProxy,
  no_proxy: noProxy,
  HWLAB_CLOUD_RUNTIME_ADAPTER: "memory",
  HWLAB_CLOUD_RUNTIME_DURABLE: "false"
};
const children = [];
const bunCommand = resolveBunCommand();
const nodeCommand = process.env.HWLAB_NODE_COMMAND || "node";

try {
  const cloud = startNode("cmd/hwlab-cloud-api/main.ts", {
    ...commonEnv,
    HWLAB_CLOUD_API_HOST: "127.0.0.1",
    HWLAB_CLOUD_API_PORT: String(cloudPort),
    HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS: "10000"
  });
  children.push(cloud);
  await waitForJson(`http://127.0.0.1:${cloudPort}/health/live`);

  if (useEdgeProxy) {
    const edge = startNode("cmd/hwlab-edge-proxy/main.mjs", {
      ...commonEnv,
      HWLAB_EDGE_HOST: "127.0.0.1",
      HWLAB_EDGE_PORT: String(edgePort),
      HWLAB_EDGE_UPSTREAM: `http://127.0.0.1:${cloudPort}`
    });
    children.push(edge);
    await waitForJson(`http://127.0.0.1:${edgePort}/health`);
  }

  const apiBaseUrl = useEdgeProxy ? `http://127.0.0.1:${edgePort}` : `http://127.0.0.1:${cloudPort}`;
  const gateway = startNode("cmd/hwlab-gateway/main.ts", {
    ...commonEnv,
    PORT: String(gatewayPort),
    HWLAB_GATEWAY_CLOUD_URL: apiBaseUrl,
    HWLAB_GATEWAY_ID: "gtw_local_demo",
    HWLAB_GATEWAY_SESSION_ID: "gws_gtw_local_demo",
    HWLAB_GATEWAY_CMD_EXEC_ENABLED: "1",
    HWLAB_GATEWAY_DEMO_OPEN: "1",
    HWLAB_GATEWAY_POLL_INTERVAL_MS: "100",
    HWLAB_GATEWAY_CMD_TIMEOUT_MS: "5000"
  });
  children.push(gateway);
  await waitForGatewaySession(apiBaseUrl, "gws_gtw_local_demo");

  const rpc = await postJson(`${apiBaseUrl}/json-rpc`, {
    jsonrpc: "2.0",
    id: "req_gateway_outbound_demo_shell",
    method: "hardware.invoke.shell",
    params: {
      projectId: "prj_mvp_topology",
      gatewaySessionId: "gws_gtw_local_demo",
      resourceId: "res_windows_host",
      capabilityId: "cap_windows_cmd_exec",
      input: {
        command: "echo hwlab-demo",
        cwd,
        timeoutMs: 5000
      }
    },
    meta: {
      traceId: "trc_gateway_outbound_demo_shell",
      actorId: "usr_gateway_outbound_demo",
      serviceId: "hwlab-cloud-api",
      environment: "dev"
    }
  });

  assert.equal(rpc.error, undefined, JSON.stringify(rpc.error ?? {}, null, 2));
  assert.equal(rpc.result?.dispatch?.shellExecuted, true);
  assert.equal(rpc.result?.dispatch?.exitCode, 0);
  assert.match(rpc.result?.dispatch?.stdout ?? "", /hwlab-demo/u);

  const slow = postJson(`${apiBaseUrl}/json-rpc`, shellRpc({
    id: "req_gateway_outbound_demo_slow",
    traceId: "trc_gateway_outbound_demo_slow",
    command: process.platform === "win32" ? "ping -n 4 127.0.0.1 >NUL" : "sleep 3",
    cwd,
    timeoutMs: 5000
  }));
  await delay(250);
  const quickStartedAt = Date.now();
  const quick = await postJson(`${apiBaseUrl}/json-rpc`, shellRpc({
    id: "req_gateway_outbound_demo_quick",
    traceId: "trc_gateway_outbound_demo_quick",
    command: "echo hwlab-quick",
    cwd,
    timeoutMs: 5000
  }));
  const quickDurationMs = Date.now() - quickStartedAt;
  assert.equal(quick.result?.dispatch?.shellExecuted, true);
  assert.equal(quick.result?.dispatch?.exitCode, 0);
  assert.match(quick.result?.dispatch?.stdout ?? "", /hwlab-quick/u);
  assert.ok(quickDurationMs < 2500, `quick request was blocked by slow request for ${quickDurationMs}ms`);

  const sessionDuringSlow = await waitForGatewaySession(apiBaseUrl, "gws_gtw_local_demo");
  assert.equal(Number.isInteger(sessionDuringSlow.inflightCount), true);
  const slowResult = await slow;
  assert.equal(slowResult.result?.dispatch?.shellExecuted, true);
  assert.equal(slowResult.result?.dispatch?.exitCode, 0);

  console.log(JSON.stringify({
    status: "pass",
    mode: useEdgeProxy ? "edge-proxy" : "direct-cloud-api",
    cloudUrl: `http://127.0.0.1:${cloudPort}`,
    apiBaseUrl,
    gatewaySessionId: "gws_gtw_local_demo",
    stdout: rpc.result.dispatch.stdout.trim(),
    nonBlockingQuickMs: quickDurationMs,
    inflightCountObserved: sessionDuringSlow.inflightCount
  }, null, 2));
} finally {
  for (const child of children.reverse()) {
    stopChild(child);
  }
}

function startNode(entrypoint, env) {
  const command = entrypoint.endsWith(".ts") ? bunCommand : nodeCommand;
  const child = spawn(command, [entrypoint], {
    cwd,
    env,
    stdio: ["ignore", "pipe", "pipe"]
  });
  child.output = "";
  child.stderrOutput = "";
  child.stdout.on("data", (chunk) => {
    child.output += chunk.toString("utf8");
  });
  child.stderr.on("data", (chunk) => {
    child.stderrOutput += chunk.toString("utf8");
  });
  child.on("exit", (code, signal) => {
    child.exitInfo = { code, signal };
  });
  return child;
}

function resolveBunCommand() {
  const candidates = [
    process.env.HWLAB_BUN_COMMAND,
    process.env.HOME ? path.join(process.env.HOME, ".bun/bin/bun") : null,
    "bun"
  ].filter(Boolean);
  return candidates.find((candidate) => !candidate.includes("/") || existsSync(candidate)) || "bun";
}

async function waitForGatewaySession(apiBaseUrl, gatewaySessionId) {
  const deadline = Date.now() + 8000;
  let last;
  while (Date.now() < deadline) {
    try {
      const body = await getJson(`${apiBaseUrl}/v1/gateway/sessions`);
      last = body;
      const session = body.sessions?.find((item) => item.gatewaySessionId === gatewaySessionId);
      if (session?.status === "online") return session;
    } catch (error) {
      last = error;
    }
    await delay(100);
  }
  throw new Error(`gateway session ${gatewaySessionId} did not become online: ${stringifyLast(last)}`);
}

async function waitForJson(url) {
  const deadline = Date.now() + 8000;
  let last;
  while (Date.now() < deadline) {
    try {
      return await getJson(url);
    } catch (error) {
      last = error;
      await delay(100);
    }
  }
  throw new Error(`endpoint did not become ready: ${url}; last=${stringifyLast(last)}`);
}

async function getJson(url) {
  return requestJson(url, {
    method: "GET",
    headers: { accept: "application/json" }
  });
}

async function postJson(url, body) {
  return requestJson(url, {
    method: "POST",
    headers: {
      accept: "application/json",
      "content-type": "application/json"
    },
    body
  });
}

function requestJson(url, { method, headers = {}, body }) {
  return new Promise((resolve, reject) => {
    const target = new URL(url);
    const client = target.protocol === "https:" ? https : http;
    const payload = body === undefined ? "" : JSON.stringify(body);
    const request = client.request(target, {
      method,
      headers: {
        ...headers,
        ...(payload ? { "content-length": Buffer.byteLength(payload) } : {})
      }
    }, (response) => {
      let text = "";
      response.setEncoding("utf8");
      response.on("data", (chunk) => {
        text += chunk;
      });
      response.on("end", () => {
        if ((response.statusCode ?? 0) < 200 || (response.statusCode ?? 0) >= 300) {
          reject(new Error(`HTTP ${response.statusCode}: ${text}`));
          return;
        }
        try {
          resolve(text ? JSON.parse(text) : null);
        } catch (error) {
          reject(new Error(`non-JSON response: ${error.message}`));
        }
      });
    });
    request.on("error", reject);
    if (payload) request.write(payload);
    request.end();
  });
}

function shellRpc({ id, traceId, command, cwd, timeoutMs }) {
  return {
    jsonrpc: "2.0",
    id,
    method: "hardware.invoke.shell",
    params: {
      projectId: "prj_mvp_topology",
      gatewaySessionId: "gws_gtw_local_demo",
      resourceId: "res_windows_host",
      capabilityId: "cap_windows_cmd_exec",
      input: {
        command,
        cwd,
        timeoutMs
      }
    },
    meta: {
      traceId,
      actorId: "usr_gateway_outbound_demo",
      serviceId: "hwlab-cloud-api",
      environment: "dev"
    }
  };
}

function freePort() {
  return new Promise((resolve, reject) => {
    const server = createServer();
    server.once("error", reject);
    server.listen(0, "127.0.0.1", () => {
      const address = server.address();
      const port = typeof address === "object" && address ? address.port : null;
      server.close(() => resolve(port));
    });
  });
}

function mergeNoProxy(...parts) {
  const values = new Set();
  for (const part of parts.flat()) {
    for (const item of String(part ?? "").split(",")) {
      const trimmed = item.trim();
      if (trimmed) values.add(trimmed);
    }
  }
  return [...values].join(",");
}

function stopChild(child) {
  if (!child.killed && child.exitCode === null) {
    child.kill();
  }
}

function stringifyLast(value) {
  if (value instanceof Error) return value.message;
  return JSON.stringify(value);
}