collabSupport/dect-relay-agent/index.js

#!/usr/bin/env node
// dect-relay-agent/index.js
//
// Data-center-resident WebSocket client that bridges the CollabSupport
// bot (running in the public cloud) to Cisco DBS-210 DECT base
// stations on the private 10.x/8 corporate network.
//
// Runtime shape:
//   1. On startup, dial `wss://<bot>/dect-relay/ws` with the shared
//      bearer token from DECT_RELAY_AGENT_TOKEN.
//   2. Send a `hello` frame declaring version + hostname + supported
//      command types.
//   3. Loop waiting for command frames from the bot. Dispatch each
//      into the (already-tested) integrations/cisco-dect/ modules
//      shared with the bot's own spike CLI (scripts/testDectBase.js).
//   4. Reply with `{id, ok, result|error, elapsedMs}` per command.
//   5. On disconnect, reconnect with exponential backoff. Restart the
//      cycle from step 2 (a fresh hello) so the bot's registry is
//      always in sync with the agent's actual capabilities.
//
// This agent does NOT store DECT credentials in transit — the bearer
// token is per-agent. DBS-210 admin creds live only in THIS process's
// .env and never leave the DC.

import 'dotenv/config';
import WebSocket from 'ws';
import os from 'node:os';
import { createDectClient } from '../integrations/cisco-dect/client.js';
import {
  triggerReboot,
  triggerRebootChain,
  triggerFactoryReset,
  triggerReconfigureDectTree,
} from '../integrations/cisco-dect/probes.js';
import {
  parseStatusXml,
  summarizeBaseHealth,
} from '../integrations/cisco-dect/statusXml.js';

// ─── Config ─────────────────────────────────────────────────────────

const CFG = {
  botUrl:     process.env.DECT_RELAY_BOT_URL,
  token:      process.env.DECT_RELAY_AGENT_TOKEN,
  hostname:   process.env.DECT_RELAY_AGENT_HOSTNAME || os.hostname(),
  dectUser:   process.env.DECT_ADMIN_USER || 'admin',
  dectPass:   process.env.DECT_ADMIN_PASSWORD,
  dectTimeout: Number(process.env.DECT_ADMIN_TIMEOUT_MS) || 30_000,
  reconnectMaxMs: Number(process.env.DECT_RELAY_RECONNECT_MAX_MS) || 30_000,
};

const AGENT_VERSION = '0.1.0';
const CAPABILITIES = [
  'collect',
  'reboot',
  'force-reboot',
  'reboot-chain',
  'force-reboot-chain',
  'factory-reset',
  'reconfigure-tree',
];

const HEARTBEAT_INTERVAL_MS = 30_000;

// ─── Logging (dependency-free; agent runs standalone) ───────────────

function log(scope, msg, level = 'info') {
  const ts = new Date().toISOString();
  const line = `[${ts}] [${level.toUpperCase()}] [${scope}] ${msg}`;
  if (level === 'error' || level === 'warn') console.error(line);
  else console.log(line);
}

// ─── Startup validation ─────────────────────────────────────────────

function assertConfig() {
  const missing = [];
  if (!CFG.botUrl)   missing.push('DECT_RELAY_BOT_URL');
  if (!CFG.token)    missing.push('DECT_RELAY_AGENT_TOKEN');
  if (!CFG.dectPass) missing.push('DECT_ADMIN_PASSWORD');
  if (missing.length > 0) {
    console.error(`Missing required env: ${missing.join(', ')}. See .env.example.`);
    process.exit(1);
  }
  if (!CFG.botUrl.startsWith('wss://') && !CFG.botUrl.startsWith('ws://')) {
    console.error(`DECT_RELAY_BOT_URL must start with wss:// (or ws:// for local dev). Got: ${CFG.botUrl}`);
    process.exit(1);
  }
  if (CFG.botUrl.startsWith('ws://') && !/(^|\.)localhost/.test(CFG.botUrl) && !/127\.0\.0\.1/.test(CFG.botUrl)) {
    log('startup', `⚠️ DECT_RELAY_BOT_URL is plain ws:// against a non-local host — bearer token would be sent in cleartext`, 'warn');
  }
}

// ─── Reconnect loop ─────────────────────────────────────────────────

