Add nodes/delegations endpoint (#5733)

* WIP

* Add /delegations endpoint

* Bump package version

* Remove node_id field

Cargo.lock

@@ -4733,7 +4733,7 @@ checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3"
 
 [[package]]
 name = "nym-api"
-version = "1.1.56"
+version = "1.1.55"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -6252,7 +6252,7 @@ dependencies = [
 
 [[package]]
 name = "nym-node-status-api"
-version = "2.2.0"
+version = "2.3.0"
 dependencies = [
  "ammonia",
  "anyhow",

nym-node-status-api/nym-node-status-api/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-node-status-api"
-version = "2.2.0"
+version = "2.3.0"
 authors.workspace = true
 repository.workspace = true
 homepage.workspace = true

nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs

@@ -1,18 +1,26 @@
 use axum::{
-    extract::{Query, State},
+    extract::{Path, Query, State},
     Json, Router,
 };
+use nym_validator_client::client::NodeId;
+use serde::Deserialize;
 use tracing::instrument;
+use utoipa::IntoParams;
 
 use crate::http::{
     error::{HttpError, HttpResult},
-    models::ExtendedNymNode,
+    models::{ExtendedNymNode, NodeDelegation},
     state::AppState,
     PagedResult, Pagination,
 };
 
 pub(crate) fn routes() -> Router<AppState> {
-    Router::new().route("/", axum::routing::get(nym_nodes))
+    Router::new()
+        .route("/", axum::routing::get(nym_nodes))
+        .route(
+            "/:node_id/delegations",
+            axum::routing::get(node_delegations),
+        )
 }
 
 #[utoipa::path(
@@ -27,7 +35,7 @@ pub(crate) fn routes() -> Router<AppState> {
         (status = 200, body = PagedResult<ExtendedNymNode>)
     )
 )]
-#[instrument(level = tracing::Level::DEBUG, skip_all, fields(page=pagination.page, size=pagination.size))]
+#[instrument(level = tracing::Level::INFO, skip_all, fields(page=pagination.page, size=pagination.size))]
 async fn nym_nodes(
     Query(pagination): Query<Pagination>,
     State(state): State<AppState>,
@@ -46,3 +54,35 @@ async fn nym_nodes(
 
     Ok(Json(PagedResult::paginate(pagination, nodes)))
 }
+
+#[allow(dead_code)] // clippy doesn't detect usage in utoipa macros
+#[derive(Deserialize, IntoParams)]
+#[into_params(parameter_in = Path)]
+struct NodeIdParam {
+    #[param(minimum = 0)]
+    node_id: NodeId,
+}
+
+#[utoipa::path(
+    tag = "Nym Explorer",
+    get,
+    params(
+        NodeIdParam
+    ),
+    path = "/{node_id}/delegations",
+    context_path = "/explorer/v3/nym-nodes",
+    responses(
+        (status = 200, body = NodeDelegation)
+    )
+)]
+#[instrument(level = tracing::Level::INFO, skip(state))]
+async fn node_delegations(
+    Path(node_id): Path<NodeId>,
+    State(state): State<AppState>,
+) -> HttpResult<Json<Vec<NodeDelegation>>> {
+    state
+        .node_delegations(node_id)
+        .await
+        .ok_or_else(|| HttpError::no_delegations_for_node(node_id))
+        .map(Json)
+}

nym-node-status-api/nym-node-status-api/src/http/error.rs

@@ -1,3 +1,4 @@
+use nym_mixnet_contract_common::NodeId;
 use std::fmt::Display;
 
 pub(crate) type HttpResult<T> = Result<T, HttpError>;
@@ -40,6 +41,14 @@ impl HttpError {
             status: axum::http::StatusCode::SERVICE_UNAVAILABLE,
         }
     }
