validations-proxy/test/relay.test.js

import assert from 'node:assert/strict';
import { once } from 'node:events';
import { mkdirSync, mkdtempSync, readFileSync, rmSync } from 'node:fs';
import { createServer } from 'node:http';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import test from 'node:test';
import { WebSocket, WebSocketServer } from 'ws';
import { createRelay, nodePublicKeyFromHex } from '../src/relay.js';

async function upstreamServer() {
  const server = new WebSocketServer({ port: 0, host: '127.0.0.1' });
  await once(server, 'listening');
  const { port } = server.address();
  return { server, url: `ws://127.0.0.1:${port}` };
}

async function stopWsServer(server) {
  for (const client of server.clients) client.terminate();
  await new Promise((resolve) => server.close(resolve));
}

async function waitFor(predicate, timeoutMs = 1000) {
  const deadline = Date.now() + timeoutMs;
  while (!predicate()) {
    if (Date.now() >= deadline) throw new Error('Timed out waiting for condition');
    await new Promise((resolve) => setTimeout(resolve, 5));
  }
}

test('serves the live dashboard at the root URL', async (t) => {
  const upstream = await upstreamServer();
  const relay = createRelay({ upstreamUrl: upstream.url, validatorListUrl: null, versionStorePath: null, logger: {} });
  const address = await relay.start({ port: 0, host: '127.0.0.1' });
  t.after(async () => {
    await relay.stop();
    await stopWsServer(upstream.server);
  });

  const response = await fetch(`http://127.0.0.1:${address.port}/`);
  assert.equal(response.status, 200);
  assert.match(response.headers.get('content-type'), /text\/html/);
  const body = await response.text();
  assert.match(body, /Validation pulse/);
  assert.match(body, /Live consensus/);
  assert.match(body, /Next flag ledger/);
  assert.match(body, /Offline VL validators/);
  assert.match(body, /Master key/);
  assert.match(body, /Domain/);
  assert.match(body, /Version/);
});

test('sends manifest domains only as dashboard metadata', async (t) => {
  const upstream = await upstreamServer();
  const relay = createRelay({ upstreamUrl: upstream.url, validatorListUrl: null, versionStorePath: null, logger: {} });
  const address = await relay.start({ port: 0, host: '127.0.0.1' });
  t.after(async () => {
    await relay.stop();
    await stopWsServer(upstream.server);
  });

  const upstreamConnection = await once(upstream.server, 'connection').then(([ws]) => ws);
  await once(upstreamConnection, 'message');
  const dashboard = new WebSocket(`ws://127.0.0.1:${address.port}/?dashboard=1`);
  await once(dashboard, 'open');

  const validation = {
    type: 'validationReceived',
    ledger_index: '789',
    data: 'signed-validation-domain',
    master_key: 'nMasterWithDomain',
  };
  const manifestMessage = once(upstreamConnection, 'message');
  upstreamConnection.send(JSON.stringify(validation));
  const [rawMessage] = await once(dashboard, 'message');
  assert.equal(rawMessage.toString(), JSON.stringify(validation));

  const [manifestData] = await manifestMessage;
  const manifestRequest = JSON.parse(manifestData);
  assert.equal(manifestRequest.command, 'manifest');
  assert.equal(manifestRequest.public_key, 'nMasterWithDomain');

  upstreamConnection.send(JSON.stringify({
    id: manifestRequest.id,
    type: 'response',
    result: { details: { domain: 'validator.example' } },
  }));
  const [metadataData] = await once(dashboard, 'message');
  const metadata = JSON.parse(metadataData);
  assert.equal(metadata.type, 'validatorMetadata');
  assert.equal(metadata.master_key, 'nMasterWithDomain');
  assert.equal(metadata.domain, 'validator.example');
});

