1pub mod cursor;
4pub mod link_header;
5pub mod next_link_body;
6pub mod offset;
7pub mod page;
8
9use faucet_core::FaucetError;
10use reqwest::header::HeaderMap;
11use schemars::JsonSchema;
12use serde::{Deserialize, Serialize};
13use serde_json::Value;
14use std::collections::HashMap;
15
16fn default_true() -> bool {
17 true
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)]
23#[serde(rename_all = "snake_case")]
24pub enum RecordCursorTarget {
25 #[default]
27 Query,
28 Body,
30}
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)]
35#[serde(rename_all = "snake_case")]
36pub enum RecordCursorAgg {
37 #[default]
39 Max,
40 Min,
42}
43
44#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
46#[serde(tag = "type")]
47pub enum PaginationStyle {
48 None,
49 Cursor {
50 next_token_path: String,
51 param_name: String,
52 },
53 CursorInBody {
63 next_token_path: String,
64 body_cursor_field: String,
65 },
66 LinkHeader,
67 NextLinkInBody {
72 next_link_path: String,
73 },
74 PageNumber {
75 param_name: String,
76 start_page: usize,
77 page_size: Option<usize>,
78 page_size_param: Option<String>,
79 },
80 Offset {
81 offset_param: String,
82 limit_param: String,
83 limit: usize,
84 total_path: Option<String>,
85 },
86 OffsetInBody {
95 offset_field: String,
96 limit_field: String,
97 limit: usize,
98 #[serde(default = "default_true")]
99 stop_when_short: bool,
100 },
101 RecordFieldCursor {
108 field: String,
110 #[serde(default)]
112 into: RecordCursorTarget,
113 param: String,
115 #[serde(default)]
117 agg: RecordCursorAgg,
118 #[serde(default = "default_true")]
120 stop_when_short: bool,
121 page_size: usize,
123 },
124}
125
126#[derive(Debug, Default)]
128pub struct PaginationState {
129 pub page: usize,
130 pub next_token: Option<String>,
131 pub offset: usize,
132 pub next_link: Option<String>,
133 #[doc(hidden)]
137 pub previous_token: Option<String>,
138 #[doc(hidden)]
142 pub previous_page_fingerprint: Option<u64>,
143 #[doc(hidden)]
149 pub current_page_is_duplicate: bool,
150 #[doc(hidden)]
154 pub record_field_cursor: Option<Value>,
155}
156
157fn body_fingerprint(body: &Value) -> u64 {
160 use std::hash::{Hash, Hasher};
161 let mut h = std::collections::hash_map::DefaultHasher::new();
162 body.to_string().hash(&mut h);
163 h.finish()
164}
165
166pub(crate) fn value_to_param_string(v: &Value) -> String {
169 match v {
170 Value::String(s) => s.clone(),
171 other => other.to_string(),
172 }
173}
174
175fn pick_cursor(agg: RecordCursorAgg, current: Value, candidate: Value) -> Value {
178 let candidate_wins = match (¤t, &candidate) {
179 (Value::Number(a), Value::Number(b)) => {
180 let (a, b) = (
181 a.as_f64().unwrap_or(f64::NAN),
182 b.as_f64().unwrap_or(f64::NAN),
183 );
184 match agg {
185 RecordCursorAgg::Max => b > a,
186 RecordCursorAgg::Min => b < a,
187 }
188 }
189 (Value::String(a), Value::String(b)) => match agg {
190 RecordCursorAgg::Max => b > a,
191 RecordCursorAgg::Min => b < a,
192 },
193 _ => true,
194 };
195 if candidate_wins { candidate } else { current }
196}
197
198impl PaginationStyle {
199 pub fn apply_params(&self, params: &mut HashMap<String, String>, state: &PaginationState) {
200 match self {
201 PaginationStyle::None => {}
202 PaginationStyle::Cursor { param_name, .. } => {
203 cursor::apply_params(params, param_name, &state.next_token);
204 }
205 PaginationStyle::CursorInBody { .. } => {}
207 PaginationStyle::LinkHeader => {}
208 PaginationStyle::NextLinkInBody { .. } => {}
209 PaginationStyle::PageNumber {
210 param_name,
211 start_page,
212 page_size,
213 page_size_param,
214 } => {
215 page::apply_params(
216 params,
217 param_name,
218 *start_page,
219 state.page,
220 *page_size,
221 page_size_param.as_deref(),
222 );
223 }
224 PaginationStyle::Offset {
225 offset_param,
226 limit_param,
227 limit,
228 ..
229 } => {
230 offset::apply_params(params, offset_param, limit_param, state.offset, *limit);
231 }
232 PaginationStyle::OffsetInBody { .. } => {}
234 PaginationStyle::RecordFieldCursor {
236 into: RecordCursorTarget::Query,
237 param,
238 ..
239 } => {
240 if let Some(cursor) = &state.record_field_cursor {
241 params.insert(param.clone(), value_to_param_string(cursor));
242 }
243 }
244 PaginationStyle::RecordFieldCursor { .. } => {}
245 }
246 }
247
248 pub fn advance(
255 &self,
256 body: &Value,
257 headers: &HeaderMap,
258 state: &mut PaginationState,
259 record_count: usize,
260 ) -> Result<bool, FaucetError> {
261 match self {
262 PaginationStyle::None => Ok(false),
263 PaginationStyle::Cursor {
264 next_token_path, ..
265 } => {
266 let has_next = cursor::advance(body, next_token_path, &mut state.next_token)?;
267 if has_next {
268 if state.next_token == state.previous_token {
269 tracing::warn!(
270 "pagination loop detected: cursor {:?} repeated — stopping",
271 state.next_token
272 );
273 return Ok(false);
274 }
275 state.previous_token = state.next_token.clone();
276 }
277 Ok(has_next)
278 }
279 PaginationStyle::CursorInBody {
280 next_token_path, ..
281 } => {
282 let has_next = cursor::advance(body, next_token_path, &mut state.next_token)?;
285 if has_next {
286 if state.next_token == state.previous_token {
287 tracing::warn!(
288 "pagination loop detected: body cursor {:?} repeated — stopping",
289 state.next_token
290 );
291 return Ok(false);
292 }
293 state.previous_token = state.next_token.clone();
294 }
295 Ok(has_next)
296 }
297 PaginationStyle::LinkHeader => match link_header::extract_next_link(headers) {
298 Some(link) => {
299 if Some(&link) == state.previous_token.as_ref() {
300 tracing::warn!(
301 "pagination loop detected: link {link:?} repeated — stopping"
302 );
303 state.next_link = None;
304 return Ok(false);
305 }
306 state.previous_token = Some(link.clone());
307 state.next_link = Some(link);
308 Ok(true)
309 }
310 None => {
311 state.next_link = None;
312 Ok(false)
313 }
314 },
315 PaginationStyle::NextLinkInBody { next_link_path } => {
316 let has_next = next_link_body::advance(body, next_link_path, &mut state.next_link)?;
317 if has_next {
318 if state.next_link == state.previous_token {
319 tracing::warn!(
320 "pagination loop detected: next_link {:?} repeated — stopping",
321 state.next_link
322 );
323 return Ok(false);
324 }
325 state.previous_token = state.next_link.clone();
326 }
327 Ok(has_next)
328 }
329 PaginationStyle::PageNumber { .. } => {
330 state.page += 1;
331 if record_count == 0 {
332 return Ok(false);
333 }
334 let fp = body_fingerprint(body);
339 if state.previous_page_fingerprint == Some(fp) {
340 tracing::warn!(
341 "pagination loop detected: PageNumber returned an identical page — stopping"
342 );
343 state.current_page_is_duplicate = true;
346 return Ok(false);
347 }
348 state.previous_page_fingerprint = Some(fp);
349 Ok(true)
350 }
351 PaginationStyle::Offset {
352 limit, total_path, ..
353 } => {
354 let has_next = offset::advance(
355 body,
356 &mut state.offset,
357 record_count,
358 *limit,
359 total_path.as_deref(),
360 )?;
361 if has_next && total_path.is_none() {
377 let fp = body_fingerprint(body);
378 if state.previous_page_fingerprint == Some(fp) {
379 tracing::warn!(
380 "pagination loop detected: Offset returned an identical page \
381 (server likely ignoring the offset parameter) — stopping"
382 );
383 state.current_page_is_duplicate = true;
385 return Ok(false);
386 }
387 state.previous_page_fingerprint = Some(fp);
388 }
389 Ok(has_next)
390 }
391 PaginationStyle::OffsetInBody {
392 limit,
393 stop_when_short,
394 ..
395 } => {
396 if record_count == 0 {
399 return Ok(false);
400 }
401 state.offset += record_count;
402 if *stop_when_short && record_count < *limit {
403 return Ok(false);
404 }
405 let fp = body_fingerprint(body);
408 if state.previous_page_fingerprint == Some(fp) {
409 tracing::warn!(
410 "pagination loop detected: OffsetInBody returned an identical page \
411 (server likely ignoring the body offset) — stopping"
412 );
413 state.current_page_is_duplicate = true;
414 return Ok(false);
415 }
416 state.previous_page_fingerprint = Some(fp);
417 Ok(true)
418 }
419 PaginationStyle::RecordFieldCursor {
420 page_size,
421 stop_when_short,
422 ..
423 } => {
424 if record_count == 0 {
428 return Ok(false);
429 }
430 if *stop_when_short && record_count < *page_size {
431 return Ok(false);
432 }
433 let cursor = state
436 .record_field_cursor
437 .as_ref()
438 .map(value_to_param_string);
439 if cursor.is_some() && cursor == state.previous_token {
440 tracing::warn!(
441 "pagination loop detected: RecordFieldCursor did not advance \
442 (cursor {cursor:?} repeated) — stopping"
443 );
444 return Ok(false);
445 }
446 state.previous_token = cursor;
447 Ok(true)
448 }
449 }
450 }
451
452 pub fn update_record_cursor(&self, records: &[Value], state: &mut PaginationState) {
457 if let PaginationStyle::RecordFieldCursor { field, agg, .. } = self {
458 let page_agg = records
459 .iter()
460 .filter_map(|r| r.get(field).cloned())
461 .reduce(|a, b| pick_cursor(*agg, a, b));
462 if let Some(page_agg) = page_agg {
463 state.record_field_cursor = Some(match state.record_field_cursor.take() {
464 Some(prev) => pick_cursor(*agg, prev, page_agg),
465 None => page_agg,
466 });
467 }
468 }
469 }
470
471 pub fn cursor_path(&self) -> Option<&str> {
475 match self {
476 PaginationStyle::Cursor {
477 next_token_path, ..
478 }
479 | PaginationStyle::CursorInBody {
480 next_token_path, ..
481 } => Some(next_token_path),
482 _ => None,
483 }
484 }
485
486 pub fn body_params(&self, state: &PaginationState) -> Vec<(String, Value)> {
492 match self {
493 PaginationStyle::CursorInBody {
494 body_cursor_field, ..
495 } => state
496 .next_token
497 .as_deref()
498 .map(|tok| vec![(body_cursor_field.clone(), Value::String(tok.to_owned()))])
499 .unwrap_or_default(),
500 PaginationStyle::OffsetInBody {
501 offset_field,
502 limit_field,
503 limit,
504 ..
505 } => vec![
506 (offset_field.clone(), Value::from(state.offset as u64)),
507 (limit_field.clone(), Value::from(*limit as u64)),
508 ],
509 PaginationStyle::RecordFieldCursor {
510 into: RecordCursorTarget::Body,
511 param,
512 ..
513 } => state
514 .record_field_cursor
515 .clone()
516 .map(|c| vec![(param.clone(), c)])
517 .unwrap_or_default(),
518 _ => Vec::new(),
519 }
520 }
521
522 pub fn body_cursor<'a>(&'a self, state: &'a PaginationState) -> Option<(&'a str, &'a str)> {
528 match self {
529 PaginationStyle::CursorInBody {
530 body_cursor_field, ..
531 } => state
532 .next_token
533 .as_deref()
534 .map(|tok| (body_cursor_field.as_str(), tok)),
535 _ => None,
536 }
537 }
538}
539
540#[cfg(test)]
541mod new_style_tests {
542 use super::*;
543 use reqwest::header::HeaderMap;
544 use serde_json::json;
545
546 fn offset_in_body() -> PaginationStyle {
547 PaginationStyle::OffsetInBody {
548 offset_field: "offset".into(),
549 limit_field: "limit".into(),
550 limit: 2,
551 stop_when_short: true,
552 }
553 }
554
555 #[test]
556 fn offset_in_body_writes_offset_and_limit_and_advances() {
557 let style = offset_in_body();
558 let mut state = PaginationState::default();
559
560 let bp = style.body_params(&state);
562 assert_eq!(
563 bp,
564 vec![("offset".into(), json!(0)), ("limit".into(), json!(2))]
565 );
566 let mut params = HashMap::new();
568 style.apply_params(&mut params, &state);
569 assert!(params.is_empty());
570
571 let body = json!([{"id": 1}, {"id": 2}]);
573 assert!(
574 style
575 .advance(&body, &HeaderMap::new(), &mut state, 2)
576 .unwrap()
577 );
578 assert_eq!(state.offset, 2);
579 let bp = style.body_params(&state);
580 assert_eq!(bp[0], ("offset".into(), json!(2)));
581
582 let body2 = json!([{"id": 3}]);
584 assert!(
585 !style
586 .advance(&body2, &HeaderMap::new(), &mut state, 1)
587 .unwrap()
588 );
589 assert_eq!(state.offset, 3);
590 }
591
592 #[test]
593 fn offset_in_body_zero_records_stops() {
594 let style = offset_in_body();
595 let mut state = PaginationState::default();
596 assert!(
597 !style
598 .advance(&json!([]), &HeaderMap::new(), &mut state, 0)
599 .unwrap()
600 );
601 }
602
603 #[test]
604 fn offset_in_body_stagnation_guard_stops_when_short_disabled() {
605 let style = PaginationStyle::OffsetInBody {
606 offset_field: "o".into(),
607 limit_field: "l".into(),
608 limit: 2,
609 stop_when_short: false,
610 };
611 let mut state = PaginationState::default();
612 let body = json!([{"id": 1}, {"id": 2}]);
613 assert!(
615 style
616 .advance(&body, &HeaderMap::new(), &mut state, 2)
617 .unwrap()
618 );
619 assert!(
621 !style
622 .advance(&body, &HeaderMap::new(), &mut state, 2)
623 .unwrap()
624 );
625 assert!(state.current_page_is_duplicate);
626 }
627
628 fn keyset(into: RecordCursorTarget) -> PaginationStyle {
629 PaginationStyle::RecordFieldCursor {
630 field: "JournalNumber".into(),
631 into,
632 param: "offset".into(),
633 agg: RecordCursorAgg::Max,
634 stop_when_short: true,
635 page_size: 2,
636 }
637 }
638
639 #[test]
640 fn record_field_cursor_computes_max_and_injects_query() {
641 let style = keyset(RecordCursorTarget::Query);
642 let mut state = PaginationState::default();
643
644 let mut params = HashMap::new();
646 style.apply_params(&mut params, &state);
647 assert!(!params.contains_key("offset"));
648
649 let page = vec![json!({"JournalNumber": 10}), json!({"JournalNumber": 25})];
650 style.update_record_cursor(&page, &mut state);
651 assert_eq!(state.record_field_cursor, Some(json!(25)));
652
653 assert!(
655 style
656 .advance(&json!({}), &HeaderMap::new(), &mut state, 2)
657 .unwrap()
658 );
659 let mut params = HashMap::new();
661 style.apply_params(&mut params, &state);
662 assert_eq!(params.get("offset").unwrap(), "25");
663
664 let page2 = vec![json!({"JournalNumber": 5})];
666 style.update_record_cursor(&page2, &mut state);
667 assert_eq!(state.record_field_cursor, Some(json!(25)));
668 }
669
670 #[test]
671 fn record_field_cursor_into_body() {
672 let style = keyset(RecordCursorTarget::Body);
673 let mut state = PaginationState::default();
674 assert!(style.body_params(&state).is_empty());
675 style.update_record_cursor(&[json!({"JournalNumber": 7})], &mut state);
676 assert_eq!(style.body_params(&state), vec![("offset".into(), json!(7))]);
677 let mut params = HashMap::new();
679 style.apply_params(&mut params, &state);
680 assert!(params.is_empty());
681 }
682
683 #[test]
684 fn record_field_cursor_stops_on_short_page_and_non_advance() {
685 let style = keyset(RecordCursorTarget::Query);
687 let mut state = PaginationState::default();
688 style.update_record_cursor(&[json!({"JournalNumber": 3})], &mut state);
689 assert!(
690 !style
691 .advance(&json!({}), &HeaderMap::new(), &mut state, 1)
692 .unwrap()
693 );
694
695 let mut state = PaginationState::default();
697 let page = vec![json!({"JournalNumber": 9}), json!({"JournalNumber": 9})];
698 style.update_record_cursor(&page, &mut state);
699 assert!(
700 style
701 .advance(&json!({}), &HeaderMap::new(), &mut state, 2)
702 .unwrap()
703 );
704 style.update_record_cursor(&page, &mut state);
706 assert!(
707 !style
708 .advance(&json!({}), &HeaderMap::new(), &mut state, 2)
709 .unwrap()
710 );
711 }
712
713 #[test]
714 fn record_field_cursor_min_agg() {
715 let style = PaginationStyle::RecordFieldCursor {
716 field: "seq".into(),
717 into: RecordCursorTarget::Query,
718 param: "before".into(),
719 agg: RecordCursorAgg::Min,
720 stop_when_short: true,
721 page_size: 2,
722 };
723 let mut state = PaginationState::default();
724 style.update_record_cursor(&[json!({"seq": 10}), json!({"seq": 4})], &mut state);
725 assert_eq!(state.record_field_cursor, Some(json!(4)));
726 style.update_record_cursor(&[json!({"seq": 2})], &mut state);
727 assert_eq!(state.record_field_cursor, Some(json!(2)));
728 }
729
730 #[test]
731 fn cursor_path_only_for_cursor_styles() {
732 assert_eq!(
733 PaginationStyle::Cursor {
734 next_token_path: "$.n".into(),
735 param_name: "c".into(),
736 }
737 .cursor_path(),
738 Some("$.n")
739 );
740 assert_eq!(
741 PaginationStyle::CursorInBody {
742 next_token_path: "$.p.next".into(),
743 body_cursor_field: "after".into(),
744 }
745 .cursor_path(),
746 Some("$.p.next")
747 );
748 assert_eq!(offset_in_body().cursor_path(), None);
749 }
750}