foxhunt/crates/risk/src/safety/unix_socket_kill_switch.rs

//! Unix Domain Socket Kill Switch Interface
//!
//! Provides external control of the kill switch via Unix domain socket at /`var/run/kill_switch`
//! for regulatory compliance and external monitoring systems integration.
//! Designed for sub-100ms emergency shutdown response times.

use chrono::Utc;
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::UnixListener as TokioUnixListener;
use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::broadcast;
use tokio::time::timeout;
use tracing::{error, info, warn};

use crate::error::{RiskError, RiskResult};
use crate::risk_types::KillSwitchScope;
use crate::safety::kill_switch::AtomicKillSwitch;

/// Unix socket commands for kill switch control
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum KillSwitchCommand {
    /// Authenticate with the kill switch system
    Authenticate {
        token: String,
        user_id: String,
        timestamp: u64,
    },
    /// Activate kill switch for specific scope (requires authentication)
    Activate {
        scope: KillSwitchScope,
        reason: String,
        cascade: bool,
        auth_token: String,
    },
    /// Deactivate kill switch for specific scope (requires authentication)
    Deactivate {
        scope: KillSwitchScope,
        auth_token: String,
    },
    /// Get current status (requires authentication)
    Status { auth_token: String },
    /// Emergency global shutdown (requires authentication)
    EmergencyShutdown { reason: String, auth_token: String },
    /// Health check (read-only, no auth required)
    HealthCheck,
}

/// Response from kill switch operations
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KillSwitchResponse {
    pub success: bool,
    pub message: String,
    pub timestamp: u64,
    pub latency_ns: u64,
}

/// Authentication session for kill switch operations
#[derive(Debug, Clone)]
struct AuthSession {
    user_id: String,
    last_used: SystemTime,
    permissions: Vec<String>,
}

/// Authentication manager for kill switch access control
#[derive(Clone)]
struct AuthManager {
    sessions: Arc<Mutex<HashMap<String, AuthSession>>>,
    master_token: String,
    session_timeout: Duration,
}

impl AuthManager {
    fn new() -> Self {
        // Generate master token from environment or secure random
        let master_token = std::env::var("KILL_SWITCH_MASTER_TOKEN").unwrap_or_else(|_| {
            error!("KILL_SWITCH_MASTER_TOKEN not set, using fallback (INSECURE!)");
            "fallback-token-change-me".to_owned()
        });

        Self {
            sessions: Arc::new(Mutex::new(HashMap::new())),
            master_token,
            session_timeout: Duration::from_secs(300), // 5 minute session timeout
        }
    }

    /// Authenticate user and create session
    fn authenticate(&self, token: &str, user_id: &str) -> Result<String, String> {
        // Check master token
        if token != self.master_token {
            return Err("Invalid authentication token".to_owned());
        }

        // Generate session token
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or_else(|_| {
                error!("Failed to get system time for kill switch authentication");
                // Fallback to a fixed timestamp to avoid panic
                0
            });

        let session_token = format!("sess_{}_{}_{}", user_id, timestamp, rand::random::<u32>());

        // Create session
        let session = AuthSession {
            user_id: user_id.to_owned(),
            last_used: SystemTime::now(),
            permissions: vec![
                "kill_switch:activate".to_owned(),
                "kill_switch:deactivate".to_owned(),
                "kill_switch:emergency".to_owned(),
            ],
        };

        // Store session
        if let Ok(mut sessions) = self.sessions.lock() {
            sessions.insert(session_token.clone(), session);
        }

        Ok(session_token)
    }

    /// Validate session token and check permissions
    fn validate_session(
        &self,
        session_token: &str,
        required_permission: &str,
    ) -> Result<String, String> {
        let mut sessions = self.sessions.lock().map_err(|_| "Session lock error")?;

        let session = sessions
            .get_mut(session_token)
            .ok_or("Invalid or expired session token")?;

        // Check session timeout
        if session
            .last_used
            .elapsed()
            .unwrap_or(Duration::from_secs(999))
            > self.session_timeout
        {
            sessions.remove(session_token);
            return Err("Session expired".to_owned());
        }

        // Check permissions
        if !session
            .permissions
            .contains(&required_permission.to_owned())
        {
            return Err(format!(
                "Insufficient permissions for {required_permission}"
            ));
        }

        // Update last used time
        session.last_used = SystemTime::now();

        Ok(session.user_id.clone())
    }

