use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

/// Emergency kill switch with multiple safety triggers
pub struct KillSwitch {
    /// Manual kill switch (set via API or panic button)
    pub manual_enabled: AtomicBool,

    /// Last block we know is safe (consensus finalized)
    pub last_safe_slot: AtomicU64,

    /// Last successful trade timestamp
    pub last_success: Arc<RwLock<Option<Instant>>>,

    /// Circuit breaker states
    pub circuit_breakers: Arc<RwLock<HashMap<String, bool>>>,

    /// Panic on critical error (kill process)
    pub panic_on_critical: bool,
}

impl KillSwitch {
    /// Check if trading should continue
    pub async fn check_safety(&self, current_slot: u64) -> Result<()> {
        // 1. Manual kill switch
        if self.manual_enabled.load(Ordering::SeqCst) {
            return Err("❌ Manual kill switch engaged".into());
        }

        // 2. Slot drift check (are we too far behind?)
        let last_safe = self.last_safe_slot.load(Ordering::SeqCst);
        if current_slot > last_safe + 100 {
            warn!("⚠️ Slot drift detected: {} slots behind",
                current_slot - last_safe);

            if current_slot > last_safe + 300 {
                self.enable_manual("Excessive slot drift");
                return Err("❌ Kill switch: Too far behind consensus".into());
            }
        }

        // 3. No successful trades in last 4 hours (system stalled?)
        if let Some(last) = *self.last_success.read().await {
            let elapsed = last.elapsed();
            if elapsed > Duration::from_secs(4 * 3600) {
                warn!("⚠️ No successful trades in {} hours",
                    elapsed.as_secs() / 3600);

                if elapsed > Duration::from_secs(6 * 3600) {
                    self.enable_manual("No trades in 6 hours");
                    return Err("❌ Kill switch: System appears stalled".into());
                }
            }
        }

        // 4. Circuit breaker check (any critical components down?)
        let breakers = self.circuit_breakers.read().await;
        let critical_down: Vec<_> = breakers.iter()
            .filter(|(k, v)| k.contains("critical") && **v)
            .map(|(k, _)| k.as_str())
            .collect();

        if !critical_down.is_empty() {
            return Err(format!(
                "❌ Kill switch: Critical components down: {:?}",
                critical_down
            ).into());
        }

        // 5. Network partition detection (check validator consensus)
        if !self.check_network_consensus().await? {
            self.enable_manual("Network partition detected");
            return Err("❌ Kill switch: Network partition detected".into());
        }

        Ok(())
    }

    /// Check if validators have consensus
    async fn check_network_consensus(&self) -> Result<bool> {
        // Query multiple RPC endpoints
        let endpoints = vec![
            "https://api.mainnet-beta.solana.com",
            "https://solana-api.projectserum.com",
            "https://rpc.ankr.com/solana",
        ];

        let mut slots = Vec::new();
        for endpoint in endpoints {
            if let Ok(slot) = get_slot(endpoint).await {
                slots.push(slot);
            }
        }

        // If slots differ by >50, network might be partitioned
        if let (Some(min), Some(max)) = (slots.iter().min(), slots.iter().max()) {
            if max - min > 50 {
                warn!("⚠️ Validator slot disagreement: {} - {}", min, max);
                return Ok(false);
            }
        }

        Ok(true)
    }

    /// Manually enable kill switch with reason
    pub fn enable_manual(&self, reason: &str) {
        error!("🚨 KILL SWITCH ENGAGED: {}", reason);
        self.manual_enabled.store(true, Ordering::SeqCst);

        if self.panic_on_critical {
            panic!("Kill switch engaged: {}", reason);
        }
    }

    /// Update last successful trade time
    pub async fn record_success(&self, slot: u64) {
        *self.last_success.write().await = Some(Instant::now());
        self.last_safe_slot.store(slot, Ordering::SeqCst);
    }
}

