//! One build: `nix build` of the check flake with its four inputs pointed at
//! this host's hyperhive, hyperhive docs and nixpkgs and at the agent's config
//! commit.
//!
//! The config repo is private, so `git` fetches it with the builder's forge
//! token. The token reaches `git` through a credential helper that reads it
//! from the child's environment: it is in no argument, no URL and no lock, so
//! no error line `nix` prints can carry it into the reply.
//!
//! ⚠️ A config's own flake inputs are fetched by the same `git`, from hosts the
//! config names. The helper is configured for the forge's URL only, and itself
//! answers only an `https` request for the forge's host, so the token never
//! goes anywhere else.

use std::path::PathBuf;
use std::process::Stdio;
use std::time::Duration;

use swarm_queue_client::builder::{BuildAgent, Reply};
use tokio::io::AsyncReadExt as _;

/// The most of `nix`'s stderr a failed build's reply carries, from its end.
const TAIL_LINES: usize = 30;
const TAIL_BYTES: usize = 4000;

/// How much of each stream is held while `nix` runs, from its end. Output
/// under this is read whole; past it, only the end is kept, so a build's log
/// cannot grow the builder's memory.
const KEEP_BYTES: usize = 64 * 1024;

/// The variable the credential helper reads the token from, in the child only.
const TOKEN_VAR: &str = "SWARM_BUILDER_GIT_TOKEN";

/// The variable holding the one host the helper answers for, in the child
/// only.
const HOST_VAR: &str = "SWARM_BUILDER_GIT_HOST";

/// What every build is run with, fixed at start.
pub struct Builder {
    /// The `nix` binary.
    pub nix: PathBuf,
    /// The check flake's directory in the store.
    pub check_flake: String,
    /// The hyperhive source this host was deployed from.
    pub hyperhive: String,
    /// The nixpkgs source this host was deployed from.
    pub nixpkgs: String,
    /// The hyperhive `docs/` source this host was deployed from.
    pub hyperhive_docs: String,
    /// `https://<forge>`, with no trailing slash and no path.
    pub forge_url: String,
    /// The org the config repos are in.
    pub config_org: String,
    /// How long one build may run.
    pub deadline: Duration,
}

impl Builder {
    /// Build `job` with `token` as the forge credential.
    pub async fn build(&self, job: &BuildAgent, token: &str) -> Reply {
        if let Err(error) = validate(job) {
            return Reply::Failed { error_tail: error };
        }
        let mut cmd = tokio::process::Command::new(&self.nix);
        cmd.args(self.args(job))
            .env(TOKEN_VAR, token)
            .env("GIT_TERMINAL_PROMPT", "0")
            .env("GIT_CONFIG_COUNT", "1")
            .env(HOST_VAR, forge_host(&self.forge_url))
            .env(
                "GIT_CONFIG_KEY_0",
                format!("credential.{}.helper", self.forge_url),
            )
            .env("GIT_CONFIG_VALUE_0", credential_helper())
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        run(cmd, self.deadline).await
    }

    /// The `nix build` arguments for `job`.
    fn args(&self, job: &BuildAgent) -> Vec<String> {
        [
            "build",
            "--no-link",
            "--print-out-paths",
            "--no-write-lock-file",
            "--option",
            "accept-flake-config",
            "false",
            "--print-build-logs",
        ]
        .into_iter()
        .map(str::to_owned)
        .chain([
            format!("{}#check", self.check_flake),
            "--override-input".to_owned(),
            "hyperhive".to_owned(),
            self.hyperhive.clone(),
            "--override-input".to_owned(),
            "nixpkgs".to_owned(),
            self.nixpkgs.clone(),
            "--override-input".to_owned(),
            "hyperhive-docs".to_owned(),
            self.hyperhive_docs.clone(),
            "--override-input".to_owned(),
            "agent".to_owned(),
            self.agent_url(job),
        ])
        .collect()
    }

    /// The config commit as a flake reference, fetched through the PR's head
    /// ref.
    fn agent_url(&self, job: &BuildAgent) -> String {
        format!(
            "git+{}/{}/{}?ref=refs/pull/{}/head&rev={}",
            self.forge_url, self.config_org, job.agent, job.pr, job.rev
        )
    }
}

