//! Async client for the `hive-priv` privileged-helper socket.
//!
//! Exposes a standalone async function per operation. Each call opens a
//! fresh connection to `/run/hive/priv.sock`, sends one JSON line, reads
//! the response, and closes. Connection-per-call is intentional: priv
//! calls are infrequent (once per rebuild step), so simplicity wins over
//! a persistent connection.

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{
    BindMount, CredentialMount, HiveFacts, InfraAction, InfraContainer, MEASURE_AGENT_DISK_TIMEOUT,
    NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream,
};
use hive_types::NixBuilds;
use std::os::fd::{AsRawFd as _, OwnedFd, RawFd};
use std::path::Path;

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Interest};
use tokio::net::UnixStream;

/// Send a single request to `hive-priv` and return the response.
/// For streaming ops use `call_streaming` instead.
pub async fn call(req: &PrivRequest) -> Result<PrivResponse> {
    call_at(Path::new(PRIV_SOCK), req).await
}

async fn call_at(sock: &Path, req: &PrivRequest) -> Result<PrivResponse> {
    let mut stream = UnixStream::connect(sock)
        .await
        .context("connect to hive-priv socket")?;
    let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
    stream
        .write_all(line.as_bytes())
        .await
        .context("send request to hive-priv")?;
    stream.shutdown().await.context("shutdown write half")?;
    let mut resp_line = String::new();
    BufReader::new(stream)
        .read_line(&mut resp_line)
        .await
        .context("read response from hive-priv")?;
    // New hive-priv sends `PrivEvent::Done(PrivResponse)` (untagged, wire-
    // identical to bare `PrivResponse`) — deserialise as `PrivEvent` to
    // handle both the old bare format and the new tagged format.
    match serde_json::from_str::<PrivEvent>(&resp_line).context("parse PrivResponse")? {
        PrivEvent::Done(resp) => Ok(resp),
        PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"),
    }
}

/// Ancillary-data buffer sized and aligned for one `SCM_RIGHTS`
/// message. `CMSG_SPACE` is not a `const fn`, so the size is a literal
/// with room to spare; the union member supplies the `cmsghdr`
/// alignment `CMSG_FIRSTHDR` requires.
#[repr(C)]
union CmsgSpace {
    _align: libc::cmsghdr,
    bytes: [u8; 32],
}

/// `sendmsg` `bytes` with `fd` attached as `SCM_RIGHTS`, returning how
/// many bytes were accepted.
///
/// The descriptor rides on this one call — ancillary data cannot be
/// sent separately from payload — so the caller must not have written
/// any of `bytes` beforehand.
fn send_with_fd(sock: RawFd, bytes: &[u8], fd: RawFd) -> std::io::Result<usize> {
    const FD_SIZE: usize = std::mem::size_of::<RawFd>();

    let mut iov = libc::iovec {
        iov_base: bytes.as_ptr().cast::<libc::c_void>().cast_mut(),
        iov_len: bytes.len(),
    };
    let mut cmsg = CmsgSpace { bytes: [0; 32] };
    // SAFETY: msghdr is a plain C struct with no invalid bit patterns;
    // every field we rely on is set immediately below.
    let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
    msg.msg_iov = &raw mut iov;
    msg.msg_iovlen = 1;
    msg.msg_control = std::ptr::addr_of_mut!(cmsg.bytes).cast();
    // SAFETY: CMSG_SPACE is a pure size computation.
    let space = unsafe { libc::CMSG_SPACE(u32::try_from(FD_SIZE).unwrap_or(4)) };
    msg.msg_controllen = space as _;

    // SAFETY: the control buffer is live, aligned, and long enough for
    // the header CMSG_SPACE just sized.
    let hdr = unsafe { libc::CMSG_FIRSTHDR(&raw const msg) };
    if hdr.is_null() {
        return Err(std::io::Error::other(
            "control buffer too small for SCM_RIGHTS",
        ));
    }
    // SAFETY: CMSG_LEN is a pure size computation; `hdr` points into
    // our own buffer, and write_unaligned tolerates its alignment.
    unsafe {
        let len = libc::CMSG_LEN(u32::try_from(FD_SIZE).unwrap_or(4));
        std::ptr::write_unaligned(
            hdr,
            libc::cmsghdr {
                cmsg_len: len as _,
                cmsg_level: libc::SOL_SOCKET,
                cmsg_type: libc::SCM_RIGHTS,
            },
        );
        // Copied in byte-wise: the control buffer is only cmsghdr-
        // aligned, so casting CMSG_DATA to a *mut RawFd would be
        // unsound even where it happens to work.
        std::ptr::copy_nonoverlapping(
            std::ptr::from_ref(&fd).cast::<u8>(),
            libc::CMSG_DATA(hdr),
            FD_SIZE,
        );
    }

    // SAFETY: `msg` points at a live iovec over `bytes` and the control
    // buffer we just filled in.
    let n = unsafe { libc::sendmsg(sock, &raw const msg, 0) };
    if n < 0 {
        return Err(std::io::Error::last_os_error());
    }
    Ok(usize::try_from(n).unwrap_or_default())
}

