json_eval_rs/jsoneval/

eval_cache.rs

use indexmap::IndexSet;
use serde_json::Value;
use std::collections::HashMap;

/// Token-version tracker for json paths
#[derive(Default, Clone)]
pub struct VersionTracker {
    versions: HashMap<String, u64>,
}

impl VersionTracker {
    pub fn new() -> Self {
        Self {
            versions: HashMap::new(),
        }
    }

    #[inline]
    pub fn get(&self, path: &str) -> u64 {
        self.versions.get(path).copied().unwrap_or(0)
    }

    #[inline]
    pub fn bump(&mut self, path: &str, source: &str) {
        let current = self.get(path);
        if crate::utils::is_debug_cache_enabled() {
            println!(
                "[store_cache] BUMPING for {} -> {} ({})",
                path,
                current + 1,
                source
            );
        }
        // We use actual data pointers here
        self.versions.insert(path.to_string(), current + 1);
    }

    /// Merge version counters from `other`, taking the **maximum** for each path.
    /// Using max (not insert) ensures that if this tracker already saw a higher version
    /// for a path (e.g., from a previous subform evaluation round), it is never downgraded.
    pub fn merge_from(&mut self, other: &VersionTracker) {
        for (k, v) in &other.versions {
            let current = self.versions.get(k).copied().unwrap_or(0);
            self.versions.insert(k.clone(), current.max(*v));
        }
    }

    /// Merge only `/$params`-prefixed version counters from `other` (max strategy).
    /// Used when giving a per-item tracker the latest schema-level param versions
    /// without absorbing data-path bumps that belong to other items.
    pub fn merge_from_params(&mut self, other: &VersionTracker) {
        for (k, v) in &other.versions {
            if k.starts_with("/$params") {
                let current = self.versions.get(k).copied().unwrap_or(0);
                self.versions.insert(k.clone(), current.max(*v));
            }
        }
    }

    /// Merge counters except paths local to a different active subform item.
    pub(crate) fn merge_excluding_prefix(&mut self, other: &VersionTracker, excluded_prefix: &str) {
        for (k, v) in &other.versions {
            if !k.starts_with(excluded_prefix) {
                let current = self.versions.get(k).copied().unwrap_or(0);
                self.versions.insert(k.clone(), current.max(*v));
            }
        }
    }

    /// Returns true if any tracked path with the given prefix has been bumped (version > 0).
    /// Used to gate table re-evaluation when item fields change without the item being new.
    pub fn any_bumped_with_prefix(&self, prefix: &str) -> bool {
        self.versions
            .iter()
            .any(|(k, &v)| k.starts_with(prefix) && v > 0)
    }

    /// Returns true if any path with the given prefix has a **higher** version than in `baseline`.
    /// Unlike `any_bumped_with_prefix`, this detects only brand-new bumps from a specific diff
    /// pass, ignoring historical bumps that were already present in the baseline.
    pub fn any_newly_bumped_with_prefix(&self, prefix: &str, baseline: &VersionTracker) -> bool {
        self.versions
            .iter()
            .any(|(k, &v)| k.starts_with(prefix) && v > baseline.get(k))
    }

    /// Returns an iterator over all (path, version) pairs, for targeted bump enumeration.
    pub fn versions(&self) -> impl Iterator<Item = (&str, &u64)> {
        self.versions.iter().map(|(k, v)| (k.as_str(), v))
    }
}

/// A cached evaluation result with the specific dependency versions it was evaluated against
#[derive(Clone)]
pub struct CacheEntry {
    pub dep_versions: HashMap<String, u64>,
    pub result: std::sync::Arc<Value>,
    /// The `active_item_index` this entry was computed under.
    /// `None` = computed during main-form evaluation (safe to reuse across all items
    /// provided the dep versions match). `Some(idx)` = computed for a specific item;
    /// Tier-2 reuse is restricted to entries whose deps are entirely `$params`-scoped.
    pub computed_for_item: Option<usize>,
}

/// Independent cache state for a single item in a subform array
#[derive(Default, Clone)]
pub struct SubformItemCache {
    pub data_versions: VersionTracker,
    pub entries: HashMap<String, CacheEntry>,
    pub item_snapshot: Value,
    /// Per-item snapshot of the evaluated schema captured after each evaluate_subform_item.
    /// Allows get_evaluated_schema_subform to return the correct per-item values without
    /// re-running the full evaluation pipeline in a shared subform context.
    pub evaluated_schema: Option<Value>,
}

