openlatch-client 0.1.18

//! `openlatch update [--check] [--apply]` — manual auto-update surface.
//!
//! Routing:
//! - `--check` (default if neither flag): probe the registry, print
//!   `{current, latest, severity}` JSON. Always exits 0.
//! - `--apply`: gate on cargo-install, probe daemon, route through
//!   `POST /admin/update` if a daemon is up; fall back to in-process
//!   apply otherwise. The CLI long-polls `/admin/update/status` for
//!   progress and treats a connection drop mid-poll as the expected
//!   restart-in-progress signal.
//!
//! See `.local/brainstorms/auto-update/PHASE-2-manual-via-rpc.md` § 5.

use std::time::Duration;

use clap::Args;
use serde_json::json;

use crate::cli::output::OutputConfig;
use crate::config;
use crate::install_state;
use crate::update;

/// Arguments for `openlatch update`.
#[derive(Args, Clone, Debug, Default)]
pub struct UpdateArgs {
    /// Probe the registry and print current/latest/severity. No swap.
    #[arg(long)]
    pub check: bool,

    /// Apply the update if one is available. Implies `--check` semantics
    /// when no update is available (idempotent exit 0).
    #[arg(long)]
    pub apply: bool,

    /// Accepted for compatibility with scripted callers. Updates apply
    /// without an interactive prompt, so this flag changes nothing today;
    /// it is kept so existing `update --apply --yes` invocations keep
    /// working.
    #[arg(long, short = 'y')]
    pub yes: bool,

    /// Force the cargo-install gate to be bypassed. Maintainer escape
    /// hatch — `cargo install`-managed binaries get rewritten by
    /// self-replace, which voids `cargo install --version` tracking.
    /// Most users should run `cargo install --force --locked
    /// openlatch-client` instead.
    #[arg(long = "force-cargo")]
    pub force_cargo: bool,
}

/// Exit codes for `openlatch update`.
///
/// 0 = success or idempotent no-op
/// 1 = generic apply failure (verify, sanity, swap, etc.)
/// 5 = OL-1505 (cargo-install refusal) or 409 already-up-to-date
/// 6 = OL-1506 (daemon expected but unreachable; we fell back but
///       could not complete in-process either)
pub fn run(args: &UpdateArgs, output: &OutputConfig) -> i32 {
    // Local current-thread tokio runtime: the CLI dispatcher is sync,
    // so spinning up a small runtime here is the cheapest way to call
    // the async `core::update` primitives without bleeding tokio into
    // every command.
    let runtime = match tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
    {
        Ok(rt) => rt,
        Err(e) => {
            output.print_info(&format!("failed to build local tokio runtime: {e}"));
            return 1;
        }
    };
    runtime.block_on(run_async(args, output))
}

async fn run_async(args: &UpdateArgs, output: &OutputConfig) -> i32 {
    let current_version = env!("CARGO_PKG_VERSION").to_string();
    // Load config once — `Config::load` already merges the
    // `OPENLATCH_NPM_REGISTRY` env override into `update.registry_origin`,
    // so re-reading the env var here would be redundant.
    let cfg = config::Config::load(None, None, false).ok();
    let registry_origin = cfg
        .as_ref()
        .map(|c| c.update.registry_origin.clone())
        .unwrap_or_else(|| "https://registry.npmjs.org".to_string());
    let port = cfg.as_ref().map(|c| c.port).unwrap_or(7443);

    if !args.apply {
        // Treat both `--check` and bare `openlatch update` as a
        // read-only registry probe. Matches `openlatch status`
        // (read-only by default).
        return run_check(output, &current_version, &registry_origin).await;
    }

    // --apply path. Cargo-install gate runs first so we never POST
    // or hit the network for a binary we can't update.
    if !args.force_cargo
        && matches!(
            install_state::detect_install_method(),
            install_state::InstallMethod::CargoInstall
        )
    {
        let err = crate::error::OlError::new(
            crate::error::ERR_UPDATE_REFUSED_CARGO_INSTALL,
            "this binary was installed via `cargo install` — auto-update would not take effect",
        )
        .with_suggestion("Run: cargo install --force --locked openlatch-client");
        output.print_error(&err);
        return 5;
    }

    match probe_daemon(port).await {
        DaemonState::RunningAndReachable { port, token } => {
            apply_via_daemon_rpc(output, &current_version, port, &token, args.force_cargo).await
        }
        DaemonState::NotRunning => {
            apply_in_process(output, &current_version, &registry_origin, args).await
        }
        DaemonState::RunningButUnauthenticated => {
            let err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_DAEMON_UNREACHABLE,
                "daemon is running but the bearer token in ~/.openlatch/daemon.token is missing or wrong",
            )
            .with_suggestion(
                "Run `openlatch init --reconfig` to regenerate, or stop the daemon and re-run.",
            );
            output.print_error(&err);
            6
        }
    }
}