    /// Clean up expired sessions
    fn cleanup_expired_sessions(&self) {
        if let Ok(mut sessions) = self.sessions.lock() {
            sessions.retain(|_, session| {
                session
                    .last_used
                    .elapsed()
                    .unwrap_or(Duration::from_secs(0))
                    <= self.session_timeout
            });
        }
    }
}

/// Unix Domain Socket Kill Switch Controller
/// Provides regulatory-compliant external control interface
pub struct UnixSocketKillSwitch {
    socket_path: String,
    kill_switch: Arc<AtomicKillSwitch>,
    emergency_shutdown: Arc<AtomicBool>,
    listener_handle: Option<tokio::task::JoinHandle<()>>,
    shutdown_sender: Option<broadcast::Sender<()>>,
    auth_manager: AuthManager,
}

impl UnixSocketKillSwitch {
    /// Create new Unix socket kill switch interface
    pub async fn new(socket_path: String, kill_switch: Arc<AtomicKillSwitch>) -> RiskResult<Self> {
        // Ensure socket directory exists and has proper permissions
        if let Some(parent) = Path::new(&socket_path).parent() {
            if !parent.exists() {
                tokio::fs::create_dir_all(parent).await.map_err(|e| {
                    RiskError::Internal(format!("Failed to create socket directory: {e}"))
                })?;

                // Set proper permissions for /var/run/kill_switch directory
                #[cfg(unix)]
                {
                    use std::os::unix::fs::PermissionsExt;
                    let perms = std::fs::Permissions::from_mode(0o755);
                    std::fs::set_permissions(parent, perms).map_err(|e| {
                        RiskError::Internal(format!("Failed to set directory permissions: {e}"))
                    })?;
                }
            }
        }

        // Remove existing socket if it exists
        if Path::new(&socket_path).exists() {
            std::fs::remove_file(&socket_path).map_err(|e| {
                RiskError::Internal(format!("Failed to remove existing socket: {e}"))
            })?;
        }

        Ok(Self {
            socket_path,
            kill_switch,
            emergency_shutdown: Arc::new(AtomicBool::new(false)),
            listener_handle: None,
            shutdown_sender: None,
            auth_manager: AuthManager::new(),
        })
    }

    /// Start the Unix socket listener for external control
    pub async fn start_listener(&mut self) -> RiskResult<()> {
        let listener = TokioUnixListener::bind(&self.socket_path)
            .map_err(|e| RiskError::Internal(format!("Failed to bind Unix socket: {e}")))?;

        // Set proper permissions (readable/writable by owner and group)
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let perms = std::fs::Permissions::from_mode(0o660);
            std::fs::set_permissions(&self.socket_path, perms).map_err(|e| {
                RiskError::Internal(format!("Failed to set socket permissions: {e}"))
            })?;
        }

