feat(proxy): record tunnel stability metrics

crates/aether-data/src/repository/proxy_nodes/memory.rs

@@ -7,10 +7,14 @@ use serde_json::{json, Map, Value};
 use uuid::Uuid;
 
 use super::types::{
-    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeHeartbeatMutation,
-    ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation, ProxyNodeReadRepository,
+    bucket_start_unix_secs, build_tunnel_error_event_detail, build_tunnel_metrics_sample,
+    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeEventQuery,
+    ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation,
+    ProxyNodeMetricsCleanupSummary, ProxyNodeMetricsStep, ProxyNodeReadRepository,
     ProxyNodeRegistrationMutation, ProxyNodeRemoteConfigMutation, ProxyNodeTrafficMutation,
-    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyNode, StoredProxyNodeEvent,
+    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyFleetMetricsBucket,
+    StoredProxyNode, StoredProxyNodeEvent, StoredProxyNodeMetricsBucket, TunnelMetricsSample,
+    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
 };
 use crate::DataLayerError;
 
@@ -18,6 +22,8 @@ use crate::DataLayerError;
 pub struct InMemoryProxyNodeRepository {
     nodes: RwLock<BTreeMap<String, StoredProxyNode>>,
     events: RwLock<Vec<StoredProxyNodeEvent>>,
+    metrics_1m: RwLock<BTreeMap<(String, u64), StoredProxyNodeMetricsBucket>>,
+    metrics_1h: RwLock<BTreeMap<(String, u64), StoredProxyNodeMetricsBucket>>,
 }
 
 impl InMemoryProxyNodeRepository {
@@ -33,6 +39,8 @@ impl InMemoryProxyNodeRepository {
                     .collect(),
             ),
             events: RwLock::new(Vec::new()),
+            metrics_1m: RwLock::new(BTreeMap::new()),
+            metrics_1h: RwLock::new(BTreeMap::new()),
         }
     }
 
@@ -49,6 +57,8 @@ impl InMemoryProxyNodeRepository {
                     .collect(),
             ),
             events: RwLock::new(events.into_iter().collect()),
+            metrics_1m: RwLock::new(BTreeMap::new()),
+            metrics_1h: RwLock::new(BTreeMap::new()),
         }
     }
 
@@ -63,6 +73,50 @@ impl InMemoryProxyNodeRepository {
         events.iter().map(|event| event.id).max().unwrap_or(0) + 1
     }
 
+    fn upsert_metrics_bucket(
+        metrics: &mut BTreeMap<(String, u64), StoredProxyNodeMetricsBucket>,
+        node_id: &str,
+        bucket_start_unix_secs: u64,
+        sample: &TunnelMetricsSample,
+    ) {
+        let key = (node_id.to_string(), bucket_start_unix_secs);
+        let bucket = metrics
+            .entry(key)
+            .or_insert_with(|| StoredProxyNodeMetricsBucket {
+                node_id: node_id.to_string(),
+                bucket_start_unix_secs,
+                samples: 0,
+                uptime_samples: 0,
+                active_connections_sum: 0,
+                active_connections_max: 0,
+                heartbeat_rtt_ms_sum: 0,
+                heartbeat_rtt_ms_max: 0,
+                connect_errors_delta: 0,
+                disconnects_delta: 0,
+                error_events_delta: 0,
+                ws_in_bytes_delta: 0,
+                ws_out_bytes_delta: 0,
+                ws_in_frames_delta: 0,
+                ws_out_frames_delta: 0,
+            });
+
+        bucket.samples += sample.samples;
+        bucket.uptime_samples += sample.uptime_samples;
+        bucket.active_connections_sum += sample.active_connections_sum;
+        bucket.active_connections_max = bucket
+            .active_connections_max
+            .max(sample.active_connections_max);
+        bucket.heartbeat_rtt_ms_sum += sample.heartbeat_rtt_ms_sum;
+        bucket.heartbeat_rtt_ms_max = bucket.heartbeat_rtt_ms_max.max(sample.heartbeat_rtt_ms_max);
+        bucket.connect_errors_delta += sample.connect_errors_delta;
+        bucket.disconnects_delta += sample.disconnects_delta;
+        bucket.error_events_delta += sample.error_events_delta;
+        bucket.ws_in_bytes_delta += sample.ws_in_bytes_delta;
+        bucket.ws_out_bytes_delta += sample.ws_out_bytes_delta;
+        bucket.ws_in_frames_delta += sample.ws_in_frames_delta;
+        bucket.ws_out_frames_delta += sample.ws_out_frames_delta;
+    }
+
     fn normalize_remote_config(
         mutation: &ProxyNodeRemoteConfigMutation,
         existing: Option<&Value>,
@@ -154,6 +208,130 @@ impl ProxyNodeReadRepository for InMemoryProxyNodeRepository {
         items.truncate(limit);
         Ok(items)
     }
+
+    async fn list_proxy_node_events_filtered(
+        &self,
+        node_id: &str,
+        query: &ProxyNodeEventQuery,
+    ) -> Result<Vec<StoredProxyNodeEvent>, DataLayerError> {
+        let events = self.events.read().expect("proxy node repository lock");
+        let mut items = events
+            .iter()
+            .filter(|event| event.node_id == node_id)
+            .filter(|event| {
+                query
+                    .from_unix_secs
+                    .map(|from| event.created_at_unix_ms.unwrap_or(0) >= from)
+                    .unwrap_or(true)
+            })
+            .filter(|event| {
+                query
+                    .to_unix_secs
+                    .map(|to| event.created_at_unix_ms.unwrap_or(u64::MAX) <= to)
+                    .unwrap_or(true)
+            })
+            .filter(|event| {
+                query
+                    .event_type
+                    .as_deref()
+                    .map(|event_type| event.event_type.eq_ignore_ascii_case(event_type))
+                    .unwrap_or(true)
+            })
+            .cloned()
+            .collect::<Vec<_>>();
+        items.sort_by(|left, right| {
+            right
+                .created_at_unix_ms
+                .unwrap_or(0)
+                .cmp(&left.created_at_unix_ms.unwrap_or(0))
+                .then(right.id.cmp(&left.id))
+        });
+        items.truncate(query.limit);
+        Ok(items)
+    }
+
+    async fn list_proxy_node_metrics(
+        &self,
+        node_id: &str,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyNodeMetricsBucket>, DataLayerError> {
+        let metrics = match step {
+            ProxyNodeMetricsStep::OneMinute => self.metrics_1m.read(),
+            ProxyNodeMetricsStep::OneHour => self.metrics_1h.read(),
+        }
+        .expect("proxy node repository lock");
+        let mut items = metrics
+            .values()
+            .filter(|bucket| bucket.node_id == node_id)
+            .filter(|bucket| bucket.bucket_start_unix_secs >= from_unix_secs)
+            .filter(|bucket| bucket.bucket_start_unix_secs <= to_unix_secs)
+            .cloned()
+            .collect::<Vec<_>>();
+        items.sort_by_key(|bucket| bucket.bucket_start_unix_secs);
+        items.truncate(limit);
+        Ok(items)
+    }
+
+    async fn list_proxy_fleet_metrics(
+        &self,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyFleetMetricsBucket>, DataLayerError> {
+        let metrics = match step {
+            ProxyNodeMetricsStep::OneMinute => self.metrics_1m.read(),
+            ProxyNodeMetricsStep::OneHour => self.metrics_1h.read(),
+        }
+        .expect("proxy node repository lock");
+        let mut grouped = BTreeMap::<u64, StoredProxyFleetMetricsBucket>::new();
+        for bucket in metrics.values() {
+            if bucket.bucket_start_unix_secs < from_unix_secs
+                || bucket.bucket_start_unix_secs > to_unix_secs
+            {
+                continue;
+            }
+            let item = grouped
+                .entry(bucket.bucket_start_unix_secs)
+                .or_insert_with(|| StoredProxyFleetMetricsBucket {
+                    bucket_start_unix_secs: bucket.bucket_start_unix_secs,
+                    samples: 0,
+                    uptime_samples: 0,
+                    active_connections_sum: 0,
+                    active_connections_max: 0,
+                    heartbeat_rtt_ms_sum: 0,
+                    heartbeat_rtt_ms_max: 0,
+                    connect_errors_delta: 0,
+                    disconnects_delta: 0,
+                    error_events_delta: 0,
+                    ws_in_bytes_delta: 0,
+                    ws_out_bytes_delta: 0,
+                    ws_in_frames_delta: 0,
+                    ws_out_frames_delta: 0,
+                });
+            item.samples += bucket.samples;
+            item.uptime_samples += bucket.uptime_samples;
+            item.active_connections_sum += bucket.active_connections_sum;
+            item.active_connections_max = item
+                .active_connections_max
+                .max(bucket.active_connections_max);
+            item.heartbeat_rtt_ms_sum += bucket.heartbeat_rtt_ms_sum;
+            item.heartbeat_rtt_ms_max = item.heartbeat_rtt_ms_max.max(bucket.heartbeat_rtt_ms_max);
+            item.connect_errors_delta += bucket.connect_errors_delta;
+            item.disconnects_delta += bucket.disconnects_delta;
+            item.error_events_delta += bucket.error_events_delta;
+            item.ws_in_bytes_delta += bucket.ws_in_bytes_delta;
+            item.ws_out_bytes_delta += bucket.ws_out_bytes_delta;
+            item.ws_in_frames_delta += bucket.ws_in_frames_delta;
+            item.ws_out_frames_delta += bucket.ws_out_frames_delta;
+        }
+        let mut items = grouped.into_values().collect::<Vec<_>>();
+        items.truncate(limit);
+        Ok(items)
+    }
 }
 
 #[async_trait]
@@ -376,65 +554,114 @@ impl ProxyNodeWriteRepository for InMemoryProxyNodeRepository {
         &self,
         mutation: &ProxyNodeHeartbeatMutation,
     ) -> Result<Option<StoredProxyNode>, DataLayerError> {
-        let mut nodes = self.nodes.write().expect("proxy node repository lock");
-        let Some(node) = nodes.get_mut(&mutation.node_id) else {
-            return Ok(None);
+        let (node, sample, now_unix_secs) = {
+            let mut nodes = self.nodes.write().expect("proxy node repository lock");
+            let Some(node) = nodes.get_mut(&mutation.node_id) else {
+                return Ok(None);
+            };
+            if !node.tunnel_mode {
+                return Err(DataLayerError::InvalidInput(
+                    "non-tunnel mode is no longer supported, please upgrade aether-proxy to use tunnel mode"
+                        .to_string(),
+                ));
+            }
+
+            let previous_proxy_metadata = node.proxy_metadata.clone();
+            let now_unix_secs = Self::now_unix_secs().unwrap_or(0);
+            let now = Some(now_unix_secs);
+            node.last_heartbeat_at_unix_secs = now;
+            if node.status != "online" || !node.tunnel_connected {
+                node.status = "online".to_string();
+                node.tunnel_connected = true;
+                node.tunnel_connected_at_unix_secs = now;
+                node.updated_at_unix_secs = now;
+            }
+
+            if let Some(value) = mutation.heartbeat_interval {
+                node.heartbeat_interval = value;
+            }
+            if let Some(value) = mutation.active_connections {
+                node.active_connections = value;
+            }
+            if let Some(value) = mutation.avg_latency_ms {
+                node.avg_latency_ms = Some(value);
+            }
+            let normalized_proxy_metadata = normalize_proxy_metadata(
+                mutation.proxy_metadata.as_ref(),
+                mutation.proxy_version.as_deref(),
+            );
+            if let Some(value) = normalized_proxy_metadata {
+                node.proxy_metadata = Some(value);
+            }
+            if let Some(value) = mutation.total_requests_delta.filter(|value| *value > 0) {
+                node.total_requests += value;
+            }
+            if let Some(value) = mutation.failed_requests_delta.filter(|value| *value > 0) {
+                node.failed_requests += value;
+            }
+            if let Some(value) = mutation.dns_failures_delta.filter(|value| *value > 0) {
+                node.dns_failures += value;
+            }
+            if let Some(value) = mutation.stream_errors_delta.filter(|value| *value > 0) {
+                node.stream_errors += value;
+            }
+            let reconciled_remote_config = reconcile_remote_config_after_heartbeat(
+                node.remote_config.as_ref(),
+                mutation.proxy_version.as_deref(),
+            );
+            if reconciled_remote_config != node.remote_config {
+                node.remote_config = reconciled_remote_config;
+                node.config_version = node.config_version.saturating_add(1);
+                node.updated_at_unix_secs = now;
+            }
+
+            let sample = build_tunnel_metrics_sample(
+                previous_proxy_metadata.as_ref(),
+                node.proxy_metadata.as_ref(),
+                node.active_connections,
+                node.tunnel_connected,
+            );
+            (node.clone(), sample, now_unix_secs)
         };
-        if !node.tunnel_mode {
-            return Err(DataLayerError::InvalidInput(
-                "non-tunnel mode is no longer supported, please upgrade aether-proxy to use tunnel mode"
-                    .to_string(),
-            ));
+
+        if let Some(sample) = sample.as_ref() {
+            Self::upsert_metrics_bucket(
+                &mut self.metrics_1m.write().expect("proxy node repository lock"),
+                &node.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneMinute),
+                sample,
+            );
+            Self::upsert_metrics_bucket(
+                &mut self.metrics_1h.write().expect("proxy node repository lock"),
+                &node.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneHour),
+                sample,
+            );
+
+            let mut events = self.events.write().expect("proxy node repository lock");
+            for error in &sample.recent_error_events {
+                let event_id = Self::next_event_id(&events);
+                events.push(StoredProxyNodeEvent {
+                    id: event_id,
+                    node_id: node.id.clone(),
+                    event_type: PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR.to_string(),
+                    detail: Some(build_tunnel_error_event_detail(error)),
+                    event_metadata: Some(json!({
+                        "source": "heartbeat",
+                        "category": error.category,
+                        "message": error.message,
+                        "timestamp_unix_secs": error.timestamp_unix_secs,
+                    })),
+                    created_at_unix_ms: Some(if error.timestamp_unix_secs == 0 {
+                        now_unix_secs
+                    } else {
+                        error.timestamp_unix_secs
+                    }),
+                });
+            }
         }
 
-        let now = Self::now_unix_secs();
-        node.last_heartbeat_at_unix_secs = now;
-        if node.status != "online" || !node.tunnel_connected {
-            node.status = "online".to_string();
-            node.tunnel_connected = true;
-            node.tunnel_connected_at_unix_secs = now;
-            node.updated_at_unix_secs = now;
-        }
-
-        if let Some(value) = mutation.heartbeat_interval {
-            node.heartbeat_interval = value;
-        }
-        if let Some(value) = mutation.active_connections {
-            node.active_connections = value;
-        }
-        if let Some(value) = mutation.avg_latency_ms {
-            node.avg_latency_ms = Some(value);
-        }
-        let normalized_proxy_metadata = normalize_proxy_metadata(
-            mutation.proxy_metadata.as_ref(),
-            mutation.proxy_version.as_deref(),
-        );
-        if let Some(value) = normalized_proxy_metadata {
-            node.proxy_metadata = Some(value);
-        }
-        if let Some(value) = mutation.total_requests_delta.filter(|value| *value > 0) {
-            node.total_requests += value;
-        }
-        if let Some(value) = mutation.failed_requests_delta.filter(|value| *value > 0) {
-            node.failed_requests += value;
-        }
-        if let Some(value) = mutation.dns_failures_delta.filter(|value| *value > 0) {
-            node.dns_failures += value;
-        }
-        if let Some(value) = mutation.stream_errors_delta.filter(|value| *value > 0) {
-            node.stream_errors += value;
-        }
-        let reconciled_remote_config = reconcile_remote_config_after_heartbeat(
-            node.remote_config.as_ref(),
-            mutation.proxy_version.as_deref(),
-        );
-        if reconciled_remote_config != node.remote_config {
-            node.remote_config = reconciled_remote_config;
-            node.config_version = node.config_version.saturating_add(1);
-            node.updated_at_unix_secs = now;
-        }
-
-        Ok(Some(node.clone()))
+        Ok(Some(node))
     }
 
     async fn record_traffic(
@@ -490,6 +717,7 @@ impl ProxyNodeWriteRepository for InMemoryProxyNodeRepository {
                     node_id: mutation.node_id.clone(),
                     event_type: event_type.to_string(),
                     detail: Some(format!("[stale_ignored] {event_detail}")),
+                    event_metadata: None,
                     created_at_unix_ms: Self::now_unix_secs(),
                 });
                 return Ok(Some(node.clone()));
@@ -513,6 +741,7 @@ impl ProxyNodeWriteRepository for InMemoryProxyNodeRepository {
             node_id: mutation.node_id.clone(),
             event_type: event_type.to_string(),
             detail: Some(event_detail),
+            event_metadata: None,
             created_at_unix_ms: Some(event_time),
         });
         Ok(Some(node.clone()))
@@ -546,6 +775,14 @@ impl ProxyNodeWriteRepository for InMemoryProxyNodeRepository {
                 .write()
                 .expect("proxy node repository lock")
                 .retain(|event| event.node_id != node_id);
+            self.metrics_1m
+                .write()
+                .expect("proxy node repository lock")
+                .retain(|(metric_node_id, _), _| metric_node_id != node_id);
+            self.metrics_1h
+                .write()
+                .expect("proxy node repository lock")
+                .retain(|(metric_node_id, _), _| metric_node_id != node_id);
         }
         Ok(removed)
     }
