foxhunt/crates/config/src/database.rs

//! PostgreSQL-based Configuration Database Module
//!
//! This module provides PostgreSQL-backed configuration storage with:
//! - Hot-reload via PostgreSQL NOTIFY/LISTEN
//! - In-memory caching with TTL for performance
//! - Type-safe configuration getters
//! - Automatic table creation and migration
//! - Configuration change notifications

use crate::{ConfigCategory, ConfigError, ConfigResult, ConfigValue, ConfigSource};
use anyhow::Context;
// Utc removed - not used in current implementation
use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Row};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, RwLock};
use tokio::time::interval;
use tracing::{debug, error, info, warn};

/// Database configuration for PostgreSQL connection
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabaseConfig {
    /// PostgreSQL connection URL
    pub url: String,
    /// Maximum number of connections in the pool
    pub max_connections: u32,
    /// Connection timeout in seconds
    pub connect_timeout: u64,
    /// Query timeout in seconds
    pub query_timeout: u64,
    /// Whether to run migrations on startup
    pub auto_migrate: bool,
    /// Enable query logging
    pub enable_query_logging: bool,
    /// Enable metrics collection
    pub enable_metrics: bool,
    /// Application name for connection identification
    pub application_name: String,
}

impl DatabaseConfig {
    /// Create database configuration from environment variables
    pub fn from_env() -> ConfigResult<Self> {
        let url = std::env::var("DATABASE_URL")
            .or_else(|_| std::env::var("FOXHUNT_POSTGRES_URL"))
            .unwrap_or_else(|_| "postgresql://postgres:password@localhost/foxhunt".to_string());

        let max_connections = std::env::var("DATABASE_MAX_CONNECTIONS")
            .unwrap_or_else(|_| "10".to_string())
            .parse()
            .unwrap_or(10);

        let connect_timeout = std::env::var("DATABASE_CONNECT_TIMEOUT")
            .unwrap_or_else(|_| "30".to_string())
            .parse()
            .unwrap_or(30);

        let query_timeout = std::env::var("DATABASE_QUERY_TIMEOUT")
            .unwrap_or_else(|_| "60".to_string())
            .parse()
            .unwrap_or(60);

        let auto_migrate = std::env::var("DATABASE_AUTO_MIGRATE")
            .unwrap_or_else(|_| "true".to_string())
            .parse()
            .unwrap_or(true);

        let enable_query_logging = std::env::var("DATABASE_ENABLE_QUERY_LOGGING")
            .unwrap_or_else(|_| "false".to_string())
            .parse()
            .unwrap_or(false);
        
        let enable_metrics = std::env::var("DATABASE_ENABLE_METRICS")
            .unwrap_or_else(|_| "true".to_string())
            .parse()
            .unwrap_or(true);
        
        let application_name = std::env::var("DATABASE_APPLICATION_NAME")
            .unwrap_or_else(|_| "foxhunt-database".to_string());
        
        Ok(Self {
            url,
            max_connections,
            connect_timeout,
            query_timeout,
            auto_migrate,
            enable_query_logging,
            enable_metrics,
            application_name,
        })
    }
}

impl Default for DatabaseConfig {
    fn default() -> Self {
        Self {
            url: "postgresql://postgres:password@localhost/foxhunt".to_string(),
            max_connections: 10,
            connect_timeout: 30,
            query_timeout: 60,
            auto_migrate: true,
            enable_query_logging: false,
            enable_metrics: true,
            application_name: "foxhunt".to_string(),
        }
    }
}

/// Cached configuration entry with TTL
#[derive(Debug, Clone)]
struct CachedConfig {
    /// The configuration value
    value: ConfigValue,
    /// When this entry was cached
    cached_at: Instant,
    /// TTL for this entry
    ttl: Duration,
}

impl CachedConfig {
    /// Check if this cached entry has expired
    fn is_expired(&self) -> bool {
        self.cached_at.elapsed() > self.ttl
    }
}