        let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1);
        self.shutdown_sender = Some(shutdown_tx);

        let kill_switch = Arc::clone(&self.kill_switch);
        let emergency_shutdown = Arc::clone(&self.emergency_shutdown);
        let socket_path = self.socket_path.clone();
        let auth_manager = self.auth_manager.clone();

        let handle = tokio::spawn(async move {
            info!(
                "Unix socket kill switch listener started on {}",
                socket_path
            );

            loop {
                tokio::select! {
                    // Handle new connections
                    result = listener.accept() => {
                        match result {
                            Ok((stream, _addr)) => {
                                let kill_switch_clone = Arc::clone(&kill_switch);
                                let emergency_shutdown_clone = Arc::clone(&emergency_shutdown);

                                let auth_manager_clone = auth_manager.clone();
                                tokio::spawn(async move {
                                    if let Err(e) = Self::handle_connection(
                                        stream,
                                        kill_switch_clone,
                                        emergency_shutdown_clone,
                                        auth_manager_clone
                                    ).await {
                                        error!("Error handling Unix socket connection: {}", e);
                                    }
                                });
                            }
                            Err(e) => {
                                error!("Failed to accept Unix socket connection: {}", e);
                            }
                        }
                    }
                    // Handle shutdown signal
                    _ = shutdown_rx.recv() => {
                        info!("Shutting down Unix socket listener");
                        break;
                    }
                }
            }

            // Cleanup socket file on shutdown
            if let Err(e) = std::fs::remove_file(&socket_path) {
                warn!("Failed to remove socket file {}: {}", socket_path, e);
            }
        });

        self.listener_handle = Some(handle);
        info!(
            "Unix socket kill switch controller started on {}",
            self.socket_path
        );

        Ok(())
    }

    /// Stop the Unix socket listener
    pub async fn stop_listener(&mut self) -> RiskResult<()> {
        if let Some(sender) = &self.shutdown_sender {
            let _ = sender.send(());
        }

        if let Some(handle) = self.listener_handle.take() {
            if let Err(e) = handle.await {
                warn!("Error waiting for listener shutdown: {}", e);
            }
        }

        info!("Unix socket kill switch controller stopped");
        Ok(())
    }

    /// Setup signal-based emergency shutdown handlers that bypass Tokio
    pub async fn setup_emergency_shutdown_signals(&self) -> RiskResult<()> {
        let emergency_shutdown = Arc::clone(&self.emergency_shutdown);
        let kill_switch = Arc::clone(&self.kill_switch);

        // Setup SIGUSR1 for emergency shutdown (bypasses Tokio)
        let mut sigusr1 = signal(SignalKind::user_defined1())
            .map_err(|e| RiskError::Internal(format!("Failed to setup SIGUSR1 handler: {e}")))?;

        let emergency_shutdown_usr1 = Arc::clone(&emergency_shutdown);
        let kill_switch_usr1 = Arc::clone(&kill_switch);

        tokio::spawn(async move {
            loop {
                sigusr1.recv().await;
                warn!("\u{1f6a8} SIGUSR1 received - EMERGENCY SHUTDOWN ACTIVATED");

                // Set emergency flag immediately
                emergency_shutdown_usr1.store(true, Ordering::SeqCst);

                // Engage global kill switch
                if let Err(e) = kill_switch_usr1
                    .engage(
                        KillSwitchScope::Global,
                        "SIGUSR1 emergency signal received".to_owned(),
                        "signal-handler".to_owned(),
                        true,
                    )
                    .await
                {
                    error!("Failed to engage kill switch via SIGUSR1: {}", e);
                }

                // Perform immediate shutdown bypassing Tokio
                Self::perform_emergency_shutdown("SIGUSR1 signal").await;
            }
        });

        // Setup SIGUSR2 for emergency shutdown with different priority
        let mut sigusr2 = signal(SignalKind::user_defined2())
            .map_err(|e| RiskError::Internal(format!("Failed to setup SIGUSR2 handler: {e}")))?;

        let emergency_shutdown_usr2 = Arc::clone(&emergency_shutdown);
        let kill_switch_usr2 = Arc::clone(&kill_switch);

        tokio::spawn(async move {
            loop {
                sigusr2.recv().await;
                warn!("\u{1f6a8} SIGUSR2 received - PRIORITY EMERGENCY SHUTDOWN");

                emergency_shutdown_usr2.store(true, Ordering::SeqCst);

                if let Err(e) = kill_switch_usr2
                    .engage(
                        KillSwitchScope::Global,
                        "SIGUSR2 priority emergency signal received".to_owned(),
                        "signal-handler".to_owned(),
                        true,
                    )
                    .await
                {
                    error!("Failed to engage kill switch via SIGUSR2: {}", e);
                }

                Self::perform_emergency_shutdown("SIGUSR2 priority signal").await;
            }
        });

        info!("Emergency shutdown signal handlers configured (SIGUSR1, SIGUSR2)");
        Ok(())
    }

    /// Check if emergency shutdown is active
    #[must_use]
    pub fn is_emergency_shutdown_active(&self) -> bool {
        self.emergency_shutdown.load(Ordering::SeqCst)
    }

    /// Handle incoming Unix socket connection
    async fn handle_connection(
        stream: tokio::net::UnixStream,
        kill_switch: Arc<AtomicKillSwitch>,
        emergency_shutdown: Arc<AtomicBool>,
        auth_manager: AuthManager,
    ) -> RiskResult<()> {
        let (stream_reader, mut stream_writer) = stream.into_split();
        let mut reader = BufReader::new(stream_reader);
        let mut line = String::new();

        // Set connection timeout for regulatory compliance
        let start_time = Instant::now();

        match timeout(Duration::from_millis(50), reader.read_line(&mut line)).await {
            Ok(Ok(_)) => {
                let latency_ns = start_time.elapsed().as_nanos() as u64;

                // Parse command
                let command: KillSwitchCommand = match serde_json::from_str(line.trim()) {
                    Ok(cmd) => cmd,
                    Err(e) => {
                        let response = KillSwitchResponse {
                            success: false,
                            message: format!("Invalid command format: {e}"),
                            timestamp: Utc::now().timestamp() as u64,
                            latency_ns,
                        };
                        Self::write_response(&mut stream_writer, response).await?;
                        return Ok(());
                    },
                };

                // Process command
                let response = Self::process_command(
                    command,
                    &kill_switch,
                    &emergency_shutdown,
                    &auth_manager,
                    latency_ns,
                )
                .await;

                Self::write_response(&mut stream_writer, response).await?;
            },
            Ok(Err(e)) => {
                error!("Error reading from Unix socket: {}", e);
            },
            Err(_) => {
                warn!("Unix socket read timeout exceeded (50ms)");
                let response = KillSwitchResponse {
                    success: false,
                    message: "Request timeout - must complete within 50ms".to_owned(),
                    timestamp: Utc::now().timestamp() as u64,
                    latency_ns: start_time.elapsed().as_nanos() as u64,
                };
                Self::write_response(&mut stream_writer, response).await?;
            },
        }

        Ok(())
    }

    /// Process kill switch command
    async fn process_command(
        command: KillSwitchCommand,
        kill_switch: &Arc<AtomicKillSwitch>,
        emergency_shutdown: &Arc<AtomicBool>,
        auth_manager: &AuthManager,
        base_latency_ns: u64,
    ) -> KillSwitchResponse {
        let start_time = Instant::now();

        let (success, message) = match command {
            // Authentication command - creates a session
            KillSwitchCommand::Authenticate {
                token,
                user_id,
                timestamp: _,
            } => match auth_manager.authenticate(&token, &user_id) {
                Ok(session_token) => {
                    info!("User {} authenticated successfully", user_id);
                    (
                        true,
                        format!("Authentication successful. Session token: {session_token}"),
                    )
                },
                Err(e) => {
                    warn!("Authentication failed for user {}: {}", user_id, e);
                    (false, format!("Authentication failed: {e}"))
                },
            },

            // Authenticated operations - require valid session
            KillSwitchCommand::Activate {
                scope,
                reason,
                cascade,
                auth_token,
            } => match auth_manager.validate_session(&auth_token, "kill_switch:activate") {
                Ok(user_id) => {
                    info!(
                        "User {} attempting to activate kill switch for {:?}",
                        user_id, scope
                    );
                    match kill_switch
                        .engage(scope.clone(), reason.clone(), user_id.clone(), cascade)
                        .await
                    {
                        Ok(()) => {
                            warn!("\u{1f6a8} KILL SWITCH ACTIVATED for {scope:?}: {reason} by user {user_id}");
                            (
                                true,
                                format!("Kill switch activated for {scope:?}: {reason}"),
                            )
                        },
                        Err(e) => (false, format!("Failed to activate kill switch: {e}")),
                    }
                },
                Err(e) => {
                    warn!("Unauthorized kill switch activation attempt: {}", e);
                    (false, format!("Authentication required: {e}"))
                },
            },

            KillSwitchCommand::Deactivate { scope, auth_token } => {
                match auth_manager.validate_session(&auth_token, "kill_switch:deactivate") {
                    Ok(user_id) => {
                        info!(
                            "User {} attempting to deactivate kill switch for {:?}",
                            user_id, scope
                        );
                        match kill_switch.deactivate(scope.clone(), user_id.clone()).await {
                            Ok(()) => {
                                info!("\u{2705} Kill switch deactivated for {scope:?} by user {user_id}");
                                (true, format!("Kill switch deactivated for {scope:?}"))
                            },
                            Err(e) => (false, format!("Failed to deactivate kill switch: {e}")),
                        }
                    },
                    Err(e) => {
                        warn!("Unauthorized kill switch deactivation attempt: {}", e);
                        (false, format!("Authentication required: {e}"))
                    },
                }
            },

            KillSwitchCommand::Status { auth_token } => {
                match auth_manager.validate_session(&auth_token, "kill_switch:activate") {
                    Ok(user_id) => {
                        info!("User {} requesting kill switch status", user_id);
                        match kill_switch.is_active().await {
                            Ok(active) => {
                                let (checks, commands) = kill_switch.get_metrics();
                                let (error_rate, failures) = kill_switch.get_health_metrics();
                                (true, format!(
                                    "Kill switch status: {} | Checks: {} | Commands: {} | Error rate: {:.2}% | Consecutive failures: {} | User: {}",
                                    if active { "ACTIVE" } else { "INACTIVE" },
                                    checks,
                                    commands,
                                    error_rate * 100.0,
                                    failures,
                                    user_id
                                ))
                            },
                            Err(e) => (false, format!("Failed to get status: {e}")),
                        }
                    },
                    Err(e) => {
                        warn!("Unauthorized status check attempt: {}", e);
                        (false, format!("Authentication required: {e}"))
                    },
                }
            },

            KillSwitchCommand::EmergencyShutdown { reason, auth_token } => {
                match auth_manager.validate_session(&auth_token, "kill_switch:emergency") {
                    Ok(user_id) => {
                        error!(
                            "\u{1f6a8}\u{1f6a8}\u{1f6a8} EMERGENCY SHUTDOWN initiated by user {}: {}",
                            user_id, reason
                        );

                        emergency_shutdown.store(true, Ordering::SeqCst);

                        // Activate global kill switch immediately
                        if let Err(e) = kill_switch
                            .engage(
                                KillSwitchScope::Global,
                                reason.clone(),
                                user_id.clone(),
                                true,
                            )
                            .await
                        {
                            error!("Failed to engage kill switch during emergency: {}", e);
                        }

                        // Trigger emergency shutdown in background
                        let reason_for_shutdown = format!("{reason} (initiated by {user_id})");
                        tokio::spawn(async move {
                            Self::perform_emergency_shutdown(&reason_for_shutdown).await;
                        });

                        (
                            true,
                            format!("Emergency shutdown initiated: {reason} by user {user_id}"),
                        )
                    },
                    Err(e) => {
                        error!("\u{1f6a8} UNAUTHORIZED EMERGENCY SHUTDOWN ATTEMPT: {}", e);
                        (
                            false,
                            format!("Authentication required for emergency shutdown: {e}"),
                        )
                    },
                }
            },

            // Health check is read-only and doesn't require authentication
            KillSwitchCommand::HealthCheck => match kill_switch.is_healthy().await {
                Ok(healthy) => (
                    healthy,
                    if healthy {
                        "System healthy"
                    } else {
                        "System unhealthy - circuit breaker triggered"
                    }
                    .to_owned(),
                ),
                Err(e) => (false, format!("Health check failed: {e}")),
            },
        };

        // Clean up expired sessions periodically
        auth_manager.cleanup_expired_sessions();

        let total_latency_ns = base_latency_ns + start_time.elapsed().as_nanos() as u64;

        KillSwitchResponse {
            success,
            message,
            timestamp: Utc::now().timestamp() as u64,
            latency_ns: total_latency_ns,
        }
    }

    /// Write response back through Unix socket writer
    async fn write_response(
        writer: &mut tokio::net::unix::OwnedWriteHalf,
        response: KillSwitchResponse,
    ) -> RiskResult<()> {
        let response_json = serde_json::to_string(&response)
            .map_err(|e| RiskError::Internal(format!("Failed to serialize response: {e}")))?;

        writer
            .write_all(response_json.as_bytes())
            .await
            .map_err(|e| RiskError::Internal(format!("Failed to write response: {e}")))?;

        writer
            .write_all(b"\n")
            .await
            .map_err(|e| RiskError::Internal(format!("Failed to write newline: {e}")))?;

        Ok(())
    }

    /// Perform emergency shutdown bypassing Tokio runtime
    /// This is the critical regulatory compliance function - must complete in <100ms
    async fn perform_emergency_shutdown(reason: &str) {
        error!("\u{1f6a8}\u{1f6a8}\u{1f6a8} EMERGENCY SHUTDOWN INITIATED: {} \u{1f6a8}\u{1f6a8}\u{1f6a8}", reason);

        // Log emergency event
        error!("Emergency shutdown timestamp: {}", Utc::now().to_rfc3339());

        // In a real implementation, this would:
        // 1. Immediately cancel all outstanding orders
        // 2. Close all positions at market
        // 3. Disconnect from all brokers
        // 4. Stop all trading algorithms
        // 5. Notify regulatory authorities
        // 6. Generate emergency audit log

        // For this implementation, we'll simulate immediate action
        tokio::time::sleep(Duration::from_millis(10)).await; // Simulated shutdown time

        error!("\u{1f6a8} EMERGENCY SHUTDOWN COMPLETE - System halted");

        // In production, this might call std::process::exit(1) to ensure immediate termination
        // std::process::exit(1);
    }
}

