//! HTTP API for serve mode, exposes health check data to consumers like MNW.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use axum::extract::{Path, Request, State as AxumState};
use axum::http::StatusCode;
use axum::middleware::{self, Next};
use axum::response::IntoResponse;
use axum::routing::get;
use axum::{Json, Router};
use serde::Serialize;
use tracing::instrument;

use crate::checks::drift::{compute_test_staleness, detect_test_duration_drift};
use crate::config::Config;
use crate::db;
use crate::peer::SharedMeshState;
use crate::types::{HealthSnapshot, LatencyBucket, LatencyStats, TestStaleness};

/// Fixed-window rate limiter, keyed per client IP.
///
/// Previously a single global counter shared across all clients (fuzz-2026-07-06
/// SERIOUS #5): one noisy source burned the whole 60/min budget for everyone.
/// Combined with the layer reorder (auth now runs before this), an unauthenticated
/// attacker can't reach the limiter at all, and an authenticated client's bucket
/// is isolated to its own IP.
#[derive(Clone)]
pub struct PerIpRateLimiter {
    windows: Arc<std::sync::Mutex<HashMap<std::net::IpAddr, (std::time::Instant, u64)>>>,
    max_per_window: u64,
    window_duration: std::time::Duration,
}

impl PerIpRateLimiter {
    pub fn new(max_per_window: u64, window_duration: std::time::Duration) -> Self {
        Self {
            windows: Arc::new(std::sync::Mutex::new(HashMap::new())),
            max_per_window,
            window_duration,
        }
    }

    pub fn try_acquire(&self, ip: std::net::IpAddr) -> bool {
        let now = std::time::Instant::now();
        let mut windows = self.windows.lock().unwrap();
        // Opportunistically evict stale buckets so the map can't grow unbounded
        // from transient/spoofed source IPs.
        windows.retain(|_, (start, _)| now.duration_since(*start) <= self.window_duration);
        let (start, count) = windows.entry(ip).or_insert((now, 0));
        if now.duration_since(*start) > self.window_duration {
            *start = now;
            *count = 1;
            true
        } else {
            *count += 1;
            *count <= self.max_per_window
        }
    }
}

/// Mint a fresh random dashboard session secret (128 bits, hex). Ephemeral: it
/// lives only for the process, so a restart invalidates any leaked cookie.
pub(crate) fn mint_dashboard_token() -> String {
    uuid::Uuid::new_v4().simple().to_string()
}

/// Shared state for the API server.
#[derive(Clone)]
pub struct ApiState {
    pub pool: sqlx::SqlitePool,
    pub config: Arc<Config>,
    pub mesh: Option<SharedMeshState>,
    pub rate_limiter: PerIpRateLimiter,
    /// Ephemeral, per-process dashboard session secret. `Some` only when the
    /// dashboard is enabled. Handed to the browser as an httpOnly cookie (never
    /// embedded in page JS, never the long-lived `api_token`), and accepted by
    /// [`require_bearer_token`] for same-origin dashboard `/api/*` calls.
    pub dashboard_token: Option<Arc<str>>,
}

/// Rate limiting middleware. Returns 429 if the request rate exceeds the limit.
/// Runs *inside* the auth layer, so only authenticated requests are counted and
/// each client IP has its own bucket.
async fn rate_limit(
    AxumState(state): AxumState<ApiState>,
    axum::extract::ConnectInfo(peer): axum::extract::ConnectInfo<std::net::SocketAddr>,
    req: Request,
    next: Next,
) -> impl IntoResponse {
    if state.rate_limiter.try_acquire(peer.ip()) {
        Ok(next.run(req).await)
    } else {
        Err((
            StatusCode::TOO_MANY_REQUESTS,
            Json(serde_json::json!({
                "error": "rate limit exceeded"
            })),
        ))
    }
}

/// Read the `pom_dash` session cookie from a request, if present.
fn dashboard_cookie(req: &Request) -> Option<String> {
    let cookies = req
        .headers()
        .get(axum::http::header::COOKIE)?
        .to_str()
        .ok()?;
    cookies.split(';').find_map(|kv| {
        let (k, v) = kv.split_once('=')?;
        (k.trim() == "pom_dash").then(|| v.trim().to_string())
    })
}

/// Bearer token authentication middleware.
/// If `api_token` is configured, requires `Authorization: Bearer <token>` on every request.
/// If no token is configured, all requests pass through.
async fn require_bearer_token(
    AxumState(state): AxumState<ApiState>,
    req: Request,
    next: Next,
) -> impl IntoResponse {
    let expected = state.config.serve.api_token.as_deref();
    let Some(expected) = expected else {
        return Ok(next.run(req).await);
    };

    use subtle::ConstantTimeEq;

    // Accept the same-origin dashboard session cookie (httpOnly, ephemeral) as an
    // alternative to the bearer token, so the dashboard never has to embed the
    // long-lived api_token in page JS (fuzz-2026-07-06 SERIOUS #4).
    if let (Some(dash), Some(cookie)) = (state.dashboard_token.as_deref(), dashboard_cookie(&req))
        && cookie.as_bytes().ct_eq(dash.as_bytes()).into()
    {
        return Ok(next.run(req).await);
    }

    let auth_header = req
        .headers()
        .get("authorization")
        .and_then(|v| v.to_str().ok());
    match auth_header {
        Some(header) if header.starts_with("Bearer ") => {
            let token = &header[7..];
            // Constant-time comparison to prevent timing side-channels
            if token.as_bytes().ct_eq(expected.as_bytes()).into() {
                Ok(next.run(req).await)
            } else {
                Err((
                    StatusCode::UNAUTHORIZED,
                    Json(serde_json::json!({
                        "error": "invalid bearer token"
                    })),
                ))
            }
        }
        _ => Err((
            StatusCode::UNAUTHORIZED,
            Json(serde_json::json!({
                "error": "missing or malformed Authorization header"
            })),
        )),
    }
}

