//! One archetype evaluation: `nix eval` of the `meta` of every entry in this
//! host's hyperhive source's `hyperhiveArchetypes` output, as JSON.
//!
//! Only `meta` is evaluated, never `module`: `meta` is plain data, so the
//! evaluation builds nothing. Both inputs are store paths, so it fetches
//! nothing and needs no forge token.

use std::path::PathBuf;
use std::process::Stdio;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use swarm_wire::builder::ArchetypesReply;
use tokio::io::AsyncReadExt as _;

/// What every evaluation is run with, fixed at start.
pub struct Evaluator {
    /// The `nix` binary.
    pub nix: PathBuf,
    /// The hyperhive source this host was deployed from.
    pub hyperhive: String,
    /// The nixpkgs source this host was deployed from, which the hyperhive
    /// source's own `nixpkgs` input is pointed at, as a hive's meta flake does.
    pub nixpkgs: String,
    /// How long one evaluation may run.
    pub deadline: Duration,
}

impl Evaluator {
    /// Evaluate the archetypes' `meta`.
    pub async fn eval(&self) -> ArchetypesReply {
        let mut cmd = tokio::process::Command::new(&self.nix);
        cmd.args(self.args())
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);
        run(cmd, &self.hyperhive, self.deadline).await
    }

    /// The `nix eval` arguments.
    fn args(&self) -> Vec<String> {
        [
            "eval",
            "--json",
            "--no-write-lock-file",
            "--option",
            "accept-flake-config",
            "false",
        ]
        .into_iter()
        .map(str::to_owned)
        .chain([
            format!("{}#hyperhiveArchetypes", self.hyperhive),
            "--override-input".to_owned(),
            "nixpkgs".to_owned(),
            self.nixpkgs.clone(),
            "--apply".to_owned(),
            "builtins.mapAttrs (_: a: a.meta)".to_owned(),
        ])
        .collect()
    }
}

/// How much of nix's stderr is held while it runs, for the error of an
/// evaluation that does not finish.
const KEEP_STDERR_BYTES: usize = 64 * 1024;

/// Run `cmd` to its end or `deadline`, log how long it took, and read the JSON
/// object it printed or the tail of what it said.
async fn run(cmd: tokio::process::Command, source: &str, deadline: Duration) -> ArchetypesReply {
    let started = Instant::now();
    let reply = run_to_end(cmd, source, deadline).await;
    log_elapsed(&reply, started.elapsed());
    reply
}

/// Log how long one evaluation took, as `elapsed_ms`.
fn log_elapsed(reply: &ArchetypesReply, elapsed: Duration) {
    let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
    match reply {
        ArchetypesReply::Evaluated { .. } => {
            tracing::info!(elapsed_ms, "archetype evaluation finished");
        }
        ArchetypesReply::Failed { .. } => {
            tracing::info!(elapsed_ms, "archetype evaluation ended in failure");
        }
    }
}

