//! Query engine wrapping a `TripleStore`.

use std::collections::HashSet;

use ontoref_triples::TripleStore;
use ontoref_types::{AttrId, EntityId, Hlc, Operation, Value};

use crate::error::Result;

/// Subscriber callback invoked once per successful `apply`.
pub type Subscriber = Box<dyn FnMut(&Operation) + Send>;

/// Query and apply pipeline over a `TripleStore`.
pub struct QueryEngine {
    triples: TripleStore,
    subscribers: Vec<Subscriber>,
}

impl QueryEngine {
    /// Wrap a triple store. The engine owns the store; queries and applies go
    /// through here.
    pub fn new(triples: TripleStore) -> Self {
        Self {
            triples,
            subscribers: Vec::new(),
        }
    }

    /// Borrow the underlying triple store (for read-only queries the engine
    /// does not yet expose).
    pub fn triples(&self) -> &TripleStore {
        &self.triples
    }

    /// Apply an operation and notify every subscriber. If the underlying
    /// `triples.apply` returns an error, subscribers are NOT invoked.
    pub fn apply(&mut self, op: &Operation) -> Result<()> {
        self.triples.apply(op)?;
        for subscriber in &mut self.subscribers {
            (subscriber)(op);
        }
        Ok(())
    }

    /// Register a subscription handler. Returns the slot index (use it to
    /// reason about delivery order; G6 does not yet support unsubscribe).
    pub fn subscribe<F>(&mut self, handler: F) -> usize
    where
        F: FnMut(&Operation) + Send + 'static,
    {
        let idx = self.subscribers.len();
        self.subscribers.push(Box::new(handler));
        idx
    }

    /// Number of registered subscribers.
    pub fn subscriber_count(&self) -> usize {
        self.subscribers.len()
    }

    /// Transitive closure starting from `root`, walking `ref_attr` until the
    /// path terminates (no more refs, attr missing, or a node we have
    /// already visited). Returns the set of entities reachable from `root`
    /// at the `(as_of, valid_at)` viewpoint, inclusive of `root`. The walk
    /// terminates on cycles because each entity is enqueued at most once.
    pub fn transitive_closure(
        &self,
        root: &EntityId,
        ref_attr: &AttrId,
        as_of: Hlc,
        valid_at: Hlc,
    ) -> Result<HashSet<EntityId>> {
        let mut visited: HashSet<EntityId> = HashSet::new();
        let mut frontier: Vec<EntityId> = vec![root.clone()];
        while let Some(node) = frontier.pop() {
            if !visited.insert(node.clone()) {
                continue;
            }
            let next = self.triples.entity_attr(&node, ref_attr, as_of, valid_at)?;
            if let Some(Value::Ref(target)) = next {
                frontier.push(target);
            }
        }
        Ok(visited)
    }
}