/// Send a request to `hive-priv` with an open file descriptor attached,
/// and return the response.
///
/// The helper receives the descriptor itself — not a path or an address
/// — so it can act on something it was handed without being told what
/// that thing is or how to reach it. Used for
/// [`PrivRequest::SendAgentSnapshotToFd`], where the descriptor is a
/// socket already connected to a peer hive's snapshot store.
///
/// ⚠️ Takes the descriptor by value and closes it as soon as the kernel
/// has it, *before* awaiting the response. That is not tidiness: a
/// socket stays open until every copy of it is closed, so a caller
/// holding one back would leave the receiving end waiting for an EOF
/// that never comes — `btrfs receive` blocks, and this side reports
/// success for a transfer the peer has not committed. Passing ownership
/// makes that mistake unrepresentable.
///
/// # Errors
///
/// Fails if the socket is unreachable, the descriptor cannot be
/// attached, or hive-priv answers with something other than a terminal
/// event.
pub async fn call_with_fd(req: &PrivRequest, fd: OwnedFd) -> Result<PrivResponse> {
    call_with_fd_at(Path::new(PRIV_SOCK), req, fd).await
}

async fn call_with_fd_at(path: &Path, req: &PrivRequest, fd: OwnedFd) -> Result<PrivResponse> {
    let mut stream = UnixStream::connect(path)
        .await
        .context("connect to hive-priv socket (fd-passing)")?;
    let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
    let bytes = line.as_bytes();

    let sock = stream.as_raw_fd();
    let raw_fd = fd.as_raw_fd();
    let sent = stream
        .async_io(Interest::WRITABLE, || send_with_fd(sock, bytes, raw_fd))
        .await
        .context("send request + descriptor to hive-priv")?;
    // The kernel has duplicated the descriptor into hive-priv's queue,
    // so our copy has done its job. Close it now, before waiting on the
    // response: see the EOF note on this function.
    drop(fd);

    // A short sendmsg is legal; the descriptor went with the first
    // call, so the tail is an ordinary write.
    if sent < bytes.len() {
        stream
            .write_all(&bytes[sent..])
            .await
            .context("send remainder of request to hive-priv")?;
    }
    stream.shutdown().await.context("shutdown write half")?;

    let mut resp_line = String::new();
    BufReader::new(stream)
        .read_line(&mut resp_line)
        .await
        .context("read response from hive-priv")?;
    match serde_json::from_str::<PrivEvent>(&resp_line).context("parse PrivResponse")? {
        PrivEvent::Done(resp) => Ok(resp),
        PrivEvent::Line(_) => bail!("unexpected stream line from non-streaming priv op"),
    }
}

/// Send a streaming request to `hive-priv`, calling `on_line` for each
/// `PrivEvent::Line` as it arrives, then returning the terminal
/// `PrivResponse`. Used for long-running ops (`create` / `update`).
pub async fn call_streaming(
    req: &PrivRequest,
    on_line: impl FnMut(PrivStream, &str),
) -> Result<PrivResponse> {
    call_streaming_at(Path::new(PRIV_SOCK), req, on_line).await
}

async fn call_streaming_at(
    path: &Path,
    req: &PrivRequest,
    mut on_line: impl FnMut(PrivStream, &str),
) -> Result<PrivResponse> {
    let mut stream = UnixStream::connect(path)
        .await
        .context("connect to hive-priv socket (streaming)")?;
    let line = serde_json::to_string(req).context("serialise PrivRequest")? + "\n";
    stream
        .write_all(line.as_bytes())
        .await
        .context("send request to hive-priv")?;
    stream.shutdown().await.context("shutdown write half")?;
    let mut reader = BufReader::new(stream);
    loop {
        let mut event_line = String::new();
        reader
            .read_line(&mut event_line)
            .await
            .context("read event from hive-priv")?;
        if event_line.is_empty() {
            bail!("hive-priv closed connection before sending Done event");
        }
        match serde_json::from_str::<PrivEvent>(&event_line).context("parse PrivEvent")? {
            PrivEvent::Line(l) => on_line(l.stream, &l.data),
            PrivEvent::Done(resp) => return Ok(resp),
        }
    }
}

pub async fn start_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::StartContainer {
        name: name.to_owned(),
    })
    .await?)
}

/// The container unit's journal lines since `since_unix`. An older hive-priv
/// answers an error (it does not know the op).
pub async fn start_failure_log(name: &str, since_unix: u64) -> Result<String> {
    let (stdout, _) = check(
        call(&PrivRequest::StartFailureLog {
            name: name.to_owned(),
            since_unix,
        })
        .await?,
    )?;
    Ok(stdout)
}

pub async fn stop_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::StopContainer {
        name: name.to_owned(),
    })
    .await?)
}

pub async fn kill_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::KillContainer {
        name: name.to_owned(),
    })
    .await?)
}

/// Streaming variant: forward stdout/stderr lines to `on_line` as they
/// arrive. Returns `Ok(())` on success; the callback is responsible for
/// appending lines to `build_logs` or otherwise capturing the output.
pub async fn update_container_streaming(
    name: &str,
    toplevel: &str,
    mut on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
    let req = update_container_request(name, toplevel);
    let first = ok(call_streaming(&req, &mut on_line).await?);
    if !older_hive_priv_refused(&first) {
        return first;
    }
    ask_older_hive_priv_to_download(name, toplevel).await?;
    ok(call_streaming(&req, &mut on_line).await?)
}

