velesdb_core/collection/graph/

edge.rs

//! Graph edge types and storage for knowledge graph relationships.
//!
//! This module provides:
//! - `GraphEdge`: A typed relationship between nodes with properties
//! - `EdgeStore`: Bidirectional index for efficient edge traversal
//!
//! CSR snapshot types are in [`super::csr_snapshot`].
//!
//! # Edge Removal Semantics
//!
//! During edge removal, the internal indexes may be temporarily inconsistent
//! while the operation is in progress. The final state is always consistent.
//! For concurrent access, use `ConcurrentEdgeStore` instead.

use super::csr_snapshot::CsrSnapshot;
use super::label_table::{LabelId, LabelTable};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;

/// A directed edge (relationship) in the knowledge graph.
///
/// Edges connect nodes and can have a label (type) and properties.
///
/// # Example
///
/// ```rust,ignore
/// use velesdb_core::collection::graph::GraphEdge;
/// use serde_json::json;
/// use std::collections::HashMap;
///
/// let mut props = HashMap::new();
/// props.insert("since".to_string(), json!("2020-01-01"));
///
/// let edge = GraphEdge::new(1, 100, 200, "KNOWS")
///     .with_properties(props);
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct GraphEdge {
    id: u64,
    source: u64,
    target: u64,
    label: String,
    properties: HashMap<String, Value>,
}

impl GraphEdge {
    /// Creates a new edge with the given ID, endpoints, and label.
    ///
    /// # Errors
    ///
    /// Returns `Error::InvalidEdgeLabel` if the label is empty or whitespace-only.
    pub fn new(id: u64, source: u64, target: u64, label: &str) -> Result<Self> {
        let trimmed = label.trim();
        if trimmed.is_empty() {
            return Err(Error::InvalidEdgeLabel(
                "Edge label cannot be empty or whitespace-only".to_string(),
            ));
        }
        Ok(Self {
            id,
            source,
            target,
            label: trimmed.to_string(),
            properties: HashMap::new(),
        })
    }

    /// Adds properties to this edge (builder pattern).
    #[must_use]
    pub fn with_properties(mut self, properties: HashMap<String, Value>) -> Self {
        self.properties = properties;
        self
    }

    /// Returns the edge ID.
    #[must_use]
    pub fn id(&self) -> u64 {
        self.id
    }

    /// Returns the source node ID.
    #[must_use]
    pub fn source(&self) -> u64 {
        self.source
    }

    /// Returns the target node ID.
    #[must_use]
    pub fn target(&self) -> u64 {
        self.target
    }

    /// Returns the edge label (relationship type).
    #[must_use]
    pub fn label(&self) -> &str {
        &self.label
    }

    /// Returns all properties of this edge.
    #[must_use]
    pub fn properties(&self) -> &HashMap<String, Value> {
        &self.properties
    }

    /// Returns a specific property value, if it exists.
    #[must_use]
    pub fn property(&self, name: &str) -> Option<&Value> {
        self.properties.get(name)
    }
}

