linear-api 0.1.0

//! Cursor pagination: [`Page`]/[`PageInfo`] (tier 1), the lazy [`paginate`]
//! stream (tier 2), and the draining [`collect_all`] helper (tier 3).

use futures::{Future, Stream, TryStreamExt};

/// Relay-style page metadata returned by every Linear connection.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PageInfo {
    /// Whether another page follows this one.
    pub has_next_page: bool,
    /// Whether a page precedes this one.
    pub has_previous_page: bool,
    /// Cursor of the first node in this page.
    pub start_cursor: Option<String>,
    /// Cursor of the last node in this page — pass as `after` to fetch the
    /// next page.
    pub end_cursor: Option<String>,
}

/// One page of results plus its pagination metadata.
#[derive(Debug, Clone)]
pub struct Page<T> {
    /// The items in this page.
    pub nodes: Vec<T>,
    /// Pagination metadata.
    pub page_info: PageInfo,
}

/// Lazily streams every item across pages by repeatedly invoking `fetch` with
/// the previous page's `end_cursor` (`None` on the first call).
///
/// Stops when the server reports `has_next_page == false`, and **also** treats
/// a repeated `end_cursor` as the end of the stream — an infinite-loop guard
/// against a misbehaving (or mocked) server.
///
/// Complexity note: page size multiplies Linear's query complexity (single
/// queries are capped at 10,000 points; the server default page size is 50).
/// Prefer modest `first` values on the underlying request.
pub fn paginate<'a, T, F, Fut>(fetch: F) -> impl Stream<Item = crate::Result<T>> + 'a
where
    T: 'a,
    F: FnMut(Option<String>) -> Fut + 'a,
    Fut: Future<Output = crate::Result<Page<T>>> + 'a,
{
    futures::stream::try_unfold(
        (fetch, None::<String>, false),
        |(mut fetch, cursor, done)| async move {
            if done {
                return Ok(None);
            }
            let page = fetch(cursor.clone()).await?;
            let next_cursor = page.page_info.end_cursor.clone();
            // Repeating the previous cursor would refetch the same page
            // forever; treat it as the end.
            let done = !page.page_info.has_next_page || next_cursor == cursor;
            Ok(Some((page, (fetch, next_cursor, done))))
        },
    )
    .map_ok(|page| futures::stream::iter(page.nodes.into_iter().map(Ok)))
    .try_flatten()
}

/// Drains [`paginate`] into a `Vec`, stopping after `limit` items when given.
///
/// Callers **should** pass a limit: an unbounded drain of a large workspace
/// both burns the request budget and multiplies query complexity (see
/// [`paginate`]).
pub async fn collect_all<'a, T, F, Fut>(fetch: F, limit: Option<usize>) -> crate::Result<Vec<T>>
where
    T: 'a,
    F: FnMut(Option<String>) -> Fut + 'a,
    Fut: Future<Output = crate::Result<Page<T>>> + 'a,
{
    let stream = paginate(fetch);
    futures::pin_mut!(stream);
    let mut items = Vec::new();
    while let Some(item) = stream.try_next().await? {
        items.push(item);
        if limit.is_some_and(|limit| items.len() >= limit) {
            break;
        }
    }
    Ok(items)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn page(nodes: Vec<u32>, end_cursor: Option<&str>, has_next: bool) -> Page<u32> {
        Page {
            nodes,
            page_info: PageInfo {
                has_next_page: has_next,
                has_previous_page: false,
                start_cursor: None,
                end_cursor: end_cursor.map(str::to_owned),
            },
        }
    }

    #[tokio::test]
    async fn paginate_walks_all_pages() {
        let calls = AtomicUsize::new(0);
        let items = collect_all(
            |cursor| {
                calls.fetch_add(1, Ordering::SeqCst);
                async move {
                    Ok(match cursor.as_deref() {
                        None => page(vec![1, 2], Some("a"), true),
                        Some("a") => page(vec![3], Some("b"), false),
                        other => panic!("unexpected cursor {other:?}"),
                    })
                }
            },
            None,
        )
        .await
        .unwrap();
        assert_eq!(items, vec![1, 2, 3]);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn paginate_stops_on_repeated_cursor() {
        let items = collect_all(
            |_cursor| async move { Ok(page(vec![1], Some("stuck"), true)) },
            None,
        )
        .await
        .unwrap();
        // First page (cursor None -> "stuck"), second page repeats "stuck":
        // the guard stops the stream instead of looping forever.
        assert_eq!(items, vec![1, 1]);
    }

    #[tokio::test]
    async fn collect_all_honors_limit() {
        let items = collect_all(
            |_cursor| async move { Ok(page(vec![1, 2, 3], Some("x"), false)) },
            Some(2),
        )
        .await
        .unwrap();
        assert_eq!(items, vec![1, 2]);
    }
}