//! Evaluating revsets (docs/revset_query_language.md §4) over a repository.
//!
//! The graph is the change metadata the state index already holds: a
//! symbol loads the closure of the heads it names, and `all()`, `X::`,
//! `dependents()` and `roots()` load every bookmark, remote bookmark, tag
//! and the working copy. Evaluation is read-only and has a time budget;
//! a query past it fails and says what to narrow.

use super::*;
use crate::ids::NodeId;
use crate::revset::{self, Expr};
use crate::state::ChangeMeta;
use std::time::{Duration, Instant};

type Set = BTreeSet<ChangeHash>;

/// The default budget for a command-line evaluation.
pub const REVSET_BUDGET: Duration = Duration::from_secs(10);

/// A revset's result with the metadata of every change it names.
#[derive(Clone, Debug)]
pub struct Revset {
    /// Newest first, as a log is ordered.
    pub hashes: Vec<ChangeHash>,
    pub metas: HashMap<ChangeHash, ChangeMeta>,
}

struct Eval<'a> {
    repo: &'a Repo,
    graph: HashMap<ChangeHash, ChangeMeta>,
    /// Which heads have had their closure loaded.
    loaded: HashSet<ChangeHash>,
    all_loaded: bool,
    dependents: Option<HashMap<ChangeHash, Vec<ChangeHash>>>,
    names: HashMap<KeyId, Option<(String, String)>>,
    /// Who authenticated each change, once asked (signature checks cost).
    auth: HashMap<ChangeHash, SignatureStatus>,
    deadline: Instant,
    budget: Duration,
    now_ms: u64,
    depth: usize,
}

fn err(msg: impl Into<String>) -> RepoError {
    RepoError::Msg(msg.into())
}

impl<'a> Eval<'a> {
    fn tick(&self) -> Result<()> {
        if Instant::now() > self.deadline {
            return Err(err(format!(
                "the revset went past its budget of {:.0} s; narrow it with a range such as main.. or a date()",
                self.budget.as_secs_f64()
            )));
        }
        Ok(())
    }

    /// Load the closure of `heads` into the graph.
    fn ensure(&mut self, heads: &[ChangeHash]) -> Result<()> {
        let fresh: Vec<ChangeHash> = heads
            .iter()
            .copied()
            .filter(|h| !self.loaded.contains(h))
            .collect();
        if fresh.is_empty() {
            return Ok(());
        }
        for (h, m) in self.repo.metas_of(&fresh)? {
            self.graph.insert(h, m);
        }
        self.loaded.extend(fresh);
        self.dependents = None;
        self.tick()
    }

    /// Load every change the repository names: bookmarks, remote
    /// bookmarks, tags, and the working copy.
    fn load_all(&mut self) -> Result<()> {
        if self.all_loaded {
            return Ok(());
        }
        let view = self.repo.view()?;
        let mut heads: Vec<ChangeHash> = Vec::new();
        for st in view.bookmarks.values() {
            heads.extend(self.repo.store().get_state(st)?.heads);
        }
        for bookmarks in view.remotes.values() {
            for st in bookmarks.values() {
                heads.extend(self.repo.store().get_state(st)?.heads);
            }
        }
        for st in view.tags.values() {
            heads.extend(self.repo.store().get_state(st)?.heads);
        }
        if let Some(wc) = self.working_copy() {
            heads.push(wc);
        }
        heads.sort();
        heads.dedup();
        self.ensure(&heads)?;
        self.all_loaded = true;
        Ok(())
    }

    fn working_copy(&self) -> Option<ChangeHash> {
        if self.repo.is_bare() {
            None
        } else {
            self.repo.working_copy().ok().map(|(h, _)| h)
        }
    }

    fn all(&mut self) -> Result<Set> {
        self.load_all()?;
        Ok(self.graph.keys().copied().collect())
    }

    fn dependents_map(&mut self) -> Result<&HashMap<ChangeHash, Vec<ChangeHash>>> {
        self.load_all()?;
        if self.dependents.is_none() {
            let mut map: HashMap<ChangeHash, Vec<ChangeHash>> = HashMap::new();
            for (h, m) in &self.graph {
                for d in &m.deps {
                    map.entry(*d).or_default().push(*h);
                }
            }
            self.dependents = Some(map);
        }
        Ok(self.dependents.as_ref().unwrap())
    }

