ggen-engine 26.10.11

//! Deterministic RDF graph over an in-memory oxigraph [`Store`].
//!
//! Provides [`DeterministicGraph`] — Turtle loading, SPARQL querying, and a
//! BLAKE3 `state_hash` computed over canonicalized, sorted N-Quads — plus a
//! [`Delta`] type describing additions/deletions between two graphs.
//!
//! Canonicalization: graphs without blank nodes are sorted lexicographically
//! by their N-Quads string form; graphs with blank nodes go through a bounded
//! color-refinement pass (5 iterations of BLAKE3 neighborhood signatures)
//! and blank nodes are relabelled `c14n{i}` in signature order so isomorphic
//! graphs hash identically regardless of blank-node labels.

use std::collections::{BTreeSet, HashMap, HashSet};

use oxigraph::{
    io::RdfFormat,
    model::{BlankNode, GraphName, NamedOrBlankNode, Quad, Term},
    sparql::{QueryResults, SparqlEvaluator},
    store::Store,
};

use crate::error::{AppError, Result};

mod ontology_batch;
pub use ontology_batch::{OntologyBatchReceipt, TurtleDocument};

/// Number of color-refinement iterations for blank-node canonicalization.
/// Five rounds are sufficient for the small ontology graphs ggen operates on.
const REFINEMENT_ITERATIONS: usize = 5;

/// An in-memory RDF store with deterministic state hashing.
pub struct DeterministicGraph {
    store: Store,
}

impl DeterministicGraph {
    /// Create an empty in-memory graph.
    ///
    /// # Errors
    /// Returns `[FM-GRAPH-001]` if the oxigraph store cannot be initialized.
    pub fn new() -> Result<Self> {
        let store = Store::new()
            .map_err(|e| AppError::fm_graph(1, format!("failed to create in-memory store: {e}")))?;
        Ok(Self { store })
    }

    /// Parse and insert Turtle content, returning the number of quads added.
    ///
    /// # Errors
    /// Returns `[FM-GRAPH-002]` on Turtle syntax or storage errors.
    pub fn insert_turtle(&self, ttl: &str) -> Result<usize> {
        let before = self
            .store
            .len()
            .map_err(|e| AppError::fm_graph(2, format!("store length unavailable: {e}")))?;
        self.store
            .load_from_slice(RdfFormat::Turtle, ttl)
            .map_err(|e| AppError::fm_graph(2, format!("turtle load failed: {e}")))?;
        let after = self
            .store
            .len()
            .map_err(|e| AppError::fm_graph(2, format!("store length unavailable: {e}")))?;
        let inserted = after.saturating_sub(before);
        self.canonicalize_blank_nodes()?;
        Ok(inserted)
    }

    /// Rewrite every blank node currently in the store to its stable
    /// `c14n{i}` label (bounded color refinement, see [`blank_node_relabel_map`]
    /// / the module docs) so the store never exposes a raw, parser-assigned
    /// blank-node id to anything downstream.
    ///
    /// This is the root-cause half of blank-node determinism: relabeling
    /// only at SPARQL projection time (also done, see
    /// [`GraphEngine::query`]'s `impl` for [`DeterministicGraph`]) cannot by
    /// itself fix a query whose `ORDER BY`/`GROUP BY`/`DISTINCT` compares a
    /// blank-node-valued variable, because SPARQL evaluation runs BEFORE
    /// projection and would still sort/dedupe on the store's raw ids.
    /// Canonicalizing the store itself, immediately after every mutation,
    /// means the SPARQL evaluator itself only ever sees stable ids, so
    /// `ORDER BY ?s` picks the same row on every independent load of
    /// byte-identical input. Called from every mutation path
    /// ([`DeterministicGraph::insert_turtle`],
    /// `ontology_batch::insert_documents`, [`Delta::apply`]) rather than
    /// once per query, so the (usually blank-node-free) common case pays
    /// this cost once per load, not once per template.
    ///
    /// A no-op (single `all_quads` scan, no store mutation) when the graph
    /// has no blank nodes — the overwhelmingly common case.
    ///
    /// # Errors
    /// Propagates [`blank_node_relabel_map`] failures, or store
    /// iteration/mutation errors (`[FM-GRAPH-004]`/`[FM-GRAPH-002]`).
    fn canonicalize_blank_nodes(&self) -> Result<()> {
        let quads = self.all_quads()?;
        let relabel = blank_node_relabel_map(&quads)?;
        if relabel.is_empty() {
            return Ok(());
        }
        let mut to_insert: Vec<Quad> = Vec::new();
        for quad in &quads {
            let touches_blank = matches!(&quad.subject, NamedOrBlankNode::BlankNode(_))
                || matches!(&quad.object, Term::BlankNode(_))
                || matches!(&quad.graph_name, GraphName::BlankNode(_));
            if !touches_blank {
                continue;
            }
            let relabeled = relabel_quad(quad, &relabel);
            if &relabeled == quad {
                // Already canonical (idempotent re-run, e.g. a second
                // `insert_turtle` call after the first already canonicalized
                // this node) — nothing to rewrite.
                continue;
            }
            self.store.remove(quad).map_err(|e| {
                AppError::fm_graph(
                    2,
                    format!("blank-node canonicalization: remove of stale quad failed: {e}"),
                )
            })?;
            to_insert.push(relabeled);
        }
        if !to_insert.is_empty() {
            self.store.extend(to_insert).map_err(|e| {
                AppError::fm_graph(
                    2,
                    format!("blank-node canonicalization: insert of relabeled quads failed: {e}"),
                )
            })?;
        }
        Ok(())
    }

    /// Execute a SPARQL query against the graph.
    ///
    /// Before executing, the query is independently parsed with
    /// [`spargebra`] (oxigraph keeps its own parsed AST private — see
    /// `query_has_graph_clause`'s doc comment) and structurally checked
    /// for a `GRAPH <...> { ... }` clause. This crate loads all RDF content
    /// into a single default graph — no named-graph ingestion path exists —
    /// so a `GRAPH` clause would otherwise always match zero triples
    /// *silently*, which is worse than refusing.
    ///
    /// # Errors
    /// - Returns `[FM-GRAPH-008]` if the query contains a `GRAPH` clause
    ///   (anywhere, including nested inside `FILTER EXISTS`/`NOT EXISTS`).
    /// - Returns `[FM-GRAPH-009]` if the query fails to parse as a
    ///   SELECT/CONSTRUCT/DESCRIBE/ASK *and* its text looks like a SPARQL
    ///   UPDATE attempt (best-effort keyword sniff — see
    ///   `looks_like_sparql_update`). SPARQL UPDATE is structurally
    ///   unreachable through this method regardless: `spargebra::Query` has
    ///   no `Update` variant and the `QueryUnit` grammar this crate parses
    ///   with never accepts UPDATE syntax, so this is strictly a clearer
    ///   error message on an already-refused query, not a new refusal path.
    /// - Returns `[FM-GRAPH-003]` on any other parse or evaluation error
    ///   (including SERVICE calls: no HTTP service-handler feature or
    ///   custom handler is configured on the evaluator, so a SERVICE clause
    ///   fails closed at evaluation time with oxigraph's own
    ///   `UnsupportedService` error, wrapped here).
    pub fn query(&self, sparql: &str) -> Result<QueryResults<'static>> {
        if let Ok(parsed) = spargebra::SparqlParser::new().parse_query(sparql) {
            if query_has_graph_clause(&parsed) {
                return Err(AppError::fm_graph(
                    8,
                    "SPARQL GRAPH clause is not supported -- ggen-engine loads all RDF \
                     content into a single default graph (no named-graph ingestion path \
                     exists); a GRAPH clause will always match zero triples silently, \
                     which is worse than refusing. \
                     Remediation: remove the GRAPH clause, or query without it against \
                     the default graph.",
                ));
            }
        }
        SparqlEvaluator::new()
            .parse_query(sparql)
            .map_err(|e| {
                if looks_like_sparql_update(sparql) {
                    AppError::fm_graph(
                        9,
                        format!(
                            "SPARQL UPDATE is not supported by ggen-engine's read-only \
                             query pipeline (only SELECT/CONSTRUCT/DESCRIBE/ASK are \
                             accepted). Remediation: this crate has no mutation path \
                             through `query()` -- load RDF via `insert_turtle` or \
                             frontmatter `rdf:`/`rdf_inline:` instead. \
                             Underlying parse error: {e}"
                        ),
                    )
                } else if let Some(err) = diagnose_bind_rebind(sparql) {
                    err
                } else {
                    let near = offending_line_hint(sparql, &e);
                    AppError::fm_graph(3, format!("SPARQL parse failed: {e}{near}"))
                }
            })?
            .on_store(&self.store)
            .execute()
            .map_err(|e| AppError::fm_graph(3, format!("SPARQL evaluation failed: {e}")))
    }

    /// Return every quad currently in the store.
    ///
    /// # Errors
    /// Returns `[FM-GRAPH-004]` if iteration over the store fails.
    pub fn all_quads(&self) -> Result<Vec<Quad>> {
        self.store
            .iter()
            .collect::<std::result::Result<Vec<Quad>, _>>()
            .map_err(|e| AppError::fm_graph(4, format!("store iteration failed: {e}")))
    }

    /// BLAKE3 hash of the canonicalized graph state.
    ///
    /// The hash is computed over the sorted canonical N-Quads strings joined
    /// by `\n`. Isomorphic graphs (including blank-node relabelings) produce
    /// the same hash; insertion order never matters.
    ///
    /// # Errors
    /// Returns `[FM-GRAPH-004]`/`[FM-GRAPH-005]` on iteration or
    /// canonicalization failures.
    pub fn state_hash(&self) -> Result<[u8; 32]> {
        let quads = self.all_quads()?;
        let lines = canonical_nquad_lines(&quads)?;
        Ok(*blake3::hash(lines.join("\n").as_bytes()).as_bytes())
    }
}

