//! A state: a dependency-closed set of changes, applied to a tree graph and
//! per-node file graphs (spec §8.3), with materialization (spec §9.5, §11).

use crate::cbor::{self, Value};
use crate::graph::{AtomKey, DepOracle, FileGraph, Materialized, Vertex};
use crate::hash::KeyId;
use crate::hash::{ChangeHash, ObjectRef};
use crate::ids::NodeId;
use crate::objects::{
    self as o, AtomRef, Change, ChangeKind, Edit, Hlc, Mode, NodeKind, ObjectError, TreeOp,
};
use crate::tree::{TreeGraph, TreeView};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::sync::{Arc, Mutex};

/// Checkpoint payload version 2 (spec §11): file graphs by reference.
pub const CHECKPOINT_INDEX_VERSION: u64 = 2;

/// A file graph encoded in fewer bytes stays inline in a version 2 index.
pub const INLINE_GRAPH_LIMIT: usize = 2048;

#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum StateError {
    #[error("change {1} depends on {0}, which is not in the state")]
    MissingDep(ChangeHash, ChangeHash),
}

/// One `Replace` edit: `(change, old blob, new blob)`.
type ReplaceRecord = (ChangeHash, Option<ObjectRef>, Option<ObjectRef>);

/// What a state remembers about a change: everything but its operations,
/// which live in the graphs once applied (and in the store).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ChangeMeta {
    pub change_id: crate::ids::ChangeId,
    pub deps: Vec<ChangeHash>,
    pub author: KeyId,
    pub hlc: Hlc,
    pub message: String,
    pub kind: ChangeKind,
}

impl ChangeMeta {
    pub fn of(c: &Change) -> ChangeMeta {
        ChangeMeta {
            change_id: c.change_id,
            deps: c.deps.clone(),
            author: c.author,
            hlc: c.hlc,
            message: c.message.clone(),
            kind: c.kind,
        }
    }
}

#[derive(Default)]
pub struct State {
    changes: HashMap<ChangeHash, ChangeMeta>,
    tree: TreeGraph,
    files: HashMap<NodeId, FileGraph>,
    replaces: HashMap<NodeId, Vec<ReplaceRecord>>,
    /// Memoised transitive dependencies. A mutex, not a `RefCell`, so a
    /// loaded state can be shared across threads by a host that caches it.
    closure: Mutex<HashMap<ChangeHash, Arc<HashSet<ChangeHash>>>>,
}