    /// `::X`: the members and their dependency closures.
    fn ancestors(&mut self, set: &Set) -> Result<Set> {
        let mut out = set.clone();
        let mut stack: Vec<ChangeHash> = set.iter().copied().collect();
        let mut n = 0usize;
        while let Some(h) = stack.pop() {
            if let Some(m) = self.graph.get(&h) {
                for d in &m.deps {
                    if out.insert(*d) {
                        stack.push(*d);
                    }
                }
            }
            n += 1;
            if n.is_multiple_of(4096) {
                self.tick()?;
            }
        }
        Ok(out)
    }

    /// `X::`: the members and everything that transitively depends on them.
    fn descendants(&mut self, set: &Set) -> Result<Set> {
        let mut out = set.clone();
        let mut stack: Vec<ChangeHash> = set.iter().copied().collect();
        let deps = self.dependents_map()?;
        while let Some(h) = stack.pop() {
            if let Some(ds) = deps.get(&h) {
                for d in ds {
                    if out.insert(*d) {
                        stack.push(*d);
                    }
                }
            }
        }
        Ok(out)
    }

    /// Members no other member depends on, directly or through others.
    fn heads(&mut self, set: &Set) -> Result<Set> {
        let mut below: HashSet<ChangeHash> = HashSet::new();
        let mut stack: Vec<ChangeHash> = set
            .iter()
            .filter_map(|h| self.graph.get(h))
            .flat_map(|m| m.deps.iter().copied())
            .collect();
        while let Some(h) = stack.pop() {
            if below.insert(h) {
                if let Some(m) = self.graph.get(&h) {
                    stack.extend(m.deps.iter().copied());
                }
            }
        }
        Ok(set.iter().copied().filter(|h| !below.contains(h)).collect())
    }

    /// Members that depend on no other member, directly or through others.
    fn roots(&mut self, set: &Set) -> Result<Set> {
        let deps = self.dependents_map()?;
        let mut above: HashSet<ChangeHash> = HashSet::new();
        let mut stack: Vec<ChangeHash> = set
            .iter()
            .filter_map(|h| deps.get(h))
            .flat_map(|ds| ds.iter().copied())
            .collect();
        while let Some(h) = stack.pop() {
            if above.insert(h) {
                if let Some(ds) = deps.get(&h) {
                    stack.extend(ds.iter().copied());
                }
            }
        }
        Ok(set.iter().copied().filter(|h| !above.contains(h)).collect())
    }

    fn symbol(&mut self, name: &str) -> Result<Set> {
        if name == "@" {
            let Some(wc) = self.working_copy() else {
                return Err(err(
                    "@ names the working copy, and a bare repository has none",
                ));
            };
            self.ensure(&[wc])?;
            return Ok([wc].into_iter().collect());
        }
        let heads = self.repo.resolve_heads(name)?;
        self.ensure(&heads)?;
        Ok(heads.into_iter().collect())
    }

    /// The identity's name and email, if its object is known.
    fn who(&mut self, key: &KeyId) -> Option<(String, String)> {
        if let Some(w) = self.names.get(key) {
            return w.clone();
        }
        let w = self.repo.identity(key).ok().map(|id| (id.name, id.email));
        self.names.insert(*key, w.clone());
        w
    }

    /// The set a predicate ranges over: what it is intersected with when
    /// that is known, else everything.
    fn universe(&mut self, within: Option<&Set>) -> Result<Set> {
        match within {
            Some(s) => Ok(s.clone()),
            None => self.all(),
        }
    }

    fn filter_metas(
        &mut self,
        within: Option<&Set>,
        mut keep: impl FnMut(&ChangeHash, &ChangeMeta) -> bool,
    ) -> Result<Set> {
        let candidates = self.universe(within)?;
        let mut out = Set::new();
        for (i, h) in candidates.iter().enumerate() {
            if let Some(m) = self.graph.get(h) {
                if keep(h, m) {
                    out.insert(*h);
                }
            }
            if i % 8192 == 8191 {
                self.tick()?;
            }
        }
        Ok(out)
    }