// In main trading loop
async fn trading_loop(kill_switch: Arc<KillSwitch>) {
    loop {
        // Check safety before EVERY trade
        let current_slot = get_current_slot().await?;
        if let Err(e) = kill_switch.check_safety(current_slot).await {
            error!("Trading halted: {}", e);

            // Send critical alert (Slack, PagerDuty, etc.)
            send_critical_alert(&e.to_string()).await;

            // Wait 5 minutes before checking again
            tokio::time::sleep(Duration::from_secs(300)).await;
            continue;
        }

        // Proceed with trading...
        match execute_opportunity(&opp).await {
            Ok(signature) => {
                kill_switch.record_success(current_slot).await;
            }
            Err(e) => {
                warn!("Trade failed: {}", e);
                // Circuit breaker logic...
            }
        }
    }
}

// POST /api/killswitch/enable
async fn enable_killswitch(
    State(kill_switch): State<Arc<KillSwitch>>
) -> Json<KillSwitchStatus> {
    kill_switch.enable_manual("Manual API call");
    Json(KillSwitchStatus {
        enabled: true,
        reason: "Manual".to_string(),
    })
}

// POST /api/killswitch/disable
async fn disable_killswitch(
    State(kill_switch): State<Arc<KillSwitch>>
) -> Json<KillSwitchStatus> {
    kill_switch.manual_enabled.store(false, Ordering::SeqCst);
    Json(KillSwitchStatus {
        enabled: false,
        reason: "Manually disabled".to_string(),
    })
}

/// Dual-mode executor: paper trading + real trading
pub enum ExecutionMode {
    PaperTrading,  // Log what WOULD happen, don't submit
    RealTrading,   // Actually submit transactions
    ParallelMode,  // Both (compare results)
}

pub struct DualModeExecutor {
    mode: ExecutionMode,
    paper_results: Arc<RwLock<Vec<PaperTradeResult>>>,
    real_results: Arc<RwLock<Vec<RealTradeResult>>>,
}

impl DualModeExecutor {
    async fn execute_opportunity(&self, opp: &Opportunity) -> Result<()> {
        match self.mode {
            ExecutionMode::PaperTrading => {
                // Simulate execution
                let result = self.simulate_trade(opp).await?;
                self.paper_results.write().await.push(result);
                info!("📄 Paper trade: {} → profit ${:.2}",
                    opp.route, result.profit);
            }

            ExecutionMode::RealTrading => {
                // Real execution
                let result = self.submit_real_trade(opp).await?;
                self.real_results.write().await.push(result);
                info!("💰 Real trade: {} → profit ${:.2}",
                    opp.route, result.profit);
            }

            ExecutionMode::ParallelMode => {
                // Both (for comparison)
                let paper = self.simulate_trade(opp).await?;
                let real = self.submit_real_trade(opp).await?;

                // Compare results
                self.compare_results(&paper, &real).await;
            }
        }

        Ok(())
    }

    async fn compare_results(
        &self,
        paper: &PaperTradeResult,
        real: &RealTradeResult
    ) {
        let diff = real.actual_profit - paper.expected_profit;

        if diff.abs() > 0.1 {
            warn!("📊 Prediction vs Reality: expected ${:.2}, got ${:.2} (diff: ${:.2})",
                paper.expected_profit, real.actual_profit, diff);
        }
    }
}

/// Simulate 1-2 other bots competing on same opportunities
async fn replay_with_competitors(
    events: Vec<MarketEvent>,
    num_competitors: usize,
) -> ReplayResults {
    let mut results = Vec::new();

    for event in events {
        let opp = detect_opportunity(&event)?;

        // Simulate competitor bots
        for i in 0..num_competitors {
            let competitor_latency = rand::range(50..300); // ms
            let our_latency = 150; // Target latency

            if competitor_latency < our_latency {
                // We lost to competitor
                results.push(TradeResult::LostToCompetitor {
                    route: opp.route.clone(),
                    their_latency: competitor_latency,
                    our_latency,
                });
                continue;
            }
        }

        // We won, execute
        results.push(execute_simulated_trade(&opp));
    }

    // Analyze results
    let win_rate = results.iter()
        .filter(|r| matches!(r, TradeResult::Success(_)))
        .count() as f64 / results.len() as f64;

    info!("Win rate with {} competitors: {:.1}%",
        num_competitors, win_rate * 100.0);

    ReplayResults { results, win_rate }
}

