mod common;

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

use fafb_q::namespace::{ManagerConfig, NamespaceManager};
use fafb_q::types::{ClientId, GroupName, NamespaceId, SubscribeStart, TopicName};

use common::{make_engine, make_envelope, make_namespace_manager};

fn ns() -> NamespaceId {
    NamespaceId("ns".into())
}

fn topic() -> TopicName {
    TopicName("test-topic".into())
}

/// 10 tasks subscribe/unsubscribe while another publishes 50 messages.
/// No deadlocks, active subscribers receive messages.
#[tokio::test]
async fn concurrent_subscribe_unsubscribe_during_publish() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));

    let publish_mgr = Arc::clone(&mgr);
    let publish_handle = tokio::spawn(async move {
        for i in 0..50u64 {
            publish_mgr
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "ns", "test-topic", b"data"),
                )
                .await
                .unwrap();
            if i % 10 == 0 {
                tokio::task::yield_now().await;
            }
        }
    });

    let mut sub_handles = Vec::new();
    for task_id in 0..10u64 {
        let m = Arc::clone(&mgr);
        sub_handles.push(tokio::spawn(async move {
            let client = ClientId(100 + task_id);
            let sub = m
                .subscribe(&ns(), &topic(), client, None, SubscribeStart::Head)
                .await
                .unwrap();
            let mut rx = sub.rx;

            let mut count = 0;
            loop {
                match tokio::time::timeout(Duration::from_millis(200), rx.recv()).await {
                    Ok(Some(_)) => count += 1,
                    _ => break,
                }
                if count >= 5 {
                    break;
                }
            }

            m.unsubscribe(&ns(), &topic(), client, None).await.unwrap();
            count
        }));
    }

    publish_handle.await.unwrap();

    let mut total_received = 0;
    for h in sub_handles {
        total_received += h.await.unwrap();
    }

    assert!(
        total_received > 0,
        "subscribers should have received some messages, got {total_received}"
    );
}

/// 5 tasks join/leave a consumer group during publish. No panic or deadlock.
#[tokio::test]
async fn concurrent_consumer_group_join_leave() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let group = GroupName("workers".into());

    let publish_mgr = Arc::clone(&mgr);
    let publish_handle = tokio::spawn(async move {
        for i in 0..50u64 {
            publish_mgr
                .publish(
                    &ns(),
                    &topic(),
                    make_envelope(i, "ns", "test-topic", b"task"),
                )
                .await
                .unwrap();
            if i % 10 == 0 {
                tokio::task::yield_now().await;
            }
        }
    });

    let mut join_handles = Vec::new();
    for task_id in 0..5u64 {
        let m = Arc::clone(&mgr);
        let g = group.clone();
        join_handles.push(tokio::spawn(async move {
            let client = ClientId(200 + task_id);
            let sub = m
                .subscribe(
                    &ns(),
                    &topic(),
                    client,
                    Some(g.clone()),
                    SubscribeStart::Head,
                )
                .await
                .unwrap();
            let mut rx = sub.rx;

            let mut count = 0;
            loop {
                match tokio::time::timeout(Duration::from_millis(300), rx.recv()).await {
                    Ok(Some(_)) => count += 1,
                    _ => break,
                }
                if count >= 5 {
                    break;
                }
            }

            m.unsubscribe(&ns(), &topic(), client, Some(g))
                .await
                .unwrap();
            count
        }));
    }

    publish_handle.await.unwrap();

    let mut total = 0;
    for h in join_handles {
        total += h.await.unwrap();
    }

    assert!(total > 0, "consumer group should have received messages");
}