async fn run_to_end(
    mut cmd: tokio::process::Command,
    source: &str,
    deadline: Duration,
) -> ArchetypesReply {
    let failed = |error_tail: String| ArchetypesReply::Failed { error_tail };
    let mut child = match cmd.spawn() {
        Ok(child) => child,
        Err(e) => return failed(format!("starting nix: {e}")),
    };
    let mut stdout = child.stdout.take().expect("piped");
    let mut stderr = child.stderr.take().expect("piped");
    let seen = Arc::new(Mutex::new(Vec::<u8>::new()));
    let mut reader = tokio::spawn({
        let seen = Arc::clone(&seen);
        async move {
            let mut chunk = [0u8; 8192];
            while let Ok(n) = stderr.read(&mut chunk).await {
                if n == 0 {
                    break;
                }
                let mut seen = seen.lock().expect("unpoisoned");
                seen.extend_from_slice(&chunk[..n]);
                if seen.len() > KEEP_STDERR_BYTES {
                    let cut = seen.len() - KEEP_STDERR_BYTES;
                    seen.drain(..cut);
                }
            }
        }
    });
    let mut out = Vec::new();
    let work = async {
        let (read, status, _) =
            tokio::join!(stdout.read_to_end(&mut out), child.wait(), &mut reader);
        read?;
        status
    };
    let status = match tokio::time::timeout(deadline, work).await {
        Err(_) => {
            reader.abort();
            let stderr = String::from_utf8_lossy(&seen.lock().expect("unpoisoned")).into_owned();
            let head = format!(
                "the evaluation did not finish within {} seconds",
                deadline.as_secs()
            );
            return failed(match crate::nix_build::last_lines(&stderr) {
                Some(tail) => format!("{head}; nix last said:\n{tail}"),
                None => head,
            });
        }
        Ok(Err(e)) => return failed(format!("waiting for nix: {e}")),
        Ok(Ok(status)) => status,
    };
    if !status.success() {
        let stderr = String::from_utf8_lossy(&seen.lock().expect("unpoisoned")).into_owned();
        return failed(crate::nix_build::tail(&stderr, status));
    }
    match serde_json::from_slice::<serde_json::Value>(&out) {
        Ok(serde_json::Value::Object(archetypes)) => ArchetypesReply::Evaluated {
            source: source.to_owned(),
            archetypes,
        },
        Ok(other) => failed(format!(
            "hyperhiveArchetypes is not an attribute set of archetypes: {other}"
        )),
        Err(e) => failed(format!("nix printed no JSON: {e}")),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tracing_subscriber::layer::SubscriberExt as _;

    fn evaluator() -> Evaluator {
        Evaluator {
            nix: PathBuf::from("nix"),
            hyperhive: "/nix/store/h-source".to_owned(),
            nixpkgs: "/nix/store/n-source".to_owned(),
            deadline: swarm_wire::builder::ARCHETYPES_DEADLINE,
        }
    }

    #[test]
    fn only_meta_is_evaluated_on_this_hosts_sources() {
        let args = evaluator().args();
        assert!(
            args.iter()
                .any(|a| a == "/nix/store/h-source#hyperhiveArchetypes")
        );
        let overridden: Vec<&str> = args
            .windows(2)
            .filter(|w| w[0] == "--override-input")
            .map(|w| w[1].as_str())
            .collect();
        assert_eq!(overridden, ["nixpkgs"]);
        let apply = args
            .windows(2)
            .find(|w| w[0] == "--apply")
            .map(|w| w[1].as_str());
        assert_eq!(apply, Some("builtins.mapAttrs (_: a: a.meta)"));
        assert!(args.iter().any(|a| a == "--no-write-lock-file"));
    }

    fn sh(script: &str, deadline: Duration) -> impl Future<Output = ArchetypesReply> {
        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, "/nix/store/h-source", deadline)
    }

    #[tokio::test]
    async fn an_evaluation_replies_with_each_archetypes_meta() {
        let reply = sh(
            r#"echo '{"reviewer":{"description":"reads"}}'"#,
            Duration::from_secs(30),
        )
        .await;
        let ArchetypesReply::Evaluated { source, archetypes } = reply else {
            panic!("{reply:?}")
        };
        assert_eq!(source, "/nix/store/h-source");
        assert_eq!(archetypes["reviewer"]["description"], "reads");
    }

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

    #[tokio::test]
    async fn output_that_is_not_an_attribute_set_is_refused() {
        for script in ["echo '[1]'", "echo 'not json'"] {
            let reply = sh(script, Duration::from_secs(30)).await;
            assert!(
                matches!(reply, ArchetypesReply::Failed { .. }),
                "{script}: {reply:?}"
            );
        }
    }

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

    /// The `(message, elapsed_ms)` of every event logged.
    #[derive(Clone, Default)]
    struct Events(Arc<Mutex<Vec<Event>>>);

    type Event = (String, Option<u64>);

    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Events {
        fn on_event(
            &self,
            event: &tracing::Event<'_>,
            _: tracing_subscriber::layer::Context<'_, S>,
        ) {
            struct Fields(String, Option<u64>);
            impl tracing::field::Visit for Fields {
                fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
                    if field.name() == "elapsed_ms" {
                        self.1 = Some(value);
                    }
                }
                fn record_debug(
                    &mut self,
                    field: &tracing::field::Field,
                    value: &dyn std::fmt::Debug,
                ) {
                    if field.name() == "message" {
                        self.0 = format!("{value:?}");
                    }
                }
            }
            let mut fields = Fields(String::new(), None);
            event.record(&mut fields);
            self.0
                .lock()
                .expect("unpoisoned")
                .push((fields.0, fields.1));
        }
    }

    #[test]
    fn the_elapsed_time_is_logged_on_success_and_on_failure() {
        let events = Events::default();
        tracing::subscriber::with_default(
            tracing_subscriber::registry().with(events.clone()),
            || {
                let evaluated = ArchetypesReply::Evaluated {
                    source: String::new(),
                    archetypes: serde_json::Map::new(),
                };
                let failed = ArchetypesReply::Failed {
                    error_tail: String::new(),
                };
                log_elapsed(&evaluated, Duration::from_millis(1234));
                log_elapsed(&failed, Duration::from_millis(56));
            },
        );
        let seen = events.0.lock().expect("unpoisoned").clone();
        assert_eq!(
            seen,
            [
                ("archetype evaluation finished".to_owned(), Some(1234)),
                ("archetype evaluation ended in failure".to_owned(), Some(56)),
            ]
        );
    }

    #[tokio::test]
    async fn a_timeout_carries_what_nix_had_printed() {
        let reply = sh(
            "echo 'copying path x' >&2; sleep 30",
            Duration::from_secs(1),
        )
        .await;
        assert!(
            matches!(&reply, ArchetypesReply::Failed { error_tail }
                if error_tail.contains("did not finish") && error_tail.ends_with("copying path x")),
            "{reply:?}"
        );
    }

    #[test]
    fn the_evaluator_waits_five_minutes() {
        assert_eq!(
            swarm_wire::builder::ARCHETYPES_DEADLINE,
            Duration::from_mins(5)
        );
    }
}