/// Test with degraded network conditions
async fn replay_with_latency(
    events: Vec<MarketEvent>,
    added_latency_ms: u64,
) -> ReplayResults {
    for event in events {
        // Add artificial delay
        tokio::time::sleep(Duration::from_millis(added_latency_ms)).await;

        let opp = detect_opportunity(&event)?;

        // Simulate how stale the quote is
        let quote_age = opp.detected_at.elapsed().as_millis();
        if quote_age > 200 {
            warn!("Quote too stale: {}ms", quote_age);
            continue; // Skip
        }

        execute_simulated_trade(&opp);
    }
}

// Test multiple latency scenarios
replay_with_latency(events.clone(), 0).await;    // Ideal
replay_with_latency(events.clone(), 100).await;  // Good
replay_with_latency(events.clone(), 200).await;  // Acceptable
replay_with_latency(events.clone(), 500).await;  // Poor

/// Simulate RPC failures, packet loss, etc.
async fn replay_with_failures(
    events: Vec<MarketEvent>,
    failure_rate: f64, // 0.0 - 1.0
) -> ReplayResults {
    for event in events {
        // Randomly inject failures
        if rand::random::<f64>() < failure_rate {
            let failure_type = rand::choose(&[
                FailureType::RpcTimeout,
                FailureType::PacketLoss,
                FailureType::InvalidQuote,
                FailureType::InsufficientBalance,
            ]);

            warn!("💥 Injected failure: {:?}", failure_type);
            continue; // Skip this opportunity
        }

        execute_simulated_trade(&detect_opportunity(&event)?);
    }
}

// Test with various failure rates
replay_with_failures(events.clone(), 0.05).await; // 5% failures
replay_with_failures(events.clone(), 0.15).await; // 15% failures
replay_with_failures(events.clone(), 0.30).await; // 30% failures

/// Monitor pool depth and auto-scale trade size
pub struct LiquidityMonitor {
    pool_depths: Arc<RwLock<HashMap<Pubkey, PoolDepth>>>,
}

#[derive(Debug, Clone)]
pub struct PoolDepth {
    pub pool_id: Pubkey,
    pub base_reserve: u64,
    pub quote_reserve: u64,
    pub last_updated: Instant,
    pub daily_volume: u64,  // Last 24h
}

impl LiquidityMonitor {
    /// Calculate optimal trade size for this pool
    pub async fn calculate_optimal_size(
        &self,
        pool_id: &Pubkey,
        desired_amount: u64,
    ) -> Result<u64> {
        let depth = self.pool_depths.read().await
            .get(pool_id)
            .ok_or("Pool not found")?
            .clone();

        // Rule: Never trade >2% of pool reserves
        let max_by_depth = (depth.base_reserve as f64 * 0.02) as u64;

        // Rule: Never exceed 5% of daily volume
        let max_by_volume = (depth.daily_volume as f64 * 0.05) as u64;

        // Rule: Absolute ceiling (10 SOL = 10 billion lamports)
        let absolute_max = 10_000_000_000;

        // Take minimum of all constraints
        let optimal = desired_amount
            .min(max_by_depth)
            .min(max_by_volume)
            .min(absolute_max);

        if optimal < desired_amount {
            warn!(
                "Scaled down trade: desired {} → optimal {} (pool depth: {}, volume: {})",
                desired_amount, optimal, depth.base_reserve, depth.daily_volume
            );
        }

        Ok(optimal)
    }

