nodedb 0.3.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

// SPDX-License-Identifier: BUSL-1.1

//! EdgeStore root — types, redb table definitions, open/close.
//!
//! Query paths (`get_edge`, `scan_*`, `put_edge_raw`) live in `scan.rs`.
//! Cascade (`delete_edges_for_node`) lives in `cascade.rs`. Bitemporal
//! write/read primitives live in `temporal/`.

use std::path::Path;
use std::sync::Arc;

use nodedb_types::TenantId;
use redb::{Database, TableDefinition};

use super::stats::GRAPH_STATS;

/// `(collection, src, label, dst)` — a base-edge identity (no version suffix).
pub(super) type BaseKey = (String, String, String, String);

/// Tenant-qualified `BaseKey`. Used when scanning across tenants.
pub(super) type TenantBaseKey = (u64, String, String, String, String);

/// Edge table: composite key `(tid, "collection\x00src\x00label\x00dst\x00{system_from:020}")` → value.
///
/// Value is either an `EdgeValuePayload` (zerompk fixarray-3) or a single-byte
/// sentinel (`TOMBSTONE_SENTINEL`, `GDPR_ERASURE_SENTINEL`).
pub(super) const EDGES: TableDefinition<(u64, &str), &[u8]> = TableDefinition::new("edges");

/// Reverse edge index: same versioned key shape as `EDGES` but with
/// `dst`/`src` swapped. Value is empty for live edges, or a sentinel for
/// soft-deleted / erased edges (symmetry with forward).
pub(super) const REVERSE_EDGES: TableDefinition<(u64, &str), &[u8]> =
    TableDefinition::new("reverse_edges");

pub(super) fn redb_err<E: std::fmt::Display>(ctx: &str, e: E) -> crate::Error {
    crate::Error::Storage {
        engine: "graph".into(),
        detail: format!("{ctx}: {e}"),
    }
}

// Re-export shared Direction from nodedb-types.
pub use nodedb_types::graph::Direction;

/// Decoded edge record yielded by `EdgeStore::scan_all_edges_decoded`:
/// `(tenant, collection, src, label, dst, properties)`. Current-state only —
/// tombstoned and GDPR-erased edges are filtered out, and only the latest
/// non-sentinel version per base key is yielded.
pub type EdgeRecord = (TenantId, String, String, String, String, Vec<u8>);

/// A single edge with its properties.
#[derive(Debug, Clone)]
pub struct Edge {
    pub collection: String,
    pub src_id: String,
    pub label: String,
    pub dst_id: String,
    pub properties: Vec<u8>,
}

/// redb-backed edge storage for the Knowledge Graph engine.
///
/// Keys are `(TenantId, versioned_composite_key)` tuples — tenant routing
/// is structural, not lexical. Each Data Plane core owns its own
/// `EdgeStore` instance; no cross-core sharing.
pub struct EdgeStore {
    pub(super) db: Arc<Database>,
}

impl EdgeStore {
    /// Open or create the edge store database at the given path.
    pub fn open(path: &Path) -> crate::Result<Self> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        let db = Database::create(path).map_err(|e| redb_err("open", e))?;

        let write_txn = db.begin_write().map_err(|e| redb_err("begin_write", e))?;
        {
            let _ = write_txn
                .open_table(EDGES)
                .map_err(|e| redb_err("open edges", e))?;
            let _ = write_txn
                .open_table(REVERSE_EDGES)
                .map_err(|e| redb_err("open reverse_edges", e))?;
            let _ = write_txn
                .open_table(GRAPH_STATS)
                .map_err(|e| redb_err("open graph_stats", e))?;
        }
        write_txn.commit().map_err(|e| redb_err("commit", e))?;

        Ok(Self { db: Arc::new(db) })
    }
}

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

    #[test]
    fn open_creates_tables_in_new_file() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("graph.redb");
        let _ = EdgeStore::open(&path).unwrap();
        assert!(path.exists());
    }

    #[test]
    fn open_is_idempotent() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("graph.redb");
        let _a = EdgeStore::open(&path).unwrap();
        drop(_a);
        let _b = EdgeStore::open(&path).unwrap();
    }
}