use serde::{Deserialize, Serialize};

use super::{Crdt, CrdtError, HybridLogicalClock, VectorClock, clock::now_ms};

//
// TreeCRDT — hierarchical tree with convergent move semantics.
//
// Based on Kleppmann et al. (2021) "A highly-available move operation for
// replicated trees". Each node has a unique HLC id, a parent pointer, a
// fractional-index position string for sibling ordering, and a string value.
//
// Key design decisions:
// - Tombstones are never GC'd (same as RGA). Children of deleted nodes are
//   preserved; the subtree becomes invisible but is restored if the delete is
//   later "undone" via a concurrent move.
// - MoveNode uses a separate op_id so a node can be moved many times; each
//   move has its own causal identity in the move log.
// - Concurrent moves are ordered by op_id. A move that would create a cycle
//   (i.e. moving a node under its own descendant) is silently discarded.
// - UpdateNode uses last-write-wins on the value field via updated_at HLC.
// - Sibling order: lexicographic on position string, ties broken by id descending
//   (higher HLC = leftmost), matching the RGA tiebreak convention.

/// A single tree node. Tombstones are GC'd after a configurable grace period.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct TreeNode {
    pub id: HybridLogicalClock,
    /// None = direct child of the implicit root.
    pub parent_id: Option<HybridLogicalClock>,
    /// Fractional index string for sibling ordering, e.g. "a0", "a0V", "b0".
    /// The caller supplies this; the SDK computes it locally.
    pub position: String,
    pub value: String,
    pub deleted: bool,
    /// HLC of the last accepted UpdateNode op. Used for LWW on value.
    pub updated_at: HybridLogicalClock,
    /// Wall-clock ms when the tombstone was created on this replica.
    /// 0 = unknown (deserialized from pre-GC data — never GC'd for safety).
    #[serde(default)]
    pub deleted_at_ms: u64,
}

/// One entry in the ordered move log — needed to detect cycles during merge.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct MoveRecord {
    /// Unique ID of this move op (a fresh HLC tick, not the node id).
    pub op_id: HybridLogicalClock,
    pub node_id: HybridLogicalClock,
    pub new_parent_id: Option<HybridLogicalClock>,
    pub new_position: String,
    /// Parent state before this move — kept so we can undo during merge replay.
    pub old_parent_id: Option<HybridLogicalClock>,
    pub old_position: String,
}

#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct TreeCrdt {
    pub nodes: Vec<TreeNode>,
    /// All MoveNode ops ever applied, sorted ascending by op_id.
    pub move_log: Vec<MoveRecord>,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub enum TreeOp {
    AddNode {
        id: HybridLogicalClock,
        parent_id: Option<HybridLogicalClock>,
        position: String,
        value: String,
    },
    MoveNode {
        /// Fresh HLC for this op — NOT the node id.
        op_id: HybridLogicalClock,
        node_id: HybridLogicalClock,
        new_parent_id: Option<HybridLogicalClock>,
        new_position: String,
    },
    UpdateNode {
        id: HybridLogicalClock,
        value: String,
        /// LWW timestamp for this update.
        updated_at: HybridLogicalClock,
    },
    DeleteNode {
        id: HybridLogicalClock,
    },
}

/// A MoveNode op that was rejected because it would have created a cycle.
/// Included in the delta so clients can emit `move_discarded` conflict events.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub struct DiscardedMove {
    /// The node that was supposed to move.
    pub node_id: HybridLogicalClock,
    /// The parent it tried to move under (None = root level).
    pub attempted_parent_id: Option<HybridLogicalClock>,
    pub attempted_position: String,
    /// The parent it actually stayed under (unchanged).
    pub actual_parent_id: Option<HybridLogicalClock>,
    pub actual_position: String,
}

#[must_use]
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct TreeDelta {
    pub ops: Vec<TreeOp>,
    /// MoveNode ops that were rejected due to cycle prevention.
    /// Empty in the common case; populated only when a cycle would have occurred.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub discarded_moves: Vec<DiscardedMove>,
}

/// Recursive tree node as delivered to clients.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct TreeNodeValue {
    /// HLC serialized as "wall_ms:logical:node_id" — avoids JS bigint issues.
    pub id: String,
    pub value: String,
    /// Fractional index string for sibling ordering (e.g. "a0", "b0").
    pub position: String,
    pub children: Vec<TreeNodeValue>,
}

/// The root-level value returned by `TreeCrdt::value()`.
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct TreeValue {
    pub roots: Vec<TreeNodeValue>,
}

