silk/

graph.rs

use std::collections::{BTreeMap, HashMap, HashSet};

use crate::clock::LamportClock;
use crate::entry::{Entry, GraphOp, Hash, Value};
use crate::ontology::{Ontology, ValidationMode};

/// A materialized node in the graph.
#[derive(Debug, Clone, PartialEq)]
pub struct Node {
    pub node_id: String,
    pub node_type: String,
    pub subtype: Option<String>,
    pub label: String,
    pub properties: BTreeMap<String, Value>,
    /// Per-property clocks for LWW conflict resolution.
    /// Each property key tracks the clock of its last write, so
    /// concurrent updates to different properties don't conflict.
    pub property_clocks: HashMap<String, LamportClock>,
    /// Clock of the entry that last modified this node.
    /// Used for add-wins semantics and label LWW.
    pub last_clock: LamportClock,
    /// Clock of the most recent AddNode for this node.
    /// Used for add-wins semantics: remove only wins if its clock
    /// is strictly greater than last_add_clock.
    pub last_add_clock: LamportClock,
    /// Whether this node has been tombstoned (removed).
    pub tombstoned: bool,
}

/// A materialized edge in the graph.
#[derive(Debug, Clone, PartialEq)]
pub struct Edge {
    pub edge_id: String,
    pub edge_type: String,
    pub source_id: String,
    pub target_id: String,
    pub properties: BTreeMap<String, Value>,
    /// Per-property clocks for LWW conflict resolution.
    pub property_clocks: HashMap<String, LamportClock>,
    pub last_clock: LamportClock,
    /// Clock of the most recent AddEdge for this edge.
    pub last_add_clock: LamportClock,
    pub tombstoned: bool,
}

/// Why an entry was quarantined (S1).
///
/// The validators produce a precise diagnosis — type name, property,
/// expected type, allowed set, constraint name — and every quarantine site
/// used to drop it one line before the operator needed it, leaving a bare
/// hash and no way to tell an unknown type from a constraint violation.
#[derive(Debug, Clone, PartialEq)]
pub struct QuarantineRecord {
    /// The operation kind that was rejected, e.g. "add_node".
    pub op_kind: String,
    /// The validator's own message.
    pub reason: String,
    /// Hex content hash of the ontology this decision was made against, so a
    /// stale diagnosis is recognizable after the schema moves.
    pub ontology_hash: String,
}

/// Materialized graph — derived from the op log.
///
/// Provides fast queries without replaying the full log.
/// Updated incrementally as new entries arrive, or rebuilt
/// from scratch by replaying the entire op log.
///
/// CRDT semantics:
/// - **Add-wins** for topology (concurrent add + remove → node/edge exists)
/// - **LWW** (Last-Writer-Wins) per property key (highest Lamport clock wins)
/// - **Tombstones** for deletes (mark as deleted, don't physically remove)
pub struct MaterializedGraph {
    /// node_id → Node
    pub nodes: HashMap<String, Node>,
    /// edge_id → Edge
    pub edges: HashMap<String, Edge>,
    /// node_id → set of outgoing edge_ids
    pub outgoing: HashMap<String, HashSet<String>>,
    /// node_id → set of incoming edge_ids
    pub incoming: HashMap<String, HashSet<String>>,
    /// node_type → set of node_ids (type index)
    pub by_type: HashMap<String, HashSet<String>>,
    /// The ontology (for validation during materialization)
    pub ontology: Ontology,
    /// The ontology this graph started from, before any `ExtendOntology` was
    /// replayed. `rebuild` resets to it so that replaying the log is
    /// idempotent: without this, an extension already folded into `ontology`
    /// is re-merged on rebuild, fails as a duplicate, and quarantines the
    /// store's own schema entry. It also makes the effective ontology a
    /// function of (base, oplog) rather than of how many rebuilds have run,
    /// which is what I-06's proof assumes when it says both peers replay
    /// "against the same evolved ontology".
    base_ontology: Ontology,
    /// R-02: entries that failed ontology validation during apply().
    /// These entries exist in the oplog (for CRDT convergence) but are
    /// invisible in the materialized graph. Grow-only within a single
    /// materialization pass. Cleared and rebuilt on `rebuild()` — this
    /// allows previously-quarantined entries to be re-evaluated when the
    /// ontology evolves (e.g., after ExtendOntology arrives via sync).
    pub quarantined: HashMap<Hash, QuarantineRecord>,
    /// S9: edges that arrived before their endpoints. Held (and reported
    /// quarantined) until both endpoints materialize, then re-evaluated with
    /// full endpoint-type validation. Previously such edges were admitted
    /// unvalidated on the promise of a rebuild that only fires when the merge
    /// batch happens to carry a schema change.
    pending_edges: HashMap<Hash, Entry>,
}

impl MaterializedGraph {
    /// Create an empty materialized graph with the given ontology.
    pub fn new(ontology: Ontology) -> Self {
        Self {
            nodes: HashMap::new(),
            edges: HashMap::new(),
            outgoing: HashMap::new(),
            incoming: HashMap::new(),
            by_type: HashMap::new(),
            base_ontology: ontology.clone(),
            ontology,
            quarantined: HashMap::new(),
            pending_edges: HashMap::new(),
        }
    }

    /// Apply a single entry to the graph (incremental materialization).
    ///
    /// R-02: Validates AddNode/AddEdge payloads against the ontology.
    /// Invalid entries are quarantined (added to `self.quarantined`) and
    /// skipped for materialization. They remain in the oplog for CRDT
    /// convergence — quarantine is a graph-layer concern, not an oplog concern.
    pub fn apply(&mut self, entry: &Entry) {
        self.apply_entry(entry, ValidationMode::Full)
    }

