use crate::EcosystemId;
use crate::error::Result;
use bytes::Bytes;
use std::future::Future;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
pub enum ListCoverage {
#[default]
Complete,
Truncated,
}
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct Paginated<T> {
pub items: Vec<T>,
pub coverage: ListCoverage,
}
impl<T> Paginated<T> {
#[must_use]
pub const fn new(items: Vec<T>, coverage: ListCoverage) -> Self {
Self { items, coverage }
}
}
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: EcosystemId,
noun: &str,
name: &str,
page: u32,
page_len: usize,
max_pages: u32,
) {
if page == max_pages && page_has_more(page_len) {
let name = crate::redact::redact_declaration_key(name);
tracing::warn!(
package = %name,
pages_fetched = max_pages,
%ecosystem,
"{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: EcosystemId,
noun: &str,
name: &str,
max_pages: u32,
mut fetch_page: F,
mut parse_page: P,
) -> Result<Paginated<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(Paginated {
items,
coverage: ListCoverage::Complete,
});
}
let mut coverage = ListCoverage::Truncated;
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) {
coverage = ListCoverage::Complete;
break 'batches;
}
warn_if_pagination_truncated(
provider,
ecosystem,
noun,
name,
current_page,
page_len,
max_pages,
);
current_page += 1;
}
page = batch_end + 1;
}
Ok(Paginated { items, coverage })
}
#[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",
EcosystemId::GitlabCi,
"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",
EcosystemId::GitlabCi,
"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",
EcosystemId::GitlabCi,
"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.items.len(), 42);
}
#[tokio::test]
async fn test_paginate_pages_stops_after_partial_page_at_custom_cap() {
let result = paginate_pages(
"GitLab",
EcosystemId::GitlabCi,
"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.items.len(), 110);
assert_eq!(result.coverage, ListCoverage::Complete);
}
#[tokio::test]
async fn test_paginate_pages_single_partial_page_is_complete() {
let result = paginate_pages(
"GitLab",
EcosystemId::GitlabCi,
"tags",
"org/repo",
5,
|_page| async move { Ok(page_json(3)) },
parse_page,
)
.await
.unwrap();
assert_eq!(result.coverage, ListCoverage::Complete);
}
#[tokio::test]
async fn test_paginate_pages_cap_reached_with_full_last_page_is_truncated() {
let result = paginate_pages(
"GitLab",
EcosystemId::GitlabCi,
"tags",
"org/repo",
3,
|_page| async move { Ok(page_json(100)) },
parse_page,
)
.await
.unwrap();
assert_eq!(result.items.len(), 300);
assert_eq!(result.coverage, ListCoverage::Truncated);
}
#[tokio::test]
async fn test_paginate_pages_partial_page_exactly_at_cap_is_complete() {
let result = paginate_pages(
"GitLab",
EcosystemId::GitlabCi,
"tags",
"org/repo",
3,
|page| async move {
match page {
1 | 2 => Ok(page_json(100)),
3 => Ok(page_json(7)),
_ => panic!("page {page} must not be fetched beyond the 3-page cap"),
}
},
parse_page,
)
.await
.unwrap();
assert_eq!(result.items.len(), 207);
assert_eq!(result.coverage, ListCoverage::Complete);
}
#[test]
fn test_list_coverage_default_is_complete() {
assert_eq!(ListCoverage::default(), ListCoverage::Complete);
}
}