arete_server/

sorted_cache.rs

//! Sorted view cache for maintaining ordered entity collections.
//!
//! This module provides incremental maintenance of sorted entity views,
//! enabling efficient windowed subscriptions (take/skip) with minimal
//! recomputation on updates.

use crate::shared_entity::{lookup, EntityFields, SharedEntity};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::cmp::Ordering;
use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;

/// A sortable key that combines the sort value with entity key for stable ordering.
/// Uses (sort_value, entity_key) tuple to ensure deterministic ordering even when
/// sort values are equal.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SortKey {
    /// The extracted sort value (as comparable bytes)
    sort_value: SortValue,
    /// Entity key for tie-breaking
    entity_key: String,
    /// Direction to apply to the sort value comparison.
    order: SortOrder,
}

impl PartialOrd for SortKey {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for SortKey {
    fn cmp(&self, other: &Self) -> Ordering {
        if self.order != other.order {
            return match (self.order, other.order) {
                (SortOrder::Asc, SortOrder::Desc) => Ordering::Less,
                (SortOrder::Desc, SortOrder::Asc) => Ordering::Greater,
                _ => Ordering::Equal,
            };
        }

        // An entity with no sort value (missing or `null`) cannot be ranked,
        // so it sorts after every entity that has one in *both* directions:
        // it must never take a window slot from a ranked entity. `desc`
        // reverses only the comparison between two present values.
        let sort_order = match (&self.sort_value, &other.sort_value, self.order) {
            (SortValue::Null, SortValue::Null, _) => Ordering::Equal,
            (SortValue::Null, _, _) => Ordering::Greater,
            (_, SortValue::Null, _) => Ordering::Less,
            (_, _, SortOrder::Asc) => self.sort_value.cmp(&other.sort_value),
            (_, _, SortOrder::Desc) => self.sort_value.cmp(&other.sort_value).reverse(),
        };

        match sort_order {
            Ordering::Equal => self.entity_key.cmp(&other.entity_key),
            other => other,
        }
    }
}

/// Comparable sort value extracted from JSON
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SortValue {
    Null,
    Bool(bool),
    Integer(i64),
    Float(OrderedFloat),
    String(String),
}

impl Ord for SortValue {
    fn cmp(&self, other: &Self) -> Ordering {
        match (self, other) {
            (SortValue::Null, SortValue::Null) => Ordering::Equal,
            (SortValue::Null, _) => Ordering::Less,
            (_, SortValue::Null) => Ordering::Greater,
            (SortValue::Bool(a), SortValue::Bool(b)) => a.cmp(b),
            (SortValue::Integer(a), SortValue::Integer(b)) => a.cmp(b),
            (SortValue::Float(a), SortValue::Float(b)) => a.cmp(b),
            (SortValue::String(a), SortValue::String(b)) => {
                compare_decimal_strings(a, b).unwrap_or_else(|| a.cmp(b))
            }
            // Cross-type comparisons: numbers < strings
            (SortValue::Integer(_), SortValue::String(_)) => Ordering::Less,
            (SortValue::String(_), SortValue::Integer(_)) => Ordering::Greater,
            (SortValue::Float(_), SortValue::String(_)) => Ordering::Less,
            (SortValue::String(_), SortValue::Float(_)) => Ordering::Greater,
            // Integer vs Float: convert to float
            (SortValue::Integer(a), SortValue::Float(b)) => OrderedFloat(*a as f64).cmp(b),
            (SortValue::Float(a), SortValue::Integer(b)) => a.cmp(&OrderedFloat(*b as f64)),
            // Bool vs others
            (SortValue::Bool(_), _) => Ordering::Less,
            (_, SortValue::Bool(_)) => Ordering::Greater,
        }
    }
}

impl PartialOrd for SortValue {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

/// Wrapper for f64 that implements Ord (treats NaN as less than all values)
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct OrderedFloat(pub f64);

impl Eq for OrderedFloat {}

impl Ord for OrderedFloat {
    fn cmp(&self, other: &Self) -> Ordering {
        self.0.partial_cmp(&other.0).unwrap_or_else(|| {
            if self.0.is_nan() && other.0.is_nan() {
                Ordering::Equal
            } else if self.0.is_nan() {
                Ordering::Less
            } else {
                Ordering::Greater
            }
        })
    }
}

impl PartialOrd for OrderedFloat {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

/// Sort order for the cache
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum SortOrder {
    Asc,
    Desc,
}

impl From<crate::materialized_view::SortOrder> for SortOrder {
    fn from(order: crate::materialized_view::SortOrder) -> Self {
        match order {
            crate::materialized_view::SortOrder::Asc => SortOrder::Asc,
            crate::materialized_view::SortOrder::Desc => SortOrder::Desc,
        }
    }
}

/// Delta representing a change to a client's windowed view
#[derive(Debug, Clone, PartialEq)]
pub enum ViewDelta {
    /// No change to the client's window
    None,
    /// Entity was added to the window
    Add { key: String, entity: Value },
    /// Entity was removed from the window
    Remove { key: String },
    /// Entity in the window was updated
    Update { key: String, entity: Value },
}

/// Sorted view cache maintaining entities in sort order.
///
/// # Bounding
///
/// The cache holds every entity it has been given. It shares an entity's
/// fields with whoever gave it (see [`SharedEntity`]) rather than copying
/// them, but it keeps them alive, so callers that feed it from a bounded
/// source (the projector and snapshot restore feed it from the LRU-capped
/// [`EntityCache`](crate::EntityCache)) should use
/// [`upsert_bounded`](Self::upsert_bounded) or
/// [`trim_to_max_entries`](Self::trim_to_max_entries) with that source's cap.
///
/// Entries are evicted from the *bottom* of the sort order, not by recency.
/// Evicting the least-recently-updated entries (mirroring the entity cache)
/// would drop a top-ranked entity that simply has not updated recently and
/// break leaderboard-style `sort` + `take` views. Evicting everything beyond
/// position `max_entries` keeps every window with `skip + take <= max_entries`
/// exact. An evicted entity re-enters on its next update, because the
/// projector re-reads the full entity from the entity cache before upserting.
///
/// Known edge: after entities are removed from (or move down out of) the top
/// of the order, a previously evicted entity that has not updated since is
/// missing from the cache until it next updates, so a window near the cap can
/// under-fill or show a lower-ranked entity in its place until then.
#[derive(Debug)]
pub struct SortedViewCache {
    /// View identifier
    view_id: String,
    /// Field path to sort by (e.g., ["id", "round_id"])
    sort_field: Vec<String>,
    /// Sort order
    order: SortOrder,
    /// Sorted entries: SortKey -> entity_key (for iteration in order)
    sorted: BTreeMap<SortKey, ()>,
    /// Entity data: entity_key -> (SortKey, entity)
    entities: HashMap<String, (SortKey, SharedEntity)>,
    /// Ordered keys cache (rebuilt on structural changes)
    keys_cache: Vec<String>,
    /// Whether keys_cache needs rebuild
    cache_dirty: bool,
}

impl SortedViewCache {
    pub fn new(view_id: String, sort_field: Vec<String>, order: SortOrder) -> Self {
        Self {
            view_id,
            sort_field,
            order,
            sorted: BTreeMap::new(),
            entities: HashMap::new(),
            keys_cache: Vec::new(),
            cache_dirty: true,
        }
    }

