//! JSON view over the Automerge document with conflict-aware merging. use anyhow::anyhow; use automerge::{ ActorId, AutoCommit, ObjId, ObjType, Prop, ReadDoc, ScalarValue, Value, transaction::Transactable, }; use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; use serde::Serialize; use serde_json::Value as JsonValue; use std::collections::BTreeSet; /// Store paths touched by one batch of document changes (`AppState::change_events`). #[derive(Debug)] pub(crate) struct ChangeSet { pub(crate) paths: Vec>, } /// Whether one path is within the other (a change at either touches both). pub(crate) fn paths_related(a: &[String], b: &[String]) -> bool { a.starts_with(b) || b.starts_with(a) } #[derive(Debug, thiserror::Error)] pub(crate) enum StoreError { #[error("path crosses a non-map value")] Conflict, #[error("value does not exist")] NotFound, /// An insert where there is no list. #[error("not a list")] NotAList, /// An index past a list's end (`length` items). #[error("index {index} is past the list's end ({length} items)")] PastTheEnd { index: usize, length: usize }, #[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), /// 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], prop: impl Into + Clone, ) -> Result, ObjId)>, StoreError> { let mut all = Vec::new(); for parent in parents { for (value, id) in doc.get_all(parent, prop.clone())? { 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, StoreError> { resolve_prop(doc, parents, key, key, log) } /// The item at `index` of `list` (concurrent writes there resolved as for a key). fn resolve_item( doc: &AutoCommit, list: &ObjId, index: usize, log: Option<&mut ConflictLog>, ) -> Result, StoreError> { if index >= doc.length(list) { return Ok(None); } resolve_prop( doc, std::slice::from_ref(list), index, &index.to_string(), log, ) } /// What `segment` names within `current`: a key of merged maps, or the item of a list /// at the index it spells (`list_index`); None for anything else, and in a scalar. fn step( doc: &AutoCommit, current: &Resolved, segment: &str, log: Option<&mut ConflictLog>, ) -> Result, StoreError> { match current { Resolved::Maps(maps) => resolve(doc, maps, segment, log), Resolved::Other(Value::Object(ObjType::List), list) => match list_index(segment) { Some(index) => resolve_item(doc, list, index, log), None => Ok(None), }, Resolved::Other(..) => Ok(None), } } /// A list index as a path names it: decimal digits, no sign, no leading zeros. pub(crate) fn list_index(segment: &str) -> Option { let digits = !segment.is_empty() && segment.bytes().all(|byte| byte.is_ascii_digit()); if !digits || (segment.len() > 1 && segment.starts_with('0')) { return None; } segment.parse().ok() } /// A negative list index (`-1` the last item): how far from the end. fn from_the_end(segment: &str) -> Option { list_index(segment.strip_prefix('-')?).filter(|back| *back > 0) } /// `path` with its negative list indexes made positive (`-1` the last item; or, as the /// last segment with `insertion`, the end: where an insert appends), if it has any; /// None if it has none. An index before the start names nothing (`NotFound`). Only /// segments within a list count: in a map, `-1` is a key like any other. pub(crate) fn positive_path( doc: &AutoCommit, path: &[String], insertion: bool, ) -> Result>, StoreError> { let mut current = Resolved::Maps(vec![automerge::ROOT]); let mut positive = path.to_vec(); let mut changed = false; for (depth, segment) in path.iter().enumerate() { let mut segment = segment.clone(); if let Resolved::Other(Value::Object(ObjType::List), list) = ¤t && let Some(back) = from_the_end(&segment) { let last = depth + 1 == path.len(); let length = doc.length(list) + usize::from(insertion && last); let index = length.checked_sub(back).ok_or(StoreError::NotFound)?; segment = index.to_string(); positive[depth] = segment.clone(); changed = true; } match step(doc, ¤t, &segment, None)? { Some(next) => current = next, // Nothing further to make positive. None => break, } } Ok(changed.then_some(positive)) } /// Resolves `key` (a map key or a list index) in `parents`; `name` is how conflict /// reports name it. fn resolve_prop( doc: &AutoCommit, parents: &[ObjId], prop: impl Into + Clone, name: &str, log: Option<&mut ConflictLog>, ) -> Result, StoreError> { let key = name; let mut all = candidates(doc, parents, prop)?; if all.is_empty() { return Ok(None); } let merged = matches!(all[0].0, Value::Object(ObjType::Map)); if all.len() > 1 && 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 { maps.iter().flat_map(|map| doc.keys(map)).collect() } fn resolved_to_json( doc: &AutoCommit, resolved: &Resolved, mut log: Option<&mut ConflictLog>, ) -> Result { 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 { 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, StoreError> { let Some(current) = resolve_logged(doc, path, log.as_deref_mut())? else { return Ok(None); }; Ok(Some(resolved_to_json(doc, ¤t, log)?)) } /// Reads a path; the empty path is the whole store. pub(crate) fn read_path( doc: &AutoCommit, path: &[String], ) -> Result, StoreError> { read_path_inner(doc, path, None) } pub(crate) fn read_path_with_conflicts( doc: &AutoCommit, path: &[String], ) -> Result)>, 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(crate) fn read_path_traced( doc: &AutoCommit, path: &[String], ) -> Result<(Option, bool), StoreError> { let mut log = ConflictLog::default(); let value = read_path_inner(doc, path, Some(&mut log))?; Ok((value, !log.reports.is_empty())) } /// Strings and bytes up to this long are in a shallow read's summaries. const SHALLOW_INLINE: usize = 256; /// What's at `path` (the empty path is the whole store), one level deep: a summary of /// it, `type` (`map`, `list`, `text`, `string`, `bytes`, `number`, `boolean`, `null`, /// or `unknown`, an Automerge scalar of a later version) /// and `size` (keys, items, characters or bytes), and its `value` where it's a short /// scalar; a map's has its keys' summaries too, under `entries`. Concurrent values /// resolved as a read resolves them; conflicts not reported. pub(crate) fn read_shallow( doc: &AutoCommit, path: &[String], ) -> Result, StoreError> { let Some(resolved) = resolve_through(doc, path)? else { return Ok(None); }; summary(doc, &resolved, true).map(Some) } /// The resolved value at `path` (the empty path is the root map), through maps and /// lists (`step`). fn resolve_through(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { resolve_logged(doc, path, None) } /// `resolve_through`, noting the conflicts on the way in `log` (as ancestors, but the /// last segment's: the requested value's), its path left at `path`. fn resolve_logged( doc: &AutoCommit, path: &[String], mut log: Option<&mut ConflictLog>, ) -> Result, StoreError> { let mut current = Resolved::Maps(vec![automerge::ROOT]); for (index, segment) in path.iter().enumerate() { 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 step(doc, ¤t, 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 { log.in_ancestors = false; } Ok(Some(current)) } /// A summary of `resolved` (`read_shallow`); with `entries`, a map's keys' too, or a /// list's items' (by index). fn summary(doc: &AutoCommit, resolved: &Resolved, entries: bool) -> Result { let mut out = serde_json::Map::new(); match resolved { Resolved::Maps(maps) => { let keys = merged_keys(doc, maps); out.insert("type".into(), "map".into()); out.insert("size".into(), keys.len().into()); if entries { let mut listed = serde_json::Map::new(); for key in keys { if let Some(child) = resolve(doc, maps, &key, None)? { listed.insert(key, summary(doc, &child, false)?); } } out.insert("entries".into(), JsonValue::Object(listed)); } } Resolved::Other(Value::Object(ObjType::Map), id) => { return summary(doc, &Resolved::Maps(vec![id.clone()]), entries); } Resolved::Other(Value::Object(ObjType::List), id) => { let length = doc.length(id); out.insert("type".into(), "list".into()); out.insert("size".into(), length.into()); if entries { let mut listed = serde_json::Map::new(); for index in 0..length { if let Some(item) = resolve_item(doc, id, index, None)? { listed.insert(index.to_string(), summary(doc, &item, false)?); } } out.insert("entries".into(), JsonValue::Object(listed)); } } Resolved::Other(Value::Object(ObjType::Text), id) => { out.insert("type".into(), "text".into()); out.insert("size".into(), doc.length(id).into()); } Resolved::Other(Value::Object(ObjType::Table), _) => { return Err(anyhow!("Automerge tables are not supported").into()); } Resolved::Other(Value::Scalar(scalar), _) => { let (kind, size) = match &**scalar { ScalarValue::Str(text) => ("string", Some(text.len())), ScalarValue::Bytes(bytes) => ("bytes", Some(bytes.len())), ScalarValue::Null => ("null", None), ScalarValue::Boolean(_) => ("boolean", None), ScalarValue::Unknown { .. } => ("unknown", None), _ => ("number", None), }; out.insert("type".into(), kind.into()); if let Some(size) = size { out.insert("size".into(), size.into()); } if kind != "unknown" && size.is_none_or(|size| size <= SHALLOW_INLINE) { out.insert("value".into(), scalar_to_json(scalar.clone().into_owned())?); } } } Ok(JsonValue::Object(out)) } /// Keys of the merged map at `path`; empty if it doesn't exist or isn't a map. pub(crate) fn keys_at(doc: &AutoCommit, path: &[String]) -> Result, StoreError> { Ok(match resolve_through(doc, path)? { Some(Resolved::Maps(maps)) => merged_keys(doc, &maps).into_iter().collect(), _ => Vec::new(), }) } /// Every concurrent value at `path`, newest first; empty if there is none. Where /// `read_path` picks one (or merges concurrent maps into one), this keeps them apart. pub(crate) fn read_candidates( doc: &AutoCommit, path: &[String], ) -> Result, StoreError> { let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; let Some(Resolved::Maps(maps)) = resolve_through(doc, parents)? else { return Ok(Vec::new()); }; candidates(doc, &maps, leaf.as_str())? .into_iter() .map(|(value, id)| match value { Value::Scalar(scalar) => scalar_to_json(scalar.into_owned()), Value::Object(_) => object_to_json(doc, &id, None), }) .collect() } /// Writes (replaces) the value at `path`, creating intermediate maps. Within a list, /// the segment is an index (`list_index`): an item there is replaced, one at the end /// appended; past the end, or anything but an index, names nothing (`NotFound`). pub(crate) fn write_path( doc: &mut AutoCommit, path: &[String], value: &JsonValue, ) -> Result<(), StoreError> { let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; match parent_for_write(doc, parents)? { Parent::Maps(maps) => { // 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)?; } Parent::List(list) => { let index = list_index(leaf).ok_or(StoreError::NotFound)?; let length = doc.length(&list); if index > length { return Err(StoreError::NotFound); } if index < length { doc.delete(&list, index)?; } write_value(doc, &list, Slot::Index(index), value)?; } } Ok(()) } /// What a write's last segment goes into. enum Parent { /// Merged maps (`[0]` takes new keys). Maps(Vec), List(ObjId), } /// The parent of a write at `parents` + one more segment: missing maps are created on /// the way; through a list only by an index of an item it has. A scalar or text on the /// way is a `Conflict`. fn parent_for_write(doc: &mut AutoCommit, parents: &[String]) -> Result { let mut current = Parent::Maps(vec![automerge::ROOT]); for segment in parents { let next = match ¤t { Parent::Maps(maps) => match resolve(doc, maps, segment, None)? { // A concurrent creation elsewhere is merged on read. None => Resolved::Maps(vec![doc.put_object( &maps[0], segment.as_str(), ObjType::Map, )?]), Some(next) => next, }, Parent::List(list) => { let index = list_index(segment).ok_or(StoreError::NotFound)?; resolve_item(doc, list, index, None)?.ok_or(StoreError::NotFound)? } }; current = match next { Resolved::Maps(maps) => Parent::Maps(maps), Resolved::Other(Value::Object(ObjType::List), list) => Parent::List(list), Resolved::Other(..) => return Err(StoreError::Conflict), }; } Ok(current) } /// Inserts `value` into a list: `path` ends in the index it goes to (the items from /// there on move up one; the list's length appends). An index past the end is /// `PastTheEnd`; a parent that isn't a list `NotAList`; anything but an index there, or /// no parent, `NotFound`. pub(crate) fn insert_path( doc: &mut AutoCommit, path: &[String], value: &JsonValue, ) -> Result<(), StoreError> { let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; let list = match resolve_through(doc, parents)? { None => return Err(StoreError::NotFound), Some(Resolved::Other(Value::Object(ObjType::List), id)) => id, Some(_) => return Err(StoreError::NotAList), }; let index = list_index(leaf).ok_or(StoreError::NotFound)?; let length = doc.length(&list); if index > length { return Err(StoreError::PastTheEnd { index, length }); } write_value(doc, &list, Slot::Index(index), value)?; Ok(()) } /// 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> { copy_path_to(src, path, dst, path) } /// Like [`copy_path`], but puts what `src` holds at `from` at `to` in `dst`. pub(crate) fn copy_path_to( src: &AutoCommit, from: &[String], dst: &mut AutoCommit, to: &[String], ) -> Result<(), StoreError> { let (leaf, parents) = to.split_last().ok_or(StoreError::NotFound)?; let Some(resolved) = resolve_through(src, from)? else { return match delete_path(dst, to, false) { Ok(()) | Err(StoreError::NotFound) => Ok(()), Err(error) => Err(error), }; }; // Into a map: what is copied are map entries (delta replay names nothing inside a // list: a list is copied whole). let Parent::Maps(maps) = parent_for_write(dst, parents)? else { return Err(StoreError::Conflict); }; delete_in_all(dst, &maps, leaf)?; let mut trace = from.to_vec(); copy_resolved( src, &resolved, dst, &maps[0], Slot::Key(leaf), &mut trace, &|_| false, ) } /// Deletes the value at `path`, through lists too, where a segment is an index /// (`list_index`; the items after it move down one). With `prune`, the maps above it /// that it leaves empty go too, up to (not including) the root or a list. pub(crate) fn delete_path( doc: &mut AutoCommit, path: &[String], prune: bool, ) -> Result<(), StoreError> { let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; // levels[d] = what path[..d] resolves to let mut levels = vec![Resolved::Maps(vec![automerge::ROOT])]; for segment in parents { let next = step(doc, levels.last().expect("non-empty"), segment, None)? .ok_or(StoreError::NotFound)?; levels.push(next); } match levels.last().expect("non-empty") { Resolved::Maps(maps) => { if resolve(doc, maps, leaf, None)?.is_none() { return Err(StoreError::NotFound); } delete_in_all(doc, maps, leaf)?; } Resolved::Other(Value::Object(ObjType::List), list) => { let index = list_index(leaf).ok_or(StoreError::NotFound)?; if index >= doc.length(list) { return Err(StoreError::NotFound); } doc.delete(list, index)?; return Ok(()); } Resolved::Other(..) => return Err(StoreError::NotFound), } if prune { // Never prune the root (depth 0), nor through a list. for depth in (1..levels.len()).rev() { let (Resolved::Maps(above), Resolved::Maps(maps)) = (&levels[depth - 1], &levels[depth]) else { break; }; if !merged_keys(doc, maps).is_empty() { break; } delete_in_all(doc, above, &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, 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) && 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, 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, 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 { 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, ) -> 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 { 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(crate) enum Resolution { MergedMaps, NewestWins, } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct CandidateReport { pub(crate) op_id: String, pub(crate) selected: bool, #[serde(skip_serializing_if = "Option::is_none")] pub(crate) value: Option, } #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct ConflictReport { pub(crate) path: Vec, /// On the way to the requested path; reported without candidate values. #[serde(skip_serializing_if = "std::ops::Not::not")] pub(crate) ancestor: bool, pub(crate) resolution: Resolution, pub(crate) candidates: Vec, } #[derive(Default)] struct ConflictLog { path: Vec, reports: Vec, /// 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 { 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::, 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( 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 } #[cfg(test)] mod tests { use super::*; use serde_json::json; fn path(at: &str) -> Vec { at.split('/').map(str::to_owned).collect() } #[test] fn paths_go_through_lists() { let mut doc = AutoCommit::new(); write_path(&mut doc, &path("list"), &json!(["b", {"k": 1}, "d"])).unwrap(); assert_eq!(read_path(&doc, &path("list/0")).unwrap(), Some(json!("b"))); assert_eq!(read_path(&doc, &path("list/1/k")).unwrap(), Some(json!(1))); // Not an index, or past the end: nothing. for missing in ["list/3", "list/x", "list/01", "list/-1", "list/1/k/0"] { assert_eq!(read_path(&doc, &path(missing)).unwrap(), None, "{missing}"); } // Replaced, and appended at the end; past it, nothing. write_path(&mut doc, &path("list/0"), &json!("a")).unwrap(); write_path(&mut doc, &path("list/3"), &json!("e")).unwrap(); write_path(&mut doc, &path("list/1/k"), &json!(2)).unwrap(); assert!(matches!( write_path(&mut doc, &path("list/9"), &json!(0)), Err(StoreError::NotFound) )); assert_eq!( read_path(&doc, &path("list")).unwrap(), Some(json!(["a", {"k": 2}, "d", "e"])) ); delete_path(&mut doc, &path("list/2"), false).unwrap(); assert_eq!( read_path(&doc, &path("list")).unwrap(), Some(json!(["a", {"k": 2}, "e"])) ); assert_eq!( read_shallow(&doc, &path("list")).unwrap().unwrap()["entries"]["1"]["type"], json!("map") ); } #[test] fn negative_indexes_count_from_the_end() { let mut doc = AutoCommit::new(); write_path(&mut doc, &path("list"), &json!(["a", "b"])).unwrap(); write_path(&mut doc, &path("map"), &json!({"-1": "a key"})).unwrap(); assert_eq!( positive_path(&doc, &path("list/-1"), false).unwrap(), Some(path("list/1")) ); // Inserting: -1 is the end. assert_eq!( positive_path(&doc, &path("list/-1"), true).unwrap(), Some(path("list/2")) ); assert!(matches!( positive_path(&doc, &path("list/-3"), false), Err(StoreError::NotFound) )); assert_eq!(positive_path(&doc, &path("map/-1"), false).unwrap(), None); assert_eq!(positive_path(&doc, &path("list/0"), false).unwrap(), None); } #[test] fn inserts_into_lists() { let mut doc = AutoCommit::new(); write_path(&mut doc, &path("list"), &json!(["b", "d"])).unwrap(); insert_path(&mut doc, &path("list/0"), &json!("a")).unwrap(); insert_path(&mut doc, &path("list/2"), &json!("c")).unwrap(); insert_path(&mut doc, &path("list/4"), &json!({"e": 5})).unwrap(); assert_eq!( read_path(&doc, &path("list")).unwrap(), Some(json!(["a", "b", "c", "d", {"e": 5}])) ); assert!(matches!( insert_path(&mut doc, &path("list/6"), &json!(1)), Err(StoreError::PastTheEnd { index: 6, length: 5 }) )); write_path(&mut doc, &path("map"), &json!({"k": 1})).unwrap(); assert!(matches!( insert_path(&mut doc, &path("map/0"), &json!(1)), Err(StoreError::NotAList) )); assert!(matches!( insert_path(&mut doc, &path("nothing/0"), &json!(1)), Err(StoreError::NotFound) )); } }