openlatch-client 0.1.18

//! `/admin/*` route group — privileged daemon endpoints, all gated by
//! the same Bearer auth as `/hooks` and `/shutdown`.
//!
//! Today this module hosts the manual auto-update RPC:
//! - `POST /admin/update` — kick off an apply pipeline. Returns 202 +
//!   a `stream_url` the CLI long-polls; the daemon owns the swap +
//!   drain + restart sequence.
//! - `GET /admin/update/status` — snapshot of the in-flight pipeline
//!   for long-poll display.
//!
//! Future privileged endpoints (severity gating override,
//! cargo-install force, etc.) belong under the same `/admin/*` prefix
//! and inherit the auth middleware via `router()` below.
//!
//! See `.local/brainstorms/auto-update/PHASE-2-manual-via-rpc.md` § 4.

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

use axum::{
    extract::State,
    http::StatusCode,
    middleware,
    response::IntoResponse,
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};

use crate::daemon::{auth, AppState};
use crate::install_state;
use crate::update::{
    self, ApplyMode, ApplyResult, ApplyStage, CheckResult, UpdateStatusKind, UpdateStatusSnapshot,
};

/// Request body for `POST /admin/update`.
#[derive(Debug, Deserialize, Default)]
#[serde(default)]
pub struct AdminUpdateRequest {
    /// Optional override of the cargo-install gate. Even an
    /// authenticated caller cannot bypass minisign verification — this
    /// only opts in to `cargo install`-managed binaries getting
    /// rewritten by self-replace, which is the user's call to make.
    pub force_cargo_install: bool,
}

/// Response body for the success path of `POST /admin/update` (HTTP 202).
#[derive(Debug, Serialize)]
pub struct AdminUpdateResponse {
    pub started: bool,
    pub from: String,
    pub to: String,
    pub stream_url: &'static str,
}

/// Failure modes for `POST /admin/update`. Each variant maps to the
/// HTTP status code in the brainstorm doc § 4.
#[derive(Debug)]
pub enum AdminError {
    /// `cargo install`-managed daemon refused to rewrite itself.
    CargoInstall,
    /// Already on the latest version.
    AlreadyUpToDate { current: String },
    /// Manifest fetch / probe failed (registry unreachable).
    CheckFailed(String),
    /// `min_supported_client` declared by the new release is greater
    /// than the running daemon's version — apply is refused.
    PreconditionFailed {
        latest: String,
        min_supported: String,
    },
    /// A previous `POST /admin/update` is still running.
    AlreadyInProgress,
}

impl IntoResponse for AdminError {
    fn into_response(self) -> axum::response::Response {
        let (status, body) = match &self {
            Self::CargoInstall => (
                StatusCode::CONFLICT,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_UPDATE_REFUSED_CARGO_INSTALL,
                        "message": "this daemon was installed via `cargo install` — auto-update would not take effect",
                        "suggestion": "Run: cargo install --force --locked openlatch-client"
                    }
                }),
            ),
            Self::AlreadyUpToDate { current } => (
                StatusCode::CONFLICT,
                serde_json::json!({
                    "idempotent": true,
                    "current": current,
                    "message": "already on the latest version"
                }),
            ),
            Self::CheckFailed(reason) => (
                StatusCode::BAD_GATEWAY,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_CLOUD_UNREACHABLE,
                        "message": format!("update check failed: {reason}")
                    }
                }),
            ),
            Self::PreconditionFailed {
                latest,
                min_supported,
            } => (
                StatusCode::PRECONDITION_FAILED,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_UPDATE_VERIFY_FAILED,
                        "message": format!(
                            "release {latest} requires client >= {min_supported}; manual `npm install -g @openlatch/client@{latest}` required"
                        ),
                        "latest": latest,
                        "min_supported": min_supported,
                    }
                }),
            ),
            Self::AlreadyInProgress => (
                StatusCode::SERVICE_UNAVAILABLE,
                serde_json::json!({
                    "error": {
                        "code": crate::error::ERR_DAEMON_START_FAILED,
                        "message": "another auto-update is already in progress"
                    }
                }),
            ),
        };
        (status, Json(body)).into_response()
    }
}

