// src/bin/chat_load_tester_consistency.rs
use futures_util::{
    SinkExt, StreamExt,
    stream::{FuturesUnordered, SplitSink, SplitStream},
};
use std::sync::{
    Arc,
    atomic::{AtomicU64, Ordering},
};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tokio::time::sleep;
use tokio_tungstenite::{
    MaybeTlsStream, WebSocketStream, connect_async, tungstenite::protocol::Message,
};
mod shared_types;
use shared_types::{ClientToServerMsg, ServerToClientMsg};


static EARLY_EXIT_COUNT: AtomicU64 = AtomicU64::new(0);

// --- Configuration ---
const SERVER_URL: &str = "ws://localhost:8080/ws";

const TOTAL_USERS: u64 = 1000;
const USERS_PER_SECOND: u64 = 50;   // Sustainable rate
const TEST_DURATION: Duration = Duration::from_secs(45);


// const TOTAL_USERS: u64 = 1000;
// const USERS_PER_SECOND: u64 = 100;  // Push it a bit
// const TEST_DURATION: Duration = Duration::from_secs(30);


// const TOTAL_USERS: u64 = 2000;
// const USERS_PER_SECOND: u64 = 200;  // Find limits
// const TEST_DURATION: Duration = Duration::from_secs(20);


const GROUP_NAME: &str = "global-chat";
const NUM_TEST_RUNS: usize = 5; // Number of consecutive tests

// --- Metrics ---
#[derive(Default)]
struct TestMetrics {
    connections_successful: AtomicU64,
    connections_failed: AtomicU64,
    logins_successful: AtomicU64,
    messages_sent: AtomicU64,
    messages_received: AtomicU64,
    round_trip_latency_us: AtomicU64,
    round_trip_count: AtomicU64,
    group_joins_attempted: AtomicU64,
    group_joins_successful: AtomicU64,
    group_joins_failed: AtomicU64,
}

#[derive(Debug, Clone)]
struct TestResult {
    run_number: usize,
    duration_secs: f32,
    connections_successful: u64,
    connections_failed: u64,
    logins_successful: u64,
    group_joins_successful: u64, // Add this
    group_joins_failed: u64,     // Add this
    messages_sent: u64,
    messages_received: u64,
    avg_latency_ms: f64,
    connection_success_rate: f64,
    login_success_rate: f64,
    group_join_success_rate: f64, // Add this
    msg_per_sec: f64,
    message_amplification: f64,
}

