path: root/src/document/store.rs

//! 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<Vec<String>>,
}

/// How `changed` names a change inside a list.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum InLists {
    /// By the item's index: `servers/0/name` (the change publisher).
    Index,
    /// By the list: `servers` (remakes replay a list whole).
    Whole,
}

/// A store path changes touched, and whether concurrent writes left a conflict there.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct Changed {
    pub(crate) path: Vec<String>,
    pub(crate) conflict: bool,
}

/// The store paths the changes from `before` to `after` touched, as the document
/// reports them (a path may come more than once). A map entry by its key, a counter
/// incremented in a map by its key, and list and text edits by the containing object;
/// inside a list as `lists` says.
pub(crate) fn changed(
    doc: &mut AutoCommit,
    before: &[automerge::ChangeHash],
    after: &[automerge::ChangeHash],
    lists: InLists,
) -> Vec<Changed> {
    doc.diff(before, after)
        .iter()
        .map(|patch| Changed {
            path: patch_path(patch, lists),
            conflict: matches!(
                patch.action,
                automerge::PatchAction::PutMap { conflict: true, .. }
            ),
        })
        .collect()
}

fn patch_path(patch: &automerge::Patch, lists: InLists) -> Vec<String> {
    use automerge::PatchAction;
    let mut path = Vec::new();
    for (_, prop) in &patch.path {
        match (prop, lists) {
            (Prop::Map(key), _) => path.push(key.clone()),
            (Prop::Seq(index), InLists::Index) => path.push(index.to_string()),
            (Prop::Seq(_), InLists::Whole) => return path,
        }
    }
    match &patch.action {
        PatchAction::PutMap { key, .. }
        | PatchAction::DeleteMap { key, .. }
        | PatchAction::Increment {
            prop: Prop::Map(key),
            ..
        } => path.push(key.clone()),
        // List and text edits: the containing object.
        _ => {}
    }
    path
}

/// 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<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],
    prop: impl Into<Prop> + Clone,
) -> Result<Vec<(Value<'static>, 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<Option<Resolved>, 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<Option<Resolved>, 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<Option<Resolved>, 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<usize> {
    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<usize> {
    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<Option<Vec<String>>, 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) = &current
            && 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, &current, &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<Prop> + Clone,
    name: &str,
    log: Option<&mut ConflictLog>,
) -> Result<Option<Resolved>, 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<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 Some(current) = resolve_logged(doc, path, log.as_deref_mut())? else {
        return Ok(None);
    };
    Ok(Some(resolved_to_json(doc, &current, log)?))
}

/// Reads a path; the empty path is the whole store.
pub(crate) fn read_path(
    doc: &AutoCommit,
    path: &[String],
) -> Result<Option<JsonValue>, StoreError> {
    read_path_inner(doc, path, None)
}

pub(crate) 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(crate) 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()))
}

/// 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<Option<JsonValue>, 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<Option<Resolved>, 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<Option<Resolved>, 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, &current, 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<JsonValue, StoreError> {
    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<Vec<String>, 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<Vec<JsonValue>, 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<ObjId>),
    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<Parent, StoreError> {
    let mut current = Parent::Maps(vec![automerge::ROOT]);
    for segment in parents {
        let next = match &current {
            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<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)
            && 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(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<JsonValue>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct ConflictReport {
    pub(crate) path: Vec<String>,
    /// 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<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
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn path(at: &str) -> Vec<String> {
        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)
        ));
    }
}