impl SubformItemCache {
    pub fn new() -> Self {
        Self {
            data_versions: VersionTracker::new(),
            entries: HashMap::new(),
            item_snapshot: Value::Null,
            evaluated_schema: None,
        }
    }
}

/// Primary cache structure for a JSON evaluation instance
#[derive(Clone)]
pub struct EvalCache {
    pub data_versions: VersionTracker,
    pub params_versions: VersionTracker,
    pub entries: HashMap<String, CacheEntry>,

    pub active_item_index: Option<usize>,
    pub subform_caches: HashMap<usize, SubformItemCache>,

    /// Monotonically increasing counter bumped whenever data_versions or params_versions change.
    /// When `eval_generation == last_evaluated_generation`, all cache entries are guaranteed valid
    /// and `evaluate_internal` can skip the full tree traversal.
    pub eval_generation: u64,
    pub last_evaluated_generation: u64,

    /// Snapshot of the last fully-diffed main-form data payload.
    /// Stored after each successful `evaluate_internal_with_new_data` call so the next
    /// invocation can avoid an extra `snapshot_data_clone()` when computing the diff.
    pub main_form_snapshot: Option<std::sync::Arc<Value>>,
    pub subform_roots: Vec<String>,
}

impl Default for EvalCache {
    fn default() -> Self {
        Self::new()
    }
}

impl EvalCache {
    pub fn new() -> Self {
        Self {
            data_versions: VersionTracker::new(),
            params_versions: VersionTracker::new(),
            entries: HashMap::new(),
            active_item_index: None,
            subform_caches: HashMap::new(),
            eval_generation: 0,
            last_evaluated_generation: u64::MAX, // force first evaluate_internal to run
            main_form_snapshot: None,
            subform_roots: Vec::new(),
        }
    }

    pub fn is_subform_dep(&self, path: &str) -> bool {
        self.subform_roots
            .iter()
            .any(|root| path == root || path.starts_with(&format!("{}/", root)))
    }

    pub fn clear(&mut self) {
        self.data_versions = VersionTracker::new();
        self.params_versions = VersionTracker::new();
        self.entries.clear();
        self.active_item_index = None;
        self.subform_caches.clear();
        self.eval_generation = 0;
        self.last_evaluated_generation = u64::MAX;
        self.main_form_snapshot = None;
    }

    /// Remove item caches for indices >= `current_count`.
    /// Call this whenever the subform array length is known to have shrunk so that
    /// stale per-item version trackers and cached entries do not linger in memory.
    pub fn prune_subform_caches(&mut self, current_count: usize) {
        self.subform_caches.retain(|&idx, _| idx < current_count);
    }

    /// Invalidate all `$params`-scoped table cache entries for a specific item.
    ///
    /// Called when a brand-new subform item is introduced so that `$params` tables
    /// that aggregate array data (e.g. aggregate tables) are forced to recompute instead
    /// of returning stale results cached from a prior main-form evaluation that ran
    /// when the item was absent (and thus saw zero/null for that item's values).
    pub fn invalidate_params_tables_for_item(&mut self, idx: usize, table_keys: &[String]) {
        // Bump params_versions so T2 global entries for these tables are stale.
        for key in table_keys {
            let data_path = crate::jsoneval::path_utils::schema_path_to_data_pointer(key);
            self.params_versions
                .bump(&data_path, "invalidate_params_tables_for_item");
            self.eval_generation += 1;
        }

        // Evict matching T1 (item-level) entries so they are not reused.
        if let Some(item_cache) = self.subform_caches.get_mut(&idx) {
            for key in table_keys {
                item_cache.entries.remove(key);
            }
        }
    }

    /// Returns true if evaluate_internal must run (versions changed since last full evaluation)
    pub fn needs_full_evaluation(&self) -> bool {
        self.eval_generation != self.last_evaluated_generation
    }

    /// Call after evaluate_internal completes successfully to mark the generation stable
    pub fn mark_evaluated(&mut self) {
        self.last_evaluated_generation = self.eval_generation;
    }

