//! Close signal race condition tests
//! Tests that closing an announcer properly signals listeners even during active listen operations.

use announcement::{Announcement, ListenError};
#[cfg(feature = "std")]
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc, Barrier,
};
#[cfg(feature = "std")]
use std::thread;
use std::time::Duration;

/// Test: Race condition test: close during blocking listen
///
/// Verifies: Verifies no race condition in close during blocking listen
/// Category: race
#[tokio::test]
async fn test_close_during_active_listen() {
    for iteration in 0..1_000 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let listener = announcement.listener();

        // Spawn task that will listen
        let listen_handle = tokio::spawn(async move { listener.listen().await });

        // Close at varying times
        match iteration % 5 {
            0 => {
                // Close immediately
                announcer.close();
            }
            1 => {
                // Short delay
                tokio::time::sleep(Duration::from_micros(1)).await;
                announcer.close();
            }
            2 => {
                // Medium delay
                tokio::time::sleep(Duration::from_micros(10)).await;
                announcer.close();
            }
            3 => {
                // Drop instead of explicit close
                drop(announcer);
            }
            _ => {
                // Yield and close
                tokio::task::yield_now().await;
                announcer.close();
            }
        }

        // Listener should see Closed error
        let result = listen_handle.await.unwrap();
        assert_eq!(result, Err(ListenError::Closed), "Iteration {}", iteration);
    }
}

#[cfg(feature = "std")]
/// Test: Race condition test: close during blocking listen
///
/// Verifies: Verifies no race condition in close during blocking listen
/// Category: race
#[test]
fn test_close_during_blocking_listen() {
    for iteration in 0..500 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let listener = announcement.listener();

        // Thread that will block
        let handle = thread::spawn(move || listener.listen_blocking());

        // Close at varying times
        match iteration % 4 {
            0 => {
                // Immediate
                announcer.close();
            }
            1 => {
                // Short delay
                thread::sleep(Duration::from_micros(100));
                announcer.close();
            }
            2 => {
                // Drop
                thread::sleep(Duration::from_micros(50));
                drop(announcer);
            }
            _ => {
                // Yield and close
                thread::yield_now();
                announcer.close();
            }
        }

        let result = handle.join().unwrap();
        assert_eq!(result, Err(ListenError::Closed));
    }
}

/// Test: Race condition test: close during timeout blocking
///
/// Verifies: Verifies no race condition in close during timeout blocking
/// Category: race
#[tokio::test]
async fn test_close_with_concurrent_listeners() {
    for _ in 0..500 {
        let (announcer, announcement) = Announcement::<i32>::new();

        let mut handles = Vec::new();

        // Spawn 10 listeners
        for _ in 0..10 {
            let listener = announcement.listener();
            handles.push(tokio::spawn(async move { listener.listen().await }));
        }

        // Give them time to set up
        tokio::time::sleep(Duration::from_micros(10)).await;

        // Close
        announcer.close();

        // All should see Closed
        for handle in handles {
            let result = handle.await.unwrap();
            assert_eq!(result, Err(ListenError::Closed));
        }
    }
}

/// Test: Race condition test: close during timeout blocking
///
/// Verifies: Verifies no race condition in close during timeout blocking
/// Category: race
#[tokio::test]
async fn test_close_in_toctou_window() {
    for iteration in 0..2_000 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let announcement_clone = announcement.clone();

        // Thread that closes at random time
        let close_handle = tokio::spawn(async move {
            // Vary timing to hit different points
            match iteration % 7 {
                0 => {}
                1 => tokio::task::yield_now().await,
                2 => tokio::time::sleep(Duration::from_nanos(1)).await,
                3 => tokio::time::sleep(Duration::from_nanos(10)).await,
                4 => tokio::time::sleep(Duration::from_nanos(100)).await,
                5 => tokio::time::sleep(Duration::from_micros(1)).await,
                _ => tokio::time::sleep(Duration::from_micros(10)).await,
            }
            announcer.close();
        });

        // Task that listens
        let listen_handle = tokio::spawn(async move {
            let listener = announcement_clone.listener();
            listener.listen().await
        });

        close_handle.await.unwrap();
        let result = listen_handle.await.unwrap();

        // Must see either Closed error or timeout (but not hang)
        assert_eq!(result, Err(ListenError::Closed));
    }
}