/// The host part of `forge_url` (`https://<host>[:port]`), as `git` names it
/// to a credential helper.
fn forge_host(forge_url: &str) -> &str {
    let rest = forge_url.strip_prefix("https://").unwrap_or(forge_url);
    rest.split('/').next().unwrap_or(rest)
}

/// A `git` credential helper that answers a `get` with the builder's login and
/// the token from [`TOKEN_VAR`] only when the request is `https` to the host in
/// [`HOST_VAR`], and with nothing otherwise.
fn credential_helper() -> String {
    format!(
        "!f() {{ test \"$1\" = get || return 0; p=; h=; \
         while IFS='=' read -r k v; do case \"$k\" in '') break ;; protocol) p=$v ;; host) h=$v ;; esac; done; \
         test -n \"${HOST_VAR}\" && test \"$p\" = https && test \"$h\" = \"${HOST_VAR}\" \
         && printf 'username=builder\\npassword=%s\\n' \"${TOKEN_VAR}\"; return 0; }}; f"
    )
}

/// Refuse a job whose fields would change the reference's meaning: an agent
/// is one `[a-z0-9-]` segment, and a rev is a full commit id.
fn validate(job: &BuildAgent) -> Result<(), String> {
    let agent_ok = !job.agent.is_empty()
        && job
            .agent
            .bytes()
            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-');
    if !agent_ok {
        return Err(format!("{:?} is not an agent name", job.agent));
    }
    let rev_ok = job.rev.len() == 40
        && job
            .rev
            .bytes()
            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b));
    if !rev_ok {
        return Err(format!("{:?} is not a full commit id", job.rev));
    }
    if job.pr == 0 {
        return Err("PR 0 does not exist".to_owned());
    }
    Ok(())
}

/// Run `cmd` to its end or `deadline`, and read the store path it printed or
/// the tail of what it said.
async fn run(mut cmd: tokio::process::Command, deadline: Duration) -> Reply {
    let mut child = match cmd.spawn() {
        Ok(child) => child,
        Err(e) => {
            return Reply::Failed {
                error_tail: format!("starting nix: {e}"),
            };
        }
    };
    let stdout = child.stdout.take().expect("piped");
    let stderr = child.stderr.take().expect("piped");
    let work = async {
        let (o, e, status) = tokio::join!(
            read_tail(stdout, KEEP_BYTES),
            read_tail(stderr, KEEP_BYTES),
            child.wait()
        );
        let (out, err) = (o?, e?);
        status.map(|s| (s, out, err))
    };
    match tokio::time::timeout(deadline, work).await {
        Err(_) => Reply::Failed {
            error_tail: format!(
                "the build did not finish within {} minutes",
                deadline.as_secs() / 60
            ),
        },
        Ok(Err(e)) => Reply::Failed {
            error_tail: format!("waiting for nix: {e}"),
        },
        Ok(Ok((status, out, err))) => {
            let out_path = String::from_utf8_lossy(&out)
                .lines()
                .rfind(|l| !l.trim().is_empty())
                .map(|l| l.trim().to_owned());
            match out_path {
                Some(out_path) if status.success() => Reply::Built { out_path },
                _ => Reply::Failed {
                    error_tail: tail(&String::from_utf8_lossy(&err), status),
                },
            }
        }
    }
}

/// The last `keep` bytes of a stream, held while it is read: at most
/// `2 * keep` bytes after each push.
struct Tail {
    buf: Vec<u8>,
    keep: usize,
    cut: bool,
}

impl Tail {
    fn new(keep: usize) -> Self {
        Self {
            buf: Vec::new(),
            keep,
            cut: false,
        }
    }

    fn push(&mut self, chunk: &[u8]) {
        self.buf.extend_from_slice(chunk);
        // Trim at twice the cap, so a copy happens once per `keep` bytes.
        if self.buf.len() > self.keep * 2 {
            self.trim();
        }
    }

    fn trim(&mut self) {
        self.buf.drain(..self.buf.len() - self.keep);
        self.cut = true;
    }

    /// The kept bytes. When the stream was longer than `keep`, the cut line
    /// at the front is dropped, so the result starts on a line boundary,
    /// unless that line is all the window holds.
    fn finish(mut self) -> Vec<u8> {
        if self.buf.len() > self.keep {
            self.trim();
        }
        if self.cut
            && let Some(nl) = self.buf.iter().position(|&b| b == b'\n')
            && nl + 1 < self.buf.len()
        {
            self.buf.drain(..=nl);
        }
        self.buf
    }
}

