relay-exporter/test/live.offchain.test.ts

import test from "node:test";
import assert from "node:assert/strict";
import { spawn, type ChildProcessByStdio } from "node:child_process";
import http from "node:http";
import type { Readable } from "node:stream";
import { setTimeout as sleep } from "node:timers/promises";

const ENABLED = process.env.LIVE_RELAY_TEST_OFFCHAIN === "1";
const SCRAPE_SAMPLES = Number.parseInt(process.env.LIVE_RELAY_TEST_SAMPLES ?? "12", 10);
const SCRAPE_EVERY_MS = Number.parseInt(process.env.LIVE_RELAY_TEST_SCRAPE_EVERY_MS ?? "1500", 10);
const PROBE_TIMEOUT_SECONDS = Number.parseInt(process.env.LIVE_RELAY_TEST_TIMEOUT_SECONDS ?? "8", 10);
const WRITE_VERIFY_READ = process.env.LIVE_RELAY_TEST_WRITE_VERIFY_READ === "1";
const ENFORCE_NO_WRITE_CONFIRM_FAILURES = process.env.LIVE_RELAY_TEST_EXPECT_NO_FAILURES === "1";

function randomPort(): number {
  return 20_000 + Math.floor(Math.random() * 10_000);
}

function httpGet(
  port: number,
  path: string,
): Promise<{ statusCode: number; body: string; headers: http.IncomingHttpHeaders }> {
  return new Promise((resolve, reject) => {
    const req = http.request(
      {
        hostname: "127.0.0.1",
        port,
        path,
        method: "GET",
        timeout: 5_000,
      },
      (res) => {
        const chunks: Buffer[] = [];
        res.on("data", (chunk) => {
          chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
        });
        res.on("end", () => {
          resolve({
            statusCode: res.statusCode ?? 0,
            body: Buffer.concat(chunks).toString("utf8"),
            headers: res.headers,
          });
        });
      },
    );
    req.on("timeout", () => {
      req.destroy(new Error(`timeout GET ${path}`));
    });
    req.on("error", reject);
    req.end();
  });
}

type ExporterChild = ChildProcessByStdio<null, Readable, Readable>;

async function waitForStartup(child: ExporterChild, port: number, timeoutMs: number): Promise<void> {
  const started = Date.now();
  while (Date.now() - started < timeoutMs) {
    if (child.exitCode !== null) {
      throw new Error(`exporter exited early with code ${child.exitCode}`);
    }
    try {
      const res = await httpGet(port, "/healthz");
      if ([200, 503].includes(res.statusCode)) return;
    } catch {
      // server not up yet
    }
    await sleep(150);
  }
  throw new Error("timeout waiting for exporter startup");
}

function metricValue(metrics: string, name: string, labels: Record<string, string>): number | null {
  const entries = Object.entries(labels);
  const labelsPattern = entries
    .map(([k, v]) => `${k}="${v.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}"`)
    .join(",");
  const linePattern = new RegExp(`^${name}\\{${labelsPattern}\\} (-?\\d+(?:\\.\\d+)?)$`, "m");
  const match = metrics.match(linePattern);
  if (!match?.[1]) return null;
  const parsed = Number.parseFloat(match[1]);
  return Number.isFinite(parsed) ? parsed : null;
}

test(
  "live exporter probe diagnostic against offchain.pub",
  { skip: !ENABLED },
  async (t) => {
    const relayInput = process.env.LIVE_RELAY_TEST_RELAYS ?? "wss://offchain.pub";
    const relays = relayInput
      .split(",")
      .map((v) => v.trim())
      .filter((v) => v.length > 0);
    assert.ok(relays.length > 0, "LIVE_RELAY_TEST_RELAYS must include at least one relay");

    let relayLabel: string;
    try {
      relayLabel = new URL(relays[0]).toString();
    } catch {
      assert.fail(`invalid first relay URL in LIVE_RELAY_TEST_RELAYS: ${relays[0]}`);
    }

    const port = randomPort();
    const child = spawn("pnpm", ["exec", "tsx", "src/index.ts"], {
      cwd: process.cwd(),
      env: {
        ...process.env,
        RELAYS: relays.join(","),
        PROBE_INTERVAL_SECONDS: "1",
        PROBE_TIMEOUT_SECONDS: String(PROBE_TIMEOUT_SECONDS),
        WRITE_CHECK_ENABLED: "true",
        WRITE_CHECK_VERIFY_READ: WRITE_VERIFY_READ ? "true" : "false",
        PORT: String(port),
        LOG_LEVEL: "error",
      },
      stdio: ["ignore", "pipe", "pipe"],
    });

    let stderr = "";
    child.stderr.on("data", (d: Buffer) => {
      stderr += d.toString("utf8");
    });

    t.after(() => {
      if (child.exitCode === null) {
        child.kill("SIGTERM");
      }
    });

    await waitForStartup(child, port, 10_000);

    let lastMetrics = "";
    for (let i = 0; i < SCRAPE_SAMPLES; i += 1) {
      const metrics = await httpGet(port, "/metrics");
      assert.equal(metrics.statusCode, 200);
      lastMetrics = metrics.body;
      await sleep(SCRAPE_EVERY_MS);
    }

    const writeConfirmOk = metricValue(lastMetrics, "nostr_relay_write_confirm_ok", { relay: relayLabel });
    const writeConfirmErrors = metricValue(lastMetrics, "nostr_relay_probe_errors_total", {
      relay: relayLabel,
      check: "write_confirm",
    });
    const successRuns = metricValue(lastMetrics, "nostr_relay_probe_runs_total", {
      relay: relayLabel,
      result: "success",
    });
    const failureRuns = metricValue(lastMetrics, "nostr_relay_probe_runs_total", {
      relay: relayLabel,
      result: "failure",
    });

    const diagnostic = {
      relay: relayLabel,
      relays,
      scrapeSamples: SCRAPE_SAMPLES,
      scrapeEveryMs: SCRAPE_EVERY_MS,
      writeVerifyRead: WRITE_VERIFY_READ,
      writeConfirmOk,
      writeConfirmErrors,
      successRuns,
      failureRuns,
      crashed: child.exitCode !== null,
    };
    console.log("[live-offchain-exporter]", JSON.stringify(diagnostic));

    assert.equal(child.exitCode, null, `exporter crashed unexpectedly: ${stderr}`);
    assert.notEqual(writeConfirmOk, null, "missing nostr_relay_write_confirm_ok metric");
    assert.ok(
      successRuns !== null || failureRuns !== null,
      "missing nostr_relay_probe_runs_total metrics",
    );

    if (ENFORCE_NO_WRITE_CONFIRM_FAILURES) {
      assert.equal(writeConfirmErrors ?? 0, 0, `write_confirm failures seen: ${JSON.stringify(diagnostic)}`);
      assert.equal(failureRuns ?? 0, 0, `probe failures seen: ${JSON.stringify(diagnostic)}`);
    }
  },
);