/// `GET /api/health`: simple health endpoint for PoM itself.
/// Not behind auth, allows external monitoring without credentials.
#[instrument]
async fn self_health() -> impl IntoResponse {
    Json(serde_json::json!({
        "status": "operational",
        "version": env!("CARGO_PKG_VERSION"),
    }))
}

/// Build the axum router for the PoM API.
pub fn router(pool: sqlx::SqlitePool, config: Config, mesh: Option<SharedMeshState>) -> Router {
    // Ephemeral dashboard session secret, minted once per process when the
    // dashboard is enabled, never the api_token, never embedded in page JS.
    let dashboard_token: Option<Arc<str>> = config
        .serve
        .dashboard
        .then(|| Arc::from(crate::api::mint_dashboard_token().as_str()));

    let state = ApiState {
        pool,
        config: Arc::new(config),
        mesh,
        rate_limiter: PerIpRateLimiter::new(60, std::time::Duration::from_mins(1)),
        dashboard_token,
    };

    // Authenticated routes. Order matters: `.layer` wraps outward, so listing
    // `require_bearer_token` LAST makes it the OUTERMOST layer, auth runs before
    // rate_limit, so an unauthenticated request is rejected before it can consume
    // any rate-limit budget (fuzz-2026-07-06 SERIOUS #5).
    let authenticated = Router::new()
        .route("/status.json", get(status_json))
        .route("/api/status", get(status_all))
        .route("/api/status/{target}", get(status_target))
        .route("/api/trends/{target}", get(trends))
        .route("/api/versions", get(versions))
        .route("/api/peer/info", get(peer_info))
        .route("/api/peer/status", get(peer_status))
        .route("/api/mesh", get(mesh_view))
        .layer(middleware::from_fn_with_state(state.clone(), rate_limit))
        .layer(middleware::from_fn_with_state(
            state.clone(),
            require_bearer_token,
        ));

    // Public routes (no auth, no rate limit)
    let public = Router::new().route("/api/health", get(self_health));

    let mut app = public.merge(authenticated);

    if state.config.serve.dashboard {
        app = app.route("/", get(crate::dashboard::dashboard_handler));
    }

    app.with_state(state)
}

// Response types

#[derive(Serialize)]
struct StatusResponse {
    /// Per-target status summaries, keyed by target config name.
    targets: HashMap<String, TargetStatus>,
}

#[derive(Serialize)]
struct TargetStatus {
    /// Human-readable display label for this target.
    label: String,
    /// Most recent health check snapshot. `None` if no checks have been recorded yet.
    latest: Option<SnapshotJson>,
    /// Last 10 health check snapshots, most recent first.
    recent: Vec<SnapshotJson>,
    /// Uptime percentage over the last 24 hours. `None` if no checks in that window.
    uptime_24h: Option<f64>,
    /// Uptime percentage over the last 7 days. `None` if no checks in that window.
    uptime_7d: Option<f64>,
    /// Latency statistics over the last 24 hours. Omitted if no operational checks exist.
    #[serde(skip_serializing_if = "Option::is_none")]
    latency_24h: Option<LatencyStats>,
    /// Latest TLS certificate check result. Omitted if TLS monitoring is not configured.
    #[serde(skip_serializing_if = "Option::is_none")]
    tls: Option<db::TlsCheckRow>,
    /// Test staleness assessment. Omitted if test running is not configured for this target.
    #[serde(skip_serializing_if = "Option::is_none")]
    test_staleness: Option<TestStaleness>,
    /// Currently open incident. Omitted if the target is not in an incident state.
    #[serde(skip_serializing_if = "Option::is_none")]
    current_incident: Option<db::IncidentRow>,
    /// Recent resolved and open incidents (up to 10). Omitted if empty.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    incidents: Vec<db::IncidentRow>,
    /// Latest route check results per path. Omitted if empty.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    route_status: Vec<RouteStatusJson>,
    /// Latest DNS check results. Omitted if empty.
    #[serde(skip_serializing_if = "Vec::is_empty")]
    dns_status: Vec<DnsStatusJson>,
    /// Latest WHOIS check result. Omitted if no WHOIS monitoring is configured.
    #[serde(skip_serializing_if = "Option::is_none")]
    whois: Option<db::WhoisCheckRow>,
    /// Test duration drift warning. Omitted if no drift detected or no test config.
    #[serde(skip_serializing_if = "Option::is_none")]
    test_duration_drift: Option<String>,
}

#[derive(Serialize)]
struct DnsStatusJson {
    name: String,
    record_type: String,
    expected: Vec<String>,
    actual: Vec<String>,
    matches: bool,
    checked_at: String,
}