/// Build the `/admin/*` router with bearer auth applied to every route.
pub fn router(state: Arc<AppState>) -> Router<Arc<AppState>> {
    Router::new()
        .route("/admin/update", post(handle_admin_update))
        .route("/admin/update/status", get(handle_admin_update_status))
        .route("/admin/inventory/status", get(handle_inventory_status))
        .route("/admin/inventory/rescan", post(handle_inventory_rescan))
        .route(
            "/admin/inventory/inspect/{source_id}",
            get(handle_inventory_inspect),
        )
        .route("/admin/inventory/projects", get(handle_inventory_projects))
        .route("/admin/inventory/ack", post(handle_inventory_ack))
        .route("/admin/auth/refresh", post(handle_admin_auth_refresh))
        .route_layer(middleware::from_fn_with_state(state, auth::bearer_auth))
}

/// `POST /admin/auth/refresh` — clear the cloud worker's `auth_error`
/// latch after a successful `openlatch auth login`. Returns 200 with the
/// new state so the CLI can confirm the daemon picked up the signal.
///
/// This is the WR-01 escape hatch the credential-poll loop used to do
/// unconditionally on every poll, which caused a 60s flap against
/// permanently-revoked keys. Now the CLI flips the flag exactly once
/// per login, the worker mirrors the change on its next poll, and a
/// stable-revoked key stops re-trying.
async fn handle_admin_auth_refresh(State(state): State<Arc<AppState>>) -> Json<serde_json::Value> {
    let was_auth_error = state
        .cloud_state
        .as_ref()
        .is_some_and(|cs| cs.clear_auth_error());
    if was_auth_error {
        let dir = crate::config::openlatch_dir();
        if let Err(e) = crate::cloud::worker::persist_cloud_state(&dir, false) {
            tracing::warn!(error = %e, "admin auth/refresh: failed to persist cloud_state.json");
        }
        tracing::info!("admin auth/refresh: cleared auth_error after CLI login");
    }
    Json(serde_json::json!({
        "auth_error": false,
        "cleared": was_auth_error,
    }))
}

/// Response body for `GET /admin/inventory/status`.
#[derive(Debug, Serialize)]
pub struct InventoryStatusResponse {
    pub enabled: bool,
    pub cache_size: usize,
    pub manifest_loaded: bool,
    pub pending_alerts: usize,
}

async fn handle_inventory_status(
    State(state): State<Arc<AppState>>,
) -> Json<InventoryStatusResponse> {
    Json(InventoryStatusResponse {
        enabled: state.config.inventory_monitor.enabled,
        cache_size: state.content_hash_cache.len(),
        manifest_loaded: state.config_monitor_request_tx.is_some(),
        pending_alerts: state.pending_alerts.pending_count(),
    })
}

/// Request body for `POST /admin/inventory/rescan`.
#[derive(Debug, Deserialize, Default)]
#[serde(default)]
pub struct InventoryRescanRequest {
    pub path: Option<std::path::PathBuf>,
}

async fn handle_inventory_rescan(
    State(state): State<Arc<AppState>>,
    body: Option<Json<InventoryRescanRequest>>,
) -> StatusCode {
    let path_filter = body.and_then(|Json(b)| b.path);
    if let Some(tx) = &state.config_monitor_request_tx {
        if tx
            .send(crate::daemon::config_monitor::ConfigChangeRequest::ManualRescan { path_filter })
            .await
            .is_err()
        {
            return StatusCode::SERVICE_UNAVAILABLE;
        }
        StatusCode::ACCEPTED
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    }
}

