//! The tree graph: nodes, placements, and modes as atoms (spec §7.1, §7.2),
//! with tree conflict detection and path assignment (spec §9.2).

use crate::hash::ChangeHash;
use crate::ids::NodeId;
use crate::objects::{Mode, NodeKind, OpRef, TreeOp};
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};

/// `(change, index into that change's tree_ops)`.
pub type OpKey = (ChangeHash, u32);

/// The graph's own maps: keyed by hashes and random ids, which need no
/// SipHash (see [`crate::fasthash`]).
type Map<K, V> = crate::fasthash::FastMap<K, V>;

/// A node's placement or mode atoms, sorted: nearly always one, held
/// without an allocation.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
enum KeySet {
    #[default]
    Empty,
    One(OpKey),
    Many(Vec<OpKey>),
}

impl KeySet {
    fn insert(&mut self, k: OpKey) {
        match self {
            KeySet::Empty => *self = KeySet::One(k),
            KeySet::One(a) if *a == k => {}
            KeySet::One(a) => {
                let a = *a;
                *self = KeySet::Many(if a < k { vec![a, k] } else { vec![k, a] });
            }
            KeySet::Many(v) => {
                if let Err(i) = v.binary_search(&k) {
                    v.insert(i, k);
                }
            }
        }
    }
    fn remove(&mut self, k: &OpKey) {
        match self {
            KeySet::One(a) if a == k => *self = KeySet::Empty,
            KeySet::Many(v) => {
                if let Ok(i) = v.binary_search(k) {
                    v.remove(i);
                }
                if v.len() == 1 {
                    *self = KeySet::One(v[0]);
                } else if v.is_empty() {
                    *self = KeySet::Empty;
                }
            }
            _ => {}
        }
    }
    fn as_slice(&self) -> &[OpKey] {
        match self {
            KeySet::Empty => &[],
            KeySet::One(a) => std::slice::from_ref(a),
            KeySet::Many(v) => v,
        }
    }
    fn iter(&self) -> std::slice::Iter<'_, OpKey> {
        self.as_slice().iter()
    }
    fn len(&self) -> usize {
        self.as_slice().len()
    }
    fn is_empty(&self) -> bool {
        matches!(self, KeySet::Empty)
    }
}

