/**
 * GENESIS KEEPS ITS BUS through the broker step (novox/hq ADR 0333 §5, issue 590).
 *
 * The installer raises the bus from its bundle and, at its broker step (step 18 of 24), adopts it as the
 * catalogue's `nats` module. Until ADR 0333 the bundle raised the bus as `mesh-broker` on docker volumes,
 * and the module keeps it as `mesh-broker-nats` on directories of the machine: adopting it would have raised
 * a second bus beside the first, so the installer refused there and genesis stopped. ADR 0333 has the bundle
 * raise the bus where the module keeps it, so the adoption replaces the server's image once and keeps
 * everything else.
 *
 * The bed runs the installer from bare twice. The first run is given a catalogue without the `nats` module,
 * so it stops at the broker step by name, before anything is registered — the moment just before the step,
 * with everything genesis wrote on the bus by then. The bed reads the bus there: its containers, and every
 * stream holding anything. The second run, given the whole catalogue, finds the steps before done and goes
 * on through the broker step. The bed reads the bus again, and passes only when (ADR 0333 §5):
 *
 *   - exactly one bus container runs, and it is mesh-broker-nats, on the module's image now;
 *   - every stream that held something before the step is still there, the same stream, as long or longer;
 *   - the mesh has heard from the first node since.
 *
 * The judgement is test/integration/bus-kept.ts, held without a lab by test/bus-kept.test.ts.
 *
 * **A bed that could not reach the step says so and does not fail**: a first run that stopped earlier than
 * the broker step — a pull the internet refused, a build that broke — proves nothing about the step, and the
 * test is skipped with the reason. A second run that stops AT the broker step is the fault, and fails. One
 * that stops after it has crossed it: what it stopped on is reported, and the bus is judged.
 *
 *   MESH_LAB_HOST_BINARY=<mesh-host>/mesh-host        MESH_LAB_BOOTSTRAP_BINARY=<mesh-host>/mesh-bootstrap
 *   MESH_LAB_BUNDLE=<mesh-host>/examples/foundation-first-node.lock
 *   MESH_LAB_CATALOG=<mesh-catalog>/modules
 *   MESH_LAB_REPOS=<a directory holding mesh-controller, mesh-tools, mesh-sdk and mesh-catalog checkouts>
 *     — served by the forge machine at each checkout's HEAD; the catalogue's is the one built from
 *   MESH_LAB_KEEP=1   leave it standing
 */
import { test, before, after } from "node:test";
import assert from "node:assert/strict";
import { cpSync, existsSync, mkdtempSync, rmSync } from "node:fs";
import { execFileSync } from "node:child_process";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { loadScenario } from "../../src/declaration/parse.ts";
import { raise } from "../../src/lifecycle/raise.ts";
import { destroy, exec, push } from "../../src/lifecycle/operate.ts";
import { bootstrapBinaryPath, hostBinaryPath } from "../../src/lifecycle/place.ts";
import { labIsUsable, destroyAll, foundationBundle } from "./harness.ts";
import { genesis, type GenesisResult } from "./genesis.ts";
import { BUS, busContainers, busNotKept, runningFrom, streamsFrom, written, type Seen } from "./bus-kept.ts";

const SCENARIO = "genesis-keeps-its-bus";
const ANCHOR = "anchor";
const FORGE = "forge";
const FORGE_ADDRESS = "192.0.2.30";
const REGISTRY = "192.0.2.10:5000";
/** The repositories the installer clones, served by the forge at each checkout's HEAD. */
const REPOS = ["mesh-controller", "mesh-tools", "mesh-sdk", "mesh-catalog"] as const;
/** The step this bed is about, as the installer names it. */
const BROKER = "broker";
/** The installer's steps before it, in its order: a first run stopped at one of these never reached it. */
const BEFORE_BROKER = ["preflight", "load", "build", "bundle", "apply", "verify", "enrol", "operator", "registry",
  "publish", "control-plane", "retire", "builder", "packages", "sdk", "base", "store"];

const capability = await labIsUsable();
const binary = hostBinaryPath();
const installer = bootstrapBinaryPath();
const bundle = process.env["MESH_LAB_BUNDLE"] ?? "";
const catalogDir = process.env["MESH_LAB_CATALOG"] ?? "";
const repos = process.env["MESH_LAB_REPOS"] ?? "";
const KEEP = !!process.env["MESH_LAB_KEEP"];
const FIXED_ID = process.env["MESH_LAB_INSTANCE_ID"] ?? (KEEP ? "genesis-keeps-its-bus-live" : undefined);

const missingRepo = REPOS.find((r) => !repos || !existsSync(resolve(repos, r, ".git")));
const skip =
  !capability.usable ? capability.why :
  !binary ? "MESH_LAB_HOST_BINARY is not set to a built mesh-host" :
  !installer ? "MESH_LAB_BOOTSTRAP_BINARY is not set to a built mesh-bootstrap" :
  !bundle || !existsSync(bundle) ? "MESH_LAB_BUNDLE is not set to the genesis bundle" :
  !catalogDir || !existsSync(join(catalogDir, "nats", "module.json")) ? "MESH_LAB_CATALOG is not set to mesh-catalog/modules" :
  missingRepo ? `MESH_LAB_REPOS holds no ${missingRepo} checkout` :
  false;

