videosdk-server-sdk 0.1.0

//! Pagination.
//!
//! List endpoints return a [`Page<T>`]. Walk pages by hand with
//! [`Page::next_page`], or consume every item across every page as a
//! [`Stream`](futures_util::Stream) via [`Page::into_stream`].
//!
//! The API is inconsistent about how it shapes list responses — a bare array,
//! a `{pageInfo, data}` envelope, a `{pagination, data}` envelope, or an
//! envelope whose array lives under a different key. [`Page`] normalizes all of
//! them into one [`PageInfo`].
//!
//! Cursors are **synthesized by the client**: an opaque base64url-encoded
//! `{page, perPage}` blob. The server does not issue them.

use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use base64::Engine as _;
use futures_util::Stream;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

use crate::error::{Error, Result};

pub(crate) type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Fetches one raw page for a resolved 1-based page number and page size.
pub(crate) type PageFetcher =
    Arc<dyn Fn(Option<u32>, Option<u32>) -> BoxFuture<'static, Result<Value>> + Send + Sync>;

/// Converts a raw JSON list item into a typed value. Most endpoints decode
/// directly; a few post-process, e.g. surfacing `id` from `callId`.
pub(crate) type ItemMapper<T> = Arc<dyn Fn(Value) -> Result<T> + Send + Sync>;

/// The pagination query parameters common to every list endpoint.
///
/// Resource-specific list params embed this.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ListParams {
    /// The 1-based page number. Defaults server-side to 1.
    pub page: Option<u32>,
    /// Items per page. The server default is 10 or 20, depending on the resource.
    pub per_page: Option<u32>,
    /// An opaque cursor from [`Page::next_cursor`]. Overrides `page` and `per_page`.
    pub cursor: Option<String>,
}

/// Normalized pagination metadata, consistent across every list endpoint.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
#[non_exhaustive]
pub struct PageInfo {
    /// The 1-based number of the current page.
    pub current_page: u32,
    /// The number of items per page.
    pub per_page: u32,
    /// The number of the last page.
    pub last_page: u32,
    /// The total item count across all pages, when the API reports it.
    pub total: u64,
}

/// A page of list results.
///
/// Holds this page's [`data`](Page::data) and a cursor to the next page.
pub struct Page<T> {
    /// The items on this page.
    pub data: Vec<T>,
    page_info: PageInfo,
    next_cursor: Option<String>,
    fetcher: Option<PageFetcher>,
    data_key: &'static str,
    mapper: Option<ItemMapper<T>>,
}

impl<T: fmt::Debug> fmt::Debug for Page<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Page")
            .field("data", &self.data)
            .field("page_info", &self.page_info)
            .field("next_cursor", &self.next_cursor)
            .finish_non_exhaustive()
    }
}

impl<T> Page<T> {
    /// This page's normalized pagination metadata.
    pub fn page_info(&self) -> PageInfo {
        self.page_info
    }

    /// The total item count across all pages, when the API reports it.
    pub fn total(&self) -> u64 {
        self.page_info.total
    }

    /// The opaque cursor for the next page, or `None` if this is the last one.
    pub fn next_cursor(&self) -> Option<&str> {
        self.next_cursor.as_deref()
    }

    /// Whether another page is available.
    pub fn has_next_page(&self) -> bool {
        self.next_cursor.is_some()
    }

    /// Splits this page into its items and the handle needed to fetch the next
    /// page, if there is one.
    ///
    /// The handle deliberately borrows nothing, so it can be held across an
    /// `.await` without requiring `T: Sync`.
    fn split(self) -> (std::vec::IntoIter<T>, Option<NextPage<T>>) {
        let next = match (self.next_cursor, self.fetcher) {
            (Some(cursor), Some(fetcher)) => Some(NextPage {
                cursor,
                fetcher,
                data_key: self.data_key,
                mapper: self.mapper,
            }),
            _ => None,
        };
        (self.data.into_iter(), next)
    }
}

/// Everything needed to fetch the page after a given one.
struct NextPage<T> {
    cursor: String,
    fetcher: PageFetcher,
    data_key: &'static str,
    mapper: Option<ItemMapper<T>>,
}

impl<T: DeserializeOwned> NextPage<T> {
    async fn fetch(self) -> Result<Page<T>> {
        let params = ListParams {
            cursor: Some(self.cursor),
            ..Default::default()
        };
        paginate(self.fetcher, &params, self.data_key, self.mapper).await
    }
}