    fn apply_entry(&mut self, entry: &Entry, mode: ValidationMode) {
        self.apply_inner(entry, mode, true)
    }

    /// Core of `apply`. `mode` is `SkipRequired` only for a checkpoint's inner
    /// ops, because `build_checkpoint_ops` emits `AddNode` with an empty
    /// property map by design (EXP-02) with the values following as separate
    /// `UpdateProperty` ops — enforcing required-presence there would
    /// quarantine every such entity (Bug 14b). S2: the bypass is limited to
    /// that one rule. Declared types, constraints and edge endpoints are
    /// enforced on checkpoint inner ops like anything else.
    ///
    /// The checkpoint's own trustworthiness is established at the merge
    /// boundary, not here: a foreign checkpoint may only enter a store whose
    /// oplog is still genesis-only (H1). Replay has no valid baseline of its
    /// own to judge against, since the ontology it carries is precisely what
    /// the rest of the replay is relative to.
    /// `adopt_checkpoint_ontology` is false only during `rebuild`'s second
    /// pass, where the effective ontology has already been folded and a
    /// checkpoint's inner `DefineOntology` must not clobber the extensions
    /// that causally follow it.
    fn apply_inner(
        &mut self,
        entry: &Entry,
        mode: ValidationMode,
        adopt_checkpoint_ontology: bool,
    ) {
        macro_rules! quarantine {
            ($kind:expr, $reason:expr) => {{
                let record = QuarantineRecord {
                    op_kind: $kind.to_string(),
                    reason: $reason.to_string(),
                    ontology_hash: hex::encode(self.ontology.content_hash()),
                };
                self.quarantined.insert(entry.hash, record);
            }};
        }
        match &entry.payload {
            GraphOp::Checkpoint { ops, op_clocks, .. } => {
                // R-08: Replay synthetic ops to restore graph state.
                // Bug 6 fix: use per-op clocks (preserves LWW metadata).
                for (i, op) in ops.iter().enumerate() {
                    // Bug 14 fix: compaction folds every ExtendOntology into the
                    // checkpoint's inner DefineOntology. Apply it — otherwise
                    // extension-typed entities fail validation and quarantine on
                    // replay, and a reopened store materializes without them.
                    // The oplog is authoritative; a declared ontology only seeds
                    // new stores.
                    if let GraphOp::DefineOntology { ontology } = op {
                        if adopt_checkpoint_ontology {
                            self.ontology = ontology.clone();
                        }
                        continue;
                    }
                    let clock = if i < op_clocks.len() {
                        LamportClock::with_values(&entry.author, op_clocks[i].0, op_clocks[i].1)
                    } else {
                        entry.clock.clone() // fallback for old checkpoints without op_clocks
                    };
                    let synthetic = Entry::new(op.clone(), vec![], vec![], clock, &entry.author);
                    self.apply_entry(&synthetic, ValidationMode::SkipRequired);
                }
                // H6: the checkpoint itself is now materialized.
                self.quarantined.remove(&entry.hash);
            }
            GraphOp::DefineOntology { .. } => {
                // Genesis — nothing to materialize.
            }
            GraphOp::ExtendOntology { extension } => {
                if let Err(e) = self.ontology.merge_extension(extension) {
                    quarantine!("extend_ontology", e);
                } else {
                    self.quarantined.remove(&entry.hash);
                }
            }
            GraphOp::AddNode {
                node_id,
                node_type,
                subtype,
                label,
                properties,
            } => {
                // R-02: validate against ontology, quarantine if invalid
                if let Err(e) = self.ontology.validate_node_mode(
                    node_type,
                    subtype.as_deref(),
                    properties,
                    mode,
                ) {
                    quarantine!("add_node", e);
                    return;
                }
                // H6: quarantine must be a function of the oplog, not of sync
                // history. An entry that validates now is not quarantined,
                // even if a previous pass rejected it.
                self.quarantined.remove(&entry.hash);
                self.apply_add_node(
                    node_id,
                    node_type,
                    subtype.as_deref(),
                    label,
                    properties,
                    &entry.clock,
                );
                // S9: an endpoint just arrived — edges held for it can proceed.
                self.retry_pending_edges();
            }
            GraphOp::AddEdge {
                edge_id,
                edge_type,
                source_id,
                target_id,
                properties,
            } => {
                // R-02: validate edge type exists.
                if !self.ontology.edge_types.contains_key(edge_type.as_str()) {
                    quarantine!(
                        "add_edge",
                        crate::ontology::ValidationError::UnknownEdgeType(edge_type.clone())
                    );
                    return;
                }
                // Bug 13 fix: validate source/target type constraints when both nodes
                // are materialized. S9: when an endpoint is missing the edge is
                // held pending instead of admitted unvalidated, because the
                // rebuild the old comment promised only fires when the batch
                // happens to carry a schema change.
                match (
                    self.nodes.get(source_id.as_str()),
                    self.nodes.get(target_id.as_str()),
                ) {
                    (Some(src), Some(tgt)) => {
                        if let Err(e) = self.ontology.validate_edge_mode(
                            edge_type,
                            &src.node_type,
                            &tgt.node_type,
                            properties,
                            mode,
                        ) {
                            quarantine!("add_edge", e);
                            return;
                        }
                    }
                    _ => {
                        self.pending_edges.insert(entry.hash, entry.clone());
                        quarantine!(
                            "add_edge",
                            format!(
                                "endpoint not yet materialized (source '{source_id}', \
                                 target '{target_id}'); held pending until both arrive"
                            )
                        );
                        return;
                    }
                }
                self.quarantined.remove(&entry.hash);
                self.apply_add_edge(
                    edge_id,
                    edge_type,
                    source_id,
                    target_id,
                    properties,
                    &entry.clock,
                );
            }
            GraphOp::UpdateProperty {
                entity_id,
                key,
                value,
            } => {
                // H2: this arm applied the value with no validation at all, so
                // a peer's UpdateProperty entered the graph regardless of the
                // local ontology — the v0.1.6 bug reopened on the sync side.
                // The entity's declared type is the one the graph already
                // holds; if it is not materialized yet there is nothing to
                // validate against and the value is accepted, exactly as the
                // local path does.
                let verdict = if let Some(node) = self.nodes.get(entity_id.as_str()) {
                    self.ontology.validate_property_update(
                        &node.node_type,
                        node.subtype.as_deref(),
                        key,
                        value,
                    )
                } else if let Some(edge) = self.edges.get(entity_id.as_str()) {
                    self.ontology
                        .validate_edge_property_update(&edge.edge_type, key, value)
                } else {
                    Ok(())
                };
                if let Err(e) = verdict {
                    quarantine!("update_property", e);
                    return;
                }
                self.quarantined.remove(&entry.hash);
                self.apply_update_property(entity_id, key, value, &entry.clock);
            }
            GraphOp::RemoveNode { node_id } => {
                self.quarantined.remove(&entry.hash);
                self.apply_remove_node(node_id, &entry.clock);
            }
            GraphOp::RemoveEdge { edge_id } => {
                self.quarantined.remove(&entry.hash);
                self.apply_remove_edge(edge_id, &entry.clock);
            }
            GraphOp::DefineLens { .. } => {
                // Reserved. No materialization — lenses are metadata, not graph state.
            }
        }
    }

