add the replace kind of remake

POST /admin/remakes accepts specs with replace kind, e.g.:

  { "kind": "replace",
    "baseHeads": ["..."],
    "spec": {
      "document": { "accounts": { ... } },
      "rules": [
        { "from": "users",       "to": "accounts" },
        { "from": "users/admin", "drop": true },
        { "from": "",            "to": "" }
      ] } }

4 files changed, 256 insertions, 4 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
index b473a2d..7db7acb 100644
--- a/src/remake/api.rs
+++ b/src/remake/api.rs
@@ -20,6 +20,10 @@
 //!    "replayWindow": {"epochs": 1, "seconds": 604800},
 //!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
 //!    "settleSecs": 480 }`
+//!
+//! `kind` is `compact` (no spec) or `replace`, whose spec is
+//! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
+//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`).
 
 use std::time::Duration;
 
@@ -83,6 +87,10 @@ struct Report {
     phase: Phase,
     #[serde(flatten)]
     record: Record,
+    /// Size of the kind's build input (e.g. a replacement document), which reports
+    /// leave out.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    content_bytes: Option<usize>,
     /// What happened to this remake on the node answering the request.
     #[serde(skip_serializing_if = "Option::is_none")]
     local: Option<LocalStatus>,
@@ -114,10 +122,12 @@ fn grace<H: Host>(host: &H) -> u64 {
     host.dormant_after().as_millis() as u64
 }
 
-fn report<H: Host>(host: &H, id: String, record: Record) -> Report {
+fn report<H: Host>(host: &H, id: String, mut record: Record) -> Report {
+    let content = std::mem::take(&mut record.proposal.content);
     Report {
         phase: record.phase(unix_millis(), grace(host)),
         local: host.remake().status(&id),
+        content_bytes: (!content.is_null()).then(|| content.to_string().len()),
         id,
         record,
     }
diff --git a/src/remake/kind.rs b/src/remake/kind.rs
index 329fd09..b709a05 100644
--- a/src/remake/kind.rs
+++ b/src/remake/kind.rs
@@ -10,13 +10,17 @@
 //! * where a write made in the old epoch after the base goes in the new one
 //!   ([`RemakeKind::translate`]).
 //!
+//! Kinds: [`Compact`] (same content, no history) and [`Replace`] (an operator's
+//! document, with rules for where later writes go).
+//!
 //! Protocol data under `_continuum` is not a kind's business: the generic part carries
 //! it forward and replays it unchanged.
 
 use automerge::AutoCommit;
+use serde::{Deserialize, Serialize};
 use serde_json::Value as JsonValue;
 
-use super::snapshot::META_ROOT;
+use super::snapshot::{canonical, META_ROOT};
 use crate::document::store::{self, StoreError};
 
 /// Name of the kind used when a request doesn't name one.
@@ -70,6 +74,7 @@ pub(crate) trait RemakeKind: Send + Sync + 'static {
 pub(crate) fn by_name(name: &str) -> Option<&'static dyn RemakeKind> {
     match name {
         "compact" => Some(&Compact),
+        "replace" => Some(&Replace),
         _ => None,
     }
 }
@@ -110,6 +115,162 @@ impl RemakeKind for Compact {
     }
 }
 
+/// Replacement: the new document is supplied by an operator, in store form (the JSON
+/// `GET /store` returns), and may be structured entirely differently. Writes made after
+/// the base go where the remake's rules say; a path no rule covers is quarantined.
+///
+/// Spec: `{ "document": {…}, "rules": [ {"from": "a/b", "to": "c"},
+/// {"from": "old", "drop": true} ] }`. Paths are slash-separated strings (`""` is the
+/// root, matching everything) or arrays of segments (for keys containing `/`). The
+/// longest matching `from` wins; the rest of the path is appended to `to`.
+pub(crate) struct Replace;
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ReplaceSpec {
+    document: JsonValue,
+    #[serde(default)]
+    rules: Vec<RuleSpec>,
+}
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RuleSpec {
+    from: PathSpec,
+    #[serde(default)]
+    to: Option<PathSpec>,
+    #[serde(default)]
+    drop: bool,
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum PathSpec {
+    Text(String),
+    Segments(Vec<String>),
+}
+
+impl PathSpec {
+    fn segments(self) -> Result<Vec<String>, String> {
+        let segments: Vec<String> = match self {
+            PathSpec::Text(text) if text.is_empty() => Vec::new(),
+            PathSpec::Text(text) => text.split('/').map(str::to_owned).collect(),
+            PathSpec::Segments(segments) => segments,
+        };
+        if segments.iter().any(String::is_empty) {
+            return Err(format!("empty segment in path {segments:?}"));
+        }
+        Ok(segments)
+    }
+}
+
+/// A rule as stored in the translation (and the epoch metadata).
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+struct Rule {
+    from: Vec<String>,
+    /// None: drop.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    to: Option<Vec<String>>,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+struct Rules {
+    rules: Vec<Rule>,
+}
+
+impl RemakeKind for Replace {
+    fn name(&self) -> &'static str {
+        "replace"
+    }
+
+    fn propose(&self, spec: &JsonValue) -> Result<Prepared, String> {
+        let spec: ReplaceSpec =
+            serde_json::from_value(spec.clone()).map_err(|error| error.to_string())?;
+
+        let JsonValue::Object(document) = &spec.document else {
+            return Err("document must be a JSON object (the store root)".to_owned());
+        };
+        if document.contains_key(META_ROOT) {
+            return Err(format!(
+                "document must not contain the reserved key {META_ROOT:?}"
+            ));
+        }
+
+        let mut rules = Vec::with_capacity(spec.rules.len());
+        for rule in spec.rules {
+            let from = rule.from.segments()?;
+            let to = match (rule.to, rule.drop) {
+                (Some(to), false) => Some(to.segments()?),
+                (None, true) => None,
+                _ => {
+                    return Err(format!(
+                        "rule for {from:?} needs exactly one of \"to\" and \"drop\""
+                    ))
+                }
+            };
+            if is_reserved(&from) || to.as_deref().is_some_and(is_reserved) {
+                return Err(format!(
+                    "rules must not touch the reserved key {META_ROOT:?}"
+                ));
+            }
+            if rules.iter().any(|existing: &Rule| existing.from == from) {
+                return Err(format!("two rules for {from:?}"));
+            }
+            rules.push(Rule { from, to });
+        }
+
+        Ok(Prepared {
+            content: canonical(spec.document),
+            translation: serde_json::to_value(Rules { rules })
+                .map_err(|error| error.to_string())?,
+        })
+    }
+
+    fn build(
+        &self,
+        _old_at_base: &AutoCommit,
+        content: &JsonValue,
+        target: &mut AutoCommit,
+    ) -> Result<(), StoreError> {
+        // Canonical again: the content went through the old epoch's document, and the
+        // operations (hence the epoch ID) must not depend on any key order.
+        let JsonValue::Object(document) = canonical(content.clone()) else {
+            return Ok(()); // propose only accepts objects
+        };
+        for (key, value) in document {
+            if key != META_ROOT {
+                store::write_path(target, &[key], &value)?;
+            }
+        }
+        Ok(())
+    }
+
+    fn translate(&self, path: &[String], translation: &JsonValue) -> Translation {
+        let Ok(Rules { rules }) = serde_json::from_value::<Rules>(translation.clone()) else {
+            return Translation::Quarantine("unreadable replacement rules".to_owned());
+        };
+        let Some(rule) = rules
+            .iter()
+            .filter(|rule| path.starts_with(&rule.from))
+            .max_by_key(|rule| rule.from.len())
+        else {
+            return Translation::Quarantine("no replacement rule covers this path".to_owned());
+        };
+        match &rule.to {
+            None => Translation::Drop,
+            Some(to) => {
+                let mut target = to.clone();
+                target.extend_from_slice(&path[rule.from.len()..]);
+                if target.as_slice() == path {
+                    Translation::Same
+                } else {
+                    Translation::To(target)
+                }
+            }
+        }
+    }
+}
+
 /// The translations of several consecutive remakes, applied in order: a node that
 /// rejoins several epochs later carries its writes through every remake in between.
 #[derive(Default, Clone)]
@@ -227,4 +388,85 @@ mod tests {
             Translation::Same
         );
     }
