ojak/federation/
collection.rs

1//! Collections, paged by cursor.
2//!
3//! The application writes one page function, and optionally a counter and
4//! the cursors of the first and last pages; Ojak writes the
5//! `OrderedCollection` at the template's URI and the `OrderedCollectionPage`s
6//! at the same URI with `?cursor=`, or the `Collection` and its
7//! `CollectionPage`s when it is [unordered](Collection::unordered).
8//!
9//! A collection whose URIs Ojak does not write, such as one paged by a query
10//! of the application's own, is a document the application builds, as a
11//! [`CollectionDocument`], and serves as an object.
12
13use super::{BoxFuture, Context, Error, activity, boxed, empty};
14use http::StatusCode;
15use serde_json::Value;
16use std::future::Future;
17use std::sync::Arc;
18use url::Url;
19
20/// One page of a collection.
21#[derive(Clone, Debug, Default, PartialEq)]
22pub struct Page {
23    /// Its items, in order: IRIs as strings, or embedded objects.
24    pub items: Vec<Value>,
25    /// The cursor of the page after this one, if there is one.
26    pub next: Option<String>,
27    /// The cursor of the page before this one, if there is one.
28    pub prev: Option<String>,
29}
30
31/// Where a paged collection starts.
32#[derive(Clone, Debug, Eq, PartialEq)]
33pub enum First {
34    /// At the page with this cursor.
35    At(String),
36    /// Nowhere a requester may see: the collection's members are hidden, and
37    /// only its count, if it has a counter, is shown.
38    Hidden,
39}
40
41type PageFn<D> = Arc<
42    dyn Fn(Context<D>, String, Option<String>) -> BoxFuture<'static, Result<Option<Page>, Error>>
43        + Send
44        + Sync,
45>;
46type CountFn<D> =
47    Arc<dyn Fn(Context<D>, String) -> BoxFuture<'static, Result<Option<u64>, Error>> + Send + Sync>;
48type FirstFn<D> = Arc<
49    dyn Fn(Context<D>, String) -> BoxFuture<'static, Result<Option<First>, Error>> + Send + Sync,
50>;
51type UriFn<D> =
52    Arc<dyn Fn(Context<D>, String) -> BoxFuture<'static, Result<Option<Url>, Error>> + Send + Sync>;
53type ContextFn = Arc<dyn Fn(&[Value]) -> Value + Send + Sync>;
54type ToQueryFn = Arc<dyn Fn(&str) -> String + Send + Sync>;
55type FromQueryFn = Arc<dyn Fn(&str) -> Option<String> + Send + Sync>;
56type LastFn<D> = Arc<
57    dyn Fn(Context<D>, String) -> BoxFuture<'static, Result<Option<String>, Error>> + Send + Sync,
58>;
59
60/// A collection's dispatchers.
61pub struct Collection<D> {
62    page: PageFn<D>,
63    count: Option<CountFn<D>>,
64    first: Option<FirstFn<D>>,
65    last: Option<LastFn<D>>,
66    uri: Option<UriFn<D>>,
67    ordered: bool,
68    context: Option<ContextFn>,
69    query: Option<(ToQueryFn, FromQueryFn)>,
70    count_pages: bool,
71}
72
73impl<D> Clone for Collection<D> {
74    fn clone(&self) -> Self {
75        Self {
76            page: self.page.clone(),
77            count: self.count.clone(),
78            first: self.first.clone(),
79            last: self.last.clone(),
80            uri: self.uri.clone(),
81            ordered: self.ordered,
82            context: self.context.clone(),
83            query: self.query.clone(),
84            count_pages: self.count_pages,
85        }
86    }
87}
88
89impl<D: Clone + Send + Sync + 'static> Collection<D> {
90    /// A collection read through `page`, which receives the identifier of
91    /// what the collection belongs to and a cursor, and returns that page, or
92    /// `None` when there is no such collection.
93    ///
94    /// Without [`Collection::first_cursor`], the collection is one document:
95    /// `page` is called once, with no cursor, and every item is in it.
96    pub fn new<F, Fut, E>(page: F) -> Self
97    where
98        F: Fn(Context<D>, String, Option<String>) -> Fut + Send + Sync + 'static,
99        Fut: Future<Output = Result<Option<Page>, E>> + Send + 'static,
100        E: Into<Error>,
101    {
102        let page = boxed(move |(context, identifier, cursor)| page(context, identifier, cursor));
103        Self {
104            page: Arc::new(move |context, identifier, cursor| page((context, identifier, cursor))),
105            count: None,
106            first: None,
107            last: None,
108            uri: None,
109            ordered: true,
110            context: None,
111            query: None,
112            count_pages: false,
113        }
114    }
115
116    /// Serve it as a `Collection`, whose pages are `CollectionPage`s and
117    /// whose members are `items`, rather than ordered: for a collection
118    /// whose order means nothing, such as an account's featured hashtags or
119    /// who liked a post.
120    #[must_use]
121    pub fn unordered(mut self) -> Self {
122        self.ordered = false;
123        self
124    }
125
126    /// The `@context` of the collection and its pages, from the members a
127    /// document has (none, for a collection that only links its pages):
128    /// for an application whose context says what its members use, as
129    /// Mastodon's names `Hashtag` only for a collection of hashtags that
130    /// has any. Without it, a document gets [`default_context`].
131    ///
132    /// [`default_context`]: super::default_context
133    #[must_use]
134    pub fn context(mut self, context: impl Fn(&[Value]) -> Value + Send + Sync + 'static) -> Self {
135        self.context = Some(Arc::new(context));
136        self
137    }
138
139    /// Write a page's cursor into its URI's query, and read it back, with
140    /// `to_query` and `from_query`, rather than as `?cursor=`: for a
141    /// collection whose pages peers already know by a query of their own,
142    /// such as Mastodon's outbox at `?page=true&max_id=`. `to_query` gives
143    /// the whole query for a cursor; `from_query` is given a request's
144    /// whole query, and says which page it asks for, or `None` for the
145    /// collection itself.
146    #[must_use]
147    pub fn page_query(
148        mut self,
149        to_query: impl Fn(&str) -> String + Send + Sync + 'static,
150        from_query: impl Fn(&str) -> Option<String> + Send + Sync + 'static,
151    ) -> Self {
152        self.query = Some((Arc::new(to_query), Arc::new(from_query)));
153        self
154    }
155
156    /// Say `totalItems` on each page as well as on the collection, from the
157    /// counter, as Mastodon's followers and following pages do.
158    #[must_use]
159    pub fn count_pages(mut self) -> Self {
160        self.count_pages = true;
161        self
162    }
163
164    /// The URI of the page at `cursor` of the collection at `collection`.
165    fn page_uri(&self, collection: &Url, cursor: &str) -> Url {
166        let mut url = collection.clone();
167        match &self.query {
168            Some((to_query, _)) => {
169                let query = to_query(cursor);
170                url.set_query((!query.is_empty()).then_some(query.as_str()));
171            }
172            None => {
173                url.query_pairs_mut().clear().append_pair("cursor", cursor);
174            }
175        }
176        url
177    }
178
179    /// `document` with the `@context` the collection gives it, if it gives
180    /// one, before its other members.
181    fn with_context(&self, document: &CollectionDocument) -> Value {
182        let value = document.to_value();
183        let Some(context) = &self.context else {
184            return value;
185        };
186        let mut with_context = serde_json::Map::new();
187        with_context.insert(
188            "@context".into(),
189            context(document.items.as_deref().unwrap_or_default()),
190        );
191        if let Value::Object(members) = value {
192            with_context.extend(members);
193        }
194        Value::Object(with_context)
195    }
196
197    /// How many items the collection has, for `totalItems`.
198    #[must_use]
199    pub fn count<F, Fut, E>(mut self, count: F) -> Self
200    where
201        F: Fn(Context<D>, String) -> Fut + Send + Sync + 'static,
202        Fut: Future<Output = Result<Option<u64>, E>> + Send + 'static,
203        E: Into<Error>,
204    {
205        let count = boxed(move |(context, identifier)| count(context, identifier));
206        self.count = Some(Arc::new(move |context, identifier| {
207            count((context, identifier))
208        }));
209        self
210    }
211
212    /// Where the collection starts, which makes it paged; `None` when there is
213    /// no such collection.
214    #[must_use]
215    pub fn first_cursor<F, Fut, E>(mut self, first: F) -> Self
216    where
217        F: Fn(Context<D>, String) -> Fut + Send + Sync + 'static,
218        Fut: Future<Output = Result<Option<First>, E>> + Send + 'static,
219        E: Into<Error>,
220    {
221        let first = boxed(move |(context, identifier)| first(context, identifier));
222        self.first = Some(Arc::new(move |context, identifier| {
223            first((context, identifier))
224        }));
225        self
226    }
227
228    /// The cursor of its last page, for `last`.
229    #[must_use]
230    pub fn last_cursor<F, Fut, E>(mut self, last: F) -> Self
231    where
232        F: Fn(Context<D>, String) -> Fut + Send + Sync + 'static,
233        Fut: Future<Output = Result<Option<String>, E>> + Send + 'static,
234        E: Into<Error>,
235    {
236        let last = boxed(move |(context, identifier)| last(context, identifier));
237        self.last = Some(Arc::new(move |context, identifier| {
238            last((context, identifier))
239        }));
240        self
241    }
242
243    /// The collection's own URI, when it is not the one it was asked at: an
244    /// actor served under two templates, as Mastodon serves an account at
245    /// `/users/{username}` and `/ap/users/{id}`, names each of its collections
246    /// by one of them. `None` keeps the URI of the template it was asked at.
247    /// The collection and its pages are named, and linked, by this URI.
248    #[must_use]
249    pub fn uri<F, Fut, E>(mut self, uri: F) -> Self
250    where
251        F: Fn(Context<D>, String) -> Fut + Send + Sync + 'static,
252        Fut: Future<Output = Result<Option<Url>, E>> + Send + 'static,
253        E: Into<Error>,
254    {
255        let uri = boxed(move |(context, identifier)| uri(context, identifier));
256        self.uri = Some(Arc::new(move |context, identifier| {
257            uri((context, identifier))
258        }));
259        self
260    }
261
262    pub(super) async fn serve(
263        &self,
264        context: &Context<D>,
265        kind: &str,
266        identifier: &str,
267        query: Option<&str>,
268    ) -> Result<http::Response<Vec<u8>>, Error> {
269        let own = match &self.uri {
270            Some(uri) => uri(context.clone(), identifier.to_owned()).await?,
271            None => None,
272        };
273        let uri = match own {
274            Some(uri) => uri,
275            None => context.collection_uri(kind, identifier)?,
276        };
277        let cursor = match &self.query {
278            Some((_, from_query)) => from_query(query.unwrap_or_default()),
279            None => query.and_then(|query| {
280                url::form_urlencoded::parse(query.as_bytes())
281                    .find(|(name, _)| name == "cursor")
282                    .map(|(_, value)| value.into_owned())
283            }),
284        };
285        let identifier = identifier.to_owned();
286
287        if let Some(cursor) = cursor {
288            let Some(page) =
289                (self.page)(context.clone(), identifier.clone(), Some(cursor.clone())).await?
290            else {
291                return Ok(empty(StatusCode::NOT_FOUND));
292            };
293            let total_items = match &self.count {
294                Some(count) if self.count_pages => {
295                    count(context.clone(), identifier.clone()).await?
296                }
297                _ => None,
298            };
299            let document = CollectionDocument {
300                id: Some(self.page_uri(&uri, &cursor)),
301                total_items,
302                ordered: self.ordered,
303                part_of: Some(uri.clone()),
304                next: page.next.map(|next| self.page_uri(&uri, &next)),
305                prev: page.prev.map(|prev| self.page_uri(&uri, &prev)),
306                items: Some(page.items),
307                ..CollectionDocument::default()
308            };
309            return Ok(activity(StatusCode::OK, self.with_context(&document)));
310        }
311
312        let Some(first) = &self.first else {
313            let Some(page) = (self.page)(context.clone(), identifier.clone(), None).await? else {
314                return Ok(empty(StatusCode::NOT_FOUND));
315            };
316            let total = match &self.count {
317                Some(count) => count(context.clone(), identifier).await?,
318                None => None,
319            }
320            .unwrap_or(page.items.len() as u64);
321            let document = CollectionDocument {
322                id: Some(uri),
323                ordered: self.ordered,
324                total_items: Some(total),
325                items: Some(page.items),
326                ..CollectionDocument::default()
327            };
328            return Ok(activity(StatusCode::OK, self.with_context(&document)));
329        };
330
331        let Some(start) = first(context.clone(), identifier.clone()).await? else {
332            return Ok(empty(StatusCode::NOT_FOUND));
333        };
334        let mut document = CollectionDocument {
335            ordered: self.ordered,
336            ..CollectionDocument::default()
337        };
338        if let Some(count) = &self.count {
339            document.total_items = count(context.clone(), identifier.clone()).await?;
340        }
341        if let First::At(cursor) = start {
342            document.first = Some(self.page_uri(&uri, &cursor).as_str().into());
343            if let Some(last) = &self.last
344                && let Some(cursor) = last(context.clone(), identifier).await?
345            {
346                document.last = Some(self.page_uri(&uri, &cursor).as_str().into());
347            }
348        }
349        document.id = Some(uri);
350        Ok(activity(StatusCode::OK, self.with_context(&document)))
351    }
352}
353
354/// An ActivityStreams collection or collection page, written as a document:
355/// what Ojak serves a [`Collection`] as, for an application that serves one
356/// itself, such as a collection paged by a query of its own or embedded in
357/// another object.
358///
359/// It is a page when it is part of a collection ([`part_of`]) or says it is
360/// ([`page`]), and ordered or not as [`ordered`] says. Its members are
361/// written whenever there are [`items`], even none; a collection that only
362/// counts its members, or only links its pages, has `None`.
363///
364/// [`part_of`]: CollectionDocument::part_of
365/// [`page`]: CollectionDocument::page
366/// [`ordered`]: CollectionDocument::ordered
367/// [`items`]: CollectionDocument::items
368#[derive(Clone, Debug, Default, PartialEq)]
369pub struct CollectionDocument {
370    /// Its `id`. An embedded page may have none.
371    pub id: Option<Url>,
372    /// An `OrderedCollection` or `OrderedCollectionPage`, whose members are
373    /// `orderedItems`, rather than a `Collection` or `CollectionPage`, whose
374    /// members are `items`.
375    pub ordered: bool,
376    /// A page, even with no [`part_of`](CollectionDocument::part_of).
377    pub page: bool,
378    /// Its `totalItems`.
379    pub total_items: Option<u64>,
380    /// Its `next` page.
381    pub next: Option<Url>,
382    /// Its `prev` page.
383    pub prev: Option<Url>,
384    /// The collection it is a page of.
385    pub part_of: Option<Url>,
386    /// Its `first` page: a URI, or the page embedded.
387    pub first: Option<Value>,
388    /// Its `last` page: a URI, or the page embedded.
389    pub last: Option<Value>,
390    /// Its members, in order: IRIs as strings, or embedded objects.
391    pub items: Option<Vec<Value>>,
392}
393
394impl CollectionDocument {
395    /// Whether it is a page rather than a collection.
396    #[must_use]
397    pub fn is_page(&self) -> bool {
398        self.page || self.part_of.is_some()
399    }
400
401    /// Its `type`.
402    #[must_use]
403    pub fn type_name(&self) -> &'static str {
404        match (self.ordered, self.is_page()) {
405            (true, true) => "OrderedCollectionPage",
406            (true, false) => "OrderedCollection",
407            (false, true) => "CollectionPage",
408            (false, false) => "Collection",
409        }
410    }
411
412    /// The document, without an `@context`: `id`, `type`, `totalItems`,
413    /// `next`, `prev`, `partOf`, `first`, `last` and the members, in that
414    /// order, each only when there is one.
415    #[must_use]
416    pub fn to_value(&self) -> Value {
417        let mut document = serde_json::Map::new();
418        if let Some(id) = &self.id {
419            document.insert("id".into(), id.as_str().into());
420        }
421        document.insert("type".into(), self.type_name().into());
422        if let Some(total) = self.total_items {
423            document.insert("totalItems".into(), total.into());
424        }
425        if let Some(next) = &self.next {
426            document.insert("next".into(), next.as_str().into());
427        }
428        if let Some(prev) = &self.prev {
429            document.insert("prev".into(), prev.as_str().into());
430        }
431        if let Some(part_of) = &self.part_of {
432            document.insert("partOf".into(), part_of.as_str().into());
433        }
434        if let Some(first) = &self.first {
435            document.insert("first".into(), first.clone());
436        }
437        if let Some(last) = &self.last {
438            document.insert("last".into(), last.clone());
439        }
440        if let Some(items) = &self.items {
441            let key = if self.ordered {
442                "orderedItems"
443            } else {
444                "items"
445            };
446            document.insert(key.into(), Value::Array(items.clone()));
447        }
448        Value::Object(document)
449    }
450}
ojak/federation/collection.rs

ojak/federation/
collection.rs
