// Install Longhorn on a test cluster and put verifiable data in it.
//
// WHAT THIS IS FOR
// ----------------
// The rolling drill's whole claim is "the bytes survived". That needs data
// whose integrity can be checked independently of Longhorn's own opinion of
// itself -- Longhorn reporting `robustness: healthy` is not evidence that the
// file is intact, it is evidence that Longhorn thinks it is.
//
// So: write a known blob, record its sha256, and re-verify THROUGH A POD
// RESTART every time. The restart is not incidental. It forces a real
// detach/attach, frequently onto a different node, which is the only way to
// show the data followed the volume rather than sitting in a page cache on the
// node that wrote it.
//
// Longhorn version is pinned to the fleet's (1.8.1) so the drill rehearses what
// production runs, not what happens to be latest.

import type { Harness, ProvisionedVm } from "./provision-harness.js";
import { log } from "./libvirt.js";
import { nodeExec, tryNodeExec, kubectlOn, kubectlJson, sleep, k8sNodeName } from "./k3s-cluster.js";

export const LONGHORN_VERSION = "1.8.1";
export const LONGHORN_NS = "longhorn-system";

/**
 * Install Longhorn via k3s's built-in helm controller.
 *
 * A HelmChart manifest dropped in the server manifests directory is how k3s
 * installs charts natively, so this needs no helm binary on the host and no
 * kubeconfig plumbing.
 *
 * The namespace is created FIRST, with privileged Pod Security labels. Longhorn
 * needs hostPath and privileged containers; under the default `restricted`
 * enforcement its DaemonSets are rejected and the install fails in a way that
 * looks like a chart problem.
 *
 * `createDefaultDiskLabeledNodes: true` matches production, and is the setting
 * that makes the node label and disk annotation load-bearing: without both, a
 * node gets no disk and silently hosts no replicas.
 */
/**
 * S3 backup target for the drill, from the environment.
 *
 * Absent means the backup/restore phase is skipped rather than silently
 * "passing" -- a backup test that quietly does nothing is worse than no test.
 * The run script exports these; see DRILL_S3_* in run-drill.sh.
 */
export interface BackupTargetOpts {
  /** e.g. s3://longhorn-drill-backup@us-east-1/ */
  target: string;
  endpoint: string;
  accessKey: string;
  secretKey: string;
}

export function backupTargetFromEnv(): BackupTargetOpts | undefined {
  const target = process.env["DRILL_S3_TARGET"];
  const endpoint = process.env["DRILL_S3_ENDPOINT"];
  const accessKey = process.env["DRILL_S3_ACCESS_KEY"];
  const secretKey = process.env["DRILL_S3_SECRET_KEY"];
  if (target === undefined || endpoint === undefined
    || accessKey === undefined || secretKey === undefined) return undefined;
  return { target, endpoint, accessKey, secretKey };
}

export const BACKUP_SECRET = "longhorn-backup-credentials";

