diff --git a/apps/charging-engine/src/credit_management.rs b/apps/charging-engine/src/credit_management.rs new file mode 100644 index 0000000..b10ac3e --- /dev/null +++ b/apps/charging-engine/src/credit_management.rs @@ -0,0 +1,138 @@ +use redis::AsyncCommands; +use tracing::{info, warn, error, debug}; + +use crate::charging_types::{SubscriberAccount, UsageEvent}; +use crate::errors::{ChargingError, ChargingResult, ErrorContext}; + +impl super::ChargingEngine { + pub async fn get_balance(&self, ip: &str) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("credit:{}", ip); + let balance: u64 = conn.get(&key).await + .context("Failed to get balance from Redis")?; + + debug!("Retrieved balance {} for IP: {}", balance, ip); + Ok(balance) + } + + pub async fn add_credit(&self, ip: &str, bytes_to_add: u64) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("credit:{}", ip); + let new_balance: u64 = conn.incr(&key, bytes_to_add).await + .context("Failed to add credit in Redis")?; + + // Set expiration if this is a new key + let _: () = conn.expire(&key, 86400).await.unwrap_or(()); + + info!("Added {} bytes credit to IP: {}, new balance: {}", bytes_to_add, ip, new_balance); + Ok(new_balance) + } + + pub async fn deduct_credit(&self, ip: &str, bytes_used: u64) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("credit:{}", ip); + + // Check current balance first + let current_balance: u64 = conn.get(&key).await.unwrap_or(0); + + if current_balance < bytes_used { + return Err(ChargingError::InsufficientCredit { + available: current_balance, + requested: bytes_used, + }); + } + + let new_balance: u64 = conn.decr(&key, bytes_used).await + .context("Failed to deduct credit in Redis")?; + + info!("Deducted {} bytes credit from IP: {}, new balance: {}", bytes_used, ip, new_balance); + Ok(new_balance) + } + + pub async fn check_credit(&self, ip: &str, bytes_requested: u64) -> ChargingResult { + let current_balance = self.get_balance(ip).await.unwrap_or(0); + Ok(current_balance >= bytes_requested) + } + + pub async fn block_user(&self, ip: &str) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("block:{}", ip); + let _: () = conn.set(&key, "blocked").await + .context("Failed to block user in Redis")?; + + // Set expiration for block (24 hours) + let _: () = conn.expire(&key, 86400).await.unwrap_or(()); + + warn!("User IP: {} has been blocked", ip); + Ok(()) + } + + pub async fn unblock_user(&self, ip: &str) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("block:{}", ip); + let _: () = conn.del(&key).await + .context("Failed to unblock user in Redis")?; + + info!("User IP: {} has been unblocked", ip); + Ok(()) + } + + pub async fn is_user_blocked(&self, ip: &str) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("block:{}", ip); + let blocked: Option = conn.get(&key).await + .context("Failed to check block status in Redis")?; + + Ok(blocked.is_some()) + } + + pub async fn get_subscriber_account(&self, imsi: &str) -> ChargingResult> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("account:{}", imsi); + let account: Option = conn.get(&key).await + .context("Failed to get subscriber account from Redis")?; + + Ok(account) + } + + pub async fn update_subscriber_account(&self, account: &SubscriberAccount) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("account:{}", account.imsi); + let _: () = conn.set(&key, account).await + .context("Failed to update subscriber account in Redis")?; + + debug!("Updated subscriber account for IMSI: {}", account.imsi); + Ok(()) + } + + pub async fn record_usage_event(&self, event: &UsageEvent) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let key = format!("usage:{}:{}", event.imsi, event.session_id); + let _: () = conn.set(&key, event).await + .context("Failed to record usage event in Redis")?; + + // Set expiration for usage events (7 days) + let _: () = conn.expire(&key, 604800).await.unwrap_or(()); + + info!("Recorded usage event for IMSI: {}, cost: ${:.4}", event.imsi, event.cost); + Ok(()) + } +} diff --git a/apps/charging-engine/src/monitoring.rs b/apps/charging-engine/src/monitoring.rs new file mode 100644 index 0000000..9d52b2e --- /dev/null +++ b/apps/charging-engine/src/monitoring.rs @@ -0,0 +1,171 @@ +use tracing::{info, debug}; +use std::time::SystemTime; + +use crate::errors::{ChargingError, ChargingResult, ErrorContext}; + +impl super::ChargingEngine { + pub async fn get_system_statistics(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let active_sessions: u64 = conn.get("stats:active_sessions").await.unwrap_or(0); + let total_accounts: u64 = conn.keys("account:*").await.unwrap_or_default().len() as u64; + let blocked_users: u64 = conn.keys("block:*").await.unwrap_or_default().len() as u64; + let low_balance_alerts: u64 = conn.keys("alert:low_balance:*").await.unwrap_or_default().len() as u64; + + let stats = SystemStats { + active_sessions, + total_accounts, + blocked_users, + low_balance_alerts, + uptime: self.get_uptime().await?, + }; + + Ok(stats) + } + + async fn get_uptime(&self) -> ChargingResult { + // This would track the actual uptime in a real implementation + // For now, return a placeholder + Ok(SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default() + .as_secs()) + } + + pub async fn health_check(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + // Test Redis connection + let _: String = conn.get("health_check").await.unwrap_or_else(|_| "ok".to_string()); + + let status = HealthStatus { + redis_connected: true, + active_sync: true, + last_sync: chrono::Utc::now(), + memory_usage: self.get_memory_usage().await?, + }; + + Ok(status) + } + + async fn get_memory_usage(&self) -> ChargingResult { + // This would get actual memory usage in a real implementation + // For now, return a placeholder + Ok(50_000_000) // 50MB placeholder + } + + pub async fn get_performance_metrics(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + // Get Redis info + let info: String = redis::cmd("INFO").query_async(&mut conn).await.unwrap_or_default(); + + // Parse Redis info for metrics (simplified) + let connected_clients = self.extract_metric(&info, "connected_clients"); + let used_memory = self.extract_metric(&info, "used_memory"); + let total_commands_processed = self.extract_metric(&info, "total_commands_processed"); + + let metrics = PerformanceMetrics { + connected_clients, + used_memory, + total_commands_processed, + requests_per_second: self.calculate_rps().await?, + average_response_time: self.calculate_avg_response_time().await?, + }; + + Ok(metrics) + } + + fn extract_metric(&self, info: &str, metric: &str) -> u64 { + info.lines() + .find(|line| line.starts_with(metric)) + .and_then(|line| line.split(':').nth(1)) + .and_then(|value| value.parse::().ok()) + .unwrap_or(0) + } + + async fn calculate_rps(&self) -> ChargingResult { + // This would calculate actual requests per second + // For now, return a placeholder + Ok(100.0) + } + + async fn calculate_avg_response_time(&self) -> ChargingResult { + // This would calculate actual average response time + // For now, return a placeholder + Ok(5.0) // 5ms + } + + pub async fn get_error_statistics(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = "error:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get error keys")?; + + let mut total_errors = 0u64; + let mut error_types = std::collections::HashMap::new(); + + for key in keys { + if let Ok(count) = conn.get::<_, u64>(&key).await { + total_errors += count; + + let error_type = key.split(':').nth(1).unwrap_or("unknown"); + *error_types.entry(error_type.to_string()).or_insert(0) += count; + } + } + + let stats = ErrorStats { + total_errors, + error_types, + last_error: self.get_last_error().await?, + }; + + Ok(stats) + } + + async fn get_last_error(&self) -> ChargingResult> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let last_error: Option = conn.get("last_error").await.unwrap_or(None); + Ok(last_error) + } +} + +#[derive(Debug, Clone)] +pub struct SystemStats { + pub active_sessions: u64, + pub total_accounts: u64, + pub blocked_users: u64, + pub low_balance_alerts: u64, + pub uptime: u64, +} + +#[derive(Debug, Clone)] +pub struct HealthStatus { + pub redis_connected: bool, + pub active_sync: bool, + pub last_sync: chrono::DateTime, + pub memory_usage: u64, +} + +#[derive(Debug, Clone)] +pub struct PerformanceMetrics { + pub connected_clients: u64, + pub used_memory: u64, + pub total_commands_processed: u64, + pub requests_per_second: f64, + pub average_response_time: f64, +} + +#[derive(Debug, Clone)] +pub struct ErrorStats { + pub total_errors: u64, + pub error_types: std::collections::HashMap, + pub last_error: Option, +} diff --git a/apps/charging-engine/src/monitoring_sync.rs b/apps/charging-engine/src/monitoring_sync.rs new file mode 100644 index 0000000..5f06134 --- /dev/null +++ b/apps/charging-engine/src/monitoring_sync.rs @@ -0,0 +1,195 @@ +use tokio::time::{interval, sleep}; +use tracing::{info, warn, error, debug}; +use std::time::Duration; + +use crate::errors::{ChargingError, ChargingResult, ErrorContext}; + +impl super::ChargingEngine { + pub async fn start_background_sync(&self) -> ChargingResult<()> { + let mut interval_timer = interval(self.sync_interval); + + info!("Starting background sync with interval: {:?}", self.sync_interval); + + loop { + interval_timer.tick().await; + + if let Err(e) = self.perform_sync().await { + error!("Background sync failed: {}", e); + } + } + } + + async fn perform_sync(&self) -> ChargingResult<()> { + debug!("Performing background sync"); + + // Sync 1: Clean up expired blocks + self.cleanup_expired_blocks().await?; + + // Sync 2: Update usage statistics + self.update_usage_statistics().await?; + + // Sync 3: Check for low balance alerts + self.check_low_balance_alerts().await?; + + // Sync 4: Apply monthly fees (once per day) + self.apply_monthly_fees_if_needed().await?; + + debug!("Background sync completed successfully"); + Ok(()) + } + + async fn cleanup_expired_blocks(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = "block:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get block keys")?; + + let mut cleaned_count = 0; + + for key in keys { + let ttl: i64 = conn.ttl(&key).await.unwrap_or(-1); + if ttl == -1 || ttl == -2 { + // Key has no expiration or doesn't exist + let _: () = conn.del(&key).await.unwrap_or(()); + cleaned_count += 1; + } + } + + if cleaned_count > 0 { + info!("Cleaned up {} expired blocks", cleaned_count); + } + + Ok(()) + } + + async fn update_usage_statistics(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + // Get total active sessions + let pattern = "session:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get session keys")?; + + let active_sessions = keys.len(); + + // Store statistics + let stats_key = "stats:active_sessions"; + let _: () = conn.set(stats_key, active_sessions).await.unwrap_or(()); + + debug!("Updated usage statistics: {} active sessions", active_sessions); + Ok(()) + } + + async fn check_low_balance_alerts(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = "credit:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get credit keys")?; + + let low_balance_threshold = 100_000_000; // 100MB in bytes + let mut low_balance_count = 0; + + for key in keys { + if let Ok(balance) = conn.get::<_, u64>(&key).await { + if balance < low_balance_threshold { + low_balance_count += 1; + warn!("Low balance alert for {}: {} bytes remaining", key, balance); + + // Store alert + let alert_key = format!("alert:low_balance:{}", key); + let _: () = conn.setex(&alert_key, 3600, "low_balance").await.unwrap_or(()); + } + } + } + + if low_balance_count > 0 { + info!("Generated {} low balance alerts", low_balance_count); + } + + Ok(()) + } + + async fn apply_monthly_fees_if_needed(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let last_fee_key = "last_monthly_fee"; + let last_fee: Option = conn.get(last_fee_key).await.unwrap_or(None); + + let now = chrono::Utc::now(); + let current_month = format!("{}-{}", now.year(), now.month()); + + if let Some(last_month) = last_fee { + if last_month == current_month { + return Ok(()); // Monthly fees already applied this month + } + } + + // Apply monthly fees + let processed = self.apply_monthly_fees().await?; + + // Update last fee timestamp + let _: () = conn.set(last_fee_key, current_month).await.unwrap_or(()); + + info!("Applied monthly fees to {} accounts for {}", processed, current_month); + Ok(()) + } + + pub async fn get_system_statistics(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let active_sessions: u64 = conn.get("stats:active_sessions").await.unwrap_or(0); + let total_accounts: u64 = conn.keys("account:*").await.unwrap_or_default().len() as u64; + let blocked_users: u64 = conn.keys("block:*").await.unwrap_or_default().len() as u64; + let low_balance_alerts: u64 = conn.keys("alert:low_balance:*").await.unwrap_or_default().len() as u64; + + let stats = SystemStats { + active_sessions, + total_accounts, + blocked_users, + low_balance_alerts, + uptime: self.get_uptime().await?, + }; + + Ok(stats) + } + + async fn get_uptime(&self) -> ChargingResult { + // This would track the actual uptime in a real implementation + // For now, return a placeholder + Ok(std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs()) + } + + pub async fn health_check(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + // Test Redis connection + let _: String = conn.get("health_check").await.unwrap_or_else(|_| "ok".to_string()); + + let status = HealthStatus { + redis_connected: true, + active_sync: true, + last_sync: chrono::Utc::now(), + memory_usage: self.get_memory_usage().await?, + }; + + Ok(status) + } + + async fn get_memory_usage(&self) -> ChargingResult { + // This would get actual memory usage in a real implementation + // For now, return a placeholder + Ok(50_000_000) // 50MB placeholder + } +} + diff --git a/apps/charging-engine/src/monitoring_types.rs b/apps/charging-engine/src/monitoring_types.rs new file mode 100644 index 0000000..b67c0a4 --- /dev/null +++ b/apps/charging-engine/src/monitoring_types.rs @@ -0,0 +1,32 @@ +#[derive(Debug, Clone)] +pub struct SystemStats { + pub active_sessions: u64, + pub total_accounts: u64, + pub blocked_users: u64, + pub low_balance_alerts: u64, + pub uptime: u64, +} + +#[derive(Debug, Clone)] +pub struct HealthStatus { + pub redis_connected: bool, + pub active_sync: bool, + pub last_sync: chrono::DateTime, + pub memory_usage: u64, +} + +#[derive(Debug, Clone)] +pub struct PerformanceMetrics { + pub connected_clients: u64, + pub used_memory: u64, + pub total_commands_processed: u64, + pub requests_per_second: f64, + pub average_response_time: f64, +} + +#[derive(Debug, Clone)] +pub struct ErrorStats { + pub total_errors: u64, + pub error_types: std::collections::HashMap, + pub last_error: Option, +} diff --git a/apps/charging-engine/src/rating_billing.rs b/apps/charging-engine/src/rating_billing.rs new file mode 100644 index 0000000..4dbf06c --- /dev/null +++ b/apps/charging-engine/src/rating_billing.rs @@ -0,0 +1,179 @@ +use chrono::Utc; +use tracing::{info, debug, warn}; + +use crate::charging_types::{UsageEvent, UsageType, RatingPlan}; +use crate::errors::{ChargingError, ChargingResult, ErrorContext}; + +impl super::ChargingEngine { + pub async fn calculate_usage_cost(&self, event: &UsageEvent) -> ChargingResult { + // Get subscriber account to determine rating plan + let account = self.get_subscriber_account(&event.imsi).await?; + let account = account.ok_or_else(|| ChargingError::SubscriberNotFound(event.imsi.clone()))?; + + // Get rating plan (simplified - in real implementation this would be more complex) + let plan = self.get_rating_plan("basic") // Default to basic plan for now + .ok_or_else(|| ChargingError::RatingPlanNotFound("basic".to_string()))?; + + 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 + } + UsageType::Voice => { + let voice_minutes = event.volume as f64 / 60.0; // Convert seconds to minutes + voice_minutes * plan.voice_rate + } + UsageType::SMS => { + event.volume as f64 * plan.sms_rate + } + }; + + debug!("Calculated cost for IMSI {}: ${:.4}", event.imsi, cost); + Ok(cost) + } + + pub async fn rate_usage(&self, mut event: UsageEvent) -> ChargingResult { + event.cost = self.calculate_usage_cost(&event).await?; + + // Update rate field + if event.volume > 0 { + event.rate = event.cost / event.volume as f64; + } + + // Record the rated event + self.record_usage_event(&event).await?; + + info!("Rated usage event for IMSI {}: ${:.4}", event.imsi, event.cost); + Ok(event) + } + + 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()); + warn!("Insufficient credit for IMSI: {}", event.imsi); + } + + // Rule 2: Check usage limits + let account = self.get_subscriber_account(&event.imsi).await?; + 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()); + 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()); + 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()); + warn!("SMS limit exceeded for IMSI: {}", event.imsi); + } + } + } + } + + // Rule 3: Check if user is blocked + if self.is_user_blocked(&event.imsi).await? { + applied_rules.push("USER_BLOCKED".to_string()); + warn!("Blocked user attempted usage: {}", event.imsi); + } + + if applied_rules.is_empty() { + applied_rules.push("ALLOWED".to_string()); + } + + Ok(applied_rules) + } + + pub async fn process_usage_event(&self, event: UsageEvent) -> ChargingResult<()> { + // Apply charging rules + 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(", "))); + } + + // Rate the usage + let rated_event = self.rate_usage(event).await?; + + // 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, + } + account.last_updated = Utc::now(); + + self.update_subscriber_account(&account).await?; + } + + info!("Processed usage event for IMSI: {}", rated_event.imsi); + Ok(()) + } + + pub async fn generate_invoice(&self, imsi: &str, billing_period: &str) -> ChargingResult { + // Get all usage events for the billing period + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = format!("usage:{}:*", imsi); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get usage keys")?; + + let mut total_cost = 0.0; + let mut usage_count = 0; + + for key in keys { + if let Ok(Some(event)) = conn.get::<_, Option>(&key).await { + // Filter by billing period (simplified) + total_cost += event.cost; + usage_count += 1; + } + } + + info!("Generated invoice for IMSI {}: ${:.4} from {} usage events", + imsi, total_cost, usage_count); + + Ok(total_cost) + } + + pub async fn apply_monthly_fees(&self) -> ChargingResult { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = "account:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get account keys")?; + + let mut processed_accounts = 0; + + for key in keys { + if let Ok(Some(mut account)) = conn.get::<_, Option>(&key).await { + // Get rating plan and apply monthly fee + if let Some(plan) = self.get_rating_plan("basic") { // Simplified + account.balance -= (plan.monthly_fee * 100.0) as i64; // Convert to cents + account.last_updated = Utc::now(); + + let _: () = conn.set(&key, &account).await.unwrap_or(()); + processed_accounts += 1; + } + } + } + + info!("Applied monthly fees to {} accounts", processed_accounts); + Ok(processed_accounts) + } +} diff --git a/apps/charging-engine/src/sync_operations.rs b/apps/charging-engine/src/sync_operations.rs new file mode 100644 index 0000000..b1cf9b8 --- /dev/null +++ b/apps/charging-engine/src/sync_operations.rs @@ -0,0 +1,142 @@ +use tokio::time::interval; +use tracing::{info, warn, error, debug}; +use std::time::Duration; + +use crate::errors::{ChargingError, ChargingResult, ErrorContext}; + +impl super::ChargingEngine { + pub async fn start_background_sync(&self) -> ChargingResult<()> { + let mut interval_timer = interval(self.sync_interval); + + info!("Starting background sync with interval: {:?}", self.sync_interval); + + loop { + interval_timer.tick().await; + + if let Err(e) = self.perform_sync().await { + error!("Background sync failed: {}", e); + } + } + } + + async fn perform_sync(&self) -> ChargingResult<()> { + debug!("Performing background sync"); + + // Sync 1: Clean up expired blocks + self.cleanup_expired_blocks().await?; + + // Sync 2: Update usage statistics + self.update_usage_statistics().await?; + + // Sync 3: Check for low balance alerts + self.check_low_balance_alerts().await?; + + // Sync 4: Apply monthly fees (once per day) + self.apply_monthly_fees_if_needed().await?; + + debug!("Background sync completed successfully"); + Ok(()) + } + + async fn cleanup_expired_blocks(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = "block:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get block keys")?; + + let mut cleaned_count = 0; + + for key in keys { + let ttl: i64 = conn.ttl(&key).await.unwrap_or(-1); + if ttl == -1 || ttl == -2 { + // Key has no expiration or doesn't exist + let _: () = conn.del(&key).await.unwrap_or(()); + cleaned_count += 1; + } + } + + if cleaned_count > 0 { + info!("Cleaned up {} expired blocks", cleaned_count); + } + + Ok(()) + } + + async fn update_usage_statistics(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + // Get total active sessions + let pattern = "session:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get session keys")?; + + let active_sessions = keys.len(); + + // Store statistics + let stats_key = "stats:active_sessions"; + let _: () = conn.set(stats_key, active_sessions).await.unwrap_or(()); + + debug!("Updated usage statistics: {} active sessions", active_sessions); + Ok(()) + } + + async fn check_low_balance_alerts(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let pattern = "credit:*".to_string(); + let keys: Vec = conn.keys(&pattern).await + .context("Failed to get credit keys")?; + + let low_balance_threshold = 100_000_000; // 100MB in bytes + let mut low_balance_count = 0; + + for key in keys { + if let Ok(balance) = conn.get::<_, u64>(&key).await { + if balance < low_balance_threshold { + low_balance_count += 1; + warn!("Low balance alert for {}: {} bytes remaining", key, balance); + + // Store alert + let alert_key = format!("alert:low_balance:{}", key); + let _: () = conn.setex(&alert_key, 3600, "low_balance").await.unwrap_or(()); + } + } + } + + if low_balance_count > 0 { + info!("Generated {} low balance alerts", low_balance_count); + } + + Ok(()) + } + + async fn apply_monthly_fees_if_needed(&self) -> ChargingResult<()> { + let mut conn = self.redis_client.get_async_connection().await + .context("Failed to get Redis connection")?; + + let last_fee_key = "last_monthly_fee"; + let last_fee: Option = conn.get(last_fee_key).await.unwrap_or(None); + + let now = chrono::Utc::now(); + let current_month = format!("{}-{}", now.year(), now.month()); + + if let Some(last_month) = last_fee { + if last_month == current_month { + return Ok(()); // Monthly fees already applied this month + } + } + + // Apply monthly fees + let processed = self.apply_monthly_fees().await?; + + // Update last fee timestamp + let _: () = conn.set(last_fee_key, current_month).await.unwrap_or(()); + + info!("Applied monthly fees to {} accounts for {}", processed, current_month); + Ok(()) + } +}