# `vmTests.<system>.swarm-archive`: runs the archive server's real units
# (../host-modules/swarm-archive.nix) and the hive's real backup script
# (../host-modules/lib/archive-backup.nix) in a VM, with restic and rclone, and
# checks what the module-eval suite can only read, on two nodes: `machine` keeps
# the repositories in a directory, `s3archive` in a bucket of a fake S3
# (`rclone serve s3`). On both:
#
# - a backup of two agents lands in one repository each, and `work/` is left out;
# - one agent's login cannot read another's repository;
# - a client cannot delete or overwrite anything, and the server-side prune
#   applies the retention, also when a client has forged a snapshot dated in
#   the future;
# - the users file follows the store: a user removed from it is refused without
#   a restart of the server.
#
# Only on `machine`: a hive that has agents and no grant backs up nothing and
# succeeds. Only on `s3archive`: the server does not start while the store
# holds no S3 key; the repositories are objects in the bucket and not on the
# host; the key is in rclone's environment and in no unit or file.
#
# The gateway hop is outside what is under test: the backup script dials
# `http://127.0.0.1:<port>` instead of `https://archive.<domain>`, and the swarm
# secret store is a stand-in `bao` that serves `kv list`, `kv get` and `login`
# from a directory. The module-eval suite covers the gateway vhost and the store
# grants.
{ pkgs }:
let
  port = 8911;
  store = "/var/lib/fake-bao";
  hive = "h1";
  agentRoot = "/var/lib/hyperhive/agents";

  # A directory is a leaf, a file in it a field: `kv get -field=<f> <path>`.
  fakeBao = pkgs.writeShellScriptBin "bao" ''
    store=${store}
    case "$1 $2" in
      "login -method=cert") echo faketoken; exit 0 ;;
      "kv list")
        path="$4"
        [ -d "$store/$path" ] || { echo "No value found at $path" >&2; exit 2; }
        ls "$store/$path" | ${pkgs.jq}/bin/jq -R . | ${pkgs.jq}/bin/jq -s .
        exit 0 ;;
      "kv put")
        [ -e "$store/$4" ] && { echo "check-and-set parameter did not match" >&2; exit 2; }
        mkdir -p "$store/$4"
        printf %s "''${5#*=}" > "$store/$4/''${5%%=*}"
        exit 0 ;;
      "kv get")
        field="''${3#-field=}"
        [ -f "$store/$4/$field" ] || { echo "No value found at $4" >&2; exit 2; }
        cat "$store/$4/$field"
        exit 0 ;;
    esac
    echo "fake bao: unsupported: $*" >&2
    exit 1
  '';

  # The stubs for the options the archive module reads from the rest of the
  # hive modules.
  hiveStubs =
    { lib, ... }:
    {
      options.services.hyperhive = {
        swarm.domain = lib.mkOption {
          type = lib.types.str;
          default = "t.local";
        };
        swarm.bao = {
          domain = lib.mkOption {
            type = lib.types.str;
            default = "bao.t.local";
          };
          port = lib.mkOption {
            type = lib.types.port;
            default = 8200;
          };
        };
        deploy.bao = {
          package = lib.mkOption {
            type = lib.types.package;
            default = fakeBao;
          };
          archiveClientCertFile = lib.mkOption {
            type = lib.types.nullOr lib.types.str;
            default = "/etc/fake-archive-cert";
          };
          archiveClientKeyFile = lib.mkOption {
            type = lib.types.nullOr lib.types.str;
            default = "/etc/fake-archive-key";
          };
          serverCaFile = lib.mkOption {
            type = lib.types.nullOr lib.types.str;
            default = null;
          };
        };
        deploy.swarm-otel.enable = lib.mkOption {
          type = lib.types.bool;
          default = false;
        };
        deploy.archive.enable = lib.mkOption {
          type = lib.types.bool;
          default = true;
        };
        otel.journalUnits = lib.mkOption {
          type = lib.types.listOf lib.types.str;
          default = [ ];
        };
        gateway = {
          enable = lib.mkOption {
            type = lib.types.bool;
            default = false;
          };
          dns.enable = lib.mkOption {
            type = lib.types.bool;
            default = false;
          };
          localNames = lib.mkOption {
            type = lib.types.listOf lib.types.str;
            default = [ ];
          };
          vhosts = lib.mkOption {
            type = lib.types.attrsOf lib.types.anything;
            default = { };
          };
        };
      };
      config.environment.etc."fake-archive-cert".text = "";
      config.environment.etc."fake-archive-key".text = "";
    };

  backupScript = import ../host-modules/lib/archive-backup.nix { inherit (pkgs) lib; } {
    inherit hive agentRoot;
    restic = pkgs.restic;
    url = "http://127.0.0.1:${toString port}";
    scope = "paths";
    paths = [
      "state"
      "claude"
      "harness"
    ];
    excludes = [ ];
    caBundle = null;
  };

  # The external S3 of the second node: `rclone serve s3` over a directory,
  # whose sub-directories are the buckets. The archive reaches it over HTTP
  # with a key the unit generates at start, as it would reach a provider. rclone
  # refuses short keys.
  s3Port = 9000;
  s3Root = "/var/lib/fake-s3";
  s3Bucket = "archive-bucket";
  s3KeyDir = "/var/lib/fake-s3-key";

  # What both nodes share: the archive module, the stubs for the hive options it
  # reads, and the hive's backup unit.
  archiveNode =
    extra:
    { pkgs, ... }:
    {
      imports = [
        hiveStubs
        ../host-modules/swarm-archive.nix
        extra
      ];

      # The newest snapshot, plus whatever is younger than three days.
      services.hyperhive.deploy.archive.retention = {
        keepWithin = "3d";
        keepLast = 1;
      };

      environment.systemPackages = [
        pkgs.curl
        pkgs.openssl
        pkgs.restic
      ];

      systemd.services.test-archive-backup = {
        path = [
          fakeBao
          pkgs.btrfs-progs
          pkgs.coreutils
          pkgs.gnugrep
          pkgs.jq
        ];
        serviceConfig = {
          Type = "oneshot";
          Environment = "RESTIC_CACHE_DIR=/var/cache/test-archive-backup";
        };
        script = backupScript;
      };
    };
