//! Debounce an async function so a burst of calls collapses into one run.
//!
//! Many calls made inside a sliding `wait` window share a single run of the
//! wrapped function. Each call returns a future. Every future in the same
//! window resolves to the result of that one shared run.
//!
//! Two modes change which arguments the function sees and how results map back
//! to callers.
//!
//! - [`leading`](Builder::leading): run the function at once on the first call
//!   of a fresh window and return that result directly. Later calls in the same
//!   window join the trailing run.
//! - [`accumulate`](Debounce::accumulate): run the function once with the list
//!   of every argument collected during the window. Each caller receives the
//!   element of the result list at its own position.
//!
//! The window is trailing edge by default. Every call resets the timer, so the
//! function runs `wait` after the last call in a burst. With `leading` set, an
//! extra run fires at the start of each fresh window.
//!
//! # Example
//!
//! ```
//! use debounce_promise::Debounce;
//!
//! # async fn run() {
//! let d = Debounce::latest(|v: &'static str| async move { Ok::<_, ()>(v) })
//!     .wait_ms(100)
//!     .build();
//! let a = d.call("foo");
//! let b = d.call("bar");
//! let c = d.call("qux");
//! // All three share one run with the last call's argument.
//! assert_eq!((a.await, b.await, c.await), (Ok("qux"), Ok("qux"), Ok("qux")));
//! # }
//! ```
//!
//! # Errors
//!
//! If the wrapped function returns `Err`, every caller in that window receives a
//! clone of the same error. The error type must be `Clone`.
//!
//! # Panics
//!
//! [`Debounce::call`] spawns timer tasks, so it must run inside a Tokio runtime.
//! A panic inside the wrapped function leaves that window's callers pending
//! forever. See [`Debounce::call`] for both cases.
//!
//! # Runtime
//!
//! Timers use [`tokio::time`]. Run inside a Tokio runtime with the `time`
//! feature on.

#![forbid(unsafe_code)]
#![warn(missing_docs)]
#![warn(missing_debug_implementations)]

use std::fmt;
use std::future::Future;
use std::sync::{Arc, Mutex};

use tokio::sync::watch;
use tokio::task::AbortHandle;
use tokio::time::{sleep, Instant};

mod builder;
mod call;
mod wait;

pub use builder::Builder;
pub use call::Call;

use wait::Wait;

/// Settled result of one window, or `None` while the run is still in flight.
///
/// The window result is always a `Vec<T>`. A latest run produces a one element
/// vec read by the only caller. An accumulate run produces one element per
/// caller, read by index.
type Slot<T, E> = Option<Result<Vec<T>, E>>;

/// State shared by every call to one debounced function.
///
/// `pending` collects arguments for the open window. `deferred` settles that
/// window and carries the timer generation that owns it. The generation rises on
/// every reschedule, so a stale timer that fires late sees a mismatch and does
/// nothing. `timer` holds the live timer task so the next call can abort it.
struct State<A, T, E> {
    last_call_at: Option<Instant>,
    pending: Vec<A>,
    deferred: Option<Deferred<T, E>>,
    generation: u64,
    timer: Option<AbortHandle>,
}

/// Settle side of one window: the channel sender plus a receiver to clone for
/// each caller, tagged with the generation that owns the pending timer.
struct Deferred<T, E> {
    sender: watch::Sender<Slot<T, E>>,
    receiver: watch::Receiver<Slot<T, E>>,
    generation: u64,
}

impl<A, T, E> State<A, T, E> {
    fn new() -> Self {
        State {
            last_call_at: None,
            pending: Vec::new(),
            deferred: None,
            generation: 0,
            timer: None,
        }
    }
}

/// A debounced wrapper around a function.
///
/// Build one with [`Debounce::latest`] or [`Debounce::accumulate`], set options
/// on the returned [`Builder`], then drive it with [`call`](Debounce::call). The
/// wrapper is cheap to clone. Clones share one window, so calls through any
/// clone coalesce together.
pub struct Debounce<A, T, E> {
    inner: Arc<Inner<A, T, E>>,
}