test('subscribes upstream and relays only validation messages', async (t) => {
  const upstream = await upstreamServer();
  const relay = createRelay({ upstreamUrl: upstream.url, validatorListUrl: null, versionStorePath: null, logger: {} });
  const address = await relay.start({ port: 0, host: '127.0.0.1' });
  t.after(async () => {
    await relay.stop();
    await stopWsServer(upstream.server);
  });

  const upstreamConnection = await once(upstream.server, 'connection').then(([ws]) => ws);
  const subscription = JSON.parse((await once(upstreamConnection, 'message'))[0]);
  assert.deepEqual(subscription.streams, ['validations']);

  const client = new WebSocket(`ws://127.0.0.1:${address.port}/`);
  await once(client, 'open');

  upstreamConnection.send(JSON.stringify({ type: 'response', status: 'success' }));
  const validation = {
    type: 'validationReceived',
    ledger_index: '123',
    signature: 'abc',
    validation_public_key: 'nSigning',
    master_key: 'nMaster',
  };
  upstreamConnection.send(JSON.stringify(validation));

  const [message] = await once(client, 'message');
  assert.equal(message.toString(), JSON.stringify(validation));
  assert.equal(relay.state.validationsRelayed, 1);
});

test('collates two upstreams, removes duplicates, and survives one disconnect', async (t) => {
  const first = await upstreamServer();
  const second = await upstreamServer();
  const firstConnected = once(first.server, 'connection');
  const secondConnected = once(second.server, 'connection');
  const relay = createRelay({
    upstreamUrls: [first.url, second.url],
    validatorListUrl: null,
    versionStorePath: null,
    reconnectMinMs: 10_000,
    logger: {},
  });
  const address = await relay.start({ port: 0, host: '127.0.0.1' });
  t.after(async () => {
    await relay.stop();
    await stopWsServer(first.server);
    await stopWsServer(second.server);
  });

  const firstConnection = (await firstConnected)[0];
  const secondConnection = (await secondConnected)[0];
  await Promise.all([once(firstConnection, 'message'), once(secondConnection, 'message')]);

  const client = new WebSocket(`ws://127.0.0.1:${address.port}/`);
  await once(client, 'open');

  const duplicate = {
    type: 'validationReceived',
    ledger_index: '200',
    data: 'signed-validation-one',
  };
  firstConnection.send(JSON.stringify(duplicate));
  const [firstMessage] = await once(client, 'message');
  assert.equal(firstMessage.toString(), JSON.stringify(duplicate));
  secondConnection.send(JSON.stringify({ ...duplicate, master_key: 'nMaster' }));

  await waitFor(() => relay.state.duplicatesDropped === 1);
  assert.equal(relay.state.validationsRelayed, 1);
  assert.equal(relay.state.duplicatesDropped, 1);

  firstConnection.terminate();
  await once(firstConnection, 'close');
  const unique = {
    type: 'validationReceived',
    ledger_index: '201',
    data: 'signed-validation-two',
  };
  secondConnection.send(JSON.stringify(unique));
  const [secondMessage] = await once(client, 'message');
  assert.equal(secondMessage.toString(), JSON.stringify(unique));
  assert.equal(relay.state.upstreamConnected, true);
  assert.equal(relay.state.validationsRelayed, 2);
});

test('rejects downstream clients without the configured bearer token', async (t) => {
  const upstream = await upstreamServer();
  const relay = createRelay({ upstreamUrl: upstream.url, validatorListUrl: null, versionStorePath: null, authToken: 'secret', logger: {} });
  const address = await relay.start({ port: 0, host: '127.0.0.1' });
  t.after(async () => {
    await relay.stop();
    await stopWsServer(upstream.server);
  });

  const client = new WebSocket(`ws://127.0.0.1:${address.port}/`);
  const [error] = await once(client, 'error');
  assert.match(error.message, /401/);
});

test('converts validator-list hex keys to Xahau master keys', () => {
  assert.equal(
    nodePublicKeyFromHex('ED02E3102D348B688CCDAF2D40FA9549E7C25EE926A15880E40D5BB0E2168A14DE'),
    'nHB45nBNgjKMssrRqaNVr2tpCq3t55J5APRRDD6ov1U41JfVFjr6',
  );
});