/// `POST /admin/update` — kick off the apply pipeline.
///
/// Returns 202 immediately if the pipeline started; the long-running
/// work runs in a spawned task. The CLI polls `GET
/// /admin/update/status` for progress and treats a connection-drop
/// mid-poll as the expected restart-in-progress signal.
async fn handle_admin_update(
    State(state): State<Arc<AppState>>,
    body: Option<Json<AdminUpdateRequest>>,
) -> Result<axum::response::Response, AdminError> {
    let req = body.map(|Json(b)| b).unwrap_or_default();
    let current_version = env!("CARGO_PKG_VERSION").to_string();

    // 1. Cargo-install gate. Done first so we never hit the network
    //    for a binary we won't update.
    if !req.force_cargo_install
        && matches!(
            install_state::detect_install_method(),
            install_state::InstallMethod::CargoInstall
        )
    {
        return Err(AdminError::CargoInstall);
    }

    // 2. Probe the manifest synchronously — this is cheap and lets us
    //    answer 409/412/502 before committing the in-progress lock.
    let registry_origin = state.config.update.registry_origin.clone();
    let download_timeout = Duration::from_secs(state.config.update.download_timeout_secs.max(1));
    let check = update::check(&current_version, &registry_origin).await;
    let (latest, severity, min_supported, _tarball_url, _tarball_integrity) = match check {
        CheckResult::UpToDate { current } => {
            return Err(AdminError::AlreadyUpToDate { current });
        }
        CheckResult::Failed { reason } => return Err(AdminError::CheckFailed(reason)),
        CheckResult::Available {
            latest,
            severity,
            min_supported,
            tarball_url,
            tarball_integrity,
            ..
        } => (
            latest,
            severity,
            min_supported,
            tarball_url,
            tarball_integrity,
        ),
    };

    if let Some(ref min) = min_supported {
        if !update::version_at_least(&current_version, min) {
            crate::telemetry::capture_global(
                crate::telemetry::Event::update_blocked_by_min_supported(
                    &current_version,
                    &latest,
                    min,
                ),
            );
            return Err(AdminError::PreconditionFailed {
                latest,
                min_supported: min.clone(),
            });
        }
    }

    // 3. Reserve the single-update slot. AcqRel is the standard pattern
    //    for compare-and-swap: Acquire on the prior load, Release on
    //    the new value so other threads see all our subsequent writes.
    if state
        .update_in_progress
        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
        .is_err()
    {
        return Err(AdminError::AlreadyInProgress);
    }

    // 4. Stamp the status snapshot to InProgress so the long-poll
    //    starts seeing real data immediately. We re-stamp at every
    //    stage from inside the spawned task.
    {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        *snap = UpdateStatusSnapshot {
            status: UpdateStatusKind::InProgress,
            stage: Some(ApplyStage::Check),
            from: Some(current_version.clone()),
            to: Some(latest.clone()),
            started_at: Some(install_state::now_rfc3339()),
            ended_at: None,
            error: None,
        };
    }

    // 5. Spawn the apply pipeline. The task owns: download → verify →
    //    extract → sanity → swap → write_sentinel → drain → restart.
    let state_for_task = state.clone();
    let from = current_version.clone();
    let to = latest.clone();
    let response_to = to.clone();
    tokio::spawn(async move {
        let opts = update::ApplyOptions {
            current_version: from.clone(),
            registry_origin,
            download_timeout,
            force_cargo_install: req.force_cargo_install,
            mode: ApplyMode::Rpc,
        };
        run_apply_in_daemon(state_for_task, opts, severity).await;
    });

    Ok((
        StatusCode::ACCEPTED,
        Json(AdminUpdateResponse {
            started: true,
            from: current_version,
            to: response_to,
            stream_url: "/admin/update/status",
        }),
    )
        .into_response())
}

/// `GET /admin/update/status` — long-poll surface for the CLI.
async fn handle_admin_update_status(
    State(state): State<Arc<AppState>>,
) -> Json<UpdateStatusSnapshot> {
    let snap = state
        .update_status
        .lock()
        .expect("status mutex poisoned")
        .clone();
    Json(snap)
}