impl<A, T, E> Clone for Debounce<A, T, E> {
    fn clone(&self) -> Self {
        Debounce {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<A, T, E> fmt::Debug for Debounce<A, T, E> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Debounce")
            .field("wait", &self.inner.wait)
            .field("leading", &self.inner.leading)
            .field("accumulate", &self.inner.mode.is_accumulate())
            .finish_non_exhaustive()
    }
}

/// Shared heart of a debounced function: the user function, the wait policy, the
/// leading flag, and the window state behind a mutex.
struct Inner<A, T, E> {
    mode: Mode<A, T, E>,
    wait: Wait,
    leading: bool,
    state: Mutex<State<A, T, E>>,
}

/// Boxed send future. Hides the concrete future type behind [`Mode`].
type BoxFuture<O> = std::pin::Pin<Box<dyn Future<Output = O> + Send>>;

/// A function that takes one argument, the latest call in the window.
type LatestFn<A, T, E> = Box<dyn Fn(A) -> BoxFuture<Result<T, E>> + Send + Sync>;

/// A function that takes the whole batch and returns one result per argument.
type BatchFn<A, T, E> = Box<dyn Fn(Vec<A>) -> BoxFuture<Result<Vec<T>, E>> + Send + Sync>;

/// Calling convention for the wrapped function.
///
/// The mode is fixed at construction. `Latest` runs with one argument. `Batch`
/// runs with the whole list. Each variant holds only the function shape it uses,
/// so there are no stub methods.
enum Mode<A, T, E> {
    /// Run with the last argument in the window.
    Latest(LatestFn<A, T, E>),
    /// Run with the full batch of arguments.
    Batch(BatchFn<A, T, E>),
}

impl<A, T, E> Mode<A, T, E> {
    /// True when the mode runs the function over the whole batch.
    fn is_accumulate(&self) -> bool {
        matches!(self, Mode::Batch(_))
    }
}

impl<A, T, E> Debounce<A, T, E>
where
    A: Send + 'static,
    T: Clone + Send + Sync + 'static,
    E: Clone + Send + Sync + 'static,
{
    /// Build a debounced wrapper that runs `f` with the latest call's argument.
    ///
    /// `f` takes one argument and returns a future. When the window flushes the
    /// wrapper runs `f` once with the argument from the last call in the window.
    /// Every caller in the window receives a clone of that result.
    ///
    /// The default wait is zero. Set it with [`Builder::wait_ms`],
    /// [`Builder::wait`], or [`Builder::wait_fn`].
    pub fn latest<F, Fut>(f: F) -> Builder<A, T, E>
    where
        F: Fn(A) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<T, E>> + Send + 'static,
    {
        let run: LatestFn<A, T, E> = Box::new(move |arg| Box::pin(f(arg)));
        Builder::from_mode(Mode::Latest(run))
    }

    /// Build a debounced wrapper that runs `f` with the full batch of arguments
    /// collected during the window.
    ///
    /// `f` takes the list of arguments and returns one result per argument. The
    /// caller at position `i` in the batch receives `result[i]`. The batch holds
    /// arguments in call order. With [`leading`](Builder::leading) set, the first
    /// call of a fresh window forms a one element batch and later calls form the
    /// trailing batch.
    ///
    /// # Panics
    ///
    /// A caller panics when its future is awaited if the returned list has no
    /// element at the caller's index.
    pub fn accumulate<F, Fut>(f: F) -> Builder<A, T, E>
    where
        F: Fn(Vec<A>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Result<Vec<T>, E>> + Send + 'static,
    {
        let run: BatchFn<A, T, E> = Box::new(move |args| Box::pin(f(args)));
        Builder::from_mode(Mode::Batch(run))
    }

    /// Call the debounced function with `arg`.
    ///
    /// Returns a future that resolves with this call's share of the window's
    /// result. The future stands on its own and can move to another task.
    /// Dropping it does not cancel the shared run.
    ///
    /// In accumulate mode the future may panic on await. See
    /// [`Debounce::accumulate`].
    ///
    /// # Panics
    ///
    /// Must run inside a Tokio runtime. The window timer and the leading run are
    /// spawned tasks, so the first hot or leading call panics with no reactor
    /// running otherwise.
    ///
    /// A panic inside the wrapped function leaves that window's callers pending
    /// forever. The flush task unwinds before it settles the channel, the sender
    /// drops, and every caller of that window parks. Later windows recover
    /// because the next call opens a fresh deferred.
    pub fn call(&self, arg: A) -> Call<T, E> {
        let now = Instant::now();
        let current_wait = self.inner.wait.get();

        let mut state = self.inner.state.lock().unwrap();

        let is_cold = match state.last_call_at {
            None => true,
            Some(prev) => now.duration_since(prev) > current_wait,
        };
        state.last_call_at = Some(now);

        if is_cold && self.inner.leading {
            drop(state);
            return self.run_leading(arg);
        }

        state.generation += 1;
        let generation = state.generation;
        let receiver = match state.deferred.as_mut() {
            Some(deferred) => {
                // Reuse the open window's channel, retag it for this timer.
                deferred.generation = generation;
                deferred.receiver.clone()
            }
            None => {
                let (sender, receiver) = watch::channel(None);
                state.deferred = Some(Deferred {
                    sender,
                    receiver: receiver.clone(),
                    generation,
                });
                receiver
            }
        };
        let index = state.pending.len();
        state.pending.push(arg);

        // Reset the window timer: drop the prior task so only one timer is live.
        if let Some(prev) = state.timer.take() {
            prev.abort();
        }
        let inner = Arc::clone(&self.inner);
        let handle = tokio::spawn(async move {
            sleep(current_wait).await;
            flush(&inner, generation).await;
        });
        state.timer = Some(handle.abort_handle());

        let accumulate = self.inner.mode.is_accumulate();
        drop(state);

        if accumulate {
            Call::indexed(receiver, index)
        } else {
            Call::direct(receiver)
        }
    }

    /// Run the leading edge call now and wrap its result in a [`Call`].
    fn run_leading(&self, arg: A) -> Call<T, E> {
        let (sender, receiver) = watch::channel(None);
        let inner = Arc::clone(&self.inner);
        let accumulate = inner.mode.is_accumulate();
        tokio::spawn(async move {
            let result = match &inner.mode {
                Mode::Latest(run) => run(arg).await.map(|v| vec![v]),
                Mode::Batch(run) => run(vec![arg]).await,
            };
            let _ = sender.send(Some(result));
        });
        // A leading call owns a one element batch, so its index is zero.
        if accumulate {
            Call::indexed(receiver, 0)
        } else {
            Call::direct(receiver)
        }
    }
}

/// Fire the wrapped function for the window owned by `generation`.
///
/// A late timer from an earlier reschedule sees a generation mismatch and
/// returns without touching state. The owning timer takes the pending arguments
/// and the deferred sender, runs the function, then settles the shared channel.
/// The window resets so the next call opens a fresh one.
async fn flush<A, T, E>(inner: &Arc<Inner<A, T, E>>, generation: u64)
where
    A: Send + 'static,
    T: Clone + Send + Sync + 'static,
    E: Clone + Send + Sync + 'static,
{
    let (args, sender) = {
        let mut state = inner.state.lock().unwrap();
        match state.deferred.as_ref() {
            Some(d) if d.generation == generation => {}
            _ => return,
        }
        let args = std::mem::take(&mut state.pending);
        let deferred = state.deferred.take().expect("owned deferred");
        state.timer = None;
        (args, deferred.sender)
    };

    let result = match &inner.mode {
        Mode::Batch(run) => run(args).await,
        Mode::Latest(run) => {
            let last = args.into_iter().next_back().expect("non-empty window");
            run(last).await.map(|v| vec![v])
        }
    };

    let _ = sender.send(Some(result));
}
