1use serde::de::DeserializeOwned;
33use serde::Deserialize;
34use tracing::{debug, warn};
35
36use crate::error::Result;
37use crate::ApiClient;
38
39#[derive(Debug, Clone, PartialEq, Eq)]
46pub enum Continuation {
47 Url(String),
50 Token { param: String, value: String },
54}
55
56#[derive(Debug, Clone, Default, PartialEq, Eq)]
58pub struct PageInfo {
59 pub total: Option<u64>,
65 pub truncated: bool,
68 pub next: Option<Continuation>,
70}
71
72pub trait Page: DeserializeOwned {
74 type Item;
75
76 fn into_parts(self) -> (Vec<Self::Item>, Option<Continuation>, Option<u64>);
81}
82
83#[derive(Debug, Clone, Deserialize)]
85pub struct BitbucketPage<T> {
86 pub values: Vec<T>,
92 #[serde(default)]
93 pub next: Option<String>,
94 #[serde(default)]
95 pub size: Option<u64>,
96}
97
98impl<T: DeserializeOwned> Page for BitbucketPage<T> {
99 type Item = T;
100
101 fn into_parts(self) -> (Vec<T>, Option<Continuation>, Option<u64>) {
102 let cursor = self
103 .next
104 .filter(|url| !url.trim().is_empty())
105 .map(Continuation::Url);
106 (self.values, cursor, self.size)
107 }
108}
109
110#[derive(Debug, Clone, Deserialize)]
112pub struct JiraPage<T> {
113 pub issues: Vec<T>,
115 #[serde(default, rename = "nextPageToken")]
116 pub next_page_token: Option<String>,
117 #[serde(default, rename = "isLast")]
118 pub is_last: Option<bool>,
119}
120
121pub const JIRA_PAGE_TOKEN_PARAM: &str = "nextPageToken";
123
124impl<T: DeserializeOwned> Page for JiraPage<T> {
125 type Item = T;
126
127 fn into_parts(self) -> (Vec<T>, Option<Continuation>, Option<u64>) {
128 let finished = self.is_last.unwrap_or(false);
132 let cursor = if finished {
133 None
134 } else {
135 self.next_page_token
136 .filter(|token| !token.trim().is_empty())
137 .map(|value| Continuation::Token {
138 param: JIRA_PAGE_TOKEN_PARAM.to_string(),
139 value,
140 })
141 };
142 (self.issues, cursor, None)
144 }
145}
146
147#[derive(Debug, Clone, Deserialize)]
159pub struct JiraOffsetPage<T> {
160 pub values: Vec<T>,
161 #[serde(default, rename = "startAt")]
162 pub start_at: Option<u64>,
163 #[serde(default, rename = "maxResults")]
164 pub max_results: Option<u64>,
165 #[serde(default)]
166 pub total: Option<u64>,
167 #[serde(default, rename = "isLast")]
168 pub is_last: Option<bool>,
169}
170
171pub const JIRA_START_AT_PARAM: &str = "startAt";
173
174impl<T: DeserializeOwned> Page for JiraOffsetPage<T> {
175 type Item = T;
176
177 fn into_parts(self) -> (Vec<T>, Option<Continuation>, Option<u64>) {
178 let start = self.start_at.unwrap_or(0);
179 let returned = self.values.len() as u64;
180 let next_offset = start + returned;
181
182 let finished = match self.is_last {
186 Some(is_last) => is_last,
187 None => match self.total {
188 Some(total) => next_offset >= total,
189 None => returned == 0,
190 },
191 };
192
193 let cursor = if finished || returned == 0 {
194 None
195 } else {
196 Some(Continuation::Token {
197 param: JIRA_START_AT_PARAM.to_string(),
198 value: next_offset.to_string(),
199 })
200 };
201
202 (self.values, cursor, self.total)
203 }
204}
205
206#[derive(Debug, Clone, Copy)]
208pub struct PageLimits {
209 pub limit: Option<usize>,
211 pub budget: usize,
218}
219
220impl PageLimits {
221 pub const DEFAULT_BUDGET: usize = 50;
222
223 pub fn new(limit: Option<usize>) -> Self {
224 Self {
225 limit,
226 budget: Self::DEFAULT_BUDGET,
227 }
228 }
229
230 pub fn from_cli_limit(limit: usize) -> Self {
237 Self::new(if limit == 0 { None } else { Some(limit) })
238 }
239
240 pub fn with_budget(mut self, budget: usize) -> Self {
241 self.budget = budget;
242 self
243 }
244}
245
246pub async fn fetch_paged<P: Page>(
255 client: &ApiClient,
256 path: &str,
257 limits: PageLimits,
258) -> Result<(Vec<P::Item>, PageInfo)> {
259 let mut items: Vec<P::Item> = Vec::new();
260 let mut info = PageInfo::default();
261 let mut request = path.to_string();
262
263 for attempt in 0..limits.budget.max(1) {
264 debug!(request = %request, attempt, "Fetching page");
265
266 let page: P = client.get(&request).await?;
267 let (page_items, cursor, total) = page.into_parts();
268
269 if total.is_some() {
270 info.total = total;
271 }
272
273 let empty_page = page_items.is_empty();
274 items.extend(page_items);
275
276 if let Some(limit) = limits.limit {
278 if items.len() >= limit {
279 let dropped = items.len() > limit;
285 items.truncate(limit);
286 info.truncated = dropped || cursor.is_some();
287 info.next = cursor;
288 return Ok((items, info));
289 }
290 }
291
292 let Some(cursor) = cursor else {
293 return Ok((items, info));
295 };
296
297 if empty_page {
300 warn!("Stopping pagination: the server returned an empty page with a cursor");
304 info.truncated = true;
305 info.next = Some(cursor);
306 return Ok((items, info));
307 }
308
309 if attempt + 1 >= limits.budget.max(1) {
310 warn!(
311 budget = limits.budget,
312 collected = items.len(),
313 "Stopping pagination: request budget exhausted; the result is incomplete"
314 );
315 info.truncated = true;
316 info.next = Some(cursor);
317 return Ok((items, info));
318 }
319
320 request = match cursor {
321 Continuation::Url(url) => url,
324 Continuation::Token { param, value } => append_query(path, ¶m, &value),
325 };
326 }
327
328 Ok((items, info))
329}
330
331fn append_query(path: &str, key: &str, value: &str) -> String {
337 let (base, query) = match path.split_once('?') {
338 Some((base, query)) => (base, Some(query)),
339 None => (path, None),
340 };
341
342 let mut pairs: Vec<String> = query
343 .map(|q| {
344 q.split('&')
345 .filter(|pair| !pair.is_empty())
346 .filter(|pair| {
347 let name = pair.split('=').next().unwrap_or("");
348 name != key
349 })
350 .map(|pair| pair.to_string())
351 .collect()
352 })
353 .unwrap_or_default();
354
355 pairs.push(format!(
356 "{}={}",
357 encode_query_component(key),
358 encode_query_component(value)
359 ));
360
361 format!("{base}?{}", pairs.join("&"))
362}
363
364fn encode_query_component(value: &str) -> String {
371 let mut out = String::with_capacity(value.len());
372 for byte in value.as_bytes() {
373 match byte {
374 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
375 out.push(*byte as char)
376 }
377 _ => out.push_str(&format!("%{byte:02X}")),
378 }
379 }
380 out
381}
382
383#[cfg(test)]
384mod tests {
385 use super::*;
386 use serde_json::json;
387
388 #[derive(Debug, Deserialize, PartialEq)]
389 struct Item {
390 name: String,
391 }
392
393 fn bb_page(value: serde_json::Value) -> BitbucketPage<Item> {
394 serde_json::from_value(value).unwrap()
395 }
396
397 fn jira_page(value: serde_json::Value) -> JiraPage<Item> {
398 serde_json::from_value(value).unwrap()
399 }
400
401 #[test]
402 fn a_bitbucket_page_yields_its_next_url() {
403 let page = bb_page(json!({
404 "values": [{"name": "a"}],
405 "next": "https://api.bitbucket.org/2.0/x?page=2",
406 "size": 7
407 }));
408 let (items, cursor, total) = page.into_parts();
409 assert_eq!(items.len(), 1);
410 assert_eq!(
411 cursor,
412 Some(Continuation::Url(
413 "https://api.bitbucket.org/2.0/x?page=2".to_string()
414 ))
415 );
416 assert_eq!(total, Some(7));
417 }
418
419 #[test]
420 fn a_bitbucket_page_without_next_is_the_last() {
421 let page = bb_page(json!({"values": [{"name": "a"}]}));
422 assert_eq!(page.into_parts().1, None);
423 }
424
425 #[test]
428 fn a_blank_next_is_not_a_cursor() {
429 let page = bb_page(json!({"values": [], "next": " "}));
430 assert_eq!(page.into_parts().1, None);
431 }
432
433 #[test]
436 fn is_last_overrides_a_trailing_jira_token() {
437 let page = jira_page(json!({
438 "issues": [{"name": "a"}],
439 "nextPageToken": "abc",
440 "isLast": true
441 }));
442 assert_eq!(page.into_parts().1, None);
443 }
444
445 #[test]
446 fn a_jira_token_carries_its_parameter_name() {
447 let page = jira_page(json!({"issues": [], "nextPageToken": "abc"}));
448 assert_eq!(
449 page.into_parts().1,
450 Some(Continuation::Token {
451 param: JIRA_PAGE_TOKEN_PARAM.to_string(),
452 value: "abc".to_string()
453 })
454 );
455 }
456
457 #[test]
460 fn jira_reports_no_total() {
461 let page = jira_page(json!({"issues": [{"name": "a"}]}));
462 assert_eq!(page.into_parts().2, None);
463 }
464
465 use wiremock::matchers::{method, path as path_matcher, query_param};
468 use wiremock::{Mock, MockServer, ResponseTemplate};
469
470 fn client_for(server: &MockServer) -> ApiClient {
471 ApiClient::new(server.uri()).unwrap()
472 }
473
474 #[tokio::test]
477 async fn a_bitbucket_collection_is_followed_to_the_end() {
478 let server = MockServer::start().await;
479 let page_two = format!("{}/items?page=2", server.uri());
480
481 Mock::given(method("GET"))
482 .and(path_matcher("/items"))
483 .and(query_param("page", "2"))
484 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
485 "values": [{"name": "c"}]
486 })))
487 .mount(&server)
488 .await;
489
490 Mock::given(method("GET"))
491 .and(path_matcher("/items"))
492 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
493 "values": [{"name": "a"}, {"name": "b"}],
494 "next": page_two,
495 "size": 3
496 })))
497 .mount(&server)
498 .await;
499
500 let (items, info) = fetch_paged::<BitbucketPage<Item>>(
501 &client_for(&server),
502 "/items",
503 PageLimits::new(None),
504 )
505 .await
506 .unwrap();
507
508 assert_eq!(items.len(), 3, "every page must be collected");
509 assert_eq!(items[2].name, "c");
510 assert!(!info.truncated, "a complete result is not truncated");
511 assert_eq!(info.total, Some(3));
512 }
513
514 #[tokio::test]
517 async fn a_limit_truncates_and_says_so() {
518 let server = MockServer::start().await;
519 let page_two = format!("{}/items?page=2", server.uri());
520
521 Mock::given(method("GET"))
522 .and(path_matcher("/items"))
523 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
524 "values": [{"name": "a"}, {"name": "b"}],
525 "next": page_two
526 })))
527 .mount(&server)
528 .await;
529
530 let (items, info) = fetch_paged::<BitbucketPage<Item>>(
531 &client_for(&server),
532 "/items",
533 PageLimits::new(Some(1)),
534 )
535 .await
536 .unwrap();
537
538 assert_eq!(items.len(), 1);
539 assert!(info.truncated, "a capped result must be labelled truncated");
540 assert!(info.next.is_some(), "and must say where it stopped");
541 }
542
543 #[tokio::test]
545 async fn the_budget_bounds_a_server_that_never_ends() {
546 let server = MockServer::start().await;
547 let forever = format!("{}/items?page=next", server.uri());
548
549 Mock::given(method("GET"))
550 .and(path_matcher("/items"))
551 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
552 "values": [{"name": "x"}],
553 "next": forever
554 })))
555 .mount(&server)
556 .await;
557
558 let (items, info) = fetch_paged::<BitbucketPage<Item>>(
559 &client_for(&server),
560 "/items",
561 PageLimits::new(None).with_budget(3),
562 )
563 .await
564 .unwrap();
565
566 assert_eq!(items.len(), 3, "one item per allowed request");
567 assert!(
568 info.truncated,
569 "budget exhaustion is truncation, not success"
570 );
571 }
572
573 #[tokio::test]
575 async fn an_empty_page_with_a_cursor_stops_the_walk() {
576 let server = MockServer::start().await;
577 let forever = format!("{}/items?page=next", server.uri());
578
579 Mock::given(method("GET"))
580 .and(path_matcher("/items"))
581 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
582 "values": [],
583 "next": forever
584 })))
585 .mount(&server)
586 .await;
587
588 let (items, _) = fetch_paged::<BitbucketPage<Item>>(
589 &client_for(&server),
590 "/items",
591 PageLimits::new(None).with_budget(20),
592 )
593 .await
594 .unwrap();
595
596 assert!(items.is_empty());
597 let requests = server.received_requests().await.unwrap_or_default();
598 assert_eq!(
599 requests.len(),
600 1,
601 "must not keep asking: {}",
602 requests.len()
603 );
604 }
605
606 #[tokio::test]
609 async fn an_offset_paged_endpoint_is_followed_to_the_end() {
610 let server = MockServer::start().await;
611
612 Mock::given(method("GET"))
613 .and(path_matcher("/project/search"))
614 .and(query_param("startAt", "2"))
615 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
616 "values": [{"name": "c"}], "startAt": 2, "total": 3
617 })))
618 .mount(&server)
619 .await;
620
621 Mock::given(method("GET"))
622 .and(path_matcher("/project/search"))
623 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
624 "values": [{"name": "a"}, {"name": "b"}], "startAt": 0, "total": 3
625 })))
626 .mount(&server)
627 .await;
628
629 let (items, info) = fetch_paged::<JiraOffsetPage<Item>>(
630 &client_for(&server),
631 "/project/search?expand=lead",
632 PageLimits::new(None),
633 )
634 .await
635 .unwrap();
636
637 assert_eq!(items.len(), 3, "every page collected");
638 assert!(!info.truncated);
639 assert_eq!(info.total, Some(3));
640
641 for request in server.received_requests().await.unwrap_or_default() {
642 let query = request.url.query().unwrap_or("");
643 assert!(
644 query.matches("startAt").count() <= 1,
645 "offset accumulated: {query}"
646 );
647 assert!(
648 query.contains("expand=lead"),
649 "the original query must survive: {query}"
650 );
651 }
652 }
653
654 #[tokio::test]
657 async fn a_jira_token_never_accumulates() {
658 let server = MockServer::start().await;
659
660 Mock::given(method("GET"))
661 .and(path_matcher("/search"))
662 .and(query_param("nextPageToken", "t2"))
663 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
664 "issues": [{"name": "c"}], "isLast": true
665 })))
666 .mount(&server)
667 .await;
668
669 Mock::given(method("GET"))
670 .and(path_matcher("/search"))
671 .and(query_param("nextPageToken", "t1"))
672 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
673 "issues": [{"name": "b"}], "nextPageToken": "t2", "isLast": false
674 })))
675 .mount(&server)
676 .await;
677
678 Mock::given(method("GET"))
679 .and(path_matcher("/search"))
680 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
681 "issues": [{"name": "a"}], "nextPageToken": "t1", "isLast": false
682 })))
683 .mount(&server)
684 .await;
685
686 let (items, info) = fetch_paged::<JiraPage<Item>>(
687 &client_for(&server),
688 "/search?jql=project%3DX",
689 PageLimits::new(None),
690 )
691 .await
692 .unwrap();
693
694 assert_eq!(items.len(), 3, "all three pages");
695 assert!(!info.truncated);
696
697 for request in server.received_requests().await.unwrap_or_default() {
698 let query = request.url.query().unwrap_or("");
699 assert!(
700 query.matches("nextPageToken").count() <= 1,
701 "token accumulated: {query}"
702 );
703 assert!(
704 query.contains("jql=project"),
705 "the original query must survive: {query}"
706 );
707 }
708 }
709
710 #[tokio::test]
713 async fn a_foreign_next_url_is_refused() {
714 let server = MockServer::start().await;
715
716 Mock::given(method("GET"))
717 .and(path_matcher("/items"))
718 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
719 "values": [{"name": "a"}],
720 "next": "https://evil.example.com/2.0/items?page=2"
721 })))
722 .mount(&server)
723 .await;
724
725 let result = fetch_paged::<BitbucketPage<Item>>(
726 &client_for(&server),
727 "/items",
728 PageLimits::new(None),
729 )
730 .await;
731
732 assert!(
733 result.is_err(),
734 "a cross-origin cursor must not be followed"
735 );
736 }
737
738 #[tokio::test]
742 async fn overshooting_the_limit_on_a_final_page_is_still_truncation() {
743 let server = MockServer::start().await;
744
745 Mock::given(method("GET"))
746 .and(path_matcher("/search"))
747 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
748 "issues": [{"name": "a"}, {"name": "b"}, {"name": "c"}],
749 "isLast": true
750 })))
751 .mount(&server)
752 .await;
753
754 let (items, info) = fetch_paged::<JiraPage<Item>>(
755 &client_for(&server),
756 "/search",
757 PageLimits::new(Some(2)),
758 )
759 .await
760 .unwrap();
761
762 assert_eq!(items.len(), 2);
763 assert!(
764 info.truncated,
765 "a page that overshot the limit dropped rows and must say so"
766 );
767 }
768
769 #[tokio::test]
771 async fn hitting_the_limit_exactly_is_not_truncation() {
772 let server = MockServer::start().await;
773
774 Mock::given(method("GET"))
775 .and(path_matcher("/search"))
776 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
777 "issues": [{"name": "a"}, {"name": "b"}],
778 "isLast": true
779 })))
780 .mount(&server)
781 .await;
782
783 let (items, info) = fetch_paged::<JiraPage<Item>>(
784 &client_for(&server),
785 "/search",
786 PageLimits::new(Some(2)),
787 )
788 .await
789 .unwrap();
790
791 assert_eq!(items.len(), 2);
792 assert!(!info.truncated, "nothing was dropped: {info:?}");
793 }
794
795 #[tokio::test]
798 async fn an_abandoned_walk_is_reported_as_truncated() {
799 let server = MockServer::start().await;
800 let forever = format!("{}/items?page=next", server.uri());
801
802 Mock::given(method("GET"))
803 .and(path_matcher("/items"))
804 .respond_with(ResponseTemplate::new(200).set_body_json(json!({
805 "values": [],
806 "next": forever
807 })))
808 .mount(&server)
809 .await;
810
811 let (_, info) = fetch_paged::<BitbucketPage<Item>>(
812 &client_for(&server),
813 "/items",
814 PageLimits::new(None),
815 )
816 .await
817 .unwrap();
818
819 assert!(info.truncated, "the server said more exists: {info:?}");
820 }
821
822 #[tokio::test]
824 async fn a_body_without_the_items_key_is_an_error() {
825 let server = MockServer::start().await;
826
827 Mock::given(method("GET"))
828 .and(path_matcher("/items"))
829 .respond_with(ResponseTemplate::new(200).set_body_json(json!({"page": 1})))
830 .mount(&server)
831 .await;
832
833 let result = fetch_paged::<BitbucketPage<Item>>(
834 &client_for(&server),
835 "/items",
836 PageLimits::new(None),
837 )
838 .await;
839
840 assert!(
841 result.is_err(),
842 "a missing values key must not read as an empty result"
843 );
844 }
845
846 #[test]
849 fn a_zero_cli_limit_means_no_limit() {
850 assert_eq!(PageLimits::from_cli_limit(0).limit, None);
851 assert_eq!(PageLimits::from_cli_limit(25).limit, Some(25));
852 }
853
854 fn offset_page(value: serde_json::Value) -> JiraOffsetPage<Item> {
855 serde_json::from_value(value).unwrap()
856 }
857
858 #[test]
859 fn an_offset_page_advances_by_what_it_returned() {
860 let page = offset_page(json!({
861 "values": [{"name": "a"}, {"name": "b"}],
862 "startAt": 0, "maxResults": 2, "total": 5
863 }));
864 let (items, cursor, total) = page.into_parts();
865 assert_eq!(items.len(), 2);
866 assert_eq!(
867 cursor,
868 Some(Continuation::Token {
869 param: JIRA_START_AT_PARAM.to_string(),
870 value: "2".to_string()
871 })
872 );
873 assert_eq!(total, Some(5));
874 }
875
876 #[test]
878 fn is_last_ends_an_offset_walk() {
879 let page = offset_page(json!({
880 "values": [{"name": "a"}], "startAt": 0, "total": 99, "isLast": true
881 }));
882 assert_eq!(page.into_parts().1, None);
883 }
884
885 #[test]
886 fn reaching_the_total_ends_an_offset_walk() {
887 let page = offset_page(json!({
888 "values": [{"name": "a"}], "startAt": 4, "total": 5
889 }));
890 assert_eq!(page.into_parts().1, None);
891 }
892
893 #[test]
896 fn an_empty_offset_page_ends_the_walk() {
897 let page = offset_page(json!({"values": [], "startAt": 10}));
898 assert_eq!(page.into_parts().1, None);
899 }
900
901 #[test]
902 fn append_query_adds_a_parameter() {
903 assert_eq!(append_query("/x", "t", "1"), "/x?t=1");
904 assert_eq!(append_query("/x?a=b", "t", "1"), "/x?a=b&t=1");
905 }
906
907 #[test]
909 fn append_query_replaces_rather_than_duplicating() {
910 let once = append_query("/x?a=b", "t", "1");
911 let twice = append_query(&once, "t", "2");
912 assert_eq!(twice, "/x?a=b&t=2");
913 assert_eq!(twice.matches("t=").count(), 1);
914 }
915
916 #[test]
917 fn append_query_encodes_its_value() {
918 assert_eq!(append_query("/x", "t", "a b&c"), "/x?t=a%20b%26c");
919 }
920}