impl TreeCrdt {
    /// Returns true if `ancestor_id` is an ancestor of `node_id` in the
    /// current (possibly partially updated) parent-pointer state.
    /// Cycles in the parent graph are guarded against with a visited set.
    fn is_ancestor(
        &self,
        mut node_id: HybridLogicalClock,
        ancestor_id: HybridLogicalClock,
    ) -> bool {
        let mut visited = std::collections::BTreeSet::new();
        loop {
            if node_id == ancestor_id {
                return true;
            }
            if !visited.insert(node_id) {
                // Guard against corrupt cycles in state (should not happen in
                // practice, but keeps the function total).
                return false;
            }
            match self.nodes.iter().find(|n| n.id == node_id) {
                Some(n) => match n.parent_id {
                    Some(pid) => node_id = pid,
                    None => return false,
                },
                None => return false,
            }
        }
    }

    /// Serialize an HLC to the wire string format used in TreeNodeValue.
    fn hlc_to_str(hlc: HybridLogicalClock) -> String {
        format!("{}:{}:{}", hlc.wall_ms, u32::from(hlc.logical), hlc.node_id)
    }

    /// Build the recursive value for one node and its visible descendants.
    fn build_node_value(&self, node: &TreeNode) -> TreeNodeValue {
        let mut children: Vec<&TreeNode> = self
            .nodes
            .iter()
            .filter(|n| !n.deleted && n.parent_id == Some(node.id))
            .collect();

        // Sort by (position asc, id desc) — lexicographic position, ties broken by HLC descending.
        children.sort_by(|a, b| a.position.cmp(&b.position).then_with(|| b.id.cmp(&a.id)));

        TreeNodeValue {
            id: Self::hlc_to_str(node.id),
            value: node.value.clone(),
            position: node.position.clone(),
            children: children.iter().map(|c| self.build_node_value(c)).collect(),
        }
    }

    /// Compact the CRDT state by removing tombstoned nodes and truncating the
    /// move log to only keep the most recent move per node.
    ///
    /// **Safety conditions (caller must ensure both before calling):**
    /// 1. All connected clients have acknowledged WAL entries up to at least
    ///    `checkpoint_seq` — tombstoned nodes whose AddNode seq < checkpoint
    ///    are safe to drop since no new inserts can reference them as parents.
    /// 2. The move log is only needed for merge replay. Once all replicas are
    ///    caught up, older entries that have been superseded by later moves on
    ///    the same node serve no purpose.
    ///
    /// Returns `(nodes_removed, move_records_removed)`.
    /// Remove tombstoned nodes and stale move-log entries.
    ///
    /// A tombstone node is removed when it is structurally safe (no live node
    /// uses it as parent) AND old enough (`deleted_at_ms > 0` and
    /// `deleted_at_ms < threshold_wall_ms`). Nodes with `deleted_at_ms == 0`
    /// are never removed — their age is unknown.
    ///
    /// Move-log compaction retains only the latest move record per node and is
    /// always safe (the current parent state is already stored on the node itself).
    pub fn compact_before(&mut self, threshold_wall_ms: u64) -> (usize, usize) {
        // Step 1: remove tombstoned nodes whose children have all been deleted
        // or moved away, and that are old enough to be safe.
        let live_parent_ids: std::collections::BTreeSet<Option<HybridLogicalClock>> = self
            .nodes
            .iter()
            .filter(|n| !n.deleted)
            .map(|n| n.parent_id)
            .collect();

        let nodes_before = self.nodes.len();
        self.nodes.retain(|n| {
            if !n.deleted {
                return true;
            }
            // Keep tombstone if structurally referenced, age unknown, or not old enough.
            live_parent_ids.contains(&Some(n.id))
                || n.deleted_at_ms == 0
                || n.deleted_at_ms >= threshold_wall_ms
        });
        let nodes_removed = nodes_before - self.nodes.len();

        // Step 2: truncate move_log — keep only the latest move per node.
        // The log is sorted ascending by op_id, so the last entry per node_id
        // is the one that matters for current parent state.
        let mut seen_nodes = std::collections::BTreeSet::new();
        // Walk backward to find the latest move per node, then retain those.
        let mut to_keep = std::collections::BTreeSet::new();
        for (i, record) in self.move_log.iter().enumerate().rev() {
            if seen_nodes.insert(record.node_id) {
                to_keep.insert(i);
            }
        }
        let records_before = self.move_log.len();
        let mut idx = 0;
        self.move_log.retain(|_| {
            let keep = to_keep.contains(&idx);
            idx += 1;
            keep
        });
        let records_removed = records_before - self.move_log.len();

        (nodes_removed, records_removed)
    }