impl Clone for State {
    fn clone(&self) -> State {
        State {
            changes: self.changes.clone(),
            tree: self.tree.clone(),
            files: self.files.clone(),
            replaces: self.replaces.clone(),
            closure: Mutex::new(self.closure.lock().unwrap().clone()),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum FileContent {
    Text {
        bytes: Vec<u8>,
        materialized: Materialized,
    },
    /// Live blobs; more than one is a replace conflict (spec §9.3).
    Opaque {
        blobs: Vec<ObjectRef>,
    },
    Dir,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FileState {
    pub node: NodeId,
    pub kind: NodeKind,
    pub mode: Mode,
    pub content: FileContent,
}

impl FileState {
    pub fn is_conflicted(&self) -> bool {
        match &self.content {
            FileContent::Text { materialized, .. } => materialized.is_conflicted(),
            FileContent::Opaque { blobs } => blobs.len() > 1,
            FileContent::Dir => false,
        }
    }
}

/// See `State::conflict_regions`.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct ConflictRegions {
    pub plain: Vec<(AtomKey, Vec<u8>)>,
    pub region_atoms: Vec<AtomKey>,
    pub contributors: Vec<ChangeHash>,
}

/// The whole tree of a state, materialized.
#[derive(Clone, Debug, Default)]
pub struct Snapshot {
    pub tree: TreeView,
    pub files: BTreeMap<String, FileState>,
}

impl Snapshot {
    pub fn is_conflicted(&self) -> bool {
        self.tree.is_conflicted() || self.files.values().any(|f| f.is_conflicted())
    }
}

fn vertex(r: AtomRef) -> Vertex {
    match r {
        AtomRef::Start => Vertex::Start,
        AtomRef::End => Vertex::End,
        AtomRef::Atom { change, index } => Vertex::Atom((change, index)),
    }
}

/// Replace self references in a tree op with the concrete change hash.
pub fn resolve_tree_op(op: &TreeOp, me: ChangeHash) -> TreeOp {
    match op {
        TreeOp::RemoveNode { node, placement } => TreeOp::RemoveNode {
            node: *node,
            placement: placement.resolve(me),
        },
        TreeOp::Move {
            node,
            old,
            parent,
            name,
        } => TreeOp::Move {
            node: *node,
            old: old.resolve(me),
            parent: *parent,
            name: name.clone(),
        },
        TreeOp::SetMode { node, old, mode } => TreeOp::SetMode {
            node: *node,
            old: old.resolve(me),
            mode: *mode,
        },
        add @ TreeOp::AddNode { .. } => add.clone(),
    }
}

impl State {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn len(&self) -> usize {
        self.changes.len()
    }
    pub fn is_empty(&self) -> bool {
        self.changes.is_empty()
    }
    pub fn contains(&self, h: &ChangeHash) -> bool {
        self.changes.contains_key(h)
    }
    pub fn get(&self, h: &ChangeHash) -> Option<&ChangeMeta> {
        self.changes.get(h)
    }
    pub fn changes(&self) -> impl Iterator<Item = (&ChangeHash, &ChangeMeta)> {
        self.changes.iter()
    }
    pub fn tree(&self) -> &TreeGraph {
        &self.tree
    }
    pub fn file(&self, node: &NodeId) -> Option<&FileGraph> {
        self.files.get(node)
    }

    /// Add a change whose dependencies are already present.
    pub fn add(&mut self, change: Change) -> Result<ChangeHash, StateError> {
        let h = change.hash();
        if self.changes.contains_key(&h) {
            return Ok(h);
        }
        for d in &change.deps {
            if !self.changes.contains_key(d) {
                return Err(StateError::MissingDep(*d, h));
            }
        }
        Ok(self.add_unchecked(change))
    }

    /// Add a change in any order; call `validate` once all are in.
    pub fn add_unchecked(&mut self, change: Change) -> ChangeHash {
        let h = change.hash();
        if self.changes.contains_key(&h) {
            return h;
        }
        let meta = ChangeMeta::of(&change);
        self.apply(h, change);
        self.changes.insert(h, meta);
        self.closure.lock().unwrap().clear();
        h
    }

    // ------------------------------------------------------------ checkpoints (spec §11)

    /// Serialize the whole state in payload version 1: change metadata,
    /// tree, file graphs, replaces, all in one value (spec §11).
    pub fn checkpoint_payload(&self) -> Vec<u8> {
        let mut nodes: Vec<&NodeId> = self.files.keys().collect();
        nodes.sort();
        let files = Value::Array(
            nodes
                .into_iter()
                .map(|n| Value::Array(vec![Value::Bytes(n.0.to_vec()), self.files[n].to_value()]))
                .collect(),
        );
        cbor::encode(&self.payload_value(o::FORMAT_VERSION, files))
    }

    /// Each file graph encoded on its own, by node, for a version 2 payload
    /// (spec §11): the graphs of large files become blobs of their own.
    pub fn checkpoint_file_graphs(&self) -> Vec<(NodeId, Vec<u8>)> {
        use rayon::prelude::*;
        let mut nodes: Vec<(&NodeId, &FileGraph)> = self.files.iter().collect();
        nodes.sort_by_key(|(n, _)| **n);
        nodes
            .into_par_iter()
            .map(|(n, g)| (*n, cbor::encode(&g.to_value())))
            .collect()
    }

    /// The payload version 2 index for the graphs of
    /// [`State::checkpoint_file_graphs`]: a graph encoded in fewer than
    /// [`INLINE_GRAPH_LIMIT`] bytes is kept inline, a larger one is named by
    /// `blob_of` (the ObjectRef of the blob holding its encoding).
    pub fn checkpoint_index(
        &self,
        graphs: &[(NodeId, Vec<u8>)],
        mut blob_of: impl FnMut(&[u8]) -> ObjectRef,
    ) -> Vec<u8> {
        let files = Value::Array(
            graphs
                .iter()
                .map(|(n, enc)| {
                    let entry = if enc.len() < INLINE_GRAPH_LIMIT {
                        cbor::decode(enc).expect("an encoded graph decodes")
                    } else {
                        Value::Bytes(blob_of(enc).0.to_vec())
                    };
                    Value::Array(vec![Value::Bytes(n.0.to_vec()), entry])
                })
                .collect(),
        );
        cbor::encode(&self.payload_value(CHECKPOINT_INDEX_VERSION, files))
    }

    /// The payload map around `files`: changes, tree and replaces as in version 1.
    fn payload_value(&self, version: u64, files: Value) -> Value {
        let mut hashes: Vec<&ChangeHash> = self.changes.keys().collect();
        hashes.sort();
        let changes = Value::Array(
            hashes
                .into_iter()
                .map(|h| {
                    let m = &self.changes[h];
                    Value::Array(vec![
                        Value::Bytes(h.0.to_vec()),
                        Value::Bytes(m.change_id.0.to_vec()),
                        Value::Array(m.deps.iter().map(|d| Value::Bytes(d.0.to_vec())).collect()),
                        Value::Text(m.message.clone()),
                        Value::Uint(m.hlc.wall_ms),
                        Value::Uint(m.hlc.logical as u64),
                        Value::Uint(m.kind.tag()),
                        Value::Bytes(m.author.0.to_vec()),
                    ])
                })
                .collect(),
        );
        let mut rnodes: Vec<&NodeId> = self.replaces.keys().collect();
        rnodes.sort();
        let opt =
            |o: &Option<ObjectRef>| o.map(|r| Value::Bytes(r.0.to_vec())).unwrap_or(Value::Null);
        let replaces = Value::Array(
            rnodes
                .into_iter()
                .map(|n| {
                    Value::Array(vec![
                        Value::Bytes(n.0.to_vec()),
                        Value::Array(
                            self.replaces[n]
                                .iter()
                                .map(|(c, o, nw)| {
                                    Value::Array(vec![Value::Bytes(c.0.to_vec()), opt(o), opt(nw)])
                                })
                                .collect(),
                        ),
                    ])
                })
                .collect(),
        );
        Value::map([
            (0, Value::Uint(version)),
            (1, changes),
            (2, self.tree.to_value()),
            (3, files),
            (4, replaces),
        ])
    }

    /// Decode a version 1 payload (the whole state in one value). A
    /// version 2 payload needs its graph blobs: use [`State::from_checkpoint`].
    pub fn from_checkpoint_payload(bytes: &[u8]) -> Result<State, ObjectError> {
        State::from_checkpoint(bytes, |_| {
            Err(ObjectError::Schema(
                "a version 2 checkpoint needs its graph blobs".into(),
            ))
        })
    }

    /// Decode a checkpoint payload of either version (spec §11). For
    /// version 2, `load` returns the contents of the named graph blobs, in
    /// order; the graphs are then decoded in parallel.
    pub fn from_checkpoint(
        bytes: &[u8],
        load: impl FnOnce(&[ObjectRef]) -> Result<Vec<Vec<u8>>, ObjectError>,
    ) -> Result<State, ObjectError> {
        use rayon::prelude::*;
        let v = cbor::decode(bytes)?;
        let m = o::map(&v, "checkpoint payload")?;
        let version = o::uint(o::field(m, 0, "version")?, "version")?;
        if version != o::FORMAT_VERSION && version != CHECKPOINT_INDEX_VERSION {
            return Err(ObjectError::Schema(format!(
                "checkpoint payload version {version} is not supported"
            )));
        }
        let mut st = State::payload_common(m)?;
        let entries = o::array(o::field(m, 3, "files")?, "files")?;
        let mut parsed: Vec<(NodeId, &Value)> = Vec::with_capacity(entries.len());
        for f in entries {
            let e = o::array(f, "file")?;
            if e.len() != 2 {
                return Err(ObjectError::Schema("file: expected [node, graph]".into()));
            }
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
            parsed.push((node, &e[1]));
        }
        // Version 2 names large graphs by blob: fetch those in one batch.
        let mut fetched: HashMap<NodeId, Vec<u8>> = HashMap::new();
        if version == CHECKPOINT_INDEX_VERSION {
            let wanted: Vec<(NodeId, ObjectRef)> = parsed
                .iter()
                .filter(|(_, e)| matches!(e, Value::Bytes(_)))
                .map(|(n, e)| Ok((*n, o::oref(e, "graph blob")?)))
                .collect::<Result<_, ObjectError>>()?;
            if !wanted.is_empty() {
                let refs: Vec<ObjectRef> = wanted.iter().map(|(_, r)| *r).collect();
                let contents = {
                    let _t = crate::timing::phase("load:   graph blobs read");
                    load(&refs)?
                };
                if contents.len() != refs.len() {
                    return Err(ObjectError::Schema("graph blobs: wrong count".into()));
                }
                fetched.extend(wanted.iter().map(|(n, _)| *n).zip(contents));
            }
        } else if parsed.iter().any(|(_, e)| matches!(e, Value::Bytes(_))) {
            return Err(ObjectError::Schema(
                "a version 1 payload holds graphs inline".into(),
            ));
        }
        // File graphs are independent; decode and build them in parallel.
        let graphs = parsed
            .par_iter()
            .map(|(node, entry)| {
                let graph = match fetched.get(node) {
                    Some(bytes) => FileGraph::from_value(&cbor::decode(bytes)?)?,
                    None => FileGraph::from_value(entry)?,
                };
                Ok((*node, graph))
            })
            .collect::<Result<Vec<_>, ObjectError>>()?;
        st.files.extend(graphs);
        Ok(st)
    }

    /// The graph blobs a payload names: empty for version 1.
    pub fn checkpoint_graph_refs(bytes: &[u8]) -> Result<Vec<ObjectRef>, ObjectError> {
        let v = cbor::decode(bytes)?;
        let m = o::map(&v, "checkpoint payload")?;
        let mut out = Vec::new();
        for f in o::array(o::field(m, 3, "files")?, "files")? {
            if let Some(entry @ Value::Bytes(_)) = o::array(f, "file")?.get(1) {
                out.push(o::oref(entry, "graph blob")?);
            }
        }
        Ok(out)
    }

    /// Changes, tree and replaces of a payload of either version.
    fn payload_common(m: &BTreeMap<u64, Value>) -> Result<State, ObjectError> {
        let mut st = State::new();
        for c in o::array(o::field(m, 1, "changes")?, "changes")? {
            let e = o::array(c, "change meta")?;
            if e.len() != 8 {
                return Err(ObjectError::Schema("change meta: expected 8 fields".into()));
            }
            let hash = o::oref(&e[0], "hash")?;
            let meta = ChangeMeta {
                change_id: crate::ids::ChangeId::from_slice(o::bytes(&e[1], "change id")?)
                    .ok_or_else(|| ObjectError::Schema("change id: expected 16 bytes".into()))?,
                deps: o::array(&e[2], "deps")?
                    .iter()
                    .map(|d| o::oref(d, "dep"))
                    .collect::<Result<Vec<_>, _>>()?,
                message: o::text(&e[3], "message")?.to_owned(),
                hlc: Hlc {
                    wall_ms: o::uint(&e[4], "wall")?,
                    logical: o::u32_(&e[5], "logical")?,
                },
                kind: ChangeKind::from_tag(o::uint(&e[6], "kind")?)?,
                author: o::oref(&e[7], "author")?,
            };
            st.changes.insert(hash, meta);
        }
        st.tree = TreeGraph::from_value(o::field(m, 2, "tree")?)?;
        let opt = |v: &Value| -> Result<Option<ObjectRef>, ObjectError> {
            if v.is_null() {
                Ok(None)
            } else {
                o::oref(v, "blob").map(Some)
            }
        };
        for r in o::array(o::field(m, 4, "replaces")?, "replaces")? {
            let e = o::array(r, "replace")?;
            if e.len() != 2 {
                return Err(ObjectError::Schema(
                    "replace: expected [node, records]".into(),
                ));
            }
            let node = NodeId::from_slice(o::bytes(&e[0], "node")?)
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))?;
            let mut recs = Vec::new();
            for rec in o::array(&e[1], "records")? {
                let x = o::array(rec, "record")?;
                if x.len() != 3 {
                    return Err(ObjectError::Schema("record: expected 3 fields".into()));
                }
                recs.push((o::oref(&x[0], "change")?, opt(&x[1])?, opt(&x[2])?));
            }
            st.replaces.insert(node, recs);
        }
        Ok(st)
    }

    pub fn validate(&self) -> Result<(), StateError> {
        for (h, c) in &self.changes {
            for d in &c.deps {
                if !self.changes.contains_key(d) {
                    return Err(StateError::MissingDep(*d, *h));
                }
            }
        }
        Ok(())
    }

    fn apply(&mut self, h: ChangeHash, c: Change) {
        for (i, op) in c.tree_ops.iter().enumerate() {
            self.tree.apply(h, i as u32, &resolve_tree_op(op, h));
        }
        let mut counters: HashMap<NodeId, u32> = HashMap::new();
        for e in c.edits {
            match e {
                Edit::Insert {
                    node,
                    after,
                    before,
                    lines,
                } => {
                    let idx = counters.entry(node).or_default();
                    let first = *idx;
                    *idx += lines.len() as u32;
                    self.files.entry(node).or_default().insert_owned(
                        h,
                        first,
                        vertex(after.resolve(h)),
                        vertex(before.resolve(h)),
                        lines,
                    );
                }
                Edit::Delete { node, atoms } => {
                    let g = self.files.entry(node).or_default();
                    for r in atoms {
                        g.delete(h, &r.resolve(h));
                    }
                }
                Edit::Replace { node, old, new } => {
                    self.replaces.entry(node).or_default().push((h, old, new));
                }
            }
        }
    }

    /// Transitive dependencies of `h`, excluding `h` itself.
    pub fn closure(&self, h: &ChangeHash) -> Arc<HashSet<ChangeHash>> {
        if let Some(c) = self.closure.lock().unwrap().get(h) {
            return c.clone();
        }
        let mut out = HashSet::new();
        let mut stack: Vec<ChangeHash> = self
            .changes
            .get(h)
            .map(|c| c.deps.clone())
            .unwrap_or_default();
        while let Some(d) = stack.pop() {
            if out.insert(d) {
                if let Some(c) = self.changes.get(&d) {
                    stack.extend(c.deps.iter().copied());
                }
            }
        }
        let rc = Arc::new(out);
        self.closure.lock().unwrap().insert(*h, rc.clone());
        rc
    }

    /// Changes that no other change depends on directly, sorted.
    pub fn heads(&self) -> Vec<ChangeHash> {
        let depended: HashSet<&ChangeHash> =
            self.changes.values().flat_map(|c| c.deps.iter()).collect();
        let mut heads: Vec<ChangeHash> = self
            .changes
            .keys()
            .filter(|h| !depended.contains(h))
            .copied()
            .collect();
        heads.sort();
        heads
    }

    /// Maximal elements among `hashes` (those not in the closure of another),
    /// sorted. One walk down from all of them at once: whatever the walk
    /// reaches is below some other element. Linear in the changes reached,
    /// where comparing every pair's closures was quadratic in the number
    /// of heads, which a bookmark of independent changes has many of.
    pub fn maximal(&self, hashes: &[ChangeHash]) -> Vec<ChangeHash> {
        let set: BTreeSet<ChangeHash> = hashes.iter().copied().collect();
        let mut below: HashSet<ChangeHash> = HashSet::new();
        let mut stack: Vec<ChangeHash> = set
            .iter()
            .filter_map(|h| self.changes.get(h))
            .flat_map(|c| c.deps.iter().copied())
            .collect();
        while let Some(d) = stack.pop() {
            if below.insert(d) {
                if let Some(c) = self.changes.get(&d) {
                    stack.extend(c.deps.iter().copied());
                }
            }
        }
        set.into_iter().filter(|h| !below.contains(h)).collect()
    }

    pub fn materialize_file(&self, node: &NodeId) -> Materialized {
        self.files
            .get(node)
            .map(|g| g.materialize(self))
            .unwrap_or_default()
    }

    /// Live lines of a text node in materialized order.
    pub fn live_lines(&self, node: &NodeId) -> Vec<(AtomKey, Vec<u8>)> {
        let Some(g) = self.files.get(node) else {
            return vec![];
        };
        let m = g.materialize(self);
        m.live_atoms()
            .into_iter()
            .map(|a| (a, g.content(&a).unwrap_or_default().to_vec()))
            .collect()
    }

    /// Live blobs of an opaque node: every `new` that no later `Replace` retired.
    pub fn live_blobs(&self, node: &NodeId) -> Vec<ObjectRef> {
        let mut live: Vec<ObjectRef> = self
            .live_replaces(node)
            .into_iter()
            .map(|(_, n)| n)
            .collect();
        live.sort();
        live.dedup();
        live
    }

    /// The replace records whose blob is live, with the change that set it
    /// (spec §9.3). A record's blob is superseded by a later replacement of
    /// that blob: one whose change depends on the record's change, as every
    /// replacement depends on the change that made its old blob current
    /// (§8.1). Two replacements that do not know each other both stay live
    /// (a replace conflict), and a file reverted to an earlier blob keeps it.
    /// A replacement without that dependency, from before the rule, is read
    /// as before: it supersedes every record that did not come after it.
    pub fn live_replaces(&self, node: &NodeId) -> Vec<(ChangeHash, ObjectRef)> {
        let Some(rs) = self.replaces.get(node) else {
            return vec![];
        };
        let knows_its_old = |s: &ChangeHash, old: &ObjectRef| {
            rs.iter()
                .any(|(q, _, n)| n.as_ref() == Some(old) && q != s && self.closure(s).contains(q))
        };
        rs.iter()
            .filter_map(|(c, _, n)| n.map(|n| (*c, n)))
            .filter(|(c, n)| {
                !rs.iter().any(|(s, o, _)| {
                    s != c
                        && *o == Some(*n)
                        && (self.closure(s).contains(c)
                            || (!knows_its_old(s, n) && !self.closure(c).contains(s)))
                })
            })
            .collect()
    }

    /// Short label for conflict markers: `changeid-prefix "message"`.
    pub fn label(&self, h: &ChangeHash) -> String {
        match self.changes.get(h) {
            Some(c) => format!(
                "{} \"{}\"",
                c.change_id.short(8),
                c.message.lines().next().unwrap_or("")
            ),
            None => format!("#{}", h.short(12)),
        }
    }

    /// For resolving a conflicted text node (spec §9.6): the live atoms that
    /// are not part of any conflict region in order, every live atom that is
    /// part of a region, and every change that contributed to a region.
    /// The changes that introduced the live blobs of an opaque node, for a
    /// replace conflict (spec §9.3).
    pub fn blob_contributors(&self, node: &NodeId, blobs: &[ObjectRef]) -> Vec<ChangeHash> {
        let mut out: Vec<ChangeHash> = Vec::new();
        for (h, _, new) in self.replaces.get(node).map(|v| v.as_slice()).unwrap_or(&[]) {
            if new.is_some_and(|b| blobs.contains(&b)) && !out.contains(h) {
                out.push(*h);
            }
        }
        out.sort();
        out
    }

    pub fn conflict_regions(&self, node: &NodeId) -> ConflictRegions {
        let mut out = ConflictRegions::default();
        let Some(g) = self.files.get(node) else {
            return out;
        };
        let m = g.materialize(self);
        let mut contributors = BTreeSet::new();
        for seg in &m.segments {
            match seg {
                crate::graph::Segment::Lines(atoms) => {
                    for a in atoms {
                        out.plain
                            .push((*a, g.content(a).unwrap_or_default().to_vec()));
                    }
                }
                crate::graph::Segment::Conflict(sides) => {
                    for side in sides {
                        for a in side {
                            out.region_atoms.push(*a);
                            contributors.insert(a.0);
                        }
                    }
                }
                crate::graph::Segment::Zombie { deleted, inserted } => {
                    for a in deleted {
                        contributors.insert(a.0);
                        contributors.extend(g.deleters(a).copied());
                    }
                    for a in inserted {
                        out.region_atoms.push(*a);
                        contributors.insert(a.0);
                    }
                }
            }
        }
        out.contributors = contributors.into_iter().collect();
        out
    }

    pub fn render_file(&self, node: &NodeId) -> (Vec<u8>, Materialized) {
        let Some(g) = self.files.get(node) else {
            return (Vec::new(), Materialized::default());
        };
        let m = g.materialize(self);
        let bytes = g.render(&m, &|h| self.label(h));
        (bytes, m)
    }

    /// Every file rendered, in parallel: files are independent, and
    /// rendering them is most of a snapshot's cost in a large tree.
    pub fn materialize(&self) -> Snapshot {
        use rayon::prelude::*;
        let tree = self.tree.view();
        let entries: Vec<(&String, &crate::tree::TreeEntry)> = tree.entries.iter().collect();
        let rendered: Vec<(String, FileState)> = entries
            .par_iter()
            .map(|(path, e)| (path.to_string(), self.file_state(e)))
            .collect();
        let files: BTreeMap<String, FileState> = rendered.into_iter().collect();
        Snapshot { tree, files }
    }

    fn file_state(&self, e: &crate::tree::TreeEntry) -> FileState {
        {
            let content = match e.kind {
                NodeKind::Dir => FileContent::Dir,
                NodeKind::Symlink | NodeKind::Subrepo => FileContent::Opaque {
                    blobs: self.live_blobs(&e.node),
                },
                NodeKind::File if e.mode.opaque => FileContent::Opaque {
                    blobs: self.live_blobs(&e.node),
                },
                NodeKind::File => {
                    let (bytes, materialized) = self.render_file(&e.node);
                    FileContent::Text {
                        bytes,
                        materialized,
                    }
                }
            };
            FileState {
                node: e.node,
                kind: e.kind,
                mode: e.mode,
                content,
            }
        }
    }
}

impl DepOracle for State {
    fn depends_on(&self, a: &ChangeHash, b: &ChangeHash) -> bool {
        a == b || self.closure(a).contains(b)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ids::ChangeId;
    use crate::objects::{AtomRange, ChangeKind, Hlc, OpRef};

    fn change(deps: Vec<ChangeHash>, tree_ops: Vec<TreeOp>, edits: Vec<Edit>) -> Change {
        let mut c = Change {
            change_id: ChangeId::random(),
            deps,
            author: ObjectRef::ZERO,
            hlc: Hlc {
                wall_ms: 1,
                logical: 0,
            },
            message: "m".into(),
            tree_ops,
            edits,
            kind: ChangeKind::Normal,
        };
        c.normalize();
        c
    }

    #[test]
    fn build_edit_and_materialize() {
        let mut st = State::new();
        let f = NodeId::random();
        let c1 = change(
            vec![],
            vec![TreeOp::AddNode {
                node: f,
                kind: NodeKind::File,
                mode: Mode::TEXT,
                parent: NodeId::ROOT,
                name: "a.txt".into(),
            }],
            vec![Edit::Insert {
                node: f,
                after: AtomRef::Start,
                before: AtomRef::End,
                lines: vec![b"one\n".to_vec(), b"two\n".to_vec()],
            }],
        );
        let h1 = st.add(c1).unwrap();
        // Depends on h1: append a line and delete "one".
        let c2 = change(
            vec![h1],
            vec![],
            vec![
                Edit::Delete {
                    node: f,
                    atoms: vec![AtomRange {
                        change: h1,
                        from: 0,
                        to: 1,
                    }],
                },
                Edit::Insert {
                    node: f,
                    after: AtomRef::Atom {
                        change: h1,
                        index: 1,
                    },
                    before: AtomRef::End,
                    lines: vec![b"three\n".to_vec()],
                },
            ],
        );
        let h2 = st.add(c2).unwrap();
        assert_eq!(st.heads(), vec![h2]);
        assert!(st.depends_on(&h2, &h1));
        assert!(!st.depends_on(&h1, &h2));
        let snap = st.materialize();
        assert!(!snap.is_conflicted());
        let a = &snap.files["a.txt"];
        assert!(matches!(&a.content, FileContent::Text { bytes, .. } if bytes == b"two\nthree\n"));
        assert_eq!(st.live_lines(&f).len(), 2);
    }

    #[test]
    fn missing_dependency_is_rejected() {
        let mut st = State::new();
        let c = change(vec![ObjectRef([1; 32])], vec![], vec![]);
        assert!(matches!(st.add(c), Err(StateError::MissingDep(..))));
    }

    #[test]
    fn self_references_and_multi_insert_indexing() {
        let mut st = State::new();
        let f = NodeId::random();
        let me = OpRef {
            change: ObjectRef::ZERO,
            index: 0,
        };
        let c1 = change(
            vec![],
            vec![
                TreeOp::AddNode {
                    node: f,
                    kind: NodeKind::File,
                    mode: Mode::TEXT,
                    parent: NodeId::ROOT,
                    name: "x".into(),
                },
                TreeOp::SetMode {
                    node: f,
                    old: me,
                    mode: Mode {
                        exec: true,
                        opaque: false,
                    },
                },
            ],
            vec![
                Edit::Insert {
                    node: f,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines: vec![b"a\n".to_vec(), b"c\n".to_vec()],
                },
                // Insert between the two lines just inserted, using self references.
                Edit::Insert {
                    node: f,
                    after: AtomRef::Atom {
                        change: ObjectRef::ZERO,
                        index: 0,
                    },
                    before: AtomRef::Atom {
                        change: ObjectRef::ZERO,
                        index: 1,
                    },
                    lines: vec![b"b\n".to_vec()],
                },
            ],
        );
        st.add(c1).unwrap();
        let snap = st.materialize();
        let x = &snap.files["x"];
        assert_eq!(
            x.mode,
            Mode {
                exec: true,
                opaque: false
            }
        );
        assert!(matches!(&x.content, FileContent::Text { bytes, .. } if bytes == b"a\nb\nc\n"));
    }

    #[test]
    fn opaque_replace_and_conflict() {
        let mut st = State::new();
        let f = NodeId::random();
        let b1 = ObjectRef([1; 32]);
        let b2 = ObjectRef([2; 32]);
        let b3 = ObjectRef([3; 32]);
        let h1 = st
            .add(change(
                vec![],
                vec![TreeOp::AddNode {
                    node: f,
                    kind: NodeKind::File,
                    mode: Mode::OPAQUE,
                    parent: NodeId::ROOT,
                    name: "img".into(),
                }],
                vec![Edit::Replace {
                    node: f,
                    old: None,
                    new: Some(b1),
                }],
            ))
            .unwrap();
        assert_eq!(st.live_blobs(&f), vec![b1]);
        st.add(change(
            vec![h1],
            vec![],
            vec![Edit::Replace {
                node: f,
                old: Some(b1),
                new: Some(b2),
            }],
        ))
        .unwrap();
        st.add(change(
            vec![h1],
            vec![],
            vec![Edit::Replace {
                node: f,
                old: Some(b1),
                new: Some(b3),
            }],
        ))
        .unwrap();
        let snap = st.materialize();
        assert!(snap.is_conflicted());
        assert!(
            matches!(&snap.files["img"].content, FileContent::Opaque { blobs } if blobs == &vec![b2, b3])
        );
    }

    #[test]
    fn checkpoint_payload_round_trip_and_replay() {
        let mut st = State::new();
        let f = NodeId::random();
        let h1 = st
            .add(change(
                vec![],
                vec![TreeOp::AddNode {
                    node: f,
                    kind: NodeKind::File,
                    mode: Mode::TEXT,
                    parent: NodeId::ROOT,
                    name: "a".into(),
                }],
                vec![Edit::Insert {
                    node: f,
                    after: AtomRef::Start,
                    before: AtomRef::End,
                    lines: vec![b"1\n".to_vec(), b"2\n".to_vec()],
                }],
            ))
            .unwrap();
        let h2 = st
            .add(change(
                vec![h1],
                vec![],
                vec![Edit::Delete {
                    node: f,
                    atoms: vec![AtomRange {
                        change: h1,
                        from: 0,
                        to: 1,
                    }],
                }],
            ))
            .unwrap();
        let payload = st.checkpoint_payload();
        let back = State::from_checkpoint_payload(&payload).unwrap();
        assert_eq!(back.len(), 2);
        assert_eq!(back.heads(), vec![h2]);
        assert_eq!(back.get(&h1).unwrap().message, "m");
        assert_eq!(back.checkpoint_payload(), payload, "deterministic");
        // Replaying a further change onto the restored state equals a full build.
        let c3 = change(
            vec![h2],
            vec![],
            vec![Edit::Insert {
                node: f,
                after: AtomRef::Atom {
                    change: h1,
                    index: 1,
                },
                before: AtomRef::End,
                lines: vec![b"3\n".to_vec()],
            }],
        );
        let mut full = st.clone();
        full.add(c3.clone()).unwrap();
        let mut restored = back;
        restored.add(c3).unwrap();
        let (a, b) = (full.materialize(), restored.materialize());
        assert_eq!(
            a.files.keys().collect::<Vec<_>>(),
            b.files.keys().collect::<Vec<_>>()
        );
        assert_eq!(a.files["a"].content, b.files["a"].content);
        assert!(
            matches!(&b.files["a"].content, FileContent::Text { bytes, .. } if bytes == b"2\n3\n")
        );
        assert!(State::from_checkpoint_payload(b"nonsense").is_err());
    }

    /// Version 2 payloads (spec §11): small graphs inline, large ones by
    /// blob; the state read back is the same as from version 1.
    #[test]
    fn checkpoint_index_round_trip_matches_version_1() {
        let mut st = State::new();
        let small = NodeId::random();
        let big = NodeId::random();
        let add = |node: NodeId, name: &str| TreeOp::AddNode {
            node,
            kind: NodeKind::File,
            mode: Mode::TEXT,
            parent: NodeId::ROOT,
            name: name.into(),
        };
        let big_lines: Vec<Vec<u8>> = (0..400)
            .map(|i| format!("line {i} of a long file\n").into_bytes())
            .collect();
        let h1 = st
            .add(change(
                vec![],
                vec![add(small, "small"), add(big, "big")],
                vec![
                    Edit::Insert {
                        node: small,
                        after: AtomRef::Start,
                        before: AtomRef::End,
                        lines: vec![b"x\n".to_vec()],
                    },
                    Edit::Insert {
                        node: big,
                        after: AtomRef::Start,
                        before: AtomRef::End,
                        lines: big_lines,
                    },
                ],
            ))
            .unwrap();
        st.add(change(
            vec![h1],
            vec![],
            vec![Edit::Delete {
                node: big,
                atoms: vec![AtomRange {
                    change: h1,
                    from: 3,
                    to: 7,
                }],
            }],
        ))
        .unwrap();

        let graphs = st.checkpoint_file_graphs();
        let mut blobs: HashMap<ObjectRef, Vec<u8>> = HashMap::new();
        let index = st.checkpoint_index(&graphs, |enc| {
            let r = crate::hash::hash_object(crate::hash::ObjectType::Blob, enc);
            blobs.insert(r, enc.to_vec());
            r
        });
        assert_eq!(blobs.len(), 1, "only the large graph is a blob of its own");
        assert_eq!(
            State::checkpoint_graph_refs(&index).unwrap(),
            blobs.keys().copied().collect::<Vec<_>>()
        );
        assert!(
            State::checkpoint_graph_refs(&st.checkpoint_payload())
                .unwrap()
                .is_empty(),
            "version 1 names none"
        );
        assert!(index.len() < st.checkpoint_payload().len());

        let back = State::from_checkpoint(&index, |refs| {
            refs.iter()
                .map(|r| {
                    blobs
                        .get(r)
                        .cloned()
                        .ok_or_else(|| ObjectError::Schema("missing".into()))
                })
                .collect()
        })
        .unwrap();
        assert_eq!(
            back.checkpoint_payload(),
            st.checkpoint_payload(),
            "the same state as version 1"
        );
        assert_eq!(
            back.materialize().files["big"].content,
            st.materialize().files["big"].content
        );
        // Deterministic: the same state gives the same index.
        assert_eq!(
            back.checkpoint_index(&back.checkpoint_file_graphs(), |enc| {
                crate::hash::hash_object(crate::hash::ObjectType::Blob, enc)
            }),
            index
        );
        // A missing graph blob is an error, not an empty file.
        assert!(
            State::from_checkpoint(&index, |_| Err(ObjectError::Schema("gone".into()))).is_err()
        );
        assert!(
            State::from_checkpoint_payload(&index).is_err(),
            "version 2 needs its blobs"
        );
        // Version 1 payloads still read, and unknown versions do not.
        assert!(State::from_checkpoint(&st.checkpoint_payload(), |_| unreachable!()).is_ok());
        let mut v3 = cbor::decode(&index).unwrap();
        if let Value::Map(m) = &mut v3 {
            m.insert(0, Value::Uint(3));
        }
        assert!(State::from_checkpoint(&cbor::encode(&v3), |_| unreachable!()).is_err());
    }

    #[test]
    fn maximal_elements() {
        let mut st = State::new();
        let h1 = st.add(change(vec![], vec![], vec![])).unwrap();
        let h2 = st.add(change(vec![h1], vec![], vec![])).unwrap();
        let h3 = st.add(change(vec![], vec![], vec![])).unwrap();
        let mut m = st.maximal(&[h1, h2, h3]);
        m.sort();
        let mut want = vec![h2, h3];
        want.sort();
        assert_eq!(m, want);
        assert_eq!(st.heads(), want);
    }

    /// The one-walk `maximal` agrees with the definition (not in the
    /// closure of another element) on random graphs, with repeated and
    /// unknown hashes among the input.
    #[test]
    fn maximal_matches_its_definition() {
        use rand::{Rng, SeedableRng};
        let mut rng = rand::rngs::StdRng::seed_from_u64(7);
        for _ in 0..20 {
            let mut st = State::new();
            let mut all: Vec<ChangeHash> = Vec::new();
            for _ in 0..60 {
                let deps: Vec<ChangeHash> =
                    all.iter().copied().filter(|_| rng.gen_bool(0.05)).collect();
                all.push(st.add(change(deps, vec![], vec![])).unwrap());
            }
            let mut pick: Vec<ChangeHash> =
                all.iter().copied().filter(|_| rng.gen_bool(0.4)).collect();
            pick.extend(pick.clone().into_iter().take(3));
            pick.push(ObjectRef([0xee; 32]));
            let set: BTreeSet<ChangeHash> = pick.iter().copied().collect();
            let want: Vec<ChangeHash> = set
                .iter()
                .filter(|h| !set.iter().any(|o| o != *h && st.closure(o).contains(h)))
                .copied()
                .collect();
            assert_eq!(st.maximal(&pick), want);
            assert_eq!(st.maximal(&all), st.heads());
        }
    }
}