foxhunt/risk/src/safety/kill_switch.rs

//! Kill switch implementations for emergency stops

use super::KillSwitchConfig;
use crate::error::{RiskError, RiskResult};
use crate::risk_types::KillSwitchScope;
use chrono::Utc;
use redis::{AsyncCommands, Client as RedisClient};
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::RwLock;

/// Atomic kill switch for emergency trading stops
#[derive(Debug)]
pub struct AtomicKillSwitch {
    triggered: Arc<AtomicBool>,
    config: KillSwitchConfig,
    redis_client: Option<RedisClient>, // Optional for tests
    scoped_triggers: Arc<RwLock<HashMap<String, bool>>>,
    // Metrics tracking
    health_check_count: Arc<AtomicU64>,
    command_count: Arc<AtomicU64>,
    failure_count: Arc<AtomicU64>,
}

impl AtomicKillSwitch {
    /// Create a new `AtomicKillSwitch` with Redis connectivity
    pub async fn new(config: KillSwitchConfig, redis_url: String) -> RiskResult<Self> {
        let redis_client = RedisClient::open(redis_url)
            .map_err(|e| RiskError::Config(format!("Failed to connect to Redis: {e}")))?;

        // Test Redis connectivity
        let mut conn = redis_client
            .get_multiplexed_async_connection()
            .await
            .map_err(|e| RiskError::Config(format!("Failed to establish Redis connection: {e}")))?;

        // Verify Redis is accessible using AsyncCommands trait
        redis::cmd("PING")
            .exec_async(&mut conn)
            .await
            .map_err(|e| RiskError::Config(format!("Redis ping failed: {e}")))?;

        Ok(Self {
            triggered: Arc::new(AtomicBool::new(false)),
            config,
            redis_client: Some(redis_client),
            scoped_triggers: Arc::new(RwLock::new(HashMap::new())),
            health_check_count: Arc::new(AtomicU64::new(0)),
            command_count: Arc::new(AtomicU64::new(0)),
            failure_count: Arc::new(AtomicU64::new(0)),
        })
    }

    /// Engage the kill switch for a specific scope
    pub async fn engage(
        &self,
        scope: KillSwitchScope,
        reason: String,
        user_id: String,
        cascade: bool,
    ) -> RiskResult<()> {
        // Track command execution
        self.command_count.fetch_add(1, Ordering::Relaxed);

        // Set local state immediately
        if scope == KillSwitchScope::Global {
            self.triggered.store(true, Ordering::SeqCst);
        } else {
            let scope_key = self.scope_to_key(&scope);
            let mut scoped = self.scoped_triggers.write().await;
            scoped.insert(scope_key.clone(), true);

            // Implement cascade logic
            if cascade {
                match &scope {
                    KillSwitchScope::Portfolio(id) => {
                        // Cascade: Portfolio halt also halts all its strategies
                        // Store cascade flag for broader halt interpretation
                        scoped.insert(format!("cascade:portfolio:{id}"), true);
                    },
                    KillSwitchScope::Strategy(id) => {
                        // Cascade: Strategy halt can affect related strategies
                        scoped.insert(format!("cascade:strategy:{id}"), true);
                    },
                    _ => {},
                }
            }
        }

        // Broadcast to Redis for distributed coordination (if available)
        if let Some(ref client) = self.redis_client {
            match client.get_multiplexed_async_connection().await {
                Ok(mut conn) => {
                    let channel = self.scope_to_channel(&scope);
                    let message = serde_json::json!({
                        "action": "engage",
                        "scope": scope,
                        "reason": reason,
                        "user_id": user_id,
                        "cascade": cascade,
                        "timestamp": Utc::now().to_rfc3339()
                    });

                    if let Err(e) = conn
                        .publish::<_, _, ()>(&channel, message.to_string())
                        .await
                    {
                        self.failure_count.fetch_add(1, Ordering::Relaxed);
                        return Err(RiskError::Config(format!(
                            "Failed to publish to Redis: {e}"
                        )));
                    }
                },
                Err(e) => {
                    self.failure_count.fetch_add(1, Ordering::Relaxed);
                    return Err(RiskError::Config(format!(
                        "Failed to get Redis connection: {e}"
                    )));
                },
            }
        }

        Ok(())
    }

