ant-core 0.10.0

use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;

use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::sse::{Event, Sse};
use axum::response::{Html, IntoResponse};
use axum::routing::{get, post};
use axum::{Json, Router};
use tokio::sync::broadcast;
use tokio::sync::RwLock;
use tokio_util::sync::CancellationToken;

use crate::error::Result;
use crate::node::binary::NoopProgress;
use crate::node::daemon::forward::runner::{ForwarderHandle, DEFAULT_POLL_INTERVAL};
use crate::node::daemon::forward::{
    apply_enable, classify_nodes, spawn_log_forwarder, ElasticsearchSink, LogForwardConfig,
    LogForwardEnableRequest, LogForwardResult, LogForwardStatus, LogSink,
};
use crate::node::daemon::health::{DiskThresholds, FleetHealth};
use crate::node::daemon::supervisor::{
    spawn_eviction_monitor, spawn_liveness_monitor, Supervisor, EVICTION_POLL_INTERVAL,
    LIVENESS_POLL_INTERVAL,
};
use crate::node::events::NodeEvent;
use crate::node::registry::NodeRegistry;
use crate::node::types::{
    AddNodeOpts, AddNodeResult, DaemonConfig, DaemonStatus, NodeInfo, NodeStarted, NodeStatus,
    NodeStatusResult, NodeStatusSummary, NodeStopped, RemoveNodeResult, ResetResult,
    StartNodeResult, StopNodeResult,
};

/// Shared application state for the daemon HTTP server.
pub struct AppState {
    pub registry: Arc<RwLock<NodeRegistry>>,
    pub supervisor: Arc<RwLock<Supervisor>>,
    pub event_tx: broadcast::Sender<NodeEvent>,
    pub start_time: Instant,
    pub config: DaemonConfig,
    /// The actual address the server bound to (resolves port 0 to real port).
    pub bound_port: u16,
    /// Latest fleet health snapshot, refreshed by the eviction monitor and served at
    /// `GET /api/v1/health`.
    pub health: Arc<RwLock<FleetHealth>>,
    /// The running log forwarder, if the user has opted in. `None` while forwarding is disabled.
    pub forwarder: Arc<RwLock<Option<ForwarderHandle>>>,
    /// The daemon's shutdown token, kept so a forwarder started later by `enable` still stops when
    /// the daemon does.
    pub shutdown: CancellationToken,
}

/// Start the daemon HTTP server.
///
/// Returns the actual address the server bound to (useful when port is 0).
pub async fn start(
    config: DaemonConfig,
    mut registry: NodeRegistry,
    shutdown: CancellationToken,
) -> Result<SocketAddr> {
    let (event_tx, _) = broadcast::channel(256);

    let addr = SocketAddr::new(config.listen_addr, config.port.unwrap_or(0));
    let listener = tokio::net::TcpListener::bind(addr)
        .await
        .map_err(|e| crate::error::Error::BindError(e.to_string()))?;
    let bound_addr = listener
        .local_addr()
        .map_err(|e| crate::error::Error::BindError(e.to_string()))?;

    // Heal any stale `version` entries in the registry. If an earlier daemon ran without the
    // upgrade-aware supervisor, the on-disk binary may have been replaced without the registry
    // being updated. We re-read each binary's version and persist any differences before the
    // supervisor comes up, so subsequent status queries reflect reality.
    reconcile_registry_versions(&mut registry).await;

    let registry = Arc::new(RwLock::new(registry));
    let supervisor = Arc::new(RwLock::new(Supervisor::new(event_tx.clone())));

    // Adopt node processes spawned by a previous daemon instance. Must run before
    // `axum::serve` starts accepting requests — the window between supervisor
    // creation and adoption is where `/api/v1/nodes/status` would otherwise report
    // live nodes as Stopped (the supervisor's default when it has no runtime entry).
    {
        let reg = registry.read().await;
        let mut sup = supervisor.write().await;
        let adopted = sup.adopt_from_registry(&reg);
        if !adopted.is_empty() {
            tracing::info!(
                "Adopted {} running node(s) from a previous daemon instance: {:?}",
                adopted.len(),
                adopted
            );
        }
    }

    let health = Arc::new(RwLock::new(FleetHealth::healthy()));

    let state = Arc::new(AppState {
        registry: registry.clone(),
        supervisor: supervisor.clone(),
        event_tx: event_tx.clone(),
        start_time: Instant::now(),
        config: config.clone(),
        bound_port: bound_addr.port(),
        health: health.clone(),
        forwarder: Arc::new(RwLock::new(None)),
        shutdown: shutdown.clone(),
    });

    // Background task: if the user has opted into beta log forwarding, resume it. The opt-in is
    // persisted rather than held in daemon memory, so restarting the daemon does not silently stop
    // shipping logs the user asked for.
    start_forwarder_if_enabled(&state).await;

    // Background task: monitor free disk space at node data directories. Refreshes the fleet health
    // snapshot every tick and auto-evicts a node (smallest data dir) on any partition that has
    // fallen to the eviction threshold while ≥2 nodes remain. The threshold is a fixed internal
    // constant (mirroring ant-node's own refuse-to-store reserve), not user-configurable.
    spawn_eviction_monitor(
        registry.clone(),
        supervisor.clone(),
        event_tx.clone(),
        health,
        DiskThresholds::default(),
        EVICTION_POLL_INTERVAL,
        shutdown.clone(),
    );

    // Background task: poll adopted nodes' PIDs for OS liveness. Daemon-spawned nodes
    // get exit detection via `monitor_node`'s owned `Child` handle; adopted nodes don't,
    // so this poll is the only way the supervisor learns when one of them exits.
    spawn_liveness_monitor(
        registry,
        supervisor,
        event_tx,
        LIVENESS_POLL_INTERVAL,
        shutdown.clone(),
    );

    let app = build_router(state.clone());

    // Write port and PID files
    write_file(&config.port_file_path, &bound_addr.port().to_string())?;
    write_file(&config.pid_file_path, &std::process::id().to_string())?;

    let port_file = config.port_file_path.clone();
    let pid_file = config.pid_file_path.clone();

    tokio::spawn(async move {
        axum::serve(listener, app)
            .with_graceful_shutdown(shutdown.cancelled_owned())
            .await
            .ok();

        // Clean up port and PID files on shutdown
        let _ = std::fs::remove_file(&port_file);
        let _ = std::fs::remove_file(&pid_file);
    });

    Ok(bound_addr)
}