let currentWs = null;
let heartbeatTimer = null;
let reconnectAttempt = 0;
let shuttingDown = false;

function scheduleReconnect() {
  if (shuttingDown) return;
  reconnectAttempt += 1;
  // Exponential backoff with jitter: 1s, 2s, 4s, 8s… capped at
  // reconnectMaxMs, plus 0–1000ms jitter to de-sync herds when
  // multiple agents restart at once (future-proofing — today there's
  // only one).
  const base = Math.min(1000 * 2 ** (reconnectAttempt - 1), CFG.reconnectMaxMs);
  const jitter = Math.floor(Math.random() * 1000);
  const delay = base + jitter;
  log('reconnect', `Attempt #${reconnectAttempt} in ${delay}ms`);
  setTimeout(connect, delay);
}

function stopHeartbeat() {
  if (heartbeatTimer) {
    clearInterval(heartbeatTimer);
    heartbeatTimer = null;
  }
}

function connect() {
  if (shuttingDown) return;
  log('connect', `Dialing ${CFG.botUrl}`);

  const ws = new WebSocket(CFG.botUrl, {
    // Preferred auth path: standard Authorization header. Some
    // reverse proxies strip it on WS upgrades; the bot accepts the
    // Sec-WebSocket-Protocol fallback too, but header is cleaner.
    headers: { Authorization: `Bearer ${CFG.token}` },
    // Handshake grace period. Bot's WSS layer should accept
    // instantly, but corporate proxies can be slow.
    handshakeTimeout: 15_000,
  });
  currentWs = ws;

  ws.on('open', () => {
    reconnectAttempt = 0;
    log('connect', 'Connected — sending hello');
    send({
      type: 'hello',
      agentVersion: AGENT_VERSION,
      hostname: CFG.hostname,
      capabilities: CAPABILITIES,
    });
    startHeartbeat();
  });

  ws.on('message', (raw) => handleMessage(raw));

  ws.on('close', (code, reason) => {
    log('connect', `Socket closed (code=${code} reason="${reason.toString()}")`);
    stopHeartbeat();
    currentWs = null;
    scheduleReconnect();
  });

  ws.on('error', (err) => {
    // 'error' can fire BEFORE 'close' on handshake failures (401,
    // TLS problems, DNS). Log it and let 'close' handle reconnection.
    log('connect', `Socket error: ${err.message}`, 'warn');
  });
}

function startHeartbeat() {
  stopHeartbeat();
  heartbeatTimer = setInterval(() => {
    if (currentWs && currentWs.readyState === WebSocket.OPEN) {
      // JSON-level ping — bot replies with `{type:'pong', at:...}`.
      // We also let the underlying ws library exchange its own
      // ping/pong frames; belt-and-suspenders because some proxies
      // strip WS control frames.
      send({ type: 'ping', at: Date.now() });
      try { currentWs.ping(); } catch { /* ignore */ }
    }
  }, HEARTBEAT_INTERVAL_MS);
  if (heartbeatTimer.unref) heartbeatTimer.unref();
}

function send(obj) {
  if (!currentWs || currentWs.readyState !== WebSocket.OPEN) return false;
  try {
    currentWs.send(JSON.stringify(obj));
    return true;
  } catch (err) {
    log('send', `Failed to send frame: ${err.message}`, 'warn');
    return false;
  }
}

// ─── Command dispatch ───────────────────────────────────────────────

async function handleMessage(raw) {
  let msg;
  try {
    msg = JSON.parse(raw.toString('utf8'));
  } catch {
    log('dispatch', `Ignoring non-JSON frame (${raw.length} bytes)`, 'warn');
    return;
  }
  if (!msg || typeof msg !== 'object') return;

  // Server-initiated JSON ping — reply with pong (also refreshes the
  // bot-side lastPongAt timestamp).
  if (msg.type === 'ping') {
    send({ type: 'pong', at: Date.now() });
    return;
  }
  if (msg.type === 'pong') return; // no-op; we just want to see it come back

  // Everything else must have an id and a command type.
  if (!msg.id) {
    log('dispatch', `Frame missing id: ${JSON.stringify(msg).slice(0, 120)}`, 'warn');
    return;
  }
  if (!msg.type) {
    replyError(msg.id, 'MALFORMED', 'command frame missing `type`');
    return;
  }
  if (!msg.baseIp) {
    replyError(msg.id, 'MALFORMED', 'command frame missing `baseIp`');
    return;
  }

  const started = Date.now();
  try {
    const result = await dispatch(msg);
    replyOk(msg.id, result, Date.now() - started);
  } catch (err) {
    const code = err?.code || 'AGENT_EXCEPTION';
    log('dispatch', `Command ${msg.type} for ${msg.baseIp} failed: ${err.message}`, 'warn');
    replyError(msg.id, code, err.message, { stack: err.stack?.split('\n')[0] });
  }
}

