//! Deploying a service target, and the glibc floor that says whether the host
//! can run what was built.

use super::DEPLOY_STAGING_ROOT;
use super::RecipeCtx;
use super::collect::ensure_glob_safe;
use super::git::expand_tilde;
use crate::topology::{DeployTarget, Kind};
use anyhow::{Context as _, Result};
use ops_exec::{Action, SyncOpts};
use std::path::{Path, PathBuf};
use std::sync::Arc;

/// Highest `GLIBC_x.y` version referenced by a built binary, and the glibc a
/// host actually has, both parsed from the text the commands print.
///
/// Native-per-architecture builds remove the cross-compile hazard `deploy.sh`
/// was written against, but not this one: fw13 tracks a newer glibc than the
/// Ubuntu 24.04 box in Hetzner, so a binary built here can reference a symbol
/// version that box does not have and fail at exec — after the unit has already
/// been restarted onto it. Comparing the two before the install is what makes
/// that a failed step instead of a downed service.
pub(super) fn max_glibc_symbol(objdump_out: &str) -> Option<(u64, u64)> {
    objdump_out
        .split(|c: char| !(c.is_ascii_digit() || c == '.' || c == '_' || c.is_ascii_alphabetic()))
        .filter_map(|tok| tok.strip_prefix("GLIBC_"))
        .filter_map(parse_glibc_version)
        .max()
}

/// Parse `2.39` (or `2.39.1`, keeping major/minor) into a comparable pair.
pub(super) fn parse_glibc_version(s: &str) -> Option<(u64, u64)> {
    let mut parts = s.split('.');
    let major = parts.next()?.parse().ok()?;
    let minor = parts.next()?.parse().ok()?;
    Some((major, minor))
}

/// The glibc version out of `ldd --version`'s first line, whose tail is the
/// version however the distro decorates the rest (`ldd (Ubuntu GLIBC
/// 2.39-0ubuntu8.8) 2.39`).
pub(super) fn glibc_from_ldd(ldd_out: &str) -> Option<(u64, u64)> {
    let first = ldd_out.lines().find(|l| !l.trim().is_empty())?;
    parse_glibc_version(first.split_whitespace().last()?)
}

impl RecipeCtx {
    /// This target's install destination, or an error naming why there is none.
    pub(super) fn deploy_target(&self) -> Result<&DeployTarget> {
        self.deploy.as_ref().ok_or_else(|| {
            anyhow::anyhow!(
                "no deploy destination for {} {}: the app is `kind = \"{}\"`, and only a \
                 service declares [[deploy]] entries",
                self.app,
                self.target,
                match self.kind {
                    Kind::App => "app",
                    Kind::Library => "library",
                    Kind::Service => "service",
                }
            )
        })
    }

    /// Compare the built binary's glibc requirement against the service host's.
    /// Returns the two versions for the recipe to log.
    pub(super) fn glibc_check(self: &Arc<Self>, binary: &str) -> Result<(String, String)> {
        let d = self.deploy_target()?.clone();
        // `objdump -T` on the build host; no symbols at all (a static binary)
        // means nothing to check, which is a pass rather than a failure.
        let (code, out) = self.run(
            &self.build_host.clone(),
            &format!(
                "objdump -T {binary} 2>/dev/null | grep -o 'GLIBC_[0-9.]*' | sort -uV || true"
            ),
        )?;
        anyhow::ensure!(code == 0, "reading glibc symbols from {binary} failed");
        let Some(needs) = max_glibc_symbol(&out) else {
            return Ok(("none".into(), "n/a".into()));
        };
        let (code, ldd) = self.run(&d.host, "ldd --version")?;
        anyhow::ensure!(
            code == 0,
            "could not read glibc version on service host `{}`",
            d.host
        );
        let has = glibc_from_ldd(&ldd).ok_or_else(|| {
            anyhow::anyhow!(
                "could not parse glibc version from `ldd --version` on `{}`",
                d.host
            )
        })?;
        anyhow::ensure!(
            needs <= has,
            "binary needs glibc {}.{} but `{}` has {}.{} — it would fail to exec after the \
             unit restarted onto it. Build on a host no newer than the service host.",
            needs.0,
            needs.1,
            d.host,
            has.0,
            has.1,
        );
        Ok((
            format!("{}.{}", needs.0, needs.1),
            format!("{}.{}", has.0, has.1),
        ))
    }