    pub fn view_id(&self) -> &str {
        &self.view_id
    }

    pub fn len(&self) -> usize {
        self.entities.len()
    }

    pub fn is_empty(&self) -> bool {
        self.entities.is_empty()
    }

    /// Insert or update an entity, returns the position where it was inserted
    ///
    /// An update merges into the entity held, keeping fields the update lacks.
    /// When it lacks none, which is the case for a whole entity, the cache
    /// shares the update's fields instead of copying them.
    pub fn upsert(&mut self, entity_key: String, entity: impl Into<SharedEntity>) -> UpsertResult {
        let entity = entity.into();
        let sort_value = self.extract_sort_value(&entity);

        // Check if entity already exists. Taken out rather than copied: every
        // branch below puts the merged entity back.
        if let Some((old_sort_key, old_entity)) = self.entities.remove(&entity_key) {
            let effective_sort_value = if matches!(sort_value, SortValue::Null)
                && !matches!(old_sort_key.sort_value, SortValue::Null)
            {
                old_sort_key.sort_value.clone()
            } else {
                sort_value
            };

            let new_sort_key = SortKey {
                sort_value: effective_sort_value,
                entity_key: entity_key.clone(),
                order: self.order,
            };

            // Merge incoming entity with existing to preserve fields not in the update
            let merged_entity = Self::merge_entity(old_entity, entity);

            if old_sort_key == new_sort_key {
                // Sort key unchanged - just update entity data
                self.entities
                    .insert(entity_key.clone(), (new_sort_key, merged_entity));
                // Position unchanged, no structural change
                let position = self.find_position(&entity_key);
                return UpsertResult::Updated { position };
            }

            // Sort key changed - need to reposition
            self.sorted.remove(&old_sort_key);
            self.sorted.insert(new_sort_key.clone(), ());
            self.entities
                .insert(entity_key.clone(), (new_sort_key, merged_entity));
            self.cache_dirty = true;

            let position = self.find_position(&entity_key);
            return UpsertResult::Inserted { position };
        }

        let new_sort_key = SortKey {
            sort_value,
            entity_key: entity_key.clone(),
            order: self.order,
        };

        self.sorted.insert(new_sort_key.clone(), ());
        self.entities
            .insert(entity_key.clone(), (new_sort_key, entity));
        self.cache_dirty = true;

        let position = self.find_position(&entity_key);

        UpsertResult::Inserted { position }
    }