fn build_router(state: Arc<AppState>) -> Router {
    use axum::http::HeaderValue;
    use tower_http::cors::{Any, CorsLayer};

    // Restrict CORS to the daemon's own origin to prevent cross-origin CSRF
    // attacks from malicious webpages. Non-browser clients (CLI, AI agents)
    // don't send Origin headers so CORS doesn't affect them.
    let origin = format!("http://127.0.0.1:{}", state.bound_port);
    let cors = CorsLayer::new()
        .allow_origin([origin.parse::<HeaderValue>().unwrap()])
        .allow_methods(Any)
        .allow_headers(Any);

    Router::new()
        .route("/console", get(get_console))
        .route("/api/v1/status", get(get_status))
        .route("/api/v1/health", get(get_health))
        .route("/api/v1/events", get(get_events))
        .route("/api/v1/nodes/status", get(get_nodes_status))
        .route("/api/v1/nodes", post(post_nodes))
        .route(
            "/api/v1/nodes/{id}",
            get(get_node_detail).delete(delete_node),
        )
        .route("/api/v1/nodes/{id}/start", post(post_start_node))
        .route("/api/v1/nodes/start-all", post(post_start_all))
        .route("/api/v1/nodes/{id}/stop", post(post_stop_node))
        .route("/api/v1/nodes/stop-all", post(post_stop_all))
        .route("/api/v1/reset", post(post_reset))
        .route("/api/v1/logs/forward", get(get_log_forward))
        .route("/api/v1/logs/forward/enable", post(post_log_forward_enable))
        .route(
            "/api/v1/logs/forward/disable",
            post(post_log_forward_disable),
        )
        .route("/api/v1/openapi.json", get(get_openapi))
        .layer(cors)
        .with_state(state)
}

async fn get_status(State(state): State<Arc<AppState>>) -> Json<DaemonStatus> {
    let registry = state.registry.read().await;
    let supervisor = state.supervisor.read().await;
    let (running, stopped, errored) = supervisor.node_counts();

    Json(DaemonStatus {
        running: true,
        pid: Some(std::process::id()),
        port: Some(state.bound_port),
        uptime_secs: Some(state.start_time.elapsed().as_secs()),
        nodes_total: registry.len() as u32,
        nodes_running: running,
        nodes_stopped: stopped,
        nodes_errored: errored,
    })
}

/// GET /api/v1/health — Current fleet health (overall level + per-check findings).
///
/// Refreshed by the eviction monitor; reflects disk pressure and the next eviction candidate.
async fn get_health(State(state): State<Arc<AppState>>) -> Json<FleetHealth> {
    Json(state.health.read().await.clone())
}

async fn get_events(
    State(state): State<Arc<AppState>>,
) -> Sse<impl futures_core::Stream<Item = std::result::Result<Event, std::convert::Infallible>>> {
    let mut rx = state.event_tx.subscribe();

    let stream = async_stream::stream! {
        loop {
            match rx.recv().await {
                Ok(event) => {
                    let event_type = event.event_type().to_string();
                    if let Ok(data) = serde_json::to_string(&event) {
                        yield Ok(Event::default().event(event_type).data(data));
                    }
                }
                Err(broadcast::error::RecvError::Lagged(_)) => continue,
                Err(broadcast::error::RecvError::Closed) => break,
            }
        }
    };

    Sse::new(stream)
}