/// Utility functions for Unix socket kill switch control
impl UnixSocketKillSwitch {
    /// Send a command to the kill switch via Unix socket (client utility)
    pub async fn send_command_to_socket(
        socket_path: &str,
        command: KillSwitchCommand,
    ) -> RiskResult<KillSwitchResponse> {
        let stream = tokio::net::UnixStream::connect(socket_path)
            .await
            .map_err(|e| {
                RiskError::Internal(format!("Failed to connect to kill switch socket: {e}"))
            })?;

        let command_json = serde_json::to_string(&command)
            .map_err(|e| RiskError::Internal(format!("Failed to serialize command: {e}")))?;

        // Split stream for reading and writing
        let (stream_reader, mut stream_writer) = stream.into_split();

        // Send command
        stream_writer
            .write_all(command_json.as_bytes())
            .await
            .map_err(|e| RiskError::Internal(format!("Failed to send command: {e}")))?;
        stream_writer
            .write_all(b"\n")
            .await
            .map_err(|e| RiskError::Internal(format!("Failed to send newline: {e}")))?;

        // Read response
        let mut reader = BufReader::new(stream_reader);
        let mut response_line = String::new();

        match timeout(
            Duration::from_millis(100),
            reader.read_line(&mut response_line),
        )
        .await
        {
            Ok(Ok(_)) => serde_json::from_str(response_line.trim())
                .map_err(|e| RiskError::Internal(format!("Failed to parse response: {e}"))),
            Ok(Err(e)) => Err(RiskError::Internal(format!("Failed to read response: {e}"))),
            Err(_) => Err(RiskError::Internal("Response timeout".to_owned())),
        }
    }