export async function installLonghorn(h: Harness, server: ProvisionedVm, opts: {
  defaultReplicas?: number;
  backup?: BackupTargetOpts;
} = {}): Promise<void> {
  const replicas = opts.defaultReplicas ?? 3;

  log(`[longhorn] creating ${LONGHORN_NS} with privileged PSA labels`);
  nodeExec(h, server, `k3s kubectl apply -f - <<'EOF'
apiVersion: v1
kind: Namespace
metadata:
  name: ${LONGHORN_NS}
  labels:
    pod-security.kubernetes.io/enforce: privileged
    pod-security.kubernetes.io/audit: privileged
    pod-security.kubernetes.io/warn: privileged
EOF`);

  if (opts.backup !== undefined) {
    // Secret BEFORE the chart: Longhorn reads it by name from a defaultSetting,
    // and a missing secret makes the first backup fail with a credential error
    // that reads like a network problem.
    //
    // AWS_ENDPOINTS is what makes this work against a non-AWS gateway; without
    // it the driver resolves real AWS endpoints.
    log(`[longhorn] configuring backup target ${opts.backup.target}`);
    // tryNodeExec, not nodeExec: nodeExec throws with the whole command in the
    // message, and this command contains the S3 secret. A credential landing in
    // a log that then gets pasted into drill evidence is not acceptable, so the
    // failure is reported without the payload.
    const secretApplied = tryNodeExec(h, server, `k3s kubectl -n ${LONGHORN_NS} apply -f - <<'EOF'
apiVersion: v1
kind: Secret
metadata:
  name: ${BACKUP_SECRET}
  namespace: ${LONGHORN_NS}
stringData:
  AWS_ACCESS_KEY_ID: ${opts.backup.accessKey}
  AWS_SECRET_ACCESS_KEY: ${opts.backup.secretKey}
  AWS_ENDPOINTS: ${opts.backup.endpoint}
EOF`);
    if (secretApplied.exitCode !== 0) {
      throw new Error(
        `[${server.name}] could not create the ${BACKUP_SECRET} secret ` +
        `(exit ${secretApplied.exitCode}). Command withheld: it contains credentials. ` +
        `stderr: ${secretApplied.stderr.trim().slice(0, 200)}`);
    }
  }

  log(`[longhorn] installing chart ${LONGHORN_VERSION} via k3s helm controller`);
  nodeExec(h, server, `cat > /var/lib/rancher/k3s/server/manifests/longhorn.yaml <<'EOF'
apiVersion: helm.cattle.io/v1
kind: HelmChart
metadata:
  name: longhorn
  namespace: kube-system
spec:
  chart: longhorn
  repo: https://charts.longhorn.io
  version: ${LONGHORN_VERSION}
  targetNamespace: ${LONGHORN_NS}
  valuesContent: |-
    defaultSettings:
      defaultDataPath: /var/lib/longhorn
      defaultReplicaCount: ${replicas}
      createDefaultDiskLabeledNodes: true
      defaultDataLocality: best-effort
      replicaSoftAntiAffinity: false
      replicaReplenishmentWaitInterval: 600
      orphanAutoDeletion: true
    persistence:
      defaultClass: true
      defaultClassReplicaCount: ${replicas}
      defaultFsType: ext4
    longhornUI:
      replicas: 1
EOF`);
}

/**
 * Every expected node has a Longhorn disk that is actually schedulable.
 *
 * Takes the VMs, not names, and resolves each one's Kubernetes name itself.
 * Longhorn keys its node records by the FQDN the kubelet registered
 * (`drill-s1.pxe-test.local`), never the libvirt domain name (`drill-s1`), so
 * a caller passing VM names matches nothing and waits out the full timeout
 * while the cluster is in fact perfectly healthy.
 */