fn update_container_request(name: &str, toplevel: &str) -> PrivRequest {
    PrivRequest::UpdateContainer {
        name: name.to_owned(),
        stream: true,
        toplevel: toplevel.to_owned(),
    }
}

/// Streaming variant: forward stdout/stderr lines to `on_line` as they
/// arrive. Returns `Ok(())` on success.
pub async fn create_container_streaming(
    name: &str,
    toplevel: &str,
    mut on_line: impl FnMut(PrivStream, &str),
) -> Result<()> {
    let req = create_container_request(name, toplevel);
    let first = ok(call_streaming(&req, &mut on_line).await?);
    if !older_hive_priv_refused(&first) {
        return first;
    }
    ask_older_hive_priv_to_download(name, toplevel).await?;
    ok(call_streaming(&req, &mut on_line).await?)
}

fn create_container_request(name: &str, toplevel: &str) -> PrivRequest {
    PrivRequest::CreateContainer {
        name: name.to_owned(),
        stream: true,
        toplevel: toplevel.to_owned(),
    }
}

/// What an older hive-priv, which applies only a path it downloaded itself,
/// says to an apply of any other.
const OLDER_PIN_REFUSAL: &str = "is not the closure realised for this agent";

/// Whether `applied` is an older hive-priv refusing an apply because it did not
/// download the path itself (hive-c0re pulled it through the nix daemon). A
/// hive-priv of the same release never refuses that way, so a retry through
/// [`ask_older_hive_priv_to_download`] happens only while the two restart out of
/// step.
fn older_hive_priv_refused(applied: &Result<()>) -> bool {
    applied
        .as_ref()
        .is_err_and(|e| format!("{e:#}").contains(OLDER_PIN_REFUSAL))
}

/// Ask an older hive-priv to download `toplevel`, as it did before, which also
/// pins the path for the apply that follows. The path is already in the store,
/// so this is a lookup.
async fn ask_older_hive_priv_to_download(name: &str, toplevel: &str) -> Result<()> {
    tracing::warn!(%name, %toplevel, "an older hive-priv refused a closure it did not download; asking it to");
    ok(call(&PrivRequest::RealiseClosure {
        agent: name.to_owned(),
        store_path: toplevel.to_owned(),
    })
    .await?)
    .context("asking the older hive-priv to download the closure")
}

/// How long hive-c0re waits for hive-priv's facts refresh: hive-priv bounds
/// the unit at 60 s.
const REFRESH_WAIT: std::time::Duration = std::time::Duration::from_secs(90);

/// Run the root unit that reads this hive's cache facts from the store and
/// writes the nix.conf fragment, and wait for it. An older hive-priv does not
/// know the op and answers an error.
///
/// # Errors
/// The call failed or did not answer in time, or the unit failed.
pub async fn refresh_cache_facts() -> Result<()> {
    within(REFRESH_WAIT, async {
        ok(call(&PrivRequest::RefreshCacheFacts).await?)
    })
    .await
}

pub async fn destroy_container(name: &str) -> Result<()> {
    ok(call(&PrivRequest::DestroyContainer {
        name: name.to_owned(),
    })
    .await?)
}

/// How long hive-c0re waits for hive-priv's `nixos-container list`. Periodic
/// workers await the listing inline, so a hung one must reach them as an
/// error they skip a cycle on, not as a wait that stops their loop.
const LIST_WAIT: std::time::Duration = std::time::Duration::from_secs(30);

/// # Errors
/// Returns an error if the hive-priv call fails or doesn't answer within
/// [`LIST_WAIT`], or `nixos-container list` exits non-zero.
pub async fn list_containers() -> Result<String> {
    let (stdout, _) = check(within(LIST_WAIT, call(&PrivRequest::ListContainers)).await?)?;
    Ok(stdout)
}

pub async fn write_nspawn_flags(
    container: &str,
    binds: &[BindMount],
    isolation: NetworkIsolation,
    load_credentials: &[CredentialMount],
    hive_facts: HiveFacts,
    private_users: Option<bool>,
    nix_builds: Option<NixBuilds>,
) -> Result<()> {
    ok(call(&write_nspawn_flags_request(
        container,
        binds,
        isolation,
        load_credentials,
        hive_facts,
        private_users,
        nix_builds,
    ))
    .await?)
}

fn write_nspawn_flags_request(
    container: &str,
    binds: &[BindMount],
    isolation: NetworkIsolation,
    load_credentials: &[CredentialMount],
    hive_facts: HiveFacts,
    private_users: Option<bool>,
    nix_builds: Option<NixBuilds>,
) -> PrivRequest {
    PrivRequest::WriteNspawnFlags {
        container: container.to_owned(),
        binds: binds.to_vec(),
        isolation,
        load_credentials: load_credentials.to_vec(),
        hive_facts: Box::new(hive_facts),
        private_users,
        nix_builds,
    }
}

/// See [`PrivRequest::EnsureAgentSocketDir`].
pub async fn ensure_agent_socket_dir(name: &str, private_users: Option<bool>) -> Result<()> {
    ok(call(&PrivRequest::EnsureAgentSocketDir {
        name: name.to_owned(),
        private_users,
    })
    .await?)
}