/// GET /api/v1/nodes/status — Get status of all registered nodes.
async fn get_nodes_status(State(state): State<Arc<AppState>>) -> Json<NodeStatusResult> {
    let registry = state.registry.read().await;
    let supervisor = state.supervisor.read().await;

    let mut nodes = Vec::new();
    let mut total_running = 0u32;
    let mut total_stopped = 0u32;

    for config in registry.list() {
        // An evicted node has no live process: its persisted marker takes precedence over any
        // runtime status the supervisor might still report.
        let status = if config.eviction.is_some() {
            NodeStatus::Evicted
        } else {
            supervisor
                .node_status(config.id)
                .unwrap_or(NodeStatus::Stopped)
        };

        match status {
            NodeStatus::Running | NodeStatus::Starting => total_running += 1,
            _ => total_stopped += 1,
        }

        let (pid, uptime_secs) = if config.eviction.is_some() {
            (None, None)
        } else {
            (
                supervisor.node_pid(config.id),
                supervisor.node_uptime_secs(config.id),
            )
        };

        nodes.push(NodeStatusSummary {
            node_id: config.id,
            name: config.service_name.clone(),
            version: config.version.clone(),
            status,
            pid,
            uptime_secs,
            eviction: config.eviction.clone(),
        });
    }

    Json(NodeStatusResult {
        nodes,
        total_running,
        total_stopped,
    })
}

/// GET /api/v1/nodes/:id — Get full detail for a single node.
async fn get_node_detail(
    State(state): State<Arc<AppState>>,
    Path(id): Path<u32>,
) -> std::result::Result<Json<NodeInfo>, (StatusCode, Json<serde_json::Value>)> {
    let registry = state.registry.read().await;
    let config = match registry.get(id) {
        Ok(config) => config.clone(),
        Err(_) => {
            return Err((
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({ "error": format!("Node not found: {id}") })),
            ))
        }
    };

    let supervisor = state.supervisor.read().await;
    // A persisted eviction marker takes precedence over any runtime status.
    let (status, pid, uptime_secs) = if config.eviction.is_some() {
        (NodeStatus::Evicted, None, None)
    } else {
        (
            supervisor.node_status(id).unwrap_or(NodeStatus::Stopped),
            supervisor.node_pid(id),
            supervisor.node_uptime_secs(id),
        )
    };

    Ok(Json(NodeInfo {
        config,
        status,
        pid,
        uptime_secs,
    }))
}

/// POST /api/v1/nodes — Add one or more nodes to the registry.
async fn post_nodes(
    State(state): State<Arc<AppState>>,
    Json(opts): Json<AddNodeOpts>,
) -> std::result::Result<(StatusCode, Json<AddNodeResult>), (StatusCode, Json<serde_json::Value>)> {
    let registry_path = state.config.registry_path.clone();
    let progress = NoopProgress;

    match crate::node::add_nodes(opts, &registry_path, &progress).await {
        Ok(result) => {
            // Update the in-memory registry to stay in sync
            let mut registry = state.registry.write().await;
            if let Ok(fresh) = NodeRegistry::load(&registry_path) {
                *registry = fresh;
            }
            Ok((StatusCode::CREATED, Json(result)))
        }
        Err(e) => Err((
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({ "error": e.to_string() })),
        )),
    }
}

/// DELETE /api/v1/nodes/:id — Remove a node from the registry.
async fn delete_node(
    State(state): State<Arc<AppState>>,
    Path(id): Path<u32>,
) -> std::result::Result<Json<RemoveNodeResult>, (StatusCode, Json<serde_json::Value>)> {
    // Prevent removing a running node (would orphan the process)
    let supervisor = state.supervisor.read().await;
    if supervisor.is_running(id) {
        return Err((
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "error": format!("Cannot remove node {id} while it is running. Stop it first."),
                "current_state": { "node_id": id, "status": "running" }
            })),
        ));
    }
    drop(supervisor);

    let registry_path = state.config.registry_path.clone();

    match crate::node::remove_node(id, &registry_path) {
        Ok(result) => {
            // Update the in-memory registry to stay in sync
            let mut registry = state.registry.write().await;
            if let Ok(fresh) = NodeRegistry::load(&registry_path) {
                *registry = fresh;
            }
            Ok(Json(result))
        }
        Err(crate::error::Error::NodeNotFound(id)) => Err((
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({ "error": format!("Node not found: {id}") })),
        )),
        Err(e) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({ "error": e.to_string() })),
        )),
    }
}

/// POST /api/v1/nodes/:id/start — Start a specific node.
async fn post_start_node(
    State(state): State<Arc<AppState>>,
    Path(id): Path<u32>,
) -> std::result::Result<Json<NodeStarted>, (StatusCode, Json<serde_json::Value>)> {
    let registry = state.registry.read().await;
    let config = match registry.get(id) {
        Ok(config) => config.clone(),
        Err(_) => {
            return Err((
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({ "error": format!("Node not found: {id}") })),
            ))
        }
    };
    drop(registry);

    // An evicted node's data directory is gone; refuse to start it (recovery is to dismiss and
    // re-add, not restart).
    if config.eviction.is_some() {
        return Err((
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "error": format!(
                    "Node {id} has been evicted and cannot be started. Dismiss it with \
                     `ant node dismiss {id}` and add a new node instead."
                ),
                "current_state": { "node_id": id, "status": "evicted" }
            })),
        ));
    }

    let supervisor_ref = state.supervisor.clone();

    // Acquire write lock once for atomic check-and-act (avoids TOCTOU race)
    let mut supervisor = state.supervisor.write().await;
    if supervisor.is_running(id) {
        let pid = supervisor.node_pid(id);
        let uptime_secs = supervisor.node_uptime_secs(id);
        return Err((
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "error": format!("Node {id} is already running"),
                "current_state": {
                    "node_id": id,
                    "status": "running",
                    "pid": pid,
                    "uptime_secs": uptime_secs,
                }
            })),
        ));
    }

    let registry_ref = state.registry.clone();
    match supervisor
        .start_node(&config, supervisor_ref, registry_ref)
        .await
    {
        Ok(started) => Ok(Json(started)),
        Err(crate::error::Error::NodeAlreadyRunning(id)) => {
            let pid = supervisor.node_pid(id);
            let uptime_secs = supervisor.node_uptime_secs(id);
            Err((
                StatusCode::CONFLICT,
                Json(serde_json::json!({
                    "error": format!("Node {id} is already running"),
                    "current_state": {
                        "node_id": id,
                        "status": "running",
                        "pid": pid,
                        "uptime_secs": uptime_secs,
                    }
                })),
            ))
        }
        Err(e) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({ "error": e.to_string() })),
        )),
    }
}