#[derive(Serialize)]
struct RouteStatusJson {
    path: String,
    status_code: i64,
    ok: bool,
    checked_at: String,
    response_time_ms: i64,
}

#[derive(Serialize)]
struct SnapshotJson {
    /// Health status as a lowercase string (e.g. "operational", "degraded").
    status: String,
    /// Timestamp of the check in RFC 3339 format.
    checked_at: String,
    /// Round-trip response time in milliseconds.
    response_time_ms: i64,
    /// Structured health details from the endpoint. Omitted when unavailable.
    #[serde(skip_serializing_if = "Option::is_none")]
    details: Option<serde_json::Value>,
    /// Error message if the check failed. Omitted on success.
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<String>,
}

impl From<HealthSnapshot> for SnapshotJson {
    fn from(s: HealthSnapshot) -> Self {
        Self {
            status: s.status.to_string(),
            checked_at: s.checked_at,
            response_time_ms: s.response_time_ms,
            details: s
                .details
                .map(|d| serde_json::to_value(d).unwrap_or_default()),
            error: s.error,
        }
    }
}

/// Build a `TargetStatus` for a single target.
#[instrument(skip_all, fields(target = %name))]
async fn build_target_status(
    pool: &sqlx::SqlitePool,
    name: &str,
    label: &str,
    config: &Config,
) -> TargetStatus {
    let recent = db::get_health_history(pool, Some(name), 10)
        .await
        .unwrap_or_default();

    // Extract the version info we need before consuming the snapshots.
    let latest_version = recent
        .first()
        .and_then(|s| s.details.as_ref())
        .and_then(|d| d.version.clone());
    let latest = recent.first().cloned().map(SnapshotJson::from);
    let recent_json: Vec<SnapshotJson> = recent.into_iter().map(SnapshotJson::from).collect();

    let uptime_24h = db::get_uptime_percent(pool, name, 24).await.unwrap_or(None);
    let uptime_7d = db::get_uptime_percent(pool, name, 168)
        .await
        .unwrap_or(None);

    // Compute 24h latency stats from operational checks
    let latency_24h = {
        let cutoff = (chrono::Utc::now() - chrono::Duration::hours(24)).to_rfc3339();
        let times = db::get_response_times(pool, name, &cutoff)
            .await
            .unwrap_or_default();
        let operational_times: Vec<i64> = times
            .iter()
            .filter(|(_, ms)| *ms > 0)
            .map(|(_, ms)| *ms)
            .collect();
        LatencyStats::from_times(&operational_times)
    };

    let tls = db::get_latest_tls_check(pool, name).await.unwrap_or(None);

    // Compute test staleness for targets with test config
    let test_staleness = if let Some(target_config) = config.get_target(name)
        && let Some(tests_config) = &target_config.tests
    {
        let current_version = latest_version.clone();

        let latest_test = db::get_latest_test_run(pool, name).await.unwrap_or(None);

        let tested_version = if let Some(ref test) = latest_test {
            db::get_version_at_time(pool, name, &test.started_at)
                .await
                .unwrap_or(None)
        } else {
            None
        };

        let staleness = compute_test_staleness(
            current_version.as_deref(),
            tested_version.as_deref(),
            latest_test.as_ref().map(|t| t.started_at.as_str()),
            tests_config.staleness_days,
        );
        Some(staleness)
    } else {
        None
    };

    // Compute test duration drift for targets with test config
    let test_duration_drift = if config
        .get_target(name)
        .and_then(|t| t.tests.as_ref())
        .is_some()
    {
        let durations = db::get_test_durations(pool, name, 13)
            .await
            .unwrap_or_default();
        detect_test_duration_drift(&durations, 10, 3, 1.5)
    } else {
        None
    };

    let current_incident = db::get_open_incident(pool, name).await.unwrap_or(None);

    let incidents = db::get_recent_incidents(pool, name, 10)
        .await
        .unwrap_or_default();

    let route_checks = db::get_latest_route_checks(pool, name)
        .await
        .unwrap_or_default();
    let expected_routes: HashSet<&str> = config
        .get_target(name)
        .map(|t| {
            t.expected_routes
                .iter()
                .map(std::string::String::as_str)
                .collect()
        })
        .unwrap_or_default();
    let route_status: Vec<RouteStatusJson> = route_checks
        .into_iter()
        .filter(|r| expected_routes.contains(r.path.as_str()))
        .map(|r| RouteStatusJson {
            path: r.path,
            status_code: r.status_code,
            ok: r.ok,
            checked_at: r.checked_at,
            response_time_ms: r.response_time_ms,
        })
        .collect();

    let dns_checks = db::get_latest_dns_checks(pool, name)
        .await
        .unwrap_or_default();
    let expected_dns: HashSet<(String, String)> = config
        .get_target(name)
        .map(|t| {
            t.dns
                .iter()
                .map(|d| (d.name.clone(), d.record_type.to_string()))
                .collect()
        })
        .unwrap_or_default();
    let dns_status: Vec<DnsStatusJson> = dns_checks
        .into_iter()
        .filter(|r| expected_dns.contains(&(r.name.clone(), r.record_type.clone())))
        .map(|r| DnsStatusJson {
            name: r.name,
            record_type: r.record_type,
            expected: serde_json::from_str(&r.expected).unwrap_or_default(),
            actual: serde_json::from_str(&r.actual).unwrap_or_default(),
            matches: r.matches,
            checked_at: r.checked_at,
        })
        .collect();

    let whois = db::get_latest_whois_check(pool, name).await.unwrap_or(None);

    TargetStatus {
        label: label.to_string(),
        latest,
        recent: recent_json,
        uptime_24h,
        uptime_7d,
        latency_24h,
        tls,
        test_staleness,
        test_duration_drift,
        current_incident,
        incidents,
        route_status,
        dns_status,
        whois,
    }
}