    pub(crate) fn ensure_active_item_cache(&mut self, idx: usize) {
        self.subform_caches
            .entry(idx)
            .or_insert_with(SubformItemCache::new);
    }

    pub fn set_active_item(&mut self, idx: usize) {
        self.active_item_index = Some(idx);
        self.ensure_active_item_cache(idx);
    }

    pub fn clear_active_item(&mut self) {
        self.active_item_index = None;
    }

    /// Recursively diffs `old` against `new` and bumps version for every changed data path scalar.
    pub fn store_snapshot_and_diff_versions(&mut self, old: &Value, new: &Value) {
        if let Some(idx) = self.active_item_index {
            self.ensure_active_item_cache(idx);
            let sub_cache = self.subform_caches.get_mut(&idx).unwrap();
            diff_and_update_versions(
                &mut sub_cache.data_versions,
                "",
                old,
                new,
                "subform store_snapshot_and_diff_versions",
            );
            sub_cache.item_snapshot = new.clone();
        } else {
            diff_and_update_versions(
                &mut self.data_versions,
                "",
                old,
                new,
                "store_snapshot_and_diff_versions",
            );
        }
    }

    pub fn get_active_snapshot(&self) -> Value {
        if let Some(idx) = self.active_item_index {
            self.subform_caches
                .get(&idx)
                .map(|c| c.item_snapshot.clone())
                .unwrap_or(Value::Null)
        } else {
            Value::Null
        }
    }

    pub fn diff_active_item(
        &mut self,
        field_key: &str,
        old_sub_data: &Value,
        new_sub_data: &Value,
    ) {
        if let Some(idx) = self.active_item_index {
            self.ensure_active_item_cache(idx);
            let sub_cache = self.subform_caches.get_mut(&idx).unwrap();

            // Diff ONLY the localized item part, skipping the massive parent tree
            let empty = Value::Null;
            let old_item = old_sub_data.get(field_key).unwrap_or(&empty);
            let new_item = new_sub_data.get(field_key).unwrap_or(&empty);

            diff_and_update_versions(
                &mut sub_cache.data_versions,
                &format!("/{}", field_key),
                old_item,
                new_item,
                format!("diff_active_item {}", field_key).as_str(),
            );
            sub_cache.item_snapshot = new_sub_data.clone();
        }
    }

    pub fn bump_data_version(&mut self, data_path: &str) {
        // Always signal that something changed so the parent's needs_full_evaluation()
        // returns true even when the bump was item-scoped.
        self.eval_generation += 1;
        if let Some(idx) = self.active_item_index {
            if let Some(cache) = self.subform_caches.get_mut(&idx) {
                cache.data_versions.bump(data_path, "bump_data_version1");
            }
        } else {
            self.data_versions.bump(data_path, "bump_data_version2");
        }
    }

    pub fn bump_params_version(&mut self, data_path: &str) {
        self.params_versions.bump(data_path, "bump_params_version");
        self.eval_generation += 1;
    }

    /// Check if the `eval_key` result can be safely bypassed because dependencies are unchanged.
    ///
    /// Two-tier lookup:
    /// - Tier 1: item-scoped entries in `subform_caches[idx]` — checked first when an active item is set
    /// - Tier 2: global `self.entries` — allows Run 1 (main form) results to be reused in Run 2 (subform)
    pub fn check_cache(&self, eval_key: &str, deps: &IndexSet<String>) -> Option<Value> {
        self.check_cache_arc(eval_key, deps)
            .map(|arc| (*arc).clone())
    }