// ---------------------------------------------------------------------------
// Structural (AST-level) GRAPH-clause detection + UPDATE-attempt sniffing
// ---------------------------------------------------------------------------
//
// Investigated route rejected: matching `format!("{:?}", parsed_query)` for
// the substring "Graph {" is unsound, not merely inelegant -- oxrdf's
// `Literal`'s derived `Debug` for its `String` variant does not escape `{`
// or `}`, so `FILTER(?x = "a Graph { thing }")` (a literal whose lexical
// value happens to contain that text) would trip a false positive with zero
// actual GRAPH clause present. The structural match below is compiler-
// checked and exhaustive instead: it can never mistake a literal's contents
// for a syntax element, because it operates on the typed AST, not its
// stringified form.
//
// oxigraph 0.5.9 does not expose the `spargebra::Query` it parses
// internally (`PreparedSparqlQuery.query` is a private field, `mod.rs`,
// with no accessor), so the only route to an inspectable AST is a second,
// independent parse of the same query text via `spargebra`'s own public
// `SparqlParser` -- added as a direct dependency pinned to the exact
// version oxigraph 0.5.9 already resolves transitively (see Cargo.toml).

/// Does `query` contain a `GRAPH <...> { ... }` clause anywhere in its
/// pattern tree, including nested inside a `FILTER EXISTS`/`FILTER NOT
/// EXISTS` sub-pattern? Delegates to [`graph_pattern_has_graph_clause`] on
/// every top-level query form's pattern.
fn query_has_graph_clause(query: &spargebra::Query) -> bool {
    match query {
        spargebra::Query::Select { pattern, .. }
        | spargebra::Query::Construct { pattern, .. }
        | spargebra::Query::Describe { pattern, .. }
        | spargebra::Query::Ask { pattern, .. } => graph_pattern_has_graph_clause(pattern),
    }
}

/// Exhaustive, compiler-checked recursive walk over `spargebra::algebra::
/// GraphPattern` looking for a `GraphPattern::Graph` variant. Also descends
/// into every attached `Expression` (`Filter`'s `expr`, `LeftJoin`'s
/// `expression`, `Extend`'s `expression`, `Group`'s aggregate expressions)
/// via [`expression_has_graph_clause`], because `FILTER EXISTS { ... }`
/// embeds its own `GraphPattern` that could itself contain a `GRAPH`
/// clause -- a purely-pattern-tree walk that ignored expressions would miss
/// that case.
fn graph_pattern_has_graph_clause(pattern: &spargebra::algebra::GraphPattern) -> bool {
    use spargebra::algebra::GraphPattern as GP;
    match pattern {
        GP::Graph { .. } => true,
        GP::Bgp { .. } | GP::Path { .. } | GP::Values { .. } => false,
        // `Lateral` is spargebra's `sep-0006`-gated variant; matched
        // unconditionally (not `#[cfg(feature = "sep-0006")]`) because
        // `cfg(feature = ...)` here would check *this* crate's own Cargo
        // features, not spargebra's -- and spargebra's `sep-0006` feature is
        // unconditionally enabled in every build of this crate via feature
        // unification with the `oxigraph` dependency, which requests it
        // unconditionally (see `oxigraph`'s own `Cargo.toml`). Confirmed by
        // `cargo check`: omitting this arm is a compile error (non-
        // exhaustive match), proving the variant really is always compiled
        // in here.
        GP::Join { left, right }
        | GP::Minus { left, right }
        | GP::Union { left, right }
        | GP::Lateral { left, right } => {
            graph_pattern_has_graph_clause(left) || graph_pattern_has_graph_clause(right)
        }
        GP::LeftJoin {
            left,
            right,
            expression,
        } => {
            graph_pattern_has_graph_clause(left)
                || graph_pattern_has_graph_clause(right)
                || expression.as_ref().is_some_and(expression_has_graph_clause)
        }
        GP::Filter { expr, inner } => {
            expression_has_graph_clause(expr) || graph_pattern_has_graph_clause(inner)
        }
        GP::Extend {
            inner, expression, ..
        } => graph_pattern_has_graph_clause(inner) || expression_has_graph_clause(expression),
        // `Service`'s own `inner` pattern (the block sent to the remote
        // endpoint) can itself contain a `GRAPH` clause; SERVICE handling
        // itself is a separate, already-fail-closed concern (see
        // `DeterministicGraph::query`'s doc comment) -- grouped into this
        // arm (identical body to OrderBy/Project/Distinct/Reduced/Slice)
        // only to make sure a nested GRAPH inside a SERVICE block is still
        // caught, not because it shares their semantics.
        GP::OrderBy { inner, .. }
        | GP::Project { inner, .. }
        | GP::Distinct { inner }
        | GP::Reduced { inner }
        | GP::Slice { inner, .. }
        | GP::Service { inner, .. } => graph_pattern_has_graph_clause(inner),
        GP::Group {
            inner, aggregates, ..
        } => {
            graph_pattern_has_graph_clause(inner)
                || aggregates
                    .iter()
                    .any(|(_, agg)| aggregate_expression_has_graph_clause(agg))
        }
    }
}

/// Recursive walk over `spargebra::algebra::Expression` looking for a
/// `FILTER EXISTS`/`FILTER NOT EXISTS` (`Expression::Exists`) whose embedded
/// pattern contains a `GRAPH` clause. Exhaustive over every `Expression`
/// variant.
fn expression_has_graph_clause(expr: &spargebra::algebra::Expression) -> bool {
    use spargebra::algebra::Expression as Ex;
    match expr {
        Ex::NamedNode(_) | Ex::Literal(_) | Ex::Variable(_) | Ex::Bound(_) => false,
        Ex::Or(a, b)
        | Ex::And(a, b)
        | Ex::Equal(a, b)
        | Ex::SameTerm(a, b)
        | Ex::Greater(a, b)
        | Ex::GreaterOrEqual(a, b)
        | Ex::Less(a, b)
        | Ex::LessOrEqual(a, b)
        | Ex::Add(a, b)
        | Ex::Subtract(a, b)
        | Ex::Multiply(a, b)
        | Ex::Divide(a, b) => expression_has_graph_clause(a) || expression_has_graph_clause(b),
        Ex::UnaryPlus(inner) | Ex::UnaryMinus(inner) | Ex::Not(inner) => {
            expression_has_graph_clause(inner)
        }
        Ex::In(inner, list) => {
            expression_has_graph_clause(inner) || list.iter().any(expression_has_graph_clause)
        }
        Ex::Exists(pattern) => graph_pattern_has_graph_clause(pattern),
        Ex::If(cond, then_branch, else_branch) => {
            expression_has_graph_clause(cond)
                || expression_has_graph_clause(then_branch)
                || expression_has_graph_clause(else_branch)
        }
        Ex::Coalesce(list) => list.iter().any(expression_has_graph_clause),
        Ex::FunctionCall(_, args) => args.iter().any(expression_has_graph_clause),
    }
}

/// `GROUP BY`/aggregate expressions can wrap an ordinary `Expression`
/// (e.g. `SUM(?x)`) that could in principle embed a `FILTER EXISTS { GRAPH
/// ... }` inside a more exotic construct; checked for completeness.
fn aggregate_expression_has_graph_clause(agg: &spargebra::algebra::AggregateExpression) -> bool {
    match agg {
        spargebra::algebra::AggregateExpression::CountSolutions { .. } => false,
        spargebra::algebra::AggregateExpression::FunctionCall { expr, .. } => {
            expression_has_graph_clause(expr)
        }
    }
}

/// Query-text keywords that signal a likely SPARQL UPDATE attempt (`INSERT
/// DATA`, `DELETE DATA`, `DELETE`/`INSERT ... WHERE`, `WITH ... DELETE/
/// INSERT`, `CLEAR`, `DROP`, `CREATE`, `LOAD`, `COPY`, `MOVE`, `ADD`
/// graph-management operations). Best-effort, message-only: this only
/// customizes the *error text* after `SparqlEvaluator::parse_query` has
/// already failed (see [`DeterministicGraph::query`]'s doc comment for why
/// UPDATE cannot reach execution regardless of what this sniff decides). A
/// false-positive "looks like UPDATE" match on a query that fails to parse
/// for an unrelated reason only costs a slightly mis-worded error message,
/// never a wrong refuse/allow decision.
const LIKELY_UPDATE_KEYWORDS: &[&str] = &[
    "INSERT", "DELETE", "WITH", "CLEAR", "DROP", "CREATE", "LOAD", "COPY", "MOVE", "ADD",
];

/// Case-insensitive, word-boundary-checked prefix match of `sparql`
/// (leading whitespace trimmed) against [`LIKELY_UPDATE_KEYWORDS`].
fn looks_like_sparql_update(sparql: &str) -> bool {
    let trimmed = sparql.trim_start();
    LIKELY_UPDATE_KEYWORDS.iter().any(|kw| {
        trimmed.len() >= kw.len()
            && trimmed[..kw.len()].eq_ignore_ascii_case(kw)
            && trimmed[kw.len()..]
                .chars()
                .next()
                .is_none_or(|c| !c.is_alphanumeric() && c != '_')
    })
}

