mirror of
https://github.com/andwati/telecom-platform.git
synced 2026-10-23 05:02:17 +00:00
refactor: Extract credit management and monitoring functionality into separate modules
- Move credit operations (get_balance, add_credit, deduct_credit, check_credit) to credit_management.rs - Move user blocking operations (block_user, unblock_user, is_user_blocked) to credit_management.rs - Move subscriber account operations (get_subscriber_account, update_subscriber_account, record_usage_event) to credit_management.rs - Move monitoring operations (get_system_statistics, health_check, get_performance
This commit is contained in:
nutcas3
@@ -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<u64> {
|
||||
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<u64> {
|
||||
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<u64> {
|
||||
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<bool> {
|
||||
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<bool> {
|
||||
let mut conn = self.redis_client.get_async_connection().await
|
||||
.context("Failed to get Redis connection")?;
|
||||
|
||||
let key = format!("block:{}", ip);
|
||||
let blocked: Option<String> = 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<Option<SubscriberAccount>> {
|
||||
let mut conn = self.redis_client.get_async_connection().await
|
||||
.context("Failed to get Redis connection")?;
|
||||
|
||||
let key = format!("account:{}", imsi);
|
||||
let account: Option<SubscriberAccount> = 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(())
|
||||
}
|
||||
}
|
||||
@@ -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<SystemStats> {
|
||||
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<u64> {
|
||||
// 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<HealthStatus> {
|
||||
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<u64> {
|
||||
// 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<PerformanceMetrics> {
|
||||
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::<u64>().ok())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
async fn calculate_rps(&self) -> ChargingResult<f64> {
|
||||
// This would calculate actual requests per second
|
||||
// For now, return a placeholder
|
||||
Ok(100.0)
|
||||
}
|
||||
|
||||
async fn calculate_avg_response_time(&self) -> ChargingResult<f64> {
|
||||
// This would calculate actual average response time
|
||||
// For now, return a placeholder
|
||||
Ok(5.0) // 5ms
|
||||
}
|
||||
|
||||
pub async fn get_error_statistics(&self) -> ChargingResult<ErrorStats> {
|
||||
let mut conn = self.redis_client.get_async_connection().await
|
||||
.context("Failed to get Redis connection")?;
|
||||
|
||||
let pattern = "error:*".to_string();
|
||||
let keys: Vec<String> = 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<Option<String>> {
|
||||
let mut conn = self.redis_client.get_async_connection().await
|
||||
.context("Failed to get Redis connection")?;
|
||||
|
||||
let last_error: Option<String> = 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<chrono::Utc>,
|
||||
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<String, u64>,
|
||||
pub last_error: Option<String>,
|
||||
}
|
||||
@@ -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<String> = 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<String> = 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<String> = 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<String> = 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<SystemStats> {
|
||||
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<u64> {
|
||||
// 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<HealthStatus> {
|
||||
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<u64> {
|
||||
// This would get actual memory usage in a real implementation
|
||||
// For now, return a placeholder
|
||||
Ok(50_000_000) // 50MB placeholder
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<chrono::Utc>,
|
||||
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<String, u64>,
|
||||
pub last_error: Option<String>,
|
||||
}
|
||||
@@ -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<f64> {
|
||||
// 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<UsageEvent> {
|
||||
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<Vec<String>> {
|
||||
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<f64> {
|
||||
// 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<String> = 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<UsageEvent>>(&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<u32> {
|
||||
let mut conn = self.redis_client.get_async_connection().await
|
||||
.context("Failed to get Redis connection")?;
|
||||
|
||||
let pattern = "account:*".to_string();
|
||||
let keys: Vec<String> = 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<crate::charging_types::SubscriberAccount>>(&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)
|
||||
}
|
||||
}
|
||||
@@ -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<String> = 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<String> = 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<String> = 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<String> = 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(())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user