collabSupport/services/dectRelayHub.js

// src/services/dectRelayHub.js
//
// Bot-side of the DECT relay: a WebSocket server that accepts ONE
// long-lived connection from a data-center-resident relay agent, plus
// a promise-based RPC API for the rest of the bot to call ("collect
// this base's status", "reboot this base"). The agent — which lives
// in `dect-relay-agent/` in this repo — makes the actual HTTPS Digest
// calls to DBS-210 base stations on the private 10.x network.
//
// Why a WebSocket at all: the bot runs in the public cloud and can't
// reach 10.x. The agent runs in the DC and can, but the DC can't
// accept unsolicited inbound connections. WSS solves both sides: the
// agent dials outbound to the bot (traversing NAT / proxy just like
// any HTTPS request), and once the socket is up the bot can push
// commands whenever it wants.
//
// Only ONE agent is expected to connect. If a second agent dials in,
// we assume it's a legitimate restart (agent redeployed, network
// blip, etc.), close the old socket, and replace it with the new one.
// This is safe because the RPC pending-map is drained + rejected on
// disconnect — any in-flight command reports back "relay disconnected"
// rather than silently hanging.
//
// Auth is a static bearer token shared between bot .env and agent
// .env. That's fine for a single trusted agent — WSS gives us
// transport-level confidentiality, and rotating the token is a
// two-line env change. If we ever need multiple agents we'd swap
// this for per-agent tokens plus an agent-id → base-list registry.

import { WebSocketServer } from 'ws';
import { randomUUID, timingSafeEqual } from 'node:crypto';

import { logger } from '../utils/logger.js';

const LOG_SCOPE = 'dect:relay-hub';

const DEFAULTS = {
  path: '/dect-relay/ws',
  rpcTimeoutMs: 15_000,   // per-command default; callers can override
  heartbeatIntervalMs: 30_000,
  heartbeatIdleTimeoutMs: 90_000,  // treat socket as dead if no pong in this long
};

// Error codes surfaced back to callers via rejected RPC promises.
// Keeping them string-typed (not numeric) so log lines stay readable.
export const RelayErrorCodes = Object.freeze({
  NOT_CONNECTED:  'RELAY_NOT_CONNECTED',
  DISCONNECTED:   'RELAY_DISCONNECTED_MID_RPC',
  TIMEOUT:        'RELAY_RPC_TIMEOUT',
  AGENT_ERROR:    'RELAY_AGENT_ERROR',    // agent returned {ok:false, error:{...}}
  MALFORMED:      'RELAY_MALFORMED_REPLY',
});

/**
 * Structured error thrown by RPC calls. Carrying a code + optional
 * detail lets callers branch on it (e.g. render "relay offline"
 * differently from "base returned 401") without regex-matching on
 * .message strings.
 */
export class DectRelayError extends Error {
  constructor(code, message, detail = null) {
    super(message);
    this.name = 'DectRelayError';
    this.code = code;
    this.detail = detail;
  }
}

/**
 * The hub itself. Not a singleton class — the module exports one
 * default instance below and that's what the bot uses. Keeping it
 * class-shaped anyway so tests can spin up an isolated hub with an
 * ephemeral port and its own token.
 */
export class DectRelayHub {
  constructor({ token, path = DEFAULTS.path } = {}) {
    if (!token || typeof token !== 'string') {
      throw new Error('DectRelayHub: token is required');
    }
    this._token = Buffer.from(token, 'utf8');
    this._path = path;
    this._socket = null;
    this._hello = null;          // last hello frame from the agent
    this._pending = new Map();   // cmdId → { resolve, reject, timer }
    this._lastPongAt = 0;
    this._heartbeatTimer = null;
    this._wss = null;
  }

  /** True when there's a live agent socket we can command. */
  isConnected() {
    return !!(this._socket && this._socket.readyState === 1 /* OPEN */);
  }

  /**
   * Snapshot of the current connection state. Safe to expose over a
   * health endpoint or /dectstatus admin page — no secrets in here.
   */
  status() {
    return {
      connected: this.isConnected(),
      agent: this._hello ? { ...this._hello } : null,
      inFlight: this._pending.size,
      lastPongMsAgo: this._lastPongAt ? Date.now() - this._lastPongAt : null,
    };
  }