export async function waitForLonghornReady(
  h: Harness,
  server: ProvisionedVm,
  vms: ProvisionedVm[],
  timeoutMs = 20 * 60_000,
): Promise<void> {
  const expectedNodes = vms.map((vm) => k8sNodeName(h, vm));
  const start = Date.now();
  let last = "";
  while (Date.now() - start < timeoutMs) {
    const r = tryNodeExec(h, server,
      `k3s kubectl -n ${LONGHORN_NS} get nodes.longhorn.io -o json`, 60_000);
    if (r.exitCode === 0) {
      try {
        const parsed = JSON.parse(r.stdout) as { items: {
          metadata: { name: string };
          status?: { diskStatus?: Record<string, { conditions?: { type: string; status: string }[] }> };
        }[] };
        const ready = parsed.items.filter((n) => {
          const disks = Object.values(n.status?.diskStatus ?? {});
          return disks.length > 0 && disks.some((d) =>
            (d.conditions ?? []).some((c) => c.type === "Schedulable" && c.status === "True"));
        }).map((n) => n.metadata.name);

        last = `${ready.length}/${expectedNodes.length} with schedulable disk: ${ready.join(",") || "none"}`;
        if (expectedNodes.every((n) => ready.includes(n))) {
          log(`[longhorn] ready after ${Math.round((Date.now() - start) / 1000)}s — ${last}`);
          return;
        }
      } catch { /* API not serving the CRD yet */ }
    }
    await sleep(15_000);
  }

  // Say WHY, per node, instead of just "not ready". A node with no disk has a
  // specific cause and the record knows it; without this the only way to find
  // out is to catch the cluster before teardown, which is a race you lose.
  const detail: string[] = [];
  try {
    const raw = kubectlJson<{ items: {
      metadata: { name: string; annotations?: Record<string, string>; labels?: Record<string, string> };
      spec?: { disks?: Record<string, unknown>; allowScheduling?: boolean };
      status?: {
        conditions?: { type: string; status: string; reason?: string; message?: string }[];
        diskStatus?: Record<string, { conditions?: { type: string; status: string; reason?: string; message?: string }[] }>;
      };
    }[] }>(h, server, `-n ${LONGHORN_NS} get nodes.longhorn.io`);

    for (const n of raw.items) {
      const disks = Object.keys(n.spec?.disks ?? {});
      const conds = (n.status?.conditions ?? [])
        .filter((c) => c.status !== "True")
        .map((c) => `${c.type}=${c.status}(${c.reason ?? ""}: ${(c.message ?? "").slice(0, 80)})`);
      const diskConds = Object.entries(n.status?.diskStatus ?? {}).flatMap(([dn, d]) =>
        (d.conditions ?? []).filter((c) => c.status !== "True")
          .map((c) => `${dn}:${c.type}=${c.status}(${c.reason ?? ""}: ${(c.message ?? "").slice(0, 80)})`));
      const ann = n.metadata.annotations?.["node.longhorn.io/default-disks-config"] ?? "<absent>";
      const lbl = n.metadata.labels?.["node.longhorn.io/create-default-disk"] ?? "<absent>";
      detail.push(
        `  ${n.metadata.name}: disks=[${disks.join(",")}] label=${lbl} ann=${ann} ` +
        `${conds.length > 0 ? `nodeConds=${conds.join(";")} ` : ""}` +
        `${diskConds.length > 0 ? `diskConds=${diskConds.join(";")}` : ""}`);
    }
  } catch (e) {
    detail.push(`  (could not read Longhorn nodes: ${e instanceof Error ? e.message : String(e)})`);
  }

  throw new Error(
    `Longhorn not ready in ${timeoutMs / 60000} min (${last}).\n` +
    `A node with no disk usually means the create-default-disk label or the\n` +
    `default-disks-config annotation is missing, or the path is not mounted.\n` +
    detail.join("\n"));
}

/** Create a PVC plus a Deployment that mounts it at /data. */
export function createDataWorkload(h: Harness, server: ProvisionedVm, opts: {
  name: string;
  sizeGi: number;
  replicas: number;
}): void {
  log(`[longhorn] creating ${opts.name} (${opts.sizeGi}Gi, ${opts.replicas} replicas)`);
  nodeExec(h, server, `k3s kubectl apply -f - <<'EOF'
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: lh-${opts.name}
provisioner: driver.longhorn.io
allowVolumeExpansion: true
parameters:
  numberOfReplicas: "${opts.replicas}"
  staleReplicaTimeout: "30"
  fsType: ext4
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: ${opts.name}
spec:
  accessModes: [ReadWriteOnce]
  storageClassName: lh-${opts.name}
  resources:
    requests:
      storage: ${opts.sizeGi}Gi
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${opts.name}
spec:
  replicas: 1
  strategy:
    type: Recreate
  selector:
    matchLabels: {app: ${opts.name}}
  template:
    metadata:
      labels: {app: ${opts.name}}
    spec:
      # Pod Security Admission enforces "restricted" on this namespace (the k3s
      # hardening group applies the label), so a bare pod spec is REJECTED
      # outright -- the Deployment reports 0/1 with no pod and no obvious
      # reason unless you read the ReplicaSet events. Every one of these
      # fields is required by "restricted"; dropping any brings it back.
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        runAsGroup: 1000
        # fsGroup so the non-root user can actually write to the Longhorn
        # volume, which is otherwise owned by root.
        fsGroup: 1000
        seccompProfile: {type: RuntimeDefault}
      containers:
        - name: shell
          image: busybox:1.36
          command: ["sh", "-c", "sleep infinity"]
          securityContext:
            allowPrivilegeEscalation: false
            capabilities: {drop: ["ALL"]}
          volumeMounts:
            - {name: data, mountPath: /data}
      volumes:
        - name: data
          persistentVolumeClaim: {claimName: ${opts.name}}
EOF`);
}