    /// Install `binary` (a path on the BUILD host) onto the service host and
    /// restart its unit, via the privileged installer the host holds a scoped
    /// sudo grant for.
    ///
    /// Bento never runs the install itself. It stages the bytes and calls a
    /// root script whose arguments are re-checked on the far side — the same
    /// shape as Sando's `install-companion.sh`, and for the same reason: the
    /// sudoers grant is then ONE auditable script rather than a broad
    /// `install`+`systemctl` grant on a production box.
    ///
    /// Only the binary moves. Config is deliberately untouched: pom's
    /// `pom-astra.toml` / `pom-hetzner.toml` differ per instance, and prod's
    /// carried a `[targets.mnw.tests]` block the repo did not have. A deploy
    /// that copies config over is how that block gets silently deleted.
    pub(super) fn deploy(self: &Arc<Self>, binary: &str) -> Result<String> {
        anyhow::ensure!(
            !self.is_cancelled(),
            "build superseded by a newer request; refusing to deploy"
        );
        // A failed earlier step bars a deploy exactly as it bars a publish. An
        // artifact that failed its gates must not reach a production host just
        // because the recipe kept running.
        let failed = self.failed_steps_snapshot();
        anyhow::ensure!(
            failed.is_empty(),
            "refusing to deploy {} {}: {} failed earlier in this run",
            self.app,
            self.version,
            failed
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>()
                .join(", "),
        );
        let d = self.deploy_target()?.clone();
        ensure_glob_safe(binary)?;

        // Stage under a fixed root the installer also insists on, so "what was
        // checked" and "what is installed" cannot drift apart.
        let staged = format!("{DEPLOY_STAGING_ROOT}/{}", self.app);
        let staged_bin = format!("{staged}/{}", self.app);
        let deploy_exec = self.exec(&d.host)?;
        anyhow::ensure!(
            deploy_exec.capabilities().permits(&Action::Deploy),
            "service host `{}` is not granted the `deploy` capability",
            d.host
        );

        self.run_ok(&d.host, &format!("mkdir -p {staged}"))?;
        if self.build_host_ssh == d.host {
            // Same box: the binary is already there. Routing it through the
            // daemon would be two transfers to end up where it started. This is
            // pom's aarch64 leg — astra builds it and astra runs it.
            self.run_ok(&d.host, &format!("cp -f {binary} {staged_bin}"))?;
        } else {
            // Build host -> daemon -> service host. Two hops because an executor
            // reaches one host; a direct host-to-host transport would mean the
            // build host holding a credential for the production box.
            let tmp = tempfile::tempdir().context("staging dir for deploy")?;
            let local = tmp.path().join(self.app.as_str());
            self.pull_for_deploy(binary, &local)?;
            let (dest, opts) = (PathBuf::from(&staged), SyncOpts::default());
            let dir = tmp.path().to_path_buf();
            self.run_bounded(&format!("stage {} on `{}`", self.app, d.host), async move {
                deploy_exec.push_dir(&dir, &dest, &opts).await
            })
            .with_context(|| format!("staging {} onto `{}`", self.app, d.host))?;
        }

        // The privileged half. Every argument is re-validated by the script,
        // which is the thing actually holding the sudo grant.
        self.run_ok(
            &d.host,
            &format!(
                "{} {staged_bin} {} {}",
                self.cfg.deploy_installer, d.install_path, d.service
            ),
        )?;
        Ok(format!(
            "{} {} installed at {} on `{}`; {} restarted",
            self.app, self.version, d.install_path, d.host, d.service
        ))
    }

    /// Fetch one file off a host into a daemon-local path for re-pushing.
    ///
    /// A local build host is read directly: `fw13` is the daemon's own box, so
    /// the file is already on this filesystem. Routing it through the
    /// artifact-pull gate instead would demand a `pull_root` covering every repo
    /// a service could be built in — today that is `~/Code/Apps`, and pom lives
    /// in `~/Code/MNW`. Widening it to `~/Code` would put `_private`, the
    /// secrets root, inside the collectable tree. This is pom's x86_64 leg.
    fn pull_for_deploy(self: &Arc<Self>, remote: &str, local: &Path) -> Result<()> {
        let host = self.build_host.clone();
        let remote_path = expand_tilde(remote);
        if self.build_host_ssh == "local" || self.build_host_ssh.is_empty() {
            std::fs::copy(&remote_path, local).with_context(|| {
                format!("staging {} from the daemon host", remote_path.display())
            })?;
            return Ok(());
        }
        let sync = self.host_sync(&host)?;
        let (src, dst, opts) = (remote_path, local.to_path_buf(), SyncOpts::default());
        self.run_bounded(&format!("fetch {remote} from `{host}`"), async move {
            sync.pull_file(&src, &dst, &opts).await
        })
        .with_context(|| format!("fetching {remote} from `{host}` to deploy"))
    }