    pub fn check_cache_arc(
        &self,
        eval_key: &str,
        deps: &IndexSet<String>,
    ) -> Option<std::sync::Arc<Value>> {
        if let Some(idx) = self.active_item_index {
            // Tier 1: item-specific entries (always safe to reuse for the same index)
            if let Some(cache) = self.subform_caches.get(&idx) {
                if let Some(hit) =
                    self.validate_entry(eval_key, deps, &cache.entries, &cache.data_versions)
                {
                    if crate::utils::is_debug_cache_enabled() {
                        println!("Cache HIT [T1 idx={}] {}", idx, eval_key);
                    }
                    return Some(hit);
                }
            }

            // Reuse only index-safe T2 entries.
            let item_data_versions = self
                .subform_caches
                .get(&idx)
                .map(|c| &c.data_versions)
                .unwrap_or(&self.data_versions);

            if let Some(entry) = self.entries.get(eval_key) {
                let depends_on_subform_item =
                    entry.dep_versions.keys().any(|p| self.is_subform_dep(p));
                if depends_on_subform_item && entry.computed_for_item != Some(idx) {
                    return None;
                }

                let has_changed_item_dep = self.subform_caches.get(&idx).is_some_and(|cache| {
                    entry.dep_versions.keys().any(|p| {
                        if p.starts_with("/$params") {
                            false
                        } else {
                            cache.data_versions.get(p) > self.data_versions.get(p)
                        }
                    })
                });
                let index_safe = match entry.computed_for_item {
                    None => !has_changed_item_dep,
                    Some(stored_idx) if stored_idx == idx => true,
                    _ => !has_changed_item_dep,
                };
                if index_safe {
                    let result =
                        self.validate_entry(eval_key, deps, &self.entries, item_data_versions);
                    if result.is_some() {
                        if crate::utils::is_debug_cache_enabled() {
                            println!(
                                "Cache HIT [T2 idx={} for={:?}] {}",
                                idx, entry.computed_for_item, eval_key
                            );
                        }
                    }
                    return result;
                }
            }

            None
        } else {
            self.validate_entry(eval_key, deps, &self.entries, &self.data_versions)
        }
    }

    /// Specialized cache check for `$params`-scoped table evaluations.
    ///
    /// Checks global cache for `$params` tables.
    pub fn check_table_cache(
        &self,
        eval_key: &str,
        deps: &IndexSet<String>,
    ) -> Option<std::sync::Arc<Value>> {
        if let Some(idx) = self.active_item_index {
            // Tier 1: item-scoped entries first (unlikely for $params tables but check anyway)
            if let Some(cache) = self.subform_caches.get(&idx) {
                if let Some(hit) =
                    self.validate_entry(eval_key, deps, &cache.entries, &cache.data_versions)
                {
                    if crate::utils::is_debug_cache_enabled() {
                        println!("Cache HIT [T1 table idx={}] {}", idx, eval_key);
                    }
                    return Some(hit);
                }
            }

            // If the table has an item dependency that was bumped for this active item,
            // the active item must not reuse the global T2 table.
            let has_changed_item_dep = self.subform_caches.get(&idx).is_some_and(|cache| {
                if let Some(entry) = self.entries.get(eval_key) {
                    entry.dep_versions.keys().any(|p| {
                        if p.starts_with("/$params") {
                            false
                        } else {
                            cache.data_versions.get(p) > self.data_versions.get(p)
                        }
                    })
                } else {
                    deps.iter().any(|dep| {
                        let p = crate::jsoneval::path_utils::schema_path_to_data_pointer(dep);
                        if p.starts_with("/$params") {
                            false
                        } else {
                            cache.data_versions.get(&p) > self.data_versions.get(&p)
                        }
                    })
                }
            });
            if has_changed_item_dep {
                return None;
            }

            let result = self.validate_entry(eval_key, deps, &self.entries, &self.data_versions);
            if result.is_some() {
                if crate::utils::is_debug_cache_enabled() {
                    println!("Cache HIT [T2 table idx={}] {}", idx, eval_key);
                }
            }
            result
        } else {
            self.validate_entry(eval_key, deps, &self.entries, &self.data_versions)
        }
    }

    fn validate_entry(
        &self,
        eval_key: &str,
        _deps: &IndexSet<String>,
        entries: &HashMap<String, CacheEntry>,
        data_versions: &VersionTracker,
    ) -> Option<std::sync::Arc<Value>> {
        let entry = entries.get(eval_key)?;
        for (data_dep_path, &cached_ver) in &entry.dep_versions {
            let current_ver = if data_dep_path.starts_with("/$params") {
                self.params_versions.get(data_dep_path)
            } else if let Some(idx) = self.active_item_index {
                self.subform_caches
                    .get(&idx)
                    .map(|c| c.data_versions.get(data_dep_path))
                    .filter(|&v| v > 0)
                    .unwrap_or_else(|| data_versions.get(data_dep_path))
            } else {
                data_versions.get(data_dep_path)
            };

            if current_ver != cached_ver {
                if crate::utils::is_debug_cache_enabled() {
                    println!(
                        "Cache MISS {}: dep {} changed ({} -> {})",
                        eval_key, data_dep_path, cached_ver, current_ver
                    );
                }
                return None;
            }
        }
        if crate::utils::is_debug_cache_enabled() {
            println!("Cache HIT {}", eval_key);
        }
        Some(std::sync::Arc::clone(&entry.result))
    }

