andwati/telecom-platform
1
0
Fork 0
mirror of https://github.com/andwati/telecom-platform.git synced 2026-10-18 11:02:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Re-enable circuit breaker implementation in charging-engine credit management and rating plans

Browse Source 
- Uncomment CircuitBreaker and CircuitBreakerError imports in credit_management.rs and rating_plans_repo.rs
- Re-enable redis_circuit_breaker execute calls for all Redis operations (get_balance, add_credit, deduct_credit, block_user, unblock_user, is_user_blocked, get_subscriber_account, update_subscriber_account, record_usage_event)
- Re-enable postgres circuit_breaker execute calls for all database operations
This commit is contained in:
nutcas3
2026-04-26 02:13:07 +03:00
parent 1d489a0a8a
commit a4aec94314
4 changed files with 76 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/charging-engine/src/charging/credit_management.rs
+47 -47
View File
@@ -1,12 +1,12 @@
use tracing::{info, warn, debug};
use super::types::{SubscriberAccount, UsageEvent};
// use crate::circuit_breaker::CircuitBreakerError;
use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError};
use crate::errors::{ChargingError, ChargingResult};
impl crate::charging::ChargingEngine {
pub async fn get_balance(&self, ip: &str) -> ChargingResult<u64> {
// // // self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -14,16 +14,16 @@ impl crate::charging::ChargingEngine {
let balance: u64 = redis::AsyncCommands::get(&mut conn, &key).await
.map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?;
debug!("Retrieved balance {} for IP: {}", balance, ip);
info!("Retrieved balance for IP: {}, balance: {} bytes", ip, balance);
Ok(balance)
// }).await.map_err(|e| match e {$
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn add_credit(&self, ip: &str, bytes_to_add: u64) -> ChargingResult<u64> {
// // self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -51,14 +51,14 @@ impl crate::charging::ChargingEngine {
info!("Added {} bytes credit to IP: {}, new balance: {}", bytes_to_add, ip, new_balance);
Ok(new_balance)
// }).await.map_err(|e| match e {$
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn deduct_credit(&self, ip: &str, bytes_used: u64) -> ChargingResult<u64> {
// // self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -80,10 +80,10 @@ impl crate::charging::ChargingEngine {
info!("Deducted {} bytes credit from IP: {}, new balance: {}", bytes_used, ip, new_balance);
Ok(new_balance)
// }).await.map_err(|e| match e {$
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn check_credit(&self, ip: &str, bytes_requested: u64) -> ChargingResult<bool> {
@@ -92,7 +92,7 @@ impl crate::charging::ChargingEngine {
}
pub async fn block_user(&self, ip: &str) -> ChargingResult<()> {
// // self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -116,14 +116,14 @@ impl crate::charging::ChargingEngine {
warn!("User IP: {} has been blocked for {} seconds", ip, block_expiration);
Ok(())
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn unblock_user(&self, ip: &str) -> ChargingResult<()> {
// self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -133,14 +133,14 @@ impl crate::charging::ChargingEngine {
info!("User IP: {} has been unblocked", ip);
Ok(())
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn is_user_blocked(&self, ip: &str) -> ChargingResult<bool> {
// self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -149,14 +149,14 @@ impl crate::charging::ChargingEngine {
.map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?;
Ok(blocked.is_some())
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn get_subscriber_account(&self, imsi: &str) -> ChargingResult<Option<SubscriberAccount>> {
// self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -166,14 +166,14 @@ impl crate::charging::ChargingEngine {
info!("Retrieved subscriber account for IMSI: {}", imsi);
Ok(account)
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn update_subscriber_account(&self, account: &SubscriberAccount) -> ChargingResult<()> {
// self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -183,14 +183,14 @@ impl crate::charging::ChargingEngine {
debug!("Updated subscriber account for IMSI: {}", account.imsi);
Ok(())
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
pub async fn record_usage_event(&self, event: &UsageEvent) -> ChargingResult<()> {
// self.redis_circuit_breaker.execute(|| async {
self.redis_circuit_breaker.execute(|| async {
let mut conn = self.redis_client.get_multiplexed_async_connection().await
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
@@ -200,9 +200,9 @@ impl crate::charging::ChargingEngine {
info!("Recorded usage event for IMSI: {}, session: {}", event.imsi, event.session_id);
Ok(())
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::RedisConnection("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
}
apps/charging-engine/src/charging/rating_plans_repo.rs
+24 -24
View File
@@ -9,14 +9,14 @@ use sqlx::postgres::{PgPool, PgPoolOptions};
use sqlx::Row;
use super::types::RatingPlan;
// use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError};
use crate::circuit_breaker::{CircuitBreaker, CircuitBreakerError};
use crate::errors::{ChargingError, ChargingResult};
/// Repository for rating plans backed by Postgres.
#[derive(Clone)]
pub struct RatingPlansRepo {
pool: PgPool,
// circuit_breaker: CircuitBreaker,
circuit_breaker: CircuitBreaker,
}
impl RatingPlansRepo {
@@ -28,7 +28,7 @@ impl RatingPlansRepo {
.await
.map_err(|e| crate::errors::ChargingError::InternalError(e.to_string()))?;
// let circuit_breaker = CircuitBreaker::new(5, std::time::Duration::from_secs(60));
let circuit_breaker = CircuitBreaker::new(5, std::time::Duration::from_secs(60));
// Idempotent schema for rating plans. api-server's GORM migration also
// creates a compatible `rating_plans` table; the column set here is
@@ -55,7 +55,7 @@ impl RatingPlansRepo {
.await
.map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?;
Ok(Self { pool })
Ok(Self { pool, circuit_breaker })
}
/// Seed the canonical basic/premium/enterprise plans if the table is empty.
@@ -106,7 +106,7 @@ impl RatingPlansRepo {
/// Fetch a single plan by id.
pub async fn get(&self, plan_id: &str) -> ChargingResult<Option<RatingPlan>> {
let plan_id = plan_id.to_string();
// self.circuit_breaker.execute(|| async move {
self.circuit_breaker.execute(|| async move {
let row = sqlx::query(
r#"
SELECT plan_id, name, data_rate, voice_rate, sms_rate,
@@ -121,15 +121,15 @@ impl RatingPlansRepo {
.map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?;
Ok(row.map(row_to_plan))
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
/// Fetch all active plans as a HashMap for fast lookup.
pub async fn list_map(&self) -> ChargingResult<HashMap<String, RatingPlan>> {
// self.circuit_breaker.execute(|| async {
self.circuit_breaker.execute(|| async {
let rows = sqlx::query(
r#"
SELECT plan_id, name, data_rate, voice_rate, sms_rate,
@@ -148,10 +148,10 @@ impl RatingPlansRepo {
out.insert(plan.plan_id.clone(), plan);
}
Ok(out)
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
/// List all active plans.
@@ -178,7 +178,7 @@ impl RatingPlansRepo {
/// Insert or update a plan.
pub async fn upsert(&self, plan: &RatingPlan) -> ChargingResult<()> {
let plan = plan.clone();
// self.circuit_breaker.execute(|| async move {
self.circuit_breaker.execute(|| async move {
sqlx::query(
r#"
INSERT INTO rating_plans
@@ -211,16 +211,16 @@ impl RatingPlansRepo {
.await
.map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?;
Ok(())
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
/// Soft-delete a plan by marking it inactive.
pub async fn deactivate(&self, plan_id: &str) -> ChargingResult<bool> {
let plan_id = plan_id.to_string();
// self.circuit_breaker.execute(|| async move {
self.circuit_breaker.execute(|| async move {
let result = sqlx::query(
r#"UPDATE rating_plans SET is_active = FALSE, updated_at = NOW() WHERE plan_id = $1"#,
)
@@ -229,10 +229,10 @@ impl RatingPlansRepo {
.await
.map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?;
Ok(result.rows_affected() > 0)
// }).await.map_err(|e| match e {
// CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
// CircuitBreakerError::Inner(e) => e,
// })
}).await.map_err(|e| match e {
CircuitBreakerError::Open => ChargingError::DatabaseError("Circuit breaker is open".to_string()),
CircuitBreakerError::Inner(e) => e,
})
}
}
apps/charging-engine/src/lib.rs
+2 -3
View File
@@ -1,9 +1,8 @@
pub mod auth;
pub mod charging;
// pub mod circuit_breaker;
pub mod errors;
pub mod circuit_breaker;
pub mod rate_limit;
pub mod handlers;
// pub mod rate_limit;
pub mod routes;
pub mod models;
pub mod monitoring;
apps/charging-engine/src/routes.rs
+3 -3
View File
@@ -5,7 +5,7 @@ use axum::{
use tower_http::cors::{Any, CorsLayer};
use crate::auth::auth_middleware;
// use crate::rate_limit::create_rate_limiter;
use crate::rate_limit::create_rate_limiter;
use crate::handlers::{
add_credit, add_rating_plan, block_user, check_credit, deduct_credit, detailed_health_check,
engine_start, engine_stop, engine_uptime, get_balance, get_error_stats, get_performance_metrics,
@@ -21,7 +21,7 @@ pub fn create_router(state: AppState) -> Router {
.allow_methods(Any)
.allow_headers(Any);
// let rate_limiter = create_rate_limiter();
let rate_limiter = create_rate_limiter();
Router::new()
// Public routes (no auth required - for packet-gateway integration)
@@ -54,7 +54,7 @@ pub fn create_router(state: AppState) -> Router {
.route("/v1/usage/process", post(process_usage))
.route("/v1/invoice/:imsi/:period", get(generate_invoice))
.layer(cors)
// .layer(rate_limiter)
.layer(rate_limiter)
.route_layer(axum::middleware::from_fn_with_state(state.auth_config.clone(), auth_middleware))
.with_state(state)
}