use crate::error::Result;
use bytes::Bytes;
use std::future::Future;
const CONCURRENCY: usize = 5;
#[must_use]
pub const fn page_has_more(page_len: usize) -> bool {
page_len >= 100
}
pub fn warn_if_pagination_truncated(
provider: &str,
ecosystem: &str,
noun: &str,
name: &str,
page: u32,
page_len: usize,
max_pages: u32,
) {
if page == max_pages && page_has_more(page_len) {
tracing::warn!(
package = name,
pages_fetched = max_pages,
"{ecosystem} {noun} pagination for '{name}' stopped at the {max_pages}-page cap \
while {provider} reported more pages available; the fetched version list may be \
truncated"
);
}
}
pub async fn paginate_pages<T, F, Fut, P>(
provider: &str,
ecosystem: &str,
noun: &str,
name: &str,
max_pages: u32,
mut fetch_page: F,
mut parse_page: P,
) -> Result<Vec<T>>
where
F: FnMut(u32) -> Fut,
Fut: Future<Output = Result<Bytes>>,
P: FnMut(&Bytes) -> Result<Vec<T>>,
{
use futures::stream::{self, StreamExt};
let mut items = Vec::new();
let first_page = fetch_page(1).await?;
let first_items = parse_page(&first_page)?;
let first_page_len = first_items.len();
items.extend(first_items);
if !page_has_more(first_page_len) {
return Ok(items);
}
let mut page = 2u32;
#[expect(
clippy::cast_possible_truncation,
reason = "CONCURRENCY is a small hardcoded constant (5), never close to u32::MAX"
)]
'batches: while page <= max_pages {
let batch_end = (page + CONCURRENCY as u32 - 1).min(max_pages);
let mut stream = stream::iter(page..=batch_end)
.map(&mut fetch_page)
.buffered(CONCURRENCY);
let mut current_page = page;
while let Some(data) = stream.next().await {
let page_items = parse_page(&data?)?;
let page_len = page_items.len();
items.extend(page_items);
if !page_has_more(page_len) {
break 'batches;
}
warn_if_pagination_truncated(
provider,
ecosystem,
noun,
name,
current_page,
page_len,
max_pages,
);
current_page += 1;
}
page = batch_end + 1;
}
Ok(items)
}
#[cfg(test)]
#[expect(
clippy::cast_possible_truncation,
reason = "#673: fixed test-fixture lengths cast to u32 never approach truncation range"
)]
mod tests {
use super::*;
#[cfg(feature = "test-util")]
use crate::test_util::capture_tracing_output_async;
#[test]
fn test_page_has_more_full_page_continues() {
assert!(page_has_more(100));
}
#[test]
fn test_page_has_more_partial_page_stops() {
assert!(!page_has_more(99));
assert!(!page_has_more(0));
}
#[cfg(feature = "test-util")]
#[tokio::test]
async fn test_warn_if_pagination_truncated_uses_passed_max_pages_not_a_constant() {
let output = capture_tracing_output_async(async {
warn_if_pagination_truncated("GitLab", "GitLab CI", "tags", "org/repo", 3, 100, 3);
})
.await;
assert!(output.contains("org/repo"), "output was: {output}");
assert!(output.contains("GitLab"), "output was: {output}");
assert!(output.contains('3'), "output was: {output}");
let silent = capture_tracing_output_async(async {
warn_if_pagination_truncated("GitLab", "GitLab CI", "tags", "org/repo", 2, 100, 3);
})
.await;
assert!(
silent.is_empty(),
"must not warn below the passed cap: {silent}"
);
}
fn page_json(count: usize) -> Bytes {
let entries: Vec<String> = (0..count).map(|i| format!(r#"{{"n":{i}}}"#)).collect();
Bytes::from(format!("[{}]", entries.join(",")))
}
fn parse_page(data: &Bytes) -> Result<Vec<u32>> {
let value: serde_json::Value = crate::parser::parse_json_checked(data)?;
Ok(value
.as_array()
.map(|arr| (0..arr.len() as u32).collect())
.unwrap_or_default())
}
#[tokio::test]
async fn test_paginate_pages_single_page_fetches_exactly_once() {
use std::sync::atomic::{AtomicU32, Ordering};
let calls = AtomicU32::new(0);
let result = paginate_pages(
"GitLab",
"GitLab CI",
"tags",
"org/repo",
30,
|page| {
calls.fetch_add(1, Ordering::SeqCst);
async move {
match page {
1 => Ok(page_json(42)),
_ => panic!("page {page} must not be fetched"),
}
}
},
parse_page,
)
.await
.unwrap();
assert_eq!(calls.load(Ordering::SeqCst), 1);
assert_eq!(result.len(), 42);
}
#[tokio::test]
async fn test_paginate_pages_stops_after_partial_page_at_custom_cap() {
let result = paginate_pages(
"GitLab",
"GitLab CI",
"tags",
"org/repo",
5,
|page| async move {
match page {
1 => Ok(page_json(100)),
2..=5 => Ok(page_json(10)),
_ => panic!("page {page} must not be fetched beyond the 5-page cap"),
}
},
parse_page,
)
.await
.unwrap();
assert_eq!(result.len(), 110);
}
}