impl<'a> IntoIterator for &'a KeySet {
    type Item = &'a OpKey;
    type IntoIter = std::slice::Iter<'a, OpKey>;
    fn into_iter(self) -> Self::IntoIter {
        self.iter()
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Placement {
    pub node: NodeId,
    pub parent: NodeId,
    pub name: String,
}

#[derive(Clone, Debug, Default)]
pub struct TreeGraph {
    kinds: Map<NodeId, NodeKind>,
    introduced: Map<NodeId, ChangeHash>,
    placements: Map<OpKey, Placement>,
    modes: Map<OpKey, (NodeId, Mode)>,
    /// Placement atoms retired by `RemoveNode`/`Move`, and mode atoms retired
    /// by `SetMode`. Kept apart because `AddNode` introduces one of each
    /// under the same key.
    deleted_placements: Map<OpKey, BTreeSet<ChangeHash>>,
    deleted_modes: Map<OpKey, BTreeSet<ChangeHash>>,
    placements_of: Map<NodeId, KeySet>,
    modes_of: Map<NodeId, KeySet>,
    /// The live subsets of `placements_of` and `modes_of`, kept as changes
    /// apply, so the view walks what is live rather than all history.
    live_placements_of: Map<NodeId, KeySet>,
    live_modes_of: Map<NodeId, KeySet>,
    /// The view, once computed, with the path of each directory in it;
    /// kept up to date as ops apply when that is simple (see
    /// `update_view`), cleared otherwise.
    view_cache: std::sync::OnceLock<CachedView>,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum TreeConflict {
    /// Two live nodes share `(parent, name)`.
    Name {
        parent: NodeId,
        name: String,
        nodes: Vec<NodeId>,
    },
    /// One node has several live placements.
    MultiPlacement {
        node: NodeId,
        placements: Vec<OpKey>,
    },
    /// A live node's parent is not live.
    Orphan { node: NodeId, parent: NodeId },
    /// A directory is placed inside its own descendant.
    Cycle { node: NodeId },
    /// Several live mode atoms.
    Mode { node: NodeId, modes: Vec<Mode> },
}

/// A node as it appears in the working copy.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TreeEntry {
    pub path: String,
    pub node: NodeId,
    pub kind: NodeKind,
    pub mode: Mode,
    /// The placement this entry materializes (for building `RemoveNode`/`Move`).
    pub placement: OpKey,
    /// The live mode atom (for building `SetMode`).
    pub mode_atom: Option<OpKey>,
}

#[derive(Clone, Debug, Default)]
struct CachedView {
    view: TreeView,
    dirs: HashMap<NodeId, String>,
}

impl CachedView {
    fn new(view: TreeView) -> CachedView {
        let mut dirs: HashMap<NodeId, String> = view
            .entries
            .values()
            .filter(|e| e.kind == NodeKind::Dir)
            .map(|e| (e.node, e.path.clone()))
            .collect();
        dirs.insert(NodeId::ROOT, String::new());
        CachedView { view, dirs }
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct TreeView {
    /// Path → entry, paths use `/` and have no leading slash; conflicted
    /// entries carry a `.oxy-conflict-<n>` suffix. Shared, so a view is
    /// cheap to clone.
    pub entries: std::sync::Arc<BTreeMap<String, TreeEntry>>,
    pub conflicts: Vec<TreeConflict>,
}

impl TreeView {
    pub fn is_conflicted(&self) -> bool {
        !self.conflicts.is_empty()
    }
    pub fn by_node(&self) -> HashMap<NodeId, &TreeEntry> {
        self.entries.values().map(|e| (e.node, e)).collect()
    }
}

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

    /// Apply tree op `index` of `change`; `OpRef` self references must already be resolved.
    pub fn apply(&mut self, change: ChangeHash, index: u32, op: &TreeOp) {
        let cached = self.view_cache.take();
        // Whether the placement or mode atom this op retires was live.
        let was_live = match op {
            TreeOp::RemoveNode { placement, .. } => self.placement_live(&op_key(*placement)),
            TreeOp::Move { old, .. } => self.placement_live(&op_key(*old)),
            TreeOp::SetMode { old, .. } => self.mode_live(&op_key(*old)),
            TreeOp::AddNode { .. } => false,
        };
        self.apply_to_graph(change, index, op);
        if let Some(mut c) = cached {
            if self.update_view(&mut c, (change, index), op, was_live) {
                #[cfg(test)]
                assert_eq!(
                    c.view,
                    self.compute_view(),
                    "view kept up to date by {op:?}"
                );
                self.view_cache = std::sync::OnceLock::from(c);
            }
        }
    }

    fn apply_to_graph(&mut self, change: ChangeHash, index: u32, op: &TreeOp) {
        let key = (change, index);
        match op {
            TreeOp::AddNode {
                node,
                kind,
                mode,
                parent,
                name,
            } => {
                self.kinds.insert(*node, *kind);
                self.introduced.insert(*node, change);
                self.placements.insert(
                    key,
                    Placement {
                        node: *node,
                        parent: *parent,
                        name: name.clone(),
                    },
                );
                self.placements_of.entry(*node).or_default().insert(key);
                self.index_placement(key);
                if *kind != NodeKind::Dir {
                    self.modes.insert(key, (*node, *mode));
                    self.modes_of.entry(*node).or_default().insert(key);
                    self.index_mode(key);
                }
            }
            TreeOp::RemoveNode { placement, .. } => {
                self.deleted_placements
                    .entry(op_key(*placement))
                    .or_default()
                    .insert(change);
                self.index_placement(op_key(*placement));
            }
            TreeOp::Move {
                node,
                old,
                parent,
                name,
            } => {
                self.deleted_placements
                    .entry(op_key(*old))
                    .or_default()
                    .insert(change);
                self.index_placement(op_key(*old));
                self.placements.insert(
                    key,
                    Placement {
                        node: *node,
                        parent: *parent,
                        name: name.clone(),
                    },
                );
                self.placements_of.entry(*node).or_default().insert(key);
                self.index_placement(key);
            }
            TreeOp::SetMode { node, old, mode } => {
                self.deleted_modes
                    .entry(op_key(*old))
                    .or_default()
                    .insert(change);
                self.index_mode(op_key(*old));
                self.modes.insert(key, (*node, *mode));
                self.modes_of.entry(*node).or_default().insert(key);
                self.index_mode(key);
            }
        }
    }

    /// Bring the cached view `c` up to date with `op` (already applied as
    /// `key`) where that is simple and the view has no conflicts; false
    /// when it must be computed again. `was_live`: whether the placement or
    /// mode atom `op` retires was live before it.
    fn update_view(&self, c: &mut CachedView, key: OpKey, op: &TreeOp, was_live: bool) -> bool {
        if !c.view.conflicts.is_empty() {
            return false;
        }
        match op {
            TreeOp::AddNode { node, .. } => self.view_add(c, *node, key),
            TreeOp::RemoveNode { node, placement } => {
                !was_live || self.view_remove(c, *node, op_key(*placement))
            }
            TreeOp::Move { node, old, .. } => {
                (!was_live || self.view_remove(c, *node, op_key(*old)))
                    && self.view_add(c, *node, key)
            }
            TreeOp::SetMode { node, .. } => {
                let live = self.live_placements(node);
                match live.as_slice() {
                    [] => true,
                    [(k, p)] => {
                        let Some(path) = c.dirs.get(&p.parent).map(|d| join(d, &p.name)) else {
                            return false;
                        };
                        let Some(mode) = self.entry_mode(node) else {
                            return false;
                        };
                        let entries = std::sync::Arc::make_mut(&mut c.view.entries);
                        match entries.get_mut(&path) {
                            Some(e) if e.node == *node && e.placement == *k => {
                                (e.mode_atom, e.mode) = mode;
                                true
                            }
                            _ => false,
                        }
                    }
                    _ => false,
                }
            }
        }
    }

    /// The view entry for `node`'s placement `key`, just made live.
    fn view_add(&self, c: &mut CachedView, node: NodeId, key: OpKey) -> bool {
        if !self.placement_live(&key) {
            return true;
        }
        if self
            .live_placements_of
            .get(&node)
            .is_none_or(|s| s.len() != 1)
        {
            return false;
        }
        let Some(p) = self.placements.get(&key) else {
            return false;
        };
        let Some(path) = c.dirs.get(&p.parent).map(|d| join(d, &p.name)) else {
            return false;
        };
        let Some((mode_atom, mode)) = self.entry_mode(&node) else {
            return false;
        };
        let kind = self.kind(&node).unwrap_or(NodeKind::File);
        let entries = std::sync::Arc::make_mut(&mut c.view.entries);
        if entries.contains_key(&path) || node == NodeId::ROOT {
            return false;
        }
        // Nothing can be placed under a directory that was not in the view
        // yet: it would have been an orphan, and the view had no conflicts.
        if kind == NodeKind::Dir {
            c.dirs.insert(node, path.clone());
        }
        entries.insert(
            path.clone(),
            TreeEntry {
                path,
                node,
                kind,
                mode,
                placement: key,
                mode_atom,
            },
        );
        true
    }

    /// Drop the view entry of `node`'s placement `key`, just retired.
    fn view_remove(&self, c: &mut CachedView, node: NodeId, key: OpKey) -> bool {
        if self.live_placements_of.contains_key(&node) {
            return false;
        }
        let Some(p) = self.placements.get(&key) else {
            return false;
        };
        let Some(path) = c.dirs.get(&p.parent).map(|d| join(d, &p.name)) else {
            return false;
        };
        let entries = std::sync::Arc::make_mut(&mut c.view.entries);
        match entries.get(&path) {
            Some(e) if e.node == node && e.placement == key => {}
            _ => return false,
        }
        if entries.get(&path).is_some_and(|e| e.kind == NodeKind::Dir) {
            let prefix = format!("{path}/");
            if entries
                .range(prefix.clone()..)
                .next()
                .is_some_and(|(k, _)| k.starts_with(&prefix))
            {
                return false;
            }
            c.dirs.remove(&node);
        }
        entries.remove(&path);
        true
    }

    /// The mode atom and mode of `node`'s entry, as `compute_view` finds
    /// them; `None` when its live modes conflict.
    fn entry_mode(&self, node: &NodeId) -> Option<(Option<OpKey>, Mode)> {
        let lm = self.live_modes(node);
        let distinct: BTreeSet<Mode> = lm.iter().map(|(_, m)| *m).collect();
        if distinct.len() > 1 {
            return None;
        }
        let mode_atom = lm.iter().map(|(k, _)| *k).max();
        let mode = mode_atom
            .and_then(|k| self.modes.get(&k).map(|(_, m)| *m))
            .unwrap_or_default();
        Some((mode_atom, mode))
    }

    /// Bring placement `k` in the live index up to date.
    fn index_placement(&mut self, k: OpKey) {
        let Some(node) = self.placements.get(&k).map(|p| p.node) else {
            return;
        };
        if self.placement_live(&k) {
            self.live_placements_of.entry(node).or_default().insert(k);
        } else if let Some(set) = self.live_placements_of.get_mut(&node) {
            set.remove(&k);
            if set.is_empty() {
                self.live_placements_of.remove(&node);
            }
        }
    }

    /// Bring mode atom `k` in the live index up to date.
    fn index_mode(&mut self, k: OpKey) {
        let Some(node) = self.modes.get(&k).map(|(n, _)| *n) else {
            return;
        };
        if self.mode_live(&k) {
            self.live_modes_of.entry(node).or_default().insert(k);
        } else if let Some(set) = self.live_modes_of.get_mut(&node) {
            set.remove(&k);
            if set.is_empty() {
                self.live_modes_of.remove(&node);
            }
        }
    }

    /// Directories no longer placed anywhere, by the (parent, name) they
    /// were last placed at, with that placement: what an import can bring
    /// back rather than make a second node for one Git path.
    pub fn dead_dirs(&self) -> HashMap<(NodeId, String), (NodeId, OpKey)> {
        let mut out: HashMap<(NodeId, String), (NodeId, OpKey)> = HashMap::new();
        for (k, p) in &self.placements {
            if self.kind(&p.node) != Some(NodeKind::Dir)
                || self.placement_live(k)
                || !self.live_placements(&p.node).is_empty()
            {
                continue;
            }
            let slot = out
                .entry((p.parent, p.name.clone()))
                .or_insert((p.node, *k));
            // The newest placement, by key: the same answer in any order.
            if *k > slot.1 {
                *slot = (p.node, *k);
            }
        }
        out
    }

    /// Every placement `node` ever had, live or not, with the changes that
    /// retired it (none while live).
    pub fn placement_history(&self, node: &NodeId) -> Vec<(OpKey, &Placement, Vec<ChangeHash>)> {
        self.placements_of
            .get(node)
            .into_iter()
            .flatten()
            .filter_map(|k| {
                let p = self.placements.get(k)?;
                let by = self
                    .deleted_placements
                    .get(k)
                    .map(|d| d.iter().copied().collect())
                    .unwrap_or_default();
                Some((*k, p, by))
            })
            .collect()
    }

    fn placement_live(&self, k: &OpKey) -> bool {
        self.deleted_placements.get(k).is_none_or(|d| d.is_empty())
    }
    fn mode_live(&self, k: &OpKey) -> bool {
        self.deleted_modes.get(k).is_none_or(|d| d.is_empty())
    }

    pub fn kind(&self, node: &NodeId) -> Option<NodeKind> {
        if node.is_root() {
            Some(NodeKind::Dir)
        } else {
            self.kinds.get(node).copied()
        }
    }

    /// The change whose `AddNode` introduced `node`.
    pub fn introducer(&self, node: &NodeId) -> Option<ChangeHash> {
        self.introduced.get(node).copied()
    }

    pub fn placement(&self, k: &OpKey) -> Option<&Placement> {
        self.placements.get(k)
    }

    pub fn live_placements(&self, node: &NodeId) -> Vec<(OpKey, &Placement)> {
        self.live_placements_of
            .get(node)
            .into_iter()
            .flat_map(|ks| ks.iter())
            .filter_map(|k| self.placements.get(k).map(|p| (*k, p)))
            .collect()
    }

    pub fn live_modes(&self, node: &NodeId) -> Vec<(OpKey, Mode)> {
        self.live_modes_of
            .get(node)
            .into_iter()
            .flat_map(|ks| ks.iter())
            .filter_map(|k| self.modes.get(k).map(|(_, m)| (*k, *m)))
            .collect()
    }

    /// Every node that has at least one live placement.
    pub fn placed_nodes(&self) -> impl Iterator<Item = NodeId> + '_ {
        self.live_placements_of.keys().copied()
    }

    /// Resolve the tree: assign paths, find conflicts.
    /// The materialized tree (spec §7.1): computed once per graph and
    /// cloned for each caller.
    pub fn view(&self) -> TreeView {
        self.view_cache
            .get_or_init(|| CachedView::new(self.compute_view()))
            .view
            .clone()
    }

    fn compute_view(&self) -> TreeView {
        let mut view = TreeView::default();
        let mut entries: BTreeMap<String, TreeEntry> = BTreeMap::new();

        // Live placements grouped by parent.
        let mut children: HashMap<NodeId, Vec<(OpKey, &Placement)>> = HashMap::new();
        for ks in self.live_placements_of.values() {
            for k in ks {
                if let Some(p) = self.placements.get(k) {
                    children.entry(p.parent).or_default().push((*k, p));
                }
            }
        }
        for c in children.values_mut() {
            c.sort_by(|a, b| a.1.name.cmp(&b.1.name).then(a.0.cmp(&b.0)));
        }

        // Multi-placement and mode conflicts.
        // Only placed nodes: one with no live placement is not in the tree.
        let mut nodes: Vec<NodeId> = self.live_placements_of.keys().copied().collect();
        nodes.sort();
        let mut multi: HashSet<NodeId> = HashSet::new();
        // Each node's mode and the atom it comes from, for the entries.
        let mut modes: HashMap<NodeId, (Option<OpKey>, Mode)> = HashMap::new();
        for n in &nodes {
            let lp = self.live_placements(n);
            // One placement (nearly every node) cannot be in two places.
            if lp.len() > 1 {
                let distinct_places: BTreeSet<(&NodeId, &str)> = lp
                    .iter()
                    .map(|(_, p)| (&p.parent, p.name.as_str()))
                    .collect();
                if distinct_places.len() > 1 {
                    multi.insert(*n);
                    view.conflicts.push(TreeConflict::MultiPlacement {
                        node: *n,
                        placements: lp.iter().map(|(k, _)| *k).collect(),
                    });
                }
            }
            let lm = self.live_modes(n);
            if lm.len() > 1 {
                let distinct_modes: BTreeSet<Mode> = lm.iter().map(|(_, m)| *m).collect();
                if distinct_modes.len() > 1 {
                    view.conflicts.push(TreeConflict::Mode {
                        node: *n,
                        modes: distinct_modes.into_iter().collect(),
                    });
                }
            }
            let mode_atom = lm.iter().map(|(k, _)| *k).max();
            let mode = mode_atom
                .and_then(|k| self.modes.get(&k).map(|(_, m)| *m))
                .unwrap_or_default();
            modes.insert(*n, (mode_atom, mode));
        }

        // BFS from the root, assigning paths.
        let mut reached: HashSet<NodeId> = HashSet::new();
        reached.insert(NodeId::ROOT);
        let mut queue: Vec<(NodeId, String)> = vec![(NodeId::ROOT, String::new())];
        let mut on_path: HashSet<NodeId> = HashSet::new();
        while let Some((dir, dir_path)) = queue.pop() {
            let Some(kids) = children.get(&dir) else {
                continue;
            };
            // Name conflicts within this directory: the children are sorted
            // by name, so each name's group is a run of them.
            let mut groups: Vec<(&str, &[(OpKey, &Placement)])> = Vec::new();
            let mut start = 0;
            for i in 1..=kids.len() {
                if i == kids.len() || kids[i].1.name != kids[start].1.name {
                    groups.push((kids[start].1.name.as_str(), &kids[start..i]));
                    start = i;
                }
            }
            for (name, group) in groups {
                let distinct: BTreeSet<NodeId> = group.iter().map(|(_, p)| p.node).collect();
                if distinct.len() > 1 {
                    view.conflicts.push(TreeConflict::Name {
                        parent: dir,
                        name: name.to_string(),
                        nodes: distinct.iter().copied().collect(),
                    });
                }
                let suffixed = distinct.len() > 1;
                for (i, (k, p)) in group.iter().enumerate() {
                    let node = p.node;
                    if node == dir || on_path.contains(&node) {
                        view.conflicts.push(TreeConflict::Cycle { node });
                        continue;
                    }
                    let needs_suffix = suffixed || multi.contains(&node);
                    let leaf = if needs_suffix {
                        format!("{name}.oxy-conflict-{}", i + 1)
                    } else {
                        name.to_string()
                    };
                    let path = if dir_path.is_empty() {
                        leaf
                    } else {
                        format!("{dir_path}/{leaf}")
                    };
                    let kind = self.kind(&node).unwrap_or(NodeKind::File);
                    let (mode_atom, mode) = modes.get(&node).copied().unwrap_or_default();
                    entries.insert(
                        path.clone(),
                        TreeEntry {
                            path: path.clone(),
                            node,
                            kind,
                            mode,
                            placement: *k,
                            mode_atom,
                        },
                    );
                    if kind == NodeKind::Dir && reached.insert(node) {
                        on_path.insert(dir);
                        queue.push((node, path));
                    } else {
                        reached.insert(node);
                    }
                }
            }
        }

        // Nodes with live placements that were never reached: orphans or cycles.
        for n in &nodes {
            if reached.contains(n) {
                continue;
            }
            let lp = self.live_placements(n);
            if lp.is_empty() {
                continue;
            }
            let mut seen = HashSet::new();
            let mut cur = *n;
            let conflict = loop {
                if !seen.insert(cur) {
                    break TreeConflict::Cycle { node: *n };
                }
                let lp = self.live_placements(&cur);
                let Some((_, p)) = lp.first() else {
                    break TreeConflict::Orphan {
                        node: *n,
                        parent: cur,
                    };
                };
                if reached.contains(&p.parent) {
                    // Parent reached but this node was not: the parent is not a
                    // directory, so this is a name/kind problem surfaced as orphan.
                    break TreeConflict::Orphan {
                        node: *n,
                        parent: p.parent,
                    };
                }
                cur = p.parent;
            };
            if !view.conflicts.contains(&conflict) {
                view.conflicts.push(conflict);
            }
        }
        view.entries = std::sync::Arc::new(entries);
        view
    }
}

impl TreeGraph {
    /// Serialize for a checkpoint.
    pub fn to_value(&self) -> crate::cbor::Value {
        use crate::cbor::Value;
        let opkey = |k: &OpKey| vec![Value::Bytes(k.0 .0.to_vec()), Value::Uint(k.1 as u64)];
        let deleters = |m: &Map<OpKey, BTreeSet<ChangeHash>>| {
            let mut ks: Vec<&OpKey> = m.keys().collect();
            ks.sort();
            Value::Array(
                ks.into_iter()
                    .map(|k| {
                        let mut v = opkey(k);
                        v.push(Value::Array(
                            m[k].iter().map(|d| Value::Bytes(d.0.to_vec())).collect(),
                        ));
                        Value::Array(v)
                    })
                    .collect(),
            )
        };
        let mut nodes: Vec<&NodeId> = self.kinds.keys().collect();
        nodes.sort();
        let mut pks: Vec<&OpKey> = self.placements.keys().collect();
        pks.sort();
        let mut mks: Vec<&OpKey> = self.modes.keys().collect();
        mks.sort();
        Value::map([
            (
                0,
                Value::Array(
                    nodes
                        .into_iter()
                        .map(|n| {
                            Value::Array(vec![
                                Value::Bytes(n.0.to_vec()),
                                Value::Uint(self.kinds[n].tag()),
                                self.introduced
                                    .get(n)
                                    .map(|h| Value::Bytes(h.0.to_vec()))
                                    .unwrap_or(Value::Null),
                            ])
                        })
                        .collect(),
                ),
            ),
            (
                1,
                Value::Array(
                    pks.into_iter()
                        .map(|k| {
                            let p = &self.placements[k];
                            let mut v = opkey(k);
                            v.extend([
                                Value::Bytes(p.node.0.to_vec()),
                                Value::Bytes(p.parent.0.to_vec()),
                                Value::Text(p.name.clone()),
                            ]);
                            Value::Array(v)
                        })
                        .collect(),
                ),
            ),
            (2, deleters(&self.deleted_placements)),
            (
                3,
                Value::Array(
                    mks.into_iter()
                        .map(|k| {
                            let (n, m) = &self.modes[k];
                            let mut v = opkey(k);
                            v.extend([Value::Bytes(n.0.to_vec()), Value::Uint(m.bits())]);
                            Value::Array(v)
                        })
                        .collect(),
                ),
            ),
            (4, deleters(&self.deleted_modes)),
        ])
    }

    pub fn from_value(v: &crate::cbor::Value) -> Result<TreeGraph, crate::objects::ObjectError> {
        use crate::objects::{array, bytes, field, map, oref, text, u32_, uint, ObjectError};
        let m = map(v, "tree graph")?;
        let node = |v: &crate::cbor::Value| {
            NodeId::from_slice(bytes(v, "node")?)
                .ok_or_else(|| ObjectError::Schema("node: expected 16 bytes".into()))
        };
        let opkey = |e: &[crate::cbor::Value]| -> Result<OpKey, ObjectError> {
            Ok((oref(&e[0], "op change")?, u32_(&e[1], "op index")?))
        };
        let mut g = TreeGraph::default();
        for n in array(field(m, 0, "nodes")?, "nodes")? {
            let e = array(n, "node entry")?;
            if e.len() != 3 {
                return Err(ObjectError::Schema("node entry: expected 3 fields".into()));
            }
            let id = node(&e[0])?;
            g.kinds
                .insert(id, NodeKind::from_tag(uint(&e[1], "kind")?)?);
            if !e[2].is_null() {
                g.introduced.insert(id, oref(&e[2], "introducer")?);
            }
        }
        for p in array(field(m, 1, "placements")?, "placements")? {
            let e = array(p, "placement")?;
            if e.len() != 5 {
                return Err(ObjectError::Schema("placement: expected 5 fields".into()));
            }
            let k = opkey(e)?;
            let pl = Placement {
                node: node(&e[2])?,
                parent: node(&e[3])?,
                name: text(&e[4], "name")?.to_owned(),
            };
            g.placements_of.entry(pl.node).or_default().insert(k);
            g.placements.insert(k, pl);
        }
        let deleters = |v: &crate::cbor::Value,
                        what: &str|
         -> Result<Map<OpKey, BTreeSet<ChangeHash>>, ObjectError> {
            let mut out = Map::default();
            for d in array(v, what)? {
                let e = array(d, what)?;
                if e.len() != 3 {
                    return Err(ObjectError::Schema(format!("{what}: expected 3 fields")));
                }
                out.insert(
                    opkey(e)?,
                    array(&e[2], "deleters")?
                        .iter()
                        .map(|x| oref(x, "deleter"))
                        .collect::<Result<BTreeSet<_>, _>>()?,
                );
            }
            Ok(out)
        };
        g.deleted_placements = deleters(field(m, 2, "deleted placements")?, "deleted placement")?;
        for md in array(field(m, 3, "modes")?, "modes")? {
            let e = array(md, "mode")?;
            if e.len() != 4 {
                return Err(ObjectError::Schema("mode: expected 4 fields".into()));
            }
            let k = opkey(e)?;
            let n = node(&e[2])?;
            g.modes
                .insert(k, (n, Mode::from_bits(uint(&e[3], "mode bits")?)?));
            g.modes_of.entry(n).or_default().insert(k);
        }
        g.deleted_modes = deleters(field(m, 4, "deleted modes")?, "deleted mode")?;
        let keys: Vec<OpKey> = g.placements.keys().copied().collect();
        for k in keys {
            g.index_placement(k);
        }
        let keys: Vec<OpKey> = g.modes.keys().copied().collect();
        for k in keys {
            g.index_mode(k);
        }
        Ok(g)
    }
}

/// The tree in the flat layout of a local index (oxygen#41), with its view
/// when that has no conflicts, so a loaded tree need not work it out again.
impl TreeGraph {
    pub(crate) fn write_flat(&self, w: &mut crate::flat::Writer) {
        let key = |w: &mut crate::flat::Writer, k: &OpKey| {
            w.object(&k.0);
            w.u32(k.1);
        };
        let mut kinds: Vec<(&NodeId, &NodeKind)> = self.kinds.iter().collect();
        kinds.sort_by_key(|(n, _)| **n);
        w.count(kinds.len());
        for (n, k) in kinds {
            w.fixed(&n.0);
            w.u8(k.tag() as u8);
        }
        let mut intro: Vec<(&NodeId, &ChangeHash)> = self.introduced.iter().collect();
        intro.sort();
        w.count(intro.len());
        for (n, c) in intro {
            w.fixed(&n.0);
            w.object(c);
        }
        let mut pl: Vec<(&OpKey, &Placement)> = self.placements.iter().collect();
        pl.sort_by_key(|(k, _)| **k);
        w.count(pl.len());
        for (k, p) in pl {
            key(w, k);
            w.fixed(&p.node.0);
            w.fixed(&p.parent.0);
            w.bytes(p.name.as_bytes());
        }
        let mut modes: Vec<(&OpKey, &(NodeId, Mode))> = self.modes.iter().collect();
        modes.sort_by_key(|(k, _)| **k);
        w.count(modes.len());
        for (k, (n, m)) in modes {
            key(w, k);
            w.fixed(&n.0);
            w.u8(m.bits() as u8);
        }
        for m in [&self.deleted_placements, &self.deleted_modes] {
            let mut ks: Vec<(&OpKey, &BTreeSet<ChangeHash>)> = m.iter().collect();
            ks.sort_by_key(|(k, _)| **k);
            w.count(ks.len());
            for (k, by) in ks {
                key(w, k);
                w.count(by.len());
                for c in by {
                    w.object(c);
                }
            }
        }
        // The view as a block of its own, decoded beside the maps.
        let outer = std::mem::take(w);
        let view = self.view();
        if view.conflicts.is_empty() {
            w.u8(1);
            w.count(view.entries.len());
            for (path, e) in view.entries.iter() {
                w.bytes(path.as_bytes());
                w.fixed(&e.node.0);
                w.u8(e.kind.tag() as u8);
                w.u8(e.mode.bits() as u8);
                key(w, &e.placement);
                match &e.mode_atom {
                    None => w.u8(0),
                    Some(k) => {
                        w.u8(1);
                        key(w, k);
                    }
                }
            }
        } else {
            w.u8(0);
        }
        let view_bytes = std::mem::replace(w, outer);
        w.bytes(&view_bytes.0);
    }

    pub(crate) fn read_flat(
        r: &mut crate::flat::Reader,
    ) -> Result<TreeGraph, crate::objects::ObjectError> {
        use crate::objects::ObjectError;
        let node = |r: &mut crate::flat::Reader| -> Result<NodeId, ObjectError> {
            Ok(NodeId(r.fixed::<16>()?))
        };
        let key = |r: &mut crate::flat::Reader| -> Result<OpKey, ObjectError> {
            Ok((r.object()?, r.u32()?))
        };
        let mut g = TreeGraph::new();
        let t = crate::timing::phase("index: tree maps");
        for _ in 0..r.count()? {
            let n = node(r)?;
            g.kinds.insert(n, NodeKind::from_tag(r.u8()? as u64)?);
        }
        for _ in 0..r.count()? {
            let n = node(r)?;
            g.introduced.insert(n, r.object()?);
        }
        for _ in 0..r.count()? {
            let k = key(r)?;
            let p = Placement {
                node: node(r)?,
                parent: node(r)?,
                name: r.text()?,
            };
            g.placements_of.entry(p.node).or_default().insert(k);
            g.placements.insert(k, p);
        }
        for _ in 0..r.count()? {
            let k = key(r)?;
            let n = node(r)?;
            g.modes.insert(k, (n, Mode::from_bits(r.u8()? as u64)?));
            g.modes_of.entry(n).or_default().insert(k);
        }
        for i in 0..2 {
            let mut m: Map<OpKey, BTreeSet<ChangeHash>> = Map::default();
            for _ in 0..r.count()? {
                let k = key(r)?;
                let mut by = BTreeSet::new();
                for _ in 0..r.count()? {
                    by.insert(r.object()?);
                }
                m.insert(k, by);
            }
            if i == 0 {
                g.deleted_placements = m;
            } else {
                g.deleted_modes = m;
            }
        }
        drop(t);
        let view_bytes = r.bytes()?;
        let (view, ()) = rayon::join(
            || Self::read_view(view_bytes),
            || {
                let _t = crate::timing::phase("index: tree live");
                let keys: Vec<OpKey> = g.placements.keys().copied().collect();
                for k in keys {
                    g.index_placement(k);
                }
                let keys: Vec<OpKey> = g.modes.keys().copied().collect();
                for k in keys {
                    g.index_mode(k);
                }
            },
        );
        if let Some(view) = view? {
            #[cfg(test)]
            assert_eq!(view, g.compute_view(), "an index's view is the tree's");
            g.view_cache = std::sync::OnceLock::from(CachedView::new(view));
        }
        Ok(g)
    }
}

impl TreeGraph {
    /// The view block of a tree's flat form, when it holds one.
    fn read_view(bytes: &[u8]) -> Result<Option<TreeView>, crate::objects::ObjectError> {
        let _t = crate::timing::phase("index: view");
        let mut r = crate::flat::Reader::new(bytes);
        if r.u8()? != 1 {
            return Ok(None);
        }
        let key = |r: &mut crate::flat::Reader| -> Result<OpKey, crate::objects::ObjectError> {
            Ok((r.object()?, r.u32()?))
        };
        let n = r.count()?;
        let mut entries = BTreeMap::new();
        for _ in 0..n {
            let path = r.text()?;
            let e = TreeEntry {
                path: path.clone(),
                node: NodeId(r.fixed::<16>()?),
                kind: NodeKind::from_tag(r.u8()? as u64)?,
                mode: Mode::from_bits(r.u8()? as u64)?,
                placement: key(&mut r)?,
                mode_atom: match r.u8()? {
                    0 => None,
                    _ => Some(key(&mut r)?),
                },
            };
            entries.insert(path, e);
        }
        Ok(Some(TreeView {
            entries: std::sync::Arc::new(entries),
            conflicts: Vec::new(),
        }))
    }
}

fn join(dir: &str, name: &str) -> String {
    if dir.is_empty() {
        name.to_string()
    } else {
        format!("{dir}/{name}")
    }
}

pub fn op_key(r: OpRef) -> OpKey {
    (r.change, r.index)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hash::ObjectRef;

    fn h(n: u8) -> ChangeHash {
        ObjectRef([n; 32])
    }
    fn nid(n: u8) -> NodeId {
        NodeId([n; 16])
    }
    fn add(g: &mut TreeGraph, c: u8, i: u32, node: u8, kind: NodeKind, parent: NodeId, name: &str) {
        g.apply(
            h(c),
            i,
            &TreeOp::AddNode {
                node: nid(node),
                kind,
                mode: Mode::TEXT,
                parent,
                name: name.into(),
            },
        );
    }
    fn paths(v: &TreeView) -> Vec<String> {
        v.entries.keys().cloned().collect()
    }

    /// Each op applied to a graph whose view is cached keeps it up to date
    /// or drops it: `apply` checks every kept view against a full one.
    #[test]
    fn cached_view_follows_ops() {
        let at = |c: u8, i: u32| OpRef {
            change: h(c),
            index: i,
        };
        let mut g = TreeGraph::new();
        let ops = [
            TreeOp::AddNode {
                node: nid(1),
                kind: NodeKind::Dir,
                mode: Mode::TEXT,
                parent: NodeId::ROOT,
                name: "src".into(),
            },
            TreeOp::AddNode {
                node: nid(2),
                kind: NodeKind::File,
                mode: Mode::TEXT,
                parent: nid(1),
                name: "a.rs".into(),
            },
            TreeOp::SetMode {
                node: nid(2),
                old: at(1, 1),
                mode: Mode {
                    exec: true,
                    opaque: false,
                },
            },
            TreeOp::Move {
                node: nid(2),
                old: at(1, 1),
                parent: NodeId::ROOT,
                name: "b.rs".into(),
            },
            // Moving an empty directory, then removing it.
            TreeOp::Move {
                node: nid(1),
                old: at(1, 0),
                parent: NodeId::ROOT,
                name: "lib".into(),
            },
            TreeOp::RemoveNode {
                node: nid(1),
                placement: at(1, 4),
            },
            // A name taken twice: a conflict, so the view is computed again.
            TreeOp::AddNode {
                node: nid(3),
                kind: NodeKind::File,
                mode: Mode::TEXT,
                parent: NodeId::ROOT,
                name: "b.rs".into(),
            },
        ];
        for (i, op) in ops.iter().enumerate() {
            let _ = g.view();
            g.apply(h(1), i as u32, op);
            assert_eq!(g.view(), g.compute_view());
        }
        let v = g.view();
        assert_eq!(paths(&v), ["b.rs.oxy-conflict-1", "b.rs.oxy-conflict-2"]);
        assert!(v.entries.values().any(|e| e.mode.exec));
    }

    #[test]
    fn nested_paths_and_rename() {
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::Dir, NodeId::ROOT, "src");
        add(&mut g, 1, 1, 2, NodeKind::File, nid(1), "main.rs");
        add(&mut g, 1, 2, 3, NodeKind::File, NodeId::ROOT, "README");
        let v = g.view();
        assert_eq!(paths(&v), ["README", "src", "src/main.rs"]);
        assert!(!v.is_conflicted());

        // Rename src/main.rs → src/lib.rs, move src → lib (subtree follows).
        g.apply(
            h(2),
            0,
            &TreeOp::Move {
                node: nid(2),
                old: OpRef {
                    change: h(1),
                    index: 1,
                },
                parent: nid(1),
                name: "lib.rs".into(),
            },
        );
        g.apply(
            h(2),
            1,
            &TreeOp::Move {
                node: nid(1),
                old: OpRef {
                    change: h(1),
                    index: 0,
                },
                parent: NodeId::ROOT,
                name: "lib".into(),
            },
        );
        let v = g.view();
        assert_eq!(paths(&v), ["README", "lib", "lib/lib.rs"]);
        assert_eq!(v.entries["lib/lib.rs"].placement, (h(2), 0));

        g.apply(
            h(3),
            0,
            &TreeOp::RemoveNode {
                node: nid(3),
                placement: OpRef {
                    change: h(1),
                    index: 2,
                },
            },
        );
        assert_eq!(paths(&g.view()), ["lib", "lib/lib.rs"]);
    }

    #[test]
    fn name_conflict_gets_suffixes() {
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::File, NodeId::ROOT, "a.txt");
        add(&mut g, 2, 0, 2, NodeKind::File, NodeId::ROOT, "a.txt");
        let v = g.view();
        assert_eq!(paths(&v), ["a.txt.oxy-conflict-1", "a.txt.oxy-conflict-2"]);
        assert_eq!(
            v.conflicts,
            vec![TreeConflict::Name {
                parent: NodeId::ROOT,
                name: "a.txt".into(),
                nodes: vec![nid(1), nid(2)]
            }]
        );
    }

