Jeppe Bundgaard
e93d3f30d2
- Implement generation detection logic for Shelly devices using model codes, metadata, and type inference. - Extend relay switch and inventory handling to include generation capabilities. - Ensure compatibility with Gen1, Gen2, and Gen3 devices for relay control and diagnostics. - Update unit tests to validate generation inference, fallback behavior, and API compatibility.
2175 lines
63 KiB
JavaScript
2175 lines
63 KiB
JavaScript
import { createHash } from "node:crypto";
|
|
import { execFile as execFileCallback, spawn as spawnCallback } 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";
|
|
|
|
const DEFAULT_VERSION = "0.1.0";
|
|
const DEFAULT_SHELL_COLS = 120;
|
|
const DEFAULT_SHELL_ROWS = 32;
|
|
const DEFAULT_CPU_SAMPLE_DELAY_MS = 150;
|
|
const DEFAULT_SHELL_EVENT_FLUSH_DELAY_MS = 40;
|
|
const DEFAULT_EDGE_AGENT_SERVICE_NAME = "truckwash-edge-agent.service";
|
|
const DEFAULT_UPDATE_VERIFICATION_TIMEOUT_SECONDS = 45;
|
|
const DEFAULT_UPDATE_VERIFY_INTERVAL_MS = 500;
|
|
const DEFAULT_UPDATE_RESTART_GRACE_MS = 150;
|
|
const DEFAULT_BROKER_RECONNECT_DELAY_MS = 1500;
|
|
const DEFAULT_SHELLY_LOCAL_HTTP_TIMEOUT_MS = 1200;
|
|
const UPDATE_VERIFY_COMMAND = "post-update-verify";
|
|
const execFile = promisify(execFileCallback);
|
|
|
|
function cloneJson(value) {
|
|
return value === undefined ? undefined : JSON.parse(JSON.stringify(value));
|
|
}
|
|
|
|
function replaceObjectContents(target, source) {
|
|
if (!target || typeof target !== "object" || Array.isArray(target) || !source || typeof source !== "object") {
|
|
return source;
|
|
}
|
|
|
|
for (const key of Object.keys(target)) {
|
|
delete target[key];
|
|
}
|
|
for (const [key, value] of Object.entries(source)) {
|
|
target[key] = value;
|
|
}
|
|
|
|
return target;
|
|
}
|
|
|
|
function formatUpdateTimestamp(date = new Date()) {
|
|
return date.toISOString();
|
|
}
|
|
|
|
function resolveUpdateVerificationTimeoutSeconds(config = {}, pendingUpdate = {}) {
|
|
const configuredTimeout = Number(
|
|
pendingUpdate.verificationTimeoutSeconds ??
|
|
config.updateVerificationTimeoutSeconds ??
|
|
DEFAULT_UPDATE_VERIFICATION_TIMEOUT_SECONDS
|
|
);
|
|
return Number.isFinite(configuredTimeout) && configuredTimeout > 0
|
|
? Math.max(5, Math.round(configuredTimeout))
|
|
: DEFAULT_UPDATE_VERIFICATION_TIMEOUT_SECONDS;
|
|
}
|
|
|
|
function resolveUpdatePaths(configPath, config = {}) {
|
|
const installDir = path.resolve(String(config.installDir || path.dirname(configPath)));
|
|
return {
|
|
installDir,
|
|
configPath: path.resolve(configPath),
|
|
agentPath: path.resolve(String(config.agentPath || path.join(installDir, "agent.mjs"))),
|
|
packagePath: path.resolve(String(config.packagePath || path.join(installDir, "package.json"))),
|
|
updatesDir: path.resolve(String(config.updatesDir || path.join(installDir, ".updates"))),
|
|
serviceName: String(config.serviceName || DEFAULT_EDGE_AGENT_SERVICE_NAME),
|
|
};
|
|
}
|
|
|
|
function sanitizeUpdateSegment(value, fallback = "update") {
|
|
const normalized = String(value || "")
|
|
.trim()
|
|
.replace(/[^a-zA-Z0-9._-]+/g, "-")
|
|
.replace(/^-+|-+$/g, "");
|
|
return normalized === "" ? fallback : normalized;
|
|
}
|
|
|
|
function buildUpdateMetadata(config = {}) {
|
|
const lastUpdate = config.lastUpdate;
|
|
if (!lastUpdate || typeof lastUpdate !== "object" || Array.isArray(lastUpdate)) {
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
state: String(lastUpdate.state || "UNKNOWN"),
|
|
target_version: lastUpdate.targetVersion ?? null,
|
|
previous_version: lastUpdate.previousVersion ?? null,
|
|
restored_version: lastUpdate.restoredVersion ?? null,
|
|
error: lastUpdate.error ?? null,
|
|
completed_at: lastUpdate.completedAt ?? null,
|
|
};
|
|
}
|
|
|
|
async function pathExists(pathToCheck) {
|
|
try {
|
|
await fs.access(pathToCheck);
|
|
return true;
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
async function ensureDirectory(pathToEnsure) {
|
|
await fs.mkdir(pathToEnsure, { recursive: true });
|
|
}
|
|
|
|
async function writeBuffer(pathToWrite, buffer) {
|
|
await fs.writeFile(pathToWrite, buffer);
|
|
}
|
|
|
|
async function backupFileIfPresent(sourcePath, destinationPath) {
|
|
if (!(await pathExists(sourcePath))) {
|
|
return false;
|
|
}
|
|
|
|
await ensureDirectory(path.dirname(destinationPath));
|
|
await fs.copyFile(sourcePath, destinationPath);
|
|
return true;
|
|
}
|
|
|
|
async function restoreFileIfPresent(sourcePath, destinationPath) {
|
|
if (!(await pathExists(sourcePath))) {
|
|
return false;
|
|
}
|
|
|
|
await ensureDirectory(path.dirname(destinationPath));
|
|
await fs.copyFile(sourcePath, destinationPath);
|
|
return true;
|
|
}
|
|
|
|
function createUpdateErrorMessage(error, fallback = "Edge agent update failed") {
|
|
return error instanceof Error ? error.message : String(error || fallback);
|
|
}
|
|
|
|
function buildTransportHeartbeatState(brokerState = {}) {
|
|
const brokerConnected = Boolean(brokerState.connected);
|
|
return {
|
|
status: "ONLINE",
|
|
metadata: {
|
|
command_transport: brokerConnected ? "BROKER_FAST_PATH" : "API_POLLING",
|
|
shell_transport: brokerConnected ? "BROKER_FAST_PATH" : "API_POLLING",
|
|
broker_connected: brokerConnected,
|
|
broker_url: brokerState.url || null,
|
|
broker_last_error: brokerState.lastError || null,
|
|
broker_last_connected_at: brokerState.lastConnectedAt || null,
|
|
broker_last_disconnected_at: brokerState.lastDisconnectedAt || null,
|
|
broker_disconnect_reason: brokerState.disconnectReason || null,
|
|
broker_last_close_code: brokerState.lastCloseCode ?? null,
|
|
broker_last_close_clean: brokerState.lastCloseClean ?? null,
|
|
},
|
|
};
|
|
}
|
|
|
|
function normalizeBrokerBaseUrl(value) {
|
|
const trimmed = String(value || "").trim().replace(/\/+$/, "");
|
|
if (trimmed === "") {
|
|
return null;
|
|
}
|
|
if (trimmed.startsWith("ws://") || trimmed.startsWith("wss://")) {
|
|
return trimmed;
|
|
}
|
|
if (trimmed.startsWith("https://")) {
|
|
return `wss://${trimmed.slice("https://".length)}`;
|
|
}
|
|
if (trimmed.startsWith("http://")) {
|
|
return `ws://${trimmed.slice("http://".length)}`;
|
|
}
|
|
return `ws://${trimmed}`;
|
|
}
|
|
|
|
function buildBrokerSocketUrl(brokerUrl, gatewayId, token) {
|
|
const baseUrl = normalizeBrokerBaseUrl(brokerUrl);
|
|
if (!baseUrl || !gatewayId || !token) {
|
|
return null;
|
|
}
|
|
|
|
const search = new URLSearchParams({
|
|
gatewayId: String(gatewayId),
|
|
token: String(token),
|
|
});
|
|
return `${baseUrl}/ws/agent?${search.toString()}`;
|
|
}
|
|
|
|
function redactBrokerSocketUrl(value) {
|
|
try {
|
|
const parsed = new URL(String(value || ""));
|
|
for (const key of ["token", "agentToken", "agent_token"]) {
|
|
if (parsed.searchParams.has(key)) {
|
|
parsed.searchParams.set(key, "***");
|
|
}
|
|
}
|
|
return parsed.toString();
|
|
} catch {
|
|
return String(value || "").replace(/([?&](?:token|agentToken|agent_token)=)[^&]+/gi, "$1***");
|
|
}
|
|
}
|
|
|
|
function normalizeBrokerSocketError(error, socketUrl) {
|
|
const message = error instanceof Error && error.message
|
|
? error.message
|
|
: error?.message
|
|
? String(error.message)
|
|
: "Broker connection failed";
|
|
const redactedUrl = redactBrokerSocketUrl(socketUrl);
|
|
return redactedUrl ? `${message} (${redactedUrl})` : message;
|
|
}
|
|
|
|
async function readSocketMessageText(data) {
|
|
if (typeof data === "string") {
|
|
return data;
|
|
}
|
|
if (data instanceof ArrayBuffer) {
|
|
return Buffer.from(data).toString("utf8");
|
|
}
|
|
if (ArrayBuffer.isView(data)) {
|
|
return Buffer.from(data.buffer, data.byteOffset, data.byteLength).toString("utf8");
|
|
}
|
|
if (typeof Blob !== "undefined" && data instanceof Blob) {
|
|
return data.text();
|
|
}
|
|
return String(data || "");
|
|
}
|
|
|
|
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,
|
|
transport: config.brokerUrl ? "HYBRID" : "API_POLLING",
|
|
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);
|
|
}
|
|
|
|
function wait(delayMs) {
|
|
return new Promise((resolve) => {
|
|
const timer = setTimeout(resolve, delayMs);
|
|
timer.unref?.();
|
|
});
|
|
}
|
|
|
|
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,
|
|
releaseChannel: claimed.release_channel || config.releaseChannel || "stable",
|
|
};
|
|
if (claimed.broker_url || config.brokerUrl) {
|
|
nextConfig.brokerUrl = claimed.broker_url || config.brokerUrl;
|
|
}
|
|
await saveConfig(configPath, nextConfig);
|
|
return nextConfig;
|
|
}
|
|
|
|
function makeHttpTimeoutError(timeoutMs) {
|
|
const error = new Error(`HTTP request timed out after ${timeoutMs}ms`);
|
|
error.code = "EDGE_AGENT_HTTP_TIMEOUT";
|
|
return error;
|
|
}
|
|
|
|
function isHttpTimeoutError(error) {
|
|
return error?.code === "EDGE_AGENT_HTTP_TIMEOUT";
|
|
}
|
|
|
|
function resolveShellyLocalHttpTimeoutMs(options = {}) {
|
|
const configured = Number(options.timeoutMs ?? process.env.EDGE_SHELLY_LOCAL_HTTP_TIMEOUT_MS);
|
|
if (Number.isFinite(configured) && configured > 0) {
|
|
return Math.max(50, Math.floor(configured));
|
|
}
|
|
|
|
return DEFAULT_SHELLY_LOCAL_HTTP_TIMEOUT_MS;
|
|
}
|
|
|
|
function resolveRelayToggleAfterSeconds(payload = {}) {
|
|
const configured = Number(payload.toggleAfter ?? payload.toggle_after ?? payload.timer);
|
|
if (!Number.isFinite(configured) || configured <= 0) {
|
|
return null;
|
|
}
|
|
|
|
return Math.floor(configured);
|
|
}
|
|
|
|
async function fetchJson(url, fetchImpl = fetch, options = {}) {
|
|
const timeoutMs = Number(options.timeoutMs || 0);
|
|
let timeout = null;
|
|
let controller = null;
|
|
const requestOptions = {};
|
|
if (timeoutMs > 0 && typeof AbortController !== "undefined") {
|
|
controller = new AbortController();
|
|
requestOptions.signal = controller.signal;
|
|
}
|
|
|
|
let response;
|
|
try {
|
|
const fetchPromise = Promise.resolve().then(() => fetchImpl(url, requestOptions));
|
|
response = timeoutMs > 0
|
|
? await Promise.race([
|
|
fetchPromise,
|
|
new Promise((_, reject) => {
|
|
timeout = setTimeout(() => {
|
|
controller?.abort();
|
|
reject(makeHttpTimeoutError(timeoutMs));
|
|
}, timeoutMs);
|
|
}),
|
|
])
|
|
: await fetchPromise;
|
|
} finally {
|
|
if (timeout !== null) {
|
|
clearTimeout(timeout);
|
|
}
|
|
}
|
|
|
|
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 [];
|
|
}
|
|
|
|
function normalizeShellyDeviceGeneration(value) {
|
|
if (Number.isInteger(value) && value > 0) {
|
|
return value;
|
|
}
|
|
|
|
if (typeof value === "number" && Number.isFinite(value) && value > 0) {
|
|
return Math.trunc(value);
|
|
}
|
|
|
|
return inferShellyDeviceGenerationFromString(value);
|
|
}
|
|
|
|
function inferShellyDeviceGenerationFromString(value) {
|
|
const normalized = String(value || "").trim();
|
|
if (!normalized) {
|
|
return null;
|
|
}
|
|
|
|
const explicit = normalized.match(/\bgen(?:eration)?\s*([1-9]\d*)\b/i);
|
|
if (explicit) {
|
|
return Number(explicit[1]);
|
|
}
|
|
|
|
const upper = normalized.toUpperCase();
|
|
const sSeries = upper.match(/\bS([3-9])(?:[A-Z0-9]+)?-[A-Z0-9-]+\b/);
|
|
if (sSeries) {
|
|
return Number(sSeries[1]);
|
|
}
|
|
|
|
if (/\b(?:SHELLY\s+)?(?:PLUS|PRO)\b/i.test(normalized) || /\bSP[A-Z0-9]+-[A-Z0-9-]+\b/.test(upper)) {
|
|
return 2;
|
|
}
|
|
|
|
if (/\bSH[A-Z0-9]+-?[A-Z0-9-]*\b/.test(upper)) {
|
|
return 1;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function resolveShellyDeviceGeneration(identity = {}) {
|
|
for (const candidate of [
|
|
identity.gen,
|
|
identity.generation,
|
|
identity.device_generation,
|
|
identity.capabilities?.generation,
|
|
]) {
|
|
const generation = normalizeShellyDeviceGeneration(candidate);
|
|
if (generation !== null) {
|
|
return generation;
|
|
}
|
|
}
|
|
|
|
for (const candidate of [
|
|
identity.model,
|
|
identity.type,
|
|
identity.app,
|
|
identity.name,
|
|
identity.id,
|
|
identity.mac,
|
|
]) {
|
|
const generation = inferShellyDeviceGenerationFromString(candidate);
|
|
if (generation !== null) {
|
|
return generation;
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function resolveShellyCommandGeneration(payload = {}) {
|
|
return resolveShellyDeviceGeneration({
|
|
gen: payload.gen ?? payload.generation ?? payload.deviceGeneration ?? payload.device_generation,
|
|
device_generation: payload.device_generation,
|
|
capabilities: payload.capabilities,
|
|
model: payload.model ?? payload.deviceModel ?? payload.device_model ?? payload.deviceType ?? payload.device_type,
|
|
type: payload.type ?? payload.deviceType ?? payload.device_type,
|
|
app: payload.app,
|
|
name: payload.name ?? payload.deviceName ?? payload.device_name,
|
|
id: payload.deviceId ?? payload.device_id ?? payload.relayId ?? payload.relay_id,
|
|
mac: payload.mac,
|
|
});
|
|
}
|
|
|
|
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, {
|
|
timeoutMs: resolveShellyLocalHttpTimeoutMs(options),
|
|
});
|
|
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: resolveShellyDeviceGeneration(identity),
|
|
},
|
|
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;
|
|
const timeoutMs = resolveShellyLocalHttpTimeoutMs(payload);
|
|
if (!ip) {
|
|
throw new Error("Missing relay local IP");
|
|
}
|
|
|
|
const attempts = [
|
|
async () => {
|
|
const rpc = await fetchJson(`http://${ip}/rpc/Switch.GetStatus?id=${channel}`, fetchImpl, { timeoutMs });
|
|
return {
|
|
online: true,
|
|
on: Boolean(rpc.output),
|
|
raw: rpc,
|
|
};
|
|
},
|
|
async () => {
|
|
const legacy = await fetchJson(`http://${ip}/relay/${channel}`, fetchImpl, { timeoutMs });
|
|
return {
|
|
online: true,
|
|
on: Boolean(legacy.ison ?? legacy.output),
|
|
raw: legacy,
|
|
};
|
|
},
|
|
];
|
|
|
|
try {
|
|
return await Promise.any(attempts.map((attempt) => attempt()));
|
|
} catch (error) {
|
|
const errors = Array.isArray(error?.errors) ? error.errors : [error];
|
|
const message = errors
|
|
.map((entry) => entry?.message || String(entry))
|
|
.filter(Boolean)
|
|
.join("; ");
|
|
const relayStatusError = new Error(message ? `Unable to read relay status: ${message}` : "Unable to read relay status");
|
|
relayStatusError.cause = error;
|
|
throw relayStatusError;
|
|
}
|
|
}
|
|
|
|
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);
|
|
const timeoutMs = resolveShellyLocalHttpTimeoutMs(payload);
|
|
const toggleAfter = resolveRelayToggleAfterSeconds(payload);
|
|
const timerQuery = toggleAfter === null ? "" : `&toggle_after=${encodeURIComponent(String(toggleAfter))}`;
|
|
const legacyTimerQuery = toggleAfter === null ? "" : `&timer=${encodeURIComponent(String(toggleAfter))}`;
|
|
const deviceGeneration = resolveShellyCommandGeneration(payload);
|
|
if (!ip) {
|
|
throw new Error("Missing relay local IP");
|
|
}
|
|
|
|
const runRpcSwitch = async () => {
|
|
const rpc = await fetchJson(`http://${ip}/rpc/Switch.Set?id=${channel}&on=${on ? "true" : "false"}${timerQuery}`, fetchImpl, { timeoutMs });
|
|
return {
|
|
online: true,
|
|
on: Boolean(rpc.output ?? on),
|
|
raw: rpc,
|
|
};
|
|
};
|
|
const runLegacySwitch = async () => {
|
|
const legacy = await fetchJson(`http://${ip}/relay/${channel}?turn=${on ? "on" : "off"}${legacyTimerQuery}`, fetchImpl, { timeoutMs });
|
|
return {
|
|
online: true,
|
|
on: Boolean(legacy.ison ?? legacy.output ?? on),
|
|
raw: legacy,
|
|
};
|
|
};
|
|
|
|
if (toggleAfter !== null) {
|
|
const attempts = deviceGeneration === 1
|
|
? [runLegacySwitch, runRpcSwitch]
|
|
: [runRpcSwitch, runLegacySwitch];
|
|
|
|
let lastError = null;
|
|
for (const attempt of attempts) {
|
|
try {
|
|
return await attempt();
|
|
} catch (error) {
|
|
if (isHttpTimeoutError(error)) {
|
|
throw error;
|
|
}
|
|
lastError = error;
|
|
}
|
|
}
|
|
|
|
throw lastError || new Error("Unable to switch relay");
|
|
}
|
|
|
|
try {
|
|
return await runRpcSwitch();
|
|
} catch (error) {
|
|
if (isHttpTimeoutError(error)) {
|
|
throw error;
|
|
}
|
|
return await runLegacySwitch();
|
|
}
|
|
}
|
|
|
|
async function fetchArtifactBuffer(url, expectedSha256, label, fetchImpl = fetch) {
|
|
if (!url) {
|
|
return null;
|
|
}
|
|
|
|
const response = await fetchImpl(url);
|
|
if (!response.ok) {
|
|
throw new Error(`${label} download failed: HTTP ${response.status}`);
|
|
}
|
|
|
|
const buffer = Buffer.from(await response.arrayBuffer());
|
|
const sha256 = createHash("sha256").update(buffer).digest("hex");
|
|
if (expectedSha256 && String(expectedSha256).toLowerCase() !== sha256.toLowerCase()) {
|
|
throw new Error(`${label} checksum mismatch`);
|
|
}
|
|
|
|
return {
|
|
buffer,
|
|
sha256,
|
|
bytes: buffer.length,
|
|
};
|
|
}
|
|
|
|
async function installAgentDependencies(installDir, execFileImpl = execFile) {
|
|
await execFileImpl("npm", ["install", "--omit=dev"], {
|
|
cwd: installDir,
|
|
encoding: "utf8",
|
|
windowsHide: true,
|
|
});
|
|
}
|
|
|
|
async function runAgentStatusPreflight(agentPath, configPath, execFileImpl = execFile) {
|
|
await execFileImpl(process.execPath, [agentPath, "status", "--config", configPath], {
|
|
encoding: "utf8",
|
|
windowsHide: true,
|
|
});
|
|
}
|
|
|
|
function buildPendingUpdateState(payload, config, paths, backupDir) {
|
|
return {
|
|
targetVersion: String(payload.targetVersion || payload.target_version || config.targetVersion || config.installedVersion || DEFAULT_VERSION),
|
|
previousVersion: String(config.installedVersion || DEFAULT_VERSION),
|
|
releaseChannel: String(payload.releaseChannel || payload.release_channel || config.releaseChannel || "stable"),
|
|
requestedAt: formatUpdateTimestamp(),
|
|
backupDir,
|
|
installDir: paths.installDir,
|
|
serviceName: paths.serviceName,
|
|
restartMode: String(payload.restartMode || config.restartMode || (process.platform === "win32" ? "spawn" : "systemd")),
|
|
verificationTimeoutSeconds: resolveUpdateVerificationTimeoutSeconds(config),
|
|
};
|
|
}
|
|
|
|
function buildPreparedUpdatePayload({
|
|
pendingUpdate,
|
|
agentArtifact,
|
|
packageArtifact,
|
|
paths,
|
|
}) {
|
|
return {
|
|
updated: true,
|
|
verification_pending: true,
|
|
target_version: pendingUpdate.targetVersion,
|
|
previous_version: pendingUpdate.previousVersion,
|
|
release_channel: pendingUpdate.releaseChannel,
|
|
restart_mode: pendingUpdate.restartMode,
|
|
verification_timeout_seconds: pendingUpdate.verificationTimeoutSeconds,
|
|
artifact_sha256: agentArtifact?.sha256 ?? null,
|
|
artifact_bytes: agentArtifact?.bytes ?? null,
|
|
package_sha256: packageArtifact?.sha256 ?? null,
|
|
package_bytes: packageArtifact?.bytes ?? null,
|
|
install_dir: paths.installDir,
|
|
};
|
|
}
|
|
|
|
async function restorePreparedUpdate(paths, pendingUpdate, originalConfig, {
|
|
execFileImpl = execFile,
|
|
liveConfig = null,
|
|
rollbackMessage = "Edge agent update failed",
|
|
} = {}) {
|
|
const backupDir = pendingUpdate?.backupDir;
|
|
if (!backupDir) {
|
|
if (originalConfig && pendingUpdate !== null) {
|
|
await saveConfig(paths.configPath, originalConfig);
|
|
if (liveConfig) {
|
|
replaceObjectContents(liveConfig, cloneJson(originalConfig));
|
|
}
|
|
}
|
|
return {
|
|
rolledBack: false,
|
|
restoredVersion: originalConfig?.installedVersion || null,
|
|
};
|
|
}
|
|
|
|
await restoreFileIfPresent(path.join(backupDir, "agent.mjs"), paths.agentPath);
|
|
await restoreFileIfPresent(path.join(backupDir, "package.json"), paths.packagePath);
|
|
|
|
try {
|
|
await installAgentDependencies(paths.installDir, execFileImpl);
|
|
} catch {
|
|
// Prefer preserving the restored files and rollback metadata over surfacing a secondary npm failure here.
|
|
}
|
|
|
|
const nextConfig = {
|
|
...(cloneJson(originalConfig) || {}),
|
|
targetVersion: pendingUpdate.previousVersion || originalConfig?.targetVersion || originalConfig?.installedVersion || DEFAULT_VERSION,
|
|
installedVersion: pendingUpdate.previousVersion || originalConfig?.installedVersion || DEFAULT_VERSION,
|
|
pendingUpdate: null,
|
|
lastUpdate: {
|
|
state: "ROLLED_BACK",
|
|
targetVersion: pendingUpdate.targetVersion ?? null,
|
|
previousVersion: pendingUpdate.previousVersion ?? null,
|
|
restoredVersion: pendingUpdate.previousVersion ?? null,
|
|
error: rollbackMessage,
|
|
completedAt: formatUpdateTimestamp(),
|
|
},
|
|
};
|
|
|
|
await saveConfig(paths.configPath, nextConfig);
|
|
if (liveConfig) {
|
|
replaceObjectContents(liveConfig, cloneJson(nextConfig));
|
|
}
|
|
|
|
return {
|
|
rolledBack: true,
|
|
restoredVersion: nextConfig.installedVersion || null,
|
|
};
|
|
}
|
|
|
|
export async function rollbackPendingUpdate(configPath, {
|
|
config = null,
|
|
execFileImpl = execFile,
|
|
liveConfig = null,
|
|
reason = "Edge agent update verification failed",
|
|
} = {}) {
|
|
const currentConfig = cloneJson(config || (await loadConfig(configPath)));
|
|
const pendingUpdate = currentConfig.pendingUpdate;
|
|
if (!pendingUpdate || typeof pendingUpdate !== "object") {
|
|
return {
|
|
rolledBack: false,
|
|
restoredVersion: currentConfig.installedVersion || null,
|
|
};
|
|
}
|
|
|
|
const paths = resolveUpdatePaths(configPath, currentConfig);
|
|
const originalConfig = {
|
|
...currentConfig,
|
|
pendingUpdate: null,
|
|
lastUpdate: currentConfig.lastUpdate ?? null,
|
|
};
|
|
originalConfig.installedVersion = pendingUpdate.previousVersion || originalConfig.installedVersion || DEFAULT_VERSION;
|
|
originalConfig.targetVersion = pendingUpdate.previousVersion || originalConfig.targetVersion || originalConfig.installedVersion;
|
|
|
|
return restorePreparedUpdate(paths, pendingUpdate, originalConfig, {
|
|
execFileImpl,
|
|
liveConfig,
|
|
rollbackMessage: reason,
|
|
});
|
|
}
|
|
|
|
export async function finalizePendingUpdateOnStartup(config, configPath, {
|
|
liveConfig = null,
|
|
} = {}) {
|
|
const pendingUpdate = config?.pendingUpdate;
|
|
if (!pendingUpdate || typeof pendingUpdate !== "object") {
|
|
return config;
|
|
}
|
|
|
|
const nextConfig = {
|
|
...cloneJson(config),
|
|
installedVersion:
|
|
pendingUpdate.targetVersion ||
|
|
config.targetVersion ||
|
|
config.installedVersion ||
|
|
DEFAULT_VERSION,
|
|
targetVersion:
|
|
pendingUpdate.targetVersion ||
|
|
config.targetVersion ||
|
|
config.installedVersion ||
|
|
DEFAULT_VERSION,
|
|
releaseChannel: pendingUpdate.releaseChannel || config.releaseChannel || "stable",
|
|
pendingUpdate: null,
|
|
lastUpdate: {
|
|
state: "COMPLETED",
|
|
targetVersion: pendingUpdate.targetVersion ?? null,
|
|
previousVersion: pendingUpdate.previousVersion ?? null,
|
|
restoredVersion: null,
|
|
error: null,
|
|
completedAt: formatUpdateTimestamp(),
|
|
},
|
|
};
|
|
|
|
await saveConfig(configPath, nextConfig);
|
|
if (liveConfig) {
|
|
replaceObjectContents(liveConfig, cloneJson(nextConfig));
|
|
return liveConfig;
|
|
}
|
|
|
|
return nextConfig;
|
|
}
|
|
|
|
function buildUpdateRestartPlan(payload, configPath, config = {}) {
|
|
const paths = resolveUpdatePaths(configPath, config);
|
|
return {
|
|
configPath: paths.configPath,
|
|
agentPath: paths.agentPath,
|
|
serviceName: String(
|
|
payload.serviceName ||
|
|
payload.service_name ||
|
|
config.serviceName ||
|
|
DEFAULT_EDGE_AGENT_SERVICE_NAME
|
|
),
|
|
restartMode: String(
|
|
payload.restartMode ||
|
|
payload.restart_mode ||
|
|
config.restartMode ||
|
|
(process.platform === "win32" ? "spawn" : "systemd")
|
|
),
|
|
restartGraceMs: DEFAULT_UPDATE_RESTART_GRACE_MS,
|
|
};
|
|
}
|
|
|
|
function buildUninstallPlan(payload = {}, configPath, config = {}) {
|
|
const paths = resolveUpdatePaths(configPath, config);
|
|
return {
|
|
configPath: paths.configPath,
|
|
serviceName: String(payload.serviceName || config.serviceName || paths.serviceName || DEFAULT_EDGE_AGENT_SERVICE_NAME),
|
|
restartMode: String(config.restartMode || (process.platform === "win32" ? "spawn" : "systemd")),
|
|
cleanupDelayMs: DEFAULT_UPDATE_RESTART_GRACE_MS,
|
|
};
|
|
}
|
|
|
|
function escapeShellArgument(value) {
|
|
return `'${String(value || "").replace(/'/g, `'\\''`)}'`;
|
|
}
|
|
|
|
async function persistUninstallState(configPath, config, { liveConfig = null } = {}) {
|
|
const currentConfig = cloneJson(config || (await loadConfig(configPath)));
|
|
const nextConfig = {
|
|
...currentConfig,
|
|
gatewayId: null,
|
|
agentToken: null,
|
|
installToken: null,
|
|
brokerUrl: null,
|
|
pendingUpdate: null,
|
|
lastUninstall: {
|
|
state: "SCHEDULED",
|
|
completedAt: formatUpdateTimestamp(),
|
|
},
|
|
};
|
|
|
|
await saveConfig(configPath, nextConfig);
|
|
if (liveConfig) {
|
|
replaceObjectContents(liveConfig, cloneJson(nextConfig));
|
|
}
|
|
|
|
return nextConfig;
|
|
}
|
|
|
|
async function scheduleAgentUninstall(uninstallPlan, {
|
|
spawnImpl = spawnCallback,
|
|
waitImpl = wait,
|
|
exitProcessImpl = (code) => process.exit(code),
|
|
} = {}) {
|
|
if (uninstallPlan.restartMode === "systemd" && process.platform !== "win32") {
|
|
const serviceName = escapeShellArgument(uninstallPlan.serviceName || DEFAULT_EDGE_AGENT_SERVICE_NAME);
|
|
const uninstallProcess = spawnImpl(
|
|
"/bin/sh",
|
|
[
|
|
"-lc",
|
|
`sleep 1; systemctl disable --now ${serviceName} >/dev/null 2>&1 || true; systemctl reset-failed ${serviceName} >/dev/null 2>&1 || true`,
|
|
],
|
|
{
|
|
detached: true,
|
|
stdio: "ignore",
|
|
windowsHide: true,
|
|
}
|
|
);
|
|
uninstallProcess.unref?.();
|
|
}
|
|
|
|
await waitImpl(uninstallPlan.cleanupDelayMs ?? DEFAULT_UPDATE_RESTART_GRACE_MS);
|
|
exitProcessImpl(0);
|
|
}
|
|
|
|
async function startDetachedUpdateVerifier(configPath, config, {
|
|
spawnImpl = spawnCallback,
|
|
} = {}) {
|
|
const paths = resolveUpdatePaths(configPath, config);
|
|
const verifier = spawnImpl(
|
|
process.execPath,
|
|
[paths.agentPath, UPDATE_VERIFY_COMMAND, "--config", paths.configPath],
|
|
{
|
|
detached: true,
|
|
stdio: "ignore",
|
|
windowsHide: true,
|
|
}
|
|
);
|
|
verifier.unref?.();
|
|
return true;
|
|
}
|
|
|
|
async function restartAgentAfterUpdate(restartPlan, {
|
|
spawnImpl = spawnCallback,
|
|
waitImpl = wait,
|
|
exitProcessImpl = (code) => process.exit(code),
|
|
} = {}) {
|
|
if (restartPlan.restartMode === "systemd" && process.platform !== "win32") {
|
|
const restartProcess = spawnImpl("systemctl", ["restart", restartPlan.serviceName], {
|
|
detached: true,
|
|
stdio: "ignore",
|
|
windowsHide: true,
|
|
});
|
|
restartProcess.unref?.();
|
|
} else {
|
|
const nextProcess = spawnImpl(process.execPath, [restartPlan.agentPath, "--config", restartPlan.configPath], {
|
|
detached: true,
|
|
stdio: "ignore",
|
|
windowsHide: true,
|
|
});
|
|
nextProcess.unref?.();
|
|
}
|
|
|
|
await waitImpl(restartPlan.restartGraceMs ?? DEFAULT_UPDATE_RESTART_GRACE_MS);
|
|
exitProcessImpl(0);
|
|
}
|
|
|
|
async function resumeAgentAfterRollback(configPath, config, {
|
|
spawnImpl = spawnCallback,
|
|
} = {}) {
|
|
const restartPlan = buildUpdateRestartPlan({}, configPath, config);
|
|
if (restartPlan.restartMode === "systemd" && process.platform !== "win32") {
|
|
const restartProcess = spawnImpl("systemctl", ["restart", restartPlan.serviceName], {
|
|
detached: true,
|
|
stdio: "ignore",
|
|
windowsHide: true,
|
|
});
|
|
restartProcess.unref?.();
|
|
return;
|
|
}
|
|
|
|
const nextProcess = spawnImpl(process.execPath, [restartPlan.agentPath, "--config", restartPlan.configPath], {
|
|
detached: true,
|
|
stdio: "ignore",
|
|
windowsHide: true,
|
|
});
|
|
nextProcess.unref?.();
|
|
}
|
|
|
|
export async function verifyPendingUpdate(configPath, {
|
|
waitImpl = wait,
|
|
execFileImpl = execFile,
|
|
spawnImpl = spawnCallback,
|
|
verifyIntervalMs = DEFAULT_UPDATE_VERIFY_INTERVAL_MS,
|
|
timeoutMs = null,
|
|
} = {}) {
|
|
const config = await loadConfig(configPath);
|
|
const pendingUpdate = config.pendingUpdate;
|
|
if (!pendingUpdate || typeof pendingUpdate !== "object") {
|
|
return {
|
|
verified: false,
|
|
skipped: true,
|
|
reason: "No pending update verification state",
|
|
};
|
|
}
|
|
|
|
const effectiveTimeoutMs =
|
|
Number.isFinite(timeoutMs) && timeoutMs !== null
|
|
? Math.max(1, Number(timeoutMs))
|
|
: resolveUpdateVerificationTimeoutSeconds(config, pendingUpdate) * 1000;
|
|
const deadline = Date.now() + effectiveTimeoutMs;
|
|
while (Date.now() < deadline) {
|
|
const latestConfig = await loadConfig(configPath).catch(() => null);
|
|
if (latestConfig && !latestConfig.pendingUpdate) {
|
|
return {
|
|
verified: true,
|
|
state: latestConfig.lastUpdate?.state || null,
|
|
};
|
|
}
|
|
|
|
await waitImpl(Math.max(25, verifyIntervalMs));
|
|
}
|
|
|
|
const rollbackReason = `Edge agent update to ${pendingUpdate.targetVersion || "unknown"} did not pass startup verification`;
|
|
const rollbackResult = await rollbackPendingUpdate(configPath, {
|
|
config,
|
|
execFileImpl,
|
|
reason: rollbackReason,
|
|
});
|
|
await resumeAgentAfterRollback(configPath, await loadConfig(configPath), {
|
|
spawnImpl,
|
|
});
|
|
|
|
return {
|
|
verified: false,
|
|
rolledBack: rollbackResult.rolledBack,
|
|
restoredVersion: rollbackResult.restoredVersion || null,
|
|
reason: rollbackReason,
|
|
};
|
|
}
|
|
|
|
export async function runUpdate(payload, fetchImpl = fetch, deps = {}) {
|
|
if (!payload.artifactUrl) {
|
|
return {
|
|
updated: false,
|
|
skipped: true,
|
|
reason: "No artifact URL provided",
|
|
};
|
|
}
|
|
|
|
const configPath = deps.configPath;
|
|
if (!configPath) {
|
|
throw new Error("Missing config path for edge agent update");
|
|
}
|
|
|
|
const liveConfig = deps.liveConfig && typeof deps.liveConfig === "object" ? deps.liveConfig : null;
|
|
const currentConfig = cloneJson(deps.config || liveConfig || (await loadConfig(configPath)));
|
|
const originalConfig = cloneJson(currentConfig);
|
|
const paths = resolveUpdatePaths(configPath, currentConfig);
|
|
const safeTargetVersion = sanitizeUpdateSegment(
|
|
payload.targetVersion || payload.target_version || currentConfig.targetVersion || currentConfig.installedVersion
|
|
);
|
|
const backupDir = path.join(paths.updatesDir, `${Date.now()}-${safeTargetVersion}`);
|
|
|
|
await ensureDirectory(paths.installDir);
|
|
await ensureDirectory(paths.updatesDir);
|
|
|
|
const agentArtifact = await fetchArtifactBuffer(
|
|
payload.artifactUrl,
|
|
payload.sha256 || payload.artifactSha256 || payload.artifact_sha256,
|
|
"Agent artifact",
|
|
fetchImpl
|
|
);
|
|
const packageArtifact = await fetchArtifactBuffer(
|
|
payload.packageUrl || payload.package_url,
|
|
payload.packageSha256 || payload.package_sha256,
|
|
"Package manifest",
|
|
fetchImpl
|
|
);
|
|
|
|
let pendingUpdate = null;
|
|
try {
|
|
await ensureDirectory(backupDir);
|
|
await backupFileIfPresent(paths.agentPath, path.join(backupDir, "agent.mjs"));
|
|
await backupFileIfPresent(paths.packagePath, path.join(backupDir, "package.json"));
|
|
|
|
if (packageArtifact) {
|
|
await writeBuffer(paths.packagePath, packageArtifact.buffer);
|
|
}
|
|
await writeBuffer(paths.agentPath, agentArtifact.buffer);
|
|
|
|
await installAgentDependencies(paths.installDir, deps.execFileImpl || execFile);
|
|
await runAgentStatusPreflight(paths.agentPath, paths.configPath, deps.execFileImpl || execFile);
|
|
|
|
pendingUpdate = buildPendingUpdateState(payload, currentConfig, paths, backupDir);
|
|
const nextConfig = {
|
|
...currentConfig,
|
|
targetVersion: pendingUpdate.targetVersion,
|
|
pendingUpdate,
|
|
lastUpdate: null,
|
|
};
|
|
await saveConfig(paths.configPath, nextConfig);
|
|
if (liveConfig) {
|
|
replaceObjectContents(liveConfig, cloneJson(nextConfig));
|
|
}
|
|
|
|
return {
|
|
__agentCommandEnvelope: true,
|
|
payload: buildPreparedUpdatePayload({
|
|
pendingUpdate,
|
|
agentArtifact,
|
|
packageArtifact,
|
|
paths,
|
|
}),
|
|
followUp: {
|
|
type: "RUN_UPDATE",
|
|
restartPlan: buildUpdateRestartPlan(pendingUpdate, paths.configPath, nextConfig),
|
|
},
|
|
};
|
|
} catch (error) {
|
|
await restorePreparedUpdate(paths, pendingUpdate || buildPendingUpdateState(payload, currentConfig, paths, backupDir), originalConfig, {
|
|
execFileImpl: deps.execFileImpl || execFile,
|
|
liveConfig,
|
|
rollbackMessage: createUpdateErrorMessage(error),
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
export async function runUninstall(payload = {}, deps = {}) {
|
|
const configPath = deps.configPath;
|
|
if (!configPath) {
|
|
throw new Error("Missing config path for edge agent uninstall");
|
|
}
|
|
|
|
const liveConfig = deps.liveConfig && typeof deps.liveConfig === "object" ? deps.liveConfig : null;
|
|
const currentConfig = cloneJson(deps.config || liveConfig || (await loadConfig(configPath)));
|
|
|
|
return {
|
|
__agentCommandEnvelope: true,
|
|
payload: {
|
|
uninstall_scheduled: true,
|
|
scheduled_at: formatUpdateTimestamp(),
|
|
service_name: String(payload.serviceName || currentConfig.serviceName || DEFAULT_EDGE_AGENT_SERVICE_NAME),
|
|
},
|
|
followUp: {
|
|
type: "UNINSTALL_AGENT",
|
|
uninstallPlan: buildUninstallPlan(payload, configPath, currentConfig),
|
|
},
|
|
};
|
|
}
|
|
|
|
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) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
sendMessage({
|
|
type: "SHELL_OUTPUT",
|
|
sessionId,
|
|
data: `Failed to start root shell: ${message}\r\n`,
|
|
});
|
|
sendMessage({ type: "SHELL_EXIT", sessionId, code: 1, reason: "shell_spawn_failed", message });
|
|
}
|
|
};
|
|
|
|
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();
|
|
};
|
|
|
|
const dispose = () => {
|
|
for (const sessionId of sessions.keys()) {
|
|
close({ sessionId });
|
|
}
|
|
};
|
|
|
|
return { open, input, resize, close, dispose };
|
|
}
|
|
|
|
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, deps);
|
|
case "UNINSTALL_AGENT":
|
|
return await runUninstall(command.payload || {}, deps);
|
|
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 lastUpdateMetadata = buildUpdateMetadata(config);
|
|
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 (lastUpdateMetadata) {
|
|
payload.metadata.last_update = lastUpdateMetadata;
|
|
}
|
|
|
|
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 pollShellActionJob(config, fetchImpl = fetch, waitSeconds = 20) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before polling shell actions");
|
|
}
|
|
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/shell-actions/poll`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
wait_seconds: waitSeconds,
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
export async function submitShellActionJobResult(config, jobId, { ok, error = null } = {}, fetchImpl = fetch) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before submitting shell action results");
|
|
}
|
|
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/shell-actions/${jobId}/result`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
ok: Boolean(ok),
|
|
error,
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
export async function submitShellSessionEvents(config, sessionId, events = [], fetchImpl = fetch) {
|
|
if (!config.gatewayId || !config.agentToken) {
|
|
throw new Error("Gateway claim must complete before submitting shell events");
|
|
}
|
|
|
|
return apiRequest(
|
|
config.apiUrl,
|
|
`/edge-agent/gateways/${config.gatewayId}/shell-sessions/${sessionId}/events`,
|
|
"POST",
|
|
{
|
|
agent_token: config.agentToken,
|
|
events,
|
|
},
|
|
fetchImpl
|
|
);
|
|
}
|
|
|
|
async function executeAgentCommandEnvelope(config, command, fetchImpl = fetch, deps = {}) {
|
|
const commandType = command?.commandType || command?.command_type;
|
|
const payload = command?.payload || {};
|
|
|
|
if (!commandType) {
|
|
return null;
|
|
}
|
|
|
|
let followUp = null;
|
|
const result = await handleAgentCommand(
|
|
{ commandType, payload },
|
|
{
|
|
...deps,
|
|
fetchImpl,
|
|
config,
|
|
configPath: payload.configPath || config.configPath || null,
|
|
liveConfig: config,
|
|
}
|
|
);
|
|
|
|
return {
|
|
commandType,
|
|
payload,
|
|
followUp: result && result.__agentCommandEnvelope === true ? (followUp = result.followUp || null, followUp) : null,
|
|
responsePayload: result && result.__agentCommandEnvelope === true ? result.payload || {} : result,
|
|
};
|
|
}
|
|
|
|
async function runAgentCommandFollowUp(
|
|
config,
|
|
execution,
|
|
followUpErrorMessage = "Edge agent restart failed after update",
|
|
deps = {}
|
|
) {
|
|
if (!execution || !execution.followUp?.type) {
|
|
return;
|
|
}
|
|
|
|
if (execution.commandType === "RUN_UPDATE" && execution.followUp.type === "RUN_UPDATE") {
|
|
try {
|
|
await startDetachedUpdateVerifier(config.configPath || execution.payload?.configPath || null, config, {
|
|
spawnImpl: deps.spawnImpl,
|
|
});
|
|
await restartAgentAfterUpdate(execution.followUp.restartPlan, {
|
|
spawnImpl: deps.spawnImpl,
|
|
waitImpl: deps.waitImpl,
|
|
exitProcessImpl: deps.exitProcessImpl,
|
|
});
|
|
} catch (followUpError) {
|
|
await rollbackPendingUpdate(config.configPath || execution.payload?.configPath || null, {
|
|
config,
|
|
liveConfig: config,
|
|
reason: createUpdateErrorMessage(followUpError, followUpErrorMessage),
|
|
}).catch(() => {});
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (execution.commandType === "UNINSTALL_AGENT" && execution.followUp.type === "UNINSTALL_AGENT") {
|
|
await persistUninstallState(config.configPath || execution.followUp.uninstallPlan?.configPath, config, {
|
|
liveConfig: config,
|
|
});
|
|
await scheduleAgentUninstall(execution.followUp.uninstallPlan, {
|
|
spawnImpl: deps.spawnImpl,
|
|
waitImpl: deps.waitImpl,
|
|
exitProcessImpl: deps.exitProcessImpl,
|
|
});
|
|
}
|
|
}
|
|
|
|
export async function processPolledCommand(config, command, fetchImpl = fetch, deps = {}) {
|
|
const jobId = command?.id;
|
|
if (!jobId) {
|
|
return null;
|
|
}
|
|
|
|
try {
|
|
const execution = await executeAgentCommandEnvelope(config, command, fetchImpl, deps);
|
|
if (!execution) {
|
|
return null;
|
|
}
|
|
|
|
await submitCommandJobResult(config, jobId, {
|
|
ok: true,
|
|
payload: execution.responsePayload,
|
|
}, fetchImpl);
|
|
await runAgentCommandFollowUp(config, execution, "Edge agent restart failed after update", deps);
|
|
|
|
return { ok: true, payload: execution.responsePayload };
|
|
} 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 };
|
|
}
|
|
}
|
|
|
|
function normalizeShellEvent(message) {
|
|
const sessionId = message?.sessionId ?? message?.session_id ?? null;
|
|
if (!sessionId) {
|
|
return null;
|
|
}
|
|
|
|
if (message.type === "SHELL_OPENED") {
|
|
return {
|
|
sessionId: String(sessionId),
|
|
event: {
|
|
type: "OPENED",
|
|
payload: {},
|
|
},
|
|
};
|
|
}
|
|
|
|
if (message.type === "SHELL_OUTPUT") {
|
|
return {
|
|
sessionId: String(sessionId),
|
|
event: {
|
|
type: "OUTPUT",
|
|
payload: {
|
|
data: String(message.data || ""),
|
|
},
|
|
},
|
|
};
|
|
}
|
|
|
|
if (message.type === "SHELL_EXIT") {
|
|
const reason = String(message.reason || "agent_exit");
|
|
return {
|
|
sessionId: String(sessionId),
|
|
event: {
|
|
type: "CLOSED",
|
|
payload: {
|
|
code: Number.isFinite(message.code) ? message.code : 0,
|
|
reason,
|
|
message: message.message ? String(message.message) : null,
|
|
},
|
|
},
|
|
};
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function createShellEventPublisher(config, fetchImpl = fetch) {
|
|
let queue = [];
|
|
let flushTimer = null;
|
|
let flushPromise = Promise.resolve();
|
|
|
|
const flush = () => {
|
|
const pending = queue;
|
|
queue = [];
|
|
|
|
if (pending.length === 0) {
|
|
return flushPromise;
|
|
}
|
|
|
|
const grouped = new Map();
|
|
for (const item of pending) {
|
|
const normalized = normalizeShellEvent(item);
|
|
if (!normalized) {
|
|
continue;
|
|
}
|
|
|
|
const events = grouped.get(normalized.sessionId) || [];
|
|
events.push(normalized.event);
|
|
grouped.set(normalized.sessionId, events);
|
|
}
|
|
|
|
flushPromise = flushPromise
|
|
.catch(() => {})
|
|
.then(async () => {
|
|
for (const [sessionId, events] of grouped.entries()) {
|
|
await submitShellSessionEvents(config, sessionId, events, fetchImpl);
|
|
}
|
|
});
|
|
|
|
return flushPromise;
|
|
};
|
|
|
|
const scheduleFlush = () => {
|
|
if (flushTimer !== null) {
|
|
return;
|
|
}
|
|
|
|
flushTimer = setTimeout(() => {
|
|
flushTimer = null;
|
|
flush().catch(() => {});
|
|
}, DEFAULT_SHELL_EVENT_FLUSH_DELAY_MS);
|
|
flushTimer.unref?.();
|
|
};
|
|
|
|
return {
|
|
publish(message) {
|
|
queue.push(message);
|
|
if (message?.type === "SHELL_OPENED" || message?.type === "SHELL_EXIT") {
|
|
if (flushTimer !== null) {
|
|
clearTimeout(flushTimer);
|
|
flushTimer = null;
|
|
}
|
|
flush().catch(() => {});
|
|
return;
|
|
}
|
|
|
|
scheduleFlush();
|
|
},
|
|
async drain() {
|
|
if (flushTimer !== null) {
|
|
clearTimeout(flushTimer);
|
|
flushTimer = null;
|
|
}
|
|
await flush().catch(() => {});
|
|
await flushPromise.catch(() => {});
|
|
},
|
|
};
|
|
}
|
|
|
|
export async function processPolledShellAction(config, action, shell, fetchImpl = fetch) {
|
|
const jobId = action?.id;
|
|
const actionType = action?.actionType || action?.action_type;
|
|
const payload = action?.payload || {};
|
|
|
|
if (!jobId || !actionType) {
|
|
return null;
|
|
}
|
|
|
|
try {
|
|
if (actionType === "OPEN") {
|
|
await shell.open(payload);
|
|
} else if (actionType === "INPUT") {
|
|
shell.input(payload);
|
|
} else if (actionType === "RESIZE") {
|
|
shell.resize(payload);
|
|
} else if (actionType === "CLOSE") {
|
|
shell.close(payload);
|
|
} else {
|
|
throw new Error(`Unsupported shell action: ${actionType}`);
|
|
}
|
|
|
|
await submitShellActionJobResult(config, jobId, {
|
|
ok: true,
|
|
}, fetchImpl);
|
|
return { ok: true };
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
await submitShellActionJobResult(config, jobId, {
|
|
ok: false,
|
|
error: message,
|
|
}, fetchImpl);
|
|
return { ok: false, error: message };
|
|
}
|
|
}
|
|
|
|
function createBrokerBridge({
|
|
config,
|
|
shell,
|
|
fetchImpl = fetch,
|
|
reconnectDelayMs = DEFAULT_BROKER_RECONNECT_DELAY_MS,
|
|
} = {}) {
|
|
const state = {
|
|
url: config?.brokerUrl || null,
|
|
connected: false,
|
|
lastError: null,
|
|
disconnectReason: null,
|
|
lastConnectedAt: null,
|
|
lastDisconnectedAt: null,
|
|
lastCloseCode: null,
|
|
lastCloseClean: null,
|
|
};
|
|
|
|
let stopped = false;
|
|
let reconnectTimer = null;
|
|
let socket = null;
|
|
|
|
const clearReconnectTimer = () => {
|
|
if (reconnectTimer) {
|
|
clearTimeout(reconnectTimer);
|
|
reconnectTimer = null;
|
|
}
|
|
};
|
|
|
|
const send = (message) => {
|
|
if (!socket || socket.readyState !== WebSocket.OPEN) {
|
|
return false;
|
|
}
|
|
|
|
socket.send(JSON.stringify(message));
|
|
return true;
|
|
};
|
|
|
|
const scheduleReconnect = () => {
|
|
if (stopped || reconnectTimer || !state.url) {
|
|
return;
|
|
}
|
|
|
|
reconnectTimer = setTimeout(() => {
|
|
reconnectTimer = null;
|
|
connect();
|
|
}, Math.max(250, reconnectDelayMs));
|
|
};
|
|
|
|
const connect = () => {
|
|
if (stopped) {
|
|
return;
|
|
}
|
|
|
|
const socketUrl = buildBrokerSocketUrl(state.url, config?.gatewayId, config?.agentToken);
|
|
if (!socketUrl) {
|
|
return;
|
|
}
|
|
|
|
try {
|
|
socket = new WebSocket(socketUrl);
|
|
} catch (error) {
|
|
state.connected = false;
|
|
state.lastError = error instanceof Error ? error.message : String(error);
|
|
state.lastDisconnectedAt = formatUpdateTimestamp();
|
|
scheduleReconnect();
|
|
return;
|
|
}
|
|
|
|
socket.onopen = () => {
|
|
state.connected = true;
|
|
state.lastError = null;
|
|
state.disconnectReason = null;
|
|
state.lastCloseCode = null;
|
|
state.lastCloseClean = null;
|
|
state.lastConnectedAt = formatUpdateTimestamp();
|
|
};
|
|
|
|
socket.onmessage = async (event) => {
|
|
try {
|
|
const raw = await readSocketMessageText(event.data);
|
|
const message = JSON.parse(raw);
|
|
|
|
if (message.type === "COMMAND") {
|
|
try {
|
|
const execution = await executeAgentCommandEnvelope(config, message, fetchImpl);
|
|
if (!execution) {
|
|
return;
|
|
}
|
|
|
|
send({
|
|
type: "COMMAND_RESULT",
|
|
commandId: message.commandId,
|
|
ok: true,
|
|
payload: execution.responsePayload,
|
|
});
|
|
await runAgentCommandFollowUp(config, execution);
|
|
} catch (error) {
|
|
send({
|
|
type: "COMMAND_RESULT",
|
|
commandId: message.commandId,
|
|
ok: false,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
});
|
|
}
|
|
return;
|
|
}
|
|
|
|
if (message.type === "OPEN_ROOT_SHELL") {
|
|
await shell.open(message.payload || {});
|
|
return;
|
|
}
|
|
if (message.type === "SHELL_INPUT") {
|
|
shell.input(message.payload || {});
|
|
return;
|
|
}
|
|
if (message.type === "RESIZE_ROOT_SHELL") {
|
|
shell.resize(message.payload || {});
|
|
return;
|
|
}
|
|
if (message.type === "CLOSE_ROOT_SHELL") {
|
|
shell.close(message.payload || {});
|
|
}
|
|
} catch {
|
|
// Ignore malformed or unsupported broker messages so polling remains authoritative.
|
|
}
|
|
};
|
|
|
|
socket.onerror = (error) => {
|
|
state.lastError = normalizeBrokerSocketError(error, socketUrl);
|
|
};
|
|
|
|
socket.onclose = (event) => {
|
|
socket = null;
|
|
state.connected = false;
|
|
state.disconnectReason = event.reason || "broker_disconnected";
|
|
state.lastCloseCode = Number.isFinite(Number(event.code)) ? Number(event.code) : null;
|
|
state.lastCloseClean = typeof event.wasClean === "boolean" ? event.wasClean : null;
|
|
if (state.lastCloseCode && state.lastCloseCode !== 1000 && !state.lastError) {
|
|
state.lastError = `Broker websocket closed with code ${state.lastCloseCode}`;
|
|
}
|
|
state.lastDisconnectedAt = formatUpdateTimestamp();
|
|
if (!stopped) {
|
|
scheduleReconnect();
|
|
}
|
|
};
|
|
};
|
|
|
|
return {
|
|
state,
|
|
start() {
|
|
connect();
|
|
},
|
|
stop() {
|
|
stopped = true;
|
|
clearReconnectTimer();
|
|
if (socket && socket.readyState <= WebSocket.OPEN) {
|
|
socket.close();
|
|
}
|
|
socket = null;
|
|
state.connected = false;
|
|
},
|
|
send,
|
|
};
|
|
}
|
|
|
|
export async function startAgent({
|
|
configPath,
|
|
fetchImpl = fetch,
|
|
collectMetricsImpl = collectSystemMetrics,
|
|
createShellBridgeImpl = createShellBridge,
|
|
} = {}) {
|
|
if (!configPath) {
|
|
throw new Error("Missing --config path");
|
|
}
|
|
|
|
let config = await loadConfig(configPath);
|
|
config.configPath = configPath;
|
|
config = await claimIfNeeded(config, configPath, fetchImpl);
|
|
config.configPath = configPath;
|
|
config = await finalizePendingUpdateOnStartup(config, configPath, {
|
|
liveConfig: config,
|
|
});
|
|
config.configPath = configPath;
|
|
const intervalMs = Number(config.heartbeatIntervalSeconds || 15) * 1000;
|
|
const commandPollTimeoutSeconds = Number(config.commandPollTimeoutSeconds || 20);
|
|
const commandPollRetryDelayMs = Number(config.commandPollRetryDelayMs || 1000);
|
|
const brokerReconnectDelayMs = Number(config.brokerReconnectDelayMs || DEFAULT_BROKER_RECONNECT_DELAY_MS);
|
|
const shellActionPollTimeoutSeconds = Number(config.shellActionPollTimeoutSeconds || 20);
|
|
const shellActionPollRetryDelayMs = Number(config.shellActionPollRetryDelayMs || 1000);
|
|
|
|
let stopped = false;
|
|
let cpuSnapshot = null;
|
|
let lastHeartbeatLatencyMs = null;
|
|
let brokerBridge = null;
|
|
const shellEventPublisher = createShellEventPublisher(config, fetchImpl);
|
|
const shell = createShellBridgeImpl((message) => {
|
|
brokerBridge?.send(message);
|
|
shellEventPublisher.publish(message);
|
|
});
|
|
brokerBridge = createBrokerBridge({
|
|
config,
|
|
shell,
|
|
fetchImpl,
|
|
reconnectDelayMs: brokerReconnectDelayMs,
|
|
});
|
|
brokerBridge.start();
|
|
|
|
const sendTransportHeartbeat = async (extra = {}) => {
|
|
const transportState = buildTransportHeartbeatState(brokerBridge?.state || { url: config.brokerUrl || null });
|
|
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 = {
|
|
...transportState,
|
|
...extra,
|
|
metadata: {
|
|
...(transportState.metadata || {}),
|
|
...extraMetadataFields,
|
|
system_metrics: metricPayload,
|
|
},
|
|
};
|
|
|
|
const response = await sendTimedHeartbeat(config, fetchImpl, heartbeatPayload);
|
|
lastHeartbeatLatencyMs = response.latencyMs;
|
|
return response.data;
|
|
};
|
|
|
|
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 runShellActionPollLoop = async () => {
|
|
while (!stopped) {
|
|
try {
|
|
const action = await pollShellActionJob(config, fetchImpl, shellActionPollTimeoutSeconds);
|
|
if (stopped) {
|
|
break;
|
|
}
|
|
|
|
if (!action) {
|
|
continue;
|
|
}
|
|
|
|
await processPolledShellAction(config, action, shell, fetchImpl);
|
|
} catch {
|
|
if (stopped) {
|
|
break;
|
|
}
|
|
|
|
await new Promise((resolve) => setTimeout(resolve, shellActionPollRetryDelayMs));
|
|
}
|
|
}
|
|
};
|
|
|
|
await sendTransportHeartbeat();
|
|
const commandPollPromise = runCommandPollLoop();
|
|
const shellActionPollPromise = runShellActionPollLoop();
|
|
|
|
const timer = setInterval(() => {
|
|
sendTransportHeartbeat().catch(() => {});
|
|
}, intervalMs);
|
|
timer.unref?.();
|
|
|
|
const stop = async () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
brokerBridge?.stop();
|
|
await shellEventPublisher.drain();
|
|
shell.dispose();
|
|
await Promise.allSettled([commandPollPromise, shellActionPollPromise]);
|
|
};
|
|
|
|
return {
|
|
brokerBridge,
|
|
commandPollPromise,
|
|
shellActionPollPromise,
|
|
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 === UPDATE_VERIFY_COMMAND && command === "start") {
|
|
command = UPDATE_VERIFY_COMMAND;
|
|
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;
|
|
}
|
|
if (command === UPDATE_VERIFY_COMMAND) {
|
|
return verifyPendingUpdate(configPath);
|
|
}
|
|
|
|
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;
|
|
});
|
|
}
|