    /// `run`, failing the step on a non-zero exit. The Rust-side twin of the
    /// recipe's `sh_ok`, for commands the deploy machinery issues itself.
    fn run_ok(self: &Arc<Self>, host: &str, cmd: &str) -> Result<String> {
        let (code, tail) = self.run(host, cmd)?;
        if code != 0 {
            self.fail_current_step();
            anyhow::bail!("command on `{host}` exited {code}: {cmd}\n{tail}");
        }
        Ok(tail)
    }
}

#[cfg(test)]
mod tests {
    use super::super::action_for;
    use super::super::build_engine;
    use super::*;
    use crate::config::Config;
    use crate::domain::{AppId, Status, Step, Version};
    use crate::ota::OtaRegistry;
    use std::sync::atomic::AtomicBool;

    /// The comparison that decides whether a binary can exec on the box that is
    /// about to be restarted onto it. Both sides are parsed out of text a tool
    /// printed, so both parsers are worth pinning: fw13 tracks a newer glibc
    /// than the Ubuntu 24.04 host in Hetzner, and getting this backwards means a
    /// dead unit rather than a failed step.
    #[test]
    fn glibc_versions_parse_from_what_the_tools_actually_print() {
        // `objdump -T | grep -o 'GLIBC_[0-9.]*'` output: highest wins, and the
        // comparison is numeric (2.9 must not beat 2.34 lexically).
        let objdump = "GLIBC_2.2.5\nGLIBC_2.34\nGLIBC_2.9\nGLIBC_2.17\n";
        assert_eq!(max_glibc_symbol(objdump), Some((2, 34)));
        // A static binary references none: nothing to check.
        assert_eq!(max_glibc_symbol(""), None);

        // `ldd --version` first line, however the distro decorates it.
        assert_eq!(
            glibc_from_ldd("ldd (Ubuntu GLIBC 2.39-0ubuntu8.8) 2.39\nCopyright...\n"),
            Some((2, 39))
        );
        assert_eq!(
            glibc_from_ldd("ldd (GNU libc) 2.41\nCopyright (C) 2025\n"),
            Some((2, 41))
        );
        assert_eq!(glibc_from_ldd(""), None);
    }

    /// A binary needing MORE than the host has is the failure this check exists
    /// for; equal and less are both fine (glibc symbol versioning is backward
    /// compatible, so an older requirement runs on a newer host).
    #[test]
    fn glibc_requirement_is_satisfied_by_equal_or_newer_only() {
        let needs = max_glibc_symbol("GLIBC_2.41").unwrap();
        assert!(needs > glibc_from_ldd("ldd (Ubuntu GLIBC 2.39) 2.39").unwrap());
        assert!(needs <= glibc_from_ldd("ldd (GNU libc) 2.41").unwrap());
        assert!(needs <= glibc_from_ldd("ldd (GNU libc) 2.42").unwrap());
        assert!(needs <= glibc_from_ldd("ldd (GNU libc) 3.0").unwrap());
    }

