//! Connects the RSS-based memory_guard to real cache eviction via HomeostasisController.
//!
//! The orchestrator bridges two systems:
//! - `memory_guard`: monitors process RSS and reports PressureLevel (Normal..Critical)
//! - `homeostasis`: decides which eviction action to take based on token utilization
//!
//! On each pressure callback, the orchestrator queries current cache utilization,
//! feeds it to HomeostasisController, and executes the recommended action.

use std::sync::{Arc, Mutex};

use super::bm25_cache::SharedBm25Cache;
use super::cache::SessionCache;
use super::homeostasis::{HomeostasisAction, HomeostasisController};
use super::memory_guard;

type SharedCache = Arc<tokio::sync::RwLock<SessionCache>>;

pub struct EvictionOrchestrator {
    cache: SharedCache,
    bm25_cache: SharedBm25Cache,
    controller: Mutex<HomeostasisController>,
    token_budget: usize,
}

impl EvictionOrchestrator {
    pub fn new(cache: SharedCache, bm25_cache: SharedBm25Cache) -> Self {
        let token_budget = super::cache::max_cache_tokens();
        Self {
            cache,
            bm25_cache,
            controller: Mutex::new(HomeostasisController::new(token_budget)),
            token_budget,
        }
    }

    /// Called by the memory_guard thread when pressure is detected.
    /// Runs on the guardian thread — must not block on async locks for too long.
    pub fn on_pressure(&self, level: memory_guard::PressureLevel) {
        if level == memory_guard::PressureLevel::Normal {
            return;
        }

        let current_tokens = self.try_read_cache_tokens();
        let bm25_bytes = super::bm25_cache::memory_usage(&self.bm25_cache);

        let effective_tokens = if bm25_bytes > 0 {
            current_tokens + bm25_bytes / 4
        } else {
            current_tokens
        };

        let action = {
            let mut ctrl = self
                .controller
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            ctrl.evaluate(effective_tokens)
        };

        // #685: the homeostasis controller reasons over *cache-token*
        // utilization, but RSS pressure can come from structures the token
        // heuristic cannot see (ANN/HNSW graph, resident trigram + graph
        // indices, transient build state). At Hard/Critical RSS the guard's
        // signal must win: enforce a floor action so eviction always reaches
        // the index caches instead of returning `None` because the session
        // cache happens to be small.
        let action = floor_action_for_rss(level, action);

        if action == HomeostasisAction::None {
            return;
        }

        tracing::info!(
            "[eviction] pressure={level:?} tokens={current_tokens}/{} bm25={:.1}MB action={action:?}",
            self.token_budget,
            bm25_bytes as f64 / 1_048_576.0,
        );

        let pressure_reduced = self.execute_action(&action, bm25_bytes);

        let mut ctrl = self
            .controller
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        ctrl.report_outcome(pressure_reduced);
    }

