api: allow reading past versions

1 files changed, 54 insertions, 8 deletions

diff --git a/src/api/store.rs b/src/api/store.rs
index 82cac30..8b09487 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -38,6 +38,43 @@ pub(crate) struct ReadQuery {
     /// level at a time.
     #[serde(default)]
     shallow: bool,
+    /// Read as the document was at these heads (comma-separated hex, as
+    /// `x-continuum-heads` gives them), not as it is now (`document::at`).
+    at: Option<String>,
+}
+
+impl ReadQuery {
+    /// The heads `at` names, if any; `400` if it doesn't name them.
+    fn at(&self) -> Result<Option<Vec<ChangeHash>>, ApiError> {
+        let Some(at) = &self.at else {
+            return Ok(None);
+        };
+        at.split(',')
+            .map(|head| head.trim().parse::<ChangeHash>())
+            .collect::<Result<Vec<_>, _>>()
+            .map(Some)
+            .map_err(|_| ApiError(StatusCode::BAD_REQUEST, Some("at: not heads".to_owned())))
+    }
+}
+
+/// Reads `doc` with `read`, as it is or, with `at`, as it was at those heads
+/// (`UnknownHeads` if it lacks one).
+macro_rules! read_at {
+    ($doc:expr, $at:expr, |$d:ident| $read:expr) => {
+        match $at {
+            None => {
+                let $d = $doc;
+                $read
+            }
+            Some(heads) => match crate::document::at::At::new($doc, heads) {
+                Some(at) => {
+                    let $d = &at;
+                    $read
+                }
+                None => Err(StoreError::UnknownHeads),
+            },
+        }
+    };
 }
 
 #[derive(Debug, Deserialize)]
@@ -72,7 +109,10 @@ pub(crate) async fn get_store_entry(
     RawQuery(raw_query): RawQuery,
 ) -> Result<Response, ApiError> {
     let path = parse_key_path(&key)?;
-    if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
+    let at = query.at()?;
+    if let Some(redirect) =
+        to_positive(&state, &path, false, at.as_deref(), raw_query.as_deref()).await?
+    {
         return Ok(redirect);
     }
     read(&state, path, &headers, &query).await
@@ -93,18 +133,19 @@ async fn read(
     }
     let accept = headers.get(ACCEPT).cloned();
     let want_conflicts = query.conflicts;
+    let at = query.at()?;
 
     let (response, source, heads) = state
         .render_read(
             root,
             query.heads,
             move |doc| {
-                if want_conflicts {
+                read_at!(doc, at.as_deref(), |doc| if want_conflicts {
                     store::read_path_with_conflicts(doc, &path)
                         .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts))))
                 } else {
                     store::read_path(doc, &path).map(|found| found.map(|value| (value, None)))
-                }
+                })
             },
             move |result| {
                 let found = result.map_err(|e| store_status(e, "Failed to read store path"))?;
@@ -134,11 +175,12 @@ async fn shallow(
     if query.conflicts {
         return Err(StatusCode::BAD_REQUEST.into());
     }
+    let at = query.at()?;
     let (response, source, heads) = state
         .render_read(
             unready_live,
             query.heads,
-            move |doc| store::read_shallow(doc, &path),
+            move |doc| read_at!(doc, at.as_deref(), |doc| store::read_shallow(doc, &path)),
             move |result| {
                 let value = result
                     .map_err(|e| store_status(e, "Failed to read store path"))?
@@ -195,7 +237,7 @@ pub(crate) async fn put_store_entry(
     }
 
     let path = parse_write_path(&state, &user, &key, "PUT")?;
-    if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
+    if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
         return Ok(redirect);
     }
     let value = body_value(&headers, &body)?;
@@ -326,7 +368,7 @@ pub(crate) async fn post_store_entry(
         return Err(StatusCode::SERVICE_UNAVAILABLE.into());
     }
     let path = parse_write_path(&state, &user, &key, "POST")?;
-    if let Some(redirect) = to_positive(&state, &path, true, raw_query.as_deref()).await? {
+    if let Some(redirect) = to_positive(&state, &path, true, None, raw_query.as_deref()).await? {
         return Ok(redirect);
     }
     let index = path.last().and_then(|leaf| store::list_index(leaf));
@@ -367,7 +409,7 @@ pub(crate) async fn delete_store_entry(
         return Err(StatusCode::BAD_REQUEST.into());
     }
 
-    if let Some(redirect) = to_positive(&state, &path, false, raw_query.as_deref()).await? {
+    if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? {
         return Ok(redirect);
     }
 
@@ -389,6 +431,7 @@ async fn to_positive(
     state: &AppState,
     path: &[String],
     insertion: bool,
+    at: Option<&[ChangeHash]>,
     query: Option<&str>,
 ) -> Result<Option<Response>, ApiError> {
     if !path.iter().any(|segment| segment.starts_with('-')) {
@@ -396,7 +439,10 @@ async fn to_positive(
     }
     let positive = {
         let replica = state.doc.read().await;
-        store::positive_path(&replica.doc, path, insertion)
+        // Counted from the end of the list as it was then, for a read at heads.
+        read_at!(&replica.doc, at, |doc| store::positive_path(
+            doc, path, insertion
+        ))
     }
     .map_err(|e| store_status(e, "Failed to resolve a negative index"))?;
     let Some(positive) = positive else {