//! Configuration management with PostgreSQL NOTIFY/LISTEN hot-reload and Redis caching use crate::config::validator::ConfigValidator; use crate::error::{GatewayConfigError, GatewayConfigResult}; use chrono::{DateTime, Utc}; use redis::aio::ConnectionManager; use serde_json::Value; use sqlx::PgPool; use std::sync::Arc; use tokio::sync::RwLock; use tracing::{debug, info}; use uuid::Uuid; /// Configuration item from database #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, sqlx::FromRow)] pub struct ConfigItem { pub id: Uuid, pub service_scope: String, pub config_key: String, pub config_value: Value, pub data_type: String, pub validation_rules: Option, pub description: Option, pub is_active: bool, pub created_at: DateTime, pub updated_at: DateTime, pub updated_by: Option, } /// Centralized configuration manager with hot-reload and caching pub struct ConfigurationManager { /// Postgre`SQL` connection pool db_pool: Arc, /// Redis connection manager for caching redis: Arc>, /// Configuration validator validator: Arc>, /// Postgre`SQL` NOTIFY listener listener: Option, } impl ConfigurationManager { /// Creates a new ConfigurationManager /// /// # Arguments /// * `db_pool` - Postgre`SQL` connection pool /// /// * `redis` - Redis connection manager pub async fn new(db_pool: PgPool, redis: ConnectionManager) -> GatewayConfigResult { Ok(Self { db_pool: Arc::new(db_pool), redis: Arc::new(RwLock::new(redis)), validator: Arc::new(RwLock::new(ConfigValidator::new())), listener: None, }) } /// Starts listening for configuration changes via Postgre`SQL` NOTIFY pub async fn start_listening(&mut self) -> GatewayConfigResult<()> { let mut listener = sqlx::postgres::PgListener::connect_with(&self.db_pool).await?; // Listen to global config updates channel listener.listen("config_updates_global").await?; info!("Started listening for configuration updates on 'config_updates_global'"); self.listener = Some(listener); Ok(()) } /// Processes configuration change notifications pub async fn handle_notifications(&mut self) -> GatewayConfigResult<()> { // Collect notifications first to avoid borrow checker issues let mut invalidations = Vec::new(); if let Some(listener) = &mut self.listener { while let Some(notification) = listener.try_recv().await? { let payload = notification.payload(); debug!("Received config notification: {}", payload); // Parse notification payload if let Ok(payload_json) = serde_json::from_str::(payload) { if let (Some(service_scope), Some(config_key)) = ( payload_json.get("service_scope").and_then(|v| v.as_str()), payload_json.get("config_key").and_then(|v| v.as_str()), ) { invalidations.push((service_scope.to_string(), config_key.to_string())); } } } } // Now invalidate caches without holding the listener borrow for (service_scope, config_key) in invalidations { self.invalidate_cache(&service_scope, &config_key).await?; info!("Invalidated cache for {}/{}", service_scope, config_key); } Ok(()) } /// Retrieves a configuration value /// /// # Arguments /// * `service_scope` - Service scope (e.g., "trading", "global") /// /// * `config_key` - Configuration key /// /// # Returns /// /// Configuration item if found pub async fn get_config( &self, service_scope: &str, config_key: &str, ) -> GatewayConfigResult { // Try Redis cache first if let Some(cached) = self.get_from_cache(service_scope, config_key).await? { debug!("Cache hit for {}/{}", service_scope, config_key); return Ok(cached); } // Cache miss - load from PostgreSQL debug!("Cache miss for {}/{}", service_scope, config_key); let config = self.load_from_db(service_scope, config_key).await?; // Update Redis cache self.set_in_cache(&config).await?; Ok(config) } /// Updates a configuration value with validation and audit logging /// /// # Arguments /// * `service_scope` - Service scope /// /// * `config_key` - Configuration key /// * `new_value` - New configuration value /// /// * `updated_by` - User ID making the change pub async fn update_config( &self, service_scope: &str, config_key: &str, new_value: Value, updated_by: &str, ) -> GatewayConfigResult<()> { // Load current configuration for validation rules and audit let current = self.load_from_db(service_scope, config_key).await?; // Validate new value { let mut validator = self.validator.write().await; validator.validate( &new_value, ¤t.data_type, current.validation_rules.as_ref(), )?; } // Start transaction let mut tx = self.db_pool.begin().await?; // Update config_settings sqlx::query( r#" UPDATE config_settings SET config_value = $1, updated_by = $2, updated_at = NOW() WHERE service_scope = $3 AND config_key = $4 "#, ) .bind(&new_value) .bind(updated_by) .bind(service_scope) .bind(config_key) .execute(&mut *tx) .await?; // Insert audit log entry sqlx::query( r#" INSERT INTO config_audit_log (action, service_scope, config_key, old_value, new_value, changed_by) VALUES ('UPDATE', $1, $2, $3, $4, $5) "#, ) .bind(service_scope) .bind(config_key) .bind(¤t.config_value) .bind(&new_value) .bind(updated_by) .execute(&mut *tx) .await?; // Commit transaction (this triggers NOTIFY via database trigger) tx.commit().await?; info!( "Updated configuration {}/{} by {}", service_scope, config_key, updated_by ); Ok(()) } /// Lists all configurations for a service scope /// /// # Arguments /// * `service_scope` - Service scope (None for all scopes) pub async fn list_configs(&self, service_scope: Option<&str>) -> GatewayConfigResult> { let configs = if let Some(scope) = service_scope { sqlx::query_as::<_, ConfigItem>( r#" SELECT * FROM config_settings WHERE service_scope = $1 AND is_active = TRUE ORDER BY config_key "#, ) .bind(scope) .fetch_all(&*self.db_pool) .await? } else { sqlx::query_as::<_, ConfigItem>( r#" SELECT * FROM config_settings WHERE is_active = TRUE ORDER BY service_scope, config_key "#, ) .fetch_all(&*self.db_pool) .await? }; Ok(configs) } /// Loads configuration from Postgre`SQL` database async fn load_from_db( &self, service_scope: &str, config_key: &str, ) -> GatewayConfigResult { let config = sqlx::query_as::<_, ConfigItem>( r#" SELECT * FROM config_settings WHERE service_scope = $1 AND config_key = $2 AND is_active = TRUE "#, ) .bind(service_scope) .bind(config_key) .fetch_optional(&*self.db_pool) .await? .ok_or_else(|| GatewayConfigError::NotFound { service_scope: service_scope.to_string(), key: config_key.to_string(), })?; Ok(config) } /// Retrieves configuration from Redis cache async fn get_from_cache( &self, service_scope: &str, config_key: &str, ) -> GatewayConfigResult> { let redis_key = format!("config:{}:{}", service_scope, config_key); let mut redis = self.redis.write().await; let cached: Option = redis::cmd("GET") .arg(&redis_key) .query_async(&mut *redis) .await .map_err(GatewayConfigError::Redis)?; if let Some(cached_json) = cached { let config: ConfigItem = serde_json::from_str(&cached_json)?; Ok(Some(config)) } else { Ok(None) } } /// Stores configuration in Redis cache async fn set_in_cache(&self, config: &ConfigItem) -> GatewayConfigResult<()> { let redis_key = format!("config:{}:{}", config.service_scope, config.config_key); let config_json = serde_json::to_string(config)?; let mut redis = self.redis.write().await; // Set with 5-minute TTL redis::cmd("SETEX") .arg(&redis_key) .arg(300) // 5 minutes .arg(&config_json) .query_async::<()>(&mut *redis) .await .map_err(GatewayConfigError::Redis)?; Ok(()) } /// Invalidates Redis cache for a configuration async fn invalidate_cache(&self, service_scope: &str, config_key: &str) -> GatewayConfigResult<()> { let redis_key = format!("config:{}:{}", service_scope, config_key); let mut redis = self.redis.write().await; redis::cmd("DEL") .arg(&redis_key) .query_async::<()>(&mut *redis) .await .map_err(GatewayConfigError::Redis)?; Ok(()) } } #[cfg(test)] mod tests { // Note: These tests require a running PostgreSQL and Redis instance // They are integration tests and should be run with --ignored flag }