spiral-sdk 0.1.7

//! Consistent pagination for resource collections.
use std::collections::HashSet;
use std::pin::Pin;
use std::sync::Arc;

use futures::future::BoxFuture;
use futures::{Stream, StreamExt, TryStreamExt, stream};

use crate::Error;
use crate::http::continuation;

/// A lazy stream of collection items, fetching the next page only when needed.
pub type ListStream<'a, T> = Pin<Box<dyn Stream<Item = Result<T, Error>> + Send + 'a>>;

/// One response from a resource collection. Tokens are opaque and retain the request's filters.
#[derive(Debug, Clone, serde::Deserialize)]
pub struct Page<T> {
    /// Items returned by this request.
    pub items: Vec<T>,
    /// Pass to `List::page_token` to resume the same query.
    pub next_page_token: Option<String>,
}

#[derive(Clone, Default)]
pub(crate) struct PageOptions {
    pub size: Option<u32>,
    pub token: Option<String>,
}

type Fetch<'a, T> =
    Arc<dyn Fn(PageOptions) -> BoxFuture<'a, Result<Page<T>, Error>> + Send + Sync + 'a>;

/// A lazy collection request. Use `send` for one page or `stream` to follow every page.
/// Constructing or configuring a request performs no I/O. Each page resolves current credentials.
pub struct List<'a, T> {
    fetch: Fetch<'a, T>,
    options: PageOptions,
}
impl<'a, T: Send + 'a> List<'a, T> {
    pub(crate) fn new(
        fetch: impl Fn(PageOptions) -> BoxFuture<'a, Result<Page<T>, Error>> + Send + Sync + 'a,
    ) -> Self {
        Self {
            fetch: Arc::new(fetch),
            options: PageOptions::default(),
        }
    }
    pub(crate) fn with_options(mut self, options: PageOptions) -> Self {
        self.options = options;
        self
    }
    /// Set the page size. Zero is rejected; the service enforces its upper limit.
    pub fn page_size(mut self, size: u32) -> Self {
        self.options.size = Some(size);
        self
    }
    /// Resume at an opaque token from a previous response to the same query.
    pub fn page_token(mut self, token: impl Into<String>) -> Self {
        self.options.token = Some(token.into());
        self
    }
    /// Fetch exactly one page, preserving an explicit continuation token in the result.
    pub async fn send(self) -> Result<Page<T>, Error> {
        if self.options.size == Some(0) || self.options.token.as_deref() == Some("") {
            return Err(Error::invalid(
                "page size and continuation token must be nonempty",
            ));
        }
        (self.fetch)(self.options).await
    }
    /// Stream items with bounded page buffering, stopping at the first error.
    /// Empty intermediate pages are followed; repeated tokens are rejected.
    pub fn stream(self) -> ListStream<'a, T> {
        let mut seen = HashSet::new();
        if let Some(token) = &self.options.token {
            seen.insert(token.clone());
        }
        stream::try_unfold((Some(self), seen), |(request, mut seen)| async move {
            let Some(request) = request else {
                return Ok::<_, Error>(None);
            };
            let fetch = Arc::clone(&request.fetch);
            let mut options = request.options.clone();
            let page = request.send().await?;
            options.token = continuation(page.next_page_token, &mut seen)?;
            let next = options.token.as_ref().map(|_| Self {
                fetch,
                options: options.clone(),
            });
            Ok(Some((
                stream::iter(page.items.into_iter().map(Ok)),
                (next, seen),
            )))
        })
        .try_flatten()
        .boxed()
    }
}