/// POST /api/v1/nodes/start-all — Start all registered nodes.
async fn post_start_all(State(state): State<Arc<AppState>>) -> Json<StartNodeResult> {
    let registry = state.registry.read().await;
    // Evicted nodes have no data directory; skip them silently rather than attempting a spawn that
    // would fail. They remain visible as `Evicted` in status until dismissed.
    let configs: Vec<_> = registry
        .list()
        .into_iter()
        .filter(|c| c.eviction.is_none())
        .cloned()
        .collect();
    drop(registry);

    let mut started = Vec::new();
    let mut failed = Vec::new();
    let mut already_running = Vec::new();

    let supervisor_ref = state.supervisor.clone();
    let registry_ref = state.registry.clone();

    for config in &configs {
        let mut supervisor = state.supervisor.write().await;
        if supervisor.is_running(config.id) {
            already_running.push(config.id);
            continue;
        }

        match supervisor
            .start_node(config, supervisor_ref.clone(), registry_ref.clone())
            .await
        {
            Ok(result) => started.push(result),
            Err(crate::error::Error::NodeAlreadyRunning(id)) => {
                already_running.push(id);
            }
            Err(e) => {
                failed.push(crate::node::types::NodeStartFailed {
                    node_id: config.id,
                    service_name: config.service_name.clone(),
                    error: e.to_string(),
                });
            }
        }
    }

    Json(StartNodeResult {
        started,
        failed,
        already_running,
    })
}

/// POST /api/v1/nodes/:id/stop — Stop a specific node.
async fn post_stop_node(
    State(state): State<Arc<AppState>>,
    Path(id): Path<u32>,
) -> std::result::Result<Json<NodeStopped>, (StatusCode, Json<serde_json::Value>)> {
    let registry = state.registry.read().await;
    let config = match registry.get(id) {
        Ok(config) => config.clone(),
        Err(_) => {
            return Err((
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({ "error": format!("Node not found: {id}") })),
            ))
        }
    };
    drop(registry);

    // An evicted node is already stopped and its data directory deleted; there is nothing to stop.
    if config.eviction.is_some() {
        return Err((
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "error": format!(
                    "Node {id} has been evicted; there is nothing to stop. Dismiss it with \
                     `ant node dismiss {id}`."
                ),
                "current_state": { "node_id": id, "status": "evicted" }
            })),
        ));
    }

    // Acquire write lock once for atomic check-and-act (avoids TOCTOU race)
    let mut supervisor = state.supervisor.write().await;
    if !supervisor.is_running(id) {
        let status = supervisor
            .node_status(id)
            .unwrap_or(crate::node::types::NodeStatus::Stopped);
        return Err((
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "error": format!("Node {id} is not running"),
                "current_state": {
                    "node_id": id,
                    "status": status,
                }
            })),
        ));
    }

    match supervisor.stop_node(id).await {
        Ok(()) => Ok(Json(NodeStopped {
            node_id: id,
            service_name: config.service_name,
        })),
        Err(crate::error::Error::NodeNotRunning(id)) => {
            let status = supervisor
                .node_status(id)
                .unwrap_or(crate::node::types::NodeStatus::Stopped);
            Err((
                StatusCode::CONFLICT,
                Json(serde_json::json!({
                    "error": format!("Node {id} is not running"),
                    "current_state": {
                        "node_id": id,
                        "status": status,
                    }
                })),
            ))
        }
        Err(e) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({ "error": e.to_string() })),
        )),
    }
}

/// POST /api/v1/nodes/stop-all — Stop all running nodes.
async fn post_stop_all(State(state): State<Arc<AppState>>) -> Json<StopNodeResult> {
    let registry = state.registry.read().await;
    // Skip evicted nodes — there is nothing to stop, and they should stay `Evicted` until dismissed.
    let configs: Vec<(u32, String)> = registry
        .list()
        .into_iter()
        .filter(|c| c.eviction.is_none())
        .map(|c| (c.id, c.service_name.clone()))
        .collect();
    drop(registry);

    let mut supervisor = state.supervisor.write().await;
    let result = supervisor.stop_all_nodes(&configs).await;

    Json(result)
}