/// Best-effort pre-parse diagnostic for the SPARQL 1.1 illegal rebind
/// pattern: `BIND(expr AS ?v)` where `?v` already appears earlier in the
/// group (or inside the BIND expression itself, before the `AS`).
///
/// This is an AST-free lexical scan, not a parser: it locates each
/// case-insensitive `BIND` keyword, walks its parenthesized expression at
/// paren depth 1 to the `AS` keyword, extracts the target `?variable`, and
/// refuses when that variable token occurs anywhere earlier in the query
/// text. Per SPARQL 1.1 grammar the `AS` variable of `BIND` must be fresh
/// in the group, so this scan is deliberately conservative in the direction
/// of *more* refusals only when a parse would have failed anyway — it runs
/// exclusively on the parse-failure path, never overrides a successful
/// parse, and therefore cannot turn a valid query into an error.
///
/// Returns the typed `[FM-GRAPH-013]` error, or `None` when no rebind
/// pattern is found (caller falls through to the generic parse error).
fn diagnose_bind_rebind(sparql: &str) -> Option<AppError> {
    let bytes = sparql.as_bytes();
    let is_name_char = |c: u8| c.is_ascii_alphanumeric() || c == b'_' || c >= 0x80;
    let mut from = 0usize;
    while let Some(rel) = sparql[from..]
        .to_ascii_lowercase()
        .find("bind")
        .map(|p| from + p)
    {
        let kw = rel;
        // Word-boundary check: must not be part of e.g. `UNBINDD` or a name.
        let before_ok = kw == 0 || !is_name_char(bytes[kw - 1]);
        let after_ok = bytes.get(kw + 4).is_none_or(|&c| !is_name_char(c));
        from = kw + 4;
        if !before_ok || !after_ok {
            continue;
        }
        // First `(` after BIND opens the expression.
        let Some(open_rel) = sparql[kw..].find('(') else {
            continue;
        };
        let open = kw + open_rel;
        // Walk to the `AS` keyword at paren depth 1, tracking the offset of
        // the deepest scan position so the target-variable freshness check
        // covers the expression text before `AS`.
        let mut depth: usize = 0;
        let mut as_pos: Option<usize> = None;
        let mut i = open;
        while i < bytes.len() {
            match bytes[i] {
                b'(' => depth += 1,
                b')' => {
                    if depth == 1 {
                        break;
                    }
                    depth -= 1;
                }
                b'A' | b'a'
                    if depth == 1
                        && sparql[i..].len() >= 2
                        && sparql[i + 1..].starts_with(['S', 's'])
                        && !is_name_char(bytes[i - 1])
                        && bytes.get(i + 2).is_none_or(|&c| !is_name_char(c)) =>
                {
                    as_pos = Some(i);
                    break;
                }
                _ => {}
            }
            i += 1;
        }
        let Some(as_pos) = as_pos else { continue };
        // Extract the target `?var` right after AS.
        let rest = &sparql[as_pos + 2..];
        let rest_trimmed = rest.trim_start();
        let target_rel = rest.len() - rest_trimmed.len();
        let target_abs = as_pos + 2 + target_rel;
        let target_name: String = {
            let mut cs = sparql[target_abs..].chars();
            match cs.next() {
                Some('?' | '$') => cs
                    .by_ref()
                    .take_while(|c| c.is_alphanumeric() || *c == '_')
                    .collect(),
                _ => continue,
            }
        };
        if target_name.is_empty() {
            continue;
        }
        // Freshness check: does `?target_name` occur anywhere strictly
        // before the AS keyword (group prefix or the BIND expression)?
        let prefix = &sparql[..as_pos];
        if var_token_present(prefix, &target_name) {
            let line_no = sparql[..kw].matches('\n').count() + 1;
            let line_start = prefix_line_start(sparql, kw);
            let line_text = sparql[line_start..].lines().next().unwrap_or("").trim();
            return Some(AppError::fm_graph(
                13,
                format!(
                    "BIND re-binds in-scope variable ?{target_name} (SPARQL 1.1 \
                     requires a fresh variable); rename the AS target. \
                     [at line {line_no}: {line_text}]"
                ),
            ));
        }
    }
    None
}

/// Byte offset of the start of the line containing byte offset `at`.
fn prefix_line_start(sparql: &str, at: usize) -> usize {
    sparql[..at].rfind('\n').map_or(0, |p| p + 1)
}

/// True when the variable token `?name` (or `$name`) appears in `text` with
/// a proper token boundary after the name characters.
fn var_token_present(text: &str, name: &str) -> bool {
    let needle_q = format!("?{name}");
    let needle_d = format!("${name}");
    let mut from = 0usize;
    while let Some(rel) = text[from..].find(&needle_q).map(|p| from + p) {
        let end = rel + needle_q.len();
        let boundary = text[end..]
            .chars()
            .next()
            .is_none_or(|c| !(c.is_alphanumeric() || c == '_'));
        if boundary {
            return true;
        }
        from = rel + 1;
    }
    let mut from = 0usize;
    while let Some(rel) = text[from..].find(&needle_d).map(|p| from + p) {
        let end = rel + needle_d.len();
        let boundary = text[end..]
            .chars()
            .next()
            .is_none_or(|c| !(c.is_alphanumeric() || c == '_'));
        if boundary {
            return true;
        }
        from = rel + 1;
    }
    false
}

/// Extract the source line named by a spargebra `error at L:C` message and
/// render a short context suffix for the generic `[FM-GRAPH-003]` error, so
/// the message carries the actual offending text even when the parser's
/// reported column points after the illegal construct.
fn offending_line_hint(sparql: &str, err: &impl std::fmt::Display) -> String {
    let rendered = err.to_string();
    let Some(at_idx) = rendered.find(" at ") else {
        return String::new();
    };
    let rest = &rendered[at_idx + 4..];
    let line_no: Option<usize> = rest
        .split(|c: char| !c.is_ascii_digit())
        .next()
        .and_then(|s| s.parse().ok());
    let Some(line_no) = line_no else {
        return String::new();
    };
    match sparql.lines().nth(line_no.saturating_sub(1)) {
        Some(line) => format!(" (near line {line_no}: {})", line.trim()),
        None => String::new(),
    }
}

// ---------------------------------------------------------------------------
// GraphEngine — the engine-neutral pipeline seam
// ---------------------------------------------------------------------------

/// One row of a SELECT result: bound variable name (bare, no `?`) →
/// datatype-aware [`EngineValue`]. No engine model types cross this seam
/// (the law kernel — `graphlaw`/Eyeron/PurRDF — and this crate's oxigraph
/// mirror exchange facts only as N-Triples strings, and oxigraph model
/// types never leave [`DeterministicGraph`]) — `EngineValue` is this
/// crate's own neutral scalar, not an oxigraph, purrdf or tera type.
pub type EngineRow = std::collections::BTreeMap<String, EngineValue>;

/// A datatype-aware scalar value coerced from an RDF term, produced by
/// `term_to_engine_value` and consumed by
/// `template::solutions_to_values` (the only place that turns this into a
/// `tera::Value`). Kept engine-neutral here — no `tera::Value` in this
/// module — matching this seam's existing "no engine model types cross
/// this seam" convention.
///
/// **Scope boundary (deliberate):** language-tagged literals and
/// `xsd:dateTime`/`xsd:date`-typed literals remain lossy plain strings —
/// the language tag is dropped and no date object is constructed. Only
/// `xsd:boolean` and the XSD integer/decimal/float datatype families get a
/// non-string coercion in this pass; see `term_to_engine_value`.
#[derive(Debug, Clone, PartialEq)]
pub enum EngineValue {
    /// Coerced from an `xsd:boolean` literal.
    Bool(bool),
    /// Coerced from an XSD integer-family literal (`xsd:integer`, `int`,
    /// `long`, `short`, `byte`, `nonNegativeInteger`, `positiveInteger`,
    /// `nonPositiveInteger`, `negativeInteger`, `unsignedLong`,
    /// `unsignedInt`, `unsignedShort`, `unsignedByte`). Values that do not
    /// fit an `i64` (e.g. a very large `xsd:unsignedLong`) fall back to
    /// [`EngineValue::String`] — the same fallback path as a malformed
    /// lexical value, since this pass cannot represent them as an `i64`.
    Int(i64),
    /// Coerced from an XSD decimal/float-family literal (`xsd:decimal`,
    /// `xsd:double`, `xsd:float`). Non-finite results (`NaN`/`Infinity`,
    /// legal `xsd:double`/`xsd:float` lexical values but not representable
    /// as a JSON number) fall back to [`EngineValue::String`].
    Float(f64),
    /// Everything else: plain strings, `xsd:string`, language-tagged
    /// literals, `xsd:dateTime`/`xsd:date`/`xsd:anyURI`/etc. (lossy — see
    /// the scope-boundary note above), NamedNode/IRI, `BlankNode` — plus the
    /// malformed-lexical-value and non-representable-value fallback paths
    /// for the other three variants.
    String(String),
}