    /// Whether upserting `entity` under `entity_key` into a cache bounded at
    /// `max_entries` would keep it: the cache already holds the key, has
    /// room, or the entity sorts before its current last entry.
    ///
    /// Lets a caller skip copying an entity that [`Self::upsert_bounded`]
    /// would evict straight away, which in a busy view is most of them.
    pub fn would_keep<E: EntityFields + ?Sized>(
        &self,
        entity_key: &str,
        entity: &E,
        max_entries: usize,
    ) -> bool {
        if self.entities.contains_key(entity_key) || self.sorted.len() < max_entries {
            return true;
        }
        let Some((last, ())) = self.sorted.last_key_value() else {
            return true;
        };
        let candidate = SortKey {
            sort_value: self.extract_sort_value(entity),
            entity_key: entity_key.to_string(),
            order: self.order,
        };
        candidate < *last
    }

    /// Upsert an entity, then evict from the bottom of the sort order so the
    /// cache holds at most `max_entries` entities.
    ///
    /// If the upserted entity itself sorts beyond `max_entries` it is evicted
    /// immediately and the returned position is `>= max_entries`.
    pub fn upsert_bounded(
        &mut self,
        entity_key: String,
        entity: impl Into<SharedEntity>,
        max_entries: usize,
    ) -> UpsertResult {
        let result = self.upsert(entity_key, entity);
        self.trim_to_max_entries(max_entries);
        result
    }

    /// Evict entities from the bottom of the sort order until at most
    /// `max_entries` remain. Returns the number of evicted entities.
    ///
    /// Each eviction is an `O(log n)` pop from the end of the ordered index,
    /// so a batch of evictions (e.g. after a bulk rebuild) costs no more than
    /// the inserts that caused it. The ordered-keys cache is truncated in place
    /// when it is current, so trimming does not force a full rebuild.
    pub fn trim_to_max_entries(&mut self, max_entries: usize) -> usize {
        let mut evicted = 0;
        while self.sorted.len() > max_entries {
            let Some((sort_key, ())) = self.sorted.pop_last() else {
                break;
            };
            self.entities.remove(&sort_key.entity_key);
            evicted += 1;
        }
        if evicted > 0 && !self.cache_dirty {
            // Only the tail of the order was removed, so the prefix is still
            // exact.
            self.keys_cache.truncate(self.sorted.len());
        }
        evicted
    }

    /// [`Self::deep_merge`] for whole entities, `_version` included. The
    /// result shares `patch`'s fields whenever merging adds nothing to them:
    /// when `base` has no field, at any depth, that `patch` lacks.
    fn merge_entity(base: SharedEntity, patch: SharedEntity) -> SharedEntity {
        let (base_fields, base_version) = base.into_parts();
        let (patch_fields, patch_version) = patch.into_parts();
        if !(base_fields.is_object() && patch_fields.is_object()) {
            return SharedEntity::from_parts(patch_fields, patch_version);
        }
        let fields = if Arc::ptr_eq(&base_fields, &patch_fields)
            || !keeps_fields(&base_fields, &patch_fields)
        {
            patch_fields
        } else {
            let base = Arc::try_unwrap(base_fields).unwrap_or_else(|shared| Value::clone(&shared));
            Arc::new(Self::deep_merge(base, Value::clone(&patch_fields)))
        };
        let version = match (base_version, patch_version) {
            (Some(base), Some(patch)) => Some(Self::deep_merge(base, patch)),
            (base, None) => base,
            (None, patch) => patch,
        };
        SharedEntity::from_parts(fields, version)
    }

    fn deep_merge(base: Value, patch: Value) -> Value {
        match (base, patch) {
            (Value::Object(mut base_map), Value::Object(patch_map)) => {
                for (key, patch_value) in patch_map {
                    if let Some(base_value) = base_map.remove(&key) {
                        base_map.insert(key, Self::deep_merge(base_value, patch_value));
                    } else {
                        base_map.insert(key, patch_value);
                    }
                }
                Value::Object(base_map)
            }
            (_, patch) => patch,
        }
    }