pub async fn write_resource_limits(
    container: &str,
    memory_max: &str,
    cpu_quota: &str,
    cpu_weight: Option<u32>,
    io_weight: Option<u32>,
) -> Result<()> {
    ok(call(&resource_limits_request(
        container, memory_max, cpu_quota, cpu_weight, io_weight,
    ))
    .await?)
}

fn resource_limits_request(
    container: &str,
    memory_max: &str,
    cpu_quota: &str,
    cpu_weight: Option<u32>,
    io_weight: Option<u32>,
) -> PrivRequest {
    PrivRequest::WriteResourceLimits {
        container: container.to_owned(),
        memory_max: memory_max.to_owned(),
        cpu_quota: cpu_quota.to_owned(),
        cpu_weight,
        io_weight,
    }
}

pub async fn remove_service_dropin(container: &str) -> Result<()> {
    ok(call(&PrivRequest::RemoveServiceDropin {
        container: container.to_owned(),
    })
    .await?)
}

pub async fn daemon_reload() -> Result<()> {
    ok(call(&PrivRequest::DaemonReload).await?)
}

pub async fn reload_gateway_nginx() -> Result<()> {
    ok(call(&PrivRequest::ReloadGatewayNginx).await?)
}

/// Create (`paused: true`) or remove (`paused: false`) the pause marker in
/// `agent_name`'s harness dir via hive-priv (running as root).
///
/// hive-c0re cannot do this itself: the harness dir is chowned to the agent
/// user on the container's first boot and left mode 0755, so this process
/// can stat the marker (that's what `Coordinator::is_paused` does) but gets
/// `EACCES` on create *and* unlink. Both directions are idempotent.
pub async fn set_agent_paused(agent_name: &str, paused: bool) -> Result<()> {
    ok(call(&PrivRequest::SetAgentPaused {
        agent_name: agent_name.to_owned(),
        paused,
    })
    .await?)
}

/// Start / stop a hive infrastructure service (`hive-gateway`) on the host
/// via `systemctl <action> <unit>`, where the unit is derived root-side from
/// the variant (`nginx.service` for the gateway). The [`InfraContainer`]
/// enum is the allowlist — hive-priv needs no name re-validation. Used by
/// the hive-wide `hivectl stop` / `start` / `restart` flow. No agent-facing
/// path exists.
pub async fn control_infra_container(container: InfraContainer, action: InfraAction) -> Result<()> {
    ok(call(&PrivRequest::ControlInfraContainer { container, action }).await?)
}

/// Ensure the agent's persistent state root is a btrfs subvolume when the
/// host FS supports it (via hive-priv, which runs as root). Idempotent and
/// progressive: a no-op when the root already exists or the FS isn't btrfs.
/// Safe to call on every provision.
pub async fn ensure_agent_subvolume(agent_name: &str) -> Result<()> {
    ok(call(&PrivRequest::EnsureAgentSubvolume {
        agent_name: agent_name.to_owned(),
    })
    .await?)
}

/// Delete the agent's state root iff it is a btrfs subvolume (purge path
/// only). No-op for plain dirs / missing paths — hive-c0re's own
/// `remove_dir_all` handles those.
pub async fn delete_agent_subvolume(agent_name: &str) -> Result<()> {
    ok(call(&PrivRequest::DeleteAgentSubvolume {
        agent_name: agent_name.to_owned(),
    })
    .await?)
}

/// Enable btrfs qgroup accounting on the agent-state filesystem (operator
/// opt-in; prerequisite for usage reads + quotas). Idempotent; no-op off
/// btrfs. Via hive-priv (root).
///
/// # Errors
/// Returns an error if the hive-priv call fails or `btrfs quota enable`
/// reports a non-zero exit.
pub async fn ensure_btrfs_quota() -> Result<()> {
    ok(call(&PrivRequest::EnsureBtrfsQuota).await?)
}

/// Read an agent state subvolume's btrfs qgroup usage, returning
/// `(referenced_bytes, exclusive_bytes)`. Via hive-priv (root).
///
/// # Errors
/// Returns an error if the hive-priv call fails, `btrfs qgroup show` exits
/// non-zero (e.g. quota not enabled — the message propagates so the caller
/// can surface it), or the output has no level-0 qgroup row to parse.
pub async fn read_subvolume_usage(agent_name: &str) -> Result<(u64, u64)> {
    let (stdout, _) = check(
        call(&PrivRequest::ReadSubvolumeUsage {
            agent_name: agent_name.to_owned(),
        })
        .await?,
    )?;
    parse_qgroup_usage(&stdout).with_context(|| format!("parse qgroup usage for {agent_name}"))
}

/// How long hive-c0re waits for hive-priv's answer to a disk measurement:
/// hive-priv's own bound plus a margin, so a hung hive-priv can't stall the
/// caller while a slow walk still arrives as hive-priv's timeout error.
const MEASURE_AGENT_DISK_WAIT: std::time::Duration =
    MEASURE_AGENT_DISK_TIMEOUT.saturating_add(std::time::Duration::from_mins(1));

