salamander/view/
indexed.rs

1//! docs/phase-1.5.md §3 — `IndexedView`: the ready-made
2//! queryable view. A primary `BTreeMap` store plus named secondary
3//! indexes, all maintained incrementally as events fan in. The `indexed_as`
4//! reverse map (§5) is the correctness core: it's what lets an update or
5//! delete remove the *old* value's index entries before adding the new
6//! ones, so `by(index, key)` never returns a phantom hit.
7
8use std::any::Any;
9use std::borrow::Borrow;
10use std::collections::{BTreeMap, BTreeSet, HashMap};
11use std::hash::Hash;
12use std::ops::RangeBounds;
13
14use super::{IndexKey, View};
15use crate::event::{Body, Event};
16use crate::projection::Projection;
17
18/// The change a `project` closure derives from one event: upsert or remove
19/// a primary key. Events irrelevant to this view yield `None` instead.
20pub enum Change<K, V> {
21    /// Insert or replace the value at the primary key `K`.
22    Put(K, V),
23    /// Remove the entry at the primary key `K`.
24    Delete(K),
25}
26
27impl<K, V> Change<K, V> {
28    /// Constructs a [`Change::Put`].
29    pub fn put(key: K, value: V) -> Self {
30        Change::Put(key, value)
31    }
32    /// Constructs a [`Change::Delete`].
33    pub fn delete(key: K) -> Self {
34        Change::Delete(key)
35    }
36}
37
38// Boxed closures, aliased so the struct fields don't trip `type_complexity`
39// and read as intent rather than punctuation.
40type Projector<K, V, B> = Box<dyn Fn(&Event<B>) -> Option<Change<K, V>>>;
41type Indexer<V> = Box<dyn Fn(&V) -> Vec<IndexKey>>;
42
43/// A live, queryable view over payloads of type `B`, keyed by `K` with
44/// values `V`. `B` is the third type parameter because the `project`
45/// closure matches on `&Event<B>` — the query-layer generalization of
46/// query-layer design §9 OQ-Q4 (the agent `EventBody` is just one `B`).
47pub struct IndexedView<K, V, B> {
48    project: Projector<K, V, B>,
49    indexers: Vec<(String, Indexer<V>)>,
50    primary: BTreeMap<K, V>,
51    indexes: HashMap<String, BTreeMap<IndexKey, BTreeSet<K>>>,
52    /// Reverse map: for each present primary key, the `(index, index_key)`
53    /// pairs it currently contributes. The §5 update/delete correctness core.
54    indexed_as: HashMap<K, Vec<(String, IndexKey)>>,
55    cursor: u64,
56}
57
58/// Builder for [`IndexedView`] — `.project(..)` (required) then any number
59/// of `.index(name, ..)`, then `.build()`.
60/// Builder for an [`IndexedView`]: supply a `project` closure and any
61/// number of named secondary indexes, then `build`.
62pub struct IndexedViewBuilder<K, V, B> {
63    project: Option<Projector<K, V, B>>,
64    indexers: Vec<(String, Indexer<V>)>,
65}
66
67impl<K, V, B> IndexedView<K, V, B> {
68    /// Starts building a view; call `project` (required) and `index`
69    /// (optional) on the returned builder.
70    pub fn builder() -> IndexedViewBuilder<K, V, B> {
71        IndexedViewBuilder {
72            project: None,
73            indexers: Vec::new(),
74        }
75    }
76}
77
78impl<K, V, B> IndexedViewBuilder<K, V, B> {
79    /// The projection function: map one event to an optional primary-key
80    /// change. This is where the caller's payload vocabulary is decoded.
81    pub fn project(mut self, f: impl Fn(&Event<B>) -> Option<Change<K, V>> + 'static) -> Self {
82        self.project = Some(Box::new(f));
83        self
84    }
85
86    /// Add a named secondary index: map a value to zero or more index keys.
87    /// A value may appear under several keys (multi-valued index) and
88    /// several values may share a key.
89    pub fn index(mut self, name: &str, f: impl Fn(&V) -> Vec<IndexKey> + 'static) -> Self {
90        self.indexers.push((name.to_string(), Box::new(f)));
91        self
92    }
93
94    /// Finishes the view. Panics if no `project` closure was supplied.
95    pub fn build(self) -> IndexedView<K, V, B> {
96        IndexedView {
97            project: self
98                .project
99                .expect("IndexedView::build requires a .project() closure"),
100            indexers: self.indexers,
101            primary: BTreeMap::new(),
102            indexes: HashMap::new(),
103            indexed_as: HashMap::new(),
104            cursor: 0,
105        }
106    }
107}
108
109// ── incremental maintenance (the fold) ───────────────────────────────────
110
111impl<K, V, B> IndexedView<K, V, B>
112where
113    K: Ord + Hash + Clone,
114{
115    /// Apply one event: run the projector, fold any resulting change into
116    /// the primary store and every index, then advance the cursor. Shared
117    /// by both the `View` and `Projection` `apply` impls so live fan-out
118    /// and on-demand replay produce byte-identical state.
119    fn record(&mut self, event: &Event<B>) {
120        if let Some(change) = (self.project)(event) {
121            match change {
122                Change::Put(k, v) => self.put(k, v),
123                Change::Delete(k) => self.delete(&k),
124            }
125        }
126        self.cursor = event.offset + 1;
127    }
128
129    fn put(&mut self, key: K, value: V) {
130        // Clear the previous value's index entries first (§5): an update
131        // from V1 to V2 must not leave V1's keys behind.
132        if let Some(stale) = self.indexed_as.remove(&key) {
133            self.prune_index_entries(&key, &stale);
134        }
135
136        let mut recorded: Vec<(String, IndexKey)> = Vec::new();
137        for (name, indexer) in &self.indexers {
138            for index_key in indexer(&value) {
139                recorded.push((name.clone(), index_key));
140            }
141        }
142        for (name, index_key) in &recorded {
143            self.indexes
144                .entry(name.clone())
145                .or_default()
146                .entry(index_key.clone())
147                .or_default()
148                .insert(key.clone());
149        }
150
151        self.indexed_as.insert(key.clone(), recorded);
152        self.primary.insert(key, value);
153    }
154
155    fn delete(&mut self, key: &K) {
156        if let Some(stale) = self.indexed_as.remove(key) {
157            self.prune_index_entries(key, &stale);
158        }
159        self.primary.remove(key);
160    }
161
162    /// Remove `key` from each `(index, index_key)` bucket in `entries`,
163    /// dropping buckets that become empty so `by` never scans dead keys.
164    fn prune_index_entries(&mut self, key: &K, entries: &[(String, IndexKey)]) {
165        for (name, index_key) in entries {
166            if let Some(index) = self.indexes.get_mut(name) {
167                if let Some(set) = index.get_mut(index_key) {
168                    set.remove(key);
169                    if set.is_empty() {
170                        index.remove(index_key);
171                    }
172                }
173            }
174        }
175    }
176}
177
178// ── the query surface (reached after downcast) ───────────────────────────
179
180impl<K, V, B> IndexedView<K, V, B>
181where
182    K: Ord,
183{
184    /// Point lookup by primary key.
185    pub fn get<Q>(&self, key: &Q) -> Option<&V>
186    where
187        K: Borrow<Q>,
188        Q: Ord + ?Sized,
189    {
190        self.primary.get(key)
191    }
192
193    /// Ordered scan over a primary-key range (keys are `Ord`, stored in a
194    /// `BTreeMap`).
195    pub fn range<T, R>(&self, range: R) -> impl Iterator<Item = (&K, &V)>
196    where
197        T: Ord + ?Sized,
198        K: Borrow<T>,
199        R: RangeBounds<T>,
200    {
201        self.primary.range(range)
202    }
203
204    /// Secondary-index lookup: every value whose *current* row maps to
205    /// `index_key` under `index`, in primary-key order. Returns empty for
206    /// an unknown index or a key with no live entries.
207    pub fn by(&self, index: &str, index_key: &[u8]) -> Vec<&V> {
208        self.indexes
209            .get(index)
210            .and_then(|idx| idx.get(index_key))
211            .into_iter()
212            .flatten()
213            .filter_map(|k| self.primary.get(k))
214            .collect()
215    }
216
217    /// Number of primary entries in the view.
218    pub fn len(&self) -> usize {
219        self.primary.len()
220    }
221
222    /// Whether the view has no primary entries.
223    pub fn is_empty(&self) -> bool {
224        self.primary.is_empty()
225    }
226}
227
228/// `prefix` is a range special case, and only well-defined for keys with a
229/// notion of "starts with" — so it's offered for the two natural key types
230/// (`String` here, `Vec<u8>` below) rather than all `K: Ord`.
231impl<V, B> IndexedView<String, V, B> {
232    /// Every entry whose key starts with `prefix`, in key order. Relies on
233    /// prefixed keys forming a contiguous block in sorted order.
234    pub fn prefix<'a>(&'a self, prefix: &'a str) -> impl Iterator<Item = (&'a String, &'a V)> + 'a {
235        self.primary
236            .range(prefix.to_string()..)
237            .take_while(move |(k, _)| k.starts_with(prefix))
238    }
239}
240
241impl<V, B> IndexedView<Vec<u8>, V, B> {
242    /// Every entry whose byte key starts with `prefix`, in key order.
243    pub fn prefix<'a>(
244        &'a self,
245        prefix: &'a [u8],
246    ) -> impl Iterator<Item = (&'a Vec<u8>, &'a V)> + 'a {
247        self.primary
248            .range(prefix.to_vec()..)
249            .take_while(move |(k, _)| k.starts_with(prefix))
250    }
251}
252
253// ── trait impls: driven by the registry (View) and by replay (Projection) ─
254
255impl<K, V, B> View<B> for IndexedView<K, V, B>
256where
257    K: Ord + Hash + Clone + 'static,
258    V: 'static,
259    B: 'static,
260{
261    fn apply(&mut self, event: &Event<B>) {
262        self.record(event);
263    }
264    fn cursor(&self) -> u64 {
265        self.cursor
266    }
267    fn as_any(&self) -> &dyn Any {
268        self
269    }
270}
271
272impl<K, V, B> Projection for IndexedView<K, V, B>
273where
274    K: Ord + Hash + Clone + 'static,
275    V: 'static,
276    B: Body,
277{
278    type Body = B;
279    type State = BTreeMap<K, V>;
280
281    fn apply(&mut self, event: &Event<B>) {
282        self.record(event);
283    }
284    fn cursor(&self) -> u64 {
285        self.cursor
286    }
287    fn state(&self) -> &Self::State {
288        &self.primary
289    }
290}
291
292#[cfg(test)]
293mod tests {
294    use super::*;
295
296    // A tiny non-serde payload — the index logic doesn't need `Body`, only
297    // the `View`/`replay` boundary does, so these tests exercise the fold
298    // directly through `record`.
299    #[derive(Clone)]
300    enum Ev {
301        Set(&'static str, u32),
302        Del(&'static str),
303    }
304
305    fn ev(offset: u64, body: Ev) -> Event<Ev> {
306        Event {
307            offset,
308            timestamp_ms: 0,
309            namespace: "ns".to_string(),
310            body,
311        }
312    }
313
314    fn view() -> IndexedView<String, u32, Ev> {
315        IndexedView::builder()
316            .project(|e: &Event<Ev>| match &e.body {
317                Ev::Set(k, v) => Some(Change::put(k.to_string(), *v)),
318                Ev::Del(k) => Some(Change::delete(k.to_string())),
319            })
320            .index("by_val", |v: &u32| vec![v.to_le_bytes().to_vec()])
321            .build()
322    }
323
324    fn k(n: u32) -> Vec<u8> {
325        n.to_le_bytes().to_vec()
326    }
327
328    #[test]
329    fn get_and_range() {
330        let mut v = view();
331        v.record(&ev(0, Ev::Set("a", 1)));
332        v.record(&ev(1, Ev::Set("b", 2)));
333        v.record(&ev(2, Ev::Set("c", 3)));
334
335        assert_eq!(v.get("a"), Some(&1));
336        assert_eq!(v.get("z"), None);
337        assert_eq!(v.len(), 3);
338        assert_eq!(v.cursor(), 3);
339
340        let scan: Vec<_> = v
341            .range("a".to_string().."c".to_string())
342            .map(|(k, val)| (k.clone(), *val))
343            .collect();
344        assert_eq!(scan, vec![("a".to_string(), 1), ("b".to_string(), 2)]);
345    }
346
347    #[test]
348    fn prefix_scan() {
349        let mut v = view();
350        v.record(&ev(0, Ev::Set("tool:a", 1)));
351        v.record(&ev(1, Ev::Set("tool:b", 2)));
352        v.record(&ev(2, Ev::Set("user:c", 3)));
353
354        let hits: Vec<_> = v.prefix("tool:").map(|(k, _)| k.clone()).collect();
355        assert_eq!(hits, vec!["tool:a".to_string(), "tool:b".to_string()]);
356    }
357
358    #[test]
359    fn by_returns_all_keys_sharing_an_index_key() {
360        let mut v = view();
361        v.record(&ev(0, Ev::Set("a", 1)));
362        v.record(&ev(1, Ev::Set("b", 1)));
363        v.record(&ev(2, Ev::Set("c", 2)));
364
365        // a and b both map to value 1, in primary-key order.
366        assert_eq!(v.by("by_val", &k(1)), vec![&1, &1]);
367        assert_eq!(v.by("by_val", &k(2)), vec![&2]);
368        assert!(v.by("by_val", &k(9)).is_empty());
369        assert!(v.by("no_such_index", &k(1)).is_empty());
370    }
371
372    #[test]
373    fn update_moves_the_secondary_index_entry() {
374        let mut v = view();
375        v.record(&ev(0, Ev::Set("a", 1)));
376        assert_eq!(v.by("by_val", &k(1)), vec![&1]);
377
378        // Update a: 1 -> 5. The old index entry must vanish, the new appear.
379        v.record(&ev(1, Ev::Set("a", 5)));
380        assert!(
381            v.by("by_val", &k(1)).is_empty(),
382            "stale index entry left behind"
383        );
384        assert_eq!(v.by("by_val", &k(5)), vec![&5]);
385        assert_eq!(v.get("a"), Some(&5));
386    }
387
388    #[test]
389    fn delete_removes_primary_and_all_index_entries() {
390        let mut v = view();
391        v.record(&ev(0, Ev::Set("a", 1)));
392        v.record(&ev(1, Ev::Set("b", 1)));
393        v.record(&ev(2, Ev::Del("a")));
394
395        assert_eq!(v.get("a"), None);
396        assert_eq!(v.by("by_val", &k(1)), vec![&1]); // only b remains
397
398        v.record(&ev(3, Ev::Del("b")));
399        assert!(v.by("by_val", &k(1)).is_empty());
400
401        // The now-empty index bucket must be pruned, and the reverse map
402        // must be empty — no orphaned bookkeeping (§5).
403        assert!(!v.indexes["by_val"].contains_key(&k(1)));
404        assert!(v.indexed_as.is_empty());
405    }
406
407    #[test]
408    fn churn_leaves_no_orphaned_bookkeeping() {
409        let mut v = view();
410        // put -> update -> update -> delete, twice, interleaved.
411        v.record(&ev(0, Ev::Set("a", 1)));
412        v.record(&ev(1, Ev::Set("a", 2)));
413        v.record(&ev(2, Ev::Set("b", 2)));
414        v.record(&ev(3, Ev::Set("a", 3)));
415        v.record(&ev(4, Ev::Del("a")));
416        v.record(&ev(5, Ev::Del("b")));
417
418        assert!(v.is_empty());
419        assert!(v.indexed_as.is_empty());
420        // Every index bucket should have been pruned as it emptied.
421        assert!(v.indexes.values().all(|idx| idx.is_empty()));
422    }
423}
salamander/view/indexed.rs

salamander/view/
indexed.rs