    /// Remove an entity, returns the position it was at
    pub fn clear(&mut self) {
        self.sorted.clear();
        self.entities.clear();
        self.cache_dirty = true;
    }

    pub fn remove(&mut self, entity_key: &str) -> Option<usize> {
        if let Some((sort_key, _)) = self.entities.remove(entity_key) {
            let position = self.find_position_by_sort_key(&sort_key);
            self.sorted.remove(&sort_key);
            self.cache_dirty = true;
            Some(position)
        } else {
            None
        }
    }

    /// Get entity by key
    pub fn get(&self, entity_key: &str) -> Option<&SharedEntity> {
        self.entities.get(entity_key).map(|(_, v)| v)
    }

    /// Get ordered keys (rebuilds cache if dirty)
    pub fn ordered_keys(&mut self) -> &[String] {
        if self.cache_dirty {
            self.rebuild_keys_cache();
        }
        &self.keys_cache
    }

    /// Get a window of entities, as copies.
    pub fn get_window(&mut self, skip: usize, take: usize) -> Vec<(String, Value)> {
        if self.cache_dirty {
            self.rebuild_keys_cache();
        }

        self.keys_cache
            .iter()
            .skip(skip)
            .take(take)
            .filter_map(|key| {
                self.entities
                    .get(key)
                    .map(|(_, v)| (key.clone(), v.to_value()))
            })
            .collect()
    }

    /// Copies of every entity in deterministic sort order.
    pub fn get_all_ordered(&mut self) -> Vec<(String, Value)> {
        self.ordered_entities()
            .into_iter()
            .map(|(key, entity)| (key, entity.into_value()))
            .collect()
    }

    /// Every entity in deterministic sort order for query-side filtering,
    /// sharing their fields with the cache rather than copying them.
    pub fn ordered_entities(&mut self) -> Vec<(String, SharedEntity)> {
        if self.cache_dirty {
            self.rebuild_keys_cache();
        }

        self.keys_cache
            .iter()
            .filter_map(|key| {
                self.entities
                    .get(key)
                    .map(|(_, entity)| (key.clone(), entity.clone()))
            })
            .collect()
    }

    /// Compute deltas for a client with a specific window
    pub fn compute_window_deltas(
        &mut self,
        old_window_keys: &[String],
        skip: usize,
        take: usize,
    ) -> Vec<ViewDelta> {
        if self.cache_dirty {
            self.rebuild_keys_cache();
        }

        let new_window_keys: Vec<&String> = self.keys_cache.iter().skip(skip).take(take).collect();

        let old_set: std::collections::HashSet<&String> = old_window_keys.iter().collect();
        let new_set: std::collections::HashSet<&String> = new_window_keys.iter().cloned().collect();

        let mut deltas = Vec::new();

        // Removed from window
        for key in old_set.difference(&new_set) {
            deltas.push(ViewDelta::Remove {
                key: (*key).clone(),
            });
        }

        // Added to window
        for key in new_set.difference(&old_set) {
            if let Some((_, entity)) = self.entities.get(*key) {
                deltas.push(ViewDelta::Add {
                    key: (*key).clone(),
                    entity: entity.to_value(),
                });
            }
        }

        deltas
    }

    fn extract_sort_value<E: EntityFields + ?Sized>(&self, entity: &E) -> SortValue {
        match lookup(entity, &self.sort_field) {
            Some(value) => value_to_sort_value(&value),
            None => SortValue::Null,
        }
    }

    fn find_position(&self, entity_key: &str) -> usize {
        if let Some((sort_key, _)) = self.entities.get(entity_key) {
            self.find_position_by_sort_key(sort_key)
        } else {
            0
        }
    }

    fn find_position_by_sort_key(&self, sort_key: &SortKey) -> usize {
        self.sorted.range(..sort_key).count()
    }