/// PostgreSQL Configuration Loader with hot-reload support
pub struct PostgresConfigLoader {
    /// PostgreSQL connection pool
    pool: PgPool,
    /// In-memory cache of configurations
    cache: Arc<RwLock<HashMap<(ConfigCategory, String), CachedConfig>>>,
    /// Default TTL for cached entries
    default_ttl: Duration,
    /// Channel for hot-reload notifications
    reload_tx: mpsc::UnboundedSender<(ConfigCategory, String)>,
    /// Receiver for hot-reload notifications (for internal use)
    reload_rx: Arc<RwLock<Option<mpsc::UnboundedReceiver<(ConfigCategory, String)>>>>,
    /// Environment for configuration loading
    environment: String,
}

impl PostgresConfigLoader {
    /// Create a new PostgreSQL configuration loader
    pub async fn new(config: DatabaseConfig, default_ttl: Duration) -> ConfigResult<Self> {
        let pool = PgPool::connect(&config.url)
            .await
            .context("Failed to connect to PostgreSQL")?;

        let environment = std::env::var("FOXHUNT_ENV").unwrap_or_else(|_| "development".to_string());

        let (reload_tx, reload_rx) = mpsc::unbounded_channel();

        let loader = Self {
            pool,
            cache: Arc::new(RwLock::new(HashMap::new())),
            default_ttl,
            reload_tx,
            reload_rx: Arc::new(RwLock::new(Some(reload_rx))),
            environment,
        };

        // Create configuration tables and triggers if needed
        if config.auto_migrate {
            loader.create_tables().await?;
        }

        // Start the hot-reload listener
        loader.start_notify_listener().await?;

        info!(
            "PostgreSQL ConfigLoader initialized for environment '{}' with TTL {:?}",
            loader.environment, default_ttl
        );

        Ok(loader)
    }

    /// Create configuration tables and triggers if they don't exist
    async fn create_tables(&self) -> ConfigResult<()> {
        let categories = [
            ConfigCategory::Trading,
            ConfigCategory::Risk,
            ConfigCategory::MarketData,
            ConfigCategory::MachineLearning,
            ConfigCategory::Brokers,
            ConfigCategory::Performance,
            ConfigCategory::Security,
            ConfigCategory::Environment,
        ];

        for category in &categories {
            let table_name = category.table_name();
            let sql = format!(
                r#"
                CREATE TABLE IF NOT EXISTS {} (
                    id SERIAL PRIMARY KEY,
                    key VARCHAR(255) NOT NULL,
                    value JSONB NOT NULL,
                    environment VARCHAR(50) NOT NULL DEFAULT 'development',
                    description TEXT,
                    is_active BOOLEAN DEFAULT TRUE,
                    created_at TIMESTAMPTZ DEFAULT NOW(),
                    updated_at TIMESTAMPTZ DEFAULT NOW(),
                    UNIQUE(key, environment)
                );
                
                CREATE INDEX IF NOT EXISTS idx_{}_key_env ON {} (key, environment);
                CREATE INDEX IF NOT EXISTS idx_{}_updated_at ON {} (updated_at);
                CREATE INDEX IF NOT EXISTS idx_{}_active ON {} (is_active) WHERE is_active = TRUE;
                
                CREATE OR REPLACE FUNCTION notify_{}_changes()
                RETURNS trigger AS $$
                BEGIN
                    PERFORM pg_notify('{}', NEW.key || ':' || NEW.environment);
                    RETURN NEW;
                END;
                $$ LANGUAGE plpgsql;
                
                DROP TRIGGER IF EXISTS {}_notify_trigger ON {};
                CREATE TRIGGER {}_notify_trigger
                    AFTER INSERT OR UPDATE ON {}
                    FOR EACH ROW EXECUTE FUNCTION notify_{}_changes();
                "#,
                table_name,
                table_name, table_name,
                table_name, table_name,
                table_name, table_name,
                table_name,
                category.notify_channel(),
                table_name, table_name,
                table_name,
                table_name,
                table_name
            );

            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .with_context(|| format!("Failed to create table {}", table_name))?;
        }

        info!("Configuration tables and triggers created successfully");
        Ok(())
    }