/// One triple of a CONSTRUCT/DESCRIBE result, pre-rendered both ways the
/// pipeline consumes it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EngineTriple {
    /// Subject in N-Triples form (`<iri>` / `_:b`).
    pub subject: String,
    /// Predicate in N-Triples form.
    pub predicate: String,
    /// Object as a plain value string (literal lexical form / bare IRI).
    pub object_value: String,
    /// The whole triple in N-Triples form without the terminating ` .`,
    /// suitable for re-insertion via [`GraphEngine::insert_turtle`].
    pub ntriples: String,
}

/// Engine-neutral SPARQL results: what the template layer and the sync
/// pipeline consume instead of any engine's own result types.
// `PartialEq`-only (no `Eq`): `EngineQueryResults::Solutions` carries
// `EngineRow` values that may be `EngineValue::Float(f64)`, and `f64` has
// no total order / no `Eq` impl in Rust. Nothing in this crate compares
// these types via `Eq`-bounded generics (only `assert_eq!`, which needs
// `PartialEq + Debug`) — verified by grepping every consumer before this
// derive was narrowed.
#[derive(Debug, Clone, PartialEq)]
pub enum EngineQueryResults {
    /// ASK result.
    Boolean(bool),
    /// SELECT rows.
    Solutions(Vec<EngineRow>),
    /// CONSTRUCT/DESCRIBE triples.
    Graph(Vec<EngineTriple>),
}

/// Outcome of a SHACL validation run at the law gate.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ShaclOutcome {
    /// `sh:conforms` — true iff no validation result was produced.
    pub conforms: bool,
    /// One human-readable line per violation, each naming its focus node.
    pub violations: Vec<String>,
}

/// Outcome of a materialization pass.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct MaterializeOutcome {
    /// Derived triples (N-Triples lines without the terminating ` .`)
    /// that were new relative to the pre-materialization state, sorted.
    pub derived: Vec<String>,
    /// Number of rules loaded into the reasoner for this pass.
    pub rules_loaded: usize,
}

/// The pipeline's graph surface: RDF facts + SPARQL plus the law-state
/// operations (N3/Datalog materialization, SHACL/ShEx gates, denials).
///
/// `state_hash` has a provided default so both engines hash identically:
/// BLAKE3 over the canonically sorted quads returned by `canonical_quads`,
/// joined by `\n` — computed ggen-side, never delegated to an engine.
pub trait GraphEngine: Send + Sync {
    /// Parse and insert Turtle content, returning the number of quads added.
    ///
    /// # Errors
    /// Fails closed on syntax or storage errors.
    fn insert_turtle(&self, ttl: &str) -> Result<usize>;

    /// Parse multiple Turtle documents independently, then atomically insert
    /// their union through one store mutation.
    ///
    /// Document-local parser state (prefixes, bases, blank-node labels) is
    /// never shared. Implementations may optimize transaction fan-in while
    /// preserving exact document attribution on refusal.
    ///
    /// # Errors
    /// Fails closed before store mutation on any document parse error, or if
    /// the atomic store commit fails.
    fn insert_turtle_documents(
        &self, documents: &[TurtleDocument<'_>],
    ) -> Result<OntologyBatchReceipt>;

    /// Execute a SPARQL query, returning engine-neutral results.
    ///
    /// # Errors
    /// Fails closed on parse or evaluation errors.
    fn query(&self, sparql: &str) -> Result<EngineQueryResults>;

    /// Sorted canonical N-Quads lines of the current state (blank nodes
    /// relabelled `c14n{i}`; see module docs).
    ///
    /// # Errors
    /// Fails closed on iteration or canonicalization errors.
    fn canonical_quads(&self) -> Result<Vec<String>>;

    /// BLAKE3 hash of the canonicalized graph state (determinism invariant:
    /// identical for isomorphic graphs regardless of engine).
    ///
    /// # Errors
    /// Propagates [`GraphEngine::canonical_quads`] failures.
    fn state_hash(&self) -> Result<[u8; 32]> {
        let lines = self.canonical_quads()?;
        Ok(*blake3::hash(lines.join("\n").as_bytes()).as_bytes())
    }

    /// Load N3/Datalog rules (N3 rule syntax), returning how many rules
    /// the document contributed.
    ///
    /// # Errors
    /// Typed refusal when the engine has no rule support, or on parse errors.
    fn load_rules(&self, rules: &str) -> Result<usize>;

    /// Load a `kh:` Knowledge Hook pack (Turtle content: `kh:Hook`/
    /// `kh:Action` individuals, e.g. a pack's `hook.ttl`), registering its
    /// hooks so a subsequent [`GraphEngine::materialize`] fires them over
    /// the union graph.
    ///
    /// # Errors
    /// Typed refusal when the engine has no knowledge-hook support, or on
    /// parse errors in the hook document.
    fn load_hook_pack(&self, hook_ttl: &str) -> Result<()>;

    /// Forward-chain all loaded rules to fixpoint, folding derived facts
    /// back into the queryable state.
    ///
    /// # Errors
    /// Typed refusal when the engine has no rule support.
    fn materialize(&self) -> Result<MaterializeOutcome>;

    /// Validate the current facts against a Turtle SHACL shapes graph.
    ///
    /// # Errors
    /// Typed refusal when the engine has no SHACL support, or on shape
    /// parse errors.
    fn validate_shacl(&self, shapes_turtle: &str) -> Result<ShaclOutcome>;

    /// Validate `(focus node, shape)` pairs against a `ShExC` schema.
    ///
    /// # Errors
    /// Typed refusal when the engine has no `ShEx` support, or on schema
    /// parse errors.
    fn validate_shex(
        &self, schema_shexc: &str, shape_map: &[(String, String)],
    ) -> Result<ShaclOutcome>;

    /// Evaluate every loaded denial rule (`{ body } => false.`) against the
    /// current (post-materialization) facts; one line per violated denial.
    ///
    /// # Errors
    /// Typed refusal when the engine has no rule support.
    fn check_denials(&self) -> Result<Vec<String>>;
}

/// Plain value string for an oxigraph term: literals as their lexical form,
/// IRIs as the bare IRI, everything else in N-Triples form. (Moved here from
/// `template.rs` so both the engine impls and the template layer share one
/// rendering.) Used for [`EngineTriple`] (CONSTRUCT/DESCRIBE) — that path
/// stays plain-string, unaffected by `term_to_engine_value`'s SELECT-row
/// coercion; see the module docs on `EngineTriple::object_value`.
///
/// `relabel` is the blank-node canonicalization map from
/// [`blank_node_relabel_map`] (the same scheme `canonical_quads` uses) —
/// a `Term::BlankNode` is rewritten through it before rendering, so a
/// blank-node-valued term is a stable string across independent loads of
/// byte-identical input instead of the store's raw parser-assigned id.
pub(crate) fn term_value(term: &Term, relabel: &HashMap<BlankNode, BlankNode>) -> String {
    match term {
        Term::Literal(lit) => lit.value().to_string(),
        Term::NamedNode(n) => n.as_str().to_string(),
        Term::BlankNode(b) => relabel.get(b).unwrap_or(b).to_string(),
        Term::Triple(triple) => triple.to_string(),
    }
}

/// `term`'s full N-Triples/Turtle/SPARQL rendering (`<iri>`, quoted literal,
/// or `_:label`), with a `Term::BlankNode` relabelled through `relabel`
/// exactly as [`term_value`] does. Used for [`EngineTriple::ntriples`], the
/// re-insertable whole-triple form — unlike [`term_value`], literals and
/// IRIs keep their full syntactic form here (quotes, datatype, angle
/// brackets), not the bare lexical/IRI string.
fn term_ntriples_relabeled(term: &Term, relabel: &HashMap<BlankNode, BlankNode>) -> String {
    match term {
        Term::BlankNode(b) => relabel.get(b).unwrap_or(b).to_string(),
        other => other.to_string(),
    }
}

/// `node`'s N-Triples/Turtle/SPARQL rendering (`<iri>` or `_:label`), with a
/// `NamedOrBlankNode::BlankNode` relabelled through `relabel`. Used for
/// [`EngineTriple::subject`] and the subject half of
/// [`EngineTriple::ntriples`].
fn named_or_blank_ntriples(
    node: &NamedOrBlankNode, relabel: &HashMap<BlankNode, BlankNode>,
) -> String {
    match node {
        NamedOrBlankNode::NamedNode(n) => n.to_string(),
        NamedOrBlankNode::BlankNode(b) => relabel.get(b).unwrap_or(b).to_string(),
    }
}

/// XSD namespace prefix, spelled out once for the datatype tables below.
const XSD: &str = "http://www.w3.org/2001/XMLSchema#";

/// `xsd:boolean` — coerced to [`EngineValue::Bool`].
const XSD_BOOLEAN: &str = "http://www.w3.org/2001/XMLSchema#boolean";

/// XSD integer-family datatypes coerced to [`EngineValue::Int`] by
/// `term_to_engine_value` (the XSD 1.1 built-in integer subtypes).
const XSD_INTEGER_DATATYPES: &[&str] = &[
    "integer",
    "int",
    "long",
    "short",
    "byte",
    "nonNegativeInteger",
    "positiveInteger",
    "nonPositiveInteger",
    "negativeInteger",
    "unsignedLong",
    "unsignedInt",
    "unsignedShort",
    "unsignedByte",
];

/// XSD decimal/float-family datatypes coerced to [`EngineValue::Float`] by
/// `term_to_engine_value`.
const XSD_FLOAT_DATATYPES: &[&str] = &["decimal", "double", "float"];

/// Datatype-aware coercion of an oxigraph [`Term`] for the SELECT-row
/// pipeline (`EngineRow`, consumed by `template::solutions_to_values`):
///
/// - `xsd:boolean` → [`EngineValue::Bool`] (`"true"`/`"1"` → `true`,
///   `"false"`/`"0"` → `false`, matching the XSD 1.1 boolean lexical space).
/// - Integer-family XSD datatypes (`xsd:integer`, `int`, `long`, `short`,
///   `byte`, `nonNegativeInteger`, `positiveInteger`, `nonPositiveInteger`,
///   `negativeInteger`, `unsignedLong`/`unsignedInt`/`unsignedShort`/
///   `unsignedByte`) → [`EngineValue::Int`] (parsed as `i64`).
/// - Decimal/float-family XSD datatypes (`xsd:decimal`, `xsd:double`,
///   `xsd:float`) → [`EngineValue::Float`] (parsed as `f64`; non-finite
///   results fall back the same way a parse failure does).
/// - Everything else (plain strings, `xsd:string`, language-tagged
///   literals, `xsd:dateTime`/`xsd:date`/`xsd:anyURI`/etc., NamedNode/IRI,
///   `BlankNode`) → [`EngineValue::String`], byte-identical to
///   [`term_value`]'s output — backward compatible with every existing
///   `{{ row.field }}`-style template access.
///
/// **Malformed-value fallback:** when a literal's lexical form does not
/// parse under its *declared* datatype (e.g. `"not-a-number"^^xsd:integer`)
/// or parses to a non-finite float, this function falls back to
/// [`EngineValue::String`] with the original lexical form — a malformed
/// literal is a data-quality issue in the source graph, not a reason to
/// hard-crash template rendering — but the fallback is never silent: a
/// `tracing::warn!` names the offending value and datatype every time it
/// triggers.
///
/// **Scope boundary (deliberate, not an oversight):** language-tagged
/// literals and `xsd:dateTime`/`xsd:date`-typed literals remain lossy plain
/// strings in this pass — the language tag is dropped, no date object is
/// constructed. See [`EngineValue`]'s own doc comment for the same note.
///
/// `relabel` is threaded straight through to [`term_value`] for the
/// NamedNode/BlankNode fallback branch below — see that function's doc
/// comment for the blank-node canonicalization it applies.
pub(crate) fn term_to_engine_value(
    term: &Term, relabel: &HashMap<BlankNode, BlankNode>,
) -> EngineValue {
    let Term::Literal(lit) = term else {
        // NamedNode/BlankNode/etc.: identical to `term_value`.
        return EngineValue::String(term_value(term, relabel));
    };
    let datatype = lit.datatype().as_str();
    let lexical = lit.value();

    if datatype == XSD_BOOLEAN {
        return match lexical {
            "true" | "1" => EngineValue::Bool(true),
            "false" | "0" => EngineValue::Bool(false),
            _ => {
                tracing::warn!(
                    value = lexical,
                    datatype,
                    "malformed xsd:boolean literal; falling back to plain string"
                );
                EngineValue::String(lexical.to_string())
            }
        };
    }

    if let Some(suffix) = datatype.strip_prefix(XSD) {
        if XSD_INTEGER_DATATYPES.contains(&suffix) {
            return match lexical.parse::<i64>() {
                Ok(n) => EngineValue::Int(n),
                Err(e) => {
                    tracing::warn!(
                        value = lexical,
                        datatype,
                        error = %e,
                        "malformed integer-family literal; falling back to plain string"
                    );
                    EngineValue::String(lexical.to_string())
                }
            };
        }
        if XSD_FLOAT_DATATYPES.contains(&suffix) {
            return match lexical.parse::<f64>() {
                Ok(n) if n.is_finite() => EngineValue::Float(n),
                Ok(n) => {
                    tracing::warn!(
                        value = lexical,
                        datatype,
                        parsed = n,
                        "non-finite decimal/float literal (NaN/Infinity); falling back to plain string"
                    );
                    EngineValue::String(lexical.to_string())
                }
                Err(e) => {
                    tracing::warn!(
                        value = lexical,
                        datatype,
                        error = %e,
                        "malformed decimal/float-family literal; falling back to plain string"
                    );
                    EngineValue::String(lexical.to_string())
                }
            };
        }
    }

    // xsd:string, language-tagged literals, dateTime/date/anyURI/etc. — the
    // documented lossy-string scope boundary.
    EngineValue::String(lexical.to_string())
}

impl GraphEngine for DeterministicGraph {
    fn insert_turtle(&self, ttl: &str) -> Result<usize> {
        DeterministicGraph::insert_turtle(self, ttl)
    }