+
+    fn replace(spec: serde_json::Value) -> Result<Prepared, String> {
+        Replace.propose(&spec)
+    }
+
+    #[test]
+    fn replace_validates_the_spec() {
+        use serde_json::json;
+        assert!(replace(json!({ "document": [] })).is_err());
+        assert!(replace(json!({ "document": { META_ROOT: {} } })).is_err());
+        assert!(replace(json!({ "document": {}, "rules": [{ "from": "a" }] })).is_err());
+        assert!(replace(
+            json!({ "document": {}, "rules": [{ "from": "a", "to": "b", "drop": true }] })
+        )
+        .is_err());
+        assert!(
+            replace(json!({ "document": {}, "rules": [{ "from": "a//b", "to": "c" }] })).is_err()
+        );
+        assert!(
+            replace(json!({ "document": {}, "rules": [{ "from": "a", "to": META_ROOT }] }))
+                .is_err()
+        );
+        assert!(replace(json!({ "document": {}, "rules": [{ "from": "a", "drop": true }, { "from": ["a"], "to": "b" }] })).is_err());
+        assert!(replace(json!({ "document": {}, "extra": 1 })).is_err());
+        assert!(replace(json!({ "document": { "k": 1 } })).is_ok());
+    }
+
+    #[test]
+    fn replace_rules_translate_by_longest_prefix() {
+        use serde_json::json;
+        let prepared = replace(json!({
+            "document": {},
+            "rules": [
+                { "from": "users", "to": "accounts" },
+                { "from": "users/admin", "drop": true },
+                { "from": ["keep", "a/b"], "to": ["keep", "a/b"] },
+            ],
+        }))
+        .unwrap();
+        let translator = Translator::single(&Replace, prepared.translation);
+
+        assert_eq!(
+            translator.translate(&path(&["users", "alice", "name"])),
+            Translation::To(path(&["accounts", "alice", "name"]))
+        );
+        assert_eq!(
+            translator.translate(&path(&["users", "admin", "x"])),
+            Translation::Drop
+        );
+        assert_eq!(
+            translator.translate(&path(&["keep", "a/b", "c"])),
+            Translation::Same
+        );
+        assert!(matches!(
+            translator.translate(&path(&["settings"])),
+            Translation::Quarantine(_)
+        ));
+    }
+
+    #[test]
+    fn replace_builds_the_document_deterministically() {
+        use serde_json::json;
+        let build = |document: serde_json::Value| {
+            let prepared = replace(json!({ "document": document })).unwrap();
+            let mut target = AutoCommit::new().with_actor(automerge::ActorId::from(&b"fixed"[..]));
+            Replace
+                .build(&AutoCommit::new(), &prepared.content, &mut target)
+                .unwrap();
+            target.commit_with(automerge::transaction::CommitOptions::default().with_time(0));
+            (
+                target.get_heads(),
+                store::read_path(&target, &[]).unwrap().unwrap(),
+            )
+        };
+        // Same content, different key order.
+        let (heads_a, view_a) = build(json!({ "b": { "y": 2, "x": 1 }, "a": [1, 2] }));
+        let (heads_b, view_b) = build(json!({ "a": [1, 2], "b": { "x": 1, "y": 2 } }));
+        assert_eq!(heads_a, heads_b);
+        assert_eq!(view_a, view_b);
+        assert_eq!(view_a, json!({ "a": [1, 2], "b": { "x": 1, "y": 2 } }));
+    }
 }
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
index e31eadf..c1014d7 100644
--- a/src/remake/mod.rs
+++ b/src/remake/mod.rs
@@ -2,7 +2,7 @@
 //! Compaction is one: the new document has the old one's content without its history.
 //!
 //! * `kind`: what differs between kinds of remake (building the new document, and
-//!   where writes made after the base go); compaction is the only kind so far.
+//!   where writes made after the base go): `compact` and `replace`.
 //! * `snapshot`: deterministic build of the next epoch's initial document.
 //! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
 //! * `coordinator`: proposal, settle period and decision on the node that was asked.
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
index 2f2dfd6..507d8ac 100644
--- a/src/remake/snapshot.rs
+++ b/src/remake/snapshot.rs
@@ -336,7 +336,7 @@ fn snapshot_actor(remake_id: &str, generation: u64) -> ActorId {
 
 /// Rebuilds every object with sorted keys, so the result is the same whichever key
 /// order serde_json happens to use (it keeps insertion order with `preserve_order`).
-fn canonical(value: JsonValue) -> JsonValue {
+pub(super) fn canonical(value: JsonValue) -> JsonValue {
     match value {
         JsonValue::Object(map) => {
             let sorted: BTreeMap<String, JsonValue> = map