    /// Emergency activation via Unix socket (for external monitoring systems)
    /// Requires authentication token
    pub async fn emergency_activate(
        socket_path: &str,
        reason: String,
        auth_token: String,
    ) -> RiskResult<KillSwitchResponse> {
        Self::send_command_to_socket(
            socket_path,
            KillSwitchCommand::EmergencyShutdown { reason, auth_token },
        )
        .await
    }

    /// Quick status check via Unix socket
    /// Requires authentication token
    pub async fn quick_status_check(
        socket_path: &str,
        auth_token: String,
    ) -> RiskResult<KillSwitchResponse> {
        Self::send_command_to_socket(socket_path, KillSwitchCommand::Status { auth_token }).await
    }

    /// Authenticate and get session token for subsequent operations
    pub async fn authenticate(
        socket_path: &str,
        master_token: String,
        user_id: String,
    ) -> RiskResult<String> {
        let response = Self::send_command_to_socket(
            socket_path,
            KillSwitchCommand::Authenticate {
                token: master_token,
                user_id,
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .map(|d| d.as_secs())
                    .unwrap_or_else(|_| {
                        error!("Failed to get system time for kill switch authentication");
                        0
                    }),
            },
        )
        .await?;

        if response.success {
            // Extract session token from response message
            if let Some(token) = response.message.split("Session token: ").nth(1) {
                Ok(token.to_owned())
            } else {
                Err(RiskError::Internal(
                    "Failed to extract session token from response".to_owned(),
                ))
            }
        } else {
            Err(RiskError::Internal(format!(
                "Authentication failed: {}",
                response.message
            )))
        }
    }
}

