Node Status API background task (#4854)

* Setup new package

* Setup DB

* Fetch & store mixnodes/GWs
- refactor db package structure
- finally solve DATABASE_URL: absolute path works best

* Additional query functionality
- missing only daily summary, which requires type refactoring

* Replace type alias tuples with structs

* Insert summary

* Add github job to build package

* Build script for sqlx

* Remove data dir
- useless now that sqlx DB sits in OUT_DIR

* PR feedback

Cargo.lock

@@ -5878,6 +5878,29 @@ dependencies = [
  "utoipa",
 ]
 
+[[package]]
+name = "nym-node-status-api"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "axum 0.7.5",
+ "chrono",
+ "clap 4.5.17",
+ "cosmwasm-std",
+ "futures-util",
+ "nym-bin-common",
+ "nym-explorer-client",
+ "nym-network-defaults",
+ "nym-validator-client",
+ "serde",
+ "serde_json",
+ "sqlx",
+ "thiserror",
+ "tokio",
+ "tracing",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "nym-node-tester-utils"
 version = "0.1.0"

Cargo.toml

@@ -120,6 +120,7 @@ members = [
     "nym-node",
     "nym-node/nym-node-http-api",
     "nym-node/nym-node-requests",
+    "nym-node-status-api",
     "nym-outfox",
     "nym-validator-rewarder",
     "tools/echo-server",
@@ -238,6 +239,7 @@ eyre = "0.6.9"
 fastrand = "2.1.1"
 flate2 = "1.0.34"
 futures = "0.3.28"
+futures-util = "0.3"
 generic-array = "0.14.7"
 getrandom = "0.2.10"
 getset = "0.1.3"

nym-node-status-api/.gitignore

@@ -0,0 +1 @@
+data/

nym-node-status-api/Cargo.toml

@@ -0,0 +1,37 @@
+# Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+# SPDX-License-Identifier: Apache-2.0
+
+[package]
+name = "nym-node-status-api"
+version = "0.1.0"
+authors.workspace = true
+repository.workspace = true
+homepage.workspace = true
+documentation.workspace = true
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+axum = { workspace = true }
+chrono = { workspace = true }
+clap = { workspace = true, features = ["cargo", "derive"] }
+cosmwasm-std = { workspace = true }
+futures-util = { workspace = true }
+nym-bin-common = { path = "../common/bin-common" }
+nym-explorer-client = { path = "../explorer-api/explorer-client" }
+nym-network-defaults = { path = "../common/network-defaults" }
+nym-validator-client = { path = "../common/client-libs/validator-client" }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
+
+[build-dependencies]
+anyhow = { workspace = true }
+tokio = { workspace = true, features = ["macros" ] }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }

nym-node-status-api/build.rs

@@ -0,0 +1,33 @@
+use anyhow::{anyhow, Result};
+use sqlx::{Connection, SqliteConnection};
+
+const SQLITE_DB_FILENAME: &str = "nym-node-status-api.sqlite";
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let out_dir = read_env_var("OUT_DIR")?;
+    let database_path = format!("sqlite://{}/{}?mode=rwc", out_dir, SQLITE_DB_FILENAME);
+
+    let mut conn = SqliteConnection::connect(&database_path).await?;
+    sqlx::migrate!("./migrations").run(&mut conn).await?;
+
+    #[cfg(target_family = "unix")]
+    println!("cargo::rustc-env=DATABASE_URL=sqlite://{}", &database_path);
+
+    #[cfg(target_family = "windows")]
+    // for some strange reason we need to add a leading `/` to the windows path even though it's
+    // not a valid windows path... but hey, it works...
+    println!("cargo::rustc-env=DATABASE_URL=sqlite:///{}", &database_path);
+
+    rerun_if_changed();
+    Ok(())
+}
+
+fn read_env_var(var: &str) -> Result<String> {
+    std::env::var(var).map_err(|_| anyhow!("You need to set {} env var", var))
+}
+
+fn rerun_if_changed() {
+    println!("cargo::rerun-if-changed=migrations");
+    println!("cargo::rerun-if-changed=src/db/queries");
+}

nym-node-status-api/launch_node_status_api.sh

