shifty_parse/

lower.rs

//! Lower a loaded shapes graph into the formalism [`Schema`].
//!
//! Every SHACL Core construct collapses into the small IR, applying the sugar
//! rules from the gap analysis (`class → path`, `minCount/maxCount → Count`,
//! per-value constraints wrapped in `∀π = ∃≤0 π.¬φ`, `xone → ∧∨¬`, …). Each
//! shape lowers to a **focus-node predicate** `φ`, so `sh:property`/`sh:node`
//! compose by conjunction. Unsupported AF constructs emit diagnostics.

use crate::diagnostics::{DiagLevel, Diagnostic};
use crate::graph::{Loaded, term_to_node};
use crate::path::parse_path;
use crate::vocab;
use oxrdf::{Literal, NamedNode, NamedOrBlankNode, Term};
use shifty_algebra::{
    Bound, NodeExpr, NodeKindSet, Path, Rule, RuleHead, Schema, Selector, Severity, Shape,
    ShapeArena, ShapeId, SparqlConstraint, SparqlConstruct, SparqlQueryKind, SparqlTarget,
    Statement, ValueType,
};
use spargebra::{Query, SparqlParser};
use std::collections::{BTreeSet, HashMap, HashSet};

pub struct Lowered {
    pub schema: Schema,
    pub diagnostics: Vec<Diagnostic>,
}

/// Lower a loaded graph into a schema plus diagnostics.
pub fn lower(g: &Loaded) -> Lowered {
    let mut l = Lowerer {
        g,
        arena: ShapeArena::new(),
        cache: HashMap::new(),
        statements: Vec::new(),
        rules: Vec::new(),
        diags: Vec::new(),
    };
    let shapes = l.discover_shapes();
    for s in &shapes {
        l.lower_shape(s);
    }
    l.lower_custom_components(&shapes);
    for s in &shapes {
        // selectors are shared by the shape's statements and its rules
        let selectors = l.target_selectors(s);
        if let Some(shape) = l.cache.get(s).copied() {
            for sel in &selectors {
                l.statements.push(Statement {
                    selector: sel.clone(),
                    shape,
                });
            }
        }
        l.parse_rules(s, &selectors);
    }
    let names = l
        .cache
        .iter()
        .filter_map(|(node, id)| match node {
            NamedOrBlankNode::NamedNode(n) => Some((*id, n.as_str().to_string())),
            NamedOrBlankNode::BlankNode(_) => None,
        })
        .collect();
    let schema = Schema {
        arena: l.arena,
        statements: l.statements,
        rules: l.rules,
        names,
    };
    schema.arena.debug_assert_finalized();
    Lowered {
        schema,
        diagnostics: l.diags,
    }
}

struct Lowerer<'a> {
    g: &'a Loaded,
    arena: ShapeArena,
    cache: HashMap<NamedOrBlankNode, ShapeId>,
    statements: Vec<Statement>,
    rules: Vec<Rule>,
    diags: Vec<Diagnostic>,
}

struct ComponentDef {
    iri: NamedNode,
    params: Vec<ParamDef>,
    node_validator: Option<ValidatorDef>,
    property_validator: Option<ValidatorDef>,
    generic_validator: Option<ValidatorDef>,
}

struct ParamDef {
    path: NamedNode,
    var: String,
    optional: bool,
}

struct ValidatorDef {
    kind: SparqlQueryKind,
    query: String,
    messages: Vec<Term>,
}

/// The local name of an IRI (after the last `#` or `/`) — the SHACL rule for a
/// parameter's pre-bound variable name (SHACL §6.2.1).
fn local_name(iri: &str) -> &str {
    iri.rsplit(['#', '/']).next().unwrap_or(iri)
}