    fn insert_turtle_documents(
        &self, documents: &[TurtleDocument<'_>],
    ) -> Result<OntologyBatchReceipt> {
        ontology_batch::insert_documents(self, documents)
    }

    fn query(&self, sparql: &str) -> Result<EngineQueryResults> {
        match DeterministicGraph::query(self, sparql)? {
            QueryResults::Boolean(b) => Ok(EngineQueryResults::Boolean(b)),
            QueryResults::Solutions(solutions) => {
                // Blank-node projection determinism: relabel any bound blank
                // node through the SAME canonicalization scheme
                // `canonical_quads`/`state_hash` already use (bounded color
                // refinement over the whole graph, see module docs), so a
                // blank-node-valued SELECT binding a template consumes is a
                // stable label across independent loads of byte-identical
                // input, not the store's raw parser-assigned internal id
                // (which oxigraph regenerates fresh on every Turtle load).
                let relabel = blank_node_relabel_map(&self.all_quads()?)?;
                let mut rows = Vec::new();
                for solution in solutions {
                    let solution = solution.map_err(|e| {
                        AppError::fm_graph(3, format!("SELECT solution iteration failed: {e}"))
                    })?;
                    let mut row = EngineRow::new();
                    for (var, term) in solution.iter() {
                        row.insert(
                            var.as_str().to_string(),
                            term_to_engine_value(term, &relabel),
                        );
                    }
                    rows.push(row);
                }
                Ok(EngineQueryResults::Solutions(rows))
            }
            QueryResults::Graph(triples) => {
                // Same canonicalization applied to CONSTRUCT/DESCRIBE
                // results -- a blank-node-valued triple position is exactly
                // as template-reachable (`EngineTriple::subject`/
                // `object_value`/`ntriples`) as a SELECT binding.
                let relabel = blank_node_relabel_map(&self.all_quads()?)?;
                let mut out = Vec::new();
                for triple in triples {
                    let triple = triple.map_err(|e| {
                        AppError::fm_graph(3, format!("CONSTRUCT triple iteration failed: {e}"))
                    })?;
                    out.push(EngineTriple {
                        subject: named_or_blank_ntriples(&triple.subject, &relabel),
                        predicate: triple.predicate.to_string(),
                        object_value: term_value(&triple.object, &relabel),
                        ntriples: format!(
                            "{} {} {}",
                            named_or_blank_ntriples(&triple.subject, &relabel),
                            triple.predicate,
                            term_ntriples_relabeled(&triple.object, &relabel)
                        ),
                    });
                }
                Ok(EngineQueryResults::Graph(out))
            }
        }
    }

    fn canonical_quads(&self) -> Result<Vec<String>> {
        canonical_nquad_lines(&self.all_quads()?)
    }

    fn load_rules(&self, _rules: &str) -> Result<usize> {
        Err(AppError::fm_law(
            1,
            "the oxigraph engine has no N3/Datalog rule support. \
             Remediation: run with the default GraphLaw engine.",
        ))
    }

    fn materialize(&self) -> Result<MaterializeOutcome> {
        Err(AppError::fm_law(
            1,
            "the oxigraph engine cannot materialize rules. \
             Remediation: run with the default GraphLaw engine.",
        ))
    }

    fn validate_shacl(&self, _shapes_turtle: &str) -> Result<ShaclOutcome> {
        Err(AppError::fm_law(
            2,
            "the oxigraph engine has no SHACL support. \
             Remediation: run with the default GraphLaw engine.",
        ))
    }

    fn validate_shex(
        &self, _schema_shexc: &str, _shape_map: &[(String, String)],
    ) -> Result<ShaclOutcome> {
        Err(AppError::fm_law(
            3,
            "the oxigraph engine has no ShEx support. \
             Remediation: run with the default GraphLaw engine.",
        ))
    }

    fn check_denials(&self) -> Result<Vec<String>> {
        Err(AppError::fm_law(
            1,
            "the oxigraph engine has no denial-rule support. \
             Remediation: run with the default GraphLaw engine.",
        ))
    }

    fn load_hook_pack(&self, _hook_ttl: &str) -> Result<()> {
        Err(AppError::fm_law(
            15,
            "the oxigraph engine has no knowledge-hook support. \
             Remediation: run with the default GraphLaw engine.",
        ))
    }
}

// ---------------------------------------------------------------------------
// GraphLawStore — graphlaw as the live law-state engine
// ---------------------------------------------------------------------------

/// The default [`GraphEngine`]: the `graphlaw` kernel (Eyeron N3 forward
/// chaining, `PurRDF` SHACL/ShEx evaluation, SPARQL `kh:` hook packs) as the
/// law-state engine — N3 rule materialization, SHACL/ShEx gates, denial
/// checks — layered over a [`DeterministicGraph`] mirror that answers
/// SPARQL 1.1 and provides the canonical BLAKE3 state hash.
///
/// Division of labor (and why): the law kernel and the oxigraph mirror are
/// one stack now (graphlaw 0.x with purrdf, both workspace path deps), and
/// facts flow between them only as N-Triples strings — `law_engine.rs`'s
/// helpers (`n3_run`/`hooks_apply`/`shacl_check`/`shex_check`/
/// `hook_pack_check`) are the single graphlaw call surface, and no engine
/// model type crosses the seam. The mirror is the *queryable* state; the
/// graphlaw kernel is the *law* state. Every derived fact enters the mirror
/// exclusively through [`GraphEngine::materialize`], i.e. through the
/// kernel — a `when:` ASK that only passes after materialization is proof
/// the reasoner is in the loop (see `tests/graphlaw_e2e.rs`).
pub struct GraphLawStore {
    mirror: DeterministicGraph,
    law: std::sync::Mutex<LawState>,
}

/// Interior law-side state. The kernel is stateless between calls: every
/// `materialize()`/`validate_*`/`check_denials` re-runs `n3_run`/
/// `hooks_apply` over the mirror's canonical N-Triples so rules always see
/// the full fact state.
#[derive(Default)]
struct LawState {
    /// Raw N3 rule documents, in load order.
    rules_src: Vec<String>,
    /// Raw `kh:` Knowledge Hook pack documents (Turtle), in load order —
    /// replayed at each `materialize()` (via `hooks_apply`) or a hook pack
    /// loaded via [`GraphEngine::load_hook_pack`] would silently vanish
    /// after the first materialize.
    hooks_src: Vec<String>,
    /// Number of N3 rules (including denial fuses) in the loaded rule
    /// documents.
    rules_loaded: usize,
}

impl GraphLawStore {
    /// Create an empty `GraphLaw` engine.
    ///
    /// # Errors
    /// Propagates mirror-store initialization failures (`[FM-GRAPH-001]`).
    pub fn new() -> Result<Self> {
        Ok(Self {
            mirror: DeterministicGraph::new()?,
            law: std::sync::Mutex::new(LawState::default()),
        })
    }

