1use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use std::cmp::Ordering;
10use std::collections::{BTreeMap, HashMap};
11
12#[derive(Debug, Clone, PartialEq, Eq)]
16pub struct SortKey {
17 sort_value: SortValue,
19 entity_key: String,
21 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#[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 (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 (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 (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#[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#[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#[derive(Debug, Clone, PartialEq)]
144pub enum ViewDelta {
145 None,
147 Add { key: String, entity: Value },
149 Remove { key: String },
151 Update { key: String, entity: Value },
153}
154
155#[derive(Debug)]
157pub struct SortedViewCache {
158 view_id: String,
160 sort_field: Vec<String>,
162 order: SortOrder,
164 sorted: BTreeMap<SortKey, ()>,
166 entities: HashMap<String, (SortKey, Value)>,
168 keys_cache: Vec<String>,
170 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 pub fn upsert(&mut self, entity_key: String, entity: Value) -> UpsertResult {
201 let sort_value = self.extract_sort_value(&entity);
202
203 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 let merged_entity = Self::deep_merge(old_entity, entity);
221
222 if old_sort_key == new_sort_key {
223 self.entities
225 .insert(entity_key.clone(), (new_sort_key, merged_entity));
226 let position = self.find_position(&entity_key);
228 return UpsertResult::Updated { position };
229 }
230
231 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 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 pub fn get(&self, entity_key: &str) -> Option<&Value> {
288 self.entities.get(entity_key).map(|(_, v)| v)
289 }
290
291 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 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 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 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 for key in old_set.difference(&new_set) {
353 deltas.push(ViewDelta::Remove {
354 key: (*key).clone(),
355 });
356 }
357
358 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#[derive(Debug, Clone, PartialEq)]
403pub enum UpsertResult {
404 Inserted { position: usize },
406 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 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 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 assert_eq!(cache.ordered_keys(), vec!["b", "c", "a"]);
541
542 cache.upsert("a".to_string(), json!({"score": 25}));
544
545 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 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 cache.upsert("e6".to_string(), json!({"id": 6}));
586
587 let deltas = cache.compute_window_deltas(&old_window, 0, 3);
590
591 assert_eq!(deltas.len(), 2);
592 assert!(deltas
594 .iter()
595 .any(|d| matches!(d, ViewDelta::Remove { key } if key == "e3")));
596 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}