    /// Apply a sequence of entries (full rematerialization from op log).
    pub fn apply_all(&mut self, entries: &[&Entry]) {
        for entry in entries {
            self.apply(entry);
        }
    }

    /// Rebuild from scratch: clear everything and replay all entries.
    pub fn rebuild(&mut self, entries: &[&Entry]) {
        self.nodes.clear();
        self.edges.clear();
        self.outgoing.clear();
        self.incoming.clear();
        self.by_type.clear();
        self.quarantined.clear();
        self.pending_edges.clear();

        // Pass 1: fold the effective ontology from the log, starting at the
        // base so each extension merges exactly once.
        //
        // Validity is then judged against the FINAL ontology, not against the
        // ontology as of each entry's position. That is what makes the
        // documented un-quarantine promise true in the ordinary case: an
        // operator receives data whose type is unknown, extends the schema,
        // and the data becomes visible — even though the extension is
        // causally later than the data it rescues. Judging each entry against
        // the schema as of its own position would re-quarantine it on every
        // replay, forever.
        self.ontology = self.base_ontology.clone();
        for entry in entries {
            match &entry.payload {
                GraphOp::DefineOntology { .. } => {}
                GraphOp::ExtendOntology { extension } => {
                    if let Err(e) = self.ontology.merge_extension(extension) {
                        self.quarantined.insert(
                            entry.hash,
                            QuarantineRecord {
                                op_kind: "extend_ontology".to_string(),
                                reason: e.to_string(),
                                ontology_hash: hex::encode(self.ontology.content_hash()),
                            },
                        );
                    }
                }
                GraphOp::Checkpoint { ops, .. } => {
                    for op in ops {
                        if let GraphOp::DefineOntology { ontology } = op {
                            self.ontology = ontology.clone();
                        }
                    }
                }
                _ => {}
            }
        }

        // Pass 2: materialize everything else against that ontology.
        // ExtendOntology entries are already folded and already judged.
        for entry in entries {
            if matches!(entry.payload, GraphOp::ExtendOntology { .. }) {
                continue;
            }
            self.apply_inner(entry, ValidationMode::Full, false);
        }
    }

    /// S9: re-evaluate edges that were waiting on an endpoint. Called whenever
    /// a node materializes. Edges whose endpoints are still missing stay
    /// pending; the rest go through full validation and are admitted or
    /// quarantined on their merits.
    fn retry_pending_edges(&mut self) {
        if self.pending_edges.is_empty() {
            return;
        }
        let ready: Vec<Entry> = self
            .pending_edges
            .values()
            .filter(|entry| match &entry.payload {
                GraphOp::AddEdge {
                    source_id,
                    target_id,
                    ..
                } => {
                    self.nodes.contains_key(source_id.as_str())
                        && self.nodes.contains_key(target_id.as_str())
                }
                _ => false,
            })
            .cloned()
            .collect();
        for entry in ready {
            self.pending_edges.remove(&entry.hash);
            self.apply_entry(&entry, ValidationMode::Full);
        }
    }

    // -- Queries --

    /// Get a node by ID (returns None if not found or tombstoned).
    pub fn get_node(&self, node_id: &str) -> Option<&Node> {
        self.nodes.get(node_id).filter(|n| !n.tombstoned)
    }

    /// Get an edge by ID (returns None if not found or tombstoned).
    pub fn get_edge(&self, edge_id: &str) -> Option<&Edge> {
        self.edges.get(edge_id).filter(|e| !e.tombstoned)
    }

    /// Query all live nodes of a given type, including descendants (RDFS rdfs9).
    /// If "entity" has children "server" and "project", querying "entity" returns all three.
    pub fn nodes_by_type(&self, node_type: &str) -> Vec<&Node> {
        let mut types = vec![node_type.to_string()];
        types.extend(
            self.ontology
                .descendants(node_type)
                .into_iter()
                .map(|s| s.to_string()),
        );
        types
            .iter()
            .flat_map(|t| self.by_type.get(t.as_str()))
            .flatten()
            .filter_map(|id| self.get_node(id))
            .collect()
    }

