initial commit

1 files changed, 161 insertions, 0 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
new file mode 100644
index 0000000..b2658a5
--- /dev/null
+++ b/src/api/peers.rs
@@ -0,0 +1,161 @@
+//! /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::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 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,
+}
+
+#[derive(Debug, Serialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct PeerListResponse {
+    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 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(),
+        });
+    }
+
+    peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
+    Ok(Json(PeerListResponse { 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)
+}