//! Push an agent snapshot to the swarm's snapshot store.
//!
//! This is the c0re half of the transport: it resolves the store,
//! opens the connection, writes the protocol header, and hands the
//! connected socket to `hive-priv`, which runs `btrfs send` straight
//! into it.
//!
//! The split is the point. Everything that requires knowing *where* the
//! store is and *what* the wire protocol looks like happens here, in the
//! unprivileged daemon. hive-priv only ever receives an already-open
//! descriptor, so the root helper never learns an address, never parses
//! a peer list, and never decides who to trust — and once `btrfs send`
//! starts, neither process is in the data path.
//!
//! There is exactly **one** store per swarm, not one per peer hive. The
//! receiver keys destinations by *agent*, so an agent that migrates
//! between hives keeps a single unbroken incremental chain; per-hive
//! stores would split that chain in two, which is the case the store
//! exists to serve.

use std::os::fd::OwnedFd;
use std::time::Duration;

use anyhow::{Context as _, Result, bail};
use hive_priv_sock::PrivRequest;
use tokio::io::AsyncWriteExt as _;
use tokio::net::TcpStream;

use crate::priv_client;

/// `host:port` of the swarm's snapshot store, rendered by
/// `nix/host-modules/hive-c0re/environment.nix` from
/// `services.hyperhive.swarm.snapshotStore`.
const STORE_ENV: &str = "HYPERHIVE_SNAPSHOT_STORE";

/// Deadline for the TCP connect to the store. An unreachable mesh address
/// otherwise waits out the kernel's own connect timeout while holding the
/// admin-socket handler.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);

/// Deadline for writing and flushing the one header line into a connected
/// socket. Shorter than the receiver's own header deadline.
const HEADER_WRITE_TIMEOUT: Duration = Duration::from_secs(15);

/// Send `snapshot` of `agent` to the swarm's snapshot store, optionally
/// as an incremental against `parent`.
///
/// # Errors
///
/// Fails when no store is configured, it is unreachable, or hive-priv
/// reports the `btrfs send` failed.
pub async fn push_agent_snapshot(agent: &str, snapshot: &str, parent: Option<&str>) -> Result<()> {
    let configured = std::env::var(STORE_ENV).ok();
    let addr = require_store_addr(configured.as_deref())?;

    let sock = open_stream(addr, agent, CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT).await?;

    // Hand the connected socket over. `call_with_fd` takes ownership and
    // closes our copy as soon as the kernel has it, so the receiver sees
    // EOF when `btrfs send` finishes rather than hanging on a descriptor
    // this process is still holding.
    let fd = detach_blocking(sock)?;

    let resp = priv_client::call_with_fd(
        &PrivRequest::SendAgentSnapshotToFd {
            agent_name: agent.to_owned(),
            snapshot_name: snapshot.to_owned(),
            parent_snapshot_name: parent.map(str::to_owned),
        },
        fd,
    )
    .await
    .with_context(|| format!("stream {agent}/{snapshot} to the swarm snapshot store"))?;

    if !resp.ok {
        bail!(
            "push of {agent}/{snapshot} failed: {}",
            resp.error.unwrap_or_else(|| resp.stderr.clone())
        );
    }
    tracing::info!(
        agent,
        snapshot,
        parent,
        store = addr,
        "pushed agent snapshot to the swarm snapshot store"
    );
    Ok(())
}

/// Connect to the store and write the protocol header.
///
/// The header names which agent this stream belongs to: a btrfs stream
/// carries the sender's own subvolume name, not the hive's notion of the
/// agent, so the receiver cannot infer it. One line, because the receiver
/// reads it with `read` and must not buffer past the newline into the byte
/// stream. Only the connect and the header are bounded; the stream body that
/// follows is `btrfs send` writing straight into the socket and may take as
/// long as the snapshot is large.
async fn open_stream(
    addr: &str,
    agent: &str,
    connect_limit: Duration,
    header_limit: Duration,
) -> Result<TcpStream> {
    let mut sock = bounded(
        connect_limit,
        "connect to the swarm snapshot store",
        TcpStream::connect(addr),
    )
    .await?
    .with_context(|| format!("connect to the swarm snapshot store at {addr}"))?;

    bounded(
        header_limit,
        "send header to the swarm snapshot store",
        async {
            sock.write_all(format!("agent {agent}\n").as_bytes())
                .await?;
            sock.flush().await
        },
    )
    .await?
    .context("send header to the swarm snapshot store")?;
    Ok(sock)
}

/// Run `fut` for at most `limit`; a timeout is an error naming `what`.
async fn bounded<T>(limit: Duration, what: &str, fut: impl Future<Output = T>) -> Result<T> {
    tokio::time::timeout(limit, fut)
        .await
        .map_err(|_| anyhow::anyhow!("{what} timed out after {limit:?}"))
}

