Skip to main content

arete_server/
sorted_cache.rs

1//! Sorted view cache for maintaining ordered entity collections.
2//!
3//! This module provides incremental maintenance of sorted entity views,
4//! enabling efficient windowed subscriptions (take/skip) with minimal
5//! recomputation on updates.
6
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::cmp::Ordering;
10use std::collections::{BTreeMap, HashMap};
11
12/// A sortable key that combines the sort value with entity key for stable ordering.
13/// Uses (sort_value, entity_key) tuple to ensure deterministic ordering even when
14/// sort values are equal.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct SortKey {
17    /// The extracted sort value (as comparable bytes)
18    sort_value: SortValue,
19    /// Entity key for tie-breaking
20    entity_key: String,
21    /// Direction to apply to the sort value comparison.
22    order: SortOrder,
23}
24
25impl PartialOrd for SortKey {
26    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
27        Some(self.cmp(other))
28    }
29}
30
31impl Ord for SortKey {
32    fn cmp(&self, other: &Self) -> Ordering {
33        if self.order != other.order {
34            return match (self.order, other.order) {
35                (SortOrder::Asc, SortOrder::Desc) => Ordering::Less,
36                (SortOrder::Desc, SortOrder::Asc) => Ordering::Greater,
37                _ => Ordering::Equal,
38            };
39        }
40
41        let sort_order = self.sort_value.cmp(&other.sort_value);
42        let sort_order = match (&self.sort_value, &other.sort_value, self.order) {
43            (SortValue::Null, _, _) | (_, SortValue::Null, _) | (_, _, SortOrder::Asc) => {
44                sort_order
45            }
46            (_, _, SortOrder::Desc) => sort_order.reverse(),
47        };
48
49        match sort_order {
50            Ordering::Equal => self.entity_key.cmp(&other.entity_key),
51            other => other,
52        }
53    }
54}
55
56/// Comparable sort value extracted from JSON
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub enum SortValue {
59    Null,
60    Bool(bool),
61    Integer(i64),
62    Float(OrderedFloat),
63    String(String),
64}
65
66impl Ord for SortValue {
67    fn cmp(&self, other: &Self) -> Ordering {
68        match (self, other) {
69            (SortValue::Null, SortValue::Null) => Ordering::Equal,
70            (SortValue::Null, _) => Ordering::Less,
71            (_, SortValue::Null) => Ordering::Greater,
72            (SortValue::Bool(a), SortValue::Bool(b)) => a.cmp(b),
73            (SortValue::Integer(a), SortValue::Integer(b)) => a.cmp(b),
74            (SortValue::Float(a), SortValue::Float(b)) => a.cmp(b),
75            (SortValue::String(a), SortValue::String(b)) => {
76                compare_decimal_strings(a, b).unwrap_or_else(|| a.cmp(b))
77            }
78            // Cross-type comparisons: numbers < strings
79            (SortValue::Integer(_), SortValue::String(_)) => Ordering::Less,
80            (SortValue::String(_), SortValue::Integer(_)) => Ordering::Greater,
81            (SortValue::Float(_), SortValue::String(_)) => Ordering::Less,
82            (SortValue::String(_), SortValue::Float(_)) => Ordering::Greater,
83            // Integer vs Float: convert to float
84            (SortValue::Integer(a), SortValue::Float(b)) => OrderedFloat(*a as f64).cmp(b),
85            (SortValue::Float(a), SortValue::Integer(b)) => a.cmp(&OrderedFloat(*b as f64)),
86            // Bool vs others
87            (SortValue::Bool(_), _) => Ordering::Less,
88            (_, SortValue::Bool(_)) => Ordering::Greater,
89        }
90    }
91}
92
93impl PartialOrd for SortValue {
94    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
95        Some(self.cmp(other))
96    }
97}
98
99/// Wrapper for f64 that implements Ord (treats NaN as less than all values)
100#[derive(Debug, Clone, Copy, PartialEq)]
101pub struct OrderedFloat(pub f64);
102
103impl Eq for OrderedFloat {}
104
105impl Ord for OrderedFloat {
106    fn cmp(&self, other: &Self) -> Ordering {
107        self.0.partial_cmp(&other.0).unwrap_or_else(|| {
108            if self.0.is_nan() && other.0.is_nan() {
109                Ordering::Equal
110            } else if self.0.is_nan() {
111                Ordering::Less
112            } else {
113                Ordering::Greater
114            }
115        })
116    }
117}
118
119impl PartialOrd for OrderedFloat {
120    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
121        Some(self.cmp(other))
122    }
123}
124
125/// Sort order for the cache
126#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
127#[serde(rename_all = "lowercase")]
128pub enum SortOrder {
129    Asc,
130    Desc,
131}
132
133impl From<crate::materialized_view::SortOrder> for SortOrder {
134    fn from(order: crate::materialized_view::SortOrder) -> Self {
135        match order {
136            crate::materialized_view::SortOrder::Asc => SortOrder::Asc,
137            crate::materialized_view::SortOrder::Desc => SortOrder::Desc,
138        }
139    }
140}
141
142/// Delta representing a change to a client's windowed view
143#[derive(Debug, Clone, PartialEq)]
144pub enum ViewDelta {
145    /// No change to the client's window
146    None,
147    /// Entity was added to the window
148    Add { key: String, entity: Value },
149    /// Entity was removed from the window
150    Remove { key: String },
151    /// Entity in the window was updated
152    Update { key: String, entity: Value },
153}
154
155/// Sorted view cache maintaining entities in sort order
156#[derive(Debug)]
157pub struct SortedViewCache {
158    /// View identifier
159    view_id: String,
160    /// Field path to sort by (e.g., ["id", "round_id"])
161    sort_field: Vec<String>,
162    /// Sort order
163    order: SortOrder,
164    /// Sorted entries: SortKey -> entity_key (for iteration in order)
165    sorted: BTreeMap<SortKey, ()>,
166    /// Entity data: entity_key -> (SortKey, Value)
167    entities: HashMap<String, (SortKey, Value)>,
168    /// Ordered keys cache (rebuilt on structural changes)
169    keys_cache: Vec<String>,
170    /// Whether keys_cache needs rebuild
171    cache_dirty: bool,
172}
173
174impl SortedViewCache {
175    pub fn new(view_id: String, sort_field: Vec<String>, order: SortOrder) -> Self {
176        Self {
177            view_id,
178            sort_field,
179            order,
180            sorted: BTreeMap::new(),
181            entities: HashMap::new(),
182            keys_cache: Vec::new(),
183            cache_dirty: true,
184        }
185    }
186
187    pub fn view_id(&self) -> &str {
188        &self.view_id
189    }
190
191    pub fn len(&self) -> usize {
192        self.entities.len()
193    }
194
195    pub fn is_empty(&self) -> bool {
196        self.entities.is_empty()
197    }
198
199    /// Insert or update an entity, returns the position where it was inserted
200    pub fn upsert(&mut self, entity_key: String, entity: Value) -> UpsertResult {
201        let sort_value = self.extract_sort_value(&entity);
202
203        // Check if entity already exists
204        if let Some((old_sort_key, old_entity)) = self.entities.get(&entity_key).cloned() {
205            let effective_sort_value = if matches!(sort_value, SortValue::Null)
206                && !matches!(old_sort_key.sort_value, SortValue::Null)
207            {
208                old_sort_key.sort_value.clone()
209            } else {
210                sort_value
211            };
212
213            let new_sort_key = SortKey {
214                sort_value: effective_sort_value,
215                entity_key: entity_key.clone(),
216                order: self.order,
217            };
218
219            // Merge incoming entity with existing to preserve fields not in the update
220            let merged_entity = Self::deep_merge(old_entity, entity);
221
222            if old_sort_key == new_sort_key {
223                // Sort key unchanged - just update entity data
224                self.entities
225                    .insert(entity_key.clone(), (new_sort_key, merged_entity));
226                // Position unchanged, no structural change
227                let position = self.find_position(&entity_key);
228                return UpsertResult::Updated { position };
229            }
230
231            // Sort key changed - need to reposition
232            self.sorted.remove(&old_sort_key);
233            self.sorted.insert(new_sort_key.clone(), ());
234            self.entities
235                .insert(entity_key.clone(), (new_sort_key, merged_entity));
236            self.cache_dirty = true;
237
238            let position = self.find_position(&entity_key);
239            return UpsertResult::Inserted { position };
240        }
241
242        let new_sort_key = SortKey {
243            sort_value,
244            entity_key: entity_key.clone(),
245            order: self.order,
246        };
247
248        self.sorted.insert(new_sort_key.clone(), ());
249        self.entities
250            .insert(entity_key.clone(), (new_sort_key, entity));
251        self.cache_dirty = true;
252
253        let position = self.find_position(&entity_key);
254
255        UpsertResult::Inserted { position }
256    }
257
258    fn deep_merge(base: Value, patch: Value) -> Value {
259        match (base, patch) {
260            (Value::Object(mut base_map), Value::Object(patch_map)) => {
261                for (key, patch_value) in patch_map {
262                    if let Some(base_value) = base_map.remove(&key) {
263                        base_map.insert(key, Self::deep_merge(base_value, patch_value));
264                    } else {
265                        base_map.insert(key, patch_value);
266                    }
267                }
268                Value::Object(base_map)
269            }
270            (_, patch) => patch,
271        }
272    }
273
274    /// Remove an entity, returns the position it was at
275    pub fn remove(&mut self, entity_key: &str) -> Option<usize> {
276        if let Some((sort_key, _)) = self.entities.remove(entity_key) {
277            let position = self.find_position_by_sort_key(&sort_key);
278            self.sorted.remove(&sort_key);
279            self.cache_dirty = true;
280            Some(position)
281        } else {
282            None
283        }
284    }
285
286    /// Get entity by key
287    pub fn get(&self, entity_key: &str) -> Option<&Value> {
288        self.entities.get(entity_key).map(|(_, v)| v)
289    }
290
291    /// Get ordered keys (rebuilds cache if dirty)
292    pub fn ordered_keys(&mut self) -> &[String] {
293        if self.cache_dirty {
294            self.rebuild_keys_cache();
295        }
296        &self.keys_cache
297    }
298
299    /// Get a window of entities
300    pub fn get_window(&mut self, skip: usize, take: usize) -> Vec<(String, Value)> {
301        if self.cache_dirty {
302            self.rebuild_keys_cache();
303        }
304
305        self.keys_cache
306            .iter()
307            .skip(skip)
308            .take(take)
309            .filter_map(|key| {
310                self.entities
311                    .get(key)
312                    .map(|(_, v)| (key.clone(), v.clone()))
313            })
314            .collect()
315    }
316
317    /// Get every entity in deterministic sort order for query-side filtering.
318    pub fn get_all_ordered(&mut self) -> Vec<(String, Value)> {
319        if self.cache_dirty {
320            self.rebuild_keys_cache();
321        }
322
323        self.keys_cache
324            .iter()
325            .filter_map(|key| {
326                self.entities
327                    .get(key)
328                    .map(|(_, value)| (key.clone(), value.clone()))
329            })
330            .collect()
331    }
332
333    /// Compute deltas for a client with a specific window
334    pub fn compute_window_deltas(
335        &mut self,
336        old_window_keys: &[String],
337        skip: usize,
338        take: usize,
339    ) -> Vec<ViewDelta> {
340        if self.cache_dirty {
341            self.rebuild_keys_cache();
342        }
343
344        let new_window_keys: Vec<&String> = self.keys_cache.iter().skip(skip).take(take).collect();
345
346        let old_set: std::collections::HashSet<&String> = old_window_keys.iter().collect();
347        let new_set: std::collections::HashSet<&String> = new_window_keys.iter().cloned().collect();
348
349        let mut deltas = Vec::new();
350
351        // Removed from window
352        for key in old_set.difference(&new_set) {
353            deltas.push(ViewDelta::Remove {
354                key: (*key).clone(),
355            });
356        }
357
358        // Added to window
359        for key in new_set.difference(&old_set) {
360            if let Some((_, entity)) = self.entities.get(*key) {
361                deltas.push(ViewDelta::Add {
362                    key: (*key).clone(),
363                    entity: entity.clone(),
364                });
365            }
366        }
367
368        deltas
369    }
370
371    fn extract_sort_value(&self, entity: &Value) -> SortValue {
372        let mut current = entity;
373        for segment in &self.sort_field {
374            match current.get(segment) {
375                Some(v) => current = v,
376                None => return SortValue::Null,
377            }
378        }
379
380        value_to_sort_value(current)
381    }
382
383    fn find_position(&self, entity_key: &str) -> usize {
384        if let Some((sort_key, _)) = self.entities.get(entity_key) {
385            self.find_position_by_sort_key(sort_key)
386        } else {
387            0
388        }
389    }
390
391    fn find_position_by_sort_key(&self, sort_key: &SortKey) -> usize {
392        self.sorted.range(..sort_key).count()
393    }
394
395    fn rebuild_keys_cache(&mut self) {
396        self.keys_cache = self.sorted.keys().map(|sk| sk.entity_key.clone()).collect();
397        self.cache_dirty = false;
398    }
399}
400
401/// Result of an upsert operation
402#[derive(Debug, Clone, PartialEq)]
403pub enum UpsertResult {
404    /// Entity was inserted at a new position
405    Inserted { position: usize },
406    /// Entity was updated (may or may not have moved)
407    Updated { position: usize },
408}
409
410fn value_to_sort_value(v: &Value) -> SortValue {
411    match v {
412        Value::Null => SortValue::Null,
413        Value::Bool(b) => SortValue::Bool(*b),
414        Value::Number(n) => {
415            if let Some(i) = n.as_i64() {
416                SortValue::Integer(i)
417            } else if let Some(f) = n.as_f64() {
418                SortValue::Float(OrderedFloat(f))
419            } else {
420                SortValue::Null
421            }
422        }
423        Value::String(s) => SortValue::String(s.clone()),
424        _ => SortValue::Null,
425    }
426}
427
428fn compare_decimal_strings(left: &str, right: &str) -> Option<Ordering> {
429    fn parts(value: &str) -> Option<(bool, &str)> {
430        let (negative, digits) = match value.strip_prefix('-') {
431            Some(digits) => (true, digits),
432            None => (false, value),
433        };
434        if digits.is_empty() || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
435            return None;
436        }
437
438        let digits = digits.trim_start_matches('0');
439        let digits = if digits.is_empty() { "0" } else { digits };
440        Some((negative && digits != "0", digits))
441    }
442
443    let (left_negative, left_digits) = parts(left)?;
444    let (right_negative, right_digits) = parts(right)?;
445
446    match (left_negative, right_negative) {
447        (true, false) => Some(Ordering::Less),
448        (false, true) => Some(Ordering::Greater),
449        _ => {
450            let magnitude = left_digits
451                .len()
452                .cmp(&right_digits.len())
453                .then_with(|| left_digits.cmp(right_digits));
454            Some(if left_negative {
455                magnitude.reverse()
456            } else {
457                magnitude
458            })
459        }
460    }
461}
462
463#[cfg(test)]
464mod tests {
465    use super::*;
466    use serde_json::json;
467
468    #[test]
469    fn test_sorted_cache_basic() {
470        let mut cache = SortedViewCache::new(
471            "test/latest".to_string(),
472            vec!["id".to_string()],
473            SortOrder::Desc,
474        );
475
476        cache.upsert("a".to_string(), json!({"id": 1, "name": "first"}));
477        cache.upsert("b".to_string(), json!({"id": 3, "name": "third"}));
478        cache.upsert("c".to_string(), json!({"id": 2, "name": "second"}));
479
480        let keys = cache.ordered_keys();
481        // Desc order: 3, 2, 1
482        assert_eq!(keys, vec!["b", "c", "a"]);
483    }
484
485    #[test]
486    fn test_sorted_cache_window() {
487        let mut cache = SortedViewCache::new(
488            "test/latest".to_string(),
489            vec!["id".to_string()],
490            SortOrder::Desc,
491        );
492
493        for i in 1..=10 {
494            cache.upsert(format!("e{}", i), json!({"id": i}));
495        }
496
497        // Desc order: 10, 9, 8, 7, 6, 5, 4, 3, 2, 1
498        let window = cache.get_window(0, 3);
499        assert_eq!(window.len(), 3);
500        assert_eq!(window[0].0, "e10");
501        assert_eq!(window[1].0, "e9");
502        assert_eq!(window[2].0, "e8");
503
504        let window = cache.get_window(3, 3);
505        assert_eq!(window[0].0, "e7");
506    }
507
508    #[test]
509    fn all_ordered_preserves_stable_sort_and_tie_breaking() {
510        let mut cache = SortedViewCache::new(
511            "test/latest".to_string(),
512            vec!["score".to_string()],
513            SortOrder::Desc,
514        );
515        cache.upsert("b".to_string(), json!({"score": 10}));
516        cache.upsert("a".to_string(), json!({"score": 10}));
517        cache.upsert("c".to_string(), json!({"score": 9}));
518
519        let keys: Vec<_> = cache
520            .get_all_ordered()
521            .into_iter()
522            .map(|(key, _)| key)
523            .collect();
524        assert_eq!(keys, ["a", "b", "c"]);
525    }
526
527    #[test]
528    fn test_sorted_cache_update_moves_position() {
529        let mut cache = SortedViewCache::new(
530            "test/latest".to_string(),
531            vec!["score".to_string()],
532            SortOrder::Desc,
533        );
534
535        cache.upsert("a".to_string(), json!({"score": 10}));
536        cache.upsert("b".to_string(), json!({"score": 20}));
537        cache.upsert("c".to_string(), json!({"score": 15}));
538
539        // Order: b(20), c(15), a(10)
540        assert_eq!(cache.ordered_keys(), vec!["b", "c", "a"]);
541
542        // Update a to have highest score
543        cache.upsert("a".to_string(), json!({"score": 25}));
544
545        // New order: a(25), b(20), c(15)
546        assert_eq!(cache.ordered_keys(), vec!["a", "b", "c"]);
547    }
548
549    #[test]
550    fn test_sorted_cache_remove() {
551        let mut cache = SortedViewCache::new(
552            "test/latest".to_string(),
553            vec!["id".to_string()],
554            SortOrder::Asc,
555        );
556
557        cache.upsert("a".to_string(), json!({"id": 1}));
558        cache.upsert("b".to_string(), json!({"id": 2}));
559        cache.upsert("c".to_string(), json!({"id": 3}));
560
561        assert_eq!(cache.len(), 3);
562
563        let pos = cache.remove("b");
564        assert_eq!(pos, Some(1));
565        assert_eq!(cache.len(), 2);
566        assert_eq!(cache.ordered_keys(), vec!["a", "c"]);
567    }
568
569    #[test]
570    fn test_compute_window_deltas() {
571        let mut cache = SortedViewCache::new(
572            "test/latest".to_string(),
573            vec!["id".to_string()],
574            SortOrder::Desc,
575        );
576
577        // Initial: 5, 4, 3, 2, 1
578        for i in 1..=5 {
579            cache.upsert(format!("e{}", i), json!({"id": i}));
580        }
581
582        let old_window: Vec<String> = vec!["e5".to_string(), "e4".to_string(), "e3".to_string()];
583
584        // Add e6 (new top)
585        cache.upsert("e6".to_string(), json!({"id": 6}));
586
587        // New order: 6, 5, 4, 3, 2, 1
588        // New top 3: e6, e5, e4
589        let deltas = cache.compute_window_deltas(&old_window, 0, 3);
590
591        assert_eq!(deltas.len(), 2);
592        // e3 removed from window
593        assert!(deltas
594            .iter()
595            .any(|d| matches!(d, ViewDelta::Remove { key } if key == "e3")));
596        // e6 added to window
597        assert!(deltas
598            .iter()
599            .any(|d| matches!(d, ViewDelta::Add { key, .. } if key == "e6")));
600    }
601
602    #[test]
603    fn test_nested_sort_field() {
604        let mut cache = SortedViewCache::new(
605            "test/latest".to_string(),
606            vec!["id".to_string(), "round_id".to_string()],
607            SortOrder::Desc,
608        );
609
610        cache.upsert("a".to_string(), json!({"id": {"round_id": 1}}));
611        cache.upsert("b".to_string(), json!({"id": {"round_id": 3}}));
612        cache.upsert("c".to_string(), json!({"id": {"round_id": 2}}));
613
614        let keys = cache.ordered_keys();
615        assert_eq!(keys, vec!["b", "c", "a"]);
616    }
617
618    #[test]
619    fn test_nested_decimal_string_sort_field() {
620        let mut cache = SortedViewCache::new(
621            "test/latest".to_string(),
622            vec!["id".to_string(), "round_id".to_string()],
623            SortOrder::Desc,
624        );
625
626        cache.upsert("9".to_string(), json!({"id": {"round_id": "9"}}));
627        cache.upsert("100".to_string(), json!({"id": {"round_id": "100"}}));
628        cache.upsert("10".to_string(), json!({"id": {"round_id": "10"}}));
629
630        assert_eq!(cache.ordered_keys(), vec!["100", "10", "9"]);
631    }
632
633    #[test]
634    fn test_descending_string_sort_field() {
635        let mut cache = SortedViewCache::new(
636            "test/latest".to_string(),
637            vec!["name".to_string()],
638            SortOrder::Desc,
639        );
640
641        cache.upsert("a".to_string(), json!({"name": "alpha"}));
642        cache.upsert("c".to_string(), json!({"name": "charlie"}));
643        cache.upsert("b".to_string(), json!({"name": "bravo"}));
644
645        assert_eq!(cache.ordered_keys(), vec!["c", "b", "a"]);
646    }
647
648    #[test]
649    fn test_update_with_missing_sort_field_preserves_position() {
650        let mut cache = SortedViewCache::new(
651            "test/latest".to_string(),
652            vec!["id".to_string(), "round_id".to_string()],
653            SortOrder::Desc,
654        );
655
656        cache.upsert(
657            "100".to_string(),
658            json!({"id": {"round_id": 100}, "data": "initial"}),
659        );
660        cache.upsert(
661            "200".to_string(),
662            json!({"id": {"round_id": 200}, "data": "initial"}),
663        );
664        cache.upsert(
665            "300".to_string(),
666            json!({"id": {"round_id": 300}, "data": "initial"}),
667        );
668
669        assert_eq!(cache.ordered_keys(), vec!["300", "200", "100"]);
670
671        cache.upsert("200".to_string(), json!({"data": "updated_without_id"}));
672
673        assert_eq!(
674            cache.ordered_keys(),
675            vec!["300", "200", "100"],
676            "Entity 200 should retain its position even when updated without sort field"
677        );
678
679        let entity = cache.get("200").unwrap();
680        assert_eq!(entity["data"], "updated_without_id");
681    }
682
683    #[test]
684    fn test_new_entity_with_missing_sort_field_gets_null_position() {
685        let mut cache = SortedViewCache::new(
686            "test/latest".to_string(),
687            vec!["id".to_string(), "round_id".to_string()],
688            SortOrder::Desc,
689        );
690
691        cache.upsert("100".to_string(), json!({"id": {"round_id": 100}}));
692        cache.upsert("200".to_string(), json!({"id": {"round_id": 200}}));
693
694        cache.upsert("new".to_string(), json!({"data": "no_sort_field"}));
695
696        let keys = cache.ordered_keys();
697        assert_eq!(
698            keys.first().unwrap(),
699            "new",
700            "New entity without sort field gets Null which sorts first (Null < any value)"
701        );
702    }
703}