diff options
Diffstat (limited to 'src/store.rs')
| -rw-r--r-- | src/store.rs | 696 |
1 files changed, 0 insertions, 696 deletions
diff --git a/src/store.rs b/src/store.rs deleted file mode 100644 index 1815635..0000000 --- a/src/store.rs +++ /dev/null @@ -1,696 +0,0 @@ -//! JSON view over the Automerge document with conflict-aware merging. - -use anyhow::anyhow; -use automerge::{ - transaction::Transactable, ActorId, AutoCommit, ObjId, ObjType, ReadDoc, ScalarValue, Value, -}; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; -use serde::Serialize; -use serde_json::Value as JsonValue; -use std::collections::BTreeSet; - -#[derive(Debug, thiserror::Error)] -pub enum StoreError { - #[error("path crosses a non-map value")] - Conflict, - #[error("value does not exist")] - NotFound, - #[error(transparent)] - Doc(#[from] automerge::AutomergeError), - #[error(transparent)] - Other(#[from] anyhow::Error), -} - -/// Automerge's total order on operations: Lamport counter, then actor. -fn op_order(id: &ObjId) -> (u64, Option<&ActorId>) { - match id { - ObjId::Root => (0, None), - ObjId::Id(counter, actor, _) => (*counter, Some(actor)), - } -} - -#[derive(Debug, Clone)] -enum Resolved { - /// Concurrent maps presented as one. Newest first; `[0]` is canonical for writes. - Maps(Vec<ObjId>), - /// Any other value: the newest concurrent candidate. - Other(Value<'static>, ObjId), -} - -/// All concurrent values of `key` across a merged set of parent maps, newest first. -fn candidates( - doc: &AutoCommit, - parents: &[ObjId], - key: &str, -) -> Result<Vec<(Value<'static>, ObjId)>, StoreError> { - let mut all = Vec::new(); - for parent in parents { - for (value, id) in doc.get_all(parent, key)? { - all.push((value.into_owned(), id)); - } - } - all.sort_by(|a, b| op_order(&b.1).cmp(&op_order(&a.1))); - Ok(all) -} - -fn resolve( - doc: &AutoCommit, - parents: &[ObjId], - key: &str, - log: Option<&mut ConflictLog>, -) -> Result<Option<Resolved>, StoreError> { - let mut all = candidates(doc, parents, key)?; - if all.is_empty() { - return Ok(None); - } - - let merged = matches!(all[0].0, Value::Object(ObjType::Map)); - - if all.len() > 1 { - if let Some(log) = log { - log.record(doc, key, &all, merged)?; - } - } - - if merged { - let maps = all - .into_iter() - .filter(|(value, _)| matches!(value, Value::Object(ObjType::Map))) - .map(|(_, id)| id) - .collect(); - Ok(Some(Resolved::Maps(maps))) - } else { - let (value, id) = all.swap_remove(0); - Ok(Some(Resolved::Other(value, id))) - } -} - -fn merged_keys(doc: &AutoCommit, maps: &[ObjId]) -> BTreeSet<String> { - maps.iter().flat_map(|map| doc.keys(map)).collect() -} - -fn resolved_to_json( - doc: &AutoCommit, - resolved: &Resolved, - mut log: Option<&mut ConflictLog>, -) -> Result<JsonValue, StoreError> { - let maps = match resolved { - Resolved::Maps(maps) => maps, - Resolved::Other(Value::Scalar(scalar), _) => { - return scalar_to_json(scalar.clone().into_owned()) - } - Resolved::Other(Value::Object(_), id) => return object_to_json(doc, id, log), - }; - - let mut out = serde_json::Map::new(); - - if let [map] = maps.as_slice() { - // Fast path: one pass; get_all only for keys Automerge flags as conflicted. - for item in doc.map_range(map, ..) { - let key = item.key.to_string(); - - let child = if item.conflict { - match resolve(doc, maps, &key, log.as_deref_mut())? { - Some(child) => child, - None => continue, - } - } else { - let id = item.id(); - match Value::from(item.value) { - Value::Object(ObjType::Map) => Resolved::Maps(vec![id]), - value => Resolved::Other(value.into_owned(), id), - } - }; - - let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; - out.insert(key, json); - } - } else { - // Several concurrent maps: union of their keys. - for key in merged_keys(doc, maps) { - let Some(child) = resolve(doc, maps, &key, log.as_deref_mut())? else { - continue; - }; - let json = nested(&mut log, &key, |log| resolved_to_json(doc, &child, log))?; - out.insert(key, json); - } - } - - Ok(JsonValue::Object(out)) -} - -fn object_to_json( - doc: &AutoCommit, - id: &ObjId, - mut log: Option<&mut ConflictLog>, -) -> Result<JsonValue, StoreError> { - match doc.object_type(id)? { - ObjType::Map => resolved_to_json(doc, &Resolved::Maps(vec![id.clone()]), log), - ObjType::List => { - let mut out = Vec::new(); - for (index, item) in doc.list_range(id, ..).enumerate() { - let child_id = item.id(); - out.push(match Value::from(item.value) { - Value::Object(_) => nested(&mut log, &index.to_string(), |log| { - object_to_json(doc, &child_id, log) - })?, - Value::Scalar(scalar) => scalar_to_json(scalar.into_owned())?, - }); - } - Ok(JsonValue::Array(out)) - } - ObjType::Text => Ok(JsonValue::String(doc.text(id)?)), - ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()), - } -} - -fn delete_in_all(doc: &mut AutoCommit, maps: &[ObjId], key: &str) -> Result<(), StoreError> { - for map in maps { - if doc.get(map, key)?.is_some() { - doc.delete(map, key)?; - } - } - Ok(()) -} - -fn read_path_inner( - doc: &AutoCommit, - path: &[String], - mut log: Option<&mut ConflictLog>, -) -> Result<Option<JsonValue>, StoreError> { - let mut current = Resolved::Maps(vec![automerge::ROOT]); - - for (index, segment) in path.iter().enumerate() { - let Resolved::Maps(maps) = ¤t else { - return Ok(None); - }; - if let Some(log) = log.as_deref_mut() { - // The last segment is the requested value itself, not an ancestor. - log.in_ancestors = index + 1 < path.len(); - } - match resolve(doc, maps, segment, log.as_deref_mut())? { - Some(next) => current = next, - None => return Ok(None), - } - if let Some(log) = log.as_deref_mut() { - log.path.push(segment.clone()); - } - } - - if let Some(log) = log.as_deref_mut() { - log.in_ancestors = false; - } - Ok(Some(resolved_to_json(doc, ¤t, log)?)) -} - -/// Reads a path; the empty path is the whole store. -pub fn read_path(doc: &AutoCommit, path: &[String]) -> Result<Option<JsonValue>, StoreError> { - read_path_inner(doc, path, None) -} - -pub fn read_path_with_conflicts( - doc: &AutoCommit, - path: &[String], -) -> Result<Option<(JsonValue, Vec<ConflictReport>)>, StoreError> { - let mut log = ConflictLog { - want_values: true, - ..Default::default() - }; - let value = read_path_inner(doc, path, Some(&mut log))?; - Ok(value.map(|value| (value, log.reports))) -} - -/// For streams: the value, plus whether the path or subtree involves any conflict. -/// Reports are collected without values, so this costs little more than read_path. -pub fn read_path_traced( - doc: &AutoCommit, - path: &[String], -) -> Result<(Option<JsonValue>, bool), StoreError> { - let mut log = ConflictLog::default(); - let value = read_path_inner(doc, path, Some(&mut log))?; - Ok((value, !log.reports.is_empty())) -} - -/// Keys of the merged map at `path`; empty if it doesn't exist or isn't a map. -pub fn keys_at(doc: &AutoCommit, path: &[String]) -> Result<Vec<String>, StoreError> { - let mut maps = vec![automerge::ROOT]; - for segment in path { - maps = match resolve(doc, &maps, segment, None)? { - Some(Resolved::Maps(next)) => next, - _ => return Ok(Vec::new()), - }; - } - Ok(merged_keys(doc, &maps).into_iter().collect()) -} - -/// Writes (replaces) the value at `path`, creating intermediate maps. -pub fn write_path( - doc: &mut AutoCommit, - path: &[String], - value: &JsonValue, -) -> Result<(), StoreError> { - let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; - let maps = ensure_maps(doc, parents)?; - - // Replace semantics: clear the key in every merged parent, or an old map - // would merge into the new value. - delete_in_all(doc, &maps, leaf)?; - json_to_automerge(doc, &maps[0], leaf, value)?; - Ok(()) -} - -/// The merged maps at `path`, creating missing maps on the way. -fn ensure_maps(doc: &mut AutoCommit, path: &[String]) -> Result<Vec<ObjId>, StoreError> { - let mut maps = vec![automerge::ROOT]; - for segment in path { - maps = match resolve(doc, &maps, segment, None)? { - Some(Resolved::Maps(next)) => next, - Some(Resolved::Other(..)) => return Err(StoreError::Conflict), - // A concurrent creation elsewhere is merged on read. - None => vec![doc.put_object(&maps[0], segment.as_str(), ObjType::Map)?], - }; - } - Ok(maps) -} - -/// The resolved value at `path` (the empty path is the root map). -fn resolve_at(doc: &AutoCommit, path: &[String]) -> Result<Option<Resolved>, StoreError> { - let mut current = Resolved::Maps(vec![automerge::ROOT]); - for segment in path { - let Resolved::Maps(maps) = ¤t else { - return Ok(None); - }; - match resolve(doc, maps, segment, None)? { - Some(next) => current = next, - None => return Ok(None), - } - } - Ok(Some(current)) -} - -/// Makes `dst` hold at `path` what `src`'s merged view holds there: a type-preserving -/// copy (see `materialize_into`) if it exists, a delete otherwise. Replaces the whole -/// subtree at `path`; the path must not be empty. -pub(crate) fn copy_path( - src: &AutoCommit, - dst: &mut AutoCommit, - path: &[String], -) -> Result<(), StoreError> { - let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; - - let Some(resolved) = resolve_at(src, path)? else { - return match delete_path(dst, path, false) { - Ok(()) | Err(StoreError::NotFound) => Ok(()), - Err(error) => Err(error), - }; - }; - - let maps = ensure_maps(dst, parents)?; - delete_in_all(dst, &maps, leaf)?; - let mut trace = path.to_vec(); - copy_resolved( - src, - &resolved, - dst, - &maps[0], - Slot::Key(leaf), - &mut trace, - &|_| false, - ) -} - -/// Deletes the value at `path`, optionally pruning ancestors that become empty. -pub fn delete_path(doc: &mut AutoCommit, path: &[String], prune: bool) -> Result<(), StoreError> { - let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; - - // levels[d] = merged maps at path[..d] - let mut levels = vec![vec![automerge::ROOT]]; - for segment in parents { - let next = match resolve(doc, levels.last().expect("non-empty"), segment, None)? { - Some(Resolved::Maps(maps)) => maps, - _ => return Err(StoreError::NotFound), - }; - levels.push(next); - } - - let maps = levels.last().expect("non-empty"); - if resolve(doc, maps, leaf, None)?.is_none() { - return Err(StoreError::NotFound); - } - delete_in_all(doc, maps, leaf)?; - - if prune { - // Never prune the root (depth 0). - for depth in (1..levels.len()).rev() { - if !merged_keys(doc, &levels[depth]).is_empty() { - break; - } - delete_in_all(doc, &levels[depth - 1], &path[depth - 1])?; - } - } - - Ok(()) -} - -// --- materialization --------------------------------------------------------- - -/// Copies the merged view of `src` into `dst` as fresh objects. -/// -/// * Conflicts collapse exactly as reads resolve them: concurrent maps merge, everything -/// else keeps the newest candidate. -/// * Automerge types survive: text stays text, counters stay counters (normalized to -/// their current value), bytes, timestamps and floats are copied bit-exact. -/// * Map keys are visited in sorted order and list elements in order, so two documents -/// with the same view produce the same operations in the same order. -/// -/// `skip` sees the path of every map entry and list element (list indices as decimal -/// strings); returning true leaves that subtree out. -pub(crate) fn materialize_into( - src: &AutoCommit, - dst: &mut AutoCommit, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - let mut path = Vec::new(); - copy_maps( - src, - &[automerge::ROOT], - dst, - &automerge::ROOT, - &mut path, - skip, - ) -} - -fn copy_maps( - src: &AutoCommit, - maps: &[ObjId], - dst: &mut AutoCommit, - target: &ObjId, - path: &mut Vec<String>, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - // merged_keys is a BTreeSet: sorted, so the operation order is deterministic. - for key in merged_keys(src, maps) { - path.push(key.clone()); - if !skip(path) { - if let Some(resolved) = resolve(src, maps, &key, None)? { - copy_resolved(src, &resolved, dst, target, Slot::Key(&key), path, skip)?; - } - } - path.pop(); - } - Ok(()) -} - -fn copy_resolved( - src: &AutoCommit, - resolved: &Resolved, - dst: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - path: &mut Vec<String>, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - match resolved { - Resolved::Maps(maps) => { - let id = put_object(dst, parent, slot, ObjType::Map)?; - copy_maps(src, maps, dst, &id, path, skip) - } - Resolved::Other(Value::Scalar(scalar), _) => { - let scalar = match scalar.as_ref() { - // A read counter carries its increment history; copy only its value, so - // equal counters produce equal operations. - ScalarValue::Counter(counter) => ScalarValue::counter(i64::from(counter.clone())), - other => other.clone(), - }; - put_scalar(dst, parent, slot, scalar) - } - Resolved::Other(Value::Object(obj_type), id) => { - copy_object(src, *obj_type, id, dst, parent, slot, path, skip) - } - } -} - -#[allow(clippy::too_many_arguments)] -fn copy_object( - src: &AutoCommit, - obj_type: ObjType, - id: &ObjId, - dst: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - path: &mut Vec<String>, - skip: &dyn Fn(&[String]) -> bool, -) -> Result<(), StoreError> { - match obj_type { - ObjType::Map => copy_resolved( - src, - &Resolved::Maps(vec![id.clone()]), - dst, - parent, - slot, - path, - skip, - ), - ObjType::List => { - let list = put_object(dst, parent, slot, ObjType::List)?; - // Skipped elements leave no gap in the copy. - let mut position = 0; - for (index, item) in src.list_range(id, ..).enumerate() { - path.push(index.to_string()); - if !skip(path) { - let child_id = item.id(); - let child = match Value::from(item.value).into_owned() { - Value::Object(ObjType::Map) => Resolved::Maps(vec![child_id]), - value => Resolved::Other(value, child_id), - }; - copy_resolved(src, &child, dst, &list, Slot::Index(position), path, skip)?; - position += 1; - } - path.pop(); - } - Ok(()) - } - ObjType::Text => { - let text = put_object(dst, parent, slot, ObjType::Text)?; - dst.splice_text(&text, 0, 0, &src.text(id)?)?; - Ok(()) - } - ObjType::Table => Err(anyhow!("Automerge tables are not supported").into()), - } -} - -/// Where a value is placed inside its parent object. -#[derive(Clone, Copy)] -enum Slot<'a> { - /// A key in a map. - Key(&'a str), - /// An index in a list. The value is inserted here, not overwritten. - Index(usize), -} - -fn put_object( - doc: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - obj_type: ObjType, -) -> Result<ObjId, StoreError> { - Ok(match slot { - Slot::Key(key) => doc.put_object(parent, key, obj_type)?, - Slot::Index(index) => doc.insert_object(parent, index, obj_type)?, - }) -} - -fn put_scalar( - doc: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - value: impl Into<ScalarValue>, -) -> Result<(), StoreError> { - match slot { - Slot::Key(key) => doc.put(parent, key, value)?, - Slot::Index(index) => doc.insert(parent, index, value)?, - } - Ok(()) -} - -fn write_value( - doc: &mut AutoCommit, - parent: &ObjId, - slot: Slot, - value: &JsonValue, -) -> Result<(), StoreError> { - match value { - JsonValue::Object(object) => { - let id = put_object(doc, parent, slot, ObjType::Map)?; - for (child_key, child_value) in object { - write_value(doc, &id, Slot::Key(child_key), child_value)?; - } - } - JsonValue::Array(items) => { - let id = put_object(doc, parent, slot, ObjType::List)?; - for (index, item) in items.iter().enumerate() { - write_value(doc, &id, Slot::Index(index), item)?; - } - } - JsonValue::String(value) => put_scalar(doc, parent, slot, value.as_str())?, - JsonValue::Bool(value) => put_scalar(doc, parent, slot, *value)?, - JsonValue::Number(value) => { - if let Some(value) = value.as_i64() { - put_scalar(doc, parent, slot, value)?; - } else if let Some(value) = value.as_u64() { - put_scalar(doc, parent, slot, value)?; - } else if let Some(value) = value.as_f64() { - put_scalar(doc, parent, slot, value)?; - } else { - return Err(anyhow!("Unsupported JSON number").into()); - } - } - JsonValue::Null => put_scalar(doc, parent, slot, ScalarValue::Null)?, - } - - Ok(()) -} - -fn json_to_automerge( - doc: &mut AutoCommit, - parent: &ObjId, - key: &str, - value: &JsonValue, -) -> Result<(), StoreError> { - write_value(doc, parent, Slot::Key(key), value) -} - -fn scalar_to_json(scalar: ScalarValue) -> Result<JsonValue, StoreError> { - match scalar { - ScalarValue::Null => Ok(JsonValue::Null), - ScalarValue::Boolean(b) => Ok(JsonValue::Bool(b)), - ScalarValue::Int(i) => Ok(JsonValue::Number(i.into())), - ScalarValue::Uint(u) => Ok(JsonValue::Number(u.into())), - ScalarValue::F64(f) => serde_json::Number::from_f64(f) - .map(JsonValue::Number) - .ok_or(anyhow!("Non-finite float cannot be represented as JSON").into()), - ScalarValue::Str(s) => Ok(JsonValue::String(s.to_string())), - ScalarValue::Bytes(b) => Ok(JsonValue::String(BASE64.encode(b))), - ScalarValue::Counter(c) => Ok(JsonValue::Number(i64::from(c).into())), - ScalarValue::Timestamp(t) => Ok(JsonValue::Number(t.into())), - ScalarValue::Unknown { .. } => Err(anyhow!("Unsupported Automerge scalar type").into()), - } -} - -// --- conflict log ------------------------------------------------------------ - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub enum Resolution { - MergedMaps, - NewestWins, -} - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct CandidateReport { - pub op_id: String, - pub selected: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub value: Option<JsonValue>, -} - -#[derive(Debug, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct ConflictReport { - pub path: Vec<String>, - /// On the way to the requested path; reported without candidate values. - #[serde(skip_serializing_if = "std::ops::Not::not")] - pub ancestor: bool, - pub resolution: Resolution, - pub candidates: Vec<CandidateReport>, -} - -#[derive(Default)] -struct ConflictLog { - path: Vec<String>, - reports: Vec<ConflictReport>, - /// Render candidate values (only within the requested subtree). - want_values: bool, - /// Currently walking the ancestors of the requested path. - in_ancestors: bool, -} - -fn op_id_string(id: &ObjId) -> String { - match id { - ObjId::Root => "_root".to_owned(), - ObjId::Id(counter, actor, _) => format!("{counter}@{actor}"), - } -} - -fn value_to_json(doc: &AutoCommit, value: &Value<'_>, id: &ObjId) -> Result<JsonValue, StoreError> { - match value { - Value::Scalar(scalar) => Ok(scalar_to_json(scalar.clone().into_owned())?), - Value::Object(_) => object_to_json(doc, id, None), - } -} - -impl ConflictLog { - fn record( - &mut self, - doc: &AutoCommit, - key: &str, - all: &[(Value<'static>, ObjId)], - merged: bool, - ) -> Result<(), StoreError> { - let with_values = self.want_values && !self.in_ancestors; - - let candidates = all - .iter() - .enumerate() - .map(|(index, (value, id))| { - let selected = if merged { - matches!(value, Value::Object(ObjType::Map)) - } else { - index == 0 - }; - Ok(CandidateReport { - op_id: op_id_string(id), - selected, - value: if with_values { - Some(value_to_json(doc, value, id)?) - } else { - None - }, - }) - }) - .collect::<Result<Vec<_>, StoreError>>()?; - - let mut path = self.path.clone(); - path.push(key.to_owned()); - - self.reports.push(ConflictReport { - path, - ancestor: self.in_ancestors, - resolution: if merged { - Resolution::MergedMaps - } else { - Resolution::NewestWins - }, - candidates, - }); - Ok(()) - } -} - -/// Runs `f` with `segment` appended to the conflict path. -fn nested<T>( - log: &mut Option<&mut ConflictLog>, - segment: &str, - f: impl FnOnce(Option<&mut ConflictLog>) -> T, -) -> T { - if let Some(log) = log.as_deref_mut() { - log.path.push(segment.to_owned()); - } - let out = f(log.as_deref_mut()); - if let Some(log) = log.as_deref_mut() { - log.path.pop(); - } - out -} |