//! AIRGAP6 + INSTALL-2 — per-product install (the generalized nornir installer).
//!
//! This is where nornir's bespoke `run_install_systemd` becomes reusable. A
//! [`Product`] (the dep-graph node) carries a [`ComponentSpec`] — the INSTALL-2
//! unit-of-installation type (name/binary/env/ports/after/restart-policy) — and
//! the [`Installer`] trait gives every product the same idempotent
//! `install / upgrade / remove / status` lifecycle, driven by the layout
//! (AIRGAP5) and rendered through [`crate::systemd`] (the extracted
//! `run_install_systemd`).
//!
//! `SystemdInstaller` is the real Linux impl: it builds a [`SystemdServiceSpec`]
//! from the [`ComponentSpec`] + [`Layout`], creates the service user, unpacks the
//! product's software, writes config pointed at the data path, opens firewall
//! ports, and installs the unit — every step **idempotent** (re-running a
//! partial install just fills the gaps). On non-unix it is an honest stub.

use std::path::PathBuf;

use anyhow::Result;
use serde::{Deserialize, Serialize};

use crate::layout::Layout;
use crate::systemd::{RestartPolicy, SystemdServiceSpec};

/// How the installer verifies a started service is actually up.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum HealthCheck {
    /// No active check — `systemctl is-active` is enough.
    #[default]
    None,
    /// TCP connect to `127.0.0.1:<port>` must succeed.
    Tcp(u16),
    /// HTTP GET `<path>` on the first port must return 2xx.
    Http(String),
}

/// INSTALL-2 — the unit of installation. Everything `run_install_systemd`
/// hardcodes for nornir, generalized so any lake product carries its own spec.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ComponentSpec {
    /// Product / service name (e.g. "holger").
    pub name: String,
    /// Human description for the unit.
    pub description: String,
    /// Relative path of the product binary INSIDE the airgap bundle (e.g.
    /// `linux-x86_64/holger`).
    pub bundle_binary: String,
    /// Args appended to the installed binary on `ExecStart=`.
    pub exec_args: Vec<String>,
    /// Service user to create + run as (e.g. "_holger").
    pub user: String,
    /// Firewall ports to open for this product.
    pub ports: Vec<u16>,
    /// Products this one must start `After=` (the dep-graph edges).
    pub after: Vec<String>,
    /// systemd restart policy.
    pub restart: RestartPolicy,
    /// Static env vars (the data path is added at install time from the layout).
    pub env: Vec<(String, String)>,
    /// How to verify the started service.
    pub health: HealthCheck,
}

impl ComponentSpec {
    pub fn new(name: impl Into<String>, bundle_binary: impl Into<String>, user: impl Into<String>) -> Self {
        let name = name.into();
        ComponentSpec {
            description: format!("{name} service"),
            bundle_binary: bundle_binary.into(),
            user: user.into(),
            exec_args: vec![],
            ports: vec![],
            after: vec![],
            restart: RestartPolicy::OnFailure,
            env: vec![],
            health: HealthCheck::None,
            name,
        }
    }

    pub fn with_arg(mut self, a: impl Into<String>) -> Self {
        self.exec_args.push(a.into());
        self
    }
    pub fn with_port(mut self, p: u16) -> Self {
        self.ports.push(p);
        self
    }
    pub fn after(mut self, dep: impl Into<String>) -> Self {
        self.after.push(dep.into());
        self
    }
    pub fn with_env(mut self, k: impl Into<String>, v: impl Into<String>) -> Self {
        self.env.push((k.into(), v.into()));
        self
    }
    pub fn with_health(mut self, h: HealthCheck) -> Self {
        self.health = h;
        self
    }

    /// The dep-graph [`Product`] view (for [`crate::orchestrate`]).
    pub fn as_product(&self) -> Product {
        Product {
            name: self.name.clone(),
            user: self.user.clone(),
            depends_on: self.after.clone(),
            ports: self.ports.clone(),
        }
    }

    /// The installed binary path THROUGH the layout's `current` symlink, so the
    /// service follows the active generation across upgrades.
    pub fn installed_binary(&self, layout: &Layout) -> PathBuf {
        layout
            .software_dir
            .join("current")
            .join(&self.name)
            .join(file_name(&self.bundle_binary))
    }