    /// Lock the law state, converting a poisoned lock into a typed refusal
    /// (never a panic).
    fn law_state(&self) -> Result<std::sync::MutexGuard<'_, LawState>> {
        self.law.lock().map_err(|_| {
            AppError::fm_law(
                4,
                "law-state lock poisoned by a prior panic; refusing to continue",
            )
        })
    }

    /// The current fact state as one N-Triples document (canonical order).
    fn mirror_ntriples(&self) -> Result<String> {
        use std::fmt::Write as _;
        let lines = canonical_nquad_lines(&self.mirror.all_quads()?)?;
        Ok(lines.iter().fold(String::new(), |mut doc, l| {
            let _ = writeln!(doc, "{l} .");
            doc
        }))
    }
}

impl GraphEngine for GraphLawStore {
    fn insert_turtle(&self, ttl: &str) -> Result<usize> {
        self.mirror.insert_turtle(ttl)
    }

    fn insert_turtle_documents(
        &self, documents: &[TurtleDocument<'_>],
    ) -> Result<OntologyBatchReceipt> {
        ontology_batch::insert_documents(&self.mirror, documents)
    }

    fn query(&self, sparql: &str) -> Result<EngineQueryResults> {
        GraphEngine::query(&self.mirror, sparql)
    }

    fn canonical_quads(&self) -> Result<Vec<String>> {
        GraphEngine::canonical_quads(&self.mirror)
    }

    fn load_rules(&self, rules: &str) -> Result<usize> {
        // Parse eagerly so a bad rule document is refused at load time,
        // not at the later materialize call (same convention as before).
        let probe = crate::law_engine::n3_run("", &[rules]).map_err(|e| match e {
            crate::law_engine::N3Failure::Load(m) | crate::law_engine::N3Failure::Reason(m) => {
                AppError::fm_law(6, format!("GraphLaw rule load refused: {m}"))
            }
        })?;
        self.law_state()?.rules_src.push(rules.to_string());
        Ok(probe.rules)
    }

    fn materialize(&self) -> Result<MaterializeOutcome> {
        let mut state = self.law_state()?;
        let nt = self.mirror_ntriples()?;
        let rules: Vec<&str> = state.rules_src.iter().map(String::as_str).collect();
        let run = crate::law_engine::n3_run(&nt, &rules).map_err(|e| match e {
            crate::law_engine::N3Failure::Load(m) => {
                AppError::fm_law(6, format!("GraphLaw rule load refused: {m}"))
            }
            crate::law_engine::N3Failure::Reason(m) => {
                AppError::fm_law(9, format!("Reasoner materialize failed: {m}"))
            }
        })?;
        state.rules_loaded = run.rules;
        if let Some(fuse) = run.fuse {
            // A denial rule fired mid-derivation; Eyeron clears the
            // derivation in that case (same convention as
            // `law_engine::check_denials`). The violation itself surfaces
            // through `check_denials`, which re-runs the same kernel.
            let _ = fuse;
            return Ok(MaterializeOutcome {
                derived: Vec::new(),
                rules_loaded: run.rules,
            });
        }

        // Fold derived facts back into the mirror through the one canonical
        // door (Turtle/N-Triples parse) — an underived/duplicate line is
        // idempotent; an unparseable derived term is a typed refusal, never
        // silently dropped.
        let before: std::collections::BTreeSet<String> =
            self.canonical_quads()?.into_iter().collect();
        if !run.derived_nt.trim().is_empty() {
            self.mirror.insert_turtle(&run.derived_nt).map_err(|e| {
                AppError::fm_law(
                    7,
                    format!(
                        "derived facts from the GraphLaw reasoner are not valid \
                         N-Triples/Turtle: {e}. Derived document:\n{}",
                        run.derived_nt
                    ),
                )
            })?;
        }

        // Fire the loaded `kh:` hook packs to fixpoint over the
        // post-derivation state; hook-derived facts enter the mirror
        // through the same canonical door (idempotent re-insert plus the
        // new firing markers/facts).
        if !state.hooks_src.is_empty() {
            let facts_now = self.mirror_ntriples()?;
            let hook_outcome = crate::law_engine::hooks_apply(&facts_now, &state.hooks_src)
                .map_err(|e| AppError::fm_law(9, format!("Reasoner materialize failed: {e}")))?;
            // A refuse-effect hook's verdict maps onto the same DENIED-class
            // surface the N3 fuse path produces: fail-closed for sync, naming
            // the hook and its `kh:reason` — never silently dropped.
            if !hook_outcome.denied.is_empty() {
                return Err(AppError::fm_law(
                    16,
                    format!(
                        "GraphLaw hook refuse verdict denied: {}",
                        hook_outcome.denied.join("; ")
                    ),
                ));
            }
            self.mirror
                .insert_turtle(&hook_outcome.state_nt)
                .map_err(|e| {
                    AppError::fm_law(
                        7,
                        format!(
                            "hook-derived facts from the graphlaw kernel are not valid \
                         N-Triples/Turtle: {e}. Hook state:\n{}",
                            hook_outcome.state_nt
                        ),
                    )
                })?;
        }

        let after = self.canonical_quads()?;
        let derived_new: Vec<String> = after.into_iter().filter(|l| !before.contains(l)).collect();

        Ok(MaterializeOutcome {
            derived: derived_new,
            rules_loaded: run.rules,
        })
    }

    fn validate_shacl(&self, shapes_turtle: &str) -> Result<ShaclOutcome> {
        // Validate against the full current fact state (post-materialization
        // if materialize ran) via the shared kernel helper.
        let nt = self.mirror_ntriples()?;
        crate::law_engine::shacl_check(&nt, shapes_turtle)
            .map_err(|e| AppError::fm_law(8, format!("SHACL shapes graph refused: {e}")))
    }

    fn validate_shex(
        &self, schema_shexc: &str, shape_map: &[(String, String)],
    ) -> Result<ShaclOutcome> {
        let nt = self.mirror_ntriples()?;
        crate::law_engine::shex_check(&nt, schema_shexc, shape_map)
            .map_err(|e| AppError::fm_law(9, format!("ShExC schema refused: {e}")))
    }

    fn check_denials(&self) -> Result<Vec<String>> {
        let state = self.law_state()?;
        let nt = self.mirror_ntriples()?;
        let rules: Vec<&str> = state.rules_src.iter().map(String::as_str).collect();
        let run = crate::law_engine::n3_run(&nt, &rules).map_err(|e| match e {
            crate::law_engine::N3Failure::Load(m) => {
                AppError::fm_law(6, format!("GraphLaw rule load refused: {m}"))
            }
            crate::law_engine::N3Failure::Reason(m) => {
                AppError::fm_law(9, format!("Reasoner materialize failed: {m}"))
            }
        })?;
        Ok(run.fuse.into_iter().collect())
    }

    fn load_hook_pack(&self, hook_ttl: &str) -> Result<()> {
        // Parse (and hook-compile) eagerly via the shared kernel helper so
        // a bad hook document is refused at load time, not at the later
        // materialize call — same convention as `load_rules` above.
        crate::law_engine::hook_pack_check(hook_ttl).map_err(|e| {
            // `kh:effect "refuse"` packs are legal since graphlaw gained
            // `Effect::Refuse`/`Verdict::Refuse`; their verdicts surface as a
            // DENIED-class outcome at materialize (fail-closed for sync), so
            // only a genuinely malformed document is refused here.
            AppError::fm_law(16, format!("GraphLaw hook pack load refused: {e}"))
        })?;
        self.law_state()?.hooks_src.push(hook_ttl.to_string());
        Ok(())
    }
}

/// A set difference between two graph states, expressed as canonical
/// N-Quads strings.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Delta {
    /// Canonical N-Quads present in the target but not the baseline (sorted).
    pub additions: Vec<String>,
    /// Canonical N-Quads present in the baseline but not the target (sorted).
    pub deletions: Vec<String>,
}

impl Delta {
    /// Compute the delta that transforms `baseline` into `target`.
    ///
    /// # Errors
    /// Propagates canonicalization/iteration failures from either graph.
    pub fn compute(baseline: &DeterministicGraph, target: &DeterministicGraph) -> Result<Self> {
        let base: BTreeSet<String> = canonical_nquad_lines(&baseline.all_quads()?)?
            .into_iter()
            .collect();
        let tgt: BTreeSet<String> = canonical_nquad_lines(&target.all_quads()?)?
            .into_iter()
            .collect();
        Ok(Self {
            additions: tgt.difference(&base).cloned().collect(),
            deletions: base.difference(&tgt).cloned().collect(),
        })
    }

