//! Concurrent churn against the hub.
//!
//! The hub deliberately diverges from the SyncKit SSE pruning pattern, and the
//! reason is a race that only appears under contention: read `receiver_count()`,
//! drop the guard, then `remove()`, and a subscriber arriving between those steps
//! is left holding a receiver whose sender has just been dropped from the map.
//! Nothing errors. The subscriber simply never receives anything again.
//!
//! A single-threaded test cannot see that, so the claim is worth nothing without
//! these. Each test here fails against the read-then-remove implementation and
//! passes against `remove_if`.

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use livechat::{ChatEvent, Hub, HubLimits, MessageId, RoomId, UserId};
use uuid::Uuid;

fn event(id: i64) -> ChatEvent {
    ChatEvent::Delete { id: MessageId(id) }
}

/// Subscribe, immediately publish, and require delivery, while other tasks churn
/// the same room's channel in and out of existence.
///
/// This is the orphaned-subscriber failure stated directly: if pruning can race a
/// subscribe, some iteration attaches to a sender that is about to be removed and
/// the publish goes to a fresh channel it is not listening to.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_subscriber_is_never_orphaned_by_a_concurrent_prune() {
    let hub = Hub::new(HubLimits {
        max_connections_per_user: 512,
        max_connections_total: 4096,
        room_buffer: 64,
    });
    let room = RoomId(Uuid::new_v4());
    let missed = Arc::new(AtomicUsize::new(0));

    let mut tasks = Vec::new();
    for _ in 0..16 {
        let hub = hub.clone();
        let missed = missed.clone();
        tasks.push(tokio::spawn(async move {
            for i in 0..200 {
                let mut sub = hub.subscribe(room, UserId(Uuid::new_v4())).unwrap();

                // Publish to our own subscription. Delivery is not optional: we
                // hold a live receiver on this room.
                let delivered = hub.publish(room, event(i));
                if delivered == 0 {
                    missed.fetch_add(1, Ordering::Relaxed);
                    continue;
                }

                match tokio::time::timeout(std::time::Duration::from_secs(5), sub.recv()).await {
                    Ok(Some(_)) => {}
                    _ => {
                        missed.fetch_add(1, Ordering::Relaxed);
                    }
                }
                // Drop here, racing every other task's subscribe.
            }
        }));
    }

    for t in tasks {
        t.await.unwrap();
    }

    assert_eq!(
        missed.load(Ordering::Relaxed),
        0,
        "a live subscriber missed a publish, which means a prune raced a subscribe"
    );
    assert_eq!(hub.connection_count(), 0);
    assert_eq!(hub.room_count(), 0, "every room channel was pruned");
}

/// Connection accounting must survive concurrent acquire and release, including
/// the rejected path. A counter that drifts upward silently shrinks the budget
/// until the process refuses connections it has room for.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn connection_counters_do_not_drift_under_contention() {
    let limits = HubLimits {
        max_connections_per_user: 2,
        max_connections_total: 16,
        room_buffer: 8,
    };
    let hub = Hub::new(limits);
    let rooms: Vec<RoomId> = (0..4).map(|_| RoomId(Uuid::new_v4())).collect();
    let users: Vec<UserId> = (0..4).map(|_| UserId(Uuid::new_v4())).collect();

    let mut tasks = Vec::new();
    for t in 0..16 {
        let hub = hub.clone();
        let room = rooms[t % rooms.len()];
        let user = users[t % users.len()];
        tasks.push(tokio::spawn(async move {
            for _ in 0..300 {
                // Most of these are rejected by one budget or the other. Every
                // rejection must leave both counters exactly as it found them.
                if let Ok(sub) = hub.subscribe(room, user) {
                    tokio::task::yield_now().await;
                    drop(sub);
                }
            }
        }));
    }

    for t in tasks {
        t.await.unwrap();
    }

    assert_eq!(
        hub.connection_count(),
        0,
        "global counter drifted; some path incremented without releasing"
    );
    assert_eq!(hub.room_count(), 0, "room channels leaked");

    // The budget must be fully available again, which it is not if the counter
    // drifted even by one.
    let mut held = Vec::new();
    for i in 0..limits.max_connections_total {
        held.push(
            hub.subscribe(rooms[i % rooms.len()], UserId(Uuid::new_v4()))
                .expect("full budget available after churn"),
        );
    }
    assert_eq!(hub.connection_count(), limits.max_connections_total);
}

/// Publishes to a room must reach every listener while others come and go.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn a_stable_listener_receives_everything_through_churn() {
    let hub = Hub::new(HubLimits {
        max_connections_per_user: 512,
        max_connections_total: 4096,
        room_buffer: 4096,
    });
    let room = RoomId(Uuid::new_v4());

    let mut stable = hub.subscribe(room, UserId(Uuid::new_v4())).unwrap();

    let churn = {
        let hub = hub.clone();
        tokio::spawn(async move {
            for _ in 0..500 {
                let s = hub.subscribe(room, UserId(Uuid::new_v4())).unwrap();
                tokio::task::yield_now().await;
                drop(s);
            }
        })
    };

    let publisher = {
        let hub = hub.clone();
        tokio::spawn(async move {
            for i in 0..500 {
                hub.publish(room, event(i));
                tokio::task::yield_now().await;
            }
        })
    };

    let mut received = 0;
    while received < 500 {
        match tokio::time::timeout(std::time::Duration::from_secs(5), stable.recv()).await {
            Ok(Some(ChatEvent::Gap)) => panic!("buffer was sized to make a gap impossible here"),
            Ok(Some(_)) => received += 1,
            Ok(None) => panic!("stable listener's room was closed out from under it"),
            Err(e) => panic!("stable listener stopped receiving after {received} events: {e:?}"),
        }
    }

    churn.await.unwrap();
    publisher.await.unwrap();
    assert_eq!(received, 500);
}