    /// DottedDB-inspired precise GC: like `compact_before` but gates tombstone
    /// removal on `floor_vc` rather than purely on wall-clock age.
    ///
    /// A tombstone with `id = HLC(wall_ms, logical, node_id)` is eligible when:
    /// 1. `floor_vc[node_id] >= logical` — every registered client has seen the add.
    /// 2. `deleted_at_ms > 0 && deleted_at_ms + grace_ms <= now_ms()`.
    /// 3. No live node references it as its parent.
    ///
    /// Move-log compaction (keep latest record per node) is always safe and
    /// unchanged from `compact_before`.
    ///
    /// Returns `(nodes_removed, move_records_removed)`.
    pub fn compact_safe(&mut self, floor_vc: &VectorClock, grace_ms: u64) -> (usize, usize) {
        let now = now_ms();
        let live_parent_ids: std::collections::BTreeSet<Option<HybridLogicalClock>> = self
            .nodes
            .iter()
            .filter(|n| !n.deleted)
            .map(|n| n.parent_id)
            .collect();

        let nodes_before = self.nodes.len();
        self.nodes.retain(|n| {
            if !n.deleted {
                return true;
            }
            if live_parent_ids.contains(&Some(n.id)) {
                return true;
            }
            if n.deleted_at_ms == 0 {
                return true;
            }
            if n.deleted_at_ms.saturating_add(grace_ms) > now {
                return true;
            }
            let all_seen = floor_vc.get(n.id.node_id) >= u32::from(n.id.logical);
            !all_seen
        });
        let nodes_removed = nodes_before - self.nodes.len();

        // Move-log compaction: retain only the latest record per node (unchanged).
        let mut seen_nodes = std::collections::BTreeSet::new();
        let mut to_keep = std::collections::BTreeSet::new();
        for (i, record) in self.move_log.iter().enumerate().rev() {
            if seen_nodes.insert(record.node_id) {
                to_keep.insert(i);
            }
        }
        let records_before = self.move_log.len();
        let mut idx = 0;
        self.move_log.retain(|_| {
            let keep = to_keep.contains(&idx);
            idx += 1;
            keep
        });
        let records_removed = records_before - self.move_log.len();

        (nodes_removed, records_removed)
    }

    /// Insert a MoveRecord into move_log, keeping it sorted by op_id ascending.
    fn insert_move_record(&mut self, record: MoveRecord) {
        let idx = self.move_log.partition_point(|r| r.op_id < record.op_id);
        self.move_log.insert(idx, record);
    }
}

impl Crdt for TreeCrdt {
    type Op = TreeOp;
    type Delta = TreeDelta;
    type Value = TreeValue;

    fn apply(&mut self, op: TreeOp) -> Result<Option<TreeDelta>, CrdtError> {
        match op {
            TreeOp::AddNode {
                id,
                parent_id,
                ref position,
                ref value,
            } => {
                // Idempotent: skip if already present.
                if self.nodes.iter().any(|n| n.id == id) {
                    return Ok(None);
                }
                self.nodes.push(TreeNode {
                    id,
                    parent_id,
                    position: position.clone(),
                    value: value.clone(),
                    deleted: false,
                    updated_at: id,
                    deleted_at_ms: 0,
                });
                Ok(Some(TreeDelta {
                    ops: vec![op],
                    ..Default::default()
                }))
            }

            TreeOp::MoveNode {
                op_id,
                node_id,
                new_parent_id,
                ref new_position,
            } => {
                // Idempotent: skip if this exact move op was already applied.
                if self.move_log.iter().any(|r| r.op_id == op_id) {
                    return Ok(None);
                }
                // Node must exist to move it.
                let node_idx = match self.nodes.iter().position(|n| n.id == node_id) {
                    Some(i) => i,
                    None => return Ok(None),
                };
                // Cycle prevention: refuse if new_parent is a descendant of node_id.
                if new_parent_id
                    .is_some_and(|new_pid| new_pid == node_id || self.is_ancestor(new_pid, node_id))
                {
                    let discarded = DiscardedMove {
                        node_id,
                        attempted_parent_id: new_parent_id,
                        attempted_position: new_position.clone(),
                        actual_parent_id: self.nodes[node_idx].parent_id,
                        actual_position: self.nodes[node_idx].position.clone(),
                    };
                    return Ok(Some(TreeDelta {
                        ops: vec![],
                        discarded_moves: vec![discarded],
                    }));
                }
                let old_parent_id = self.nodes[node_idx].parent_id;
                let old_position = self.nodes[node_idx].position.clone();

                self.nodes[node_idx].parent_id = new_parent_id;
                self.nodes[node_idx].position = new_position.clone();

                self.insert_move_record(MoveRecord {
                    op_id,
                    node_id,
                    new_parent_id,
                    new_position: new_position.clone(),
                    old_parent_id,
                    old_position,
                });

                Ok(Some(TreeDelta {
                    ops: vec![op],
                    ..Default::default()
                }))
            }

            TreeOp::UpdateNode {
                id,
                ref value,
                updated_at,
            } => {
                match self.nodes.iter_mut().find(|n| n.id == id) {
                    None => Ok(None),
                    Some(node) => {
                        // LWW: only accept if this update is strictly newer.
                        if updated_at <= node.updated_at {
                            return Ok(None);
                        }
                        node.value = value.clone();
                        node.updated_at = updated_at;
                        Ok(Some(TreeDelta {
                            ops: vec![op],
                            ..Default::default()
                        }))
                    }
                }
            }

            TreeOp::DeleteNode { id } => match self.nodes.iter_mut().find(|n| n.id == id) {
                None => Ok(None),
                Some(node) if node.deleted => Ok(None),
                Some(node) => {
                    node.deleted = true;
                    node.deleted_at_ms = now_ms();
                    Ok(Some(TreeDelta {
                        ops: vec![TreeOp::DeleteNode { id }],
                        ..Default::default()
                    }))
                }
            },
        }
    }