impl<T: DeserializeOwned + Send + 'static> Page<T> {
    /// Fetches the next page, or returns `Ok(None)` if there is none.
    pub async fn next_page(&self) -> Result<Option<Page<T>>> {
        let (Some(cursor), Some(fetcher)) = (self.next_cursor.clone(), self.fetcher.clone()) else {
            return Ok(None);
        };
        NextPage {
            cursor,
            fetcher,
            data_key: self.data_key,
            mapper: self.mapper.clone(),
        }
        .fetch()
        .await
        .map(Some)
    }

    /// Consumes every item across every page, fetching subsequent pages on demand.
    ///
    /// The stream yields an `Err` at most once, and stops after it.
    ///
    /// ```no_run
    /// # use futures_util::StreamExt;
    /// # async fn f(page: videosdk::Page<String>) -> Result<(), videosdk::Error> {
    /// let mut items = Box::pin(page.into_stream());
    /// while let Some(item) = items.next().await {
    ///     println!("{}", item?);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub fn into_stream(self) -> impl Stream<Item = Result<T>> + Send {
        struct State<T> {
            items: std::vec::IntoIter<T>,
            next: Option<NextPage<T>>,
        }

        let (items, next) = self.split();

        futures_util::stream::unfold(Some(State { items, next }), |state| async move {
            let mut state = state?;
            loop {
                if let Some(item) = state.items.next() {
                    return Some((Ok(item), Some(state)));
                }
                match state.next.take()?.fetch().await {
                    Ok(page) => {
                        let (items, next) = page.split();
                        state = State { items, next };
                    }
                    Err(err) => return Some((Err(err), None)),
                }
            }
        })
    }
}

/// Performs one list request and wraps it in a [`Page`] that can fetch
/// subsequent pages. `data_key` names the response array field, usually `data`.
pub(crate) async fn paginate<T: DeserializeOwned>(
    fetcher: PageFetcher,
    params: &ListParams,
    data_key: &'static str,
    mapper: Option<ItemMapper<T>>,
) -> Result<Page<T>> {
    let (page, per_page) = resolve_page_params(params);
    let raw = fetcher(page, per_page).await?;
    let (items, page_info) = normalize_raw_page(&raw, data_key);

    let mut data = Vec::with_capacity(items.len());
    for item in items {
        data.push(match &mapper {
            Some(map) => map(item)?,
            None => serde_json::from_value(item)
                .map_err(|e| Error::decode(format!("list item in `{data_key}`"), e))?,
        });
    }

    // A next page exists only when the server says so *and* this page had items.
    let has_next = page_info.current_page < page_info.last_page && !data.is_empty();
    let next_cursor = has_next.then(|| {
        encode_cursor(&CursorState {
            page: page_info.current_page + 1,
            per_page: page_info.per_page,
        })
    });

    Ok(Page {
        data,
        page_info,
        next_cursor,
        fetcher: Some(fetcher),
        data_key,
        mapper,
    })
}

/// Drives a list endpoint into a stream over every item of every page.
///
/// This is what each resource's `list_stream` returns. An error from the very
/// first fetch is surfaced through the stream rather than eagerly, so the
/// method itself needs no `async`.
pub(crate) fn auto_page<T: DeserializeOwned + Send + 'static>(
    fetcher: PageFetcher,
    params: ListParams,
    data_key: &'static str,
    mapper: Option<ItemMapper<T>>,
) -> impl Stream<Item = Result<T>> + Send {
    use futures_util::StreamExt as _;

    futures_util::stream::once(async move { paginate(fetcher, &params, data_key, mapper).await })
        .flat_map(|result| match result {
            Ok(page) => page.into_stream().left_stream(),
            Err(err) => futures_util::stream::once(async move { Err(err) }).right_stream(),
        })
}

/// Wraps an already-materialized slice in a single, complete page. Used by
/// endpoints that return a flat array, or that drain every page eagerly.
pub(crate) fn static_page<T>(data: Vec<T>) -> Page<T> {
    let n = data.len();
    Page {
        data,
        page_info: PageInfo {
            current_page: 1,
            per_page: n as u32,
            last_page: 1,
            total: n as u64,
        },
        next_cursor: None,
        fetcher: None,
        data_key: "data",
        mapper: None,
    }
}

/* ------------------------------ normalization ------------------------------ */