    /// Start the PostgreSQL NOTIFY listener for hot-reload
    async fn start_notify_listener(&self) -> ConfigResult<()> {
        let pool = self.pool.clone();
        let reload_tx = self.reload_tx.clone();

        tokio::spawn(async move {
            let mut listener = match sqlx::postgres::PgListener::connect_with(&pool).await {
                Ok(listener) => listener,
                Err(e) => {
                    error!("Failed to create NOTIFY listener: {}", e);
                    return;
                }
            };

            // Subscribe to all configuration change channels
            let categories = [
                ConfigCategory::Trading,
                ConfigCategory::Risk,
                ConfigCategory::MarketData,
                ConfigCategory::MachineLearning,
                ConfigCategory::Brokers,
                ConfigCategory::Performance,
                ConfigCategory::Security,
                ConfigCategory::Environment,
            ];

            for category in &categories {
                if let Err(e) = listener.listen(category.notify_channel()).await {
                    error!(
                        "Failed to listen on channel {}: {}",
                        category.notify_channel(),
                        e
                    );
                    return;
                }
            }

            info!("NOTIFY listener started for configuration hot-reload");

            loop {
                match listener.recv().await {
                    Ok(notification) => {
                        let channel = notification.channel();
                        let payload = notification.payload();

                        debug!("Received NOTIFY on channel {}: {}", channel, payload);

                        // Determine which category was updated
                        let category = match channel {
                            "config_trading_changes" => ConfigCategory::Trading,
                            "config_risk_changes" => ConfigCategory::Risk,
                            "config_market_data_changes" => ConfigCategory::MarketData,
                            "config_ml_changes" => ConfigCategory::MachineLearning,
                            "config_broker_changes" => ConfigCategory::Brokers,
                            "config_performance_changes" => ConfigCategory::Performance,
                            "config_security_changes" => ConfigCategory::Security,
                            "config_environment_changes" => ConfigCategory::Environment,
                            _ => {
                                warn!("Unknown notification channel: {}", channel);
                                continue;
                            }
                        };

                        // Extract key from payload (format: "key:environment")
                        let key = payload.split(':').next().unwrap_or(payload).to_string();

                        // Send reload notification
                        if let Err(e) = reload_tx.send((category, key)) {
                            error!("Failed to send reload notification: {}", e);
                            break;
                        }
                    }
                    Err(e) => {
                        error!("Error receiving NOTIFY: {}", e);
                        tokio::time::sleep(Duration::from_secs(1)).await;
                    }
                }
            }
        });

        // Start cache cleanup task
        self.start_cache_cleanup().await;

        Ok(())
    }

    /// Start background task to clean up expired cache entries
    async fn start_cache_cleanup(&self) {
        let cache = self.cache.clone();
        let cleanup_interval = self.default_ttl / 4; // Clean up 4x more frequently than TTL

        tokio::spawn(async move {
            let mut interval = interval(cleanup_interval);

            loop {
                interval.tick().await;

                let mut cache_guard = cache.write().await;
                let initial_size = cache_guard.len();

                cache_guard.retain(|_, cached| !cached.is_expired());

                let final_size = cache_guard.len();
                if initial_size != final_size {
                    debug!(
                        "Cache cleanup: removed {} expired entries",
                        initial_size - final_size
                    );
                }
            }
        });
    }

