feat(analytics): fix formatting issues

Author: jagan-jaya

src/analytics/events.rs

@@ -25,12 +25,7 @@ pub struct RoutingEvent {
 }
 
 impl RoutingEvent {
-    pub fn new(
-        merchant_id: String,
-        request_id: String,
-        endpoint: String,
-        method: String,
-    ) -> Self {
+    pub fn new(merchant_id: String, request_id: String, endpoint: String, method: String) -> Self {
         Self {
             event_id: Uuid::new_v4().to_string(),
             merchant_id,
@@ -137,17 +132,21 @@ impl RoutingEvent {
         if !self.response_payload.is_empty() {
             if let Ok(response_json) = serde_json::from_str::<Value>(&self.response_payload) {
                 // Try to extract gateway from various possible response structures
-                if let Some(gateway) = response_json.get("gateway")
+                if let Some(gateway) = response_json
+                    .get("gateway")
                     .or_else(|| response_json.get("selected_gateway"))
                     .or_else(|| response_json.get("connector"))
-                    .and_then(|v| v.as_str()) {
+                    .and_then(|v| v.as_str())
+                {
                     self.gateway_selected = Some(gateway.to_string());
                 }
 
                 // Try to extract routing algorithm ID
-                if let Some(algo_id) = response_json.get("routing_algorithm_id")
+                if let Some(algo_id) = response_json
+                    .get("routing_algorithm_id")
                     .or_else(|| response_json.get("algorithm_id"))
-                    .and_then(|v| v.as_str()) {
+                    .and_then(|v| v.as_str())
+                {
                     self.routing_algorithm_id = Some(algo_id.to_string());
                 }
             }
@@ -159,10 +158,12 @@ impl RoutingEvent {
     pub fn extract_error_from_response(&mut self) -> AnalyticsResult<()> {
         if self.status_code >= 400 && !self.response_payload.is_empty() {
             if let Ok(response_json) = serde_json::from_str::<Value>(&self.response_payload) {
-                if let Some(error) = response_json.get("error")
+                if let Some(error) = response_json
+                    .get("error")
                     .or_else(|| response_json.get("message"))
                     .or_else(|| response_json.get("error_message"))
-                    .and_then(|v| v.as_str()) {
+                    .and_then(|v| v.as_str())
+                {
                     self.error_message = Some(error.to_string());
                 }
             }
@@ -199,7 +200,10 @@ fn extract_ip_address(request: &Request) -> Option<String> {
         .and_then(|v| v.to_str().ok())
         .map(|s| {
             // Take the first IP if there are multiple (comma-separated)
-            s.split(',').next().map(|ip| ip.trim().to_string()).unwrap_or_else(|| String::new())
+            s.split(',')
+                .next()
+                .map(|ip| ip.trim().to_string())
+                .unwrap_or_else(|| String::new())
         })
 }
 
@@ -274,6 +278,9 @@ mod tests {
 
         event.extract_error_from_response().unwrap();
 
-        assert_eq!(event.error_message, Some("Gateway not available".to_string()));
+        assert_eq!(
+            event.error_message,
+            Some("Gateway not available".to_string())
+        );
     }
 }

src/analytics/kafka_producer.rs

@@ -2,7 +2,7 @@ use crate::analytics::{AnalyticsError, AnalyticsResult, KafkaConfig, RoutingEven
 use kafka::producer::{Producer, Record, RequiredAcks};
 use std::time::Duration;
 use tokio::sync::mpsc;
-use tracing::{error, info, warn, debug};
+use tracing::{debug, error, info, warn};
 
 #[derive(Clone)]
 pub struct KafkaProducer {
@@ -13,39 +13,48 @@ pub struct KafkaProducer {
 impl KafkaProducer {
     pub fn new(config: KafkaConfig) -> AnalyticsResult<Self> {
         let topic = format!("{}-routing-events", config.topic_prefix);
-        
+
         // Validate broker configuration
         if config.brokers.is_empty() {
             return Err(AnalyticsError::Configuration(
-                "No Kafka brokers configured".to_string()
+                "No Kafka brokers configured".to_string(),
             ));
         }
-        
-        debug!("Initializing Kafka producer with brokers: {:?}", config.brokers);
-        
+
+        debug!(
+            "Initializing Kafka producer with brokers: {:?}",
+            config.brokers
+        );
+
         Ok(Self { config, topic })
     }
 
     /// Test Kafka connectivity
     pub async fn test_connection(&self) -> AnalyticsResult<()> {
-        debug!("Testing Kafka connection to brokers: {:?}", self.config.brokers);
-        
+        debug!(
+            "Testing Kafka connection to brokers: {:?}",
+            self.config.brokers
+        );
+
         let producer = Producer::from_hosts(self.config.brokers.clone())
             .with_ack_timeout(Duration::from_secs(5))
             .with_required_acks(RequiredAcks::One)
             .create()
             .map_err(|e| {
-                error!("Failed to create Kafka producer for connection test: {:?}", e);
+                error!(
+                    "Failed to create Kafka producer for connection test: {:?}",
+                    e
+                );
                 AnalyticsError::Kafka(e)
             })?;
-        
+
         info!("Kafka connection test successful");
         Ok(())
     }
 
     pub async fn send_event(&self, event: &RoutingEventData) -> AnalyticsResult<()> {
         let json_data = serde_json::to_string(event)?;
-        
+
         // Create producer with configuration
         let mut producer = Producer::from_hosts(self.config.brokers.clone())
             .with_ack_timeout(Duration::from_secs(1))
@@ -54,11 +63,10 @@ impl KafkaProducer {
             .map_err(AnalyticsError::Kafka)?;
 
         // Send the record
-        let record = Record::from_key_value(&self.topic, event.event_id.as_bytes(), json_data.as_bytes());
-        
-        producer
-            .send(&record)
-            .map_err(AnalyticsError::Kafka)?;
+        let record =
+            Record::from_key_value(&self.topic, event.event_id.as_bytes(), json_data.as_bytes());
+
+        producer.send(&record).map_err(AnalyticsError::Kafka)?;
 
         Ok(())
     }
@@ -80,15 +88,28 @@ impl KafkaProducer {
 
         for (index, event) in events.iter().enumerate() {
             let json_data = serde_json::to_string(event)?;
-            let record = Record::from_key_value(&self.topic, event.event_id.as_bytes(), json_data.as_bytes());
-            
+            let record = Record::from_key_value(
+                &self.topic,
+                event.event_id.as_bytes(),
+                json_data.as_bytes(),
+            );
+
             if let Err(e) = producer.send(&record) {
-                error!("Failed to send event {} of {} to Kafka: {:?}", index + 1, events.len(), e);
+                error!(
+                    "Failed to send event {} of {} to Kafka: {:?}",
+                    index + 1,
+                    events.len(),
+                    e
+                );
                 return Err(AnalyticsError::Kafka(e));
             }
         }
 
-        info!("Successfully sent {} events to Kafka topic: {}", events.len(), self.topic);
+        info!(
+            "Successfully sent {} events to Kafka topic: {}",
+            events.len(),
+            self.topic
+        );
         Ok(())
     }
 
@@ -97,7 +118,10 @@ impl KafkaProducer {
         match self.send_events_batch(events).await {
             Ok(()) => true,
             Err(e) => {
-                warn!("Failed to send events batch to Kafka, continuing without analytics: {:?}", e);
+                warn!(
+                    "Failed to send events batch to Kafka, continuing without analytics: {:?}",
+                    e
+                );
                 false
             }
         }
@@ -126,7 +150,7 @@ impl KafkaProducer {
                         match event {
                             Some(event) => {
                                 batch.push(event);
-                                
+
                                 // Flush if batch is full
                                 if batch.len() >= batch_size {
                                     let success = producer.send_events_batch_graceful(&batch).await;
@@ -135,7 +159,7 @@ impl KafkaProducer {
                                     } else {
                                         consecutive_failures += 1;
                                         if consecutive_failures >= max_consecutive_failures {
-                                            warn!("Too many consecutive Kafka failures ({}), continuing to collect events but not sending", 
+                                            warn!("Too many consecutive Kafka failures ({}), continuing to collect events but not sending",
                                                   consecutive_failures);
                                         }
                                     }
@@ -154,7 +178,7 @@ impl KafkaProducer {
                             }
                         }
                     }
-                    
+
                     // Timeout-based flush
                     _ = tokio::time::sleep_until(last_flush + batch_timeout) => {
                         if !batch.is_empty() {
@@ -164,7 +188,7 @@ impl KafkaProducer {
                             } else {
                                 consecutive_failures += 1;
                                 if consecutive_failures >= max_consecutive_failures {
-                                    warn!("Too many consecutive Kafka failures ({}), continuing to collect events but not sending", 
+                                    warn!("Too many consecutive Kafka failures ({}), continuing to collect events but not sending",
                                           consecutive_failures);
                                 }
                             }

src/analytics/middleware.rs

@@ -24,14 +24,14 @@ pub async fn analytics_middleware(
     }
 
     let start_time = Instant::now();
-    
+
     // Extract request information
     let method = request.method().to_string();
     let endpoint = path.to_string();
-    
+
     // Extract merchant ID from request headers or body (simplified for now)
     let merchant_id = extract_merchant_id(&request).unwrap_or("public".to_string());
-    
+
     // Get the tenant app state to access analytics client
     let tenant_app_state = match global_app_state.get_app_state_of_tenant(&merchant_id).await {
         Ok(state) => state,
@@ -46,62 +46,66 @@ pub async fn analytics_middleware(
             }
         }
     };
-    
+
     // Create routing event
     let mut routing_event = RoutingEvent::from_request(&request, merchant_id.clone());
-    
+
     // Extract request body for logging
     let (request_parts, body) = request.into_parts();
     let body_bytes = match body.collect().await {
         Ok(collected) => collected.to_bytes(),
         Err(_) => Bytes::new(),
     };
-    
+
     let request_payload = String::from_utf8_lossy(&body_bytes).to_string();
     routing_event = routing_event.with_request_payload(&request_payload);
-    
+
     // Reconstruct request with body
     let request = Request::from_parts(request_parts, Body::from(body_bytes));
-    
+
     // Process the request
     let response = next.run(request).await;
-    
+
     // Calculate processing time
     let processing_time = start_time.elapsed().as_millis() as u32;
-    
+
     // Extract response information
     let status_code = response.status().as_u16();
-    
-    // Extract response body for logging. Note: This operation can lead to high memory usage 
+
+    // Extract response body for logging. Note: This operation can lead to high memory usage
     let (response_parts, body) = response.into_parts();
     let body_bytes = match body.collect().await {
         Ok(collected) => collected.to_bytes(),
         Err(_) => Bytes::new(),
     };
-    
+
     let response_payload = String::from_utf8_lossy(&body_bytes).to_string();
-    
+
     // Complete the routing event
     routing_event = routing_event
         .with_response_payload(&response_payload)
         .with_status_code(status_code)
         .with_processing_time(processing_time);
-    
+
     // Extract gateway information from response
     if let Err(e) = routing_event.extract_gateway_from_response() {
         warn!("Failed to extract gateway from response: {:?}", e);
     }
-    
+
     // Extract error information if status indicates failure
     if let Err(e) = routing_event.extract_error_from_response() {
         warn!("Failed to extract error from response: {:?}", e);
     }
-    
+
     // Send event to analytics (async, non-blocking)
-    if let Err(e) = tenant_app_state.analytics_client.track_routing_event(routing_event).await {
+    if let Err(e) = tenant_app_state
+        .analytics_client
+        .track_routing_event(routing_event)
+        .await
+    {
         error!("Failed to track routing event: {:?}", e);
     }
-    
+
     // Reconstruct response
     Response::from_parts(response_parts, Body::from(body_bytes))
 }
@@ -119,14 +123,14 @@ fn extract_merchant_id(request: &Request) -> Option<String> {
             return Some(merchant_id_str.to_string());
         }
     }
-    
+
     // Try x-tenant-id header as fallback
     if let Some(tenant_id) = request.headers().get("x-tenant-id") {
         if let Ok(tenant_id_str) = tenant_id.to_str() {
             return Some(tenant_id_str.to_string());
         }
     }
-    
+
     // Default to "public" tenant
     Some("public".to_string())
 }
@@ -146,15 +150,15 @@ mod tests {
     #[test]
     fn test_extract_merchant_id_from_header() {
         use axum::http::{HeaderMap, HeaderValue};
-        
+
         let mut headers = HeaderMap::new();
         headers.insert("x-merchant-id", HeaderValue::from_static("merchant-123"));
-        
+
         let request = Request::builder()
             .uri("/routing/evaluate")
             .body(Body::empty())
             .unwrap();
-        
+
         // Note: This test would need to be adjusted to work with the actual request structure
         // For now, it's a placeholder to show the testing approach
     }