/// Turn a tokio socket into a descriptor safe to hand to `btrfs send`.
///
/// tokio sockets are `O_NONBLOCK`, and `into_std` does not clear it. The
/// flag lives on the open file description, which `hive-priv` shares with
/// us through `SCM_RIGHTS`, so `btrfs send` would write to a
/// non-blocking socket and get `EAGAIN` whenever the send buffer is full.
fn detach_blocking(sock: TcpStream) -> Result<OwnedFd> {
    let std_sock = sock
        .into_std()
        .context("detach snapshot-store socket for hand-off")?;
    std_sock
        .set_nonblocking(false)
        .context("switch snapshot-store socket to blocking mode for hand-off")?;
    Ok(std_sock.into())
}

/// Require a configured store address.
///
/// Split out of [`push_agent_snapshot`] so the "not configured" path is
/// testable without setting a process-wide env var, which would race
/// other tests.
///
/// # Errors
///
/// Fails when unset, naming the option to set. There is deliberately no
/// fallback: an address is a deployment fact this host cannot derive,
/// and guessing one means pushing an agent's state at whatever happens
/// to answer.
fn require_store_addr(configured: Option<&str>) -> Result<&str> {
    match configured {
        Some(addr) if !addr.is_empty() => Ok(addr),
        _ => bail!(
            "no snapshot store configured for this swarm — set \
             services.hyperhive.swarm.snapshotStore.address (and .port if it \
             differs from the default) on this host"
        ),
    }
}

#[cfg(test)]
mod tests {
    use std::os::fd::AsRawFd as _;

    use tokio::net::{TcpListener, TcpStream};

    use std::time::Duration;

    use super::{
        CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT, bounded, detach_blocking, open_stream,
        require_store_addr,
    };

    #[tokio::test]
    async fn the_handed_off_descriptor_is_blocking() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let sock = TcpStream::connect(listener.local_addr().expect("addr"))
            .await
            .expect("connect");

        let fd = detach_blocking(sock).expect("detach");

        // SAFETY: `fd` is a live descriptor owned by this test.
        let flags = unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_GETFL) };
        assert!(flags >= 0, "F_GETFL failed");
        assert_eq!(
            flags & libc::O_NONBLOCK,
            0,
            "btrfs send would get EAGAIN on a full socket buffer"
        );
    }

    #[tokio::test]
    async fn a_future_that_never_finishes_errors_at_the_deadline() {
        let err = bounded(
            Duration::from_millis(50),
            "connect",
            std::future::pending::<()>(),
        )
        .await
        .expect_err("pending must time out");
        assert!(
            format!("{err:#}").contains("connect timed out after"),
            "{err:#}"
        );
    }

    #[tokio::test]
    async fn a_header_the_peer_never_drains_errors_at_the_deadline() {
        // The peer accepts and holds the connection without reading, so the
        // header cannot drain once the socket buffers fill; a name this
        // large exceeds any loopback buffer.
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr").to_string();
        let peer = tokio::spawn(async move {
            let (held, _) = listener.accept().await.expect("accept");
            std::future::pending::<()>().await;
            drop(held);
        });
        let huge = "a".repeat(128 * 1024 * 1024);
        let err = open_stream(&addr, &huge, CONNECT_TIMEOUT, Duration::from_millis(200))
            .await
            .expect_err("an undrained header must time out");
        peer.abort();
        let msg = format!("{err:#}");
        assert!(
            msg.contains("send header to the swarm snapshot store timed out after"),
            "{msg}"
        );
    }

    #[tokio::test]
    async fn a_reachable_peer_gets_the_header() {
        use tokio::io::AsyncReadExt as _;
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr").to_string();
        let sock = open_stream(&addr, "alice", CONNECT_TIMEOUT, HEADER_WRITE_TIMEOUT)
            .await
            .expect("open");
        let (mut peer, _) = listener.accept().await.expect("accept");
        drop(sock);
        let mut got = String::new();
        peer.read_to_string(&mut got).await.expect("read");
        assert_eq!(got, "agent alice\n");
    }

    #[test]
    fn a_configured_address_passes_through() {
        assert_eq!(
            require_store_addr(Some("10.100.0.1:51821")).expect("configured"),
            "10.100.0.1:51821"
        );
    }

    #[test]
    fn an_unconfigured_store_names_the_option_and_never_guesses() {
        let err = require_store_addr(None).expect_err("no store means no push");
        let msg = format!("{err:#}");
        assert!(msg.contains("swarm.snapshotStore.address"), "{msg}");
        // A guessed address or port is the failure this is guarding
        // against: it would push an agent's state at whatever answers.
        assert!(!msg.contains("51821"), "must not suggest a value: {msg}");
    }

    #[test]
    fn an_empty_value_is_treated_as_unset() {
        // An empty env var is a misconfiguration, not an address; it
        // must fail the same way rather than reaching connect().
        let err = require_store_addr(Some("")).expect_err("empty is not an address");
        assert!(format!("{err:#}").contains("no snapshot store configured"));
    }
}