fn num_or(map: &Map<String, Value>, key: &str, default: u64) -> u64 {
    map.get(key)
        .and_then(Value::as_f64)
        .filter(|n| *n >= 0.0)
        .map(|n| n as u64)
        .unwrap_or(default)
}

fn single_page_info(n: usize) -> PageInfo {
    PageInfo {
        current_page: 1,
        per_page: n as u32,
        last_page: 1,
        total: n as u64,
    }
}

/// Extracts the item array and pagination metadata from a raw list response.
///
/// Handles the standard `{pageInfo, data}` envelope (whose fields may be keyed
/// either `currentPage`/`lastPage` or `page`/`totalPages`), the
/// `{pagination, data}` variant, alternate data keys, and bare arrays.
pub(crate) fn normalize_raw_page(raw: &Value, data_key: &str) -> (Vec<Value>, PageInfo) {
    let obj = match raw {
        Value::Null => return (Vec::new(), single_page_info(0)),
        Value::Array(items) => {
            let items = items.clone();
            let info = single_page_info(items.len());
            return (items, info);
        }
        Value::Object(obj) => obj,
        _ => return (Vec::new(), single_page_info(0)),
    };

    // A present-but-non-array `data_key` does not fall back to `data`.
    let items: Vec<Value> = match obj.get(data_key).or_else(|| obj.get("data")) {
        Some(Value::Array(items)) => items.clone(),
        _ => Vec::new(),
    };
    let n = items.len();

    if let Some(Value::Object(info)) = obj.get("pageInfo") {
        // Some endpoints, e.g. agent console logs, key this `page`/`totalPages`.
        let page_info = PageInfo {
            current_page: num_or(info, "currentPage", num_or(info, "page", 1)) as u32,
            per_page: num_or(info, "perPage", n as u64) as u32,
            last_page: num_or(info, "lastPage", num_or(info, "totalPages", 1)) as u32,
            total: num_or(info, "total", n as u64),
        };
        return (items, page_info);
    }

    if let Some(Value::Object(info)) = obj.get("pagination") {
        let page_info = PageInfo {
            current_page: num_or(info, "page", 1) as u32,
            per_page: num_or(info, "perPage", n as u64) as u32,
            last_page: num_or(info, "totalPages", 1) as u32,
            total: num_or(info, "total", n as u64),
        };
        return (items, page_info);
    }

    (items, single_page_info(n))
}

/* --------------------------------- cursors --------------------------------- */

fn is_zero(n: &u32) -> bool {
    *n == 0
}

/// The decoded contents of a cursor. Serialized shape must stay wire-compatible
/// with the TypeScript and Go SDKs.
#[derive(Debug, Serialize, Deserialize)]
struct CursorState {
    page: u32,
    #[serde(rename = "perPage", default, skip_serializing_if = "is_zero")]
    per_page: u32,
}

fn encode_cursor(state: &CursorState) -> String {
    let json = serde_json::to_vec(state).unwrap_or_else(|_| b"{\"page\":1}".to_vec());
    URL_SAFE_NO_PAD.encode(json)
}

fn decode_cursor(cursor: &str) -> CursorState {
    let fallback = CursorState {
        page: 1,
        per_page: 0,
    };
    let Ok(bytes) = URL_SAFE_NO_PAD.decode(cursor) else {
        return fallback;
    };
    match serde_json::from_slice::<CursorState>(&bytes) {
        Ok(state) if state.page >= 1 => state,
        _ => fallback,
    }
}