/// Measure an agent's disk usage in bytes (`state/` + `work/` + container
/// rootfs, one bounded `du` run by hive-priv as root).
///
/// # Errors
/// Returns an error if the hive-priv call fails or doesn't answer within
/// [`MEASURE_AGENT_DISK_WAIT`], the `du` walk times out or exits non-zero,
/// the agent has none of the paths, or the returned total isn't a number.
pub async fn measure_agent_disk(agent_name: &str) -> Result<u64> {
    let resp = within(
        MEASURE_AGENT_DISK_WAIT,
        call(&PrivRequest::MeasureAgentDisk {
            agent_name: agent_name.to_owned(),
        }),
    )
    .await?;
    let (stdout, _) = check(resp)?;
    stdout
        .trim()
        .parse()
        .with_context(|| format!("parse disk usage for {agent_name}: {stdout:?}"))
}

/// Await `fut`, failing if it takes longer than `limit`.
async fn within<T>(
    limit: std::time::Duration,
    fut: impl std::future::Future<Output = Result<T>>,
) -> Result<T> {
    tokio::time::timeout(limit, fut)
        .await
        .map_err(|_| anyhow::anyhow!("hive-priv did not answer within {}s", limit.as_secs()))?
}

/// Set or clear a btrfs qgroup size limit on an agent's state subvolume.
/// `limit_bytes = None` clears it. Requires quota enabled. Via hive-priv.
///
/// # Errors
/// Returns an error if the hive-priv call fails or `btrfs qgroup limit`
/// reports a non-zero exit (e.g. quota not enabled).
pub async fn set_subvolume_quota(agent_name: &str, limit_bytes: Option<u64>) -> Result<()> {
    ok(call(&PrivRequest::SetSubvolumeQuota {
        agent_name: agent_name.to_owned(),
        limit_bytes,
    })
    .await?)
}

/// Convert an existing plain-dir agent state root into a btrfs subvolume in
/// place (operator opt-in; via hive-priv as root). The caller must stop the
/// agent first (so its state bind-mount is gone) and restart it after.
/// Idempotent: a no-op when the root is already a subvolume.
///
/// # Errors
/// Returns an error if the hive-priv call fails, the state dir is missing,
/// the FS isn't btrfs, or the migration (subvolume create / copy / swap)
/// fails — in which case the original dir is left untouched.
pub async fn upgrade_agent_subvolume(agent_name: &str) -> Result<()> {
    ok(call(&PrivRequest::UpgradeAgentSubvolume {
        agent_name: agent_name.to_owned(),
    })
    .await?)
}

/// Create a read-only btrfs snapshot of an agent's state subvolume (via
/// hive-priv as root) — the first step of `hivectl migrate`'s send/receive
/// path. Returns the snapshot's absolute host path. Fails if the agent's
/// state root isn't a subvolume yet, or a snapshot with the same
/// `snapshot_name` already exists.
///
/// # Errors
/// Returns an error if the hive-priv call fails, the state dir isn't a
/// btrfs subvolume, or the snapshot already exists.
pub async fn snapshot_agent_subvolume(agent_name: &str, snapshot_name: &str) -> Result<String> {
    let (stdout, _) = check(
        call(&PrivRequest::SnapshotAgentSubvolume {
            agent_name: agent_name.to_owned(),
            snapshot_name: snapshot_name.to_owned(),
        })
        .await?,
    )?;
    Ok(stdout)
}

/// Delete a previously-created read-only agent-state snapshot (cleanup
/// counterpart to [`snapshot_agent_subvolume`]). No-op if the snapshot
/// doesn't exist. Via hive-priv as root.
///
/// # Errors
/// Returns an error if the hive-priv call fails or the underlying
/// `btrfs subvolume delete` fails.
pub async fn delete_agent_snapshot(agent_name: &str, snapshot_name: &str) -> Result<()> {
    ok(call(&PrivRequest::DeleteAgentSnapshot {
        agent_name: agent_name.to_owned(),
        snapshot_name: snapshot_name.to_owned(),
    })
    .await?)
}

/// Stream a read-only agent snapshot to a local file via `btrfs send`
/// (optionally incremental against `parent_snapshot_name`). Returns the
/// full path of the written file under `MIGRATE_STAGING_ROOT`. Via
/// hive-priv as root. See [`PrivRequest::SendAgentSnapshotToFile`].
///
/// # Errors
/// Returns an error if the hive-priv call fails, the snapshot (or parent)
/// doesn't exist, or the destination file already exists.
pub async fn send_agent_snapshot_to_file(
    agent_name: &str,
    snapshot_name: &str,
    parent_snapshot_name: Option<&str>,
    dest_file_name: &str,
) -> Result<String> {
    let req = send_snapshot_request(
        agent_name,
        snapshot_name,
        parent_snapshot_name,
        dest_file_name,
    );
    let (stdout, _) = check(call(&req).await?)?;
    Ok(stdout)
}

fn send_snapshot_request(
    agent_name: &str,
    snapshot_name: &str,
    parent_snapshot_name: Option<&str>,
    dest_file_name: &str,
) -> PrivRequest {
    PrivRequest::SendAgentSnapshotToFile {
        agent_name: agent_name.to_owned(),
        snapshot_name: snapshot_name.to_owned(),
        parent_snapshot_name: parent_snapshot_name.map(str::to_owned),
        dest_file_name: dest_file_name.to_owned(),
    }
}