async fn run_check(output: &OutputConfig, current: &str, registry: &str) -> i32 {
    // `--check` is read-only and idempotent; install-state.json is
    // updated by the apply paths only (the daemon's auto-update worker
    // bumps `last_check_at` on its periodic poll in P3).
    let result = update::check(current, registry).await;
    match result {
        update::CheckResult::UpToDate { current } => {
            output.print_json(&json!({"current": current, "latest": null}));
            0
        }
        update::CheckResult::Available {
            current,
            latest,
            severity,
            min_supported,
            ..
        } => {
            output.print_json(&json!({
                "current": current,
                "latest": latest,
                "severity": severity.as_str(),
                "min_supported_client": min_supported,
            }));
            0
        }
        update::CheckResult::Failed { reason } => {
            output.print_info(&format!("update check failed: {reason}"));
            // Per the doc's idempotency table, `--check` always exits 0
            // — failure surfaces in the message, not the exit code.
            0
        }
    }
}

async fn apply_in_process(
    output: &OutputConfig,
    current: &str,
    registry: &str,
    args: &UpdateArgs,
) -> i32 {
    let opts = update::ApplyOptions {
        current_version: current.to_string(),
        registry_origin: registry.to_string(),
        download_timeout: Duration::from_secs(60),
        force_cargo_install: args.force_cargo,
        mode: update::ApplyMode::InProcess,
    };
    match update::apply_local(opts).await {
        update::ApplyResult::Applied { from, to, .. } => {
            output.print_info(&format!("Updated {from} → {to} (no daemon was running)"));
            output.print_json(&json!({"from": from, "to": to, "applied": true}));
            0
        }
        update::ApplyResult::UpToDate { current } => {
            output.print_info(&format!("Already on the latest version ({current})"));
            output.print_json(&json!({"current": current, "idempotent": true}));
            0
        }
        update::ApplyResult::RefusedCargoInstall { suggestion } => {
            let err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_REFUSED_CARGO_INSTALL,
                "this binary was installed via `cargo install` — auto-update would not take effect",
            )
            .with_suggestion(suggestion);
            output.print_error(&err);
            5
        }
        update::ApplyResult::Failed { stage, reason } => {
            let err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_VERIFY_FAILED,
                format!("auto-update failed at stage `{}`: {reason}", stage.as_str()),
            );
            output.print_error(&err);
            1
        }
    }
}

async fn apply_via_daemon_rpc(
    output: &OutputConfig,
    current: &str,
    port: u16,
    token: &str,
    force_cargo: bool,
) -> i32 {
    let client = match reqwest::Client::builder()
        .timeout(Duration::from_secs(15))
        .use_rustls_tls()
        .build()
    {
        Ok(c) => c,
        Err(e) => {
            output.print_info(&format!("failed to build HTTP client: {e}"));
            return 1;
        }
    };
    let admin_url = format!("http://127.0.0.1:{port}/admin/update");
    let status_url = format!("{admin_url}/status");

    // 1. POST /admin/update.
    let post_resp = match client
        .post(&admin_url)
        .bearer_auth(token)
        .json(&json!({"force_cargo_install": force_cargo}))
        .send()
        .await
    {
        Ok(r) => r,
        Err(e) => {
            let err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_DAEMON_UNREACHABLE,
                format!("daemon RPC failed: {e}"),
            )
            .with_suggestion("Is the daemon still running? Try `openlatch status`.");
            output.print_error(&err);
            return 6;
        }
    };
    let status = post_resp.status();
    if status != reqwest::StatusCode::ACCEPTED {
        // Map daemon error responses to CLI exit codes.
        let body: serde_json::Value = post_resp.json().await.unwrap_or(serde_json::Value::Null);
        return handle_daemon_error(output, current, status, &body);
    }

    let post_body: serde_json::Value = post_resp.json().await.unwrap_or(serde_json::Value::Null);
    let from = post_body
        .get("from")
        .and_then(|v| v.as_str())
        .unwrap_or(current)
        .to_string();
    let to = post_body
        .get("to")
        .and_then(|v| v.as_str())
        .unwrap_or("?")
        .to_string();
    output.print_info(&format!("Updating {from} → {to}…"));

    // 2. Long-poll status until completed/failed OR daemon stops responding.
    let started = std::time::Instant::now();
    let max_wait = Duration::from_secs(120);
    let poll_interval = Duration::from_secs(1);
    loop {
        if started.elapsed() > max_wait {
            output.print_info("daemon long-poll timed out — the update may still be in progress");
            return 1;
        }
        let resp = client.get(&status_url).bearer_auth(token).send().await;
        match resp {
            Ok(r) if r.status().is_success() => {
                let body: serde_json::Value = r.json().await.unwrap_or(serde_json::Value::Null);
                let status_str = body
                    .get("status")
                    .and_then(|v| v.as_str())
                    .unwrap_or("in_progress");
                match status_str {
                    "completed" => {
                        output.print_info(&format!("Updated {from} → {to}"));
                        output.print_json(&json!({"from": from, "to": to, "applied": true}));
                        return 0;
                    }
                    "failed" => {
                        let stage = body
                            .get("stage")
                            .and_then(|v| v.as_str())
                            .unwrap_or("unknown");
                        let reason = body.get("error").and_then(|v| v.as_str()).unwrap_or("");
                        let err = crate::error::OlError::new(
                            crate::error::ERR_UPDATE_VERIFY_FAILED,
                            format!("update failed at stage `{stage}`: {reason}"),
                        );
                        output.print_error(&err);
                        return 1;
                    }
                    _ => {
                        // in_progress or idle — keep polling.
                        tokio::time::sleep(poll_interval).await;
                    }
                }
            }
            Ok(_) | Err(_) => {
                // Daemon stopped responding mid-poll — this is the
                // expected outcome on Unix where `execv` keeps the PID
                // but momentarily drops the listener, and on Windows
                // where the daemon spawn-detached + exited. Wait for
                // the new daemon to come back up and verify the
                // version bump.
                if let Some(new_version) = wait_for_daemon_version(port, &to).await {
                    output.print_info(&format!("Updated {from} → {new_version}"));
                    output.print_json(&json!({
                        "from": from,
                        "to": new_version,
                        "applied": true,
                    }));
                    return 0;
                }
                let err = crate::error::OlError::new(
                    crate::error::ERR_UPDATE_DAEMON_UNREACHABLE,
                    "daemon did not return after the update — check `openlatch status`",
                );
                output.print_error(&err);
                return 1;
            }
        }
    }
}