    /// Store the newly evaluated value and snapshot the dependency versions.
    ///
    /// Stores result in active cache tier.
    pub fn store_cache(&mut self, eval_key: &str, deps: &IndexSet<String>, result: Value) {
        self.store_cache_arc(eval_key, deps, std::sync::Arc::new(result));
    }

    /// Store the newly evaluated value and snapshot the dependency versions (zero-copy Arc).
    pub fn store_cache_arc(
        &mut self,
        eval_key: &str,
        deps: &IndexSet<String>,
        result: std::sync::Arc<Value>,
    ) {
        // Snapshot dependency versions.
        let mut dep_versions = HashMap::with_capacity(deps.len());
        {
            let data_versions = if let Some(idx) = self.active_item_index {
                self.ensure_active_item_cache(idx);
                &self.subform_caches[&idx].data_versions
            } else {
                &self.data_versions
            };

            for dep in deps {
                let data_dep_path = crate::jsoneval::path_utils::schema_path_to_data_pointer(dep);
                let ver = if data_dep_path.starts_with("/$params") {
                    self.params_versions.get(&data_dep_path)
                } else {
                    data_versions.get(&data_dep_path)
                };
                dep_versions.insert(data_dep_path.into_owned(), ver);
            }
        }

        // Store with current item scope.
        let computed_for_item = self.active_item_index;

        // Bump $params versions only when result changes.
        if eval_key.starts_with("#/$params") {
            let existing_result: Option<&Value> = if let Some(idx) = self.active_item_index {
                // Prefer canonical T2 result.
                self.entries
                    .get(eval_key)
                    .map(|e| e.result.as_ref())
                    .or_else(|| {
                        self.subform_caches
                            .get(&idx)
                            .and_then(|c| c.entries.get(eval_key))
                            .map(|e| e.result.as_ref())
                    })
            } else {
                self.entries.get(eval_key).map(|e| e.result.as_ref())
            };

            let value_changed = existing_result.map_or(true, |r| r != result.as_ref());

            if value_changed {
                let data_path = crate::jsoneval::path_utils::schema_path_to_data_pointer(eval_key);

                // Bump the explicit path and its table-level parent.
                // Stop at slash_count < 3 — never bump /$params/others or /$params itself.
                let mut current_path = data_path.as_ref();
                let mut slash_count = current_path.matches('/').count();

                while slash_count >= 3 {
                    self.params_versions.bump(current_path, "store_cache");
                    if let Some(last_slash) = current_path.rfind('/') {
                        current_path = &current_path[..last_slash];
                        slash_count -= 1;
                    } else {
                        break;
                    }
                }

                self.eval_generation += 1;
            }
        }

        let entry = CacheEntry {
            dep_versions,
            result: std::sync::Arc::clone(&result),
            computed_for_item,
        };

        if let Some(idx) = self.active_item_index {
            // Store item-scoped: isolates per-rider entries so riders with different data don't collide
            self.subform_caches
                .get_mut(&idx)
                .unwrap()
                .entries
                .insert(eval_key.to_string(), entry.clone());

            // Promote $params tables to T2 with parent versions.
            if eval_key.starts_with("#/$params") {
                let t2_dep_versions: HashMap<String, u64> = entry
                    .dep_versions
                    .iter()
                    .map(|(path, &item_ver)| {
                        let parent_ver = if path.starts_with("/$params") {
                            item_ver // params_versions are global — same for both
                        } else {
                            // Use parent version.
                            self.data_versions.get(path)
                        };
                        (path.clone(), parent_ver)
                    })
                    .collect();

                let t2_entry = CacheEntry {
                    dep_versions: t2_dep_versions,
                    result: std::sync::Arc::clone(&entry.result),
                    computed_for_item,
                };
                self.entries.insert(eval_key.to_string(), t2_entry);
            }
        } else {
            self.entries.insert(eval_key.to_string(), entry);
        }
    }
}

