// Form and inspect a real multi-node k3s cluster from an integration test.
//
// WHY THIS EXISTS
// ---------------
// Nothing in the repo decides which node bootstraps a cluster and which nodes
// join it. The production generator has the right discriminator --
// generateServerConfig emits `cluster-init: true` when no server URL is set and
// `server:`+`token:` when one is -- but the choice is an INPUT, and the only
// caller that ever supplied it was a hand-run labsim shell script.
//
// So a test that wants three servers in one cluster has to sequence it. That
// sequencing is here rather than in a test file because the rolling drill is
// not the only thing that will want it: an etcd-recovery test (recoverEtcdMember
// still has no caller), a dual-stack conversion and any HA upgrade test all
// need exactly these primitives.
//
// Everything drives K3sModule, not shell. That is deliberate and is the same
// rule labsim follows by shelling out to the compiled render-config binary: a
// test that reimplements the install proves the test works, not the product.

import { execSync } from "node:child_process";
import type { Role } from "../../../src/shared/src/types/state.js";
import { K3sModule } from "../../../src/modules/modules/k3s/src/k3s-module.js";
import type { ModuleContext } from "../../../src/modules/src/types.js";
import { log } from "./libvirt.js";
import { parseEtcdMembers, type EtcdMember } from "../../../src/modules/modules/k3s/src/operations/etcd-members.js";
import { sshExec } from "./ssh.js";
import type { Harness, ProvisionedVm } from "./provision-harness.js";

export const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

/**
 * A pre-shared cluster token, the way labsim does it.
 *
 * The alternative is scraping /var/lib/rancher/k3s/server/node-token off the
 * first server, and nothing in the repo does that. It is also the wrong shape
 * for a rolling migration: the first server gets wiped and rejoins, so the
 * token has to outlive it and be knowable before it exists.
 */
export const DRILL_TOKEN = "lab-drill-shared-token";

/**
 * k3s operations run unprefixed `mkdir -p /etc/rancher/k3s` and
 * `systemctl restart k3s`, so they need root. install.ks.ts provisions
 * `sshkey --username=root` and sets PermitRootLogin prohibit-password, so root
 * SSH works on every machine this harness builds. The harness's own `lab` user
 * stays for the sudo-based reads (readLvs, markers).
 */
export const MODULE_SSH_USER = "root";

/**
 * The k3s version the drill runs, pinned to what the fleet runs.
 *
 * get.k3s.io installs LATEST, and latest is currently two minor versions ahead
 * of production (v1.36.4 vs v1.34.6). Two things follow, and both matter:
 *
 *  - a rehearsal on a different minor version is rehearsing a different
 *    system, and Longhorn in particular is only qualified against a range of
 *    Kubernetes versions;
 *  - it is the same trap production would hit. A node reinstalled mid-migration
 *    comes back on latest while its peers stay put, and Kubernetes supports
 *    only ONE minor of skew between control-plane members.
 *
 * Override with LAB_K3S_VERSION to test an upgrade deliberately.
 */
export const PINNED_K3S_VERSION = process.env["LAB_K3S_VERSION"] ?? "v1.34.6+k3s1";

/** Build the ModuleContext K3sModule expects for one VM. */
export function moduleCtx(
  h: Harness,
  vm: ProvisionedVm,
  role: Role,
  config: Record<string, unknown> = {},
): ModuleContext {
  return {
    hostname: vm.name,
    ip: vm.ip,
    role,
    os: "fedora-43",
    arch: "x86_64",
    sshUser: MODULE_SSH_USER,
    sshKeyPath: h.sshKeyPath,
    config,
  };
}

/** Run a command on a node as root, throwing with context on failure. */
export function nodeExec(h: Harness, vm: ProvisionedVm, cmd: string, timeoutMs = 120_000): string {
  const r = sshExec(vm.ip, MODULE_SSH_USER, cmd, { keyPath: h.sshKeyPath, timeout: timeoutMs });
  if (r.exitCode !== 0) {
    throw new Error(`[${vm.name}] ${cmd} failed (${r.exitCode}): ${r.stderr.trim() || r.stdout.trim()}`);
  }
  return r.stdout;
}

/** Same, but returns the result instead of throwing. */
export function tryNodeExec(h: Harness, vm: ProvisionedVm, cmd: string, timeoutMs = 120_000) {
  return sshExec(vm.ip, MODULE_SSH_USER, cmd, { keyPath: h.sshKeyPath, timeout: timeoutMs });
}

/**
 * The name Kubernetes knows this VM by.
 *
 * NOT the libvirt domain name. k3s registers a node under its FQDN, so a VM
 * called `smoke-a` appears as `smoke-a.pxe-test.local`. Passing the VM name to
 * `kubectl get node` silently finds nothing, and every wait built on it times
 * out instead of failing usefully. Read from the node itself, the same way the
 * production operations do (`hostname -f`), so the two can never disagree.
 */
