//! The future returned by [`Debounce::call`](crate::Debounce::call).

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

use tokio::sync::watch;

use crate::Slot;

/// How a caller reads its result out of the settled window.
///
/// A window settles to a `Vec<T>`. A latest run produces a one element vec, read
/// at index 0. An accumulate run produces one element per caller, read at the
/// caller's index. One discriminant covers both, so there is no dead arm.
#[derive(Clone, Copy)]
enum Pick {
    /// Take element 0, the single shared value.
    Direct,
    /// Take the batch element at this index.
    Indexed(usize),
}

/// Future for one call to a debounced function.
///
/// Resolves to `Result<T, E>` when the window settles. Many of these can wait on
/// the same window. Each gets a clone of the shared result, or its own batch
/// element in accumulate mode.
///
/// The future watches a [`watch`] channel and resolves the first time the slot
/// holds a settled value. A sender dropped before settling leaves the caller
/// pending forever, matching a deferred that never settles.
pub struct Call<T, E> {
    inner: Pin<Box<dyn Future<Output = Result<T, E>> + Send>>,
}

impl<T, E> std::fmt::Debug for Call<T, E> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Call").finish_non_exhaustive()
    }
}

impl<T, E> Call<T, E>
where
    T: Clone + Send + Sync + 'static,
    E: Clone + Send + Sync + 'static,
{
    /// A caller that takes the single shared value.
    pub(crate) fn direct(receiver: watch::Receiver<Slot<T, E>>) -> Self {
        Self::wait(receiver, Pick::Direct)
    }

    /// A caller that takes the batch element at `index`.
    pub(crate) fn indexed(receiver: watch::Receiver<Slot<T, E>>, index: usize) -> Self {
        Self::wait(receiver, Pick::Indexed(index))
    }

    /// Build the await loop: read the slot, wait for a change, read again.
    fn wait(mut receiver: watch::Receiver<Slot<T, E>>, pick: Pick) -> Self {
        Call {
            inner: Box::pin(async move {
                loop {
                    if let Some(settled) = receiver.borrow_and_update().as_ref() {
                        return resolve(settled, pick);
                    }
                    if receiver.changed().await.is_err() {
                        // Sender dropped without settling. Park forever.
                        std::future::pending::<()>().await;
                    }
                }
            }),
        }
    }
}

impl<T, E> Future for Call<T, E> {
    type Output = Result<T, E>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        self.inner.as_mut().poll(cx)
    }
}

/// Resolve a caller's output from a settled slot and its pick strategy.
///
/// The slot holds a `Vec<T>`. Direct reads element 0. Indexed reads element `i`.
fn resolve<T, E>(settled: &Result<Vec<T>, E>, pick: Pick) -> Result<T, E>
where
    T: Clone,
    E: Clone,
{
    match settled {
        Ok(values) => {
            let i = match pick {
                Pick::Direct => 0,
                Pick::Indexed(i) => i,
            };
            Ok(values
                .get(i)
                .cloned()
                .expect("window result missing this caller's index"))
        }
        Err(e) => Err(e.clone()),
    }
}