/// FIX 6: Reduced from 100 subscribers x 50 messages to 20 x 10.
/// Fan-out to 20 individual subscribers + 3 consumer groups, 10 published messages.
/// Each individual subscriber gets all 10. Each group collectively gets all 10.
#[tokio::test]
async fn fanout_individual_and_groups() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));

    let mut individual_rxs = Vec::new();
    for i in 0..20u64 {
        let sub = mgr
            .subscribe(&ns(), &topic(), ClientId(i), None, SubscribeStart::Head)
            .await
            .unwrap();
        individual_rxs.push(sub.rx);
    }

    let mut group_rxs: Vec<Vec<_>> = Vec::new();
    for g in 0..3u32 {
        let group = GroupName(format!("group-{g}"));
        let mut members = Vec::new();
        for m in 0..3u64 {
            let client = ClientId(1000 + g as u64 * 10 + m);
            let sub = mgr
                .subscribe(
                    &ns(),
                    &topic(),
                    client,
                    Some(group.clone()),
                    SubscribeStart::Head,
                )
                .await
                .unwrap();
            members.push(sub.rx);
        }
        group_rxs.push(members);
    }

    for i in 0..10u64 {
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(i, "ns", "test-topic", format!("msg-{i}").as_bytes()),
        )
        .await
        .unwrap();
    }

    let timeout = Duration::from_secs(3);

    let mut ind_handles = Vec::new();
    for mut rx in individual_rxs {
        ind_handles.push(tokio::spawn(async move {
            let mut count = 0;
            for _ in 0..10 {
                match tokio::time::timeout(timeout, rx.recv()).await {
                    Ok(Some(_)) => count += 1,
                    _ => break,
                }
            }
            count
        }));
    }

    for h in ind_handles {
        let count = h.await.unwrap();
        assert_eq!(
            count, 10,
            "each individual subscriber should get all 10 messages"
        );
    }

    for members in group_rxs {
        let mut group_handles = Vec::new();
        for mut rx in members {
            group_handles.push(tokio::spawn(async move {
                let mut count = 0;
                loop {
                    match tokio::time::timeout(Duration::from_secs(2), rx.recv()).await {
                        Ok(Some(_)) => count += 1,
                        _ => break,
                    }
                }
                count
            }));
        }

        let mut group_total = 0;
        for h in group_handles {
            group_total += h.await.unwrap();
        }
        assert_eq!(
            group_total, 10,
            "each consumer group should collectively get all 10 messages"
        );
    }
}

/// One subscriber with capacity 1 that never reads. Other subscribers still receive messages.
#[tokio::test]
async fn slow_subscriber_doesnt_block_others() {
    let engine = make_engine();
    let mgr = Arc::new(fafb_q::namespace::DefaultNamespaceManager::new(
        engine,
        vec![ns()],
        common::test_shared_config(),
        fafb_q::namespace::ManagerConfig::default(),
    ));

    let _slow_sub = mgr
        .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
        .await
        .unwrap();

    let fast_sub = mgr
        .subscribe(&ns(), &topic(), ClientId(2), None, SubscribeStart::Head)
        .await
        .unwrap();
    let mut fast_rx = fast_sub.rx;

    for i in 0..20u64 {
        mgr.publish(
            &ns(),
            &topic(),
            make_envelope(i, "ns", "test-topic", format!("msg-{i}").as_bytes()),
        )
        .await
        .unwrap();
    }

    let mut fast_count = 0;
    for _ in 0..20 {
        match tokio::time::timeout(Duration::from_secs(3), fast_rx.recv()).await {
            Ok(Some(_)) => fast_count += 1,
            _ => break,
        }
    }

    assert_eq!(
        fast_count, 20,
        "fast subscriber should receive all messages regardless of slow subscriber"
    );
}

/// Verify that delivery_count flows correctly through ack on first delivery
/// and after inflight timeout escalation (sweep increments delivery_count → ack).
#[tokio::test]
async fn delivery_attempts_tracked_through_ack() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(Arc::clone(&engine), vec![ns()]));

    // Set a very short inflight timeout so the sweep escalates quickly.
    mgr.set_inflight_timeout(Duration::from_millis(50)).await;

    let sub = mgr
        .subscribe(&ns(), &topic(), ClientId(1), None, SubscribeStart::Head)
        .await
        .unwrap();
    let mut rx = sub.rx;

    // --- Test 1: Ack on first delivery (delivery_count=1) ---
    mgr.publish(
        &ns(),
        &topic(),
        make_envelope(1, "ns", "test-topic", b"first-try"),
    )
    .await
    .unwrap();

    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive message")
        .expect("channel open");
    // Ack immediately — delivery_count=1 in histogram.
    mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
        .await
        .unwrap();

    // --- Test 2: Let inflight timeout sweep escalate delivery_count, then ack ---
    mgr.publish(
        &ns(),
        &topic(),
        make_envelope(2, "ns", "test-topic", b"retry-me"),
    )
    .await
    .unwrap();

    // Receive first delivery — don't ack immediately.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive message")
        .expect("channel open");

    // Wait for the inflight sweep to fire (50ms timeout + 1s sweep interval).
    // The sweep increments delivery_count from 1 → 2 and resets the timer.
    tokio::time::sleep(Duration::from_millis(1200)).await;

    // Ack after sweep — delivery_count=2 in histogram (sweep escalated it).
    mgr.ack(&ns(), &topic(), ClientId(1), msg.envelope.id)
        .await
        .unwrap();
}