/** Wait for the workload's pod to be Running and its volume attached. */
export async function waitForWorkload(
  h: Harness, server: ProvisionedVm, name: string, timeoutMs = 10 * 60_000,
): Promise<void> {
  const start = Date.now();
  let last = "";
  while (Date.now() - start < timeoutMs) {
    const r = tryNodeExec(h, server,
      `k3s kubectl get pods -l app=${name} -o jsonpath='{.items[*].status.phase}'`);
    last = r.stdout.trim();
    if (last === "Running") {
      // Ready, not merely scheduled: the volume mount happens before Ready.
      const ready = tryNodeExec(h, server,
        `k3s kubectl get pods -l app=${name} -o jsonpath='{.items[*].status.containerStatuses[*].ready}'`);
      if (ready.stdout.trim() === "true") return;
    }
    await sleep(5_000);
  }
  throw new Error(`workload ${name} not running in ${timeoutMs / 60000} min (last phase: ${last || "none"})`);
}

function podOf(h: Harness, server: ProvisionedVm, name: string): string {
  const pod = kubectlOn(h, server, `get pods -l app=${name} -o jsonpath='{.items[0].metadata.name}'`).trim();
  if (!pod) throw new Error(`no pod for workload ${name}`);
  return pod;
}

/**
 * Write a random blob, fsync it, and record its digest.
 *
 * conv=fsync matters: without it the write can still be in the guest's page
 * cache, and a checksum taken immediately afterwards proves nothing about what
 * reached the replicas.
 */
export async function writeBlob(h: Harness, server: ProvisionedVm, opts: {
  name: string; mib: number;
}): Promise<string> {
  const pod = podOf(h, server, opts.name);
  log(`[longhorn] writing ${opts.mib} MiB to ${opts.name}`);
  nodeExec(h, server,
    `k3s kubectl exec ${pod} -- sh -c ` +
    `"dd if=/dev/urandom of=/data/blob bs=1M count=${opts.mib} conv=fsync 2>/dev/null && ` +
    `cd /data && sha256sum blob > blob.sha256 && sync"`,
    15 * 60_000);

  const digest = kubectlOn(h, server, `exec ${pod} -- cat /data/blob.sha256`).trim().split(/\s+/)[0] ?? "";
  if (!/^[0-9a-f]{64}$/.test(digest)) throw new Error(`bad digest for ${opts.name}: ${digest}`);
  log(`[longhorn] ${opts.name} digest ${digest.slice(0, 16)}...`);
  return digest;
}

/**
 * Restart the pod, then verify the blob.
 *
 * The restart is the point. It detaches and reattaches the volume, so a pass
 * means the bytes came back from Longhorn rather than from wherever the writer
 * left them.
 */
export async function verifyBlob(h: Harness, server: ProvisionedVm, opts: {
  name: string; expectDigest: string;
}): Promise<void> {
  const before = podOf(h, server, opts.name);
  kubectlOn(h, server, `delete pod ${before} --wait=true --timeout=5m`, 6 * 60_000);
  await waitForWorkload(h, server, opts.name);
  const after = podOf(h, server, opts.name);
  if (after === before) throw new Error(`${opts.name}: pod was not actually replaced`);

  // sha256sum -c proves the file matches what was recorded IN the volume...
  const check = tryNodeExec(h, server,
    `k3s kubectl exec ${after} -- sh -c "cd /data && sha256sum -c blob.sha256"`, 15 * 60_000);
  if (check.exitCode !== 0) {
    throw new Error(`${opts.name}: DATA CORRUPTION — sha256sum -c failed: ${check.stdout}${check.stderr}`);
  }

  // ...and this proves the recorded digest is still the one we started with,
  // which -c alone would not catch if both file and checksum were replaced.
  const actual = kubectlOn(h, server,
    `exec ${after} -- sh -c "cd /data && sha256sum blob"`, 15 * 60_000).trim().split(/\s+/)[0];
  if (actual !== opts.expectDigest) {
    throw new Error(`${opts.name}: DIGEST CHANGED — expected ${opts.expectDigest}, got ${actual}`);
  }
  log(`[longhorn] ${opts.name} verified intact across a detach/attach (pod ${before} -> ${after})`);
}