let instanceId = "";
/** Why the bed never reached the step, or "" when it did. */
let unreached = "";
let first: GenesisResult | undefined;
let second: GenesisResult | undefined;
let seenBefore: Seen | undefined;
let seenAfter: Seen | undefined;
let nodeLine = "";
const heads: Record<string, string> = {};
const report: string[] = [];

async function on(machine: string, command: string, timeoutMs?: number): Promise<{ out: string; ok: boolean }> {
  const { stdout } = await exec(instanceId, machine, ["sh", "-c", `exec 2>&1\n${command}\necho "__exit=$?"`], timeoutMs);
  const marker = stdout.lastIndexOf("__exit=");
  if (marker < 0) return { out: stdout, ok: false };
  return { out: stdout.slice(0, marker), ok: stdout.slice(marker + 7).trim() === "0" };
}
async function must(machine: string, command: string, timeoutMs?: number): Promise<string> {
  const { out, ok } = await on(machine, command, timeoutMs);
  if (!ok) throw new Error(`${machine}: ${command}\n${out}`);
  return out;
}

/** Each repository at its checkout's HEAD, served from the forge machine over git's protocol as `main`. */
async function raiseForge(): Promise<void> {
  const dir = mkdtempSync(join(tmpdir(), "mesh-lab-590-forge-"));
  try {
    for (const repo of REPOS) {
      const checkout = resolve(repos, repo);
      heads[repo] = execFileSync("git", ["-C", checkout, "rev-parse", "HEAD"], { encoding: "utf8" }).trim();
      const bare = join(dir, `${repo}.git`);
      execFileSync("git", ["init", "-q", "--bare", bare]);
      // No collection after the push: one run in the background rewrites the objects while they are carried.
      execFileSync("git", ["-C", bare, "config", "receive.autogc", "false"]);
      execFileSync("git", ["-C", checkout, "push", "-q", "--force", bare, "HEAD:refs/heads/main"], { stdio: "pipe" });
    }
    const tar = join(tmpdir(), `mesh-lab-590-forge-${process.pid}.tar`);
    execFileSync("tar", ["-cf", tar, "-C", dir, "."]);
    await push(instanceId, FORGE, tar, "/tmp/forge.tar");
    rmSync(tar, { force: true });
  } finally {
    rmSync(dir, { recursive: true, force: true });
  }
  // Owned by the daemon's user, or git refuses to serve it ("dubious ownership") and says nothing.
  await must(FORGE, `mkdir -p /srv/git && tar --no-same-owner -xf /tmp/forge.tar -C /srv/git && chown -R root:root /srv/git && ` +
    `git daemon --base-path=/srv/git --export-all --reuseaddr --detach --listen=${FORGE_ADDRESS} --pid-file=/run/git-daemon.pid`);
  for (let i = 0; ; i++) {
    if ((await on(ANCHOR, `git ls-remote git://${FORGE_ADDRESS}/mesh-catalog.git main`)).ok) break;
    if (i >= 30) throw new Error(`the anchor cannot reach the forge at git://${FORGE_ADDRESS}/`);
    await new Promise((r) => setTimeout(r, 2_000));
  }
}

/** What the anchor shows of its bus: the containers it runs, and every stream the bus holds. */
async function look(): Promise<Seen> {
  const ps = await must(ANCHOR, `docker ps --format '{{.Names}}\t{{.Image}}\t{{.Ports}}'`);
  // The monitor, published on the anchor's loopback by the bundle and by the module alike.
  let jsz = "";
  for (let i = 0; ; i++) {
    const asked = await on(ANCHOR, `curl -sf --max-time 10 'http://127.0.0.1:8222/jsz?accounts=true&streams=true'`);
    if (asked.ok) { jsz = asked.out; break; }
    if (i >= 30) throw new Error(`the bus's monitor does not answer on the anchor: ${asked.out}`);
    await new Promise((r) => setTimeout(r, 2_000));
  }
  return { running: runningFrom(ps), streams: streamsFrom(jsz) };
}

function stoppedAt(r: GenesisResult): string {
  return r.step.replace(/^step \d+ of \d+, /, "");
}

