zc2 0.0.28

//! Deployment reconciler: makes the hub's desired "environments" — pre-
//! loaded, warmed images ready to execute a remote function — true on THIS
//! host, reports progress back to the hub, rolls back on failure, and bills
//! served `/serve/:deployment_id` requests by execution time on every
//! verified (2xx) response.
//!
//! Contract with the hub (`zak-dashboard`, being built in parallel):
//!   GET  {api_url}/api/broker/deployments/desired
//!   POST {api_url}/api/broker/deployments/{id}/status
//!
//! Both are authenticated the same way the rest of the broker's dashboard
//! calls are (`X-Broker-Api-Key`), plus `X-Node-Pubkey` identifying which
//! host is reporting — a node only ever reconciles ITS OWN desired set, and
//! the hub uses the pubkey to tell nodes apart the same way it already does
//! for `/api/broker/node/register` (see `node_sync.rs`).

pub mod expose;
pub mod reconcile;
pub mod runtime;
pub mod serve;
pub mod state;

use std::sync::Arc;
use std::time::Duration;

use serde::{Deserialize, Serialize};

// Re-exported so external callers (and doc links) can name these without
// reaching into the submodules directly; `#[allow(unused_imports)]` because
// nothing in THIS crate currently needs the short path — they're still part
// of the module's public surface (used by name in doc comments) and by any
// future caller (e.g. a dashboard-shaped integration test) that wants
// `deploy::Action` rather than `deploy::reconcile::Action`.
#[allow(unused_imports)]
pub use reconcile::{plan, plan_failure, Action};
#[allow(unused_imports)]
pub use runtime::{runtime_available, ContainerRuntime, DockerCli, RunSpec};
#[allow(unused_imports)]
pub use state::DeploymentRecord;
pub use state::LocalState;

/// One port the deployment's container listens on.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Port {
    pub container: u16,
    #[serde(default = "default_protocol")]
    pub protocol: String,
}

fn default_protocol() -> String {
    "http".to_string()
}

/// How to probe a running container for readiness.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Healthcheck {
    #[serde(rename = "type")]
    pub kind: String, // "http" | "tcp"
    pub port: u16,
    #[serde(default)]
    pub path: Option<String>,
    #[serde(default = "default_timeout_s")]
    pub timeout_s: u64,
}

fn default_timeout_s() -> u64 {
    60
}

/// What a running environment may reach: for now this is surfaced only
/// (local state + `zc deployments`), and its `grants` are handed to the
/// container as env vars so the process inside can see them, but nothing on
/// this node enforces them yet — enforcement is a hub/dashboard concern.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ServiceAccount {
    pub name: String,
    #[serde(default)]
    pub grants: Vec<String>,
}

/// The deployment this one is replacing, as the hub describes it. Usually
/// `previous.id == this.id` (a version bump of the same deployment); when it
/// names a DIFFERENT id, the hub is describing a slot handoff — e.g. a
/// blue/green cutover where this deployment takes over from another one
/// entirely — and this node stops that other deployment's container once
/// this one is healthy (see `reconcile::plan`).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PreviousDeployment {
    pub id: String,
    pub image: String,
    #[serde(default)]
    pub cmd: Option<String>,
    #[serde(default)]
    pub ports: Vec<Port>,
    #[serde(default)]
    pub env: std::collections::HashMap<String, String>,
    #[serde(default)]
    pub healthcheck: Option<Healthcheck>,
}

/// One entry of `GET /api/broker/deployments/desired`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Desired {
    pub id: String,
    pub version: u64,
    pub name: String,
    pub image: String,
    #[serde(default)]
    pub cmd: Option<String>,
    #[serde(default)]
    pub ports: Vec<Port>,
    #[serde(default)]
    pub env: std::collections::HashMap<String, String>,
    #[serde(default)]
    pub healthcheck: Option<Healthcheck>,
    /// Credits per second of execution — billed per served request on
    /// `duration_ms` actually spent proxied into the environment (see
    /// `deploy::serve`), not a flat per-call price. The hub has already
    /// resolved this to an effective value for auto-priced deployments.
    pub price_per_second: f64,
    pub desired_state: String, // "running" | "stopped"
    /// What this environment may reach; surfaced only for now (see
    /// `ServiceAccount`'s doc comment) — not enforced on this node.
    #[serde(default)]
    pub service_account: Option<ServiceAccount>,
    /// `true`: keep the environment's container running and ready.
    /// `false`: allowed to stop it when idle and cold-start on the next
    /// `/serve` call — NOT YET IMPLEMENTED, treated as `true` (see
    /// `reconcile::drive`'s `Start` handling).
    #[serde(default = "default_warm")]
    pub warm: bool,
    #[serde(default)]
    pub previous: Option<Box<PreviousDeployment>>,
}