    /// Get a configuration value with caching
    pub async fn get_config<T>(&self, category: ConfigCategory, key: &str) -> ConfigResult<Option<T>>
    where
        T: for<'de> Deserialize<'de>,
    {
        let cache_key = (category.clone(), key.to_string());

        // Check cache first
        {
            let cache_guard = self.cache.read().await;
            if let Some(cached) = cache_guard.get(&cache_key) {
                if !cached.is_expired() {
                    debug!("Cache hit for {}.{}", category.table_name(), key);
                    return Ok(Some(serde_json::from_value(cached.value.value.clone())?));
                }
            }
        }

        // Cache miss or expired - fetch from database
        debug!(
            "Cache miss for {}.{}, fetching from database",
            category.table_name(),
            key
        );

        let table_name = category.table_name();
        let sql = format!(
            "SELECT key, value, updated_at, description, is_active FROM {} 
             WHERE key = $1 AND environment = $2 AND is_active = TRUE 
             ORDER BY updated_at DESC LIMIT 1",
            table_name
        );

        let row = sqlx::query(&sql)
            .bind(key)
            .bind(&self.environment)
            .fetch_optional(&self.pool)
            .await
            .with_context(|| format!("Failed to fetch config {}.{}", table_name, key))?;

        if let Some(row) = row {
            let config_value = ConfigValue {
                key: row.try_get("key")?,
                value: row.try_get("value")?,
                category: category.clone(),
                environment: self.environment.clone(),
                updated_at: row.try_get("updated_at")?,
                description: row.try_get("description")?,
                is_active: row.try_get("is_active")?,
                source: ConfigSource::Database,
            };

            // Cache the result
            let cached = CachedConfig {
                value: config_value.clone(),
                cached_at: Instant::now(),
                ttl: self.default_ttl,
            };

            {
                let mut cache_guard = self.cache.write().await;
                cache_guard.insert(cache_key, cached);
            }

            Ok(Some(serde_json::from_value(config_value.value)?))
        } else {
            Ok(None)
        }
    }

    /// Set a configuration value
    pub async fn set_config<T>(
        &self,
        category: ConfigCategory,
        key: &str,
        value: &T,
        description: Option<&str>,
    ) -> ConfigResult<()>
    where
        T: Serialize,
    {
        self.set_config_for_environment(category, key, value, &self.environment, description)
            .await
    }

    /// Set a configuration value for a specific environment
    pub async fn set_config_for_environment<T>(
        &self,
        category: ConfigCategory,
        key: &str,
        value: &T,
        environment: &str,
        description: Option<&str>,
    ) -> ConfigResult<()>
    where
        T: Serialize,
    {
        let json_value = serde_json::to_value(value)?;
        let table_name = category.table_name();

        let sql = format!(
            "INSERT INTO {} (key, value, environment, description, updated_at) 
             VALUES ($1, $2, $3, $4, NOW()) 
             ON CONFLICT (key, environment) 
             DO UPDATE SET value = $2, description = $4, updated_at = NOW()",
            table_name
        );

        sqlx::query(&sql)
            .bind(key)
            .bind(&json_value)
            .bind(environment)
            .bind(description)
            .execute(&self.pool)
            .await
            .with_context(|| format!("Failed to set config {}.{}", table_name, key))?;

        // Invalidate cache entry
        let cache_key = (category, key.to_string());
        {
            let mut cache_guard = self.cache.write().await;
            cache_guard.remove(&cache_key);
        }

        info!("Updated configuration {}.{} for {}", table_name, key, environment);
        Ok(())
    }

    /// Get all configurations for a category
    pub async fn get_category_configs(&self, category: ConfigCategory) -> ConfigResult<Vec<ConfigValue>> {
        self.get_category_configs_for_environment(category, &self.environment).await
    }

    /// Get all configurations for a category and environment
    pub async fn get_category_configs_for_environment(
        &self, 
        category: ConfigCategory, 
        environment: &str
    ) -> ConfigResult<Vec<ConfigValue>> {
        let table_name = category.table_name();
        let sql = format!(
            "SELECT key, value, updated_at, description, is_active FROM {} 
             WHERE environment = $1 AND is_active = TRUE 
             ORDER BY key",
            table_name
        );

        let rows = sqlx::query(&sql)
            .bind(environment)
            .fetch_all(&self.pool)
            .await
            .with_context(|| format!("Failed to fetch configs for category {}", table_name))?;

        let mut configs = Vec::new();
        for row in rows {
            configs.push(ConfigValue {
                key: row.try_get("key")?,
                value: row.try_get("value")?,
                category: category.clone(),
                environment: environment.to_string(),
                updated_at: row.try_get("updated_at")?,
                description: row.try_get("description")?,
                is_active: row.try_get("is_active")?,
                source: ConfigSource::Database,
            });
        }

        Ok(configs)
    }