    fn merge(&mut self, other: &TreeCrdt) {
        // Step 1: merge node set (excluding parent pointers — those come from move log replay).
        for node in &other.nodes {
            match self.nodes.iter_mut().find(|n| n.id == node.id) {
                Some(existing) => {
                    // Tombstone propagates.
                    if node.deleted {
                        existing.deleted = true;
                    }
                    // LWW on value.
                    if node.updated_at > existing.updated_at {
                        existing.value = node.value.clone();
                        existing.updated_at = node.updated_at;
                    }
                }
                None => {
                    self.nodes.push(node.clone());
                }
            }
        }

        // Step 2: build the combined move log (self + other), sorted by op_id.
        let mut combined_log: Vec<MoveRecord> = self.move_log.clone();
        for record in &other.move_log {
            if !combined_log.iter().any(|r| r.op_id == record.op_id) {
                let idx = combined_log.partition_point(|r| r.op_id < record.op_id);
                combined_log.insert(idx, record.clone());
            }
        }

        if combined_log == self.move_log {
            // No new move ops — parent pointers are already correct.
            return;
        }

        // Step 3: reset all parent pointers to the original AddNode state, then
        // replay the combined log in op_id order. This ensures all replicas
        // converge to the same tree regardless of the order they received moves.
        //
        // "Original state" = the parent_id that came from the AddNode op,
        // which we recover by resetting to the AddNode's parent (stored in the
        // first MoveRecord's old_parent_id if it exists, else the node's
        // current parent when no moves have been applied). Since we keep
        // AddNode parent_id in the node struct at creation time, and moves
        // only update parent_id, we need to reconstruct it.
        //
        // Simplest correct approach: reset to AddNode parent (None for
        // root-level, or the parent given at creation). We approximate this by
        // rewinding to the old_parent_id of the earliest MoveRecord per node,
        // then replaying from there.

        // Build a map: node_id -> original parent_id (before any moves).
        let mut original_parents: std::collections::BTreeMap<
            HybridLogicalClock,
            Option<HybridLogicalClock>,
        > = std::collections::BTreeMap::new();
        for node in &self.nodes {
            original_parents.insert(node.id, node.parent_id);
        }
        // Walk the combined log forward to find the oldest record per node.
        // The old_parent_id of the first move on each node is the AddNode parent.
        let mut first_move_seen: std::collections::BTreeSet<HybridLogicalClock> =
            std::collections::BTreeSet::new();
        for record in &combined_log {
            if first_move_seen.insert(record.node_id) {
                original_parents.insert(record.node_id, record.old_parent_id);
            }
        }

        // Reset parent pointers to original AddNode state.
        for node in &mut self.nodes {
            if let Some(orig) = original_parents.get(&node.id) {
                node.parent_id = *orig;
            }
        }

        // Clear move log — we will rebuild it via replay.
        self.move_log.clear();

        // Replay all moves in op_id order with cycle detection.
        for record in &combined_log {
            let _ = self.apply(TreeOp::MoveNode {
                op_id: record.op_id,
                node_id: record.node_id,
                new_parent_id: record.new_parent_id,
                new_position: record.new_position.clone(),
            });
        }
    }

    fn merge_delta(&mut self, delta: TreeDelta) {
        for op in delta.ops {
            let _ = self.apply(op);
        }
    }