    /// Changes touching any node whose path matches `glob` in the
    /// evaluation state: the working copy's base, or the bookmark's state
    /// in a bare repository. Nodes keep their identity through renames.
    fn touches(&mut self, within: Option<&Set>, glob: &str) -> Result<Set> {
        let matcher = globset::GlobBuilder::new(glob)
            .literal_separator(false)
            .build()
            .map_err(|e| err(format!("touches({glob:?}): {e}")))?
            .compile_matcher();
        let heads = match self.working_copy() {
            Some(wc) => self
                .graph
                .get(&wc)
                .map(|m| m.deps.clone())
                .or_else(|| self.repo.working_copy().ok().map(|(_, c)| c.deps))
                .unwrap_or_default(),
            None => self.repo.bookmark_heads()?,
        };
        let state = self.repo.load_state(&heads)?;
        let nodes: HashSet<NodeId> = state
            .tree()
            .view()
            .entries
            .iter()
            .filter(|(path, _)| matcher.is_match(path))
            .map(|(_, e)| e.node)
            .collect();
        if nodes.is_empty() {
            return Ok(Set::new());
        }
        let candidates = self.universe(within)?;
        let mut out = Set::new();
        for (i, h) in candidates.iter().enumerate() {
            let Ok(c) = self.repo.store().get_change(h) else {
                continue;
            };
            if c.tree_ops.iter().any(|o| nodes.contains(&o.node()))
                || c.edits.iter().any(|e| nodes.contains(&e.node()))
            {
                out.insert(*h);
            }
            if i % 256 == 255 {
                self.tick()?;
            }
        }
        Ok(out)
    }