/// `GET /api/status`: JSON summary for all targets.
#[instrument(skip_all)]
async fn status_all(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
    let mut targets = HashMap::new();

    for name in state.config.target_names() {
        if let Some(target_config) = state.config.get_target(&name) {
            let status =
                build_target_status(&state.pool, &name, &target_config.label, &state.config).await;
            targets.insert(name, status);
        }
    }

    Json(StatusResponse { targets })
}

/// `GET /api/status/{target}`: JSON summary for a single target.
#[instrument(skip_all, fields(target = %target))]
async fn status_target(
    AxumState(state): AxumState<ApiState>,
    Path(target): Path<String>,
) -> impl IntoResponse {
    let Some(target_config) = state.config.get_target(&target) else {
        return Err((
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({
                "error": format!("unknown target: {target}")
            })),
        ));
    };

    let status =
        build_target_status(&state.pool, &target, &target_config.label, &state.config).await;
    Ok(Json(status))
}

/// `GET /status.json`: every monitored target restated in the shared
/// cross-service payload the release viewer renders. See `crate::status`.
#[instrument(skip_all)]
async fn status_json(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
    Json(status_payload(&state.pool, &state.config).await)
}

/// The `/status.json` payload, built straight from the database.
///
/// Public because the MCP tools serve the same payload for the local instance
/// without going through HTTP: a session asking what is live should not need a
/// running `pom serve` on its own machine to read its own database. Reading a
/// *remote* instance is what the HTTP path is for.
pub async fn status_payload(pool: &sqlx::SqlitePool, config: &Config) -> ops_status::Payload {
    let targets = build_status_view(pool, config).await;
    crate::status::payload(&targets, chrono::Utc::now())
}

