// Load tests for rust-fafb-q.
//
// All tests are #[ignore] so they don't run in normal `cargo test`.
// Run with: cargo test --test load_test -- --ignored --nocapture
//
// Scale via env vars:
//   LOAD_TEST_CONNECTIONS  — number of client connections (default: 100)
//   LOAD_TEST_DURATION_SECS — test duration in seconds (default: 30)

mod common;

use std::sync::Arc;
use std::time::Duration;

use common::{
    ClientPool, LatencyStats, NsConfig, TEST_KEY, TEST_NS, TestClient, start_test_tcp_server,
    start_test_tcp_server_multi_ns,
};
use fafb_q::codec::Frame;
use fafb_q::namespace::manager::NamespaceQuota;
use fafb_q::queue::BackpressurePolicy;
use fafb_q::queue::{EngineConfig, InMemoryQueueEngine, QueueEngine};
use fafb_q::storage::StorageBackend;
use fafb_q::storage::mock::MockStorage;
use fafb_q::types::{MessageEnvelope, MessageId, NamespaceId, TopicName, TopicOffset};

fn conn_count() -> usize {
    std::env::var("LOAD_TEST_CONNECTIONS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(100)
}

fn test_duration() -> Duration {
    let secs = std::env::var("LOAD_TEST_DURATION_SECS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(30);
    Duration::from_secs(secs)
}

fn make_envelope(id: u64, ns: &str, topic: &str, payload: &[u8]) -> MessageEnvelope {
    MessageEnvelope::new(
        MessageId(id + 1),
        NamespaceId(ns.into()),
        TopicName(topic.into()),
        payload.to_vec(),
        fafb_q::types::now_millis(),
    )
}

/// Sustained load: N publishers + N subscribers publish for D seconds.
///
/// Asserts: all published messages are delivered, p99 stable, no panics.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn sustained_load() {
    let n = conn_count();
    let duration = test_duration();
    let server = start_test_tcp_server().await;

    // Publisher pool
    let mut publishers = ClientPool::connect(&server.addr, n, TEST_NS, TEST_KEY).await;

    // Subscriber pool — separate connections
    let mut subscribers = ClientPool::connect(&server.addr, n, TEST_NS, TEST_KEY).await;

    // Subscribe all subscribers
    let topic = "sustained-topic";
    let collect_duration = duration + Duration::from_secs(5);
    let sub_handle = tokio::spawn(async move {
        let (received, out_of_order) = subscribers
            .subscribe_and_collect(topic, collect_duration)
            .await;
        (received, out_of_order, subscribers)
    });

    // Give subscribers time to register
    tokio::time::sleep(Duration::from_millis(500)).await;

    // Publish at 1000 msgs/sec total
    let (published, latencies) = publishers
        .sustained_publish(topic, 1000, duration, 64)
        .await;

    // Wait for delivery to settle
    tokio::time::sleep(Duration::from_secs(3)).await;

    let (received, out_of_order, _subscribers) = sub_handle.await.unwrap();
    let stats = LatencyStats::from_samples(latencies);

    println!("--- sustained_load results ---");
    println!("connections: {n} publishers + {n} subscribers");
    println!("duration: {}s", duration.as_secs());
    println!("published: {published}");
    println!("received (total across subscribers): {received}");
    println!("out-of-order deliveries: {out_of_order}");
    println!(
        "publish latency: p50={:.3}ms p95={:.3}ms p99={:.3}ms max={:.3}ms",
        stats.p50 * 1000.0,
        stats.p95 * 1000.0,
        stats.p99 * 1000.0,
        stats.max * 1000.0
    );

    assert!(published > 0, "should have published messages");
    // p99 should be under 500ms for local testing
    assert!(
        stats.p99 < 0.5,
        "p99 latency {:.3}ms exceeds 500ms",
        stats.p99 * 1000.0
    );
    // Note: with multiple concurrent publishers, message IDs interleave at
    // the server's global counter, so per-subscriber message ID ordering is
    // not guaranteed. Offset ordering (the real invariant) is tested in unit
    // tests. Here we just report the count as diagnostic data.
    if out_of_order > 0 {
        println!(
            "note: {out_of_order} out-of-order message IDs (expected with {} concurrent publishers)",
            n
        );
    }

    server.shutdown();
}

/// Burst connections: 2x max_connections simultaneously.
///
/// Asserts: half accepted, half rejected cleanly, survivors functional.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn burst_connections() {
    let max_conns = 50;
    let burst = max_conns * 2;

    let ns_config = NsConfig {
        namespace: "burst-ns".to_string(),
        api_key: "burst-key".to_string(),
        quota: NamespaceQuota {
            max_connections: Some(max_conns),
            ..Default::default()
        },
    };
    let server = start_test_tcp_server_multi_ns(vec![ns_config]).await;

    // Try to connect `burst` clients simultaneously
    let mut handles = tokio::task::JoinSet::new();
    for _ in 0..burst {
        let addr = server.addr.clone();
        handles.spawn(async move {
            let mut client = TestClient::connect(&addr).await;
            client
                .send(Frame::Connect {
                    version: 1,
                    namespace: NamespaceId("burst-ns".into()),
                    api_key: fafb_q::types::ApiKey("burst-key".into()),
                    capabilities: 0,
                })
                .await;
            let ack = client.recv().await;
            match ack {
                Frame::ConnectAck { success: true, .. } => Ok(client),
                Frame::Error { .. } => Err("rejected"),
                Frame::ConnectAck { success: false, .. } => Err("connect failed"),
                _other => Err("unexpected frame"),
            }
        });
    }

    let mut accepted = 0u64;
    let mut rejected = 0u64;
    let mut survivors = Vec::new();

    while let Some(result) = handles.join_next().await {
        match result {
            Ok(Ok(client)) => {
                accepted += 1;
                survivors.push(client);
            }
            Ok(Err(_)) => {
                rejected += 1;
            }
            Err(e) => {
                panic!("connection task panicked: {e}");
            }
        }
    }

    println!("--- burst_connections results ---");
    println!("attempted: {burst}");
    println!("accepted: {accepted}");
    println!("rejected: {rejected}");

    // At least some should be accepted (up to max_conns), at least some rejected
    assert!(
        accepted > 0 && accepted <= max_conns as u64,
        "expected 1..={max_conns} accepted, got {accepted}"
    );
    assert!(rejected > 0, "expected some rejected connections");

    // Survivors should be functional — publish a message
    if let Some(mut client) = survivors.into_iter().next() {
        client
            .send(Frame::Publish {
                topic: TopicName("burst-test".into()),
                payload: bytes::Bytes::from_static(b"alive"),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let ack = client.recv_timeout(Duration::from_secs(2)).await;
        assert!(
            matches!(ack, Some(Frame::PublishAck { .. })),
            "survivor should be able to publish"
        );
    }

    server.shutdown();
}

/// Slow consumer isolation: 50 fast + 1 slow subscriber.
///
/// Asserts: fast subs get all msgs, slow has drops, fast p99 < 100ms.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn slow_consumer_isolation() {
    let server = start_test_tcp_server().await;
    let topic = "slow-topic";
    let msg_count = 1000u64;

    // 50 fast subscribers
    let mut fast_clients = Vec::new();
    for _ in 0..50 {
        let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
        client
            .send(Frame::Subscribe {
                topic: TopicName(topic.into()),
                group: None,
            })
            .await;
        let _ack = client.recv().await;
        fast_clients.push(client);
    }

    // 1 slow subscriber (will not read from its channel)
    let mut slow_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    slow_client
        .send(Frame::Subscribe {
            topic: TopicName(topic.into()),
            group: None,
        })
        .await;
    let _ack = slow_client.recv().await;
    // Keep slow_client alive (but never read from it) to simulate a slow consumer
    let _slow_client = slow_client;

    // Publish messages
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..msg_count {
        publisher
            .send(Frame::Publish {
                topic: TopicName(topic.into()),
                payload: bytes::Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ack = publisher.recv().await;
    }

    // Collect from fast subscribers
    let timeout = Duration::from_secs(10);
    let mut fast_received = Vec::new();
    let mut fast_latencies = Vec::new();

    for mut client in fast_clients {
        let mut count = 0u64;
        let start = std::time::Instant::now();
        loop {
            match client.recv_timeout(Duration::from_millis(500)).await {
                Some(Frame::Deliver { .. }) => {
                    count += 1;
                    if count >= msg_count {
                        break;
                    }
                }
                _ => {
                    if start.elapsed() > timeout {
                        break;
                    }
                }
            }
        }
        fast_latencies.push(start.elapsed().as_secs_f64());
        fast_received.push(count);
    }

    let min_received = fast_received.iter().copied().min().unwrap_or(0);
    let max_received = fast_received.iter().copied().max().unwrap_or(0);
    let avg_received = fast_received.iter().sum::<u64>() as f64 / fast_received.len() as f64;

    println!("--- slow_consumer_isolation results ---");
    println!("published: {msg_count}");
    println!("fast subscribers: 50");
    println!("fast min/avg/max received: {min_received}/{avg_received:.0}/{max_received}");

    // The key invariant: fast subscribers receive all/nearly all messages
    // even with a slow subscriber on the same topic. This validates that
    // per-subscriber delivery task isolation works correctly.
    assert!(
        min_received >= msg_count * 9 / 10,
        "fast subscribers should receive >=90% of messages, got min={min_received}/{msg_count}"
    );

    // We intentionally don't drain the slow subscriber here — doing so
    // creates a concurrent reader that feeds back into the delivery task,
    // defeating the "slow" simulation. The slow subscriber's internal state
    // (is_slow, total_drops) is validated by unit tests in manager.rs.

    server.shutdown();
}

/// Memory pressure: engine with 10MB budget, publish until exhausted.
///
/// Asserts: ResourceExhausted error, process alive, recovery after compaction.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn memory_pressure() {
    let budget = 10 * 1024 * 1024; // 10MB
    let engine = Arc::new(InMemoryQueueEngine::new(
        Arc::new(MockStorage::new()),
        EngineConfig {
            max_depth: 1_000_000,
            backpressure: BackpressurePolicy::Reject,
            topic_max_bytes: 0,
            global_max_memory_bytes: budget,
            flush_interval_ms: 100,
            ..Default::default()
        },
    ));

    let ns = NamespaceId("pressure-ns".into());
    let topic = TopicName("pressure-topic".into());

    // Publish until exhausted
    let payload = vec![0xABu8; 1024]; // 1KB per message
    let mut published = 0u64;
    let mut exhausted = false;

    for i in 0..100_000u64 {
        match engine
            .publish(
                &ns,
                &topic,
                make_envelope(i, "pressure-ns", "pressure-topic", &payload),
            )
            .await
        {
            Ok(_) => published += 1,
            Err(e) => {
                assert!(
                    e.to_string().contains("global memory budget exceeded"),
                    "expected ResourceExhausted, got: {e}"
                );
                exhausted = true;
                break;
            }
        }
    }

    println!("--- memory_pressure results ---");
    println!("budget: {budget} bytes");
    println!("published before exhaustion: {published}");
    println!("exhausted: {exhausted}");

    assert!(exhausted, "should have hit memory budget");
    assert!(published > 0, "should have published some messages");

    // Compact to free memory
    let half = TopicOffset(published / 2);
    let result = engine.compact(&ns, &topic, half).await.unwrap();
    println!(
        "compacted: freed {} messages, {} bytes",
        result.messages_freed, result.bytes_freed
    );

    // Should be able to publish again after compaction
    let recovery = engine
        .publish(
            &ns,
            &topic,
            make_envelope(published, "pressure-ns", "pressure-topic", &payload),
        )
        .await;
    assert!(
        recovery.is_ok(),
        "should be able to publish after compaction"
    );
}

/// Namespace isolation: 10 namespaces, 1 fills quota.
///
/// Asserts: other 9 unaffected, aggressive gets quota error.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn namespace_isolation() {
    let ns_count = 10;
    let quota_limit = 100u64;

    let ns_configs: Vec<NsConfig> = (0..ns_count)
        .map(|i| NsConfig {
            namespace: format!("ns-{i}"),
            api_key: format!("key-{i}"),
            quota: NamespaceQuota {
                max_total_messages: Some(quota_limit),
                ..Default::default()
            },
        })
        .collect();

    let server = start_test_tcp_server_multi_ns(ns_configs).await;

    // Fill namespace 0 to its quota
    let mut aggressor = TestClient::connect_and_auth(&server.addr, "ns-0", "key-0").await;

    let mut published = 0u64;
    let mut got_quota_error = false;
    for i in 0..200u64 {
        aggressor
            .send(Frame::Publish {
                topic: TopicName("fill-topic".into()),
                payload: bytes::Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let resp = aggressor.recv().await;
        match resp {
            Frame::PublishAck { .. } => published += 1,
            Frame::Error { .. } => {
                got_quota_error = true;
                break;
            }
            _ => {}
        }
    }

    println!("--- namespace_isolation results ---");
    println!("aggressor (ns-0) published: {published}");
    println!("aggressor got quota error: {got_quota_error}");

    assert!(got_quota_error, "aggressor should hit quota limit");

    // Other namespaces should be unaffected
    for i in 1..ns_count {
        let ns = format!("ns-{i}");
        let key = format!("key-{i}");
        let mut client = TestClient::connect_and_auth(&server.addr, &ns, &key).await;
        client
            .send(Frame::Publish {
                topic: TopicName("test-topic".into()),
                payload: bytes::Bytes::from_static(b"hello"),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let resp = client.recv().await;
        assert!(
            matches!(resp, Frame::PublishAck { .. }),
            "ns-{i} should be able to publish, got: {resp:?}"
        );
    }

    server.shutdown();
}

/// Storage durability: 10K messages -> drop engine -> replay from same dir.
///
/// Asserts: replayed count >= published.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore]
async fn storage_durability() {
    use fafb_q::storage::FsStorage;

    let tmp_dir = tempfile::TempDir::new().unwrap();
    let data_dir = tmp_dir.path().join("durability-data");

    let msg_count = 10_000u64;
    let ns = NamespaceId("durable-ns".into());
    let topic = TopicName("durable-topic".into());

    // Phase 1: Publish with FsStorage
    {
        let storage = Arc::new(FsStorage::new(&data_dir));
        storage.init().await.unwrap();

        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::clone(&storage),
            EngineConfig {
                max_depth: 100_000,
                backpressure: BackpressurePolicy::Reject,
                topic_max_bytes: 0,
                global_max_memory_bytes: 0,
                flush_interval_ms: 50,
                ..Default::default()
            },
        ));
        engine.init().await.unwrap();

        for i in 0..msg_count {
            engine
                .publish(
                    &ns,
                    &topic,
                    make_envelope(i, "durable-ns", "durable-topic", &[0xABu8; 64]),
                )
                .await
                .unwrap();
        }

        // Flush to ensure data is on disk
        storage.flush(&ns, &topic).await.unwrap();
        engine.shutdown().await.unwrap();
    }

    // Phase 2: New engine, replay from same directory
    {
        let storage = Arc::new(FsStorage::new(&data_dir));
        storage.init().await.unwrap();

        let engine = Arc::new(InMemoryQueueEngine::new(
            Arc::clone(&storage),
            EngineConfig {
                max_depth: 100_000,
                backpressure: BackpressurePolicy::Reject,
                topic_max_bytes: 0,
                global_max_memory_bytes: 0,
                flush_interval_ms: 50,
                ..Default::default()
            },
        ));
        engine.init().await.unwrap();

        let head = engine.head_offset(&ns, &topic).await.unwrap();
        let replayed = head.0;

        println!("--- storage_durability results ---");
        println!("published: {msg_count}");
        println!("replayed: {replayed}");

        assert!(
            replayed >= msg_count,
            "replayed {replayed} should be >= published {msg_count}"
        );

        // Verify a sample of messages
        for offset in [0, msg_count / 2, msg_count - 1] {
            let msg = engine
                .consume_from(&ns, &topic, TopicOffset(offset))
                .await
                .unwrap();
            assert!(
                msg.is_some(),
                "message at offset {offset} should exist after replay"
            );
        }

        engine.shutdown().await.unwrap();
    }
}

fn throughput_env(name: &str, default: usize) -> usize {
    std::env::var(name)
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(default)
}

/// Full-scale throughput test matching a specific production scenario:
///
///   10 namespaces × 50 topics = 500 topics total
///   100 publishers (10/ns) round-robin across topics in their namespace
///   1000 consumers (100/ns, 2/topic via individual fan-out)
///   Target: 1000 msgs/sec/topic = 500K msgs/sec total ingest
///   Expected delivery: 2× fan-out = 1M deliveries/sec
///
/// Each publisher sends at (rate_per_topic × topics_per_ns / pubs_per_ns) =
/// (1000 × 50 / 10) = 5000 msgs/sec, cycling through all topics.
///
/// Override defaults via env vars:
///   THROUGHPUT_NAMESPACES=10   THROUGHPUT_TOPICS_PER_NS=50
///   THROUGHPUT_PUBLISHERS=100  THROUGHPUT_CONSUMERS=1000
///   THROUGHPUT_RATE_PER_TOPIC=1000  THROUGHPUT_DURATION_SECS=30
///   THROUGHPUT_PAYLOAD_SIZE=128
///
/// Run with:
///   cargo test --test load_test -- --ignored throughput_full_scale --nocapture
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore]
async fn throughput_full_scale() {
    // --- Configuration ---
    let num_ns = throughput_env("THROUGHPUT_NAMESPACES", 10);
    let topics_per_ns = throughput_env("THROUGHPUT_TOPICS_PER_NS", 50);
    let total_pubs = throughput_env("THROUGHPUT_PUBLISHERS", 100);
    let total_cons = throughput_env("THROUGHPUT_CONSUMERS", 1000);
    let rate_per_topic = throughput_env("THROUGHPUT_RATE_PER_TOPIC", 1000);
    let duration_secs = throughput_env("THROUGHPUT_DURATION_SECS", 30);
    let payload_size = throughput_env("THROUGHPUT_PAYLOAD_SIZE", 128);

    let duration = Duration::from_secs(duration_secs as u64);
    let pubs_per_ns = total_pubs / num_ns.max(1);
    let cons_per_ns = total_cons / num_ns.max(1);
    let cons_per_topic = cons_per_ns / topics_per_ns.max(1);
    let total_topics = num_ns * topics_per_ns;
    let target_total_rate = rate_per_topic * total_topics;

    println!("--- throughput_full_scale configuration ---");
    println!("namespaces:     {num_ns}");
    println!("topics/ns:      {topics_per_ns} (total: {total_topics})");
    println!("publishers:     {total_pubs} ({pubs_per_ns}/ns)");
    println!("consumers:      {total_cons} ({cons_per_ns}/ns, {cons_per_topic}/topic)");
    println!(
        "target rate:    {rate_per_topic} msgs/s/topic = {target_total_rate} msgs/s total ingest"
    );
    println!(
        "expected delivery: {cons_per_topic}x fan-out = {} msgs/s",
        target_total_rate * cons_per_topic
    );
    println!("duration:       {duration_secs}s");
    println!("payload:        {payload_size}B");
    println!();

    // --- Server setup ---
    let ns_configs: Vec<NsConfig> = (0..num_ns)
        .map(|i| NsConfig {
            namespace: format!("ns-{i}"),
            api_key: format!("throughput-key-ns-{i:03}"),
            quota: NamespaceQuota::default(),
        })
        .collect();
    let server = start_test_tcp_server_multi_ns(ns_configs).await;

    // --- Phase 1: Connect & subscribe consumers ---
    //
    // Each consumer subscribes to one topic (round-robin assignment).
    // Consumer deadline extends past publish duration to drain in-flight messages.
    let consumer_deadline = tokio::time::Instant::now() + duration + Duration::from_secs(15);
    let mut consumer_tasks = tokio::task::JoinSet::new();

    for ns_idx in 0..num_ns {
        let ns_name = format!("ns-{ns_idx}");
        let key = format!("throughput-key-ns-{ns_idx:03}");

        for c in 0..cons_per_ns {
            let topic_idx = c % topics_per_ns;
            let addr = server.addr.clone();
            let ns = ns_name.clone();
            let k = key.clone();

            consumer_tasks.spawn(async move {
                let mut client = TestClient::connect_and_auth(&addr, &ns, &k).await;
                client
                    .send(Frame::Subscribe {
                        topic: TopicName(format!("topic-{topic_idx}")),
                        group: None,
                    })
                    .await;
                let _ack = client.recv().await;

                let mut count = 0u64;
                loop {
                    if tokio::time::Instant::now() >= consumer_deadline {
                        break;
                    }
                    match client.recv_timeout(Duration::from_millis(100)).await {
                        Some(Frame::Deliver { .. }) => count += 1,
                        _ => {}
                    }
                }
                (ns_idx, topic_idx, count)
            });
        }
    }

    // Let all subscriptions register before publishing.
    tokio::time::sleep(Duration::from_secs(2)).await;
    println!("{total_cons} consumers connected and subscribed");

    // --- Phase 2: Publishers run for `duration` ---
    let publish_wall_start = std::time::Instant::now();
    let publisher_deadline = tokio::time::Instant::now() + duration;
    let mut publisher_tasks = tokio::task::JoinSet::new();
    let payload_bytes = bytes::Bytes::from(vec![0xABu8; payload_size]);

    for ns_idx in 0..num_ns {
        let ns_name = format!("ns-{ns_idx}");
        let key = format!("throughput-key-ns-{ns_idx:03}");
        // Pre-compute topic names to avoid per-message allocations.
        let topics: Vec<TopicName> = (0..topics_per_ns)
            .map(|t| TopicName(format!("topic-{t}")))
            .collect();

        for p in 0..pubs_per_ns {
            let addr = server.addr.clone();
            let ns = ns_name.clone();
            let k = key.clone();
            let topics = topics.clone();
            let payload = payload_bytes.clone();

            publisher_tasks.spawn(async move {
                let mut client = TestClient::connect_and_auth(&addr, &ns, &k).await;

                // This publisher round-robins across all topics in its namespace.
                // Rate = (rate_per_topic × topics_per_ns) / pubs_per_ns msgs/sec.
                let msgs_per_sec = (rate_per_topic * topics_per_ns) / pubs_per_ns.max(1);
                let interval_ns = 1_000_000_000u64 / (msgs_per_sec as u64).max(1);
                let mut ticker = tokio::time::interval(Duration::from_nanos(interval_ns));
                ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

                let mut count = 0u64;
                let mut errors = 0u64;
                let mut latencies = Vec::with_capacity(msgs_per_sec * duration_secs + 1);
                let mut t_idx = p; // stagger starting topic per publisher

                loop {
                    ticker.tick().await;
                    if tokio::time::Instant::now() >= publisher_deadline {
                        break;
                    }

                    let topic = topics[t_idx % topics_per_ns].clone();
                    t_idx += 1;

                    let start = std::time::Instant::now();
                    client
                        .send(Frame::Publish {
                            topic,
                            payload: payload.clone(),
                            headers: vec![],
                            ttl_ms: 0,
                        })
                        .await;
                    match client.recv().await {
                        Frame::PublishAck { .. } => {
                            count += 1;
                            latencies.push(start.elapsed().as_secs_f64());
                        }
                        Frame::Error { message, .. } => {
                            errors += 1;
                            if errors <= 3 {
                                eprintln!("  publisher ns-{ns_idx}#{p}: {message}");
                            }
                        }
                        _ => {}
                    }
                }
                (ns_idx, count, errors, latencies)
            });
        }
    }

    println!("{total_pubs} publishers connected — publishing for {duration_secs}s...");

    // --- Phase 3: Collect publisher results ---
    let mut total_published = 0u64;
    let mut total_pub_errors = 0u64;
    let mut ns_published = vec![0u64; num_ns];
    let mut all_latencies: Vec<f64> = Vec::new();

    while let Some(result) = publisher_tasks.join_next().await {
        match result {
            Ok((ns_idx, count, errs, latencies)) => {
                total_published += count;
                total_pub_errors += errs;
                ns_published[ns_idx] += count;
                all_latencies.extend(latencies);
            }
            Err(e) => eprintln!("publisher task panicked: {e}"),
        }
    }

    let publish_elapsed = publish_wall_start.elapsed();
    println!(
        "publishers done: {} msgs in {:.1}s ({} errors)",
        total_published,
        publish_elapsed.as_secs_f64(),
        total_pub_errors,
    );
    println!("waiting for delivery to settle...");

    // --- Phase 4: Collect consumer results ---
    let mut total_received = 0u64;
    let mut ns_received = vec![0u64; num_ns];
    let mut per_topic: std::collections::HashMap<(usize, usize), u64> =
        std::collections::HashMap::new();

    while let Some(result) = consumer_tasks.join_next().await {
        match result {
            Ok((ns_idx, topic_idx, count)) => {
                total_received += count;
                ns_received[ns_idx] += count;
                *per_topic.entry((ns_idx, topic_idx)).or_default() += count;
            }
            Err(e) => eprintln!("consumer task panicked: {e}"),
        }
    }

    // --- Compute stats ---
    let stats = LatencyStats::from_samples(all_latencies);
    let actual_pub_rate = total_published as f64 / publish_elapsed.as_secs_f64();
    let expected_deliveries = total_published * cons_per_topic as u64;
    let delivery_pct = if expected_deliveries > 0 {
        total_received as f64 / expected_deliveries as f64 * 100.0
    } else {
        0.0
    };
    let actual_delivery_rate = total_received as f64 / publish_elapsed.as_secs_f64();

    // --- Report ---
    println!();
    println!("=======================================");
    println!("  throughput_full_scale results");
    println!("=======================================");
    println!(
        "publish elapsed:      {:.1}s",
        publish_elapsed.as_secs_f64()
    );
    println!("total published:      {total_published}");
    println!(
        "actual publish rate:  {:.0} msgs/s ({:.1}% of {target_total_rate} target)",
        actual_pub_rate,
        actual_pub_rate / target_total_rate.max(1) as f64 * 100.0
    );
    println!("publish errors:       {total_pub_errors}");
    println!();
    println!("total delivered:      {total_received}");
    println!("expected deliveries:  {expected_deliveries}");
    println!("delivery ratio:       {delivery_pct:.1}%");
    println!("actual delivery rate: {:.0} msgs/s", actual_delivery_rate);
    println!();
    println!(
        "publish latency:      p50={:.3}ms  p95={:.3}ms  p99={:.3}ms  max={:.3}ms",
        stats.p50 * 1000.0,
        stats.p95 * 1000.0,
        stats.p99 * 1000.0,
        stats.max * 1000.0,
    );

    // Per-namespace breakdown
    println!();
    println!("per-namespace:");
    for ns_idx in 0..num_ns {
        let rate = ns_published[ns_idx] as f64 / publish_elapsed.as_secs_f64();
        println!(
            "  ns-{ns_idx:<2}: pub={:>9} ({:>8.0}/s)  del={:>9}",
            ns_published[ns_idx], rate, ns_received[ns_idx],
        );
    }

    // Per-topic sample (first namespace, first 10 topics)
    if num_ns > 0 && topics_per_ns > 0 {
        println!();
        println!("topic sample (ns-0, first 10):");
        let expected_per_topic =
            ns_published[0] as f64 / topics_per_ns as f64 * cons_per_topic as f64;
        for t in 0..topics_per_ns.min(10) {
            let delivered = per_topic.get(&(0, t)).copied().unwrap_or(0);
            println!(
                "  topic-{t:<2}: delivered={delivered:>9}  (expected ~{expected_per_topic:.0})"
            );
        }
    }
    println!("=======================================");

    // --- Assertions ---
    assert!(total_published > 0, "should have published messages");

    // Synchronous publish (send → wait for ack) caps each connection at
    // ~1/round-trip msgs/sec.  With 100 publishers the practical ceiling is
    // roughly 100K-150K msgs/s regardless of server capacity.  Assert that
    // we reach at least 10% of the target to catch regressions without
    // failing on the inherent protocol overhead.
    let min_pub_rate = target_total_rate as f64 * 0.10;
    assert!(
        actual_pub_rate > min_pub_rate,
        "publish rate {:.0} < 10% of target ({min_pub_rate:.0})",
        actual_pub_rate,
    );

    // Every published message must be delivered to every subscriber on that
    // topic.  100% is the goal; allow ≥ 95% to account for timing at the
    // tail of the settle window.
    assert!(
        delivery_pct > 95.0,
        "delivery ratio {delivery_pct:.1}% < 95%",
    );

    // p99 publish latency under 100ms (generous for varied hardware).
    assert!(
        stats.p99 < 0.100,
        "p99 publish latency {:.3}ms > 100ms",
        stats.p99 * 1000.0,
    );

    server.shutdown();
}