fn handle_daemon_error(
    output: &OutputConfig,
    current: &str,
    status: reqwest::StatusCode,
    body: &serde_json::Value,
) -> i32 {
    let message = body
        .pointer("/error/message")
        .and_then(|v| v.as_str())
        .unwrap_or("daemon rejected update");
    let suggestion = body
        .pointer("/error/suggestion")
        .and_then(|v| v.as_str())
        .map(String::from);

    match status.as_u16() {
        409 => {
            // Either cargo-install refusal or already-up-to-date.
            if body.get("idempotent").and_then(|v| v.as_bool()) == Some(true) {
                let cur = body
                    .get("current")
                    .and_then(|v| v.as_str())
                    .unwrap_or(current);
                output.print_info(&format!("Already on the latest version ({cur})"));
                output.print_json(&json!({"current": cur, "idempotent": true}));
                return 0;
            }
            let mut err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_REFUSED_CARGO_INSTALL,
                message.to_string(),
            );
            if let Some(s) = suggestion {
                err = err.with_suggestion(s);
            }
            output.print_error(&err);
            5
        }
        412 => {
            let err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_VERIFY_FAILED,
                message.to_string(),
            );
            output.print_error(&err);
            1
        }
        503 => {
            let err = crate::error::OlError::new(
                crate::error::ERR_DAEMON_START_FAILED,
                message.to_string(),
            );
            output.print_error(&err);
            1
        }
        _ => {
            let err = crate::error::OlError::new(
                crate::error::ERR_UPDATE_VERIFY_FAILED,
                format!("daemon rejected update (HTTP {status}): {message}"),
            );
            output.print_error(&err);
            1
        }
    }
}

/// Poll `/health` until the daemon responds AND reports the expected
/// new version, or until a 60-second budget elapses. Returns the
/// reported version on success.
async fn wait_for_daemon_version(port: u16, expected: &str) -> Option<String> {
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(2))
        .use_rustls_tls()
        .build()
        .ok()?;
    let url = format!("http://127.0.0.1:{port}/health");
    let deadline = std::time::Instant::now() + Duration::from_secs(60);
    while std::time::Instant::now() < deadline {
        if let Ok(resp) = client.get(&url).send().await {
            if let Ok(body) = resp.json::<serde_json::Value>().await {
                if let Some(v) = body.get("version").and_then(|v| v.as_str()) {
                    if v == expected || v.starts_with(expected) {
                        return Some(v.to_string());
                    }
                }
            }
        }
        tokio::time::sleep(Duration::from_secs(2)).await;
    }
    None
}

enum DaemonState {
    RunningAndReachable { port: u16, token: String },
    NotRunning,
    RunningButUnauthenticated,
}

/// `/health` is unauthenticated — use it to detect "is something
/// listening" before trying the bearer-protected `/admin` endpoint.
async fn probe_daemon(port: u16) -> DaemonState {
    let Ok(client) = reqwest::Client::builder()
        .timeout(Duration::from_secs(2))
        .use_rustls_tls()
        .build()
    else {
        return DaemonState::NotRunning;
    };
    let health_url = format!("http://127.0.0.1:{port}/health");
    if client.get(&health_url).send().await.is_err() {
        return DaemonState::NotRunning;
    }

    let token_path = config::openlatch_dir().join("daemon.token");
    let token = match std::fs::read_to_string(&token_path) {
        Ok(t) => t.trim().to_string(),
        Err(_) => return DaemonState::RunningButUnauthenticated,
    };
    if token.is_empty() {
        return DaemonState::RunningButUnauthenticated;
    }
    DaemonState::RunningAndReachable { port, token }
}