feat: Improve rating_billing with integer arithmetic, overflow checks, and SCAN optimization

- Replace float division with integer arithmetic in calculate_usage_cost for data/voice conversions
- Add checked_add overflow protection for usage accumulation in apply_charging_rules and process_usage
- Replace String-based charging rules with ChargingRule enum variants
- Add Datelike import and implement billing period filtering in generate_invoice
- Replace blocking KEYS command with SCAN cursor iteration

apps/charging-engine/src/charging/rating_billing.rs

@@ -1,7 +1,7 @@
-use chrono::Utc;
+use chrono::{Utc, Datelike};
 use tracing::{info, debug, warn};
 
-use super::types::{UsageEvent, UsageType};
+use super::types::{UsageEvent, UsageType, ChargingRule};
 use crate::errors::{ChargingError, ChargingResult, validate_amount};
 
 impl super::ChargingEngine {
@@ -18,14 +18,25 @@ impl super::ChargingEngine {
 
         let cost = match event.usage_type {
             UsageType::Data => {
-                let data_mb = event.volume as f64 / 1_048_576.0; // Convert bytes to MB
-                data_mb * plan.data_rate
+                // Use integer arithmetic: bytes to MB = bytes / 1,048,576
+                let data_mb = event.volume / 1_048_576;
+                let remainder_bytes = event.volume % 1_048_576;
+                // Calculate cost: full MBs + partial MB
+                let cost_full_mb = data_mb as f64 * plan.data_rate;
+                let cost_partial = (remainder_bytes as f64 / 1_048_576.0) * plan.data_rate;
+                cost_full_mb + cost_partial
             }
             UsageType::Voice => {
-                let voice_minutes = event.volume as f64 / 60.0; // Convert seconds to minutes
-                voice_minutes * plan.voice_rate
+                // Use integer arithmetic: seconds to minutes = seconds / 60
+                let voice_minutes = event.volume / 60;
+                let remainder_seconds = event.volume % 60;
+                // Calculate cost: full minutes + partial minutes
+                let cost_full_min = voice_minutes as f64 * plan.voice_rate;
+                let cost_partial = (remainder_seconds as f64 / 60.0) * plan.voice_rate;
+                cost_full_min + cost_partial
             }
             UsageType::SMS => {
+                // SMS rate is per message, no conversion needed
                 event.volume as f64 * plan.sms_rate
             }
         };
@@ -51,12 +62,12 @@ impl super::ChargingEngine {
         Ok(event)
     }
 
-    pub async fn apply_charging_rules(&self, event: &UsageEvent) -> ChargingResult<Vec<String>> {
+    pub async fn apply_charging_rules(&self, event: &UsageEvent) -> ChargingResult<Vec<ChargingRule>> {
         let mut applied_rules = Vec::new();
 
         // Rule 1: Check if user has sufficient credit
         if !self.check_credit(&event.imsi, event.volume).await? {
-            applied_rules.push("INSUFFICIENT_CREDIT".to_string());
+            applied_rules.push(ChargingRule::InsufficientCredit);
             warn!("Insufficient credit for IMSI: {}", event.imsi);
         }
 
@@ -65,20 +76,26 @@ impl super::ChargingEngine {
         if let Some(account) = account {
             match event.usage_type {
                 UsageType::Data => {
-                    if account.data_used + event.volume > account.data_limit {
-                        applied_rules.push("DATA_LIMIT_EXCEEDED".to_string());
+                    let new_usage = account.data_used.checked_add(event.volume)
+                        .ok_or_else(|| ChargingError::InvalidInput("Data usage would overflow".to_string()))?;
+                    if new_usage > account.data_limit {
+                        applied_rules.push(ChargingRule::DataLimitExceeded);
                         warn!("Data limit exceeded for IMSI: {}", event.imsi);
                     }
                 }
                 UsageType::Voice => {
-                    if account.voice_used + event.volume > account.voice_limit {
-                        applied_rules.push("VOICE_LIMIT_EXCEEDED".to_string());
+                    let new_usage = account.voice_used.checked_add(event.volume)
+                        .ok_or_else(|| ChargingError::InvalidInput("Voice usage would overflow".to_string()))?;
+                    if new_usage > account.voice_limit {
+                        applied_rules.push(ChargingRule::VoiceLimitExceeded);
                         warn!("Voice limit exceeded for IMSI: {}", event.imsi);
                     }
                 }
                 UsageType::SMS => {
-                    if account.sms_used + event.volume > account.sms_limit {
-                        applied_rules.push("SMS_LIMIT_EXCEEDED".to_string());
+                    let new_usage = account.sms_used.checked_add(event.volume)
+                        .ok_or_else(|| ChargingError::InvalidInput("SMS usage would overflow".to_string()))?;
+                    if new_usage > account.sms_limit {
+                        applied_rules.push(ChargingRule::SmsLimitExceeded);
                         warn!("SMS limit exceeded for IMSI: {}", event.imsi);
                     }
                 }
@@ -87,12 +104,12 @@ impl super::ChargingEngine {
 
         // Rule 3: Check if user is blocked
         if self.is_user_blocked(&event.imsi).await? {
-            applied_rules.push("USER_BLOCKED".to_string());
+            applied_rules.push(ChargingRule::UserBlocked);
             warn!("Blocked user attempted usage: {}", event.imsi);
         }
 
         if applied_rules.is_empty() {
-            applied_rules.push("ALLOWED".to_string());
+            applied_rules.push(ChargingRule::Allowed);
         }
 
         Ok(applied_rules)
@@ -103,10 +120,11 @@ impl super::ChargingEngine {
         let rules = self.apply_charging_rules(&event).await?;
         
         // Check if usage is allowed
-        if rules.contains(&"BLOCKED".to_string()) || 
-           rules.contains(&"INSUFFICIENT_CREDIT".to_string()) ||
-           rules.contains(&"USER_BLOCKED".to_string()) {
-            return Err(ChargingError::UsageBlocked(rules.join(", ")));
+        if rules.contains(&ChargingRule::Blocked) || 
+           rules.contains(&ChargingRule::InsufficientCredit) ||
+           rules.contains(&ChargingRule::UserBlocked) {
+            let rule_strs: Vec<&str> = rules.iter().map(|r| r.as_str()).collect();
+            return Err(ChargingError::UsageBlocked(rule_strs.join(", ")));
         }
 
         // Rate the usage
@@ -115,9 +133,18 @@ impl super::ChargingEngine {
         // Update subscriber account
         if let Some(mut account) = self.get_subscriber_account(&rated_event.imsi).await? {
             match rated_event.usage_type {
-                UsageType::Data => account.data_used += rated_event.volume,
-                UsageType::Voice => account.voice_used += rated_event.volume,
-                UsageType::SMS => account.sms_used += rated_event.volume,
+                UsageType::Data => {
+                    account.data_used = account.data_used.checked_add(rated_event.volume)
+                        .ok_or_else(|| ChargingError::InvalidInput("Data usage would overflow".to_string()))?;
+                }
+                UsageType::Voice => {
+                    account.voice_used = account.voice_used.checked_add(rated_event.volume)
+                        .ok_or_else(|| ChargingError::InvalidInput("Voice usage would overflow".to_string()))?;
+                }
+                UsageType::SMS => {
+                    account.sms_used = account.sms_used.checked_add(rated_event.volume)
+                        .ok_or_else(|| ChargingError::InvalidInput("SMS usage would overflow".to_string()))?;
+                }
             }
             account.last_updated = Utc::now();
             
@@ -128,27 +155,59 @@ impl super::ChargingEngine {
         Ok(())
     }
 
-    pub async fn generate_invoice(&self, imsi: &str, _billing_period: &str) -> ChargingResult<f64> {
+    pub async fn generate_invoice(&self, imsi: &str, billing_period: &str) -> ChargingResult<f64> {
         let mut conn = self.redis_client.get_multiplexed_async_connection().await
             .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
 
         let pattern = format!("usage:{}:*", imsi);
-        let keys: Vec<String> = redis::AsyncCommands::keys(&mut conn, &pattern).await
-            .map_err(|e| ChargingError::RedisOperation(e.to_string()))?;
+        let mut keys: Vec<String> = Vec::new();
+        
+        // Use SCAN instead of KEYS to avoid blocking Redis
+        let mut cursor = 0u64;
+        loop {
+            let (next_cursor, batch_keys): (u64, Vec<String>) = redis::cmd("SCAN")
+                .arg(cursor)
+                .arg("MATCH")
+                .arg(&pattern)
+                .arg("COUNT")
+                .arg(100)
+                .query_async(&mut conn)
+                .await
+                .map_err(|e| ChargingError::RedisOperation(e.to_string()))?;
+            
+            keys.extend(batch_keys);
+            cursor = next_cursor;
+            
+            if cursor == 0 {
+                break;
+            }
+        }
+
+        // Parse billing period (format: YYYY-MM)
+        let (year, month) = if billing_period.len() >= 7 {
+            let parts: Vec<&str> = billing_period.split('-').collect();
+            let year = parts[0].parse::<i32>().unwrap_or(0);
+            let month = parts.get(1).and_then(|m| m.parse::<u32>().ok()).unwrap_or(0);
+            (year, month)
+        } else {
+            return Err(ChargingError::InvalidInput("Invalid billing period format. Use YYYY-MM".to_string()));
+        };
 
         let mut total_cost = 0.0;
         let mut usage_count = 0;
 
         for key in keys {
             if let Ok(Some(event)) = redis::AsyncCommands::get::<_, Option<UsageEvent>>(&mut conn, &key).await {
-                // Filter by billing period (simplified)
-                total_cost += event.cost;
-                usage_count += 1;
+                // Filter by billing period
+                if event.timestamp.year() == year && event.timestamp.month() == month {
+                    total_cost += event.cost;
+                    usage_count += 1;
+                }
             }
         }
 
-        info!("Generated invoice for IMSI {}: ${:.4} from {} usage events", 
-              imsi, total_cost, usage_count);
+        info!("Generated invoice for IMSI {}: ${:.4} from {} usage events (period: {})", 
+              imsi, total_cost, usage_count, billing_period);
         
         Ok(total_cost)
     }
@@ -158,8 +217,28 @@ impl super::ChargingEngine {
             .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
 
         let pattern = "account:*".to_string();
-        let keys: Vec<String> = redis::AsyncCommands::keys(&mut conn, &pattern).await
-            .map_err(|e| ChargingError::RedisOperation(e.to_string()))?;
+        let mut keys: Vec<String> = Vec::new();
+        
+        // Use SCAN instead of KEYS to avoid blocking Redis
+        let mut cursor = 0u64;
+        loop {
+            let (next_cursor, batch_keys): (u64, Vec<String>) = redis::cmd("SCAN")
+                .arg(cursor)
+                .arg("MATCH")
+                .arg(&pattern)
+                .arg("COUNT")
+                .arg(100)
+                .query_async(&mut conn)
+                .await
+                .map_err(|e| ChargingError::RedisOperation(e.to_string()))?;
+            
+            keys.extend(batch_keys);
+            cursor = next_cursor;
+            
+            if cursor == 0 {
+                break;
+            }
+        }
 
         let mut processed_accounts = 0;