zc2 0.0.29

//! `zc agent` configuration (spec §6.1-6.5). `from_env` is the only place that
//! reads the environment; tests build configs with `for_dirs`.

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

pub const DEFAULT_PORT: u16 = 4720;
pub const DEFAULT_BROKER_PORT: u16 = 9000;

/// Every interval the agent uses. Defaults are the spec's values; integration
/// tests shrink them.
#[derive(Debug, Clone)]
pub struct Timings {
    pub reconcile: Duration,
    pub drain_poll: Duration,
    pub drain_timeout: Duration,
    pub kill_grace: Duration,
    /// How long a SIGTERMed broker gets to flush its WAL before SIGKILL.
    pub broker_stop_wait: Duration,
    pub hub_poll: Duration,
    pub hub_timeout: Duration,
    pub shutdown_wait: Duration,
}

impl Default for Timings {
    fn default() -> Self {
        Self {
            reconcile: Duration::from_secs(2),
            drain_poll: Duration::from_secs(1),
            drain_timeout: Duration::from_secs(300),
            kill_grace: Duration::from_secs(10),
            broker_stop_wait: Duration::from_secs(30),
            hub_poll: Duration::from_secs(60),
            hub_timeout: Duration::from_secs(10),
            shutdown_wait: Duration::from_secs(30),
        }
    }
}

#[derive(Debug, Clone)]
pub struct AgentConfig {
    /// `~/.zakuro` (or `$ZAKURO_HOME`): credentials, node_key.
    pub state_dir: PathBuf,
    /// `<state_dir>/agent`.
    pub dir: PathBuf,
    pub port: u16,
    pub broker_port: u16,
    pub base_worker_port: u16,
    pub max_workers: u32,
    /// Hub base URL used when the credentials file carries no `api_url`.
    pub api_url: String,
    /// Broker command; `{port}` is replaced by `broker_port`.
    pub broker_argv: Vec<String>,
    /// Worker command template with `{port}` and `{name}`.
    pub worker_template: Vec<String>,
    /// True when `ZAKURO_AGENT_WORKER_CMD` (or a test) replaced the worker command:
    /// the uv / zakuro-dir prerequisites no longer apply.
    pub worker_template_overridden: bool,
    /// Extra environment for every child (tests isolate the broker's WAL/home here).
    pub child_env: Vec<(String, String)>,
    /// Legacy zakuro-worker-dir candidates (`../zak-zakuro` next to the cwd,
    /// `/opt/code/ZAK/zak-zakuro`, `<home>/zak-zakuro`), captured once by
    /// `from_env` (see `crate::up::legacy_zakuro_dir_candidates_from_env`).
    /// Empty in `for_dirs`, so a test built that way can never see the real
    /// host's `$HOME`, cwd or `/opt/code/ZAK` checkout -- only `state_dir`
    /// and `<state_dir>/env` matter for it.
    pub worker_dir_legacy_candidates: Vec<PathBuf>,
    /// `ZAKURO_WORKER_DIR` from the process environment, captured once by
    /// `from_env`. `None` in `for_dirs`, so a test built that way never reads
    /// the real process environment for the worker-dir lookup either.
    pub worker_dir_env: Option<String>,
    pub timings: Timings,
}

/// `uv run --extra worker python -m zakuro.worker.server --port {port} --worker-name {name}`.
pub fn default_worker_template() -> Vec<String> {
    crate::up::worker_argv(0, "{name}")
        .into_iter()
        .map(|a| if a == "0" { "{port}".to_string() } else { a })
        .collect()
}

/// `ZAKURO_AGENT_MAX_WORKERS` when it is a positive integer, else half the
/// logical CPUs, never less than 1.
pub fn parse_max_workers(env: Option<&str>, logical_cpus: usize) -> u32 {
    env.and_then(|v| v.trim().parse::<u32>().ok())
        .filter(|n| *n > 0)
        .unwrap_or_else(|| (logical_cpus / 2).max(1) as u32)
}

pub fn parse_port(env: Option<&str>) -> u16 {
    env.and_then(|v| v.trim().parse().ok())
        .unwrap_or(DEFAULT_PORT)
}

impl AgentConfig {
    /// Spec defaults for a given state dir, reading nothing from the environment.
    pub fn for_dirs(state_dir: PathBuf) -> Self {
        let logical = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(2);
        let exe = std::env::current_exe()
            .map(|p| p.to_string_lossy().to_string())
            .unwrap_or_else(|_| "zc".to_string());
        Self {
            dir: state_dir.join("agent"),
            state_dir,
            port: DEFAULT_PORT,
            broker_port: DEFAULT_BROKER_PORT,
            base_worker_port: crate::up::BASE_WORKER_PORT,
            max_workers: parse_max_workers(None, logical),
            api_url: crate::credentials::PROD_API_URL.to_string(),
            broker_argv: vec![exe, "broker".to_string(), "{port}".to_string()],
            worker_template: default_worker_template(),
            worker_template_overridden: false,
            child_env: vec![],
            worker_dir_legacy_candidates: vec![],
            worker_dir_env: None,
            timings: Timings::default(),
        }
    }