/// Read each target's current signals out of the database into the pure view the
/// payload mapping consumes. The DB lives here; `crate::status::payload` stays a
/// pure function of `(targets, now)`.
async fn build_status_view(
    pool: &sqlx::SqlitePool,
    config: &Config,
) -> Vec<crate::status::TargetView> {
    let mut targets = Vec::new();

    for name in config.target_names() {
        let Some(target_config) = config.get_target(&name) else {
            continue;
        };

        let health = db::get_latest_health(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|s| crate::status::HealthView {
                status: s.status,
                checked_at: s.checked_at,
                version: s.details.and_then(|d| d.version),
                error: s.error,
            });

        let uptime_24h = db::get_uptime_percent(pool, &name, 24)
            .await
            .unwrap_or(None);

        let latency_avg_ms = {
            let cutoff = (chrono::Utc::now() - chrono::Duration::hours(24)).to_rfc3339();
            let times = db::get_response_times(pool, &name, &cutoff)
                .await
                .unwrap_or_default();
            let operational: Vec<i64> = times
                .iter()
                .filter(|(_, ms)| *ms > 0)
                .map(|(_, ms)| *ms)
                .collect();
            LatencyStats::from_times(&operational).map(|s| s.avg_ms)
        };

        let tls = db::get_latest_tls_check(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|r| crate::status::TlsView {
                valid: r.valid,
                days_remaining: r.days_remaining,
                checked_at: r.checked_at,
                error: r.error,
            });

        let incident = db::get_open_incident(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|i| crate::status::IncidentView {
                from_status: i.from_status,
                to_status: i.to_status,
                started_at: i.started_at,
            });

        let whois = db::get_latest_whois_check(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|w| crate::status::WhoisView {
                days_remaining: w.days_remaining,
                checked_at: w.checked_at,
                error: w.error,
            });

        let mut backups = Vec::new();
        if let Some(backup_config) = &target_config.backups {
            for database in &backup_config.databases {
                if let Ok(Some(row)) = db::get_latest_backup_check(pool, &name, database).await {
                    backups.push(crate::status::BackupView {
                        database: row.database_name,
                        status: row.status,
                        age_hours: row.age_hours,
                        checked_at: row.checked_at,
                        error: row.error,
                    });
                }
            }
        }

        let scan_pipeline = db::get_latest_scan_pipeline_check(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|s| crate::status::ScanView {
                issues: s.issue_list(),
                status: s.status,
                checked_at: s.checked_at,
                error: s.error,
            });

        let systemd = db::get_latest_systemd_check(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|s| crate::status::SystemdView {
                issues: s.issue_list(),
                status: s.status,
                checked_at: s.checked_at,
                error: s.error,
            });

        let synckit_fleet = db::get_latest_synckit_fleet_check(pool, &name)
            .await
            .ok()
            .flatten()
            .map(|f| crate::status::SyncKitFleetView {
                versions: f
                    .version_list()
                    .into_iter()
                    .map(|v| (v.client_version, v.devices))
                    .collect(),
                devices: f.devices,
                window_days: f.window_days,
                checked_at: f.checked_at,
                error: f.error,
            });

        // Tests: the latest run plus PoM's staleness verdict, sourced exactly as
        // build_target_status does (version at test time vs current version).
        let tests = if let Some(tests_config) = &target_config.tests {
            let latest_test = db::get_latest_test_run(pool, &name).await.ok().flatten();
            let current_version = health.as_ref().and_then(|h| h.version.clone());
            let tested_version = if let Some(test) = &latest_test {
                db::get_version_at_time(pool, &name, &test.started_at)
                    .await
                    .ok()
                    .flatten()
            } else {
                None
            };
            let staleness = compute_test_staleness(
                current_version.as_deref(),
                tested_version.as_deref(),
                latest_test.as_ref().map(|t| t.started_at.as_str()),
                tests_config.staleness_days,
            );
            Some(crate::status::TestsView {
                ran: latest_test.is_some(),
                passed: latest_test.as_ref().is_some_and(|t| t.passed),
                total_passed: latest_test.as_ref().and_then(|t| t.summary.total_passed),
                total_failed: latest_test.as_ref().and_then(|t| t.summary.total_failed),
                started_at: latest_test.as_ref().map(|t| t.started_at.clone()),
                stale: staleness.stale,
                stale_reason: staleness.reason,
            })
        } else {
            None
        };

        // DNS: one entry per monitored record. Absent config yields no rows and
        // therefore no condition.
        let dns = {
            let rows = db::get_latest_dns_checks(pool, &name)
                .await
                .unwrap_or_default();
            (!rows.is_empty()).then(|| crate::status::DnsView {
                checked_at: rows.iter().map(|r| r.checked_at.clone()).max(),
                records: rows
                    .into_iter()
                    .map(|r| crate::status::DnsRecordView {
                        name: r.name,
                        record_type: r.record_type,
                        matches: r.matches,
                        error: r.error,
                    })
                    .collect(),
            })
        };

        // CORS: one entry per monitored URL.
        let cors = {
            let rows = db::get_latest_cors_checks(pool, &name)
                .await
                .unwrap_or_default();
            (!rows.is_empty()).then(|| crate::status::CorsView {
                checked_at: rows.iter().map(|r| r.checked_at.clone()).max(),
                checks: rows
                    .into_iter()
                    .map(|r| crate::status::CorsCheckView {
                        url: r.url,
                        origin: r.origin,
                        passes: r.passes,
                        error: r.error,
                    })
                    .collect(),
            })
        };

        targets.push(crate::status::TargetView {
            name,
            label: target_config.label.clone(),
            health_configured: target_config.health.is_some(),
            health,
            uptime_24h,
            latency_avg_ms,
            tls,
            incident,
            whois,
            backups,
            scan_pipeline,
            systemd,
            synckit_fleet,
            tests,
            dns,
            cors,
        });
    }

    targets
}

// Peer endpoints

/// `GET /api/peer/info`: Returns this instance's identity info.
#[instrument(skip_all)]
async fn peer_info(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
    let Some(ref mesh) = state.mesh else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({
                "error": "peer mesh not enabled"
            })),
        ));
    };

    let mesh_state = mesh.read().await;
    Ok(Json(
        serde_json::to_value(&mesh_state.instance).unwrap_or_default(),
    ))
}

/// `GET /api/peer/status`: This instance's full view: own info + target statuses + peer summaries.
#[instrument(skip_all)]
async fn peer_status(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
    let Some(ref mesh) = state.mesh else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({
                "error": "peer mesh not enabled"
            })),
        ));
    };

    // Collect mesh data under lock, then drop lock before DB queries
    let (instance, peers) = {
        let mesh_state = mesh.read().await;
        let instance = mesh_state.instance.clone();
        let peers: HashMap<String, serde_json::Value> = mesh_state
            .peers
            .iter()
            .map(|(name, peer)| {
                (
                    name.clone(),
                    serde_json::json!({
                        "status": peer.status,
                        "last_seen": peer.last_seen,
                        "latency_ms": peer.latency_ms,
                    }),
                )
            })
            .collect();
        (instance, peers)
    };

    // Build target statuses (DB queries with no lock held)
    let mut targets = HashMap::new();
    for name in state.config.target_names() {
        if let Some(target_config) = state.config.get_target(&name)
            && let Ok(Some(latest)) = db::get_latest_health(&state.pool, &name).await
        {
            targets.insert(
                name,
                serde_json::json!({
                    "label": target_config.label,
                    "status": latest.status.to_string(),
                    "response_time_ms": latest.response_time_ms,
                    "checked_at": latest.checked_at,
                }),
            );
        }
    }

    Ok(Json(serde_json::json!({
        "instance": instance,
        "targets": targets,
        "peers": peers,
    })))
}