/// Spawned-task body that drives the apply pipeline inside the daemon
/// process. Owns the drain + restart sequence.
///
/// Reachable to the auto-update worker so the same code path runs
/// whether the apply was triggered manually via the RPC or by the
/// background worker. The caller is responsible for taking the
/// `update_in_progress` lock; this function releases it on every
/// failure path.
pub(crate) async fn run_apply_in_daemon(
    state: Arc<AppState>,
    opts: update::ApplyOptions,
    severity: update::Severity,
) {
    let started = std::time::Instant::now();
    let started_at = install_state::now_rfc3339();

    let stamp_stage = |stage: ApplyStage| {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        snap.stage = Some(stage);
    };
    let mark_failed = |stage: ApplyStage, reason: String| {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        snap.status = UpdateStatusKind::Failed;
        snap.stage = Some(stage);
        snap.error = Some(reason);
        snap.ended_at = Some(install_state::now_rfc3339());
    };
    let mark_completed = |duration_ms: u64| {
        let mut snap = state.update_status.lock().expect("status mutex poisoned");
        snap.status = UpdateStatusKind::Completed;
        snap.stage = None;
        snap.ended_at = Some(install_state::now_rfc3339());
        snap.error = None;
        // Duration is in telemetry; the snapshot doesn't carry it.
        let _ = duration_ms;
    };

    stamp_stage(ApplyStage::Check);
    let artefacts = match update::prepare_swap_artefacts(&opts).await {
        Ok(a) => a,
        Err(ApplyResult::UpToDate { current }) => {
            // Race window: between the synchronous check above and now,
            // we somehow ended up still on the latest. Treat as a no-op
            // success.
            tracing::info!(target: "update", current = %current, "concurrent check found us up-to-date — releasing lock");
            mark_completed(started.elapsed().as_millis() as u64);
            release_lock(&state);
            return;
        }
        Err(ApplyResult::RefusedCargoInstall { suggestion }) => {
            mark_failed(ApplyStage::Check, suggestion);
            release_lock(&state);
            return;
        }
        Err(ApplyResult::Failed { stage, reason }) => {
            mark_failed(stage, reason);
            release_lock(&state);
            return;
        }
        Err(ApplyResult::Applied { .. }) => unreachable!("prepare can't return Applied"),
    };

    // Stage: swap.
    stamp_stage(ApplyStage::Swap);
    let hook_path = match update::locate_hook_binary() {
        Ok(p) => p,
        Err(e) => {
            mark_failed(ApplyStage::Swap, format!("locate hook: {e}"));
            release_lock(&state);
            return;
        }
    };
    let _swap_handle =
        match update::perform_swap(&artefacts.staging_exe, &artefacts.staging_hook, &hook_path) {
            Ok(h) => h,
            Err(e) => {
                mark_failed(ApplyStage::Swap, e.to_string());
                release_lock(&state);
                return;
            }
        };

    // Stage: write sentinel BEFORE we let axum drop connections so the
    // new daemon can pick up the breadcrumb if anything between here
    // and post-restart healthz fails.
    //
    // **A sentinel-write failure is fatal post-swap.** `should_rollback`
    // on the next start short-circuits to `false` when the sentinel is
    // missing — so without it, the supervisor-restart-loop safety net
    // is silently disabled. Re-execing into a binary we can no longer
    // safely roll back from is unacceptable. Best-effort undo the swap
    // (the `.bak` siblings are still on disk) and bail; the user
    // continues running the working OLD binary and gets a loud,
    // actionable error.
    let sentinel = update::UpdateSentinel {
        from: artefacts.from.clone(),
        to: artefacts.to.clone(),
        applied_at: started_at.clone(),
    };
    if let Err(e) = update::write_sentinel(&sentinel) {
        let rollback = update::rollback_from_bak();
        tracing::error!(
            target: "update",
            error = %e,
            rollback_error = ?rollback.as_ref().err(),
            "sentinel write failed post-swap; rolled back the swap to keep the safety net intact",
        );
        mark_failed(
            ApplyStage::Swap,
            format!("sentinel write failed post-swap: {e}"),
        );
        release_lock(&state);
        return;
    }

    // Stage: drain. Trigger graceful shutdown of axum, then race a 5s
    // budget against the listener actually closing. The 1s force-close
    // window mimics the brainstorm doc § 4.
    stamp_stage(ApplyStage::Drain);
    tracing::info!(target: "update", "draining axum prior to restart");
    state.admin_shutdown_request.notify_waiters();
    let drain_deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    tokio::time::sleep_until(drain_deadline).await;
    tokio::time::sleep(Duration::from_secs(1)).await;

    // Telemetry: completion is captured *before* the exec call —
    // execv on Unix never returns, and we want the event to land while
    // the runtime is still alive. Failure of restart_into_new_binary
    // is reported via mark_failed below for the rare case it returns.
    let duration_ms = started.elapsed().as_millis().min(u128::from(u64::MAX)) as u64;
    crate::telemetry::capture_global(crate::telemetry::Event::update_completed(
        &artefacts.from,
        &artefacts.to,
        severity.as_str(),
        opts.mode.as_str(),
        true,
        Some(duration_ms),
        false,
    ));
    mark_completed(duration_ms);

    // Stage: restart. On success this never returns on Unix
    // (execv reuses the PID); on Windows the daemon spawns a detached
    // child and exits the current process. If it does return (rare
    // failure mode like ENOEXEC), we mark_failed so the status surface
    // catches it and leave the daemon running on the OLD binary —
    // the new binary is live on disk and a manual restart picks it up.
    stamp_stage(ApplyStage::Restart);
    if let Err(e) = update::restart_into_new_binary() {
        mark_failed(ApplyStage::Restart, e);
        release_lock(&state);
    }
}