/// Recursive helper to walk JSON structures and bump specific leaf versions where they differ
pub(crate) fn diff_and_update_versions(
    tracker: &mut VersionTracker,
    pointer: &str,
    old: &Value,
    new: &Value,
    source: &str,
) {
    let mut pointer_buf = String::with_capacity(128);
    pointer_buf.push_str(pointer);
    diff_and_update_versions_internal(tracker, &mut pointer_buf, old, new, source);
}

#[cfg(test)]
mod cache_tests {
    use super::{CacheEntry, EvalCache};
    use indexmap::IndexSet;
    use serde_json::json;
    use std::collections::HashMap;
    use std::sync::Arc;

    #[test]
    fn unchanged_active_item_reuses_global_table_with_item_dependency() {
        let mut cache = EvalCache::new();
        cache.set_active_item(1);

        let eval_key = "#/$params/references/LOOKUP_RATE";
        let deps = IndexSet::from_iter(["#/items/properties/benefit".to_string()]);
        cache.entries.insert(
            eval_key.to_string(),
            CacheEntry {
                dep_versions: HashMap::from([("/items/benefit".to_string(), 0)]),
                result: Arc::new(json!([{"rate": 97}])),
                computed_for_item: None,
            },
        );

        assert_eq!(
            cache.check_table_cache(eval_key, &deps),
            Some(Arc::new(json!([{"rate": 97}]))),
            "a scoped alias may reuse the parent result for its unchanged canonical item"
        );
    }

    #[test]
    fn changed_active_item_does_not_reuse_global_table_with_item_dependency() {
        let mut cache = EvalCache::new();
        cache.set_active_item(1);
        cache
            .subform_caches
            .get_mut(&1)
            .expect("active item cache must exist")
            .data_versions
            .bump("/items/benefit", "test item input change");

        let eval_key = "#/$params/references/LOOKUP_RATE";
        let deps = IndexSet::from_iter(["#/items/properties/benefit".to_string()]);
        cache.entries.insert(
            eval_key.to_string(),
            CacheEntry {
                dep_versions: HashMap::from([("/items/benefit".to_string(), 0)]),
                result: Arc::new(json!([{"rate": 97}])),
                computed_for_item: None,
            },
        );

        assert!(
            cache.check_table_cache(eval_key, &deps).is_none(),
            "a changed item input must force item-scoped table recomputation"
        );
    }

    #[test]
    fn active_item_reuses_global_table_with_only_params_dependencies() {
        let mut cache = EvalCache::new();
        cache.set_active_item(1);

        let eval_key = "#/$params/references/SHARED_RATE";
        let deps = IndexSet::from_iter(["#/$params/others/currency".to_string()]);
        cache.entries.insert(
            eval_key.to_string(),
            CacheEntry {
                dep_versions: HashMap::from([("/$params/others/currency".to_string(), 0)]),
                result: Arc::new(json!([{"rate": 10}])),
                computed_for_item: None,
            },
        );

        assert_eq!(
            cache.check_table_cache(eval_key, &deps),
            Some(Arc::new(json!([{"rate": 10}])))
        );
    }
}