    /// Build the [`SystemdServiceSpec`] for this product against `layout`,
    /// pointing the data env at `data_dir/<name>` (the AIRGAP5 contract). This is
    /// the bridge from the generic [`ComponentSpec`] to the extracted
    /// `run_install_systemd` renderer.
    pub fn to_systemd_spec(&self, layout: &Layout) -> SystemdServiceSpec {
        let data = layout.product_data(&self.name);
        let bin = self.installed_binary(layout);
        let mut spec = SystemdServiceSpec::new(
            &self.name,
            &self.description,
            bin.to_string_lossy().into_owned(),
            &self.user,
            data.to_string_lossy().into_owned(),
        )
        .with_restart(self.restart)
        // Standard data-path env every product gets (the configure(data) step).
        .with_env(
            format!("{}_DATA", self.name.to_uppercase().replace('-', "_")),
            data.to_string_lossy().into_owned(),
        );
        for a in &self.exec_args {
            spec = spec.with_arg(a);
        }
        for dep in &self.after {
            // dep is a product name → the systemd unit name.
            spec = spec.after(format!("{dep}.service"));
        }
        for (k, v) in &self.env {
            spec = spec.with_env(k, v);
        }
        spec
    }

    /// INSTALL-TREE bridge — build the backend-NEUTRAL [`crate::service::ServiceSpec`]
    /// for this product against `layout` for a chosen [`crate::service::Backend`]
    /// (sibling to [`ComponentSpec::to_systemd_spec`], which is systemd-only). This
    /// is how "create the service" goes through the init-neutral tree
    /// ([`crate::service`]) — systemd + OpenRC + … — rather than hard-coupling to
    /// systemd.
    ///
    /// Wiring: `working_dir` = the product's software dir THROUGH `current` (so it
    /// follows the active generation), `exec` = the installed binary, `user` from
    /// the spec, and the AIRGAP5 data+log contract exposed as env
    /// `DATA_DIR`/`LOG_DIR` (plus the legacy `<NAME>_DATA` key `to_systemd_spec`
    /// emits, kept for continuity). `after` → ordering edges, each health/listen
    /// port → a declared socket, restart carried through.
    ///
    /// SEAM: the neutral [`crate::systemd::SystemdServiceSpec`] this later projects
    /// onto has no `StateDirectory=`/`LogsDirectory=` field, so on systemd the data
    /// and log dirs are surfaced as `Environment=` (and `WorkingDirectory=`), not
    /// the native `StateDirectory=`/`LogsDirectory=` directives — the dirs are
    /// provisioned explicitly by [`Layout::prepare_product`] regardless.
    pub fn to_service_spec(
        &self,
        layout: &Layout,
        backend: crate::service::Backend,
    ) -> crate::service::ServiceSpec {
        use crate::service::{Backend, ServiceSpec, Socket};
        use crate::systemd::Scope;

        let data = layout.product_data(&self.name);
        let log = layout.product_log(&self.name);
        // WorkingDirectory follows `current` so the running service tracks the
        // active generation across an upgrade swap (same law as installed_binary).
        let working = layout.software_dir.join("current").join(&self.name);
        // The lifecycle path (`install::install_product`) unpacks the WHOLE product
        // tree with `extract_all`, which PRESERVES the archive's internal layout —
        // so the exec is the full `bundle_binary` relative path under the product's
        // software dir (e.g. `current/holger/linux-x86_64/holger`). This differs on
        // purpose from `to_systemd_spec`/`SystemdInstaller`, whose `extract_one`
        // FLATTENS the binary to `current/holger/holger` (`installed_binary`).
        let bin = working.join(&self.bundle_binary);

        let scope = match backend {
            Backend::SystemdUser => Scope::User,
            _ => Scope::System,
        };

        let mut spec = ServiceSpec::new(&self.name, &self.description, bin.to_string_lossy())
            .user(&self.user)
            .scope(scope)
            .working_dir(working.to_string_lossy())
            .restart(self.restart)
            // The AIRGAP5 data + log contract, backend-neutral (env in every
            // renderer; systemd inlines it, OpenRC writes /etc/conf.d/<name>).
            .env("DATA_DIR", data.to_string_lossy())
            .env("LOG_DIR", log.to_string_lossy())
            // Legacy per-product data key (matches to_systemd_spec) — kept so a
            // product reading `<NAME>_DATA` keeps working on either path.
            .env(
                format!("{}_DATA", self.name.to_uppercase().replace('-', "_")),
                data.to_string_lossy(),
            );
        for a in &self.exec_args {
            spec = spec.arg(a);
        }
        for (k, v) in &self.env {
            spec = spec.env(k, v);
        }
        for dep in &self.after {
            spec = spec.after(dep);
        }
        for p in &self.ports {
            spec = spec.socket(Socket::tcp(*p));
        }
        spec
    }
}