  /**
   * Attach the WebSocket upgrade handler to a Node http.Server. Must
   * be called during startup, AFTER app.listen() returns the http
   * server. Express doesn't upgrade sockets itself, so we hook the
   * 'upgrade' event manually and route just our path — anything else
   * (e.g. a future webhook that needs its own upgrade) can add its
   * own listener without conflict.
   *
   * @param {import('node:http').Server} httpServer
   */
  attachTo(httpServer) {
    if (this._wss) throw new Error('DectRelayHub: already attached');

    // noServer:true → we do the upgrade dance manually so we can
    // enforce auth BEFORE ws does its handshake. Otherwise ws would
    // 101 first and then we'd have to close, which is uglier + wastes
    // a round-trip on every unauthorized probe.
    this._wss = new WebSocketServer({ noServer: true });

    httpServer.on('upgrade', (req, socket, head) => {
      // Only handle our path; leave others alone so future upgrades
      // don't collide.
      const url = req.url || '';
      // Match with or without a trailing slash / query string.
      const cleanPath = url.split('?')[0].replace(/\/$/, '');
      if (cleanPath !== this._path.replace(/\/$/, '')) return;

      if (!this._checkAuth(req)) {
        logger(LOG_SCOPE, `Unauthorized upgrade attempt from ${req.socket.remoteAddress}`, 'warn');
        socket.write('HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\n\r\n');
        socket.destroy();
        return;
      }

      this._wss.handleUpgrade(req, socket, head, (ws) => this._adoptAgent(ws, req));
    });

    logger(LOG_SCOPE, `WebSocket upgrade handler attached at ${this._path}`);
  }

  /**
   * Constant-time bearer-token comparison. The bearer must be sent
   * as `Authorization: Bearer <token>` on the WSS upgrade request.
   * We also accept `Sec-WebSocket-Protocol: bearer.<token>` as a
   * fallback because some proxies strip Authorization headers on
   * upgrade requests — this hides the token in a spec-compliant
   * subprotocol string that isn't logged by most access logs.
   */
  _checkAuth(req) {
    const presented = extractBearer(req);
    if (!presented) return false;
    const buf = Buffer.from(presented, 'utf8');
    if (buf.length !== this._token.length) return false;
    try {
      return timingSafeEqual(buf, this._token);
    } catch {
      return false;
    }
  }

  _adoptAgent(ws, req) {
    const from = req.socket.remoteAddress || 'unknown';

    // Replace any existing socket: we only want ONE agent at a time.
    if (this._socket) {
      logger(LOG_SCOPE, `Replacing existing agent socket with new connection from ${from}`, 'warn');
      try { this._socket.close(1000, 'replaced by newer agent'); } catch { /* ignore */ }
      this._socket = null;
      this._hello = null;
    }

    this._socket = ws;
    this._lastPongAt = Date.now();
    logger(LOG_SCOPE, `Agent connected from ${from}`);

    // Every handler closes over `ws` so we can distinguish the socket
    // that fired the event from `this._socket` — critical when a
    // second agent replaces the first: the replaced socket's async
    // 'close' event will fire AFTER we've swapped in the new socket,
    // and without this guard it would wipe out the replacement.
    ws.on('message', (raw) => this._onMessage(ws, raw));
    ws.on('close', (code, reasonBuf) => this._onSocketClose(ws, code, reasonBuf?.toString?.() || ''));
    ws.on('error', (err) => logger(LOG_SCOPE, `Agent socket error: ${err.message}`, 'error'));
    ws.on('pong', () => { if (ws === this._socket) this._lastPongAt = Date.now(); });

    this._startHeartbeat();
  }

  _startHeartbeat() {
    this._stopHeartbeat();
    this._heartbeatTimer = setInterval(() => {
      if (!this.isConnected()) return;
      // If we haven't seen a pong in too long, assume dead. Node's ws
      // client won't detect a wedged TCP for many minutes; this
      // heartbeat is how we recover in under 90s.
      if (Date.now() - this._lastPongAt > DEFAULTS.heartbeatIdleTimeoutMs) {
        logger(LOG_SCOPE, 'No pong in idle timeout — terminating agent socket', 'warn');
        try { this._socket.terminate(); } catch { /* ignore */ }
        return;
      }
      try { this._socket.ping(); } catch { /* ignore, will surface via 'error' */ }
    }, DEFAULTS.heartbeatIntervalMs);
    // Don't hold the event loop open on process exit.
    if (this._heartbeatTimer.unref) this._heartbeatTimer.unref();
  }

  _stopHeartbeat() {
    if (this._heartbeatTimer) {
      clearInterval(this._heartbeatTimer);
      this._heartbeatTimer = null;
    }
  }