impl Lowerer<'_> {
    fn diag(&mut self, level: DiagLevel, msg: impl Into<String>, subj: &NamedOrBlankNode) {
        self.diags
            .push(Diagnostic::new(level, msg, Some(subj.to_string())));
    }

    /// Subjects that are declared shapes: typed NodeShape/PropertyShape, or
    /// carrying `sh:path` or a target predicate. Referenced-only shapes are
    /// pulled in on demand during lowering. Sorted for deterministic output.
    fn discover_shapes(&self) -> Vec<NamedOrBlankNode> {
        let mut found: HashSet<NamedOrBlankNode> = HashSet::new();
        for triple in self.g.graph.iter() {
            let p = triple.predicate;
            let is_target = p == vocab::SH_TARGET_NODE
                || p == vocab::SH_TARGET_CLASS
                || p == vocab::SH_TARGET_SUBJECTS_OF
                || p == vocab::SH_TARGET_OBJECTS_OF
                || p == vocab::SH_TARGET;
            if p == vocab::SH_PATH || p == vocab::SH_SPARQL || p == vocab::SH_RULE || is_target {
                found.insert(triple.subject.into_owned());
            }
            if p == vocab::RDF_TYPE
                && let Term::NamedNode(ty) = triple.object.into_owned()
                && (ty.as_ref() == vocab::SH_NODE_SHAPE || ty.as_ref() == vocab::SH_PROPERTY_SHAPE)
            {
                found.insert(triple.subject.into_owned());
            }
        }
        let mut shapes: Vec<NamedOrBlankNode> = found.into_iter().collect();
        shapes.sort_by_key(|n| n.to_string());
        shapes
    }

    fn lower_shape(&mut self, s: &NamedOrBlankNode) -> ShapeId {
        if let Some(id) = self.cache.get(s) {
            return *id;
        }
        let id = self.arena.reserve();
        self.cache.insert(s.clone(), id);

        if self.bool_prop(s, vocab::SH_DEACTIVATED) {
            self.arena.set(id, Shape::Top);
            return id;
        }

        let path = self.parse_shape_path(s);
        let mut conjuncts: Vec<ShapeId> = Vec::new();

        // Value-scoped constraints: each applies to every value node along the
        // path (or to the focus node directly when there is no path).
        let value = self.collect_value_constraints(s);
        if !value.is_empty() {
            let value_phi = self.arena.and(value);
            match &path {
                Some(p) => {
                    // ∀π.φ  ≡  ∃≤0 π.¬φ
                    let neg = self.arena.not(value_phi);
                    let c = self.arena.count(p.clone(), None, Some(0), neg);
                    conjuncts.push(c);
                }
                None => conjuncts.push(value_phi),
            }
        }

        self.collect_path_constraints(s, path.as_ref(), &mut conjuncts);

        if self.bool_prop(s, vocab::SH_CLOSED) {
            let q = self.closed_allowed(s);
            let c = self.arena.insert(Shape::Closed(q));
            conjuncts.push(c);
        }

        for constraint_term in self.g.objects(s, vocab::SH_SPARQL) {
            let Some(constraint_node) = term_to_node(&constraint_term) else {
                self.diag(DiagLevel::Error, "sh:sparql must reference a resource", s);
                continue;
            };
            let parsed = if let Some(Term::Literal(query)) =
                self.g.object(&constraint_node, vocab::SH_SELECT)
            {
                self.canonical_sparql(&constraint_node, query.value(), ExpectedQuery::Select)
                    .map(|query| (SparqlQueryKind::Select, query))
            } else if let Some(Term::Literal(query)) =
                self.g.object(&constraint_node, vocab::SH_ASK)
            {
                self.canonical_sparql(&constraint_node, query.value(), ExpectedQuery::Ask)
                    .map(|query| (SparqlQueryKind::Ask, query))
            } else {
                self.diag(
                    DiagLevel::Error,
                    "sh:sparql constraint requires sh:select or sh:ask",
                    &constraint_node,
                );
                None
            };
            if let Some((kind, query)) = parsed {
                let shape = Some(match s {
                    NamedOrBlankNode::NamedNode(n) => Term::NamedNode(n.clone()),
                    NamedOrBlankNode::BlankNode(b) => Term::BlankNode(b.clone()),
                });
                // `sh:message` on the SPARQL constraint takes precedence; absent
                // that, fall back to the owning shape's `sh:message` (SHACL §5.2.1).
                let mut messages: Vec<Term> = self.g.objects(&constraint_node, vocab::SH_MESSAGE);
                if messages.is_empty() {
                    messages = self.g.objects(s, vocab::SH_MESSAGE);
                }
                let constraint = SparqlConstraint {
                    kind,
                    query,
                    path: path.clone(),
                    shape,
                    messages,
                    extra_bindings: Vec::new(),
                    bind_value_to_this: false,
                };
                conjuncts.push(self.arena.insert(Shape::Sparql(constraint)));
            }
        }

        // sh:expression (SHACL-AF §5): the node expression must evaluate to
        // `true` with the focus node as `?this`. Focus-scoped, so it joins the
        // conjuncts directly rather than under a `∀π` wrapper.
        for expr_term in self.g.objects(s, vocab::SH_EXPRESSION) {
            if let Some(expr) = self.parse_node_expr(expr_term, s) {
                if node_expr_has_function(&expr) {
                    // SPARQL functions inside expressions need shapes-graph
                    // lookups the validation evaluators don't perform; refuse
                    // rather than silently under-constrain.
                    self.diag(
                        DiagLevel::Unsupported,
                        "sh:expression with a function call is not yet evaluated",
                        s,
                    );
                } else {
                    conjuncts.push(self.arena.insert(Shape::Expression(expr)));
                }
            }
        }

        let body = if conjuncts.is_empty() {
            self.arena.top()
        } else if conjuncts.len() == 1 {
            if conjuncts[0] == id {
                self.arena.top()
            } else {
                conjuncts[0]
            }
        } else {
            self.arena.insert(Shape::And(conjuncts))
        };
        self.arena.set(
            id,
            Shape::Annotated {
                severity: self.severity(s),
                shape: body,
            },
        );
        id
    }

    fn severity(&self, shape: &NamedOrBlankNode) -> Severity {
        match self.g.object(shape, vocab::SH_SEVERITY) {
            Some(Term::NamedNode(value)) => Severity::from_named_node(value),
            _ => Severity::Violation,
        }
    }

    fn collect_value_constraints(&mut self, s: &NamedOrBlankNode) -> Vec<ShapeId> {
        let mut value: Vec<ShapeId> = Vec::new();

        // sh:class C  ≡  ∃≥1 (rdf:type/rdfs:subClassOf*) . test(C)
        for c in self.g.objects(s, vocab::SH_CLASS) {
            let tn = self.arena.insert(Shape::TestConst(c));
            let cc = self.arena.count(class_path(), Some(1), None, tn);
            value.push(cc);
        }

        // sh:datatype
        for d in self.g.objects(s, vocab::SH_DATATYPE) {
            if let Term::NamedNode(n) = d {
                let id = self.arena.insert(Shape::TestType(ValueType::Datatype(n)));
                value.push(id);
            }
        }

        // sh:nodeKind
        for k in self.g.objects(s, vocab::SH_NODE_KIND) {
            if let Some(set) = map_node_kind(&k) {
                let id = self.arena.insert(Shape::TestKind(set));
                value.push(id);
            } else {
                self.diag(DiagLevel::Warning, "unrecognized sh:nodeKind value", s);
            }
        }

        // numeric range (combine the four bounds into one facet)
        let lo = self
            .lit(s, vocab::SH_MIN_INCLUSIVE)
            .map(|value| Bound {
                value,
                inclusive: true,
            })
            .or_else(|| {
                self.lit(s, vocab::SH_MIN_EXCLUSIVE).map(|value| Bound {
                    value,
                    inclusive: false,
                })
            });
        let hi = self
            .lit(s, vocab::SH_MAX_INCLUSIVE)
            .map(|value| Bound {
                value,
                inclusive: true,
            })
            .or_else(|| {
                self.lit(s, vocab::SH_MAX_EXCLUSIVE).map(|value| Bound {
                    value,
                    inclusive: false,
                })
            });
        if lo.is_some() || hi.is_some() {
            let id = self
                .arena
                .insert(Shape::TestType(ValueType::NumericRange { lo, hi }));
            value.push(id);
        }

        // length
        let min_len = self.int(s, vocab::SH_MIN_LENGTH);
        let max_len = self.int(s, vocab::SH_MAX_LENGTH);
        if min_len.is_some() || max_len.is_some() {
            let id = self.arena.insert(Shape::TestType(ValueType::Length {
                min: min_len,
                max: max_len,
            }));
            value.push(id);
        }

        // pattern (+ flags)
        let flags = self
            .lit(s, vocab::SH_FLAGS)
            .map(|l| l.value().to_string())
            .unwrap_or_default();
        for pat in self.g.objects(s, vocab::SH_PATTERN) {
            if let Term::Literal(l) = pat {
                let id = self.arena.insert(Shape::TestType(ValueType::Pattern {
                    regex: l.value().to_string(),
                    flags: flags.clone(),
                }));
                value.push(id);
            }
        }

        // sh:languageIn
        for li in self.g.objects(s, vocab::SH_LANGUAGE_IN) {
            let langs: Vec<String> = self
                .g
                .read_list(&li)
                .into_iter()
                .filter_map(|m| match m {
                    Term::Literal(l) => Some(l.value().to_string()),
                    _ => None,
                })
                .collect();
            let id = self.arena.insert(Shape::TestType(ValueType::LangIn(langs)));
            value.push(id);
        }

        // sh:in  ≡  ⋁ test(member)
        for inl in self.g.objects(s, vocab::SH_IN) {
            let alts: Vec<ShapeId> = self
                .g
                .read_list(&inl)
                .into_iter()
                .map(|m| self.arena.insert(Shape::TestConst(m)))
                .collect();
            let or = self.arena.or(alts);
            value.push(or);
        }

        // sh:node — each value node must conform to the referenced shape
        for n in self.g.objects(s, vocab::SH_NODE) {
            if let Some(nn) = term_to_node(&n) {
                let id = self.lower_shape(&nn);
                value.push(id);
            }
        }

        // sh:property — like sh:node, each *value node* must conform to the
        // referenced property shape (so on a property shape it is scoped under
        // ∀path, not applied to the focus node directly).
        for prop in self.g.objects(s, vocab::SH_PROPERTY) {
            if let Some(pn) = term_to_node(&prop) {
                let id = self.lower_shape(&pn);
                value.push(id);
            }
        }

        // sh:not
        for n in self.g.objects(s, vocab::SH_NOT) {
            if let Some(nn) = term_to_node(&n) {
                let id = self.lower_shape(&nn);
                let neg = self.arena.not(id);
                value.push(neg);
            }
        }

        // sh:and / sh:or / sh:xone (each object is an rdf:list of shapes)
        for l in self.g.objects(s, vocab::SH_AND) {
            let ids = self.lower_shape_list(&l);
            let a = self.arena.and(ids);
            value.push(a);
        }
        for l in self.g.objects(s, vocab::SH_OR) {
            let ids = self.lower_shape_list(&l);
            let o = self.arena.or(ids);
            value.push(o);
        }
        for l in self.g.objects(s, vocab::SH_XONE) {
            let ids = self.lower_shape_list(&l);
            let x = self.arena.xone(ids);
            value.push(x);
        }

        value
    }

    /// Path-level constraints (cardinality, qualified counts, property pairs,
    /// hasValue, uniqueLang). Most require a path; without one they are ignored
    /// with a diagnostic, except `sh:hasValue` which applies to the focus node.
    fn collect_path_constraints(
        &mut self,
        s: &NamedOrBlankNode,
        path: Option<&Path>,
        conjuncts: &mut Vec<ShapeId>,
    ) {
        let need_path = |me: &mut Self, what: &str| {
            me.diag(DiagLevel::Warning, format!("{what} ignored: no sh:path"), s);
        };

        let min_count = self.int(s, vocab::SH_MIN_COUNT);
        let max_count = self.int(s, vocab::SH_MAX_COUNT);
        if min_count.is_some() || max_count.is_some() {
            match path {
                Some(p) => {
                    let top = self.arena.top();
                    let c = self.arena.count(p.clone(), min_count, max_count, top);
                    conjuncts.push(c);
                }
                None => need_path(self, "sh:minCount/sh:maxCount"),
            }
        }

        // sh:hasValue
        for v in self.g.objects(s, vocab::SH_HAS_VALUE) {
            match path {
                Some(p) => {
                    let tc = self.arena.insert(Shape::TestConst(v));
                    let c = self.arena.count(p.clone(), Some(1), None, tc);
                    conjuncts.push(c);
                }
                None => {
                    let tc = self.arena.insert(Shape::TestConst(v));
                    conjuncts.push(tc);
                }
            }
        }

        // sh:qualifiedValueShape + qualifiedMin/MaxCount
        for q in self.g.objects(s, vocab::SH_QUALIFIED_VALUE_SHAPE) {
            if let Some(qn) = term_to_node(&q) {
                let qmin = self.int(s, vocab::SH_QUALIFIED_MIN_COUNT);
                let qmax = self.int(s, vocab::SH_QUALIFIED_MAX_COUNT);
                match path {
                    Some(p) => {
                        let mut qualifiers = vec![self.lower_shape(&qn)];
                        if self.bool_prop(s, vocab::SH_QUALIFIED_VALUE_SHAPES_DISJOINT) {
                            for sibling in self.sibling_qualified_shapes(s, &qn) {
                                let sibling = self.lower_shape(&sibling);
                                qualifiers.push(self.arena.not(sibling));
                            }
                        }
                        let qualifier = self.arena.and(qualifiers);
                        let c = self.arena.count(p.clone(), qmin, qmax, qualifier);
                        conjuncts.push(c);
                    }
                    None => need_path(self, "sh:qualifiedValueShape"),
                }
            }
        }

        // property-pair constraints
        let pairs = [
            (vocab::SH_EQUALS, "equals"),
            (vocab::SH_DISJOINT, "disjoint"),
            (vocab::SH_LESS_THAN, "lessThan"),
            (vocab::SH_LESS_THAN_OR_EQUALS, "lessThanOrEquals"),
        ];
        for (pred, name) in pairs {
            for other in self.g.objects(s, pred) {
                let Term::NamedNode(op) = other else { continue };
                match path {
                    Some(p) => {
                        let shape = match name {
                            "equals" => Shape::Eq(p.clone(), op),
                            "disjoint" => Shape::Disj(p.clone(), op),
                            "lessThan" => Shape::Lt(p.clone(), op),
                            _ => Shape::Le(p.clone(), op),
                        };
                        let c = self.arena.insert(shape);
                        conjuncts.push(c);
                    }
                    None if matches!(name, "equals" | "disjoint") => {
                        let shape = if name == "equals" {
                            Shape::Eq(Path::Id, op)
                        } else {
                            Shape::Disj(Path::Id, op)
                        };
                        let c = self.arena.insert(shape);
                        conjuncts.push(c);
                    }
                    None => need_path(self, &format!("sh:{name}")),
                }
            }
        }

        // sh:uniqueLang
        if self.bool_prop(s, vocab::SH_UNIQUE_LANG) {
            match path {
                Some(p) => {
                    let c = self.arena.insert(Shape::UniqueLang(p.clone()));
                    conjuncts.push(c);
                }
                None => need_path(self, "sh:uniqueLang"),
            }
        }
    }

    /// The target selectors of a shape (used by both its statements and rules).
    fn target_selectors(&mut self, s: &NamedOrBlankNode) -> Vec<Selector> {
        let mut sels = Vec::new();

        for c in self.g.objects(s, vocab::SH_TARGET_NODE) {
            sels.push(Selector::IsConst(c));
        }
        for c in self.g.objects(s, vocab::SH_TARGET_CLASS) {
            sels.push(self.class_selector(c));
        }
        for p in self.g.objects(s, vocab::SH_TARGET_SUBJECTS_OF) {
            if let Term::NamedNode(n) = p {
                sels.push(Selector::HasOut(n));
            }
        }
        for p in self.g.objects(s, vocab::SH_TARGET_OBJECTS_OF) {
            if let Term::NamedNode(n) = p {
                sels.push(Selector::HasIn(n));
            }
        }

        // implicit class target: a shape that is also an rdfs:Class / owl:Class
        if (self.g.is_instance_of(s, vocab::RDFS_CLASS)
            || self.g.is_instance_of(s, vocab::OWL_CLASS))
            && let NamedOrBlankNode::NamedNode(n) = s
        {
            sels.push(self.class_selector(Term::NamedNode(n.clone())));
        }

        for target_term in self.g.objects(s, vocab::SH_TARGET) {
            let Some(target_node) = term_to_node(&target_term) else {
                self.diag(DiagLevel::Error, "sh:target must reference a resource", s);
                continue;
            };
            match self.g.object(&target_node, vocab::SH_SELECT) {
                Some(Term::Literal(query)) => {
                    if let Some(query) =
                        self.canonical_sparql(&target_node, query.value(), ExpectedQuery::Select)
                    {
                        sels.push(Selector::Sparql(SparqlTarget { query }));
                    }
                }
                _ => self.diag(
                    DiagLevel::Unsupported,
                    "custom sh:target without sh:select is not yet lowered",
                    &target_node,
                ),
            }
        }

        sels
    }

    /// Lower SPARQL-based custom constraint components (SHACL §6.3) into the
    /// algebra. For each shape that *activates* a component (supplies a value for
    /// every mandatory parameter), a `Shape::Sparql` node is appended to that
    /// shape's conjunction. The parameter values from the activating shape are
    /// stored as `extra_bindings` so `compile_constraint` can pre-substitute them
    /// (§6.2.3) before binding `$this` per focus node.
    ///
    /// Components with an invalid SPARQL validator query emit `DiagLevel::Error`
    /// and are skipped (the constraint is not enforced for that activation).
    fn lower_custom_components(&mut self, shapes: &[NamedOrBlankNode]) {
        let components = self.collect_component_defs();
        if components.is_empty() {
            return;
        }
        for s in shapes {
            let Some(&shape_id) = self.cache.get(s) else {
                continue;
            };
            let is_property_shape = self.g.object(s, vocab::SH_PATH).is_some();
            let path = if is_property_shape {
                self.parse_shape_path(s)
            } else {
                None
            };
            let shape_term = match s {
                NamedOrBlankNode::NamedNode(n) => Some(Term::NamedNode(n.clone())),
                NamedOrBlankNode::BlankNode(b) => Some(Term::BlankNode(b.clone())),
            };

            for component in &components {
                // Activation: every mandatory parameter must be present on `s`.
                let mut extra_bindings: Vec<(String, Term)> = Vec::new();
                let mut activated = true;
                for p in &component.params {
                    if let Some(value) = self.g.object(s, p.path.as_ref()) {
                        extra_bindings.push((p.var.clone(), value));
                    } else if !p.optional {
                        activated = false;
                        break;
                    }
                }
                if !activated {
                    continue;
                }

                // Pick the right validator for the shape kind.
                let validator = if is_property_shape {
                    component
                        .property_validator
                        .as_ref()
                        .or(component.generic_validator.as_ref())
                } else {
                    component
                        .node_validator
                        .as_ref()
                        .or(component.generic_validator.as_ref())
                };
                let Some(validator) = validator else { continue };

                let constraint = SparqlConstraint {
                    kind: validator.kind,
                    query: validator.query.clone(),
                    path: path.clone(),
                    shape: shape_term.clone(),
                    messages: validator.messages.clone(),
                    extra_bindings,
                    // For node-shape activations (no sh:path), $value equals the
                    // focus node, so rename ?value → ?this before binding per focus.
                    bind_value_to_this: !is_property_shape
                        && validator.kind == SparqlQueryKind::Ask,
                };
                let raw_id = self.arena.insert(Shape::Sparql(constraint));
                // sh:SPARQLAskValidator (SHACL §6.2.3): ASK=true means the
                // constraint is *satisfied*. The algebra evaluator treats
                // ASK=true as a violation (matching sh:sparql semantics), so
                // we wrap in Not to restore the correct polarity. SELECT
                // validators are violation-per-row, same as sh:sparql — no inversion.
                let sparql_id = if validator.kind == SparqlQueryKind::Ask {
                    self.arena.not(raw_id)
                } else {
                    raw_id
                };

                // Conjoin with the shape's existing body.
                let (severity, inner) = match self.arena.get(shape_id) {
                    Shape::Annotated {
                        severity,
                        shape: inner,
                    } => (severity.clone(), *inner),
                    _ => continue, // defensive; lower_shape always produces Annotated
                };
                let combined = self.arena.and(vec![inner, sparql_id]);
                self.arena.set(
                    shape_id,
                    Shape::Annotated {
                        severity,
                        shape: combined,
                    },
                );
            }
        }
    }

    /// Collect all custom constraint component definitions from the shapes graph:
    /// named subjects that carry `sh:parameter` and at least one validator
    /// predicate (`sh:validator`, `sh:nodeValidator`, `sh:propertyValidator`).
    fn collect_component_defs(&mut self) -> Vec<ComponentDef> {
        let mut out: Vec<ComponentDef> = Vec::new();
        let mut seen = HashSet::new();
        for triple in self.g.graph.triples_for_predicate(vocab::SH_PARAMETER) {
            let subject = triple.subject.into_owned();
            if !seen.insert(subject.clone()) {
                continue;
            }
            let NamedOrBlankNode::NamedNode(iri) = &subject else {
                continue;
            };
            let node_validator = self
                .g
                .object(&subject, vocab::SH_NODE_VALIDATOR)
                .and_then(|v| self.parse_validator_def(&v));
            let property_validator = self
                .g
                .object(&subject, vocab::SH_PROPERTY_VALIDATOR)
                .and_then(|v| self.parse_validator_def(&v));
            let generic_validator = self
                .g
                .object(&subject, vocab::SH_VALIDATOR)
                .and_then(|v| self.parse_validator_def(&v));
            if node_validator.is_none()
                && property_validator.is_none()
                && generic_validator.is_none()
            {
                continue; // not a constraint component (e.g. a sh:SPARQLFunction)
            }
            let mut params: Vec<ParamDef> = Vec::new();
            for p in self.g.objects(&subject, vocab::SH_PARAMETER) {
                let Some(pn) = term_to_node(&p) else { continue };
                let Some(Term::NamedNode(path)) = self.g.object(&pn, vocab::SH_PATH) else {
                    continue;
                };
                let var = local_name(path.as_str()).to_string();
                let optional = matches!(
                    self.g.object(&pn, vocab::SH_OPTIONAL),
                    Some(Term::Literal(ref l)) if l.value() == "true"
                );
                params.push(ParamDef {
                    path,
                    var,
                    optional,
                });
            }
            if params.is_empty() {
                continue;
            }
            out.push(ComponentDef {
                iri: iri.clone(),
                params,
                node_validator,
                property_validator,
                generic_validator,
            });
        }
        out.sort_by(|a, b| a.iri.as_str().cmp(b.iri.as_str()));
        out
    }

    /// Parse a validator node into a `ValidatorDef`, canonicalizing its SPARQL
    /// query. Returns `None` (and emits an error diagnostic) if the query is
    /// invalid SPARQL.
    fn parse_validator_def(&mut self, node: &Term) -> Option<ValidatorDef> {
        let node = term_to_node(node)?;
        let (kind, raw) = if let Some(Term::Literal(q)) = self.g.object(&node, vocab::SH_ASK) {
            (SparqlQueryKind::Ask, q.value().to_string())
        } else if let Some(Term::Literal(q)) = self.g.object(&node, vocab::SH_SELECT) {
            (SparqlQueryKind::Select, q.value().to_string())
        } else {
            return None; // no query body — not a SPARQL validator
        };
        let canonical = match canonical_sparql_query(self.g, &node, &raw) {
            Ok((_, c)) => c,
            Err(msg) => {
                self.diag(
                    DiagLevel::Error,
                    format!("invalid SPARQL in validator: {msg}"),
                    &node,
                );
                return None;
            }
        };
        let messages = self.g.objects(&node, vocab::SH_MESSAGE);
        Some(ValidatorDef {
            kind,
            query: canonical,
            messages,
        })
    }

    /// Lower the `sh:rule`s of a shape (SHACL-AF). A rule fires on the shape's
    /// targets, so we emit one [`Rule`] per selector.
    fn parse_rules(&mut self, s: &NamedOrBlankNode, selectors: &[Selector]) {
        for rule_term in self.g.objects(s, vocab::SH_RULE) {
            let Some(rn) = term_to_node(&rule_term) else {
                continue;
            };
            let Some(head) = self.parse_rule_head(&rn) else {
                continue;
            };

            let conditions: Vec<ShapeId> = self
                .g
                .objects(&rn, vocab::SH_CONDITION)
                .iter()
                .filter_map(term_to_node)
                .map(|c| self.lower_shape(&c))
                .collect();
            let order = self.order(&rn);
            let deactivated = self.bool_prop(&rn, vocab::SH_DEACTIVATED);

            for sel in selectors {
                self.rules.push(Rule {
                    selector: sel.clone(),
                    conditions: conditions.clone(),
                    head: head.clone(),
                    order,
                    deactivated,
                });
            }
        }
    }

    /// Qualified value shapes attached through the same parent `sh:property`
    /// declaration, excluding the current qualified shape itself.
    fn sibling_qualified_shapes(
        &self,
        shape: &NamedOrBlankNode,
        qualifier: &NamedOrBlankNode,
    ) -> Vec<NamedOrBlankNode> {
        let mut siblings = HashSet::new();
        for triple in self.g.graph.triples_for_predicate(vocab::SH_PROPERTY) {
            if term_to_node(&triple.object.into_owned()).as_ref() != Some(shape) {
                continue;
            }
            let parent = triple.subject.into_owned();
            for property in self.g.objects(&parent, vocab::SH_PROPERTY) {
                let Some(property) = term_to_node(&property) else {
                    continue;
                };
                for sibling in self.g.objects(&property, vocab::SH_QUALIFIED_VALUE_SHAPE) {
                    if let Some(sibling) = term_to_node(&sibling) {
                        siblings.insert(sibling);
                    }
                }
            }
        }
        siblings.remove(qualifier);
        let mut siblings: Vec<_> = siblings.into_iter().collect();
        siblings.sort_by_key(|node| node.to_string());
        siblings
    }

    fn parse_rule_head(&mut self, rn: &NamedOrBlankNode) -> Option<RuleHead> {
        // sh:SPARQLRule — parse and canonicalize the CONSTRUCT while retaining
        // an opaque algebra leaf for later query rewriting.
        if let Some(Term::Literal(q)) = self.g.object(rn, vocab::SH_CONSTRUCT) {
            let query = self.canonical_sparql(rn, q.value(), ExpectedQuery::Construct)?;
            return Some(RuleHead::Sparql(SparqlConstruct { query }));
        }
        // sh:TripleRule — subject/predicate/object node expressions
        let (subj, pred, obj) = (
            self.g.object(rn, vocab::SH_SUBJECT),
            self.g.object(rn, vocab::SH_PREDICATE),
            self.g.object(rn, vocab::SH_OBJECT),
        );
        if subj.is_none() && pred.is_none() && obj.is_none() {
            self.diag(DiagLevel::Unsupported, "unrecognized sh:rule head", rn);
            return None;
        }
        let (Some(subj), Some(pred), Some(obj)) = (subj, pred, obj) else {
            self.diag(
                DiagLevel::Error,
                "sh:TripleRule missing subject/predicate/object",
                rn,
            );
            return None;
        };
        Some(RuleHead::Triple {
            subject: self.parse_node_expr(subj, rn)?,
            predicate: self.parse_node_expr(pred, rn)?,
            object: self.parse_node_expr(obj, rn)?,
        })
    }

    /// Parse a node expression (SHACL-AF §6). Handles `sh:this`, constants,
    /// path expressions, filter / intersection / union expressions, and SPARQL
    /// function calls `[ ex:fn (arg …) ]`.
    fn parse_node_expr(&mut self, term: Term, owner: &NamedOrBlankNode) -> Option<NodeExpr> {
        match &term {
            Term::NamedNode(n) if n.as_ref() == vocab::SH_THIS => Some(NodeExpr::This),
            Term::NamedNode(_) | Term::Literal(_) => Some(NodeExpr::Constant(term)),
            Term::BlankNode(_) => {
                let node = term_to_node(&term).expect("blank node");
                if let Some(path_term) = self.g.object(&node, vocab::SH_PATH) {
                    match parse_path(self.g, &path_term) {
                        Ok(path) => Some(NodeExpr::Path(path)),
                        Err(e) => {
                            self.diag(
                                DiagLevel::Error,
                                format!("invalid node-expression path: {e}"),
                                owner,
                            );
                            None
                        }
                    }
                } else if self.g.object(&node, vocab::SH_FILTER_SHAPE).is_some() {
                    self.parse_filter_expr(&node, owner)
                } else if let Some(list) = self.g.object(&node, vocab::SH_INTERSECTION) {
                    self.parse_set_expr(&list, owner)
                        .map(NodeExpr::Intersection)
                } else if let Some(list) = self.g.object(&node, vocab::SH_UNION) {
                    self.parse_set_expr(&list, owner).map(NodeExpr::Union)
                } else if let Some(expr) = self.try_function_call(&node, owner) {
                    Some(expr)
                } else {
                    self.diag(
                        DiagLevel::Unsupported,
                        "complex node expression not yet lowered",
                        owner,
                    );
                    None
                }
            }
        }
    }

    /// `sh:filterShape` + `sh:nodes` — a filter node expression (SHACL-AF §6.4):
    /// the value nodes of `sh:nodes` that conform to `sh:filterShape`.
    fn parse_filter_expr(
        &mut self,
        node: &NamedOrBlankNode,
        owner: &NamedOrBlankNode,
    ) -> Option<NodeExpr> {
        let Some(shape_term) = self.g.object(node, vocab::SH_FILTER_SHAPE) else {
            self.diag(DiagLevel::Error, "sh:filterShape missing a value", owner);
            return None;
        };
        let Some(shape_node) = term_to_node(&shape_term) else {
            self.diag(
                DiagLevel::Error,
                "sh:filterShape must reference a shape",
                owner,
            );
            return None;
        };
        let Some(nodes_term) = self.g.object(node, vocab::SH_NODES) else {
            self.diag(
                DiagLevel::Error,
                "filter expression missing sh:nodes",
                owner,
            );
            return None;
        };
        let input = self.parse_node_expr(nodes_term, owner)?;
        let shape = self.lower_shape(&shape_node);
        Some(NodeExpr::Filter {
            input: Box::new(input),
            shape,
        })
    }

    /// Parse the RDF list of an `sh:intersection` / `sh:union` node expression
    /// (SHACL-AF §6.5–6.6) into its member node expressions. Returns `None` if
    /// the list is empty or any member fails to parse (already diagnosed).
    fn parse_set_expr(
        &mut self,
        list_head: &Term,
        owner: &NamedOrBlankNode,
    ) -> Option<Vec<NodeExpr>> {
        let members = self.g.read_list(list_head);
        if members.is_empty() {
            self.diag(
                DiagLevel::Error,
                "sh:intersection/sh:union expects a non-empty list",
                owner,
            );
            return None;
        }
        let n = members.len();
        let exprs: Vec<NodeExpr> = members
            .into_iter()
            .filter_map(|t| self.parse_node_expr(t, owner))
            .collect();
        if exprs.len() != n {
            return None;
        }
        Some(exprs)
    }

    /// Detect a function-call node expression `[ ex:fn ( arg1 arg2 … ) ]`.
    ///
    /// The blank node must have exactly one non-SHACL/RDF/RDFS/OWL predicate;
    /// its object must be an RDF list of argument node expressions.
    fn try_function_call(
        &mut self,
        node: &NamedOrBlankNode,
        owner: &NamedOrBlankNode,
    ) -> Option<NodeExpr> {
        let func_preds: Vec<(NamedNode, Term)> = self
            .g
            .graph
            .triples_for_subject(node)
            .map(|t| (t.predicate.into_owned(), t.object.into_owned()))
            .filter(|(p, _)| {
                let s = p.as_str();
                !s.starts_with(vocab::SH)
                    && !s.starts_with(vocab::RDF)
                    && !s.starts_with(vocab::RDFS)
                    && !s.starts_with(vocab::OWL)
            })
            .collect();

        if func_preds.len() != 1 {
            return None;
        }
        let (func_iri, list_head) = func_preds.into_iter().next().unwrap();
        let arg_terms = self.g.read_list(&list_head);
        let n = arg_terms.len();
        let args: Vec<NodeExpr> = arg_terms
            .into_iter()
            .filter_map(|t| self.parse_node_expr(t, owner))
            .collect();
        if args.len() != n {
            return None;
        }
        Some(NodeExpr::Function {
            iri: func_iri,
            args,
        })
    }

    fn order(&self, s: &NamedOrBlankNode) -> Option<i64> {
        match self.g.object(s, vocab::SH_ORDER) {
            Some(Term::Literal(l)) => l.value().parse().ok(),
            _ => None,
        }
    }

    /// `∃≥1 (rdf:type/rdfs:subClassOf*) . test(class)` as a selector.
    fn class_selector(&mut self, class: Term) -> Selector {
        let tn = self.arena.insert(Shape::TestConst(class));
        Selector::HasPath(class_path(), tn)
    }

    fn lower_shape_list(&mut self, list_head: &Term) -> Vec<ShapeId> {
        self.g
            .read_list(list_head)
            .into_iter()
            .filter_map(|m| term_to_node(&m))
            .map(|n| self.lower_shape(&n))
            .collect()
    }

    fn parse_shape_path(&mut self, s: &NamedOrBlankNode) -> Option<Path> {
        let term = self.g.object(s, vocab::SH_PATH)?;
        match parse_path(self.g, &term) {
            Ok(p) => Some(p),
            Err(e) => {
                self.diag(DiagLevel::Error, format!("invalid sh:path: {e}"), s);
                None
            }
        }
    }

    fn closed_allowed(&self, s: &NamedOrBlankNode) -> BTreeSet<oxrdf::NamedNode> {
        let mut q = BTreeSet::new();
        for prop in self.g.objects(s, vocab::SH_PROPERTY) {
            if let Some(pn) = term_to_node(&prop)
                && let Some(Term::NamedNode(n)) = self.g.object(&pn, vocab::SH_PATH)
            {
                q.insert(n);
            }
        }
        for ip in self.g.objects(s, vocab::SH_IGNORED_PROPERTIES) {
            for m in self.g.read_list(&ip) {
                if let Term::NamedNode(n) = m {
                    q.insert(n);
                }
            }
        }
        q
    }

    fn bool_prop(&self, s: &NamedOrBlankNode, pred: oxrdf::NamedNodeRef) -> bool {
        matches!(self.g.object(s, pred), Some(Term::Literal(l)) if l.value() == "true")
    }

    fn int(&self, s: &NamedOrBlankNode, pred: oxrdf::NamedNodeRef) -> Option<u64> {
        match self.g.object(s, pred) {
            Some(Term::Literal(l)) => l.value().parse().ok(),
            _ => None,
        }
    }

    fn lit(&self, s: &NamedOrBlankNode, pred: oxrdf::NamedNodeRef) -> Option<Literal> {
        match self.g.object(s, pred) {
            Some(Term::Literal(l)) => Some(l),
            _ => None,
        }
    }

    /// Parse a SHACL SPARQL query once, resolving both document prefixes and
    /// `sh:prefixes` declarations. `Query::to_string` expands prefix names, so
    /// the IR remains self-contained and can be reparsed or rewritten later.
    fn canonical_sparql(
        &mut self,
        owner: &NamedOrBlankNode,
        raw: &str,
        expected: ExpectedQuery,
    ) -> Option<String> {
        let (query, canonical) = match canonical_sparql_query(self.g, owner, raw) {
            Ok(result) => result,
            Err(message) => {
                self.diag(DiagLevel::Error, message, owner);
                return None;
            }
        };
        let actual = match &query {
            Query::Select { .. } => ExpectedQuery::Select,
            Query::Ask { .. } => ExpectedQuery::Ask,
            Query::Construct { .. } => ExpectedQuery::Construct,
            Query::Describe { .. } => ExpectedQuery::Describe,
        };
        if actual != expected {
            self.diag(
                DiagLevel::Error,
                format!("expected SPARQL {expected}, found {actual}"),
                owner,
            );
            return None;
        }
        Some(canonical)
    }
}