// ── Wildcard subscribe integration tests (Phase 20.3) ───────────────

/// Subscribe to `orders.*`, publish to `orders.us` and `orders.eu`, receive
/// messages from both topics on the same delivery channel.
#[tokio::test]
async fn wildcard_subscribe_receives_from_multiple_topics() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create two topics that match the pattern.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"us-msg"),
    )
    .await
    .unwrap();
    mgr.publish(
        &ns(),
        &TopicName("orders.eu".into()),
        make_envelope(2, "ns", "orders.eu", b"eu-msg"),
    )
    .await
    .unwrap();

    // Subscribe to wildcard pattern.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 2, "should match 2 existing topics");
    let mut rx = sub.rx;

    // Publish another message to each topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(3, "ns", "orders.us", b"us-msg-2"),
    )
    .await
    .unwrap();
    mgr.publish(
        &ns(),
        &TopicName("orders.eu".into()),
        make_envelope(4, "ns", "orders.eu", b"eu-msg-2"),
    )
    .await
    .unwrap();

    // Collect messages — should receive from both topics.
    let mut received_topics = std::collections::HashSet::new();
    for _ in 0..2 {
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("should receive message")
            .expect("channel open");
        received_topics.insert(msg.envelope.topic.0.clone());
    }
    assert!(
        received_topics.contains("orders.us"),
        "should receive from orders.us"
    );
    assert!(
        received_topics.contains("orders.eu"),
        "should receive from orders.eu"
    );
}

/// Subscribe to `events.>`, publish to multi-level nested topics, receive both.
#[tokio::test]
async fn wildcard_gt_subscribe_receives_nested_topics() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create multi-level topics.
    mgr.publish(
        &ns(),
        &TopicName("events.user.created".into()),
        make_envelope(1, "ns", "events.user.created", b"created"),
    )
    .await
    .unwrap();
    mgr.publish(
        &ns(),
        &TopicName("events.order.updated.v2".into()),
        make_envelope(2, "ns", "events.order.updated.v2", b"updated"),
    )
    .await
    .unwrap();

    // Subscribe with `>` pattern.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("events.>".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 2, "should match 2 existing topics");
    let mut rx = sub.rx;

    // Publish new messages.
    mgr.publish(
        &ns(),
        &TopicName("events.user.created".into()),
        make_envelope(3, "ns", "events.user.created", b"created-2"),
    )
    .await
    .unwrap();
    mgr.publish(
        &ns(),
        &TopicName("events.order.updated.v2".into()),
        make_envelope(4, "ns", "events.order.updated.v2", b"updated-2"),
    )
    .await
    .unwrap();

    // Collect messages.
    let mut received_topics = std::collections::HashSet::new();
    for _ in 0..2 {
        let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
            .await
            .expect("should receive message")
            .expect("channel open");
        received_topics.insert(msg.envelope.topic.0.clone());
    }
    assert!(
        received_topics.contains("events.user.created"),
        "should receive from events.user.created"
    );
    assert!(
        received_topics.contains("events.order.updated.v2"),
        "should receive from events.order.updated.v2"
    );
}

/// Unsubscribing a wildcard pattern stops delivery from all matched topics.
#[tokio::test]
async fn wildcard_unsubscribe_stops_all_delivery() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create topics.
    for t in &["orders.us", "orders.eu"] {
        mgr.publish(
            &ns(),
            &TopicName(t.to_string()),
            make_envelope(1, "ns", t, b"init"),
        )
        .await
        .unwrap();
    }

    // Subscribe to wildcard.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx = sub.rx;

    // Unsubscribe the wildcard pattern.
    mgr.unsubscribe(&ns(), &TopicName("orders.*".into()), client, None)
        .await
        .unwrap();

    // Small delay to let forwarders shut down.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Publish again — should NOT be received.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(10, "ns", "orders.us", b"after-unsub"),
    )
    .await
    .unwrap();

    let result = tokio::time::timeout(Duration::from_millis(300), rx.recv()).await;
    assert!(
        result.is_err() || result.unwrap().is_none(),
        "should not receive messages after wildcard unsubscribe"
    );
}