    /// Query all live nodes of a given subtype.
    pub fn nodes_by_subtype(&self, subtype: &str) -> Vec<&Node> {
        self.nodes
            .values()
            .filter(|n| !n.tombstoned && n.subtype.as_deref() == Some(subtype))
            .collect()
    }

    /// Query nodes by a property value.
    pub fn nodes_by_property(&self, key: &str, value: &Value) -> Vec<&Node> {
        self.nodes
            .values()
            .filter(|n| !n.tombstoned && n.properties.get(key) == Some(value))
            .collect()
    }

    /// Get outgoing edges for a node (only live edges with live endpoints).
    pub fn outgoing_edges(&self, node_id: &str) -> Vec<&Edge> {
        match self.outgoing.get(node_id) {
            Some(edge_ids) => edge_ids
                .iter()
                .filter_map(|eid| self.get_edge(eid))
                .filter(|e| self.is_node_live(&e.target_id))
                .collect(),
            None => vec![],
        }
    }

    /// Get incoming edges for a node (only live edges with live endpoints).
    pub fn incoming_edges(&self, node_id: &str) -> Vec<&Edge> {
        match self.incoming.get(node_id) {
            Some(edge_ids) => edge_ids
                .iter()
                .filter_map(|eid| self.get_edge(eid))
                .filter(|e| self.is_node_live(&e.source_id))
                .collect(),
            None => vec![],
        }
    }

    /// Approximate heap memory used by the materialized graph (bytes).
    /// Uses fixed overhead estimates per node/edge. Does not account for heap
    /// allocations behind String/Vec in property values or allocator fragmentation.
    /// Actual memory may be 2-3x higher for string-heavy graphs.
    pub fn estimated_memory_bytes(&self) -> usize {
        let mut total = 0;
        // Nodes: id string + type string + label + properties + clocks + overhead
        for node in self.nodes.values() {
            total += node.node_id.len() + node.node_type.len() + node.label.len();
            total += node.subtype.as_ref().map_or(0, |s| s.len());
            // Properties: key + estimated value size + clock per property
            for (k, v) in &node.properties {
                total += k.len() + std::mem::size_of_val(v) + 48; // key + value + clock overhead
            }
            total += 128; // fixed struct overhead (clocks, bools, HashMap shells)
        }
        // Edges: similar structure
        for edge in self.edges.values() {
            total += edge.edge_id.len() + edge.edge_type.len();
            total += edge.source_id.len() + edge.target_id.len();
            for (k, v) in &edge.properties {
                total += k.len() + std::mem::size_of_val(v) + 48;
            }
            total += 128;
        }
        // Adjacency indexes: outgoing + incoming (id strings + HashSet overhead)
        for (k, set) in &self.outgoing {
            total += k.len() + set.len() * 32;
        }
        for (k, set) in &self.incoming {
            total += k.len() + set.len() * 32;
        }
        // Type index
        for (k, set) in &self.by_type {
            total += k.len() + set.len() * 32;
        }
        // Quarantine set
        total += self.quarantined.len() * 48;
        total
    }

    /// All live nodes.
    pub fn all_nodes(&self) -> Vec<&Node> {
        self.nodes.values().filter(|n| !n.tombstoned).collect()
    }

    /// All live edges (with live endpoints).
    pub fn all_edges(&self) -> Vec<&Edge> {
        self.edges
            .values()
            .filter(|e| {
                !e.tombstoned && self.is_node_live(&e.source_id) && self.is_node_live(&e.target_id)
            })
            .collect()
    }

    /// Neighbors of a node (connected via outgoing edges).
    pub fn neighbors(&self, node_id: &str) -> Vec<&str> {
        self.outgoing_edges(node_id)
            .iter()
            .map(|e| e.target_id.as_str())
            .collect()
    }

    /// Reverse neighbors (connected via incoming edges).
    pub fn reverse_neighbors(&self, node_id: &str) -> Vec<&str> {
        self.incoming_edges(node_id)
            .iter()
            .map(|e| e.source_id.as_str())
            .collect()
    }

    // -- CRDT application helpers --

    fn apply_add_node(
        &mut self,
        node_id: &str,
        node_type: &str,
        subtype: Option<&str>,
        label: &str,
        properties: &BTreeMap<String, Value>,
        clock: &LamportClock,
    ) {
        if let Some(existing) = self.nodes.get_mut(node_id) {
            // Add-wins: always resurrect from tombstone.
            existing.tombstoned = false;
            // Track the latest add clock for add-wins semantics.
            if clock_wins(clock, &existing.last_add_clock) {
                existing.last_add_clock = clock.clone();
            }
            // LWW merge for label, subtype, and properties.
            if clock_wins(clock, &existing.last_clock) {
                existing.label = label.to_string();
                existing.subtype = subtype.map(|s| s.to_string());
                existing.last_clock = clock.clone();
            }
            merge_properties_lww(
                &mut existing.properties,
                &mut existing.property_clocks,
                properties,
                clock,
            );
        } else {
            let property_clocks: HashMap<String, LamportClock> = properties
                .keys()
                .map(|k| (k.clone(), clock.clone()))
                .collect();
            let node = Node {
                node_id: node_id.to_string(),
                node_type: node_type.to_string(),
                subtype: subtype.map(|s| s.to_string()),
                label: label.to_string(),
                properties: properties.clone(),
                property_clocks,
                last_clock: clock.clone(),
                last_add_clock: clock.clone(),
                tombstoned: false,
            };
            self.by_type
                .entry(node_type.to_string())
                .or_default()
                .insert(node_id.to_string());
            self.nodes.insert(node_id.to_string(), node);
        }
    }

