fizzy_sdk/
pagination.rs

1//! Pages, the cursors between them, and the walks that read them all.
2
3use std::ops::Deref;
4
5use futures_util::Stream;
6use futures_util::stream::{self, StreamExt};
7use serde::de::DeserializeOwned;
8use serde_json::Value;
9use url::Url;
10
11use crate::client::{Client, Response};
12use crate::error::Error;
13use crate::http::Method;
14use crate::observability::OperationInfo;
15use crate::operation::{Operation, RetryPolicy};
16use crate::security::is_same_origin;
17
18/// One page of a paginated read, with the cursor Fizzy handed out for the next one.
19///
20/// The page derefs to its value, so `page.iter()` on a page of boards reads the same as it
21/// would on the `Vec<Board>` itself.
22#[derive(Debug, Clone)]
23pub struct Page<T> {
24    value: T,
25    next_url: Option<Url>,
26    next_cursor: Option<String>,
27    total_count: Option<u64>,
28    /// What the read that produced this page announced itself as, so the reads that walk
29    /// on from it can say the same.
30    info: OperationInfo,
31    /// The retry policy the first page was read under, so every page after it is too.
32    retry: Option<RetryPolicy>,
33}
34
35/// What it takes to read the page after one already handed out: where it is, what the
36/// read announces itself as, and the policy it goes under. Taken off a page so the page
37/// itself can be yielded before the next is fetched.
38#[derive(Debug, Clone)]
39struct Cursor {
40    next_url: Option<Url>,
41    info: OperationInfo,
42    retry: Option<RetryPolicy>,
43}
44
45impl<T> Page<T> {
46    pub(crate) fn new(
47        value: T,
48        response: &Response,
49        info: OperationInfo,
50        retry: Option<RetryPolicy>,
51    ) -> Page<T> {
52        let next_url = response
53            .headers
54            .get("link")
55            .and_then(|value| value.to_str().ok())
56            .and_then(next_link)
57            .and_then(|target| response.url.join(&target).ok());
58        let next_cursor = next_url.as_ref().and_then(|url| {
59            url.query_pairs()
60                .find(|(name, _)| name == "page")
61                .map(|(_, value)| value.into_owned())
62        });
63        let total_count = response
64            .headers
65            .get("x-total-count")
66            .and_then(|value| value.to_str().ok())
67            .and_then(|value| value.trim().parse().ok());
68        Page {
69            value,
70            next_url,
71            next_cursor,
72            total_count,
73            info,
74            retry,
75        }
76    }
77
78    fn cursor(&self) -> Cursor {
79        Cursor {
80            next_url: self.next_url.clone(),
81            info: self.info.clone(),
82            retry: self.retry.clone(),
83        }
84    }
85
86    /// The page's contents, owned.
87    pub fn into_inner(self) -> T {
88        self.value
89    }
90
91    /// The page's contents.
92    pub fn value(&self) -> &T {
93        &self.value
94    }
95
96    /// The opaque cursor for the page after this one, to pass as `page` on the same read.
97    pub fn next_page(&self) -> Option<&str> {
98        self.next_cursor.as_deref()
99    }
100
101    /// The URL of the page after this one, as Fizzy's `Link` header named it.
102    pub fn next_url(&self) -> Option<&Url> {
103        self.next_url.as_ref()
104    }
105
106    /// Whether Fizzy named a page after this one.
107    pub fn has_next(&self) -> bool {
108        self.next_url.is_some()
109    }
110
111    /// The `X-Total-Count` header, when the read carried one.
112    pub fn total_count(&self) -> Option<u64> {
113        self.total_count
114    }
115
116    /// The same page with its contents transformed.
117    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> Page<U> {
118        Page {
119            value: f(self.value),
120            next_url: self.next_url,
121            next_cursor: self.next_cursor,
122            total_count: self.total_count,
123            info: self.info,
124            retry: self.retry,
125        }
126    }
127}
128
129impl<T> Deref for Page<T> {
130    type Target = T;
131
132    fn deref(&self) -> &T {
133        &self.value
134    }
135}
136
137impl Client {
138    /// Reads the page after the given one, or `None` when Fizzy named no next page. A
139    /// `Link` header pointing off the Fizzy origin is refused rather than followed. The
140    /// read announces itself as the operation the first page came from and is sent under
141    /// the same retry policy, so a whole walk shows up as one thing and is retried as one.
142    pub async fn next_page<T: DeserializeOwned>(
143        &self,
144        page: &Page<T>,
145    ) -> Result<Option<Page<T>>, Error> {
146        self.page_after(&page.cursor()).await
147    }
148
149    async fn page_after<T: DeserializeOwned>(
150        &self,
151        cursor: &Cursor,
152    ) -> Result<Option<Page<T>>, Error> {
153        match &cursor.next_url {
154            None => Ok(None),
155            Some(next) if !is_same_origin(next, self.base_url()) => Err(Error::usage(format!(
156                "pagination Link header points to a different origin: {next}"
157            ))),
158            Some(next) => {
159                let mut operation = Operation::at(Method::GET, next.clone());
160                operation.info(cursor.info.clone());
161                if let Some(retry) = &cursor.retry {
162                    operation.retry(retry.clone());
163                }
164                self.send_page(operation).await.map(Some)
165            }
166        }
167    }
168
169    /// Reads every page after the first, up to the client's page limit, calling `visit`
170    /// with each one. Stops early when `visit` answers `false`.
171    pub async fn each_page<T: DeserializeOwned>(
172        &self,
173        first: Page<T>,
174        mut visit: impl FnMut(&Page<T>) -> bool,
175    ) -> Result<(), Error> {
176        let mut page = first;
177        let mut count = 1;
178        while visit(&page) && count < self.max_pages() {
179            match self.next_page(&page).await? {
180                Some(next) => page = next,
181                None => break,
182            }
183            count += 1;
184        }
185        Ok(())
186    }
187
188    /// The first page and every one after it, read lazily as the stream is polled, up to
189    /// the client's page limit. A page in hand is yielded before the next is fetched, so a
190    /// consumer that stops early never pays for a page it did not read, and a page that
191    /// fails to read ends the stream with its error after the ones before it.
192    pub fn pages<'a, T: DeserializeOwned + 'a>(
193        &'a self,
194        first: Page<T>,
195    ) -> impl Stream<Item = Result<Page<T>, Error>> + 'a {
196        let max_pages = self.max_pages();
197        stream::try_unfold(
198            (Some(first), None::<Cursor>, 0usize),
199            move |(pending, cursor, read)| async move {
200                let page = match (pending, cursor) {
201                    (Some(page), _) => page,
202                    (None, Some(cursor)) if read < max_pages => {
203                        match self.page_after(&cursor).await? {
204                            Some(page) => page,
205                            None => return Ok(None),
206                        }
207                    }
208                    (None, _) => return Ok(None),
209                };
210                let cursor = page.cursor();
211                Ok(Some((page, (None, Some(cursor), read + 1))))
212            },
213        )
214    }
215
216    /// Every item on every page, read lazily as the stream is polled.
217    pub fn items<'a, T: DeserializeOwned + 'a>(
218        &'a self,
219        first: Page<Vec<T>>,
220    ) -> impl Stream<Item = Result<T, Error>> + 'a {
221        self.pages(first).flat_map(|page| match page {
222            Ok(page) => stream::iter(page.into_inner().into_iter().map(Ok)).left_stream(),
223            Err(error) => stream::once(async move { Err(error) }).right_stream(),
224        })
225    }
226
227    /// Reads a paginated path to its end and hands back the items of every page as one
228    /// list. Each page has to decode as a JSON array. Use this for the paths the model
229    /// does not cover; a modelled read walks with [`Client::each_page`] or
230    /// [`Client::pages`], which keep the records typed.
231    pub async fn get_all(&self, path: &str) -> Result<Vec<Value>, Error> {
232        self.get_all_with_limit(path, 0).await
233    }
234
235    /// Reads a paginated path until `limit` items are in hand, or to its end when `limit`
236    /// is zero. The last page is trimmed to land on exactly `limit`.
237    pub async fn get_all_with_limit(&self, path: &str, limit: usize) -> Result<Vec<Value>, Error> {
238        let operation = self.raw(Method::GET, path)?;
239        self.collect_all(operation, limit).await
240    }
241
242    pub(crate) async fn collect_all(
243        &self,
244        mut operation: Operation,
245        limit: usize,
246    ) -> Result<Vec<Value>, Error> {
247        let started_at = self.url_for(&operation)?;
248        let retry = operation.retry.clone();
249        let mut collected: Vec<Value> = Vec::new();
250        let mut pages = 0;
251
252        loop {
253            let response = self.execute(operation).await?;
254            collected.extend(response.json::<Vec<Value>>()?);
255            pages += 1;
256
257            if limit > 0 && collected.len() >= limit {
258                collected.truncate(limit);
259                break;
260            }
261            match Client::next_page_url(&response, &started_at)? {
262                Some(next) if pages < self.max_pages() => {
263                    operation = Operation::at(Method::GET, next);
264                    if let Some(retry) = &retry {
265                        operation.retry(retry.clone());
266                    }
267                }
268                Some(_) => {
269                    crate::trace::warn(&format!("pagination capped at {} pages", self.max_pages()));
270                    break;
271                }
272                None => break,
273            }
274        }
275        Ok(collected)
276    }
277
278    /// The page after this one, as the `Link` header named it, resolved against the answer
279    /// it came in. A target off the origin the walk started on is refused rather than
280    /// followed: the header is the server's to write, and following it would carry the
281    /// credentials somewhere they were never meant to go.
282    fn next_page_url(response: &Response, started_at: &Url) -> Result<Option<Url>, Error> {
283        match response.header("link").and_then(next_link) {
284            None => Ok(None),
285            Some(target) => {
286                let next = response.url.join(&target)?;
287                if is_same_origin(&next, started_at) {
288                    Ok(Some(next))
289                } else {
290                    Err(Error::usage(format!(
291                        "pagination Link header points to a different origin: {next}"
292                    )))
293                }
294            }
295        }
296    }
297}
298
299/// The target of the `rel="next"` link in an RFC 8288 `Link` header. Targets are read
300/// between angle brackets, so commas inside a URL do not split it, and `rel` is a
301/// space-separated set matched case-insensitively.
302pub fn next_link(header: &str) -> Option<String> {
303    let mut remaining = header;
304    while let Some(start) = remaining.find('<') {
305        let after_start = &remaining[start + 1..];
306        let end = after_start.find('>')?;
307        let target = &after_start[..end];
308        let rest = &after_start[end + 1..];
309        let params_end = rest.find('<').unwrap_or(rest.len());
310        // The link-values are comma-separated, so the parameters of this one end at the
311        // comma before the next `<`, however the header is spaced.
312        let params = rest[..params_end].trim().trim_end_matches(',');
313        if link_is_next(params) {
314            return Some(target.to_string());
315        }
316        remaining = &rest[params_end..];
317    }
318    None
319}
320
321fn link_is_next(params: &str) -> bool {
322    params.split(';').any(|param| {
323        let mut parts = param.splitn(2, '=');
324        let name = parts.next().unwrap_or_default().trim();
325        let value = parts.next().unwrap_or_default().trim().trim_matches('"');
326        name.eq_ignore_ascii_case("rel")
327            && value
328                .split_whitespace()
329                .any(|rel| rel.eq_ignore_ascii_case("next"))
330    })
331}
332
333#[cfg(test)]
334mod tests {
335    use super::*;
336
337    #[test]
338    fn finds_the_next_link_among_others() {
339        let header = r#"<https://fizzy.do/999/boards.json?page=1>; rel="prev", <https://fizzy.do/999/boards.json?page=3>; rel="next""#;
340        assert_eq!(
341            next_link(header).as_deref(),
342            Some("https://fizzy.do/999/boards.json?page=3")
343        );
344    }
345
346    #[test]
347    fn finds_the_next_link_whichever_order_the_relations_come_in() {
348        assert_eq!(
349            next_link(r#"</p2>; rel="next", </p9>; rel="last""#).as_deref(),
350            Some("/p2")
351        );
352        assert_eq!(
353            next_link(r#"</p9>; rel="last",</p2>; rel="next""#).as_deref(),
354            Some("/p2")
355        );
356        assert_eq!(next_link(r#"</p9>; rel="last", </p1>; rel="first""#), None);
357    }
358
359    #[test]
360    fn matches_rel_sets_and_case() {
361        assert_eq!(
362            next_link(r#"</x?page=2>; REL="prev next""#).as_deref(),
363            Some("/x?page=2")
364        );
365        assert_eq!(next_link(r#"</x?page=2>; rel="last""#), None);
366        assert_eq!(next_link(""), None);
367    }
368}
fizzy_sdk/pagination.rs

fizzy_sdk/
pagination.rs