@@ -598,6 +835,28 @@ impl ProxyNodeWriteRepository for InMemoryProxyNodeRepository {
         }
         Ok(())
     }
+
+    async fn cleanup_proxy_node_metrics(
+        &self,
+        retain_1m_from_unix_secs: u64,
+        retain_1h_from_unix_secs: u64,
+    ) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
+        let mut metrics_1m = self.metrics_1m.write().expect("proxy node repository lock");
+        let before_1m = metrics_1m.len();
+        metrics_1m.retain(|(_, bucket_start), _| *bucket_start >= retain_1m_from_unix_secs);
+        let deleted_1m_rows = before_1m.saturating_sub(metrics_1m.len());
+        drop(metrics_1m);
+
+        let mut metrics_1h = self.metrics_1h.write().expect("proxy node repository lock");
+        let before_1h = metrics_1h.len();
+        metrics_1h.retain(|(_, bucket_start), _| *bucket_start >= retain_1h_from_unix_secs);
+        let deleted_1h_rows = before_1h.saturating_sub(metrics_1h.len());
+
+        Ok(ProxyNodeMetricsCleanupSummary {
+            deleted_1m_rows,
+            deleted_1h_rows,
+        })
+    }
 }
 
 #[cfg(test)]
@@ -750,6 +1009,7 @@ mod tests {
                     node_id: "node-1".to_string(),
                     event_type: "connected".to_string(),
                     detail: Some("older".to_string()),
+                    event_metadata: None,
                     created_at_unix_ms: Some(1_710_000_000),
                 },
                 StoredProxyNodeEvent {
@@ -757,6 +1017,7 @@ mod tests {
                     node_id: "node-1".to_string(),
                     event_type: "disconnected".to_string(),
                     detail: Some("newer".to_string()),
+                    event_metadata: None,
                     created_at_unix_ms: Some(1_710_000_100),
                 },
             ],

crates/aether-data/src/repository/proxy_nodes/mod.rs

@@ -9,11 +9,14 @@ pub use mysql::MysqlProxyNodeReadRepository;
 pub use postgres::SqlxProxyNodeRepository;
 pub use sqlite::SqliteProxyNodeReadRepository;
 pub use types::{
+    bucket_start_unix_secs, build_tunnel_error_event_detail, build_tunnel_metrics_sample,
     normalize_proxy_node_scheduling_state, proxy_node_accepts_new_tunnels, proxy_reported_version,
     reconcile_remote_config_after_heartbeat, remote_config_scheduling_state,
-    remote_config_upgrade_target, ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation,
-    ProxyNodeManualUpdateMutation, ProxyNodeReadRepository, ProxyNodeRegistrationMutation,
+    remote_config_upgrade_target, ProxyNodeEventQuery, ProxyNodeHeartbeatMutation,
+    ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation, ProxyNodeMetricsCleanupSummary,
+    ProxyNodeMetricsStep, ProxyNodeReadRepository, ProxyNodeRegistrationMutation,
     ProxyNodeRemoteConfigMutation, ProxyNodeTrafficMutation, ProxyNodeTunnelStatusMutation,
-    ProxyNodeWriteRepository, StoredProxyNode, StoredProxyNodeEvent,
+    ProxyNodeWriteRepository, StoredProxyFleetMetricsBucket, StoredProxyNode, StoredProxyNodeEvent,
+    StoredProxyNodeMetricsBucket, TunnelErrorEventRecord, PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
     PROXY_NODE_SCHEDULING_STATE_CORDONED, PROXY_NODE_SCHEDULING_STATE_DRAINING,
 };

crates/aether-data/src/repository/proxy_nodes/mysql.rs

@@ -2,10 +2,14 @@ use async_trait::async_trait;
 use sqlx::{mysql::MySqlRow, Row};
 
 use super::types::{
-    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeHeartbeatMutation,
-    ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation, ProxyNodeReadRepository,
+    bucket_start_unix_secs, build_tunnel_error_event_detail, build_tunnel_metrics_sample,
+    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeEventQuery,
+    ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation,
+    ProxyNodeMetricsCleanupSummary, ProxyNodeMetricsStep, ProxyNodeReadRepository,
     ProxyNodeRegistrationMutation, ProxyNodeRemoteConfigMutation, ProxyNodeTrafficMutation,
-    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyNode, StoredProxyNodeEvent,
+    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyFleetMetricsBucket,
+    StoredProxyNode, StoredProxyNodeEvent, StoredProxyNodeMetricsBucket,
+    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
 };
 use crate::driver::mysql::MysqlPool;
 use crate::error::SqlResultExt;
@@ -148,17 +152,22 @@ ON DUPLICATE KEY UPDATE
         node_id: &str,
         event_type: &str,
         detail: Option<&str>,
+        event_metadata: Option<&serde_json::Value>,
         created_at_unix_secs: Option<u64>,
     ) -> Result<(), DataLayerError> {
         sqlx::query(
             r#"
-INSERT INTO proxy_node_events (node_id, event_type, detail, created_at)
-VALUES (?, ?, ?, ?)
+INSERT INTO proxy_node_events (node_id, event_type, detail, event_metadata, created_at)
+VALUES (?, ?, ?, ?, ?)
 "#,
         )
         .bind(node_id)
         .bind(event_type)
         .bind(detail)
+        .bind(optional_json_to_string(
+            &event_metadata.cloned(),
+            "proxy_node_events.event_metadata",
+        )?)
         .bind(created_at_unix_secs.unwrap_or_else(current_unix_secs) as i64)
         .execute(&self.pool)
         .await
@@ -166,6 +175,70 @@ VALUES (?, ?, ?, ?)
         Ok(())
     }
 
