kimun_notes/server_client/
observer.rs

1//! The [`IndexObserver`] the client registers on the vault, and the dirty-set it
2//! feeds. The dirty-set is deliberately best-effort in-memory: a lost entry
3//! costs a reconciliation pass, not a lost update.
4
5use std::collections::HashMap;
6use std::sync::{Arc, Mutex};
7
8use kimun_core::{IndexObserver, NoteChange, nfs::VaultPath};
9
10/// The pending change for a note. The latest event wins, so an Upsert followed
11/// by a Delete on the same path collapses to Delete (and vice-versa). `Upsert`
12/// carries the note's content hash from the (post-commit) change event, which
13/// matches the chunks the index holds for it at drain time.
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum DirtyOp {
16    Upsert(u64),
17    Delete,
18}
19
20/// Set of notes changed since the last drain, keyed by path (latest op per path).
21#[derive(Debug, Default)]
22pub struct DirtySet {
23    inner: Mutex<HashMap<VaultPath, DirtyOp>>,
24}
25
26impl DirtySet {
27    /// Records a change, overwriting any earlier pending op for the same note.
28    pub fn record(&self, change: &NoteChange) {
29        let mut map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
30        match change {
31            NoteChange::Upsert { path, hash } => map.insert(path.clone(), DirtyOp::Upsert(*hash)),
32            NoteChange::Delete { path } => map.insert(path.clone(), DirtyOp::Delete),
33        };
34    }
35
36    /// Takes and clears all pending ops for flushing.
37    pub fn drain(&self) -> Vec<(VaultPath, DirtyOp)> {
38        let mut map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
39        std::mem::take(&mut *map).into_iter().collect()
40    }
41
42    /// Puts back ops that failed to flush, without clobbering a newer op that
43    /// was recorded for the same note while the flush was in flight.
44    pub fn requeue(&self, items: impl IntoIterator<Item = (VaultPath, DirtyOp)>) {
45        let mut map = self.inner.lock().unwrap_or_else(|e| e.into_inner());
46        for (path, op) in items {
47            map.entry(path).or_insert(op);
48        }
49    }
50
51    pub fn len(&self) -> usize {
52        self.inner.lock().unwrap_or_else(|e| e.into_inner()).len()
53    }
54
55    pub fn is_empty(&self) -> bool {
56        self.len() == 0
57    }
58}
59
60/// Registered on the vault; folds every note change into the shared [`DirtySet`].
61#[derive(Debug)]
62pub struct RagObserver {
63    dirty: Arc<DirtySet>,
64}
65
66impl RagObserver {
67    pub fn new(dirty: Arc<DirtySet>) -> Self {
68        Self { dirty }
69    }
70}
71
72impl IndexObserver for RagObserver {
73    fn on_change(&self, change: &NoteChange) {
74        self.dirty.record(change);
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use super::*;
81
82    fn upsert(p: &str) -> NoteChange {
83        NoteChange::Upsert {
84            path: VaultPath::new(p),
85            hash: 1,
86        }
87    }
88    fn delete(p: &str) -> NoteChange {
89        NoteChange::Delete {
90            path: VaultPath::new(p),
91        }
92    }
93
94    #[test]
95    fn latest_op_wins_per_path() {
96        let set = DirtySet::default();
97        set.record(&upsert("a.md"));
98        set.record(&delete("a.md")); // delete supersedes the upsert
99        let drained = set.drain();
100        assert_eq!(drained, vec![(VaultPath::new("a.md"), DirtyOp::Delete)]);
101
102        set.record(&delete("b.md"));
103        set.record(&upsert("b.md")); // recreate supersedes the delete
104        assert_eq!(
105            set.drain(),
106            vec![(VaultPath::new("b.md"), DirtyOp::Upsert(1))]
107        );
108    }
109
110    #[test]
111    fn drain_clears() {
112        let set = DirtySet::default();
113        set.record(&upsert("a.md"));
114        assert_eq!(set.len(), 1);
115        let _ = set.drain();
116        assert!(set.is_empty());
117    }
118
119    #[test]
120    fn requeue_does_not_clobber_newer_op() {
121        let set = DirtySet::default();
122        // Flush of "a.md" as Upsert failed; meanwhile a Delete arrived.
123        set.record(&delete("a.md"));
124        set.requeue([(VaultPath::new("a.md"), DirtyOp::Upsert(1))]);
125        // The newer Delete must survive.
126        assert_eq!(set.drain(), vec![(VaultPath::new("a.md"), DirtyOp::Delete)]);
127    }
128
129    #[test]
130    fn observer_records_into_shared_set() {
131        let dirty = Arc::new(DirtySet::default());
132        let observer = RagObserver::new(dirty.clone());
133        observer.on_change(&upsert("n.md"));
134        assert_eq!(dirty.len(), 1);
135    }
136}
kimun_notes/server_client/observer.rs

kimun_notes/server_client/
observer.rs