/// `GET /api/mesh`: Aggregated view: self + each peer's cached status.
#[instrument(skip_all)]
async fn mesh_view(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
    let Some(ref mesh) = state.mesh else {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({
                "error": "peer mesh not enabled"
            })),
        ));
    };

    // Collect all mesh data under lock, then drop lock before DB queries
    let (instance, own_peers_json, peer_entries) = {
        let mesh_state = mesh.read().await;
        let instance = mesh_state.instance.clone();
        let own_peers: HashMap<String, serde_json::Value> = mesh_state
            .peers
            .iter()
            .map(|(name, peer)| {
                (
                    name.clone(),
                    serde_json::json!({
                        "status": peer.status,
                        "last_seen": peer.last_seen,
                        "latency_ms": peer.latency_ms,
                    }),
                )
            })
            .collect();
        let peer_entries: Vec<(String, Option<serde_json::Value>, serde_json::Value)> = mesh_state
            .peers
            .iter()
            .map(|(name, peer)| {
                let fallback = serde_json::json!({
                    "status": peer.status,
                    "last_seen": peer.last_seen,
                    "error": "no status data cached",
                });
                (name.clone(), peer.status_data.clone(), fallback)
            })
            .collect();
        (instance, own_peers, peer_entries)
    };

    // Build target statuses (DB queries with no lock held)
    let mut targets = HashMap::new();
    for name in state.config.target_names() {
        if let Some(target_config) = state.config.get_target(&name)
            && let Ok(Some(latest)) = db::get_latest_health(&state.pool, &name).await
        {
            targets.insert(
                name,
                serde_json::json!({
                    "label": target_config.label,
                    "status": latest.status.to_string(),
                    "response_time_ms": latest.response_time_ms,
                    "checked_at": latest.checked_at,
                }),
            );
        }
    }

    let self_entry = serde_json::json!({
        "instance": instance,
        "targets": targets,
        "peers": own_peers_json,
    });

    let mut instances = serde_json::Map::new();
    instances.insert(instance.name.clone(), self_entry);

    for (name, status_data, fallback) in peer_entries {
        // Re-project the peer's cached status into a FIXED schema instead of
        // re-serving its raw JSON verbatim: a compromised/MITM'd peer otherwise
        // injects arbitrary structure into /api/mesh consumers (mesh poisoning,
        // fuzz-2026-07-06 #6). Values are still the peer's to report; the shape is
        // ours. Terminal rendering additionally scrubs the values (display::scrub).
        let entry = status_data.as_ref().map_or(fallback, reproject_peer_status);
        instances.insert(name, entry);
    }

    Ok(Json(serde_json::json!({
        "instances": instances,
    })))
}

/// Extract only the known fields from an untrusted peer's cached status blob,
/// dropping any attacker-injected extra structure. Preserves exactly the paths
/// the mesh consumers (`display::format_mesh`, the dashboard) read.
fn reproject_peer_status(raw: &serde_json::Value) -> serde_json::Value {
    let instance = raw.get("instance").map(|i| {
        serde_json::json!({
            "id": i.get("id").and_then(|v| v.as_str()),
            "name": i.get("name").and_then(|v| v.as_str()),
            "version": i.get("version").and_then(|v| v.as_str()),
            "started_at": i.get("started_at").and_then(|v| v.as_str()),
            "targets": i.get("targets")
                .and_then(|v| v.as_array())
                .map(|a| a.iter().filter_map(|t| t.as_str()).collect::<Vec<_>>())
                .unwrap_or_default(),
        })
    });

    let project_map = |key: &str, f: &dyn Fn(&serde_json::Value) -> serde_json::Value| {
        let mut out = serde_json::Map::new();
        if let Some(obj) = raw.get(key).and_then(|v| v.as_object()) {
            for (k, v) in obj {
                out.insert(k.clone(), f(v));
            }
        }
        serde_json::Value::Object(out)
    };

    let targets = project_map("targets", &|t| {
        serde_json::json!({
            "label": t.get("label").and_then(|v| v.as_str()),
            "status": t.get("status").and_then(|v| v.as_str()),
            "response_time_ms": t.get("response_time_ms").and_then(serde_json::Value::as_i64),
            "checked_at": t.get("checked_at").and_then(|v| v.as_str()),
        })
    });

    let peers = project_map("peers", &|p| {
        serde_json::json!({
            "status": p.get("status").and_then(|v| v.as_str()),
            "last_seen": p.get("last_seen").and_then(|v| v.as_str()),
            "latency_ms": p.get("latency_ms").and_then(serde_json::Value::as_u64),
        })
    });

    serde_json::json!({ "instance": instance, "targets": targets, "peers": peers })
}

// Trends endpoint

#[derive(Serialize, serde::Deserialize)]
pub struct TrendResponse {
    /// Target config name this trend data belongs to.
    pub target: String,
    /// Requested time window in hours (from query param, default 24).
    pub window_hours: u64,
    /// Requested bucket width in minutes (from query param, default 60).
    pub bucket_minutes: u64,
    /// Per-bucket latency statistics within the requested window.
    pub buckets: Vec<LatencyBucket>,
    /// Aggregate latency statistics across the entire requested window.
    pub overall: Option<LatencyStats>,
    /// 7-day baseline latency statistics for drift comparison.
    pub baseline: Option<LatencyStats>,
}