/// Storage for graph edges with bidirectional indexing.
///
/// Provides O(1) access to edges by ID and O(degree) access to
/// outgoing/incoming edges for any node.
///
/// # Index Structure (EPIC-019 US-003)
///
/// - `by_label`: Secondary index for O(k) label-based queries
/// - `outgoing_by_label`: Composite index (source, label) for O(k) filtered traversal
///
/// # CSR Snapshot (G1)
///
/// After loading from disk or after explicit `build_read_snapshot()`, the
/// `csr_snapshot` field provides zero-copy `&[u64]` access to neighbor
/// target IDs and edge IDs. Writes invalidate the snapshot automatically.
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct EdgeStore {
    /// All edges indexed by ID
    pub(super) edges: HashMap<u64, GraphEdge>,
    /// Outgoing edges: source_id -> Vec<edge_id>
    pub(super) outgoing: HashMap<u64, Vec<u64>>,
    /// Incoming edges: target_id -> Vec<edge_id>
    pub(super) incoming: HashMap<u64, Vec<u64>>,
    /// Secondary index: interned label -> Vec<edge_id> for fast label queries.
    ///
    /// `serde(skip)`: keyed by [`LabelId`], which is only meaningful against
    /// this store's `label_table` — both are rebuilt in `load_from_file`
    /// (#2089). Old snapshots that still serialize the String-keyed maps
    /// load fine: postcard's `from_bytes` ignores trailing bytes.
    #[serde(skip)]
    pub(super) by_label: HashMap<LabelId, Vec<u64>>,
    /// Composite index: (source_id, label) -> Vec<edge_id> for fast filtered
    /// traversal. `serde(skip)`: same rebuild contract as `by_label`.
    #[serde(skip)]
    pub(super) outgoing_by_label: HashMap<(u64, LabelId), Vec<u64>>,
    /// Composite index: (target_id, label) -> Vec<edge_id> — the incoming
    /// mirror of `outgoing_by_label`, so `<-[:TYPE]-` patterns stop paying
    /// O(in-degree) full-edge clones on super-nodes.
    ///
    /// `serde(skip)`: the snapshot format is postcard (not self-describing),
    /// so a new serialized field would break every existing edge_store.bin.
    /// The index is fully derivable and rebuilt in `load_from_file`.
    #[serde(skip)]
    pub(super) incoming_by_label: HashMap<(u64, LabelId), Vec<u64>>,
    /// Interning table mapping label strings to the [`LabelId`]s that key the
    /// three label indices above. Populated by `insert_edge`, rebuilt with
    /// the indices in `load_from_file` (#2089 — wires the table the doc
    /// comment always promised; labels are no longer stored per edge entry).
    #[serde(skip)]
    pub(super) label_table: LabelTable,
    /// Zero-copy CSR snapshot for BFS traversal (G1).
    /// Built on-demand via `build_read_snapshot()`, invalidated by writes.
    #[serde(skip)]
    pub(super) csr_snapshot: Option<CsrSnapshot>,
}

impl EdgeStore {
    /// Creates a new empty edge store.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Creates an edge store with pre-allocated capacity for better performance.
    ///
    /// Pre-allocating reduces memory reallocation overhead when inserting many edges.
    /// With 10M edges, this can reduce peak memory usage by ~2x and improve insert throughput.
    ///
    /// Note: when accessed through the sharded `ConcurrentEdgeStore`, an edge
    /// whose endpoints hash to different shards is stored in **both** shards
    /// (outgoing + incoming halves), which offsets part of that saving — with
    /// many shards this applies to nearly all edges.
    ///
    /// # Arguments
    ///
    /// * `expected_edges` - Expected number of edges to store
    /// * `expected_nodes` - Expected number of unique nodes (sources + targets)
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // For a graph with ~1M edges and ~100K nodes
    /// let store = EdgeStore::with_capacity(1_000_000, 100_000);
    /// ```
    #[must_use]
    pub fn with_capacity(expected_edges: usize, expected_nodes: usize) -> Self {
        // Estimate ~10 unique labels typical for knowledge graphs
        let expected_labels = 10usize;
        // Use saturating_mul to prevent overflow for extreme inputs
        let outgoing_by_label_cap = expected_nodes
            .saturating_mul(expected_labels)
            .saturating_div(10);
        Self {
            edges: HashMap::with_capacity(expected_edges),
            outgoing: HashMap::with_capacity(expected_nodes),
            incoming: HashMap::with_capacity(expected_nodes),
            by_label: HashMap::with_capacity(expected_labels),
            outgoing_by_label: HashMap::with_capacity(outgoing_by_label_cap),
            incoming_by_label: HashMap::with_capacity(outgoing_by_label_cap),
            label_table: LabelTable::with_capacity(expected_labels),
            csr_snapshot: None,
        }
    }

    /// Adds an edge to the store.
    ///
    /// Creates bidirectional index entries for efficient traversal.
    /// Also maintains label-based secondary indices (EPIC-019 US-003).
    ///
    /// # Errors
    ///
    /// Returns `Error::EdgeExists` if an edge with the same ID already exists.
    pub fn add_edge(&mut self, edge: GraphEdge) -> Result<()> {
        self.insert_edge(edge, true, true)
    }

    /// Adds an edge with only the outgoing index (for cross-shard storage).
    ///
    /// Used by `ConcurrentEdgeStore` when source and target are in different shards.
    /// The edge is stored and indexed by source node only.
    ///
    /// # Errors
    ///
    /// Returns `Error::EdgeExists` if an edge with the same ID already exists.
    pub fn add_edge_outgoing_only(&mut self, edge: GraphEdge) -> Result<()> {
        self.insert_edge(edge, true, false)
    }