    /// Apply this delta to `graph`: remove the deletions, insert the additions.
    ///
    /// Deletions are matched by canonical N-Quads string against the graph's
    /// current canonicalization; additions are parsed as N-Quads and inserted.
    ///
    /// # Errors
    /// Returns `[FM-GRAPH-006]` if a deletion does not match any quad in the
    /// graph (fail closed), or on parse/storage errors while applying.
    pub fn apply(&self, graph: &DeterministicGraph) -> Result<()> {
        if !self.deletions.is_empty() {
            let quads = graph.all_quads()?;
            let pairs = canonical_pairs(&quads)?;
            let by_canonical: HashMap<&str, &Quad> =
                pairs.iter().map(|(s, q)| (s.as_str(), q)).collect();
            for del in &self.deletions {
                let quad = by_canonical.get(del.as_str()).ok_or_else(|| {
                    AppError::fm_graph(
                        6,
                        format!("deletion not present in graph (refusing partial apply): {del}"),
                    )
                })?;
                graph.store.remove(quad.as_ref()).map_err(|e| {
                    AppError::fm_graph(6, format!("failed to remove quad `{del}`: {e}"))
                })?;
            }
        }
        if !self.additions.is_empty() {
            // `Quad`'s Display form omits the terminating ` .` required by
            // the N-Quads grammar; add it per line before parsing.
            use std::fmt::Write as _;
            let doc: String = self.additions.iter().fold(String::new(), |mut doc, a| {
                let _ = writeln!(doc, "{a} .");
                doc
            });
            graph
                .store
                .load_from_slice(RdfFormat::NQuads, doc.as_str())
                .map_err(|e| {
                    AppError::fm_graph(6, format!("failed to insert delta additions: {e}"))
                })?;
        }
        // Same blank-node canonicalization every other mutation path runs
        // (see `DeterministicGraph::canonicalize_blank_nodes`'s doc
        // comment) — this method writes straight to `graph.store`,
        // bypassing `insert_turtle`, so it must call this itself.
        graph.canonicalize_blank_nodes()?;
        Ok(())
    }

    /// The inverse delta: additions and deletions swapped, so that applying
    /// `self` then `self.inverse()` is a net no-op.
    #[must_use]
    pub fn inverse(&self) -> Self {
        Self {
            additions: self.deletions.clone(),
            deletions: self.additions.clone(),
        }
    }

    /// Compose two deltas: the result of applying `self` first, then `other`.
    ///
    /// Cancellation semantics: an addition in `self` removed again by a
    /// deletion in `other` vanishes from the composite (and vice versa).
    /// Output vectors are sorted and deduplicated (deterministic).
    #[must_use]
    pub fn compose(&self, other: &Delta) -> Self {
        let a1: BTreeSet<&String> = self.additions.iter().collect();
        let d1: BTreeSet<&String> = self.deletions.iter().collect();
        let a2: BTreeSet<&String> = other.additions.iter().collect();
        let d2: BTreeSet<&String> = other.deletions.iter().collect();

        // Net additions: self's additions that survive other's deletions,
        // plus other's additions that are not merely undoing self's deletions.
        let additions: BTreeSet<String> = a1
            .iter()
            .filter(|s| !d2.contains(**s))
            .chain(a2.iter().filter(|s| !d1.contains(**s)))
            .map(|s| (**s).clone())
            .collect();
        // Net deletions, symmetrically.
        let deletions: BTreeSet<String> = d1
            .iter()
            .filter(|s| !a2.contains(**s))
            .chain(d2.iter().filter(|s| !a1.contains(**s)))
            .map(|s| (**s).clone())
            .collect();

        Self {
            additions: additions.into_iter().collect(),
            deletions: deletions.into_iter().collect(),
        }
    }

    /// True when the delta contains no additions and no deletions.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.additions.is_empty() && self.deletions.is_empty()
    }

    /// BLAKE3 hash of the delta: sorted additions prefixed `+`, then sorted
    /// deletions prefixed `-`.
    #[must_use]
    pub fn hash(&self) -> [u8; 32] {
        let mut additions = self.additions.clone();
        additions.sort();
        let mut deletions = self.deletions.clone();
        deletions.sort();
        let mut hasher = blake3::Hasher::new();
        for a in &additions {
            hasher.update(b"+");
            hasher.update(a.as_bytes());
        }
        for d in &deletions {
            hasher.update(b"-");
            hasher.update(d.as_bytes());
        }
        *hasher.finalize().as_bytes()
    }
}

/// Sorted canonical N-Quads lines for a quad slice.
fn canonical_nquad_lines(quads: &[Quad]) -> Result<Vec<String>> {
    Ok(canonical_pairs(quads)?
        .into_iter()
        .map(|(s, _)| s)
        .collect())
}

/// Canonicalize quads, returning `(canonical N-Quads string, original quad)`
/// pairs sorted by the canonical string.
///
/// Blank nodes are relabelled `c14n{i}` after bounded color refinement so
/// the canonical strings are stable across blank-node renamings.
fn canonical_pairs(quads: &[Quad]) -> Result<Vec<(String, Quad)>> {
    let relabel = blank_node_relabel_map(quads)?;

    let mut pairs: Vec<(String, Quad)> = quads
        .iter()
        .map(|q| (relabel_quad(q, &relabel).to_string(), q.clone()))
        .collect();
    pairs.sort_by(|a, b| a.0.cmp(&b.0));
    Ok(pairs)
}

/// Compute the blank-node canonicalization map for `quads`: every blank node
/// appearing anywhere in the slice mapped to its stable `c14n{i}` label via
/// bounded color refinement (see [`canonical_blank_node_map`]).
///
/// This is the single shared entry point for blank-node canonicalization in
/// this module — [`canonical_pairs`] (receipt/content hashing) and
/// [`GraphEngine::query`]'s SELECT/CONSTRUCT projection (template-reachable
/// blank-node values) both call this instead of each maintaining their own
/// relabeling logic, so a blank node's canonical label is identical whether
/// it reaches a receipt hash or a template.
///
/// # Errors
/// Propagates [`canonical_blank_node_map`] failures (a `c14n{i}` label
/// rejected as an invalid blank-node identifier — practically unreachable
/// for any finite `i`, but not `unwrap`-away-able).
fn blank_node_relabel_map(quads: &[Quad]) -> Result<HashMap<BlankNode, BlankNode>> {
    let blank_nodes = collect_blank_nodes(quads);
    if blank_nodes.is_empty() {
        Ok(HashMap::new())
    } else {
        canonical_blank_node_map(quads, &blank_nodes)
    }
}

/// Collect every blank node appearing in subject, object, or graph position.
fn collect_blank_nodes(quads: &[Quad]) -> HashSet<BlankNode> {
    let mut blanks = HashSet::new();
    for q in quads {
        if let NamedOrBlankNode::BlankNode(b) = &q.subject {
            blanks.insert(b.clone());
        }
        if let Term::BlankNode(b) = &q.object {
            blanks.insert(b.clone());
        }
        if let GraphName::BlankNode(b) = &q.graph_name {
            blanks.insert(b.clone());
        }
    }
    blanks
}

/// Bounded color refinement: compute stable BLAKE3 neighborhood signatures
/// for every blank node, then map each to a `c14n{i}` label in signature
/// order (ties broken by original label for determinism within this graph).
fn canonical_blank_node_map(
    quads: &[Quad], blank_nodes: &HashSet<BlankNode>,
) -> Result<HashMap<BlankNode, BlankNode>> {
    let mut labels: HashMap<BlankNode, String> = blank_nodes
        .iter()
        .map(|b| (b.clone(), "bnode".to_string()))
        .collect();

    for _ in 0..REFINEMENT_ITERATIONS {
        let mut next: HashMap<BlankNode, String> = HashMap::new();
        for bnode in blank_nodes {
            let mut neighborhood: Vec<String> = quads
                .iter()
                .filter(|q| quad_touches(q, bnode))
                .map(|q| neighborhood_line(q, bnode, &labels))
                .collect();
            neighborhood.sort();
            let signature = blake3::hash(neighborhood.join("\n").as_bytes())
                .to_hex()
                .to_string();
            next.insert(bnode.clone(), signature);
        }
        labels = next;
    }

    let mut ordered: Vec<&BlankNode> = blank_nodes.iter().collect();
    ordered.sort_by(|a, b| {
        let la = labels.get(*a).map_or("", String::as_str);
        let lb = labels.get(*b).map_or("", String::as_str);
        la.cmp(lb).then_with(|| a.to_string().cmp(&b.to_string()))
    });

    let mut map = HashMap::new();
    for (idx, bnode) in ordered.into_iter().enumerate() {
        let canonical = BlankNode::new(format!("c14n{idx}")).map_err(|e| {
            AppError::fm_graph(5, format!("canonical blank node label rejected: {e}"))
        })?;
        map.insert(bnode.clone(), canonical);
    }
    Ok(map)
}

