diff --git a/apps/charging-engine/src/charging/credit_management.rs b/apps/charging-engine/src/charging/credit_management.rs index f37e863..bf16bde 100644 --- a/apps/charging-engine/src/charging/credit_management.rs +++ b/apps/charging-engine/src/charging/credit_management.rs @@ -1,73 +1,89 @@ use tracing::{info, warn, debug}; use super::types::{SubscriberAccount, UsageEvent}; +use crate::circuit_breaker::CircuitBreakerError; use crate::errors::{ChargingError, ChargingResult}; impl crate::charging::ChargingEngine { pub async fn get_balance(&self, ip: &str) -> ChargingResult { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("credit:{}", ip); - let balance: u64 = redis::AsyncCommands::get(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("credit:{}", ip); + 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); - Ok(balance) + debug!("Retrieved balance {} for IP: {}", balance, ip); + Ok(balance) + }).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 { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("credit:{}", ip); - - // Check for potential overflow before incrementing - let current_balance: Option = redis::AsyncCommands::get(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - - if let Some(existing_balance) = current_balance { - if existing_balance.checked_add(bytes_to_add).is_none() { - return Err(ChargingError::InvalidInput("Credit addition would overflow".to_string())); + let key = format!("credit:{}", ip); + + // Check for potential overflow before incrementing + let current_balance: Option = redis::AsyncCommands::get(&mut conn, &key).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + + if let Some(existing_balance) = current_balance { + if existing_balance.checked_add(bytes_to_add).is_none() { + return Err(ChargingError::InvalidInput("Credit addition would overflow".to_string())); + } } - } - - let new_balance: u64 = redis::AsyncCommands::incr(&mut conn, &key, bytes_to_add).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + + let new_balance: u64 = redis::AsyncCommands::incr(&mut conn, &key, bytes_to_add).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - // Set expiration if this is a new key (configurable via CREDIT_EXPIRATION_SECONDS, defaults to 24 hours) - let credit_expiration: u64 = std::env::var("CREDIT_EXPIRATION_SECONDS") - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(86400); - let _: () = redis::AsyncCommands::expire(&mut conn, &key, credit_expiration as i64).await.unwrap_or(()); + // Set expiration if this is a new key (configurable via CREDIT_EXPIRATION_SECONDS, defaults to 24 hours) + let credit_expiration: u64 = std::env::var("CREDIT_EXPIRATION_SECONDS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(86400); + let _: () = redis::AsyncCommands::expire(&mut conn, &key, credit_expiration as i64).await.unwrap_or(()); - info!("Added {} bytes credit to IP: {}, new balance: {}", bytes_to_add, ip, new_balance); - Ok(new_balance) + 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, + }) } pub async fn deduct_credit(&self, ip: &str, bytes_used: u64) -> ChargingResult { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("credit:{}", ip); - - // Check current balance first - let current_balance: u64 = redis::AsyncCommands::get(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - - if current_balance < bytes_used { - return Err(ChargingError::InsufficientCredit { - available: current_balance, - requested: bytes_used, - }); - } + let key = format!("credit:{}", ip); + + // Check current balance first + let current_balance: u64 = redis::AsyncCommands::get(&mut conn, &key).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + + if current_balance < bytes_used { + return Err(ChargingError::InsufficientCredit { + available: current_balance, + requested: bytes_used, + }); + } - let new_balance: u64 = redis::AsyncCommands::decr(&mut conn, &key, bytes_used).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let new_balance: u64 = redis::AsyncCommands::decr(&mut conn, &key, bytes_used).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - info!("Deducted {} bytes credit from IP: {}, new balance: {}", bytes_used, ip, new_balance); - Ok(new_balance) + 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, + }) } pub async fn check_credit(&self, ip: &str, bytes_requested: u64) -> ChargingResult { @@ -76,87 +92,117 @@ impl crate::charging::ChargingEngine { } pub async fn block_user(&self, ip: &str) -> ChargingResult<()> { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("block:{}", ip); - let blocked: Option = redis::AsyncCommands::get(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("block:{}", ip); + let blocked: Option = redis::AsyncCommands::get(&mut conn, &key).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - if blocked.is_some() { - return Err(ChargingError::UsageBlocked(format!("User IP {} is already blocked", ip))); - } + if blocked.is_some() { + return Err(ChargingError::UsageBlocked(format!("User IP {} is already blocked", ip))); + } - let _: () = redis::AsyncCommands::set(&mut conn, &key, "blocked").await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let _: () = redis::AsyncCommands::set(&mut conn, &key, "blocked").await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - // Set expiration for block (configurable via BLOCK_EXPIRATION_SECONDS, defaults to 24 hours) - let block_expiration: u64 = std::env::var("BLOCK_EXPIRATION_SECONDS") - .ok() - .and_then(|s| s.parse().ok()) - .unwrap_or(86400); - let _: () = redis::AsyncCommands::expire(&mut conn, &key, block_expiration as i64).await.unwrap_or(()); + // Set expiration for block (configurable via BLOCK_EXPIRATION_SECONDS, defaults to 24 hours) + let block_expiration: u64 = std::env::var("BLOCK_EXPIRATION_SECONDS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(86400); + let _: () = redis::AsyncCommands::expire(&mut conn, &key, block_expiration as i64).await.unwrap_or(()); - warn!("User IP: {} has been blocked for {} seconds", ip, block_expiration); - Ok(()) + 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, + }) } pub async fn unblock_user(&self, ip: &str) -> ChargingResult<()> { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("block:{}", ip); - let _: () = redis::AsyncCommands::del(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("block:{}", ip); + let _: () = redis::AsyncCommands::del(&mut conn, &key).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - info!("User IP: {} has been unblocked", ip); - Ok(()) + 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, + }) } pub async fn is_user_blocked(&self, ip: &str) -> ChargingResult { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("block:{}", ip); - let blocked: Option = redis::AsyncCommands::get(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("block:{}", ip); + let blocked: Option = redis::AsyncCommands::get(&mut conn, &key).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - Ok(blocked.is_some()) + Ok(blocked.is_some()) + }).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> { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("account:{}", imsi); - let account: Option = redis::AsyncCommands::get(&mut conn, &key).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("account:{}", imsi); + let account: Option = 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) + 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, + }) } pub async fn update_subscriber_account(&self, account: &SubscriberAccount) -> ChargingResult<()> { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("account:{}", account.imsi); - let _: () = redis::AsyncCommands::set(&mut conn, &key, account).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("account:{}", account.imsi); + let _: () = redis::AsyncCommands::set(&mut conn, &key, account).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - debug!("Updated subscriber account for IMSI: {}", account.imsi); - Ok(()) + 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, + }) } pub async fn record_usage_event(&self, event: &UsageEvent) -> ChargingResult<()> { - let mut conn = self.redis_client.get_multiplexed_async_connection().await - .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + 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()))?; - let key = format!("usage:{}:{}", event.imsi, event.session_id); - let _: () = redis::AsyncCommands::set(&mut conn, &key, event).await - .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; + let key = format!("usage:{}:{}", event.imsi, event.session_id); + let _: () = redis::AsyncCommands::set(&mut conn, &key, event).await + .map_err(|e| crate::errors::ChargingError::RedisOperation(e.to_string()))?; - info!("Recorded usage event for IMSI: {}, session: {}", event.imsi, event.session_id); - Ok(()) + 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, + }) } } diff --git a/apps/charging-engine/src/charging/engine.rs b/apps/charging-engine/src/charging/engine.rs index 8cb513c..f3ce5aa 100644 --- a/apps/charging-engine/src/charging/engine.rs +++ b/apps/charging-engine/src/charging/engine.rs @@ -4,6 +4,7 @@ use tracing::info; use super::rating_plans_repo::RatingPlansRepo; use super::types::RatingPlan; +use crate::circuit_breaker::CircuitBreaker; use crate::errors::ChargingResult; /// ChargingEngine coordinates Redis-backed hot-path state (balances, sessions) @@ -13,6 +14,8 @@ pub struct ChargingEngine { pub(crate) plans: RatingPlansRepo, pub(crate) sync_interval: Duration, pub(crate) startup_time: SystemTime, + pub(crate) redis_circuit_breaker: CircuitBreaker, + pub(crate) postgres_circuit_breaker: CircuitBreaker, } impl ChargingEngine { @@ -27,11 +30,32 @@ impl ChargingEngine { let redis_client = redis::Client::open(redis_url) .map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?; + // Initialize circuit breakers with configurable thresholds + let redis_cb_threshold: u32 = std::env::var("REDIS_CB_THRESHOLD") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(5); + let redis_cb_timeout: u64 = std::env::var("REDIS_CB_TIMEOUT") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(60); + + let postgres_cb_threshold: u32 = std::env::var("POSTGRES_CB_THRESHOLD") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(5); + let postgres_cb_timeout: u64 = std::env::var("POSTGRES_CB_TIMEOUT") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(60); + Ok(Self { redis_client, plans, sync_interval: Duration::from_secs(sync_interval_secs), startup_time: SystemTime::now(), + redis_circuit_breaker: CircuitBreaker::new(redis_cb_threshold, Duration::from_secs(redis_cb_timeout)), + postgres_circuit_breaker: CircuitBreaker::new(postgres_cb_threshold, Duration::from_secs(postgres_cb_timeout)), }) } diff --git a/apps/charging-engine/src/charging/rating_plans_repo.rs b/apps/charging-engine/src/charging/rating_plans_repo.rs index e7a90d3..559a8f3 100644 --- a/apps/charging-engine/src/charging/rating_plans_repo.rs +++ b/apps/charging-engine/src/charging/rating_plans_repo.rs @@ -9,12 +9,14 @@ use sqlx::postgres::{PgPool, PgPoolOptions}; use sqlx::Row; use super::types::RatingPlan; -use crate::errors::ChargingResult; +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, } impl RatingPlansRepo { @@ -26,6 +28,17 @@ impl RatingPlansRepo { .await .map_err(|e| crate::errors::ChargingError::InternalError(e.to_string()))?; + // Initialize circuit breaker for PostgreSQL operations + let postgres_cb_threshold: u32 = std::env::var("POSTGRES_CB_THRESHOLD") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(5); + let postgres_cb_timeout: u64 = std::env::var("POSTGRES_CB_TIMEOUT") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(60); + let circuit_breaker = CircuitBreaker::new(postgres_cb_threshold, std::time::Duration::from_secs(postgres_cb_timeout)); + // Idempotent schema for rating plans. api-server's GORM migration also // creates a compatible `rating_plans` table; the column set here is // intentionally a strict subset that both services agree on. @@ -51,7 +64,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. @@ -101,42 +114,53 @@ impl RatingPlansRepo { /// Fetch a single plan by id. pub async fn get(&self, plan_id: &str) -> ChargingResult> { - let row = sqlx::query( - r#" - SELECT plan_id, name, data_rate, voice_rate, sms_rate, - monthly_fee, data_limit, voice_limit, sms_limit - FROM rating_plans - WHERE plan_id = $1 AND is_active = TRUE - "#, - ) - .bind(plan_id) - .fetch_optional(&self.pool) - .await - .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; + let plan_id = plan_id.to_string(); + self.circuit_breaker.execute(async move { + let row = sqlx::query( + r#" + SELECT plan_id, name, data_rate, voice_rate, sms_rate, + monthly_fee, data_limit, voice_limit, sms_limit + FROM rating_plans + WHERE plan_id = $1 AND is_active = TRUE + "#, + ) + .bind(&plan_id) + .fetch_optional(&self.pool) + .await + .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; - Ok(row.map(row_to_plan)) + 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, + }) } /// Fetch all active plans as a HashMap for fast lookup. pub async fn list_map(&self) -> ChargingResult> { - let rows = sqlx::query( - r#" - SELECT plan_id, name, data_rate, voice_rate, sms_rate, - monthly_fee, data_limit, voice_limit, sms_limit - FROM rating_plans - WHERE is_active = TRUE - "#, - ) - .fetch_all(&self.pool) - .await - .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; + self.circuit_breaker.execute(async { + let rows = sqlx::query( + r#" + SELECT plan_id, name, data_rate, voice_rate, sms_rate, + monthly_fee, data_limit, voice_limit, sms_limit + FROM rating_plans + WHERE is_active = TRUE + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; - let mut out = HashMap::with_capacity(rows.len()); - for row in rows { - let plan = row_to_plan(row); - out.insert(plan.plan_id.clone(), plan); - } - Ok(out) + let mut out = HashMap::with_capacity(rows.len()); + for row in rows { + let plan = row_to_plan(row); + 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, + }) } /// List all active plans. @@ -162,50 +186,62 @@ impl RatingPlansRepo { /// Insert or update a plan. pub async fn upsert(&self, plan: &RatingPlan) -> ChargingResult<()> { - sqlx::query( - r#" - INSERT INTO rating_plans - (plan_id, name, data_rate, voice_rate, sms_rate, - monthly_fee, data_limit, voice_limit, sms_limit, is_active, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, TRUE, NOW()) - ON CONFLICT (plan_id) DO UPDATE SET - name = EXCLUDED.name, - data_rate = EXCLUDED.data_rate, - voice_rate = EXCLUDED.voice_rate, - sms_rate = EXCLUDED.sms_rate, - monthly_fee = EXCLUDED.monthly_fee, - data_limit = EXCLUDED.data_limit, - voice_limit = EXCLUDED.voice_limit, - sms_limit = EXCLUDED.sms_limit, - is_active = TRUE, - updated_at = NOW() - "#, - ) - .bind(&plan.plan_id) - .bind(&plan.name) - .bind(plan.data_rate) - .bind(plan.voice_rate) - .bind(plan.sms_rate) - .bind(plan.monthly_fee) - .bind(plan.data_limit as i64) - .bind(plan.voice_limit as i64) - .bind(plan.sms_limit as i64) - .execute(&self.pool) - .await - .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; - Ok(()) + let plan = plan.clone(); + self.circuit_breaker.execute(async move { + sqlx::query( + r#" + INSERT INTO rating_plans + (plan_id, name, data_rate, voice_rate, sms_rate, + monthly_fee, data_limit, voice_limit, sms_limit, is_active, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, TRUE, NOW()) + ON CONFLICT (plan_id) DO UPDATE SET + name = EXCLUDED.name, + data_rate = EXCLUDED.data_rate, + voice_rate = EXCLUDED.voice_rate, + sms_rate = EXCLUDED.sms_rate, + monthly_fee = EXCLUDED.monthly_fee, + data_limit = EXCLUDED.data_limit, + voice_limit = EXCLUDED.voice_limit, + sms_limit = EXCLUDED.sms_limit, + is_active = TRUE, + updated_at = NOW() + "#, + ) + .bind(&plan.plan_id) + .bind(&plan.name) + .bind(plan.data_rate) + .bind(plan.voice_rate) + .bind(plan.sms_rate) + .bind(plan.monthly_fee) + .bind(plan.data_limit as i64) + .bind(plan.voice_limit as i64) + .bind(plan.sms_limit as i64) + .execute(&self.pool) + .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, + }) } /// Soft-delete a plan by marking it inactive. pub async fn deactivate(&self, plan_id: &str) -> ChargingResult { - let result = sqlx::query( - r#"UPDATE rating_plans SET is_active = FALSE, updated_at = NOW() WHERE plan_id = $1"#, - ) - .bind(plan_id) - .execute(&self.pool) - .await - .map_err(|e| crate::errors::ChargingError::DatabaseError(e.to_string()))?; - Ok(result.rows_affected() > 0) + let plan_id = plan_id.to_string(); + 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"#, + ) + .bind(&plan_id) + .execute(&self.pool) + .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, + }) } } diff --git a/apps/charging-engine/src/lib.rs b/apps/charging-engine/src/lib.rs index 99cb2d1..6d9cfec 100644 --- a/apps/charging-engine/src/lib.rs +++ b/apps/charging-engine/src/lib.rs @@ -1,5 +1,6 @@ pub mod auth; pub mod charging; +pub mod circuit_breaker; pub mod errors; pub mod handlers; pub mod routes;