911 lines
26 KiB
JavaScript
911 lines
26 KiB
JavaScript
import { createHash } from "node:crypto";
|
|
import { execFile as execFileCallback } from "node:child_process";
|
|
import { promises as fs } from "node:fs";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import process from "node:process";
|
|
import { fileURLToPath } from "node:url";
|
|
import { promisify } from "node:util";
|
|
import WebSocket from "ws";
|
|
|
|
const DEFAULT_VERSION = "0.1.0";
|
|
const DEFAULT_RECONNECT_DELAY_MS = 5000;
|
|
const DEFAULT_SHELL_COLS = 120;
|
|
const DEFAULT_SHELL_ROWS = 32;
|
|
const DEFAULT_CPU_SAMPLE_DELAY_MS = 150;
|
|
const execFile = promisify(execFileCallback);
|
|
|
|
function buildBrokerHeartbeatState(isConnected) {
|
|
return isConnected
|
|
? {
|
|
status: "ONLINE",
|
|
metadata: { broker_connected: true },
|
|
}
|
|
: {
|
|
status: "DEGRADED",
|
|
metadata: { broker_connected: false },
|
|
};
|
|
}
|
|
|
|
export async function loadConfig(configPath) {
|
|
const raw = await fs.readFile(configPath, "utf8");
|
|
return JSON.parse(raw);
|
|
}
|
|
|
|
export async function saveConfig(configPath, config) {
|
|
await fs.writeFile(configPath, JSON.stringify(config, null, 2));
|
|
}
|
|
|
|
export function buildStatusReport(configPath, config) {
|
|
const heartbeatIntervalSeconds = Number(config.heartbeatIntervalSeconds || 15);
|
|
const claimed = Boolean(config.gatewayId && config.agentToken);
|
|
|
|
return {
|
|
command: "status",
|
|
configPath,
|
|
claimed,
|
|
state: claimed ? "CLAIMED" : "PENDING_CLAIM",
|
|
gatewayId: config.gatewayId ?? null,
|
|
apiUrl: config.apiUrl ?? null,
|
|
brokerUrl: config.brokerUrl ?? null,
|
|
brokerWsUrl: config.brokerUrl ? normalizeBrokerWsUrl(config.brokerUrl) : null,
|
|
hostname: config.hostname || os.hostname(),
|
|
releaseChannel: config.releaseChannel || "stable",
|
|
installedVersion: config.installedVersion || DEFAULT_VERSION,
|
|
targetVersion: config.targetVersion || config.installedVersion || DEFAULT_VERSION,
|
|
heartbeatIntervalSeconds: Number.isFinite(heartbeatIntervalSeconds) && heartbeatIntervalSeconds > 0
|
|
? heartbeatIntervalSeconds
|
|
: 15,
|
|
hasInstallToken: Boolean(config.installToken),
|
|
};
|
|
}
|
|
|
|
export async function getAgentStatus(configPath) {
|
|
if (!configPath) {
|
|
throw new Error("Missing --config path");
|
|
}
|
|
|
|
const config = await loadConfig(configPath);
|
|
return buildStatusReport(configPath, config);
|
|
}
|
|
|
|
export function normalizeBrokerWsUrl(brokerUrl) {
|
|
if (!brokerUrl) {
|
|
throw new Error("Missing broker URL");
|
|
}
|
|
if (brokerUrl.startsWith("ws://") || brokerUrl.startsWith("wss://")) {
|
|
return brokerUrl;
|
|
}
|
|
return brokerUrl.replace(/^http:/i, "ws:").replace(/^https:/i, "wss:");
|
|
}
|
|
|
|
function wait(delayMs) {
|
|
return new Promise((resolve) => setTimeout(resolve, delayMs));
|
|
}
|
|
|
|
function clampPercentage(value) {
|
|
if (!Number.isFinite(value)) {
|
|
return null;
|
|
}
|
|
|
|
return Math.max(0, Math.min(100, value));
|
|
}
|
|
|
|
function snapshotCpuTimes() {
|
|
return os.cpus().reduce(
|
|
(accumulator, cpu) => {
|
|
const total = Object.values(cpu.times).reduce((sum, current) => sum + current, 0);
|
|
return {
|
|
idle: accumulator.idle + cpu.times.idle,
|
|
total: accumulator.total + total,
|
|
};
|
|
},
|
|
{ idle: 0, total: 0 }
|
|
);
|
|
}
|
|
|
|
function calculateCpuUsagePercent(previousSnapshot, currentSnapshot) {
|
|
if (!previousSnapshot || !currentSnapshot) {
|
|
return null;
|
|
}
|
|
|
|
const totalDelta = currentSnapshot.total - previousSnapshot.total;
|
|
const idleDelta = currentSnapshot.idle - previousSnapshot.idle;
|
|
if (totalDelta <= 0) {
|
|
return null;
|
|
}
|
|
|
|
return clampPercentage(((totalDelta - idleDelta) / totalDelta) * 100);
|
|
}
|
|
|
|
async function sampleCpuUsage(previousSnapshot = null, sleepImpl = wait) {
|
|
if (previousSnapshot) {
|
|
const currentSnapshot = snapshotCpuTimes();
|
|
return {
|
|
snapshot: currentSnapshot,
|
|
usagePct: calculateCpuUsagePercent(previousSnapshot, currentSnapshot),
|
|
};
|
|
}
|
|
|
|
const initialSnapshot = snapshotCpuTimes();
|
|
await sleepImpl(DEFAULT_CPU_SAMPLE_DELAY_MS);
|
|
const sampledSnapshot = snapshotCpuTimes();
|
|
return {
|
|
snapshot: sampledSnapshot,
|
|
usagePct: calculateCpuUsagePercent(initialSnapshot, sampledSnapshot),
|
|
};
|
|
}
|
|
|
|
function buildMemoryMetrics() {
|
|
const totalBytes = os.totalmem();
|
|
const freeBytes = os.freemem();
|
|
const usedBytes = Math.max(0, totalBytes - freeBytes);
|
|
|
|
return {
|
|
memory_total_bytes: totalBytes,
|
|
memory_used_bytes: usedBytes,
|
|
memory_usage_pct: totalBytes > 0 ? clampPercentage((usedBytes / totalBytes) * 100) : null,
|
|
};
|
|
}
|
|
|
|
async function readDiskMetrics(execFileImpl = execFile) {
|
|
if (process.platform === "win32") {
|
|
return {
|
|
disk_total_bytes: null,
|
|
disk_used_bytes: null,
|
|
disk_usage_pct: null,
|
|
disk_mount: null,
|
|
};
|
|
}
|
|
|
|
try {
|
|
const { stdout } = await execFileImpl("df", ["-Pk", "/"], {
|
|
encoding: "utf8",
|
|
windowsHide: true,
|
|
});
|
|
const lines = String(stdout)
|
|
.split(/\r?\n/)
|
|
.map((line) => line.trim())
|
|
.filter(Boolean);
|
|
const dataLine = lines.at(-1);
|
|
if (!dataLine) {
|
|
throw new Error("Missing disk usage output");
|
|
}
|
|
|
|
const segments = dataLine.split(/\s+/);
|
|
if (segments.length < 6) {
|
|
throw new Error("Unexpected disk usage output");
|
|
}
|
|
|
|
const totalBytes = Number(segments[1]) * 1024;
|
|
const usedBytes = Number(segments[2]) * 1024;
|
|
const usagePct = clampPercentage(Number.parseFloat(String(segments[4]).replace("%", "")));
|
|
|
|
return {
|
|
disk_total_bytes: Number.isFinite(totalBytes) ? totalBytes : null,
|
|
disk_used_bytes: Number.isFinite(usedBytes) ? usedBytes : null,
|
|
disk_usage_pct: usagePct,
|
|
disk_mount: segments.slice(5).join(" ") || null,
|
|
};
|
|
} catch {
|
|
return {
|
|
disk_total_bytes: null,
|
|
disk_used_bytes: null,
|
|
disk_usage_pct: null,
|
|
disk_mount: null,
|
|
};
|
|
}
|
|
}
|
|
|
|
export async function collectSystemMetrics({
|
|
previousCpuSnapshot = null,
|
|
execFileImpl = execFile,
|
|
sleepImpl = wait,
|
|
latencyMs = null,
|
|
} = {}) {
|
|
const [cpuMetric, diskMetrics] = await Promise.all([
|
|
sampleCpuUsage(previousCpuSnapshot, sleepImpl),
|
|
readDiskMetrics(execFileImpl),
|
|
]);
|
|
|
|
return {
|
|
cpuSnapshot: cpuMetric.snapshot,
|
|
metrics: {
|
|
latency_ms: Number.isFinite(latencyMs) ? Math.max(0, Math.round(latencyMs)) : null,
|
|
cpu_usage_pct: cpuMetric.usagePct,
|
|
cpu_core_count: os.cpus().length,
|
|
load_average_1m: process.platform === "win32" ? null : os.loadavg()[0],
|
|
...buildMemoryMetrics(),
|
|
...diskMetrics,
|
|
},
|
|
};
|
|
}
|
|
|
|
export async function apiRequest(baseUrl, path, method = "POST", body = {}, fetchImpl = fetch) {
|
|
const response = await fetchImpl(`${String(baseUrl).replace(/\/$/, "")}${path}`, {
|
|
method,
|
|
headers: {
|
|
"content-type": "application/json",
|
|
},
|
|
body: method === "GET" ? undefined : JSON.stringify(body),
|
|
});
|
|
|
|
const json = await response.json();
|
|
if (!response.ok) {
|
|
throw new Error(json?.data?.message || json?.message || `HTTP ${response.status}`);
|
|
}
|
|
|
|
return json.data ?? json;
|
|
}
|
|
|
|
export async function claimIfNeeded(config, configPath, fetchImpl = fetch) {
|
|
if (config.gatewayId && config.agentToken) {
|
|
return config;
|
|
}
|
|
|
|
const claimed = await apiRequest(config.apiUrl, "/edge-agent/claim", "POST", {
|
|
token: config.installToken,
|
|
hostname: config.hostname || os.hostname(),
|
|
installed_version: config.installedVersion || DEFAULT_VERSION,
|
|
metadata: {
|
|
platform: process.platform,
|
|
arch: process.arch,
|
|
},
|
|
}, fetchImpl);
|
|
|
|
const nextConfig = {
|
|
...config,
|
|
gatewayId: claimed.gateway.id,
|
|
agentToken: claimed.agent_token,
|
|
brokerUrl: claimed.broker_url || config.brokerUrl,
|
|
releaseChannel: claimed.release_channel || config.releaseChannel || "stable",
|
|
};
|
|
await saveConfig(configPath, nextConfig);
|
|
return nextConfig;
|
|
}
|
|
|
|
async function fetchJson(url, fetchImpl = fetch) {
|
|
const response = await fetchImpl(url);
|
|
if (!response.ok) {
|
|
throw new Error(`HTTP ${response.status}`);
|
|
}
|
|
return response.json();
|
|
}
|
|
|
|
function expandCandidateIps(options = {}) {
|
|
if (Array.isArray(options.candidateIps) && options.candidateIps.length > 0) {
|
|
return options.candidateIps;
|
|
}
|
|
|
|
if (typeof options.subnetPrefix === "string") {
|
|
const start = Number.isInteger(options.startHost) ? options.startHost : 1;
|
|
const end = Number.isInteger(options.endHost) ? options.endHost : 20;
|
|
const ips = [];
|
|
for (let host = start; host <= end; host += 1) {
|
|
ips.push(`${options.subnetPrefix}.${host}`);
|
|
}
|
|
return ips;
|
|
}
|
|
|
|
return [];
|
|
}
|
|
|
|
export async function discoverShellyDevices(options = {}, fetchImpl = fetch) {
|
|
const candidateIps = expandCandidateIps(options);
|
|
const discovered = [];
|
|
|
|
await Promise.all(candidateIps.map(async (ip) => {
|
|
try {
|
|
const identity = await fetchJson(`http://${ip}/shelly`, fetchImpl);
|
|
discovered.push({
|
|
id: identity.mac || identity.id || ip,
|
|
device_id: identity.mac || identity.id || ip,
|
|
local_ip: ip,
|
|
model: identity.model || identity.type || "Shelly",
|
|
channel_count: Number(identity.num_outputs || identity.num_switches || 1),
|
|
online: true,
|
|
capabilities: {
|
|
generation: identity.gen || null,
|
|
},
|
|
metadata: identity,
|
|
});
|
|
} catch {
|
|
// Ignore non-responsive candidates during opportunistic discovery.
|
|
}
|
|
}));
|
|
|
|
return discovered;
|
|
}
|
|
|
|
export async function getRelayStatus(payload, fetchImpl = fetch) {
|
|
const ip = payload.localIp || payload.local_ip || payload.ip;
|
|
const channel = Number.isInteger(payload.channel) ? payload.channel : 0;
|
|
if (!ip) {
|
|
throw new Error("Missing relay local IP");
|
|
}
|
|
|
|
try {
|
|
const rpc = await fetchJson(`http://${ip}/rpc/Switch.GetStatus?id=${channel}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(rpc.output),
|
|
raw: rpc,
|
|
};
|
|
} catch {
|
|
const legacy = await fetchJson(`http://${ip}/relay/${channel}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(legacy.ison ?? legacy.output),
|
|
raw: legacy,
|
|
};
|
|
}
|
|
}
|
|
|
|
export async function setRelayState(payload, fetchImpl = fetch) {
|
|
const ip = payload.localIp || payload.local_ip || payload.ip;
|
|
const channel = Number.isInteger(payload.channel) ? payload.channel : 0;
|
|
const on = Boolean(payload.on);
|
|
if (!ip) {
|
|
throw new Error("Missing relay local IP");
|
|
}
|
|
|
|
try {
|
|
const rpc = await fetchJson(`http://${ip}/rpc/Switch.Set?id=${channel}&on=${on ? "true" : "false"}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(rpc.output ?? on),
|
|
raw: rpc,
|
|
};
|
|
} catch {
|
|
const legacy = await fetchJson(`http://${ip}/relay/${channel}?turn=${on ? "on" : "off"}`, fetchImpl);
|
|
return {
|
|
online: true,
|
|
on: Boolean(legacy.ison ?? legacy.output ?? on),
|
|
raw: legacy,
|
|
};
|
|
}
|
|
}
|
|
|
|
export async function runUpdate(payload, fetchImpl = fetch) {
|
|
if (!payload.artifactUrl) {
|
|
return {
|
|
updated: false,
|
|
skipped: true,
|
|
reason: "No artifact URL provided",
|
|
};
|
|
}
|
|
|
|
const response = await fetchImpl(payload.artifactUrl);
|
|
if (!response.ok) {
|
|
throw new Error(`Artifact download failed: HTTP ${response.status}`);
|
|
}
|
|
const buffer = Buffer.from(await response.arrayBuffer());
|
|
const sha256 = createHash("sha256").update(buffer).digest("hex");
|
|
if (payload.sha256 && String(payload.sha256).toLowerCase() !== sha256.toLowerCase()) {
|
|
throw new Error("Artifact checksum mismatch");
|
|
}
|
|
|
|
return {
|
|
updated: true,
|
|
sha256,
|
|
bytes: buffer.length,
|
|
};
|
|
}
|
|
|
|
function defaultShellCommand() {
|
|
if (process.platform === "win32") {
|
|
return { command: process.env.ComSpec || "cmd.exe", args: [] };
|
|
}
|
|
return { command: process.env.SHELL || "/bin/sh", args: [] };
|
|
}
|
|
|
|
function normalizeShellSize(value, fallback) {
|
|
const numeric = Number(value);
|
|
return Number.isFinite(numeric) && numeric > 0 ? Math.round(numeric) : fallback;
|
|
}
|
|
|
|
async function createDefaultPtyProcess(options = {}) {
|
|
const nodePtyModule = await import("node-pty");
|
|
const spawnPty =
|
|
nodePtyModule.spawn ??
|
|
nodePtyModule.default?.spawn ??
|
|
nodePtyModule.default;
|
|
|
|
if (typeof spawnPty !== "function") {
|
|
throw new Error("node-pty spawn is unavailable");
|
|
}
|
|
|
|
return spawnPty(options.command, options.args || [], {
|
|
name: options.env?.TERM || "xterm-256color",
|
|
cols: normalizeShellSize(options.cols, DEFAULT_SHELL_COLS),
|
|
rows: normalizeShellSize(options.rows, DEFAULT_SHELL_ROWS),
|
|
cwd: options.cwd || process.cwd(),
|
|
env: options.env || process.env,
|
|
});
|
|
}
|
|
|
|
export function createShellBridge(sendMessage, { createPtyProcess = createDefaultPtyProcess } = {}) {
|
|
const sessions = new Map();
|
|
|
|
const open = async (payload = {}) => {
|
|
const sessionId = String(payload.sessionId || "");
|
|
if (sessionId === "") {
|
|
return;
|
|
}
|
|
|
|
const existingSession = sessions.get(sessionId);
|
|
if (existingSession) {
|
|
existingSession.pty.kill();
|
|
sessions.delete(sessionId);
|
|
}
|
|
|
|
const shell = payload.shellCommand
|
|
? { command: payload.shellCommand, args: payload.shellArgs || [] }
|
|
: defaultShellCommand();
|
|
|
|
try {
|
|
const pty = await Promise.resolve(createPtyProcess({
|
|
command: shell.command,
|
|
args: shell.args,
|
|
cwd: payload.cwd || process.cwd(),
|
|
cols: normalizeShellSize(payload.cols, DEFAULT_SHELL_COLS),
|
|
rows: normalizeShellSize(payload.rows, DEFAULT_SHELL_ROWS),
|
|
env: {
|
|
...process.env,
|
|
TERM: payload.term || process.env.TERM || "xterm-256color",
|
|
},
|
|
}));
|
|
|
|
const sessionRecord = {
|
|
pty,
|
|
dataSubscription: null,
|
|
exitSubscription: null,
|
|
};
|
|
|
|
sessionRecord.dataSubscription = pty.onData((data) => {
|
|
sendMessage({ type: "SHELL_OUTPUT", sessionId, data: String(data || "") });
|
|
});
|
|
|
|
sessionRecord.exitSubscription = pty.onExit(({ exitCode }) => {
|
|
sessions.delete(sessionId);
|
|
sessionRecord.dataSubscription?.dispose?.();
|
|
sessionRecord.exitSubscription?.dispose?.();
|
|
sendMessage({ type: "SHELL_EXIT", sessionId, code: Number.isFinite(exitCode) ? exitCode : 0 });
|
|
});
|
|
|
|
sessions.set(sessionId, sessionRecord);
|
|
sendMessage({ type: "SHELL_OPENED", sessionId });
|
|
} catch (error) {
|
|
sendMessage({
|
|
type: "SHELL_OUTPUT",
|
|
sessionId,
|
|
data: `Failed to start root shell: ${error instanceof Error ? error.message : String(error)}\r\n`,
|
|
});
|
|
sendMessage({ type: "SHELL_EXIT", sessionId, code: 1 });
|
|
}
|
|
};
|
|
|
|
const input = (payload = {}) => {
|
|
const sessionRecord = sessions.get(String(payload.sessionId || ""));
|
|
if (!sessionRecord) {
|
|
return;
|
|
}
|
|
sessionRecord.pty.write(String(payload.data || ""));
|
|
};
|
|
|
|
const resize = (payload = {}) => {
|
|
const sessionRecord = sessions.get(String(payload.sessionId || ""));
|
|
if (!sessionRecord) {
|
|
return;
|
|
}
|
|
|
|
sessionRecord.pty.resize(
|
|
normalizeShellSize(payload.cols, DEFAULT_SHELL_COLS),
|
|
normalizeShellSize(payload.rows, DEFAULT_SHELL_ROWS)
|
|
);
|
|
};
|
|
|
|
const close = (payload = {}) => {
|
|
const sessionId = String(payload.sessionId || "");
|
|
const sessionRecord = sessions.get(sessionId);
|
|
if (!sessionRecord) {
|
|
return;
|
|
}
|
|
|
|
sessionRecord.pty.kill();
|
|
};
|
|
|
|
return { open, input, resize, close };
|
|
}
|
|
|
|
export async function handleAgentCommand(command, deps = {}) {
|
|
const fetchImpl = deps.fetchImpl || fetch;
|
|
switch (command.commandType) {
|
|
case "DISCOVER_SHELLY":
|
|
return { inventory: await discoverShellyDevices(command.payload || {}, fetchImpl) };
|
|
case "GET_RELAY_STATUS":
|
|
return await getRelayStatus(command.payload || {}, fetchImpl);
|
|
case "SET_RELAY_STATE":
|
|
return await setRelayState(command.payload || {}, fetchImpl);
|
|
case "RUN_UPDATE":
|
|
return await runUpdate(command.payload || {}, fetchImpl);
|
|
case "RESTART_AGENT":
|
|
return { restarted: true };
|
|
case "REBOOT_HOST":
|
|
return { rebooted: true };
|
|
default:
|
|
throw new Error(`Unsupported agent command: ${command.commandType}`);
|
|
}
|
|
}
|
|
|
|
export function buildHeartbeatPayload(config, extra = {}) {
|
|
const payload = {
|
|
hostname: os.hostname(),
|
|
installed_version: config.installedVersion || DEFAULT_VERSION,
|
|
target_version: config.targetVersion || config.installedVersion || DEFAULT_VERSION,
|
|
status: extra.status || "ONLINE",
|
|
metadata: {
|
|
release_channel: config.releaseChannel || "stable",
|
|
...extra.metadata,
|
|
},
|
|
};
|
|
|
|
if (Object.prototype.hasOwnProperty.call(extra, "discovery_status")) {
|
|
payload.discovery_status = extra.discovery_status;
|
|
}
|
|
|
|
if (Array.isArray(extra.inventory)) {
|
|
payload.inventory = extra.inventory;
|
|
}
|
|
|
|
return payload;
|
|
}
|
|
|
|
export async function sendHeartbeat(config, fetchImpl = fetch, extra = {}) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before heartbeat");
|
|
}
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/heartbeat`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
...buildHeartbeatPayload(config, extra),
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
async function sendTimedHeartbeat(config, fetchImpl = fetch, extra = {}) {
|
|
const startedAt = Date.now();
|
|
const data = await sendHeartbeat(config, fetchImpl, extra);
|
|
|
|
return {
|
|
data,
|
|
latencyMs: Math.max(0, Date.now() - startedAt),
|
|
};
|
|
}
|
|
|
|
export async function pollCommandJob(config, fetchImpl = fetch, waitSeconds = 20) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before polling commands");
|
|
}
|
|
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/commands/poll`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
wait_seconds: waitSeconds,
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
export async function submitCommandJobResult(config, jobId, { ok, payload = {}, error = null } = {}, fetchImpl = fetch) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before submitting command results");
|
|
}
|
|
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/commands/${jobId}/result`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
ok: Boolean(ok),
|
|
payload,
|
|
error,
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
export async function processPolledCommand(config, command, fetchImpl = fetch) {
|
|
const jobId = command?.id;
|
|
const commandType = command?.commandType || command?.command_type;
|
|
const payload = command?.payload || {};
|
|
|
|
if (!jobId || !commandType) {
|
|
return null;
|
|
}
|
|
|
|
try {
|
|
const result = await handleAgentCommand({ commandType, payload }, { fetchImpl });
|
|
await submitCommandJobResult(config, jobId, {
|
|
ok: true,
|
|
payload: result,
|
|
}, fetchImpl);
|
|
return { ok: true, payload: result };
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
await submitCommandJobResult(config, jobId, {
|
|
ok: false,
|
|
error: message,
|
|
}, fetchImpl);
|
|
return { ok: false, error: message };
|
|
}
|
|
}
|
|
|
|
export function connectBroker(
|
|
config,
|
|
{
|
|
fetchImpl = fetch,
|
|
wsFactory = (url) => new WebSocket(url),
|
|
createShellBridgeImpl = createShellBridge,
|
|
onOpen = null,
|
|
onClose = null,
|
|
onError = null,
|
|
} = {}
|
|
) {
|
|
const shell = createShellBridgeImpl((message) => {
|
|
if (socket.readyState === WebSocket.OPEN) {
|
|
socket.send(JSON.stringify(message));
|
|
}
|
|
});
|
|
const wsUrl = `${normalizeBrokerWsUrl(config.brokerUrl)}/ws/agent?gatewayId=${encodeURIComponent(config.gatewayId)}&token=${encodeURIComponent(config.agentToken)}`;
|
|
const socket = wsFactory(wsUrl);
|
|
|
|
socket.on("open", () => {
|
|
onOpen?.(socket);
|
|
});
|
|
|
|
socket.on("close", (...args) => {
|
|
onClose?.(...args);
|
|
});
|
|
|
|
socket.on("error", (error) => {
|
|
onError?.(error);
|
|
});
|
|
|
|
socket.on("message", async (raw) => {
|
|
const message = JSON.parse(raw.toString());
|
|
try {
|
|
if (message.type === "COMMAND") {
|
|
const normalizedPayload =
|
|
message.payload && typeof message.payload === "object" && message.payload.payload
|
|
? message.payload.payload
|
|
: (message.payload || {});
|
|
const payload = await handleAgentCommand({
|
|
commandType: message.commandType,
|
|
payload: normalizedPayload,
|
|
}, { fetchImpl });
|
|
socket.send(JSON.stringify({
|
|
type: "COMMAND_RESULT",
|
|
commandId: message.commandId,
|
|
ok: true,
|
|
payload,
|
|
}));
|
|
return;
|
|
}
|
|
|
|
if (message.type === "OPEN_ROOT_SHELL") {
|
|
await shell.open(message.payload || {});
|
|
} else if (message.type === "SHELL_INPUT") {
|
|
shell.input(message.payload || {});
|
|
} else if (message.type === "RESIZE_ROOT_SHELL") {
|
|
shell.resize(message.payload || {});
|
|
} else if (message.type === "CLOSE_ROOT_SHELL") {
|
|
shell.close(message.payload || {});
|
|
}
|
|
} catch (error) {
|
|
if (message.type === "COMMAND") {
|
|
socket.send(JSON.stringify({
|
|
type: "COMMAND_RESULT",
|
|
commandId: message.commandId,
|
|
ok: false,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
}));
|
|
}
|
|
}
|
|
});
|
|
|
|
return socket;
|
|
}
|
|
|
|
export async function startAgent({
|
|
configPath,
|
|
fetchImpl = fetch,
|
|
wsFactory,
|
|
collectMetricsImpl = collectSystemMetrics,
|
|
} = {}) {
|
|
if (!configPath) {
|
|
throw new Error("Missing --config path");
|
|
}
|
|
|
|
let config = await loadConfig(configPath);
|
|
config = await claimIfNeeded(config, configPath, fetchImpl);
|
|
const intervalMs = Number(config.heartbeatIntervalSeconds || 15) * 1000;
|
|
const reconnectDelayMs = Number(config.reconnectDelayMs || DEFAULT_RECONNECT_DELAY_MS);
|
|
const commandPollTimeoutSeconds = Number(config.commandPollTimeoutSeconds || 20);
|
|
const commandPollRetryDelayMs = Number(config.commandPollRetryDelayMs || 1000);
|
|
|
|
let brokerConnected = false;
|
|
let socket = null;
|
|
let reconnectTimer = null;
|
|
let stopped = false;
|
|
let cpuSnapshot = null;
|
|
let lastHeartbeatLatencyMs = null;
|
|
|
|
const sendBrokerAwareHeartbeat = async (extra = {}) => {
|
|
const brokerState = buildBrokerHeartbeatState(brokerConnected);
|
|
const collectedMetrics = await collectMetricsImpl({
|
|
previousCpuSnapshot: cpuSnapshot,
|
|
latencyMs: lastHeartbeatLatencyMs,
|
|
});
|
|
cpuSnapshot = collectedMetrics.cpuSnapshot ?? cpuSnapshot;
|
|
|
|
const extraMetadata =
|
|
extra.metadata && typeof extra.metadata === "object" && !Array.isArray(extra.metadata)
|
|
? extra.metadata
|
|
: {};
|
|
const { system_metrics: extraSystemMetrics, ...extraMetadataFields } = extraMetadata;
|
|
const metricPayload = {
|
|
...(collectedMetrics.metrics || {}),
|
|
...(extraSystemMetrics && typeof extraSystemMetrics === "object" ? extraSystemMetrics : {}),
|
|
};
|
|
|
|
const heartbeatPayload = {
|
|
...brokerState,
|
|
...extra,
|
|
metadata: {
|
|
...(brokerState.metadata || {}),
|
|
...extraMetadataFields,
|
|
system_metrics: metricPayload,
|
|
},
|
|
};
|
|
|
|
const response = await sendTimedHeartbeat(config, fetchImpl, heartbeatPayload);
|
|
lastHeartbeatLatencyMs = response.latencyMs;
|
|
return response.data;
|
|
};
|
|
|
|
const scheduleReconnect = () => {
|
|
if (stopped || reconnectTimer) {
|
|
return;
|
|
}
|
|
|
|
reconnectTimer = setTimeout(() => {
|
|
reconnectTimer = null;
|
|
openBrokerConnection();
|
|
}, reconnectDelayMs);
|
|
};
|
|
|
|
const runCommandPollLoop = async () => {
|
|
while (!stopped) {
|
|
try {
|
|
const command = await pollCommandJob(config, fetchImpl, commandPollTimeoutSeconds);
|
|
if (stopped) {
|
|
break;
|
|
}
|
|
|
|
if (!command) {
|
|
continue;
|
|
}
|
|
|
|
await processPolledCommand(config, command, fetchImpl);
|
|
} catch {
|
|
if (stopped) {
|
|
break;
|
|
}
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, commandPollRetryDelayMs));
|
|
}
|
|
}
|
|
};
|
|
|
|
const openBrokerConnection = () => {
|
|
if (stopped) {
|
|
return;
|
|
}
|
|
|
|
socket = connectBroker(config, {
|
|
fetchImpl,
|
|
wsFactory,
|
|
onOpen: () => {
|
|
brokerConnected = true;
|
|
sendBrokerAwareHeartbeat().catch(() => {});
|
|
},
|
|
onClose: () => {
|
|
brokerConnected = false;
|
|
sendBrokerAwareHeartbeat().catch(() => {});
|
|
scheduleReconnect();
|
|
},
|
|
onError: () => {
|
|
brokerConnected = false;
|
|
},
|
|
});
|
|
};
|
|
|
|
await sendBrokerAwareHeartbeat();
|
|
openBrokerConnection();
|
|
const commandPollPromise = runCommandPollLoop();
|
|
|
|
const timer = setInterval(() => {
|
|
sendBrokerAwareHeartbeat().catch(() => {});
|
|
}, intervalMs);
|
|
|
|
const stop = () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
if (reconnectTimer) {
|
|
clearTimeout(reconnectTimer);
|
|
reconnectTimer = null;
|
|
}
|
|
socket?.close?.();
|
|
};
|
|
|
|
return {
|
|
commandPollPromise,
|
|
get socket() {
|
|
return socket;
|
|
},
|
|
timer,
|
|
config,
|
|
stop,
|
|
};
|
|
}
|
|
|
|
export function parseCliArgs(argv = process.argv.slice(2)) {
|
|
let command = "start";
|
|
let configPath = null;
|
|
|
|
for (let index = 0; index < argv.length; index += 1) {
|
|
const arg = argv[index];
|
|
if (arg === "status" && command === "start") {
|
|
command = "status";
|
|
continue;
|
|
}
|
|
if (arg === "--config") {
|
|
configPath = argv[index + 1] || null;
|
|
index += 1;
|
|
continue;
|
|
}
|
|
}
|
|
|
|
return { command, configPath };
|
|
}
|
|
|
|
export async function runCli(argv = process.argv.slice(2)) {
|
|
const { command, configPath } = parseCliArgs(argv);
|
|
if (command === "status") {
|
|
const report = await getAgentStatus(configPath);
|
|
process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
|
|
return report;
|
|
}
|
|
|
|
return startAgent({ configPath });
|
|
}
|
|
|
|
const currentModulePath = fileURLToPath(import.meta.url);
|
|
const invokedModulePath = process.argv[1] ? path.resolve(process.argv[1]) : null;
|
|
|
|
if (invokedModulePath && currentModulePath === invokedModulePath) {
|
|
runCli().catch((error) => {
|
|
console.error(error instanceof Error ? error.message : String(error));
|
|
process.exitCode = 1;
|
|
});
|
|
}
|