//! `hivectl agent <name> subvol` — btrfs state-subvolume ops: migrate a
//! plain-dir agent state root to a subvolume (`upgrade`), and snapshot
//! create/delete/send.

use std::path::Path;

use anyhow::{Context as _, Result, bail};

use crate::cli::{SnapshotCmd, SubvolCmd};
use crate::dag_progress::wait_for_nodes;
use crate::util::{agent_exists, daemon_request};

/// A [`LifecycleScope`](hive_host_sock::LifecycleScope) targeting exactly one
/// agent by name (no infra containers, no all-agents flag).
fn single_agent_scope(name: &str) -> hive_host_sock::LifecycleScope {
    hive_host_sock::LifecycleScope {
        agents: false,
        agent_names: vec![name.to_owned()],
        ci: false,
        forge: false,
        gateway: false,
        matrix: false,
    }
}

/// `subvol upgrade <agent>` — migrate an existing plain-dir agent state root
/// to a btrfs subvolume. Composed client-side (like `restart`): stop the
/// agent so its state bind-mount is released, run the privileged in-place
/// migration via hive-priv, then restart it. The restart is attempted
/// regardless of the migration outcome so a failed migration never leaves
/// the agent down; the migration error (if any) is surfaced afterwards.
/// Route a `hivectl agent <name> subvol …` subcommand. Split out of
/// `main`'s top-level match so the CLI router stays within the clippy
/// line budget and the subvolume-op subcommands are dispatched in one
/// place. `name` is hoisted in from the parent `agent <name>` command.
pub(crate) async fn dispatch_subvol(socket: &Path, name: &str, cmd: SubvolCmd) -> Result<()> {
    match cmd {
        SubvolCmd::Upgrade { yes } => subvol_upgrade(socket, name, yes).await,
        SubvolCmd::Snapshot { cmd } => match cmd {
            SnapshotCmd::Create { label } => subvol_snapshot_create(socket, name, label).await,
            SnapshotCmd::Delete { label } => subvol_snapshot_delete(socket, name, &label).await,
            SnapshotCmd::Send {
                label,
                parent,
                dest,
            } => subvol_snapshot_send(socket, name, &label, parent.as_deref(), &dest).await,
            SnapshotCmd::Push { label, parent } => {
                subvol_snapshot_push(socket, name, &label, parent.as_deref()).await
            }
        },
    }
}