fn diff_and_update_versions_internal(
    tracker: &mut VersionTracker,
    pointer: &mut String,
    old: &Value,
    new: &Value,
    source: &str,
) {
    if old == new {
        return;
    }

    if crate::utils::is_debug_cache_enabled() {
        println!(
            "[diff_and_update_versions_internal] {} pointer={}, old={:?}, new={:?}",
            source, pointer, old, new
        );
    }

    match (old, new) {
        (Value::Object(a), Value::Object(b)) => {
            for (key, a_val) in a {
                if key == "$params" {
                    continue;
                }
                let b_val = b.get(key).unwrap_or(&Value::Null);
                if a_val == b_val {
                    continue;
                }

                let old_len = pointer.len();
                pointer.push('/');
                if key.contains('~') || key.contains('/') {
                    let escaped_key = key.replace('~', "~0").replace('/', "~1");
                    pointer.push_str(&escaped_key);
                } else {
                    pointer.push_str(key);
                }
                diff_and_update_versions_internal(tracker, pointer, a_val, b_val, source);
                pointer.truncate(old_len);
            }

            for (key, b_val) in b {
                if key == "$params" || a.contains_key(key) {
                    continue;
                }
                if b_val.is_null() {
                    continue;
                }

                let old_len = pointer.len();
                pointer.push('/');
                if key.contains('~') || key.contains('/') {
                    let escaped_key = key.replace('~', "~0").replace('/', "~1");
                    pointer.push_str(&escaped_key);
                } else {
                    pointer.push_str(key);
                }
                diff_and_update_versions_internal(tracker, pointer, &Value::Null, b_val, source);
                pointer.truncate(old_len);
            }
        }
        (Value::Array(a), Value::Array(b)) => {
            if a != b {
                tracker.bump(pointer, source);
            }
            let max_len = a.len().max(b.len());
            for i in 0..max_len {
                let a_val = a.get(i).unwrap_or(&Value::Null);
                let b_val = b.get(i).unwrap_or(&Value::Null);
                if a_val == b_val {
                    continue;
                }
                let old_len = pointer.len();
                use std::fmt::Write;
                write!(pointer, "/{}", i).unwrap();
                diff_and_update_versions_internal(tracker, pointer, a_val, b_val, source);
                pointer.truncate(old_len);
            }
        }
        (old_val, new_val) => {
            if old_val != new_val {
                if crate::utils::is_debug_cache_enabled() {
                    println!(
                        "[store_cache] Catch-all for {}: old={}, new={}",
                        pointer,
                        match old_val {
                            Value::Null => "Null",
                            Value::Bool(_) => "Bool",
                            Value::Number(_) => "Number",
                            Value::String(_) => "String",
                            Value::Array(_) => "Array",
                            Value::Object(_) => "Object",
                        },
                        match new_val {
                            Value::Null => "Null",
                            Value::Bool(_) => "Bool",
                            Value::Number(_) => "Number",
                            Value::String(_) => "String",
                            Value::Array(_) => "Array",
                            Value::Object(_) => "Object",
                        }
                    );
                }
                tracker.bump(pointer, "diff_and_update_versions_internal");

                // If either side contains nested structures (e.g. Object replaced by Null, or vice versa)
                // we must recursively bump all paths inside them so targeted cache entries invalidate.
                if old_val.is_object() || old_val.is_array() {
                    traverse_and_bump(tracker, pointer, old_val);
                }
                if new_val.is_object() || new_val.is_array() {
                    traverse_and_bump(tracker, pointer, new_val);
                }
            }
        }
    }
}

/// Recursively traverses a value and bumps the version for every nested path.
/// Used when a structural type mismatch occurs (e.g., Object -> Null) so that
/// cache entries depending on nested fields are correctly invalidated.
#[cfg(test)]
mod tests {
    use super::VersionTracker;

    #[test]
    fn merge_excluding_prefix_keeps_item_versions_isolated() {
        let mut item = VersionTracker::new();
        item.bump("/items/flag", "test");

        let mut parent = VersionTracker::new();
        parent.bump("/users/profile/name", "test");
        parent.bump("/items/flag", "test");

        item.merge_excluding_prefix(&parent, "/items/");

        assert_eq!(item.get("/users/profile/name"), 1);
        assert_eq!(
            item.get("/items/flag"),
            1,
            "another item's parent-tracker bump must not alter this item's version"
        );
    }
}

fn traverse_and_bump(tracker: &mut VersionTracker, pointer: &mut String, val: &Value) {
    match val {
        Value::Object(map) => {
            for (key, v) in map {
                if key == "$params" {
                    continue; // Skip the special top-level params branch if it leaked here
                }
                let escaped_key = key.replace('~', "~0").replace('/', "~1");
                let old_len = pointer.len();
                pointer.push('/');
                pointer.push_str(&escaped_key);
                tracker.bump(pointer, "traverse_and_bump1");
                traverse_and_bump(tracker, pointer, v);
                pointer.truncate(old_len);
            }
        }
        Value::Array(arr) => {
            for (i, v) in arr.iter().enumerate() {
                let old_len = pointer.len();
                use std::fmt::Write;
                write!(pointer, "/{}", i).unwrap();
                tracker.bump(pointer, "traverse_and_bump2");
                traverse_and_bump(tracker, pointer, v);
                pointer.truncate(old_len);
            }
        }
        _ => {}
    }
}