    fn apply_add_edge(
        &mut self,
        edge_id: &str,
        edge_type: &str,
        source_id: &str,
        target_id: &str,
        properties: &BTreeMap<String, Value>,
        clock: &LamportClock,
    ) {
        if let Some(existing) = self.edges.get_mut(edge_id) {
            // Add-wins: always resurrect if tombstoned.
            existing.tombstoned = false;
            if clock_wins(clock, &existing.last_add_clock) {
                existing.last_add_clock = clock.clone();
            }
            if clock_wins(clock, &existing.last_clock) {
                existing.last_clock = clock.clone();
            }
            merge_properties_lww(
                &mut existing.properties,
                &mut existing.property_clocks,
                properties,
                clock,
            );
        } else {
            let property_clocks: HashMap<String, LamportClock> = properties
                .keys()
                .map(|k| (k.clone(), clock.clone()))
                .collect();
            let edge = Edge {
                edge_id: edge_id.to_string(),
                edge_type: edge_type.to_string(),
                source_id: source_id.to_string(),
                target_id: target_id.to_string(),
                properties: properties.clone(),
                property_clocks,
                last_clock: clock.clone(),
                last_add_clock: clock.clone(),
                tombstoned: false,
            };
            self.outgoing
                .entry(source_id.to_string())
                .or_default()
                .insert(edge_id.to_string());
            self.incoming
                .entry(target_id.to_string())
                .or_default()
                .insert(edge_id.to_string());
            self.edges.insert(edge_id.to_string(), edge);
        }
    }

    fn apply_update_property(
        &mut self,
        entity_id: &str,
        key: &str,
        value: &Value,
        clock: &LamportClock,
    ) {
        // Try node first, then edge. Per-property LWW: each key competes
        // only with other writes to the same key, not the entire entity.
        if let Some(node) = self.nodes.get_mut(entity_id) {
            let dominated = node
                .property_clocks
                .get(key)
                .map(|c| clock_wins(clock, c))
                .unwrap_or(true);
            if dominated {
                node.properties.insert(key.to_string(), value.clone());
                node.property_clocks.insert(key.to_string(), clock.clone());
            }
            // Update entity-level clock for add-wins tracking.
            if clock_wins(clock, &node.last_clock) {
                node.last_clock = clock.clone();
            }
        } else if let Some(edge) = self.edges.get_mut(entity_id) {
            let dominated = edge
                .property_clocks
                .get(key)
                .map(|c| clock_wins(clock, c))
                .unwrap_or(true);
            if dominated {
                edge.properties.insert(key.to_string(), value.clone());
                edge.property_clocks.insert(key.to_string(), clock.clone());
            }
            if clock_wins(clock, &edge.last_clock) {
                edge.last_clock = clock.clone();
            }
        }
        // If entity not found, silently ignore (may arrive out of order in sync).
    }

    fn apply_remove_node(&mut self, node_id: &str, clock: &LamportClock) {
        if let Some(node) = self.nodes.get_mut(node_id) {
            // Add-wins: only tombstone if the remove clock is strictly greater
            // than the last add clock. If a concurrent (or later) add exists,
            // the node stays alive.
            if clock_wins(clock, &node.last_add_clock) {
                node.tombstoned = true;
                node.last_clock = clock.clone();
            }
        }
        // Tombstoning a node doesn't physically remove edges — they just become
        // invisible via is_node_live() checks in queries.
    }

    fn apply_remove_edge(&mut self, edge_id: &str, clock: &LamportClock) {
        if let Some(edge) = self.edges.get_mut(edge_id) {
            // Add-wins: only tombstone if remove clock > last add clock.
            if clock_wins(clock, &edge.last_add_clock) {
                edge.tombstoned = true;
                edge.last_clock = clock.clone();
            }
        }
    }

    fn is_node_live(&self, node_id: &str) -> bool {
        self.nodes
            .get(node_id)
            .map(|n| !n.tombstoned)
            .unwrap_or(false)
    }
}

/// Per-property LWW merge: each property from `new_props` competes with
/// existing properties. Higher clock wins per key.
fn merge_properties_lww(
    existing_props: &mut BTreeMap<String, Value>,
    existing_clocks: &mut HashMap<String, LamportClock>,
    new_props: &BTreeMap<String, Value>,
    clock: &LamportClock,
) {
    for (k, v) in new_props {
        let dominated = existing_clocks
            .get(k)
            .map(|c| clock_wins(clock, c))
            .unwrap_or(true);
        if dominated {
            existing_props.insert(k.clone(), v.clone());
            existing_clocks.insert(k.clone(), clock.clone());
        }
    }
}