/// Parse `(referenced, exclusive)` bytes from `btrfs qgroup show -f --raw`
/// output (a qgroup row is `<id-with-slash> <rfer> <excl> …`).
///
/// `-f <path>` already restricts the listing to qgroups impacting that path
/// (excluding ancestral qgroups — see btrfs-qgroup-show(8)), so it never
/// mixes in other agents' subvolumes. Among the rows it returns we select
/// the **level-0** qgroup (`0/<subvolid>`) — the subvolume's own automatic
/// usage qgroup — rather than blindly taking the last line. Picking the
/// `0/` leaf is unambiguous even if an operator has assigned the subvolume
/// to a higher-level aggregate qgroup (`1/<id>`, …) that `-F` would surface.
fn parse_qgroup_usage(out: &str) -> Result<(u64, u64)> {
    for line in out.lines() {
        let cols: Vec<&str> = line.split_whitespace().collect();
        if cols.len() >= 3
            && cols[0].starts_with("0/")
            && let (Ok(rfer), Ok(excl)) = (cols[1].parse::<u64>(), cols[2].parse::<u64>())
        {
            return Ok((rfer, excl));
        }
    }
    bail!("no level-0 qgroup data row in `btrfs qgroup show` output: {out:?}")
}

fn check(resp: PrivResponse) -> Result<(String, String)> {
    if resp.ok {
        Ok((resp.stdout, resp.stderr))
    } else {
        bail!(
            "{}",
            resp.error.as_deref().unwrap_or("hive-priv returned error")
        )
    }
}