  _onSocketClose(sourceSocket, code, reason) {
    // If this event is for a socket that's already been replaced by
    // a newer connection, log at debug and skip the state reset —
    // otherwise we'd wipe out the replacement socket we just adopted.
    if (this._socket && this._socket !== sourceSocket) {
      logger(LOG_SCOPE, `Ignoring close from replaced socket (code=${code})`, 'debug');
      return;
    }
    logger(LOG_SCOPE, `Agent disconnected (code=${code}, reason="${reason}")`);
    this._socket = null;
    this._hello = null;
    this._stopHeartbeat();
    // Reject any in-flight RPCs so callers don't hang forever.
    for (const [, entry] of this._pending) {
      clearTimeout(entry.timer);
      entry.reject(new DectRelayError(
        RelayErrorCodes.DISCONNECTED,
        'DECT relay disconnected while command was in flight',
      ));
    }
    this._pending.clear();
  }

  _onMessage(sourceSocket, raw) {
    // Drop late messages from a replaced socket (see _onSocketClose).
    if (sourceSocket !== this._socket) return;
    let msg;
    try {
      msg = JSON.parse(raw.toString('utf8'));
    } catch {
      logger(LOG_SCOPE, `Ignoring non-JSON frame from agent (${raw.length} bytes)`, 'warn');
      return;
    }
    if (!msg || typeof msg !== 'object') {
      logger(LOG_SCOPE, 'Ignoring non-object frame from agent', 'warn');
      return;
    }

    if (msg.type === 'hello') {
      this._hello = {
        agentVersion: msg.agentVersion || 'unknown',
        hostname: msg.hostname || null,
        capabilities: Array.isArray(msg.capabilities) ? msg.capabilities.slice() : [],
        receivedAt: new Date().toISOString(),
      };
      logger(LOG_SCOPE, `Agent hello: version=${this._hello.agentVersion} host=${this._hello.hostname} caps=${this._hello.capabilities.join(',')}`);
      return;
    }
    if (msg.type === 'ping') {
      // Explicit JSON-level ping (in addition to the ws-level ping
      // frames). Reply with an equivalent pong so a symmetric agent
      // can verify liveness without relying on the ws framing.
      this._sendRaw({ type: 'pong', at: Date.now() });
      return;
    }
    if (msg.type === 'pong') {
      this._lastPongAt = Date.now();
      return;
    }

    // Otherwise it's an RPC reply for a pending command.
    if (!msg.id) {
      logger(LOG_SCOPE, `Ignoring frame with no id and unknown type ${msg.type}`, 'warn');
      return;
    }
    const entry = this._pending.get(msg.id);
    if (!entry) {
      // Late reply after timeout — safe to drop.
      logger(LOG_SCOPE, `Reply for unknown/expired cmd id ${msg.id} — dropping`, 'debug');
      return;
    }
    this._pending.delete(msg.id);
    clearTimeout(entry.timer);

    if (msg.ok === true) {
      entry.resolve({ result: msg.result, elapsedMs: msg.elapsedMs || null });
    } else if (msg.ok === false) {
      const err = msg.error || {};
      entry.reject(new DectRelayError(
        err.code || RelayErrorCodes.AGENT_ERROR,
        err.message || 'Agent reported an error',
        err,
      ));
    } else {
      entry.reject(new DectRelayError(
        RelayErrorCodes.MALFORMED,
        `Agent reply missing ok field for id ${msg.id}`,
        msg,
      ));
    }
  }

  _sendRaw(obj) {
    if (!this.isConnected()) return false;
    try {
      this._socket.send(JSON.stringify(obj));
      return true;
    } catch (err) {
      logger(LOG_SCOPE, `Failed to send frame: ${err.message}`, 'warn');
      return false;
    }
  }

  /**
   * Send an RPC command to the agent. Returns { result, elapsedMs }
   * on success, throws DectRelayError on failure. Timeouts and
   * disconnects are surfaced as rejections — callers should always
   * try/catch or use .catch().
   *
   * @param {object} payload            command frame WITHOUT id (added here)
   * @param {object} [opts]
   * @param {number} [opts.timeoutMs]   per-call override; default 15s
   */
  rpc(payload, { timeoutMs = DEFAULTS.rpcTimeoutMs } = {}) {
    return new Promise((resolve, reject) => {
      if (!this.isConnected()) {
        reject(new DectRelayError(
          RelayErrorCodes.NOT_CONNECTED,
          'DECT relay is not connected — data-center agent may be offline',
        ));
        return;
      }
      const id = `cmd_${randomUUID()}`;
      const timer = setTimeout(() => {
        this._pending.delete(id);
        reject(new DectRelayError(
          RelayErrorCodes.TIMEOUT,
          `DECT relay RPC ${payload.type || '(no type)'} timed out after ${timeoutMs}ms`,
        ));
      }, timeoutMs);
      // Do NOT unref this timer — we want the process to stay alive
      // until every in-flight RPC has resolved or timed out.

      this._pending.set(id, { resolve, reject, timer });

      const ok = this._sendRaw({ id, ...payload });
      if (!ok) {
        this._pending.delete(id);
        clearTimeout(timer);
        reject(new DectRelayError(
          RelayErrorCodes.NOT_CONNECTED,
          'Failed to send frame (socket may have just closed)',
        ));
      }
    });
  }