    /// Check if trading is allowed for a specific scope
    ///
    /// FAIL-SAFE MODE: If unable to verify status (lock contention), trading is BLOCKED
    /// This ensures safety - we never allow trading when we cannot confirm it's safe.
    #[must_use]
    pub fn is_trading_allowed(&self, scope: &KillSwitchScope) -> bool {
        // Check global kill switch first
        if self.triggered.load(Ordering::SeqCst) {
            return false;
        }

        // Check scoped kill switches
        match scope {
            KillSwitchScope::Global => true, // Already checked above
            _ => {
                let scope_key = self.scope_to_key(scope);
                // FAIL-SAFE: If we can't read the lock (contention), block trading
                // This is safer than allowing trading when we cannot verify status
                if let Ok(scoped) = self.scoped_triggers.try_read() {
                    // Check both the specific scope and cascade flags
                    let is_blocked = scoped.get(&scope_key).copied().unwrap_or(false);

                    // Check cascade halts that might affect this scope
                    let cascade_blocked = match scope {
                        KillSwitchScope::Strategy(_id) => {
                            // Check if parent portfolio has cascade halt
                            scoped.iter().any(|(k, &v)| {
                                v && k.starts_with("cascade:portfolio:")
                                // In production, would check if strategy belongs to halted portfolio
                            })
                        },
                        _ => false,
                    };

                    !is_blocked && !cascade_blocked
                } else {
                    // FAIL-SAFE: Cannot verify status -> block trading
                    false
                }
            },
        }
    }

    /// Trigger the kill switch
    pub fn trigger(&self) {
        self.triggered.store(true, Ordering::SeqCst);
    }

    /// Check if kill switch is triggered
    #[must_use]
    pub fn is_triggered(&self) -> bool {
        self.triggered.load(Ordering::SeqCst)
    }

    /// Reset the kill switch
    pub async fn reset(&self, scope: Option<KillSwitchScope>) -> RiskResult<()> {
        // Track command execution
        self.command_count.fetch_add(1, Ordering::Relaxed);

        match scope {
            Some(KillSwitchScope::Global) | None => {
                self.triggered.store(false, Ordering::SeqCst);
            },
            Some(ref s) => {
                let scope_key = self.scope_to_key(s);
                let mut scoped = self.scoped_triggers.write().await;
                scoped.remove(&scope_key);

                // Also remove cascade flags for this scope
                match s {
                    KillSwitchScope::Portfolio(id) => {
                        scoped.remove(&format!("cascade:portfolio:{id}"));
                    },
                    KillSwitchScope::Strategy(id) => {
                        scoped.remove(&format!("cascade:strategy:{id}"));
                    },
                    _ => {},
                }
            },
        }

        // Broadcast reset to Redis (if available)
        if let Some(scope) = scope {
            if let Some(ref client) = self.redis_client {
                match client.get_multiplexed_async_connection().await {
                    Ok(mut conn) => {
                        let channel = self.scope_to_channel(&scope);
                        let message = serde_json::json!({
                            "action": "reset",
                            "scope": scope,
                            "timestamp": Utc::now().to_rfc3339()
                        });

                        if let Err(e) = conn
                            .publish::<_, _, ()>(&channel, message.to_string())
                            .await
                        {
                            self.failure_count.fetch_add(1, Ordering::Relaxed);
                            return Err(RiskError::Config(format!(
                                "Failed to publish reset to Redis: {e}"
                            )));
                        }
                    },
                    Err(e) => {
                        self.failure_count.fetch_add(1, Ordering::Relaxed);
                        return Err(RiskError::Config(format!(
                            "Failed to get Redis connection: {e}"
                        )));
                    },
                }
            }
        }

        Ok(())
    }