/// POST /api/v1/reset — Reset all node state.
async fn post_reset(
    State(state): State<Arc<AppState>>,
) -> std::result::Result<Json<ResetResult>, (StatusCode, Json<serde_json::Value>)> {
    // Hold write lock for atomic check-and-act (prevents nodes being started
    // between the running check and the reset operation)
    let supervisor = state.supervisor.write().await;
    let (running, _, _) = supervisor.node_counts();
    if running > 0 {
        return Err((
            StatusCode::CONFLICT,
            Json(serde_json::json!({
                "error": format!("Cannot reset while nodes are running ({running} node(s) still running). Stop all nodes first."),
                "nodes_running": running,
            })),
        ));
    }
    drop(supervisor);

    let registry_path = state.config.registry_path.clone();

    match crate::node::reset(&registry_path) {
        Ok(result) => {
            // Update the in-memory registry to stay in sync
            let mut registry = state.registry.write().await;
            if let Ok(fresh) = NodeRegistry::load(&registry_path) {
                *registry = fresh;
            }
            Ok(Json(result))
        }
        Err(e) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({ "error": e.to_string() })),
        )),
    }
}

/// Load the persisted forwarding config, treating an unreadable one as disabled.
///
/// The daemon must come up whatever state that file is in; a broken config is reported through the
/// status endpoint rather than by refusing to start.
fn load_forward_config() -> LogForwardConfig {
    LogForwardConfig::default_path()
        .and_then(|path| LogForwardConfig::load(&path))
        .unwrap_or_else(|error| {
            tracing::warn!("log forwarding: could not read the saved config: {error}");
            LogForwardConfig::disabled()
        })
}

/// Build the sink a config describes.
fn build_sink(config: &LogForwardConfig) -> Result<Arc<dyn LogSink>> {
    let sink = ElasticsearchSink::new(config.endpoint_base(), &config.token)?;
    Ok(Arc::new(sink))
}

/// Start a forwarder for the persisted config, if forwarding is enabled.
async fn start_forwarder_if_enabled(state: &Arc<AppState>) {
    let config = load_forward_config();
    if !config.enabled {
        return;
    }
    if let Err(error) = start_forwarder(state, config).await {
        tracing::warn!("log forwarding: could not start: {error}");
    }
}

/// Replace any running forwarder with one for `config`.
async fn start_forwarder(state: &Arc<AppState>, config: LogForwardConfig) -> Result<()> {
    let sink = build_sink(&config)?;
    let offsets_path = crate::node::daemon::forward::OffsetStore::default_path()?;
    let endpoint = sink.describe();

    let handle = spawn_log_forwarder(
        state.registry.clone(),
        config,
        sink,
        offsets_path,
        DEFAULT_POLL_INTERVAL,
        state.shutdown.clone(),
    );

    let mut slot = state.forwarder.write().await;
    if let Some(previous) = slot.replace(handle) {
        // Awaited, not merely signalled: two forwarders tailing the same files and shipping to the
        // same endpoint would otherwise overlap for the length of the old one's retry ladder.
        previous.stop_and_wait().await;
    }
    tracing::info!("log forwarding: shipping node logs to {endpoint}");
    Ok(())
}

/// Build the status response, merging persisted config with the live forwarder's counters.
///
/// The node lists always come from the registry rather than the forwarder's snapshot. They are
/// derived from registry state, not runtime state, and reading them live keeps `status` consistent
/// with what `enable` just reported — a snapshot taken before the forwarder's first poll would
/// otherwise show no nodes at all a moment after `enable` listed them.
async fn forward_status(state: &Arc<AppState>) -> LogForwardStatus {
    let config = load_forward_config();
    let mut status = LogForwardStatus::inactive(&config);

    {
        let registry = state.registry.read().await;
        let (forwarding, skipped) = classify_nodes(&registry);
        status.nodes_forwarding = forwarding;
        status.nodes_skipped = skipped;
    }

    let slot = state.forwarder.read().await;
    if let Some(handle) = slot.as_ref().filter(|handle| !handle.is_stopped()) {
        status.active = true;
        status.stats = handle.snapshot().await.stats;
    }

    status
}

/// GET /api/v1/logs/forward — Whether beta log forwarding is on, and what it is doing.
async fn get_log_forward(State(state): State<Arc<AppState>>) -> Json<LogForwardStatus> {
    Json(forward_status(&state).await)
}

/// POST /api/v1/logs/forward/enable — Opt into forwarding node logs to the beta endpoint.
///
/// Running this is the consent act. It is idempotent: enabling while already enabled re-reads the
/// request, restarts the forwarder against it, and reports `already_in_state`.
async fn post_log_forward_enable(
    State(state): State<Arc<AppState>>,
    Json(request): Json<LogForwardEnableRequest>,
) -> std::result::Result<Json<LogForwardResult>, (StatusCode, Json<serde_json::Value>)> {
    let stored = load_forward_config();
    let was_enabled = stored.enabled;

    let config = apply_enable(&stored, &request).map_err(|error| {
        (
            StatusCode::BAD_REQUEST,
            Json(serde_json::json!({ "error": error.to_string() })),
        )
    })?;

    let path = LogForwardConfig::default_path().map_err(internal_error)?;
    config.save(&path).map_err(internal_error)?;

    start_forwarder(&state, config.clone())
        .await
        .map_err(internal_error)?;

    let registry = state.registry.read().await;
    let (nodes_forwarding, nodes_skipped) = classify_nodes(&registry);

    Ok(Json(LogForwardResult {
        enabled: true,
        already_in_state: was_enabled,
        endpoint: config.endpoint.clone(),
        min_level: config.min_level,
        nodes_forwarding,
        nodes_skipped,
        pending_daemon_start: false,
    }))
}