in
pkgs.testers.runNixOSTest {
  name = "swarm-archive";

  nodes.machine = archiveNode { };

  nodes.s3archive = archiveNode {
    services.hyperhive.deploy.archive = {
      backend = "s3";
      s3 = {
        endpoint = "http://127.0.0.1:${toString s3Port}";
        bucket = s3Bucket;
        prefix = "pfx";
        region = "us-east-1";
      };
    };

    systemd.services.fake-s3 = {
      wantedBy = [ "multi-user.target" ];
      before = [ "swarm-archive.service" ];
      environment.RCLONE_CONFIG = "/dev/null";
      script = ''
        mkdir -p ${s3Root}/${s3Bucket}
        install -d -m 0700 ${s3KeyDir}
        if [ ! -e ${s3KeyDir}/id ]; then
          head -c 15 /dev/urandom | base32 | tr -d = > ${s3KeyDir}/id
          head -c 24 /dev/urandom | base32 | tr -d = > ${s3KeyDir}/secret
        fi
        exec ${pkgs.rclone}/bin/rclone serve s3 --addr 127.0.0.1:${toString s3Port} \
          --auth-key "$(cat ${s3KeyDir}/id),$(cat ${s3KeyDir}/secret)" ${s3Root}
      '';
    };
  };

  testScript = ''
    store = "${store}"
    url = "http://127.0.0.1:${toString port}"
    agents = ["alpha", "beta"]

    def sha_line(m, agent, password):
        return m.succeed(
            f"printf '%s:{{SHA}}%s' {agent} "
            f"\"$(printf %s {password} | openssl dgst -sha1 -binary | base64)\""
        )

    def put(m, path, field, value):
        m.succeed(f"mkdir -p {store}/{path} && printf %s '{value}' > {store}/{path}/{field}")

    def status(m, agent, repo, path="config", method="GET"):
        return m.succeed(
            f"curl -s -o /dev/null -w '%{{http_code}}' -X {method} "
            f"-u {agent}:restpw-{agent} {url}/{repo}/{path}"
        )

    def restic(agent, repo, args):
        return (
            f"RESTIC_PASSWORD=resticpw-{repo} restic "
            f"-r rest:http://{agent}:restpw-{agent}@127.0.0.1:${toString port}/{repo}/ {args}"
        )

    # What swarm-controller writes: a user line and a restic password per agent
    # for the archive, a REST password and the restic password for the hive.
    def grant_agents(m):
        for agent in agents:
            put(m, f"secret/swarm/services/archive/users/{agent}", "htpasswd", sha_line(m, agent, f"restpw-{agent}"))
            put(m, f"secret/swarm/services/archive/restic/{agent}", "password", f"resticpw-{agent}")
            put(m, f"secret/swarm/hives/${hive}/archive/{agent}", "rest-password", f"restpw-{agent}")
            put(m, f"secret/swarm/hives/${hive}/archive/{agent}", "restic-password", f"resticpw-{agent}")
            m.succeed(
                f"mkdir -p ${agentRoot}/{agent}/state ${agentRoot}/{agent}/claude "
                f"${agentRoot}/{agent}/harness ${agentRoot}/{agent}/work"
            )
            m.succeed(f"echo {agent}-state > ${agentRoot}/{agent}/state/data")
            m.succeed(f"echo {agent}-work > ${agentRoot}/{agent}/work/scratch")

    # What the archive promises, whichever backend holds the repositories.
    def archive_suite(m, tag):
        with subtest(f"{tag}: the server starts with no users and admits nobody"):
            m.succeed("test -e /run/swarm-archive-auth/htpasswd")
            assert status(m, "alpha", "alpha") == "401", status(m, "alpha", "alpha")

        grant_agents(m)

        with subtest(f"{tag}: the users file follows the store without a restart of the server"):
            pid = m.succeed("systemctl show -p MainPID --value swarm-archive").strip()
            m.succeed("systemctl start swarm-archive-users")
            assert m.succeed("systemctl show -p MainPID --value swarm-archive").strip() == pid
            assert status(m, "alpha", "alpha") in ("200", "404"), status(m, "alpha", "alpha")
            assert status(m, "beta", "beta") in ("200", "404"), status(m, "beta", "beta")
            # Wrong password: the control that the 200 above is a login, not an open door.
            m.succeed(f"test $(curl -s -o /dev/null -w '%{{http_code}}' -u alpha:wrong {url}/alpha/config) = 401")

        with subtest(f"{tag}: the users file holds hashes, readable by the server only"):
            m.succeed("grep -q '{SHA}' /run/swarm-archive-auth/htpasswd")
            m.fail("grep -q restpw /run/swarm-archive-auth/htpasswd")
            m.succeed("test \"$(stat -c '%a %U:%G' /run/swarm-archive-auth/htpasswd)\" = '640 root:swarm-archive'")

        with subtest(f"{tag}: the hive's backup puts each agent in its own repository, without work/"):
            m.succeed("systemctl start test-archive-backup")
            for agent in agents:
                snaps = m.succeed(restic(agent, agent, "snapshots --json"))
                assert snaps.count('"id"') == 1, snaps
                listing = m.succeed(restic(agent, agent, "ls latest"))
                assert f"/{agent}/state/data" in listing, listing
                assert "scratch" not in listing, listing

        with subtest(f"{tag}: one agent's login cannot reach another's repository"):
            assert status(m, "beta", "alpha") == "403", status(m, "beta", "alpha")
            m.fail(restic("beta", "alpha", "snapshots"))
            assert status(m, "alpha", "alpha") == "200", status(m, "alpha", "alpha")
            assert status(m, "alpha", "beta") == "403", status(m, "alpha", "beta")

        with subtest(f"{tag}: a client cannot delete or overwrite"):
            snap = m.succeed(restic("alpha", "alpha", "snapshots --json")).split('"id":"')[1].split('"')[0]
            # An old snapshot for the prune to drop. The backup is also the presence control: a lock, which a backup creates and removes, is deletable.
            m.succeed(restic("alpha", "alpha", "backup --host alpha --time '2000-01-01 00:00:00' ${agentRoot}/alpha/state"))
            assert status(m, "alpha", "alpha", f"snapshots/{snap}", "DELETE") == "403"
            assert status(m, "alpha", "alpha", "config", "POST") == "403"
            m.fail(restic("alpha", "alpha", "forget --keep-last 1 --prune"))
            assert m.succeed(restic("alpha", "alpha", "snapshots --json")).count('"id"') == 2

        with subtest(f"{tag}: the server-side prune applies the retention"):
            m.succeed("systemctl start swarm-archive-prune")
            for agent in agents:
                assert m.succeed(restic(agent, agent, "snapshots --json")).count('"id"') == 1
            # The old snapshot went, the one from the hive's backup stayed.
            assert snap in m.succeed(restic("alpha", "alpha", "snapshots --json"))

        with subtest(f"{tag}: a snapshot dated in the future cannot push the real ones out of the retention"):
            # restic counts --keep-within from the newest snapshot, and the client sets its time.
            m.succeed(restic("alpha", "alpha", "backup --host alpha --time '2100-01-01 00:00:00' ${agentRoot}/alpha/state"))
            assert m.succeed(restic("alpha", "alpha", "snapshots --json")).count('"id"') == 2
            m.succeed("systemctl start swarm-archive-prune")
            left = m.succeed(restic("alpha", "alpha", "snapshots --json"))
            assert left.count('"id"') == 1 and snap in left, left
            m.succeed("journalctl -u swarm-archive-prune | grep -q 'archive prune warning: alpha has snapshots dated after now'")
            # beta's repository had nothing forged, and says nothing.
            m.fail("journalctl -u swarm-archive-prune | grep -q 'archive prune warning: beta'")

        with subtest(f"{tag}: a user removed from the store is refused without a restart"):
            m.succeed(f"rm -r {store}/secret/swarm/services/archive/users/beta")
            m.succeed("systemctl start swarm-archive-users")
            assert status(m, "beta", "beta") == "401", status(m, "beta", "beta")
            assert status(m, "alpha", "alpha") == "200", status(m, "alpha", "alpha")

    start_all()
    machine.wait_for_unit("multi-user.target")
    machine.wait_for_open_port(${toString port})

    archive_suite(machine, "local")

    with subtest("the repositories are under the data directory"):
        machine.succeed("test -d /var/lib/swarm-archive/repos/alpha")

    with subtest("the archive announces itself once, so swarm-controller starts granting"):
        machine.succeed(f"test \"$(cat {store}/secret/swarm/services/archive/ready/ready)\" = 1")
        machine.succeed("systemctl start swarm-archive-users")

    with subtest("a hive with agents and no grant backs up nothing and does not fail"):
        machine.succeed(f"rm -r {store}/secret/swarm/hives/${hive}")
        machine.succeed("systemctl start test-archive-backup")
        machine.succeed("journalctl -u test-archive-backup | grep -q 'granted this hive no archive credentials'")
        machine.fail("journalctl -u test-archive-backup | grep -q 'archive backup failed'")

    # The same archive on the second node, with its repositories on the fake S3.
    s3archive.wait_for_unit("fake-s3.service")
    s3_key_id = s3archive.succeed("cat ${s3KeyDir}/id").strip()
    s3_secret_key = s3archive.succeed("cat ${s3KeyDir}/secret").strip()

    with subtest("s3: without the key in the store the server does not start"):
        s3archive.wait_until_succeeds("journalctl -u swarm-archive | grep -q 'No value found'")
        s3archive.fail("systemctl is-active --quiet swarm-archive")
        s3archive.fail(f"curl -sf --max-time 2 -o /dev/null {url}/")

    with subtest("s3: the server starts once the store holds the key"):
        put(s3archive, "secret/swarm/services/archive/s3/access-key-id", "value", s3_key_id)
        put(s3archive, "secret/swarm/services/archive/s3/secret-access-key", "value", s3_secret_key)
        s3archive.wait_for_open_port(${toString port})

    archive_suite(s3archive, "s3")

    with subtest("s3: the repositories are in the bucket and not on this host"):
        for agent in agents:
            s3archive.succeed(f"test -f ${s3Root}/${s3Bucket}/pfx/{agent}/config")
        s3archive.succeed("find ${s3Root}/${s3Bucket}/pfx/alpha/snapshots -type f | grep -q .")
        s3archive.succeed("find ${s3Root}/${s3Bucket}/pfx/alpha/data -type f | grep -q .")
        s3archive.fail("test -e /var/lib/swarm-archive/repos")

    with subtest("s3: the key is in rclone's environment and in no unit"):
        pid = s3archive.succeed("systemctl show -p MainPID --value swarm-archive").strip()
        env = s3archive.succeed(f"tr '\\0' '\\n' < /proc/{pid}/environ")
        assert f"RCLONE_S3_ACCESS_KEY_ID={s3_key_id}" in env, env
        assert f"RCLONE_S3_SECRET_ACCESS_KEY={s3_secret_key}" in env, env
        # The connection settings are in the unit text, the control that the search
        # for the key below looks at the units.
        units = "swarm-archive swarm-archive-prune swarm-archive-users"
        s3archive.succeed(f"systemctl cat {units} | grep -q RCLONE_S3_ENDPOINT")
        s3archive.fail(f"systemctl cat {units} | grep -q {s3_key_id}")
        s3archive.fail(f"systemctl cat {units} | grep -q {s3_secret_key}")
        wrapper = s3archive.succeed("systemctl show -p ExecStart --value swarm-archive | grep -o '/nix/store/[^ ;]*swarm-archive-serve' | head -1").strip()
        s3archive.succeed(f"grep -q 'kv get' {wrapper}")
        s3archive.fail(f"grep -q {s3_key_id} {wrapper}")
        s3archive.fail(f"grep -rq {s3_key_id} /run/swarm-archive-auth /var/cache/swarm-archive-prune")
  '';
}