fn release_lock(state: &AppState) {
    state.update_in_progress.store(false, Ordering::Release);
}

// ---------------------------------------------------------------------------
// Inventory P2 admin endpoints — inspect / projects / ignore / unignore / ack
// ---------------------------------------------------------------------------

#[derive(Debug, Serialize)]
pub struct InventoryInspectResponse {
    /// `{agent}:{kind}:{path_hash}` echoed back for stable client display.
    pub source_id: String,
    /// `Some` when the source is currently in the cache; `None` otherwise.
    pub cache_entry: Option<InventoryInspectEntry>,
    /// Pending alerts that match this source_id (delivered or undelivered).
    pub alerts: Vec<crate::daemon::config_monitor::PendingAlert>,
}

#[derive(Debug, Serialize)]
pub struct InventoryInspectEntry {
    pub agent: String,
    pub kind: String,
    pub content_hash: String,
    pub path_hash: String,
    pub path: String,
}

async fn handle_inventory_inspect(
    State(state): State<Arc<AppState>>,
    axum::extract::Path(source_id): axum::extract::Path<String>,
) -> Json<InventoryInspectResponse> {
    let cache_entry = state
        .content_hash_cache
        .snapshot()
        .into_iter()
        .find_map(|e| {
            let path_hash_hex = hex::encode(e.path_hash);
            let id = format!("{}:{}:{}", e.agent, e.kind, path_hash_hex);
            if id == source_id {
                Some(InventoryInspectEntry {
                    agent: e.agent,
                    kind: e.kind,
                    content_hash: hex::encode(e.content_hash),
                    path_hash: path_hash_hex,
                    path: e.path.display().to_string(),
                })
            } else {
                None
            }
        });
    let alerts: Vec<_> = state
        .pending_alerts
        .snapshot()
        .into_iter()
        .filter(|a| a.source_id == source_id)
        .collect();
    Json(InventoryInspectResponse {
        source_id,
        cache_entry,
        alerts,
    })
}

#[derive(Debug, Serialize)]
pub struct InventoryProjectsResponse {
    pub projects: Vec<String>,
}

async fn handle_inventory_projects(
    State(state): State<Arc<AppState>>,
) -> Json<InventoryProjectsResponse> {
    // Project roots are inferred from cwd hashes seen so far; the cache
    // doesn't index them directly, so we surface the unique parent
    // directories of every cached entry as a heuristic. This is enough
    // for `openlatch inventory projects` to show the user where config
    // is actively being watched without us tracking a separate set.
    let mut roots: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
    for e in state.content_hash_cache.snapshot() {
        if let Some(parent) = e.path.parent() {
            roots.insert(parent.display().to_string());
        }
    }
    Json(InventoryProjectsResponse {
        projects: roots.into_iter().collect(),
    })
}

#[derive(Debug, Deserialize, Default)]
#[serde(default)]
pub struct InventoryAckRequest {
    pub alert_id: Option<String>,
}

async fn handle_inventory_ack(
    State(state): State<Arc<AppState>>,
    body: Option<Json<InventoryAckRequest>>,
) -> Json<serde_json::Value> {
    let req = body.map(|Json(b)| b).unwrap_or_default();
    let removed = state.pending_alerts.ack(req.alert_id.as_deref());
    Json(serde_json::json!({
        "acknowledged": removed,
        "alert_id": req.alert_id,
    }))
}