dregs 1.0.0

//! Pages, and the machinery behind every `list()` and `list_all()`.

use std::collections::VecDeque;
use std::fmt;
use std::ops::Deref;
use std::pin::Pin;

use futures_util::Stream;
use reqwest::Method;
use serde_json::Value;

use crate::client::Client;
use crate::error::Result;
use crate::models::{boolean_or, integer};

/// A stream of every item a `list_all()` call matches, fetching pages as it goes.
///
/// Read it with `StreamExt::next` or `TryStreamExt::try_next` from
/// [`futures`](https://docs.rs/futures) or [`futures-util`](https://docs.rs/futures-util). A page
/// that fails to load ends the stream with that error.
pub type ItemStream<T> = Pin<Box<dyn Stream<Item = Result<T>> + Send + 'static>>;

/// Which way a list is sorted.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Order {
    /// Smallest, earliest, or first in the alphabet first.
    Asc,
    /// Largest, latest, or last in the alphabet first.
    Desc,
}

impl Order {
    /// The name the API uses for the order, `"ASC"` or `"DESC"`.
    pub fn as_str(self) -> &'static str {
        match self {
            Order::Asc => "ASC",
            Order::Desc => "DESC",
        }
    }
}

impl fmt::Display for Order {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(self.as_str())
    }
}

/// One page of a list call.
///
/// Dereferences to a slice of the items on this page, and iterates over them. Pages are numbered
/// from 0. To walk every page, use the namespace's `list_all()` rather than looping on
/// `page_number` yourself.
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub struct Page<T> {
    /// The items on this page.
    pub items: Vec<T>,
    /// This page's number, counting from 0.
    pub page_number: u32,
    /// The most items a page holds.
    pub page_size: u32,
    /// How many pages there are in all.
    pub total_pages: u64,
    /// How many items there are across every page.
    pub total_elements: u64,
    /// Whether this is the first page.
    pub first: bool,
    /// Whether this is the last page.
    pub last: bool,
    /// The page as received.
    pub raw: Value,
}

impl<T> Page<T> {
    /// Whether another page follows this one.
    pub fn has_next(&self) -> bool {
        !self.last && !self.items.is_empty()
    }

    /// The items as a vector.
    pub fn into_vec(self) -> Vec<T> {
        self.items
    }

    pub(crate) fn from_value(raw: Value, parse: fn(Value) -> T) -> Self {
        let items = raw
            .get("content")
            .and_then(Value::as_array)
            .into_iter()
            .flatten()
            .filter(|item| item.is_object())
            .cloned()
            .map(parse)
            .collect();

        Page {
            items,
            page_number: count(&raw, "pageNumber"),
            page_size: count(&raw, "pageSize"),
            total_pages: count(&raw, "totalPages"),
            total_elements: count(&raw, "totalElements"),
            first: boolean_or(&raw, "first", true),
            last: boolean_or(&raw, "last", true),
            raw,
        }
    }
}

impl<T> Deref for Page<T> {
    type Target = [T];

    fn deref(&self) -> &[T] {
        &self.items
    }
}

impl<T> IntoIterator for Page<T> {
    type Item = T;
    type IntoIter = std::vec::IntoIter<T>;

    fn into_iter(self) -> Self::IntoIter {
        self.items.into_iter()
    }
}

impl<'a, T> IntoIterator for &'a Page<T> {
    type Item = &'a T;
    type IntoIter = std::slice::Iter<'a, T>;

    fn into_iter(self) -> Self::IntoIter {
        self.items.iter()
    }
}

/// A count that cannot be negative, reading anything missing or out of range as 0.
fn count<N: TryFrom<i64> + Default>(raw: &Value, key: &str) -> N {
    integer(raw, key)
        .and_then(|value| N::try_from(value).ok())
        .unwrap_or_default()
}

