mirror of
https://github.com/andwati/telecom-platform.git
synced 2026-10-23 21:22:16 +00:00
feat: Wrap Redis and PostgreSQL operations in circuit breakers for fault tolerance
- Add CircuitBreakerError import and error mapping to ChargingError::RedisConnection/DatabaseError - Add redis_circuit_breaker and postgres_circuit_breaker fields to ChargingEngine struct - Add circuit_breaker field to RatingPlansRepo struct - Wrap all Redis operations (get_balance, add_credit, deduct_credit, block_user, unblock_user, is_user_blocked, get_subscriber_account, update_subscriber_account, record_usage
This commit is contained in:
nutcas3
@@ -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<u64> {
|
||||
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<u64> {
|
||||
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<u64> = 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<u64> = 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<u64> {
|
||||
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<bool> {
|
||||
@@ -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<String> = 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<String> = 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<bool> {
|
||||
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<String> = 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<String> = 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<Option<SubscriberAccount>> {
|
||||
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<SubscriberAccount> = 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<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)
|
||||
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,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)),
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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<Option<RatingPlan>> {
|
||||
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<HashMap<String, RatingPlan>> {
|
||||
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<bool> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
pub mod auth;
|
||||
pub mod charging;
|
||||
pub mod circuit_breaker;
|
||||
pub mod errors;
|
||||
pub mod handlers;
|
||||
pub mod routes;
|
||||
|
||||
Reference in New Issue
Block a user