/// The dep-graph node view of a product (name + service user + edges + ports).
/// Kept distinct from [`ComponentSpec`] so [`crate::orchestrate`] can topo-sort
/// without the full install detail.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Product {
    pub name: String,
    pub user: String,
    pub depends_on: Vec<String>,
    pub ports: Vec<u16>,
}

/// The reusable installer lifecycle — generalizes nornir's `run_install_systemd`
/// / `run_uninstall`. Idempotent by contract.
pub trait Installer {
    /// Install (or upgrade — same code path, atomic) `spec` against `layout`,
    /// reading the product binary from `bundle` via the given archive engine.
    fn install(
        &self,
        spec: &ComponentSpec,
        layout: &Layout,
        bundle: &std::path::Path,
        archive: &dyn crate::archive::Archive,
    ) -> Result<()>;
    /// Stop + remove the product's unit (idempotent).
    fn remove(&self, spec: &ComponentSpec) -> Result<()>;
    /// Whether the product's unit is currently active.
    fn status(&self, spec: &ComponentSpec) -> Result<bool>;
}

/// The real Linux installer: layout-driven unpack + systemd unit (reusing
/// [`crate::systemd::install_systemd`]) + firewall ports. Idempotent.
#[derive(Debug, Clone, Default)]
pub struct SystemdInstaller;

#[cfg(unix)]
impl Installer for SystemdInstaller {
    fn install(
        &self,
        spec: &ComponentSpec,
        layout: &Layout,
        bundle: &std::path::Path,
        archive: &dyn crate::archive::Archive,
    ) -> Result<()> {
        use anyhow::Context;
        // 1. Layout: prepare data + version dirs (idempotent), per-product subdir.
        layout.prepare()?;
        let prod_sw = layout.product_software(&spec.name);
        std::fs::create_dir_all(&prod_sw)
            .with_context(|| format!("create product software dir {}", prod_sw.display()))?;
        std::fs::create_dir_all(layout.product_data(&spec.name)).ok();

        // 2. Verify-on-receipt BEFORE trusting a single byte of the bundle (the
        //    airgap law): recompute the manifest chain + every entry's content hash
        //    and refuse to extract an unverified/tampered bundle.
        crate::fetch::fetch_and_verify(bundle, archive)
            .with_context(|| format!("verify bundle {} before install", bundle.display()))?;

        // 3. Unpack: stream the product binary out of the bundle (no full
        //    decompress for znippy) to the version dir, then activate `current`.
        let dest = prod_sw.join(file_name(&spec.bundle_binary));
        archive.extract_one(bundle, &spec.bundle_binary, &dest)?;
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&dest, std::fs::Permissions::from_mode(0o755)).ok();
        }
        layout.activate()?;

        // 3b. Config: the rendered unit's `ExecStart` may point at a `--config
        //     <path>` (e.g. holger's `/etc/holger/holger.toml`). If nothing writes
        //     that file the unit fails to start. Materialize a minimal, data-path-
        //     aware config so the service the installer just declared can actually
        //     boot — idempotent: an existing (operator-edited) config is left alone.
        write_referenced_config(spec, layout)?;

        // 4. systemd unit (reuse the extracted run_install_systemd). Needs root;
        //    surfaces a clear sudo error otherwise.
        let svc = spec.to_systemd_spec(layout);
        crate::systemd::install_systemd(&svc)?;

        // 5. Firewall: open the product's ports (idempotent — checks first).
        open_ports(&spec.ports)?;
        Ok(())
    }

    fn remove(&self, spec: &ComponentSpec) -> Result<()> {
        // 100% Rust (no `systemctl`): systemd D-Bus.
        let unit = format!("{}.service", spec.name);
        let sd = crate::systemd::Systemd::connect()?;
        let _ = sd.stop(&unit);
        let _ = sd.disable(&unit);
        let _ = std::fs::remove_file(format!("/etc/systemd/system/{unit}"));
        sd.reload()?;
        Ok(())
    }

    fn status(&self, spec: &ComponentSpec) -> Result<bool> {
        crate::systemd::Systemd::connect()?.is_active(&format!("{}.service", spec.name))
    }
}