// ── Wildcard auto-subscribe integration tests (Phase 20.4) ──────────

/// Subscribe to wildcard `orders.*`, then publish to a new topic `orders.eu`
/// that didn't exist at subscribe time — verify the message is auto-delivered.
#[tokio::test]
async fn wildcard_auto_subscribe_new_topic_delivers() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Subscribe to wildcard pattern — no matching topics exist yet.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 0);
    let mut rx = sub.rx;

    // Publish to a new topic that matches the wildcard.
    mgr.publish(
        &ns(),
        &TopicName("orders.eu".into()),
        make_envelope(1, "ns", "orders.eu", b"hello-eu"),
    )
    .await
    .unwrap();

    // The wildcard subscriber should receive the message via auto-subscribe.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive auto-subscribed message")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"hello-eu");
}

/// Subscribe to wildcard with 3 existing topics, then create a 4th.
/// Verify all 4 topics deliver messages.
#[tokio::test]
async fn wildcard_auto_subscribe_with_existing_topics() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create 3 topics by publishing to them.
    for region in &["us", "eu", "jp"] {
        let topic = TopicName(format!("orders.{region}"));
        mgr.publish(&ns(), &topic, make_envelope(1, "ns", &topic.0, b"init"))
            .await
            .unwrap();
    }

    // Subscribe to wildcard — should match the 3 existing topics.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 3);
    let mut rx = sub.rx;

    // Publish to a 4th topic that didn't exist at subscribe time.
    mgr.publish(
        &ns(),
        &TopicName("orders.apac".into()),
        make_envelope(10, "ns", "orders.apac", b"hello-apac"),
    )
    .await
    .unwrap();

    // Should receive the auto-subscribed message.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive auto-subscribed message from new topic")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"hello-apac");

    // Now publish to one of the original topics — should also deliver.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(11, "ns", "orders.us", b"hello-us"),
    )
    .await
    .unwrap();

    let msg2 = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive message from existing topic")
        .expect("channel open");
    assert_eq!(msg2.envelope.payload, b"hello-us");
}

/// Auto-subscribe respects the `max_topics_per_wildcard` limit.
#[tokio::test]
async fn wildcard_auto_subscribe_respects_limit() {
    use fafb_q::namespace::DefaultNamespaceManager;

    let engine = make_engine();
    let mgr = Arc::new(DefaultNamespaceManager::new(
        engine,
        vec![ns()],
        common::test_shared_config(),
        ManagerConfig {
            max_topics_per_wildcard: 2,
            ..Default::default()
        },
    ));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create 2 topics (at the limit).
    for region in &["us", "eu"] {
        let topic = TopicName(format!("orders.{region}"));
        mgr.publish(&ns(), &topic, make_envelope(1, "ns", &topic.0, b"init"))
            .await
            .unwrap();
    }

    // Subscribe — 2 matches, at the limit.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 2);
    let mut rx = sub.rx;

    // Create a 3rd topic — should be skipped due to limit.
    mgr.publish(
        &ns(),
        &TopicName("orders.jp".into()),
        make_envelope(10, "ns", "orders.jp", b"hello-jp"),
    )
    .await
    .unwrap();

    // Should NOT receive the message from orders.jp (over limit).
    let result = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await;
    assert!(
        result.is_err(),
        "should not receive message from topic exceeding max_topics_per_wildcard"
    );
}

/// Publishing to an existing topic does not trigger wildcard auto-subscribe
/// (hot path optimization — trie lookup only on new topic creation).
#[tokio::test]
async fn wildcard_existing_topic_publish_no_auto_subscribe() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init"),
    )
    .await
    .unwrap();

    // Subscribe to wildcard — matches existing topic.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 1);
    let mut rx = sub.rx;

    // Publish again to the same existing topic — this should NOT trigger
    // the auto-subscribe path (topic_created == false).
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(2, "ns", "orders.us", b"hello-again"),
    )
    .await
    .unwrap();

    // Should still receive the message via the existing subscription.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive message from existing subscription")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"hello-again");
}