    fn rebuild_keys_cache(&mut self) {
        self.keys_cache = self.sorted.keys().map(|sk| sk.entity_key.clone()).collect();
        self.cache_dirty = false;
    }
}

/// Result of an upsert operation
#[derive(Debug, Clone, PartialEq)]
pub enum UpsertResult {
    /// Entity was inserted at a new position
    Inserted { position: usize },
    /// Entity was updated (may or may not have moved)
    Updated { position: usize },
}

/// Whether merging `patch` into `base` keeps a field `patch` lacks: whether
/// `base` has one, in an object where `patch` has an object too.
fn keeps_fields(base: &Value, patch: &Value) -> bool {
    match (base, patch) {
        (Value::Object(base), Value::Object(patch)) => base
            .iter()
            .any(|(key, base)| patch.get(key).is_none_or(|patch| keeps_fields(base, patch))),
        _ => false,
    }
}

fn value_to_sort_value(v: &Value) -> SortValue {
    match v {
        Value::Null => SortValue::Null,
        Value::Bool(b) => SortValue::Bool(*b),
        Value::Number(n) => {
            if let Some(i) = n.as_i64() {
                SortValue::Integer(i)
            } else if let Some(f) = n.as_f64() {
                SortValue::Float(OrderedFloat(f))
            } else {
                SortValue::Null
            }
        }
        Value::String(s) => SortValue::String(s.clone()),
        _ => SortValue::Null,
    }
}

fn compare_decimal_strings(left: &str, right: &str) -> Option<Ordering> {
    fn parts(value: &str) -> Option<(bool, &str)> {
        let (negative, digits) = match value.strip_prefix('-') {
            Some(digits) => (true, digits),
            None => (false, value),
        };
        if digits.is_empty() || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
            return None;
        }

        let digits = digits.trim_start_matches('0');
        let digits = if digits.is_empty() { "0" } else { digits };
        Some((negative && digits != "0", digits))
    }

    let (left_negative, left_digits) = parts(left)?;
    let (right_negative, right_digits) = parts(right)?;

    match (left_negative, right_negative) {
        (true, false) => Some(Ordering::Less),
        (false, true) => Some(Ordering::Greater),
        _ => {
            let magnitude = left_digits
                .len()
                .cmp(&right_digits.len())
                .then_with(|| left_digits.cmp(right_digits));
            Some(if left_negative {
                magnitude.reverse()
            } else {
                magnitude
            })
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn test_sorted_cache_basic() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string()],
            SortOrder::Desc,
        );

        cache.upsert("a".to_string(), json!({"id": 1, "name": "first"}));
        cache.upsert("b".to_string(), json!({"id": 3, "name": "third"}));
        cache.upsert("c".to_string(), json!({"id": 2, "name": "second"}));

        let keys = cache.ordered_keys();
        // Desc order: 3, 2, 1
        assert_eq!(keys, vec!["b", "c", "a"]);
    }