const nodeNameCache = new Map<string, string>();
export function k8sNodeName(h: Harness, vm: ProvisionedVm): string {
  const cached = nodeNameCache.get(vm.name);
  if (cached !== undefined) return cached;
  const name = nodeExec(h, vm, "hostname -f 2>/dev/null || hostname").trim();
  if (name === "") throw new Error(`[${vm.name}] could not resolve node name`);
  nodeNameCache.set(vm.name, name);
  return name;
}

/** Forget cached names — a reinstalled machine re-registers. */
export function forgetNodeName(vm: ProvisionedVm): void {
  nodeNameCache.delete(vm.name);
}

/** kubectl against a server node. */
export function kubectlOn(h: Harness, server: ProvisionedVm, args: string, timeoutMs = 120_000): string {
  return nodeExec(h, server, `k3s kubectl ${args}`, timeoutMs);
}

export function kubectlJson<T>(h: Harness, server: ProvisionedVm, args: string): T {
  return JSON.parse(kubectlOn(h, server, `${args} -o json`)) as T;
}

/**
 * Wait until a server's API answers /ping.
 *
 * Straight from labsim-k8s-etcd.sh, whose comment is the whole justification:
 * a joiner that starts before the first server has finished bootstrapping etcd
 * races it, and the failure looks like etcd instability rather than a startup
 * order problem. Probed from the HOST, so it does not depend on the joining
 * node being reachable yet.
 */
export async function waitForApiPing(vm: ProvisionedVm, timeoutMs = 450_000): Promise<void> {
  const start = Date.now();
  let lastErr = "";
  while (Date.now() - start < timeoutMs) {
    try {
      const out = execSync(
        `curl -sk --max-time 5 https://${vm.ip}:6443/ping || true`,
        { encoding: "utf-8", timeout: 15_000 },
      ).trim();
      if (out === "pong") {
        log(`[${vm.name}] API answering after ${Math.round((Date.now() - start) / 1000)}s`);
        return;
      }
      lastErr = out;
    } catch (e) {
      lastErr = e instanceof Error ? e.message : String(e);
    }
    await sleep(5_000);
  }
  throw new Error(`[${vm.name}] API never answered /ping in ${timeoutMs / 1000}s (last: ${lastErr || "no response"})`);
}

/** Wait for a node to appear Ready in the cluster, as seen from `server`. */
export async function waitForNodeReady(
  h: Harness,
  server: ProvisionedVm,
  /** The VM to wait for; its Kubernetes name is resolved from the machine. */
  target: ProvisionedVm,
  // Generous: a node is NotReady until its CNI is up, and on the cluster-init
  // server that means waiting for Cilium to install and pull its images. Five
  // minutes is enough on a warm host and not enough on a cold one, and the
  // cost of being wrong is a wasted multi-hour run.
  timeoutMs = 900_000,
): Promise<void> {
  const nodeName = k8sNodeName(h, target);
  const start = Date.now();
  let last = "";
  while (Date.now() - start < timeoutMs) {
    const r = tryNodeExec(h, server,
      `k3s kubectl get node ${nodeName} -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}'`);
    last = r.stdout.trim();
    if (last === "True") {
      log(`[${nodeName}] Ready after ${Math.round((Date.now() - start) / 1000)}s`);
      return;
    }
    await sleep(5_000);
  }
  throw new Error(`[${nodeName}] never became Ready in ${timeoutMs / 1000}s (last status: ${last || "absent"})`);
}

/**
 * etcdctl invocation.
 *
 * NOT `k3s etcdctl` -- that is not a subcommand ("No help topic for 'etcdctl'",
 * verified on a live v1.34 node). The binary comes from the `etcd` rpm, which
 * our kickstart does NOT install, so ensureEtcdctl() has to put it there.
 * Cert paths verified against a real node.
 */
const ETCD_BASE = "/var/lib/rancher/k3s/server/tls/etcd";
const ETCDCTL =
  `ETCDCTL_API=3 etcdctl ` +
  `--cacert=${ETCD_BASE}/server-ca.crt ` +
  `--cert=${ETCD_BASE}/server-client.crt ` +
  `--key=${ETCD_BASE}/server-client.key ` +
  `--endpoints=https://127.0.0.1:2379`;

/** Install etcdctl if the node does not have it. Idempotent, ~30s once. */
export function ensureEtcdctl(h: Harness, server: ProvisionedVm): void {
  const has = tryNodeExec(h, server, "command -v etcdctl");
  if (has.exitCode === 0) return;
  log(`[${server.name}] installing etcdctl (the etcd rpm; kickstart does not include it)`);
  nodeExec(h, server, "dnf install -y etcd >/dev/null 2>&1 || dnf install -y etcd-ctl >/dev/null 2>&1", 300_000);
  nodeExec(h, server, "command -v etcdctl");
}