+    async fn upsert_metrics_bucket(
+        &self,
+        table: &str,
+        node_id: &str,
+        bucket_start: u64,
+        sample: &super::types::TunnelMetricsSample,
+    ) -> Result<(), DataLayerError> {
+        sqlx::query(&format!(
+            r#"
+INSERT INTO {table} (
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON DUPLICATE KEY UPDATE
+  samples = samples + VALUES(samples),
+  uptime_samples = uptime_samples + VALUES(uptime_samples),
+  active_connections_sum = active_connections_sum + VALUES(active_connections_sum),
+  active_connections_max = GREATEST(active_connections_max, VALUES(active_connections_max)),
+  heartbeat_rtt_ms_sum = heartbeat_rtt_ms_sum + VALUES(heartbeat_rtt_ms_sum),
+  heartbeat_rtt_ms_max = GREATEST(heartbeat_rtt_ms_max, VALUES(heartbeat_rtt_ms_max)),
+  connect_errors_delta = connect_errors_delta + VALUES(connect_errors_delta),
+  disconnects_delta = disconnects_delta + VALUES(disconnects_delta),
+  error_events_delta = error_events_delta + VALUES(error_events_delta),
+  ws_in_bytes_delta = ws_in_bytes_delta + VALUES(ws_in_bytes_delta),
+  ws_out_bytes_delta = ws_out_bytes_delta + VALUES(ws_out_bytes_delta),
+  ws_in_frames_delta = ws_in_frames_delta + VALUES(ws_in_frames_delta),
+  ws_out_frames_delta = ws_out_frames_delta + VALUES(ws_out_frames_delta)
+"#
+        ))
+        .bind(node_id)
+        .bind(i64::try_from(bucket_start).unwrap_or(i64::MAX))
+        .bind(sample.samples)
+        .bind(sample.uptime_samples)
+        .bind(sample.active_connections_sum)
+        .bind(sample.active_connections_max)
+        .bind(sample.heartbeat_rtt_ms_sum)
+        .bind(sample.heartbeat_rtt_ms_max)
+        .bind(sample.connect_errors_delta)
+        .bind(sample.disconnects_delta)
+        .bind(sample.error_events_delta)
+        .bind(sample.ws_in_bytes_delta)
+        .bind(sample.ws_out_bytes_delta)
+        .bind(sample.ws_in_frames_delta)
+        .bind(sample.ws_out_frames_delta)
+        .execute(&self.pool)
+        .await
+        .map_sql_err()?;
+        Ok(())
+    }
+
     fn normalize_remote_config(
         mutation: &ProxyNodeRemoteConfigMutation,
         existing: Option<&serde_json::Value>,
@@ -298,6 +371,7 @@ SELECT
   node_id,
   event_type,
   detail,
+  event_metadata,
   created_at AS created_at_unix_ms
 FROM proxy_node_events
 WHERE node_id = ?
@@ -312,6 +386,152 @@ LIMIT ?
         .map_sql_err()?;
         rows.iter().map(map_proxy_node_event_row).collect()
     }
+
+    async fn list_proxy_node_events_filtered(
+        &self,
+        node_id: &str,
+        query: &ProxyNodeEventQuery,
+    ) -> Result<Vec<StoredProxyNodeEvent>, DataLayerError> {
+        let rows = sqlx::query(
+            r#"
+SELECT
+  id,
+  node_id,
+  event_type,
+  detail,
+  event_metadata,
+  created_at AS created_at_unix_ms
+FROM proxy_node_events
+WHERE node_id = ?
+  AND (? IS NULL OR created_at >= ?)
+  AND (? IS NULL OR created_at <= ?)
+  AND (? IS NULL OR LOWER(event_type) = LOWER(?))
+ORDER BY created_at DESC, id DESC
+LIMIT ?
+"#,
+        )
+        .bind(node_id)
+        .bind(
+            query
+                .from_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(
+            query
+                .from_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(
+            query
+                .to_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(
+            query
+                .to_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(query.event_type.as_deref())
+        .bind(query.event_type.as_deref())
+        .bind(i64::try_from(query.limit).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await
+        .map_sql_err()?;
+        rows.iter().map(map_proxy_node_event_row).collect()
+    }
+
+    async fn list_proxy_node_metrics(
+        &self,
+        node_id: &str,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyNodeMetricsBucket>, DataLayerError> {
+        let table = match step {
+            ProxyNodeMetricsStep::OneMinute => "proxy_node_metrics_1m",
+            ProxyNodeMetricsStep::OneHour => "proxy_node_metrics_1h",
+        };
+        let rows = sqlx::query(&format!(
+            r#"
+SELECT
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+FROM {table}
+WHERE node_id = ?
+  AND bucket_start_unix_secs >= ?
+  AND bucket_start_unix_secs <= ?
+ORDER BY bucket_start_unix_secs ASC
+LIMIT ?
+"#
+        ))
+        .bind(node_id)
+        .bind(i64::try_from(from_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(to_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(limit).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await
+        .map_sql_err()?;
+        rows.iter().map(map_proxy_node_metric_row).collect()
+    }
+
+    async fn list_proxy_fleet_metrics(
+        &self,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyFleetMetricsBucket>, DataLayerError> {
+        let table = match step {
+            ProxyNodeMetricsStep::OneMinute => "proxy_node_metrics_1m",
+            ProxyNodeMetricsStep::OneHour => "proxy_node_metrics_1h",
+        };
+        let rows = sqlx::query(&format!(
+            r#"
+SELECT
+  bucket_start_unix_secs,
+  SUM(samples) AS samples,
+  SUM(uptime_samples) AS uptime_samples,
+  SUM(active_connections_sum) AS active_connections_sum,
+  MAX(active_connections_max) AS active_connections_max,
+  SUM(heartbeat_rtt_ms_sum) AS heartbeat_rtt_ms_sum,
+  MAX(heartbeat_rtt_ms_max) AS heartbeat_rtt_ms_max,
+  SUM(connect_errors_delta) AS connect_errors_delta,
+  SUM(disconnects_delta) AS disconnects_delta,
+  SUM(error_events_delta) AS error_events_delta,
+  SUM(ws_in_bytes_delta) AS ws_in_bytes_delta,
+  SUM(ws_out_bytes_delta) AS ws_out_bytes_delta,
+  SUM(ws_in_frames_delta) AS ws_in_frames_delta,
+  SUM(ws_out_frames_delta) AS ws_out_frames_delta
+FROM {table}
+WHERE bucket_start_unix_secs >= ?
+  AND bucket_start_unix_secs <= ?
+GROUP BY bucket_start_unix_secs
+ORDER BY bucket_start_unix_secs ASC
+LIMIT ?
+"#
+        ))
+        .bind(i64::try_from(from_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(to_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(limit).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await
+        .map_sql_err()?;
+        rows.iter().map(map_proxy_fleet_metric_row).collect()
+    }
 }
 
 #[async_trait]
@@ -540,7 +760,9 @@ WHERE is_manual = 0
             ));
         }
 
-        let now = Some(current_unix_secs());
+        let previous_proxy_metadata = node.proxy_metadata.clone();
+        let now_unix_secs = current_unix_secs();
+        let now = Some(now_unix_secs);
         node.last_heartbeat_at_unix_secs = now;
         if node.status != "online" || !node.tunnel_connected {
             node.status = "online".to_string();
@@ -584,7 +806,52 @@ WHERE is_manual = 0
             node.config_version = node.config_version.saturating_add(1);
             node.updated_at_unix_secs = now;
         }
+        let tunnel_metrics_sample = build_tunnel_metrics_sample(
+            previous_proxy_metadata.as_ref(),
+            node.proxy_metadata.as_ref(),
+            node.active_connections,
+            node.tunnel_connected,
+        );
         self.upsert_node(&node).await?;
+
+        if let Some(sample) = tunnel_metrics_sample.as_ref() {
+            self.upsert_metrics_bucket(
+                "proxy_node_metrics_1m",
+                &node.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneMinute),
+                sample,
+            )
+            .await?;
+            self.upsert_metrics_bucket(
+                "proxy_node_metrics_1h",
+                &node.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneHour),
+                sample,
+            )
+            .await?;
+
+            for error in &sample.recent_error_events {
+                let detail = build_tunnel_error_event_detail(error);
+                let event_metadata = serde_json::json!({
+                    "source": "heartbeat",
+                    "category": error.category,
+                    "message": error.message,
+                    "timestamp_unix_secs": error.timestamp_unix_secs,
+                });
+                self.insert_event(
+                    &node.id,
+                    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
+                    Some(detail.as_str()),
+                    Some(&event_metadata),
+                    Some(if error.timestamp_unix_secs == 0 {
+                        now_unix_secs
+                    } else {
+                        error.timestamp_unix_secs
+                    }),
+                )
+                .await?;
+            }
+        }
         Ok(Some(node))
     }
 
@@ -638,6 +905,7 @@ WHERE is_manual = 0
                 &mutation.node_id,
                 event_type,
                 Some(&format!("[stale_ignored] {event_detail}")),
+                None,
                 Some(current_unix_secs()),
             )
             .await?;
@@ -660,6 +928,7 @@ WHERE is_manual = 0
             &mutation.node_id,
             event_type,
             Some(&event_detail),
+            None,
             Some(event_time),
         )
         .await?;
@@ -691,6 +960,16 @@ WHERE is_manual = 0
                 .execute(&self.pool)
                 .await
                 .map_sql_err()?;
+            sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE node_id = ?")
+                .bind(node_id)
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?;
+            sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE node_id = ?")
+                .bind(node_id)
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?;
             sqlx::query("DELETE FROM proxy_nodes WHERE id = ?")
                 .bind(node_id)
                 .execute(&self.pool)
@@ -747,6 +1026,33 @@ WHERE is_manual = 0
         node.updated_at_unix_secs = Some(current_unix_secs());
         self.upsert_node(&node).await
     }
+
+    async fn cleanup_proxy_node_metrics(
+        &self,
+        retain_1m_from_unix_secs: u64,
+        retain_1h_from_unix_secs: u64,
+    ) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
+        let deleted_1m =
+            sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE bucket_start_unix_secs < ?")
+                .bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?
+                .rows_affected() as usize;
+
+        let deleted_1h =
+            sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE bucket_start_unix_secs < ?")
+                .bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?
+                .rows_affected() as usize;
+
+        Ok(ProxyNodeMetricsCleanupSummary {
+            deleted_1m_rows: deleted_1m,
+            deleted_1h_rows: deleted_1h,
+        })
+    }
 }
 
 fn optional_unix_secs(value: Option<i64>) -> Option<u64> {
@@ -861,10 +1167,63 @@ fn map_proxy_node_event_row(row: &MySqlRow) -> Result<StoredProxyNodeEvent, Data
         node_id: row.try_get("node_id").map_sql_err()?,
         event_type: row.try_get("event_type").map_sql_err()?,
         detail: row.try_get("detail").map_sql_err()?,
+        event_metadata: optional_json_from_string(
+            row.try_get("event_metadata").map_sql_err()?,
+            "proxy_node_events.event_metadata",
+        )?,
         created_at_unix_ms: optional_unix_secs(row.try_get("created_at_unix_ms").map_sql_err()?),
     })
 }
 