    /// Every deploy host function fails with the app's KIND as the reason when
    /// there is no destination, rather than with a missing-host error from
    /// somewhere deeper. A recipe calling `deploy()` on a library is a recipe
    /// written against the wrong kind, and the message should say so.
    #[tokio::test]
    async fn deploy_host_fns_explain_a_missing_destination_by_kind() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::Library,
            1,
            Arc::new(std::collections::HashMap::new()),
            Arc::new(std::collections::HashMap::new()),
            None,
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));
        let engine = build_engine(&ctx);
        for call in [
            "deploy_host()",
            "service_name()",
            "install_path()",
            "health_url()",
            r#"deploy("/tmp/x")"#,
        ] {
            let err = engine.eval::<String>(call).unwrap_err().to_string();
            assert!(
                err.contains("library") && err.contains("no deploy destination"),
                "`{call}` must fail on the kind, got: {err}"
            );
        }
    }

    /// A service host is addressed on the DEPLOY plane whatever step is open.
    ///
    /// The subtle one. Actions are normally derived from the step, which is
    /// right for a build host — the step is what that host is being asked to do.
    /// A service host is granted `deploy`/`restart` and must never be granted
    /// `build`, so the same rule would have `glibc_check` ask it for `build`
    /// during a `verify` step and get denied for a reason unrelated to what was
    /// attempted. `verify` is the step that check belongs in, so without this
    /// routing the glibc gate cannot run at all.
    #[tokio::test]
    async fn a_service_host_is_addressed_on_the_deploy_plane_in_any_step() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        sqlx::query(
            "INSERT INTO builds (id, app, version, status, created_at) \
             VALUES (1, 'demo', '0.1.0', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO target_runs (id, build_id, app, version, target, status, started_at) \
             VALUES (1, 1, 'demo', '0.1.0', 'linux/x86_64', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();

        let deploy = crate::topology::DeployTarget {
            target: "linux/x86_64".parse().unwrap(),
            host: "local".into(),
            port: None,
            install_path: "/usr/local/bin/demo".into(),
            service: "demo.service".into(),
            health_url: None,
        };
        let mut execs: crate::state::ExecutorMap = std::collections::HashMap::new();
        execs.insert("local".into(), crate::state::build_deploy_executor(&deploy));
        // The service host's grant is exactly deploy + restart. If this ever
        // widens to include `build`, the test below stops proving anything.
        assert!(!execs["local"].capabilities().permits(&Action::Build));
        assert!(execs["local"].capabilities().permits(&Action::Deploy));

        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::Service,
            1,
            Arc::new(execs),
            Arc::new(std::collections::HashMap::new()),
            Some(deploy),
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));

        let ctx_blocking = ctx.clone();
        tokio::task::spawn_blocking(move || {
            // `verify` on a service derives Action::Build — which the service
            // host does not grant. The command must still run.
            ctx_blocking.begin_step(Step::Verify).unwrap();
            assert_eq!(
                action_for(Step::Verify, Kind::Service),
                Action::Build,
                "the step's own action is the one that would be denied",
            );
            let (code, out) = ctx_blocking
                .run("local", "echo reached-the-service-host")
                .expect("a service host must be reachable during a verify step");
            assert_eq!(code, 0, "{out}");
            assert!(out.contains("reached-the-service-host"), "{out}");
        })
        .await
        .unwrap();
    }

    /// A step that finalized `Failed` bars the deploy, exactly as it bars a
    /// publish. Without this, a recipe that inspects `sh(...).code` and carries
    /// on regardless still lands a binary on a production host — the precise
    /// hazard a pipeline exists to remove. The check is the ledger, not the
    /// control flow, so it holds whether or not the recipe noticed.
    #[tokio::test]
    async fn a_failed_step_bars_the_deploy() {
        let dir = tempfile::tempdir().unwrap();
        let cfg = Arc::new(Config::for_tests(dir.path()));
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let deploy = crate::topology::DeployTarget {
            target: "linux/x86_64".parse().unwrap(),
            host: "local".into(),
            port: None,
            install_path: "/usr/local/bin/demo".into(),
            service: "demo.service".into(),
            health_url: None,
        };
        // A real build + target run, so the step rows this test finalizes have
        // the parents the schema requires.
        sqlx::query(
            "INSERT INTO builds (id, app, version, status, created_at) \
             VALUES (1, 'demo', '0.1.0', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO target_runs (id, build_id, app, version, target, status, started_at) \
             VALUES (1, 1, 'demo', '0.1.0', 'linux/x86_64', 'running', '2026-07-30T00:00:00Z')",
        )
        .execute(&pool)
        .await
        .unwrap();

        let mut execs: crate::state::ExecutorMap = std::collections::HashMap::new();
        execs.insert("local".into(), crate::state::build_deploy_executor(&deploy));
        let ctx = Arc::new(RecipeCtx::new(
            AppId::new("demo"),
            Version::parse("0.1.0").unwrap(),
            "linux/x86_64".parse().unwrap(),
            "fw13".into(),
            "local".into(),
            "v0.1.0".into(),
            "/tmp".into(),
            vec![],
            Kind::Service,
            1,
            Arc::new(execs),
            Arc::new(std::collections::HashMap::new()),
            Some(deploy),
            pool,
            crate::events::channel(),
            cfg,
            Arc::new(OtaRegistry::standard("https://makenot.work")),
            tokio::runtime::Handle::current(),
            Arc::new(AtomicBool::new(false)),
            None,
        ));

        // A gate ran, failed, and the recipe did not abort — the swallowed
        // failure. Finalizing it is what puts it in the ledger.
        let ctx_blocking = ctx.clone();
        tokio::task::spawn_blocking(move || {
            ctx_blocking.begin_step(Step::Prebuild).unwrap();
            ctx_blocking.fail_current_step();
            ctx_blocking.finish_step(Status::Ok).unwrap();

            let err = ctx_blocking.deploy("/tmp/demo").unwrap_err().to_string();
            assert!(
                err.contains("refusing to deploy") && err.contains("prebuild"),
                "must refuse and name the failed step, got: {err}"
            );
        })
        .await
        .unwrap();
    }
}