collabSupport/tests/dectRelayHub.test.js

// Integration tests for services/dectRelayHub.js.
//
// These spin up a real HTTP server on an ephemeral port, attach the
// hub, and connect a real `ws` client that plays the role of the
// dect-relay-agent. This gives us end-to-end coverage of the auth
// path, the wire protocol, RPC correlation, timeouts, and clean
// disconnect handling — none of which we can meaningfully test with
// pure mocks.
//
// Every test creates its own hub + server so they can run in parallel
// without port conflicts. All servers are torn down in the test's
// finally block so a failing test can't leak file descriptors.

import test from 'node:test';
import assert from 'node:assert/strict';
import http from 'node:http';
import WebSocket from 'ws';

import { DectRelayHub, RelayErrorCodes, DectRelayError } from '../services/dectRelayHub.js';

const TEST_TOKEN = 'super-secret-token-for-tests';

// ─── Test harness ───────────────────────────────────────────────────

/**
 * Spin up an HTTP server with the hub attached on an ephemeral port.
 * Returns { hub, port, closeAll }. Caller MUST call closeAll() (in a
 * try/finally) to release the port + socket handles.
 */
async function makeHub(token = TEST_TOKEN) {
  const server = http.createServer((req, res) => {
    res.writeHead(404); res.end();
  });
  const hub = new DectRelayHub({ token });
  hub.attachTo(server);
  await new Promise((res) => server.listen(0, '127.0.0.1', res));
  const { port } = server.address();
  return {
    hub,
    port,
    async closeAll() {
      await hub.close();
      await new Promise((res) => server.close(() => res()));
    },
  };
}

/**
 * Open a WebSocket client to the hub. Optional bearer overrides the
 * default token — useful for the "wrong token" test.
 */
function connectAgent(port, { bearer = TEST_TOKEN, useProtocol = false } = {}) {
  const url = `ws://127.0.0.1:${port}/dect-relay/ws`;
  const opts = useProtocol
    ? { headers: {}, protocol: `bearer.${bearer}` }
    : { headers: { Authorization: `Bearer ${bearer}` } };
  return new WebSocket(url, opts.protocol ? opts.protocol : undefined, {
    headers: opts.headers,
    handshakeTimeout: 3000,
  });
}

function waitOpen(ws) {
  return new Promise((resolve, reject) => {
    ws.once('open', resolve);
    ws.once('error', reject);
  });
}

function waitClose(ws) {
  return new Promise((resolve) => ws.once('close', (code, reason) => resolve({ code, reason: reason?.toString() || '' })));
}

// A tiny agent that immediately replies to every command with the
// given handler. Handler receives the parsed inbound message and
// returns either { ok:true, result:{...} } or throws.
function attachAutoAgent(ws, handler) {
  ws.on('message', async (raw) => {
    const msg = JSON.parse(raw.toString('utf8'));
    if (msg.type === 'ping') { ws.send(JSON.stringify({ type: 'pong', at: Date.now() })); return; }
    if (!msg.id) return;
    try {
      const result = await handler(msg);
      ws.send(JSON.stringify({ id: msg.id, ok: true, result, elapsedMs: 1 }));
    } catch (err) {
      ws.send(JSON.stringify({
        id: msg.id, ok: false,
        error: { code: err.code || 'AUTO_AGENT_ERR', message: err.message },
      }));
    }
  });
}

// ─── isConnected / status ───────────────────────────────────────────

test('hub: isConnected is false with no agent', async () => {
  const { hub, closeAll } = await makeHub();
  try {
    assert.equal(hub.isConnected(), false);
    assert.equal(hub.status().connected, false);
    assert.equal(hub.status().agent, null);
    assert.equal(hub.status().inFlight, 0);
  } finally {
    await closeAll();
  }
});

test('hub: rpc without connection rejects with NOT_CONNECTED', async () => {
  const { hub, closeAll } = await makeHub();
  try {
    await assert.rejects(
      hub.collect('10.0.0.100'),
      (err) => err instanceof DectRelayError && err.code === RelayErrorCodes.NOT_CONNECTED,
    );
  } finally {
    await closeAll();
  }
});

// ─── Auth ──────────────────────────────────────────────────────────

test('hub: rejects upgrade with no bearer', async () => {
  const { port, closeAll } = await makeHub();
  try {
    const ws = new WebSocket(`ws://127.0.0.1:${port}/dect-relay/ws`, {
      handshakeTimeout: 3000,
    });
    // Server writes a raw 401 before the WS handshake completes.
    // ws throws 'Unexpected server response: 401' as an error.
    await assert.rejects(waitOpen(ws), /401/);
  } finally {
    await closeAll();
  }
});

test('hub: rejects upgrade with wrong bearer', async () => {
  const { port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port, { bearer: 'wrong-token' });
    await assert.rejects(waitOpen(ws), /401/);
  } finally {
    await closeAll();
  }
});

test('hub: accepts upgrade with correct bearer via Authorization header', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    // Give the hub a tick to record the adoption.
    await new Promise((r) => setImmediate(r));
    assert.equal(hub.isConnected(), true);
    ws.close();
    await waitClose(ws);
  } finally {
    await closeAll();
  }
});