fn ok(resp: PrivResponse) -> Result<()> {
    check(resp)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{
        BindMount, CredentialMount, HiveFacts, NetworkIsolation, NixBuilds, PrivRequest,
        PrivResponse, PrivStream, call_at, call_streaming_at, call_with_fd_at, check,
        create_container_request, older_hive_priv_refused, parse_qgroup_usage,
        resource_limits_request, send_snapshot_request, update_container_request, within,
        write_nspawn_flags_request,
    };
    use anyhow::Context as _;
    use serde_json::{Value, json};
    use std::io::{Read as _, Write as _};
    use std::os::fd::{AsRawFd as _, FromRawFd as _, OwnedFd, RawFd};
    use std::os::unix::net::{UnixListener, UnixStream};
    use std::time::Duration;

    fn wire(req: &PrivRequest) -> Value {
        serde_json::to_value(req).expect("a request serialises")
    }

    const REFUSAL: &str = "atlas: /nix/store/x is not the closure realised for this agent (none)";

    /// An older hive-priv's refusal of a path it did not download is the
    /// signal to ask it to download; a success and any other failure are not.
    #[test]
    fn only_an_older_hive_privs_pin_refusal_asks_for_a_download() {
        assert!(older_hive_priv_refused(&Err(anyhow::anyhow!(REFUSAL))));
        // As the streaming call words it: the error wraps hive-priv's text.
        assert!(older_hive_priv_refused(
            &Err(anyhow::anyhow!(REFUSAL)).context("nixos-container create")
        ));
        assert!(!older_hive_priv_refused(&Ok(())));
        for other in [
            "nixos-container create failed",
            "atlas: refusing /nix/store/x: has no signature by the swarm cache's key",
            "connect to hive-priv socket",
        ] {
            assert!(
                !older_hive_priv_refused(&Err(anyhow::anyhow!(other))),
                "{other}"
            );
        }
    }

    #[test]
    fn the_cache_facts_refresh_names_no_unit() {
        assert_eq!(
            wire(&PrivRequest::RefreshCacheFacts),
            json!({"op": "refresh_cache_facts"})
        );
    }

    #[test]
    fn create_and_update_keep_the_name_and_toplevel_apart_and_stream() {
        assert_eq!(
            wire(&create_container_request("alice", "/nix/store/x")),
            json!({"op": "create_container", "name": "alice", "stream": true, "toplevel": "/nix/store/x"}),
        );
        assert_eq!(
            wire(&update_container_request("alice", "/nix/store/x")),
            json!({"op": "update_container", "name": "alice", "stream": true, "toplevel": "/nix/store/x"}),
        );
    }

    #[test]
    fn resource_limits_put_each_value_in_its_own_field() {
        let got = wire(&resource_limits_request(
            "alice",
            "4G",
            "200%",
            Some(50),
            None,
        ));
        assert_eq!(got["op"], "write_resource_limits");
        assert_eq!(got["container"], "alice");
        assert_eq!(got["memory_max"], "4G");
        assert_eq!(got["cpu_quota"], "200%");
        assert_eq!(got["cpu_weight"], 50);
        assert_eq!(got["io_weight"], Value::Null);
    }

    #[test]
    fn nspawn_flags_carry_every_argument_to_its_field() {
        let binds = [BindMount {
            host_path: "/host/a".into(),
            container_path: "/ctr/a".into(),
            read_only: true,
        }];
        let creds = [CredentialMount {
            name: "otel-headers".into(),
            host_path: "/run/secrets/otel".into(),
        }];
        let isolation = NetworkIsolation {
            bridge: "hive0".into(),
            gateway_ip: "10.42.0.1".into(),
        };
        let facts = HiveFacts {
            name: Some("pr1ma".into()),
            ..HiveFacts::default()
        };
        let got = wire(&write_nspawn_flags_request(
            "alice",
            &binds,
            isolation.clone(),
            &creds,
            facts.clone(),
            Some(true),
            Some(NixBuilds::Disabled),
        ));
        assert_eq!(got["nix_builds"], "none");
        let unset = wire(&write_nspawn_flags_request(
            "alice",
            &binds,
            isolation,
            &creds,
            facts,
            Some(true),
            None,
        ));
        assert_eq!(unset["nix_builds"], Value::Null);
        assert_eq!(got["op"], "write_nspawn_flags");
        assert_eq!(got["container"], "alice");
        assert_eq!(
            got["binds"],
            json!([{"host_path": "/host/a", "container_path": "/ctr/a", "read_only": true}])
        );
        assert_eq!(
            got["isolation"],
            json!({"bridge": "hive0", "gateway_ip": "10.42.0.1"})
        );
        assert_eq!(
            got["load_credentials"],
            json!([{"name": "otel-headers", "host_path": "/run/secrets/otel"}])
        );
        assert_eq!(got["hive_facts"]["name"], "pr1ma");
        assert_eq!(got["private_users"], true);
    }

    #[test]
    fn a_snapshot_send_names_the_parent_only_when_given() {
        let full = wire(&send_snapshot_request("alice", "snap-2", None, "out.btrfs"));
        assert_eq!(full["agent_name"], "alice");
        assert_eq!(full["snapshot_name"], "snap-2");
        assert_eq!(full["dest_file_name"], "out.btrfs");
        assert_eq!(full["parent_snapshot_name"], Value::Null);

        let incremental = wire(&send_snapshot_request(
            "alice",
            "snap-2",
            Some("snap-1"),
            "out.btrfs",
        ));
        assert_eq!(incremental["parent_snapshot_name"], "snap-1");
        assert_eq!(incremental["snapshot_name"], "snap-2");
    }

    #[test]
    fn the_level_zero_qgroup_row_gives_the_usage() {
        let out = "qgroupid rfer excl\n\
                   -------- ---- ----\n\
                   1/100 999 888\n\
                   0/257 4096 2048\n";
        assert_eq!(parse_qgroup_usage(out).expect("a 0/ row"), (4096, 2048));
    }

    #[test]
    fn output_with_no_level_zero_row_is_an_error() {
        let err = parse_qgroup_usage("qgroupid rfer excl\n1/100 1 2\n").expect_err("no 0/ row");
        assert!(err.to_string().contains("no level-0 qgroup"), "{err}");
        // A 0/ row whose columns are not numbers is skipped, not misread.
        parse_qgroup_usage("0/257 x y\n").expect_err("non-numeric columns");
    }

    #[test]
    fn a_refused_response_surfaces_the_error_text() {
        let refused = PrivResponse {
            ok: false,
            stdout: String::new(),
            stderr: String::new(),
            error: Some("bad path".into()),
        };
        assert_eq!(check(refused).expect_err("refused").to_string(), "bad path");

        let silent = PrivResponse {
            ok: false,
            stdout: String::new(),
            stderr: String::new(),
            error: None,
        };
        let err = check(silent).expect_err("refused");
        assert_eq!(err.to_string(), "hive-priv returned error");

        let fine = PrivResponse {
            ok: true,
            stdout: "out".into(),
            stderr: "err".into(),
            error: None,
        };
        assert_eq!(check(fine).expect("ok"), ("out".into(), "err".into()));
    }

    #[tokio::test]
    async fn a_silent_hive_priv_times_out_as_an_error() {
        let err = within(
            Duration::from_millis(20),
            std::future::pending::<Result<u64, _>>(),
        )
        .await
        .expect_err("a call that never answers must fail");
        assert!(err.to_string().contains("did not answer"), "{err}");
    }

    #[tokio::test]
    async fn an_answer_inside_the_limit_passes_through() {
        let got = within(Duration::from_secs(5), async { Ok(7_u64) }).await;
        assert_eq!(got.expect("answered in time"), 7);
    }

    /// A stand-in for hive-priv on a tempdir socket: it accepts one
    /// connection, reads the request until the client shuts down its write
    /// half, writes `reply`, and hands back what it read.
    fn fake_priv(
        reply: &'static str,
    ) -> (
        tempfile::TempDir,
        std::path::PathBuf,
        std::thread::JoinHandle<String>,
    ) {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("priv.sock");
        let listener = UnixListener::bind(&path).expect("bind");
        let server = std::thread::spawn(move || {
            let (mut conn, _) = listener.accept().expect("accept");
            let mut got = String::new();
            conn.read_to_string(&mut got).expect("read request");
            conn.write_all(reply.as_bytes()).expect("write reply");
            got
        });
        (dir, path, server)
    }

    const DONE: &str = "{\"ok\":true,\"stdout\":\"hi\",\"stderr\":\"\"}\n";

    #[tokio::test]
    async fn a_call_sends_one_json_line_and_returns_the_done_event() {
        let (_dir, path, server) = fake_priv(DONE);
        let resp = call_at(&path, &PrivRequest::DaemonReload)
            .await
            .expect("answered");
        assert!(resp.ok);
        assert_eq!(resp.stdout, "hi");
        let sent = server.join().expect("server");
        assert_eq!(sent, "{\"op\":\"daemon_reload\"}\n");
    }

    #[tokio::test]
    async fn a_call_to_a_missing_socket_names_the_connect_step() {
        let dir = tempfile::tempdir().expect("tempdir");
        let err = call_at(&dir.path().join("none.sock"), &PrivRequest::DaemonReload)
            .await
            .expect_err("nothing listens there");
        assert!(
            err.to_string().contains("connect to hive-priv socket"),
            "{err}"
        );
    }

    #[tokio::test]
    async fn a_stream_line_answering_a_plain_call_is_an_error() {
        let (_dir, path, server) = fake_priv("{\"stream\":\"stdout\",\"data\":\"x\"}\n");
        let err = call_at(&path, &PrivRequest::DaemonReload)
            .await
            .expect_err("a non-streaming op must end in Done");
        assert!(err.to_string().contains("unexpected stream line"), "{err}");
        server.join().expect("server");
    }

    #[tokio::test]
    async fn a_garbled_answer_is_a_parse_error() {
        let (_dir, path, server) = fake_priv("not json\n");
        let err = call_at(&path, &PrivRequest::DaemonReload)
            .await
            .expect_err("not an event");
        assert!(err.to_string().contains("parse PrivResponse"), "{err}");
        server.join().expect("server");
    }

    #[tokio::test]
    async fn streaming_hands_each_line_over_in_order_before_the_done_event() {
        let (_dir, path, server) = fake_priv(
            "{\"stream\":\"stdout\",\"data\":\"one\"}\n\
             {\"stream\":\"stderr\",\"data\":\"two\"}\n\
             {\"ok\":true,\"stdout\":\"\",\"stderr\":\"\"}\n",
        );
        let mut seen = Vec::new();
        let resp = call_streaming_at(&path, &PrivRequest::DaemonReload, |s, d| {
            seen.push((s, d.to_owned()));
        })
        .await
        .expect("done");
        assert!(resp.ok);
        assert_eq!(
            seen,
            vec![
                (PrivStream::Stdout, "one".to_owned()),
                (PrivStream::Stderr, "two".to_owned()),
            ]
        );
        server.join().expect("server");
    }

    #[tokio::test]
    async fn streaming_that_ends_without_a_done_event_is_an_error() {
        let (_dir, path, server) = fake_priv("{\"stream\":\"stdout\",\"data\":\"one\"}\n");
        let err = call_streaming_at(&path, &PrivRequest::DaemonReload, |_, _| {})
            .await
            .expect_err("closed early");
        assert!(err.to_string().contains("before sending Done"), "{err}");
        server.join().expect("server");
    }

    /// `recvmsg` the request and the one descriptor that rides with it.
    fn recv_with_fd(conn: &UnixStream) -> (Vec<u8>, OwnedFd) {
        let mut buf = [0_u8; 256];
        let mut iov = libc::iovec {
            iov_base: buf.as_mut_ptr().cast(),
            iov_len: buf.len(),
        };
        // u64 elements give the control buffer the 8-byte alignment a
        // `cmsghdr` read needs; a byte array would be misaligned.
        let mut cmsg = [0_u64; 8];
        // SAFETY: an all-zero msghdr is valid; every pointer set below
        // refers to a buffer that outlives the recvmsg call.
        let (n, fd) = unsafe {
            let mut msg: libc::msghdr = std::mem::zeroed();
            msg.msg_iov = &raw mut iov;
            msg.msg_iovlen = 1;
            msg.msg_control = cmsg.as_mut_ptr().cast();
            msg.msg_controllen = std::mem::size_of_val(&cmsg);
            let n = libc::recvmsg(conn.as_raw_fd(), &raw mut msg, 0);
            assert!(n > 0, "recvmsg failed");
            let hdr = libc::CMSG_FIRSTHDR(&raw const msg);
            assert!(!hdr.is_null(), "no ancillary data arrived");
            assert_eq!((*hdr).cmsg_level, libc::SOL_SOCKET);
            assert_eq!((*hdr).cmsg_type, libc::SCM_RIGHTS);
            let fd = std::ptr::read_unaligned(libc::CMSG_DATA(hdr).cast::<RawFd>());
            (n, OwnedFd::from_raw_fd(fd))
        };
        (buf[..usize::try_from(n).expect("positive")].to_vec(), fd)
    }

    #[tokio::test]
    async fn a_descriptor_arrives_with_the_request_and_is_closed_on_our_side() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("priv.sock");
        let listener = UnixListener::bind(&path).expect("bind");
        let server = std::thread::spawn(move || {
            let (mut conn, _) = listener.accept().expect("accept");
            let (request, fd) = recv_with_fd(&conn);
            let mut rest = String::new();
            conn.read_to_string(&mut rest).expect("read to EOF");
            conn.write_all(DONE.as_bytes()).expect("reply");
            (request, fd)
        });

        let (mut read_end, write_end) = std::io::pipe().expect("pipe");
        let resp = call_with_fd_at(&path, &PrivRequest::DaemonReload, write_end.into())
            .await
            .expect("answered");
        assert!(resp.ok);

        let (request, received) = server.join().expect("server");
        assert_eq!(request, b"{\"op\":\"daemon_reload\"}\n");
        // The descriptor that arrived is the pipe's write end, and our
        // copy is already closed: writing through the received one and
        // then dropping it is what ends the read.
        let mut write_end = std::fs::File::from(received);
        write_end.write_all(b"ping").expect("write via received fd");
        drop(write_end);
        let mut out = String::new();
        read_end
            .read_to_string(&mut out)
            .expect("EOF after both ends closed");
        assert_eq!(out, "ping");
    }
}