    /// Convert scope to Redis channel name
    fn scope_to_channel(&self, scope: &KillSwitchScope) -> String {
        match scope {
            KillSwitchScope::Global => self.config.global_channel.clone(),
            KillSwitchScope::Portfolio(id) => {
                format!("{}:portfolio:{}", self.config.strategy_channel_prefix, id)
            },
            KillSwitchScope::Strategy(id) => {
                format!("{}:{}", self.config.strategy_channel_prefix, id)
            },
            KillSwitchScope::Instrument(id) => {
                format!("{}:instrument:{}", self.config.symbol_channel_prefix, id)
            },
            KillSwitchScope::Symbol(id) => format!("{}:{}", self.config.symbol_channel_prefix, id),
            KillSwitchScope::Account(id) => {
                format!("{}:account:{}", self.config.strategy_channel_prefix, id)
            },
        }
    }

    /// Convert scope to internal key
    fn scope_to_key(&self, scope: &KillSwitchScope) -> String {
        match scope {
            KillSwitchScope::Global => "global".to_owned(),
            KillSwitchScope::Portfolio(id) => format!("portfolio:{id}"),
            KillSwitchScope::Strategy(id) => format!("strategy:{id}"),
            KillSwitchScope::Instrument(id) => format!("instrument:{id}"),
            KillSwitchScope::Symbol(id) => format!("symbol:{id}"),
            KillSwitchScope::Account(id) => format!("account:{id}"),
        }
    }

    /// Activate global kill switch
    pub async fn activate_global(&self, reason: String, user: String) -> RiskResult<()> {
        self.engage(KillSwitchScope::Global, reason, user, true)
            .await
    }

    /// Deactivate a scoped kill switch
    pub async fn deactivate(&self, scope: KillSwitchScope, _user_id: String) -> RiskResult<()> {
        self.reset(Some(scope)).await
    }

    /// Check if the kill switch is active for any scope
    pub async fn is_active(&self) -> RiskResult<bool> {
        Ok(self.triggered.load(Ordering::SeqCst) || {
            if let Ok(scoped) = self.scoped_triggers.try_read() {
                scoped.values().any(|&active| active)
            } else {
                false
            }
        })
    }

    /// Check health status of the kill switch
    pub async fn is_healthy(&self) -> RiskResult<bool> {
        // Track health check
        self.health_check_count.fetch_add(1, Ordering::Relaxed);

        // If Redis is configured, try to ping it
        if let Some(ref client) = self.redis_client {
            if let Ok(mut conn) = client.get_multiplexed_async_connection().await {
                if let Ok(()) = redis::cmd("PING").exec_async(&mut conn).await {
                    Ok(true)
                } else {
                    self.failure_count.fetch_add(1, Ordering::Relaxed);
                    Ok(false)
                }
            } else {
                self.failure_count.fetch_add(1, Ordering::Relaxed);
                Ok(false)
            }
        } else {
            // No Redis configured, consider healthy (test mode)
            Ok(true)
        }
    }

    /// Get operational metrics
    ///
    /// Returns (`health_checks`, commands) - actual tracked values
    /// - `health_checks`: Total number of health checks performed
    /// - commands: Total number of kill switch commands executed (engage/reset)
    #[must_use]
    pub fn get_metrics(&self) -> (u64, u64) {
        let checks = self.health_check_count.load(Ordering::Relaxed);
        let commands = self.command_count.load(Ordering::Relaxed);
        (checks, commands)
    }

    /// Get health metrics
    ///
    /// Returns (`error_rate`, failures) - actual tracked values
    /// - `error_rate`: Ratio of failures to total operations
    /// - failures: Total number of failed operations
    #[must_use]
    pub fn get_health_metrics(&self) -> (f64, u64) {
        let failures = self.failure_count.load(Ordering::Relaxed);
        let total_ops = self.command_count.load(Ordering::Relaxed);

        let error_rate = if total_ops > 0 {
            failures as f64 / total_ops as f64
        } else {
            0.0
        };

        (error_rate, failures)
    }

    /// Activate a scoped kill switch (alias for engage)
    pub async fn activate(
        &self,
        scope: KillSwitchScope,
        reason: String,
        user_id: String,
        cascade: bool,
    ) -> RiskResult<()> {
        self.engage(scope, reason, user_id, cascade).await
    }