    /// Adds an edge with only the incoming index (for cross-shard storage).
    ///
    /// Used by `ConcurrentEdgeStore` when source and target are in different shards.
    /// The edge is stored and indexed by target node only.
    /// Note: Label indices are maintained by the source shard in `ConcurrentEdgeStore`.
    ///
    /// # Errors
    ///
    /// Returns `Error::EdgeExists` if an edge with the same ID already exists.
    pub fn add_edge_incoming_only(&mut self, edge: GraphEdge) -> Result<()> {
        self.insert_edge(edge, false, true)
    }

    /// Shared implementation for all `add_edge*` variants.
    ///
    /// Validates uniqueness, populates the requested directional indices,
    /// and stores the edge. Label indices (`by_label`, `outgoing_by_label`)
    /// are maintained only when `index_outgoing` is `true` (source shard
    /// owns label indices in the concurrent model).
    fn insert_edge(
        &mut self,
        edge: GraphEdge,
        index_outgoing: bool,
        index_incoming: bool,
    ) -> Result<()> {
        let id = edge.id();
        if self.edges.contains_key(&id) {
            return Err(Error::EdgeExists(id));
        }

        // One interning lookup per insert; no per-edge label allocation (#2089).
        let label_id = self.intern_label(edge.label())?;

        if index_outgoing {
            let source = edge.source();
            self.outgoing.entry(source).or_default().push(id);
            // Label indices are owned by the source shard (US-003)
            self.by_label.entry(label_id).or_default().push(id);
            self.outgoing_by_label
                .entry((source, label_id))
                .or_default()
                .push(id);
        }

        if index_incoming {
            let target = edge.target();
            self.incoming.entry(target).or_default().push(id);
            self.incoming_by_label
                .entry((target, label_id))
                .or_default()
                .push(id);
        }

        self.edges.insert(id, edge);
        // Invalidate CSR snapshot — writes make it stale (G1).
        self.csr_snapshot = None;
        Ok(())
    }

    /// Returns the total number of edges in the store.
    #[must_use]
    pub fn edge_count(&self) -> usize {
        self.edges.len()
    }

    /// Returns the count of edges where this shard is the source (for accurate cross-shard counting).
    #[must_use]
    pub fn outgoing_edge_count(&self) -> usize {
        self.outgoing.values().map(Vec::len).sum()
    }

    /// Gets an edge by its ID.
    #[must_use]
    pub fn get_edge(&self, id: u64) -> Option<&GraphEdge> {
        self.edges.get(&id)
    }

    /// Gets all outgoing edges from a node.
    #[must_use]
    pub fn get_outgoing(&self, node_id: u64) -> Vec<&GraphEdge> {
        self.resolve_edge_ids(self.outgoing.get(&node_id))
    }

    /// Invokes `f` for each outgoing edge from `node_id` without allocating a `Vec`.
    ///
    /// Prefer this over [`get_outgoing`](Self::get_outgoing) in hot loops (e.g. BFS
    /// frontiers) where the caller processes edges inline rather than collecting them.
    #[inline]
    pub fn for_each_outgoing<F: FnMut(&GraphEdge)>(&self, node_id: u64, mut f: F) {
        if let Some(ids) = self.outgoing.get(&node_id) {
            for id in ids {
                if let Some(edge) = self.edges.get(id) {
                    f(edge);
                }
            }
        }
    }

    /// Returns the number of outgoing edges from `node_id` without materializing them.
    #[must_use]
    #[inline]
    pub fn outgoing_degree(&self, node_id: u64) -> usize {
        self.outgoing.get(&node_id).map_or(0, Vec::len)
    }

    /// Returns the number of incoming edges to `node_id` without materializing them.
    #[must_use]
    #[inline]
    pub fn incoming_degree(&self, node_id: u64) -> usize {
        self.incoming.get(&node_id).map_or(0, Vec::len)
    }

    /// Gets all incoming edges to a node.
    #[must_use]
    pub fn get_incoming(&self, node_id: u64) -> Vec<&GraphEdge> {
        self.resolve_edge_ids(self.incoming.get(&node_id))
    }

    /// Gets at most `cap` outgoing edges from a node.
    ///
    /// The bound is applied to the index BEFORE any edge is resolved, so the
    /// work and the allocation are O(cap) — never O(degree). This is what
    /// makes reading a super-node affordable: [`Self::get_outgoing`] on a
    /// million-edge node materializes a million entries even when the caller
    /// keeps only the first 64 (#1820).
    #[must_use]
    pub fn get_outgoing_bounded(&self, node_id: u64, cap: usize) -> Vec<&GraphEdge> {
        self.resolve_edge_ids_bounded(self.outgoing.get(&node_id), cap)
    }