test('hub: accepts upgrade via Sec-WebSocket-Protocol fallback', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port, { useProtocol: true });
    await waitOpen(ws);
    await new Promise((r) => setImmediate(r));
    assert.equal(hub.isConnected(), true);
    ws.close();
    await waitClose(ws);
  } finally {
    await closeAll();
  }
});

// ─── Hello frame ────────────────────────────────────────────────────

test('hub: records agent hello frame into status()', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    ws.send(JSON.stringify({
      type: 'hello', agentVersion: '9.9.9', hostname: 'test-host',
      capabilities: ['collect', 'reboot'],
    }));
    // Wait until hub processes it (message events are queued).
    await new Promise((r) => setTimeout(r, 20));
    const s = hub.status();
    assert.equal(s.connected, true);
    assert.equal(s.agent.agentVersion, '9.9.9');
    assert.equal(s.agent.hostname, 'test-host');
    assert.deepEqual(s.agent.capabilities, ['collect', 'reboot']);
    ws.close();
    await waitClose(ws);
  } finally {
    await closeAll();
  }
});

// ─── RPC correlation ────────────────────────────────────────────────

test('hub: RPC round-trip resolves with agent result', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    attachAutoAgent(ws, async (msg) => {
      assert.equal(msg.type, 'collect');
      assert.equal(msg.baseIp, '10.4.11.87');
      return { parsed: { device: { macAddress: 'aa:bb:cc:dd:ee:ff' } }, verdict: { healthy: true } };
    });
    const { result } = await hub.collect('10.4.11.87');
    assert.equal(result.parsed.device.macAddress, 'aa:bb:cc:dd:ee:ff');
    assert.equal(result.verdict.healthy, true);
    ws.close();
    await waitClose(ws);
  } finally {
    await closeAll();
  }
});

test('hub: RPC error from agent surfaces as DectRelayError with code', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    attachAutoAgent(ws, async () => {
      const err = new Error('base rejected credentials');
      err.code = 'DIGEST_401';
      throw err;
    });
    await assert.rejects(
      hub.collect('10.4.11.87'),
      (err) => err instanceof DectRelayError && err.code === 'DIGEST_401'
        && /base rejected credentials/i.test(err.message),
    );
  } finally {
    await closeAll();
  }
});

test('hub: multiple concurrent RPCs correlate by id, not order', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    // Delay short IPs less than long IPs, deliberately reversing
    // response order relative to send order.
    ws.on('message', (raw) => {
      const msg = JSON.parse(raw.toString('utf8'));
      if (!msg.id) return;
      const delay = msg.baseIp === '10.0.0.1' ? 40 : 5;
      setTimeout(() => {
        ws.send(JSON.stringify({
          id: msg.id, ok: true, result: { echo: msg.baseIp }, elapsedMs: delay,
        }));
      }, delay);
    });
    const [a, b] = await Promise.all([
      hub.collect('10.0.0.1'),  // slower
      hub.collect('10.0.0.2'),  // faster
    ]);
    assert.equal(a.result.echo, '10.0.0.1');
    assert.equal(b.result.echo, '10.0.0.2');
  } finally {
    await closeAll();
  }
});

// ─── Timeout ────────────────────────────────────────────────────────

test('hub: RPC that never gets a reply times out with TIMEOUT code', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    // Silent agent: acknowledge nothing.
    ws.on('message', () => { /* intentionally do nothing */ });
    await assert.rejects(
      hub.collect('10.0.0.1', { timeoutMs: 50 }),
      (err) => err instanceof DectRelayError && err.code === RelayErrorCodes.TIMEOUT,
    );
  } finally {
    await closeAll();
  }
});

// ─── Mid-flight disconnect ──────────────────────────────────────────

test('hub: agent disconnect mid-RPC rejects the pending promise', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    // Close the socket the moment we receive a command.
    ws.on('message', () => ws.close(1000, 'test'));
    await assert.rejects(
      hub.collect('10.0.0.1', { timeoutMs: 2000 }),
      (err) => err instanceof DectRelayError && err.code === RelayErrorCodes.DISCONNECTED,
    );
  } finally {
    await closeAll();
  }
});

// ─── Second agent replaces first ────────────────────────────────────

test('hub: second agent connection replaces the first (with clean close)', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const wsA = connectAgent(port);
    await waitOpen(wsA);
    const closedA = waitClose(wsA);

    const wsB = connectAgent(port);
    await waitOpen(wsB);

    // wsA should have been closed by the hub with reason "replaced".
    const closeInfo = await closedA;
    assert.equal(closeInfo.code, 1000);
    assert.match(closeInfo.reason, /replaced/i);

    // The hub is still connected — to wsB now.
    assert.equal(hub.isConnected(), true);
    wsB.close();
    await waitClose(wsB);
  } finally {
    await closeAll();
  }
});

// ─── execAction routing ────────────────────────────────────────────

test('hub: execAction routes action name into type field', async () => {
  const { hub, port, closeAll } = await makeHub();
  try {
    const ws = connectAgent(port);
    await waitOpen(ws);
    let observed = null;
    attachAutoAgent(ws, async (msg) => {
      observed = msg;
      return { ok: 'done' };
    });
    await hub.execAction('10.0.0.1', 'reboot', { forced: false });
    assert.equal(observed.type, 'reboot');
    assert.equal(observed.baseIp, '10.0.0.1');
    assert.equal(observed.forced, false);
  } finally {
    await closeAll();
  }
});