feat: node stats (#85)

* feat: node stats

* fix: remove ip

docker-compose.yaml

@@ -32,6 +32,7 @@ services:
       DATABASE_CACHE_DIR: /var/cache/cheburcheck/databases
       DATABASE_RETRY_INTERVAL_SECONDS: "${DATABASE_RETRY_INTERVAL_SECONDS:-300}"
       MQTT_ADMIN_TOKEN: "${MQTT_ADMIN_TOKEN}"
+      NODE_STATS_KEY: "${NODE_STATS_KEY}"
       MQTT_HOST: rmqtt
       MQTT_PORT: 11883
       PROBE_TRACEROUTE_ENABLED: "${PROBE_TRACEROUTE_ENABLED:-false}"

nginx.conf

@@ -19,7 +19,7 @@ server {
     listen 80;
     server_name _;
 
-    set_real_ip_from 172.18.0.0/16;
+    set_real_ip_from 172.16.0.0/12;
     real_ip_header X-Forwarded-For;
     real_ip_recursive on;
 
@@ -58,6 +58,9 @@ server {
     }
 
     location / {
+        proxy_buffer_size       32k;
+        proxy_buffers           8 32k;
+        proxy_busy_buffers_size 64k;
         proxy_pass http://frontend;
     }
 }

website/migrations/20260824000000_reporter_connection_info.sql

@@ -0,0 +1,2 @@
+ALTER TABLE reporters
+    ADD COLUMN IF NOT EXISTS last_connected_at TIMESTAMPTZ;

website/src/api.rs

@@ -2,12 +2,15 @@
 mod check_endpoint;
 #[path = "api/feedback.rs"]
 mod feedback_endpoint;
+#[path = "api/node_stats.rs"]
+mod nodes_endpoint;
 mod probe;
 mod rate_limit;
 mod status;
 
 pub use check_endpoint::check;
 pub use feedback_endpoint::feedback;
+pub use nodes_endpoint::node_stats;
 pub use probe::probe_query;
 pub use rate_limit::{build_probe_rate_limiter, build_rate_limiter};
 pub use status::{get_system_status, healthcheck};

website/src/api/node_stats.rs

@@ -0,0 +1,192 @@
+use crate::mqtt::{MqttPublisher, ProbeStatusSnapshot};
+use rocket::http::Status;
+use rocket::request::{FromRequest, Outcome};
+use rocket::serde::json::Json;
+use rocket::{Request, State};
+use serde::Serialize;
+use sqlx::PgPool;
+use sqlx::types::chrono::{DateTime, Utc};
+use std::collections::HashMap;
+
+#[derive(sqlx::FromRow)]
+struct ProbeMetadata {
+    id: i32,
+    name: String,
+    region: Option<String>,
+    provider: Option<String>,
+    asn: Option<String>,
+    last_connected_at: Option<DateTime<Utc>>,
+}
+
+pub struct NodeStatsKey;
+
+#[rocket::async_trait]
+impl<'r> FromRequest<'r> for NodeStatsKey {
+    type Error = ();
+
+    async fn from_request(request: &'r Request<'_>) -> Outcome<Self, Self::Error> {
+        let expected_key = std::env::var("NODE_STATS_KEY")
+            .ok()
+            .filter(|key| !key.is_empty());
+        let provided_key = request
+            .headers()
+            .get_one("Authorization")
+            .and_then(|header| header.strip_prefix("Bearer "));
+
+        if valid_key(provided_key, expected_key.as_deref()) {
+            Outcome::Success(Self)
+        } else {
+            Outcome::Error((Status::Unauthorized, ()))
+        }
+    }
+}
+
+#[derive(Debug, Serialize, PartialEq, Eq)]
+pub struct NodeStatsResponse {
+    nodes: Vec<NodeStatus>,
+}
+
+#[derive(Debug, Serialize, PartialEq, Eq)]
+struct NodeStatus {
+    probe_id: i32,
+    name: String,
+    region: Option<String>,
+    provider: Option<String>,
+    asn: Option<String>,
+    connected_at: Option<DateTime<Utc>>,
+    online: bool,
+    version: Option<String>,
+    dpi_hop_v4: Option<u8>,
+    dpi_hop_v6: Option<u8>,
+}
+
+#[get("/nodes")]
+pub async fn node_stats(
+    _key: NodeStatsKey,
+    pool: &State<PgPool>,
+    mqtt: &State<MqttPublisher>,
+) -> Result<Json<NodeStatsResponse>, Status> {
+    let probes = sqlx::query_as::<_, ProbeMetadata>(
+        "SELECT id, name, region, provider, asn, last_connected_at
+         FROM reporters
+         ORDER BY id",
+    )
+    .fetch_all(&**pool)
+    .await
+    .map_err(|error| {
+        log::error!("failed to load probe metadata for node stats: {error}");
+        Status::InternalServerError
+    })?;
+
+    let statuses = mqtt.probe_statuses().await;
+    Ok(Json(NodeStatsResponse {
+        nodes: build_node_statuses(probes, &statuses),
+    }))
+}
+
+fn valid_key(provided: Option<&str>, expected: Option<&str>) -> bool {
+    provided
+        .zip(expected)
+        .is_some_and(|(provided, expected)| provided == expected && !provided.is_empty())
+}
+
+fn build_node_statuses(
+    probes: Vec<ProbeMetadata>,
+    statuses: &HashMap<String, ProbeStatusSnapshot>,
+) -> Vec<NodeStatus> {
+    probes
+        .into_iter()
+        .map(|probe| {
+            let status = statuses.get(&probe.id.to_string());
+            NodeStatus {
+                probe_id: probe.id,
+                name: probe.name,
+                region: probe.region,
+                provider: probe.provider,
+                asn: probe.asn,
+                connected_at: probe.last_connected_at,
+                online: status.is_some_and(|status| status.online),
+                version: status.map(|status| status.version.clone()),
+                dpi_hop_v4: status.and_then(|status| status.dpi_hop_v4),
+                dpi_hop_v6: status.and_then(|status| status.dpi_hop_v6),
+            }
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn requires_a_configured_matching_key() {
+        assert!(valid_key(Some("secret"), Some("secret")));
+        assert!(!valid_key(Some("wrong"), Some("secret")));
+        assert!(!valid_key(None, Some("secret")));
+        assert!(!valid_key(Some("secret"), None));
+        assert!(!valid_key(Some(""), Some("")));
+    }
+
+    #[test]
+    fn builds_online_and_offline_json_nodes() {
+        let probes = vec![
+            ProbeMetadata {
+                id: 1,
+                name: "Home router".to_string(),
+                region: Some("Ural".to_string()),
+                provider: Some("ExampleNet".to_string()),
+                asn: Some("AS64500".to_string()),
+                last_connected_at: Some("2026-08-24T12:34:56Z".parse().unwrap()),
+            },
+            ProbeMetadata {
+                id: 2,
+                name: "Offline".to_string(),
+                region: None,
+                provider: None,
+                asn: None,
+                last_connected_at: Some("2026-08-23T12:34:56Z".parse().unwrap()),
+            },
+        ];
+        let statuses = HashMap::from([(
+            "1".to_string(),
+            ProbeStatusSnapshot {
+                online: true,
+                version: "1.2.3".to_string(),
+                dpi_hop_v4: Some(5),
+                dpi_hop_v6: None,
+            },
+        )]);
+
+        let nodes = build_node_statuses(probes, &statuses);
+
+        assert_eq!(
+            nodes,
+            vec![
+                NodeStatus {
+                    probe_id: 1,
+                    name: "Home router".to_string(),
+                    region: Some("Ural".to_string()),
+                    provider: Some("ExampleNet".to_string()),
+                    asn: Some("AS64500".to_string()),
+                    connected_at: Some("2026-08-24T12:34:56Z".parse().unwrap()),
+                    online: true,
+                    version: Some("1.2.3".to_string()),
+                    dpi_hop_v4: Some(5),
+                    dpi_hop_v6: None,
+                },
+                NodeStatus {
+                    probe_id: 2,
+                    name: "Offline".to_string(),
+                    region: None,
+                    provider: None,
+                    asn: None,
+                    connected_at: Some("2026-08-23T12:34:56Z".parse().unwrap()),
+                    online: false,
+                    version: None,
+                    dpi_hop_v4: None,
+                    dpi_hop_v6: None,
+                },
+            ]
+        );
+    }
+}

website/src/main.rs

@@ -100,6 +100,7 @@ async fn rocket() -> _ {
                 api::healthcheck,
                 api::feedback,
                 api::get_system_status,
+                api::node_stats,
                 whitelist::histogram
             ],
         )

website/src/mqtt.rs

@@ -8,7 +8,7 @@ use rumqttc::{AsyncClient, Event as MqttEvent, Incoming, MqttOptions, QoS};
 use serde::Deserialize;
 use sqlx::types::Uuid;
 use sqlx::types::chrono::Utc;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::fmt;
 use std::net::IpAddr;
 use std::sync::Arc;
@@ -53,13 +53,22 @@ impl fmt::Display for PublishError {
 pub struct MqttPublisher {
     client: Option<AsyncClient>,
     sessions: Arc<rocket::tokio::sync::RwLock<HashMap<String, ProbeResultSender>>>,
-    online_probes: Arc<rocket::tokio::sync::RwLock<HashSet<String>>>,
+    probe_statuses: ProbeStatuses,
     probe_config: Arc<ProbeConfig>,
     task_timeout_ms: u64,
 }
 
 type ProbeResultSender = rocket::tokio::sync::broadcast::Sender<ProbeResultEvent>;
 pub type ProbeResultReceiver = rocket::tokio::sync::broadcast::Receiver<ProbeResultEvent>;
+type ProbeStatuses = Arc<rocket::tokio::sync::RwLock<HashMap<String, ProbeStatusSnapshot>>>;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ProbeStatusSnapshot {
+    pub online: bool,
+    pub version: String,
+    pub dpi_hop_v4: Option<u8>,
+    pub dpi_hop_v6: Option<u8>,
+}
 
 #[derive(Deserialize)]
 struct ProbeHostsFile {
@@ -87,7 +96,7 @@ struct ProbeHostEntry {
 impl MqttPublisher {
     pub fn start_from_env() -> Self {
         let sessions = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new()));
-        let online_probes = Arc::new(rocket::tokio::sync::RwLock::new(HashSet::new()));
+        let probe_statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new()));
         let task_timeout_ms = task_timeout_ms_from_env();
         let probe_config = Arc::new(load_probe_config(task_timeout_ms).unwrap_or_else(|error| {
             warn!("failed to load probe config: {error}");
@@ -110,7 +119,7 @@ impl MqttPublisher {
                 return Self {
                     client: None,
                     sessions,
-                    online_probes,
+                    probe_statuses,
                     probe_config,
                     task_timeout_ms: task_timeout_ms_from_env(),
                 };
@@ -132,7 +141,7 @@ impl MqttPublisher {
 
         let (client, mut eventloop) = AsyncClient::new(options, 100);
         let event_sessions = sessions.clone();
-        let event_online_probes = online_probes.clone();
+        let event_probe_statuses = probe_statuses.clone();
         let config_client = client.clone();
         let event_probe_config = probe_config.clone();
         rocket::tokio::spawn(async move {
@@ -149,7 +158,7 @@ impl MqttPublisher {
                         dispatch_probe_result(&event_sessions, &publish.topic, &publish.payload)
                             .await;
                         dispatch_probe_status(
-                            &event_online_probes,
+                            &event_probe_statuses,
                             &publish.topic,
                             &publish.payload,
                         )
@@ -178,7 +187,7 @@ impl MqttPublisher {
         Self {
             client: Some(client),
             sessions,
-            online_probes,
+            probe_statuses,
             probe_config,
             task_timeout_ms,
         }
@@ -189,7 +198,16 @@ impl MqttPublisher {
     }
 
     pub async fn online_probe_count(&self) -> usize {
-        self.online_probes.read().await.len()
+        self.probe_statuses
+            .read()
+            .await
+            .values()
+            .filter(|status| status.online)
+            .count()
+    }
+
+    pub async fn probe_statuses(&self) -> HashMap<String, ProbeStatusSnapshot> {
+        self.probe_statuses.read().await.clone()
     }
 
     pub fn probe_config(&self) -> Arc<ProbeConfig> {
@@ -320,17 +338,13 @@ fn parse_probe_hosts(contents: &str) -> Result<ParsedProbeConfig, PublishError>
     })
 }
 
-async fn dispatch_probe_status(
-    online_probes: &Arc<rocket::tokio::sync::RwLock<HashSet<String>>>,
-    topic: &str,
-    payload: &[u8],
-) {
+async fn dispatch_probe_status(probe_statuses: &ProbeStatuses, topic: &str, payload: &[u8]) {
     let Some(probe_id) = parse_probe_status_topic(topic) else {
         return;
     };
 
     if payload.is_empty() {
-        online_probes.write().await.remove(probe_id);
+        probe_statuses.write().await.remove(probe_id);
         return;
     }
 
@@ -342,12 +356,23 @@ async fn dispatch_probe_status(
         }
     };
 
-    let mut online_probes = online_probes.write().await;
-    if status.online {
-        online_probes.insert(probe_id.to_string());
-    } else {
-        online_probes.remove(probe_id);
+    if status.probe_id != probe_id {
+        warn!(
+            "ignoring probe status on {topic}: payload probe_id {} does not match topic",
+            status.probe_id
+        );
+        return;
     }
+
+    probe_statuses.write().await.insert(
+        probe_id.to_string(),
+        ProbeStatusSnapshot {
+            online: status.online,
+            version: status.version.to_string(),
+            dpi_hop_v4: status.dpi_hop_v4,
+            dpi_hop_v6: status.dpi_hop_v6,
+        },
+    );
 }
 
 async fn dispatch_probe_result(
@@ -417,3 +442,61 @@ fn task_timeout_ms_from_env() -> u64 {
         .and_then(|timeout| timeout.parse().ok())
         .unwrap_or(15_000)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[rocket::async_test]
+    async fn status_snapshot_keeps_offline_node_metadata() {
+        let statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new()));
+
+        dispatch_probe_status(
+            &statuses,
+            "probe/status/v1/42",
+            br#"{"online":false,"probe_id":"42","version":"1.2.3","dpi_hop_v4":4,"dpi_hop_v6":6}"#,
+        )
+        .await;
+
+        assert_eq!(
+            statuses.read().await.get("42"),
+            Some(&ProbeStatusSnapshot {
+                online: false,
+                version: "1.2.3".to_string(),
+                dpi_hop_v4: Some(4),
+                dpi_hop_v6: Some(6),
+            })
+        );
+    }
+
+    #[rocket::async_test]
+    async fn empty_retained_status_removes_snapshot() {
+        let statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::from([(
+            "42".to_string(),
+            ProbeStatusSnapshot {
+                online: true,
+                version: "1.2.3".to_string(),
+                dpi_hop_v4: None,
+                dpi_hop_v6: None,
+            },
+        )])));
+
+        dispatch_probe_status(&statuses, "probe/status/v1/42", b"").await;
+
+        assert!(statuses.read().await.is_empty());
+    }
+
+    #[rocket::async_test]
+    async fn mismatched_payload_id_is_ignored() {
+        let statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new()));
+
+        dispatch_probe_status(
+            &statuses,
+            "probe/status/v1/42",
+            br#"{"online":true,"probe_id":"7","version":"1.2.3"}"#,
+        )
+        .await;
+
+        assert!(statuses.read().await.is_empty());
+    }
+}

website/src/mqtt_auth.rs

@@ -70,15 +70,24 @@ pub async fn auth(
         return Json(MqttAuthResponse::deny());
     }
 
-    let reporter_id =
-        sqlx::query_scalar::<_, i32>("SELECT id FROM reporters WHERE token = $1 LIMIT 1")
-            .bind(request.password)
-            .fetch_optional(&**pool)
-            .await
-            .ok()
-            .flatten();
+    let Ok(reporter_id) = request.clientid.parse::<i32>() else {
+        return Json(MqttAuthResponse::deny());
+    };
+    let authenticated = sqlx::query_scalar::<_, i32>(
+        "UPDATE reporters
+         SET last_connected_at = NOW()
+         WHERE id = $1 AND token = $2
+         RETURNING id",
+    )
+    .bind(reporter_id)
+    .bind(request.password)
+    .fetch_optional(&**pool)
+    .await
+    .ok()
+    .flatten()
+    .is_some();
 
-    if reporter_id.is_some_and(|id| id.to_string() == request.clientid) {
+    if authenticated {
         Json(MqttAuthResponse::allow())
     } else {
         Json(MqttAuthResponse::deny())