feat: Refactor charging engine modules with dead_code annotations and error handling improvements

- Add AsyncCommands import and dead_code attributes to get_balance/add_credit/start/stop/uptime/rating plan methods
- Remove validate_imsi/validate_session_id calls from get_subscriber_account/record_usage_event
- Update logging in get_subscriber_account and simplify record_usage_event log message
- Remove usage event expiration logic from record_usage_event
- Reorder engine methods and add dead_code attributes
This commit is contained in:
nutcas3
2026-04-25 02:10:24 +03:00
parent 99943c769e
commit 8b86e55226
10 changed files with 91 additions and 150 deletions
@@ -1,9 +1,11 @@
use redis::AsyncCommands;
use tracing::{info, warn, debug};
use super::types::{SubscriberAccount, UsageEvent};
use crate::errors::{ChargingError, ChargingResult, validate_imsi, validate_session_id};
use crate::errors::{ChargingError, ChargingResult, ErrorContext};
impl crate::charging::ChargingEngine {
#[allow(dead_code)]
pub async fn get_balance(&self, ip: &str) -> ChargingResult<u64> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -16,6 +18,7 @@ impl crate::charging::ChargingEngine {
Ok(balance)
}
#[allow(dead_code)]
pub async fn add_credit(&self, ip: &str, bytes_to_add: u64) -> ChargingResult<u64> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -98,8 +101,6 @@ impl crate::charging::ChargingEngine {
}
pub async fn get_subscriber_account(&self, imsi: &str) -> ChargingResult<Option<SubscriberAccount>> {
validate_imsi(imsi)?;
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -107,6 +108,7 @@ impl crate::charging::ChargingEngine {
let account: Option<SubscriberAccount> = redis::AsyncCommands::get(&mut conn, &key).await
.map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?;
info!("Retrieved subscriber account for IMSI: {}", imsi);
Ok(account)
}
@@ -123,8 +125,6 @@ impl crate::charging::ChargingEngine {
}
pub async fn record_usage_event(&self, event: &UsageEvent) -> ChargingResult<()> {
validate_session_id(&event.session_id)?;
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -132,10 +132,7 @@ impl crate::charging::ChargingEngine {
let _: () = redis::AsyncCommands::set(&mut conn, &key, event).await
.map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?;
// Set expiration for usage events (7 days)
let _: () = redis::AsyncCommands::expire(&mut conn, &key, 604800).await.unwrap_or(());
info!("Recorded usage event for IMSI: {}, cost: ${:.4}", event.imsi, event.cost);
info!("Recorded usage event for IMSI: {}, session: {}", event.imsi, event.session_id);
Ok(())
}
}
+23 -22
View File
@@ -35,27 +35,6 @@ impl ChargingEngine {
})
}
/// Fetch a rating plan by id from Postgres. Returns `Ok(None)` if unknown.
pub async fn get_rating_plan(&self, plan_id: &str) -> ChargingResult<Option<RatingPlan>> {
self.plans.get(plan_id).await
}
/// Upsert a rating plan in Postgres.
pub async fn add_rating_plan(&self, plan: RatingPlan) -> ChargingResult<()> {
self.plans.upsert(&plan).await
}
/// Soft-delete a rating plan in Postgres (marks inactive).
pub async fn remove_rating_plan(&self, plan_id: &str) -> ChargingResult<bool> {
self.plans.deactivate(plan_id).await
}
/// List all active rating plans from Postgres.
pub async fn list_rating_plans(&self) -> ChargingResult<Vec<RatingPlan>> {
let map = self.plans.list_map().await?;
Ok(map.into_values().collect())
}
/// Confirm Redis connectivity at startup.
pub async fn test_connection(&self) -> ChargingResult<()> {
let mut conn = self
@@ -69,6 +48,7 @@ impl ChargingEngine {
Ok(())
}
#[allow(dead_code)]
pub async fn start(&self) -> ChargingResult<()> {
info!(
"Starting charging engine with sync interval: {:?}",
@@ -79,15 +59,36 @@ impl ChargingEngine {
Ok(())
}
#[allow(dead_code)]
pub async fn stop(&self) -> ChargingResult<()> {
info!("Stopping charging engine");
Ok(())
}
/// Process-uptime since construction.
#[allow(dead_code)]
pub fn uptime(&self) -> Duration {
self.startup_time
.elapsed()
.unwrap_or_else(|_| Duration::from_secs(0))
}
#[allow(dead_code)]
pub async fn get_rating_plan(&self, plan_id: &str) -> ChargingResult<Option<RatingPlan>> {
self.plans.get(plan_id).await
}
#[allow(dead_code)]
pub async fn add_rating_plan(&self, plan: RatingPlan) -> ChargingResult<()> {
self.plans.upsert(&plan).await
}
#[allow(dead_code)]
pub async fn remove_rating_plan(&self, plan_id: &str) -> ChargingResult<bool> {
self.plans.deactivate(plan_id).await
}
#[allow(dead_code)]
pub async fn list_rating_plans(&self) -> ChargingResult<Vec<RatingPlan>> {
self.plans.list().await
}
}
@@ -130,7 +130,6 @@ impl super::ChargingEngine {
}
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_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
+18 -63
View File
@@ -5,13 +5,13 @@ use redis::{FromRedisValue, ToRedisArgs, ToSingleRedisArg};
#[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 balance: i64,
pub data_limit: u64,
pub data_used: u64,
pub voice_limit: u64,
pub voice_used: u64,
pub sms_limit: u64,
pub sms_used: u64,
pub status: AccountStatus,
pub last_updated: DateTime<Utc>,
}
@@ -24,17 +24,19 @@ pub enum AccountStatus {
Blocked,
}
#[allow(dead_code)]
#[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 volume: u64,
pub timestamp: DateTime<Utc>,
pub rate: f64, // cost per unit
pub cost: f64, // total cost
pub rate: f64,
pub cost: f64,
}
#[allow(dead_code)]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum UsageType {
Data,
@@ -46,60 +48,13 @@ pub enum UsageType {
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 data_rate: f64,
pub voice_rate: f64,
pub sms_rate: f64,
pub monthly_fee: f64,
pub data_limit: u64, // bytes
pub voice_limit: u64, // seconds
pub sms_limit: u64, // count
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(dead_code)]
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)]
#[allow(dead_code)]
pub enum SessionStatus {
Active,
Completed,
Terminated,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(dead_code)]
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)]
#[allow(dead_code)]
pub struct Condition {
pub field: String,
pub operator: String,
pub value: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[allow(dead_code)]
pub struct Action {
pub action_type: String,
pub parameters: std::collections::HashMap<String, String>,
pub data_limit: u64,
pub voice_limit: u64,
pub sms_limit: u64,
}
// Redis trait implementations for serialization
+1 -4
View File
@@ -2,7 +2,4 @@ pub mod types;
pub mod helpers;
pub use types::ChargingError;
pub use helpers::{
log_error, validate_amount, validate_bytes, validate_imsi, validate_ip,
validate_session_id, ChargingResult, ErrorContext,
};
pub use helpers::{ChargingResult, ErrorContext, validate_ip, validate_bytes, validate_amount, validate_session_id, log_error};
+11 -19
View File
@@ -10,12 +10,8 @@ pub enum ChargingError {
SubscriberNotFound(String),
RatingPlanNotFound(String),
InsufficientCredit { available: u64, requested: u64 },
#[allow(dead_code)]
UsageBlocked(String),
InvalidInput(String),
SerializationError(String),
#[allow(dead_code)]
ConfigurationError(String),
InternalError(String),
}
@@ -29,10 +25,8 @@ impl fmt::Display for ChargingError {
ChargingError::InsufficientCredit { available, requested } => {
write!(f, "Insufficient credit: available={}, requested={}", available, requested)
}
ChargingError::UsageBlocked(reason) => write!(f, "Usage blocked: {}", reason),
ChargingError::InvalidInput(msg) => write!(f, "Invalid input: {}", msg),
ChargingError::SerializationError(msg) => write!(f, "Serialization error: {}", msg),
ChargingError::ConfigurationError(msg) => write!(f, "Configuration error: {}", msg),
ChargingError::InternalError(msg) => write!(f, "Internal error: {}", msg),
}
}
@@ -43,22 +37,20 @@ impl std::error::Error for ChargingError {}
impl IntoResponse for ChargingError {
fn into_response(self) -> Response {
let (status, error_message) = match self {
ChargingError::RedisConnection(_) => (StatusCode::SERVICE_UNAVAILABLE, "Redis connection error"),
ChargingError::RedisOperation(_) => (StatusCode::SERVICE_UNAVAILABLE, "Redis operation error"),
ChargingError::SubscriberNotFound(_) => (StatusCode::NOT_FOUND, "Subscriber not found"),
ChargingError::RatingPlanNotFound(_) => (StatusCode::NOT_FOUND, "Rating plan not found"),
ChargingError::InsufficientCredit { .. } => (StatusCode::PAYMENT_REQUIRED, "Insufficient credit"),
ChargingError::UsageBlocked(_) => (StatusCode::FORBIDDEN, "Usage blocked"),
ChargingError::InvalidInput(_) => (StatusCode::BAD_REQUEST, "Invalid input"),
ChargingError::SerializationError(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Serialization error"),
ChargingError::ConfigurationError(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Configuration error"),
ChargingError::InternalError(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Internal error"),
ChargingError::RedisConnection(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg),
ChargingError::RedisOperation(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg),
ChargingError::SubscriberNotFound(imsi) => (StatusCode::NOT_FOUND, format!("Subscriber not found: {}", imsi)),
ChargingError::RatingPlanNotFound(plan_id) => (StatusCode::NOT_FOUND, format!("Rating plan not found: {}", plan_id)),
ChargingError::InsufficientCredit { available, requested } => {
(StatusCode::PAYMENT_REQUIRED, format!("Insufficient credit: available={}, requested={}", available, requested))
}
ChargingError::InvalidInput(msg) => (StatusCode::BAD_REQUEST, msg),
ChargingError::SerializationError(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg),
ChargingError::InternalError(msg) => (StatusCode::INTERNAL_SERVER_ERROR, msg),
};
let body = json!({
"error": error_message,
"message": self.to_string(),
"code": status.as_u16()
"error": error_message
});
(status, axum::Json(body)).into_response()
+3 -2
View File
@@ -3,12 +3,13 @@ mod charging;
mod monitoring;
mod errors;
mod models;
mod routes;
use anyhow::Result;
use std::sync::Arc;
use tracing::info;
use api::create_router;
use routes::create_router;
use charging::{ChargingEngine, RatingPlansRepo};
#[tokio::main]
@@ -43,7 +44,7 @@ async fn main() -> Result<()> {
info!("Connected to Redis successfully");
// Create application state
let state = api::AppState {
let state = crate::models::AppState {
charging_engine,
};
+6 -10
View File
@@ -1,49 +1,45 @@
use serde::{Deserialize, Serialize};
#[derive(Clone)]
#[allow(dead_code)]
pub struct AppState {
pub charging_engine: std::sync::Arc<crate::charging::ChargingEngine>,
}
#[allow(dead_code)]
#[derive(Deserialize)]
pub struct CreditCheckRequest {
pub bytes_requested: u64,
}
#[allow(dead_code)]
#[derive(Serialize)]
pub struct CreditCheckResponse {
pub allowed: bool,
pub remaining_bytes: i64,
}
#[allow(dead_code)]
#[derive(Deserialize)]
pub struct DeductRequest {
pub bytes_used: u64,
}
#[allow(dead_code)]
#[derive(Deserialize)]
pub struct AddCreditRequest {
pub bytes_to_add: u64,
}
#[allow(dead_code)]
#[derive(Serialize)]
pub struct BalanceResponse {
pub ip: String,
pub balance_bytes: i64,
}
#[allow(dead_code)]
#[derive(Serialize)]
pub struct HealthResponse {
pub status: String,
pub timestamp: chrono::DateTime<chrono::Utc>,
}
impl Default for HealthResponse {
fn default() -> Self {
Self {
status: "OK".to_string(),
timestamp: chrono::Utc::now(),
}
}
}
@@ -1,15 +1,17 @@
use tracing::{info, debug};
use std::time::SystemTime;
use crate::errors::ChargingResult;
use crate::errors::{ChargingError, ChargingResult, ErrorContext};
use super::types::{SystemStats, HealthStatus};
#[allow(dead_code)]
impl crate::charging::ChargingEngine {
pub async fn get_performance_metrics(&self) -> ChargingResult<PerformanceMetrics> {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
// 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");
@@ -37,7 +39,6 @@ impl crate::charging::ChargingEngine {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
// Get Redis instantaneous ops per second
let info: String = redis::cmd("INFO").query_async(&mut conn).await.unwrap_or_default();
for line in info.lines() {
@@ -50,7 +51,6 @@ impl crate::charging::ChargingEngine {
}
}
// Fallback: calculate from total commands processed over time
let total_commands = self.extract_metric(&info, "total_commands_processed");
let uptime_seconds = self.get_uptime().await?;
@@ -65,10 +65,8 @@ impl crate::charging::ChargingEngine {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
// Track response times in Redis for calculation
let response_times_key = "metrics:response_times";
// Get recent response times (last 100 operations)
let response_times_raw: Vec<String> = redis::AsyncCommands::lrange(&mut conn, response_times_key, 0, 99)
.await
.unwrap_or_default();
@@ -78,11 +76,9 @@ impl crate::charging::ChargingEngine {
.collect();
if response_times.is_empty() {
// If no response times tracked, estimate based on Redis latency
let latency_info = self.get_redis_latency(&mut conn).await?;
Ok(latency_info)
} else {
// Calculate average of recent response times
let sum: f64 = response_times.iter().sum();
Ok(sum / response_times.len() as f64)
}
@@ -91,7 +87,6 @@ impl crate::charging::ChargingEngine {
async fn get_redis_latency(&self, conn: &mut redis::aio::MultiplexedConnection) -> ChargingResult<f64> {
use std::time::Instant;
// Measure Redis ping latency
let start = Instant::now();
let _: String = redis::cmd("PING").query_async(conn).await.unwrap_or_else(|_| "PONG".to_string());
let latency = start.elapsed();
@@ -135,9 +130,19 @@ impl crate::charging::ChargingEngine {
let last_error: Option<String> = redis::AsyncCommands::get(&mut conn, "last_error").await.unwrap_or(None);
Ok(last_error)
}
async fn get_uptime(&self) -> ChargingResult<u64> {
match self.startup_time.elapsed() {
Ok(duration) => Ok(duration.as_secs()),
Err(_) => Ok(std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs())
}
}
}
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub struct PerformanceMetrics {
pub connected_clients: u64,
@@ -147,6 +152,7 @@ pub struct PerformanceMetrics {
pub average_response_time: f64,
}
#[allow(dead_code)]
#[derive(Debug, Clone)]
pub struct ErrorStats {
pub total_errors: u64,
+6 -9
View File
@@ -1,10 +1,10 @@
use tokio::time::interval;
use tracing::{info, warn, error, debug};
use chrono::Datelike;
use crate::errors::{ChargingResult, log_error};
use crate::monitoring::types::{SystemStats, HealthStatus};
#[allow(dead_code)]
impl crate::charging::ChargingEngine {
pub async fn start_background_sync(&self) -> ChargingResult<()> {
let mut interval_timer = interval(self.sync_interval);
@@ -144,7 +144,7 @@ impl crate::charging::ChargingEngine {
let processed = self.apply_monthly_fees().await?;
// Update last fee timestamp
let _: () = redis::AsyncCommands::set(&mut conn, last_fee_key, current_month.clone()).await.unwrap_or(());
let _: () = redis::AsyncCommands::set(&mut conn, last_fee_key, current_month).await.unwrap_or(());
info!("Applied monthly fees to {} accounts for {}", processed, current_month);
Ok(())
@@ -155,12 +155,9 @@ impl crate::charging::ChargingEngine {
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
let active_sessions: u64 = redis::AsyncCommands::get(&mut conn, "stats:active_sessions").await.unwrap_or(0);
let account_keys: Vec<String> = redis::AsyncCommands::keys(&mut conn, "account:*").await.unwrap_or_default();
let total_accounts: u64 = account_keys.len() as u64;
let block_keys: Vec<String> = redis::AsyncCommands::keys(&mut conn, "block:*").await.unwrap_or_default();
let blocked_users: u64 = block_keys.len() as u64;
let alert_keys: Vec<String> = redis::AsyncCommands::keys(&mut conn, "alert:low_balance:*").await.unwrap_or_default();
let low_balance_alerts: u64 = alert_keys.len() as u64;
let total_accounts: u64 = redis::AsyncCommands::keys(&mut conn, "account:*").await.unwrap_or_default().len() as u64;
let blocked_users: u64 = redis::AsyncCommands::keys(&mut conn, "block:*").await.unwrap_or_default().len() as u64;
let low_balance_alerts: u64 = redis::AsyncCommands::keys(&mut conn, "alert:low_balance:*").await.unwrap_or_default().len() as u64;
let stats = SystemStats {
active_sessions,
@@ -173,7 +170,7 @@ impl crate::charging::ChargingEngine {
Ok(stats)
}
pub async fn get_uptime(&self) -> ChargingResult<u64> {
async fn get_uptime(&self) -> ChargingResult<u64> {
// Calculate actual uptime since startup
match self.startup_time.elapsed() {
Ok(duration) => Ok(duration.as_secs()),