/// Read `reader` to its end, keeping only its last `keep` bytes (see [`Tail`]).
async fn read_tail(
    mut reader: impl tokio::io::AsyncRead + Unpin,
    keep: usize,
) -> std::io::Result<Vec<u8>> {
    let mut tail = Tail::new(keep);
    let mut chunk = vec![0u8; 8192];
    loop {
        let n = reader.read(&mut chunk).await?;
        if n == 0 {
            return Ok(tail.finish());
        }
        tail.push(&chunk[..n]);
    }
}

/// The last [`TAIL_LINES`] lines of `stderr`, at most [`TAIL_BYTES`] bytes,
/// or the exit status when it said nothing.
pub(crate) fn tail(stderr: &str, status: std::process::ExitStatus) -> String {
    let lines: Vec<&str> = stderr.lines().filter(|l| !l.trim().is_empty()).collect();
    if lines.is_empty() {
        return format!("nix exited with {status} and printed nothing");
    }
    let joined = lines[lines.len().saturating_sub(TAIL_LINES)..].join("\n");
    let mut start = joined.len().saturating_sub(TAIL_BYTES);
    while !joined.is_char_boundary(start) {
        start += 1;
    }
    joined[start..].to_owned()
}

#[cfg(test)]
mod tests {
    use super::*;
    use swarm_queue_client::builder::BUILD_DEADLINE;

    fn job() -> BuildAgent {
        BuildAgent {
            agent: "atlas".to_owned(),
            rev: "0123456789abcdef0123456789abcdef01234567".to_owned(),
            pr: 7,
        }
    }

    fn builder() -> Builder {
        Builder {
            check_flake: "/nix/store/c-check".to_owned(),
            hyperhive: "/nix/store/h-source".to_owned(),
            nixpkgs: "/nix/store/n-source".to_owned(),
            hyperhive_docs: "/nix/store/d-docs-source".to_owned(),
            forge_url: "https://forge.example".to_owned(),
            config_org: "agent-configs".to_owned(),
            nix: PathBuf::from("nix"),
            deadline: BUILD_DEADLINE,
        }
    }

    #[test]
    fn the_agent_is_fetched_at_its_commit_through_the_prs_ref() {
        assert_eq!(
            builder().agent_url(&job()),
            "git+https://forge.example/agent-configs/atlas?ref=refs/pull/7/head\
             &rev=0123456789abcdef0123456789abcdef01234567"
        );
    }

    #[test]
    fn every_input_is_overridden_and_no_lock_is_written() {
        let args = builder().args(&job());
        let overridden: Vec<&str> = args
            .windows(2)
            .filter(|w| w[0] == "--override-input")
            .map(|w| w[1].as_str())
            .collect();
        assert_eq!(
            overridden,
            ["hyperhive", "nixpkgs", "hyperhive-docs", "agent"]
        );
        assert!(
            args.windows(2)
                .any(|w| w[0] == "hyperhive-docs" && w[1] == "/nix/store/d-docs-source")
        );
        assert!(args.iter().any(|a| a == "/nix/store/c-check#check"));
        assert!(args.iter().any(|a| a == "--no-write-lock-file"));
    }

    /// What the helper prints for `op` and a request with `protocol` and
    /// `host`, run the way `git` runs a `!` helper.
    fn helper_answer(op: &str, protocol: &str, host: &str) -> String {
        use std::io::Write as _;

        let helper = credential_helper();
        let script = format!("{} {op}", helper.trim_start_matches('!'));
        let mut child = std::process::Command::new("sh")
            .args(["-c", &script])
            .env(TOKEN_VAR, "s3cret")
            .env(HOST_VAR, "forge.example:3000")
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .spawn()
            .expect("sh runs");
        // The helper exits without reading the request for any op but `get`,
        // so the write can hit EPIPE. `git` ignores EPIPE here too.
        let written = write!(
            child.stdin.take().expect("piped"),
            "protocol={protocol}\nhost={host}\npath=agent-configs/atlas\n\n"
        );
        if let Err(e) = written {
            assert_eq!(e.kind(), std::io::ErrorKind::BrokenPipe, "{e}");
        }
        let out = child.wait_with_output().expect("the helper exits");
        assert!(out.status.success());
        String::from_utf8(out.stdout).expect("utf-8")
    }

