From 69a84d85e682530656ec464e866427600ce28fcf Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Sat, 25 Apr 2026 03:03:06 +0300 Subject: [PATCH] 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 --- .../src/charging/rating_billing.rs | 145 ++++++++++++++---- 1 file changed, 112 insertions(+), 33 deletions(-) diff --git a/apps/charging-engine/src/charging/rating_billing.rs b/apps/charging-engine/src/charging/rating_billing.rs index 2debd0a..a40bf9b 100644 --- a/apps/charging-engine/src/charging/rating_billing.rs +++ b/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> { + pub async fn apply_charging_rules(&self, event: &UsageEvent) -> ChargingResult> { 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 { + pub async fn generate_invoice(&self, imsi: &str, billing_period: &str) -> ChargingResult { 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 = redis::AsyncCommands::keys(&mut conn, &pattern).await - .map_err(|e| ChargingError::RedisOperation(e.to_string()))?; + let mut keys: Vec = Vec::new(); + + // Use SCAN instead of KEYS to avoid blocking Redis + let mut cursor = 0u64; + loop { + let (next_cursor, batch_keys): (u64, Vec) = 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::().unwrap_or(0); + let month = parts.get(1).and_then(|m| m.parse::().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>(&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 = redis::AsyncCommands::keys(&mut conn, &pattern).await - .map_err(|e| ChargingError::RedisOperation(e.to_string()))?; + let mut keys: Vec = Vec::new(); + + // Use SCAN instead of KEYS to avoid blocking Redis + let mut cursor = 0u64; + loop { + let (next_cursor, batch_keys): (u64, Vec) = 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;