    /// Subscribe to configuration changes (returns receiver for hot-reload notifications)
    pub async fn subscribe_to_changes(
        &self,
    ) -> ConfigResult<mpsc::UnboundedReceiver<(ConfigCategory, String)>> {
        let mut reload_rx_guard = self.reload_rx.write().await;
        reload_rx_guard
            .take()
            .ok_or_else(|| ConfigError::ValidationError {
                message: "Configuration change subscription already taken".to_string(),
            })
    }

    /// Get cache statistics
    pub async fn cache_stats(&self) -> (usize, usize) {
        let cache_guard = self.cache.read().await;
        let total = cache_guard.len();
        let expired = cache_guard.values().filter(|c| c.is_expired()).count();
        (total, expired)
    }

    /// Clear the entire cache
    pub async fn clear_cache(&self) {
        let mut cache_guard = self.cache.write().await;
        let size = cache_guard.len();
        cache_guard.clear();
        info!("Cleared {} entries from configuration cache", size);
    }

    /// Get current environment
    pub fn environment(&self) -> &str {
        &self.environment
    }

    /// Test database connection
    pub async fn test_connection(&self) -> ConfigResult<()> {
        sqlx::query("SELECT 1")
            .fetch_one(&self.pool)
            .await
            .context("Failed to test database connection")?;
        Ok(())
    }
}

/// Type-safe configuration getters for common configuration parameters
impl PostgresConfigLoader {
    /// Get trading configuration
    pub async fn get_trading_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::Trading, key).await
    }

    /// Get risk configuration
    pub async fn get_risk_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::Risk, key).await
    }

    /// Get market data configuration
    pub async fn get_market_data_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::MarketData, key).await
    }

    /// Get machine learning configuration
    pub async fn get_ml_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::MachineLearning, key).await
    }

    /// Get broker configuration
    pub async fn get_broker_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::Brokers, key).await
    }

    /// Get performance configuration
    pub async fn get_performance_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::Performance, key).await
    }

    /// Get security configuration
    pub async fn get_security_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::Security, key).await
    }

    /// Get environment configuration
    pub async fn get_environment_config(&self, key: &str) -> ConfigResult<Option<serde_json::Value>> {
        self.get_config(ConfigCategory::Environment, key).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    #[test]
    fn test_database_config_from_env() {
        std::env::set_var("DATABASE_URL", "postgresql://test:test@localhost/test");
        std::env::set_var("DATABASE_MAX_CONNECTIONS", "5");
        std::env::set_var("DATABASE_CONNECT_TIMEOUT", "10");

        let config = DatabaseConfig::from_env().unwrap();
        assert_eq!(config.url, "postgresql://test:test@localhost/test");
        assert_eq!(config.max_connections, 5);
        assert_eq!(config.connect_timeout, 10);

        std::env::remove_var("DATABASE_URL");
        std::env::remove_var("DATABASE_MAX_CONNECTIONS");
        std::env::remove_var("DATABASE_CONNECT_TIMEOUT");
    }

    #[test]
    fn test_cached_config_expiry() {
        let config_value = ConfigValue {
            key: "test".to_string(),
            value: serde_json::json!("test_value"),
            category: ConfigCategory::Trading,
            environment: "test".to_string(),
            updated_at: Utc::now(),
            description: None,
            is_active: true,
            source: ConfigSource::Database,
        };

        let cached = CachedConfig {
            value: config_value.clone(),
            cached_at: Instant::now() - Duration::from_secs(10),
            ttl: Duration::from_secs(5),
        };

        assert!(cached.is_expired());

        let fresh_cached = CachedConfig {
            value: config_value,
            cached_at: Instant::now(),
            ttl: Duration::from_secs(60),
        };

        assert!(!fresh_cached.is_expired());
    }
}