/// Does `quad` mention `bnode` in any position?
fn quad_touches(quad: &Quad, bnode: &BlankNode) -> bool {
    matches!(&quad.subject, NamedOrBlankNode::BlankNode(b) if b == bnode)
        || matches!(&quad.object, Term::BlankNode(b) if b == bnode)
        || matches!(&quad.graph_name, GraphName::BlankNode(b) if b == bnode)
}

/// Render one neighborhood line for the signature of `bnode`, substituting
/// `_:self` for the node itself and current labels for other blank nodes.
fn neighborhood_line(
    quad: &Quad, bnode: &BlankNode, labels: &HashMap<BlankNode, String>,
) -> String {
    let blank_repr = |b: &BlankNode| -> String {
        if b == bnode {
            "_:self".to_string()
        } else {
            format!("_:{}", labels.get(b).map_or("", String::as_str))
        }
    };
    let s = match &quad.subject {
        NamedOrBlankNode::BlankNode(b) => blank_repr(b),
        NamedOrBlankNode::NamedNode(n) => n.to_string(),
    };
    let o = match &quad.object {
        Term::BlankNode(b) => blank_repr(b),
        other => other.to_string(),
    };
    let g = match &quad.graph_name {
        GraphName::DefaultGraph => String::new(),
        GraphName::NamedNode(n) => n.to_string(),
        GraphName::BlankNode(b) => blank_repr(b),
    };
    format!("{s} {} {o} {g}", quad.predicate)
}

/// Rewrite a quad's blank nodes through the canonical relabeling map.
/// Nodes absent from the map (i.e. non-blank positions) pass through.
fn relabel_quad(quad: &Quad, map: &HashMap<BlankNode, BlankNode>) -> Quad {
    let subject = match &quad.subject {
        NamedOrBlankNode::BlankNode(b) => {
            NamedOrBlankNode::BlankNode(map.get(b).cloned().unwrap_or_else(|| b.clone()))
        }
        named @ NamedOrBlankNode::NamedNode(_) => named.clone(),
    };
    let object = match &quad.object {
        Term::BlankNode(b) => Term::BlankNode(map.get(b).cloned().unwrap_or_else(|| b.clone())),
        other => other.clone(),
    };
    let graph_name = match &quad.graph_name {
        GraphName::BlankNode(b) => {
            GraphName::BlankNode(map.get(b).cloned().unwrap_or_else(|| b.clone()))
        }
        other => other.clone(),
    };
    Quad::new(subject, quad.predicate.clone(), object, graph_name)
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    fn delta(adds: &[&str], dels: &[&str]) -> Delta {
        Delta {
            additions: adds.iter().map(ToString::to_string).collect(),
            deletions: dels.iter().map(ToString::to_string).collect(),
        }
    }

    #[test]
    fn compose_with_inverse_is_empty_and_hashes_as_empty() {
        let d = delta(
            &["<http://e/a> <http://e/p> \"1\""],
            &["<http://e/b> <http://e/p> \"2\""],
        );
        let net = d.compose(&d.inverse());
        assert!(net.is_empty(), "delta ∘ delta⁻¹ must cancel: {net:?}");
        assert_eq!(
            net.hash(),
            Delta::default().hash(),
            "empty-composite hash must equal empty delta hash"
        );
    }

    #[test]
    fn inverse_is_an_involution() {
        let d = delta(&["a", "b"], &["c"]);
        assert_eq!(d.inverse().inverse(), d);
    }

    #[test]
    fn compose_cancels_crosswise_and_keeps_survivors_sorted() {
        // self adds x, deletes y; other deletes x (cancel), adds y (cancel),
        // and adds z (survives).
        let d1 = delta(&["x"], &["y"]);
        let d2 = delta(&["z", "y"], &["x"]);
        let net = d1.compose(&d2);
        assert_eq!(net.additions, vec!["z".to_string()]);
        assert!(net.deletions.is_empty());
    }

    #[test]
    fn empty_delta_is_empty() {
        assert!(Delta::default().is_empty());
        assert!(!delta(&["a"], &[]).is_empty());
    }

    // -- GraphLawStore ------------------------------------------------------

    const DOG_TTL: &str = r"
        @prefix ex: <http://example.org/> .
        ex:rex a ex:Dog .
    ";

    const DOG_RULE: &str = "@prefix ex: <http://example.org/>. {?s a ex:Dog} => {?s a ex:Animal}.";

    #[test]
    fn graphlaw_materialize_derives_facts_visible_to_sparql() {
        let store = GraphLawStore::new().expect("store");
        GraphEngine::insert_turtle(&store, DOG_TTL).expect("ttl");
        // Before materialization the ASK is false.
        let ask = "ASK { <http://example.org/rex> a <http://example.org/Animal> }";
        assert_eq!(
            store.query(ask).expect("ask"),
            EngineQueryResults::Boolean(false)
        );
        let rules = store.load_rules(DOG_RULE).expect("rules");
        assert_eq!(rules, 1, "one rule in the document");
        let outcome = store.materialize().expect("materialize");
        assert_eq!(outcome.rules_loaded, 1);
        assert_eq!(
            outcome.derived.len(),
            1,
            "one derived fact: {:?}",
            outcome.derived
        );
        assert_eq!(
            store.query(ask).expect("ask"),
            EngineQueryResults::Boolean(true),
            "derived fact must be visible to SPARQL after materialization"
        );
    }

    #[test]
    fn graphlaw_state_hash_matches_oxigraph_for_same_facts() {
        // Determinism invariant: both engines hash identical fact states
        // identically (BLAKE3 over canonical quads, computed ggen-side).
        let law = GraphLawStore::new().expect("law");
        GraphEngine::insert_turtle(&law, DOG_TTL).expect("ttl");
        let oxi = DeterministicGraph::new().expect("oxi");
        oxi.insert_turtle(DOG_TTL).expect("ttl");
        assert_eq!(
            GraphEngine::state_hash(&law).expect("law hash"),
            GraphEngine::state_hash(&oxi).expect("oxi hash"),
        );
    }

    #[test]
    fn graphlaw_check_denials_reports_violation() {
        let store = GraphLawStore::new().expect("store");
        GraphEngine::insert_turtle(
            &store,
            "@prefix ex: <http://example.org/> . ex:x a ex:Forbidden .",
        )
        .expect("ttl");
        store
            .load_rules("@prefix ex: <http://example.org/>. {?s a ex:Forbidden} => false.")
            .expect("denial rule");
        store.materialize().expect("materialize");
        let denials = store.check_denials().expect("denials");
        assert_eq!(denials.len(), 1, "one violated denial: {denials:?}");
        assert!(denials[0].contains("DENIED"), "{denials:?}");
    }

    #[test]
    fn graphlaw_validate_shacl_flags_focus_node() {
        let store = GraphLawStore::new().expect("store");
        GraphEngine::insert_turtle(
            &store,
            "@prefix ex: <http://example.org/> . ex:rex a ex:Dog .",
        )
        .expect("ttl");
        let shapes = r"
            @prefix sh: <http://www.w3.org/ns/shacl#> .
            @prefix ex: <http://example.org/> .
            ex:DogShape a sh:NodeShape ;
                sh:targetClass ex:Dog ;
                sh:property [ sh:path ex:name ; sh:minCount 1 ] .
        ";
        let outcome = store.validate_shacl(shapes).expect("shacl");
        assert!(!outcome.conforms, "rex has no ex:name");
        assert!(
            outcome.violations.iter().any(|v| v.contains("rex")),
            "violation must name the focus node: {:?}",
            outcome.violations
        );
        // Conforming after the fix.
        GraphEngine::insert_turtle(
            &store,
            "@prefix ex: <http://example.org/> . ex:rex ex:name \"Rex\" .",
        )
        .expect("ttl");
        let outcome = store.validate_shacl(shapes).expect("shacl");
        assert!(outcome.conforms, "{:?}", outcome.violations);
    }

    #[test]
    fn oxigraph_engine_refuses_law_ops_with_typed_fm_law() {
        let g = DeterministicGraph::new().expect("graph");
        let err =
            GraphEngine::load_rules(&g, "{?s ?p ?o} => {?s ?p ?o}.").expect_err("must refuse");
        assert!(err.to_string().contains("FM-LAW-001"), "{err}");
        let err = GraphEngine::materialize(&g).expect_err("must refuse");
        assert!(err.to_string().contains("FM-LAW-001"), "{err}");
        let err = GraphEngine::validate_shacl(&g, "").expect_err("must refuse");
        assert!(err.to_string().contains("FM-LAW-002"), "{err}");
    }
}