⎇ 4d50bfc5879839a835e26ffda3bbc7ae3b927cc7
crates/modules/nakui/core/src/store.rs
Historia
Quién tocó qué
crudo
chore: absorbe nakui (ERP matemático) en modules/nakui
1use serde_json::Value;
2use sha2::{Digest, Sha256};
3use std::collections::HashMap;
4use thiserror::Error;
5use uuid::Uuid;
6
7use crate::delta::FieldOp;
8
9#[derive(Debug, Clone, Error)]
10pub enum StoreError {
11 #[error("entity {0} id {1} not found")]
12 NotFound(String, Uuid),
13 #[error("entity {0} id {1} already exists")]
14 Conflict(String, Uuid),
15 #[error("set on non-object record at {0} {1}")]
16 NotAnObject(String, Uuid),
17 /// Backend-specific transient or systemic failure (network, disk,
18 /// driver). Distinct from the data-shape errors above.
19 #[error("backend error: {0}")]
20 Backend(String),
21}
22
23pub trait Store {
24 fn load(&self, entity: &str, id: Uuid) -> Option<Value>;
25
26 /// Insert or replace a record without going through the morphism
27 /// pipeline. Represents external/boundary input — the source of
28 /// records that didn't originate from a kernel-validated event.
29 fn seed(&mut self, entity: &str, id: Uuid, data: Value);
30
31 /// Read-only check: would `apply(ops)` succeed against current state?
32 /// Does NOT mutate. The kernel runs this as the last step of `compute`
33 /// so that, by the time we log an event, the apply is contractually
34 /// guaranteed to land.
35 fn apply_dry_run(&self, ops: &[FieldOp]) -> Result<(), StoreError>;
36
37 fn apply(&mut self, ops: &[FieldOp]) -> Result<(), StoreError>;
38
39 /// Drop every record. Used by `reconcile` to wipe a stale store before
40 /// replaying the log. Must leave the store in the same state it would
41 /// be in immediately after construction. Implementors that override
42 /// `last_applied_seq` must reset that marker here too — a cleared
43 /// store has applied nothing.
44 fn clear(&mut self) -> Result<(), StoreError>;
45
46 /// The last log seq whose effects are reflected in this store, if
47 /// the store can persist that fact. Default `Ok(None)` covers
48 /// transient backends. The startup path uses this to skip the full
49 /// replay when the store is verifiably already in sync with the log.
50 fn last_applied_seq(&self) -> Result<Option<u64>, StoreError> {
51 Ok(None)
52 }
53
54 /// Persist the marker after a successful apply / seed / replay.
55 /// Best-effort: callers ignore failures here because a stale marker
56 /// only costs an extra full replay on next startup, never
57 /// correctness — full replay starts with `clear()`, so it tolerates
58 /// any prior state. Default impl is a no-op for transient backends.
59 fn set_last_applied_seq(&mut self, _seq: u64) -> Result<(), StoreError> {
60 Ok(())
61 }
62
63 /// Enumerate every record in canonical order: sorted first by entity
64 /// name, then by id bytes. The canonical order is what makes
65 /// `hash_state` reproducible — without it two stores with the same
66 /// records would hash differently depending on insertion order.
67 ///
68 /// Returns owned `Value`s. For an in-memory backend this clones; for
69 /// a remote backend it materializes a snapshot. V1 chooses simplicity
70 /// over streaming — the hash and drift-comparison use cases need to
71 /// see all records anyway, and an iterator over a Vec keeps the
72 /// trait method object-safe and free of async lifetime concerns.
73 fn iter(&self) -> Result<Box<dyn Iterator<Item = (String, Uuid, Value)> + '_>, StoreError>;
74
75 /// Deterministic SHA-256 of the store's full state. Two stores with
76 /// the same records (regardless of how they got there or which
77 /// backend they live in) produce the same hash; any drift produces
78 /// a different one. The default impl is the contract — backends
79 /// should only override it for backend-native acceleration (e.g.
80 /// server-side table digests), and an override must produce the
81 /// same bytes as the default.
82 ///
83 /// Framing per record:
84 /// entity_bytes | 0x00 | id_bytes | 0x00 | canonical_value_hash
85 /// The length prefix on entity/id prevents (entity="ab", id="c")
86 /// from colliding with (entity="a", id="bc"). The value bytes are
87 /// produced by `hash_value`, which walks the JSON tree with
88 /// type-tagged framing — that decouples the hash from
89 /// `serde_json::to_vec`'s representation choices (especially
90 /// integer-valued floats vs ints) so cross-backend comparison
91 /// works.
92 fn hash_state(&self) -> Result<[u8; 32], StoreError> {
93 let mut hasher = Sha256::new();
94 for (entity, id, value) in self.iter()? {
95 hasher.update(entity.as_bytes());
96 hasher.update([0u8]);
97 hasher.update(id.as_bytes());
98 hasher.update([0u8]);
99 hash_value(&mut hasher, &value);
100 }
101 Ok(hasher.finalize().into())
102 }
103}
104
105/// Canonical hash of a `serde_json::Value`. Type-tagged so a string
106/// "true" can't collide with the boolean `true`; length-prefixed so
107/// concatenation can't shift bytes between fields. Numbers normalize:
108/// any integer-valued number (i64, u64, or a finite f64 with no
109/// fractional part) is hashed as an i128 — that's what makes
110/// cross-backend equality work, since SurrealDB may round-trip
111/// what the caller wrote as `100_i64` back as the same numeric value
112/// without us needing to commit to a wire-format-specific
113/// representation.
114pub fn hash_value(hasher: &mut Sha256, v: &Value) {
115 match v {
116 Value::Null => hasher.update([TAG_NULL]),
117 Value::Bool(b) => {
118 hasher.update([TAG_BOOL]);
119 hasher.update([*b as u8]);
120 }
121 Value::Number(n) => {
122 if let Some(i) = n.as_i64() {
123 hash_int(hasher, i as i128);
124 } else if let Some(u) = n.as_u64() {
125 hash_int(hasher, u as i128);
126 } else if let Some(f) = n.as_f64() {
127 // Integer-valued floats canonicalize to int. Anything
128 // else (fractions, NaN, infinities) hashes as the raw
129 // f64 bit pattern — that's still deterministic, just
130 // not normalized.
131 if f.is_finite()
132 && f.fract() == 0.0
133 && f >= I128_MIN_AS_F64
134 && f <= I128_MAX_AS_F64
135 {
136 hash_int(hasher, f as i128);
137 } else {
138 hasher.update([TAG_FLOAT]);
139 hasher.update(f.to_bits().to_le_bytes());
140 }
141 } else {
142 // serde_json::Number guarantees one of the above; this
143 // branch only fires if a future variant appears.
144 hasher.update([TAG_FLOAT]);
145 hasher.update(f64::NAN.to_bits().to_le_bytes());
146 }
147 }
148 Value::String(s) => {
149 hasher.update([TAG_STRING]);
150 hasher.update((s.len() as u64).to_le_bytes());
151 hasher.update(s.as_bytes());
152 }
153 Value::Array(arr) => {
154 hasher.update([TAG_ARRAY]);
155 hasher.update((arr.len() as u64).to_le_bytes());
156 for item in arr {
157 hash_value(hasher, item);
158 }
159 }
160 Value::Object(map) => {
161 hasher.update([TAG_OBJECT]);
162 hasher.update((map.len() as u64).to_le_bytes());
163 // serde_json::Map without `preserve_order` is BTreeMap
164 // (alphabetical). We sort defensively in case the build
165 // pulls in `preserve_order` transitively from a future dep.
166 let mut keys: Vec<&String> = map.keys().collect();
167 keys.sort();
168 for k in keys {
169 hasher.update((k.len() as u64).to_le_bytes());
170 hasher.update(k.as_bytes());
171 hash_value(hasher, &map[k]);
172 }
173 }
174 }
175}
176
177fn hash_int(hasher: &mut Sha256, n: i128) {
178 hasher.update([TAG_INT]);
179 hasher.update(n.to_le_bytes());
180}
181
182const TAG_NULL: u8 = 0;
183const TAG_BOOL: u8 = 1;
184const TAG_INT: u8 = 2;
185const TAG_FLOAT: u8 = 3;
186const TAG_STRING: u8 = 4;
187const TAG_ARRAY: u8 = 5;
188const TAG_OBJECT: u8 = 6;
189
190// f64 can't represent i128::MAX exactly; the cast truncates upward to
191// the next representable f64. Use those as the comparison bounds so
192// `f as i128` stays well-defined.
193const I128_MIN_AS_F64: f64 = -1.7014118346046923e38;
194const I128_MAX_AS_F64: f64 = 1.7014118346046923e38;
195
196#[derive(Debug, Default, Clone, PartialEq)]
197pub struct MemoryStore {
198 records: HashMap<String, HashMap<Uuid, Value>>,
199 /// Last log seq whose effects are reflected here. In-process only —
200 /// resets to `None` on construction or `clear`. The skip-replay
201 /// optimization in `nakui run` benefits the persistent backends;
202 /// for `MemoryStore` it's harmless bookkeeping (process restart =
203 /// new store = `None`, which forces full replay).
204 last_applied: Option<u64>,
205}
206
207impl MemoryStore {
208 pub fn new() -> Self {
209 Self::default()
210 }
211
212 /// Borrow the internal records map. Used by `Snapshot::from_memory_store`
213 /// to capture state for snapshot persistence.
214 pub fn records(&self) -> &HashMap<String, HashMap<Uuid, Value>> {
215 &self.records
216 }
217}
218
219impl Store for MemoryStore {
220 fn load(&self, entity: &str, id: Uuid) -> Option<Value> {
221 self.records.get(entity)?.get(&id).cloned()
222 }
223
224 fn seed(&mut self, entity: &str, id: Uuid, data: Value) {
225 self.records
226 .entry(entity.to_string())
227 .or_default()
228 .insert(id, data);
229 }
230
231 fn apply_dry_run(&self, ops: &[FieldOp]) -> Result<(), StoreError> {
232 for op in ops {
233 match op {
234 FieldOp::Set { path, .. } => {
235 match self.records.get(&path.entity).and_then(|m| m.get(&path.id)) {
236 None => {
237 return Err(StoreError::NotFound(path.entity.clone(), path.id));
238 }
239 Some(Value::Object(_)) => {}
240 Some(_) => {
241 return Err(StoreError::NotAnObject(path.entity.clone(), path.id));
242 }
243 }
244 }
245 FieldOp::Create { entity, id, .. } => {
246 if self
247 .records
248 .get(entity)
249 .and_then(|m| m.get(id))
250 .is_some()
251 {
252 return Err(StoreError::Conflict(entity.clone(), *id));
253 }
254 }
255 FieldOp::Delete { entity, id } => {
256 if self
257 .records
258 .get(entity)
259 .and_then(|m| m.get(id))
260 .is_none()
261 {
262 return Err(StoreError::NotFound(entity.clone(), *id));
263 }
264 }
265 }
266 }
267 Ok(())
268 }
269
270 fn apply(&mut self, ops: &[FieldOp]) -> Result<(), StoreError> {
271 self.apply_dry_run(ops)?;
272 for op in ops {
273 match op {
274 FieldOp::Set { path, value } => {
275 let rec = self
276 .records
277 .get_mut(&path.entity)
278 .and_then(|m| m.get_mut(&path.id))
279 .expect("validated by dry_run");
280 let map = match rec {
281 Value::Object(m) => m,
282 _ => unreachable!("dry_run guards against non-object"),
283 };
284 map.insert(path.field.clone(), value.clone());
285 }
286 FieldOp::Create { entity, id, data } => {
287 self.records
288 .entry(entity.clone())
289 .or_default()
290 .insert(*id, data.clone());
291 }
292 FieldOp::Delete { entity, id } => {
293 self.records
294 .get_mut(entity)
295 .expect("validated by dry_run")
296 .remove(id);
297 }
298 }
299 }
300 Ok(())
301 }
302
303 fn clear(&mut self) -> Result<(), StoreError> {
304 self.records.clear();
305 self.last_applied = None;
306 Ok(())
307 }
308
309 fn last_applied_seq(&self) -> Result<Option<u64>, StoreError> {
310 Ok(self.last_applied)
311 }
312
313 fn set_last_applied_seq(&mut self, seq: u64) -> Result<(), StoreError> {
314 self.last_applied = Some(seq);
315 Ok(())
316 }
317
318 fn iter(&self) -> Result<Box<dyn Iterator<Item = (String, Uuid, Value)> + '_>, StoreError> {
319 let mut out: Vec<(String, Uuid, Value)> = self
320 .records
321 .iter()
322 .flat_map(|(entity, m)| {
323 m.iter()
324 .map(move |(id, v)| (entity.clone(), *id, v.clone()))
325 })
326 .collect();
327 out.sort_by(|a, b| a.0.cmp(&b.0).then_with(|| a.1.as_bytes().cmp(b.1.as_bytes())));
328 Ok(Box::new(out.into_iter()))
329 }
330}
331
332#[cfg(test)]
333mod tests {
334 use super::*;
335 use crate::delta::{FieldOp, FieldPath};
336 use serde_json::json;
337
338 #[test]
339 fn dry_run_rejects_set_on_non_object() {
340 let mut store = MemoryStore::new();
341 let id = Uuid::new_v4();
342 store.seed("Caja", id, json!(42)); // not an object
343 let op = FieldOp::Set {
344 path: FieldPath {
345 entity: "Caja".into(),
346 id,
347 field: "saldo".into(),
348 },
349 value: json!(100),
350 };
351 match store.apply_dry_run(&[op.clone()]) {
352 Err(StoreError::NotAnObject(e, i)) => {
353 assert_eq!(e, "Caja");
354 assert_eq!(i, id);
355 }
356 other => panic!("expected NotAnObject, got {:?}", other),
357 }
358 // apply must reject too without panicking.
359 assert!(matches!(
360 store.apply(&[op]),
361 Err(StoreError::NotAnObject(_, _))
362 ));
363 }
364
365 #[test]
366 fn dry_run_rejects_create_conflict() {
367 let mut store = MemoryStore::new();
368 let id = Uuid::new_v4();
369 store.seed("Caja", id, json!({"id": id.to_string()}));
370 let op = FieldOp::Create {
371 entity: "Caja".into(),
372 id,
373 data: json!({"id": id.to_string()}),
374 };
375 assert!(matches!(
376 store.apply_dry_run(&[op]),
377 Err(StoreError::Conflict(_, _))
378 ));
379 }
380
381 #[test]
382 fn dry_run_passes_for_valid_set() {
383 let mut store = MemoryStore::new();
384 let id = Uuid::new_v4();
385 store.seed("Caja", id, json!({"saldo": 100, "currency": "USD"}));
386 let op = FieldOp::Set {
387 path: FieldPath {
388 entity: "Caja".into(),
389 id,
390 field: "saldo".into(),
391 },
392 value: json!(150),
393 };
394 assert!(store.apply_dry_run(&[op]).is_ok());
395 }
396
397 #[test]
398 fn iter_returns_canonical_order_regardless_of_insertion() {
399 let a = Uuid::new_v4();
400 let b = Uuid::new_v4();
401 let c = Uuid::new_v4();
402
403 let mut s1 = MemoryStore::new();
404 s1.seed("Caja", a, json!({"id": a.to_string(), "x": 1}));
405 s1.seed("Movimiento", c, json!({"id": c.to_string(), "y": 3}));
406 s1.seed("Caja", b, json!({"id": b.to_string(), "x": 2}));
407
408 let mut s2 = MemoryStore::new();
409 s2.seed("Movimiento", c, json!({"id": c.to_string(), "y": 3}));
410 s2.seed("Caja", b, json!({"id": b.to_string(), "x": 2}));
411 s2.seed("Caja", a, json!({"id": a.to_string(), "x": 1}));
412
413 let r1: Vec<_> = s1.iter().unwrap().collect();
414 let r2: Vec<_> = s2.iter().unwrap().collect();
415 assert_eq!(r1, r2, "iter order must be insertion-independent");
416
417 // Entities lexicographically sorted (Caja before Movimiento).
418 let entities: Vec<&str> = r1.iter().map(|(e, _, _)| e.as_str()).collect();
419 assert_eq!(entities, vec!["Caja", "Caja", "Movimiento"]);
420
421 // Within Caja, ids in byte order.
422 let caja_ids: Vec<Uuid> = r1
423 .iter()
424 .filter(|(e, _, _)| e == "Caja")
425 .map(|(_, id, _)| *id)
426 .collect();
427 let mut expected = vec![a, b];
428 expected.sort_by(|x, y| x.as_bytes().cmp(y.as_bytes()));
429 assert_eq!(caja_ids, expected);
430 }
431
432 #[test]
433 fn hash_state_is_deterministic_and_independent_of_insertion_order() {
434 let a = Uuid::new_v4();
435 let b = Uuid::new_v4();
436
437 let mut s1 = MemoryStore::new();
438 s1.seed("Caja", a, json!({"id": a.to_string(), "saldo": 100}));
439 s1.seed("Caja", b, json!({"id": b.to_string(), "saldo": 200}));
440
441 let mut s2 = MemoryStore::new();
442 s2.seed("Caja", b, json!({"id": b.to_string(), "saldo": 200}));
443 s2.seed("Caja", a, json!({"id": a.to_string(), "saldo": 100}));
444
445 assert_eq!(
446 s1.hash_state().unwrap(),
447 s2.hash_state().unwrap(),
448 "equal state must hash identically regardless of how it was built"
449 );
450 }
451
452 #[test]
453 fn hash_state_changes_when_state_changes() {
454 let a = Uuid::new_v4();
455
456 let mut s1 = MemoryStore::new();
457 s1.seed("Caja", a, json!({"id": a.to_string(), "saldo": 100}));
458
459 let mut s2 = MemoryStore::new();
460 s2.seed("Caja", a, json!({"id": a.to_string(), "saldo": 101}));
461
462 assert_ne!(
463 s1.hash_state().unwrap(),
464 s2.hash_state().unwrap(),
465 "off-by-one in a single field must produce a different hash"
466 );
467
468 // Adding a record changes the hash too.
469 let mut s3 = MemoryStore::new();
470 s3.seed("Caja", a, json!({"id": a.to_string(), "saldo": 100}));
471 s3.seed("Caja", Uuid::new_v4(), json!({"id": "extra", "saldo": 0}));
472 assert_ne!(s1.hash_state().unwrap(), s3.hash_state().unwrap());
473 }
474
475 #[test]
476 fn last_applied_seq_round_trips_and_resets_on_clear() {
477 let mut store = MemoryStore::new();
478 assert_eq!(
479 store.last_applied_seq().unwrap(),
480 None,
481 "fresh MemoryStore has no marker"
482 );
483 store.set_last_applied_seq(5).unwrap();
484 assert_eq!(store.last_applied_seq().unwrap(), Some(5));
485 store.set_last_applied_seq(12).unwrap();
486 assert_eq!(store.last_applied_seq().unwrap(), Some(12));
487 store.clear().unwrap();
488 assert_eq!(
489 store.last_applied_seq().unwrap(),
490 None,
491 "clear() resets the marker — a cleared store has applied nothing"
492 );
493 }
494
495 #[test]
496 fn integer_and_integer_valued_float_hash_identically() {
497 // The cross-backend property: the same numeric value, written
498 // by a backend as i64 vs read back as integer-valued f64,
499 // must hash the same.
500 let int_value = json!({"saldo": 100_i64});
501 let float_value = json!({"saldo": 100.0_f64});
502
503 let mut h_int = sha2::Sha256::new();
504 super::hash_value(&mut h_int, &int_value);
505 let mut h_float = sha2::Sha256::new();
506 super::hash_value(&mut h_float, &float_value);
507 assert_eq!(
508 h_int.finalize(),
509 h_float.finalize(),
510 "integer-valued numbers must canonicalize regardless of source representation"
511 );
512 }
513
514 #[test]
515 fn fractional_floats_do_not_canonicalize_to_int() {
516 // Floats with fractional parts must remain floats — collapsing
517 // 100.5 into 100 would hide real differences.
518 let int_value = json!({"x": 100_i64});
519 let frac_value = json!({"x": 100.5_f64});
520
521 let mut h_int = sha2::Sha256::new();
522 super::hash_value(&mut h_int, &int_value);
523 let mut h_frac = sha2::Sha256::new();
524 super::hash_value(&mut h_frac, &frac_value);
525 assert_ne!(
526 h_int.finalize(),
527 h_frac.finalize(),
528 "100 and 100.5 must hash differently"
529 );
530 }
531
532 #[test]
533 fn same_object_with_different_insertion_order_hashes_same() {
534 // serde_json::Map is BTreeMap by default but we sort defensively
535 // in case `preserve_order` is enabled by some transitive dep.
536 let mut m1 = serde_json::Map::new();
537 m1.insert("a".into(), json!(1));
538 m1.insert("b".into(), json!(2));
539 m1.insert("c".into(), json!(3));
540 let mut m2 = serde_json::Map::new();
541 m2.insert("c".into(), json!(3));
542 m2.insert("a".into(), json!(1));
543 m2.insert("b".into(), json!(2));
544
545 let mut h1 = sha2::Sha256::new();
546 super::hash_value(&mut h1, &Value::Object(m1));
547 let mut h2 = sha2::Sha256::new();
548 super::hash_value(&mut h2, &Value::Object(m2));
549 assert_eq!(h1.finalize(), h2.finalize());
550 }
551
552 #[test]
553 fn type_tagged_framing_distinguishes_string_from_number() {
554 // The string "42" must not collide with the number 42.
555 let str_v = json!("42");
556 let num_v = json!(42);
557 let mut h_str = sha2::Sha256::new();
558 super::hash_value(&mut h_str, &str_v);
559 let mut h_num = sha2::Sha256::new();
560 super::hash_value(&mut h_num, &num_v);
561 assert_ne!(h_str.finalize(), h_num.finalize());
562
563 // Bool true must not collide with the number 1.
564 let bool_v = json!(true);
565 let one_v = json!(1);
566 let mut h_bool = sha2::Sha256::new();
567 super::hash_value(&mut h_bool, &bool_v);
568 let mut h_one = sha2::Sha256::new();
569 super::hash_value(&mut h_one, &one_v);
570 assert_ne!(h_bool.finalize(), h_one.finalize());
571 }
572
573 #[test]
574 fn empty_store_has_a_well_defined_hash() {
575 let s1 = MemoryStore::new();
576 let s2 = MemoryStore::new();
577 assert_eq!(s1.hash_state().unwrap(), s2.hash_state().unwrap());
578 // The empty hash is the SHA-256 of an empty input — fix the
579 // expected bytes so an accidental framing change in `hash_state`
580 // can't silently sail through.
581 let expected = hex_decode(
582 "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
583 );
584 assert_eq!(s1.hash_state().unwrap().to_vec(), expected);
585 }
586
587 fn hex_decode(s: &str) -> Vec<u8> {
588 (0..s.len())
589 .step_by(2)
590 .map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("hex"))
591 .collect()
592 }
593}