    /// Start monitoring — logs health status and Redis connectivity state
    pub async fn start_monitoring(&self) -> RiskResult<()> {
        let has_redis = self.redis_client.is_some();
        let triggered = self.triggered.load(Ordering::Relaxed);
        let health_checks = self.health_check_count.load(Ordering::Relaxed);
        let failures = self.failure_count.load(Ordering::Relaxed);

        tracing::info!(
            redis_connected = has_redis,
            kill_switch_triggered = triggered,
            health_checks_completed = health_checks,
            failure_count = failures,
            "Kill switch monitoring started"
        );

        // TODO: spawn a tokio background task that periodically checks Redis
        // connectivity and logs health status (interval from config.health_check_interval).

        Ok(())
    }

    /// Stop monitoring — logs final health statistics before shutdown
    pub async fn stop_monitoring(&self) -> RiskResult<()> {
        let (error_rate, failures) = self.get_health_metrics();
        let health_checks = self.health_check_count.load(Ordering::Relaxed);
        let triggered = self.triggered.load(Ordering::Relaxed);

        tracing::info!(
            kill_switch_triggered = triggered,
            health_checks_completed = health_checks,
            total_failures = failures,
            error_rate = error_rate,
            "Kill switch monitoring stopped"
        );

        Ok(())
    }

    /// Create a test-only kill switch without Redis dependency
    /// Available for both internal and external tests
    pub fn new_test(config: KillSwitchConfig) -> Self {
        // No Redis client for tests - operations will be no-ops
        Self {
            triggered: Arc::new(AtomicBool::new(false)),
            config,
            redis_client: None,
            scoped_triggers: Arc::new(RwLock::new(HashMap::new())),
            health_check_count: Arc::new(AtomicU64::new(0)),
            command_count: Arc::new(AtomicU64::new(0)),
            failure_count: Arc::new(AtomicU64::new(0)),
        }
    }
}

/// Trading gate for controlled market access
pub struct TradingGate {
    open: Arc<AtomicBool>,
}

impl TradingGate {
    #[must_use]
    pub fn new(initially_open: bool) -> Self {
        Self {
            open: Arc::new(AtomicBool::new(initially_open)),
        }
    }

    pub fn open(&self) {
        self.open.store(true, Ordering::SeqCst);
    }

    pub fn close(&self) {
        self.open.store(false, Ordering::SeqCst);
    }

    #[must_use]
    pub fn is_open(&self) -> bool {
        self.open.load(Ordering::SeqCst)
    }
}