#[cfg(test)]
#[allow(
    clippy::assertions_on_result_states,
    clippy::expect_used,
    clippy::str_to_string
)]
mod tests {
    use super::*;
    use crate::safety::kill_switch::AtomicKillSwitch;
    use crate::safety::KillSwitchConfig;
    use tempfile::tempdir;

    async fn create_test_setup() -> RiskResult<(UnixSocketKillSwitch, String, tempfile::TempDir)> {
        let temp_dir = tempdir().map_err(|e| RiskError::Internal(e.to_string()))?;
        let socket_path = temp_dir.path().join("test_kill_switch.sock");
        let socket_path_str = socket_path.to_string_lossy().to_string();

        let config = KillSwitchConfig::default();
        // Use test constructor to avoid Redis dependency
        let kill_switch = Arc::new(AtomicKillSwitch::new_test(config));

        let unix_socket_kill_switch =
            UnixSocketKillSwitch::new(socket_path_str.clone(), kill_switch).await?;

        Ok((unix_socket_kill_switch, socket_path_str, temp_dir))
    }

    #[tokio::test]
    async fn test_unix_socket_creation() -> RiskResult<()> {
        let (unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        // Verify socket path is set correctly
        assert_eq!(unix_socket_kill_switch.socket_path, socket_path);
        assert!(!unix_socket_kill_switch.is_emergency_shutdown_active());

        Ok(())
    }

    #[tokio::test]
    async fn test_socket_listener_lifecycle() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, _, _temp_dir) = create_test_setup().await?;