fn resolve_page_params(params: &ListParams) -> (Option<u32>, Option<u32>) {
    if let Some(cursor) = params.cursor.as_deref().filter(|c| !c.is_empty()) {
        let state = decode_cursor(cursor);
        let per_page = (state.per_page > 0).then_some(state.per_page);
        return (Some(state.page), per_page);
    }
    (params.page, params.per_page)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn normalizes_a_bare_array() {
        let (items, info) = normalize_raw_page(&json!([{"id": 1}, {"id": 2}]), "data");
        assert_eq!(items.len(), 2);
        assert_eq!(info, single_page_info(2));
    }

    #[test]
    fn normalizes_a_page_info_envelope() {
        let raw = json!({
            "pageInfo": {"currentPage": 2, "perPage": 10, "lastPage": 5, "total": 47},
            "data": [{"id": 1}],
        });
        let (items, info) = normalize_raw_page(&raw, "data");
        assert_eq!(items.len(), 1);
        assert_eq!(
            info,
            PageInfo {
                current_page: 2,
                per_page: 10,
                last_page: 5,
                total: 47
            }
        );
    }

    #[test]
    fn page_info_accepts_the_page_and_total_pages_aliases() {
        // The agent console-logs endpoint keys pageInfo this way.
        let raw = json!({"pageInfo": {"page": 3, "totalPages": 9}, "data": [{"id": 1}]});
        let (_, info) = normalize_raw_page(&raw, "data");
        assert_eq!(info.current_page, 3);
        assert_eq!(info.last_page, 9);
        assert_eq!(info.per_page, 1, "falls back to the item count");
        assert_eq!(info.total, 1);
    }

    #[test]
    fn normalizes_a_pagination_envelope() {
        let raw = json!({
            "pagination": {"page": 1, "perPage": 20, "totalPages": 3, "total": 55},
            "data": [{"id": 1}],
        });
        let (_, info) = normalize_raw_page(&raw, "data");
        assert_eq!(
            info,
            PageInfo {
                current_page: 1,
                per_page: 20,
                last_page: 3,
                total: 55
            }
        );
    }

    #[test]
    fn honors_an_alternate_data_key() {
        let raw = json!({"recordings": [{"id": 1}, {"id": 2}]});
        let (items, info) = normalize_raw_page(&raw, "recordings");
        assert_eq!(items.len(), 2);
        assert_eq!(info.total, 2);
    }

    #[test]
    fn falls_back_to_data_when_the_alternate_key_is_absent() {
        let raw = json!({"data": [{"id": 1}]});
        let (items, _) = normalize_raw_page(&raw, "recordings");
        assert_eq!(items.len(), 1);
    }

    #[test]
    fn envelope_without_metadata_is_a_single_page() {
        let (items, info) = normalize_raw_page(&json!({"data": [{"id": 1}]}), "data");
        assert_eq!(items.len(), 1);
        assert_eq!(info, single_page_info(1));
    }

    #[test]
    fn tolerates_null_and_unexpected_shapes() {
        assert_eq!(normalize_raw_page(&Value::Null, "data").0.len(), 0);
        assert_eq!(normalize_raw_page(&json!(42), "data").0.len(), 0);
        assert_eq!(
            normalize_raw_page(&json!({"data": "nope"}), "data").0.len(),
            0
        );
    }

    #[test]
    fn cursor_round_trips() {
        let encoded = encode_cursor(&CursorState {
            page: 4,
            per_page: 25,
        });
        let decoded = decode_cursor(&encoded);
        assert_eq!(decoded.page, 4);
        assert_eq!(decoded.per_page, 25);
    }

    #[test]
    fn cursor_omits_a_zero_per_page() {
        let encoded = encode_cursor(&CursorState {
            page: 2,
            per_page: 0,
        });
        let json = String::from_utf8(URL_SAFE_NO_PAD.decode(&encoded).unwrap()).unwrap();
        assert_eq!(json, r#"{"page":2}"#);
    }

    #[test]
    fn malformed_cursors_fall_back_to_page_one() {
        assert_eq!(decode_cursor("!!!not base64!!!").page, 1);
        assert_eq!(decode_cursor(&URL_SAFE_NO_PAD.encode("null")).page, 1);
        assert_eq!(
            decode_cursor(&URL_SAFE_NO_PAD.encode(r#"{"page":0}"#)).page,
            1
        );
    }

    #[test]
    fn cursor_overrides_explicit_page_params() {
        let cursor = encode_cursor(&CursorState {
            page: 7,
            per_page: 5,
        });
        let params = ListParams {
            page: Some(1),
            per_page: Some(100),
            cursor: Some(cursor),
        };
        assert_eq!(resolve_page_params(&params), (Some(7), Some(5)));
    }

    #[test]
    fn an_empty_cursor_is_ignored() {
        let params = ListParams {
            page: Some(2),
            per_page: None,
            cursor: Some(String::new()),
        };
        assert_eq!(resolve_page_params(&params), (Some(2), None));
    }

    #[test]
    fn static_page_has_no_next() {
        let page = static_page(vec![1, 2, 3]);
        assert!(!page.has_next_page());
        assert_eq!(page.total(), 3);
        assert_eq!(page.page_info().per_page, 3);
    }
}