/// Subscribe to `events.>` and verify auto-subscribe works for deeply
/// nested new topics.
#[tokio::test]
async fn wildcard_gt_auto_subscribe_nested_topics() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Subscribe to `events.>` — no topics exist.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("events.>".into()),
            client,
            None,
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 0);
    let mut rx = sub.rx;

    // Publish to a deeply nested topic.
    mgr.publish(
        &ns(),
        &TopicName("events.user.created".into()),
        make_envelope(1, "ns", "events.user.created", b"user-event"),
    )
    .await
    .unwrap();

    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive deeply nested topic via >")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"user-event");

    // Publish to another deeply nested topic.
    mgr.publish(
        &ns(),
        &TopicName("events.order.updated.v2".into()),
        make_envelope(2, "ns", "events.order.updated.v2", b"order-event"),
    )
    .await
    .unwrap();

    let msg2 = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive second nested topic via >")
        .expect("channel open");
    assert_eq!(msg2.envelope.payload, b"order-event");
}

// ── Wildcard consumer group integration tests (Phase 20.5) ──────────

/// Two consumers in a consumer group subscribed to `orders.*` with 2 existing
/// topics. Messages are round-robined across group members per topic.
#[tokio::test]
async fn wildcard_consumer_group_round_robin() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client1 = ClientId(1);
    let client2 = ClientId(2);

    mgr.connect(&ns(), client1).await.unwrap();
    mgr.connect(&ns(), client2).await.unwrap();

    // Pre-create two topics.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init-us"),
    )
    .await
    .unwrap();
    mgr.publish(
        &ns(),
        &TopicName("orders.eu".into()),
        make_envelope(2, "ns", "orders.eu", b"init-eu"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    // Client 1 subscribes to wildcard with group.
    let sub1 = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client1,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub1.matched_count, 2);
    let mut rx1 = sub1.rx;

    // Client 2 joins the same wildcard group.
    let sub2 = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client2,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub2.matched_count, 2);
    let mut rx2 = sub2.rx;

    // Publish messages to both topics.
    for i in 0..4u64 {
        mgr.publish(
            &ns(),
            &TopicName("orders.us".into()),
            make_envelope(10 + i, "ns", "orders.us", format!("us-{i}").as_bytes()),
        )
        .await
        .unwrap();
    }
    for i in 0..4u64 {
        mgr.publish(
            &ns(),
            &TopicName("orders.eu".into()),
            make_envelope(20 + i, "ns", "orders.eu", format!("eu-{i}").as_bytes()),
        )
        .await
        .unwrap();
    }

    // Collect messages from both clients with a timeout.
    let mut c1_msgs = Vec::new();
    let mut c2_msgs = Vec::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        if c1_msgs.len() + c2_msgs.len() >= 8 {
            break;
        }
        tokio::select! {
            msg = rx1.recv() => {
                if let Some(m) = msg {
                    c1_msgs.push(String::from_utf8_lossy(&m.envelope.payload).to_string());
                }
            }
            msg = rx2.recv() => {
                if let Some(m) = msg {
                    c2_msgs.push(String::from_utf8_lossy(&m.envelope.payload).to_string());
                }
            }
            _ = tokio::time::sleep_until(deadline) => {
                break;
            }
        }
    }

    // Both clients should have received some messages (round-robin).
    assert!(
        !c1_msgs.is_empty() && !c2_msgs.is_empty(),
        "both group members should receive messages: c1={}, c2={}",
        c1_msgs.len(),
        c2_msgs.len()
    );
    // Total should be 8 (4 from each topic).
    assert_eq!(
        c1_msgs.len() + c2_msgs.len(),
        8,
        "all messages should be delivered exactly once across group members"
    );
}

/// Wildcard group subscribe → new topic created → verify new topic is
/// automatically added to the group.
#[tokio::test]
async fn wildcard_consumer_group_auto_subscribe_new_topic() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client1 = ClientId(1);

    mgr.connect(&ns(), client1).await.unwrap();

    let group = GroupName("g1".into());

    // Subscribe to wildcard with group — no matching topics yet.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client1,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub.matched_count, 0);
    let mut rx = sub.rx;

    // Publish to a new topic matching the wildcard.
    mgr.publish(
        &ns(),
        &TopicName("orders.apac".into()),
        make_envelope(1, "ns", "orders.apac", b"hello-apac"),
    )
    .await
    .unwrap();

    // The wildcard group subscriber should receive the message via auto-subscribe.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive auto-subscribed group message")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"hello-apac");
}