async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> {
    if !agent_exists(socket, name).await? {
        bail!("no agent named {name:?} (no state dir under the agents root)");
    }
    if !yes {
        bail!(
            "`subvol upgrade {name}` stops the agent, migrates its state dir to a btrfs \
             subvolume, then restarts it. Re-run with --yes to proceed."
        );
    }

    println!("stopping {name} (releasing its state bind-mount)…");
    let stop_resp = crate::client::request(
        socket,
        hive_host_sock::HostRequest::Stop {
            scope: single_agent_scope(name),
            graceful: false,
        },
    )
    .await
    .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !stop_resp.ok {
        bail!(
            "stop {name}: {}",
            stop_resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    // The stop is a queued job now — the migration below snapshots +
    // swaps the state dir and MUST NOT run under a live bind mount, so
    // wait for the stop to actually execute before touching anything.
    wait_for_nodes(socket, stop_resp.queued_dags.unwrap_or_default(), false)
        .await
        .with_context(|| format!("waiting for {name} to stop before the migration"))?;

    println!("migrating {name} state dir to a btrfs subvolume…");
    let upgrade = daemon_request(
        socket,
        hive_host_sock::HostRequest::UpgradeSubvolume {
            name: crate::util::parse_ident(name)?,
        },
        "upgrade",
    )
    .await;

    // Always attempt the restart, even if the migration failed — don't leave
    // the agent down. Capture the result rather than `?`-ing it so a
    // start-side failure (incl. the IPC call itself erroring) can't mask the
    // migration outcome below.
    println!("starting {name}…");
    let start_result = crate::client::request(
        socket,
        hive_host_sock::HostRequest::Start {
            scope: single_agent_scope(name),
        },
    )
    .await;

    // Surface the migration outcome FIRST — it's the meaningful result and
    // must not be shadowed by a restart-side failure. On migration failure the
    // original state dir is untouched (the priv op rolls back before the swap).
    upgrade.with_context(|| format!("upgrade {name} state subvolume"))?;

    // Migration succeeded; now surface any restart problem — either the IPC
    // call erroring, or the daemon reporting a failed start. The migration is
    // done regardless, so point at the manual recovery.
    let start_resp = start_result.with_context(|| {
        format!(
            "{name} migrated to a btrfs subvolume, but the restart request to the daemon \
             socket {} failed — run `hivectl start --agent {name}` to bring it back up",
            socket.display()
        )
    })?;
    if !start_resp.ok {
        bail!(
            "{name} migrated to a btrfs subvolume, but restarting it failed: {} — run \
             `hivectl start --agent {name}` to retry",
            start_resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    wait_for_nodes(socket, start_resp.queued_dags.unwrap_or_default(), false)
        .await
        .with_context(|| {
            format!(
                "{name} migrated to a btrfs subvolume, but its restart job failed — run \
                 `hivectl start --agent {name}` to retry"
            )
        })?;
    println!("upgraded {name} to a btrfs subvolume and restarted it");
    Ok(())
}

/// `subvol snapshot create <agent> --label <label>` — create a read-only
/// btrfs snapshot of an agent's state subvolume. Unlike `upgrade`, this does
/// NOT stop the agent: btrfs snapshots are atomic + consistent to take
/// against a live subvolume. `label` is mandatory, must start with
/// `hive-`, and is otherwise restricted to `[A-Za-z0-9_-]` (no `.` at all
/// — mara: "we are making up the rules here, lets go strict"). hive-priv
/// enforces the same rules server-side, so this check is
/// belt-and-suspenders (fail fast client-side with a clear message).
async fn subvol_snapshot_create(socket: &Path, name: &str, label: String) -> Result<()> {
    if !agent_exists(socket, name).await? {
        bail!("no agent named {name:?} (no state dir under the agents root)");
    }
    if !label.starts_with("hive-") {
        bail!("snapshot label {label:?} must start with \"hive-\"");
    }
    if !label
        .bytes()
        .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-'))
    {
        bail!(
            "snapshot label {label:?} must be [A-Za-z0-9_-] only (no \".\" — hive-priv rejects it)"
        );
    }
    // The daemon returns the snapshot's host path as a message line, which
    // `daemon_request` prints — same bare-path output as before.
    daemon_request(
        socket,
        hive_host_sock::HostRequest::SnapshotSubvolume {
            name: crate::util::parse_ident(name)?,
            label,
        },
        "snapshot",
    )
    .await
}

/// `subvol snapshot delete <agent> <label>` — remove a snapshot created by
/// `subvol snapshot create`.
async fn subvol_snapshot_delete(socket: &Path, name: &str, label: &str) -> Result<()> {
    daemon_request(
        socket,
        hive_host_sock::HostRequest::DeleteSnapshot {
            name: crate::util::parse_ident(name)?,
            label: label.to_owned(),
        },
        "snapshot delete",
    )
    .await?;
    println!("deleted snapshot {label:?} for {name}");
    Ok(())
}

/// `subvol snapshot send <agent> <label> [--parent <label>] --dest <file>` —
/// export a snapshot to a local file via `btrfs send`. `dest` must be a
/// bare filename (hive-priv rejects anything else via the credential-name
/// charset check, which excludes `/`), written under the migrate-staging
/// dir. Prints the resulting file's host path.
async fn subvol_snapshot_send(
    socket: &Path,
    name: &str,
    label: &str,
    parent: Option<&str>,
    dest: &str,
) -> Result<()> {
    // The daemon returns the written file's host path as a message line,
    // which `daemon_request` prints — same bare-path output as before.
    daemon_request(
        socket,
        hive_host_sock::HostRequest::SendSnapshot {
            name: crate::util::parse_ident(name)?,
            label: label.to_owned(),
            parent: parent.map(str::to_owned),
            dest: dest.to_owned(),
        },
        "snapshot send",
    )
    .await
}

/// `subvol snapshot push <agent> <label> [--parent <label>]` — stream a
/// snapshot to the swarm's snapshot store over the mesh.
///
/// The network sibling of `send`. Nothing is staged on this host, so
/// there is no path to print: success is silent apart from the
/// confirmation below. No destination argument — a swarm has one store,
/// and the daemon reads its address from the host config.
async fn subvol_snapshot_push(
    socket: &Path,
    name: &str,
    label: &str,
    parent: Option<&str>,
) -> Result<()> {
    daemon_request(
        socket,
        hive_host_sock::HostRequest::PushSnapshot {
            name: crate::util::parse_ident(name)?,
            label: label.to_owned(),
            parent: parent.map(str::to_owned),
        },
        "snapshot push",
    )
    .await?;
    match parent {
        Some(p) => {
            println!(
                "pushed {name} snapshot {label:?} (incremental from {p:?}) to the swarm store"
            );
        }
        None => println!("pushed {name} snapshot {label:?} (full) to the swarm store"),
    }
    Ok(())
}