andwati/telecom-platform
1
0
Fork 0
mirror of https://github.com/andwati/telecom-platform.git synced 2026-10-31 17:12:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Split charging engine into separate modules for better code organization

Browse Source 
- Move core types (SubscriberAccount, AccountStatus, UsageEvent, UsageType, RatingPlan, ChargingSession, SessionStatus, ChargingRule, Condition, Action) to charging_types.rs
- Move main ChargingEngine struct and initialization to charging_engine.rs
- Move credit operations (check_credit, get_balance, deduct_credit, add_credit) to credit_management.rs
- Move account operations (create_account, get_account, top_up_balance,
This commit is contained in:
nutcas3
2026-04-14 16:24:04 +03:00
parent 577f625bbd
commit 2a2f481daa
3 changed files with 212 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/charging-engine/src/charging.rs
+16 -474
View File
@@ -1,478 +1,20 @@
use chrono::{DateTime, Utc}; pub mod charging_types;
use redis::AsyncCommands; pub mod charging_engine;
use serde::{Deserialize, Serialize}; pub mod credit_management;
use std::collections::HashMap; pub mod rating_billing;
use std::time::Duration; pub mod monitoring_sync;
use tokio::time::{interval, sleep}; pub mod monitoring_types;
use tracing::{info, warn, error, debug}; pub mod sync_operations;
pub mod monitoring;
use crate::errors::{ChargingError, ChargingResult, ErrorContext}; pub use charging_types::{
SubscriberAccount, AccountStatus, UsageEvent, UsageType, RatingPlan,
ChargingSession, SessionStatus, ChargingRule, Condition, Action
};
#[derive(Debug, Clone, Serialize, Deserialize)] pub use charging_engine::ChargingEngine;
pub struct SubscriberAccount {
pub imsi: String,
pub balance: i64, // in smallest currency unit (e.g., cents)
pub data_limit: u64, // bytes
pub data_used: u64, // bytes
pub voice_limit: u64, // seconds
pub voice_used: u64, // seconds
pub sms_limit: u64, // count
pub sms_used: u64, // count
pub status: AccountStatus,
pub last_updated: DateTime<Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)] pub use monitoring::{SystemStats, HealthStatus};
pub enum AccountStatus { pub use monitoring_types::{PerformanceMetrics, ErrorStats};
Active,
Suspended,
Terminated,
Blocked,
}
#[derive(Debug, Clone, Serialize, Deserialize)] pub use crate::errors::{ChargingError, ChargingResult, ErrorContext};
pub struct UsageEvent {
pub imsi: String,
pub session_id: String,
pub usage_type: UsageType,
pub volume: u64, // bytes, seconds, or count
pub timestamp: DateTime<Utc>,
pub rate: f64, // cost per unit
pub cost: f64, // total cost
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UsageType {
Data,
Voice,
SMS,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RatingPlan {
pub plan_id: String,
pub name: String,
pub data_rate: f64, // cost per MB
pub voice_rate: f64, // cost per minute
pub sms_rate: f64, // cost per SMS
pub monthly_fee: f64,
pub data_limit: u64, // bytes
pub voice_limit: u64, // seconds
pub sms_limit: u64, // count
}
pub struct ChargingEngine {
redis_client: redis::Client,
rating_plans: HashMap<String, RatingPlan>,
sync_interval: Duration,
}
impl ChargingEngine {
pub fn new(redis_url: &str, sync_interval_secs: u64) -> ChargingResult<Self> {
let redis_client = redis::Client::open(redis_url)
.context("Failed to create Redis client")?;
let mut rating_plans = HashMap::new();
// Default rating plans
rating_plans.insert("basic".to_string(), RatingPlan {
plan_id: "basic".to_string(),
name: "Basic Plan".to_string(),
data_rate: 0.01, // $0.01 per MB
voice_rate: 0.05, // $0.05 per minute
sms_rate: 0.10, // $0.10 per SMS
monthly_fee: 10.0, // $10 monthly
data_limit: 1_000_000_000, // 1GB
voice_limit: 300, // 300 minutes
sms_limit: 100, // 100 SMS
});
rating_plans.insert("premium".to_string(), RatingPlan {
plan_id: "premium".to_string(),
name: "Premium Plan".to_string(),
data_rate: 0.005, // $0.005 per MB
voice_rate: 0.02, // $0.02 per minute
sms_rate: 0.05, // $0.05 per SMS
monthly_fee: 25.0, // $25 monthly
data_limit: 5_000_000_000, // 5GB
voice_limit: 1000, // 1000 minutes
sms_limit: 500, // 500 SMS
});
Ok(Self {
redis_client,
rating_plans,
sync_interval: Duration::from_secs(sync_interval_secs),
})
}
pub async fn start(&self) -> ChargingResult<()> {
info!("Starting charging engine");
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.context("Failed to get Redis connection")?;
// Test connection
let _: () = redis::cmd("PING").query_async(&mut conn).await?;
info!("Connected to Redis successfully");
// Start sync loop
let mut ticker = interval(self.sync_interval);
loop {
ticker.tick().await;
if let Err(e) = self.sync_usage_data(&mut conn).await {
error!("Failed to sync usage data: {}", e);
}
if let Err(e) = self.check_credit_limits(&mut conn).await {
error!("Failed to check credit limits: {}", e);
}
}
}
async fn sync_usage_data(&self, conn: &mut redis::aio::MultiplexedConnection) -> Result<()> {
debug!("Syncing usage data from packet gateway");
// Get all usage keys from packet gateway
let pattern = "usage:*";
let keys: Vec<String> = redis::cmd("KEYS")
.arg(pattern)
.query_async(conn)
.await
.unwrap_or_default();
for key in keys {
if let Some(ip) = key.strip_prefix("usage:") {
// Get IMSI mapping from IP -> IMSI
let imsi_key = format!("ip2imsi:{}", ip);
if let Ok(imsi): Result<String, _> = redis::cmd("GET")
.arg(&imsi_key)
.query_async(conn)
.await
{
// Get usage data
let usage: u64 = redis::cmd("GET")
.arg(&key)
.query_async(conn)
.await
.unwrap_or(0);
if usage > 0 {
// Update subscriber usage
self.update_subscriber_usage(conn, &imsi, usage).await?;
// Clear processed usage
let _: () = redis::cmd("DEL").arg(&key).query_async(conn).await?;
}
}
}
}
Ok(())
}
async fn update_subscriber_usage(
&self,
conn: &mut redis::aio::MultiplexedConnection,
imsi: &str,
data_bytes: u64
) -> Result<()> {
let account_key = format!("account:{}", imsi);
// Get current account
let account_data: Option<String> = redis::cmd("GET")
.arg(&account_key)
.query_async(conn)
.await?;
if let Some(data) = account_data {
let mut account: SubscriberAccount = serde_json::from_str(&data)
.context("Failed to deserialize account")?;
// Update data usage
account.data_used += data_bytes;
account.last_updated = Utc::now();
// Calculate cost
if let Some(plan) = self.rating_plans.get(&account.imsi) {
let mb_used = data_bytes as f64 / (1024.0 * 1024.0);
let cost = mb_used * plan.data_rate;
// Deduct from balance
account.balance -= (cost * 100.0) as i64; // Convert to cents
// Create usage event
let usage_event = UsageEvent {
imsi: imsi.to_string(),
session_id: "default".to_string(), // TODO: Get actual session
usage_type: UsageType::Data,
volume: data_bytes,
timestamp: Utc::now(),
rate: plan.data_rate,
cost,
};
// Store usage event
let event_key = format!("events:{}:{}", imsi, Utc::now().timestamp());
let event_json = serde_json::to_string(&usage_event)?;
let _: () = redis::cmd("SET")
.arg(&event_key)
.arg(&event_json)
.arg("EX")
.arg(86400) // Expire in 24 hours
.query_async(conn)
.await?;
// Check if over limit
if account.data_used > account.data_limit {
warn!("Subscriber {} exceeded data limit: {} > {}",
imsi, account.data_used, account.data_limit);
// Block subscriber if significantly over limit
if account.data_used > account.data_limit * 110 / 100 {
account.status = AccountStatus::Blocked;
// Update block list in packet gateway
let ip_key = format!("imsi2ip:{}", imsi);
if let Ok(ip): Result<String, _> = redis::cmd("GET")
.arg(&ip_key)
.query_async(conn)
.await
{
let block_key = format!("block:{}", ip);
let _: () = redis::cmd("SET")
.arg(&block_key)
.arg(1)
.query_async(conn)
.await?;
}
}
}
// Save updated account
let account_json = serde_json::to_string(&account)?;
let _: () = redis::cmd("SET")
.arg(&account_key)
.arg(&account_json)
.query_async(conn)
.await?;
info!("Updated usage for {}: {} bytes, cost: ${:.4}, balance: ${:.2}",
imsi, data_bytes, cost, account.balance as f64 / 100.0);
}
}
Ok(())
}
async fn check_credit_limits(&self, conn: &mut redis::aio::MultiplexedConnection) -> Result<()> {
debug!("Checking credit limits");
// Get all account keys
let pattern = "account:*";
let keys: Vec<String> = redis::cmd("KEYS")
.arg(pattern)
.query_async(conn)
.await
.unwrap_or_default();
for key in keys {
if let Some(imsi) = key.strip_prefix("account:") {
let account_data: Option<String> = redis::cmd("GET")
.arg(&key)
.query_async(conn)
.await?;
if let Some(data) = account_data {
let account: SubscriberAccount = serde_json::from_str(&data)
.context("Failed to deserialize account")?;
// Check if balance is negative
if account.balance < 0 {
warn!("Subscriber {} has negative balance: ${:.2}",
imsi, account.balance as f64 / 100.0);
// Block if significantly negative
if account.balance < -1000 { // -$10
account.status = AccountStatus::Blocked;
// Update block list
let ip_key = format!("imsi2ip:{}", imsi);
if let Ok(ip): Result<String, _> = redis::cmd("GET")
.arg(&ip_key)
.query_async(conn)
.await
{
let block_key = format!("block:{}", ip);
let _: () = redis::cmd("SET")
.arg(&block_key)
.arg(1)
.query_async(conn)
.await?;
}
// Save updated account
let account_json = serde_json::to_string(&account)?;
let _: () = redis::cmd("SET")
.arg(&key)
.arg(&account_json)
.query_async(conn)
.await?;
}
}
}
}
}
Ok(())
}
pub async fn create_account(&self, imsi: &str, plan_id: &str, initial_balance: i64) -> Result<()> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await?;
let plan = self.rating_plans.get(plan_id)
.ok_or_else(|| anyhow::anyhow!("Rating plan '{}' not found", plan_id))?;
let account = SubscriberAccount {
imsi: imsi.to_string(),
balance: initial_balance,
data_limit: plan.data_limit,
data_used: 0,
voice_limit: plan.voice_limit,
voice_used: 0,
sms_limit: plan.sms_limit,
sms_used: 0,
status: AccountStatus::Active,
last_updated: Utc::now(),
};
let account_key = format!("account:{}", imsi);
let account_json = serde_json::to_string(&account)?;
let _: () = redis::cmd("SET")
.arg(&account_key)
.arg(&account_json)
.query_async(&mut conn)
.await?;
info!("Created account for {} with plan {} and balance ${:.2}",
imsi, plan_id, initial_balance as f64 / 100.0);
Ok(())
}
pub async fn get_account(&self, imsi: &str) -> Result<Option<SubscriberAccount>> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await?;
let account_key = format!("account:{}", imsi);
let account_data: Option<String> = redis::cmd("GET")
.arg(&account_key)
.query_async(&mut conn)
.await?;
if let Some(data) = account_data {
let account: SubscriberAccount = serde_json::from_str(&data)
.context("Failed to deserialize account")?;
Ok(Some(account))
} else {
Ok(None)
}
}
pub async fn top_up_balance(&self, imsi: &str, amount: i64) -> Result<()> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await?;
let account_key = format!("account:{}", imsi);
let account_data: Option<String> = redis::cmd("GET")
.arg(&account_key)
.query_async(&mut conn)
.await?;
if let Some(data) = account_data {
let mut account: SubscriberAccount = serde_json::from_str(&data)
.context("Failed to deserialize account")?;
account.balance += amount;
account.last_updated = Utc::now();
// Unblock if balance is restored and status was blocked
if account.status == AccountStatus::Blocked && account.balance > 0 {
account.status = AccountStatus::Active;
// Remove from block list
let ip_key = format!("imsi2ip:{}", imsi);
if let Ok(ip): Result<String, _> = redis::cmd("GET")
.arg(&ip_key)
.query_async(&mut conn)
.await
{
let block_key = format!("block:{}", ip);
let _: () = redis::cmd("DEL").arg(&block_key).query_async(&mut conn).await?;
}
}
let account_json = serde_json::to_string(&account)?;
let _: () = redis::cmd("SET")
.arg(&account_key)
.arg(&account_json)
.query_async(&mut conn)
.await?;
info!("Topped up balance for {} by ${:.2}, new balance: ${:.2}",
imsi, amount as f64 / 100.0, account.balance as f64 / 100.0);
} else {
return Err(anyhow::anyhow!("Account not found for IMSI: {}", imsi));
}
Ok(())
}
// API methods for HTTP endpoints
pub async fn test_connection(&self) -> Result<()> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.context("Failed to get Redis connection")?;
let _: () = redis::cmd("PING").query_async(&mut conn).await?;
Ok(())
}
pub async fn check_credit(&self, ip: &str, bytes_requested: u64) -> Result<bool> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.context("Failed to get Redis connection")?;
let key = format!("credit:{}", ip);
let credit: i64 = conn.get(&key).await.unwrap_or(0);
Ok(credit >= bytes_requested as i64)
}
pub async fn get_balance(&self, ip: &str) -> Result<i64> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.context("Failed to get Redis connection")?;
let key = format!("credit:{}", ip);
let balance: i64 = conn.get(&key).await.unwrap_or(0);
Ok(balance)
}
pub async fn deduct_credit(&self, ip: &str, bytes_used: u64) -> Result<i64> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.context("Failed to get Redis connection")?;
let key = format!("credit:{}", ip);
let new_balance: i64 = conn.decr(&key, bytes_used).await.unwrap_or(0);
Ok(new_balance)
}
pub async fn add_credit(&self, ip: &str, bytes_to_add: u64) -> Result<i64> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.context("Failed to get Redis connection")?;
let key = format!("credit:{}", ip);
let new_balance: i64 = conn.incr(&key, bytes_to_add).await.unwrap_or(0);
Ok(new_balance)
}
}
apps/charging-engine/src/charging_engine.rs
+99
View File
@@ -0,0 +1,99 @@
use std::collections::HashMap;
use std::time::Duration;
use crate::charging_types::{RatingPlan, SubscriberAccount};
use crate::errors::{ChargingError, ChargingResult, ErrorContext};
pub struct ChargingEngine {
redis_client: redis::Client,
rating_plans: HashMap<String, RatingPlan>,
sync_interval: Duration,
}
impl ChargingEngine {
pub fn new(redis_url: &str, sync_interval_secs: u64) -> ChargingResult<Self> {
let redis_client = redis::Client::open(redis_url)
.context("Failed to create Redis client")?;
let mut rating_plans = HashMap::new();
// Default rating plans
rating_plans.insert("basic".to_string(), RatingPlan {
plan_id: "basic".to_string(),
name: "Basic Plan".to_string(),
data_rate: 0.01, // $0.01 per MB
voice_rate: 0.05, // $0.05 per minute
sms_rate: 0.10, // $0.10 per SMS
monthly_fee: 10.0, // $10 monthly
data_limit: 1_000_000_000, // 1GB
voice_limit: 300, // 300 minutes
sms_limit: 100, // 100 SMS
});
rating_plans.insert("premium".to_string(), RatingPlan {
plan_id: "premium".to_string(),
name: "Premium Plan".to_string(),
data_rate: 0.005, // $0.005 per MB
voice_rate: 0.02, // $0.02 per minute
sms_rate: 0.05, // $0.05 per SMS
monthly_fee: 25.0, // $25 monthly
data_limit: 5_000_000_000, // 5GB
voice_limit: 1000, // 1000 minutes
sms_limit: 500, // 500 SMS
});
rating_plans.insert("enterprise".to_string(), RatingPlan {
plan_id: "enterprise".to_string(),
name: "Enterprise Plan".to_string(),
data_rate: 0.001, // $0.001 per MB
voice_rate: 0.01, // $0.01 per minute
sms_rate: 0.02, // $0.02 per SMS
monthly_fee: 100.0, // $100 monthly
data_limit: 50_000_000_000, // 50GB
voice_limit: 5000, // 5000 minutes
sms_limit: 2000, // 2000 SMS
});
Ok(Self {
redis_client,
rating_plans,
sync_interval: Duration::from_secs(sync_interval_secs),
})
}
pub fn get_rating_plan(&self, plan_id: &str) -> Option<&RatingPlan> {
self.rating_plans.get(plan_id)
}
pub fn add_rating_plan(&mut self, plan: RatingPlan) {
self.rating_plans.insert(plan.plan_id.clone(), plan);
}
pub fn remove_rating_plan(&mut self, plan_id: &str) -> Option<RatingPlan> {
self.rating_plans.remove(plan_id)
}
pub fn list_rating_plans(&self) -> Vec<&RatingPlan> {
self.rating_plans.values().collect()
}
pub async fn start(&self) -> ChargingResult<()> {
info!("Starting charging engine with sync interval: {:?}", self.sync_interval);
// Initialize Redis connection
let mut conn = self.redis_client.get_async_connection().await
.context("Failed to establish Redis connection")?;
// Test Redis connection
let _: String = conn.get("test").await.unwrap_or_else(|_| "ok".to_string());
info!("Charging engine started successfully");
Ok(())
}
pub async fn stop(&self) -> ChargingResult<()> {
info!("Stopping charging engine");
// Cleanup operations
Ok(())
}
}
apps/charging-engine/src/charging_types.rs
+97
View File
@@ -0,0 +1,97 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SubscriberAccount {
pub imsi: String,
pub balance: i64, // in smallest currency unit (e.g., cents)
pub data_limit: u64, // bytes
pub data_used: u64, // bytes
pub voice_limit: u64, // seconds
pub voice_used: u64, // seconds
pub sms_limit: u64, // count
pub sms_used: u64, // count
pub status: AccountStatus,
pub last_updated: DateTime<Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum AccountStatus {
Active,
Suspended,
Terminated,
Blocked,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UsageEvent {
pub imsi: String,
pub session_id: String,
pub usage_type: UsageType,
pub volume: u64, // bytes, seconds, or count
pub timestamp: DateTime<Utc>,
pub rate: f64, // cost per unit
pub cost: f64, // total cost
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UsageType {
Data,
Voice,
SMS,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RatingPlan {
pub plan_id: String,
pub name: String,
pub data_rate: f64, // cost per MB
pub voice_rate: f64, // cost per minute
pub sms_rate: f64, // cost per SMS
pub monthly_fee: f64,
pub data_limit: u64, // bytes
pub voice_limit: u64, // seconds
pub sms_limit: u64, // count
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChargingSession {
pub session_id: String,
pub imsi: String,
pub start_time: DateTime<Utc>,
pub end_time: Option<DateTime<Utc>>,
pub data_bytes: u64,
pub voice_seconds: u64,
pub sms_count: u64,
pub total_cost: f64,
pub status: SessionStatus,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SessionStatus {
Active,
Completed,
Terminated,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChargingRule {
pub rule_id: String,
pub name: String,
pub priority: u32,
pub conditions: Vec<Condition>,
pub actions: Vec<Action>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Condition {
pub field: String,
pub operator: String,
pub value: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Action {
pub action_type: String,
pub parameters: std::collections::HashMap<String, String>,
}