    fn text_arg<'e>(&self, name: &str, args: &'e [Expr], i: usize) -> Result<&'e str> {
        match args.get(i) {
            Some(Expr::Symbol(s)) => Ok(s),
            _ => Err(err(format!("{name}() takes text, quoted if it has spaces"))),
        }
    }

    fn arity(&self, name: &str, args: &[Expr], n: usize) -> Result<()> {
        if args.len() != n {
            return Err(err(format!(
                "{name}() takes {n} argument{}, not {}",
                if n == 1 { "" } else { "s" },
                args.len()
            )));
        }
        Ok(())
    }

    fn call(&mut self, name: &str, args: &[Expr], within: Option<&Set>) -> Result<Set> {
        match name {
            "all" => {
                self.arity(name, args, 0)?;
                self.universe(within)
            }
            "none" => {
                self.arity(name, args, 0)?;
                Ok(Set::new())
            }
            "deps" => {
                self.arity(name, args, 1)?;
                let x = self.eval(&args[0], None)?;
                let mut out = Set::new();
                for h in &x {
                    if let Some(m) = self.graph.get(h) {
                        out.extend(m.deps.iter().copied());
                    }
                }
                Ok(out)
            }
            "dependents" => {
                self.arity(name, args, 1)?;
                let x = self.eval(&args[0], None)?;
                let deps = self.dependents_map()?;
                let mut out = Set::new();
                for h in &x {
                    if let Some(ds) = deps.get(h) {
                        out.extend(ds.iter().copied());
                    }
                }
                Ok(out)
            }
            "heads" => {
                self.arity(name, args, 1)?;
                let x = self.eval(&args[0], None)?;
                self.heads(&x)
            }
            "roots" => {
                self.arity(name, args, 1)?;
                let x = self.eval(&args[0], None)?;
                self.roots(&x)
            }
            "latest" => {
                self.arity(name, args, 2)?;
                let x = self.eval(&args[0], None)?;
                let n: usize = self
                    .text_arg(name, args, 1)?
                    .parse()
                    .map_err(|_| err("latest(X, n) takes a count"))?;
                let mut members: Vec<ChangeHash> = x.into_iter().collect();
                self.sort_newest_first(&mut members);
                Ok(members.into_iter().take(n).collect())
            }
            "author" => {
                self.arity(name, args, 1)?;
                let who = self.text_arg(name, args, 0)?.to_string();
                let mut matches: HashMap<KeyId, bool> = HashMap::new();
                self.filter_metas_with(within, |ev, _, m| {
                    if let Some(hit) = matches.get(&m.author) {
                        return *hit;
                    }
                    let hit = ev.names_key(&who, &m.author);
                    matches.insert(m.author, hit);
                    hit
                })
            }
            "description" => {
                self.arity(name, args, 1)?;
                let text = self.text_arg(name, args, 0)?.to_string();
                if text.starts_with("re:") {
                    return Err(err(
                        "description(re:…) is not supported yet; give a substring",
                    ));
                }
                let needle = text.to_lowercase();
                self.filter_metas(within, |_, m| m.message.to_lowercase().contains(&needle))
            }
            "date" => {
                self.arity(name, args, 1)?;
                let range = revset::parse_date_range(self.text_arg(name, args, 0)?, self.now_ms)
                    .map_err(err)?;
                self.filter_metas(within, |_, m| range.contains(m.hlc.wall_ms))
            }
            "kind" => {
                self.arity(name, args, 1)?;
                let kind = match self.text_arg(name, args, 0)? {
                    "normal" => ChangeKind::Normal,
                    "resolution" => ChangeKind::Resolution,
                    "working-copy" | "working_copy" => ChangeKind::WorkingCopy,
                    other => {
                        return Err(err(format!(
                            "kind({other:?}): use normal, resolution or working-copy"
                        )))
                    }
                };
                self.filter_metas(within, |_, m| m.kind == kind)
            }
            "touches" => {
                self.arity(name, args, 1)?;
                let glob = self.text_arg(name, args, 0)?.to_string();
                self.touches(within, &glob)
            }
            // Trust (§2.4): what authenticated a change, and what vouched
            // for the states it is in.
            "signed_by" => {
                self.arity(name, args, 1)?;
                let who = self.text_arg(name, args, 0)?.to_string();
                self.filter_metas_with(within, |ev, h, m| match ev.auth(h, m) {
                    SignatureStatus::Valid { key, .. } => ev.names_key(&who, &key),
                    _ => false,
                })
            }
            "delegated" => {
                self.arity(name, args, 0)?;
                self.filter_metas_with(within, |ev, h, m| {
                    !m.author.is_zero()
                        && matches!(ev.auth(h, m), SignatureStatus::Valid { key, .. } if key != m.author)
                })
            }
            "unsigned" => {
                self.arity(name, args, 0)?;
                self.filter_metas_with(within, |ev, h, m| {
                    m.kind != ChangeKind::WorkingCopy
                        && !matches!(ev.auth(h, m), SignatureStatus::Valid { .. })
                })
            }
            "revoked" => {
                self.arity(name, args, 0)?;
                self.filter_metas_with(within, |ev, h, m| {
                    matches!(ev.auth(h, m), SignatureStatus::Valid { revoked: true, .. })
                })
            }
            "attested" => {
                self.arity(name, args, 1)?;
                let kind = self.text_arg(name, args, 0)?.to_string();
                self.attested(within, &kind)
            }
            "built" => {
                self.arity(name, args, 1)?;
                let job = self.text_arg(name, args, 0)?.to_string();
                self.built(within, &job)
            }
            other => {
                // A saved revset: `oxy config revset.NAME 'EXPR'` makes NAME().
                if let Some(saved) = self.repo.config.revsets.get(other).cloned() {
                    self.arity(name, args, 0)?;
                    if self.depth >= 16 {
                        return Err(err(format!("revset.{other} refers to itself")));
                    }
                    let e = revset::parse(&saved)
                        .map_err(|e| err(format!("revset.{other} = {saved:?}: {e}")))?;
                    self.depth += 1;
                    let r = self.eval(&e, within);
                    self.depth -= 1;
                    return r;
                }
                Err(err(format!(
                    "unknown function {other}(); see docs/revset_query_language.md"
                )))
            }
        }
    }

    /// The authentication of a change: its signature or seal, checked once.
    fn auth(&mut self, h: &ChangeHash, m: &ChangeMeta) -> SignatureStatus {
        if let Some(s) = self.auth.get(h) {
            return s.clone();
        }
        let s = self
            .repo
            .authenticate_change_known(h, Some(m))
            .unwrap_or_else(|e| SignatureStatus::Invalid(e.to_string()));
        self.auth.insert(*h, s.clone());
        s
    }

    /// Whether `who` names key `key`: by hex prefix, `me`, or the name or
    /// email of its identity.
    fn names_key(&mut self, who: &str, key: &KeyId) -> bool {
        (who == "me" && Some(*key) == self.repo.config.author_key)
            || key.starts_with_hex(who)
            || self.who(key).is_some_and(|(n, e)| {
                let needle = who.to_lowercase();
                n.to_lowercase().contains(&needle) || e.to_lowercase().contains(&needle)
            })
    }

    /// The closure of the states whose heads these are, within `within`.
    fn closure_of(&mut self, mut heads: Vec<ChangeHash>, within: Option<&Set>) -> Result<Set> {
        heads.sort();
        heads.dedup();
        self.ensure(&heads)?;
        let set: Set = heads.into_iter().collect();
        let closure = self.ancestors(&set)?;
        Ok(match within {
            Some(w) => closure.intersection(w).copied().collect(),
            None => closure,
        })
    }

    /// The closure of every state that carries an attestation of `kind`:
    /// `tag` (tagged states), `receipt` (states a runner reported on) or
    /// `ref` (states a remote attested, as its tracking bookmarks hold).
    fn attested(&mut self, within: Option<&Set>, kind: &str) -> Result<Set> {
        let mut heads: Vec<ChangeHash> = Vec::new();
        match kind {
            "tag" => {
                for t in self.repo.tags()? {
                    heads.extend(t.heads);
                }
            }
            "receipt" => {
                for st in self.repo.store().receipt_states()? {
                    if let Ok(s) = self.repo.store().get_state(&st) {
                        heads.extend(s.heads);
                    }
                }
            }
            "ref" => {
                let view = self.repo.view()?;
                for bookmarks in view.remotes.values() {
                    for st in bookmarks.values() {
                        heads.extend(self.repo.store().get_state(st)?.heads);
                    }
                }
            }
            other => return Err(err(format!("attested({other:?}): use tag, receipt or ref"))),
        }
        self.closure_of(heads, within)
    }

    /// The closure of every state with a valid receipt for Flow job `job`
    /// from a runner whose key is not revoked.
    fn built(&mut self, within: Option<&Set>, job: &str) -> Result<Set> {
        let mut heads: Vec<ChangeHash> = Vec::new();
        for st in self.repo.store().receipt_states()? {
            let Ok(state) = self.repo.store().get_state(&st) else {
                continue;
            };
            let good = self
                .repo
                .receipts(&state.heads)?
                .into_iter()
                .any(|r| r.signed.receipt.job_name == job && r.problem.is_none() && !r.revoked);
            if good {
                heads.extend(state.heads);
            }
            self.tick()?;
        }
        self.closure_of(heads, within)
    }

    /// `filter_metas` for a test that needs the evaluator (identity lookups).
    fn filter_metas_with(
        &mut self,
        within: Option<&Set>,
        mut keep: impl FnMut(&mut Eval<'a>, &ChangeHash, &ChangeMeta) -> bool,
    ) -> Result<Set> {
        let candidates = self.universe(within)?;
        let mut out = Set::new();
        for (i, h) in candidates.iter().enumerate() {
            if let Some(m) = self.graph.get(h).cloned() {
                if keep(self, h, &m) {
                    out.insert(*h);
                }
            }
            if i % 8192 == 8191 {
                self.tick()?;
            }
        }
        Ok(out)
    }

    /// Whether an expression selects by a property, so that evaluating it
    /// within the other side of an intersection saves scanning everything.
    fn is_predicate(e: &Expr) -> bool {
        match e {
            Expr::Call(name, _) => matches!(
                name.as_str(),
                "author"
                    | "description"
                    | "date"
                    | "kind"
                    | "touches"
                    | "all"
                    | "signed_by"
                    | "delegated"
                    | "unsigned"
                    | "revoked"
                    | "attested"
                    | "built"
            ),
            Expr::Intersection(a, b) => Self::is_predicate(a) || Self::is_predicate(b),
            Expr::Union(a, b) => Self::is_predicate(a) && Self::is_predicate(b),
            Expr::Difference(a, _) => Self::is_predicate(a),
            _ => false,
        }
    }

    fn eval(&mut self, e: &Expr, within: Option<&Set>) -> Result<Set> {
        self.tick()?;
        match e {
            Expr::Symbol(s) => self.symbol(s),
            Expr::Ancestors(x) => {
                let x = self.eval(x, None)?;
                self.ancestors(&x)
            }
            Expr::Descendants(x) => {
                let x = self.eval(x, None)?;
                self.descendants(&x)
            }
            Expr::Between(x, y) => {
                let x = self.eval(x, None)?;
                let y = self.eval(y, None)?;
                let down = self.descendants(&x)?;
                let up = self.ancestors(&y)?;
                Ok(down.intersection(&up).copied().collect())
            }
            Expr::Range(x, y) => {
                let x = self.eval(x, None)?;
                let y = self.eval(y, None)?;
                let from = self.ancestors(&x)?;
                let to = self.ancestors(&y)?;
                Ok(to.difference(&from).copied().collect())
            }
            Expr::Union(a, b) => {
                let mut a = self.eval(a, within)?;
                a.extend(self.eval(b, within)?);
                Ok(a)
            }
            Expr::Intersection(a, b) => {
                let (first, second) = if Self::is_predicate(a) && !Self::is_predicate(b) {
                    (b, a)
                } else {
                    (a, b)
                };
                let left = self.eval(first, within)?;
                let right = self.eval(second, Some(&left))?;
                Ok(left.intersection(&right).copied().collect())
            }
            Expr::Difference(a, b) => {
                let left = self.eval(a, within)?;
                let right = self.eval(b, Some(&left))?;
                Ok(left.difference(&right).copied().collect())
            }
            Expr::Call(name, args) => self.call(name, args, within),
        }
    }

    fn sort_newest_first(&self, hashes: &mut [ChangeHash]) {
        hashes.sort_by(|a, b| {
            let (ma, mb) = (self.graph.get(a), self.graph.get(b));
            mb.map(|m| m.hlc).cmp(&ma.map(|m| m.hlc)).then(b.cmp(a))
        });
    }
}