export interface VolumeState {
  pv: string;
  state: string;
  robustness: string;
  wanted: number;
  replicaNodes: string[];
  healthyReplicas: number;
}

/** Longhorn's view of the volume backing a PVC. */
export function volumeState(h: Harness, server: ProvisionedVm, pvcName: string): VolumeState {
  const pv = kubectlOn(h, server, `get pvc ${pvcName} -o jsonpath='{.spec.volumeName}'`).trim();
  if (!pv) throw new Error(`PVC ${pvcName} has no bound volume`);

  const vol = kubectlJson<{
    spec: { numberOfReplicas?: number };
    status?: { state?: string; robustness?: string };
  }>(h, server, `-n ${LONGHORN_NS} get volumes.longhorn.io ${pv}`);

  const reps = kubectlJson<{ items: {
    spec: { nodeID?: string; volumeName?: string };
    status?: { currentState?: string; started?: boolean };
  }[] }>(h, server, `-n ${LONGHORN_NS} get replicas.longhorn.io`);

  const mine = reps.items.filter((r) => r.spec.volumeName === pv);
  return {
    pv,
    state: vol.status?.state ?? "unknown",
    robustness: vol.status?.robustness ?? "unknown",
    wanted: vol.spec.numberOfReplicas ?? 0,
    replicaNodes: mine.map((r) => r.spec.nodeID ?? "").filter(Boolean).sort(),
    healthyReplicas: mine.filter((r) => r.status?.currentState === "running").length,
  };
}

/**
 * Wait for a volume to be healthy on the expected number of distinct nodes.
 *
 * Logs rebuild progress while waiting, so a slow rebuild reads as progress
 * rather than as an opaque hour of silence.
 */
export async function waitForVolumeHealthy(h: Harness, server: ProvisionedVm, opts: {
  pvc: string; expectReplicas: number; timeoutMs?: number;
}): Promise<VolumeState> {
  const timeoutMs = opts.timeoutMs ?? 60 * 60_000;
  const start = Date.now();
  let last = "";
  while (Date.now() - start < timeoutMs) {
    let st: VolumeState | null = null;
    try { st = volumeState(h, server, opts.pvc); } catch { /* transient */ }

    if (st) {
      const distinct = new Set(st.replicaNodes).size;
      last = `${st.state}/${st.robustness} on ${distinct} node(s): ${st.replicaNodes.join(",")}`;
      if (st.robustness === "healthy" && distinct >= opts.expectReplicas) {
        log(`[longhorn] ${opts.pvc} healthy on ${distinct} nodes after ${Math.round((Date.now() - start) / 1000)}s`);
        return st;
      }
      const prog = tryNodeExec(h, server,
        `k3s kubectl -n ${LONGHORN_NS} get engines.longhorn.io -o jsonpath=` +
        `'{range .items[*]}{.spec.volumeName}{" "}{.status.rebuildStatus}{"\\n"}{end}'`);
      const line = prog.stdout.split("\n").find((l) => l.includes(st!.pv) && l.trim() !== st!.pv);
      log(`[longhorn] ${opts.pvc}: ${last}${line ? ` | rebuild ${line.replace(st.pv, "").trim()}` : ""}`);
    }
    await sleep(20_000);
  }
  throw new Error(`${opts.pvc} not healthy on ${opts.expectReplicas} nodes in ${timeoutMs / 60000} min (last: ${last})`);
}

