mirror of
https://github.com/andwati/telecom-platform.git
synced 2026-10-20 11:54:25 +00:00
feat: Refactor ChargingEngine to use RatingPlansRepo for Postgres-backed rating plans and multiplexed Redis connections
- Replace in-memory HashMap rating plans with RatingPlansRepo for Postgres-backed storage - Update constructor to accept RatingPlansRepo and remove hardcoded default plans (basic/premium/enterprise) - Make redis_client/plans/sync_interval/startup_time fields public(crate) for module access - Convert get_rating_plan/add_rating_plan/remove_rating_plan/list_rating_plans to async methods using
This commit is contained in:
@@ -1,101 +1,93 @@
|
||||
use std::collections::HashMap;
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use super::types::{RatingPlan, SubscriberAccount};
|
||||
use crate::errors::{ChargingError, ChargingResult, ErrorContext};
|
||||
use tracing::info;
|
||||
|
||||
use super::rating_plans_repo::RatingPlansRepo;
|
||||
use super::types::RatingPlan;
|
||||
use crate::errors::{ChargingResult, ErrorContext, ChargingError};
|
||||
|
||||
/// ChargingEngine coordinates Redis-backed hot-path state (balances, sessions)
|
||||
/// and Postgres-backed configuration (rating plans) via `RatingPlansRepo`.
|
||||
pub struct ChargingEngine {
|
||||
redis_client: redis::Client,
|
||||
rating_plans: HashMap<String, RatingPlan>,
|
||||
sync_interval: Duration,
|
||||
startup_time: SystemTime,
|
||||
pub(crate) redis_client: redis::Client,
|
||||
pub(crate) plans: RatingPlansRepo,
|
||||
pub(crate) sync_interval: Duration,
|
||||
pub(crate) startup_time: SystemTime,
|
||||
}
|
||||
|
||||
impl ChargingEngine {
|
||||
pub fn new(redis_url: &str, sync_interval_secs: u64) -> ChargingResult<Self> {
|
||||
/// Construct the engine with a Redis hot store and a Postgres-backed plan repo.
|
||||
/// The plan repo migration and default seeding must be done by the caller via
|
||||
/// [`RatingPlansRepo::connect`] + [`RatingPlansRepo::seed_defaults`].
|
||||
pub fn new(
|
||||
redis_url: &str,
|
||||
plans: RatingPlansRepo,
|
||||
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
|
||||
});
|
||||
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
|
||||
|
||||
Ok(Self {
|
||||
redis_client,
|
||||
rating_plans,
|
||||
plans,
|
||||
sync_interval: Duration::from_secs(sync_interval_secs),
|
||||
startup_time: SystemTime::now(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn get_rating_plan(&self, plan_id: &str) -> Option<&RatingPlan> {
|
||||
self.rating_plans.get(plan_id)
|
||||
/// 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
|
||||
}
|
||||
|
||||
pub fn add_rating_plan(&mut self, plan: RatingPlan) {
|
||||
self.rating_plans.insert(plan.plan_id.clone(), plan);
|
||||
/// Upsert a rating plan in Postgres.
|
||||
pub async fn add_rating_plan(&self, plan: RatingPlan) -> ChargingResult<()> {
|
||||
self.plans.upsert(&plan).await
|
||||
}
|
||||
|
||||
pub fn remove_rating_plan(&mut self, plan_id: &str) -> Option<RatingPlan> {
|
||||
self.rating_plans.remove(plan_id)
|
||||
/// 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
|
||||
}
|
||||
|
||||
pub fn list_rating_plans(&self) -> Vec<&RatingPlan> {
|
||||
self.rating_plans.values().collect()
|
||||
/// 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
|
||||
.redis_client
|
||||
.get_multiplexed_async_connection()
|
||||
.await
|
||||
.map_err(|e| crate::errors::ChargingError::RedisConnection(e.to_string()))?;
|
||||
let _: String = redis::AsyncCommands::get(&mut conn, "startup_probe")
|
||||
.await
|
||||
.unwrap_or_else(|_| "ok".to_string());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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!(
|
||||
"Starting charging engine with sync interval: {:?}",
|
||||
self.sync_interval
|
||||
);
|
||||
self.test_connection().await?;
|
||||
info!("Charging engine started successfully");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn stop(&self) -> ChargingResult<()> {
|
||||
info!("Stopping charging engine");
|
||||
// Cleanup operations
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Process-uptime since construction.
|
||||
pub fn uptime(&self) -> Duration {
|
||||
self.startup_time
|
||||
.elapsed()
|
||||
.unwrap_or_else(|_| Duration::from_secs(0))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user