    #[test]
    fn the_helper_gives_the_token_to_the_forge_alone() {
        assert_eq!(
            helper_answer("get", "https", "forge.example:3000"),
            "username=builder\npassword=s3cret\n"
        );
        for (op, protocol, host) in [
            ("get", "https", "github.com"),
            ("get", "https", "forge.example"),
            ("get", "https", "forge.example:3000.evil.example"),
            ("get", "http", "forge.example:3000"),
            ("store", "https", "forge.example:3000"),
        ] {
            assert_eq!(
                helper_answer(op, protocol, host),
                "",
                "{op} {protocol}://{host}"
            );
        }
    }

    #[test]
    fn the_helper_is_configured_for_the_forge_url_only() {
        assert_eq!(
            forge_host("https://forge.example:3000"),
            "forge.example:3000"
        );
        assert_eq!(forge_host("https://forge.example/sub"), "forge.example");
    }

    #[test]
    fn the_token_is_in_no_argument() {
        let args = builder().args(&job()).join(" ");
        assert!(!args.contains("password"));
        assert!(credential_helper().contains(&format!("${TOKEN_VAR}")));
    }

    #[test]
    fn a_job_that_would_bend_the_reference_is_refused() {
        for bad in [
            BuildAgent {
                agent: "a?rev=x".to_owned(),
                ..job()
            },
            BuildAgent {
                agent: String::new(),
                ..job()
            },
            BuildAgent {
                rev: "main".to_owned(),
                ..job()
            },
            BuildAgent { pr: 0, ..job() },
        ] {
            assert!(validate(&bad).is_err(), "{bad:?}");
        }
        assert!(validate(&job()).is_ok());
    }

    #[test]
    fn the_tail_keeps_the_end_of_a_long_log() {
        use std::os::unix::process::ExitStatusExt as _;

        let log = (0..100)
            .map(|i| format!("line {i}"))
            .collect::<Vec<_>>()
            .join("\n");
        let t = tail(&log, std::process::ExitStatus::from_raw(256));
        assert!(t.ends_with("line 99"));
        assert!(!t.contains("line 69\n"));
        assert!(t.starts_with("line 70"));
        let quiet = tail("", std::process::ExitStatus::from_raw(256));
        assert!(quiet.contains("printed nothing"));
    }

    fn sh(script: &str, deadline: Duration) -> impl Future<Output = Reply> {
        let mut cmd = tokio::process::Command::new("sh");
        cmd.args(["-c", script])
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        run(cmd, deadline)
    }

    #[tokio::test]
    async fn a_success_replies_with_the_last_path_printed() {
        let reply = sh(
            "echo warming >&2; echo /nix/store/abc-nixos-system",
            Duration::from_secs(30),
        )
        .await;
        assert_eq!(
            reply,
            Reply::Built {
                out_path: "/nix/store/abc-nixos-system".to_owned()
            }
        );
    }

    #[tokio::test]
    async fn a_failure_replies_with_what_nix_said_last() {
        let reply = sh(
            "echo first >&2; echo 'error: attribute missing' >&2; exit 1",
            Duration::from_secs(30),
        )
        .await;
        let Reply::Failed { error_tail } = reply else {
            panic!("{reply:?}")
        };
        assert!(error_tail.ends_with("error: attribute missing"));
    }