    #[test]
    fn test_sorted_cache_window() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string()],
            SortOrder::Desc,
        );

        for i in 1..=10 {
            cache.upsert(format!("e{}", i), json!({"id": i}));
        }

        // Desc order: 10, 9, 8, 7, 6, 5, 4, 3, 2, 1
        let window = cache.get_window(0, 3);
        assert_eq!(window.len(), 3);
        assert_eq!(window[0].0, "e10");
        assert_eq!(window[1].0, "e9");
        assert_eq!(window[2].0, "e8");

        let window = cache.get_window(3, 3);
        assert_eq!(window[0].0, "e7");
    }

    #[test]
    fn all_ordered_preserves_stable_sort_and_tie_breaking() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );
        cache.upsert("b".to_string(), json!({"score": 10}));
        cache.upsert("a".to_string(), json!({"score": 10}));
        cache.upsert("c".to_string(), json!({"score": 9}));

        let keys: Vec<_> = cache
            .get_all_ordered()
            .into_iter()
            .map(|(key, _)| key)
            .collect();
        assert_eq!(keys, ["a", "b", "c"]);
    }

    #[test]
    fn test_sorted_cache_update_moves_position() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        cache.upsert("a".to_string(), json!({"score": 10}));
        cache.upsert("b".to_string(), json!({"score": 20}));
        cache.upsert("c".to_string(), json!({"score": 15}));

        // Order: b(20), c(15), a(10)
        assert_eq!(cache.ordered_keys(), vec!["b", "c", "a"]);

        // Update a to have highest score
        cache.upsert("a".to_string(), json!({"score": 25}));

        // New order: a(25), b(20), c(15)
        assert_eq!(cache.ordered_keys(), vec!["a", "b", "c"]);
    }

    #[test]
    fn test_sorted_cache_remove() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string()],
            SortOrder::Asc,
        );

        cache.upsert("a".to_string(), json!({"id": 1}));
        cache.upsert("b".to_string(), json!({"id": 2}));
        cache.upsert("c".to_string(), json!({"id": 3}));

        assert_eq!(cache.len(), 3);

        let pos = cache.remove("b");
        assert_eq!(pos, Some(1));
        assert_eq!(cache.len(), 2);
        assert_eq!(cache.ordered_keys(), vec!["a", "c"]);
    }

    #[test]
    fn test_compute_window_deltas() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string()],
            SortOrder::Desc,
        );

        // Initial: 5, 4, 3, 2, 1
        for i in 1..=5 {
            cache.upsert(format!("e{}", i), json!({"id": i}));
        }

        let old_window: Vec<String> = vec!["e5".to_string(), "e4".to_string(), "e3".to_string()];

        // Add e6 (new top)
        cache.upsert("e6".to_string(), json!({"id": 6}));

        // New order: 6, 5, 4, 3, 2, 1
        // New top 3: e6, e5, e4
        let deltas = cache.compute_window_deltas(&old_window, 0, 3);

        assert_eq!(deltas.len(), 2);
        // e3 removed from window
        assert!(deltas
            .iter()
            .any(|d| matches!(d, ViewDelta::Remove { key } if key == "e3")));
        // e6 added to window
        assert!(deltas
            .iter()
            .any(|d| matches!(d, ViewDelta::Add { key, .. } if key == "e6")));
    }

    fn keys(cache: &mut SortedViewCache) -> Vec<String> {
        cache.ordered_keys().to_vec()
    }

    #[test]
    fn bounded_upsert_evicts_bottom_of_desc_order() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        for i in 1..=10 {
            cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 4);
            assert!(cache.len() <= 4);
        }

        assert_eq!(cache.len(), 4);
        assert_eq!(keys(&mut cache), ["e10", "e9", "e8", "e7"]);
        assert!(cache.get("e1").is_none());
        assert!(cache.get("e6").is_none());
    }

    #[test]
    fn would_keep_matches_what_a_bounded_upsert_keeps() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );
        for i in 1..=10 {
            let entity = json!({"score": i});
            let expected = cache.would_keep(&format!("e{i}"), &entity, 4);
            cache.upsert_bounded(format!("e{i}"), entity, 4);
            assert!(expected, "an entity above a full cache's tail is kept");
        }
        // e7 is the tail of [e10, e9, e8, e7].
        for (key, score, kept) in [("e0", 0, false), ("e6", 6, false), ("e11", 11, true)] {
            let entity = json!({"score": score});
            assert_eq!(cache.would_keep(key, &entity, 4), kept, "{key}");
            let mut copy = SortedViewCache::new(
                "test/top".to_string(),
                vec!["score".to_string()],
                SortOrder::Desc,
            );
            for existing in keys(&mut cache) {
                let value = cache.get(&existing).unwrap().clone();
                copy.upsert_bounded(existing, value, 4);
            }
            copy.upsert_bounded(key.to_string(), entity, 4);
            assert_eq!(copy.get(key).is_some(), kept, "{key}");
        }
        // A held entity is always kept, even when its new value sorts last.
        assert!(cache.would_keep("e8", &json!({"score": -1}), 4));
    }

    #[test]
    fn bounded_upsert_evicts_bottom_of_asc_order() {
        let mut cache = SortedViewCache::new(
            "test/bottom".to_string(),
            vec!["score".to_string()],
            SortOrder::Asc,
        );

        for i in (1..=10).rev() {
            cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 4);
            assert!(cache.len() <= 4);
        }

        assert_eq!(cache.len(), 4);
        assert_eq!(keys(&mut cache), ["e1", "e2", "e3", "e4"]);
        assert!(cache.get("e10").is_none());
    }

    #[test]
    fn bounded_upsert_does_not_evict_stale_top_entities() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        // The leader is inserted first and never updated again; recency-based
        // eviction would drop it.
        cache.upsert_bounded("leader".to_string(), json!({"score": 1_000}), 3);
        for i in 1..=20 {
            cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
        }

        assert_eq!(keys(&mut cache), ["leader", "e20", "e19"]);
    }

    #[test]
    fn windows_within_cap_match_unbounded_cache() {
        let mut bounded = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );
        let mut unbounded = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        // Scores arrive out of order within each round and every entity moves
        // up on each later round. Entities never move down, so nothing
        // evicted can belong back inside the cap (see the edge-case test
        // below for what happens when they do).
        for i in 0..200u64 {
            let key = format!("e{}", i % 60);
            let score = (i / 60) * 1_000 + (i * 37) % 101;
            let entity = json!({"score": score, "n": i});
            bounded.upsert_bounded(key.clone(), entity.clone(), 25);
            unbounded.upsert(key, entity);
        }

        assert_eq!(bounded.len(), 25);
        for (skip, take) in [(0, 25), (0, 10), (5, 20), (24, 1)] {
            assert_eq!(
                bounded.get_window(skip, take),
                unbounded.get_window(skip, take),
                "window skip={skip} take={take}"
            );
        }
    }

    #[test]
    fn evicted_entity_is_missing_after_top_moves_down_until_it_updates() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        for i in 1..=4 {
            cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
        }
        assert_eq!(keys(&mut cache), ["e4", "e3", "e2"]);

        // The leader drops to the bottom; e1 would now rank third but was
        // evicted and has not updated, so e4 holds third place instead.
        cache.upsert_bounded("e4".to_string(), json!({"score": 0}), 3);
        assert_eq!(keys(&mut cache), ["e3", "e2", "e4"]);

        // Once e1 updates it re-enters at its correct position.
        cache.upsert_bounded("e1".to_string(), json!({"score": 1}), 3);
        assert_eq!(keys(&mut cache), ["e3", "e2", "e1"]);
    }

    #[test]
    fn evicted_entity_reenters_when_upserted_again() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        for i in 1..=5 {
            cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
        }
        assert!(cache.get("e1").is_none());

        let result = cache.upsert_bounded("e1".to_string(), json!({"score": 100}), 3);
        assert_eq!(result, UpsertResult::Inserted { position: 0 });
        assert_eq!(keys(&mut cache), ["e1", "e5", "e4"]);
        assert_eq!(cache.len(), 3);
    }

    #[test]
    fn upsert_below_full_cap_is_evicted_immediately() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        for i in 10..=12 {
            cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
        }
        let result = cache.upsert_bounded("low".to_string(), json!({"score": 1}), 3);

        assert_eq!(result, UpsertResult::Inserted { position: 3 });
        assert!(cache.get("low").is_none());
        assert_eq!(keys(&mut cache), ["e12", "e11", "e10"]);
    }

    #[test]
    fn trim_keeps_keys_cache_and_entities_consistent() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );

        for i in 1..=10 {
            cache.upsert(format!("e{i}"), json!({"score": i}));
        }
        // Build the keys cache so the trim takes the in-place truncate path.
        assert_eq!(cache.ordered_keys().len(), 10);

        assert_eq!(cache.trim_to_max_entries(4), 6);
        assert_eq!(cache.trim_to_max_entries(4), 0);
        assert_eq!(cache.len(), 4);
        assert_eq!(keys(&mut cache), ["e10", "e9", "e8", "e7"]);
        assert_eq!(cache.get_all_ordered().len(), 4);
        assert_eq!(cache.remove("e7"), Some(3));
        assert_eq!(keys(&mut cache), ["e10", "e9", "e8"]);
    }

    #[test]
    fn test_nested_sort_field() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string(), "round_id".to_string()],
            SortOrder::Desc,
        );

        cache.upsert("a".to_string(), json!({"id": {"round_id": 1}}));
        cache.upsert("b".to_string(), json!({"id": {"round_id": 3}}));
        cache.upsert("c".to_string(), json!({"id": {"round_id": 2}}));

        let keys = cache.ordered_keys();
        assert_eq!(keys, vec!["b", "c", "a"]);
    }

    #[test]
    fn test_nested_decimal_string_sort_field() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string(), "round_id".to_string()],
            SortOrder::Desc,
        );

        cache.upsert("9".to_string(), json!({"id": {"round_id": "9"}}));
        cache.upsert("100".to_string(), json!({"id": {"round_id": "100"}}));
        cache.upsert("10".to_string(), json!({"id": {"round_id": "10"}}));

        assert_eq!(cache.ordered_keys(), vec!["100", "10", "9"]);
    }

    #[test]
    fn test_descending_string_sort_field() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["name".to_string()],
            SortOrder::Desc,
        );

        cache.upsert("a".to_string(), json!({"name": "alpha"}));
        cache.upsert("c".to_string(), json!({"name": "charlie"}));
        cache.upsert("b".to_string(), json!({"name": "bravo"}));

        assert_eq!(cache.ordered_keys(), vec!["c", "b", "a"]);
    }

    #[test]
    fn test_update_with_missing_sort_field_preserves_position() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string(), "round_id".to_string()],
            SortOrder::Desc,
        );

        cache.upsert(
            "100".to_string(),
            json!({"id": {"round_id": 100}, "data": "initial"}),
        );
        cache.upsert(
            "200".to_string(),
            json!({"id": {"round_id": 200}, "data": "initial"}),
        );
        cache.upsert(
            "300".to_string(),
            json!({"id": {"round_id": 300}, "data": "initial"}),
        );

        assert_eq!(cache.ordered_keys(), vec!["300", "200", "100"]);

        cache.upsert("200".to_string(), json!({"data": "updated_without_id"}));

        assert_eq!(
            cache.ordered_keys(),
            vec!["300", "200", "100"],
            "Entity 200 should retain its position even when updated without sort field"
        );

        let entity = cache.get("200").unwrap();
        assert_eq!(entity["data"], "updated_without_id");
    }

    #[test]
    fn test_new_entity_with_missing_sort_field_sorts_last() {
        for order in [SortOrder::Desc, SortOrder::Asc] {
            let mut cache = SortedViewCache::new(
                "test/latest".to_string(),
                vec!["id".to_string(), "round_id".to_string()],
                order,
            );

            cache.upsert("100".to_string(), json!({"id": {"round_id": 100}}));
            cache.upsert("200".to_string(), json!({"id": {"round_id": 200}}));
            cache.upsert("new".to_string(), json!({"data": "no_sort_field"}));
            cache.upsert("nil".to_string(), json!({"id": {"round_id": null}}));

            let ranked = match order {
                SortOrder::Desc => ["200", "100"],
                SortOrder::Asc => ["100", "200"],
            };
            assert_eq!(
                cache.ordered_keys(),
                [ranked[0], ranked[1], "new", "nil"],
                "{order:?}: an entity without a sort value never outranks one with it; \
                 among themselves they order by key"
            );
        }
    }

    /// The live `latest` failure: a window full of ranked entities must not
    /// admit an unranked one, since it would evict a real member.
    #[test]
    fn a_full_window_does_not_admit_an_entity_without_a_sort_value() {
        let mut cache = SortedViewCache::new(
            "test/latest".to_string(),
            vec!["id".to_string(), "round_id".to_string()],
            SortOrder::Desc,
        );
        for round in [100, 200, 300] {
            cache.upsert_bounded(round.to_string(), json!({"id": {"round_id": round}}), 3);
        }
        let partial = json!({"metrics": {"checkpoint_count": 7}});
        assert!(!cache.would_keep("1", &partial, 3));
        cache.upsert_bounded("1".to_string(), partial, 3);
        assert_eq!(cache.ordered_keys(), ["300", "200", "100"]);
    }

    /// A whole entity, holding every field the cached copy has, is kept as
    /// given: the cache shares its fields instead of copying them.
    #[test]
    fn a_whole_update_shares_the_given_entity() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );
        cache.upsert(
            "a".to_string(),
            json!({"score": 1, "name": "a", "_version": "e:1"}),
        );
        let update =
            SharedEntity::new(json!({"score": 2, "name": "a", "extra": true, "_version": "e:2"}));
        cache.upsert("a".to_string(), update.clone());
        let held = cache.get("a").unwrap();
        assert!(held.shares_fields_with(&update));
        assert_eq!(*held, update);
    }

    /// An update that lacks a field the cached copy has keeps that field, in
    /// a copy of its own: the entity it was given is left as it was.
    #[test]
    fn an_update_lacking_held_fields_merges_into_its_own_copy() {
        let mut cache = SortedViewCache::new(
            "test/top".to_string(),
            vec!["score".to_string()],
            SortOrder::Desc,
        );
        cache.upsert(
            "a".to_string(),
            json!({"score": 1, "nested": {"x": 1, "y": 1}, "_version": "e:1"}),
        );
        let update = SharedEntity::new(json!({"score": 2, "nested": {"x": 2}}));
        cache.upsert("a".to_string(), update.clone());
        let held = cache.get("a").unwrap();
        assert!(!held.shares_fields_with(&update));
        assert_eq!(
            held.to_value(),
            json!({"score": 2, "nested": {"x": 2, "y": 1}, "_version": "e:1"})
        );
        assert_eq!(update.to_value(), json!({"score": 2, "nested": {"x": 2}}));
    }

    /// `_version` sorts like any other field, though copies keep it apart
    /// from the fields they share.
    #[test]
    fn a_version_sort_reads_each_copys_own_version() {
        let mut cache = SortedViewCache::new(
            "test/versions".to_string(),
            vec!["_version".to_string()],
            SortOrder::Desc,
        );
        let first = SharedEntity::new(json!({"id": 1, "_version": "e:1"}));
        let (fields, _) = first.clone().into_parts();
        cache.upsert("a".to_string(), first);
        cache.upsert(
            "b".to_string(),
            SharedEntity::from_parts(fields, Some(json!("e:2"))),
        );
        assert_eq!(cache.ordered_keys(), ["b", "a"]);
    }
}