#[cfg(not(unix))]
impl Installer for SystemdInstaller {
    fn install(
        &self,
        _spec: &ComponentSpec,
        _layout: &Layout,
        _bundle: &std::path::Path,
        _archive: &dyn crate::archive::Archive,
    ) -> Result<()> {
        anyhow::bail!("SystemdInstaller is linux-only")
    }
    fn remove(&self, _spec: &ComponentSpec) -> Result<()> {
        anyhow::bail!("SystemdInstaller is linux-only")
    }
    fn status(&self, _spec: &ComponentSpec) -> Result<bool> {
        anyhow::bail!("SystemdInstaller is linux-only")
    }
}

/// Open `ports` in the host firewall, idempotently. Prefers **firewalld over
/// D-Bus** (the README's "talks over D-Bus directly, no shell" contract, mirroring
/// [`crate::systemd::Systemd`]); only when firewalld is absent does it fall back to
/// `ufw` — which has NO D-Bus surface, so that path stays a best-effort subprocess.
/// A missing firewall daemon is a non-fatal skip (an air-gapped box may run none).
#[cfg(unix)]
fn open_ports(ports: &[u16]) -> Result<()> {
    if ports.is_empty() {
        return Ok(());
    }
    // firewalld over D-Bus first (no shell). Handles everything when present.
    if firewalld_add_ports(ports) {
        return Ok(());
    }
    // ufw fallback — best-effort, no D-Bus API exists for ufw (deploy subprocess).
    let have_ufw = std::process::Command::new("ufw")
        .arg("--version")
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false);
    for p in ports {
        if have_ufw {
            let _ = std::process::Command::new("ufw").args(["allow", &format!("{p}/tcp")]).status();
        } else {
            eprintln!("   ⚠ no firewall (firewalld D-Bus / ufw) — skipped opening port {p}");
        }
    }
    Ok(())
}

/// Open `ports` via firewalld's D-Bus API (`org.fedoraproject.FirewallD1`), 100%
/// Rust over `zbus` — no `firewall-cmd` subprocess, mirroring the
/// [`crate::systemd::Systemd`] pattern. Returns `false` (so the caller falls back
/// to ufw) when firewalld is not running, i.e. its well-known name is unowned on
/// the system bus. Idempotent: an already-enabled port surfaces as a best-effort
/// no-op. The empty zone string targets firewalld's default zone; `timeout = 0`
/// makes the rule non-expiring.
#[cfg(unix)]
fn firewalld_add_ports(ports: &[u16]) -> bool {
    let Ok(conn) = zbus::blocking::Connection::system() else {
        return false;
    };
    // Detect firewalld: is its bus name owned? (Graceful fallback when absent.)
    let owned = conn
        .call_method(
            Some("org.freedesktop.DBus"),
            "/org/freedesktop/DBus",
            Some("org.freedesktop.DBus"),
            "NameHasOwner",
            &("org.fedoraproject.FirewallD1",),
        )
        .ok()
        .and_then(|m| m.body().deserialize::<bool>().ok())
        .unwrap_or(false);
    if !owned {
        return false;
    }
    for p in ports {
        // org.fedoraproject.FirewallD1.zone.addPort(zone, port, protocol, timeout):
        // empty zone = default zone, timeout 0 = no expiry. Best-effort per port
        // (an ALREADY_ENABLED port is not fatal — mirrors the old idempotent skip).
        let port = p.to_string();
        let _ = conn.call_method(
            Some("org.fedoraproject.FirewallD1"),
            "/org/fedoraproject/FirewallD1",
            Some("org.fedoraproject.FirewallD1.zone"),
            "addPort",
            &("", port.as_str(), "tcp", 0i32),
        );
    }
    true
}

impl ComponentSpec {
    /// INSTALL-4 — holger's canonical [`ComponentSpec`]. holger is NOT a code
    /// dependency of nornir (a lake product installs itself the same way), so this
    /// is the reusable *spec* that lets `nornir airgap component add holger` render holger's
    /// unit + would-install end-to-end without nornir linking holger.
    ///
    /// `serve --config /etc/holger/holger.toml`, runs as `_holger`, after
    /// `nornir-server`, listens on 7070, TCP-health on the same port.
    pub fn holger() -> ComponentSpec {
        ComponentSpec::new("holger", "linux-x86_64/holger", "_holger")
            .with_arg("serve")
            .with_arg("--config")
            .with_arg("/etc/holger/holger.toml")
            .with_port(7070)
            .after("nornir-server")
            .with_health(HealthCheck::Tcp(7070))
    }