    fn execute_action(&self, action: &HomeostasisAction, bm25_bytes: usize) -> bool {
        match action {
            HomeostasisAction::None => true,

            HomeostasisAction::TrimOutputs => {
                let trimmed = self.try_write_cache(SessionCache::trim_compressed_outputs);
                tracing::info!("[eviction] trimmed compressed outputs from {trimmed} entries");
                trimmed > 0
            }

            HomeostasisAction::EvictProbationary { .. } => {
                let evicted = self.try_write_cache(|cache| {
                    let n = cache.evict_probationary();
                    cache.trim_shared_blocks();
                    n
                });
                tracing::info!("[eviction] evicted {evicted} probationary entries");
                evicted > 0
            }

            HomeostasisAction::UnloadIndices => {
                if bm25_bytes > 0 {
                    super::bm25_cache::unload(&self.bm25_cache);
                }
                let content_freed = super::content_cache::memory_usage_bytes();
                super::content_cache::clear();
                // #685: the ANN/HNSW graph, resident trigram indices and the
                // materialized graph indexes were invisible to eviction — on
                // embedding-heavy sessions they dominated RSS while the
                // guardian could only trim the (small) session cache.
                let ann_freed = super::ann_cache::memory_usage_bytes();
                super::ann_cache::clear();
                super::search_index::clear_resident();
                super::graph_cache::invalidate(None);
                let trimmed = self.try_write_cache(SessionCache::trim_compressed_outputs);
                memory_guard::jemalloc_purge();
                tracing::info!(
                    "[eviction] unloaded indices (bm25={:.1}MB + content={:.1}MB + ann={:.1}MB freed, \
                     search/graph residents dropped, {trimmed} outputs trimmed)",
                    bm25_bytes as f64 / 1_048_576.0,
                    content_freed as f64 / 1_048_576.0,
                    ann_freed as f64 / 1_048_576.0,
                );
                bm25_bytes > 0 || content_freed > 0 || ann_freed > 0 || trimmed > 0
            }

            HomeostasisAction::EvictProtected { target_tokens } => {
                self.try_write_cache(|cache| cache.evict_to_budget(*target_tokens));
                memory_guard::jemalloc_purge();
                tracing::info!(
                    "[eviction] evicted protected entries to budget {target_tokens} tokens"
                );
                true
            }

            HomeostasisAction::EmergencyDrop => {
                let cleared = self.try_write_cache(SessionCache::clear);
                super::bm25_cache::unload(&self.bm25_cache);
                super::content_cache::clear();
                // #685: emergency must reach every resident structure.
                super::ann_cache::clear();
                super::search_index::clear_resident();
                super::graph_cache::invalidate(None);
                memory_guard::jemalloc_purge();
                tracing::warn!(
                    "[eviction] EMERGENCY: cleared {cleared} cache entries + unloaded all indices \
                     (bm25, content, ann, search, graph)"
                );
                true
            }
        }
    }

    /// Try to read current token count without blocking.
    /// Falls back to budget (assumes full) if the lock is contended.
    fn try_read_cache_tokens(&self) -> usize {
        match self.cache.try_read() {
            Ok(guard) => guard.total_cached_tokens(),
            Err(_) => self.token_budget,
        }
    }

    /// Try to write to the cache. Returns default value if lock is contended.
    fn try_write_cache<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&mut SessionCache) -> R,
        R: Default,
    {
        if let Ok(mut guard) = self.cache.try_write() {
            f(&mut guard)
        } else {
            tracing::debug!("[eviction] cache write lock contended, skipping");
            R::default()
        }
    }
}

