//! Loaded states cached by their heads (hub spec §16.3). A state is a pure
//! function of its heads, so entries never go stale; the cache is only
//! bounded. Shared across threads because `oxy_core::state::State` is
//! `Send + Sync`.

use oxy_core::hash::ChangeHash;
use oxy_core::repo::Repo;
use oxy_core::state::{Snapshot, State};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

/// A state and its materialised snapshot.
pub struct Loaded {
    pub heads: Vec<ChangeHash>,
    pub state: State,
    pub snapshot: Snapshot,
}

#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct Key {
    repo: [u8; 16],
    heads: Vec<ChangeHash>,
}

struct Entry {
    loaded: Arc<Loaded>,
    last_used: u64,
}

/// What the bookmark lists and a change's "in bookmarks" need of a state:
/// small, so many are kept, and computed without filling the state cache.
/// A repository with more bookmarks than the state cache holds otherwise
/// cycled through it on every such page, so nothing stayed cached
/// (metalgear's 128 bookmarks: 7.7 s per change page, and every page of
/// every repository slower after it).
pub struct Summary {
    pub changes: std::collections::HashSet<ChangeHash>,
    pub latest_ms: u64,
    pub conflicted: bool,
}

/// Summaries kept; each is the state's change set, a few hundred kilobytes
/// for a long history.
const SUMMARIES: usize = 2048;

/// Summaries by heads, each with when it was last used, and the clock.
type Summaries = (HashMap<Key, (Arc<Summary>, u64)>, u64);

/// A bounded least-recently-used cache of loaded states.
pub struct StateCache {
    capacity: usize,
    inner: Mutex<(HashMap<Key, Entry>, u64)>,
    summaries: Mutex<Summaries>,
}

impl StateCache {
    pub fn new(capacity: usize) -> StateCache {
        StateCache { capacity: capacity.max(1), inner: Mutex::new((HashMap::new(), 0)), summaries: Mutex::new((HashMap::new(), 0)) }
    }

    /// The summary of the state with these heads: from the state cache when
    /// it holds the state, else loaded and dropped, the state cache left as
    /// it was.
    pub fn summary(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Summary>, oxy_core::repo::RepoError> {
        let mut sorted = heads.to_vec();
        sorted.sort();
        sorted.dedup();
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
        {
            let mut guard = self.summaries.lock().unwrap();
            let (map, tick) = &mut *guard;
            *tick += 1;
            if let Some(e) = map.get_mut(&key) {
                e.1 = *tick;
                return Ok(e.0.clone());
            }
        }
        let held = self.inner.lock().unwrap().0.get(&key).map(|e| e.loaded.clone());
        let summary = Arc::new(match held {
            Some(l) => summarise(&l.state, &l.snapshot),
            None => {
                let state = repo.load_state(&sorted)?;
                let snapshot = state.materialize();
                summarise(&state, &snapshot)
            }
        });
        let mut guard = self.summaries.lock().unwrap();
        let (map, tick) = &mut *guard;
        if map.len() >= SUMMARIES {
            if let Some(oldest) = map.iter().min_by_key(|(_, e)| e.1).map(|(k, _)| k.clone()) {
                map.remove(&oldest);
            }
        }
        map.insert(key, (summary.clone(), *tick));
        Ok(summary)
    }

    /// The state with these heads, loaded once per hub process.
    pub fn load(&self, repo: &Repo, heads: &[ChangeHash]) -> Result<Arc<Loaded>, oxy_core::repo::RepoError> {
        let mut sorted = heads.to_vec();
        sorted.sort();
        sorted.dedup();
        let key = Key { repo: repo.id()?, heads: sorted.clone() };
        {
            let mut guard = self.inner.lock().unwrap();
            let (map, tick) = &mut *guard;
            *tick += 1;
            if let Some(e) = map.get_mut(&key) {
                e.last_used = *tick;
                return Ok(e.loaded.clone());
            }
        }
        let state = repo.load_state(&sorted)?;
        let snapshot = state.materialize();
        let loaded = Arc::new(Loaded { heads: sorted, state, snapshot });
        let mut guard = self.inner.lock().unwrap();
        let (map, tick) = &mut *guard;
        if map.len() >= self.capacity {
            if let Some(oldest) = map.iter().min_by_key(|(_, e)| e.last_used).map(|(k, _)| k.clone()) {
                map.remove(&oldest);
            }
        }
        map.insert(key, Entry { loaded: loaded.clone(), last_used: *tick });
        Ok(loaded)
    }

    pub fn len(&self) -> usize {
        self.inner.lock().unwrap().0.len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }
}

fn summarise(state: &State, snapshot: &Snapshot) -> Summary {
    Summary {
        changes: state.changes().map(|(h, _)| *h).collect(),
        latest_ms: state.changes().map(|(_, m)| m.hlc.wall_ms).max().unwrap_or(0),
        conflicted: snapshot.is_conflicted(),
    }
}

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

    #[test]
    fn caches_by_heads_and_evicts_the_oldest() {
        let dir = tempfile::tempdir().unwrap();
        let repo = Repo::init_bare(dir.path()).unwrap();
        let cache = StateCache::new(2);
        let a = cache.load(&repo, &[]).unwrap();
        let b = cache.load(&repo, &[]).unwrap();
        assert!(Arc::ptr_eq(&a, &b), "same heads, same entry");
        assert_eq!(cache.len(), 1);
        let h1 = oxy_core::hash::ObjectRef([1; 32]);
        let h2 = oxy_core::hash::ObjectRef([2; 32]);
        // Unknown heads fail to load and are not cached.
        assert!(cache.load(&repo, &[h1]).is_err());
        assert_eq!(cache.len(), 1);
        let _ = (h2, dir);
    }

    #[test]
    fn summaries_leave_the_state_cache_alone() {
        let dir = tempfile::tempdir().unwrap();
        let repo = Repo::init_bare(dir.path()).unwrap();
        let cache = StateCache::new(2);
        let a = cache.summary(&repo, &[]).unwrap();
        assert_eq!(cache.len(), 0, "a summary does not fill the state cache");
        assert!(Arc::ptr_eq(&a, &cache.summary(&repo, &[]).unwrap()), "kept");
        assert!(a.changes.is_empty() && !a.conflicted);
        assert!(cache.summary(&repo, &[oxy_core::hash::ObjectRef([3; 32])]).is_err());
    }
}