before(async () => {
  if (skip) return;
  const raised = await raise(loadScenario(`scenarios/${SCENARIO}.yml`), {
    onProgress: (m) => console.log(`raise: ${m}`),
    ...(FIXED_ID ? { instanceId: FIXED_ID } : {}),
  });
  instanceId = raised.instanceId;
  console.log(`INSTANCE ${instanceId}${KEEP ? " (KEEP — will be left standing)" : ""}`);

  try {
    await raiseForge();
    report.push(`  forge             ${REPOS.map((r) => `${r} ${heads[r]?.slice(0, 8)}`).join(", ")}`);
  } catch (err) {
    unreached = `the forge was not raised, so the installer never ran: ${(err as Error).message}`;
    return;
  }

  const forge = (repo: string) => `git://${FORGE_ADDRESS}/${repo}.git`;
  const run = (catalogue: string) => genesis({
    instanceId,
    node: ANCHOR,
    installer: installer as string,
    catalogDir: catalogue,
    catalogueModules: ["distribution", "build-agent", "postgres"],
    bundleTemplate: foundationBundle(bundle, []),
    registry: REGISTRY,
    source: forge("mesh-controller"),
    sourceRef: heads["mesh-controller"] as string,
    toolsSource: forge("mesh-tools"),
    sdkSource: forge("mesh-sdk"),
    catalogSource: forge("mesh-catalog"),
    // A refusal is the answer, and a re-run is the second half of this bed, not a retry.
    attempts: 1,
    verify: false,
    log: (m) => console.log(m),
  });

  // 1. From bare to the broker step, and no further: the catalogue the installer reads has no nats module.
  const withoutTheBus = mkdtempSync(join(tmpdir(), "mesh-lab-590-catalogue-"));
  try {
    const nats = join(catalogDir, "nats");
    cpSync(catalogDir, withoutTheBus, { recursive: true, filter: (src) => src !== nats && !src.startsWith(nats + "/") });
    first = await run(withoutTheBus);
  } catch (err) {
    unreached = `the installer could not be run: ${(err as Error).message}`;
    return;
  } finally {
    rmSync(withoutTheBus, { recursive: true, force: true });
  }
  report.push(...first.report);
  if (first.ok || stoppedAt(first) !== BROKER) {
    unreached = first.ok
      ? "the first run finished without the nats module, so the broker step never stopped it"
      : `the first run stopped at ${first.step || "no step it named"}, before the broker step:\n${first.why}`;
    return;
  }
  if (!/nats module's manifest could not be read/.test(first.said)) {
    unreached = `the first run stopped at the broker step for another reason than the missing module:\n${first.why}`;
    return;
  }
  seenBefore = await look();
  report.push(`  before the step   ${busContainers(seenBefore.running).map((b) => `${b.name} on ${b.image}`).join("; ")}`);
  report.push(`                    ${written(seenBefore).length} stream(s) holding something: ` +
    written(seenBefore).map((s) => `${s.name} (last ${s.lastSeq})`).join(", "));
  if (written(seenBefore).length === 0) {
    unreached = "nothing was written on the bus before the broker step, so whether its streams survive it says nothing";
    return;
  }

  // 2. The whole catalogue: the steps before are done already, and the broker step adopts the bus.
  try {
    second = await run(catalogDir);
  } catch (err) {
    unreached = `the installer could not be run again: ${(err as Error).message}`;
    return;
  }
  report.push(...second.report);
  if (!second.ok && BEFORE_BROKER.includes(stoppedAt(second))) {
    unreached = `the second run stopped at ${second.step}, before the broker step it was run again for:\n${second.why}`;
    return;
  }
  seenAfter = await look();
  report.push(`  after the step    ${busContainers(seenAfter.running).map((b) => `${b.name} on ${b.image}`).join("; ") || "no bus"}`);
  const nodes = await on(ANCHOR, `docker exec mesh-controller /mesh-controller node list`, 60_000);
  nodeLine = nodes.out.split("\n").map((l) => l.trim()).find((l) => l.startsWith(`${ANCHOR} `)) ?? "";
  report.push(`  node list         ${nodeLine || nodes.out.trim()}`);
}, { timeout: 7_200_000 });

after(async () => {
  if (KEEP) {
    console.log(`\nLEFT STANDING: ${instanceId} — not destroyed (MESH_LAB_KEEP).`);
    return;
  }
  if (instanceId) await destroy(instanceId);
  await destroyAll(`${SCENARIO}-`);
}, { timeout: 900_000 });

test("genesis adopts its bus as the nats module in place: one bus, its streams kept, the node connected", {
  skip, timeout: 7_200_000,
}, async (t) => {
  console.log(report.join("\n"));
  if (unreached) {
    t.skip(`ADR 0333 §5 NOT REACHED: ${unreached}`);
    return;
  }
  const ran = second as GenesisResult;
  // Stopped AT the broker step: the adoption itself refused or failed — the fault, not a missed bed.
  assert.ok(ran.ok || stoppedAt(ran) !== BROKER,
    `genesis stopped at its broker step and did not adopt the bus:\n${ran.why}`);
  if (!ran.ok) {
    console.log(`\nthe second run crossed the broker step and stopped later, at ${ran.step} — not this bed's ` +
      `question:\n${ran.why}`);
  }
  const wrong = busNotKept(seenBefore as Seen, seenAfter as Seen, nodeLine, ran.said);
  assert.deepEqual(wrong, [], `genesis did not keep its bus through the broker step:\n  - ${wrong.join("\n  - ")}`);
  console.log(`\nVERDICT: one bus, ${BUS}, on the module's image; ${written(seenBefore as Seen).length} stream(s) ` +
    `kept; ${ANCHOR} connected.`);
});