/// LWW comparison: returns true if `new_clock` wins over `existing_clock`.
/// Uses HybridClock total ordering: (physical_ms, logical, id).
fn clock_wins(new_clock: &LamportClock, existing_clock: &LamportClock) -> bool {
    new_clock.cmp_order(existing_clock) == std::cmp::Ordering::Greater
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::entry::Entry;
    use crate::ontology::{EdgeTypeDef, NodeTypeDef};

    fn test_ontology() -> Ontology {
        Ontology {
            node_types: BTreeMap::from([
                (
                    "entity".into(),
                    NodeTypeDef {
                        description: None,
                        properties: BTreeMap::new(),
                        subtypes: None,
                        parent_type: None,
                    },
                ),
                (
                    "signal".into(),
                    NodeTypeDef {
                        description: None,
                        properties: BTreeMap::new(),
                        subtypes: None,
                        parent_type: None,
                    },
                ),
            ]),
            edge_types: BTreeMap::from([
                (
                    "RUNS_ON".into(),
                    EdgeTypeDef {
                        description: None,
                        source_types: vec!["entity".into()],
                        target_types: vec!["entity".into()],
                        properties: BTreeMap::new(),
                    },
                ),
                (
                    "OBSERVES".into(),
                    EdgeTypeDef {
                        description: None,
                        source_types: vec!["signal".into()],
                        target_types: vec!["entity".into()],
                        properties: BTreeMap::new(),
                    },
                ),
            ]),
        }
    }

    fn make_entry(op: GraphOp, clock_time: u64, author: &str) -> Entry {
        Entry::new(
            op,
            vec![],
            vec![],
            LamportClock::with_values(author, clock_time, 0),
            author,
        )
    }

    // -- test_graph.rs spec from docs/silk.md --

    #[test]
    fn add_node_appears_in_query() {
        let mut g = MaterializedGraph::new(test_ontology());
        let entry = make_entry(
            GraphOp::AddNode {
                node_id: "server-1".into(),
                node_type: "entity".into(),
                label: "Server 1".into(),
                properties: BTreeMap::from([("ip".into(), Value::String("10.0.0.1".into()))]),
                subtype: None,
            },
            1,
            "inst-a",
        );
        g.apply(&entry);

        let node = g.get_node("server-1").unwrap();
        assert_eq!(node.node_type, "entity");
        assert_eq!(node.label, "Server 1");
        assert_eq!(
            node.properties.get("ip"),
            Some(&Value::String("10.0.0.1".into()))
        );
    }

    #[test]
    fn add_edge_creates_adjacency() {
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "svc".into(),
                node_type: "entity".into(),
                label: "svc".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "srv".into(),
                node_type: "entity".into(),
                label: "srv".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            2,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddEdge {
                edge_id: "e1".into(),
                edge_type: "RUNS_ON".into(),
                source_id: "svc".into(),
                target_id: "srv".into(),
                properties: BTreeMap::new(),
            },
            3,
            "inst-a",
        ));

        // Both endpoints know about the edge.
        let out = g.outgoing_edges("svc");
        assert_eq!(out.len(), 1);
        assert_eq!(out[0].target_id, "srv");

        let inc = g.incoming_edges("srv");
        assert_eq!(inc.len(), 1);
        assert_eq!(inc[0].source_id, "svc");

        assert_eq!(g.neighbors("svc"), vec!["srv"]);
    }

    #[test]
    fn update_property_reflected() {
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "cpu".into(),
                value: Value::Float(85.5),
            },
            2,
            "inst-a",
        ));

        let node = g.get_node("s1").unwrap();
        assert_eq!(node.properties.get("cpu"), Some(&Value::Float(85.5)));
    }

    #[test]
    fn remove_node_cascades_edges() {
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "a".into(),
                node_type: "entity".into(),
                label: "a".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "b".into(),
                node_type: "entity".into(),
                label: "b".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            2,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddEdge {
                edge_id: "e1".into(),
                edge_type: "RUNS_ON".into(),
                source_id: "a".into(),
                target_id: "b".into(),
                properties: BTreeMap::new(),
            },
            3,
            "inst-a",
        ));
        assert_eq!(g.all_edges().len(), 1);

        // Remove node 'b' — edge becomes invisible (dangling target).
        g.apply(&make_entry(
            GraphOp::RemoveNode {
                node_id: "b".into(),
            },
            4,
            "inst-a",
        ));
        assert!(g.get_node("b").is_none());
        // Edge still exists but not returned by all_edges (target tombstoned).
        assert_eq!(g.all_edges().len(), 0);
        // Outgoing from 'a' also filters out dangling edges.
        assert_eq!(g.outgoing_edges("a").len(), 0);
    }

    #[test]
    fn remove_edge_preserves_nodes() {
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "a".into(),
                node_type: "entity".into(),
                label: "a".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "b".into(),
                node_type: "entity".into(),
                label: "b".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            2,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddEdge {
                edge_id: "e1".into(),
                edge_type: "RUNS_ON".into(),
                source_id: "a".into(),
                target_id: "b".into(),
                properties: BTreeMap::new(),
            },
            3,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::RemoveEdge {
                edge_id: "e1".into(),
            },
            4,
            "inst-a",
        ));

        // Nodes still exist.
        assert!(g.get_node("a").is_some());
        assert!(g.get_node("b").is_some());
        // Edge is gone.
        assert!(g.get_edge("e1").is_none());
        assert_eq!(g.all_edges().len(), 0);
    }

    #[test]
    fn query_by_type_filters() {
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s2".into(),
                node_type: "entity".into(),
                label: "s2".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            2,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "alert".into(),
                node_type: "signal".into(),
                label: "alert".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            3,
            "inst-a",
        ));

        let entities = g.nodes_by_type("entity");
        assert_eq!(entities.len(), 2);
        let signals = g.nodes_by_type("signal");
        assert_eq!(signals.len(), 1);
        assert_eq!(signals[0].node_id, "alert");
    }

    #[test]
    fn query_by_property_filters() {
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::from([("status".into(), Value::String("alive".into()))]),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s2".into(),
                node_type: "entity".into(),
                label: "s2".into(),
                properties: BTreeMap::from([("status".into(), Value::String("dead".into()))]),
                subtype: None,
            },
            2,
            "inst-a",
        ));

        let alive = g.nodes_by_property("status", &Value::String("alive".into()));
        assert_eq!(alive.len(), 1);
        assert_eq!(alive[0].node_id, "s1");
    }

    #[test]
    fn materialization_from_empty() {
        // Build graph incrementally.
        let mut g1 = MaterializedGraph::new(test_ontology());
        let entries = vec![
            make_entry(
                GraphOp::DefineOntology {
                    ontology: test_ontology(),
                },
                0,
                "inst-a",
            ),
            make_entry(
                GraphOp::AddNode {
                    node_id: "a".into(),
                    node_type: "entity".into(),
                    label: "a".into(),
                    properties: BTreeMap::new(),
                    subtype: None,
                },
                1,
                "inst-a",
            ),
            make_entry(
                GraphOp::AddNode {
                    node_id: "b".into(),
                    node_type: "entity".into(),
                    label: "b".into(),
                    properties: BTreeMap::new(),
                    subtype: None,
                },
                2,
                "inst-a",
            ),
            make_entry(
                GraphOp::AddEdge {
                    edge_id: "e1".into(),
                    edge_type: "RUNS_ON".into(),
                    source_id: "a".into(),
                    target_id: "b".into(),
                    properties: BTreeMap::new(),
                },
                3,
                "inst-a",
            ),
        ];
        for e in &entries {
            g1.apply(e);
        }

        // Rebuild from scratch.
        let mut g2 = MaterializedGraph::new(test_ontology());
        let refs: Vec<&Entry> = entries.iter().collect();
        g2.rebuild(&refs);

        // Same result.
        assert_eq!(g1.all_nodes().len(), g2.all_nodes().len());
        assert_eq!(g1.all_edges().len(), g2.all_edges().len());
        for node in g1.all_nodes() {
            let n2 = g2.get_node(&node.node_id).unwrap();
            assert_eq!(node.node_type, n2.node_type);
            assert_eq!(node.properties, n2.properties);
        }
    }

    #[test]
    fn incremental_equals_full() {
        let entries = vec![
            make_entry(
                GraphOp::DefineOntology {
                    ontology: test_ontology(),
                },
                0,
                "inst-a",
            ),
            make_entry(
                GraphOp::AddNode {
                    node_id: "a".into(),
                    node_type: "entity".into(),
                    label: "a".into(),
                    properties: BTreeMap::from([("x".into(), Value::Int(1))]),
                    subtype: None,
                },
                1,
                "inst-a",
            ),
            make_entry(
                GraphOp::UpdateProperty {
                    entity_id: "a".into(),
                    key: "x".into(),
                    value: Value::Int(2),
                },
                2,
                "inst-a",
            ),
            make_entry(
                GraphOp::AddNode {
                    node_id: "b".into(),
                    node_type: "entity".into(),
                    label: "b".into(),
                    properties: BTreeMap::new(),
                    subtype: None,
                },
                3,
                "inst-a",
            ),
            make_entry(
                GraphOp::AddEdge {
                    edge_id: "e1".into(),
                    edge_type: "RUNS_ON".into(),
                    source_id: "a".into(),
                    target_id: "b".into(),
                    properties: BTreeMap::new(),
                },
                4,
                "inst-a",
            ),
            make_entry(
                GraphOp::RemoveEdge {
                    edge_id: "e1".into(),
                },
                5,
                "inst-a",
            ),
        ];

        // Incremental.
        let mut g_inc = MaterializedGraph::new(test_ontology());
        for e in &entries {
            g_inc.apply(e);
        }

        // Full replay.
        let mut g_full = MaterializedGraph::new(test_ontology());
        let refs: Vec<&Entry> = entries.iter().collect();
        g_full.rebuild(&refs);

        // Property should be 2 (updated).
        assert_eq!(
            g_inc.get_node("a").unwrap().properties.get("x"),
            Some(&Value::Int(2))
        );
        assert_eq!(
            g_full.get_node("a").unwrap().properties.get("x"),
            Some(&Value::Int(2))
        );
        // Edge should be removed.
        assert_eq!(g_inc.all_edges().len(), 0);
        assert_eq!(g_full.all_edges().len(), 0);
    }

    #[test]
    fn lww_concurrent_property_update() {
        // Two instances update the same property — higher clock wins.
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        // inst-a sets status=alive at time 2
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "status".into(),
                value: Value::String("alive".into()),
            },
            2,
            "inst-a",
        ));
        // inst-b sets status=dead at time 3 — wins (higher clock)
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "status".into(),
                value: Value::String("dead".into()),
            },
            3,
            "inst-b",
        ));
        assert_eq!(
            g.get_node("s1").unwrap().properties.get("status"),
            Some(&Value::String("dead".into()))
        );
    }

    #[test]
    fn lww_tiebreak_by_instance_id() {
        // Same clock time — higher instance ID wins.
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        // Both at physical_ms=5, logical=0. Lower id wins → "inst-a" wins.
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "x".into(),
                value: Value::Int(1),
            },
            5,
            "inst-a",
        ));
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "x".into(),
                value: Value::Int(2),
            },
            5,
            "inst-b",
        ));
        // inst-a has lower id → wins the tiebreak → value stays Int(1).
        assert_eq!(
            g.get_node("s1").unwrap().properties.get("x"),
            Some(&Value::Int(1))
        );
    }

    #[test]
    fn lww_per_property_concurrent_different_keys() {
        // Two instances concurrently update DIFFERENT properties at the same
        // clock time. Both updates must be accepted — they don't conflict.
        // This requires per-property LWW, not node-level LWW.
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::from([
                    ("x".into(), Value::Int(0)),
                    ("y".into(), Value::Int(0)),
                ]),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        // inst-a updates "x" at time 3
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "x".into(),
                value: Value::Int(42),
            },
            3,
            "inst-a",
        ));
        // inst-b updates "y" at time 3 (concurrent, different property)
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "y".into(),
                value: Value::Int(99),
            },
            3,
            "inst-b",
        ));

        let node = g.get_node("s1").unwrap();
        // Both updates must be applied — no conflict.
        assert_eq!(
            node.properties.get("x"),
            Some(&Value::Int(42)),
            "update to 'x' must not be rejected by concurrent update to 'y'"
        );
        assert_eq!(
            node.properties.get("y"),
            Some(&Value::Int(99)),
            "update to 'y' must not be rejected by concurrent update to 'x'"
        );
    }

    #[test]
    fn lww_per_property_order_independent() {
        // Same scenario but applied in reverse order — result must be identical.
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::from([
                    ("x".into(), Value::Int(0)),
                    ("y".into(), Value::Int(0)),
                ]),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        // Apply inst-b first this time
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "y".into(),
                value: Value::Int(99),
            },
            3,
            "inst-b",
        ));
        g.apply(&make_entry(
            GraphOp::UpdateProperty {
                entity_id: "s1".into(),
                key: "x".into(),
                value: Value::Int(42),
            },
            3,
            "inst-a",
        ));

        let node = g.get_node("s1").unwrap();
        assert_eq!(node.properties.get("x"), Some(&Value::Int(42)));
        assert_eq!(node.properties.get("y"), Some(&Value::Int(99)));
    }

    #[test]
    fn add_wins_over_remove() {
        // Concurrent add + remove → node should exist (add-wins).
        let mut g = MaterializedGraph::new(test_ontology());
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            1,
            "inst-a",
        ));
        // Remove at time 2.
        g.apply(&make_entry(
            GraphOp::RemoveNode {
                node_id: "s1".into(),
            },
            2,
            "inst-a",
        ));
        assert!(g.get_node("s1").is_none());

        // Re-add at time 3 (add-wins — resurrects).
        g.apply(&make_entry(
            GraphOp::AddNode {
                node_id: "s1".into(),
                node_type: "entity".into(),
                label: "s1 v2".into(),
                properties: BTreeMap::new(),
                subtype: None,
            },
            3,
            "inst-b",
        ));
        let node = g.get_node("s1").unwrap();
        assert_eq!(node.label, "s1 v2");
        assert!(!node.tombstoned);
    }

    /// Bug 14: a checkpoint's inner DefineOntology carries the merged ontology
    /// (compaction folds ExtendOntology entries into it). Replay must apply it,
    /// or entities typed by an extension quarantine and vanish from the graph.
    #[test]
    fn checkpoint_replay_applies_inner_define_ontology() {
        // Base ontology: "entity" only — no "signal".
        let mut base = test_ontology();
        base.node_types.remove("signal");
        base.edge_types.remove("OBSERVES");
        let mut g = MaterializedGraph::new(base);

        let checkpoint = make_entry(
            GraphOp::Checkpoint {
                ops: vec![
                    GraphOp::DefineOntology {
                        ontology: test_ontology(), // merged: has "signal"
                    },
                    GraphOp::AddNode {
                        node_id: "n1".into(),
                        node_type: "entity".into(),
                        label: "base-typed".into(),
                        properties: BTreeMap::new(),
                        subtype: None,
                    },
                    GraphOp::AddNode {
                        node_id: "s1".into(),
                        node_type: "signal".into(),
                        label: "extension-typed".into(),
                        properties: BTreeMap::new(),
                        subtype: None,
                    },
                ],
                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
                compacted_at_physical_ms: 1,
                compacted_at_logical: 2,
            },
            1,
            "inst-a",
        );
        g.apply(&checkpoint);

        assert!(g.get_node("n1").is_some());
        assert!(g.get_node("s1").is_some(), "extension-typed node lost");
        assert!(g.ontology.node_types.contains_key("signal"));
        assert!(g.quarantined.is_empty());
    }

    /// Bug 14b: build_checkpoint_ops emits AddNode with empty properties
    /// (EXP-02: per-property clocks ride in separate UpdateProperty ops).
    /// Replay must trust checkpoint inner ops — validating them fails any
    /// required property against the empty map and quarantines the node.
    #[test]
    fn checkpoint_replay_trusts_inner_ops() {
        use crate::ontology::{PropertyDef, ValueType};

        let mut ont = test_ontology();
        ont.node_types.get_mut("entity").unwrap().properties.insert(
            "name".into(),
            PropertyDef {
                value_type: ValueType::String,
                required: true,
                description: None,
                constraints: None,
            },
        );
        let mut g = MaterializedGraph::new(ont.clone());

        let checkpoint = make_entry(
            GraphOp::Checkpoint {
                ops: vec![
                    GraphOp::DefineOntology { ontology: ont },
                    GraphOp::AddNode {
                        node_id: "n1".into(),
                        node_type: "entity".into(),
                        label: "req".into(),
                        properties: BTreeMap::new(), // empty by design
                        subtype: None,
                    },
                    GraphOp::UpdateProperty {
                        entity_id: "n1".into(),
                        key: "name".into(),
                        value: Value::String("x".into()),
                    },
                ],
                op_clocks: vec![(1, 0), (1, 1), (1, 2)],
                compacted_at_physical_ms: 1,
                compacted_at_logical: 2,
            },
            1,
            "inst-a",
        );
        g.apply(&checkpoint);

        let node = g.get_node("n1").expect("required-property node lost");
        assert_eq!(
            node.properties.get("name"),
            Some(&Value::String("x".into()))
        );
        assert!(g.quarantined.is_empty());
    }
}