/**
 * Wait until Longhorn reports the backup target usable.
 *
 * `available: true` is the appliance saying it could reach the endpoint AND
 * authenticate. Backing up without checking this first produces a failure that
 * looks like a volume problem.
 */
export async function setBackupTarget(
  h: Harness, server: ProvisionedVm, backup: BackupTargetOpts, timeoutMs = 5 * 60_000,
): Promise<void> {
  // PATCH the CR; do NOT set defaultSettings.backupTarget.
  //
  // In Longhorn 1.8 the backup target moved out of settings and onto
  // `backuptargets.longhorn.io/default`. The helm value is SILENTLY IGNORED --
  // no `backup-target` Setting object is created at all, the default
  // BackupTarget keeps an empty URL, and everything reports success. Observed
  // on a fresh 1.8.1 install here and confirmed against the live fleet.
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const exists = tryNodeExec(h, server,
      `k3s kubectl -n ${LONGHORN_NS} get backuptargets.longhorn.io default -o name`);
    if (exists.exitCode === 0 && exists.stdout.trim() !== "") break;
    await sleep(5_000);
  }

  const patch = JSON.stringify({
    spec: { backupTargetURL: backup.target, credentialSecret: BACKUP_SECRET, pollInterval: "5m0s" },
  });
  nodeExec(h, server,
    `k3s kubectl -n ${LONGHORN_NS} patch backuptargets.longhorn.io default --type=merge -p '${patch}'`);
  log(`[longhorn] backup target set to ${backup.target}`);
}

export async function waitForBackupTarget(
  h: Harness, server: ProvisionedVm, timeoutMs = 5 * 60_000,
): Promise<void> {
  const start = Date.now();
  let last = "";
  while (Date.now() - start < timeoutMs) {
    const r = tryNodeExec(h, server,
      `k3s kubectl -n ${LONGHORN_NS} get backuptargets.longhorn.io default ` +
      `-o jsonpath='{.spec.backupTargetURL}|{.status.available}|{.status.conditions[*].message}'`);
    last = r.stdout.trim();
    if (r.exitCode === 0 && last.includes("|true|")) {
      log(`[longhorn] backup target available after ${Math.round((Date.now() - start) / 1000)}s`);
      return;
    }
    await sleep(10_000);
  }
  throw new Error(
    `Backup target not available in ${timeoutMs / 60000} min (url|available|conditions = ${last}). ` +
    `Longhorn could not reach or authenticate to the endpoint.`);
}

/**
 * Snapshot a volume and back it up, returning the backup's name.
 *
 * Longhorn's own CRs, not the CSI VolumeSnapshot API -- this cluster has no
 * snapshot.storage.k8s.io resources at all, so VolumeSnapshot/VolumeSnapshotClass
 * simply do not exist here. Any backup tooling has to use these.
 */
export async function backupVolume(
  h: Harness, server: ProvisionedVm, pvc: string, timeoutMs = 20 * 60_000,
): Promise<{ backup: string; volume: string }> {
  const pv = volumeState(h, server, pvc).pv;
  const snapName = `drill-snap-${Date.now().toString(36)}`;

  log(`[longhorn] snapshotting ${pvc} (${pv})`);
  nodeExec(h, server, `k3s kubectl -n ${LONGHORN_NS} apply -f - <<'EOF'
apiVersion: longhorn.io/v1beta2
kind: Snapshot
metadata:
  name: ${snapName}
  namespace: ${LONGHORN_NS}
spec:
  volume: ${pv}
  createSnapshot: true
EOF`);

  // The snapshot has to exist before a backup can reference it.
  const snapDeadline = Date.now() + 5 * 60_000;
  while (Date.now() < snapDeadline) {
    const r = tryNodeExec(h, server,
      `k3s kubectl -n ${LONGHORN_NS} get snapshots.longhorn.io ${snapName} -o jsonpath='{.status.readyToUse}'`);
    if (r.stdout.trim() === "true") break;
    await sleep(5_000);
  }

  const backupName = `drill-bak-${Date.now().toString(36)}`;
  log(`[longhorn] backing up ${pvc} to the backup target`);
  nodeExec(h, server, `k3s kubectl -n ${LONGHORN_NS} apply -f - <<'EOF'
apiVersion: longhorn.io/v1beta2
kind: Backup
metadata:
  name: ${backupName}
  namespace: ${LONGHORN_NS}
  labels:
    backup-volume: ${pv}
spec:
  snapshotName: ${snapName}
EOF`);

  const start = Date.now();
  let last = "";
  while (Date.now() - start < timeoutMs) {
    const r = tryNodeExec(h, server,
      `k3s kubectl -n ${LONGHORN_NS} get backups.longhorn.io ${backupName} ` +
      `-o jsonpath='{.status.state}|{.status.progress}|{.status.error}'`);
    last = r.stdout.trim();
    if (last.startsWith("Completed")) {
      log(`[longhorn] backup ${backupName} completed after ${Math.round((Date.now() - start) / 1000)}s`);
      return { backup: backupName, volume: pv };
    }
    if (last.includes("Error")) {
      throw new Error(`Backup ${backupName} failed: ${last}`);
    }
    await sleep(10_000);
  }
  throw new Error(`Backup ${backupName} did not complete in ${timeoutMs / 60000} min (state|progress|error = ${last})`);
}