/// POST /api/v1/logs/forward/disable — Stop forwarding.
///
/// Nothing else about any node changes: no restart, no argument change, no data touched.
async fn post_log_forward_disable(
    State(state): State<Arc<AppState>>,
) -> std::result::Result<Json<LogForwardResult>, (StatusCode, Json<serde_json::Value>)> {
    let mut config = load_forward_config();
    let was_enabled = config.enabled;

    config.enabled = false;
    let path = LogForwardConfig::default_path().map_err(internal_error)?;
    config.save(&path).map_err(internal_error)?;

    // Awaited rather than signalled. `disable` is a revocation of consent, so it must not return
    // — and the CLI must not print "Log forwarding stopped" — while a request is still in flight.
    if let Some(handle) = state.forwarder.write().await.take() {
        handle.stop_and_wait().await;
    }

    Ok(Json(LogForwardResult {
        enabled: false,
        already_in_state: !was_enabled,
        endpoint: config.endpoint.clone(),
        min_level: config.min_level,
        nodes_forwarding: Vec::new(),
        nodes_skipped: Vec::new(),
        pending_daemon_start: false,
    }))
}

fn internal_error(error: crate::error::Error) -> (StatusCode, Json<serde_json::Value>) {
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(serde_json::json!({ "error": error.to_string() })),
    )
}