pub(crate) fn default_warm() -> bool {
    true
}

#[derive(Debug, Deserialize)]
struct DesiredResponse {
    deployments: Vec<Desired>,
}

/// `GET {api_url}/api/broker/deployments/desired` for this node.
pub fn fetch_desired(
    api_url: &str,
    api_key: &str,
    node_pubkey: &str,
) -> Result<Vec<Desired>, String> {
    let endpoint = format!(
        "{}/api/broker/deployments/desired",
        api_url.trim_end_matches('/')
    );
    let resp = ureq::get(&endpoint)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("X-Node-Pubkey", node_pubkey)
        .call()
        .map_err(|e| e.to_string())?;
    if resp.status().as_u16() != 200 {
        return Err(format!("fetch_desired HTTP {}", resp.status().as_u16()));
    }
    let text = resp
        .into_body()
        .read_to_string()
        .map_err(|e| e.to_string())?;
    let parsed: DesiredResponse = serde_json::from_str(&text).map_err(|e| e.to_string())?;
    Ok(parsed.deployments)
}

/// The body of `POST {api_url}/api/broker/deployments/{id}/status`.
/// `mesh_endpoint` is always present, `null` when the deployment isn't
/// exposed: each report replaces the hub's row, so leaving the key out would
/// not mean "unchanged".
#[allow(clippy::too_many_arguments)]
fn status_payload(
    node_pubkey: &str,
    version: u64,
    phase: &str,
    message: &str,
    attempt: u32,
    container_id: Option<&str>,
    endpoint: Option<&str>,
    mesh_endpoint: Option<&str>,
) -> serde_json::Value {
    serde_json::json!({
        "node_pubkey": node_pubkey,
        "version": version,
        "phase": phase,
        "message": message,
        "attempt": attempt,
        "container_id": container_id,
        "endpoint": endpoint,
        "mesh_endpoint": mesh_endpoint,
    })
}

/// `POST {api_url}/api/broker/deployments/{id}/status`.
#[allow(clippy::too_many_arguments)]
pub fn report_status(
    api_url: &str,
    api_key: &str,
    node_pubkey: &str,
    id: &str,
    version: u64,
    phase: &str,
    message: &str,
    attempt: u32,
    container_id: Option<&str>,
    endpoint: Option<&str>,
    mesh_endpoint: Option<&str>,
) {
    let url = format!(
        "{}/api/broker/deployments/{}/status",
        api_url.trim_end_matches('/'),
        id
    );
    let payload = status_payload(
        node_pubkey,
        version,
        phase,
        message,
        attempt,
        container_id,
        endpoint,
        mesh_endpoint,
    );
    let body = serde_json::to_string(&payload).unwrap_or_default();
    let result = ureq::post(&url)
        .config()
        .timeout_global(Some(Duration::from_secs(10)))
        .http_status_as_error(false)
        .build()
        .header("X-Broker-Api-Key", api_key)
        .header("X-Node-Pubkey", node_pubkey)
        .header("Content-Type", "application/json")
        .send(body.as_str());
    if let Err(e) = result {
        eprintln!("  [DEPLOY] failed to report status for {id}: {e}");
    }
}

/// `reconcile::StatusReporter` backed by the real hub HTTP contract above.
pub struct HttpReporter<'a> {
    pub api_url: &'a str,
    pub api_key: &'a str,
    pub node_pubkey: &'a str,
}

impl reconcile::StatusReporter for HttpReporter<'_> {
    fn report(
        &self,
        id: &str,
        version: u64,
        phase: &str,
        message: &str,
        attempt: u32,
        container_id: Option<&str>,
        endpoint: Option<&str>,
        mesh_endpoint: Option<&str>,
    ) {
        report_status(
            self.api_url,
            self.api_key,
            self.node_pubkey,
            id,
            version,
            phase,
            message,
            attempt,
            container_id,
            endpoint,
            mesh_endpoint,
        );
    }
}