/// `GET /api/trends/{target}?hours=24&bucket_minutes=60`: latency trend data.
#[instrument(skip_all, fields(target = %target))]
async fn trends(
    AxumState(state): AxumState<ApiState>,
    Path(target): Path<String>,
    axum::extract::Query(params): axum::extract::Query<TrendQueryParams>,
) -> impl IntoResponse {
    let Some(_target_config) = state.config.get_target(&target) else {
        return Err((
            StatusCode::NOT_FOUND,
            Json(serde_json::json!({
                "error": format!("unknown target: {target}")
            })),
        ));
    };

    let response = build_trends(
        &state.pool,
        &target,
        params.hours.unwrap_or(24),
        params.bucket_minutes.unwrap_or(60),
    )
    .await;

    Ok(Json(response))
}

/// The latency trend for one target: per-bucket stats within the window, the
/// window aggregate, and a 7-day baseline to read it against.
///
/// Public for the same reason as [`status_payload`]: the MCP tools serve the
/// local instance's answer without requiring a running daemon on this machine.
pub async fn build_trends(
    pool: &sqlx::SqlitePool,
    target: &str,
    hours: u64,
    bucket_minutes: u64,
) -> TrendResponse {
    let cutoff = (chrono::Utc::now() - chrono::Duration::hours(hours as i64)).to_rfc3339();
    let times = db::get_response_times(pool, target, &cutoff)
        .await
        .unwrap_or_default();

    let operational_times: Vec<i64> = times
        .iter()
        .filter(|(_, ms)| *ms > 0)
        .map(|(_, ms)| *ms)
        .collect();
    let overall = LatencyStats::from_times(&operational_times);

    let operational_data: Vec<(String, i64)> =
        times.into_iter().filter(|(_, ms)| *ms > 0).collect();
    let buckets = LatencyStats::bucket_by_time(&operational_data, bucket_minutes);

    // 7d baseline for reference
    let baseline_cutoff = (chrono::Utc::now() - chrono::Duration::hours(168)).to_rfc3339();
    let baseline_times = db::get_response_times(pool, target, &baseline_cutoff)
        .await
        .unwrap_or_default();
    let baseline_operational: Vec<i64> = baseline_times
        .iter()
        .filter(|(_, ms)| *ms > 0)
        .map(|(_, ms)| *ms)
        .collect();
    let baseline = LatencyStats::from_times(&baseline_operational);

    TrendResponse {
        target: target.to_string(),
        window_hours: hours,
        bucket_minutes,
        buckets,
        overall,
        baseline,
    }
}

/// `GET /api/versions`: what every target is running, and how far behind.
///
/// The commits-behind column is measured against a checkout on *this* host, so
/// an instance without the repo serves the rest of the row and leaves that one
/// blank. Reading it from a peer therefore answers "what is live there", not
/// "how far behind is it here".
#[instrument(skip_all)]
async fn versions(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
    match crate::versions::collect(&state.pool, &state.config).await {
        Ok(rows) => Ok(Json(rows)),
        Err(e) => Err((
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(serde_json::json!({ "error": e.to_string() })),
        )),
    }
}

#[derive(serde::Deserialize)]
struct TrendQueryParams {
    /// Time window to query, in hours. Defaults to 24 if omitted.
    hours: Option<u64>,
    /// Width of each latency bucket, in minutes. Defaults to 60 if omitted.
    bucket_minutes: Option<u64>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use axum::http::Request as HttpRequest;
    use tower::ServiceExt;

    fn test_config(api_token: Option<&str>) -> Config {
        let mut config = Config {
            serve: crate::config::ServeConfig::default(),
            instance: crate::config::InstanceConfig::default(),
            targets: HashMap::new(),
            peers: HashMap::new(),
            alerts: None,
        };
        config.serve.api_token = api_token.map(std::string::ToString::to_string);
        config
    }

