feat: request & RPC ID analytics

src/analytics/message_info.rs

@@ -1,5 +1,6 @@
 use {
     crate::{handlers::RpcQueryParams, json_rpc::JsonRpcRequest, providers::ProviderKind},
+    hyper::HeaderMap,
     parquet_derive::ParquetRecordWriter,
     serde::{Deserialize, Serialize},
     std::sync::Arc,
@@ -16,6 +17,9 @@ pub struct MessageInfo {
     pub method: Arc<str>,
     pub source: String,
 
+    pub request_id: Option<String>,
+    pub rpc_id: String,
+
     pub origin: Option<String>,
     pub provider: String,
 
@@ -32,6 +36,7 @@ impl MessageInfo {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         query_params: &RpcQueryParams,
+        headers: &HeaderMap,
         request: &JsonRpcRequest,
         region: Option<Vec<String>>,
         country: Option<Arc<str>>,
@@ -53,6 +58,12 @@ impl MessageInfo {
                 .unwrap_or(&MessageSource::Rpc)
                 .to_string(),
 
+            request_id: headers
+                .get("x-request-id")
+                .and_then(|v| v.to_str().ok())
+                .map(|v| v.to_string()),
+            rpc_id: request.id.to_string(),
+
             origin,
             provider: provider.to_string(),
 
@@ -65,6 +76,7 @@ impl MessageInfo {
     }
 }
 
+// Note: these are all INTERNAL sources (except Rpc). While technically the user can override this via query param currently, this is just a technical limitation of the implementation here.
 #[derive(Debug, Clone, EnumString, Display, Deserialize, PartialEq)]
 #[strum(serialize_all = "snake_case")]
 #[serde(rename_all = "snake_case")]

src/handlers/balance.rs

@@ -248,7 +248,7 @@ async fn handler_internal(
         let (country, continent, region) = state
             .analytics
             .lookup_geo_data(
-                network::get_forwarded_ip(headers).unwrap_or_else(|| connect_info.0.ip()),
+                network::get_forwarded_ip(&headers).unwrap_or_else(|| connect_info.0.ip()),
             )
             .map(|geo| (geo.country, geo.continent, geo.region))
             .unwrap_or((None, None, None));

src/handlers/chain_agnostic/route.rs

@@ -553,7 +553,7 @@ async fn handler_internal(
         let (country, continent, region) = state
             .analytics
             .lookup_geo_data(
-                network::get_forwarded_ip(headers).unwrap_or_else(|| connect_info.0.ip()),
+                network::get_forwarded_ip(&headers).unwrap_or_else(|| connect_info.0.ip()),
             )
             .map(|geo| (geo.country, geo.continent, geo.region))
             .unwrap_or((None, None, None));

src/handlers/history.rs

@@ -203,7 +203,7 @@ async fn handler_internal(
 
     let (country, continent, region) = state
         .analytics
-        .lookup_geo_data(network::get_forwarded_ip(headers).unwrap_or_else(|| connect_info.0.ip()))
+        .lookup_geo_data(network::get_forwarded_ip(&headers).unwrap_or_else(|| connect_info.0.ip()))
         .map(|geo| (geo.country, geo.continent, geo.region))
         .unwrap_or((None, None, None));
 

src/handlers/identity.rs

@@ -134,7 +134,7 @@ async fn handler_internal(
         let (country, continent, region) = state
             .analytics
             .lookup_geo_data(
-                network::get_forwarded_ip(headers).unwrap_or_else(|| connect_info.0.ip()),
+                network::get_forwarded_ip(&headers).unwrap_or_else(|| connect_info.0.ip()),
             )
             .map(|geo| (geo.country, geo.continent, geo.region))
             .unwrap_or((None, None, None));

src/handlers/mod.rs

@@ -98,7 +98,7 @@ pub async fn rate_limit_middleware<B>(
     next: Next<B>,
 ) -> Response {
     let headers = req.headers().clone();
-    let ip = match network::get_forwarded_ip(headers.clone()) {
+    let ip = match network::get_forwarded_ip(&headers) {
         Some(ip) => ip.to_string(),
         None => {
             error!(

src/handlers/profile/register.rs

@@ -195,7 +195,7 @@ pub async fn handler_internal(
         let (country, continent, region) = state
             .analytics
             .lookup_geo_data(
-                network::get_forwarded_ip(headers).unwrap_or_else(|| connect_info.0.ip()),
+                network::get_forwarded_ip(&headers).unwrap_or_else(|| connect_info.0.ip()),
             )
             .map(|geo| (geo.country, geo.continent, geo.region))
             .unwrap_or((None, None, None));

src/handlers/proxy.rs

@@ -196,12 +196,13 @@ pub async fn rpc_provider_call(
     if let Ok(rpc_request) = serde_json::from_slice(&body) {
         let (country, continent, region) = state
             .analytics
-            .lookup_geo_data(network::get_forwarded_ip(headers).unwrap_or_else(|| addr.ip()))
+            .lookup_geo_data(network::get_forwarded_ip(&headers).unwrap_or_else(|| addr.ip()))
             .map(|geo| (geo.country, geo.continent, geo.region))
             .unwrap_or((None, None, None));
 
         state.analytics.message(MessageInfo::new(
             &query_params,
+            &headers,
             &rpc_request,
             region,
             country,

src/utils/network.rs

@@ -66,7 +66,7 @@ fn is_public_ip_addr(addr: IpAddr) -> bool {
     RESERVED_NETWORKS.iter().all(|range| !range.contains(&addr))
 }
 
-pub fn get_forwarded_ip(headers: HeaderMap) -> Option<IpAddr> {
+pub fn get_forwarded_ip(headers: &HeaderMap) -> Option<IpAddr> {
     headers
         .get("X-Forwarded-For")
         .and_then(|header| header.to_str().ok())
@@ -84,7 +84,7 @@ mod tests {
         let mut headers_single = HeaderMap::new();
         headers_single.insert("X-Forwarded-For", "10.128.128.1".parse().unwrap());
         assert_eq!(
-            get_forwarded_ip(headers_single).unwrap(),
+            get_forwarded_ip(&headers_single).unwrap(),
             "10.128.128.1".parse::<IpAddr>().unwrap()
         );
 
@@ -95,7 +95,7 @@ mod tests {
             "10.128.128.1, 10.128.128.2".parse().unwrap(),
         );
         assert_eq!(
-            get_forwarded_ip(headers_multiple).unwrap(),
+            get_forwarded_ip(&headers_multiple).unwrap(),
             "10.128.128.2".parse::<IpAddr>().unwrap()
         );
     }