//! /admin/stats: how this node is doing, in numbers. JSON, or the Prometheus text
//! format when the client asks for it (`prometheus`).

mod prometheus;

use axum::{
    Json,
    extract::State,
    http::{HeaderMap, header},
    response::{IntoResponse, Response},
};
use serde::Serialize;

use super::peers::EpochInfo;
use crate::state::{AppState, Incidents};
use crate::swarm::{members, stability, topology};
use crate::transport::traffic::PeerTraffic;

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct StatsResponse {
    public_key: String,
    alias: Option<String>,
    uptime_secs: u64,
    ready: bool,
    epoch: EpochInfo,
    document: DocumentStats,
    catch_up: CatchUpStats,
    topology: TopologyStats,
    stability: StabilityStats,
    /// HELLO_REPLYs to an earlier run of ours, ignored.
    stale_hello_replies: u64,
    /// All peers together, including ones no longer listed in /peers.
    traffic: PeerTraffic,
    /// The peers listed in /peers together (a forgotten peer's are gone).
    incidents: Incidents,
    propagation: PropagationStats,
}

/// How long changes written elsewhere took from their commit there until applied here,
/// once caught up. Includes the clock skew between the writer and us.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct PropagationStats {
    count: u64,
    sum_ms: u64,
    /// Changes stamped later than our clock on arrival (writer's clock ahead).
    ahead: u64,
    /// Not cumulative; the last bucket has no bound.
    buckets: Vec<PropagationBucket>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct PropagationBucket {
    le_ms: Option<u64>,
    count: u64,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct DocumentStats {
    heads: usize,
    /// Most heads after applying a sync message (concurrent changes not yet merged).
    max_heads: usize,
    /// Size as saved (compressed): about what a joining node downloads. Measured in the
    /// background, at least a minute apart (longer for large documents); absent until
    /// first measured.
    saved_bytes: Option<u64>,
    /// How long ago `saved_bytes` was measured.
    saved_bytes_secs_ago: Option<u64>,
    /// How long the last measurement took (copy and save), in milliseconds.
    saved_bytes_took_ms: Option<u64>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct CatchUpStats {
    /// Catch-up sources chosen (more than one: it switched).
    sources: u32,
    /// Seconds from start until the initial sync was done; absent until then (and on
    /// the genesis node).
    done_after_secs: Option<f64>,
    /// Sync payload received until then.
    sync_bytes: Option<u64>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct TopologyStats {
    k: usize,
    /// Absent until ready with a member list.
    view: Option<TopologyView>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct TopologyView {
    /// Registered members, us included.
    members: usize,
    /// Members other than us not judged unreachable.
    reachable: usize,
    stable: usize,
    neighbours: usize,
    stable_neighbours: usize,
    full_mesh: bool,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct StabilityStats {
    /// As published in the member registry.
    tier: &'static str,
    /// Fraction of the observed time (up to a day) with a live peer.
    connected_fraction: f64,
    observed_secs: u64,
}

/// `GET /admin/stats`.
pub(crate) async fn get_stats(State(state): State<AppState>, headers: HeaderMap) -> Response {
    let stats = collect(&state).await;
    let accept = headers
        .get(header::ACCEPT)
        .and_then(|value| value.to_str().ok());
    if prometheus::wanted(accept) {
        let body = prometheus::render(&state, &stats).await;
        ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response()
    } else {
        Json(stats).into_response()
    }
}

async fn collect(state: &AppState) -> StatsResponse {
    let node = state.stats.snapshot();

    let (heads, tier) = {
        let mut replica = state.doc.write().await;
        let heads = replica.doc.get_heads().len();
        (heads, members::tier(&replica.doc, state.local_id()))
    };
    let saved = state.stats.saved_size();

    let (connected_fraction, observed) = stability::measured(state);

    let mut incidents = Incidents::default();
    let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect();
    for peer_state in peers {
        incidents.add(&peer_state.lock().await.incidents);
    }

    StatsResponse {
        public_key: state.local_id().to_owned(),
        alias: state.alias.clone(),
        uptime_secs: node.uptime.as_secs(),
        ready: state.is_ready().await,
        epoch: state.local_epoch().into(),
        document: DocumentStats {
            heads,
            max_heads: node.max_heads.max(heads),
            saved_bytes: saved.map(|s| s.bytes),
            saved_bytes_secs_ago: saved.map(|s| s.at.elapsed().as_secs()),
            saved_bytes_took_ms: saved.map(|s| s.took.as_millis() as u64),
        },
        catch_up: CatchUpStats {
            sources: node.catch_up_sources,
            done_after_secs: node.caught_up.map(|c| c.after.as_secs_f64()),
            sync_bytes: node.caught_up.map(|c| c.sync_bytes),
        },
        topology: TopologyStats {
            k: state.neighbours_k,
            view: topology::summary(state).map(|s| TopologyView {
                members: s.members,
                reachable: s.reachable,
                stable: s.stable,
                neighbours: s.neighbours,
                stable_neighbours: s.stable_neighbours,
                full_mesh: s.full_mesh,
            }),
        },
        stability: StabilityStats {
            tier: tier.as_str(),
            connected_fraction,
            observed_secs: observed.as_secs(),
        },
        stale_hello_replies: node.stale_hello_replies,
        traffic: state.traffic.total(),
        incidents,
        propagation: PropagationStats {
            count: node.propagation.count,
            sum_ms: node.propagation.sum_ms,
            ahead: node.propagation.ahead,
            buckets: node
                .propagation
                .buckets()
                .map(|(le_ms, count)| PropagationBucket { le_ms, count })
                .collect(),
        },
    }
}