async fn get_openapi() -> impl IntoResponse {
    // TODO: Migrate to utoipa-generated OpenAPI spec. Types already derive
    // utoipa::ToSchema but this spec is still hand-written JSON.
    let spec = serde_json::json!({
        "openapi": "3.1.0",
        "info": {
            "title": "Ant Daemon API",
            "version": "0.1.0",
            "description": "REST API for the ant node management daemon"
        },
        "paths": {
            "/api/v1/status": {
                "get": {
                    "summary": "Daemon status",
                    "description": "Returns daemon health, uptime, and node count summary",
                    "responses": {
                        "200": {
                            "description": "Daemon status",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/DaemonStatus" }
                                }
                            }
                        }
                    }
                }
            },
            "/api/v1/events": {
                "get": {
                    "summary": "Event stream",
                    "description": "SSE stream of real-time node events",
                    "responses": {
                        "200": {
                            "description": "SSE event stream"
                        }
                    }
                }
            },
            "/api/v1/nodes": {
                "post": {
                    "summary": "Add nodes",
                    "description": "Add one or more nodes to the registry",
                    "requestBody": {
                        "required": true,
                        "content": {
                            "application/json": {
                                "schema": { "$ref": "#/components/schemas/AddNodeOpts" }
                            }
                        }
                    },
                    "responses": {
                        "201": {
                            "description": "Nodes added",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/AddNodeResult" }
                                }
                            }
                        },
                        "400": {
                            "description": "Invalid request"
                        }
                    }
                }
            },
            "/api/v1/nodes/{id}": {
                "delete": {
                    "summary": "Remove node",
                    "description": "Remove a node from the registry",
                    "parameters": [{
                        "name": "id",
                        "in": "path",
                        "required": true,
                        "schema": { "type": "integer" }
                    }],
                    "responses": {
                        "200": {
                            "description": "Node removed",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/RemoveNodeResult" }
                                }
                            }
                        },
                        "404": {
                            "description": "Node not found"
                        }
                    }
                }
            },
            "/api/v1/nodes/{id}/start": {
                "post": {
                    "summary": "Start a node",
                    "description": "Start a specific node by ID. Returns 409 if already running with current_state.",
                    "parameters": [{
                        "name": "id",
                        "in": "path",
                        "required": true,
                        "schema": { "type": "integer" }
                    }],
                    "responses": {
                        "200": {
                            "description": "Node started",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/NodeStarted" }
                                }
                            }
                        },
                        "404": {
                            "description": "Node not found"
                        },
                        "409": {
                            "description": "Node already running (includes current_state)"
                        },
                        "500": {
                            "description": "Failed to start node"
                        }
                    }
                }
            },
            "/api/v1/nodes/start-all": {
                "post": {
                    "summary": "Start all nodes",
                    "description": "Start all registered nodes. Returns per-node results.",
                    "responses": {
                        "200": {
                            "description": "Start results",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/StartNodeResult" }
                                }
                            }
                        }
                    }
                }
            },
            "/api/v1/nodes/{id}/stop": {
                "post": {
                    "summary": "Stop a node",
                    "description": "Stop a specific node by ID. Returns 409 if already stopped with current_state.",
                    "parameters": [{
                        "name": "id",
                        "in": "path",
                        "required": true,
                        "schema": { "type": "integer" }
                    }],
                    "responses": {
                        "200": {
                            "description": "Node stopped",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/NodeStopped" }
                                }
                            }
                        },
                        "404": {
                            "description": "Node not found"
                        },
                        "409": {
                            "description": "Node not running (includes current_state)"
                        },
                        "500": {
                            "description": "Failed to stop node"
                        }
                    }
                }
            },
            "/api/v1/nodes/stop-all": {
                "post": {
                    "summary": "Stop all nodes",
                    "description": "Stop all running nodes. Returns per-node results.",
                    "responses": {
                        "200": {
                            "description": "Stop results",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/StopNodeResult" }
                                }
                            }
                        }
                    }
                }
            },
            "/api/v1/reset": {
                "post": {
                    "summary": "Reset all node state",
                    "description": "Remove all node data directories, log directories, and clear the registry. Fails if any nodes are running.",
                    "responses": {
                        "200": {
                            "description": "Reset successful",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/ResetResult" }
                                }
                            }
                        },
                        "409": {
                            "description": "Nodes still running"
                        }
                    }
                }
            },
            "/api/v1/logs/forward": {
                "get": {
                    "summary": "Log forwarding status",
                    "description": "Whether beta log forwarding is enabled, which nodes are being tailed, which are skipped for having no log directory, and delivery counters. Never returns the write token, only a fingerprint of it.",
                    "responses": {
                        "200": {
                            "description": "Forwarding status",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/LogForwardStatus" }
                                }
                            }
                        }
                    }
                }
            },
            "/api/v1/logs/forward/enable": {
                "post": {
                    "summary": "Enable log forwarding",
                    "description": "Opt into forwarding managed nodes' logs to the beta endpoint. This call is the consent act. Omitted fields reuse the stored configuration, so re-enabling after a disable needs no arguments. Idempotent.",
                    "requestBody": {
                        "required": false,
                        "content": {
                            "application/json": {
                                "schema": { "$ref": "#/components/schemas/LogForwardEnableRequest" }
                            }
                        }
                    },
                    "responses": {
                        "200": {
                            "description": "Forwarding enabled",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/LogForwardResult" }
                                }
                            }
                        },
                        "400": {
                            "description": "No token has ever been supplied, or the endpoint is not an http(s) URL"
                        }
                    }
                }
            },
            "/api/v1/logs/forward/disable": {
                "post": {
                    "summary": "Disable log forwarding",
                    "description": "Stop forwarding. Nothing else about any node changes: no restart, no argument change, no data touched. Idempotent.",
                    "responses": {
                        "200": {
                            "description": "Forwarding disabled",
                            "content": {
                                "application/json": {
                                    "schema": { "$ref": "#/components/schemas/LogForwardResult" }
                                }
                            }
                        }
                    }
                }
            }
        },
        "components": {
            "schemas": {
                "LogLevel": {
                    "type": "string",
                    "enum": ["trace", "debug", "info", "warn", "error"]
                },
                "ForwardingNode": {
                    "type": "object",
                    "properties": {
                        "node_id": { "type": "integer" },
                        "service": { "type": "string" },
                        "log_dir": { "type": "string" }
                    }
                },
                "SkippedNode": {
                    "type": "object",
                    "description": "A node that cannot be forwarded, with the reason. Almost always a node added without --log-dir-path, which writes no log files at all.",
                    "properties": {
                        "node_id": { "type": "integer" },
                        "service": { "type": "string" },
                        "reason": { "type": "string" }
                    }
                },
                "ForwardStats": {
                    "type": "object",
                    "description": "Counters since the daemon started; reset on restart.",
                    "properties": {
                        "events_forwarded": { "type": "integer" },
                        "events_dropped_by_level": { "type": "integer" },
                        "events_dropped_by_overflow": { "type": "integer" },
                        "batches_sent": { "type": "integer" },
                        "batches_failed": { "type": "integer" },
                        "last_success_unix": { "type": "integer", "nullable": true },
                        "last_error": { "type": "string", "nullable": true }
                    }
                },
                "LogForwardStatus": {
                    "type": "object",
                    "properties": {
                        "enabled": { "type": "boolean" },
                        "endpoint": { "type": "string" },
                        "index_prefix": { "type": "string" },
                        "min_level": { "$ref": "#/components/schemas/LogLevel" },
                        "token_fingerprint": { "type": "string", "nullable": true, "description": "Short non-reversible identifier for the configured token. The token itself is never returned." },
                        "active": { "type": "boolean", "description": "Whether the background forwarder is currently running." },
                        "nodes_forwarding": { "type": "array", "items": { "$ref": "#/components/schemas/ForwardingNode" } },
                        "nodes_skipped": { "type": "array", "items": { "$ref": "#/components/schemas/SkippedNode" } },
                        "stats": { "$ref": "#/components/schemas/ForwardStats" }
                    }
                },
                "LogForwardEnableRequest": {
                    "type": "object",
                    "properties": {
                        "token": { "type": "string", "nullable": true, "description": "Write-only Elasticsearch API key. Reuses the stored one when omitted." },
                        "endpoint": { "type": "string", "nullable": true },
                        "min_level": { "$ref": "#/components/schemas/LogLevel", "nullable": true }
                    }
                },
                "LogForwardResult": {
                    "type": "object",
                    "properties": {
                        "enabled": { "type": "boolean" },
                        "already_in_state": { "type": "boolean" },
                        "endpoint": { "type": "string" },
                        "min_level": { "$ref": "#/components/schemas/LogLevel" },
                        "nodes_forwarding": { "type": "array", "items": { "$ref": "#/components/schemas/ForwardingNode" } },
                        "nodes_skipped": { "type": "array", "items": { "$ref": "#/components/schemas/SkippedNode" } },
                        "pending_daemon_start": { "type": "boolean", "description": "Set when the config was saved but no forwarder could be started because the daemon is not running." }
                    }
                },
                "DaemonStatus": {
                    "type": "object",
                    "properties": {
                        "running": { "type": "boolean" },
                        "pid": { "type": "integer", "nullable": true },
                        "port": { "type": "integer", "nullable": true },
                        "uptime_secs": { "type": "integer", "nullable": true },
                        "nodes_total": { "type": "integer" },
                        "nodes_running": { "type": "integer" },
                        "nodes_stopped": { "type": "integer" },
                        "nodes_errored": { "type": "integer" }
                    }
                }
            }
        }
    });
    Json(spec)
}