/// Two consumers join a wildcard group, then a new topic is created.
/// Both members should receive messages from the new topic via group delivery.
#[tokio::test]
async fn wildcard_consumer_group_auto_subscribe_both_members() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client1 = ClientId(1);
    let client2 = ClientId(2);

    mgr.connect(&ns(), client1).await.unwrap();
    mgr.connect(&ns(), client2).await.unwrap();

    let group = GroupName("g1".into());

    // Both clients subscribe to wildcard with group.
    let sub1 = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client1,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx1 = sub1.rx;

    let sub2 = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client2,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx2 = sub2.rx;

    // Publish 4 messages to a new topic.
    for i in 0..4u64 {
        mgr.publish(
            &ns(),
            &TopicName("orders.new".into()),
            make_envelope(i, "ns", "orders.new", format!("msg-{i}").as_bytes()),
        )
        .await
        .unwrap();
    }

    // Collect messages from both clients.
    let mut total = 0;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        if total >= 4 {
            break;
        }
        tokio::select! {
            msg = rx1.recv() => {
                if msg.is_some() { total += 1; }
            }
            msg = rx2.recv() => {
                if msg.is_some() { total += 1; }
            }
            _ = tokio::time::sleep_until(deadline) => {
                break;
            }
        }
    }

    assert_eq!(
        total, 4,
        "all 4 messages should be delivered across group members"
    );
}

/// Consumer group ack routing works correctly with wildcard subscriptions.
/// Messages from different wildcard-matched topics are acked via the correct
/// per-topic group state.
#[tokio::test]
async fn wildcard_consumer_group_ack_routing() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create topics.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init-us"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    // Subscribe with wildcard group.
    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx = sub.rx;

    // Publish a message.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(2, "ns", "orders.us", b"to-ack"),
    )
    .await
    .unwrap();

    // Receive and ack it.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive message")
        .expect("channel open");

    // Ack routes to the correct topic's group state.
    let ack_result = mgr
        .ack(&ns(), &msg.envelope.topic, client, msg.envelope.id)
        .await;
    assert!(
        ack_result.is_ok(),
        "ack should succeed for wildcard group message"
    );
}

/// Unsubscribe from a wildcard group pattern cleans up all per-topic groups.
#[tokio::test]
async fn wildcard_consumer_group_unsubscribe() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx = sub.rx;

    // Unsubscribe.
    mgr.unsubscribe(
        &ns(),
        &TopicName("orders.*".into()),
        client,
        Some(group.clone()),
    )
    .await
    .unwrap();

    // Publish — should not be delivered.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(2, "ns", "orders.us", b"after-unsub"),
    )
    .await
    .unwrap();

    let result = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await;
    assert!(
        result.is_err() || result.unwrap().is_none(),
        "should not receive message after wildcard group unsubscribe"
    );
}

/// Disconnect cleans up wildcard group subscriptions correctly.
#[tokio::test]
async fn wildcard_consumer_group_disconnect_cleanup() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client1 = ClientId(1);
    let client2 = ClientId(2);

    mgr.connect(&ns(), client1).await.unwrap();
    mgr.connect(&ns(), client2).await.unwrap();

    // Pre-create topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    // Both clients join wildcard group.
    let _sub1 = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client1,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();

    let sub2 = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client2,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx2 = sub2.rx;

    // Disconnect client 1 — client 2 should still be in the group.
    mgr.disconnect(&client1).await.unwrap();

    // Publish — client 2 should still receive.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(2, "ns", "orders.us", b"after-disconnect"),
    )
    .await
    .unwrap();

    let msg = tokio::time::timeout(Duration::from_secs(2), rx2.recv())
        .await
        .expect("remaining group member should receive message")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"after-disconnect");
}

/// Nack routing works correctly with wildcard consumer groups.
/// The nack path uses the same `find_client_group()` as ack — verify it
/// succeeds and the message is dead-lettered.
#[tokio::test]
async fn wildcard_consumer_group_nack_routing() {
    use fafb_q::namespace::NackResult;

    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    let sub = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    let mut rx = sub.rx;

    // Publish a message.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(2, "ns", "orders.us", b"to-nack"),
    )
    .await
    .unwrap();

    // Receive and nack it.
    let msg = tokio::time::timeout(Duration::from_secs(2), rx.recv())
        .await
        .expect("should receive message")
        .expect("channel open");

    let nack_result = mgr
        .nack(
            &ns(),
            &msg.envelope.topic,
            client,
            msg.envelope.id,
            "test rejection".into(),
        )
        .await;
    assert!(
        nack_result.is_ok(),
        "nack should succeed for wildcard group message"
    );
    assert_eq!(nack_result.unwrap(), NackResult::DeadLettered);
}

