//! Edge cases beyond the source suite: default wait, the cold boundary, leading
//! re-arm, instance isolation, batch reset across windows, the fan-out
//! guarantee, and the park-forever path on a panic in the run.

mod common;

use std::convert::Infallible;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use debounce_promise::Debounce;
use tokio::time::timeout;

use common::{sleep, Counter};

#[tokio::test]
async fn default_wait_zero_still_coalesces() {
    // A3: with the default zero wait a synchronous burst still collapses to one
    // trailing run on the next tick, resolving to the last argument.
    let calls = Counter::new();
    let c = calls.clone();
    let debounced = Debounce::latest(move |v: &'static str| {
        let c = c.clone();
        async move {
            c.bump();
            Ok::<_, Infallible>(v)
        }
    })
    .build();

    let a = debounced.call("foo");
    let b = debounced.call("bar");
    let last = debounced.call("qux");

    assert_eq!(a.await, Ok("qux"));
    assert_eq!(b.await, Ok("qux"));
    assert_eq!(last.await, Ok("qux"));
    assert_eq!(calls.get(), 1);
}

#[tokio::test(start_paused = true)]
async fn cold_boundary_uses_strict_greater_than() {
    // A4: a call exactly `wait` after the previous one is still hot. Only a gap
    // strictly greater than `wait` resets to cold and fires a fresh leading run.
    let calls = Counter::new();
    let c = calls.clone();
    let debounced = Debounce::latest(move |_: i32| {
        let c = c.clone();
        async move { Ok::<_, Infallible>(c.bump()) }
    })
    .wait_ms(100)
    .leading(true)
    .build();

    // First call is cold: leading fires.
    let _ = debounced.call(1).await;
    assert_eq!(calls.get(), 1);

    // Advance exactly the wait. Elapsed == wait is not > wait, so still hot. This
    // call joins the trailing window, no new leading.
    tokio::time::advance(Duration::from_millis(100)).await;
    let trailing = debounced.call(2);
    assert_eq!(calls.get(), 1);

    // Drive the trailing window to its flush, which adds one run.
    let _ = trailing.await;
    assert_eq!(calls.get(), 2);

    // Now a gap strictly greater than wait. Cold again, leading fires.
    tokio::time::advance(Duration::from_millis(101)).await;
    let _ = debounced.call(3).await;
    assert_eq!(calls.get(), 3);
}

#[tokio::test]
async fn leading_re_arms_after_an_idle_gap() {
    // A5: a leading call fires, the window passes, a second leading call fires
    // again. Two separate windows, count goes 1 then 2.
    let calls = Counter::new();
    let c = calls.clone();
    let debounced = Debounce::latest(move |_: i32| {
        let c = c.clone();
        async move { Ok::<_, Infallible>(c.bump()) }
    })
    .wait_ms(10)
    .leading(true)
    .build();

    let _ = debounced.call(1).await;
    assert_eq!(calls.get(), 1);

    sleep(30).await;

    let _ = debounced.call(2).await;
    assert_eq!(calls.get(), 2);
}

#[tokio::test]
async fn separate_instances_keep_independent_state() {
    // A6: two wrappers do not share a window. Interleaved calls do not cross.
    let count_a = Arc::new(AtomicUsize::new(0));
    let ca = Arc::clone(&count_a);
    let a = Debounce::latest(move |v: &'static str| {
        let ca = Arc::clone(&ca);
        async move {
            ca.fetch_add(1, Ordering::SeqCst);
            Ok::<_, Infallible>(v)
        }
    })
    .wait_ms(20)
    .build();

    let count_b = Arc::new(AtomicUsize::new(0));
    let cb = Arc::clone(&count_b);
    let b = Debounce::latest(move |v: &'static str| {
        let cb = Arc::clone(&cb);
        async move {
            cb.fetch_add(1, Ordering::SeqCst);
            Ok::<_, Infallible>(v)
        }
    })
    .wait_ms(20)
    .build();

    let a1 = a.call("a1");
    let b1 = b.call("b1");
    let a2 = a.call("a2");
    let b2 = b.call("b2");

    assert_eq!(a1.await, Ok("a2"));
    assert_eq!(a2.await, Ok("a2"));
    assert_eq!(b1.await, Ok("b2"));
    assert_eq!(b2.await, Ok("b2"));

    assert_eq!(count_a.load(Ordering::SeqCst), 1);
    assert_eq!(count_b.load(Ordering::SeqCst), 1);
}

#[tokio::test]
async fn accumulate_batch_resets_across_windows() {
    // A8: two full accumulate windows back to back. The second batch does not
    // carry the first window's arguments.
    let seen: Arc<Mutex<Vec<Vec<i32>>>> = Arc::new(Mutex::new(Vec::new()));
    let s = Arc::clone(&seen);
    let square = Debounce::accumulate(move |batch: Vec<i32>| {
        let s = Arc::clone(&s);
        async move {
            s.lock().unwrap().push(batch.clone());
            Ok::<_, Infallible>(batch.iter().map(|x| x * x).collect::<Vec<_>>())
        }
    })
    .wait_ms(10)
    .build();

    let one = square.call(1);
    let two = square.call(2);
    sleep(20).await;
    assert_eq!(one.await, Ok(1));
    assert_eq!(two.await, Ok(4));

    let three = square.call(3);
    let four = square.call(4);
    sleep(20).await;
    assert_eq!(three.await, Ok(9));
    assert_eq!(four.await, Ok(16));

    assert_eq!(*seen.lock().unwrap(), vec![vec![1, 2], vec![3, 4]]);
}

#[tokio::test]
async fn dropping_one_caller_does_not_cancel_the_shared_run() {
    // The run is driven by a spawned timer, not by any single caller's future.
    // Drop the first caller before the flush. A surviving caller still resolves
    // with the shared result. This guards the fan-out claim in the call rustdoc.
    let d = Debounce::latest(|v: &'static str| async move { Ok::<_, Infallible>(v) })
        .wait_ms(10)
        .build();

    let a = d.call("foo");
    let b = d.call("bar");
    drop(a);

    sleep(20).await;

    assert_eq!(b.await, Ok("bar"));
}

#[tokio::test]
async fn a_panic_in_the_run_leaves_callers_parked_forever() {
    // If the run panics rather than returning Err, the spawned flush task unwinds
    // before it settles the channel. The sender drops and the caller parks. This
    // pins the documented behavior. A change that settled callers on sender drop
    // would fail here and force a docs update.
    //
    // The flush task prints a panic message during this run. That is expected and
    // does not fail the test.
    let d = Debounce::latest(|_: i32| async move {
        panic!("boom in flush");
        #[allow(unreachable_code)]
        Ok::<i32, Infallible>(0)
    })
    .wait_ms(10)
    .build();

    let a = d.call(1);
    sleep(20).await;

    let r = timeout(Duration::from_millis(50), a).await;
    assert!(
        r.is_err(),
        "caller must remain pending after a panic in the run"
    );
}
