//! /peers handlers.

use std::sync::Arc;

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

use super::store_status;
use crate::peer::forgotten_path;
use crate::replica::Epoch;
use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState};
use crate::store;

#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "lowercase")]
pub(crate) enum PeerLivenessStatus {
    Active,
    Dormant,
}

#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub(crate) struct EpochInfo {
    generation: u64,
    /// Hex-encoded snapshot hash; all zeros for the genesis epoch.
    id: String,
}

impl From<Epoch> for EpochInfo {
    fn from(epoch: Epoch) -> Self {
        Self {
            generation: epoch.generation,
            id: epoch.id_hex(),
        }
    }
}

#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PeerStatus {
    public_key: String,
    validated_cluster: Option<String>,
    status: PeerLivenessStatus,
    handshake_complete: bool,
    last_seen_secs_ago: Option<u64>,
    dormant_for_secs: Option<u64>,
    has_pending_delivery: bool,
    /// Epoch the peer last advertised; absent until it has sent a message.
    epoch: Option<EpochInfo>,
    /// Whether that epoch equals ours, i.e. whether we sync with this peer.
    same_epoch: Option<bool>,
}

#[derive(Debug, Serialize, ToSchema)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PeerListResponse {
    local_epoch: EpochInfo,
    peers: Vec<PeerStatus>,
}

#[derive(Debug, Deserialize)]
pub(crate) struct DeletePeerQuery {
    #[serde(default)]
    force: bool,
}

#[utoipa::path(
    get,
    path = "/peers",
    tag = "Peers",
    responses(
        (
            status = 200,
            description = "Lists known peers and their local replication status",
            body = PeerListResponse
        )
    )
)]
pub(crate) async fn list_peers(
    State(state): State<AppState>,
) -> Result<Json<PeerListResponse>, StatusCode> {
    let snapshot: Vec<(String, Arc<Mutex<PeerState>>)> = state
        .peer_states
        .read()
        .await
        .iter()
        .map(|(pk, m)| (pk.clone(), Arc::clone(m)))
        .collect();

    let now = Instant::now();
    let local_epoch = state.local_epoch();
    let mut peers = Vec::with_capacity(snapshot.len());

    for (public_key, peer_state_mutex) in snapshot {
        let ps = peer_state_mutex.lock().await;

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

        peers.push(PeerStatus {
            public_key,
            validated_cluster: match &ps.membership {
                PeerMembership::Unknown => None,
                PeerMembership::Member { cluster_id } => Some(cluster_id.clone()),
            },
            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,
            has_pending_delivery: ps.pending_outbound_sync.is_some(),
            epoch: ps.epoch.map(EpochInfo::from),
            same_epoch: ps.epoch.map(|epoch| epoch == local_epoch),
        });
    }

    peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
    Ok(Json(PeerListResponse {
        local_epoch: local_epoch.into(),
        peers,
    }))
}

#[utoipa::path(
    delete,
    path = "/peers/{peer}",
    tag = "Peers",
    params(
        ("peer" = String, Path, description = "Peer public key"),
        ("force" = Option<bool>, Query, description = "Also forget a peer that is currently alive; it returns on its next message")
    ),
    responses(
        (status = 204, description = "Cluster-wide forget recorded"),
        (status = 400, description = "Not a public key, or the local node's own key"),
        (status = 409, description = "Peer is alive; pass force=true"),
        (status = 503, description = "Node is not ready")
    )
)]
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);
    }

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

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

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

    if state.peer_states.write().await.remove(&peer).is_some() {
        warn!(peer = %peer, "Peer forgotten cluster-wide by operator");
    }

    Ok(StatusCode::NO_CONTENT)
}