nedb_engine/
graph.rs

1// SPDX-FileCopyrightText: 2026 INTERCHAINED LLC
2// SPDX-License-Identifier: BUSL-1.1
3// NEDB · © 2026 INTERCHAINED LLC × Eth-Interchained × Vex (Claude Opus 5)
4
5//! DAG edge store — typed directed edges between node hashes.
6//!
7//! Layout: `graph/{from_hash}/{edge_type}/{to_hash}`
8//! Existence of the file = the edge exists. No file content needed.
9//!
10//! This makes TRACE queries pure filesystem traversal:
11//!   FROM nodes TRACE caused_by → list dir graph/{hash}/caused_by/
12//!   Each entry is the hash of a causal predecessor node.
13//!   Follow recursively until limit reached or no more edges.
14//!
15//! Write is atomic (create file). Read is readdir. Both are O(degree).
16//! No global lock. Multiple threads can add edges concurrently.
17
18use std::fs;
19use std::path::{Path, PathBuf};
20use std::sync::Arc;
21use anyhow::Result;
22
23pub struct GraphStore {
24    root: PathBuf,
25    /// In-memory edges: (from, edge_type) → Set<to>. None = disk-backed.
26    mem:  Option<Arc<dashmap::DashMap<(String, String), std::collections::HashSet<String>>>>,
27}
28
29impl GraphStore {
30    pub fn new(db_root: &Path) -> Result<Self> {
31        let root = db_root.join("graph");
32        fs::create_dir_all(&root)?;
33        Ok(Self { root, mem: None })
34    }
35
36    /// Create a pure in-memory graph store — no disk I/O.
37    pub fn in_memory() -> Self {
38        Self {
39            root: PathBuf::from(":memory:"),
40            mem:  Some(Arc::new(dashmap::DashMap::new())),
41        }
42    }
43
44    fn edge_path(&self, from: &str, edge_type: &str, to: &str) -> PathBuf {
45        self.root.join(from).join(edge_type).join(to)
46    }
47
48    /// Add a directed edge: from → to with the given type label.
49    pub fn add_edge(&self, from: &str, edge_type: &str, to: &str) -> Result<()> {
50        if let Some(ref mem) = self.mem {
51            mem.entry((from.to_string(), edge_type.to_string()))
52               .or_default()
53               .insert(to.to_string());
54            return Ok(());
55        }
56        let path = self.edge_path(from, edge_type, to);
57        fs::create_dir_all(path.parent().unwrap())?;
58        if !path.exists() {
59            fs::write(&path, b"")?;
60        }
61        Ok(())
62    }
63
64    /// Get all outgoing edges of a given type from a node.
65    pub fn outgoing(&self, from: &str, edge_type: &str) -> Vec<String> {
66        if let Some(ref mem) = self.mem {
67            return mem.get(&(from.to_string(), edge_type.to_string()))
68                .map(|s| s.iter().cloned().collect())
69                .unwrap_or_default();
70        }
71        let dir = self.root.join(from).join(edge_type);
72        fs::read_dir(&dir)
73            .into_iter()
74            .flatten()
75            .filter_map(|e| e.ok())
76            .map(|e| e.file_name().to_string_lossy().to_string())
77            .collect()
78    }
79
80    /// Get all incoming edges of a given type to a node (reverse lookup).
81    /// This requires scanning the `{reverse_edge_type}` edges stored when writing.
82    pub fn incoming(&self, to: &str, reverse_edge_type: &str) -> Vec<String> {
83        self.outgoing(to, reverse_edge_type)
84    }
85
86    /// TRACE: walk the DAG from `start` following `edge_type` edges.
87    /// Returns hashes in BFS order, up to `limit`.
88    pub fn trace(
89        &self,
90        start: &str,
91        edge_type: &str,
92        reverse: bool,
93        limit: usize,
94    ) -> Vec<String> {
95        let mut result = Vec::new();
96        let mut queue  = vec![start.to_string()];
97        let mut seen   = std::collections::HashSet::new();
98        seen.insert(start.to_string());
99
100        while !queue.is_empty() && result.len() < limit {
101            let current = queue.remove(0);
102            result.push(current.clone());
103
104            let next_hashes = if reverse {
105                // "reverse" means traverse the reverse-edge (e.g. "caused" instead of "caused_by")
106                let rev_type = format!("{}_rev", edge_type);
107                self.outgoing(&current, &rev_type)
108            } else {
109                self.outgoing(&current, edge_type)
110            };
111
112            for next in next_hashes {
113                if !seen.contains(&next) {
114                    seen.insert(next.clone());
115                    queue.push(next);
116                }
117            }
118        }
119        result
120    }
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126    use tempfile::tempdir;
127
128    #[test]
129    fn add_and_traverse_edge() {
130        let dir = tempdir().unwrap();
131        let g = GraphStore::new(dir.path()).unwrap();
132
133        g.add_edge("hash_c", "caused_by", "hash_b").unwrap();
134        g.add_edge("hash_b", "caused_by", "hash_a").unwrap();
135
136        let trace = g.trace("hash_c", "caused_by", false, 10);
137        assert_eq!(trace, vec!["hash_c", "hash_b", "hash_a"]);
138    }
139
140    #[test]
141    fn idempotent_edge() {
142        let dir = tempdir().unwrap();
143        let g = GraphStore::new(dir.path()).unwrap();
144        g.add_edge("a", "caused_by", "b").unwrap();
145        g.add_edge("a", "caused_by", "b").unwrap();   // second call is no-op
146        assert_eq!(g.outgoing("a", "caused_by").len(), 1);
147    }
148}
nedb_engine/graph.rs

nedb_engine/
graph.rs