    /// A `nix` stand-in: runs `body` as a shell script, ignoring its arguments.
    fn fake_nix(tag: &str, body: &str) -> PathBuf {
        use std::os::unix::fs::PermissionsExt as _;

        let dir = std::env::temp_dir().join(format!(
            "swarm-builder-fake-nix-{tag}-{}",
            std::process::id()
        ));
        std::fs::create_dir_all(&dir).expect("scratch dir");
        let path = dir.join("nix");
        std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("write fake nix");
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).expect("chmod");
        path
    }

    /// Builds with `nix`, retrying while the kernel refuses to exec the script
    /// just written (`ETXTBSY`, when another test's fork briefly inherits its
    /// open write handle).
    async fn build_with(nix: PathBuf) -> Reply {
        let builder = Builder { nix, ..builder() };
        for _ in 0..20 {
            let reply = builder.build(&job(), "token").await;
            match &reply {
                Reply::Failed { error_tail } if error_tail.contains("Text file busy") => {
                    tokio::time::sleep(Duration::from_millis(50)).await;
                }
                _ => return reply,
            }
        }
        panic!("the fake nix stayed busy")
    }

    #[tokio::test]
    async fn eight_megabytes_of_log_leave_only_the_tail_in_the_reply() {
        let nix = fake_nix(
            "big",
            "i=0; while [ $i -lt 8 ]; do \
               head -c 1048576 /dev/zero | tr '\\0' 'x' >&2; echo >&2; i=$((i+1)); done\n\
             printf 'noise\\n' >&2\n\
             for i in 1 2 3; do echo \"final line $i\" >&2; done\n\
             exit 1",
        );
        let Reply::Failed { error_tail } = build_with(nix).await else {
            panic!("expected a failure")
        };
        assert_eq!(
            error_tail,
            "noise\nfinal line 1\nfinal line 2\nfinal line 3"
        );
    }

    #[tokio::test]
    async fn eight_megabytes_on_stdout_keep_the_last_path() {
        let nix = fake_nix(
            "bigout",
            "head -c 8388608 /dev/zero | tr '\\0' 'y'; echo\n\
             echo /nix/store/abc-nixos-system",
        );
        assert_eq!(
            build_with(nix).await,
            Reply::Built {
                out_path: "/nix/store/abc-nixos-system".to_owned()
            }
        );
    }

    #[tokio::test]
    async fn output_under_the_cap_reaches_the_reply_whole() {
        let nix = fake_nix(
            "small",
            "echo 'first line' >&2; echo >&2; echo '  second line' >&2; \
             echo 'a long stdout' ; echo /nix/store/abc-nixos-system; echo",
        );
        assert_eq!(
            build_with(nix).await,
            Reply::Built {
                out_path: "/nix/store/abc-nixos-system".to_owned()
            }
        );
        let failing = fake_nix(
            "smallfail",
            "echo 'first line' >&2; echo >&2; echo '  second line' >&2; exit 1",
        );
        let Reply::Failed { error_tail } = build_with(failing).await else {
            panic!("expected a failure")
        };
        assert_eq!(error_tail, "first line\n  second line");
    }

    #[tokio::test]
    async fn read_tail_keeps_the_end_and_starts_on_a_line() {
        let data = "aaaa\nbbbb\ncccc\ndddd\n";
        let whole = read_tail(data.as_bytes(), 100).await.expect("read");
        assert_eq!(whole, data.as_bytes());
        let cut = read_tail(data.as_bytes(), 12).await.expect("read");
        assert_eq!(String::from_utf8(cut).expect("utf-8"), "cccc\ndddd\n");
        let one_long_line = read_tail(&b"x".repeat(50)[..], 12).await.expect("read");
        assert_eq!(one_long_line, b"x".repeat(12));
        let long_last_line = read_tail(format!("{}\n", "x".repeat(50)).as_bytes(), 12)
            .await
            .expect("read");
        assert_eq!(long_last_line, format!("{}\n", "x".repeat(11)).as_bytes());
    }

    #[test]
    fn the_tail_never_holds_more_than_twice_the_cap_while_reading() {
        let mut tail = Tail::new(KEEP_BYTES);
        let mut peak = 0;
        for _ in 0..1024 {
            tail.push(&[b'x'; 8192]);
            peak = peak.max(tail.buf.len());
        }
        assert!(peak > KEEP_BYTES, "control: the buffer does grow: {peak}");
        assert!(peak <= 2 * KEEP_BYTES, "{peak} bytes held");
    }

    #[tokio::test]
    async fn read_tail_holds_at_most_the_cap_of_an_eight_megabyte_stream() {
        let mut stream = b"line\n".repeat(8 * 1024 * 1024 / 5);
        stream.extend_from_slice(b"the end\n");
        let kept = read_tail(&stream[..], KEEP_BYTES).await.expect("read");
        assert!(kept.len() <= KEEP_BYTES, "{} bytes kept", kept.len());
        assert!(kept.ends_with(b"line\nthe end\n"));
        assert!(kept.starts_with(b"line\n"));
    }

    #[tokio::test]
    async fn a_build_past_its_deadline_is_stopped() {
        let reply = sh("sleep 30", Duration::from_millis(200)).await;
        assert!(
            matches!(&reply, Reply::Failed { error_tail } if error_tail.contains("did not finish")),
            "{reply:?}"
        );
    }
}