feat(nakui): Fase 1 del ERP — FieldKind Select + AutoId, seed inyecta id
1//! Implementación de [`MetaBackend`] para Nakui — compone
2//! `nakui_core::store::MemoryStore`, `event_log::EventLog`, los
3//! `Executor`s por módulo, y la lógica de auto-compaction.
4//!
5//! Es lo único que sabe de Nakui en el binario nuevo. El widget de
6//! UI no toca ninguno de estos tipos directamente.
7
8use std::collections::BTreeMap;
9use std::path::{Path, PathBuf};
10use std::sync::{Arc, Mutex};
11
12use serde_json::{json, Value};
13use uuid::Uuid;
14
15use nahual_meta_runtime::{MetaBackend, WriteOutcome};
16use nakui_core::delta::{FieldOp, FieldPath};
17use nakui_core::event_log::{
18 execute_and_log_with_recovery, replay_with_snapshot_into, EventLog, LogEntry, Snapshot,
19};
20use nakui_core::executor::Executor;
21use nakui_core::store::{MemoryStore, Store};
22
23/// Path del snapshot sibling del log:
24/// `nakui-ui-state.jsonl` ↔ `nakui-ui-state.snap.json`.
25pub fn snapshot_path_for(log_path: &Path) -> PathBuf {
26 log_path.with_extension("snap.json")
27}
28
29/// Si el log file tiene >= `threshold` entries, captura un snapshot
30/// del store actual y compacta el log dejando 1 entry como anchor del
31/// cursor. Idempotente abajo del threshold o con < 2 entries.
32///
33/// Ver el doc original (commit del runtime compact) para detalles
34/// sobre el anchor invariant. Re-locado acá porque es detalle del
35/// backend, no del widget.
36pub fn maybe_compact_log(
37 log: &mut EventLog,
38 snap_path: &Path,
39 store: &MemoryStore,
40 threshold: usize,
41) -> Result<Option<String>, String> {
42 if threshold == 0 {
43 return Ok(None);
44 }
45 let entry_count = log
46 .entries()
47 .map_err(|e| format!("read entries: {e}"))?
48 .len();
49 if entry_count < threshold || entry_count < 2 {
50 return Ok(None);
51 }
52 let snap_seq = log.next_seq() - 1;
53 let through = log.next_seq() - 2;
54 let snap = Snapshot::from_memory_store(store, snap_seq);
55 snap.write(snap_path)
56 .map_err(|e| format!("write snapshot {}: {e}", snap_path.display()))?;
57 log.compact_through(through)
58 .map_err(|e| format!("compact_through({through}): {e}"))?;
59 Ok(Some(format!(
60 "auto-compact: snapshot @ seq {snap_seq}, {} entries dropped (1 anchor kept)",
61 entry_count - 1
62 )))
63}
64
65/// Estado inicial del backend tras abrir el log + cargar snapshot
66/// + replay. Devuelto desde [`NakuiBackend::open`] para que el caller
67/// (typicamente `main.rs`) acumule mensajes informativos al banner.
68pub struct OpenStatus {
69 /// Mensaje "log X cargado: next_seq=N (snapshot @ seq K)" o similar.
70 pub init_toast: Option<String>,
71 /// Errores no-fatales acumulados (snapshot corrupto, replay falló,
72 /// log inaccesible). El backend igualmente queda usable
73 /// (eventualmente in-memory only si log_arc es None).
74 pub load_error: Option<String>,
75}
76
77/// Backend Nakui: WAL persistente + MemoryStore + executors por
78/// módulo + auto-compaction.
79///
80/// Implementa [`MetaBackend`] proyectando cada operación al
81/// pipeline de nakui-core (compute → log → apply para morphisms;
82/// log → apply para seed/edit/delete).
83pub struct NakuiBackend {
84 /// Store compartido (Arc para que el render pueda hacer reads
85 /// sin bloquear writes; el lock interno serializa).
86 store: Arc<Mutex<MemoryStore>>,
87 /// Log persistente. `None` si abrir falló — el backend degrada
88 /// a in-memory only (writes no se persisten; reads siguen).
89 event_log: Option<Arc<Mutex<EventLog>>>,
90 /// Executors indexados por `module.id`. Los módulos sin
91 /// `nakui_module_dir` no aparecen acá; sus llamadas a
92 /// `morphism()` rebotan con error claro.
93 executors: BTreeMap<String, Arc<Executor>>,
94 /// Path del snapshot (cacheado del init).
95 snap_path: PathBuf,
96 /// Threshold de auto-compaction. `0` = desactivado.
97 snapshot_threshold: usize,
98 /// Contador de writes desde el último compact. Se resetea al
99 /// disparar compact.
100 writes_since_compact: u64,
101}
102
103impl NakuiBackend {
104 /// Abre/crea el log en `log_path`, intenta cargar el snapshot
105 /// sibling, hace replay al store. Si el log no abre, degrada a
106 /// in-memory only. Ningún error es fatal — los mensajes se
107 /// devuelven en `OpenStatus` para que el caller los acumule.
108 ///
109 /// `executors` se pasan ya cargados (la lógica de qué módulos
110 /// declaran `nakui_module_dir` es responsabilidad del caller).
111 pub fn open(
112 log_path: PathBuf,
113 snapshot_threshold: usize,
114 executors: BTreeMap<String, Arc<Executor>>,
115 ) -> (Self, OpenStatus) {
116 let snap_path = snapshot_path_for(&log_path);
117 let mut store = MemoryStore::new();
118 let mut init_toast: Option<String> = None;
119 let mut load_error: Option<String> = None;
120
121 // Cargar snapshot (si existe).
122 let snapshot: Option<Snapshot> = match Snapshot::load(&snap_path) {
123 Ok(s) => s,
124 Err(e) => {
125 load_error = Some(format!(
126 "snapshot {}: {e} — full replay",
127 snap_path.display()
128 ));
129 None
130 }
131 };
132
133 let event_log = match EventLog::open(&log_path) {
134 Ok(mut log) => {
135 match replay_with_snapshot_into(&log, snapshot.as_ref(), &mut store) {
136 Ok(()) => {
137 let n = log.next_seq();
138 let from_snap = snapshot
139 .as_ref()
140 .map(|s| format!(" (snapshot @ seq {})", s.seq))
141 .unwrap_or_default();
142 if n > 0 {
143 init_toast = Some(format!(
144 "log {} cargado: next_seq={n}{from_snap}",
145 log_path.display()
146 ));
147 } else {
148 init_toast = Some(format!("log nuevo en {}", log_path.display()));
149 }
150
151 // Auto-compact si pasamos el threshold.
152 match maybe_compact_log(&mut log, &snap_path, &store, snapshot_threshold) {
153 Ok(Some(msg)) => {
154 let prev = init_toast.unwrap_or_default();
155 init_toast = Some(format!("{prev}; {msg}"));
156 }
157 Ok(None) => {}
158 Err(e) => {
159 let msg = format!("auto-compact: {e}");
160 load_error = Some(match load_error {
161 Some(p) => format!("{p}; {msg}"),
162 None => msg,
163 });
164 }
165 }
166 Some(Arc::new(Mutex::new(log)))
167 }
168 Err(e) => {
169 let msg = format!(
170 "replay del log {} falló: {e} — running in-memory",
171 log_path.display()
172 );
173 load_error = Some(match load_error {
174 Some(p) => format!("{p}; {msg}"),
175 None => msg,
176 });
177 None
178 }
179 }
180 }
181 Err(e) => {
182 let msg = format!(
183 "abrir log {}: {e} — running in-memory only",
184 log_path.display()
185 );
186 load_error = Some(match load_error {
187 Some(p) => format!("{p}; {msg}"),
188 None => msg,
189 });
190 None
191 }
192 };
193
194 let backend = NakuiBackend {
195 store: Arc::new(Mutex::new(store)),
196 event_log,
197 executors,
198 snap_path,
199 snapshot_threshold,
200 writes_since_compact: 0,
201 };
202 (
203 backend,
204 OpenStatus {
205 init_toast,
206 load_error,
207 },
208 )
209 }
210
211 /// Increment + check del threshold; si cruza, captura snapshot
212 /// + compacta. Devuelve el mensaje de status para concatenar al
213 /// `WriteOutcome.post_status`.
214 fn tick_compact(&mut self) -> Option<String> {
215 if self.snapshot_threshold == 0 {
216 return None;
217 }
218 self.writes_since_compact += 1;
219 if self.writes_since_compact < self.snapshot_threshold as u64 {
220 return None;
221 }
222 let log_arc = self.event_log.as_ref()?.clone();
223 let mut log = match log_arc.lock() {
224 Ok(l) => l,
225 Err(_) => return Some("auto-compact skip: log mutex envenenado".into()),
226 };
227 let store = match self.store.lock() {
228 Ok(s) => s,
229 Err(_) => return Some("auto-compact skip: store mutex envenenado".into()),
230 };
231 match maybe_compact_log(&mut log, &self.snap_path, &store, self.snapshot_threshold) {
232 Ok(Some(msg)) => {
233 self.writes_since_compact = 0;
234 Some(msg)
235 }
236 Ok(None) => {
237 self.writes_since_compact = 0;
238 None
239 }
240 Err(e) => Some(format!("auto-compact: {e}")),
241 }
242 }
243
244 /// Helper: append una entry al log si está disponible. Errors si
245 /// el lock falla o el append falla.
246 fn append_log(&self, entry: LogEntry) -> Result<(), String> {
247 let Some(log_arc) = self.event_log.as_ref() else {
248 return Ok(()); // in-memory mode, no log.
249 };
250 let mut log = log_arc
251 .lock()
252 .map_err(|_| "log mutex envenenado".to_string())?;
253 log.append(entry).map_err(|e| format!("append al log: {e}"))
254 }
255}
256
257impl MetaBackend for NakuiBackend {
258 fn list_records(&self, entity: &str) -> Vec<(Uuid, Value)> {
259 let store = match self.store.lock() {
260 Ok(g) => g,
261 Err(_) => return Vec::new(),
262 };
263 let it = match store.iter() {
264 Ok(i) => i,
265 Err(_) => return Vec::new(),
266 };
267 let mut out: Vec<(Uuid, Value)> = it
268 .filter(|(e, _, _)| e == entity)
269 .map(|(_, id, v)| (id, v))
270 .collect();
271 out.sort_by(|a, b| a.0.as_bytes().cmp(b.0.as_bytes()));
272 out
273 }
274
275 fn load_record(&self, entity: &str, id: Uuid) -> Option<Value> {
276 self.store.lock().ok()?.load(entity, id)
277 }
278
279 fn seed(
280 &mut self,
281 entity: &str,
282 data: serde_json::Map<String, Value>,
283 ) -> Result<WriteOutcome, String> {
284 let id = Uuid::new_v4();
285 // El `id` de la entity = la clave del store. Inyectarlo en el
286 // record hace que `data.id` y la clave coincidan — los schemas
287 // Nickel suelen declarar `id | String` y los morfismos lo leen.
288 let mut data = data;
289 data.insert("id".to_string(), Value::String(id.to_string()));
290 let value = Value::Object(data);
291 // WAL: log primero, store después.
292 if self.event_log.is_some() {
293 let seq = {
294 let log_arc = self.event_log.as_ref().expect("checked above").clone();
295 let log = log_arc
296 .lock()
297 .map_err(|_| "log mutex envenenado".to_string())?;
298 log.next_seq()
299 };
300 self.append_log(LogEntry::Seed {
301 seq,
302 entity: entity.to_string(),
303 id,
304 data: value.clone(),
305 schema_hash: None,
306 })?;
307 }
308 let mut store = self
309 .store
310 .lock()
311 .map_err(|_| "store mutex envenenado".to_string())?;
312 store.seed(entity, id, value);
313 drop(store);
314 let post_status = self.tick_compact();
315 Ok(WriteOutcome {
316 id: Some(id),
317 changed: 1,
318 post_status,
319 })
320 }
321
322 fn update(
323 &mut self,
324 entity: &str,
325 id: Uuid,
326 set: serde_json::Map<String, Value>,
327 clear: Vec<String>,
328 ) -> Result<WriteOutcome, String> {
329 if set.is_empty() && clear.is_empty() {
330 return Ok(WriteOutcome::no_change(id));
331 }
332 // Construir ops: Set primero, después Clear (la sem es
333 // independiente del orden, pero estable mejor para diff).
334 let mut ops: Vec<FieldOp> = set
335 .iter()
336 .map(|(field, value)| FieldOp::Set {
337 path: FieldPath {
338 entity: entity.to_string(),
339 id,
340 field: field.clone(),
341 },
342 value: value.clone(),
343 })
344 .collect();
345 for field in &clear {
346 ops.push(FieldOp::Clear {
347 path: FieldPath {
348 entity: entity.to_string(),
349 id,
350 field: field.clone(),
351 },
352 });
353 }
354 let changed = set.len() + clear.len();
355
356 // Log: Morphism { ui.edit_record, ops, params: {entity, id, fields, cleared} }.
357 if self.event_log.is_some() {
358 let seq = {
359 let log_arc = self.event_log.as_ref().expect("checked").clone();
360 let log = log_arc
361 .lock()
362 .map_err(|_| "log mutex envenenado".to_string())?;
363 log.next_seq()
364 };
365 let mut params = serde_json::Map::new();
366 params.insert("entity".into(), json!(entity));
367 params.insert("id".into(), json!(id.to_string()));
368 if !set.is_empty() {
369 params.insert("fields".into(), Value::Object(set.clone()));
370 }
371 if !clear.is_empty() {
372 params.insert(
373 "cleared".into(),
374 Value::Array(clear.iter().map(|s| json!(s)).collect()),
375 );
376 }
377 self.append_log(LogEntry::Morphism {
378 seq,
379 morphism: "ui.edit_record".into(),
380 inputs: Default::default(),
381 params: Value::Object(params),
382 ops: ops.clone(),
383 schema_hash: None,
384 })?;
385 }
386 let mut store = self
387 .store
388 .lock()
389 .map_err(|_| "store mutex envenenado".to_string())?;
390 store
391 .apply(&ops)
392 .map_err(|e| format!("apply edit ops: {e}"))?;
393 drop(store);
394 let post_status = self.tick_compact();
395 Ok(WriteOutcome {
396 id: Some(id),
397 changed,
398 post_status,
399 })
400 }
401
402 fn delete(&mut self, entity: &str, id: Uuid) -> Result<WriteOutcome, String> {
403 let ops = vec![FieldOp::Delete {
404 entity: entity.to_string(),
405 id,
406 }];
407 if self.event_log.is_some() {
408 let seq = {
409 let log_arc = self.event_log.as_ref().expect("checked").clone();
410 let log = log_arc
411 .lock()
412 .map_err(|_| "log mutex envenenado".to_string())?;
413 log.next_seq()
414 };
415 self.append_log(LogEntry::Morphism {
416 seq,
417 morphism: "ui.delete_record".into(),
418 inputs: Default::default(),
419 params: json!({ "entity": entity, "id": id.to_string() }),
420 ops: ops.clone(),
421 schema_hash: None,
422 })?;
423 }
424 let mut store = self
425 .store
426 .lock()
427 .map_err(|_| "store mutex envenenado".to_string())?;
428 store
429 .apply(&ops)
430 .map_err(|e| format!("apply Delete: {e}"))?;
431 drop(store);
432 let post_status = self.tick_compact();
433 Ok(WriteOutcome {
434 id: Some(id),
435 changed: 1,
436 post_status,
437 })
438 }
439
440 fn morphism(
441 &mut self,
442 module_id: &str,
443 name: &str,
444 inputs: BTreeMap<String, Uuid>,
445 params: Value,
446 ) -> Result<WriteOutcome, String> {
447 let executor = self
448 .executors
449 .get(module_id)
450 .ok_or_else(|| {
451 format!(
452 "módulo '{module_id}' no tiene executor nakui (falta nakui_module_dir o falló la carga)"
453 )
454 })?
455 .clone();
456 let log_arc = self
457 .event_log
458 .as_ref()
459 .ok_or_else(|| "morphism requiere event log activo".to_string())?
460 .clone();
461
462 let inputs_owned: Vec<(String, Uuid)> = inputs.into_iter().collect();
463 let inputs_ref: Vec<(&str, Uuid)> = inputs_owned
464 .iter()
465 .map(|(r, id)| (r.as_str(), *id))
466 .collect();
467
468 let mut log = log_arc
469 .lock()
470 .map_err(|_| "log mutex envenenado".to_string())?;
471 let mut store = self
472 .store
473 .lock()
474 .map_err(|_| "store mutex envenenado".to_string())?;
475
476 let ops = execute_and_log_with_recovery(
477 &executor,
478 &mut *store,
479 &mut *log,
480 name,
481 &inputs_ref,
482 params,
483 )
484 .map_err(|e| format!("{e}"))?;
485 drop(store);
486 drop(log);
487 let post_status = self.tick_compact();
488 Ok(WriteOutcome {
489 id: None,
490 changed: ops.len(),
491 post_status,
492 })
493 }
494}
495
496#[cfg(test)]
497mod tests {
498 //! Tests del impl `NakuiBackend` contra el contrato del trait.
499 //! Exercises seed/load/list/update/delete sin GPUI ni morphism.
500 //! El path de morphism está cubierto por
501 //! `morphism_pipeline_executes_real_sales_vender` en main.rs.
502
503 use super::*;
504 use serde_json::json;
505
506 fn open_in_tempdir() -> (NakuiBackend, tempfile::TempDir) {
507 let dir = tempfile::tempdir().unwrap();
508 let log_path = dir.path().join("log.jsonl");
509 let (backend, _status) = NakuiBackend::open(log_path, 0, BTreeMap::new());
510 (backend, dir)
511 }
512
513 fn map_of(items: &[(&str, Value)]) -> serde_json::Map<String, Value> {
514 items
515 .iter()
516 .map(|(k, v)| (k.to_string(), v.clone()))
517 .collect()
518 }
519
520 #[test]
521 fn seed_then_load_round_trip_via_trait() {
522 let (mut b, _dir) = open_in_tempdir();
523 let out = b
524 .seed("Customer", map_of(&[("name", json!("Acme"))]))
525 .unwrap();
526 let id = out.id.unwrap();
527 assert_eq!(out.changed, 1);
528 let rec = b.load_record("Customer", id).unwrap();
529 assert_eq!(rec.get("name"), Some(&json!("Acme")));
530 }
531
532 #[test]
533 fn update_set_then_clear_via_trait() {
534 let (mut b, _dir) = open_in_tempdir();
535 let id = b
536 .seed("X", map_of(&[("a", json!(1)), ("b", json!(2))]))
537 .unwrap()
538 .id
539 .unwrap();
540
541 let out = b
542 .update("X", id, map_of(&[("a", json!(10))]), vec!["b".into()])
543 .unwrap();
544 assert_eq!(out.changed, 2, "1 set + 1 clear = 2 cambios");
545
546 let rec = b.load_record("X", id).unwrap();
547 assert_eq!(rec.get("a"), Some(&json!(10)));
548 assert!(rec.get("b").is_none());
549 }
550
551 #[test]
552 fn update_no_op_returns_no_change() {
553 let (mut b, _dir) = open_in_tempdir();
554 let id = b.seed("X", map_of(&[("a", json!(1))])).unwrap().id.unwrap();
555 let out = b.update("X", id, serde_json::Map::new(), vec![]).unwrap();
556 assert_eq!(out, WriteOutcome::no_change(id));
557 }
558
559 #[test]
560 fn delete_via_trait_then_load_returns_none() {
561 let (mut b, _dir) = open_in_tempdir();
562 let id = b.seed("X", map_of(&[("a", json!(1))])).unwrap().id.unwrap();
563 b.delete("X", id).unwrap();
564 assert!(b.load_record("X", id).is_none());
565 }
566
567 #[test]
568 fn list_records_returns_seeded_in_id_order() {
569 let (mut b, _dir) = open_in_tempdir();
570 let _ = b.seed("X", map_of(&[("k", json!(1))])).unwrap();
571 let _ = b.seed("X", map_of(&[("k", json!(2))])).unwrap();
572 let _ = b.seed("Y", map_of(&[("k", json!(3))])).unwrap();
573 assert_eq!(b.list_records("X").len(), 2);
574 assert_eq!(b.list_records("Y").len(), 1);
575 assert!(b.list_records("Z").is_empty());
576 }
577
578 #[test]
579 fn morphism_without_executor_errors_clearly() {
580 let (mut b, _dir) = open_in_tempdir();
581 let err = b
582 .morphism("missing", "vender", BTreeMap::new(), json!({}))
583 .unwrap_err();
584 assert!(
585 err.contains("missing"),
586 "msg debe mencionar el módulo: {err}"
587 );
588 assert!(err.contains("nakui_module_dir") || err.contains("executor"));
589 }
590
591 #[test]
592 fn tick_compact_writes_snapshot_after_threshold() {
593 // threshold=3: tras 3 writes debería haber compactado.
594 let dir = tempfile::tempdir().unwrap();
595 let log_path = dir.path().join("log.jsonl");
596 let snap_path = snapshot_path_for(&log_path);
597 let (mut b, _) = NakuiBackend::open(log_path, 3, BTreeMap::new());
598
599 for _ in 0..3 {
600 let _ = b.seed("X", map_of(&[("k", json!(1))])).unwrap();
601 }
602 // El último seed debería traer un post_status del compact.
603 // (En la 3ra llamada el contador llega a 3 y dispara.)
604 // Verificamos que el snapshot file exists.
605 assert!(snap_path.exists(), "snap debería haberse escrito");
606 }
607}