weirflow 0.1.0

use std::cmp::{Ordering, Reverse};
use std::collections::BinaryHeap;

/// Shared LRU heap for backend-resident graph caches.
///
/// Domain caches own validation, upload, free, and byte accounting. This type
/// owns only the duplicated resident-cache recency heap, stale-entry skipping,
/// and bounded compaction policy.
#[derive(Clone, Debug)]
pub(crate) struct ResidentCacheLru<K> {
    heap: BinaryHeap<Reverse<ResidentCacheLruEntry<K>>>,
    serial: u64,
}

impl<K> PartialEq for ResidentCacheLru<K> {
    fn eq(&self, other: &Self) -> bool {
        self.serial == other.serial && self.heap.len() == other.heap.len()
    }
}

impl<K> Default for ResidentCacheLru<K> {
    fn default() -> Self {
        Self {
            heap: BinaryHeap::new(),
            serial: 0,
        }
    }
}

impl<K> ResidentCacheLru<K>
where
    K: Clone,
{
    /// Number of live and stale heap records currently retained.
    #[must_use]
    #[cfg(test)]
    pub(crate) fn len(&self) -> usize {
        self.heap.len()
    }

    /// Allocated heap capacity retained for hot-path amortization tests.
    #[must_use]
    #[cfg(test)]
    pub(crate) fn capacity(&self) -> usize {
        self.heap.capacity()
    }

    /// Remove every heap record and reset insertion serials.
    pub(crate) fn clear(&mut self) {
        self.heap.clear();
        self.serial = 0;
    }

    /// Record that `key` was observed at `last_seen`.
    pub(crate) fn record<I>(
        &mut self,
        key: K,
        last_seen: u64,
        live_entries: I,
        label: &str,
    ) -> Result<(), String>
    where
        I: ExactSizeIterator<Item = (K, u64)>,
    {
        self.serial = self.serial.checked_add(1).ok_or_else(|| {
            format!(
                "{label} LRU serial overflowed u64. Fix: rebuild the resident cache before reuse."
            )
        })?;
        self.reserve_slot(label)?;
        self.heap.push(Reverse(ResidentCacheLruEntry {
            last_seen,
            serial: self.serial,
            key,
        }));
        self.compact_if_needed(live_entries, label)
    }

    /// Pop the least recently seen key that is still live.
    pub(crate) fn pop_valid<F, G>(&mut self, mut live_last_seen: F, fallback: G) -> Option<K>
    where
        F: FnMut(&K) -> Option<u64>,
        G: FnOnce() -> Option<K>,
    {
        while let Some(Reverse(candidate)) = self.heap.pop() {
            if live_last_seen(&candidate.key)
                .is_some_and(|last_seen| last_seen == candidate.last_seen)
            {
                return Some(candidate.key);
            }
        }
        fallback()
    }

    fn compact_if_needed<I>(&mut self, live_entries: I, label: &str) -> Result<(), String>
    where
        I: ExactSizeIterator<Item = (K, u64)>,
    {
        let live_count = live_entries.len();
        let stale_limit = live_count
            .checked_mul(4)
            .and_then(|value| value.checked_add(32))
            .ok_or_else(|| {
                format!("{label} LRU compaction threshold overflowed usize. Fix: rebuild the resident cache before reuse.")
            })?;
        if self.heap.len() <= stale_limit {
            return Ok(());
        }

        self.heap.clear();
        if self.heap.capacity() < live_count {
            self.heap
                .try_reserve(live_count - self.heap.capacity())
                .map_err(|error| {
                    format!(
                        "{label} LRU compaction could not reserve {live_count} live entry slot(s): {error}. Fix: lower resident cache capacity or shard the cache."
                    )
                })?;
        }
        self.serial = 0;
        for (key, last_seen) in live_entries {
            self.serial = self.serial.checked_add(1).ok_or_else(|| {
                format!("{label} LRU serial overflowed during compaction. Fix: rebuild the resident cache before reuse.")
            })?;
            self.heap.push(Reverse(ResidentCacheLruEntry {
                last_seen,
                serial: self.serial,
                key,
            }));
        }
        Ok(())
    }

    fn reserve_slot(&mut self, label: &str) -> Result<(), String> {
        let needed = self.heap.len().checked_add(1).ok_or_else(|| {
            format!("{label} LRU entry count overflowed usize. Fix: rebuild the resident cache before reuse.")
        })?;
        if self.heap.capacity() >= needed {
            return Ok(());
        }
        let current = self.heap.capacity();
        let target = current
            .max(32)
            .checked_mul(2)
            .map(|doubled| doubled.max(needed))
            .ok_or_else(|| {
                format!("{label} LRU capacity growth overflowed usize. Fix: rebuild the resident cache before reuse.")
            })?;
        self.heap
            .try_reserve(target - current)
            .map_err(|error| {
                format!(
                    "{label} LRU could not reserve capacity for {target} entry slot(s): {error}. Fix: lower resident cache pressure or shard the workload."
                )
            })
    }
}

#[derive(Clone, Debug)]
struct ResidentCacheLruEntry<K> {
    last_seen: u64,
    serial: u64,
    key: K,
}

impl<K> PartialEq for ResidentCacheLruEntry<K> {
    fn eq(&self, other: &Self) -> bool {
        self.last_seen == other.last_seen && self.serial == other.serial
    }
}

impl<K> Eq for ResidentCacheLruEntry<K> {}

impl<K> Ord for ResidentCacheLruEntry<K> {
    fn cmp(&self, other: &Self) -> Ordering {
        self.last_seen
            .cmp(&other.last_seen)
            .then_with(|| self.serial.cmp(&other.serial))
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    #[test]
    fn lru_skips_stale_records_and_returns_oldest_live_key() {
        let mut lru = ResidentCacheLru::default();
        lru.record("a", 1, [("a", 1)].into_iter(), "test cache")
            .unwrap();
        lru.record("a", 2, [("a", 2)].into_iter(), "test cache")
            .unwrap();
        lru.record("b", 3, [("a", 2), ("b", 3)].into_iter(), "test cache")
            .unwrap();

        let live = HashMap::from([("a", 2), ("b", 3)]);
        assert_eq!(
            lru.pop_valid(|key| live.get(key).copied(), || None),
            Some("a")
        );
    }

    #[test]
    fn lru_compacts_stale_records_at_cache_scale() {
        let mut lru = ResidentCacheLru::default();
        for tick in 1..80 {
            lru.record("hot", tick, [("hot", tick)].into_iter(), "test cache")
                .unwrap();
        }
        assert!(
            lru.len() <= 36,
            "single-entry cache should compact stale LRU records instead of growing without bound"
        );
    }

    #[test]
    fn lru_reserves_in_chunks_for_hot_resident_hits() {
        let mut lru = ResidentCacheLru::default();
        lru.record("hot", 1, [("hot", 1)].into_iter(), "test cache")
            .unwrap();
        assert!(
            lru.capacity() >= 64,
            "Fix: resident LRU must reserve amortized chunks, not one slot per hot hit."
        );
    }
}