//! /peers handlers.

use std::collections::BTreeMap;

use axum::{
    Json,
    extract::{Path, Query, State},
    http::StatusCode,
};
use serde::{Deserialize, Serialize};
use tokio::time::Instant;
use tracing::warn;

use super::store_status;
use crate::document::replica::EpochRecord;
use crate::document::store;
use crate::locks::MutexExt;
use crate::state::{AppState, HelloState, Incidents, Liveness};
use crate::swarm::peer::forgotten_path;
use crate::transport::Destination;
use crate::transport::paths::ClockOffset;
use crate::transport::traffic::PeerTraffic;

#[derive(Debug, Serialize)]
#[serde(rename_all = "lowercase")]
pub(crate) enum PeerLivenessStatus {
    /// Nothing heard from it since this node started.
    Unknown,
    Active,
    Dormant,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PeerStatus {
    /// The name the peer gave itself in the member list, if any (not unique).
    pub(crate) alias: Option<String>,
    /// Where this node reaches it, or would (`Transport::addresses_of`).
    pub(crate) addresses: Vec<String>,
    /// Where messages to it go now: the address, whether the peer proved itself there,
    /// and the connection there (libp2p: direct or relayed). Absent if nowhere.
    pub(crate) destination: Option<Destination>,
    pub(crate) validated_swarm: Option<String>,
    pub(crate) status: PeerLivenessStatus,
    pub(crate) handshake_complete: bool,
    pub(crate) last_seen_secs_ago: Option<u64>,
    pub(crate) dormant_for_secs: Option<u64>,
    /// Since a sync message from it was applied (or the epoch changed).
    pub(crate) last_sync_secs_ago: Option<u64>,
    /// How long its heads have differed from ours with no sync message from it in
    /// between; absent while they match.
    pub(crate) diverged_for_secs: Option<u64>,
    pub(crate) has_pending_delivery: bool,
    /// Epoch the peer last advertised; absent until it has sent a message.
    pub(crate) epoch: Option<EpochRecord>,
    /// Whether that epoch equals ours, i.e. whether we sync with this peer.
    pub(crate) same_epoch: Option<bool>,
    /// Number of heads in its last keepalive; absent before one, or when it had more
    /// than 32 (then only their digest and some of them are sent).
    pub(crate) heads: Option<usize>,
    /// We keep it alive and sync with it on our own; other peers we only answer.
    pub(crate) neighbour: bool,
    /// The members it said it reaches (`swarm::reach`), by node ID, while that holds;
    /// absent before it said so in this session. Keys of members not in our member
    /// list are left out.
    pub(crate) reaches: Option<Vec<String>>,
    /// Messages and payload bytes exchanged with it, by kind, since we started (or
    /// since it was last listed).
    pub(crate) traffic: PeerTraffic,
    pub(crate) incidents: Incidents,
    /// Smoothed sync throughput (`swarm::throughput`); absent before the first large
    /// exchange.
    pub(crate) sync_throughput: Option<ThroughputInfo>,
    /// How far its clock is ahead of ours, give or take (`Transport::clock_offset`);
    /// absent before a PONG from it.
    pub(crate) clock: Option<ClockOffset>,
    /// How long its sync messages take to come (`swarm::hop`); absent before one came
    /// while we knew its clock.
    pub(crate) hop_delay: Option<HopDelayInfo>,
}

/// How long a peer's sync messages take to come, by its clock and ours, its offset
/// taken off: good to half the round trip (the offset's error), the same for every
/// message while the offset stands, so the minutes' means show a trend exactly.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct HopDelayInfo {
    /// Messages measured since we started.
    pub(crate) count: u64,
    pub(crate) mean_ms: u64,
    /// The bucket bounds by which half, and 99%, came (`/admin/stats`' propagation
    /// buckets); absent if beyond the last.
    pub(crate) half_within_ms: Option<u64>,
    pub(crate) p99_within_ms: Option<u64>,
    /// Messages that took less than nothing by the clocks: the offset's error was larger
    /// (counted as 0 above).
    pub(crate) below_zero: u64,
    /// The mean of each of the last 15 minutes, oldest first (the last is the current
    /// one); null for a minute without messages. May be negative, as `belowZero`.
    pub(crate) by_minute_ms: Vec<Option<i64>>,
    /// Half the round trip to it now, from PINGs (which wait behind nothing): what a
    /// message would take with nothing waiting anywhere; what the delay takes beyond it
    /// waits in queues (and is the offset's error). Absent without a recent round trip.
    pub(crate) half_rtt_ms: Option<u64>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ThroughputInfo {
    pub(crate) bytes_per_sec: f64,
    pub(crate) samples: u32,
    /// Age of the last sample.
    pub(crate) secs_ago: u64,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PeerListResponse {
    /// Our own alias (`--alias`).
    pub(crate) local_alias: Option<String>,
    pub(crate) local_epoch: EpochRecord,
    /// Number of heads of our document. Several heads are concurrent changes no later
    /// change has merged yet.
    pub(crate) local_heads: usize,
    /// Where this node can be reached (`Transport::local_addresses`).
    pub(crate) local_addresses: Vec<String>,
    /// The members we reach (`swarm::reach`), as we last told our neighbours.
    pub(crate) local_reaches: Vec<String>,
    /// By node ID.
    pub(crate) peers: BTreeMap<String, PeerStatus>,
}

#[derive(Debug, Deserialize)]
pub(crate) struct DeletePeerQuery {
    /// The checks to override (`force`): `alive`.
    force: Option<String>,
}

/// `GET /peers` (`docs/openapi.yaml`).
pub(crate) async fn list_peers(
    State(state): State<AppState>,
) -> Result<Json<PeerListResponse>, StatusCode> {
    Ok(Json(peer_list(&state).await))
}

/// What `GET /peers` answers, for others to show too (the dashboard).
pub(crate) async fn peer_list(state: &AppState) -> PeerListResponse {
    let state = state.clone();
    let snapshot = state.peers_snapshot().await;

    let now = Instant::now();
    let local_epoch = state.local_epoch();
    let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc);
    let names = crate::swarm::reach::names(&state).await;
    let named = |keys: Vec<u64>| -> Vec<String> {
        let mut named: Vec<String> = keys
            .iter()
            .filter_map(|key| names.get(key).cloned())
            .collect();
        named.sort();
        named
    };
    let mut peers = BTreeMap::new();

    for (peer, peer_state_mutex) in snapshot {
        let neighbour = crate::swarm::topology::is_neighbour(&state, &peer);
        let traffic = state
            .traffic
            .node(&peer, &state.transport.identities_of(&peer));
        let sync_throughput = state
            .throughput
            .lock_or_recover()
            .get(&peer)
            .map(|estimate| ThroughputInfo {
                bytes_per_sec: estimate.bytes_per_sec.round(),
                samples: estimate.samples,
                secs_ago: now.saturating_duration_since(estimate.at).as_secs(),
            });
        let ps = peer_state_mutex.lock().await;

        let (status, dormant_for_secs) = match ps.liveness {
            Liveness::Unknown => (PeerLivenessStatus::Unknown, None),
            Liveness::Active => (PeerLivenessStatus::Active, None),
            Liveness::Dormant { since } => (
                PeerLivenessStatus::Dormant,
                Some(now.saturating_duration_since(since).as_secs()),
            ),
        };

        let alias = aliases.remove(&peer);
        let addresses = state.transport.addresses_of(&peer);
        let destination = state.transport.destination(&peer);
        let clock = state.transport.clock_offset(&peer);
        let hops = &ps.hop_delays;
        let hop_delay = (hops.all.count > 0).then(|| HopDelayInfo {
            count: hops.all.count,
            mean_ms: hops.all.sum_ms / hops.all.count,
            half_within_ms: hops.all.within(0.5),
            p99_within_ms: hops.all.within(0.99),
            below_zero: hops.all.ahead,
            by_minute_ms: hops.by_minute(now),
            half_rtt_ms: state
                .lock_rtt()
                .effective(&peer, now, state.timing.rtt_fresh_for())
                .map(|rtt| rtt.as_millis() as u64 / 2),
        });
        peers.insert(
            peer,
            PeerStatus {
                alias,
                addresses,
                destination,
                validated_swarm: ps.standing.swarm_id().map(str::to_owned),
                status,
                handshake_complete: ps.hello_state == HelloState::Established,
                last_seen_secs_ago: ps
                    .last_seen
                    .map(|t| now.saturating_duration_since(t).as_secs()),
                dormant_for_secs,
                last_sync_secs_ago: ps
                    .last_sync_rx
                    .map(|t| now.saturating_duration_since(t).as_secs()),
                diverged_for_secs: ps
                    .diverged_since
                    .map(|t| now.saturating_duration_since(t).as_secs()),
                has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(),
                epoch: ps.epoch.map(EpochRecord::from),
                same_epoch: ps.epoch.map(|epoch| epoch == local_epoch),
                heads: ps.reported_heads,
                neighbour,
                reaches: crate::swarm::reach::of(ps.reaches.as_ref()).map(&named),
                traffic,
                incidents: ps.incidents,
                sync_throughput,
                clock,
                hop_delay,
            },
        );
    }

    let local_heads = state.doc.read().await.heads().len();
    PeerListResponse {
        local_alias: state.alias.clone(),
        local_addresses: state.transport.local_addresses(),
        local_epoch: local_epoch.into(),
        local_reaches: named(crate::swarm::reach::published(&state)),
        local_heads,
        peers,
    }
}

/// `DELETE /peers/{pk}`: forgets a peer swarm-wide.
pub(crate) async fn delete_peer(
    State(state): State<AppState>,
    Path(peer): Path<String>,
    Query(query): Query<DeletePeerQuery>,
) -> Result<StatusCode, StatusCode> {
    if !state.is_ready().await {
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }
    if peer.is_empty() || peer == state.local_id() {
        return Err(StatusCode::BAD_REQUEST);
    }
    let force = crate::force::Force::parse(query.force.as_deref(), &["alive"])
        .map_err(|_| StatusCode::BAD_REQUEST)?;

    // Unknown locally is fine: other nodes may still know the peer.
    if let Some(peer_state_mutex) = state.get_peer_state(&peer).await
        && peer_state_mutex
            .lock()
            .await
            .is_alive(&state.timing, Instant::now())
        && !force.allows("alive")
    {
        return Err(StatusCode::CONFLICT);
    }

    let record =
        serde_json::json!({ "at": state.clock.now_millis().as_u64(), "by": state.local_id() });

    let peer_clone = peer.clone();
    state
        .mutate_doc(move |doc| {
            store::write_path(doc, &forgotten_path(&peer_clone), &record)?;
            crate::swarm::members::unregister(doc, &peer_clone)
        })
        .await
        .map_err(|e| store_status(e, "Failed to write forget record").status())?;

    // Unless forced, a peer that came alive since the check above stays: it cancels
    // the record.
    let forgotten = if force.allows("alive") {
        state.peer_states.write().await.remove(&peer).is_some()
    } else {
        crate::swarm::peer::forget_unless_alive(&state, &peer, Instant::now()).await
    };
    if forgotten {
        warn!(peer = %peer, "Peer forgotten swarm-wide by operator");
    }

    Ok(StatusCode::NO_CONTENT)
}