/**
 * Route one command to the right helper. Every branch returns a
 * plain JS object that will be JSON-serialized as the `result`
 * field of the reply frame.
 */
async function dispatch(cmd) {
  const client = createDectClient({
    host: cmd.baseIp,
    user: CFG.dectUser,
    password: CFG.dectPass,
    timeoutMs: CFG.dectTimeout,
  });

  switch (cmd.type) {
    case 'collect': {
      const started = Date.now();
      const resp = await client.get('/admin/status.xml');
      const elapsedMs = Date.now() - started;
      if (resp.status !== 200 || typeof resp.data !== 'string' || !resp.data.trim()) {
        const err = new Error(`base returned status=${resp.status} (${resp.data?.length || 0} bytes)`);
        err.code = 'BASE_BAD_STATUS';
        throw err;
      }
      const parsed = parseStatusXml(resp.data);
      const verdict = summarizeBaseHealth(parsed);
      return { parsed, verdict, fetchedInMs: elapsedMs };
    }

    // All mutating actions are one-shot GETs (see integrations/cisco-
    // dect/probes.js). They come back as either `{dryRun:true,...}`
    // (which we never pass here — dryRun is always false from the
    // bot) or `{dryRun:false, kind, planned, result}`. We surface
    // `planned` + `result` so the bot can log the CSRF'd URL and
    // whether the base responded 200.
    case 'reboot':               return await triggerReboot(client,        { forced: false, dryRun: false });
    case 'force-reboot':         return await triggerReboot(client,        { forced: true,  dryRun: false });
    case 'reboot-chain':         return await triggerRebootChain(client,   { forced: false, dryRun: false });
    case 'force-reboot-chain':   return await triggerRebootChain(client,   { forced: true,  dryRun: false });
    case 'factory-reset':        return await triggerFactoryReset(client,  { dryRun: false });
    case 'reconfigure-tree':     return await triggerReconfigureDectTree(client, { dryRun: false });

    default: {
      const err = new Error(`unknown command type: ${cmd.type}`);
      err.code = 'UNKNOWN_COMMAND';
      throw err;
    }
  }
}

function replyOk(id, result, elapsedMs) {
  send({ id, ok: true, result, elapsedMs });
}

function replyError(id, code, message, detail = null) {
  send({ id, ok: false, error: { code, message, ...(detail || {}) } });
}

// ─── Signal handling ────────────────────────────────────────────────

async function shutdown(reason) {
  if (shuttingDown) return;
  shuttingDown = true;
  log('shutdown', `Shutting down — ${reason}`);
  stopHeartbeat();
  if (currentWs) {
    try { currentWs.close(1000, 'agent shutdown'); } catch { /* ignore */ }
  }
  // Give the close frame a moment to flush before exit. 500ms is
  // enough for local + LAN cases and doesn't meaningfully delay
  // container restarts.
  setTimeout(() => process.exit(0), 500);
}

process.on('SIGINT',  () => shutdown('SIGINT'));
process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('uncaughtException', (err) => {
  log('uncaught', `${err.message}\n${err.stack}`, 'error');
  shutdown('uncaughtException');
});
process.on('unhandledRejection', (reason) => {
  const msg = reason instanceof Error ? `${reason.message}\n${reason.stack}` : String(reason);
  log('uncaught', msg, 'error');
  shutdown('unhandledRejection');
});

// ─── Start ──────────────────────────────────────────────────────────

assertConfig();
log('startup', `dect-relay-agent v${AGENT_VERSION} — hostname=${CFG.hostname}, bot=${CFG.botUrl}`);
connect();