/// The paging options every list call shares.
#[derive(Debug, Clone, Default)]
pub(crate) struct Paging {
    pub(crate) page: u32,
    pub(crate) page_size: Option<u32>,
    pub(crate) sort: Option<String>,
    pub(crate) order: Option<Order>,
}

/// The setters for [`Paging`], shared by every list builder.
macro_rules! paging_setters {
    () => {
        /// The page to return, counting from 0. `list()` returns this page, and `list_all()`
        /// starts walking from it. Defaults to 0.
        pub fn page(mut self, page: u32) -> Self {
            self.paging.page = page;
            self
        }

        /// Items per page, at most 100. Defaults to the API's page size.
        pub fn page_size(mut self, page_size: u32) -> Self {
            self.paging.page_size = Some(page_size);
            self
        }

        /// The field to sort on, by its API name, such as `"lastTrackedAt"`.
        pub fn sort(mut self, sort: impl Into<String>) -> Self {
            self.paging.sort = Some(sort.into());
            self
        }

        /// Which way to sort. Each list has its own default, normally newest first.
        pub fn order(mut self, order: $crate::Order) -> Self {
            self.paging.order = Some(order);
            self
        }
    };
}

pub(crate) use paging_setters;

/// A list call, described once so `list()` and `list_all()` cannot drift.
pub(crate) struct ListCall<T> {
    pub(crate) path: &'static str,
    pub(crate) filters: Vec<(&'static str, String)>,
    pub(crate) paging: Paging,
    pub(crate) parse: fn(Value) -> T,
}

impl<T> ListCall<T> {
    fn query(&self, number: u32) -> Vec<(&'static str, String)> {
        let mut query = self.filters.clone();

        query.push(("pageNumber", number.to_string()));

        if let Some(page_size) = self.paging.page_size {
            query.push(("pageSize", page_size.to_string()));
        }

        if let Some(sort) = &self.paging.sort {
            query.push(("sort", sort.clone()));
        }

        if let Some(order) = self.paging.order {
            query.push(("order", order.as_str().to_owned()));
        }

        query
    }

    async fn fetch(&self, client: &Client, number: u32) -> Result<Page<T>> {
        let payload = client
            .request(Method::GET, self.path, &self.query(number), None)
            .await?;

        Ok(Page::from_value(
            payload.filter(Value::is_object).unwrap_or(Value::Null),
            self.parse,
        ))
    }

    /// Returns the page the call's paging names.
    pub(crate) async fn page(self, client: &Client) -> Result<Page<T>> {
        self.fetch(client, self.paging.page).await
    }
}

impl<T: Send + 'static> ListCall<T> {
    /// Walks every page from the one the call's paging names, stopping after the last page or at
    /// the first empty one.
    pub(crate) fn walk(self, client: &Client) -> ItemStream<T> {
        let state = Walk {
            next: Some(self.paging.page),
            buffer: VecDeque::new(),
            client: client.clone(),
            call: self,
        };

        Box::pin(futures_util::stream::unfold(
            state,
            |mut state| async move {
                loop {
                    if let Some(item) = state.buffer.pop_front() {
                        return Some((Ok(item), state));
                    }

                    let number = state.next?;

                    match state.call.fetch(&state.client, number).await {
                        Ok(page) => {
                            state.next = if page.has_next() {
                                number.checked_add(1)
                            } else {
                                None
                            };
                            state.buffer = page.items.into();
                        }
                        Err(error) => {
                            state.next = None;
                            return Some((Err(error), state));
                        }
                    }
                }
            },
        ))
    }
}

struct Walk<T> {
    next: Option<u32>,
    buffer: VecDeque<T>,
    client: Client,
    call: ListCall<T>,
}

/// Adds `value` to `filters` under `name` when it is set.
pub(crate) fn filter(
    filters: &mut Vec<(&'static str, String)>,
    name: &'static str,
    value: Option<impl ToString>,
) {
    if let Some(value) = value {
        filters.push((name, value.to_string()));
    }
}