#[cfg(feature = "std")]
/// Test: Race condition test: close during timeout blocking
///
/// Verifies: Verifies no race condition in close during timeout blocking
/// Category: race
#[test]
fn test_close_during_timeout_blocking() {
    for iteration in 0..500 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let listener = announcement.listener();

        let handle =
            thread::spawn(move || listener.listen_timeout_blocking(Duration::from_secs(2)));

        // Close at varying times
        if iteration % 2 == 0 {
            thread::sleep(Duration::from_micros(50));
        }
        announcer.close();

        let result = handle.join().unwrap();
        assert_eq!(result, Err(ListenError::Closed));
    }
}

/// Test: Race condition test: close with barrier sync
///
/// Verifies: Verifies no race condition in close with barrier sync
/// Category: race
#[tokio::test]
async fn test_rapid_close_listen_cycles() {
    for _ in 0..1_000 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let listener = announcement.listener();

        let listen_handle = tokio::spawn(async move { listener.listen().await });

        // Close immediately
        announcer.close();

        let result = listen_handle.await.unwrap();
        assert_eq!(result, Err(ListenError::Closed));
    }
}

#[cfg(feature = "std")]
/// Test: Race condition test: close with barrier sync
///
/// Verifies: Verifies no race condition in close with barrier sync
/// Category: race
#[test]
fn test_close_with_barrier_sync() {
    for _ in 0..500 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let barrier = Arc::new(Barrier::new(11)); // 10 listeners + closer

        let mut handles = Vec::new();

        // 10 threads that will block
        for _ in 0..10 {
            let listener = announcement.listener();
            let barrier_clone = barrier.clone();

            handles.push(thread::spawn(move || {
                barrier_clone.wait(); // Sync start
                listener.listen_blocking()
            }));
        }

        // Closer thread
        let closer = thread::spawn(move || {
            barrier.wait(); // Sync with listeners
                            // Close immediately after barrier
            announcer.close();
        });

        closer.join().unwrap();

        // All listeners should see Closed
        for handle in handles {
            let result = handle.join().unwrap();
            assert_eq!(result, Err(ListenError::Closed));
        }
    }
}

/// Test: Stress test: close propagation stress
///
/// Verifies: Verifies system handles close propagation stress correctly under stress
/// Category: race
#[tokio::test]
async fn test_close_detection_is_closed() {
    let (announcer, announcement) = Announcement::<i32>::new();
    let listener = announcement.listener();

    assert!(!announcement.is_closed());
    assert!(!listener.is_closed());

    // Close in background
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(10)).await;
        announcer.close();
    });

    // Poll until closed
    let start = tokio::time::Instant::now();
    loop {
        if listener.is_closed() {
            break;
        }
        if start.elapsed() > Duration::from_secs(1) {
            panic!("is_closed() never became true");
        }
        tokio::time::sleep(Duration::from_micros(100)).await;
    }

    // Now listen() should return Closed immediately
    let result = listener.listen().await;
    assert_eq!(result, Err(ListenError::Closed));
}

#[cfg(feature = "std")]
/// Test: Stress test: close propagation stress
///
/// Verifies: Verifies system handles close propagation stress correctly under stress
/// Category: race
#[test]
fn test_close_propagation_stress() {
    let running = Arc::new(AtomicBool::new(true));

    let mut handles = Vec::new();

    // Run for 5 seconds
    for _ in 0..10 {
        let running_clone = running.clone();

        handles.push(thread::spawn(move || {
            let mut count = 0;

            while running_clone.load(Ordering::Relaxed) {
                let (announcer, announcement) = Announcement::<i32>::new();
                let listener = announcement.listener();

                // Spawn blocking thread
                let h = thread::spawn(move || listener.listen_blocking());

                // Close quickly
                thread::yield_now();
                announcer.close();

                let result = h.join().unwrap();
                assert_eq!(result, Err(ListenError::Closed));

                count += 1;
            }

            count
        }));
    }

    thread::sleep(Duration::from_secs(5));
    running.store(false, Ordering::Relaxed);

    let total: usize = handles.into_iter().map(|h| h.join().unwrap()).sum();
    println!("Close propagation stress: {} successful cycles", total);
}

#[tokio::test]
async fn test_close_with_try_listen_checks() {
    for _ in 0..1_000 {
        let (announcer, announcement) = Announcement::<i32>::new();
        let listener = announcement.listener();

        // Task doing rapid try_listen
        let check_handle = tokio::spawn(async move {
            for _ in 0..100 {
                let _ = listener.try_listen();
                tokio::task::yield_now().await;
            }
            listener
        });

        // Close
        tokio::task::yield_now().await;
        announcer.close();

        // Get listener back and verify it sees closed
        let listener = check_handle.await.unwrap();
        assert!(listener.is_closed());
        assert_eq!(listener.try_listen(), None);
    }
}