    #[tokio::test]
    async fn no_token_configured_allows_all_requests() {
        let pool = crate::db::connect_in_memory().await.unwrap();
        let app = router(pool, test_config(None), None);

        let resp = app
            .oneshot(with_connect_info("/api/status", None))
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn valid_token_allows_request() {
        let pool = crate::db::connect_in_memory().await.unwrap();
        let app = router(pool, test_config(Some("secret123")), None);

        let resp = app
            .oneshot(with_connect_info("/api/status", Some("Bearer secret123")))
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn dashboard_uses_cookie_not_embedded_token() {
        // SERIOUS #4: GET / must NOT ship the api_token in the page, and must set
        // an httpOnly session cookie that authenticates the dashboard's /api/* calls.
        let pool = crate::db::connect_in_memory().await.unwrap();
        let mut config = test_config(Some("supersecret-token"));
        config.serve.dashboard = true;
        let app = router(pool, config, None);

        let req = HttpRequest::builder().uri("/").body(Body::empty()).unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let cookie_hdr = resp
            .headers()
            .get(axum::http::header::SET_COOKIE)
            .expect("dashboard must set a session cookie")
            .to_str()
            .unwrap()
            .to_string();
        assert!(cookie_hdr.contains("pom_dash="));
        assert!(cookie_hdr.contains("HttpOnly"));
        assert!(
            !cookie_hdr.contains("supersecret-token"),
            "cookie must not be the api_token"
        );

        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
            .await
            .unwrap();
        let html = String::from_utf8_lossy(&body);
        assert!(
            !html.contains("supersecret-token"),
            "the api_token must never appear in served HTML"
        );

        // The issued cookie authenticates an /api/* call without any bearer token.
        let dash = cookie_hdr.split(';').next().unwrap().trim().to_string(); // "pom_dash=<value>"
        let mut api_req = HttpRequest::builder()
            .uri("/api/status")
            .header(axum::http::header::COOKIE, dash)
            .body(Body::empty())
            .unwrap();
        api_req
            .extensions_mut()
            .insert(axum::extract::ConnectInfo(std::net::SocketAddr::from((
                [127, 0, 0, 1],
                40001,
            ))));
        let api_resp = app.oneshot(api_req).await.unwrap();
        assert_eq!(
            api_resp.status(),
            StatusCode::OK,
            "dashboard cookie must authenticate /api/*"
        );
    }

    #[tokio::test]
    async fn wrong_token_returns_401() {
        let pool = crate::db::connect_in_memory().await.unwrap();
        let app = router(pool, test_config(Some("secret123")), None);

        let req = HttpRequest::builder()
            .uri("/api/status")
            .header("authorization", "Bearer wrong-token")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn missing_header_returns_401() {
        let pool = crate::db::connect_in_memory().await.unwrap();
        let app = router(pool, test_config(Some("secret123")), None);

        let req = HttpRequest::builder()
            .uri("/api/status")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn malformed_header_returns_401() {
        let pool = crate::db::connect_in_memory().await.unwrap();
        let app = router(pool, test_config(Some("secret123")), None);

        let req = HttpRequest::builder()
            .uri("/api/status")
            .header("authorization", "Basic dXNlcjpwYXNz")
            .body(Body::empty())
            .unwrap();
        let resp = app.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    fn ip(n: u8) -> std::net::IpAddr {
        std::net::IpAddr::V4(std::net::Ipv4Addr::new(10, 0, 0, n))
    }

    #[test]
    fn reproject_drops_injected_structure() {
        // #6: a compromised peer stuffs extra structure into its status blob.
        let hostile = serde_json::json!({
            "instance": { "id": "abc", "version": "9.9", "evil_field": {"x": 1} },
            "targets": { "mnw": { "status": "operational", "response_time_ms": 5, "evil": "inject" } },
            "peers": { "p2": { "status": "up", "latency_ms": 3 } },
            "top_level_injection": [1, 2, 3]
        });
        let clean = reproject_peer_status(&hostile);

        // Only the fixed top-level keys survive.
        let obj = clean.as_object().unwrap();
        let mut keys: Vec<&String> = obj.keys().collect();
        keys.sort();
        assert_eq!(keys, vec!["instance", "peers", "targets"]);
        assert!(clean.get("top_level_injection").is_none());

        // Known values are preserved; injected sibling keys are gone.
        assert_eq!(clean["instance"]["id"], "abc");
        assert_eq!(clean["instance"]["version"], "9.9");
        assert!(clean["instance"].get("evil_field").is_none());
        assert_eq!(clean["targets"]["mnw"]["status"], "operational");
        assert!(clean["targets"]["mnw"].get("evil").is_none());
        assert_eq!(clean["peers"]["p2"]["latency_ms"], 3);
    }

    /// Build a GET request carrying a `ConnectInfo<SocketAddr>` extension, which
    /// the real server injects via `into_make_service_with_connect_info` but
    /// `oneshot` does not, the rate-limit layer extracts it.
    fn with_connect_info(uri: &str, bearer: Option<&str>) -> HttpRequest<Body> {
        let mut b = HttpRequest::builder().uri(uri);
        if let Some(h) = bearer {
            b = b.header("authorization", h);
        }
        let mut req = b.body(Body::empty()).unwrap();
        req.extensions_mut()
            .insert(axum::extract::ConnectInfo(std::net::SocketAddr::from((
                [127, 0, 0, 1],
                40000,
            ))));
        req
    }

    #[test]
    fn rate_limiter_allows_within_limit() {
        let limiter = PerIpRateLimiter::new(3, std::time::Duration::from_mins(1));
        assert!(limiter.try_acquire(ip(1)));
        assert!(limiter.try_acquire(ip(1)));
        assert!(limiter.try_acquire(ip(1)));
    }

    #[test]
    fn rate_limiter_blocks_over_limit() {
        let limiter = PerIpRateLimiter::new(2, std::time::Duration::from_mins(1));
        assert!(limiter.try_acquire(ip(1)));
        assert!(limiter.try_acquire(ip(1)));
        assert!(!limiter.try_acquire(ip(1)));
    }

    #[test]
    fn rate_limiter_isolates_clients_by_ip() {
        // SERIOUS #5: one client exhausting its bucket must not affect another.
        let limiter = PerIpRateLimiter::new(1, std::time::Duration::from_mins(1));
        assert!(limiter.try_acquire(ip(1)));
        assert!(
            !limiter.try_acquire(ip(1)),
            "ip(1) is now over its own limit"
        );
        assert!(limiter.try_acquire(ip(2)), "ip(2) has its own fresh bucket");
    }

    #[tokio::test]
    async fn rate_limiter_resets_after_window() {
        let limiter = PerIpRateLimiter::new(1, std::time::Duration::from_millis(10));
        assert!(limiter.try_acquire(ip(1)));
        assert!(!limiter.try_acquire(ip(1)));
        tokio::time::sleep(std::time::Duration::from_millis(15)).await;
        assert!(limiter.try_acquire(ip(1)));
    }
}