/// Build the canonical, prefix-expanded form of a SHACL SPARQL query string.
///
/// Resolves the document base IRI, document-level prefixes, and the
/// `sh:prefixes` / `sh:declare` chains (following `owl:imports`) declared on
/// `owner`, parses `raw`, and returns the parsed query together with its
/// canonical string form. `Query::to_string` expands prefix names, so the
/// result is self-contained and can be reparsed without external declarations.
///
/// Errors are returned as messages so callers can decide how to surface them:
/// the lowerer routes them to diagnostics; the report validator drops the
/// offending constraint, matching the lowering path.
pub fn canonical_sparql_query(
    g: &Loaded,
    owner: &NamedOrBlankNode,
    raw: &str,
) -> Result<(Query, String), String> {
    let mut parser = SparqlParser::new();
    if let Some(base) = &g.base {
        parser = parser
            .with_base_iri(base)
            .map_err(|e| format!("invalid SPARQL base IRI: {e}"))?;
    }
    for (prefix, namespace) in &g.prefixes {
        parser = parser
            .with_prefix(prefix, namespace)
            .map_err(|e| format!("invalid SPARQL prefix declaration {prefix}: {e}"))?;
    }
    let mut prefix_sources: Vec<NamedOrBlankNode> = g
        .objects(owner, vocab::SH_PREFIXES)
        .iter()
        .filter_map(term_to_node)
        .collect();
    let mut seen_sources = HashSet::new();
    while let Some(source) = prefix_sources.pop() {
        if !seen_sources.insert(source.clone()) {
            continue;
        }
        prefix_sources.extend(
            g.objects(&source, vocab::OWL_IMPORTS)
                .iter()
                .filter_map(term_to_node),
        );
        for declaration_term in g.objects(&source, vocab::SH_DECLARE) {
            let Some(declaration) = term_to_node(&declaration_term) else {
                continue;
            };
            let (Some(Term::Literal(prefix)), Some(Term::Literal(namespace))) = (
                g.object(&declaration, vocab::SH_PREFIX),
                g.object(&declaration, vocab::SH_NAMESPACE),
            ) else {
                continue;
            };
            parser = parser
                .with_prefix(prefix.value(), namespace.value())
                .map_err(|e| format!("invalid SHACL SPARQL prefix declaration: {e}"))?;
        }
    }
    let query = parser
        .parse_query(raw)
        .map_err(|e| format!("invalid SPARQL query: {e}"))?;
    let canonical = query.to_string();
    Ok((query, canonical))
}