  /** Convenience: fetch parsed status.xml for a given base IP. */
  collect(baseIp, opts) {
    return this.rpc({ type: 'collect', baseIp }, opts);
  }

  /**
   * Convenience: execute one of the mutating actions the agent
   * exposes (reboot / force-reboot / reboot-chain / force-reboot-chain
   * / factory-reset / reconfigure-tree). The agent is the audit
   * boundary for these — the bot's own audit sink STILL records the
   * intent (see commands/dectStatus.js in Phase 2), but the agent
   * logs the actual HTTP call.
   */
  execAction(baseIp, action, extra = {}, opts) {
    return this.rpc({ type: action, baseIp, ...extra }, opts);
  }

  /**
   * Shut everything down. Called from graceful-shutdown paths.
   * Safe to call when nothing is attached.
   *
   * Terminates every socket the WSS layer is still tracking rather
   * than relying on graceful close — a straggling client (e.g. an
   * agent whose TCP is wedged after a NAT reboot) would otherwise
   * block `wss.close()`'s callback and hang the shutdown path.
   */
  async close() {
    this._stopHeartbeat();
    if (this._socket) {
      try { this._socket.close(1001, 'bot shutting down'); } catch { /* ignore */ }
      this._socket = null;
    }
    if (this._wss) {
      // Force-close any client the wss is still tracking. Without
      // this, wss.close() waits indefinitely for all clients to
      // disconnect on their own — fine in the happy path, but tests
      // and NAT wedges both cause hangs.
      for (const client of this._wss.clients) {
        try { client.terminate(); } catch { /* ignore */ }
      }
      await new Promise((res) => this._wss.close(() => res()));
      this._wss = null;
    }
    for (const [, entry] of this._pending) {
      clearTimeout(entry.timer);
      entry.reject(new DectRelayError(
        RelayErrorCodes.DISCONNECTED,
        'Bot shutting down',
      ));
    }
    this._pending.clear();
  }
}

// ─── Helpers ────────────────────────────────────────────────────────

/**
 * Pull the bearer token out of the upgrade request. Order tried:
 *   1. `Authorization: Bearer <token>` header (canonical).
 *   2. `Sec-WebSocket-Protocol: bearer.<token>` (proxy-friendly).
 * Returns null if neither is present or well-formed.
 */
function extractBearer(req) {
  const auth = req.headers['authorization'];
  if (typeof auth === 'string') {
    const m = auth.match(/^Bearer\s+(\S+)\s*$/i);
    if (m) return m[1];
  }
  const proto = req.headers['sec-websocket-protocol'];
  if (typeof proto === 'string') {
    for (const part of proto.split(',')) {
      const trimmed = part.trim();
      if (trimmed.startsWith('bearer.')) return trimmed.slice('bearer.'.length);
    }
  }
  return null;
}

// ─── Default singleton for the bot to use ───────────────────────────

let _defaultHub = null;

/**
 * Return the process-wide DectRelayHub, constructing it on first
 * access using env config. Throws if DECT_RELAY_AGENT_TOKEN is not
 * set — surfaces the missing config at startup rather than silently
 * being non-functional.
 */
export function getDectRelayHub() {
  if (_defaultHub) return _defaultHub;
  const token = process.env.DECT_RELAY_AGENT_TOKEN;
  if (!token) {
    throw new Error(
      'DECT_RELAY_AGENT_TOKEN is not set. Add it to your .env and share ' +
      'the same value with the dect-relay-agent. Until then, DECT commands ' +
      'will fail with RELAY_NOT_CONNECTED.',
    );
  }
  _defaultHub = new DectRelayHub({ token, path: process.env.DECT_RELAY_PATH || DEFAULTS.path });
  return _defaultHub;
}

// Test-only: reset the singleton. Not exported from an index barrel;
// only imported by unit tests that need isolation.
export function _resetDectRelayHubForTests() { _defaultHub = null; }