async fn get_console() -> Html<&'static str> {
    Html(include_str!("console.html"))
}

fn write_file(path: &PathBuf, contents: &str) -> Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    std::fs::write(path, contents)?;
    Ok(())
}

/// Refresh each registered node's `version` against what its on-disk binary reports.
///
/// Intended as a one-time pass at daemon startup to heal registries left in a stale state by
/// earlier daemon versions that didn't track auto-upgrades. Missing binaries and transient
/// `--version` failures are silently skipped so daemon startup never aborts on this.
async fn reconcile_registry_versions(registry: &mut NodeRegistry) {
    let node_ids: Vec<u32> = registry.list().iter().map(|c| c.id).collect();
    let mut changed = false;

    for id in node_ids {
        let (binary_path, recorded_version) = match registry.get(id) {
            Ok(c) => (c.binary_path.clone(), c.version.clone()),
            Err(_) => continue,
        };

        if !binary_path.exists() {
            continue;
        }

        let Ok(disk_version) = crate::node::binary::extract_version(&binary_path).await else {
            continue;
        };

        if disk_version == recorded_version {
            continue;
        }

        if let Ok(entry) = registry.get_mut(id) {
            entry.version = disk_version;
            changed = true;
        }
    }

    if changed {
        let _ = registry.save();
    }
}

#[cfg(all(test, unix))]
mod tests {
    use super::*;
    use crate::node::registry::NodeRegistry;
    use crate::node::types::{EvmNetwork, NodeConfig};
    use std::collections::HashMap;
    use std::os::unix::fs::PermissionsExt;

    fn write_fake_binary(path: &std::path::Path, stdout: &str) {
        let script = format!("#!/bin/sh\nprintf '%s\\n' '{stdout}'\n");
        std::fs::write(path, script).unwrap();
        let mut perm = std::fs::metadata(path).unwrap().permissions();
        perm.set_mode(0o755);
        std::fs::set_permissions(path, perm).unwrap();
    }

    fn seed_config(binary_path: PathBuf, version: &str, data_dir: PathBuf) -> NodeConfig {
        NodeConfig {
            id: 0,
            service_name: String::new(),
            rewards_address: "0x0".into(),
            data_dir,
            log_dir: None,
            node_port: None,
            binary_path,
            version: version.into(),
            env_variables: HashMap::new(),
            bootstrap_peers: vec![],
            upgrade_channel: None,
            evm_network: EvmNetwork::default(),
            eviction: None,
        }
    }

    #[tokio::test]
    async fn reconcile_updates_stale_version_and_persists() {
        let tmp = tempfile::tempdir().unwrap();
        let reg_path = tmp.path().join("registry.json");
        let bin_path = tmp.path().join("ant-node");
        write_fake_binary(&bin_path, "ant-node 0.10.11-rc.1");

        let mut registry = NodeRegistry::load(&reg_path).unwrap();
        let id = registry.add(seed_config(
            bin_path.clone(),
            "0.10.1",
            tmp.path().join("data"),
        ));
        registry.save().unwrap();

        reconcile_registry_versions(&mut registry).await;

        assert_eq!(registry.get(id).unwrap().version, "0.10.11-rc.1");

        let reloaded = NodeRegistry::load(&reg_path).unwrap();
        assert_eq!(reloaded.get(id).unwrap().version, "0.10.11-rc.1");
    }

    #[tokio::test]
    async fn reconcile_leaves_matching_version_alone() {
        let tmp = tempfile::tempdir().unwrap();
        let reg_path = tmp.path().join("registry.json");
        let bin_path = tmp.path().join("ant-node");
        write_fake_binary(&bin_path, "ant-node 0.10.1");

        let mut registry = NodeRegistry::load(&reg_path).unwrap();
        let id = registry.add(seed_config(
            bin_path.clone(),
            "0.10.1",
            tmp.path().join("data"),
        ));

        reconcile_registry_versions(&mut registry).await;

        assert_eq!(registry.get(id).unwrap().version, "0.10.1");
    }

    #[tokio::test]
    async fn reconcile_skips_missing_binary() {
        let tmp = tempfile::tempdir().unwrap();
        let reg_path = tmp.path().join("registry.json");

        let mut registry = NodeRegistry::load(&reg_path).unwrap();
        let id = registry.add(seed_config(
            tmp.path().join("does-not-exist"),
            "0.10.1",
            tmp.path().join("data"),
        ));

        reconcile_registry_versions(&mut registry).await;

        assert_eq!(registry.get(id).unwrap().version, "0.10.1");
    }
}