+fn map_proxy_node_metric_row(
+    row: &MySqlRow,
+) -> Result<StoredProxyNodeMetricsBucket, DataLayerError> {
+    Ok(StoredProxyNodeMetricsBucket {
+        node_id: row.try_get("node_id").map_sql_err()?,
+        bucket_start_unix_secs: optional_unix_secs(
+            row.try_get("bucket_start_unix_secs").map_sql_err()?,
+        )
+        .unwrap_or_default(),
+        samples: row.try_get("samples").map_sql_err()?,
+        uptime_samples: row.try_get("uptime_samples").map_sql_err()?,
+        active_connections_sum: row.try_get("active_connections_sum").map_sql_err()?,
+        active_connections_max: row.try_get("active_connections_max").map_sql_err()?,
+        heartbeat_rtt_ms_sum: row.try_get("heartbeat_rtt_ms_sum").map_sql_err()?,
+        heartbeat_rtt_ms_max: row.try_get("heartbeat_rtt_ms_max").map_sql_err()?,
+        connect_errors_delta: row.try_get("connect_errors_delta").map_sql_err()?,
+        disconnects_delta: row.try_get("disconnects_delta").map_sql_err()?,
+        error_events_delta: row.try_get("error_events_delta").map_sql_err()?,
+        ws_in_bytes_delta: row.try_get("ws_in_bytes_delta").map_sql_err()?,
+        ws_out_bytes_delta: row.try_get("ws_out_bytes_delta").map_sql_err()?,
+        ws_in_frames_delta: row.try_get("ws_in_frames_delta").map_sql_err()?,
+        ws_out_frames_delta: row.try_get("ws_out_frames_delta").map_sql_err()?,
+    })
+}
+
+fn map_proxy_fleet_metric_row(
+    row: &MySqlRow,
+) -> Result<StoredProxyFleetMetricsBucket, DataLayerError> {
+    Ok(StoredProxyFleetMetricsBucket {
+        bucket_start_unix_secs: optional_unix_secs(
+            row.try_get("bucket_start_unix_secs").map_sql_err()?,
+        )
+        .unwrap_or_default(),
+        samples: row.try_get("samples").map_sql_err()?,
+        uptime_samples: row.try_get("uptime_samples").map_sql_err()?,
+        active_connections_sum: row.try_get("active_connections_sum").map_sql_err()?,
+        active_connections_max: row.try_get("active_connections_max").map_sql_err()?,
+        heartbeat_rtt_ms_sum: row.try_get("heartbeat_rtt_ms_sum").map_sql_err()?,
+        heartbeat_rtt_ms_max: row.try_get("heartbeat_rtt_ms_max").map_sql_err()?,
+        connect_errors_delta: row.try_get("connect_errors_delta").map_sql_err()?,
+        disconnects_delta: row.try_get("disconnects_delta").map_sql_err()?,
+        error_events_delta: row.try_get("error_events_delta").map_sql_err()?,
+        ws_in_bytes_delta: row.try_get("ws_in_bytes_delta").map_sql_err()?,
+        ws_out_bytes_delta: row.try_get("ws_out_bytes_delta").map_sql_err()?,
+        ws_in_frames_delta: row.try_get("ws_in_frames_delta").map_sql_err()?,
+        ws_out_frames_delta: row.try_get("ws_out_frames_delta").map_sql_err()?,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::MysqlProxyNodeReadRepository;

crates/aether-data/src/repository/proxy_nodes/postgres.rs

@@ -4,10 +4,14 @@ use sha2::{Digest, Sha256};
 use sqlx::{postgres::PgRow, PgPool, Row};
 
 use super::types::{
-    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeHeartbeatMutation,
-    ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation, ProxyNodeReadRepository,
+    bucket_start_unix_secs, build_tunnel_error_event_detail, build_tunnel_metrics_sample,
+    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeEventQuery,
+    ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation,
+    ProxyNodeMetricsCleanupSummary, ProxyNodeMetricsStep, ProxyNodeReadRepository,
     ProxyNodeRegistrationMutation, ProxyNodeRemoteConfigMutation, ProxyNodeTrafficMutation,
-    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyNode, StoredProxyNodeEvent,
+    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyFleetMetricsBucket,
+    StoredProxyNode, StoredProxyNodeEvent, StoredProxyNodeMetricsBucket, TunnelMetricsSample,
+    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
 };
 use crate::{
     error::{postgres_error, SqlxResultExt},
@@ -91,6 +95,7 @@ SELECT
   node_id,
   CAST(event_type AS TEXT) AS event_type,
   detail,
+  event_metadata,
   EXTRACT(EPOCH FROM created_at)::bigint AS created_at_unix_ms
 FROM proxy_node_events
 WHERE node_id = $1
@@ -98,6 +103,23 @@ ORDER BY created_at DESC, id DESC
 LIMIT $2
 "#;
 
+const LIST_PROXY_NODE_EVENTS_FILTERED_SQL: &str = r#"
+SELECT
+  id,
+  node_id,
+  CAST(event_type AS TEXT) AS event_type,
+  detail,
+  event_metadata,
+  EXTRACT(EPOCH FROM created_at)::bigint AS created_at_unix_ms
+FROM proxy_node_events
+WHERE node_id = $1
+  AND ($2::double precision IS NULL OR created_at >= TO_TIMESTAMP($2::double precision))
+  AND ($3::double precision IS NULL OR created_at <= TO_TIMESTAMP($3::double precision))
+  AND ($4::text IS NULL OR LOWER(CAST(event_type AS TEXT)) = LOWER($4::text))
+ORDER BY created_at DESC, id DESC
+LIMIT $5
+"#;
+
 const APPLY_HEARTBEAT_SQL: &str = r#"
 UPDATE proxy_nodes
 SET
@@ -381,6 +403,188 @@ WHERE id = $4
   AND is_manual = TRUE
 "#;
 
+const INSERT_PROXY_NODE_EVENT_SQL: &str = r#"
+INSERT INTO proxy_node_events (node_id, event_type, detail, event_metadata, created_at)
+VALUES (
+  $1,
+  $2,
+  $3,
+  $4::json,
+  CASE
+    WHEN $5::double precision IS NULL THEN NOW()
+    ELSE TO_TIMESTAMP($5::double precision)
+  END
+)
+"#;
+
+const UPSERT_PROXY_NODE_METRICS_1M_SQL: &str = r#"
+INSERT INTO proxy_node_metrics_1m (
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
+ON CONFLICT (node_id, bucket_start_unix_secs) DO UPDATE SET
+  samples = proxy_node_metrics_1m.samples + EXCLUDED.samples,
+  uptime_samples = proxy_node_metrics_1m.uptime_samples + EXCLUDED.uptime_samples,
+  active_connections_sum = proxy_node_metrics_1m.active_connections_sum + EXCLUDED.active_connections_sum,
+  active_connections_max = GREATEST(proxy_node_metrics_1m.active_connections_max, EXCLUDED.active_connections_max),
+  heartbeat_rtt_ms_sum = proxy_node_metrics_1m.heartbeat_rtt_ms_sum + EXCLUDED.heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max = GREATEST(proxy_node_metrics_1m.heartbeat_rtt_ms_max, EXCLUDED.heartbeat_rtt_ms_max),
+  connect_errors_delta = proxy_node_metrics_1m.connect_errors_delta + EXCLUDED.connect_errors_delta,
+  disconnects_delta = proxy_node_metrics_1m.disconnects_delta + EXCLUDED.disconnects_delta,
+  error_events_delta = proxy_node_metrics_1m.error_events_delta + EXCLUDED.error_events_delta,
+  ws_in_bytes_delta = proxy_node_metrics_1m.ws_in_bytes_delta + EXCLUDED.ws_in_bytes_delta,
+  ws_out_bytes_delta = proxy_node_metrics_1m.ws_out_bytes_delta + EXCLUDED.ws_out_bytes_delta,
+  ws_in_frames_delta = proxy_node_metrics_1m.ws_in_frames_delta + EXCLUDED.ws_in_frames_delta,
+  ws_out_frames_delta = proxy_node_metrics_1m.ws_out_frames_delta + EXCLUDED.ws_out_frames_delta
+"#;
+
+const UPSERT_PROXY_NODE_METRICS_1H_SQL: &str = r#"
+INSERT INTO proxy_node_metrics_1h (
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
+ON CONFLICT (node_id, bucket_start_unix_secs) DO UPDATE SET
+  samples = proxy_node_metrics_1h.samples + EXCLUDED.samples,
+  uptime_samples = proxy_node_metrics_1h.uptime_samples + EXCLUDED.uptime_samples,
+  active_connections_sum = proxy_node_metrics_1h.active_connections_sum + EXCLUDED.active_connections_sum,
+  active_connections_max = GREATEST(proxy_node_metrics_1h.active_connections_max, EXCLUDED.active_connections_max),
+  heartbeat_rtt_ms_sum = proxy_node_metrics_1h.heartbeat_rtt_ms_sum + EXCLUDED.heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max = GREATEST(proxy_node_metrics_1h.heartbeat_rtt_ms_max, EXCLUDED.heartbeat_rtt_ms_max),
+  connect_errors_delta = proxy_node_metrics_1h.connect_errors_delta + EXCLUDED.connect_errors_delta,
+  disconnects_delta = proxy_node_metrics_1h.disconnects_delta + EXCLUDED.disconnects_delta,
+  error_events_delta = proxy_node_metrics_1h.error_events_delta + EXCLUDED.error_events_delta,
+  ws_in_bytes_delta = proxy_node_metrics_1h.ws_in_bytes_delta + EXCLUDED.ws_in_bytes_delta,
+  ws_out_bytes_delta = proxy_node_metrics_1h.ws_out_bytes_delta + EXCLUDED.ws_out_bytes_delta,
+  ws_in_frames_delta = proxy_node_metrics_1h.ws_in_frames_delta + EXCLUDED.ws_in_frames_delta,
+  ws_out_frames_delta = proxy_node_metrics_1h.ws_out_frames_delta + EXCLUDED.ws_out_frames_delta
+"#;
+
+const LIST_PROXY_NODE_METRICS_1M_SQL: &str = r#"
+SELECT
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+FROM proxy_node_metrics_1m
+WHERE node_id = $1
+  AND bucket_start_unix_secs >= $2
+  AND bucket_start_unix_secs <= $3
+ORDER BY bucket_start_unix_secs ASC
+LIMIT $4
+"#;
+
+const LIST_PROXY_NODE_METRICS_1H_SQL: &str = r#"
+SELECT
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+FROM proxy_node_metrics_1h
+WHERE node_id = $1
+  AND bucket_start_unix_secs >= $2
+  AND bucket_start_unix_secs <= $3
+ORDER BY bucket_start_unix_secs ASC
+LIMIT $4
+"#;
+
+const LIST_PROXY_FLEET_METRICS_1M_SQL: &str = r#"
+SELECT
+  bucket_start_unix_secs,
+  SUM(samples) AS samples,
+  SUM(uptime_samples) AS uptime_samples,
+  SUM(active_connections_sum) AS active_connections_sum,
+  MAX(active_connections_max) AS active_connections_max,
+  SUM(heartbeat_rtt_ms_sum) AS heartbeat_rtt_ms_sum,
+  MAX(heartbeat_rtt_ms_max) AS heartbeat_rtt_ms_max,
+  SUM(connect_errors_delta) AS connect_errors_delta,
+  SUM(disconnects_delta) AS disconnects_delta,
+  SUM(error_events_delta) AS error_events_delta,
+  SUM(ws_in_bytes_delta) AS ws_in_bytes_delta,
+  SUM(ws_out_bytes_delta) AS ws_out_bytes_delta,
+  SUM(ws_in_frames_delta) AS ws_in_frames_delta,
+  SUM(ws_out_frames_delta) AS ws_out_frames_delta
+FROM proxy_node_metrics_1m
+WHERE bucket_start_unix_secs >= $1
+  AND bucket_start_unix_secs <= $2
+GROUP BY bucket_start_unix_secs
+ORDER BY bucket_start_unix_secs ASC
+LIMIT $3
+"#;
+
+const LIST_PROXY_FLEET_METRICS_1H_SQL: &str = r#"
+SELECT
+  bucket_start_unix_secs,
+  SUM(samples) AS samples,
+  SUM(uptime_samples) AS uptime_samples,
+  SUM(active_connections_sum) AS active_connections_sum,
+  MAX(active_connections_max) AS active_connections_max,
+  SUM(heartbeat_rtt_ms_sum) AS heartbeat_rtt_ms_sum,
+  MAX(heartbeat_rtt_ms_max) AS heartbeat_rtt_ms_max,
+  SUM(connect_errors_delta) AS connect_errors_delta,
+  SUM(disconnects_delta) AS disconnects_delta,
+  SUM(error_events_delta) AS error_events_delta,
+  SUM(ws_in_bytes_delta) AS ws_in_bytes_delta,
+  SUM(ws_out_bytes_delta) AS ws_out_bytes_delta,
+  SUM(ws_in_frames_delta) AS ws_in_frames_delta,
+  SUM(ws_out_frames_delta) AS ws_out_frames_delta
+FROM proxy_node_metrics_1h
+WHERE bucket_start_unix_secs >= $1
+  AND bucket_start_unix_secs <= $2
+GROUP BY bucket_start_unix_secs
+ORDER BY bucket_start_unix_secs ASC
+LIMIT $3
+"#;
+
 #[derive(Debug, Clone)]
 pub struct SqlxProxyNodeRepository {
     pool: PgPool,
@@ -446,12 +650,111 @@ impl SqlxProxyNodeRepository {
             node_id: row.try_get("node_id").map_postgres_err()?,
             event_type: row.try_get("event_type").map_postgres_err()?,
             detail: row.try_get("detail").map_postgres_err()?,
+            event_metadata: row.try_get("event_metadata").map_postgres_err()?,
             created_at_unix_ms: Self::optional_unix_secs(
                 row.try_get("created_at_unix_ms").map_postgres_err()?,
             ),
         })
     }
 
+    fn row_to_node_metric(row: &PgRow) -> Result<StoredProxyNodeMetricsBucket, DataLayerError> {
+        Ok(StoredProxyNodeMetricsBucket {
+            node_id: row.try_get("node_id").map_postgres_err()?,
+            bucket_start_unix_secs: Self::optional_unix_secs(
+                row.try_get("bucket_start_unix_secs").map_postgres_err()?,
+            )
+            .unwrap_or_default(),
+            samples: row.try_get("samples").map_postgres_err()?,
+            uptime_samples: row.try_get("uptime_samples").map_postgres_err()?,
+            active_connections_sum: row.try_get("active_connections_sum").map_postgres_err()?,
+            active_connections_max: row.try_get("active_connections_max").map_postgres_err()?,
+            heartbeat_rtt_ms_sum: row.try_get("heartbeat_rtt_ms_sum").map_postgres_err()?,
+            heartbeat_rtt_ms_max: row.try_get("heartbeat_rtt_ms_max").map_postgres_err()?,
+            connect_errors_delta: row.try_get("connect_errors_delta").map_postgres_err()?,
+            disconnects_delta: row.try_get("disconnects_delta").map_postgres_err()?,
+            error_events_delta: row.try_get("error_events_delta").map_postgres_err()?,
+            ws_in_bytes_delta: row.try_get("ws_in_bytes_delta").map_postgres_err()?,
+            ws_out_bytes_delta: row.try_get("ws_out_bytes_delta").map_postgres_err()?,
+            ws_in_frames_delta: row.try_get("ws_in_frames_delta").map_postgres_err()?,
+            ws_out_frames_delta: row.try_get("ws_out_frames_delta").map_postgres_err()?,
+        })
+    }
+
+    fn row_to_fleet_metric(row: &PgRow) -> Result<StoredProxyFleetMetricsBucket, DataLayerError> {
+        Ok(StoredProxyFleetMetricsBucket {
+            bucket_start_unix_secs: Self::optional_unix_secs(
+                row.try_get("bucket_start_unix_secs").map_postgres_err()?,
+            )
+            .unwrap_or_default(),
+            samples: row.try_get("samples").map_postgres_err()?,
+            uptime_samples: row.try_get("uptime_samples").map_postgres_err()?,
+            active_connections_sum: row.try_get("active_connections_sum").map_postgres_err()?,
+            active_connections_max: row.try_get("active_connections_max").map_postgres_err()?,
+            heartbeat_rtt_ms_sum: row.try_get("heartbeat_rtt_ms_sum").map_postgres_err()?,
+            heartbeat_rtt_ms_max: row.try_get("heartbeat_rtt_ms_max").map_postgres_err()?,
+            connect_errors_delta: row.try_get("connect_errors_delta").map_postgres_err()?,
+            disconnects_delta: row.try_get("disconnects_delta").map_postgres_err()?,
+            error_events_delta: row.try_get("error_events_delta").map_postgres_err()?,
+            ws_in_bytes_delta: row.try_get("ws_in_bytes_delta").map_postgres_err()?,
+            ws_out_bytes_delta: row.try_get("ws_out_bytes_delta").map_postgres_err()?,
+            ws_in_frames_delta: row.try_get("ws_in_frames_delta").map_postgres_err()?,
+            ws_out_frames_delta: row.try_get("ws_out_frames_delta").map_postgres_err()?,
+        })
+    }
+
+    async fn insert_event(
+        &self,
+        node_id: &str,
+        event_type: &str,
+        detail: Option<&str>,
+        event_metadata: Option<&serde_json::Value>,
+        created_at_unix_secs: Option<u64>,
+    ) -> Result<(), DataLayerError> {
+        sqlx::query(INSERT_PROXY_NODE_EVENT_SQL)
+            .bind(node_id)
+            .bind(event_type)
+            .bind(detail)
+            .bind(event_metadata)
+            .bind(created_at_unix_secs.map(|value| value as f64))
+            .execute(&self.pool)
+            .await
+            .map_postgres_err()?;
+        Ok(())
+    }
+
+    async fn upsert_metrics_bucket(
+        &self,
+        step: ProxyNodeMetricsStep,
+        node_id: &str,
+        bucket_start: u64,
+        sample: &TunnelMetricsSample,
+    ) -> Result<(), DataLayerError> {
+        let sql = match step {
+            ProxyNodeMetricsStep::OneMinute => UPSERT_PROXY_NODE_METRICS_1M_SQL,
+            ProxyNodeMetricsStep::OneHour => UPSERT_PROXY_NODE_METRICS_1H_SQL,
+        };
+        sqlx::query(sql)
+            .bind(node_id)
+            .bind(i64::try_from(bucket_start).unwrap_or(i64::MAX))
+            .bind(sample.samples)
+            .bind(sample.uptime_samples)
+            .bind(sample.active_connections_sum)
+            .bind(sample.active_connections_max)
+            .bind(sample.heartbeat_rtt_ms_sum)
+            .bind(sample.heartbeat_rtt_ms_max)
+            .bind(sample.connect_errors_delta)
+            .bind(sample.disconnects_delta)
+            .bind(sample.error_events_delta)
+            .bind(sample.ws_in_bytes_delta)
+            .bind(sample.ws_out_bytes_delta)
+            .bind(sample.ws_in_frames_delta)
+            .bind(sample.ws_out_frames_delta)
+            .execute(&self.pool)
+            .await
+            .map_postgres_err()?;
+        Ok(())
+    }
+
     fn registration_lock_key(ip: &str, port: i32) -> i64 {
         let mut hasher = Sha256::new();
         hasher.update(ip.as_bytes());
@@ -602,6 +905,73 @@ impl ProxyNodeReadRepository for SqlxProxyNodeRepository {
         }
         Ok(items)
     }
+
+    async fn list_proxy_node_events_filtered(
+        &self,
+        node_id: &str,
+        query: &ProxyNodeEventQuery,
+    ) -> Result<Vec<StoredProxyNodeEvent>, DataLayerError> {
+        let mut rows = sqlx::query(LIST_PROXY_NODE_EVENTS_FILTERED_SQL)
+            .bind(node_id)
+            .bind(query.from_unix_secs.map(|value| value as f64))
+            .bind(query.to_unix_secs.map(|value| value as f64))
+            .bind(query.event_type.as_deref())
+            .bind(i64::try_from(query.limit).unwrap_or(i64::MAX))
+            .fetch(&self.pool);
+        let mut items = Vec::new();
+        while let Some(row) = rows.try_next().await.map_postgres_err()? {
+            items.push(Self::row_to_event(&row)?);
+        }
+        Ok(items)
+    }
+
+    async fn list_proxy_node_metrics(
+        &self,
+        node_id: &str,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyNodeMetricsBucket>, DataLayerError> {
+        let sql = match step {
+            ProxyNodeMetricsStep::OneMinute => LIST_PROXY_NODE_METRICS_1M_SQL,
+            ProxyNodeMetricsStep::OneHour => LIST_PROXY_NODE_METRICS_1H_SQL,
+        };
+        let mut rows = sqlx::query(sql)
+            .bind(node_id)
+            .bind(i64::try_from(from_unix_secs).unwrap_or(i64::MAX))
+            .bind(i64::try_from(to_unix_secs).unwrap_or(i64::MAX))
+            .bind(i64::try_from(limit).unwrap_or(i64::MAX))
+            .fetch(&self.pool);
+        let mut items = Vec::new();
+        while let Some(row) = rows.try_next().await.map_postgres_err()? {
+            items.push(Self::row_to_node_metric(&row)?);
+        }
+        Ok(items)
+    }
+
+    async fn list_proxy_fleet_metrics(
+        &self,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyFleetMetricsBucket>, DataLayerError> {
+        let sql = match step {
+            ProxyNodeMetricsStep::OneMinute => LIST_PROXY_FLEET_METRICS_1M_SQL,
+            ProxyNodeMetricsStep::OneHour => LIST_PROXY_FLEET_METRICS_1H_SQL,
+        };
+        let mut rows = sqlx::query(sql)
+            .bind(i64::try_from(from_unix_secs).unwrap_or(i64::MAX))
+            .bind(i64::try_from(to_unix_secs).unwrap_or(i64::MAX))
+            .bind(i64::try_from(limit).unwrap_or(i64::MAX))
+            .fetch(&self.pool);
+        let mut items = Vec::new();
+        while let Some(row) = rows.try_next().await.map_postgres_err()? {
+            items.push(Self::row_to_fleet_metric(&row)?);
+        }
+        Ok(items)
+    }
 }
 
 #[async_trait]
@@ -822,6 +1192,54 @@ impl ProxyNodeWriteRepository for SqlxProxyNodeRepository {
         let Some(updated) = updated else {
             return Ok(None);
         };
+        let now_unix_secs = updated
+            .last_heartbeat_at_unix_secs
+            .unwrap_or_else(|| chrono::Utc::now().timestamp().max(0) as u64);
+        let tunnel_metrics_sample = build_tunnel_metrics_sample(
+            existing.proxy_metadata.as_ref(),
+            updated.proxy_metadata.as_ref(),
+            updated.active_connections,
+            updated.tunnel_connected,
+        );
+
+        if let Some(sample) = tunnel_metrics_sample.as_ref() {
+            self.upsert_metrics_bucket(
+                ProxyNodeMetricsStep::OneMinute,
+                &updated.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneMinute),
+                sample,
+            )
+            .await?;
+            self.upsert_metrics_bucket(
+                ProxyNodeMetricsStep::OneHour,
+                &updated.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneHour),
+                sample,
+            )
+            .await?;
+
+            for error in &sample.recent_error_events {
+                let detail = build_tunnel_error_event_detail(error);
+                let event_metadata = serde_json::json!({
+                    "source": "heartbeat",
+                    "category": error.category,
+                    "message": error.message,
+                    "timestamp_unix_secs": error.timestamp_unix_secs,
+                });
+                self.insert_event(
+                    &updated.id,
+                    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
+                    Some(detail.as_str()),
+                    Some(&event_metadata),
+                    Some(if error.timestamp_unix_secs == 0 {
+                        now_unix_secs
+                    } else {
+                        error.timestamp_unix_secs
+                    }),
+                )
+                .await?;
+            }
+        }
 
         if reconcile_remote_config_after_heartbeat(
             updated.remote_config.as_ref(),
@@ -890,23 +1308,15 @@ impl ProxyNodeWriteRepository for SqlxProxyNodeRepository {
             .zip(observed_at_unix_secs)
             .is_some_and(|(last_transition, observed_at)| observed_at < last_transition)
         {
-            sqlx::query(
-                r#"
-INSERT INTO proxy_node_events (node_id, event_type, detail, created_at)
-VALUES (
-  $1,
-  $2,
-  $3,
-  NOW()
-)
-"#,
-            )
-            .bind(&mutation.node_id)
-            .bind(event_type)
-            .bind(format!("[stale_ignored] {event_detail}"))
-            .execute(&mut *tx)
-            .await
-            .map_postgres_err()?;
+            sqlx::query(INSERT_PROXY_NODE_EVENT_SQL)
+                .bind(&mutation.node_id)
+                .bind(event_type)
+                .bind(format!("[stale_ignored] {event_detail}"))
+                .bind(None::<serde_json::Value>)
+                .bind(None::<f64>)
+                .execute(&mut *tx)
+                .await
+                .map_postgres_err()?;
             tx.commit().await.map_err(postgres_error)?;
             return self.find_proxy_node(&mutation.node_id).await;
         }
@@ -942,27 +1352,15 @@ WHERE id = $1
         .await
         .map_postgres_err()?;
 
-        sqlx::query(
-            r#"
-INSERT INTO proxy_node_events (node_id, event_type, detail, created_at)
-VALUES (
-  $1,
-  $2,
-  $3,
-  CASE
-    WHEN $4::double precision IS NULL THEN NOW()
-    ELSE TO_TIMESTAMP($4::double precision)
-  END
-)
-"#,
-        )
-        .bind(&mutation.node_id)
-        .bind(event_type)
-        .bind(event_detail)
-        .bind(observed_at_unix_secs.map(|value| value as f64))
-        .execute(&mut *tx)
-        .await
-        .map_postgres_err()?;
+        sqlx::query(INSERT_PROXY_NODE_EVENT_SQL)
+            .bind(&mutation.node_id)
+            .bind(event_type)
+            .bind(event_detail)
+            .bind(None::<serde_json::Value>)
+            .bind(observed_at_unix_secs.map(|value| value as f64))
+            .execute(&mut *tx)
+            .await
+            .map_postgres_err()?;
 
         tx.commit().await.map_err(postgres_error)?;
         self.find_proxy_node(&mutation.node_id).await
@@ -1045,6 +1443,33 @@ VALUES (
             .map_postgres_err()?;
         Ok(())
     }
+
+    async fn cleanup_proxy_node_metrics(
+        &self,
+        retain_1m_from_unix_secs: u64,
+        retain_1h_from_unix_secs: u64,
+    ) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
+        let deleted_1m =
+            sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE bucket_start_unix_secs < $1")
+                .bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
+                .execute(&self.pool)
+                .await
+                .map_postgres_err()?
+                .rows_affected() as usize;
+
+        let deleted_1h =
+            sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE bucket_start_unix_secs < $1")
+                .bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
+                .execute(&self.pool)
+                .await
+                .map_postgres_err()?
+                .rows_affected() as usize;
+
+        Ok(ProxyNodeMetricsCleanupSummary {
+            deleted_1m_rows: deleted_1m,
+            deleted_1h_rows: deleted_1h,
+        })
+    }
 }
 
 #[cfg(test)]

crates/aether-data/src/repository/proxy_nodes/sqlite.rs

@@ -2,10 +2,14 @@ use async_trait::async_trait;
 use sqlx::{sqlite::SqliteRow, Row};
 
 use super::types::{
-    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeHeartbeatMutation,
-    ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation, ProxyNodeReadRepository,
+    bucket_start_unix_secs, build_tunnel_error_event_detail, build_tunnel_metrics_sample,
+    normalize_proxy_metadata, reconcile_remote_config_after_heartbeat, ProxyNodeEventQuery,
+    ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation,
+    ProxyNodeMetricsCleanupSummary, ProxyNodeMetricsStep, ProxyNodeReadRepository,
     ProxyNodeRegistrationMutation, ProxyNodeRemoteConfigMutation, ProxyNodeTrafficMutation,
-    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyNode, StoredProxyNodeEvent,
+    ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository, StoredProxyFleetMetricsBucket,
+    StoredProxyNode, StoredProxyNodeEvent, StoredProxyNodeMetricsBucket,
+    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
 };
 use crate::driver::sqlite::SqlitePool;
 use crate::error::SqlResultExt;
@@ -148,17 +152,22 @@ ON CONFLICT(id) DO UPDATE SET
         node_id: &str,
         event_type: &str,
         detail: Option<&str>,
+        event_metadata: Option<&serde_json::Value>,
         created_at_unix_secs: Option<u64>,
     ) -> Result<(), DataLayerError> {
         sqlx::query(
             r#"
-INSERT INTO proxy_node_events (node_id, event_type, detail, created_at)
-VALUES (?, ?, ?, ?)
+INSERT INTO proxy_node_events (node_id, event_type, detail, event_metadata, created_at)
+VALUES (?, ?, ?, ?, ?)
 "#,
         )
         .bind(node_id)
         .bind(event_type)
         .bind(detail)
+        .bind(optional_json_to_string(
+            &event_metadata.cloned(),
+            "proxy_node_events.event_metadata",
+        )?)
         .bind(created_at_unix_secs.unwrap_or_else(current_unix_secs) as i64)
         .execute(&self.pool)
         .await
@@ -166,6 +175,70 @@ VALUES (?, ?, ?, ?)
         Ok(())
     }
 
+    async fn upsert_metrics_bucket(
+        &self,
+        table: &str,
+        node_id: &str,
+        bucket_start: u64,
+        sample: &super::types::TunnelMetricsSample,
+    ) -> Result<(), DataLayerError> {
+        sqlx::query(&format!(
+            r#"
+INSERT INTO {table} (
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON CONFLICT(node_id, bucket_start_unix_secs) DO UPDATE SET
+  samples = {table}.samples + excluded.samples,
+  uptime_samples = {table}.uptime_samples + excluded.uptime_samples,
+  active_connections_sum = {table}.active_connections_sum + excluded.active_connections_sum,
+  active_connections_max = MAX({table}.active_connections_max, excluded.active_connections_max),
+  heartbeat_rtt_ms_sum = {table}.heartbeat_rtt_ms_sum + excluded.heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max = MAX({table}.heartbeat_rtt_ms_max, excluded.heartbeat_rtt_ms_max),
+  connect_errors_delta = {table}.connect_errors_delta + excluded.connect_errors_delta,
+  disconnects_delta = {table}.disconnects_delta + excluded.disconnects_delta,
+  error_events_delta = {table}.error_events_delta + excluded.error_events_delta,
+  ws_in_bytes_delta = {table}.ws_in_bytes_delta + excluded.ws_in_bytes_delta,
+  ws_out_bytes_delta = {table}.ws_out_bytes_delta + excluded.ws_out_bytes_delta,
+  ws_in_frames_delta = {table}.ws_in_frames_delta + excluded.ws_in_frames_delta,
+  ws_out_frames_delta = {table}.ws_out_frames_delta + excluded.ws_out_frames_delta
+"#
+        ))
+        .bind(node_id)
+        .bind(i64::try_from(bucket_start).unwrap_or(i64::MAX))
+        .bind(sample.samples)
+        .bind(sample.uptime_samples)
+        .bind(sample.active_connections_sum)
+        .bind(sample.active_connections_max)
+        .bind(sample.heartbeat_rtt_ms_sum)
+        .bind(sample.heartbeat_rtt_ms_max)
+        .bind(sample.connect_errors_delta)
+        .bind(sample.disconnects_delta)
+        .bind(sample.error_events_delta)
+        .bind(sample.ws_in_bytes_delta)
+        .bind(sample.ws_out_bytes_delta)
+        .bind(sample.ws_in_frames_delta)
+        .bind(sample.ws_out_frames_delta)
+        .execute(&self.pool)
+        .await
+        .map_sql_err()?;
+        Ok(())
+    }
+
     fn normalize_remote_config(
         mutation: &ProxyNodeRemoteConfigMutation,
         existing: Option<&serde_json::Value>,
@@ -298,6 +371,7 @@ SELECT
   node_id,
   event_type,
   detail,
+  event_metadata,
   created_at AS created_at_unix_ms
 FROM proxy_node_events
 WHERE node_id = ?
@@ -312,6 +386,152 @@ LIMIT ?
         .map_sql_err()?;
         rows.iter().map(map_proxy_node_event_row).collect()
     }
+
+    async fn list_proxy_node_events_filtered(
+        &self,
+        node_id: &str,
+        query: &ProxyNodeEventQuery,
+    ) -> Result<Vec<StoredProxyNodeEvent>, DataLayerError> {
+        let rows = sqlx::query(
+            r#"
+SELECT
+  id,
+  node_id,
+  event_type,
+  detail,
+  event_metadata,
+  created_at AS created_at_unix_ms
+FROM proxy_node_events
+WHERE node_id = ?
+  AND (? IS NULL OR created_at >= ?)
+  AND (? IS NULL OR created_at <= ?)
+  AND (? IS NULL OR LOWER(event_type) = LOWER(?))
+ORDER BY created_at DESC, id DESC
+LIMIT ?
+"#,
+        )
+        .bind(node_id)
+        .bind(
+            query
+                .from_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(
+            query
+                .from_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(
+            query
+                .to_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(
+            query
+                .to_unix_secs
+                .map(|v| i64::try_from(v).unwrap_or(i64::MAX)),
+        )
+        .bind(query.event_type.as_deref())
+        .bind(query.event_type.as_deref())
+        .bind(i64::try_from(query.limit).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await
+        .map_sql_err()?;
+        rows.iter().map(map_proxy_node_event_row).collect()
+    }
+
+    async fn list_proxy_node_metrics(
+        &self,
+        node_id: &str,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyNodeMetricsBucket>, DataLayerError> {
+        let table = match step {
+            ProxyNodeMetricsStep::OneMinute => "proxy_node_metrics_1m",
+            ProxyNodeMetricsStep::OneHour => "proxy_node_metrics_1h",
+        };
+        let rows = sqlx::query(&format!(
+            r#"
+SELECT
+  node_id,
+  bucket_start_unix_secs,
+  samples,
+  uptime_samples,
+  active_connections_sum,
+  active_connections_max,
+  heartbeat_rtt_ms_sum,
+  heartbeat_rtt_ms_max,
+  connect_errors_delta,
+  disconnects_delta,
+  error_events_delta,
+  ws_in_bytes_delta,
+  ws_out_bytes_delta,
+  ws_in_frames_delta,
+  ws_out_frames_delta
+FROM {table}
+WHERE node_id = ?
+  AND bucket_start_unix_secs >= ?
+  AND bucket_start_unix_secs <= ?
+ORDER BY bucket_start_unix_secs ASC
+LIMIT ?
+"#
+        ))
+        .bind(node_id)
+        .bind(i64::try_from(from_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(to_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(limit).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await
+        .map_sql_err()?;
+        rows.iter().map(map_proxy_node_metric_row).collect()
+    }
+
+    async fn list_proxy_fleet_metrics(
+        &self,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyFleetMetricsBucket>, DataLayerError> {
+        let table = match step {
+            ProxyNodeMetricsStep::OneMinute => "proxy_node_metrics_1m",
+            ProxyNodeMetricsStep::OneHour => "proxy_node_metrics_1h",
+        };
+        let rows = sqlx::query(&format!(
+            r#"
+SELECT
+  bucket_start_unix_secs,
+  SUM(samples) AS samples,
+  SUM(uptime_samples) AS uptime_samples,
+  SUM(active_connections_sum) AS active_connections_sum,
+  MAX(active_connections_max) AS active_connections_max,
+  SUM(heartbeat_rtt_ms_sum) AS heartbeat_rtt_ms_sum,
+  MAX(heartbeat_rtt_ms_max) AS heartbeat_rtt_ms_max,
+  SUM(connect_errors_delta) AS connect_errors_delta,
+  SUM(disconnects_delta) AS disconnects_delta,
+  SUM(error_events_delta) AS error_events_delta,
+  SUM(ws_in_bytes_delta) AS ws_in_bytes_delta,
+  SUM(ws_out_bytes_delta) AS ws_out_bytes_delta,
+  SUM(ws_in_frames_delta) AS ws_in_frames_delta,
+  SUM(ws_out_frames_delta) AS ws_out_frames_delta
+FROM {table}
+WHERE bucket_start_unix_secs >= ?
+  AND bucket_start_unix_secs <= ?
+GROUP BY bucket_start_unix_secs
+ORDER BY bucket_start_unix_secs ASC
+LIMIT ?
+"#
+        ))
+        .bind(i64::try_from(from_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(to_unix_secs).unwrap_or(i64::MAX))
+        .bind(i64::try_from(limit).unwrap_or(i64::MAX))
+        .fetch_all(&self.pool)
+        .await
+        .map_sql_err()?;
+        rows.iter().map(map_proxy_fleet_metric_row).collect()
+    }
 }
 
 #[async_trait]
@@ -540,7 +760,9 @@ WHERE is_manual = 0
             ));
         }
 
-        let now = Some(current_unix_secs());
+        let previous_proxy_metadata = node.proxy_metadata.clone();
+        let now_unix_secs = current_unix_secs();
+        let now = Some(now_unix_secs);
         node.last_heartbeat_at_unix_secs = now;
         if node.status != "online" || !node.tunnel_connected {
             node.status = "online".to_string();
@@ -584,7 +806,55 @@ WHERE is_manual = 0
             node.config_version = node.config_version.saturating_add(1);
             node.updated_at_unix_secs = now;
         }
+
+        let tunnel_metrics_sample = build_tunnel_metrics_sample(
+            previous_proxy_metadata.as_ref(),
+            node.proxy_metadata.as_ref(),
+            node.active_connections,
+            node.tunnel_connected,
+        );
+
         self.upsert_node(&node).await?;
+
+        if let Some(sample) = tunnel_metrics_sample.as_ref() {
+            self.upsert_metrics_bucket(
+                "proxy_node_metrics_1m",
+                &node.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneMinute),
+                sample,
+            )
+            .await?;
+            self.upsert_metrics_bucket(
+                "proxy_node_metrics_1h",
+                &node.id,
+                bucket_start_unix_secs(now_unix_secs, ProxyNodeMetricsStep::OneHour),
+                sample,
+            )
+            .await?;
+
+            for error in &sample.recent_error_events {
+                let detail = build_tunnel_error_event_detail(error);
+                let event_metadata = serde_json::json!({
+                    "source": "heartbeat",
+                    "category": error.category,
+                    "message": error.message,
+                    "timestamp_unix_secs": error.timestamp_unix_secs,
+                });
+                self.insert_event(
+                    &node.id,
+                    PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
+                    Some(detail.as_str()),
+                    Some(&event_metadata),
+                    Some(if error.timestamp_unix_secs == 0 {
+                        now_unix_secs
+                    } else {
+                        error.timestamp_unix_secs
+                    }),
+                )
+                .await?;
+            }
+        }
+
         Ok(Some(node))
     }
 
@@ -638,6 +908,7 @@ WHERE is_manual = 0
                 &mutation.node_id,
                 event_type,
                 Some(&format!("[stale_ignored] {event_detail}")),
+                None,
                 Some(current_unix_secs()),
             )
             .await?;
@@ -660,6 +931,7 @@ WHERE is_manual = 0
             &mutation.node_id,
             event_type,
             Some(&event_detail),
+            None,
             Some(event_time),
         )
         .await?;
@@ -691,6 +963,16 @@ WHERE is_manual = 0
                 .execute(&self.pool)
                 .await
                 .map_sql_err()?;
+            sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE node_id = ?")
+                .bind(node_id)
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?;
+            sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE node_id = ?")
+                .bind(node_id)
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?;
             sqlx::query("DELETE FROM proxy_nodes WHERE id = ?")
                 .bind(node_id)
                 .execute(&self.pool)
@@ -747,6 +1029,33 @@ WHERE is_manual = 0
         node.updated_at_unix_secs = Some(current_unix_secs());
         self.upsert_node(&node).await
     }
+
+    async fn cleanup_proxy_node_metrics(
+        &self,
+        retain_1m_from_unix_secs: u64,
+        retain_1h_from_unix_secs: u64,
+    ) -> Result<ProxyNodeMetricsCleanupSummary, DataLayerError> {
+        let deleted_1m =
+            sqlx::query("DELETE FROM proxy_node_metrics_1m WHERE bucket_start_unix_secs < ?")
+                .bind(i64::try_from(retain_1m_from_unix_secs).unwrap_or(i64::MAX))
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?
+                .rows_affected() as usize;
+
+        let deleted_1h =
+            sqlx::query("DELETE FROM proxy_node_metrics_1h WHERE bucket_start_unix_secs < ?")
+                .bind(i64::try_from(retain_1h_from_unix_secs).unwrap_or(i64::MAX))
+                .execute(&self.pool)
+                .await
+                .map_sql_err()?
+                .rows_affected() as usize;
+
+        Ok(ProxyNodeMetricsCleanupSummary {
+            deleted_1m_rows: deleted_1m,
+            deleted_1h_rows: deleted_1h,
+        })
+    }
 }
 
 fn optional_unix_secs(value: Option<i64>) -> Option<u64> {
@@ -861,18 +1170,73 @@ fn map_proxy_node_event_row(row: &SqliteRow) -> Result<StoredProxyNodeEvent, Dat
         node_id: row.try_get("node_id").map_sql_err()?,
         event_type: row.try_get("event_type").map_sql_err()?,
         detail: row.try_get("detail").map_sql_err()?,
+        event_metadata: optional_json_from_string(
+            row.try_get("event_metadata").map_sql_err()?,
+            "proxy_node_events.event_metadata",
+        )?,
         created_at_unix_ms: optional_unix_secs(row.try_get("created_at_unix_ms").map_sql_err()?),
     })
 }
 
+fn map_proxy_node_metric_row(
+    row: &SqliteRow,
+) -> Result<StoredProxyNodeMetricsBucket, DataLayerError> {
+    Ok(StoredProxyNodeMetricsBucket {
+        node_id: row.try_get("node_id").map_sql_err()?,
+        bucket_start_unix_secs: optional_unix_secs(
+            row.try_get("bucket_start_unix_secs").map_sql_err()?,
+        )
+        .unwrap_or_default(),
+        samples: row.try_get("samples").map_sql_err()?,
+        uptime_samples: row.try_get("uptime_samples").map_sql_err()?,
+        active_connections_sum: row.try_get("active_connections_sum").map_sql_err()?,
+        active_connections_max: row.try_get("active_connections_max").map_sql_err()?,
+        heartbeat_rtt_ms_sum: row.try_get("heartbeat_rtt_ms_sum").map_sql_err()?,
+        heartbeat_rtt_ms_max: row.try_get("heartbeat_rtt_ms_max").map_sql_err()?,
+        connect_errors_delta: row.try_get("connect_errors_delta").map_sql_err()?,
+        disconnects_delta: row.try_get("disconnects_delta").map_sql_err()?,
+        error_events_delta: row.try_get("error_events_delta").map_sql_err()?,
+        ws_in_bytes_delta: row.try_get("ws_in_bytes_delta").map_sql_err()?,
+        ws_out_bytes_delta: row.try_get("ws_out_bytes_delta").map_sql_err()?,
+        ws_in_frames_delta: row.try_get("ws_in_frames_delta").map_sql_err()?,
+        ws_out_frames_delta: row.try_get("ws_out_frames_delta").map_sql_err()?,
+    })
+}
+
+fn map_proxy_fleet_metric_row(
+    row: &SqliteRow,
+) -> Result<StoredProxyFleetMetricsBucket, DataLayerError> {
+    Ok(StoredProxyFleetMetricsBucket {
+        bucket_start_unix_secs: optional_unix_secs(
+            row.try_get("bucket_start_unix_secs").map_sql_err()?,
+        )
+        .unwrap_or_default(),
+        samples: row.try_get("samples").map_sql_err()?,
+        uptime_samples: row.try_get("uptime_samples").map_sql_err()?,
+        active_connections_sum: row.try_get("active_connections_sum").map_sql_err()?,
+        active_connections_max: row.try_get("active_connections_max").map_sql_err()?,
+        heartbeat_rtt_ms_sum: row.try_get("heartbeat_rtt_ms_sum").map_sql_err()?,
+        heartbeat_rtt_ms_max: row.try_get("heartbeat_rtt_ms_max").map_sql_err()?,
+        connect_errors_delta: row.try_get("connect_errors_delta").map_sql_err()?,
+        disconnects_delta: row.try_get("disconnects_delta").map_sql_err()?,
+        error_events_delta: row.try_get("error_events_delta").map_sql_err()?,
+        ws_in_bytes_delta: row.try_get("ws_in_bytes_delta").map_sql_err()?,
+        ws_out_bytes_delta: row.try_get("ws_out_bytes_delta").map_sql_err()?,
+        ws_in_frames_delta: row.try_get("ws_in_frames_delta").map_sql_err()?,
+        ws_out_frames_delta: row.try_get("ws_out_frames_delta").map_sql_err()?,
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::SqliteProxyNodeReadRepository;
     use crate::lifecycle::migrate::run_sqlite_migrations;
     use crate::repository::proxy_nodes::{
-        ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation, ProxyNodeManualUpdateMutation,
-        ProxyNodeReadRepository, ProxyNodeRegistrationMutation, ProxyNodeRemoteConfigMutation,
-        ProxyNodeTrafficMutation, ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository,
+        ProxyNodeEventQuery, ProxyNodeHeartbeatMutation, ProxyNodeManualCreateMutation,
+        ProxyNodeManualUpdateMutation, ProxyNodeMetricsStep, ProxyNodeReadRepository,
+        ProxyNodeRegistrationMutation, ProxyNodeRemoteConfigMutation, ProxyNodeTrafficMutation,
+        ProxyNodeTunnelStatusMutation, ProxyNodeWriteRepository,
+        PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR,
     };
     use serde_json::json;
 
@@ -1142,4 +1506,131 @@ VALUES ('node-1', 'registered', 'ok', 3)
             .expect("manual node should delete")
             .is_some());
     }
+
+    #[tokio::test]
+    async fn sqlite_repository_aggregates_proxy_node_metrics_and_filters_events() {
+        let pool = sqlx::sqlite::SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect("sqlite::memory:")
+            .await
+            .expect("sqlite pool should connect");
+        run_sqlite_migrations(&pool)
+            .await
+            .expect("sqlite migrations should run");
+
+        let repository = SqliteProxyNodeReadRepository::new(pool);
+        let registered = repository
+            .register_node(&ProxyNodeRegistrationMutation {
+                name: "tunnel-1".to_string(),
+                ip: "10.0.0.1".to_string(),
+                port: 7000,
+                region: None,
+                heartbeat_interval: 30,
+                active_connections: Some(0),
+                total_requests: Some(0),
+                avg_latency_ms: None,
+                hardware_info: None,
+                estimated_max_concurrency: None,
+                proxy_metadata: None,
+                proxy_version: Some("1.0.0".to_string()),
+                registered_by: None,
+                tunnel_mode: true,
+            })
+            .await
+            .expect("node should register");
+        let now = super::current_unix_secs();
+        repository
+            .apply_heartbeat(&ProxyNodeHeartbeatMutation {
+                node_id: registered.id.clone(),
+                heartbeat_interval: Some(30),
+                active_connections: Some(5),
+                total_requests_delta: None,
+                avg_latency_ms: None,
+                failed_requests_delta: None,
+                dns_failures_delta: None,
+                stream_errors_delta: None,
+                proxy_metadata: Some(json!({
+                    "tunnel_metrics": {
+                        "connect_errors": 4,
+                        "disconnects": 1,
+                        "error_events_total": 1,
+                        "ws_in_bytes": 100,
+                        "ws_out_bytes": 200,
+                        "ws_in_frames": 3,
+                        "ws_out_frames": 6,
+                        "heartbeat_rtt_last_ms": 33
+                    },
+                    "recent_tunnel_errors": [{
+                        "timestamp_unix_secs": now,
+                        "category": "tcp_connect_timeout",
+                        "message": "timeout"
+                    }]
+                })),
+                proxy_version: Some("1.0.0".to_string()),
+            })
+            .await
+            .expect("heartbeat should apply")
+            .expect("node should exist");
+
+        let metrics = repository
+            .list_proxy_node_metrics(
+                &registered.id,
+                ProxyNodeMetricsStep::OneMinute,
+                now.saturating_sub(120),
+                now.saturating_add(120),
+                10,
+            )
+            .await
+            .expect("metrics should list");
+        assert_eq!(metrics.len(), 1);
+        assert_eq!(metrics[0].samples, 1);
+        assert_eq!(metrics[0].uptime_samples, 1);
+        assert_eq!(metrics[0].active_connections_max, 5);
+        assert_eq!(metrics[0].heartbeat_rtt_ms_sum, 33);
+        assert_eq!(metrics[0].connect_errors_delta, 4);
+        assert_eq!(metrics[0].ws_out_frames_delta, 6);
+
+        let fleet = repository
+            .list_proxy_fleet_metrics(
+                ProxyNodeMetricsStep::OneMinute,
+                now.saturating_sub(120),
+                now.saturating_add(120),
+                10,
+            )
+            .await
+            .expect("fleet metrics should list");
+        assert_eq!(fleet.len(), 1);
+        assert_eq!(fleet[0].samples, 1);
+        assert_eq!(fleet[0].error_events_delta, 1);
+
+        let events = repository
+            .list_proxy_node_events_filtered(
+                &registered.id,
+                &ProxyNodeEventQuery {
+                    limit: 10,
+                    from_unix_secs: Some(now.saturating_sub(120)),
+                    to_unix_secs: Some(now.saturating_add(120)),
+                    event_type: Some(PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR.to_string()),
+                },
+            )
+            .await
+            .expect("events should list");
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].event_type, PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR);
+        assert_eq!(
+            events[0]
+                .event_metadata
+                .as_ref()
+                .and_then(|value| value.get("category"))
+                .and_then(serde_json::Value::as_str),
+            Some("tcp_connect_timeout")
+        );
+
+        let cleanup = repository
+            .cleanup_proxy_node_metrics(now.saturating_add(1), now.saturating_add(1))
+            .await
+            .expect("cleanup should run");
+        assert_eq!(cleanup.deleted_1m_rows, 1);
+        assert_eq!(cleanup.deleted_1h_rows, 1);
+    }
 }

crates/aether-data/src/repository/proxy_nodes/types.rs

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+use serde_json::Value;
 
 #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct StoredProxyNode {
@@ -239,9 +240,186 @@ pub struct StoredProxyNodeEvent {
     pub node_id: String,
     pub event_type: String,
     pub detail: Option<String>,
+    pub event_metadata: Option<serde_json::Value>,
     pub created_at_unix_ms: Option<u64>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct ProxyNodeEventQuery {
+    pub limit: usize,
+    pub from_unix_secs: Option<u64>,
+    pub to_unix_secs: Option<u64>,
+    pub event_type: Option<String>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum ProxyNodeMetricsStep {
+    OneMinute,
+    OneHour,
+}
+
+impl ProxyNodeMetricsStep {
+    pub fn bucket_size_secs(self) -> u64 {
+        match self {
+            Self::OneMinute => 60,
+            Self::OneHour => 3_600,
+        }
+    }
+
+    pub fn as_api_value(self) -> &'static str {
+        match self {
+            Self::OneMinute => "1m",
+            Self::OneHour => "1h",
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
+pub struct StoredProxyNodeMetricsBucket {
+    pub node_id: String,
+    pub bucket_start_unix_secs: u64,
+    pub samples: i64,
+    pub uptime_samples: i64,
+    pub active_connections_sum: i64,
+    pub active_connections_max: i64,
+    pub heartbeat_rtt_ms_sum: i64,
+    pub heartbeat_rtt_ms_max: i64,
+    pub connect_errors_delta: i64,
+    pub disconnects_delta: i64,
+    pub error_events_delta: i64,
+    pub ws_in_bytes_delta: i64,
+    pub ws_out_bytes_delta: i64,
+    pub ws_in_frames_delta: i64,
+    pub ws_out_frames_delta: i64,
+}
+
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
+pub struct StoredProxyFleetMetricsBucket {
+    pub bucket_start_unix_secs: u64,
+    pub samples: i64,
+    pub uptime_samples: i64,
+    pub active_connections_sum: i64,
+    pub active_connections_max: i64,
+    pub heartbeat_rtt_ms_sum: i64,
+    pub heartbeat_rtt_ms_max: i64,
+    pub connect_errors_delta: i64,
+    pub disconnects_delta: i64,
+    pub error_events_delta: i64,
+    pub ws_in_bytes_delta: i64,
+    pub ws_out_bytes_delta: i64,
+    pub ws_in_frames_delta: i64,
+    pub ws_out_frames_delta: i64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
+pub struct ProxyNodeMetricsCleanupSummary {
+    pub deleted_1m_rows: usize,
+    pub deleted_1h_rows: usize,
+}
+
+pub const PROXY_NODE_EVENT_TYPE_TUNNEL_ERROR: &str = "tunnel_err";
+
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct TunnelErrorEventRecord {
+    pub timestamp_unix_secs: u64,
+    pub category: String,
+    pub message: String,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct TunnelMetricsCounters {
+    pub connect_errors: u64,
+    pub disconnects: u64,
+    pub error_events_total: u64,
+    pub ws_in_bytes: u64,
+    pub ws_out_bytes: u64,
+    pub ws_in_frames: u64,
+    pub ws_out_frames: u64,
+    pub heartbeat_rtt_last_ms: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TunnelMetricsSample {
+    pub samples: i64,
+    pub uptime_samples: i64,
+    pub active_connections_sum: i64,
+    pub active_connections_max: i64,
+    pub heartbeat_rtt_ms_sum: i64,
+    pub heartbeat_rtt_ms_max: i64,
+    pub connect_errors_delta: i64,
+    pub disconnects_delta: i64,
+    pub error_events_delta: i64,
+    pub ws_in_bytes_delta: i64,
+    pub ws_out_bytes_delta: i64,
+    pub ws_in_frames_delta: i64,
+    pub ws_out_frames_delta: i64,
+    pub recent_error_events: Vec<TunnelErrorEventRecord>,
+}
+
+pub fn bucket_start_unix_secs(timestamp_unix_secs: u64, step: ProxyNodeMetricsStep) -> u64 {
+    let size = step.bucket_size_secs();
+    timestamp_unix_secs / size * size
+}
+
+pub fn build_tunnel_metrics_sample(
+    previous_proxy_metadata: Option<&Value>,
+    current_proxy_metadata: Option<&Value>,
+    active_connections: i32,
+    tunnel_connected: bool,
+) -> Option<TunnelMetricsSample> {
+    let current = extract_tunnel_metrics_counters(current_proxy_metadata)?;
+    let previous = extract_tunnel_metrics_counters(previous_proxy_metadata);
+    let current_recent_errors = extract_recent_tunnel_errors(current_proxy_metadata);
+
+    let connect_errors_delta =
+        counter_delta_u64(previous.map(|v| v.connect_errors), current.connect_errors);
+    let disconnects_delta = counter_delta_u64(previous.map(|v| v.disconnects), current.disconnects);
+    let error_events_delta = counter_delta_u64(
+        previous.map(|v| v.error_events_total),
+        current.error_events_total,
+    );
+    let ws_in_bytes_delta = counter_delta_u64(previous.map(|v| v.ws_in_bytes), current.ws_in_bytes);
+    let ws_out_bytes_delta =
+        counter_delta_u64(previous.map(|v| v.ws_out_bytes), current.ws_out_bytes);
+    let ws_in_frames_delta =
+        counter_delta_u64(previous.map(|v| v.ws_in_frames), current.ws_in_frames);
+    let ws_out_frames_delta =
+        counter_delta_u64(previous.map(|v| v.ws_out_frames), current.ws_out_frames);
+
+    let take_recent = usize::try_from(error_events_delta).unwrap_or(usize::MAX);
+    let recent_error_events = if take_recent == 0 {
+        Vec::new()
+    } else {
+        let capture = take_recent.min(current_recent_errors.len());
+        let from = current_recent_errors.len().saturating_sub(capture);
+        current_recent_errors[from..].to_vec()
+    };
+
+    let active_connections = i64::from(active_connections.max(0));
+    let heartbeat_rtt_last_ms = i64::try_from(current.heartbeat_rtt_last_ms).unwrap_or(i64::MAX);
+
+    Some(TunnelMetricsSample {
+        samples: 1,
+        uptime_samples: if tunnel_connected { 1 } else { 0 },
+        active_connections_sum: active_connections,
+        active_connections_max: active_connections,
+        heartbeat_rtt_ms_sum: heartbeat_rtt_last_ms,
+        heartbeat_rtt_ms_max: heartbeat_rtt_last_ms,
+        connect_errors_delta: i64::try_from(connect_errors_delta).unwrap_or(i64::MAX),
+        disconnects_delta: i64::try_from(disconnects_delta).unwrap_or(i64::MAX),
+        error_events_delta: i64::try_from(error_events_delta).unwrap_or(i64::MAX),
+        ws_in_bytes_delta: i64::try_from(ws_in_bytes_delta).unwrap_or(i64::MAX),
+        ws_out_bytes_delta: i64::try_from(ws_out_bytes_delta).unwrap_or(i64::MAX),
+        ws_in_frames_delta: i64::try_from(ws_in_frames_delta).unwrap_or(i64::MAX),
+        ws_out_frames_delta: i64::try_from(ws_out_frames_delta).unwrap_or(i64::MAX),
+        recent_error_events,
+    })
+}
+
+pub fn build_tunnel_error_event_detail(event: &TunnelErrorEventRecord) -> String {
+    format!("[{}] {}", event.category, event.message)
+}
+
 pub fn normalize_proxy_metadata(
     proxy_metadata: Option<&serde_json::Value>,
     proxy_version: Option<&str>,
@@ -276,6 +454,71 @@ pub fn normalize_proxy_metadata(
     }
 }
 
+fn extract_tunnel_metrics_counters(
+    proxy_metadata: Option<&Value>,
+) -> Option<TunnelMetricsCounters> {
+    let tunnel_metrics = proxy_metadata
+        .and_then(Value::as_object)
+        .and_then(|metadata| metadata.get("tunnel_metrics"))
+        .and_then(Value::as_object)?;
+
+    Some(TunnelMetricsCounters {
+        connect_errors: json_u64(tunnel_metrics.get("connect_errors")).unwrap_or(0),
+        disconnects: json_u64(tunnel_metrics.get("disconnects")).unwrap_or(0),
+        error_events_total: json_u64(tunnel_metrics.get("error_events_total")).unwrap_or(0),
+        ws_in_bytes: json_u64(tunnel_metrics.get("ws_in_bytes")).unwrap_or(0),
+        ws_out_bytes: json_u64(tunnel_metrics.get("ws_out_bytes")).unwrap_or(0),
+        ws_in_frames: json_u64(tunnel_metrics.get("ws_in_frames")).unwrap_or(0),
+        ws_out_frames: json_u64(tunnel_metrics.get("ws_out_frames")).unwrap_or(0),
+        heartbeat_rtt_last_ms: json_u64(tunnel_metrics.get("heartbeat_rtt_last_ms")).unwrap_or(0),
+    })
+}
+
+fn extract_recent_tunnel_errors(proxy_metadata: Option<&Value>) -> Vec<TunnelErrorEventRecord> {
+    proxy_metadata
+        .and_then(Value::as_object)
+        .and_then(|metadata| metadata.get("recent_tunnel_errors"))
+        .and_then(Value::as_array)
+        .map(|items| {
+            items
+                .iter()
+                .filter_map(|item| {
+                    let item = item.as_object()?;
+                    Some(TunnelErrorEventRecord {
+                        timestamp_unix_secs: json_u64(item.get("timestamp_unix_secs"))
+                            .unwrap_or_default(),
+                        category: item
+                            .get("category")
+                            .and_then(Value::as_str)
+                            .unwrap_or("unknown")
+                            .to_string(),
+                        message: item
+                            .get("message")
+                            .and_then(Value::as_str)
+                            .unwrap_or("n/a")
+                            .to_string(),
+                    })
+                })
+                .collect::<Vec<_>>()
+        })
+        .unwrap_or_default()
+}
+
+fn json_u64(value: Option<&Value>) -> Option<u64> {
+    value.and_then(|value| {
+        value
+            .as_u64()
+            .or_else(|| value.as_i64().and_then(|n| (n >= 0).then_some(n as u64)))
+    })
+}
+
+fn counter_delta_u64(previous: Option<u64>, current: u64) -> u64 {
+    match previous {
+        Some(previous) if current >= previous => current - previous,
+        Some(_) | None => current,
+    }
+}
+
 fn normalize_proxy_version_label(value: &str) -> Option<String> {
     let trimmed = value.trim();
     if trimmed.is_empty() {
@@ -375,6 +618,42 @@ pub trait ProxyNodeReadRepository: Send + Sync {
         node_id: &str,
         limit: usize,
     ) -> Result<Vec<StoredProxyNodeEvent>, crate::DataLayerError>;
+
+    async fn list_proxy_node_events_filtered(
+        &self,
+        node_id: &str,
+        query: &ProxyNodeEventQuery,
+    ) -> Result<Vec<StoredProxyNodeEvent>, crate::DataLayerError> {
+        let mut items = self.list_proxy_node_events(node_id, query.limit).await?;
+        if let Some(from_unix_secs) = query.from_unix_secs {
+            items.retain(|item| item.created_at_unix_ms.unwrap_or(0) >= from_unix_secs);
+        }
+        if let Some(to_unix_secs) = query.to_unix_secs {
+            items.retain(|item| item.created_at_unix_ms.unwrap_or(u64::MAX) <= to_unix_secs);
+        }
+        if let Some(event_type) = query.event_type.as_deref() {
+            items.retain(|item| item.event_type.eq_ignore_ascii_case(event_type));
+        }
+        items.truncate(query.limit);
+        Ok(items)
+    }
+
+    async fn list_proxy_node_metrics(
+        &self,
+        node_id: &str,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyNodeMetricsBucket>, crate::DataLayerError>;
+
+    async fn list_proxy_fleet_metrics(
+        &self,
+        step: ProxyNodeMetricsStep,
+        from_unix_secs: u64,
+        to_unix_secs: u64,
+        limit: usize,
+    ) -> Result<Vec<StoredProxyFleetMetricsBucket>, crate::DataLayerError>;
 }
 
 #[async_trait]
@@ -433,6 +712,12 @@ pub trait ProxyNodeWriteRepository: Send + Sync {
         failed_delta: i64,
         latency_ms: Option<i64>,
     ) -> Result<(), crate::DataLayerError>;
+
+    async fn cleanup_proxy_node_metrics(
+        &self,
+        retain_1m_from_unix_secs: u64,
+        retain_1h_from_unix_secs: u64,
+    ) -> Result<ProxyNodeMetricsCleanupSummary, crate::DataLayerError>;
 }
 
 #[cfg(test)]
@@ -440,9 +725,10 @@ mod tests {
     use serde_json::json;
 
     use super::{
-        normalize_proxy_node_scheduling_state, proxy_node_accepts_new_tunnels,
-        proxy_reported_version, reconcile_remote_config_after_heartbeat,
-        remote_config_scheduling_state, remote_config_upgrade_target, StoredProxyNode,
+        bucket_start_unix_secs, build_tunnel_metrics_sample, normalize_proxy_node_scheduling_state,
+        proxy_node_accepts_new_tunnels, proxy_reported_version,
+        reconcile_remote_config_after_heartbeat, remote_config_scheduling_state,
+        remote_config_upgrade_target, ProxyNodeMetricsStep, StoredProxyNode,
     };
 
     #[test]
@@ -527,4 +813,63 @@ mod tests {
 
         assert!(!proxy_node_accepts_new_tunnels(&node));
     }
+
+    #[test]
+    fn builds_tunnel_metrics_sample_with_reset_safe_counter_deltas() {
+        let previous = json!({
+            "tunnel_metrics": {
+                "connect_errors": 10,
+                "disconnects": 5,
+                "error_events_total": 7,
+                "ws_in_bytes": 1_000,
+                "ws_out_bytes": 2_000,
+                "ws_in_frames": 10,
+                "ws_out_frames": 20,
+                "heartbeat_rtt_last_ms": 30
+            }
+        });
+        let current = json!({
+            "tunnel_metrics": {
+                "connect_errors": 12,
+                "disconnects": 2,
+                "error_events_total": 9,
+                "ws_in_bytes": 1_500,
+                "ws_out_bytes": 100,
+                "ws_in_frames": 11,
+                "ws_out_frames": 3,
+                "heartbeat_rtt_last_ms": 44
+            },
+            "recent_tunnel_errors": [
+                {"timestamp_unix_secs": 100, "category": "older", "message": "old"},
+                {"timestamp_unix_secs": 101, "category": "newer", "message": "new"}
+            ]
+        });
+
+        let sample = build_tunnel_metrics_sample(Some(&previous), Some(&current), 4, true)
+            .expect("sample should build");
+        assert_eq!(sample.samples, 1);
+        assert_eq!(sample.uptime_samples, 1);
+        assert_eq!(sample.active_connections_sum, 4);
+        assert_eq!(sample.heartbeat_rtt_ms_sum, 44);
+        assert_eq!(sample.connect_errors_delta, 2);
+        assert_eq!(sample.disconnects_delta, 2);
+        assert_eq!(sample.error_events_delta, 2);
+        assert_eq!(sample.ws_in_bytes_delta, 500);
+        assert_eq!(sample.ws_out_bytes_delta, 100);
+        assert_eq!(sample.ws_out_frames_delta, 3);
+        assert_eq!(sample.recent_error_events.len(), 2);
+        assert_eq!(sample.recent_error_events[0].category, "older");
+    }
+
+    #[test]
+    fn maps_timestamps_to_metric_buckets() {
+        assert_eq!(
+            bucket_start_unix_secs(1_710_000_119, ProxyNodeMetricsStep::OneMinute),
+            1_710_000_060
+        );
+        assert_eq!(
+            bucket_start_unix_secs(1_710_003_999, ProxyNodeMetricsStep::OneHour),
+            1_710_003_600
+        );
+    }
 }