    /// INSTALL-5 — the tunnr appliance [`ComponentSpec`]. tunnr is the pure-Rust
    /// KVM appliance; like holger it is not a code dependency, just an installable
    /// spec. `run --appliance`, runs as `_tunnr`, listens on 8443 (HTTPS admin),
    /// HTTP-health on `/healthz`.
    ///
    /// Delta delivery: tunnr ships as a large UEFI appliance image, so the
    /// upgrade path is a **binary delta** — `skidbladnir`'s
    /// [`crate::archive::Archive`] is the seam where znippy's zero-copy content
    /// cache + delta lands ([`crate::archive::ZnippyArchive`], `znippy` feature).
    /// With the `znippy` feature on, `upgrade --delta` ships only the changed
    /// blobs; without it, [`delta_supported`] is `false` and it falls back to a
    /// full re-install.
    pub fn tunnr() -> ComponentSpec {
        ComponentSpec::new("tunnr", "linux-x86_64/tunnr", "_tunnr")
            .with_arg("run")
            .with_arg("--appliance")
            .with_port(8443)
            .after("nornir-server")
            .with_health(HealthCheck::Http("/healthz".into()))
    }

    /// nornir's own server as a [`ComponentSpec`] (the dep-graph root most other
    /// products sit `after`). Mirrors `run_install_systemd`'s shape so the generic
    /// `add` path and the bespoke `nornir install add` describe the same service —
    /// and the SAME `nornir-server.service` unit (name == `nornir-server`).
    /// `nornir install add` stays the canonical installer for nornir itself; this
    /// is here so the registry is complete and `airgap component add nornir-server`
    /// works for a lake-style bring-up.
    pub fn nornir_server() -> ComponentSpec {
        ComponentSpec::new("nornir-server", "linux-x86_64/nornir-server", "nornir")
            .with_port(7878)
            .with_health(HealthCheck::Tcp(7878))
    }
}

/// Whether the binary-delta upgrade path is implemented. The delta seam lives in
/// [`crate::archive`] (znippy zero-copy patch); until that engine is wired this
/// returns `false` and callers fall back to a full re-install. Honest stub for
/// INSTALL-5's delta-package delivery.
pub fn delta_supported() -> bool {
    cfg!(feature = "znippy")
}

/// INSTALL-3/4/5 — the built-in component registry. Name → canonical
/// [`ComponentSpec`], so `nornir airgap component add <name>` (and the test matrix) can
/// mechanically discover every installable lake product. Keeping the specs here
/// (not in each product's crate) is deliberate: holger/tunnr are NOT nornir code
/// deps — they are *described*, not linked.
pub fn registry() -> Vec<ComponentSpec> {
    vec![
        ComponentSpec::nornir_server(),
        ComponentSpec::holger(),
        ComponentSpec::tunnr(),
    ]
}

/// Look up a registered [`ComponentSpec`] by product name.
pub fn lookup(name: &str) -> Option<ComponentSpec> {
    registry().into_iter().find(|c| c.name == name)
}

/// The registered component names (for CLI help / discovery).
pub fn registered_names() -> Vec<String> {
    registry().into_iter().map(|c| c.name).collect()
}

fn file_name(path: &str) -> String {
    path.rsplit('/').next().unwrap_or(path).to_string()
}

/// The `--config <path>` (or `--config=<path>`) the product's `ExecStart` args
/// reference, if any. This is the file the started unit will try to read — so the
/// installer must create it (see [`write_referenced_config`]).
fn config_arg_path(spec: &ComponentSpec) -> Option<PathBuf> {
    let mut it = spec.exec_args.iter();
    while let Some(a) = it.next() {
        if let Some(rest) = a.strip_prefix("--config=") {
            return Some(PathBuf::from(rest));
        }
        if a == "--config" {
            return it.next().map(PathBuf::from);
        }
    }
    None
}