impl Repo {
    /// Evaluate a revset within [`REVSET_BUDGET`].
    pub fn revset(&self, src: &str) -> Result<Revset> {
        self.revset_with_budget(src, REVSET_BUDGET)
    }

    /// Evaluate a revset: the changes it selects, newest first, with
    /// their metadata. Read-only. Fails past `budget`.
    pub fn revset_with_budget(&self, src: &str, budget: Duration) -> Result<Revset> {
        let expr = revset::parse(src).map_err(|e| err(format!("revset {src:?}: {e}")))?;
        let mut ev = Eval {
            repo: self,
            graph: HashMap::new(),
            loaded: HashSet::new(),
            all_loaded: false,
            dependents: None,
            names: HashMap::new(),
            auth: HashMap::new(),
            deadline: Instant::now() + budget,
            budget,
            now_ms: Hlc::now().wall_ms,
            depth: 0,
        };
        let set = ev.eval(&expr, None)?;
        let mut hashes: Vec<ChangeHash> = set.into_iter().collect();
        ev.sort_newest_first(&mut hashes);
        let metas = hashes
            .iter()
            .filter_map(|h| ev.graph.get(h).map(|m| (*h, m.clone())))
            .collect();
        Ok(Revset { hashes, metas })
    }

    /// The log entries a revset selects, newest first. A head of the
    /// current bookmark is marked as in `log`.
    pub fn log_revset(&self, src: &str) -> Result<Vec<LogEntry>> {
        let r = self.revset(src)?;
        let heads: BTreeSet<ChangeHash> = self.bookmark_heads()?.into_iter().collect();
        let wc = if self.is_bare() {
            None
        } else {
            self.working_copy().ok().map(|(h, _)| h)
        };
        let mut entries: Vec<LogEntry> = r
            .hashes
            .iter()
            .filter_map(|h| {
                r.metas.get(h).map(|m| LogEntry {
                    hash: *h,
                    change: m.clone(),
                    is_working_copy: Some(*h) == wc,
                    is_head: heads.contains(h),
                })
            })
            .collect();
        // The working copy first, as `log` shows it.
        entries.sort_by_key(|e| !e.is_working_copy);
        Ok(entries)
    }
}