    /// Gets at most `cap` incoming edges to a node — the mirror of
    /// [`Self::get_outgoing_bounded`], with the same O(cap) guarantee.
    #[must_use]
    pub fn get_incoming_bounded(&self, node_id: u64, cap: usize) -> Vec<&GraphEdge> {
        self.resolve_edge_ids_bounded(self.incoming.get(&node_id), cap)
    }

    /// [`Self::resolve_edge_ids`] with the id list truncated FIRST — the cap
    /// bounds the scan itself, not just the result. A dangling id inside the
    /// scanned window (defensively skipped, as in the unbounded resolver) is
    /// not replaced by scanning further: the O(cap) guarantee outranks
    /// returning exactly `cap` entries.
    #[inline]
    fn resolve_edge_ids_bounded(&self, ids: Option<&Vec<u64>>, cap: usize) -> Vec<&GraphEdge> {
        ids.map(|ids| {
            ids.iter()
                .take(cap)
                .filter_map(|id| self.edges.get(id))
                .collect()
        })
        .unwrap_or_default()
    }

    /// Gets outgoing edges filtered by label using composite index - O(k) where k = result count.
    ///
    /// Uses the `outgoing_by_label` composite index for fast lookup instead of
    /// iterating through all outgoing edges (EPIC-019 US-003).
    #[must_use]
    pub fn get_outgoing_by_label(&self, node_id: u64, label: &str) -> Vec<&GraphEdge> {
        let Some(label_id) = self.label_table.get_id(label) else {
            return Vec::new();
        };
        self.resolve_edge_ids(self.outgoing_by_label.get(&(node_id, label_id)))
    }

    /// Gets all edges with a specific label - O(k) where k = result count.
    ///
    /// Uses the `by_label` secondary index for fast lookup (EPIC-019 US-003).
    #[must_use]
    pub fn get_edges_by_label(&self, label: &str) -> Vec<&GraphEdge> {
        let Some(label_id) = self.label_table.get_id(label) else {
            return Vec::new();
        };
        self.resolve_edge_ids(self.by_label.get(&label_id))
    }

    /// Resolves edge IDs from an index entry into edge references.
    ///
    /// Shared lookup pattern used by `get_outgoing`, `get_incoming`,
    /// `get_outgoing_by_label`, and `get_edges_by_label`.
    #[inline]
    fn resolve_edge_ids(&self, ids: Option<&Vec<u64>>) -> Vec<&GraphEdge> {
        ids.map(|ids| ids.iter().filter_map(|id| self.edges.get(id)).collect())
            .unwrap_or_default()
    }

    /// Gets incoming edges filtered by label.
    #[must_use]
    pub fn get_incoming_by_label(&self, node_id: u64, label: &str) -> Vec<&GraphEdge> {
        let Some(label_id) = self.label_table.get_id(label) else {
            return Vec::new();
        };
        self.resolve_edge_ids(self.incoming_by_label.get(&(node_id, label_id)))
    }

    /// Interns `label` in this store's table, mapping the (practically
    /// unreachable) `u32::MAX`-distinct-labels overflow to [`Error::Overflow`].
    fn intern_label(&mut self, label: &str) -> Result<LabelId> {
        intern_or_overflow(&mut self.label_table, label)
    }

    /// Rebuilds the (unserialized) label table and the three label indices
    /// from the live edges — called after loading a postcard snapshot.
    ///
    /// Interning cannot overflow here: every label was already interned once
    /// when its edge was first inserted, so the vocabulary fits `u32` by
    /// construction; an overflow is reported all the same rather than
    /// silently dropping index entries.
    ///
    /// # Errors
    ///
    /// Returns [`Error::Overflow`] if the number of distinct labels exceeds
    /// the `u32` domain of [`LabelId`].
    pub(super) fn rebuild_label_indexes(&mut self) -> Result<()> {
        // Destructured so the borrow checker sees the disjoint fields: the
        // edge/index maps are read while the table and label maps are written.
        let Self {
            edges,
            outgoing,
            incoming,
            by_label,
            outgoing_by_label,
            incoming_by_label,
            label_table,
            ..
        } = self;
        by_label.clear();
        outgoing_by_label.clear();
        incoming_by_label.clear();

        // Iterate nodes in sorted order (same discipline as SnapshotBuilder):
        // HashMap iteration order is randomized per instance, so an unsorted
        // walk would hand every reload a different by-label result order.
        for &node in &sorted_keys(outgoing) {
            for &id in &outgoing[&node] {
                if let Some(edge) = edges.get(&id) {
                    let label_id = intern_or_overflow(label_table, edge.label())?;
                    by_label.entry(label_id).or_default().push(id);
                    outgoing_by_label
                        .entry((edge.source(), label_id))
                        .or_default()
                        .push(id);
                }
            }
        }
        rebuild_incoming_labels(edges, incoming, incoming_by_label, label_table)
    }