/// Mixed exact + wildcard group on the same topic: both should be in the
/// same consumer group, and auto-subscribe only adds wildcard members.
#[tokio::test]
async fn wildcard_group_mixed_with_exact_group() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client_exact = ClientId(1);
    let client_wildcard = ClientId(2);

    mgr.connect(&ns(), client_exact).await.unwrap();
    mgr.connect(&ns(), client_wildcard).await.unwrap();

    // Pre-create topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    // Client 1: exact-match subscribe to orders.us with group g1.
    let sub_exact = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.us".into()),
            client_exact,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub_exact.matched_count, 0);
    let mut rx_exact = sub_exact.rx;

    // Client 2: wildcard subscribe to orders.* with group g1.
    let sub_wc = mgr
        .subscribe(
            &ns(),
            &TopicName("orders.*".into()),
            client_wildcard,
            Some(group.clone()),
            SubscribeStart::Head,
        )
        .await
        .unwrap();
    assert_eq!(sub_wc.matched_count, 1);
    let mut rx_wc = sub_wc.rx;

    // Publish messages to orders.us — both clients should participate via round-robin.
    for i in 0..4u64 {
        mgr.publish(
            &ns(),
            &TopicName("orders.us".into()),
            make_envelope(10 + i, "ns", "orders.us", format!("msg-{i}").as_bytes()),
        )
        .await
        .unwrap();
    }

    let mut exact_count = 0;
    let mut wc_count = 0;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
    loop {
        if exact_count + wc_count >= 4 {
            break;
        }
        tokio::select! {
            msg = rx_exact.recv() => {
                if msg.is_some() { exact_count += 1; }
            }
            msg = rx_wc.recv() => {
                if msg.is_some() { wc_count += 1; }
            }
            _ = tokio::time::sleep_until(deadline) => {
                break;
            }
        }
    }

    assert_eq!(
        exact_count + wc_count,
        4,
        "all 4 messages should be delivered across group members"
    );
    assert!(
        exact_count > 0 && wc_count > 0,
        "both exact and wildcard members should receive messages: exact={exact_count}, wc={wc_count}"
    );

    // Now publish to a new topic orders.eu — only wildcard client should get it.
    mgr.publish(
        &ns(),
        &TopicName("orders.eu".into()),
        make_envelope(20, "ns", "orders.eu", b"eu-msg"),
    )
    .await
    .unwrap();

    let msg = tokio::time::timeout(Duration::from_secs(2), rx_wc.recv())
        .await
        .expect("wildcard client should receive from new auto-subscribed topic")
        .expect("channel open");
    assert_eq!(msg.envelope.payload, b"eu-msg");

    // Exact client should NOT receive from orders.eu (they only subscribed to orders.us).
    let result = tokio::time::timeout(Duration::from_millis(500), rx_exact.recv()).await;
    assert!(
        result.is_err(),
        "exact-match client should not receive from auto-subscribed topic"
    );
}

/// Unsubscribing a specific topic within a wildcard group returns InvalidOperation.
#[tokio::test]
async fn wildcard_group_unsubscribe_specific_topic_rejected() {
    let engine = make_engine();
    let mgr = Arc::new(make_namespace_manager(engine, vec![ns()]));
    let client = ClientId(1);

    mgr.connect(&ns(), client).await.unwrap();

    // Pre-create topic.
    mgr.publish(
        &ns(),
        &TopicName("orders.us".into()),
        make_envelope(1, "ns", "orders.us", b"init"),
    )
    .await
    .unwrap();

    let group = GroupName("g1".into());

    mgr.subscribe(
        &ns(),
        &TopicName("orders.*".into()),
        client,
        Some(group.clone()),
        SubscribeStart::Head,
    )
    .await
    .unwrap();

    // Try to unsubscribe from a specific topic within the wildcard — should be rejected.
    let result = mgr
        .unsubscribe(
            &ns(),
            &TopicName("orders.us".into()),
            client,
            Some(group.clone()),
        )
        .await;
    assert!(
        result.is_err(),
        "unsubscribing specific topic within wildcard group should fail"
    );
    let err = result.unwrap_err();
    assert!(
        format!("{err:?}").contains("InvalidOperation"),
        "error should be InvalidOperation, got: {err:?}"
    );
}