impl TestResult {
    fn from_metrics(run_number: usize, metrics: &TestMetrics, duration: Duration) -> Self {
        let connections_successful = metrics.connections_successful.load(Ordering::Relaxed);
        let connections_failed = metrics.connections_failed.load(Ordering::Relaxed);
        let logins_successful = metrics.logins_successful.load(Ordering::Relaxed);
        let messages_sent = metrics.messages_sent.load(Ordering::Relaxed);
        let messages_received = metrics.messages_received.load(Ordering::Relaxed);
        let rt_count = metrics.round_trip_count.load(Ordering::Relaxed);
        let rt_total_us = metrics.round_trip_latency_us.load(Ordering::Relaxed);

        let avg_latency_ms = if rt_count > 0 {
            (rt_total_us as f64 / rt_count as f64) / 1000.0
        } else {
            0.0
        };

        let connection_success_rate = (connections_successful as f64 / TOTAL_USERS as f64) * 100.0;
        let login_success_rate = if connections_successful > 0 {
            (logins_successful as f64 / connections_successful as f64) * 100.0
        } else {
            0.0
        };

        let msg_per_sec = messages_sent as f64 / duration.as_secs_f64();
        let message_amplification = if messages_sent > 0 {
            messages_received as f64 / messages_sent as f64
        } else {
            0.0
        };

        let group_joins_successful = metrics.group_joins_successful.load(Ordering::Relaxed);
        let group_joins_failed = metrics.group_joins_failed.load(Ordering::Relaxed);
        let group_joins_attempted = metrics.group_joins_attempted.load(Ordering::Relaxed);

        let group_join_success_rate = if group_joins_attempted > 0 {
            (group_joins_successful as f64 / group_joins_attempted as f64) * 100.0
        } else {
            0.0
        };

        Self {
            run_number,
            duration_secs: duration.as_secs_f32(),
            connections_successful,
            connections_failed,
            logins_successful,
            messages_sent,
            messages_received,
            avg_latency_ms,
            connection_success_rate,
            login_success_rate,
            msg_per_sec,
            message_amplification,
            group_joins_successful,
            group_joins_failed,
            group_join_success_rate,
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("🚀 Starting Chat Server Consistency Load Test...");
    println!("   Running {} consecutive tests", NUM_TEST_RUNS);
    println!(
        "   Each test: {} users over {} seconds, connecting at {}/sec.",
        TOTAL_USERS,
        TEST_DURATION.as_secs(),
        USERS_PER_SECOND
    );
    println!("{}", "=".repeat(60));
    println!();

    let mut results = Vec::new();

    for run in 1..=NUM_TEST_RUNS {
        println!("🔄 Starting Test Run {} of {}", run, NUM_TEST_RUNS);

        if run > 1 {
            if let Err(e) = reset_server().await {
                println!("⚠️ Failed to reset server: {}", e);
            }
            // Wait a bit for reset to complete
            sleep(Duration::from_secs(2)).await;
        }

        let result = run_single_test(run).await;
        results.push(result.clone());

        print_single_result(&result);

        if run < NUM_TEST_RUNS {
            // Much longer cooldown for 1000 users with high message volume
            let cooldown_secs = if TOTAL_USERS >= 1000 { 30 } else { 10 };
            println!("⏳ Waiting {} seconds before next test...\n", cooldown_secs);
            sleep(Duration::from_secs(cooldown_secs)).await;
        }
    }

    println!();
    println!("{}", "=".repeat(60));
    println!("📊 CONSISTENCY ANALYSIS");
    println!("{}", "=".repeat(60));

    analyze_consistency(&results);

    Ok(())
}

fn print_single_result(result: &TestResult) {
    println!("📈 Test Run {} Results:", result.run_number);
    println!("   Duration: {:.2}s", result.duration_secs);
    println!(
        "   Connections: {} successful, {} failed ({:.1}% success)",
        result.connections_successful, result.connections_failed, result.connection_success_rate
    );
    println!(
        "   Logins: {} successful ({:.1}% of connections)",
        result.logins_successful, result.login_success_rate
    );
    println!(
        "   Messages: {} sent, {} received ({:.1}x amplification)",
        result.messages_sent, result.messages_received, result.message_amplification
    );
    println!("   Throughput: {:.1} msg/sec", result.msg_per_sec);
    println!("   Latency: {:.1}ms average", result.avg_latency_ms);
    println!();
}

fn analyze_consistency(results: &[TestResult]) {
    if results.is_empty() {
        return;
    }

    // Calculate statistics
    let connection_rates: Vec<f64> = results.iter().map(|r| r.connection_success_rate).collect();
    let login_rates: Vec<f64> = results.iter().map(|r| r.login_success_rate).collect();
    let latencies: Vec<f64> = results.iter().map(|r| r.avg_latency_ms).collect();
    let throughputs: Vec<f64> = results.iter().map(|r| r.msg_per_sec).collect();
    let amplifications: Vec<f64> = results.iter().map(|r| r.message_amplification).collect();

    println!("📊 SUMMARY TABLE:");
    println!("┌─────┬──────────┬────────┬─────────┬───────────┬──────────┐");
    println!("│ Run │ Conn %   │ Login %│ Lat(ms) │ Msg/sec   │ Ampl     │");
    println!("├─────┼──────────┼────────┼─────────┼───────────┼──────────┤");

    for result in results {
        println!(
            "│ {:3} │ {:8.1} │ {:6.1} │ {:7.1} │ {:9.1} │ {:8.1} │",
            result.run_number,
            result.connection_success_rate,
            result.login_success_rate,
            result.avg_latency_ms,
            result.msg_per_sec,
            result.message_amplification
        );
    }
    println!("└─────┴──────────┴────────┴─────────┴───────────┴──────────┘");

    println!("\n📈 CONSISTENCY METRICS:");

    // Connection success rate
    let (conn_avg, conn_std, conn_cv) = calculate_stats(&connection_rates);
    println!("Connection Success Rate:");
    println!(
        "  Average: {:.1}%  StdDev: {:.2}%  CV: {:.1}%",
        conn_avg, conn_std, conn_cv
    );
    print_consistency_rating(conn_cv, "connection reliability");

    // Login success rate
    let (login_avg, login_std, login_cv) = calculate_stats(&login_rates);
    println!("Login Success Rate:");
    println!(
        "  Average: {:.1}%  StdDev: {:.2}%  CV: {:.1}%",
        login_avg, login_std, login_cv
    );
    print_consistency_rating(login_cv, "login reliability");

    // Latency
    let (lat_avg, lat_std, lat_cv) = calculate_stats(&latencies);
    println!("Average Latency:");
    println!(
        "  Average: {:.1}ms  StdDev: {:.2}ms  CV: {:.1}%",
        lat_avg, lat_std, lat_cv
    );
    print_consistency_rating(lat_cv, "latency consistency");

    // Throughput
    let (thr_avg, thr_std, thr_cv) = calculate_stats(&throughputs);
    println!("Throughput:");
    println!(
        "  Average: {:.1} msg/s  StdDev: {:.2} msg/s  CV: {:.1}%",
        thr_avg, thr_std, thr_cv
    );
    print_consistency_rating(thr_cv, "throughput consistency");

    // Message amplification
    let (amp_avg, amp_std, amp_cv) = calculate_stats(&amplifications);
    println!("Message Amplification:");
    println!(
        "  Average: {:.1}x  StdDev: {:.2}x  CV: {:.1}%",
        amp_avg, amp_std, amp_cv
    );
    print_consistency_rating(amp_cv, "broadcast consistency");

    println!("\n🎯 OVERALL ASSESSMENT:");
    let overall_cv = (conn_cv + login_cv + lat_cv + thr_cv + amp_cv) / 5.0;
    print_overall_rating(overall_cv);

    // Identify problematic runs
    identify_outliers(results);
}

fn calculate_stats(values: &[f64]) -> (f64, f64, f64) {
    let n = values.len() as f64;
    let mean = values.iter().sum::<f64>() / n;
    let variance = values.iter().map(|x| (x - mean).powi(2)).sum::<f64>() / n;
    let std_dev = variance.sqrt();
    let cv = if mean != 0.0 {
        (std_dev / mean.abs()) * 100.0
    } else {
        0.0
    };
    (mean, std_dev, cv)
}

fn print_consistency_rating(cv: f64, metric_name: &str) {
    let rating = match cv {
        cv if cv < 5.0 => "🟢 EXCELLENT",
        cv if cv < 10.0 => "🟡 GOOD",
        cv if cv < 20.0 => "🟠 FAIR",
        _ => "🔴 POOR",
    };
    println!("  Consistency: {} ({})", rating, metric_name);
}

fn print_overall_rating(overall_cv: f64) {
    let (rating, recommendation) = match overall_cv {
        cv if cv < 5.0 => (
            "🟢 EXCELLENT",
            "System is highly consistent and production-ready",
        ),
        cv if cv < 10.0 => ("🟡 GOOD", "System is stable with minor variations"),
        cv if cv < 20.0 => (
            "🟠 FAIR",
            "System has noticeable inconsistencies, investigate bottlenecks",
        ),
        _ => ("🔴 POOR", "System is unreliable, requires optimization"),
    };

    println!("Overall Consistency: {} (CV: {:.1}%)", rating, overall_cv);
    println!("Recommendation: {}", recommendation);
}

fn identify_outliers(results: &[TestResult]) {
    println!("\n🔍 OUTLIER ANALYSIS:");

    // Find runs with significantly different performance
    let latencies: Vec<f64> = results.iter().map(|r| r.avg_latency_ms).collect();
    let (lat_mean, lat_std, _) = calculate_stats(&latencies);

    let login_rates: Vec<f64> = results.iter().map(|r| r.login_success_rate).collect();
    let (login_mean, login_std, _) = calculate_stats(&login_rates);

    let mut outliers_found = false;

    for result in results {
        let mut issues = Vec::new();

        if (result.avg_latency_ms - lat_mean).abs() > 2.0 * lat_std {
            issues.push(format!(
                "High latency variance ({:.1}ms vs {:.1}ms avg)",
                result.avg_latency_ms, lat_mean
            ));
        }

        if (result.login_success_rate - login_mean).abs() > 2.0 * login_std {
            issues.push(format!(
                "Login rate variance ({:.1}% vs {:.1}% avg)",
                result.login_success_rate, login_mean
            ));
        }

        if result.connection_success_rate < 95.0 {
            issues.push("Low connection success rate".to_string());
        }

        if result.login_success_rate < 90.0 {
            issues.push("Low login success rate".to_string());
        }

        if !issues.is_empty() {
            outliers_found = true;
            println!("⚠️  Run {}: {}", result.run_number, issues.join(", "));
        }
    }

    if !outliers_found {
        println!("✅ No significant outliers detected");
    }
}

// Define aliases for the writer and reader types for clarity
type WsWriter = SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>;
type WsReader = SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>;

async fn simulate_user(user_id: u64, metrics: Arc<TestMetrics>) {
    // --- 1. Connect ---
    // Retry connection up to 3 times
    let mut attempts = 0;
    let ws_stream = loop {
        attempts += 1;
        match connect_async(SERVER_URL).await {
            Ok((stream, _)) => {
                metrics.connections_successful.fetch_add(1, Ordering::Relaxed);
                break stream;
            }
            Err(e) => {
                if attempts >= 3 {
                    metrics.connections_failed.fetch_add(1, Ordering::Relaxed);
                    return;
                }
                // Wait before retry with exponential backoff
                sleep(Duration::from_millis(100 * attempts)).await;
            }
        }
    };

    let (write, mut read): (WsWriter, WsReader) = ws_stream.split();
    let write = Arc::new(Mutex::new(write));
    let write_clone = Arc::clone(&write);

    let ping_task = tokio::spawn(async move {
        let mut ping_interval = tokio::time::interval(Duration::from_secs(10));
        loop {
            ping_interval.tick().await;

            let ping_msg = ClientToServerMsg::Ping {
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap()
                    .as_micros(),
            };

            if let Ok(bytes) = postcard::to_allocvec(&ping_msg) {
                if write_clone
                    .lock()
                    .await
                    .send(Message::Binary(bytes.into()))
                    .await
                    .is_err()
                {
                    break;
                }
            }
        }
    });

    // --- 2. Login ---
    let login_msg = ClientToServerMsg::Login {
        username: format!("user_{}", user_id),
    };
    if let Ok(bytes) = postcard::to_allocvec(&login_msg) {
        if write
            .lock()
            .await
            .send(Message::Binary(bytes.into()))
            .await
            .is_err()
        {
            ping_task.abort();
            return;
        }
    }

    let login_confirmed = tokio::time::timeout(Duration::from_secs(5), async {
        while let Some(Ok(Message::Binary(bin))) = read.next().await {
            if let Ok(ServerToClientMsg::LoginSuccess { .. }) = postcard::from_bytes(&bin) {
                return true;
            }
        }
        false
    })
    .await
    .unwrap_or(false);

    if !login_confirmed {
        ping_task.abort();
        // Close the connection properly
        let _ = write.lock().await.close().await;
        return;
    }
    metrics.logins_successful.fetch_add(1, Ordering::Relaxed);

    // --- 3. Join Group ---
    let join_msg = ClientToServerMsg::JoinGroup {
        group: GROUP_NAME.to_string(),
    };

    metrics.group_joins_attempted.fetch_add(1, Ordering::Relaxed);

    if let Ok(bytes) = postcard::to_allocvec(&join_msg) {
        if write
            .lock()
            .await
            .send(Message::Binary(bytes.into()))
            .await
            .is_err()
        {
            metrics.group_joins_failed.fetch_add(1, Ordering::Relaxed);
            ping_task.abort();
            let _ = write.lock().await.close().await;
            return;
        }
    }

    // Wait for join confirmation
    let join_confirmed = tokio::time::timeout(Duration::from_secs(5), async {
        while let Some(Ok(Message::Binary(bin))) = read.next().await {
            match postcard::from_bytes::<ServerToClientMsg>(&bin) {
                Ok(ServerToClientMsg::JoinedGroup { group }) if group == GROUP_NAME => {
                    return true;
                }
                Ok(ServerToClientMsg::Error { .. }) => {
                    return false;
                }
                _ => continue,
            }
        }
        false
    })
    .await
    .unwrap_or(false);

    if join_confirmed {
        metrics.group_joins_successful.fetch_add(1, Ordering::Relaxed);
    } else {
        metrics.group_joins_failed.fetch_add(1, Ordering::Relaxed);
        ping_task.abort();
        let _ = write.lock().await.close().await;
        return;
    }

    // --- 4. Messaging loop ---
    let metrics_clone_reader = Arc::clone(&metrics);
    let reader_task = tokio::spawn(async move {
        while let Some(Ok(Message::Binary(bin))) = read.next().await {
            metrics_clone_reader
                .messages_received
                .fetch_add(1, Ordering::Relaxed);

            if let Ok(ServerToClientMsg::Pong { timestamp }) = postcard::from_bytes(&bin) {
                let now_micros = SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap()
                    .as_micros();
                let rtt_us = now_micros.saturating_sub(timestamp);
                if rtt_us < 1_000_000 {
                    metrics_clone_reader
                        .round_trip_latency_us
                        .fetch_add(rtt_us as u64, Ordering::Relaxed);
                    metrics_clone_reader
                        .round_trip_count
                        .fetch_add(1, Ordering::Relaxed);
                }
            }
        }
    });

    let test_start_time = Instant::now();
    let mut message_interval = tokio::time::interval(Duration::from_millis(500));

    while test_start_time.elapsed() < TEST_DURATION {
        message_interval.tick().await;

        let msg = if metrics.messages_sent.load(Ordering::Relaxed) % 5 == 0 {
            ClientToServerMsg::Ping {
                timestamp: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap()
                    .as_micros(),
            }
        } else {
            ClientToServerMsg::SendGroup {
                group: GROUP_NAME.to_string(),
                text: "hello group".to_string(),
            }
        };

        if let Ok(bytes) = postcard::to_allocvec(&msg) {
            if write
                .lock()
                .await
                .send(Message::Binary(bytes.into()))
                .await
                .is_ok()
            {
                metrics.messages_sent.fetch_add(1, Ordering::Relaxed);
            } else {
                break;
            }
        }
    }

    let messaging_duration = test_start_time.elapsed();
    if messaging_duration < TEST_DURATION - Duration::from_secs(5) {
        // User exited early - sample the logging to reduce spam
        let count = EARLY_EXIT_COUNT.fetch_add(1, Ordering::Relaxed);
        
        // Only log every 50th early exit, or the first one
        if count == 0 || count % 50 == 0 {
            eprintln!(
                "⚠️ {} users exited early (user {} after {:.1}s, expected ~{}s)",
                count + 1,
                user_id,
                messaging_duration.as_secs_f32(),
                TEST_DURATION.as_secs()
            );
        }
    }


    // --- 5. PROPER CLEANUP ---
    // Stop background tasks
    ping_task.abort();
    reader_task.abort();

    // Send WebSocket close frame
    {
        let mut writer = write.lock().await;
        
        // Send close frame
        let _ = writer.send(Message::Close(None)).await;
        
        // Flush and close
        let _ = writer.flush().await;
        let _ = writer.close().await;
    }
}

async fn reset_server() -> Result<(), Box<dyn std::error::Error>> {
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(60))
        .build()?;

    println!("🔄 Resetting server state...");

    // Check health first
    for attempt in 0..5 {
        match client.get("http://localhost:8080/health")
            .timeout(Duration::from_secs(2))
            .send()
            .await
        {
            Ok(resp) if resp.status().is_success() => break,
            _ if attempt < 4 => {
                println!("   Waiting for server to respond... ({}/5)", attempt + 1);
                sleep(Duration::from_secs(2)).await;
            }
            _ => {
                println!("❌ Server not responding after 10s");
                return Err("Server unresponsive".into());
            }
        }
    }

    // Request reset
    match client.post("http://localhost:8080/admin/reset").send().await {
        Ok(resp) if resp.status().is_success() => {
            println!("   Reset signal sent");
        }
        Ok(resp) => {
            println!("   Reset returned status: {}", resp.status());
        }
        Err(e) => {
            println!("   Reset request failed: {}", e);
            // Server might be overwhelmed, wait and retry
            tokio::time::sleep(Duration::from_secs(2)).await;
            client.post("http://localhost:8080/admin/reset").send().await?;
        }
    }

    // Poll until connections reach 0
    println!("   Waiting for connections to drain...");
    for i in 0..60 {
        sleep(Duration::from_secs(1)).await;
        
        if let Ok(resp) = client.get("http://localhost:8080/stats")
            .timeout(Duration::from_secs(2))
            .send()
            .await
        {
            if let Ok(stats) = resp.json::<serde_json::Value>().await {
                let connections = stats["total_connections"].as_u64().unwrap_or(u64::MAX);
                print!("\r   Connections: {} ({}s elapsed)    ", connections, i + 1);
                std::io::Write::flush(&mut std::io::stdout()).unwrap();
                
                if connections == 0 {
                    println!("\n✅ Reset complete");
                    // Extra wait for OS to release ports
                    sleep(Duration::from_secs(3)).await;
                    return Ok(());
                }
            }
        }
    }

    for i in 1..=30 {  // 30 seconds max
        tokio::time::sleep(Duration::from_secs(1)).await;
        
        match client.get("http://localhost:8080/health").send().await {
            Ok(resp) if resp.status().is_success() => {
                // Check connection count from response or separate endpoint
                println!("   Server healthy after {}s", i);
                return Ok(());
            }
            Ok(_) => {
                println!("   Waiting for server to respond... ({}/30)", i);
            }
            Err(_) => {
                println!("   Server not responding... ({}/30)", i);
            }
        }
    }
    
    println!("\n⚠️ Connections still remaining, waiting longer...");
    sleep(Duration::from_secs(10)).await;
    
    Ok(())
}

async fn run_single_test(run_number: usize) -> TestResult {

    EARLY_EXIT_COUNT.store(0, Ordering::Relaxed);
    
    let metrics = Arc::new(TestMetrics::default());
    let mut user_handles: Vec<tokio::task::JoinHandle<()>> = Vec::with_capacity(TOTAL_USERS as usize);

    let connect_interval = if USERS_PER_SECOND > 0 {
        Duration::from_millis(1000 / USERS_PER_SECOND)
    } else {
        Duration::from_secs(0)
    };

    let test_start_time = Instant::now();

    // Start progress reporter
    let metrics_clone = metrics.clone();
    let progress_task = tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(2));
        loop {
            interval.tick().await;

            let attempted = metrics_clone.group_joins_attempted.load(Ordering::Relaxed);
            let successful = metrics_clone.group_joins_successful.load(Ordering::Relaxed);

            if attempted > 0 {
                let success_rate = (successful as f64 / attempted as f64) * 100.0;
                print!(
                    "\r⏳ Progress: {}/{} users joined ({:.1}% success rate)",
                    successful, TOTAL_USERS, success_rate
                );
                std::io::Write::flush(&mut std::io::stdout()).unwrap();
            }

            if attempted >= TOTAL_USERS {
                break;
            }
        }
    });

    // Spawn user simulation tasks
    for i in 0..TOTAL_USERS {
        let metrics_clone = Arc::clone(&metrics);
        let handle = tokio::spawn(async move {
            simulate_user(i, metrics_clone).await;
        });
        user_handles.push(handle);

        if USERS_PER_SECOND > 0 {
            sleep(connect_interval).await;
        }
    }

    // Wait for ALL user tasks to complete (including cleanup)
    for handle in user_handles {
        let _ = handle.await;
    }

    // Stop progress reporter
    progress_task.abort();

    // Print summary
    let total_attempted = metrics.group_joins_attempted.load(Ordering::Relaxed);
    let total_successful = metrics.group_joins_successful.load(Ordering::Relaxed);
    let total_failed = metrics.group_joins_failed.load(Ordering::Relaxed);

    println!(
        "\n✅ Join Summary: {}/{} successful, {} failed ({:.1}% success rate)",
        total_successful,
        total_attempted,
        total_failed,
        if total_attempted > 0 {
            (total_successful as f64 / total_attempted as f64) * 100.0
        } else {
            0.0
        }
    );

    // Wait for TCP cleanup
    println!("   Waiting for TCP cleanup...");
    sleep(Duration::from_secs(2)).await;

    let elapsed = test_start_time.elapsed();
    TestResult::from_metrics(run_number, &metrics, elapsed)
}