@@ -0,0 +1,64 @@
+#!/bin/bash
+
+set -e
+
+function usage() {
+    echo "Usage: $0 [-ci]"
+    echo "  -c Clear DB and re-initialize it before launching the binary."
+    echo "  -i Only initialize and prepare database, env vars then exit without"
+    echo "     launching"
+    exit 0
+}
+
+function init_db() {
+    rm -rf data/*
+    # https://github.com/launchbadge/sqlx/blob/main/sqlx-cli/README.md
+    cargo sqlx database drop -y
+
+    cargo sqlx database create
+    cargo sqlx migrate run
+    cargo sqlx prepare
+
+    echo "Fresh database ready!"
+}
+
+# export DATABASE_URL as absolute path due to this
+# https://github.com/launchbadge/sqlx/issues/3099
+db_filename="nym-node-status-api.sqlite"
+script_abs_path=$(realpath "$0")
+package_dir=$(dirname "$script_abs_path")
+db_abs_path="$package_dir/data/$db_filename"
+dotenv_file="$package_dir/.env"
+echo "DATABASE_URL=sqlite://$db_abs_path" > "$dotenv_file"
+
+export RUST_LOG=${RUST_LOG:-debug}
+
+# export DATABASE_URL from .env file
+set -a && source "$dotenv_file" && set +a
+
+clear_db=false
+init_only=false
+
+while getopts "ci" opt; do
+    case ${opt} in
+    c)
+        clear_db=true
+        ;;
+    i)
+        init_only=true
+        ;;
+    \?)
+        usage
+        ;;
+    esac
+done
+
+if [ "$clear_db" = true ] || [ "$init_only" = true ]; then
+    init_db
+fi
+
+if [ "$init_only" = true ]; then
+    exit 0
+fi
+
+cargo run --package nym-node-status-api

nym-node-status-api/migrations/000_init.sql

@@ -0,0 +1,59 @@
+CREATE TABLE gateways
+(
+  id                      INTEGER PRIMARY KEY AUTOINCREMENT,
+  gateway_identity_key    VARCHAR NOT NULL UNIQUE,
+  self_described          VARCHAR,
+  explorer_pretty_bond    VARCHAR,
+  last_probe_result       VARCHAR,
+  last_probe_log          VARCHAR,
+  config_score            INTEGER NOT NULL DEFAULT (0),
+  config_score_successes  REAL    NOT NULL DEFAULT (0),
+  config_score_samples    REAL    NOT NULL DEFAULT (0),
+  routing_score           INTEGER NOT NULL DEFAULT (0),
+  routing_score_successes REAL    NOT NULL DEFAULT (0),
+  routing_score_samples   REAL    NOT NULL DEFAULT (0),
+  test_run_samples        REAL    NOT NULL DEFAULT (0),
+  last_testrun_utc        INTEGER,
+  last_updated_utc        INTEGER NOT NULL,
+  bonded INTEGER CHECK (bonded in (0, 1)) NOT NULL DEFAULT 0,
+  blacklisted INTEGER CHECK (bonded in (0, 1)) NOT NULL DEFAULT 0,
+  performance INTEGER NOT NULL DEFAULT 0
+);
+
+CREATE INDEX idx_gateway_description_gateway_identity_key ON gateways (gateway_identity_key);
+
+
+CREATE TABLE mixnodes (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    identity_key VARCHAR NOT NULL UNIQUE,
+    mix_id INTEGER NOT NULL UNIQUE,
+    bonded INTEGER CHECK (bonded in (0, 1)) NOT NULL DEFAULT 0,
+    total_stake INTEGER NOT NULL,
+    host VARCHAR NOT NULL,
+    http_api_port INTEGER NOT NULL,
+    blacklisted INTEGER CHECK (blacklisted in (0, 1)) NOT NULL DEFAULT 0,
+    full_details VARCHAR,
+    self_described VARCHAR,
+    last_updated_utc INTEGER NOT NULL
+  , is_dp_delegatee INTEGER CHECK (is_dp_delegatee IN (0, 1)) NOT NULL DEFAULT 0);
+CREATE INDEX idx_mixnodes_mix_id ON mixnodes (mix_id);
+CREATE INDEX idx_mixnodes_identity_key ON mixnodes (identity_key);
+
+
+CREATE TABLE summary
+(
+  key              VARCHAR PRIMARY KEY,
+  value_json       VARCHAR,
+  last_updated_utc INTEGER NOT NULL
+);
+
+
+CREATE TABLE summary_history
+(
+  id            INTEGER PRIMARY KEY AUTOINCREMENT,
+  date          VARCHAR UNIQUE NOT NULL,
+  timestamp_utc INTEGER NOT NULL,
+  value_json    VARCHAR
+);
+CREATE INDEX idx_summary_history_timestamp_utc ON summary_history (timestamp_utc);
+CREATE INDEX idx_summary_history_date ON summary_history (date);

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

@@ -0,0 +1,17 @@
+use clap::Parser;
+use nym_bin_common::bin_info;
+use std::sync::OnceLock;
+
+// Helper for passing LONG_VERSION to clap
+fn pretty_build_info_static() -> &'static str {
+    static PRETTY_BUILD_INFORMATION: OnceLock<String> = OnceLock::new();
+    PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print())
+}
+
+#[derive(Parser, Debug)]
+#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)]
+pub(crate) struct Cli {
+    /// Path pointing to an env file that configures the Nym API.
+    #[clap(short, long)]
+    pub(crate) config_env_file: Option<std::path::PathBuf>,
+}

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

@@ -0,0 +1,40 @@
+use std::str::FromStr;
+
+use crate::read_env_var;
+use anyhow::{anyhow, Result};
+use sqlx::{migrate::Migrator, sqlite::SqliteConnectOptions, ConnectOptions, SqlitePool};
+pub(crate) const DATABASE_URL_ENV_VAR: &str = "DATABASE_URL";
+static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
+
+pub(crate) mod models;
+pub(crate) mod queries;
+
+pub(crate) type DbPool = SqlitePool;
+
+pub(crate) struct Storage {
+    pool: DbPool,
+}
+
+impl Storage {
+    pub async fn init() -> Result<Self> {
+        let connection_url = read_env_var(DATABASE_URL_ENV_VAR)?;
+        let connect_options = {
+            let connect_options = SqliteConnectOptions::from_str(&connection_url)?;
+            let mut connect_options = connect_options.create_if_missing(true);
+            let connect_options = connect_options.disable_statement_logging();
+            (*connect_options).clone()
+        };
+
+        let pool = sqlx::SqlitePool::connect_with(connect_options)
+            .await
+            .map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?;
+
+        MIGRATOR.run(&pool).await?;
+
+        Ok(Storage { pool })
+    }
+
+    pub async fn pool(&self) -> &DbPool {
+        &self.pool
+    }
+}

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

@@ -0,0 +1,128 @@
+use serde::{Deserialize, Serialize};
+
+pub(crate) struct GatewayRecord {
+    pub(crate) identity_key: String,
+    pub(crate) bonded: bool,
+    pub(crate) blacklisted: bool,
+    pub(crate) self_described: Option<String>,
+    pub(crate) explorer_pretty_bond: Option<String>,
+    pub(crate) last_updated_utc: i64,
+    pub(crate) performance: u8,
+}
+
+pub(crate) struct MixnodeRecord {
+    pub(crate) mix_id: u32,
+    pub(crate) identity_key: String,
+    pub(crate) bonded: bool,
+    pub(crate) total_stake: i64,
+    pub(crate) host: String,
+    pub(crate) http_port: u16,
+    pub(crate) blacklisted: bool,
+    pub(crate) full_details: String,
+    pub(crate) self_described: Option<String>,
+    pub(crate) last_updated_utc: i64,
+    pub(crate) is_dp_delegatee: bool,
+}
+
+#[allow(unused)]
+#[derive(Debug, Clone)]
+pub(crate) struct BondedStatusDto {
+    pub(crate) id: i64,
+    pub(crate) identity_key: String,
+    pub(crate) bonded: bool,
+}
+
+#[allow(unused)]
+#[derive(Debug, Clone, Default)]
+pub(crate) struct SummaryDto {
+    pub(crate) key: String,
+    pub(crate) value_json: String,
+    pub(crate) last_updated_utc: i64,
+}
+
+pub(crate) const MIXNODES_BONDED_COUNT: &str = "mixnodes.bonded.count";
+pub(crate) const MIXNODES_BONDED_ACTIVE: &str = "mixnodes.bonded.active";
+pub(crate) const MIXNODES_BONDED_INACTIVE: &str = "mixnodes.bonded.inactive";
+pub(crate) const MIXNODES_BONDED_RESERVE: &str = "mixnodes.bonded.reserve";
+pub(crate) const MIXNODES_BLACKLISTED_COUNT: &str = "mixnodes.blacklisted.count";
+
+pub(crate) const GATEWAYS_BONDED_COUNT: &str = "gateways.bonded.count";
+pub(crate) const GATEWAYS_EXPLORER_COUNT: &str = "gateways.explorer.count";
+pub(crate) const GATEWAYS_BLACKLISTED_COUNT: &str = "gateways.blacklisted.count";
+
+pub(crate) const MIXNODES_HISTORICAL_COUNT: &str = "mixnodes.historical.count";
+pub(crate) const GATEWAYS_HISTORICAL_COUNT: &str = "gateways.historical.count";
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub(crate) struct NetworkSummary {
+    pub(crate) mixnodes: mixnode::MixnodeSummary,
+    pub(crate) gateways: gateway::GatewaySummary,
+}
+
+pub(crate) mod mixnode {
+    use super::*;
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct MixnodeSummary {
+        pub(crate) bonded: MixnodeSummaryBonded,
+        pub(crate) blacklisted: MixnodeSummaryBlacklisted,
+        pub(crate) historical: MixnodeSummaryHistorical,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct MixnodeSummaryBonded {
+        pub(crate) count: i32,
+        pub(crate) active: i32,
+        pub(crate) inactive: i32,
+        pub(crate) reserve: i32,
+        pub(crate) last_updated_utc: String,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct MixnodeSummaryBlacklisted {
+        pub(crate) count: i32,
+        pub(crate) last_updated_utc: String,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct MixnodeSummaryHistorical {
+        pub(crate) count: i32,
+        pub(crate) last_updated_utc: String,
+    }
+}
+
+pub(crate) mod gateway {
+    use super::*;
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct GatewaySummary {
+        pub(crate) bonded: GatewaySummaryBonded,
+        pub(crate) blacklisted: GatewaySummaryBlacklisted,
+        pub(crate) historical: GatewaySummaryHistorical,
+        pub(crate) explorer: GatewaySummaryExplorer,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct GatewaySummaryExplorer {
+        pub(crate) count: i32,
+        pub(crate) last_updated_utc: String,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct GatewaySummaryBonded {
+        pub(crate) count: i32,
+        pub(crate) last_updated_utc: String,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct GatewaySummaryHistorical {
+        pub(crate) count: i32,
+        pub(crate) last_updated_utc: String,
+    }
+
+    #[derive(Debug, Clone, Deserialize, Serialize)]
+    pub(crate) struct GatewaySummaryBlacklisted {
+        pub(crate) count: i32,
+        pub(crate) last_updated_utc: String,
+    }
+}

nym-node-status-api/src/db/queries/gateways.rs

@@ -0,0 +1,116 @@
+use crate::db::{
+    models::{BondedStatusDto, GatewayRecord},
+    DbPool,
+};
+use futures_util::TryStreamExt;
+use nym_validator_client::models::DescribedGateway;
+
+pub(crate) async fn insert_gateways(
+    pool: &DbPool,
+    gateways: Vec<GatewayRecord>,
+) -> anyhow::Result<()> {
+    let mut db = pool.acquire().await?;
+    for record in gateways {
+        sqlx::query!(
+            "INSERT INTO gateways
+                (gateway_identity_key, bonded, blacklisted,
+                    self_described, explorer_pretty_bond,
+                    last_updated_utc, performance)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(gateway_identity_key) DO UPDATE SET
+                bonded=excluded.bonded,
+                blacklisted=excluded.blacklisted,
+                self_described=excluded.self_described,
+                explorer_pretty_bond=excluded.explorer_pretty_bond,
+                last_updated_utc=excluded.last_updated_utc,
+                performance = excluded.performance;",
+            record.identity_key,
+            record.bonded,
+            record.blacklisted,
+            record.self_described,
+            record.explorer_pretty_bond,
+            record.last_updated_utc,
+            record.performance
+        )
+        .execute(&mut *db)
+        .await?;
+    }
+
+    Ok(())
+}
+
+pub(crate) async fn write_blacklisted_gateways_to_db<'a, I>(
+    pool: &DbPool,
+    gateways: I,
+) -> anyhow::Result<()>
+where
+    I: Iterator<Item = &'a String>,
+{
+    let mut conn = pool.acquire().await?;
+    for gateway_identity_key in gateways {
+        sqlx::query!(
+            "UPDATE gateways
+             SET blacklisted = true
+             WHERE gateway_identity_key = ?;",
+            gateway_identity_key,
+        )
+        .execute(&mut *conn)
+        .await?;
+    }
+
+    Ok(())
+}
+
+/// Ensure all gateways that are set as bonded, are still bonded
+pub(crate) async fn ensure_gateways_still_bonded(
+    pool: &DbPool,
+    gateways: &[DescribedGateway],
+) -> anyhow::Result<usize> {
+    let bonded_gateways_rows = get_all_bonded_gateways_row_ids_by_status(pool, true).await?;
+    let unbonded_gateways_rows = bonded_gateways_rows.iter().filter(|v| {
+        !gateways
+            .iter()
+            .any(|bonded| *bonded.bond.identity() == v.identity_key)
+    });
+
+    let recently_unbonded_gateways = unbonded_gateways_rows.to_owned().count();
+    let last_updated_utc = chrono::offset::Utc::now().timestamp();
+    let mut transaction = pool.begin().await?;
+    for row in unbonded_gateways_rows {
+        sqlx::query!(
+            "UPDATE gateways
+                SET bonded = ?, last_updated_utc = ?
+                WHERE id = ?;",
+            false,
+            last_updated_utc,
+            row.id,
+        )
+        .execute(&mut *transaction)
+        .await?;
+    }
+    transaction.commit().await?;
+
+    Ok(recently_unbonded_gateways)
+}
+
+async fn get_all_bonded_gateways_row_ids_by_status(
+    pool: &DbPool,
+    status: bool,
+) -> anyhow::Result<Vec<BondedStatusDto>> {
+    let mut conn = pool.acquire().await?;
+    let items = sqlx::query_as!(
+        BondedStatusDto,
+        r#"SELECT
+            id as "id!",
+            gateway_identity_key as "identity_key!",
+            bonded as "bonded: bool"
+         FROM gateways
+         WHERE bonded = ?"#,
+        status,
+    )
+    .fetch(&mut *conn)
+    .try_collect::<Vec<_>>()
+    .await?;
+
+    Ok(items)
+}

nym-node-status-api/src/db/queries/misc.rs

@@ -0,0 +1,86 @@
+use crate::db::{models::NetworkSummary, DbPool};
+use chrono::{DateTime, Utc};
+
+/// take `last_updated` instead of calculating it so that `summary` matches
+/// `daily_summary`
+pub(crate) async fn insert_summaries(
+    pool: &DbPool,
+    summaries: &[(&str, &usize)],
+    summary: &NetworkSummary,
+    last_updated: DateTime<Utc>,
+) -> anyhow::Result<()> {
+    insert_summary(pool, summaries, last_updated).await?;
+
+    insert_summary_history(pool, summary, last_updated).await?;
+
+    Ok(())
+}
+
+async fn insert_summary(
+    pool: &DbPool,
+    summaries: &[(&str, &usize)],
+    last_updated: DateTime<Utc>,
+) -> anyhow::Result<()> {
+    let timestamp = last_updated.timestamp();
+    let mut tx = pool.begin().await?;
+
+    for (kind, value) in summaries {
+        let value = value.to_string();
+        sqlx::query!(
+            "INSERT INTO summary
+                    (key, value_json, last_updated_utc)
+                    VALUES (?, ?, ?)
+                    ON CONFLICT(key) DO UPDATE SET
+                    value_json=excluded.value_json,
+                    last_updated_utc=excluded.last_updated_utc;",
+            kind,
+            value,
+            timestamp
+        )
+        .execute(&mut tx)
+        .await
+        .map_err(|err| {
+            tracing::error!("Failed to insert data for {kind}: {err}, aborting transaction",);
+            err
+        })?;
+    }
+
+    Ok(())
+}
+
+/// For `<date_N>`, `summary_history` is updated with fresh data on every
+/// iteration.
+///
+/// After UTC midnight, summary is inserted for `<date_N+1>` and last entry for
+/// `<date_N>` stays there forever.
+///
+/// This is not aggregate data, it's a set of latest data points
+async fn insert_summary_history(
+    pool: &DbPool,
+    summary: &NetworkSummary,
+    last_updated: DateTime<Utc>,
+) -> anyhow::Result<()> {
+    let mut conn = pool.acquire().await?;
+
+    let value_json = serde_json::to_string(&summary)?;
+    let timestamp = last_updated.timestamp();
+    let now_rfc3339 = last_updated.to_rfc3339();
+    // YYYY-MM-DD, without time
+    let date = &now_rfc3339[..10];
+
+    sqlx::query!(
+        "INSERT INTO summary_history
+                (date, timestamp_utc, value_json)
+                VALUES (?, ?, ?)
+                ON CONFLICT(date) DO UPDATE SET
+                timestamp_utc=excluded.timestamp_utc,
+                value_json=excluded.value_json;",
+        date,
+        timestamp,
+        value_json
+    )
+    .execute(&mut *conn)
+    .await?;
+
+    Ok(())
+}

nym-node-status-api/src/db/queries/mixnodes.rs

@@ -0,0 +1,101 @@
+use futures_util::TryStreamExt;
+use nym_validator_client::models::MixNodeBondAnnotated;
+
+use crate::db::{
+    models::{BondedStatusDto, MixnodeRecord},
+    DbPool,
+};
+
+pub(crate) async fn insert_mixnodes(
+    pool: &DbPool,
+    mixnodes: Vec<MixnodeRecord>,
+) -> anyhow::Result<()> {
+    let mut conn = pool.acquire().await?;
+
+    for record in mixnodes.iter() {
+        // https://www.sqlite.org/lang_upsert.html
+        sqlx::query!(
+            "INSERT INTO mixnodes
+                (mix_id, identity_key, bonded, total_stake,
+                    host, http_api_port, blacklisted, full_details,
+                    self_described, last_updated_utc, is_dp_delegatee)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(mix_id) DO UPDATE SET
+                bonded=excluded.bonded,
+                total_stake=excluded.total_stake, host=excluded.host,
+                http_api_port=excluded.http_api_port,blacklisted=excluded.blacklisted,
+                full_details=excluded.full_details,self_described=excluded.self_described,
+                last_updated_utc=excluded.last_updated_utc,
+                is_dp_delegatee = excluded.is_dp_delegatee;",
+            record.mix_id,
+            record.identity_key,
+            record.bonded,
+            record.total_stake,
+            record.host,
+            record.http_port,
+            record.blacklisted,
+            record.full_details,
+            record.self_described,
+            record.last_updated_utc,
+            record.is_dp_delegatee
+        )
+        .execute(&mut *conn)
+        .await?;
+    }
+
+    Ok(())
+}
+
+/// Ensure all mixnodes that are set as bonded, are still bonded
+pub(crate) async fn ensure_mixnodes_still_bonded(
+    pool: &DbPool,
+    mixnodes: &[MixNodeBondAnnotated],
+) -> anyhow::Result<usize> {
+    let bonded_mixnodes_rows = get_all_bonded_mixnodes_row_ids_by_status(pool, true).await?;
+    let unbonded_mixnodes_rows = bonded_mixnodes_rows.iter().filter(|v| {
+        !mixnodes
+            .iter()
+            .any(|bonded| *bonded.mixnode_details.bond_information.identity() == v.identity_key)
+    });
+
+    let recently_unbonded_mixnodes = unbonded_mixnodes_rows.to_owned().count();
+    let last_updated_utc = chrono::offset::Utc::now().timestamp();
+    let mut transaction = pool.begin().await?;
+    for row in unbonded_mixnodes_rows {
+        sqlx::query!(
+            "UPDATE mixnodes
+                SET bonded = ?, last_updated_utc = ?
+                WHERE id = ?;",
+            false,
+            last_updated_utc,
+            row.id,
+        )
+        .execute(&mut *transaction)
+        .await?;
+    }
+    transaction.commit().await?;
+
+    Ok(recently_unbonded_mixnodes)
+}
+
+async fn get_all_bonded_mixnodes_row_ids_by_status(
+    pool: &DbPool,
+    status: bool,
+) -> anyhow::Result<Vec<BondedStatusDto>> {
+    let mut conn = pool.acquire().await?;
+    let items = sqlx::query_as!(
+        BondedStatusDto,
+        r#"SELECT
+            id as "id!",
+            identity_key as "identity_key!",
+            bonded as "bonded: bool"
+         FROM mixnodes
+         WHERE bonded = ?"#,
+        status,
+    )
+    .fetch(&mut *conn)
+    .try_collect::<Vec<_>>()
+    .await?;
+
+    Ok(items)
+}

nym-node-status-api/src/db/queries/mod.rs

@@ -0,0 +1,9 @@
+mod gateways;
+mod misc;
+mod mixnodes;
+
+pub(crate) use gateways::{
+    ensure_gateways_still_bonded, insert_gateways, write_blacklisted_gateways_to_db,
+};
+pub(crate) use misc::insert_summaries;
+pub(crate) use mixnodes::{ensure_mixnodes_still_bonded, insert_mixnodes};

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

@@ -0,0 +1,41 @@
+use tracing::level_filters::LevelFilter;
+use tracing_subscriber::{filter::Directive, EnvFilter};
+
+pub(crate) fn setup_tracing_logger() {
+    fn directive_checked(directive: String) -> Directive {
+        directive.parse().expect("Failed to parse log directive")
+    }
+
+    let log_builder = tracing_subscriber::fmt()
+        // Use a more compact, abbreviated log format
+        .compact()
+        // Display source code file paths
+        .with_file(true)
+        // Display source code line numbers
+        .with_line_number(true)
+        // Don't display the event's target (module path)
+        .with_target(false);
+
+    let mut filter = EnvFilter::builder()
+        // if RUST_LOG isn't set, set default level
+        .with_default_directive(LevelFilter::INFO.into())
+        .from_env_lossy();
+    // these crates are more granularly filtered
+    let filter_crates = [
+        "nym_bin_common",
+        "nym_explorer_client",
+        "nym_network_defaults",
+        "nym_validator_client",
+        "reqwest",
+        "rustls",
+        "hyper",
+        "sqlx",
+        "h2",
+        "tendermint_rpc",
+    ];
+    for crate_name in filter_crates {
+        filter = filter.add_directive(directive_checked(format!("{}=warn", crate_name)));
+    }
+
+    log_builder.with_env_filter(filter).init();
+}

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

@@ -0,0 +1,38 @@
+use anyhow::anyhow;
+use clap::Parser;
+use nym_network_defaults::setup_env;
+
+mod cli;
+mod db;
+mod logging;
+mod monitor;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    logging::setup_tracing_logger();
+
+    let args = cli::Cli::parse();
+    // if dotenv file is present, load its values
+    // otherwise, default to mainnet
+    setup_env(args.config_env_file.as_ref());
+    tracing::debug!("{:?}", std::env::var("NETWORK_NAME"));
+    tracing::debug!("{:?}", std::env::var("EXPLORER_API"));
+    tracing::debug!("{:?}", std::env::var("NYM_API"));
+
+    let storage = db::Storage::init().await?;
+    monitor::spawn_in_background(storage)
+        .await
+        .expect("Monitor task failed");
+    tracing::info!("Started server");
+
+    Ok(())
+}
+
+pub(crate) fn read_env_var(env_var: &str) -> anyhow::Result<String> {
+    std::env::var(env_var)
+        .map(|value| {
+            tracing::trace!("{}={}", env_var, value);
+            value
+        })
+        .map_err(|_| anyhow!("You need to set {}", env_var))
+}

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

@@ -0,0 +1,374 @@
+use crate::db::models::{
+    gateway, mixnode, GatewayRecord, MixnodeRecord, NetworkSummary, GATEWAYS_BLACKLISTED_COUNT,
+    GATEWAYS_BONDED_COUNT, GATEWAYS_EXPLORER_COUNT, GATEWAYS_HISTORICAL_COUNT,
+    MIXNODES_BLACKLISTED_COUNT, MIXNODES_BONDED_ACTIVE, MIXNODES_BONDED_COUNT,
+    MIXNODES_BONDED_INACTIVE, MIXNODES_BONDED_RESERVE, MIXNODES_HISTORICAL_COUNT,
+};
+use crate::db::{queries, DbPool, Storage};
+use anyhow::anyhow;
+use cosmwasm_std::Decimal;
+use nym_explorer_client::{ExplorerClient, PrettyDetailedGatewayBond};
+use nym_network_defaults::NymNetworkDetails;
+use nym_validator_client::client::NymApiClientExt;
+use nym_validator_client::models::{DescribedGateway, DescribedMixNode, MixNodeBondAnnotated};
+use nym_validator_client::nym_nodes::SkimmedNode;
+use nym_validator_client::nyxd::contract_traits::PagedMixnetQueryClient;
+use nym_validator_client::nyxd::{AccountId, NyxdClient};
+use nym_validator_client::NymApiClient;
+use std::collections::HashSet;
+use std::str::FromStr;
+use tokio::task::JoinHandle;
+use tokio::time::Duration;
+
+const REFRESH_DELAY: Duration = Duration::from_secs(60 * 5);
+const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60);
+static DELEGATION_PROGRAM_WALLET: &str = "n1rnxpdpx3kldygsklfft0gech7fhfcux4zst5lw";
+
+// TODO dz: query many NYM APIs:
+// multiple instances running directory cache, ask sachin
+pub(crate) fn spawn_in_background(storage: Storage) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        let db_pool = storage.pool().await;
+        let network_defaults = nym_network_defaults::NymNetworkDetails::new_from_env();
+
+        loop {
+            tracing::info!("Refreshing node info...");
+
+            if let Err(e) = run(db_pool, &network_defaults).await {
+                tracing::error!(
+                    "Monitor run failed: {e}, retrying in {}s...",
+                    FAILURE_RETRY_DELAY.as_secs()
+                );
+                tokio::time::sleep(FAILURE_RETRY_DELAY).await;
+            } else {
+                tracing::info!(
+                    "Info successfully collected, sleeping for {}s...",
+                    REFRESH_DELAY.as_secs()
+                );
+                tokio::time::sleep(REFRESH_DELAY).await;
+            }
+        }
+    })
+}
+
+async fn run(pool: &DbPool, network_details: &NymNetworkDetails) -> anyhow::Result<()> {
+    let default_api_url = network_details
+        .endpoints
+        .first()
+        .expect("rust sdk mainnet default incorrectly configured")
+        .api_url()
+        .clone()
+        .expect("rust sdk mainnet default missing api_url");
+    let default_explorer_url = network_details.explorer_api.clone().map(|url| {
+        url.parse()
+            .expect("rust sdk mainnet default explorer url not parseable")
+    });
+
+    let default_explorer_url =
+        default_explorer_url.expect("explorer url missing in network config");
+    let explorer_client = ExplorerClient::new(default_explorer_url)?;
+    let explorer_gateways = explorer_client.get_gateways().await?;
+
+    let api_client = NymApiClient::new(default_api_url);
+    let gateways = api_client.get_cached_described_gateways().await?;
+    tracing::debug!("Fetched {} gateways", gateways.len());
+    let skimmed_gateways = api_client.get_basic_gateways(None).await?;
+
+    let mixnodes = api_client.get_cached_mixnodes().await?;
+    tracing::debug!("Fetched {} mixnodes", mixnodes.len());
+
+    // TODO dz can we calculate blacklisted GWs from their performance?
+    // where do we get their performance?
+    let gateways_blacklisted = api_client
+        .nym_api
+        .get_gateways_blacklisted()
+        .await
+        .map(|vec| vec.into_iter().collect::<HashSet<_>>())?;
+
+    // Cached mixnodes don't include blacklisted nodes
+    // We need that to calculate the total locked tokens later
+    let mixnodes = api_client
+        .nym_api
+        .get_mixnodes_detailed_unfiltered()
+        .await?;
+    let mixnodes_described = api_client.nym_api.get_mixnodes_described().await?;
+    let mixnodes_active = api_client.nym_api.get_active_mixnodes().await?;
+    let delegation_program_members = get_delegation_program_details(network_details).await?;
+
+    // keep stats for later
+    let count_bonded_mixnodes = mixnodes.len();
+    let count_bonded_gateways = gateways.len();
+    let count_explorer_gateways = explorer_gateways.len();
+    let count_bonded_mixnodes_active = mixnodes_active.len();
+
+    let gateway_records = prepare_gateway_data(
+        &gateways,
+        &gateways_blacklisted,
+        explorer_gateways,
+        skimmed_gateways,
+    )?;
+    queries::insert_gateways(pool, gateway_records)
+        .await
+        .map(|_| {
+            tracing::debug!("Gateway info written to DB!");
+        })?;
+
+    // instead of counting blacklisted GWs returned from API cache, count from the active set
+    let count_gateways_blacklisted = gateways
+        .iter()
+        .filter(|gw| {
+            let gw_identity = gw.bond.identity();
+            gateways_blacklisted.contains(gw_identity)
+        })
+        .count();
+
+    if count_gateways_blacklisted > 0 {
+        queries::write_blacklisted_gateways_to_db(pool, gateways_blacklisted.iter())
+            .await
+            .map(|_| {
+                tracing::debug!(
+                    "Gateway blacklist info written to DB! {} blacklisted by Nym API",
+                    count_gateways_blacklisted
+                )
+            })?;
+    }
+
+    let mixnode_records =
+        prepare_mixnode_data(&mixnodes, mixnodes_described, delegation_program_members)?;
+    queries::insert_mixnodes(pool, mixnode_records)
+        .await
+        .map(|_| {
+            tracing::debug!("Mixnode info written to DB!");
+        })?;
+
+    let count_mixnodes_blacklisted = mixnodes.iter().filter(|elem| elem.blacklisted).count();
+
+    let recently_unbonded_gateways = queries::ensure_gateways_still_bonded(pool, &gateways).await?;
+    let recently_unbonded_mixnodes = queries::ensure_mixnodes_still_bonded(pool, &mixnodes).await?;
+
+    let count_bonded_mixnodes_reserve = 0; // TODO: NymAPI doesn't report the reserve set size
+    let count_bonded_mixnodes_inactive = count_bonded_mixnodes - count_bonded_mixnodes_active;
+
+    let (all_historical_gateways, all_historical_mixnodes) = calculate_stats(pool).await?;
+
+    //
+    // write summary keys and values to table
+    //
+
+    let nodes_summary = vec![
+        (MIXNODES_BONDED_COUNT, &count_bonded_mixnodes),
+        (MIXNODES_BONDED_ACTIVE, &count_bonded_mixnodes_active),
+        (MIXNODES_BONDED_INACTIVE, &count_bonded_mixnodes_inactive),
+        (MIXNODES_BONDED_RESERVE, &count_bonded_mixnodes_reserve),
+        (MIXNODES_BLACKLISTED_COUNT, &count_mixnodes_blacklisted),
+        (GATEWAYS_BONDED_COUNT, &count_bonded_gateways),
+        (GATEWAYS_EXPLORER_COUNT, &count_explorer_gateways),
+        (MIXNODES_HISTORICAL_COUNT, &all_historical_mixnodes),
+        (GATEWAYS_HISTORICAL_COUNT, &all_historical_gateways),
+        (GATEWAYS_BLACKLISTED_COUNT, &count_gateways_blacklisted),
+    ];
+
+    // TODO dz do we need signed int in type definition? maybe because of API?
+    let last_updated = chrono::offset::Utc::now();
+    let last_updated_utc = last_updated.timestamp().to_string();
+    let network_summary = NetworkSummary {
+        mixnodes: mixnode::MixnodeSummary {
+            bonded: mixnode::MixnodeSummaryBonded {
+                count: count_bonded_mixnodes as i32,
+                active: count_bonded_mixnodes_active as i32,
+                inactive: count_bonded_mixnodes_inactive as i32,
+                reserve: count_bonded_mixnodes_reserve as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+            blacklisted: mixnode::MixnodeSummaryBlacklisted {
+                count: count_mixnodes_blacklisted as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+            historical: mixnode::MixnodeSummaryHistorical {
+                count: all_historical_mixnodes as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+        },
+        gateways: gateway::GatewaySummary {
+            bonded: gateway::GatewaySummaryBonded {
+                count: count_bonded_gateways as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+            blacklisted: gateway::GatewaySummaryBlacklisted {
+                count: count_gateways_blacklisted as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+            historical: gateway::GatewaySummaryHistorical {
+                count: all_historical_gateways as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+            explorer: gateway::GatewaySummaryExplorer {
+                count: count_explorer_gateways as i32,
+                last_updated_utc: last_updated_utc.to_owned(),
+            },
+        },
+    };
+
+    queries::insert_summaries(pool, &nodes_summary, &network_summary, last_updated).await?;
+
+    let mut log_lines: Vec<String> = vec![];
+    for (key, value) in nodes_summary.iter() {
+        log_lines.push(format!("{} = {}", key, value));
+    }
+    log_lines.push(format!(
+        "recently_unbonded_mixnodes = {}",
+        recently_unbonded_mixnodes
+    ));
+    log_lines.push(format!(
+        "recently_unbonded_gateways = {}",
+        recently_unbonded_gateways
+    ));
+
+    tracing::info!("Directory summary: \n{}", log_lines.join("\n"));
+
+    Ok(())
+}
+
+fn prepare_gateway_data(
+    gateways: &[DescribedGateway],
+    gateways_blacklisted: &HashSet<String>,
+    explorer_gateways: Vec<PrettyDetailedGatewayBond>,
+    skimmed_gateways: Vec<SkimmedNode>,
+) -> anyhow::Result<Vec<GatewayRecord>> {
+    let mut gateway_records = Vec::new();
+
+    for gateway in gateways {
+        let identity_key = gateway.bond.identity();
+        let bonded = true;
+        let last_updated_utc = chrono::offset::Utc::now().timestamp();
+        let blacklisted = gateways_blacklisted.contains(identity_key);
+
+        let self_described = gateway
+            .self_described
+            .as_ref()
+            .and_then(|v| serde_json::to_string(&v).ok());
+
+        let explorer_pretty_bond = explorer_gateways
+            .iter()
+            .find(|g| g.gateway.identity_key.eq(identity_key));
+        let explorer_pretty_bond = explorer_pretty_bond.and_then(|g| serde_json::to_string(g).ok());
+
+        let performance = skimmed_gateways
+            .iter()
+            .find(|g| g.ed25519_identity_pubkey.eq(identity_key))
+            .map(|g| g.performance)
+            .unwrap_or_default()
+            .round_to_integer();
+
+        gateway_records.push(GatewayRecord {
+            identity_key: identity_key.to_owned(),
+            bonded,
+            blacklisted,
+            self_described,
+            explorer_pretty_bond,
+            last_updated_utc,
+            performance,
+        });
+    }
+
+    Ok(gateway_records)
+}
+
+fn prepare_mixnode_data(
+    mixnodes: &[MixNodeBondAnnotated],
+    mixnodes_described: Vec<DescribedMixNode>,
+    delegation_program_members: Vec<u32>,
+) -> anyhow::Result<Vec<MixnodeRecord>> {
+    let mut mixnode_records = Vec::new();
+
+    for mixnode in mixnodes {
+        let mix_id = mixnode.mix_id();
+        let identity_key = mixnode.identity_key();
+        let bonded = true;
+        let total_stake = decimal_to_i64(mixnode.mixnode_details.total_stake());
+        let blacklisted = mixnode.blacklisted;
+        let node_info = mixnode.mix_node();
+        let host = node_info.host.clone();
+        let http_port = node_info.http_api_port;
+        // Contains all the information including what's above
+        let full_details = serde_json::to_string(&mixnode)?;
+
+        let mixnode_described = mixnodes_described.iter().find(|m| m.bond.mix_id == mix_id);
+        let self_described = mixnode_described.and_then(|v| serde_json::to_string(v).ok());
+        let is_dp_delegatee = delegation_program_members.contains(&mix_id);
+
+        let last_updated_utc = chrono::offset::Utc::now().timestamp();
+
+        mixnode_records.push(MixnodeRecord {
+            mix_id,
+            identity_key: identity_key.to_owned(),
+            bonded,
+            total_stake,
+            host,
+            http_port,
+            blacklisted,
+            full_details,
+            self_described,
+            last_updated_utc,
+            is_dp_delegatee,
+        });
+    }
+
+    Ok(mixnode_records)
+}
+
+async fn calculate_stats(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"#)
+        .fetch_one(&mut *conn)
+        .await? as usize;
+
+    let all_historical_mixnodes = sqlx::query_scalar!(r#"SELECT count(id) FROM mixnodes"#)
+        .fetch_one(&mut *conn)
+        .await? as usize;
+
+    Ok((all_historical_gateways, all_historical_mixnodes))
+}
+
+async fn get_delegation_program_details(
+    network_details: &NymNetworkDetails,
+) -> anyhow::Result<Vec<u32>> {
+    let config = nym_validator_client::nyxd::Config::try_from_nym_network_details(network_details)?;
+
+    // TODO dz should this be configurable?
+    let client = NyxdClient::connect(config, "https://rpc.nymtech.net")
+        .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.mix_id)
+        .collect();
+
+    Ok(mix_ids)
+}
+
+fn decimal_to_i64(decimal: Decimal) -> i64 {
+    // Convert the underlying Uint128 to a u128
+    let atomics = decimal.atomics().u128();
+    let precision = 1_000_000_000_000_000_000u128;
+
+    // Get the fractional part
+    let fractional = atomics % precision;
+
+    // Get the integer part
+    let integer = atomics / precision;
+
+    // Combine them into a float
+    let float_value = integer as f64 + (fractional as f64 / 1_000_000_000_000_000_000_f64);
+
+    // Limit to 6 decimal places
+    let rounded_value = (float_value * 1_000_000.0).round() / 1_000_000.0;
+
+    rounded_value as i64
+}