/**
 * etcd membership, read with etcdctl against k3s's own certs.
 *
 * Read rather than inferred from `kubectl get nodes`: a node object can be
 * deleted while its etcd member survives, and that stale member is exactly the
 * failure a rolling reinstall creates. Counting nodes would not see it.
 */
export function etcdMembers(h: Harness, server: ProvisionedVm): EtcdMember[] {
  // NOT `-w json`. Two reasons, and the second is the dangerous one:
  //
  //  1. `member remove` requires the ID in HEX; the JSON writer emits decimal.
  //  2. etcd member IDs are uint64. JSON.parse turns them into JS numbers,
  //     which silently lose precision above 2^53 -- the ID that comes back is
  //     an APPROXIMATION of the real one. Passing that to `member remove`
  //     fails with "value out of range" if you are lucky, and in principle
  //     could match some other member if you are not. This happens at the one
  //     moment it must not: the node is already stopped and about to be wiped.
  //
  // The default writer prints the ID as hex text, so it never becomes a
  // number at all:
  //   825852d3683bc406, started, worker2-...-5f7ad0c9, https://...:2380, ...
  return parseEtcdMembers(nodeExec(h, server, `${ETCDCTL} member list`));
}


/** Are all etcd endpoints healthy? Answers the question quorum actually asks. */
export function etcdHealthy(h: Harness, server: ProvisionedVm): boolean {
  const r = tryNodeExec(h, server, `${ETCDCTL} endpoint health`);
  return r.exitCode === 0 && r.stdout.includes("is healthy");
}

/**
 * Remove an etcd member by node name.
 *
 * Must happen BEFORE the node is wiped. k3s-install.ts removes the local
 * datastore when a server joins, so a reinstalled node rejoins as a NEW member
 * while the old record persists: 3 members becomes 4, quorum rises from 2 to 3,
 * and the cluster that tolerated one failure now tolerates none.
 */
export function removeEtcdMember(h: Harness, peer: ProvisionedVm, nodeName: string): boolean {
  const members = etcdMembers(h, peer);
  const target = members.find((m) => m.node === nodeName || m.name === nodeName);
  if (!target) {
    log(`[etcd] no member for ${nodeName} (have: ${members.map((m) => m.node).join(",")})`);
    return false;
  }
  nodeExec(h, peer, `${ETCDCTL} member remove ${target.id}`);
  log(`[etcd] removed member ${target.name} (${target.id})`);
  return true;
}

export interface ClusterSnapshot {
  nodes: { name: string; ready: boolean; roles: string }[];
  readyCount: number;
  etcdMembers: number;
  etcdHealthy: boolean;
}

/** One read of cluster state, in the shape labsim's cmd_status prints. */
export function clusterSnapshot(h: Harness, server: ProvisionedVm): ClusterSnapshot {
  const raw = kubectlJson<{ items: {
    metadata: { name: string; labels?: Record<string, string> };
    status: { conditions?: { type: string; status: string }[] };
  }[] }>(h, server, "get nodes");

  const nodes = raw.items.map((n) => ({
    name: n.metadata.name,
    ready: (n.status.conditions ?? []).some((c) => c.type === "Ready" && c.status === "True"),
    roles: Object.keys(n.metadata.labels ?? {})
      .filter((k) => k.startsWith("node-role.kubernetes.io/"))
      .map((k) => k.replace("node-role.kubernetes.io/", ""))
      .sort().join(",") || "<none>",
  }));

  let members = 0;
  let healthy = false;
  try {
    members = etcdMembers(h, server).length;
    healthy = etcdHealthy(h, server);
  } catch { /* an agent, or the API is down — report zero rather than throw */ }

  return { nodes, readyCount: nodes.filter((n) => n.ready).length, etcdMembers: members, etcdHealthy: healthy };
}

/**
 * Print a snapshot. Called liberally during a drill: the VMs are destroyed at
 * the end, so this transcript is the only surviving record of what happened.
 */
export function printSnapshot(label: string, s: ClusterSnapshot): void {
  log(`[snapshot] ${label}: ${s.readyCount}/${s.nodes.length} Ready, ` +
      `etcd members=${s.etcdMembers} healthy=${s.etcdHealthy}`);
  for (const n of s.nodes) log(`             ${n.ready ? "Ready   " : "NotReady"} ${n.name} (${n.roles})`);
}

/**
 * Install k3s on one node via the production module.
 *
 * Pass no serverUrl to bootstrap the cluster (the token still seeds), or a
 * serverUrl to join an existing one.
 */
