//! Accumulate mode: batch the arguments, map results back by index.

mod common;

use std::convert::Infallible;
use std::sync::{Arc, Mutex};

use debounce_promise::Debounce;

use common::sleep;

/// A recorder for the batches the wrapped function receives, in order.
type Batches = Arc<Mutex<Vec<Vec<i32>>>>;

#[tokio::test]
async fn accumulates_arguments() {
    // The source uses [[1],[2],[3]] because each call's arguments form a tuple.
    // With one argument per call the batch is [1, 2, 3]. Each caller gets the
    // square of its own input: 1, 4, 9.
    let seen: Batches = 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);
    let three = square.call(3);

    sleep(20).await;

    assert_eq!(one.await, Ok(1));
    assert_eq!(two.await, Ok(4));
    assert_eq!(three.await, Ok(9));
    assert_eq!(*seen.lock().unwrap(), vec![vec![1, 2, 3]]);
}

#[tokio::test]
async fn accumulate_works_with_leading() {
    // Leading splits the batch: the first call forms [1], the rest form [2, 3].
    let seen: Batches = 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)
    .leading(true)
    .build();

    let one = square.call(1);
    let two = square.call(2);
    let three = square.call(3);

    sleep(20).await;

    assert_eq!(one.await, Ok(1));
    assert_eq!(two.await, Ok(4));
    assert_eq!(three.await, Ok(9));
    assert_eq!(*seen.lock().unwrap(), vec![vec![1], vec![2, 3]]);
}

#[tokio::test]
async fn accumulate_works_with_non_promise_return_value() {
    // Same split as above. The function returns a ready value with no async work.
    let seen: Batches = Arc::new(Mutex::new(Vec::new()));
    let s = Arc::clone(&seen);
    let square = Debounce::accumulate(move |batch: Vec<i32>| {
        let s = Arc::clone(&s);
        let squared: Vec<i32> = batch.iter().map(|x| x * x).collect();
        s.lock().unwrap().push(batch);
        async move { Ok::<_, Infallible>(squared) }
    })
    .wait_ms(10)
    .leading(true)
    .build();

    let one = square.call(1);
    let two = square.call(2);
    let three = square.call(3);

    sleep(20).await;

    assert_eq!(one.await, Ok(1));
    assert_eq!(two.await, Ok(4));
    assert_eq!(three.await, Ok(9));
    assert_eq!(*seen.lock().unwrap(), vec![vec![1], vec![2, 3]]);
}

#[tokio::test]
async fn maps_each_caller_to_its_own_batch_index() {
    // Square hides the index mapping: input equals output at every position, so a
    // swapped or dropped index would still pass. Offset each input by 100 so the
    // output differs from the input. Caller i must read element i, not its own
    // argument. For batch [1, 2, 3] returning [101, 102, 103]:
    //   caller 1 -> Ok(101), caller 2 -> Ok(102), caller 3 -> Ok(103).
    let offset = Debounce::accumulate(move |batch: Vec<i32>| async move {
        Ok::<_, Infallible>(batch.iter().map(|x| x + 100).collect::<Vec<_>>())
    })
    .wait_ms(10)
    .build();

    let one = offset.call(1);
    let two = offset.call(2);
    let three = offset.call(3);

    sleep(20).await;

    assert_eq!(one.await, Ok(101));
    assert_eq!(two.await, Ok(102));
    assert_eq!(three.await, Ok(103));
}

#[tokio::test]
async fn maps_each_caller_to_its_own_index_across_the_leading_split() {
    // Leading splits the batch into [1] and [2, 3]. With the offset shape the
    // leading caller reads element 0 of [101], and the two trailing callers read
    // elements 0 and 1 of [102, 103]. A mapping bug that crossed the split or
    // read the wrong index would surface here.
    let offset = Debounce::accumulate(move |batch: Vec<i32>| async move {
        Ok::<_, Infallible>(batch.iter().map(|x| x + 100).collect::<Vec<_>>())
    })
    .wait_ms(10)
    .leading(true)
    .build();

    let one = offset.call(1);
    let two = offset.call(2);
    let three = offset.call(3);

    sleep(20).await;

    assert_eq!(one.await, Ok(101));
    assert_eq!(two.await, Ok(102));
    assert_eq!(three.await, Ok(103));
}