    /// Update pool depth from on-chain data
    pub async fn update_pool_depth(&self, pool_id: Pubkey) -> Result<()> {
        let account_data = fetch_pool_account(&pool_id).await?;
        let reserves = parse_pool_reserves(&account_data)?;

        let depth = PoolDepth {
            pool_id,
            base_reserve: reserves.base,
            quote_reserve: reserves.quote,
            last_updated: Instant::now(),
            daily_volume: self.calculate_daily_volume(&pool_id).await?,
        };

        self.pool_depths.write().await.insert(pool_id, depth);
        Ok(())
    }

    /// Track 24h volume
    async fn calculate_daily_volume(&self, pool_id: &Pubkey) -> Result<u64> {
        // Query historical trades or use external API
        // For now, estimate from pool size
        let depth = self.pool_depths.read().await
            .get(pool_id)
            .cloned()
            .unwrap_or_default();

        // Heuristic: Daily volume ≈ 20-50% of pool TVL
        let estimated_volume = (depth.base_reserve as f64 * 0.30) as u64;
        Ok(estimated_volume)
    }
}

async fn evaluate_opportunity(
    opp: &Opportunity,
    liquidity_monitor: &LiquidityMonitor,
) -> Result<AdjustedOpportunity> {
    // Calculate optimal trade size based on liquidity
    let optimal_amount = liquidity_monitor
        .calculate_optimal_size(&opp.pool_id, opp.amount)
        .await?;

    // Recalculate profit with adjusted amount
    let adjusted_profit = calculate_profit(
        opp.route,
        optimal_amount,
        opp.min_output
    )?;

    // Only execute if still profitable after scaling
    if adjusted_profit < MIN_PROFIT_THRESHOLD {
        return Err("Not profitable after liquidity scaling".into());
    }

    Ok(AdjustedOpportunity {
        amount: optimal_amount,
        expected_profit: adjusted_profit,
        ..opp.clone()
    })
}

/// Track bundle submission quality
#[derive(Debug, Clone)]
pub struct BundleQualityMetrics {
    pub submitted_count: u64,
    pub landed_count: u64,
    pub dropped_count: u64,
    pub average_tip: u64,
    pub tip_to_profit_ratio: f64,  // Should be <0.3 (tip <30% of profit)
    pub competitor_tips: Vec<u64>,  // What others are paying
    pub success_rate_by_tip_percentile: HashMap<String, f64>,
}

impl BundleQualityMetrics {
    /// Calculate if we're being outbid
    pub fn analyze_competitive_position(&self) -> CompetitiveAnalysis {
        let our_avg_tip = self.average_tip;
        let competitor_p50 = percentile(&self.competitor_tips, 0.50);
        let competitor_p95 = percentile(&self.competitor_tips, 0.95);

        CompetitiveAnalysis {
            our_position: if our_avg_tip < competitor_p50 {
                TipPosition::BelowMedian
            } else if our_avg_tip < competitor_p95 {
                TipPosition::AboveMedian
            } else {
                TipPosition::TopTier
            },
            recommendation: if self.success_rate() < 0.60
                && our_avg_tip < competitor_p50 {
                "Increase tip bidding to match competition"
            } else if self.tip_to_profit_ratio > 0.35 {
                "Reduce tips, margin too low"
            } else {
                "Current tip strategy is optimal"
            },
        }
    }

    /// Monitor for reputation risk
    pub fn check_reputation_risk(&self) -> Option<String> {
        let success_rate = self.success_rate();
        let recent_drops = self.dropped_count_last_hour();

        // Red flag: >50% drops in last hour
        if recent_drops > 10 && success_rate < 0.50 {
            return Some(format!(
                "⚠️ High drop rate: {}% success, {} drops/hour",
                (success_rate * 100.0) as u32,
                recent_drops
            ));
        }

        // Red flag: Too many bundles submitted (spam detection)
        if self.submitted_count_last_hour() > 100 {
            return Some("⚠️ Submitting >100 bundles/hour (spam risk)".into());
        }

        None
    }

    fn success_rate(&self) -> f64 {
        self.landed_count as f64 / self.submitted_count as f64
    }
}