export async function installK3sNode(h: Harness, vm: ProvisionedVm, opts: {
  role: Role;
  serverUrl?: string;
  token?: string;
}): Promise<void> {
  const mod = new K3sModule();
  const config: Record<string, unknown> = {
    k3sToken: opts.token ?? DRILL_TOKEN,
    k3sVersion: PINNED_K3S_VERSION,
    ...(opts.serverUrl === undefined ? {} : { k3sServerUrl: opts.serverUrl }),
  };

  log(`[${vm.name}] installing k3s (${opts.role}${opts.serverUrl ? ` -> ${opts.serverUrl}` : " — cluster-init"})`);
  const install = await mod.install(moduleCtx(h, vm, opts.role, config));
  for (const line of install.output) log(`  [${vm.name}] ${line}`);
  if (!install.success) {
    throw new Error(`[${vm.name}] k3s install failed: ${install.errors.join("; ")}`);
  }

  const configure = await mod.configure(moduleCtx(h, vm, opts.role, config));
  for (const line of configure.output) log(`  [${vm.name}] ${line}`);
  if (!configure.success) {
    throw new Error(`[${vm.name}] k3s configure failed: ${configure.errors.join("; ")}`);
  }
}

export function serverUrlOf(vm: ProvisionedVm): string {
  return `https://${vm.ip}:6443`;
}

/**
 * Form a cluster from a list of server VMs, strictly in order.
 *
 * The first bootstraps etcd and installs the CNI (k3s-module runs networking
 * only on the cluster-init node -- joiners get Cilium via its DaemonSet). The
 * rest join one at a time, each gated on the previous node being Ready.
 *
 * Serial on purpose. labsim gets away with starting joiners together because
 * its nodes are cloud-init VMs that poll /ping themselves; here the few extra
 * minutes buy determinism in the one place where etcd bootstrap genuinely
 * races.
 */
export async function formCluster(h: Harness, servers: ProvisionedVm[], opts: {
  role?: Role;
  token?: string;
} = {}): Promise<void> {
  if (servers.length === 0) throw new Error("formCluster: no servers");
  const role = opts.role ?? "k8s-combo";
  const token = opts.token ?? DRILL_TOKEN;
  const first = servers[0]!;

  await installK3sNode(h, first, { role, token });
  await waitForApiPing(first);
  await waitForNodeReady(h, first, first);
  ensureEtcdctl(h, first);
  printSnapshot("after cluster-init", clusterSnapshot(h, first));

  for (const vm of servers.slice(1)) {
    await waitForApiPing(first);
    await installK3sNode(h, vm, { role, token, serverUrl: serverUrlOf(first) });
    await waitForNodeReady(h, first, vm);
    ensureEtcdctl(h, vm);
    printSnapshot(`after joining ${vm.name}`, clusterSnapshot(h, first));
  }
}

/** Join an agent. Agents cannot start without a server URL and token. */
export async function joinAgent(h: Harness, vm: ProvisionedVm, opts: {
  server: ProvisionedVm;
  role?: Role;
  token?: string;
}): Promise<void> {
  await waitForApiPing(opts.server);
  await installK3sNode(h, vm, {
    role: opts.role ?? "k8s-worker",
    token: opts.token ?? DRILL_TOKEN,
    serverUrl: serverUrlOf(opts.server),
  });
  await waitForNodeReady(h, opts.server, vm);
}

/**
 * A healthy server that is NOT the node being worked on.
 *
 * Needed because the first server is itself rolled during a migration: after
 * it is wiped it must rejoin against a survivor, so "the first node" cannot be
 * the join target and cannot be where cluster state is read from.
 */
export function pickJoinTarget(
  h: Harness,
  servers: ProvisionedVm[],
  exclude: ProvisionedVm,
): ProvisionedVm {
  for (const vm of servers) {
    if (vm.name === exclude.name) continue;
    const r = tryNodeExec(h, vm, "k3s kubectl get --raw /readyz");
    if (r.exitCode === 0 && r.stdout.trim() === "ok") return vm;
  }
  throw new Error(`No healthy server other than ${exclude.name} — cluster has no quorum to join against`);
}

/** Every server reports /readyz ok. Asserted positively, not inferred. */
export function allServersReady(h: Harness, servers: ProvisionedVm[]): { ok: boolean; detail: string } {
  const results = servers.map((vm) => {
    const r = tryNodeExec(h, vm, "k3s kubectl get --raw /readyz");
    return `${vm.name}=${r.exitCode === 0 ? r.stdout.trim() : `ERR(${r.exitCode})`}`;
  });
  return { ok: results.every((r) => r.endsWith("=ok")), detail: results.join(" ") };
}