nedb_engine/
store.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//! Content-addressed object store — the foundation of NEDB v2.
6//!
7//! Every document version is stored as an immutable, encrypted, BLAKE2b-hashed
8//! object at `objects/{hash[0:2]}/{hash[2:]}`. Once written, objects never change.
9//!
10//! Uncorruptable by design:
11//! - Writes are atomic (write to .tmp → rename)
12//! - Every read verifies the BLAKE2b hash of the content
13//! - A partial write leaves a .tmp file that is ignored on startup
14//! - There is no single mutable file that can be partially overwritten
15
16use std::fs;
17use std::path::{Path, PathBuf};
18use std::sync::Arc;
19use anyhow::{bail, Context, Result};
20use serde::{Deserialize, Serialize};
21use blake2::{Blake2b512, Digest};
22
23/// A single versioned document node in the DAG.
24#[derive(Debug, Clone, Serialize, Deserialize)]
25pub struct Node {
26    /// User-supplied document ID (e.g. "618000", "abc-token-id")
27    pub id:         String,
28    /// Collection name (e.g. "blocks", "itsl_ops")
29    pub coll:       String,
30    /// Monotonic global sequence number assigned at write time
31    pub seq:        u64,
32    /// The document payload (arbitrary JSON)
33    pub data:       serde_json::Value,
34    /// BLAKE2b hash of the previous version of this document (version chain)
35    pub prev:       Option<String>,
36    /// BLAKE2b hashes of nodes that causally led to this write
37    pub caused_by:  Vec<String>,
38    /// Unix timestamp (seconds since epoch)
39    pub ts:         f64,
40    /// Bi-temporal valid-from (ISO 8601)
41    pub valid_from: Option<String>,
42    /// Bi-temporal valid-to   (ISO 8601); None = still valid
43    pub valid_to:   Option<String>,
44    /// BLAKE2b hash of this node's encrypted content (set after writing)
45    #[serde(skip_serializing_if = "String::is_empty", default)]
46    pub hash:       String,
47}
48
49/// Encryption key material (AES-256-GCM).
50/// In v1 this was called DEK; the structure is the same.
51#[derive(Clone)]
52pub struct Dek(pub [u8; 32]);
53
54impl Dek {
55    pub fn from_tmk(tmk: &[u8; 32], salt: &[u8]) -> Self {
56        use sha2::{Sha256, Digest as _};
57        let mut h = Sha256::new();
58        h.update(tmk);
59        h.update(salt);
60        let result = h.finalize();
61        let mut key = [0u8; 32];
62        key.copy_from_slice(&result[..32]);
63        Dek(key)
64    }
65}
66
67fn blake2b(data: &[u8]) -> String {
68    let mut h = Blake2b512::new();
69    h.update(data);
70    hex::encode(&h.finalize()[..32])   // use first 32 bytes → 64 hex chars
71}
72
73/// NEDB v3 opt-in: the `--dag-v3` flag sets `NEDB_DAG_V3`, which switches the
74/// object substrate to the packed segment store. Default off → byte-for-byte v2.
75fn dag_v3_enabled() -> bool {
76    std::env::var("NEDB_DAG_V3")
77        .map(|v| {
78            let v = v.trim();
79            v == "1" || v.eq_ignore_ascii_case("true")
80                     || v.eq_ignore_ascii_case("on")
81                     || v.eq_ignore_ascii_case("yes")
82        })
83        .unwrap_or(false)
84}
85
86fn encrypt(data: &[u8], dek: &Dek) -> Result<Vec<u8>> {
87    use aes_gcm::{Aes256Gcm, KeyInit, aead::{Aead, OsRng, rand_core::RngCore}};
88    let cipher = Aes256Gcm::new_from_slice(&dek.0)?;
89    let mut nonce_bytes = [0u8; 12];
90    OsRng.fill_bytes(&mut nonce_bytes);
91    let nonce = aes_gcm::Nonce::from(nonce_bytes);
92    let ciphertext = cipher.encrypt(&nonce, data)
93        .map_err(|e| anyhow::anyhow!("encrypt failed: {:?}", e))?;
94    // Format: 12-byte nonce || ciphertext
95    let mut out = nonce_bytes.to_vec();
96    out.extend_from_slice(&ciphertext);
97    Ok(out)
98}
99
100fn decrypt(data: &[u8], dek: &Dek) -> Result<Vec<u8>> {
101    use aes_gcm::{Aes256Gcm, KeyInit, aead::Aead};
102    if data.len() < 12 { bail!("ciphertext too short"); }
103    let (nonce_bytes, ciphertext) = data.split_at(12);
104    let cipher = Aes256Gcm::new_from_slice(&dek.0)?;
105    let nonce = aes_gcm::Nonce::from_slice(nonce_bytes);
106    cipher.decrypt(nonce, ciphertext)
107        .map_err(|e| anyhow::anyhow!("decrypt failed: {:?}", e))
108}
109
110/// Content-addressed, encrypted, tamper-evident object store.
111pub struct ObjectStore {
112    root: PathBuf,
113    dek:  Option<Dek>,
114    /// In-memory store: hash → raw bytes. None = disk-backed (normal mode).
115    mem:  Option<Arc<dashmap::DashMap<String, Vec<u8>>>>,
116    /// NEDB v3 packed substrate. Some = segment mode (NEDB_DAG_V3 / --dag-v3);
117    /// new writes go to segments, reads fall back to loose v2 objects.
118    seg:  Option<crate::segment::SegmentStore>,
119}
120
121impl ObjectStore {
122    pub fn new(db_root: &Path, dek: Option<Dek>) -> Result<Self> {
123        let root = db_root.join("objects");
124        fs::create_dir_all(&root)
125            .context("create objects/ dir")?;
126        // v3 opt-in: bring up the packed segment substrate (and rebuild its
127        // index by scanning existing segments). Off by default → loose objects.
128        let seg = if dag_v3_enabled() {
129            Some(crate::segment::SegmentStore::open(&root)?)
130        } else {
131            None
132        };
133        Ok(Self { root, dek, mem: None, seg })
134    }
135
136    /// Create a pure in-memory object store — no disk, no files.
137    pub fn in_memory() -> Self {
138        Self {
139            root: PathBuf::from(":memory:"),
140            dek:  None,
141            mem:  Some(Arc::new(dashmap::DashMap::new())),
142            seg:  None,
143        }
144    }
145
146    /// Write a node. Returns the content hash (the node's permanent ID in the DAG).
147    pub fn write(&self, node: &mut Node) -> Result<String> {
148        // The content hash is taken over the node's CONTENT, never over its own
149        // hash field (which is circular). `hash` is `skip_serializing_if =
150        // "String::is_empty"`, so a *fresh* node already excludes it — but a
151        // node being re-written (node.hash already set from a prior write) would
152        // serialize the populated hash and produce a different content hash,
153        // breaking idempotency. Clear it first so first-write and re-write
154        // serialize byte-for-byte identical content.
155        node.hash = String::new();
156        let raw = serde_json::to_vec(node)?;
157        let content = match &self.dek {
158            Some(dek) => encrypt(&raw, dek)?,
159            None      => raw,
160        };
161        let hash = blake2b(&content);
162
163        if let Some(ref mem) = self.mem {
164            // In-memory: store in DashMap — idempotent
165            mem.entry(hash.clone()).or_insert_with(|| content);
166        } else if let Some(ref seg) = self.seg {
167            // v3: append into a packed segment (one fsync per batch via sync()).
168            seg.put(&hash, &content)?;
169        } else {
170            // v2: loose object file, written atomically via tmp → rename
171            let dir  = self.root.join(&hash[..2]);
172            fs::create_dir_all(&dir)?;
173            let path = dir.join(&hash[2..]);
174            if !path.exists() {
175                let tmp = path.with_extension("tmp");
176                fs::write(&tmp, &content)?;
177                fs::rename(&tmp, &path)?;
178            }
179        }
180        node.hash = hash.clone();
181        Ok(hash)
182    }
183
184    /// Read and verify a node by its hash. Returns error on hash mismatch (tamper).
185    pub fn read(&self, hash: &str) -> Result<Node> {
186        if hash.len() < 3 {
187            anyhow::bail!("invalid object hash (too short): {:?}", hash);
188        }
189
190        // In-memory mode (tests).
191        if let Some(ref mem) = self.mem {
192            let content = mem.get(hash)
193                .map(|v| v.clone())
194                .ok_or_else(|| anyhow::anyhow!("object not found in memory: {}", hash))?;
195            return self.decode(content, hash);
196        }
197
198        // v3 segment mode: try segments first (self-verifying), then fall back
199        // to the loose-object path so existing v2 data stays readable.
200        if let Some(ref seg) = self.seg {
201            if let Some(content) = seg.get(hash)? {
202                return self.decode(content, hash);
203            }
204            // miss → fall through to loose objects (dual-read migration)
205        }
206
207        // v2 loose object.
208        let path = self.root.join(&hash[..2]).join(&hash[2..]);
209        let c = fs::read(&path).with_context(|| format!("read object {}", hash))?;
210        // Hash verification — any bit rot or tampering is caught here
211        let actual = blake2b(&c);
212        if actual != hash {
213            bail!("object {} tampered: expected {} got {}", hash, hash, actual);
214        }
215        self.decode(c, hash)
216    }
217
218    /// Decrypt (if a DEK is set) and deserialize raw content bytes into a Node.
219    /// Hash verification is the caller's responsibility (done before this for the
220    /// loose path; inside SegmentStore::get for the segment path; trusted for mem).
221    fn decode(&self, content: Vec<u8>, hash: &str) -> Result<Node> {
222        let raw = match &self.dek {
223            Some(dek) => decrypt(&content, dek)?,
224            None      => content,
225        };
226        let mut node: Node = serde_json::from_slice(&raw)
227            .context("deserialize node")?;
228        if node.hash.is_empty() {
229            node.hash = hash.to_string();
230        }
231        Ok(node)
232    }
233
234    /// List all object hashes (for startup index rebuild / verify).
235    pub fn all_hashes(&self) -> Box<dyn Iterator<Item = String> + '_> {
236        // In-memory: collect from DashMap
237        if let Some(ref mem) = self.mem {
238            let hashes: Vec<String> = mem.iter().map(|e| e.key().clone()).collect();
239            return Box::new(hashes.into_iter());
240        }
241
242        // v3: union of packed-segment hashes and any loose v2 objects (deduped),
243        // skipping the segments/ subdir during the loose walk.
244        if let Some(ref seg) = self.seg {
245            let mut seen: std::collections::HashSet<String> =
246                seg.all_hashes().into_iter().collect();
247            if let Ok(rd) = fs::read_dir(&self.root) {
248                for prefix_dir in rd.flatten() {
249                    if !prefix_dir.file_type().map(|t| t.is_dir()).unwrap_or(false) { continue; }
250                    let prefix = prefix_dir.file_name().to_string_lossy().to_string();
251                    if prefix.len() != 2 { continue; } // skip "segments" and non-prefix dirs
252                    if let Ok(rd2) = fs::read_dir(prefix_dir.path()) {
253                        for e in rd2.flatten() {
254                            let name = e.file_name().to_string_lossy().to_string();
255                            if name.ends_with(".tmp") { continue; }
256                            seen.insert(format!("{}{}", prefix, name));
257                        }
258                    }
259                }
260            }
261            return Box::new(seen.into_iter());
262        }
263
264        // v2 (default): lazy walk of the objects/ directory tree (unchanged).
265        let root = self.root.clone();
266        Box::new(fs::read_dir(&root)
267            .into_iter()
268            .flatten()
269            .filter_map(|e| e.ok())
270            .filter(|e| e.file_type().map(|t| t.is_dir()).unwrap_or(false))
271            .flat_map(move |prefix_dir| {
272                let prefix = prefix_dir.file_name().to_string_lossy().to_string();
273                fs::read_dir(prefix_dir.path())
274                    .into_iter()
275                    .flatten()
276                    .filter_map(|e| e.ok())
277                    .filter_map(move |e| {
278                        let name = e.file_name().to_string_lossy().to_string();
279                        if name.ends_with(".tmp") { return None; }
280                        Some(format!("{}{}", prefix, name))
281                    })
282            }))
283    }
284
285    /// Flush durable state for the active segment (v3): one fsync per batch,
286    /// wired into Db::flush_all(). No-op for loose-object and in-memory modes.
287    pub fn sync(&self) -> Result<()> {
288        if let Some(ref seg) = self.seg {
289            seg.sync()?;
290        }
291        Ok(())
292    }
293
294    /// Compact the packed segment store (v3), keeping only objects whose hash is
295    /// in `live` and reclaiming the rest. No-op (zeroed stats) for loose-object
296    /// and in-memory modes.
297    pub fn compact(&self, live: &std::collections::HashSet<String>) -> Result<crate::segment::CompactStats> {
298        match self.seg {
299            Some(ref seg) => seg.compact(live),
300            None => Ok(crate::segment::CompactStats::default()),
301        }
302    }
303
304    /// Verify all objects. Returns (ok_count, tampered_hashes).
305    pub fn verify_all(&self) -> (usize, Vec<String>) {
306        use rayon::prelude::*;
307        let hashes: Vec<String> = self.all_hashes().collect();
308        let results: Vec<(bool, String)> = hashes.par_iter().map(|h| {
309            (self.read(h).is_ok(), h.clone())
310        }).collect();
311        let ok = results.iter().filter(|(ok, _)| *ok).count();
312        let bad: Vec<String> = results.into_iter()
313            .filter(|(ok, _)| !*ok)
314            .map(|(_, h)| h)
315            .collect();
316        (ok, bad)
317    }
318}
319
320#[cfg(test)]
321mod tests {
322    use super::*;
323    use tempfile::tempdir;
324
325    fn make_node(id: &str, coll: &str, seq: u64) -> Node {
326        Node {
327            id: id.to_string(), coll: coll.to_string(), seq,
328            data: serde_json::json!({"height": seq, "hash": "0000abc"}),
329            prev: None, caused_by: vec![], ts: 1718400000.0,
330            valid_from: None, valid_to: None, hash: String::new(),
331        }
332    }
333
334    #[test]
335    fn write_read_roundtrip() {
336        let dir = tempdir().unwrap();
337        let store = ObjectStore::new(dir.path(), None).unwrap();
338        let mut node = make_node("1", "blocks", 1);
339        let hash = store.write(&mut node).unwrap();
340        assert_eq!(hash.len(), 64);
341        let read_back = store.read(&hash).unwrap();
342        assert_eq!(read_back.id, "1");
343        assert_eq!(read_back.coll, "blocks");
344    }
345
346    #[test]
347    fn write_is_idempotent() {
348        let dir = tempdir().unwrap();
349        let store = ObjectStore::new(dir.path(), None).unwrap();
350        let mut node = make_node("1", "blocks", 1);
351        let h1 = store.write(&mut node).unwrap();
352        let h2 = store.write(&mut node).unwrap();
353        assert_eq!(h1, h2);
354    }
355
356    #[test]
357    fn tamper_detected() {
358        let dir = tempdir().unwrap();
359        let store = ObjectStore::new(dir.path(), None).unwrap();
360        let mut node = make_node("1", "blocks", 1);
361        let hash = store.write(&mut node).unwrap();
362        // Corrupt the object file
363        let path = dir.path().join("objects").join(&hash[..2]).join(&hash[2..]);
364        let mut content = fs::read(&path).unwrap();
365        content[10] ^= 0xff;
366        fs::write(&path, content).unwrap();
367        assert!(store.read(&hash).is_err());
368    }
369}
nedb_engine/store.rs

nedb_engine/
store.rs