/**
 * Restore a backup into a brand-new PVC and workload.
 *
 * Uses the `fromBackup` StorageClass parameter, which is the supported route
 * when the CSI VolumeSnapshot API is unavailable. The restored volume is a
 * separate volume with its own replicas -- nothing about the original is
 * touched, so a failed restore cannot damage the source.
 */
export async function restoreFromBackup(h: Harness, server: ProvisionedVm, opts: {
  backup: string;
  name: string;
  sizeGi: number;
}): Promise<void> {
  const url = nodeExec(h, server,
    `k3s kubectl -n ${LONGHORN_NS} get backups.longhorn.io ${opts.backup} -o jsonpath='{.status.url}'`).trim();
  if (url === "") throw new Error(`Backup ${opts.backup} has no status.url; cannot restore from it`);
  log(`[longhorn] restoring ${opts.backup} into ${opts.name}`);

  nodeExec(h, server, `k3s kubectl apply -f - <<'EOF'
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: lh-${opts.name}
provisioner: driver.longhorn.io
allowVolumeExpansion: true
parameters:
  numberOfReplicas: "2"
  staleReplicaTimeout: "30"
  fsType: ext4
  fromBackup: "${url}"
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: ${opts.name}
spec:
  accessModes: [ReadWriteOnce]
  storageClassName: lh-${opts.name}
  resources:
    requests:
      storage: ${opts.sizeGi}Gi
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ${opts.name}
spec:
  replicas: 1
  strategy:
    type: Recreate
  selector:
    matchLabels: {app: ${opts.name}}
  template:
    metadata:
      labels: {app: ${opts.name}}
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        runAsGroup: 1000
        fsGroup: 1000
        seccompProfile: {type: RuntimeDefault}
      containers:
        - name: shell
          image: busybox:1.36
          command: ["sh", "-c", "sleep infinity"]
          securityContext:
            allowPrivilegeEscalation: false
            capabilities: {drop: ["ALL"]}
          volumeMounts:
            - {name: data, mountPath: /data}
      volumes:
        - name: data
          persistentVolumeClaim: {claimName: ${opts.name}}
EOF`);
}

/**
 * Read a blob's digest from a workload, without restarting it.
 *
 * Used on a RESTORED volume, where the point is to compare bytes against the
 * original's recorded digest rather than to prove durability again.
 */
export function readBlobDigest(h: Harness, server: ProvisionedVm, name: string): string {
  const pod = kubectlOn(h, server, `get pods -l app=${name} -o jsonpath='{.items[0].metadata.name}'`).trim();
  if (pod === "") throw new Error(`no pod for workload ${name}`);
  const out = kubectlOn(h, server,
    `exec ${pod} -- sh -c "cd /data && sha256sum blob"`, 15 * 60_000).trim();
  return out.split(/\s+/)[0] ?? "";
}