    /// Checks if an edge with the given ID exists.
    #[must_use]
    pub fn contains_edge(&self, edge_id: u64) -> bool {
        self.edges.contains_key(&edge_id)
    }

    /// Returns the number of edges in the store.
    #[must_use]
    pub fn len(&self) -> usize {
        self.edges.len()
    }

    /// Returns true if the store contains no edges.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.edges.is_empty()
    }

    /// Returns all edges in the store.
    #[must_use]
    pub fn all_edges(&self) -> Vec<&GraphEdge> {
        self.edges.values().collect()
    }

    /// Returns all outgoing source node IDs (keys of the outgoing index).
    ///
    /// Used by [`SnapshotBuilder`](super::csr_snapshot::SnapshotBuilder) to
    /// enumerate source nodes for CSR construction.
    #[must_use]
    pub(crate) fn outgoing_keys(&self) -> Vec<u64> {
        self.outgoing.keys().copied().collect()
    }

    /// Returns the total number of outgoing edge entries across all nodes.
    ///
    /// Used by [`SnapshotBuilder`](super::csr_snapshot::SnapshotBuilder) for
    /// pre-allocation.
    #[must_use]
    pub(crate) fn total_outgoing_edges(&self) -> usize {
        self.outgoing.values().map(Vec::len).sum()
    }

    /// Invokes `f` for each outgoing edge from `node_id` (by edge object).
    ///
    /// Used by [`SnapshotBuilder`](super::csr_snapshot::SnapshotBuilder) to
    /// iterate edges without exposing internal index structure.
    pub(crate) fn for_each_outgoing_edge<F: FnMut(&GraphEdge)>(&self, node_id: u64, mut f: F) {
        if let Some(ids) = self.outgoing.get(&node_id) {
            for id in ids {
                if let Some(edge) = self.edges.get(id) {
                    f(edge);
                }
            }
        }
    }
}

/// Maps [`LabelTable::intern`]'s overflow (more than `u32::MAX` distinct
/// labels — unreachable in practice) to [`Error::Overflow`].
fn intern_or_overflow(table: &mut LabelTable, label: &str) -> Result<LabelId> {
    table
        .intern(label)
        .map_err(|e| Error::Overflow(e.to_string()))
}

/// The incoming half of [`EdgeStore::rebuild_label_indexes`], split out to
/// keep both halves under the complexity budget. Takes the already-borrowed
/// disjoint fields rather than `&mut self` so the caller's destructuring
/// still satisfies the borrow checker.
fn rebuild_incoming_labels(
    edges: &HashMap<u64, GraphEdge>,
    incoming: &HashMap<u64, Vec<u64>>,
    incoming_by_label: &mut HashMap<(u64, LabelId), Vec<u64>>,
    label_table: &mut LabelTable,
) -> Result<()> {
    // Sorted for the same determinism reason as the outgoing half.
    for &node in &sorted_keys(incoming) {
        for &id in &incoming[&node] {
            if let Some(edge) = edges.get(&id) {
                let label_id = intern_or_overflow(label_table, edge.label())?;
                incoming_by_label
                    .entry((edge.target(), label_id))
                    .or_default()
                    .push(id);
            }
        }
    }
    Ok(())
}

/// The map's keys, sorted — so index rebuilds walk nodes in a stable order
/// instead of the per-instance-random `HashMap` iteration order.
fn sorted_keys<V>(map: &HashMap<u64, V>) -> Vec<u64> {
    let mut keys: Vec<u64> = map.keys().copied().collect();
    keys.sort_unstable();
    keys
}

// Edge removal operations are in `edge_removal.rs`.
// CSR snapshot methods and persistence are in `edge_persistence.rs`.