/// Write the config file the product's `ExecStart` points at (`--config <path>`)
/// so the unit can start. Minimal + data-path-aware (the AIRGAP5 contract): the
/// data dir the service also gets via `<NAME>_DATA`. Idempotent — an already
/// present config (e.g. an operator's) is never clobbered.
#[cfg(unix)]
fn write_referenced_config(spec: &ComponentSpec, layout: &Layout) -> Result<()> {
    use anyhow::Context;
    let Some(cfg) = config_arg_path(spec) else {
        return Ok(());
    };
    if cfg.exists() {
        return Ok(());
    }
    if let Some(parent) = cfg.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("create config dir {}", parent.display()))?;
    }
    let data = layout.product_data(&spec.name);
    let mut body = format!(
        "# Auto-generated by the skidbladnir installer for `{}`.\n\
         # Minimal defaults so the unit can start; edit as needed — the installer\n\
         # will not overwrite an existing config on re-run.\n\
         data_dir = \"{}\"\n",
        spec.name,
        data.display()
    );
    if let Some(port) = spec.ports.first() {
        body.push_str(&format!("listen_port = {port}\n"));
    }
    std::fs::write(&cfg, body).with_context(|| format!("write config {}", cfg.display()))?;
    Ok(())
}

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

    /// LAW 1 (inject-assert): a ComponentSpec renders a systemd spec whose
    /// ExecStart points at the layout's `current` software path, whose data env
    /// points at the data dir, and whose After= carries the dep edges.
    #[test]
    fn component_spec_renders_layout_driven_systemd_spec() {
        let layout = Layout {
            data_dir: "/var/lib".into(),
            log_dir: "/var/log".into(),
            software_dir: "/opt/lake".into(),
            version: "20250505".into(),
        };
        let spec = ComponentSpec::holger();
        let svc = spec.to_systemd_spec(&layout);

        assert_eq!(svc.service_name, "holger");
        assert_eq!(svc.user, "_holger");
        // ExecStart goes through `current` (follows the active generation).
        let exec = svc.exec_start();
        assert!(exec.contains("/opt/lake/current/holger/holger"), "got: {exec}");
        assert!(exec.ends_with("serve --config /etc/holger/holger.toml"));
        // Data env points at the data dir.
        assert!(svc
            .environment
            .iter()
            .any(|e| e.key == "HOLGER_DATA" && e.value == "/var/lib/holger"));
        // After edge → the dep's unit.
        assert!(svc.after.contains(&"nornir-server.service".to_string()));
    }

    /// LAW 1: `to_service_spec` wires the NEUTRAL service spec — exec + working dir
    /// through `current` (the WHOLE-tree `bundle_binary` path, since the lifecycle
    /// path unfolds the full product), the AIRGAP5 DATA_DIR + LOG_DIR env, the user,
    /// after-edges and the health port as a socket — and renders on BOTH backends.
    #[test]
    fn component_spec_renders_neutral_service_spec_with_data_and_log() {
        use crate::service::{renderer_for, Backend, RestartPolicy};
        let layout = Layout {
            data_dir: "/var/lib".into(),
            log_dir: "/var/log".into(),
            software_dir: "/opt/lake".into(),
            version: "20250505".into(),
        };
        let spec = ComponentSpec::holger();

        // The neutral spec carries data+log as env and execs through current.
        let svc = spec.to_service_spec(&layout, Backend::SystemdSystem);
        assert_eq!(svc.name, "holger");
        assert_eq!(svc.user.as_deref(), Some("_holger"));
        assert_eq!(svc.working_dir.as_deref(), Some("/opt/lake/current/holger"));
        // exec is the FULL bundle_binary path under current/<name> (extract_all).
        assert_eq!(svc.exec, "/opt/lake/current/holger/linux-x86_64/holger");
        let env = |k: &str| svc.env.iter().find(|(ek, _)| ek == k).map(|(_, v)| v.as_str());
        assert_eq!(env("DATA_DIR"), Some("/var/lib/holger"));
        assert_eq!(env("LOG_DIR"), Some("/var/log/holger"));
        assert_eq!(env("HOLGER_DATA"), Some("/var/lib/holger"));
        assert!(svc.after.contains(&"nornir-server".to_string()));
        assert_eq!(svc.restart, RestartPolicy::OnFailure);
        assert!(svc.sockets.iter().any(|s| s.port == 7070), "health port → socket");

        // user scope flips the neutral scope; both backends render.
        let user = spec.to_service_spec(&layout, Backend::SystemdUser);
        assert_eq!(user.scope, crate::systemd::Scope::User, "systemd-user → user scope");

        let sd = renderer_for(Backend::SystemdSystem).unwrap().render(&svc);
        assert!(sd.files[0].contents.contains("Environment=LOG_DIR=/var/log/holger"));
        let rc = renderer_for(Backend::OpenRc).unwrap().render(&svc);
        assert!(rc.file_ending("/etc/conf.d/holger").unwrap().contains("LOG_DIR=\"/var/log/holger\""));
    }

    /// INSTALL bugfix — the `--config <path>` the ExecStart references is actually
    /// materialized so the unit can start, and the write is idempotent (an existing
    /// operator config is never clobbered). Also covers the `--config=<path>` form.
    #[cfg(unix)]
    #[test]
    fn write_referenced_config_materializes_and_is_idempotent() {
        let tmp = tempfile::tempdir().unwrap();
        let cfg = tmp.path().join("etc/svc/svc.toml");
        let layout = Layout {
            data_dir: tmp.path().join("data"),
            log_dir: tmp.path().join("log"),
            software_dir: tmp.path().join("sw"),
            version: "20250505".into(),
        };
        let spec = ComponentSpec::new("svc", "linux-x86_64/svc", "_svc")
            .with_arg("serve")
            .with_arg("--config")
            .with_arg(cfg.to_string_lossy().into_owned())
            .with_port(9000);

        // The arg parser finds the referenced path (space form).
        assert_eq!(config_arg_path(&spec).as_deref(), Some(cfg.as_path()));

        // First install writes a data-path-aware config the unit can read.
        write_referenced_config(&spec, &layout).unwrap();
        let body = std::fs::read_to_string(&cfg).unwrap();
        assert!(body.contains(&format!("data_dir = \"{}\"", layout.product_data("svc").display())));
        assert!(body.contains("listen_port = 9000"));

        // Idempotent: an operator edit survives a re-run (never clobbered).
        std::fs::write(&cfg, "data_dir = \"/custom\"\n").unwrap();
        write_referenced_config(&spec, &layout).unwrap();
        assert_eq!(std::fs::read_to_string(&cfg).unwrap(), "data_dir = \"/custom\"\n");

        // The `--config=<path>` form is recognized too.
        let inline = ComponentSpec::new("svc2", "b", "_svc2")
            .with_arg(format!("--config={}", tmp.path().join("etc/svc2.toml").display()));
        assert_eq!(
            config_arg_path(&inline),
            Some(tmp.path().join("etc/svc2.toml"))
        );

        // No `--config` → nothing to write (Ok, no-op).
        let none = ComponentSpec::new("svc3", "b", "_svc3").with_arg("run");
        assert!(config_arg_path(&none).is_none());
        assert!(write_referenced_config(&none, &layout).is_ok());
    }

    /// The ComponentSpec ↔ Product views are consistent (orchestrate reads the
    /// Product; install reads the ComponentSpec — same name/user/edges/ports).
    #[test]
    fn component_spec_to_product_is_consistent() {
        let spec = ComponentSpec::holger();
        let p = spec.as_product();
        assert_eq!(p.name, "holger");
        assert_eq!(p.user, "_holger");
        assert_eq!(p.depends_on, vec!["nornir-server"]);
        assert_eq!(p.ports, vec![7070]);
    }

    /// ComponentSpec round-trips through JSON (embeddable in the airgap bundle).
    #[test]
    fn component_spec_json_roundtrip() {
        let spec = ComponentSpec::holger().with_env("HOLGER_TLS", "/etc/holger/holger.crt");
        let back: ComponentSpec =
            serde_json::from_str(&serde_json::to_string(&spec).unwrap()).unwrap();
        assert_eq!(back, spec);
        assert_eq!(back.health, HealthCheck::Tcp(7070));
    }

    // ───────────────────── INSTALL-6 inject-assert tests ─────────────────────

    /// INSTALL-3/4 — the built-in registry enumerates the lake products by name
    /// and `lookup` returns the canonical spec (the discovery contract the
    /// `nornir airgap component add <name>` verb resolves against). Inject the names,
    /// assert the exact resolved specs.
    #[test]
    fn registry_enumerates_and_looks_up_components() {
        let names = registered_names();
        assert!(names.contains(&"holger".to_string()));
        assert!(names.contains(&"tunnr".to_string()));
        assert!(names.contains(&"nornir-server".to_string()));

        // lookup resolves to the exact canonical spec.
        let holger = lookup("holger").expect("holger registered");
        assert_eq!(holger, ComponentSpec::holger());
        assert_eq!(holger.user, "_holger");
        assert_eq!(holger.ports, vec![7070]);
        assert_eq!(holger.after, vec!["nornir-server".to_string()]);

        // unknown product → None (the CLI turns this into a clear error).
        assert!(lookup("does-not-exist").is_none());
    }

    /// INSTALL-4 — holger's spec renders the EXACT systemd unit text holger would
    /// install end-to-end (the "render unit → would-install" demo). Asserts the
    /// full unit body, not just "didn't panic".
    #[test]
    fn holger_add_renders_exact_unit_text() {
        use crate::systemd::systemd_unit;
        let layout = Layout {
            data_dir: "/var/lib".into(),
            log_dir: "/var/log".into(),
            software_dir: "/opt/lake".into(),
            version: "20250505".into(),
        };
        let svc = ComponentSpec::holger().to_systemd_spec(&layout);
        let unit = systemd_unit(&svc);

        let expected = "\
[Unit]
Description=holger service
After=network-online.target nornir-server.service
Wants=network-online.target

[Service]
Type=simple
User=_holger
Group=_holger
WorkingDirectory=/var/lib/holger
Environment=HOLGER_DATA=/var/lib/holger
ExecStart=/opt/lake/current/holger/holger serve --config /etc/holger/holger.toml
Restart=on-failure
RestartSec=2
NoNewPrivileges=true

[Install]
WantedBy=multi-user.target
";
        assert_eq!(unit, expected, "holger unit text must match byte-for-byte");
    }

    /// INSTALL-5 — tunnr's appliance spec renders its unit with the appliance args,
    /// HTTPS admin port, and the dep edge; delta delivery is wired iff the `znippy`
    /// feature is built in (otherwise an honest full-reinstall fallback).
    #[test]
    fn tunnr_spec_renders_appliance_unit_and_delta_tracks_feature() {
        use crate::systemd::systemd_unit;
        let layout = Layout {
            data_dir: "/var/lib".into(),
            log_dir: "/var/log".into(),
            software_dir: "/opt/lake".into(),
            version: "20250505".into(),
        };
        let spec = ComponentSpec::tunnr();
        assert_eq!(spec.ports, vec![8443]);
        assert_eq!(spec.health, HealthCheck::Http("/healthz".into()));

        let unit = systemd_unit(&spec.to_systemd_spec(&layout));
        assert!(
            unit.contains("ExecStart=/opt/lake/current/tunnr/tunnr run --appliance"),
            "got: {unit}"
        );
        assert!(unit.contains("User=_tunnr"));
        assert!(unit.contains("After=network-online.target nornir-server.service"));

        // Delta delivery is the znippy seam: real with the `znippy` feature, an
        // honest full-reinstall fallback without it. The capability must track the
        // feature exactly (no lying either way).
        assert_eq!(
            delta_supported(),
            cfg!(feature = "znippy"),
            "delta_supported() must mirror the znippy feature"
        );
    }

    /// INSTALL-6 — `upgrade` idempotence at the layout level: installing the SAME
    /// version twice is a no-op-equivalent (the spec → unit text is byte-stable
    /// across renders), and a NEW version reuses the spec to produce a unit that
    /// still points through `current` (so the atomic swap, not the unit, is what
    /// changes). This is the spec-level invariant the SystemdInstaller relies on.
    #[test]
    fn upgrade_is_idempotent_at_spec_level() {
        use crate::systemd::systemd_unit;
        let spec = ComponentSpec::holger();

        let v1 = Layout {
            data_dir: "/var/lib".into(),
            log_dir: "/var/log".into(),
            software_dir: "/opt/lake".into(),
            version: "20250505".into(),
        };
        let v2 = Layout {
            version: "20250606".into(),
            ..v1.clone()
        };

        let u1 = systemd_unit(&spec.to_systemd_spec(&v1));
        let u1_again = systemd_unit(&spec.to_systemd_spec(&v1));
        // Re-rendering the same (spec, layout) is byte-identical → idempotent write.
        assert_eq!(u1, u1_again, "same spec+version renders identical unit");

        // A new version renders the SAME unit (ExecStart goes through `current`),
        // so an upgrade is purely the atomic symlink swap — the unit never churns.
        let u2 = systemd_unit(&spec.to_systemd_spec(&v2));
        assert_eq!(
            u1, u2,
            "unit is version-independent (points through current) → upgrade = swap"
        );
        assert!(u2.contains("/opt/lake/current/holger/holger"));
    }
}