// Re-export the full UnixSocketKillSwitch from its dedicated module
// (the thin stub that was here has been removed to eliminate duplication)
pub use crate::safety::unix_socket_kill_switch::UnixSocketKillSwitch;

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

    fn create_test_kill_switch() -> AtomicKillSwitch {
        let config = KillSwitchConfig::default();
        AtomicKillSwitch::new_test(config)
    }

    #[tokio::test]
    async fn test_kill_switch_creation() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();
        assert!(!kill_switch.is_triggered());
        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_trigger() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Initially not triggered
        assert!(!kill_switch.is_triggered());

        // Trigger it
        kill_switch.trigger();
        assert!(kill_switch.is_triggered());

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_prevents_trading() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Initially trading allowed
        assert!(kill_switch.is_trading_allowed(&KillSwitchScope::Global));

        // Trigger kill switch
        kill_switch.trigger();

        // Now trading should be blocked
        assert!(!kill_switch.is_trading_allowed(&KillSwitchScope::Global));

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_global_activation() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        kill_switch
            .activate_global("Test emergency".to_string(), "test_user".to_string())
            .await?;

        assert!(kill_switch.is_active().await?);
        assert!(!kill_switch.is_trading_allowed(&KillSwitchScope::Global));

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_scoped_activation() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Activate for specific symbol
        kill_switch
            .engage(
                KillSwitchScope::Symbol("AAPL".to_string()),
                "Symbol-specific halt".to_string(),
                "test_user".to_string(),
                false,
            )
            .await?;

        // Global should still be allowed
        assert!(kill_switch.is_trading_allowed(&KillSwitchScope::Global));

        // But symbol should be blocked
        assert!(!kill_switch.is_trading_allowed(&KillSwitchScope::Symbol("AAPL".to_string())));

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_reset() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Trigger and verify
        kill_switch.trigger();
        assert!(kill_switch.is_triggered());

        // Reset
        kill_switch.reset(Some(KillSwitchScope::Global)).await?;

        // Should be cleared
        assert!(!kill_switch.is_triggered());

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_scoped_reset() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        let scope = KillSwitchScope::Account("test_account".to_string());

        // Activate scoped kill switch
        kill_switch
            .engage(scope.clone(), "Test".to_string(), "user".to_string(), false)
            .await?;

        // Verify blocked
        assert!(!kill_switch.is_trading_allowed(&scope));

        // Reset specific scope
        kill_switch.reset(Some(scope.clone())).await?;

        // Should be allowed now
        assert!(kill_switch.is_trading_allowed(&scope));

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_multiple_scopes() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Activate multiple scopes
        kill_switch
            .engage(
                KillSwitchScope::Symbol("AAPL".to_string()),
                "Test".to_string(),
                "user".to_string(),
                false,
            )
            .await?;

        kill_switch
            .engage(
                KillSwitchScope::Account("account1".to_string()),
                "Test".to_string(),
                "user".to_string(),
                false,
            )
            .await?;

        // Both should be blocked
        assert!(!kill_switch.is_trading_allowed(&KillSwitchScope::Symbol("AAPL".to_string())));
        assert!(!kill_switch.is_trading_allowed(&KillSwitchScope::Account("account1".to_string())));

        // But other scopes should be allowed
        assert!(kill_switch.is_trading_allowed(&KillSwitchScope::Symbol("GOOGL".to_string())));

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_cascade_behavior() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Activate with cascade=true
        kill_switch
            .engage(
                KillSwitchScope::Portfolio("portfolio1".to_string()),
                "Cascade test".to_string(),
                "user".to_string(),
                true,
            )
            .await?;

        // Verify activation
        assert!(kill_switch.is_active().await?);

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_deactivate() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        let scope = KillSwitchScope::Strategy("strategy1".to_string());

        // Activate
        kill_switch
            .activate(scope.clone(), "Test".to_string(), "user".to_string(), false)
            .await?;

        // Deactivate
        kill_switch
            .deactivate(scope.clone(), "user".to_string())
            .await?;

        // Should be allowed
        assert!(kill_switch.is_trading_allowed(&scope));

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_health_check() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Health check should pass
        assert!(kill_switch.is_healthy().await?);

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_metrics() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        let (checks, commands) = kill_switch.get_metrics();

        // Initial metrics (currently simplified)
        assert_eq!(checks, 0);
        assert_eq!(commands, 0);

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_health_metrics() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        let (error_rate, failures) = kill_switch.get_health_metrics();

        // Initial health metrics (currently simplified)
        assert_eq!(error_rate, 0.0);
        assert_eq!(failures, 0);

        Ok(())
    }

    #[tokio::test]
    async fn test_kill_switch_monitoring_lifecycle() -> RiskResult<()> {
        let kill_switch = create_test_kill_switch();

        // Start monitoring
        kill_switch.start_monitoring().await?;

        // Stop monitoring
        kill_switch.stop_monitoring().await?;

        Ok(())
    }

    #[tokio::test]
    async fn test_trading_gate_operations() -> RiskResult<()> {
        let gate = TradingGate::new(true);

        assert!(gate.is_open());

        gate.close();
        assert!(!gate.is_open());

        gate.open();
        assert!(gate.is_open());

        Ok(())
    }

    #[tokio::test]
    async fn test_unix_socket_kill_switch() -> RiskResult<()> {
        // Test trigger/is_triggered via AtomicKillSwitch directly
        // (UnixSocketKillSwitch requires an async listener and Arc<AtomicKillSwitch>)
        let atomic_switch = create_test_kill_switch();

        assert!(!atomic_switch.is_triggered());

        atomic_switch.trigger();
        assert!(atomic_switch.is_triggered());

        Ok(())
    }
}