#[derive(Clone, Copy, PartialEq, Eq)]
enum ExpectedQuery {
    Select,
    Ask,
    Construct,
    Describe,
}

impl std::fmt::Display for ExpectedQuery {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(match self {
            Self::Select => "SELECT",
            Self::Ask => "ASK",
            Self::Construct => "CONSTRUCT",
            Self::Describe => "DESCRIBE",
        })
    }
}

fn class_path() -> Path {
    Path::seq(vec![
        Path::Pred(vocab::rdf_type()),
        Path::star(Path::Pred(vocab::rdfs_subclassof())),
    ])
}

/// Whether a node expression contains a SPARQL function application anywhere in
/// its tree. Such expressions cannot yet be evaluated in the validation paths
/// (they need shapes-graph function lookups), so expression constraints over
/// them are diagnosed rather than lowered.
fn node_expr_has_function(e: &NodeExpr) -> bool {
    match e {
        NodeExpr::Function { .. } => true,
        NodeExpr::Filter { input, .. } => node_expr_has_function(input),
        NodeExpr::Intersection(es) | NodeExpr::Union(es) => es.iter().any(node_expr_has_function),
        NodeExpr::This | NodeExpr::Constant(_) | NodeExpr::Path(_) => false,
    }
}

fn map_node_kind(term: &Term) -> Option<NodeKindSet> {
    let Term::NamedNode(n) = term else {
        return None;
    };
    let r = n.as_ref();
    Some(if r == vocab::SH_IRI {
        NodeKindSet::IRI
    } else if r == vocab::SH_BLANK_NODE {
        NodeKindSet::BLANK_NODE
    } else if r == vocab::SH_LITERAL {
        NodeKindSet::LITERAL
    } else if r == vocab::SH_BLANK_NODE_OR_IRI {
        NodeKindSet::BLANK_NODE_OR_IRI
    } else if r == vocab::SH_BLANK_NODE_OR_LITERAL {
        NodeKindSet::BLANK_NODE_OR_LITERAL
    } else if r == vocab::SH_IRI_OR_LITERAL {
        NodeKindSet::IRI_OR_LITERAL
    } else {
        return None;
    })
}