/// One reconcile pass: fetch desired state from the hub and drive local
/// docker state toward it. Called from the broker's periodic tick loop (see
/// `server.rs`, "Periodic worker sync"); a no-op when `api_url`/`api_key`
/// aren't configured — a broker with no dashboard has nothing to reconcile
/// against, same precondition `WORKER_SYNC` and the roster refresh use.
pub fn tick(state: &Arc<crate::broker::BrokerState>) {
    let (Some(api_url), Some(api_key)) = (&state.config.api_url, &state.config.api_key) else {
        return;
    };
    let node_pubkey = state.node_key.public_b64();

    let desired = match fetch_desired(api_url, api_key, &node_pubkey) {
        Ok(d) => d,
        Err(e) => {
            eprintln!("  [DEPLOY] failed to fetch desired deployments: {e}");
            return;
        }
    };
    if desired.is_empty() {
        return;
    }

    let rt = DockerCli;
    if !rt.available() {
        // Nothing this node can do — tell the hub once per tick rather than
        // silently leaving every deployment stuck "pending" from its side.
        for d in &desired {
            if d.desired_state == "running" {
                report_status(
                    api_url,
                    api_key,
                    &node_pubkey,
                    &d.id,
                    d.version,
                    "failed",
                    "no container runtime on this host",
                    1,
                    None,
                    None,
                    None,
                );
            }
        }
        return;
    }

    let mut local = LocalState::load();
    let reporter = HttpReporter {
        api_url,
        api_key,
        node_pubkey: &node_pubkey,
    };
    let mesh_ip = expose::exposure_ip(
        crate::broker::discovery::get_mesh_ip(),
        std::env::var("ZAKURO_MESH_EXPOSE").ok().as_deref(),
    );
    if mesh_ip.is_some() {
        // Who may use the mesh forwarder: the owner's addresses (zc#212).
        expose::refresh_allowlist(api_url, api_key);
    }
    reconcile::drive_and_expose(
        &rt,
        &reporter,
        expose::forwarder(),
        mesh_ip.as_deref(),
        &desired,
        &mut local,
    );
}

/// `zc deployments` — print this node's local reconciler state (from
/// `deployments.json`) plus a `docker ps` view of the containers it created,
/// for debugging a stuck or unexpected deployment without reaching for
/// `sqlite3`/dashboard access.
pub fn print_deployments() {
    let local = LocalState::load();
    if local.deployments.is_empty() {
        println!("No local deployment state (nothing reconciled on this node yet).");
    } else {
        println!(
            "{:<24} {:>7}  {:<12} {:<22} {:<22} {:<6} {:<20} CONTAINER",
            "DEPLOYMENT", "VERSION", "PHASE", "ENDPOINT", "MESH", "WARM", "SERVICE_ACCOUNT"
        );
        let mut ids: Vec<&String> = local.deployments.keys().collect();
        ids.sort();
        for id in ids {
            let rec = &local.deployments[id];
            let service_account = rec
                .service_account
                .as_ref()
                .map(|sa| format!("{}[{}]", sa.name, sa.grants.join(",")))
                .unwrap_or_else(|| "-".to_string());
            println!(
                "{:<24} {:>7}  {:<12} {:<22} {:<22} {:<6} {:<20} {}",
                id,
                rec.version,
                rec.phase,
                rec.endpoint.as_deref().unwrap_or("-"),
                rec.mesh_endpoint.as_deref().unwrap_or("-"),
                rec.warm,
                service_account,
                rec.container_id.as_deref().unwrap_or("-"),
            );
        }
    }

    println!();
    let rt = DockerCli;
    if !rt.available() {
        println!("(docker not available on this host)");
        return;
    }
    match std::process::Command::new("docker")
        .args([
            "ps",
            "-a",
            "--filter",
            "label=zakuro.deployment",
            "--format",
            "table {{.Names}}\t{{.Status}}\t{{.Ports}}",
        ])
        .output()
    {
        Ok(out) if out.status.success() => {
            print!("{}", String::from_utf8_lossy(&out.stdout));
        }
        Ok(out) => {
            eprintln!(
                "docker ps failed: {}",
                String::from_utf8_lossy(&out.stderr).trim()
            );
        }
        Err(e) => eprintln!("docker ps failed: {e}"),
    }
}

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

    #[test]
    fn status_payload_carries_the_mesh_endpoint() {
        let payload = status_payload(
            "pk",
            3,
            "healthy",
            "",
            1,
            Some("c1"),
            Some("172.17.0.3:8888"),
            Some("10.13.13.22:8888"),
        );
        assert_eq!(payload["endpoint"], "172.17.0.3:8888");
        assert_eq!(payload["mesh_endpoint"], "10.13.13.22:8888");
    }

    #[test]
    fn status_payload_sends_an_explicit_null_when_not_exposed() {
        let payload = status_payload("pk", 3, "stopped", "", 1, None, None, None);
        assert!(payload.get("mesh_endpoint").is_some_and(|v| v.is_null()));
    }
}