sergio/takana

1//! El diario de mutaciones. Ver `docs/05-journal.md`.
2//!
3//! Formato: archivo único de **JSON-líneas append-only** (`mutations.jsonl`). Cada línea es
4//! un `MutationEvent` autónomo; `cat` / `grep` / `jq` los procesan sin parseo extra.
5//!
6//! Esta crate es **lógica pura**: nada de fanotify ni de mounts. El watcher real vive en
7//! `hammerd` y empuja eventos aquí vía `Journal::record`. El hook desde
8//! `takana_overlay::commit` también empuja por aquí.
9//!
10//! Diseño rationale (SDD 05):
11//! - El diario no bloquea acciones; sólo observa. Pero también acepta eventos "trazables"
12//!   (de un `hydrate` o `commit`) que llegan con `actor.source = HammerHydrate { artifact }`.
13//! - Las "mutaciones opacas" (pisada manual sin rastro de artefacto) llegan con
14//!   `actor.source = External { pid, uid }`. Eso permite que `takana export` distinga
15//!   replays reproducibles de cambios que sólo se pueden compartir como warnings.
16
17use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
18use std::path::PathBuf;
19
20use serde::{Deserialize, Serialize};
21
22/// Operación sobre el FHS. Ver SDD 05 §2 — la lista evolucionará con los eventos que el
23/// watcher fanotify aprenda a distinguir.
24#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
25#[serde(rename_all = "kebab-case")]
26pub enum MutationOp {
27    /// Archivo nuevo donde antes no había nada.
28    Create,
29    /// Pisado: el archivo existía y fue sustituido (mismo path, distinto contenido o inode).
30    Replace,
31    /// Edición en sitio: el archivo cambió sin que cambiara el path (típico de configs en /etc).
32    Edit,
33    /// Borrado.
34    Delete,
35}
36
37impl MutationOp {
38    pub fn as_str(&self) -> &'static str {
39        match self {
40            MutationOp::Create => "create",
41            MutationOp::Replace => "replace",
42            MutationOp::Edit => "edit",
43            MutationOp::Delete => "delete",
44        }
45    }
46}
47
48/// Quién originó la mutación. Distinguir trazable vs opaca es lo que más tarde permite
49/// que `takana export` produzca un `.swm` reproducible (artefactos) más una lista de
50/// warnings (pisadas manuales).
51#[derive(Debug, Clone, Serialize, Deserialize)]
52#[serde(tag = "kind", rename_all = "kebab-case")]
53pub enum Source {
54    /// Vino de un `takana hydrate <artifact>` — la mutación es replay-able a partir del hash.
55    HammerHydrate { artifact: String },
56    /// Vino de un `takana commit <overlay>` — origen indirecto: el `actor` fue el overlay,
57    /// y `artifact` es opcional (puede no haber, si la promoción fue de un edit manual).
58    HammerCommit {
59        overlay: String,
60        #[serde(default, skip_serializing_if = "Option::is_none")]
61        artifact: Option<String>,
62    },
63    /// Pisada manual fuera del flujo de takana. La grabamos sin juzgar — el SDD insiste en
64    /// que es información, no error.
65    External,
66}
67
68#[derive(Debug, Clone, Serialize, Deserialize)]
69pub struct Actor {
70    pub source: Source,
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub pid: Option<u32>,
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    pub uid: Option<u32>,
75}
76
77/// Un evento atómico — una línea del diario.
78#[derive(Debug, Clone, Serialize, Deserialize)]
79pub struct MutationEvent {
80    /// RFC 3339 UTC. El watcher fija el timestamp en el momento de la observación.
81    pub ts: String,
82    pub op: MutationOp,
83    pub path: PathBuf,
84    pub by: Actor,
85    /// Opcional: hash blake3 del contenido del archivo tras la mutación. Permite a `export`
86    /// detectar mutaciones idempotentes (mismo contenido tras N replaces → un sólo evento).
87    #[serde(default, skip_serializing_if = "Option::is_none")]
88    pub content_hash: Option<String>,
89    /// Nota libre (humana). Útil para que la IA explique su intención.
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub note: Option<String>,
92}
93
94/// Handle del diario: una ruta a `mutations.jsonl`. Las operaciones son thread-safe en el
95/// sentido de que cada `record` abre + append + cierra; suficiente para el throughput
96/// esperado (cambios humanos, no log-de-tracing).
97#[derive(Debug, Clone)]
98pub struct Journal {
99    pub dir: PathBuf,
100}
101
102#[derive(Debug, thiserror::Error)]
103pub enum Error {
104    #[error("io: {0}")]
105    Io(#[from] std::io::Error),
106    #[error("json: {0}")]
107    Json(#[from] serde_json::Error),
108    #[error("journal: {0}")]
109    Other(String),
110}
111
112pub type Result<T> = std::result::Result<T, Error>;
113
114impl Journal {
115    /// Abre (creando el directorio si hace falta) un diario bajo `dir`. El archivo
116    /// `mutations.jsonl` se crea en el primer `record`.
117    pub fn open(dir: impl Into<PathBuf>) -> Result<Self> {
118        let dir = dir.into();
119        std::fs::create_dir_all(&dir)?;
120        Ok(Self { dir })
121    }
122
123    pub fn file_path(&self) -> PathBuf {
124        self.dir.join("mutations.jsonl")
125    }
126
127    /// Último evento registrado para `path` (o `None` si nunca apareció). Naïve: relee el
128    /// diario; suficiente para la cadencia humana de Fase 3 (cuando crezca, un índice por
129    /// path lo acelera).
130    pub fn last_for_path(&self, path: &std::path::Path) -> Result<Option<MutationEvent>> {
131        let all = self.read_all()?;
132        Ok(all.into_iter().rev().find(|e| e.path == path))
133    }
134
135    /// Como [`record`], pero **de-dup idempotente**: si la mutación deja el archivo en el
136    /// mismo estado que el último evento del diario para ese path, no la registra. Devuelve
137    /// `true` si escribió un evento, `false` si lo consideró redundante.
138    ///
139    /// Dos casos de redundancia (ver SDD 05 — el diario es información, no ruido):
140    /// 1. **Contenido idéntico:** el `content_hash` nuevo coincide con el último de ese path
141    ///    (p. ej. un editor que reescribe el archivo sin cambiarlo, o un replay del mismo
142    ///    artefacto).
143    /// 2. **Borrado repetido:** un `Delete` sobre un path cuyo último evento ya era `Delete`.
144    pub fn record_dedup(&self, ev: &MutationEvent) -> Result<bool> {
145        if let Some(last) = self.last_for_path(&ev.path)? {
146            let idempotent_content = matches!(
147                (&ev.content_hash, &last.content_hash),
148                (Some(a), Some(b)) if a == b
149            );
150            let repeated_delete =
151                ev.op == MutationOp::Delete && last.op == MutationOp::Delete;
152            if idempotent_content || repeated_delete {
153                tracing::debug!(path = %ev.path.display(), "journal: evento idempotente, omitido");
154                return Ok(false);
155            }
156        }
157        self.record(ev)?;
158        Ok(true)
159    }
160
161    /// Append-only. Una línea JSON + `\n`. Maneja la creación del archivo si no existe.
162    pub fn record(&self, ev: &MutationEvent) -> Result<()> {
163        let line = serde_json::to_string(ev)?;
164        let mut f = std::fs::OpenOptions::new()
165            .create(true)
166            .append(true)
167            .open(self.file_path())?;
168        f.write_all(line.as_bytes())?;
169        f.write_all(b"\n")?;
170        f.sync_data()?;
171        Ok(())
172    }
173
174    /// Lee todos los eventos en orden de aparición. Líneas vacías o malformadas se loguean
175    /// y se saltan — el diario es append-only pero queremos resilencia a truncados de
176    /// emergencia (kernel-panic en medio de una escritura).
177    pub fn read_all(&self) -> Result<Vec<MutationEvent>> {
178        let path = self.file_path();
179        if !path.is_file() {
180            return Ok(Vec::new());
181        }
182        let f = std::fs::File::open(&path)?;
183        let rdr = BufReader::new(f);
184        let mut out = Vec::new();
185        for (i, line) in rdr.lines().enumerate() {
186            let line = line?;
187            if line.trim().is_empty() {
188                continue;
189            }
190            match serde_json::from_str::<MutationEvent>(&line) {
191                Ok(ev) => out.push(ev),
192                Err(e) => {
193                    tracing::warn!(line = i + 1, error = %e, "journal: línea malformada, saltando");
194                }
195            }
196        }
197        Ok(out)
198    }
199
200    /// Lee los últimos `n` eventos sin cargar todo el archivo. Implementación naïve por
201    /// ahora (delega a `read_all` y trunca); cuando el archivo crezca añadiremos un índice
202    /// por bloques.
203    pub fn tail(&self, n: usize) -> Result<Vec<MutationEvent>> {
204        let mut all = self.read_all()?;
205        if all.len() > n {
206            all.drain(..(all.len() - n));
207        }
208        Ok(all)
209    }
210
211    /// Devuelve el offset en bytes del fin de archivo actual. Útil para implementar un
212    /// "follow" en CLI: tras un read inicial, hacemos polling desde este offset.
213    pub fn end_offset(&self) -> Result<u64> {
214        let path = self.file_path();
215        if !path.is_file() {
216            return Ok(0);
217        }
218        let m = std::fs::metadata(&path)?;
219        Ok(m.len())
220    }
221
222    /// Lee eventos desde un offset en bytes hasta el final actual. Devuelve los eventos y el
223    /// nuevo offset (para usar en la siguiente iteración del follow).
224    pub fn read_from_offset(&self, offset: u64) -> Result<(Vec<MutationEvent>, u64)> {
225        let path = self.file_path();
226        if !path.is_file() {
227            return Ok((Vec::new(), 0));
228        }
229        let mut f = std::fs::File::open(&path)?;
230        f.seek(SeekFrom::Start(offset))?;
231        let mut buf = String::new();
232        f.read_to_string(&mut buf)?;
233        let new_offset = offset + buf.len() as u64;
234        let mut out = Vec::new();
235        for line in buf.lines() {
236            if line.trim().is_empty() {
237                continue;
238            }
239            match serde_json::from_str::<MutationEvent>(line) {
240                Ok(ev) => out.push(ev),
241                Err(e) => tracing::warn!(error = %e, "journal: línea malformada en tail"),
242            }
243        }
244        Ok((out, new_offset))
245    }
246}
247
248/// Hash BLAKE3 del contenido de un archivo, con prefijo `b3:`. Es un blake3 **plano** del
249/// contenido (lo que daría `b3sum`), distinto del `ArtifactHash::of_inputs` de takana-core
250/// (que prefija longitudes para identidad de artefactos). Aquí queremos "¿el archivo cambió?".
251pub fn content_hash_of(bytes: &[u8]) -> String {
252    format!("b3:{}", blake3::hash(bytes).to_hex())
253}
254
255/// Hashea el contenido de `path` tras una mutación. Devuelve `None` si el archivo ya no existe
256/// (p. ej. fue borrado, o es lo que el watcher ve como `... (deleted)`); propaga otros errores
257/// de IO.
258pub fn hash_file(path: &std::path::Path) -> Result<Option<String>> {
259    match std::fs::read(path) {
260        Ok(bytes) => Ok(Some(content_hash_of(&bytes))),
261        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
262        Err(e) => Err(e.into()),
263    }
264}
265
266/// Construye un timestamp RFC 3339 UTC del instante actual usando std-only (sin chrono).
267pub fn now_rfc3339() -> String {
268    use std::time::SystemTime;
269    let d = SystemTime::now()
270        .duration_since(SystemTime::UNIX_EPOCH)
271        .unwrap_or_default();
272    let secs = d.as_secs() as i64;
273    let nanos = d.subsec_nanos();
274    let (year, month, day, hour, minute, second) = unix_to_ymdhms(secs);
275    format!(
276        "{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}.{:09}Z",
277        nanos
278    )
279}
280
281/// Conversión epoch → (year, month, day, hour, minute, second) UTC. Algoritmo de Howard
282/// Hinnant (`civil_from_days`), válido para todo rango representable por i64.
283fn unix_to_ymdhms(secs: i64) -> (i32, u32, u32, u32, u32, u32) {
284    let day_secs = 86_400i64;
285    let mut days = secs.div_euclid(day_secs);
286    let time_of_day = secs.rem_euclid(day_secs) as u32;
287    let hour = time_of_day / 3600;
288    let minute = (time_of_day / 60) % 60;
289    let second = time_of_day % 60;
290
291    days += 719_468; // shift al epoch del algoritmo (0000-03-01)
292    let era = if days >= 0 { days } else { days - 146_096 } / 146_097;
293    let doe = (days - era * 146_097) as u64; // [0, 146_096]
294    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; // [0, 399]
295    let y = yoe as i64 + era * 400;
296    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); // [0, 365]
297    let mp = (5 * doy + 2) / 153; // [0, 11]
298    let day = (doy - (153 * mp + 2) / 5 + 1) as u32; // [1, 31]
299    let month = if mp < 10 { mp + 3 } else { mp - 9 } as u32; // [1, 12]
300    let year = (y + if month <= 2 { 1 } else { 0 }) as i32;
301
302    (year, month, day, hour, minute, second)
303}
304
305#[cfg(test)]
306mod tests {
307    use super::*;
308
309    fn ev(op: MutationOp, path: &str) -> MutationEvent {
310        MutationEvent {
311            ts: now_rfc3339(),
312            op,
313            path: PathBuf::from(path),
314            by: Actor {
315                source: Source::External,
316                pid: Some(1234),
317                uid: Some(1000),
318            },
319            content_hash: None,
320            note: None,
321        }
322    }
323
324    #[test]
325    fn record_and_read_roundtrip() {
326        let d = tempfile::tempdir().unwrap();
327        let j = Journal::open(d.path()).unwrap();
328        j.record(&ev(MutationOp::Replace, "/bin/grep")).unwrap();
329        j.record(&ev(MutationOp::Edit, "/etc/hosts")).unwrap();
330        let read = j.read_all().unwrap();
331        assert_eq!(read.len(), 2);
332        assert_eq!(read[0].op, MutationOp::Replace);
333        assert_eq!(read[0].path, PathBuf::from("/bin/grep"));
334        assert_eq!(read[1].op, MutationOp::Edit);
335    }
336
337    #[test]
338    fn empty_journal_reads_empty() {
339        let d = tempfile::tempdir().unwrap();
340        let j = Journal::open(d.path()).unwrap();
341        assert!(j.read_all().unwrap().is_empty());
342        assert_eq!(j.end_offset().unwrap(), 0);
343    }
344
345    #[test]
346    fn tail_returns_last_n() {
347        let d = tempfile::tempdir().unwrap();
348        let j = Journal::open(d.path()).unwrap();
349        for i in 0..5 {
350            j.record(&ev(MutationOp::Create, &format!("/etc/{i}"))).unwrap();
351        }
352        let last = j.tail(2).unwrap();
353        assert_eq!(last.len(), 2);
354        assert_eq!(last[0].path, PathBuf::from("/etc/3"));
355        assert_eq!(last[1].path, PathBuf::from("/etc/4"));
356    }
357
358    #[test]
359    fn follow_via_offset_yields_new_events_only() {
360        let d = tempfile::tempdir().unwrap();
361        let j = Journal::open(d.path()).unwrap();
362        j.record(&ev(MutationOp::Create, "/etc/a")).unwrap();
363        let cursor = j.end_offset().unwrap();
364        j.record(&ev(MutationOp::Create, "/etc/b")).unwrap();
365        j.record(&ev(MutationOp::Create, "/etc/c")).unwrap();
366        let (news, new_cursor) = j.read_from_offset(cursor).unwrap();
367        assert_eq!(news.len(), 2);
368        assert_eq!(news[0].path, PathBuf::from("/etc/b"));
369        assert_eq!(news[1].path, PathBuf::from("/etc/c"));
370        assert_eq!(new_cursor, j.end_offset().unwrap());
371        // Una segunda llamada desde el nuevo cursor devuelve vacío.
372        let (more, _) = j.read_from_offset(new_cursor).unwrap();
373        assert!(more.is_empty());
374    }
375
376    #[test]
377    fn malformed_line_is_skipped_not_fatal() {
378        let d = tempfile::tempdir().unwrap();
379        let j = Journal::open(d.path()).unwrap();
380        j.record(&ev(MutationOp::Create, "/etc/a")).unwrap();
381        // Plantamos basura en medio (kernel panic mid-write).
382        let mut f = std::fs::OpenOptions::new().append(true).open(j.file_path()).unwrap();
383        f.write_all(b"{esto no es JSON valido\n").unwrap();
384        f.write_all(b"\n").unwrap(); // línea vacía
385        drop(f);
386        j.record(&ev(MutationOp::Create, "/etc/b")).unwrap();
387        let read = j.read_all().unwrap();
388        assert_eq!(read.len(), 2, "líneas válidas se preservan");
389    }
390
391    fn ev_hash(op: MutationOp, path: &str, hash: Option<&str>) -> MutationEvent {
392        let mut e = ev(op, path);
393        e.content_hash = hash.map(|s| s.to_string());
394        e
395    }
396
397    #[test]
398    fn content_hash_is_plain_blake3_with_prefix() {
399        let h = content_hash_of(b"hola hammer\n");
400        assert!(h.starts_with("b3:"));
401        // Determinista y sensible al contenido.
402        assert_eq!(h, content_hash_of(b"hola hammer\n"));
403        assert_ne!(h, content_hash_of(b"otra cosa"));
404    }
405
406    #[test]
407    fn hash_file_some_and_none() {
408        let d = tempfile::tempdir().unwrap();
409        let f = d.path().join("x");
410        std::fs::write(&f, b"abc").unwrap();
411        assert_eq!(hash_file(&f).unwrap(), Some(content_hash_of(b"abc")));
412        assert_eq!(hash_file(&d.path().join("noexiste")).unwrap(), None);
413    }
414
415    #[test]
416    fn dedup_skips_identical_content() {
417        let d = tempfile::tempdir().unwrap();
418        let j = Journal::open(d.path()).unwrap();
419        let h = content_hash_of(b"v1");
420        // Primer evento: se registra.
421        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/bin/x", Some(&h))).unwrap());
422        // Mismo contenido otra vez: idempotente, se omite.
423        assert!(!j.record_dedup(&ev_hash(MutationOp::Replace, "/bin/x", Some(&h))).unwrap());
424        // Contenido distinto: se registra.
425        let h2 = content_hash_of(b"v2");
426        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/bin/x", Some(&h2))).unwrap());
427        // Vuelta a v1 (no es el ÚLTIMO estado): se registra (no colapsamos histórico).
428        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/bin/x", Some(&h))).unwrap());
429
430        let all = j.read_all().unwrap();
431        assert_eq!(all.len(), 3, "v1, v2, v1 — sólo se omitió el v1→v1 consecutivo");
432    }
433
434    #[test]
435    fn dedup_is_per_path() {
436        let d = tempfile::tempdir().unwrap();
437        let j = Journal::open(d.path()).unwrap();
438        let h = content_hash_of(b"same");
439        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/bin/a", Some(&h))).unwrap());
440        // Mismo hash pero OTRO path: no es redundante.
441        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/bin/b", Some(&h))).unwrap());
442        assert_eq!(j.read_all().unwrap().len(), 2);
443    }
444
445    #[test]
446    fn dedup_collapses_repeated_delete() {
447        let d = tempfile::tempdir().unwrap();
448        let j = Journal::open(d.path()).unwrap();
449        assert!(j.record_dedup(&ev_hash(MutationOp::Delete, "/etc/x", None)).unwrap());
450        assert!(!j.record_dedup(&ev_hash(MutationOp::Delete, "/etc/x", None)).unwrap());
451        assert_eq!(j.read_all().unwrap().len(), 1);
452    }
453
454    #[test]
455    fn dedup_does_not_collapse_hashless_replaces() {
456        // Dos Replace sin content_hash NO se consideran idempotentes (no sabemos el contenido).
457        let d = tempfile::tempdir().unwrap();
458        let j = Journal::open(d.path()).unwrap();
459        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/etc/x", None)).unwrap());
460        assert!(j.record_dedup(&ev_hash(MutationOp::Replace, "/etc/x", None)).unwrap());
461        assert_eq!(j.read_all().unwrap().len(), 2);
462    }
463
464    #[test]
465    fn last_for_path_finds_latest() {
466        let d = tempfile::tempdir().unwrap();
467        let j = Journal::open(d.path()).unwrap();
468        j.record(&ev(MutationOp::Create, "/etc/a")).unwrap();
469        j.record(&ev(MutationOp::Replace, "/etc/a")).unwrap();
470        j.record(&ev(MutationOp::Edit, "/etc/b")).unwrap();
471        assert_eq!(j.last_for_path(&PathBuf::from("/etc/a")).unwrap().unwrap().op, MutationOp::Replace);
472        assert_eq!(j.last_for_path(&PathBuf::from("/etc/b")).unwrap().unwrap().op, MutationOp::Edit);
473        assert!(j.last_for_path(&PathBuf::from("/etc/z")).unwrap().is_none());
474    }
475
476    #[test]
477    fn source_serializes_with_tag() {
478        let e = ev(MutationOp::Replace, "/bin/grep");
479        let j = serde_json::to_string(&e).unwrap();
480        assert!(j.contains(r#""kind":"external""#), "got {j}");
481
482        let mut e2 = e.clone();
483        e2.by.source = Source::HammerHydrate {
484            artifact: "b3:abc".into(),
485        };
486        let j2 = serde_json::to_string(&e2).unwrap();
487        assert!(j2.contains(r#""kind":"hammer-hydrate""#), "got {j2}");
488        assert!(j2.contains(r#""artifact":"b3:abc""#), "got {j2}");
489    }
490
491    #[test]
492    fn rfc3339_format_basics() {
493        let ts = now_rfc3339();
494        // Verifica forma `YYYY-MM-DDThh:mm:ss.nnnnnnnnnZ` minimal.
495        assert!(ts.len() >= 29);
496        assert_eq!(&ts[4..5], "-");
497        assert_eq!(&ts[7..8], "-");
498        assert_eq!(&ts[10..11], "T");
499        assert_eq!(&ts[13..14], ":");
500        assert_eq!(&ts[16..17], ":");
501        assert!(ts.ends_with('Z'));
502        // Año razonable (2026 actualmente; este test sólo verifica que no salimos del rango
503        // 1970..3000).
504        let year: i32 = ts[..4].parse().unwrap();
505        assert!((1970..3000).contains(&year), "year = {year}");
506    }
507}