    /// The real agent's configuration (`zc agent run`).
    pub fn from_env() -> Result<Self, String> {
        let state_dir = crate::credentials::dir().ok_or("no HOME or ZAKURO_HOME")?;
        let mut c = Self::for_dirs(state_dir);
        let var = |k: &str| std::env::var(k).ok();
        c.port = parse_port(var("ZAKURO_AGENT_PORT").as_deref());
        let logical = std::thread::available_parallelism()
            .map(|n| n.get())
            .unwrap_or(2);
        c.max_workers = parse_max_workers(var("ZAKURO_AGENT_MAX_WORKERS").as_deref(), logical);
        c.api_url = crate::credentials::default_api_url();
        if let Some(cmd) = var("ZAKURO_AGENT_WORKER_CMD").filter(|s| !s.trim().is_empty()) {
            c.worker_template = cmd.split_whitespace().map(String::from).collect();
            c.worker_template_overridden = true;
        }
        // The zakuro worker directory itself is not resolved here: it is
        // resolved fresh on every reconcile tick instead (see
        // `Core::reconcile_once`), so a fresh clone or an edited
        // `<zc dir>/env` is picked up without an agent restart. Only the
        // legacy candidate *paths*, and the process-env `ZAKURO_WORKER_DIR`
        // value itself, are captured once, here, from the real environment.
        c.worker_dir_legacy_candidates = crate::up::legacy_zakuro_dir_candidates_from_env();
        c.worker_dir_env = var("ZAKURO_WORKER_DIR");
        Ok(c)
    }

    fn node_key(&self) -> crate::broker::node_identity::NodeKey {
        crate::broker::node_identity::NodeKey::load_or_create_in(Some(self.state_dir.clone()))
    }

    /// First 8 characters of this node's fingerprint: the `{fp8}` in `{fp8}-w{i}`.
    pub fn node_fp8(&self) -> String {
        self.node_key().fingerprint()[..8].to_string()
    }

    /// This node's `broker_pubkey`: how the hub summary's devices name this Mac.
    pub fn node_pubkey(&self) -> String {
        self.node_key().public_b64()
    }
}

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

    #[test]
    fn max_workers_is_half_the_cpus_at_least_one_unless_overridden() {
        assert_eq!(parse_max_workers(None, 10), 5);
        assert_eq!(parse_max_workers(None, 1), 1);
        assert_eq!(parse_max_workers(Some("3"), 10), 3);
        assert_eq!(
            parse_max_workers(Some("0"), 10),
            5,
            "0 is not a usable override"
        );
        assert_eq!(parse_max_workers(Some("x"), 10), 5);
    }

    #[test]
    fn port_defaults_to_4720() {
        assert_eq!(parse_port(None), 4720);
        assert_eq!(parse_port(Some("5000")), 5000);
        assert_eq!(parse_port(Some("nope")), 4720);
    }

    #[test]
    fn defaults_match_the_spec() {
        let c = AgentConfig::for_dirs(std::env::temp_dir().join("zc-agent-cfg"));
        assert_eq!(c.port, 4720);
        assert_eq!(c.broker_port, 9000);
        assert_eq!(c.base_worker_port, 3960);
        assert!(c.dir.ends_with("agent"));
        let t = Timings::default();
        assert_eq!(t.reconcile.as_secs(), 2);
        assert_eq!(t.drain_poll.as_secs(), 1);
        assert_eq!(t.drain_timeout.as_secs(), 300);
        assert_eq!(t.kill_grace.as_secs(), 10);
        assert_eq!(t.hub_poll.as_secs(), 60);
        assert_eq!(t.hub_timeout.as_secs(), 10);
        assert_eq!(t.shutdown_wait.as_secs(), 30);
        assert_eq!(
            default_worker_template(),
            crate::up::worker_argv(0, "x")
                .into_iter()
                .map(|a| match a.as_str() {
                    "0" => "{port}".to_string(),
                    "x" => "{name}".to_string(),
                    _ => a,
                })
                .collect::<Vec<_>>()
        );
    }

    #[test]
    fn fp8_is_the_first_eight_fingerprint_chars_of_the_node_key_in_state_dir() {
        let dir = crate::agent::files::tests::tmp("fp8");
        let c = AgentConfig::for_dirs(dir.clone());
        let key = crate::broker::node_identity::NodeKey::load_or_create_in(Some(dir));
        assert_eq!(c.node_fp8(), key.fingerprint()[..8]);
        assert_eq!(c.node_pubkey(), key.public_b64());
    }
}