    fn delta_since(&self, since: &VectorClock) -> Option<TreeDelta> {
        let mut ops: Vec<TreeOp> = Vec::new();

        for node in &self.nodes {
            let seen = since.get(node.id.node_id);
            let logical = u32::from(node.id.logical);

            if seen < logical {
                ops.push(TreeOp::AddNode {
                    id: node.id,
                    parent_id: node.parent_id,
                    position: node.position.clone(),
                    // Emit the original value; if updated_at > id, we emit an
                    // UpdateNode below (after this block) to carry the latest value.
                    value: node.value.clone(),
                });
                if node.deleted {
                    ops.push(TreeOp::DeleteNode { id: node.id });
                }
            } else if node.updated_at != node.id {
                // Node was seen (AddNode received), but a subsequent UpdateNode may
                // not have been. Emit UpdateNode if the client hasn't seen updated_at.
                let seen_update = since.get(node.updated_at.node_id);
                let logical_update = u32::from(node.updated_at.logical);
                if seen_update < logical_update {
                    ops.push(TreeOp::UpdateNode {
                        id: node.id,
                        value: node.value.clone(),
                        updated_at: node.updated_at,
                    });
                }
            }
        }

        // Include unseen move ops.
        for record in &self.move_log {
            let seen = since.get(record.op_id.node_id);
            let logical = u32::from(record.op_id.logical);
            if seen < logical {
                ops.push(TreeOp::MoveNode {
                    op_id: record.op_id,
                    node_id: record.node_id,
                    new_parent_id: record.new_parent_id,
                    new_position: record.new_position.clone(),
                });
            }
        }

        if ops.is_empty() {
            None
        } else {
            Some(TreeDelta {
                ops,
                ..Default::default()
            })
        }
    }

    fn value(&self) -> TreeValue {
        // Collect root nodes (parent_id == None, not deleted).
        let mut roots: Vec<&TreeNode> = self
            .nodes
            .iter()
            .filter(|n| !n.deleted && n.parent_id.is_none())
            .collect();

        roots.sort_by(|a, b| a.position.cmp(&b.position).then_with(|| b.id.cmp(&a.id)));

        TreeValue {
            roots: roots.iter().map(|n| self.build_node_value(n)).collect(),
        }
    }

