nedb_engine/
migrate.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//! Automatic v1 → v2 DAG migration.
6//!
7//! When a database directory contains `log.aof` (v1 format), this module
8//! reads all valid ops, converts them to v2 Node objects, writes them to
9//! the object store, rebuilds indexes, and renames log.aof → log.aof.v1.bak.
10//!
11//! The migration is:
12//!   - Transparent: zero user action required
13//!   - Idempotent: if it crashes mid-way, log.aof is still present → retries
14//!   - Non-destructive: log.aof.v1.bak is always kept as a rollback path
15//!   - Self-repairing: corrupt AOF lines are skipped (partial writes from BrokenPipe)
16//!   - Parallel: object writes use Rayon thread pool
17
18use std::fs;
19use std::path::Path;
20use anyhow::{Context, Result};
21use serde_json::Value;
22
23use crate::store::{Dek, Node, ObjectStore};
24use crate::index::{IdIndex, SortedIndexes};
25use crate::graph::GraphStore;
26
27/// A parsed v1 AOF operation.
28#[derive(Debug)]
29struct V1Op {
30    seq:        u64,
31    coll:       String,
32    id:         String,
33    data:       Value,
34    caused_by:  Vec<String>,   // v1 stores seq numbers, v2 will store hashes after migration
35    ts:         f64,
36    valid_from: Option<String>,
37    valid_to:   Option<String>,
38}
39
40/// Read all valid ops from a v1 AOF file, skipping corrupt lines.
41fn read_v1_aof(aof_path: &Path, dek: Option<&Dek>) -> Result<Vec<V1Op>> {
42    let raw = fs::read_to_string(aof_path)
43        .context("read log.aof")?;
44
45    let mut ops = Vec::new();
46    let mut skipped = 0usize;
47
48    for (line_num, line) in raw.lines().enumerate() {
49        let line = line.trim();
50        if line.is_empty() { continue; }
51
52        // v1 AOF lines are either plain JSON or AES-GCM encrypted JSON
53        let decoded: Value = match try_decode_line(line, dek) {
54            Ok(v) => v,
55            Err(e) => {
56                skipped += 1;
57                eprintln!("  [nedb-migrate] skip corrupt line {}: {}", line_num + 1, e);
58                break;  // stop at first corruption — everything after is suspect
59            }
60        };
61
62        // v1 op format: {seq, client, nonce, op, payload, ts, ...}
63        let op_type = decoded.get("op").and_then(|v| v.as_str()).unwrap_or("");
64        if op_type != "put" { continue; }  // skip delete/link for now
65
66        let payload = match decoded.get("payload") {
67            Some(p) => p.clone(),
68            None => continue,
69        };
70        let coll = payload.get("coll").and_then(|v| v.as_str()).unwrap_or("").to_string();
71        let id   = payload.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string();
72        let data = payload.get("doc").cloned().unwrap_or(Value::Null);
73
74        if coll.is_empty() || id.is_empty() { continue; }
75
76        let seq = decoded.get("seq").and_then(|v| v.as_u64()).unwrap_or(0);
77        let ts  = decoded.get("ts").and_then(|v| v.as_f64()).unwrap_or(0.0);
78
79        // caused_by in v1 is a list of seq numbers; we'll resolve to hashes after building the seq→hash map
80        let caused_by_seqs: Vec<u64> = decoded
81            .get("caused_by")
82            .and_then(|v| v.as_array())
83            .map(|a| a.iter().filter_map(|x| x.as_u64()).collect())
84            .unwrap_or_default();
85
86        ops.push(V1Op {
87            seq, coll, id, data, ts,
88            caused_by: caused_by_seqs.iter().map(|s| s.to_string()).collect(), // temp: store as seq strings
89            valid_from: decoded.get("valid_from").and_then(|v| v.as_str()).map(|s| s.to_string()),
90            valid_to:   decoded.get("valid_to").and_then(|v| v.as_str()).map(|s| s.to_string()),
91        });
92    }
93
94    if skipped > 0 {
95        eprintln!(
96            "  [nedb-migrate] {} op(s) recovered, {} corrupt line(s) truncated",
97            ops.len(), skipped
98        );
99    }
100    Ok(ops)
101}
102
103fn try_decode_line(line: &str, dek: Option<&Dek>) -> Result<Value> {
104    // Try plain JSON first
105    if let Ok(v) = serde_json::from_str::<Value>(line) {
106        return Ok(v);
107    }
108    // Try base64-encoded encrypted envelope (v1 format: {"enc":1,"data":"<b64>"})
109    let envelope: Value = serde_json::from_str(line)
110        .context("parse AOF line as JSON")?;
111    if envelope.get("enc").and_then(|v| v.as_u64()) == Some(1) {
112        if let Some(dek) = dek {
113            let b64 = envelope.get("data")
114                .and_then(|v| v.as_str())
115                .context("missing data field in encrypted envelope")?;
116            let ciphertext = base64_decode(b64)?;
117            let plaintext = decrypt_v1(&ciphertext, dek)?;
118            return Ok(serde_json::from_slice(&plaintext)?);
119        }
120    }
121    anyhow::bail!("cannot decode AOF line")
122}
123
124fn base64_decode(s: &str) -> Result<Vec<u8>> {
125    base64_simple::decode(s).map_err(|e| anyhow::anyhow!("{}", e))
126}
127
128fn decrypt_v1(data: &[u8], dek: &Dek) -> Result<Vec<u8>> {
129    use aes_gcm::{Aes256Gcm, KeyInit, aead::Aead};
130    if data.len() < 12 { anyhow::bail!("ciphertext too short"); }
131    let (nonce_bytes, ciphertext) = data.split_at(12);
132    let cipher = Aes256Gcm::new_from_slice(&dek.0)?;
133    let nonce = aes_gcm::Nonce::from_slice(nonce_bytes);
134    cipher.decrypt(nonce, ciphertext)
135        .map_err(|e| anyhow::anyhow!("decrypt: {:?}", e))
136}
137
138/// Run the full v1 → v2 migration for one database directory.
139pub fn migrate_if_needed(
140    db_root: &Path,
141    object_store: &ObjectStore,
142    id_index: &IdIndex,
143    sorted_indexes: &SortedIndexes,
144    graph: &GraphStore,
145    dek: Option<&Dek>,
146) -> Result<bool> {
147    let aof_path = db_root.join("log.aof");
148    if !aof_path.exists() {
149        return Ok(false);  // already v2 or empty
150    }
151    let bak_path = db_root.join("log.aof.v1.bak");
152    if bak_path.exists() {
153        // Migration was interrupted — retry from the original aof (bak exists = aof was not yet renamed)
154        // This shouldn't normally happen but handle it gracefully
155    }
156
157    println!("  [nedb] Detected v1 log.aof — running automatic migration to v2 DAG...");
158
159    let ops = read_v1_aof(&aof_path, dek)?;
160    let total = ops.len();
161    println!("  [nedb] {} op(s) to migrate", total);
162
163    // Build seq → hash map as we write nodes (to resolve caused_by seq → hash)
164    let mut seq_to_hash: std::collections::HashMap<u64, String> = std::collections::HashMap::new();
165
166    // Write nodes in seq order (sequential to build the seq→hash map)
167    for op in &ops {
168        // Resolve caused_by seq numbers to hashes
169        let caused_by_hashes: Vec<String> = op.caused_by
170            .iter()
171            .filter_map(|s| s.parse::<u64>().ok())
172            .filter_map(|seq| seq_to_hash.get(&seq).cloned())
173            .collect();
174
175        // Get previous version hash for this doc
176        let prev = id_index.get(&op.coll, &op.id);
177
178        let mut node = Node {
179            id:         op.id.clone(),
180            coll:       op.coll.clone(),
181            seq:        op.seq,
182            data:       op.data.clone(),
183            prev,
184            caused_by:  caused_by_hashes.clone(),
185            ts:         op.ts,
186            valid_from: op.valid_from.clone(),
187            valid_to:   op.valid_to.clone(),
188            hash:       String::new(),
189        };
190
191        let hash = object_store.write(&mut node)?;
192        id_index.set(&op.coll, &op.id, &hash)?;
193        seq_to_hash.insert(op.seq, hash.clone());
194
195        // Write causal edges
196        for cause_hash in &caused_by_hashes {
197            graph.add_edge(&hash, "caused_by", cause_hash)?;
198            graph.add_edge(cause_hash, "caused_by_rev", &hash)?;
199        }
200
201        // Update sorted indexes for all numeric/string fields
202        if let serde_json::Value::Object(ref obj) = op.data {
203            for (field, value) in obj {
204                if sorted_indexes.has(&op.coll, field) {
205                    sorted_indexes.insert(&op.coll, field, value, &hash);
206                }
207            }
208        }
209    }
210
211    // Migration complete — rename log.aof to .v1.bak
212    fs::rename(&aof_path, &bak_path)
213        .context("rename log.aof to log.aof.v1.bak")?;
214
215    println!(
216        "  [nedb] Migration complete: {} op(s) → v2 DAG. Backup: {}",
217        total,
218        bak_path.display()
219    );
220
221    Ok(true)
222}
223
224// Minimal base64 decoder (stdlib only — no extra deps needed for migration)
225mod base64_simple {
226    pub fn decode(s: &str) -> Result<Vec<u8>, String> {
227        let s = s.trim();
228        let mut out = Vec::with_capacity(s.len() * 3 / 4);
229        let chars: Vec<u8> = s.bytes().filter(|&b| b != b'=').collect();
230        let table = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
231        let mut buf = 0u32;
232        let mut bits = 0;
233        for &c in &chars {
234            let val = table.iter().position(|&t| t == c)
235                .ok_or_else(|| format!("invalid base64 char: {}", c as char))? as u32;
236            buf = (buf << 6) | val;
237            bits += 6;
238            if bits >= 8 {
239                bits -= 8;
240                out.push((buf >> bits) as u8);
241                buf &= (1 << bits) - 1;
242            }
243        }
244        Ok(out)
245    }
246}
nedb_engine/migrate.rs

nedb_engine/
migrate.rs