+
+    pub(crate) fn no_delegations_for_node(node_id: NodeId) -> Self {
+        Self {
+            message: serde_json::json!({"message": format!("No delegation data for node_id={}", node_id)})
+                .to_string(),
+            status: axum::http::StatusCode::NOT_FOUND,
+        }
+    }
 }
 
 impl axum::response::IntoResponse for HttpError {

nym-node-status-api/nym-node-status-api/src/http/models.rs

@@ -1,4 +1,5 @@
-use cosmwasm_std::Decimal;
+use cosmwasm_std::{Addr, Coin, Decimal};
+use nym_mixnet_contract_common::CoinSchema;
 use nym_node_requests::api::v1::node::models::NodeDescription;
 use nym_validator_client::client::NodeId;
 use serde::{Deserialize, Serialize};
@@ -120,3 +121,27 @@ pub struct SessionStats {
     pub mixnet_sessions: Option<serde_json::Value>,
     pub unknown_sessions: Option<serde_json::Value>,
 }
+
+#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)]
+pub struct NodeDelegation {
+    #[schema(value_type = CoinSchema)]
+    pub amount: Coin,
+    pub cumulative_reward_ratio: String,
+    pub block_height: u64,
+    #[schema(value_type = String)]
+    pub owner: Addr,
+    #[schema(value_type = Option<String>)]
+    pub proxy: Option<Addr>,
+}
+
+impl From<nym_mixnet_contract_common::Delegation> for NodeDelegation {
+    fn from(value: nym_mixnet_contract_common::Delegation) -> Self {
+        Self {
+            amount: value.amount,
+            cumulative_reward_ratio: value.cumulative_reward_ratio.to_string(),
+            block_height: value.height,
+            owner: value.owner,
+            proxy: value.proxy,
+        }
+    }
+}

nym-node-status-api/nym-node-status-api/src/http/server.rs

@@ -1,13 +1,14 @@
 use axum::Router;
 use core::net::SocketAddr;
 use nym_crypto::asymmetric::ed25519::PublicKey;
-use tokio::{net::TcpListener, task::JoinHandle};
+use std::sync::Arc;
+use tokio::{net::TcpListener, sync::RwLock, task::JoinHandle};
 use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
 
 use crate::{
     db::DbPool,
     http::{api::RouterBuilder, state::AppState},
-    monitor::NodeGeoCache,
+    monitor::{DelegationsCache, NodeGeoCache},
 };
 
 /// Return handles that allow for graceful shutdown of server + awaiting its
@@ -19,6 +20,7 @@ pub(crate) async fn start_http_api(
     agent_key_list: Vec<PublicKey>,
     agent_max_count: i64,
     node_geocache: NodeGeoCache,
+    node_delegations: Arc<RwLock<DelegationsCache>>,
 ) -> anyhow::Result<ShutdownHandles> {
     let router_builder = RouterBuilder::with_default_routes();
 
@@ -28,6 +30,7 @@ pub(crate) async fn start_http_api(
         agent_key_list,
         agent_max_count,
         node_geocache,
+        node_delegations,
     )
     .await;
     let router = router_builder.with_state(state);

nym-node-status-api/nym-node-status-api/src/http/state.rs

@@ -1,17 +1,17 @@
-use std::{collections::HashMap, sync::Arc, time::Duration};
-
 use cosmwasm_std::Decimal;
 use moka::{future::Cache, Entry};
 use nym_contracts_common::NaiveFloat;
 use nym_crypto::asymmetric::ed25519::PublicKey;
+use nym_mixnet_contract_common::NodeId;
 use nym_validator_client::nym_api::SkimmedNode;
+use std::{collections::HashMap, sync::Arc, time::Duration};
 use tokio::sync::RwLock;
 use tracing::instrument;
 
 use crate::{
     db::{queries, DbPool},
     http::models::{DailyStats, ExtendedNymNode, Gateway, Mixnode, NodeGeoData, SummaryHistory},
-    monitor::NodeGeoCache,
+    monitor::{DelegationsCache, NodeGeoCache},
 };
 
 use super::models::SessionStats;
@@ -23,6 +23,7 @@ pub(crate) struct AppState {
     agent_key_list: Vec<PublicKey>,
     agent_max_count: i64,
     node_geocache: NodeGeoCache,
+    node_delegations: Arc<RwLock<DelegationsCache>>,
 }
 
 impl AppState {
@@ -32,6 +33,7 @@ impl AppState {
         agent_key_list: Vec<PublicKey>,
         agent_max_count: i64,
         node_geocache: NodeGeoCache,
+        node_delegations: Arc<RwLock<DelegationsCache>>,
     ) -> Self {
         Self {
             db_pool,
@@ -39,6 +41,7 @@ impl AppState {
             agent_key_list,
             agent_max_count,
             node_geocache,
+            node_delegations,
         }
     }
 
@@ -61,6 +64,16 @@ impl AppState {
     pub(crate) fn node_geocache(&self) -> NodeGeoCache {
         self.node_geocache.clone()
     }
+
+    pub(crate) async fn node_delegations(
+        &self,
+        node_id: NodeId,
+    ) -> Option<Vec<super::models::NodeDelegation>> {
+        self.node_delegations
+            .read()
+            .await
+            .delegations_owned(node_id)
+    }
 }
 
 static GATEWAYS_LIST_KEY: &str = "gateways";

nym-node-status-api/nym-node-status-api/src/main.rs

@@ -1,6 +1,9 @@
+use crate::monitor::DelegationsCache;
 use clap::Parser;
 use nym_crypto::asymmetric::ed25519::PublicKey;
 use nym_task::signal::wait_for_signal;
+use nym_validator_client::nyxd::NyxdClient;
+use std::sync::Arc;
 
 mod cli;
 mod db;
@@ -41,19 +44,27 @@ async fn main() -> anyhow::Result<()> {
     let geocache = moka::future::Cache::builder()
         .time_to_live(args.geodata_ttl)
         .build();
+    let delegations_cache = DelegationsCache::new();
 
     // Start the monitor
     let args_clone = args.clone();
     let geocache_clone = geocache.clone();
+    let delegations_cache_clone = Arc::clone(&delegations_cache);
+    let config = nym_validator_client::nyxd::Config::try_from_nym_network_details(
+        &nym_network_defaults::NymNetworkDetails::new_from_env(),
+    )?;
+    let nyxd_client = NyxdClient::connect(config, args.nyxd_addr.as_str())
+        .map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?;
 
     tokio::spawn(async move {
         monitor::spawn_in_background(
             db_pool,
             args_clone.nym_api_client_timeout,
-            args_clone.nyxd_addr,
+            nyxd_client,
             args_clone.monitor_refresh_interval,
             args_clone.ipinfo_api_token,
             geocache_clone,
+            delegations_cache_clone,
         )
         .await;
         tracing::info!("Started monitor task");
@@ -74,6 +85,7 @@ async fn main() -> anyhow::Result<()> {
         agent_key_list.to_owned(),
         args.max_agent_count,
         geocache,
+        delegations_cache,
     )
     .await
     .expect("Failed to start server");

nym-node-status-api/nym-node-status-api/src/monitor/mod.rs

@@ -12,25 +12,30 @@ use crate::utils::{decimal_to_i64, LogError, NumericalCheckedCast};
 use anyhow::anyhow;
 use moka::future::Cache;
 use nym_network_defaults::NymNetworkDetails;
-use nym_validator_client::client::{NodeId, NymApiClientExt, NymNodeDetails};
-use nym_validator_client::models::{
-    LegacyDescribedMixNode, MixNodeBondAnnotated, NymNodeDescription,
+use nym_validator_client::{
+    client::{NodeId, NymApiClientExt, NymNodeDetails},
+    models::{LegacyDescribedMixNode, MixNodeBondAnnotated, NymNodeDescription},
 };
-use nym_validator_client::nym_nodes::{NodeRole, SkimmedNode};
-use nym_validator_client::nyxd::contract_traits::PagedMixnetQueryClient;
-use nym_validator_client::nyxd::{AccountId, NyxdClient};
-use nym_validator_client::NymApiClient;
-use reqwest::Url;
-use std::collections::{HashMap, HashSet};
-use std::str::FromStr;
-use tokio::time::Duration;
+use nym_validator_client::{
+    nym_nodes::{NodeRole, SkimmedNode},
+    nyxd::{contract_traits::PagedMixnetQueryClient, AccountId},
+    NymApiClient, QueryHttpRpcNyxdClient,
+};
+use std::{
+    collections::{HashMap, HashSet},
+    str::FromStr,
+    sync::Arc,
+};
+use tokio::{sync::RwLock, time::Duration};
 use tracing::instrument;
 
 pub(crate) use geodata::IpInfoClient;
+pub(crate) use node_delegations::DelegationsCache;
 
 mod geodata;
+mod node_delegations;
 
-const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60);
+const MONITOR_FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60);
 static DELEGATION_PROGRAM_WALLET: &str = "n1rnxpdpx3kldygsklfft0gech7fhfcux4zst5lw";
 pub(crate) type NodeGeoCache = Cache<NodeId, Location>;
 
@@ -38,9 +43,10 @@ struct Monitor {
     db_pool: DbPool,
     network_details: NymNetworkDetails,
     nym_api_client_timeout: Duration,
-    nyxd_addr: Url,
+    nyxd_client: QueryHttpRpcNyxdClient,
     ipinfo: IpInfoClient,
     geocache: NodeGeoCache,
+    node_delegations: Arc<RwLock<DelegationsCache>>,
 }
 
 // TODO dz: query many NYM APIs:
@@ -49,19 +55,22 @@ struct Monitor {
 pub(crate) async fn spawn_in_background(
     db_pool: DbPool,
     nym_api_client_timeout: Duration,
-    nyxd_addr: Url,
+    nyxd_client: nym_validator_client::QueryHttpRpcNyxdClient,
     refresh_interval: Duration,
     ipinfo_api_token: String,
     geocache: NodeGeoCache,
+    node_delegations: Arc<RwLock<DelegationsCache>>,
 ) {
     let ipinfo = IpInfoClient::new(ipinfo_api_token.clone());
+
     let mut monitor = Monitor {
         db_pool,
         network_details: nym_network_defaults::NymNetworkDetails::new_from_env(),
         nym_api_client_timeout,
-        nyxd_addr,
+        nyxd_client,
         ipinfo,
         geocache,
+        node_delegations,
     };
 
     loop {
@@ -70,10 +79,9 @@ pub(crate) async fn spawn_in_background(
         if let Err(e) = monitor.run().await {
             tracing::error!(
                 "Monitor run failed: {e}, retrying in {}s...",
-                FAILURE_RETRY_DELAY.as_secs()
+                MONITOR_FAILURE_RETRY_DELAY.as_secs()
             );
-            // TODO dz implement some sort of backoff
-            tokio::time::sleep(FAILURE_RETRY_DELAY).await;
+            tokio::time::sleep(MONITOR_FAILURE_RETRY_DELAY).await;
         } else {
             tracing::info!(
                 "Info successfully collected, sleeping for {}s...",
@@ -193,8 +201,7 @@ impl Monitor {
             .nodes
             .data;
 
-        let delegation_program_members =
-            get_delegation_program_details(&self.network_details, &self.nyxd_addr).await?;
+        let delegation_program_members = self.get_delegation_program_details().await?;
 
         // keep stats for later
         let assigned_entry_count = nym_nodes
@@ -233,7 +240,9 @@ impl Monitor {
                 tracing::debug!("{} mixnode info written to DB!", mixnodes_count);
             })?;
 
-        let (all_historical_gateways, all_historical_mixnodes) = calculate_stats(&pool).await?;
+        self.refresh_node_delegations(&bonded_nym_nodes).await;
+
+        let (all_historical_gateways, all_historical_mixnodes) = historical_count(&pool).await?;
 
         //
         // write summary keys and values to table
@@ -457,9 +466,34 @@ impl Monitor {
             }
         }
     }
+
+    #[instrument(level = "info", skip_all)]
+    async fn refresh_node_delegations(&mut self, bonded_nodes: &HashMap<NodeId, NymNodeDetails>) {
+        let delegations_per_node = node_delegations::refresh(&self.nyxd_client, bonded_nodes).await;
+
+        // update after refreshing all to avoid holding write lock for too long
+        *self.node_delegations.write().await = delegations_per_node;
+    }
+
+    async fn get_delegation_program_details(&self) -> anyhow::Result<Vec<NodeId>> {
+        let account_id = AccountId::from_str(DELEGATION_PROGRAM_WALLET)
+            .map_err(|e| anyhow!("Invalid bech32 address: {}", e))?;
+
+        let delegations = self
+            .nyxd_client
+            .get_all_delegator_delegations(&account_id)
+            .await?;
+
+        let mix_ids: Vec<NodeId> = delegations
+            .iter()
+            .map(|delegation| delegation.node_id)
+            .collect();
+
+        Ok(mix_ids)
+    }
 }
 
-async fn calculate_stats(pool: &DbPool) -> anyhow::Result<(usize, usize)> {
+async fn historical_count(pool: &DbPool) -> anyhow::Result<(usize, usize)> {
     let mut conn = pool.acquire().await?;
 
     let all_historical_gateways = sqlx::query_scalar!(r#"SELECT count(id) FROM gateways"#)
@@ -474,25 +508,3 @@ async fn calculate_stats(pool: &DbPool) -> anyhow::Result<(usize, usize)> {
 
     Ok((all_historical_gateways, all_historical_mixnodes))
 }
-
-async fn get_delegation_program_details(
-    network_details: &NymNetworkDetails,
-    nyxd_addr: &Url,
-) -> anyhow::Result<Vec<u32>> {
-    let config = nym_validator_client::nyxd::Config::try_from_nym_network_details(network_details)?;
-
-    let client = NyxdClient::connect(config, nyxd_addr.as_str())
-        .map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?;
-
-    let account_id = AccountId::from_str(DELEGATION_PROGRAM_WALLET)
-        .map_err(|e| anyhow!("Invalid bech32 address: {}", e))?;
-
-    let delegations = client.get_all_delegator_delegations(&account_id).await?;
-
-    let mix_ids: Vec<u32> = delegations
-        .iter()
-        .map(|delegation| delegation.node_id)
-        .collect();
-
-    Ok(mix_ids)
-}

nym-node-status-api/nym-node-status-api/src/monitor/node_delegations.rs

@@ -0,0 +1,53 @@
+use nym_mixnet_contract_common::{NodeId, NymNodeDetails};
+use nym_validator_client::{nyxd::contract_traits::PagedMixnetQueryClient, QueryHttpRpcNyxdClient};
+use std::{collections::HashMap, sync::Arc};
+use tokio::{sync::RwLock, time::Instant};
+use tracing::{info, warn};
+
+// abstracts away data structure that holds delegations
+#[derive(Clone, Debug)]
+pub(crate) struct DelegationsCache {
+    pub inner: HashMap<NodeId, Vec<crate::http::models::NodeDelegation>>,
+}
+
+impl DelegationsCache {
+    pub(crate) fn new() -> Arc<RwLock<Self>> {
+        let a = Self {
+            inner: HashMap::new(),
+        };
+        Arc::new(RwLock::new(a))
+    }
+
+    pub(crate) fn delegations_owned(
+        &self,
+        node_id: NodeId,
+    ) -> Option<Vec<crate::http::models::NodeDelegation>> {
+        self.inner.get(&node_id).cloned()
+    }
+}
+
+pub(super) async fn refresh(
+    client: &QueryHttpRpcNyxdClient,
+    bonded_nodes: &HashMap<NodeId, NymNodeDetails>,
+) -> DelegationsCache {
+    info!("👥 Refreshing {} node delegations...", bonded_nodes.len());
+    let now = Instant::now();
+
+    let mut delegations_per_node = HashMap::new();
+    for node_id in bonded_nodes.keys() {
+        if let Ok(delegations) = client
+            .get_all_single_mixnode_delegations(*node_id)
+            .await
+            .inspect_err(|err| warn!("Failed to get delegations for {}: {}", node_id, err))
+        {
+            delegations_per_node
+                .insert(*node_id, delegations.into_iter().map(From::from).collect());
+        }
+    }
+    let time_taken = Instant::now() - now;
+    info!("👥 Node delegations refreshed in {}s", time_taken.as_secs(),);
+
+    DelegationsCache {
+        inner: delegations_per_node,
+    }
+}