    fn is_empty(&self) -> bool {
        self.nodes.iter().all(|n| n.deleted)
    }
}

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

    fn hlc(wall_ms: u64, logical: u16, node_id: u64) -> HybridLogicalClock {
        HybridLogicalClock {
            wall_ms,
            logical,
            node_id,
        }
    }

    fn add(
        tree: &mut TreeCrdt,
        id: HybridLogicalClock,
        parent_id: Option<HybridLogicalClock>,
        pos: &str,
        val: &str,
    ) {
        tree.apply(TreeOp::AddNode {
            id,
            parent_id,
            position: pos.to_string(),
            value: val.to_string(),
        })
        .unwrap();
    }

    #[allow(dead_code)]
    fn root_ids(tree: &TreeCrdt) -> Vec<String> {
        tree.value().roots.iter().map(|n| n.id.clone()).collect()
    }

    #[test]
    fn add_node_idempotent() {
        let mut t = TreeCrdt::default();
        let id = hlc(1, 1, 1);
        add(&mut t, id, None, "a0", "Root");
        let delta = t
            .apply(TreeOp::AddNode {
                id,
                parent_id: None,
                position: "a0".into(),
                value: "Root".into(),
            })
            .unwrap();
        assert!(delta.is_none(), "duplicate add must be a no-op");
        assert_eq!(t.nodes.len(), 1);
    }

    #[test]
    fn add_children_ordered() {
        let mut t = TreeCrdt::default();
        let parent = hlc(1, 1, 1);
        let c1 = hlc(1, 2, 1);
        let c2 = hlc(1, 3, 1);
        let c3 = hlc(1, 4, 1);
        add(&mut t, parent, None, "a0", "Parent");
        add(&mut t, c1, Some(parent), "b0", "C");
        add(&mut t, c2, Some(parent), "a0", "A");
        add(&mut t, c3, Some(parent), "a0", "B"); // same position as c2, higher id → comes first

        let children = &t.value().roots[0].children;
        assert_eq!(children.len(), 3);
        // a0 entries first (c3 before c2 by id desc), then b0
        assert_eq!(children[0].value, "B");
        assert_eq!(children[1].value, "A");
        assert_eq!(children[2].value, "C");
    }

    #[test]
    fn delete_node_tombstone() {
        let mut t = TreeCrdt::default();
        let id = hlc(1, 1, 1);
        add(&mut t, id, None, "a0", "Root");
        t.apply(TreeOp::DeleteNode { id }).unwrap();

        assert!(t.is_empty());
        assert_eq!(t.nodes.len(), 1); // tombstone kept
    }

    #[test]
    fn delete_preserves_children() {
        let mut t = TreeCrdt::default();
        let parent = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        add(&mut t, parent, None, "a0", "Parent");
        add(&mut t, child, Some(parent), "a0", "Child");

        t.apply(TreeOp::DeleteNode { id: parent }).unwrap();

        // Child still exists but parent is deleted → child's parent_id is
        // still Some(parent). value() hides both since parent is deleted
        // and child is orphaned (parent not in visible tree).
        let v = t.value();
        assert!(v.roots.is_empty(), "deleted parent must not appear");
    }

    #[test]
    fn move_node_basic() {
        let mut t = TreeCrdt::default();
        let r1 = hlc(1, 1, 1);
        let r2 = hlc(1, 2, 1);
        let child = hlc(1, 3, 1);
        add(&mut t, r1, None, "a0", "R1");
        add(&mut t, r2, None, "b0", "R2");
        add(&mut t, child, Some(r1), "a0", "Child");

        let op_id = hlc(2, 1, 1);
        t.apply(TreeOp::MoveNode {
            op_id,
            node_id: child,
            new_parent_id: Some(r2),
            new_position: "a0".into(),
        })
        .unwrap();

        let v = t.value();
        assert_eq!(v.roots[0].children.len(), 0, "r1 should have no children");
        assert_eq!(v.roots[1].children.len(), 1, "r2 should have child");
        assert_eq!(v.roots[1].children[0].value, "Child");
    }

    #[test]
    fn move_node_cycle_prevention() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        let grandchild = hlc(1, 3, 1);
        add(&mut t, root, None, "a0", "Root");
        add(&mut t, child, Some(root), "a0", "Child");
        add(&mut t, grandchild, Some(child), "a0", "Grandchild");

        // Try to move root under grandchild — would create a cycle.
        let op_id = hlc(2, 1, 1);
        let delta = t
            .apply(TreeOp::MoveNode {
                op_id,
                node_id: root,
                new_parent_id: Some(grandchild),
                new_position: "a0".into(),
            })
            .unwrap();

        // cycle-creating move returns a delta with one discarded_move entry (no ops)
        let delta = delta.expect("cycle move must return a delta with discarded_moves");
        assert!(delta.ops.is_empty(), "discarded move must have no ops");
        assert_eq!(delta.discarded_moves.len(), 1);
        assert_eq!(delta.discarded_moves[0].node_id, root);
        assert_eq!(
            delta.discarded_moves[0].attempted_parent_id,
            Some(grandchild)
        );
        // root is still at the top level
        assert_eq!(t.value().roots.len(), 1);
        assert_eq!(t.value().roots[0].value, "Root");
    }

    #[test]
    fn move_node_self_cycle_prevention() {
        let mut t = TreeCrdt::default();
        let id = hlc(1, 1, 1);
        add(&mut t, id, None, "a0", "Node");

        let delta = t
            .apply(TreeOp::MoveNode {
                op_id: hlc(2, 1, 1),
                node_id: id,
                new_parent_id: Some(id), // move under itself
                new_position: "a0".into(),
            })
            .unwrap();
        let delta = delta.expect("self-cycle move must return a delta with discarded_moves");
        assert!(delta.ops.is_empty());
        assert_eq!(delta.discarded_moves.len(), 1);
        assert_eq!(delta.discarded_moves[0].node_id, id);
    }

    #[test]
    fn concurrent_moves_converge() {
        // Two replicas start with the same state.
        // Replica 1 moves child under r2.
        // Replica 2 moves child under r3.
        // After merge both should have the same parent for child.
        let mut base = TreeCrdt::default();
        let r1 = hlc(1, 1, 1);
        let r2 = hlc(1, 2, 1);
        let r3 = hlc(1, 3, 1);
        let child = hlc(1, 4, 1);
        add(&mut base, r1, None, "a0", "R1");
        add(&mut base, r2, None, "b0", "R2");
        add(&mut base, r3, None, "c0", "R3");
        add(&mut base, child, Some(r1), "a0", "Child");

        let mut t1 = base.clone();
        let mut t2 = base.clone();

        // op1 (node 1, higher HLC) moves child to r2
        t1.apply(TreeOp::MoveNode {
            op_id: hlc(10, 1, 1),
            node_id: child,
            new_parent_id: Some(r2),
            new_position: "a0".into(),
        })
        .unwrap();

        // op2 (node 2, lower HLC) moves child to r3
        t2.apply(TreeOp::MoveNode {
            op_id: hlc(10, 1, 2),
            node_id: child,
            new_parent_id: Some(r3),
            new_position: "a0".into(),
        })
        .unwrap();

        let mut m1 = t1.clone();
        m1.merge(&t2);
        let mut m2 = t2.clone();
        m2.merge(&t1);

        // Both replicas must agree on where child ended up.
        let child_parent_m1 = m1.nodes.iter().find(|n| n.id == child).unwrap().parent_id;
        let child_parent_m2 = m2.nodes.iter().find(|n| n.id == child).unwrap().parent_id;
        assert_eq!(
            child_parent_m1, child_parent_m2,
            "concurrent moves must converge"
        );
    }

    #[test]
    fn update_node_lww() {
        let mut t = TreeCrdt::default();
        let id = hlc(1, 1, 1);
        add(&mut t, id, None, "a0", "Original");

        // Newer update wins.
        t.apply(TreeOp::UpdateNode {
            id,
            value: "Updated".into(),
            updated_at: hlc(2, 1, 1),
        })
        .unwrap();
        assert_eq!(t.value().roots[0].value, "Updated");

        // Older update is ignored.
        let delta = t
            .apply(TreeOp::UpdateNode {
                id,
                value: "Stale".into(),
                updated_at: hlc(1, 2, 1),
            })
            .unwrap();
        assert!(delta.is_none(), "stale update must be ignored");
        assert_eq!(t.value().roots[0].value, "Updated");
    }

    #[test]
    fn delta_since_empty_vc() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        add(&mut t, root, None, "a0", "Root");
        add(&mut t, child, Some(root), "a0", "Child");

        let vc = VectorClock::new();
        let delta = t.delta_since(&vc).unwrap();
        // Two AddNode ops.
        assert_eq!(delta.ops.len(), 2);
    }

    #[test]
    fn delta_since_no_changes() {
        let mut t = TreeCrdt::default();
        let id = hlc(1, 1, 1);
        add(&mut t, id, None, "a0", "Root");

        // Build a VC that has seen this node (increment to match logical counter).
        let mut vc = VectorClock::new();
        for _ in 0..u32::from(id.logical) {
            vc.increment(id.node_id);
        }

        let delta = t.delta_since(&vc);
        assert!(delta.is_none(), "client is up-to-date, delta must be None");
    }

    #[test]
    fn msgpack_roundtrip() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        add(&mut t, root, None, "a0", "Root");
        add(&mut t, child, Some(root), "a0", "Child");
        t.apply(TreeOp::MoveNode {
            op_id: hlc(2, 1, 1),
            node_id: child,
            new_parent_id: None,
            new_position: "b0".into(),
        })
        .unwrap();

        let bytes = rmp_serde::encode::to_vec_named(&t).unwrap();
        let t2: TreeCrdt = rmp_serde::decode::from_slice(&bytes).unwrap();
        assert_eq!(t, t2);
    }

    #[test]
    fn compact_removes_isolated_tombstone_nodes() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        add(&mut t, root, None, "a0", "Root");
        add(&mut t, child, Some(root), "a0", "Child");

        // Delete child — it has no children of its own.
        t.apply(TreeOp::DeleteNode { id: child }).unwrap();

        let (nodes_removed, _) = t.compact_before(u64::MAX);
        assert_eq!(
            nodes_removed, 1,
            "deleted leaf with no live children must be GC'd"
        );
        // Root is still live.
        assert_eq!(t.nodes.len(), 1);
        assert_eq!(t.value().roots[0].value, "Root");
    }

    #[test]
    fn compact_keeps_tombstone_with_live_children() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        let grandchild = hlc(1, 3, 1);
        add(&mut t, root, None, "a0", "Root");
        add(&mut t, child, Some(root), "a0", "Child");
        add(&mut t, grandchild, Some(child), "a0", "Grandchild");

        // Delete child — grandchild is still live under it.
        t.apply(TreeOp::DeleteNode { id: child }).unwrap();

        let (nodes_removed, _) = t.compact_before(u64::MAX);
        // Child must be kept because grandchild.parent_id == child.
        assert_eq!(nodes_removed, 0, "tombstone with live child must be kept");
        assert_eq!(t.nodes.len(), 3);
    }

    #[test]
    fn compact_truncates_move_log_to_latest_per_node() {
        let mut t = TreeCrdt::default();
        let r1 = hlc(1, 1, 1);
        let r2 = hlc(1, 2, 1);
        let child = hlc(1, 3, 1);
        add(&mut t, r1, None, "a0", "R1");
        add(&mut t, r2, None, "b0", "R2");
        add(&mut t, child, Some(r1), "a0", "Child");

        // Move child three times.
        t.apply(TreeOp::MoveNode {
            op_id: hlc(2, 1, 1),
            node_id: child,
            new_parent_id: Some(r2),
            new_position: "a0".into(),
        })
        .unwrap();
        t.apply(TreeOp::MoveNode {
            op_id: hlc(3, 1, 1),
            node_id: child,
            new_parent_id: None,
            new_position: "c0".into(),
        })
        .unwrap();
        t.apply(TreeOp::MoveNode {
            op_id: hlc(4, 1, 1),
            node_id: child,
            new_parent_id: Some(r1),
            new_position: "z0".into(),
        })
        .unwrap();

        assert_eq!(t.move_log.len(), 3);

        let (_, records_removed) = t.compact_before(u64::MAX);
        assert_eq!(
            records_removed, 2,
            "only latest move per node should be kept"
        );
        assert_eq!(t.move_log.len(), 1);
        // The remaining record should be the last move (op_id = hlc(4,1,1)).
        assert_eq!(t.move_log[0].op_id, hlc(4, 1, 1));
        // Tree structure must be unchanged.
        assert_eq!(t.value().roots[0].children[0].value, "Child");
    }

    #[test]
    fn compact_noop_when_nothing_to_gc() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        add(&mut t, root, None, "a0", "Root");
        let (nodes, records) = t.compact_before(u64::MAX);
        assert_eq!(nodes, 0);
        assert_eq!(records, 0);
    }


    fn make_floor_vc(node_id: u64, up_to_logical: u16) -> VectorClock {
        let mut vc = VectorClock::new();
        for _ in 0..u32::from(up_to_logical) {
            vc.increment(node_id);
        }
        vc
    }

    fn backdate_deleted(t: &mut TreeCrdt, id: HybridLogicalClock) {
        if let Some(n) = t.nodes.iter_mut().find(|n| n.id == id) {
            n.deleted_at_ms = 1; // very old
        }
    }

    #[test]
    fn compact_safe_removes_isolated_tombstone_all_clients_seen() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1); // node_id=1, logical=1
        add(&mut t, root, None, "a0", "Root");
        t.apply(TreeOp::DeleteNode { id: root }).unwrap();
        backdate_deleted(&mut t, root);

        // floor_vc covers node_id=1 up to logical=1 → all clients have seen the insert
        let floor = make_floor_vc(1, 1);
        let (nodes_removed, _) = t.compact_safe(&floor, 0);
        assert_eq!(nodes_removed, 1, "isolated tombstone seen by all should be GC'd");
        assert_eq!(t.nodes.len(), 0);
    }

    #[test]
    fn compact_safe_keeps_tombstone_if_client_has_not_seen_insert() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        add(&mut t, root, None, "a0", "Root");
        t.apply(TreeOp::DeleteNode { id: root }).unwrap();
        backdate_deleted(&mut t, root);

        // floor_vc is empty → no client has seen anything
        let floor = VectorClock::new();
        let (nodes_removed, _) = t.compact_safe(&floor, 0);
        assert_eq!(nodes_removed, 0, "tombstone must be kept when floor doesn't cover its insert");
    }

    #[test]
    fn compact_safe_keeps_tombstone_with_live_children() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        let child = hlc(1, 2, 1);
        add(&mut t, root, None, "a0", "Root");
        add(&mut t, child, Some(root), "a0", "Child");
        t.apply(TreeOp::DeleteNode { id: root }).unwrap();
        backdate_deleted(&mut t, root);

        // floor covers both nodes
        let floor = make_floor_vc(1, 2);
        let (nodes_removed, _) = t.compact_safe(&floor, 0);
        assert_eq!(nodes_removed, 0, "parent tombstone with live children must not be GC'd");
    }

    #[test]
    fn compact_safe_keeps_tombstone_within_grace_window() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        add(&mut t, root, None, "a0", "Root");
        t.apply(TreeOp::DeleteNode { id: root }).unwrap();
        // Set deleted_at_ms to 1 second ago — within any reasonable grace window.
        if let Some(n) = t.nodes.iter_mut().find(|n| n.id == root) {
            n.deleted_at_ms = now_ms().saturating_sub(1_000);
        }

        let floor = make_floor_vc(1, 1);
        let grace_24h = 24 * 60 * 60 * 1_000;
        let (nodes_removed, _) = t.compact_safe(&floor, grace_24h);
        assert_eq!(nodes_removed, 0, "tombstone within grace window must not be GC'd");
    }

    #[test]
    fn compact_safe_keeps_tombstone_with_zero_deleted_at() {
        let mut t = TreeCrdt::default();
        let root = hlc(1, 1, 1);
        add(&mut t, root, None, "a0", "Root");
        t.apply(TreeOp::DeleteNode { id: root }).unwrap();
        // Force deleted_at_ms to 0 (simulates a tombstone from before we tracked deletion time).
        if let Some(n) = t.nodes.iter_mut().find(|n| n.id == root) {
            n.deleted_at_ms = 0;
        }

        let floor = make_floor_vc(1, 1);
        let (nodes_removed, _) = t.compact_safe(&floor, 0);
        assert_eq!(nodes_removed, 0, "tombstone with unknown age must never be GC'd");
    }
}