    #[test]
    fn concurrent_renames_are_multi_placement() {
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::File, NodeId::ROOT, "a");
        g.apply(
            h(2),
            0,
            &TreeOp::Move {
                node: nid(1),
                old: OpRef {
                    change: h(1),
                    index: 0,
                },
                parent: NodeId::ROOT,
                name: "b".into(),
            },
        );
        g.apply(
            h(3),
            0,
            &TreeOp::Move {
                node: nid(1),
                old: OpRef {
                    change: h(1),
                    index: 0,
                },
                parent: NodeId::ROOT,
                name: "c".into(),
            },
        );
        let v = g.view();
        assert_eq!(paths(&v), ["b.oxy-conflict-1", "c.oxy-conflict-1"]);
        assert!(
            matches!(v.conflicts[0], TreeConflict::MultiPlacement { node, .. } if node == nid(1))
        );
    }

    #[test]
    fn orphan_and_cycle() {
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::Dir, NodeId::ROOT, "d");
        add(&mut g, 1, 1, 2, NodeKind::File, nid(1), "f");
        // Concurrently: 2 removes d, 3 adds d/g.
        g.apply(
            h(2),
            0,
            &TreeOp::RemoveNode {
                node: nid(1),
                placement: OpRef {
                    change: h(1),
                    index: 0,
                },
            },
        );
        add(&mut g, 3, 0, 3, NodeKind::File, nid(1), "g");
        let v = g.view();
        assert_eq!(paths(&v), Vec::<String>::new());
        assert!(v.conflicts.contains(&TreeConflict::Orphan {
            node: nid(2),
            parent: nid(1)
        }));
        assert!(v.conflicts.contains(&TreeConflict::Orphan {
            node: nid(3),
            parent: nid(1)
        }));

        // Cycle: a/b, then concurrently move a under b and b under a's parent... simplest: move a into b.
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::Dir, NodeId::ROOT, "a");
        add(&mut g, 1, 1, 2, NodeKind::Dir, nid(1), "b");
        g.apply(
            h(2),
            0,
            &TreeOp::Move {
                node: nid(1),
                old: OpRef {
                    change: h(1),
                    index: 0,
                },
                parent: nid(2),
                name: "a".into(),
            },
        );
        let v = g.view();
        assert!(
            v.conflicts
                .iter()
                .any(|c| matches!(c, TreeConflict::Cycle { .. })),
            "{:?}",
            v.conflicts
        );
    }

    #[test]
    fn serialization_round_trip() {
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::Dir, NodeId::ROOT, "src");
        add(&mut g, 1, 1, 2, NodeKind::File, nid(1), "main.rs");
        g.apply(
            h(2),
            0,
            &TreeOp::Move {
                node: nid(2),
                old: OpRef {
                    change: h(1),
                    index: 1,
                },
                parent: nid(1),
                name: "lib.rs".into(),
            },
        );
        g.apply(
            h(3),
            0,
            &TreeOp::SetMode {
                node: nid(2),
                old: OpRef {
                    change: h(1),
                    index: 1,
                },
                mode: Mode {
                    exec: true,
                    opaque: false,
                },
            },
        );
        let bytes = crate::cbor::encode(&g.to_value());
        let back = TreeGraph::from_value(&crate::cbor::decode(&bytes).unwrap()).unwrap();
        assert_eq!(back.view(), g.view());
        assert_eq!(back.introducer(&nid(2)), Some(h(1)));
        assert_eq!(crate::cbor::encode(&back.to_value()), bytes);
    }

    #[test]
    fn mode_conflict_and_resolution() {
        let mut g = TreeGraph::new();
        add(&mut g, 1, 0, 1, NodeKind::File, NodeId::ROOT, "run.sh");
        g.apply(
            h(2),
            0,
            &TreeOp::SetMode {
                node: nid(1),
                old: OpRef {
                    change: h(1),
                    index: 0,
                },
                mode: Mode {
                    exec: true,
                    opaque: false,
                },
            },
        );
        g.apply(
            h(3),
            0,
            &TreeOp::SetMode {
                node: nid(1),
                old: OpRef {
                    change: h(1),
                    index: 0,
                },
                mode: Mode {
                    exec: false,
                    opaque: true,
                },
            },
        );
        let v = g.view();
        assert!(matches!(v.conflicts[0], TreeConflict::Mode { .. }));
        // Materialized with the greatest mode atom.
        assert_eq!(
            v.entries["run.sh"].mode,
            Mode {
                exec: false,
                opaque: true
            }
        );
        // The placement introduced by the same AddNode is unaffected.
        assert_eq!(v.entries["run.sh"].placement, (h(1), 0));
        // Resolution: retire both live mode atoms; identical live values do not conflict.
        g.apply(
            h(4),
            0,
            &TreeOp::SetMode {
                node: nid(1),
                old: OpRef {
                    change: h(2),
                    index: 0,
                },
                mode: Mode {
                    exec: true,
                    opaque: false,
                },
            },
        );
        g.apply(
            h(4),
            1,
            &TreeOp::SetMode {
                node: nid(1),
                old: OpRef {
                    change: h(3),
                    index: 0,
                },
                mode: Mode {
                    exec: true,
                    opaque: false,
                },
            },
        );
        let v = g.view();
        assert!(!v.is_conflicted(), "{:?}", v.conflicts);
        assert_eq!(
            v.entries["run.sh"].mode,
            Mode {
                exec: true,
                opaque: false
            }
        );
    }
}