ferrtable 0.0.2

use std::{collections::VecDeque, pin::Pin};

use futures::prelude::*;
use reqwest::RequestBuilder;
use serde::de::DeserializeOwned;

use crate::errors::ExecutionError;

/// An Airtable API request type with cursor-based pagination (in Airtable API
/// parlance, "offset" pagination).
pub(crate) trait PaginatedQuery<T, R>: Clone
where
    T: Clone,
    R: PaginatedResponse<T>,
{
    // TODO: docs
    fn get_offset(&self) -> Option<String>;

    fn set_offset(&mut self, value: Option<String>);

    fn get_req_builder(&self) -> RequestBuilder;
}

pub(crate) trait PaginatedResponse<T>: Clone + DeserializeOwned
where
    T: Clone,
{
    fn get_offset(&self) -> Option<String>;

    fn get_items(&self) -> VecDeque<T>;
}

struct StreamState<Q, T>
where
    Q: Clone,
    T: Clone,
{
    buffered: VecDeque<T>,
    query: Q,
    started: bool,
}

/// Acts similarly to a `?` operator, but for the result stream. Upon an error,
/// it short-circuit returns the error as the final item in the stream.
macro_rules! handle_stream_err {
    ($fallible:expr, state = $state:expr) => {
        match $fallible {
            Ok(value) => value,
            Err(err) => {
                $state.query.set_offset(None);
                return Some((
                    Err(ExecutionError::from(err)),
                    StreamState {
                        buffered: VecDeque::new(),
                        started: true,
                        query: $state.query,
                    },
                ));
            }
        }
    };
}

// This could be brought into PaginatedQuery as a default implementation, but
// that forces that the traits in this module be exposed outside of the crate
// and additionally results in worse client ergonomics overall.
pub(crate) fn execute_paginated<T, R>(
    query: impl PaginatedQuery<T, R>,
) -> Pin<Box<impl Stream<Item = Result<T, ExecutionError>>>>
where
    T: Clone,
    R: PaginatedResponse<T>,
{
    // Stream has to be pinned to the heap so that the closure inside
    // doesn't need to implement Unpin (which I don't think it can).
    Box::pin(futures::stream::unfold(
        StreamState {
            buffered: VecDeque::new(),
            query,
            started: false,
        },
        |mut state| async move {
            if let Some(value) = state.buffered.pop_front() {
                // Iterate through a pre-loaded page.
                return Some((Ok(value), state));
            }
            if state.query.get_offset().is_some() || !state.started {
                // Fetch the next page.
                state.started = true;
                let http_resp = handle_stream_err!(
                    handle_stream_err!(state.query.get_req_builder().send().await, state = state)
                        .error_for_status(),
                    state = state
                );
                let deserialized_resp: R =
                    handle_stream_err!(http_resp.json().await, state = state);
                state.buffered = deserialized_resp.get_items();
                state.query.set_offset(deserialized_resp.get_offset());
                if let Some(value) = state.buffered.pop_front() {
                    // Yield the first item from the newly fetched page.
                    return Some((Ok(value), state));
                }
            }
            // No more items buffered and no subsequent page to fetch.
            None
        },
    ))
}