/// RSS floor (#685): guarantee a minimum eviction strength for a given guard
/// pressure level, regardless of the token-based homeostasis verdict. Token
/// utilization measures only the session cache; the structures that actually
/// blew up in #685 (HNSW graph, resident indices, build transients) are
/// invisible to it — Hard/Critical RSS must always at least unload indices.
fn floor_action_for_rss(
    level: memory_guard::PressureLevel,
    action: HomeostasisAction,
) -> HomeostasisAction {
    let rank = |a: &HomeostasisAction| match a {
        HomeostasisAction::None => 0u8,
        HomeostasisAction::TrimOutputs => 1,
        HomeostasisAction::EvictProbationary { .. } => 2,
        HomeostasisAction::UnloadIndices => 3,
        HomeostasisAction::EvictProtected { .. } => 4,
        HomeostasisAction::EmergencyDrop => 5,
    };
    let floor = match level {
        memory_guard::PressureLevel::Critical => HomeostasisAction::EmergencyDrop,
        memory_guard::PressureLevel::Hard => HomeostasisAction::UnloadIndices,
        // Soft/Medium RSS: trust the graduated token-based controller, but do
        // not let real pressure end in a no-op.
        memory_guard::PressureLevel::Soft | memory_guard::PressureLevel::Medium => {
            HomeostasisAction::TrimOutputs
        }
        memory_guard::PressureLevel::Normal => HomeostasisAction::None,
    };
    if rank(&action) >= rank(&floor) {
        action
    } else {
        floor
    }
}

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

    fn make_orchestrator() -> EvictionOrchestrator {
        let cache = Arc::new(tokio::sync::RwLock::new(SessionCache::new()));
        let bm25_cache: SharedBm25Cache = Arc::new(std::sync::Mutex::new(None));
        EvictionOrchestrator::new(cache, bm25_cache)
    }

    #[test]
    fn normal_pressure_is_noop() {
        let orch = make_orchestrator();
        orch.on_pressure(memory_guard::PressureLevel::Normal);
    }

    #[test]
    fn soft_pressure_with_empty_cache_is_noop() {
        let orch = make_orchestrator();
        orch.on_pressure(memory_guard::PressureLevel::Soft);
    }

    #[test]
    fn emergency_clears_cache() {
        let cache = Arc::new(tokio::sync::RwLock::new(SessionCache::new()));
        {
            let mut c = cache.blocking_write();
            c.store("/a.rs", "fn a() {}");
            c.store("/b.rs", "fn b() {}");
        }
        let bm25: SharedBm25Cache = Arc::new(std::sync::Mutex::new(None));
        let orch = EvictionOrchestrator {
            cache: cache.clone(),
            bm25_cache: bm25,
            controller: Mutex::new(HomeostasisController::new(100)),
            token_budget: 100,
        };

        orch.execute_action(&HomeostasisAction::EmergencyDrop, 0);
        let c = cache.blocking_read();
        assert_eq!(c.total_cached_tokens(), 0);
    }

    #[test]
    fn trim_outputs_clears_compressed() {
        let cache = Arc::new(tokio::sync::RwLock::new(SessionCache::new()));
        {
            let mut c = cache.blocking_write();
            c.store("/a.rs", "fn main() {}");
            c.set_compressed("/a.rs", "map", "compressed map".to_string());
        }
        let bm25: SharedBm25Cache = Arc::new(std::sync::Mutex::new(None));
        let orch = EvictionOrchestrator {
            cache: cache.clone(),
            bm25_cache: bm25,
            controller: Mutex::new(HomeostasisController::new(100_000)),
            token_budget: 100_000,
        };

        let result = orch.execute_action(&HomeostasisAction::TrimOutputs, 0);
        assert!(result);
        let c = cache.blocking_read();
        assert!(c.get_compressed("/a.rs", "map").is_none());
    }

    #[test]
    fn rss_floor_escalates_weak_actions_under_hard_pressure() {
        // Session cache nearly empty → controller says None; Hard RSS must
        // still unload indices (#685: token heuristic blind to index RAM).
        let floored =
            floor_action_for_rss(memory_guard::PressureLevel::Hard, HomeostasisAction::None);
        assert_eq!(floored, HomeostasisAction::UnloadIndices);

        let floored = floor_action_for_rss(
            memory_guard::PressureLevel::Critical,
            HomeostasisAction::TrimOutputs,
        );
        assert_eq!(floored, HomeostasisAction::EmergencyDrop);

        let floored =
            floor_action_for_rss(memory_guard::PressureLevel::Soft, HomeostasisAction::None);
        assert_eq!(floored, HomeostasisAction::TrimOutputs);
    }

    #[test]
    fn rss_floor_keeps_stronger_controller_actions() {
        // The controller may already demand more than the floor — keep it.
        let strong = HomeostasisAction::EvictProtected {
            target_tokens: 1_000,
        };
        let kept = floor_action_for_rss(memory_guard::PressureLevel::Hard, strong.clone());
        assert_eq!(kept, strong);

        let normal =
            floor_action_for_rss(memory_guard::PressureLevel::Normal, HomeostasisAction::None);
        assert_eq!(normal, HomeostasisAction::None);
    }

    #[test]
    fn evict_probationary_removes_single_reads() {
        let cache = Arc::new(tokio::sync::RwLock::new(SessionCache::new()));
        {
            let mut c = cache.blocking_write();
            c.store("/once.rs", "fn once() {}");
            c.store("/twice.rs", "fn twice() {}");
            c.store("/twice.rs", "fn twice() {}"); // second read → read_count=2
        }
        let bm25: SharedBm25Cache = Arc::new(std::sync::Mutex::new(None));
        let orch = EvictionOrchestrator {
            cache: cache.clone(),
            bm25_cache: bm25,
            controller: Mutex::new(HomeostasisController::new(100_000)),
            token_budget: 100_000,
        };

        let result = orch.execute_action(
            &HomeostasisAction::EvictProbationary { target_tokens: 0 },
            0,
        );
        assert!(result);
        let c = cache.blocking_read();
        assert!(c.get("/once.rs").is_none());
        assert!(c.get("/twice.rs").is_some());
    }
}