        // Start listener
        unix_socket_kill_switch.start_listener().await?;
        assert!(unix_socket_kill_switch.listener_handle.is_some());

        // Stop listener
        unix_socket_kill_switch.stop_listener().await?;
        assert!(unix_socket_kill_switch.listener_handle.is_none());

        Ok(())
    }

    #[tokio::test]
    async fn test_command_processing() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        // Start listener
        unix_socket_kill_switch.start_listener().await?;

        // Give listener time to start
        tokio::time::sleep(Duration::from_millis(50)).await;

        // First authenticate to get session token
        let auth_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Authenticate {
                token: "fallback-token-change-me".to_string(),
                user_id: "test_user".to_string(),
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .expect("System time should be after UNIX epoch in test")
                    .as_secs(),
            },
        )
        .await?;

        assert!(auth_response.success);
        assert!(auth_response.message.contains("Authentication successful"));

        // Extract session token from response
        let session_token = auth_response
            .message
            .split("Session token: ")
            .nth(1)
            .expect("Auth response should contain session token")
            .to_string();

        // Test status command with authentication
        let response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Status {
                auth_token: session_token,
            },
        )
        .await?;

        assert!(response.success);
        assert!(response.message.contains("Kill switch status"));
        assert!(response.latency_ns > 0);

        // Stop listener
        unix_socket_kill_switch.stop_listener().await?;

        Ok(())
    }

    #[tokio::test]
    async fn test_emergency_shutdown_command() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        unix_socket_kill_switch.start_listener().await?;
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Authenticate first
        let auth_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Authenticate {
                token: "fallback-token-change-me".to_string(),
                user_id: "emergency_user".to_string(),
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .expect("System time should be after UNIX epoch in test")
                    .as_secs(),
            },
        )
        .await?;

        assert!(auth_response.success);
        let session_token = auth_response
            .message
            .split("Session token: ")
            .nth(1)
            .expect("Auth response should contain session token")
            .to_string();

        // Test emergency shutdown with authentication
        let response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::EmergencyShutdown {
                reason: "Test emergency".to_string(),
                auth_token: session_token,
            },
        )
        .await?;

        assert!(response.success);
        assert!(response.message.contains("Emergency shutdown initiated"));
        assert!(unix_socket_kill_switch.is_emergency_shutdown_active());

        unix_socket_kill_switch.stop_listener().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_activate_deactivate_commands() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        unix_socket_kill_switch.start_listener().await?;
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Authenticate first
        let auth_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Authenticate {
                token: "fallback-token-change-me".to_string(),
                user_id: "test_operator".to_string(),
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .expect("System time should be after UNIX epoch in test")
                    .as_secs(),
            },
        )
        .await?;

        assert!(auth_response.success);
        let session_token = auth_response
            .message
            .split("Session token: ")
            .nth(1)
            .expect("Auth response should contain session token")
            .to_string();

        // Activate kill switch
        let activate_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Activate {
                scope: KillSwitchScope::Symbol("AAPL".to_string()),
                reason: "Test activation".to_string(),
                cascade: false,
                auth_token: session_token.clone(),
            },
        )
        .await?;

        assert!(activate_response.success);
        assert!(activate_response.message.contains("Kill switch activated"));

        // Deactivate kill switch
        let deactivate_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Deactivate {
                scope: KillSwitchScope::Symbol("AAPL".to_string()),
                auth_token: session_token,
            },
        )
        .await?;

        assert!(deactivate_response.success);
        assert!(deactivate_response
            .message
            .contains("Kill switch deactivated"));

        unix_socket_kill_switch.stop_listener().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_health_check_command() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        unix_socket_kill_switch.start_listener().await?;
        tokio::time::sleep(Duration::from_millis(50)).await;

        let response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::HealthCheck,
        )
        .await?;

        assert!(response.success);
        assert!(response.message.contains("healthy"));

        unix_socket_kill_switch.stop_listener().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_signal_handler_setup() -> RiskResult<()> {
        let (unix_socket_kill_switch, _, _temp_dir) = create_test_setup().await?;

        // Setup signal handlers (this should not fail)
        let result = unix_socket_kill_switch
            .setup_emergency_shutdown_signals()
            .await;
        assert!(result.is_ok());

        Ok(())
    }

    #[tokio::test]
    async fn test_utility_functions() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        unix_socket_kill_switch.start_listener().await?;
        tokio::time::sleep(Duration::from_millis(50)).await;

        // First authenticate to get session token
        let auth_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Authenticate {
                token: "fallback-token-change-me".to_string(),
                user_id: "utility_user".to_string(),
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .expect("System time should be after UNIX epoch in test")
                    .as_secs(),
            },
        )
        .await?;

        assert!(auth_response.success);
        let session_token = auth_response
            .message
            .split("Session token: ")
            .nth(1)
            .expect("Auth response should contain session token")
            .to_string();

        // Test utility function for status check with authentication
        let status_response = UnixSocketKillSwitch::send_command_to_socket(
            &socket_path,
            KillSwitchCommand::Status {
                auth_token: session_token,
            },
        )
        .await?;
        assert!(status_response.success);

        unix_socket_kill_switch.stop_listener().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_connection_timeout() -> RiskResult<()> {
        let (mut unix_socket_kill_switch, socket_path, _temp_dir) = create_test_setup().await?;

        unix_socket_kill_switch.start_listener().await?;
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Connect but don't send data (should timeout)
        let _stream = tokio::net::UnixStream::connect(&socket_path).await?;

        // Wait for timeout to occur
        tokio::time::sleep(Duration::from_millis(100)).await;

        unix_socket_kill_switch.stop_listener().await?;
        Ok(())
    }
}