test('publishes listed validators before they send a validation', async (t) => {
  const publicKey = 'ED02E3102D348B688CCDAF2D40FA9549E7C25EE926A15880E40D5BB0E2168A14DE';
  const masterKey = nodePublicKeyFromHex(publicKey);
  const validatorList = createServer((req, res) => {
    const blob = Buffer.from(JSON.stringify({ validators: [{ validation_public_key: publicKey }] })).toString('base64');
    res.writeHead(200, { 'content-type': 'application/json' });
    res.end(JSON.stringify({ blob }));
  });
  validatorList.listen(0, '127.0.0.1');
  await once(validatorList, 'listening');
  const listPort = validatorList.address().port;
  const upstream = await upstreamServer();
  const upstreamConnected = once(upstream.server, 'connection');
  const relay = createRelay({
    upstreamUrl: upstream.url,
    validatorListUrl: `http://127.0.0.1:${listPort}/`,
    versionStorePath: null,
    logger: {},
  });
  const address = await relay.start({ port: 0, host: '127.0.0.1' });
  const upstreamConnection = (await upstreamConnected)[0];
  await once(upstreamConnection, 'message');
  t.after(async () => {
    await relay.stop();
    await stopWsServer(upstream.server);
    await new Promise((resolve) => validatorList.close(resolve));
  });

  const dashboard = new WebSocket(`ws://127.0.0.1:${address.port}/?dashboard=1`);
  const metadataPromise = once(dashboard, 'message');
  await once(dashboard, 'open');
  const metadata = JSON.parse((await metadataPromise)[0]);
  assert.equal(metadata.type, 'validatorMetadata');
  assert.equal(metadata.master_key, masterKey);
  assert.equal(metadata.listed, true);
  assert.equal(metadata.offline, true);
  assert.equal(metadata.last_seen, undefined);
  assert.ok(Date.parse(metadata.monitoring_since));

  const onlineMetadata = new Promise((resolve) => {
    const listener = (data) => {
      const message = JSON.parse(data);
      if (message.type === 'validatorMetadata' && message.master_key === masterKey && message.offline === false) {
        dashboard.off('message', listener);
        resolve(message);
      }
    };
    dashboard.on('message', listener);
  });
  upstreamConnection.send(JSON.stringify({
    type: 'validationReceived',
    master_key: masterKey,
    ledger_index: '901',
    data: 'listed-validator-is-online',
  }));
  const online = await onlineMetadata;
  assert.equal(online.last_seen.length > 0, true);
});

test('persists and reloads validator versions', async (t) => {
  const dataDirectory = fileURLToPath(new URL('../data', import.meta.url));
  mkdirSync(dataDirectory, { recursive: true });
  const directory = mkdtempSync(join(dataDirectory, 'test-'));
  const storePath = join(directory, 'versions.json');
  const upstream = await upstreamServer();
  t.after(async () => {
    await stopWsServer(upstream.server);
    rmSync(directory, { recursive: true, force: true });
  });

  const firstRelay = createRelay({
    upstreamUrl: upstream.url,
    validatorListUrl: null,
    versionStorePath: storePath,
    logger: {},
  });
  await firstRelay.start({ port: 0, host: '127.0.0.1' });
  const firstUpstream = await once(upstream.server, 'connection').then(([ws]) => ws);
  await once(firstUpstream, 'message');
  firstUpstream.send(JSON.stringify({
    type: 'validationReceived',
    master_key: 'nPersistedMaster',
    ledger_index: '900',
    server_version: '1745991531128425009',
  }));
  await waitFor(() => firstRelay.state.validationsRelayed === 1);
  await firstRelay.stop();

  const stored = JSON.parse(readFileSync(storePath, 'utf8'));
  assert.equal(stored.versions.nPersistedMaster, '1745991531128425009');

  const secondRelay = createRelay({
    upstreamUrl: upstream.url,
    validatorListUrl: null,
    versionStorePath: storePath,
    logger: {},
  });
  const address = await secondRelay.start({ port: 0, host: '127.0.0.1' });
  const dashboard = new WebSocket(`ws://127.0.0.1:${address.port}/?dashboard=1`);
  const metadataPromise = once(dashboard, 'message');
  await once(dashboard, 'open');
  const [metadataData] = await metadataPromise;
  const metadata = JSON.parse(metadataData);
  assert.equal(metadata.master_key, 'nPersistedMaster');
  assert.equal(metadata.server_version, '1745991531128425009');
  await secondRelay.stop();
});