rete-core 0.3.4

//! Ingestion: parse RDF text (N-Triples / N-Quads / Turtle) and assemble a
//! complete `.rete` file image. Shared by the CLI's `build`/`validate` commands
//! and the wasm bindings (the playground's in-browser builder).
//!
//! The N-Triples/N-Quads reader is line-based and keeps terms as their exact
//! canonical token strings (`<iri>`, `_:bnode`, `"lit"`, `"lit"^^<dt>`,
//! `"lit"@lang`) so they double as dictionary keys and so a `query` can match
//! by the same string. This is not a full RDF 1.1 parser — it covers the
//! canonical N-Triples surface (and is deliberately tolerant of IRIs a strict
//! parser would reject), enough for v0 ingestion. Turtle goes through `oxttl`.
//!
//! That tolerance is *measured*, not silent: pass an [`IriAudit`] to any of the
//! `*_audited` entry points and the parse counts every IRI outside the
//! N-Triples `IRIREF` grammar / RFC 3987 (see [`crate::iri`]) — or, with
//! `strict`, refuses the input on the first one. `rete build` always audits.
//!
//! Assembly degrades with the build features: without the `compression`
//! feature (e.g. on wasm) sections are written with the `NONE` codec — larger
//! files, byte-compatible readers.

use crate::{
    build_pyramid_meta_algo, Dictionary, DictionaryBuilder, GraphIndexBuilder, PyramidAlgo,
    DEFAULT_TILE_BUDGET,
};

/// A parsed triple as three canonical term tokens.
pub type RawTriple = (String, String, String);

/// A parsed quad: the triple plus an optional graph term (`None` = default graph).
pub type RawQuad = (String, String, String, Option<String>);

/// Which **surface** a `<< … >>` in Turtle/TriG input is read as — the input
/// half of `rete export --quoted-triple-syntax`, same vocabulary, same two
/// values.
///
/// The flag exists because the two standards give one piece of syntax two
/// meanings, and no amount of looking at the bytes can tell them apart:
///
/// | written            | under [`RdfStar`](Self::RdfStar) | under [`Rdf12`](Self::Rdf12)          |
/// |--------------------|----------------------------------|---------------------------------------|
/// | `<< s p o >>`      | a **quoted triple**, one term    | a **reifier**: `_:r rdf:reifies <<( s p o )>>`, plus a blank node standing where it was |
/// | `<<( s p o )>>`    | a syntax error                   | a **triple term**, object position only |
/// | `{\| … \|}`         | a syntax error                   | an annotation on the statement before it |
///
/// One file, two readings, two different graphs — see the round-trip matrix in
/// `crates/rete-cli/tests/quoted_triple_surfaces.rs`, which asserts both.
///
/// `<<( s p o )>>` is unambiguous: it is RDF 1.2's and nothing else's. That
/// asymmetry is why [`RdfStar`](Self::RdfStar) is the **default**. Reading an
/// RDF-star file as RDF 1.2 silently yields a different graph; reading an
/// RDF 1.2 file as RDF-star is a hard parse error that names the flag. Only one
/// of the two mistakes is survivable, so the default is the one that makes the
/// other mistake loud.
///
/// **N-Triples and N-Quads ignore this entirely.** Their reader is rete's own
/// (`take_term`), it has accepted both `<< s p o >>` and `<<( s p o )>>` since
/// #262, and RDF 1.2 N-Triples has no reifier syntax for `<< … >>` to be — so
/// there is no ambiguity to resolve and nothing to choose. The same goes for
/// RDF/XML.
///
/// Whichever surface reads the file, what gets **stored** is the same canonical
/// token `<<s p o>>`. rete's term model is a superset of both: a triple term is
/// a term, and RDF 1.2 reification is ordinary RDF — a blank node, a predicate
/// and a term — which rete already stored before this flag existed.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default, Hash)]
pub enum QuotedTripleSurface {
    /// `<< s p o >>` is a **quoted triple** (RDF-star), in subject or object
    /// position — rete's own storage token. The default, and byte-for-byte the
    /// behaviour every rete release before this flag had.
    #[default]
    RdfStar,
    /// `<<( s p o )>>` is a **triple term** and `<< s p o >>` is a **reifier**
    /// (RDF 1.2), with `{| … |}` annotations read too.
    Rdf12,
}

impl QuotedTripleSurface {
    /// Parse the `--quoted-triple-syntax` value, `None` for anything else.
    /// Shared by `rete build`, `rete validate` and `rete export` so the flag
    /// cannot come to mean two things in two directions.
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF12 => Some(Self::Rdf12),
            crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF_STAR => Some(Self::RdfStar),
            _ => None,
        }
    }

    /// The flag spelling of this surface — the inverse of [`Self::parse`].
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Rdf12 => crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF12,
            Self::RdfStar => crate::card_derive::QUOTED_TRIPLE_SURFACE_RDF_STAR,
        }
    }
}

/// The one-line hint appended to a Turtle/TriG parse error when the input looks
/// like RDF 1.2 and was read as RDF-star — i.e. the exact mistake
/// [`QuotedTripleSurface`]'s default makes, made loud.
const RDF12_SURFACE_HINT: &str = "\nhint: this input contains `<<(`, the RDF 1.2 triple-term \
     syntax — which is what `rete export` writes by default. Reading it takes \
     `--quoted-triple-syntax rdf12`; the default, `rdf-star`, reads `<< s p o >>` as a quoted \
     triple instead.";

/// Does `text` look like it carries RDF 1.2 triple terms? A deliberately dumb
/// substring test: `<<(` cannot begin anything else in Turtle, and this is only
/// ever consulted to *explain a parse that already failed*, never to decide how
/// to read a file. A `<<(` inside a string literal would make it a false
/// positive on some other error's hint, which costs one confusing sentence and
/// nothing else.
fn looks_like_rdf12(text: &str) -> bool {
    text.contains("<<(")
}

/// Why an ingest failed. Parser errors from `oxttl`/`oxrdfxml` and I/O errors
/// are flattened to their `Display` string rather than wrapped: the variants
/// then carry no foreign types, which keeps the error usable from the wasm
/// bindings and stable across upstream parser upgrades.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum IngestError {
    /// A malformed N-Triples/N-Quads line: its 1-based line number and a short
    /// reason (`"bad subject"`, `"missing trailing '.'"`, …).
    #[error("line {0}: {1}")]
    Line(usize, &'static str),
    /// The Turtle parser rejected the input.
    #[error("turtle: {0}")]
    Turtle(String),
    /// The TriG parser rejected the input.
    #[error("trig: {0}")]
    TriG(String),
    /// The RDF/XML parser rejected the input.
    #[error("rdf/xml: {0}")]
    RdfXml(String),
    /// The requested format is not one this crate can parse.
    #[error("unknown input format: {0} (expected nt, nq, ttl, trig, or rdf/xml)")]
    UnknownFormat(String),
    /// [`QuotedTripleSurface::Rdf12`] was asked for on a Turtle/TriG input, but
    /// this build of `rete-core` has the `rdf12-turtle` feature off, so the
    /// RDF 1.2 reader was not compiled in. Refused by name rather than read as
    /// RDF-star, which would be the wrong graph told in silence.
    #[error(
        "this build cannot read the RDF 1.2 Turtle/TriG surface: rete-core was compiled without \
         the `rdf12-turtle` feature (N-Triples/N-Quads still accept `<<( s p o )>>`)"
    )]
    Rdf12TurtleUnavailable,
    /// The input could not be read (streaming builds surface the path here).
    #[error("io: {0}")]
    Io(String),
    /// An IRI outside the N-Triples/N-Quads `IRIREF` grammar / RFC 3987, refused
    /// because the caller asked for a strict [`IriAudit`] (`rete build
    /// --strict`). Without `strict` the same finding is only counted.
    /// `location` is `line N` for the line-based syntaxes, the format name for
    /// the others (oxttl reports its own positions).
    #[error("{location}: invalid IRI {token} — {reason}")]
    InvalidIri {
        /// Where it was seen: `line 42`, or the format name.
        location: String,
        /// The offending term token, verbatim.
        token: String,
        /// Which rule it breaks ([`crate::iri::IriDefect::reason`]).
        reason: &'static str,
    },
}

/// Policy **and** tally for the invalid-IRI check that runs during ingest.
///
/// rete's reader is deliberately tolerant of IRIs a strict parser rejects (see
/// [`crate::iri`]), which is what lets it hold every published dataset — and
/// also what let three of the `scholar/` exports produce N-Quads Oxigraph
/// refuses to load. Passing an audit makes the tolerance *visible*: the build
/// still succeeds and stores exactly what it was given, but it can now say how
/// much of what it stored is not valid RDF. `strict` turns the same finding into
/// a refusal, on the first offending statement.
#[derive(Debug, Default)]
pub struct IriAudit {
    /// Fail the parse on the first invalid IRI instead of counting it.
    pub strict: bool,
    /// What the parse saw. Constant memory — see [`crate::iri::IriReport`].
    pub report: crate::iri::IriReport,
}

impl IriAudit {
    /// Count invalid IRIs; never fail (`rete build`'s default).
    pub fn counting() -> Self {
        Self::default()
    }

    /// Refuse the input at the first invalid IRI (`rete build --strict`).
    pub fn refusing() -> Self {
        Self {
            strict: true,
            report: crate::iri::IriReport::default(),
        }
    }
}

/// Run one statement past the audit: record any invalid IRI, and in `strict`
/// mode turn it into an error naming the offending token. `location` is only
/// called on the error path, so the common case allocates nothing.
fn audit_quad<F: FnOnce() -> String>(
    audit: Option<&mut IriAudit>,
    s: &str,
    p: &str,
    o: &str,
    g: Option<&str>,
    location: F,
) -> Result<(), IngestError> {
    let Some(a) = audit else { return Ok(()) };
    let Some(defect) = a.report.observe_quad(s, p, o, g) else {
        return Ok(());
    };
    if !a.strict {
        return Ok(());
    }
    let token = [Some(s), Some(p), Some(o), g]
        .into_iter()
        .flatten()
        .find(|t| crate::iri::term_defect(t).is_some())
        .unwrap_or("<?>")
        .to_string();
    Err(IngestError::InvalidIri {
        location: location(),
        token,
        reason: defect.reason(),
    })
}

/// Estimate the statement count of an N-Triples/N-Quads text from its newline
/// count, so the output `Vec` can be pre-sized — a big build otherwise pays
/// repeated `Vec` doublings, each briefly holding ~2× the (large) spine. An
/// over-estimate by a few blank/comment lines is harmless (it's a capacity hint).
fn estimate_statements(input: &str) -> usize {
    bytecount_newlines(input).max(1)
}

/// Count `\n` bytes — one linear pass, far cheaper than the parse it sizes.
fn bytecount_newlines(input: &str) -> usize {
    input.as_bytes().iter().filter(|&&b| b == b'\n').count()
}

/// Parse one N-Quads line into a quad, or `None` for a blank/comment line.
/// Shared by the whole-text [`parse_quads`] and the streaming [`parse_reader`].
fn parse_nq_line(
    raw: &str,
    lineno: usize,
    audit: Option<&mut IriAudit>,
) -> Result<Option<RawQuad>, IngestError> {
    let line = raw.trim();
    if line.is_empty() || line.starts_with('#') {
        return Ok(None);
    }
    let stripped = line
        .strip_suffix('.')
        .ok_or(IngestError::Line(lineno, "missing trailing '.'"))?
        .trim_end();
    let (s, rest) = take_term(stripped).ok_or(IngestError::Line(lineno, "bad subject"))?;
    let (p, rest) =
        take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad predicate"))?;
    let (o, rest) = take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad object"))?;
    let rest = rest.trim();
    let graph = if rest.is_empty() {
        None
    } else {
        let (g, tail) = take_term(rest).ok_or(IngestError::Line(lineno, "bad graph"))?;
        if !tail.trim().is_empty() {
            return Err(IngestError::Line(lineno, "trailing content after graph"));
        }
        Some(g)
    };
    audit_quad(audit, &s, &p, &o, graph.as_deref(), || {
        format!("line {lineno}")
    })?;
    Ok(Some((s, p, o, graph)))
}

/// Parse one N-Triples line into a triple, or `None` for a blank/comment line.
fn parse_nt_line(
    raw: &str,
    lineno: usize,
    audit: Option<&mut IriAudit>,
) -> Result<Option<RawTriple>, IngestError> {
    let line = raw.trim();
    if line.is_empty() || line.starts_with('#') {
        return Ok(None);
    }
    let stripped = line
        .strip_suffix('.')
        .ok_or(IngestError::Line(lineno, "missing trailing '.'"))?
        .trim_end();
    let (s, rest) = take_term(stripped).ok_or(IngestError::Line(lineno, "bad subject"))?;
    let (p, rest) =
        take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad predicate"))?;
    let (o, rest) = take_term(rest.trim_start()).ok_or(IngestError::Line(lineno, "bad object"))?;
    if !rest.trim().is_empty() {
        return Err(IngestError::Line(lineno, "trailing content after object"));
    }
    audit_quad(audit, &s, &p, &o, None, || format!("line {lineno}"))?;
    Ok(Some((s, p, o)))
}

/// Parse N-Quads text: `subject predicate object [graph] .` per line.
pub fn parse_quads(input: &str) -> Result<Vec<RawQuad>, IngestError> {
    parse_quads_audited(input, None)
}

/// [`parse_quads`] with an optional [`IriAudit`].
pub fn parse_quads_audited(
    input: &str,
    mut audit: Option<&mut IriAudit>,
) -> Result<Vec<RawQuad>, IngestError> {
    let mut out = Vec::with_capacity(estimate_statements(input));
    for (i, raw) in input.lines().enumerate() {
        if let Some(q) = parse_nq_line(raw, i + 1, audit.as_deref_mut())? {
            out.push(q);
        }
    }
    Ok(out)
}

/// Parse N-Triples text into raw term-token triples, skipping blank/comment lines.
pub fn parse(input: &str) -> Result<Vec<RawTriple>, IngestError> {
    parse_audited(input, None)
}

/// [`parse`] with an optional [`IriAudit`].
pub fn parse_audited(
    input: &str,
    mut audit: Option<&mut IriAudit>,
) -> Result<Vec<RawTriple>, IngestError> {
    let mut out = Vec::with_capacity(estimate_statements(input));
    for (i, raw) in input.lines().enumerate() {
        if let Some(t) = parse_nt_line(raw, i + 1, audit.as_deref_mut())? {
            out.push(t);
        }
    }
    Ok(out)
}

/// **Stream-parse** N-Triples (`"nt"`) or N-Quads (`"nq"`) from a reader, one
/// line at a time, so the whole input text is **never resident** — the big-build
/// memory win over reading the file into a `String` first (each line String is
/// transient, freed every iteration). `cap` pre-sizes the output `Vec` (e.g.
/// `file_len / 64`) to avoid reallocation doublings of the large spine. Turtle is
/// not streamable here (oxttl needs the whole input); callers use the text path
/// for `"ttl"`.
pub fn parse_reader<R: std::io::BufRead>(
    reader: R,
    format: &str,
    cap: usize,
) -> Result<Vec<RawQuad>, IngestError> {
    parse_reader_audited(reader, format, cap, None)
}

/// [`parse_reader`] with an optional [`IriAudit`].
pub fn parse_reader_audited<R: std::io::BufRead>(
    reader: R,
    format: &str,
    cap: usize,
    audit: Option<&mut IriAudit>,
) -> Result<Vec<RawQuad>, IngestError> {
    parse_reader_audited_surface(reader, format, cap, audit, QuotedTripleSurface::default())
}

/// [`parse_reader_audited`] reading `<< … >>` in the given
/// [`QuotedTripleSurface`] (Turtle/TriG only — see the enum).
pub fn parse_reader_audited_surface<R: std::io::BufRead>(
    reader: R,
    format: &str,
    cap: usize,
    audit: Option<&mut IriAudit>,
    surface: QuotedTripleSurface,
) -> Result<Vec<RawQuad>, IngestError> {
    let mut out = Vec::with_capacity(cap);
    stream_reader_audited_surface(reader, format, audit, surface, &mut |q| out.push(q))?;
    Ok(out)
}

/// **Stream** N-Triples (`"nt"`) / N-Quads (`"nq"`) from a reader, invoking `f`
/// once per parsed quad — like [`parse_reader`] but **without collecting** into a
/// `Vec`. Each quad's term Strings are owned by `f` (and dropped when it returns
/// if it doesn't retain them), so a caller that only needs to *observe* every
/// term — e.g. the two-pass [`assemble_dataset_streaming`] building its
/// dictionary — never materializes the whole quad multiset. The big-graph,
/// low-RAM ingest primitive. Blank/comment lines are skipped; a parse error stops
/// the stream and is returned.
///
/// Every format streams: the line-based ones a line at a time, Turtle / TriG /
/// RDF-XML through their `oxttl`/`oxrdfxml` reader parsers, which pull from the
/// reader incrementally. Nothing here holds the input text — so a 60 GB Turtle
/// file is as ingestible as a 60 GB `.nt` one, which is what lets the external
/// (memory-bounded) build accept them.
pub fn stream_reader<R: std::io::BufRead>(
    reader: R,
    format: &str,
    f: &mut dyn FnMut(RawQuad),
) -> Result<(), IngestError> {
    stream_reader_audited(reader, format, None, f)
}

/// [`stream_reader`] with an optional [`IriAudit`].
///
/// Every syntax is audited, not only the line-based ones. `oxttl`/`oxrdfxml`
/// validate IRIs themselves and reject an invalid one before it reaches us, so
/// in practice the Turtle/TriG/RDF-XML tallies stay at zero — but auditing them
/// too means the count is a property of the *file*, not of which reader happened
/// to produce it, and a future parser change cannot silently reopen the hole.
pub fn stream_reader_audited<R: std::io::BufRead>(
    reader: R,
    format: &str,
    audit: Option<&mut IriAudit>,
    f: &mut dyn FnMut(RawQuad),
) -> Result<(), IngestError> {
    stream_reader_audited_surface(reader, format, audit, QuotedTripleSurface::default(), f)
}

/// [`stream_reader_audited`] reading `<< … >>` in the given
/// [`QuotedTripleSurface`].
///
/// The surface reaches only the `"ttl"` and `"trig"` arms: the line-based
/// readers are rete's own and take both spellings unconditionally, and RDF/XML
/// has no syntax for either. Under [`QuotedTripleSurface::Rdf12`] Turtle/TriG go
/// through `oxttl` 0.2 instead of 0.1, and the RDF 1.2 triple terms it hands
/// back (`<<( s p o )>>`) are folded to rete's stored token on the way out, so
/// the dictionary a file lands in does not depend on which reader read it.
///
/// **This entry point does not carry the RDF 1.2 hint** that the text path adds
/// to a failed RDF-star parse: it is handed a reader, not a string, and sniffing
/// the input would mean buffering the 60 GB file this function exists to avoid
/// buffering. `rete build --memory-budget-mb file.ttl` therefore reports the raw
/// parser error; every other route reports the hint.
pub fn stream_reader_audited_surface<R: std::io::BufRead>(
    reader: R,
    format: &str,
    mut audit: Option<&mut IriAudit>,
    surface: QuotedTripleSurface,
    f: &mut dyn FnMut(RawQuad),
) -> Result<(), IngestError> {
    if surface == QuotedTripleSurface::Rdf12 && matches!(format, "ttl" | "trig") {
        return rdf12::stream(reader, format, audit, f);
    }
    match format {
        "nt" | "nq" => {
            for (i, line) in reader.lines().enumerate() {
                let line = line.map_err(|e| IngestError::Io(e.to_string()))?;
                if format == "nq" {
                    if let Some(q) = parse_nq_line(&line, i + 1, audit.as_deref_mut())? {
                        f(q);
                    }
                } else if let Some((s, p, o)) = parse_nt_line(&line, i + 1, audit.as_deref_mut())? {
                    f((s, p, o, None));
                }
            }
            Ok(())
        }
        "ttl" => turtle_star(reader, audit, f),
        "trig" => trig_star(reader, audit, f),
        "rdfxml" => {
            for r in oxrdfxml::RdfXmlParser::new().for_reader(reader) {
                let t = r.map_err(|e| IngestError::RdfXml(e.to_string()))?;
                let q = (
                    t.subject.to_string(),
                    t.predicate.to_string(),
                    t.object.to_string(),
                    None,
                );
                audit_quad(audit.as_deref_mut(), &q.0, &q.1, &q.2, None, || {
                    "rdf/xml".into()
                })?;
                f(q);
            }
            Ok(())
        }
        other => Err(IngestError::UnknownFormat(other.to_string())),
    }
}

/// Stream **RDF-star** Turtle: `with_quoted_triples` accepts `<< s p o >>` in
/// subject/object position, and oxrdf 0.2's `Term::Triple` Displays as the
/// canonical `<<…>>` token rete's own N-Triples-star tokenizer emits — so no
/// translation is needed on the way out.
///
/// Its own function, and called directly from both the reader path and the text
/// path, so that the wasm engine — which reaches only the text path and has no
/// RDF 1.2 reader compiled in — links this and nothing else.
fn turtle_star<R: std::io::BufRead>(
    reader: R,
    mut audit: Option<&mut IriAudit>,
    f: &mut dyn FnMut(RawQuad),
) -> Result<(), IngestError> {
    for r in oxttl::TurtleParser::new()
        .with_quoted_triples()
        .for_reader(reader)
    {
        let t = r.map_err(|e| IngestError::Turtle(e.to_string()))?;
        let q = (
            t.subject.to_string(),
            t.predicate.to_string(),
            t.object.to_string(),
            None,
        );
        audit_quad(audit.as_deref_mut(), &q.0, &q.1, &q.2, None, || {
            "turtle".into()
        })?;
        f(q);
    }
    Ok(())
}

/// [`turtle_star`] for TriG — Turtle plus named-graph blocks (`<g> { … }`), the
/// shape most large RDF dumps ship in.
fn trig_star<R: std::io::BufRead>(
    reader: R,
    mut audit: Option<&mut IriAudit>,
    f: &mut dyn FnMut(RawQuad),
) -> Result<(), IngestError> {
    for r in oxttl::TriGParser::new()
        .with_quoted_triples()
        .for_reader(reader)
    {
        let q = r.map_err(|e| IngestError::TriG(e.to_string()))?;
        let q = (
            q.subject.to_string(),
            q.predicate.to_string(),
            q.object.to_string(),
            graph_token(&q.graph_name),
        );
        audit_quad(
            audit.as_deref_mut(),
            &q.0,
            &q.1,
            &q.2,
            q.3.as_deref(),
            || "trig".into(),
        )?;
        f(q);
    }
    Ok(())
}

/// The canonical graph token for a parsed quad — `None` for the default graph,
/// otherwise the same `<iri>` / `_:b` spelling the N-Quads reader produces, so
/// TriG and N-Quads inputs land identical dictionary keys. `GraphName`'s own
/// `Display` renders the default graph as `DEFAULT`, which must never become a
/// term, hence the explicit match.
fn graph_token(g: &oxrdf::GraphName) -> Option<String> {
    match g {
        oxrdf::GraphName::DefaultGraph => None,
        other => Some(other.to_string()),
    }
}

/// The **RDF 1.2** Turtle/TriG reader — `oxttl` 0.2 with its `rdf-12` feature,
/// living beside the `oxttl` 0.1 reader above rather than replacing it.
///
/// Everything RDF 1.2 adds over RDF 1.1 arrives through this one module:
/// `<<( s p o )>>` triple terms, `<< s p o >>` reifiers (which expand to
/// `_:r rdf:reifies <<( s p o )>>` plus the statement that mentioned them),
/// `{| … |}` annotations, and `"…"@lang--dir` directional literals. None of it
/// needs a storage change, which is the point: a reifier is a blank node, an
/// IRI and a term, and rete has stored those since v0.
///
/// The one translation is at the term boundary. `oxrdf` 0.3 renders a triple
/// term as `<<( s p o )>>`; rete's dictionary key is `<<s p o>>`. [`take_term`]
/// already reads both — that is what made RDF 1.2 N-Quads ingestible in #262 —
/// so the fold is a re-scan of a token oxttl has already validated.
#[cfg(feature = "rdf12-turtle")]
mod rdf12 {
    use super::{audit_quad, take_term, IngestError, IriAudit, RawQuad};

    /// Fold `oxrdf` 0.3's `<<( s p o )>>` rendering of a triple term into rete's
    /// stored `<<s p o>>`, recursively (a nested triple term folds with it).
    /// Every other token — the overwhelming majority, and all of them in a file
    /// with no triple terms — is returned untouched, not even copied.
    pub(super) fn fold_triple_term(t: String) -> String {
        if !t.starts_with("<<(") {
            return t;
        }
        match take_term(&t) {
            Some((token, rest)) if rest.trim().is_empty() => token,
            // oxttl only ever hands back a well-formed term, so this is
            // unreachable; returning the original beats mangling it silently.
            _ => t,
        }
    }

    /// `super::graph_token` for `oxrdf` 0.3's `GraphName` — the same rule, a
    /// different crate version, which is exactly why it cannot be shared.
    fn graph_token(g: &oxrdf12::GraphName) -> Option<String> {
        match g {
            oxrdf12::GraphName::DefaultGraph => None,
            other => Some(other.to_string()),
        }
    }

    /// Stream one RDF 1.2 Turtle (`"ttl"`) or TriG (`"trig"`) input, auditing
    /// every quad the way the RDF-star reader does.
    pub(super) fn stream<R: std::io::BufRead>(
        reader: R,
        format: &str,
        mut audit: Option<&mut IriAudit>,
        f: &mut dyn FnMut(RawQuad),
    ) -> Result<(), IngestError> {
        if format == "ttl" {
            for r in oxttl12::TurtleParser::new().for_reader(reader) {
                let t = r.map_err(|e| IngestError::Turtle(e.to_string()))?;
                let q = (
                    t.subject.to_string(),
                    t.predicate.to_string(),
                    fold_triple_term(t.object.to_string()),
                    None,
                );
                audit_quad(audit.as_deref_mut(), &q.0, &q.1, &q.2, None, || {
                    "turtle".into()
                })?;
                f(q);
            }
        } else {
            for r in oxttl12::TriGParser::new().for_reader(reader) {
                let q = r.map_err(|e| IngestError::TriG(e.to_string()))?;
                let q = (
                    q.subject.to_string(),
                    q.predicate.to_string(),
                    fold_triple_term(q.object.to_string()),
                    graph_token(&q.graph_name),
                );
                audit_quad(
                    audit.as_deref_mut(),
                    &q.0,
                    &q.1,
                    &q.2,
                    q.3.as_deref(),
                    || "trig".into(),
                )?;
                f(q);
            }
        }
        Ok(())
    }
}

/// The stand-in for [`rdf12`] in a build with `rdf12-turtle` off: it refuses by
/// name. Reading the file as RDF-star instead would answer a question nobody
/// asked, with a different graph.
#[cfg(not(feature = "rdf12-turtle"))]
mod rdf12 {
    use super::{IngestError, IriAudit, RawQuad};

    pub(super) fn stream<R: std::io::BufRead>(
        _reader: R,
        _format: &str,
        _audit: Option<&mut IriAudit>,
        _f: &mut dyn FnMut(RawQuad),
    ) -> Result<(), IngestError> {
        Err(IngestError::Rdf12TurtleUnavailable)
    }
}

/// Parse Turtle into canonical N-Triples-token triples via oxttl, reading
/// `<< s p o >>` as an RDF-star quoted triple. [`parse_turtle_surface`] chooses.
pub fn parse_turtle(text: &str) -> Result<Vec<RawTriple>, IngestError> {
    parse_turtle_surface(text, QuotedTripleSurface::default())
}

/// [`parse_turtle`] in the given [`QuotedTripleSurface`].
pub fn parse_turtle_surface(
    text: &str,
    surface: QuotedTripleSurface,
) -> Result<Vec<RawTriple>, IngestError> {
    Ok(parse_text_surface(text, "ttl", surface)?
        .into_iter()
        .map(|(s, p, o, _)| (s, p, o))
        .collect())
}

/// Parse TriG into canonical quads via oxttl — Turtle plus named-graph blocks
/// (`<g> { … }`), the shape most large RDF dumps ship in. Reads `<< s p o >>` as
/// an RDF-star quoted triple; [`parse_trig_surface`] chooses.
pub fn parse_trig(text: &str) -> Result<Vec<RawQuad>, IngestError> {
    parse_trig_surface(text, QuotedTripleSurface::default())
}

/// [`parse_trig`] in the given [`QuotedTripleSurface`].
pub fn parse_trig_surface(
    text: &str,
    surface: QuotedTripleSurface,
) -> Result<Vec<RawQuad>, IngestError> {
    parse_text_surface(text, "trig", surface)
}

/// The shared Turtle/TriG text path: stream the string through whichever reader
/// `surface` selects, and — this is the part only the text path can do, because
/// only it holds the input — say so when an RDF-star parse fails on what is
/// plainly RDF 1.2.
///
/// It does not audit: the Turtle/TriG text callers audit over the finished quads
/// (see [`parse_statements_audited_surface`]), and auditing here too would count
/// every IRI twice.
///
/// The RDF-star arm calls `oxttl` 0.1 **directly** rather than routing through
/// [`stream_reader_audited_surface`]. That is not style: this is the only ingest
/// path the wasm engine reaches, and going through the generic reader would
/// monomorphize it over `&[u8]` and pull the whole `match format` — every
/// syntax's reader — into a module that previously linked one Turtle parser.
/// Measured at +17 KB of wasm for code no browser can run, since the RDF 1.2
/// reader is not compiled there at all. Keeping the arm direct keeps the
/// browser artifacts what they were.
fn parse_text_surface(
    text: &str,
    format: &str,
    surface: QuotedTripleSurface,
) -> Result<Vec<RawQuad>, IngestError> {
    let mut out = Vec::new();
    let r = if surface == QuotedTripleSurface::Rdf12 {
        rdf12::stream(text.as_bytes(), format, None, &mut |q| out.push(q))
    } else if format == "ttl" {
        turtle_star(text.as_bytes(), None, &mut |q| out.push(q))
    } else {
        trig_star(text.as_bytes(), None, &mut |q| out.push(q))
    };
    match r {
        Ok(()) => Ok(out),
        // The default surface met `<<(`: that is an RDF 1.2 file being read as
        // RDF-star, the one mistake this flag exists to prevent, and the parse
        // error alone ("unexpected character") would send the reader hunting for
        // a typo that is not there.
        Err(IngestError::Turtle(m))
            if surface == QuotedTripleSurface::RdfStar && looks_like_rdf12(text) =>
        {
            Err(IngestError::Turtle(format!("{m}{RDF12_SURFACE_HINT}")))
        }
        Err(IngestError::TriG(m))
            if surface == QuotedTripleSurface::RdfStar && looks_like_rdf12(text) =>
        {
            Err(IngestError::TriG(format!("{m}{RDF12_SURFACE_HINT}")))
        }
        Err(e) => Err(e),
    }
}

/// Parse RDF/XML into canonical N-Triples-token triples via oxrdfxml. This is how
/// most OWL ontologies ship (`.rdf`/`.owl`/`.xml` with an `rdf:RDF` root) — so rete
/// ingests them directly, no external conversion. (OWL/XML — the non-RDF functional
/// XML serialization — is a different language; convert it with owlready2 first.)
pub fn parse_rdfxml(text: &str) -> Result<Vec<RawTriple>, IngestError> {
    let mut out = Vec::new();
    for r in oxrdfxml::RdfXmlParser::new().for_reader(text.as_bytes()) {
        let t = r.map_err(|e| IngestError::RdfXml(e.to_string()))?;
        out.push((
            t.subject.to_string(),
            t.predicate.to_string(),
            t.object.to_string(),
        ));
    }
    Ok(out)
}

/// Parse one text input by format name (`"nt"`, `"nq"`, `"ttl"`, `"trig"`, or
/// `"rdfxml"`) into quads (triples land in the default graph).
pub fn parse_statements(text: &str, format: &str) -> Result<Vec<RawQuad>, IngestError> {
    parse_statements_audited(text, format, None)
}

/// [`parse_statements`] with an optional [`IriAudit`]. The line-based syntaxes
/// audit as they parse (so `strict` fails on the offending line); the others are
/// audited over the parsed quads, which is where `oxttl` has already had its say.
pub fn parse_statements_audited(
    text: &str,
    format: &str,
    audit: Option<&mut IriAudit>,
) -> Result<Vec<RawQuad>, IngestError> {
    parse_statements_audited_surface(text, format, audit, QuotedTripleSurface::default())
}

/// [`parse_statements_audited`] reading `<< … >>` in the given
/// [`QuotedTripleSurface`]. This is the entry point `rete build` and
/// `rete validate` use for everything but a streamed line-based input, and the
/// one that turns "RDF 1.2 file, RDF-star reader" into an error that names the
/// flag instead of a bare syntax complaint.
pub fn parse_statements_audited_surface(
    text: &str,
    format: &str,
    mut audit: Option<&mut IriAudit>,
    surface: QuotedTripleSurface,
) -> Result<Vec<RawQuad>, IngestError> {
    let quads = match format {
        "nq" => return parse_quads_audited(text, audit),
        "nt" => {
            return Ok(parse_audited(text, audit)?
                .into_iter()
                .map(|(s, p, o)| (s, p, o, None))
                .collect())
        }
        // Straight to the quad path for both: going via `parse_turtle_surface`
        // would build a `Vec<RawQuad>`, rebuild it as a `Vec<RawTriple>` to
        // satisfy that function's signature, and rebuild it as quads again —
        // three copies of every term String on the biggest input rete takes.
        "ttl" | "trig" => parse_text_surface(text, format, surface)?,
        "rdfxml" => parse_rdfxml(text)?
            .into_iter()
            .map(|(s, p, o)| (s, p, o, None))
            .collect(),
        other => return Err(IngestError::UnknownFormat(other.to_string())),
    };
    for (s, p, o, g) in &quads {
        audit_quad(audit.as_deref_mut(), s, p, o, g.as_deref(), || {
            format.to_string()
        })?;
    }
    Ok(quads)
}

/// Take one term from the front of `s`, returning `(term, remainder)`.
pub(crate) fn take_term(s: &str) -> Option<(String, &str)> {
    let bytes = s.as_bytes();
    let first = *bytes.first()?;
    match first {
        // A quoted triple / triple term, in EITHER surface — the inner terms are
        // themselves terms (so this recurses; nesting works). `<<` starts with `<`
        // and an IRI scan would stop at the first inner `>`, so it must be handled
        // before the plain-IRI case.
        //   * RDF-star:  `<< subject predicate object >>`
        //   * RDF 1.2:   `<<( subject predicate object )>>`  (triple term)
        // Both re-emit the SAME canonical token `<<s p o>>`, so a file written in
        // either surface — and a query written either way — dedupe and match. This
        // makes RDF 1.2 N-Triples interoperable with the RDF-star we already store,
        // with no format change and no dependency swap.
        b'<' if bytes.get(1) == Some(&b'<') => {
            // Distinguish `<<(` (RDF 1.2) from `<<` (RDF-star) by the char after `<<`.
            let (inner, rdf12) = match s[2..].strip_prefix('(') {
                Some(after) => (after.trim_start(), true),
                None => (s[2..].trim_start(), false),
            };
            let (subj, r) = take_term(inner)?;
            let (pred, r) = take_term(r.trim_start())?;
            let (obj, r) = take_term(r.trim_start())?;
            let r = r.trim_start();
            let rest = if rdf12 {
                r.strip_prefix(")>>")?
            } else {
                r.strip_prefix(">>")?
            };
            // Canonical surface = oxrdf's `Triple` Display: `<<s p o>>` (tight
            // brackets, single spaces between components) — the same token for both
            // input surfaces, so N-Triples-star, Turtle-star, and RDF 1.2 triple
            // terms all resolve to one dictionary entry.
            Some((format!("<<{subj} {pred} {obj}>>"), rest))
        }
        b'<' => {
            // IRI ref: up to the closing '>'.
            let end = s.find('>')?;
            Some((s[..=end].to_string(), &s[end + 1..]))
        }
        b'_' => {
            // Blank node: up to whitespace.
            let end = s.find(char::is_whitespace).unwrap_or(s.len());
            Some((s[..end].to_string(), &s[end..]))
        }
        b'"' => {
            // Literal: closing unescaped quote, then optional ^^<dt> or @lang.
            let mut i = 1;
            let b = s.as_bytes();
            while i < b.len() {
                match b[i] {
                    b'\\' => i += 2, // skip escaped char
                    b'"' => break,
                    _ => i += 1,
                }
            }
            if i >= b.len() {
                return None; // unterminated
            }
            let mut end = i + 1; // past closing quote
            if s[end..].starts_with("^^<") {
                let close = s[end..].find('>')? + end;
                end = close + 1;
            } else if s[end..].starts_with('@') {
                // Language tag: '@' then BCP-47 subtags `[a-zA-Z0-9-]+`. Stop at
                // the first char that can't be part of a tag — normally the
                // whitespace before the predicate, but ALSO the `>>` that closes
                // a quoted triple when this literal is its object (`"x"@en>>`),
                // where there is no separating whitespace.
                let mut j = end + 1; // past '@'
                while j < b.len() && (b[j].is_ascii_alphanumeric() || b[j] == b'-') {
                    j += 1;
                }
                end = j;
            }
            Some((s[..end].to_string(), &s[end..]))
        }
        _ => None,
    }
}

/// Split a canonical quoted-triple token `<<s p o>>` (RDF-star) into its three
/// component term tokens, or `None` if `t` is not a quoted triple. Reuses
/// [`take_term`] for term-boundary scanning, so nested quoting parses correctly.
/// The inverse of the `<<…>>` construction; used by the SUBJECT/PREDICATE/OBJECT
/// SPARQL-star builtins.
pub(crate) fn quoted_triple_parts(t: &str) -> Option<(String, String, String)> {
    let inner = t.strip_prefix("<<")?.strip_suffix(">>")?.trim();
    let (s, r) = take_term(inner)?;
    let (p, r) = take_term(r.trim_start())?;
    let (o, r) = take_term(r.trim_start())?;
    if !r.trim().is_empty() {
        return None;
    }
    Some((s, p, o))
}

/// Counts describing an assembled file, for status lines and UIs.
///
/// In the [`BuildStats`] a metadata callback receives, `statements` and
/// `default_triples` are the counts **ingested** — the input multiset, before
/// the index deduplicates it. In the `BuildStats` an assemble function
/// *returns*, they are the counts actually **written** (see
/// [`FinalCounts`]), so a status line reports what the file holds. Use
/// [`FinalCounts`] rather than these when a payload must agree with the header.
#[derive(Debug, Clone, Copy)]
pub struct BuildStats {
    /// Total statements, across the default graph and every named one.
    pub statements: usize,
    /// Statements that landed in the default graph.
    pub default_triples: usize,
    /// How many named graphs the input mentioned; one index is written per graph.
    pub named_graphs: usize,
    /// Distinct terms in the shared dictionary.
    pub terms: usize,
    /// Levels in the community pyramid, or `0` when the build skipped it.
    pub pyramid_levels: u16,
}

/// The **deduplicated** counts of the file being written — what the header
/// records, and therefore what any embedded metadata has to report.
///
/// The input multiset is not the file: every permutation index sorts and
/// deduplicates, so a harvest that pages with overlapping windows writes fewer
/// quads than it ingested. These counts are known only once the indexes exist,
/// which is why a payload that carries them is derived in two stages (see
/// [`IntoMetadata`]).
#[derive(Debug, Clone, Copy)]
pub struct FinalCounts {
    /// Unique triples in the default graph.
    pub default_triples: u64,
    /// Unique quads across the default graph and every named graph — exactly
    /// the header's `quad_count`.
    pub quads: u64,
}

/// What a build's metadata callback may return.
///
/// A `Vec<u8>` is the payload verbatim — the common case (no metadata, or a
/// payload that carries no counts). [`DeferredMetadata`] is a payload that
/// *does* carry counts: the callback derives everything it needs while the
/// source quads are still resident, and returns a finalizer that the writer
/// calls with the [`FinalCounts`], which only exist once the indexes have
/// deduplicated the input. Without that second stage a card can only report the
/// ingested count, which over-reports any input containing duplicates.
pub trait IntoMetadata {
    /// Produce the metadata section payload for a file with these counts.
    fn into_metadata(self, counts: FinalCounts) -> Vec<u8>;
}

impl IntoMetadata for Vec<u8> {
    fn into_metadata(self, _counts: FinalCounts) -> Vec<u8> {
        self
    }
}

/// A metadata payload whose final form needs the written (deduplicated) counts.
/// See [`IntoMetadata`].
pub struct DeferredMetadata(Box<dyn FnOnce(FinalCounts) -> Vec<u8>>);

impl DeferredMetadata {
    /// Defer `f` until the indexes are built and the true counts are known.
    pub fn new(f: impl FnOnce(FinalCounts) -> Vec<u8> + 'static) -> Self {
        DeferredMetadata(Box::new(f))
    }

    /// No metadata section — byte-identical to a metadata-free build. Lets one
    /// `match` arm opt out while the other defers, without a type mismatch.
    pub fn none() -> Self {
        DeferredMetadata::new(|_| Vec::new())
    }
}

impl IntoMetadata for DeferredMetadata {
    fn into_metadata(self, counts: FinalCounts) -> Vec<u8> {
        (self.0)(counts)
    }
}

/// Assemble a complete `.rete` file image from parsed quads: one shared
/// dictionary, the default-graph index, one index per named graph, and the
/// community pyramid. `metadata` is the opaque metadata-section payload (the
/// CLI puts a JSON Dataset Card there); pass `&[]` for none — that is
/// byte-identical to a metadata-free build.
pub fn assemble_dataset(quads: Vec<RawQuad>, metadata: &[u8]) -> (Vec<u8>, BuildStats) {
    let blob = metadata.to_vec();
    assemble_dataset_with(quads, move |_, _| blob)
}

/// Like [`assemble_dataset`], but the metadata payload is derived from the
/// [`BuildStats`] and the source quads while they are still resident — for
/// metadata that describes the graph (the Dataset Card). Returning an empty
/// `Vec` is byte-identical to a metadata-free build; return a
/// [`DeferredMetadata`] for a payload that must also carry the file's
/// deduplicated counts, which exist only once the indexes are built.
pub fn assemble_dataset_with<M: IntoMetadata>(
    quads: Vec<RawQuad>,
    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
) -> (Vec<u8>, BuildStats) {
    assemble_dataset_with_opts(quads, true, false, None, metadata)
}

/// Like [`assemble_dataset_with`], but `with_pyramid = false` skips the Louvain
/// community pyramid entirely — no pyramid section is written (header length 0).
/// SPARQL / SHACL / triple / reachability queries don't use the pyramid, so a
/// pyramid-less file is fully queryable and markedly smaller (the pyramid is the
/// largest section on highly-connected graphs). Only the community / summary /
/// progressive paths need it.
pub fn assemble_dataset_with_opts<M: IntoMetadata>(
    quads: Vec<RawQuad>,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
) -> (Vec<u8>, BuildStats) {
    assemble_dataset_with_opts_algo(
        quads,
        with_pyramid,
        with_text_index,
        type_override,
        PyramidAlgo::Louvain,
        metadata,
    )
}

/// Like [`assemble_dataset_with_opts`], but selects the community [`PyramidAlgo`]
/// (the in-memory build path for `rete build --pyramid-algo …`).
#[allow(clippy::too_many_arguments)]
pub fn assemble_dataset_with_opts_algo<M: IntoMetadata>(
    quads: Vec<RawQuad>,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    algo: PyramidAlgo,
    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
) -> (Vec<u8>, BuildStats) {
    assemble_dataset_with_perms(
        quads,
        with_pyramid,
        with_text_index,
        type_override,
        algo,
        crate::index::PermSet::ALL,
        metadata,
    )
}

/// Like [`assemble_dataset_with_opts_algo`], but writes only the permutations in
/// `perms` (`rete build --permutations 3`). Every other byte of the file is
/// unchanged; [`crate::index::PermSet::ALL`] reproduces the default build
/// exactly, down to the header byte.
#[allow(clippy::too_many_arguments)]
pub fn assemble_dataset_with_perms<M: IntoMetadata>(
    quads: Vec<RawQuad>,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    algo: PyramidAlgo,
    perms: crate::index::PermSet,
    metadata: impl FnOnce(&BuildStats, &[RawQuad]) -> M,
) -> (Vec<u8>, BuildStats) {
    use std::collections::BTreeMap;

    let mut db = DictionaryBuilder::new();
    for (s, p, o, _) in &quads {
        db.observe(s, p, o);
    }
    let dict = db.build();

    let mut default_triples = Vec::new();
    let mut named: BTreeMap<String, Vec<(u32, u32, u32)>> = BTreeMap::new();
    for (s, p, o, g) in &quads {
        let t = dict.encode(s, p, o).expect("observed term");
        match g {
            None => default_triples.push(t),
            Some(graph) => named.entry(graph.clone()).or_default().push(t),
        }
    }

    // Derive the metadata blob (the Dataset Card) from the raw quads NOW, while
    // they are resident — then DROP them before the memory-heavy pyramid + index
    // phases. On a big build the string quads are the largest working set (every
    // term an owned String, heavily duplicated) and are fully redundant with the
    // dictionary + id-triples once encoded, so freeing them here cuts peak RAM by
    // their whole size. `pyramid_levels` is not known yet (0 in the callback);
    // it is filled into the returned `stats` by `finish_assembly`, and no metadata
    // callback depends on it (the card derives from the quads + term/graph counts).
    let stats = BuildStats {
        statements: quads.len(),
        default_triples: default_triples.len(),
        named_graphs: named.len(),
        terms: dict.term_count() as usize,
        pyramid_levels: 0,
    };
    let pending = metadata(&stats, &quads);
    drop(quads);

    finish_assembly(
        dict,
        default_triples,
        named,
        with_pyramid,
        with_text_index,
        type_override,
        algo,
        perms,
        pending,
        stats,
    )
}

/// **Two-pass, low-RAM** assembly: build a `.rete` by **streaming** the input(s)
/// twice instead of holding every parsed quad in memory. `stream` is invoked
/// **twice** and MUST replay the exact same quads in the same order each time —
/// pass 1 observes every term into the dictionary; pass 2 encodes them to
/// id-triples. The raw string quads (every term an owned String, heavily
/// duplicated — by far the largest working set on a big graph) are **never
/// collected**, so peak RAM is bounded by the dictionary + id-triples + index
/// rather than the string-quad multiset. Output is **byte-identical** to
/// [`assemble_dataset_with_opts`] on the same quads (same dictionary, same
/// id-triples in file order, same downstream pipeline).
///
/// The metadata callback derives the Dataset Card from the built dictionary +
/// default-graph id-triples (resolving terms through the dictionary), since the
/// raw quads were never retained. `stream` propagates parse/IO errors.
pub fn assemble_dataset_streaming<S, M: IntoMetadata>(
    stream: S,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
) -> Result<(Vec<u8>, BuildStats), IngestError>
where
    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
{
    assemble_dataset_streaming_algo(
        stream,
        with_pyramid,
        with_text_index,
        type_override,
        PyramidAlgo::Louvain,
        metadata,
    )
}

/// Like [`assemble_dataset_streaming`], but selects the community [`PyramidAlgo`]
/// (the streaming, low-RAM build path for `rete build --pyramid-algo …`).
#[allow(clippy::too_many_arguments)]
pub fn assemble_dataset_streaming_algo<S, M: IntoMetadata>(
    stream: S,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    algo: PyramidAlgo,
    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
) -> Result<(Vec<u8>, BuildStats), IngestError>
where
    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
{
    assemble_dataset_streaming_with_perms(
        stream,
        with_pyramid,
        with_text_index,
        type_override,
        algo,
        crate::index::PermSet::ALL,
        metadata,
    )
}

/// Like [`assemble_dataset_streaming_algo`], but writes only the permutations in
/// `perms` — the two-pass low-RAM twin of [`assemble_dataset_with_perms`].
#[allow(clippy::too_many_arguments)]
pub fn assemble_dataset_streaming_with_perms<S, M: IntoMetadata>(
    mut stream: S,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    algo: PyramidAlgo,
    perms: crate::index::PermSet,
    metadata: impl FnOnce(&BuildStats, &Dictionary, &[(u32, u32, u32)]) -> M,
) -> Result<(Vec<u8>, BuildStats), IngestError>
where
    S: FnMut(&mut dyn FnMut(RawQuad)) -> Result<(), IngestError>,
{
    use std::collections::BTreeMap;

    // Pass 1: observe every term (the dictionary dedups; the string quads are
    // freed line by line and never collected).
    let mut db = DictionaryBuilder::new();
    stream(&mut |(s, p, o, _g)| db.observe(&s, &p, &o))?;
    let dict = db.build();

    // Pass 2: encode each quad to an id-triple, bucketing named graphs.
    let mut default_triples: Vec<(u32, u32, u32)> = Vec::new();
    let mut named: BTreeMap<String, Vec<(u32, u32, u32)>> = BTreeMap::new();
    stream(&mut |(s, p, o, g)| {
        let t = dict.encode(&s, &p, &o).expect("observed term");
        match g {
            None => default_triples.push(t),
            Some(graph) => named.entry(graph).or_default().push(t),
        }
    })?;

    let statements = default_triples.len() + named.values().map(Vec::len).sum::<usize>();
    let stats = BuildStats {
        statements,
        default_triples: default_triples.len(),
        named_graphs: named.len(),
        terms: dict.term_count() as usize,
        pyramid_levels: 0,
    };
    let pending = metadata(&stats, &dict, &default_triples);
    Ok(finish_assembly(
        dict,
        default_triples,
        named,
        with_pyramid,
        with_text_index,
        type_override,
        algo,
        perms,
        pending,
        stats,
    ))
}

/// The shared tail of every build path: from a finished dictionary + encoded
/// id-triples (default graph + named), build the community pyramid, the optional
/// full-text index, the permutation indexes, and serialize the file image. Takes
/// the id-triples **by value** so they are freed as the permutations consume them.
#[allow(clippy::too_many_arguments)]
fn finish_assembly<M: IntoMetadata>(
    dict: Dictionary,
    default_triples: Vec<(u32, u32, u32)>,
    named: std::collections::BTreeMap<String, Vec<(u32, u32, u32)>>,
    with_pyramid: bool,
    with_text_index: bool,
    type_override: Option<&str>,
    algo: PyramidAlgo,
    perms: crate::index::PermSet,
    pending: M,
    mut stats: BuildStats,
) -> (Vec<u8>, BuildStats) {
    let has_named = !named.is_empty();

    // The pyramid and the full-text index are built FIRST, while the dictionary and
    // the default id-triples are both still resident — both need the two together.
    let (meta, levels) = if with_pyramid {
        build_pyramid_meta_algo(
            &dict,
            &default_triples,
            DEFAULT_TILE_BUDGET,
            type_override,
            algo,
        )
    } else {
        (Vec::new(), 0)
    };
    stats.pyramid_levels = levels;

    let text_index = if with_text_index {
        crate::file::compute_text_index(&dict, &default_triples)
    } else {
        Vec::new()
    };

    // From here the dictionary is only needed for its own serialized bytes. Encode
    // it, capture the header term count, then DROP it before building the
    // permutation indexes — which work purely on id-triples and never touch the
    // dictionary. On a large graph this frees the single biggest resident structure
    // (the dictionary) right when the index sort needs the headroom. The output is
    // byte-for-byte identical to serializing the dictionary inline.
    let codec = crate::file::writer_codec();
    let dict_container = crate::file::encode_dict_container(&dict, codec);
    let term_count = dict.term_count() as u64;
    let has_quoted_triples = dict.has_quoted_triples();
    drop(dict);

    // Build each graph's six permutations one-at-a-time on a large graph (a single
    // permuted copy resident at a time, each sort still parallel) instead of all
    // six concurrently; below the threshold the faster all-permutations-parallel
    // build is used (its 6x transient copy is negligible there). `build_seq` is
    // byte-identical to `build`.
    let build_index = |triples: Vec<(u32, u32, u32)>| -> crate::GraphIndex {
        let n = triples.len();
        let b = GraphIndexBuilder::from_triples(triples).with_perms(perms);
        if n > LOWMEM_TRIPLE_THRESHOLD {
            b.build_seq()
        } else {
            b.build()
        }
    };
    let def = build_index(default_triples);
    let named_indexes: Vec<(String, crate::GraphIndex)> = named
        .into_iter()
        .map(|(g, ts)| (g, build_index(ts)))
        .collect();

    // Only now is the DEDUPLICATED size of the file known: every index sorts and
    // dedups, so an input containing duplicate statements (a harvest that pages
    // with overlapping windows, a merge of overlapping dumps) writes fewer quads
    // than it ingested. `write_dataset_from_parts` stamps exactly this sum into
    // the header, so a metadata payload finalized with it cannot disagree with
    // the header — which is what an embedded Dataset Card used to do.
    let counts = FinalCounts {
        default_triples: def.triple_count() as u64,
        quads: def.triple_count() as u64
            + named_indexes
                .iter()
                .map(|(_, ix)| ix.triple_count() as u64)
                .sum::<u64>(),
    };
    let blob = pending.into_metadata(counts);
    // Report what was WRITTEN, not what was read: the returned stats drive the
    // CLI's "wrote …: N quads" line, which described the file, not the input.
    stats.statements = counts.quads as usize;
    stats.default_triples = counts.default_triples as usize;

    let bytes = crate::file::write_dataset_from_parts(
        &dict_container,
        term_count,
        &def,
        &named_indexes,
        has_named,
        has_quoted_triples,
        &meta,
        levels,
        &blob,
        &text_index,
        codec,
    );
    (bytes, stats)
}

/// Above this default-graph triple count, the index is built one permutation at a
/// time (low peak RAM) instead of all six in parallel. Chosen so typical builds
/// keep the faster parallel path while multi-hundred-million-triple graphs stay
/// within a bounded memory budget.
const LOWMEM_TRIPLE_THRESHOLD: usize = 30_000_000;

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

    #[test]
    fn parses_iris_bnodes_literals() {
        let input = r#"
            # a comment
            <http://ex/Alice> <http://ex/knows> <http://ex/Bob> .
            <http://ex/Alice> <http://ex/age> "30"^^<http://www.w3.org/2001/XMLSchema#integer> .
            <http://ex/Bob> <http://ex/label> "Bob"@en .
            _:b0 <http://ex/p> "plain" .
        "#;
        let t = parse(input).unwrap();
        assert_eq!(t.len(), 4);
        assert_eq!(t[0].0, "<http://ex/Alice>");
        assert_eq!(t[0].2, "<http://ex/Bob>");
        assert_eq!(t[1].2, "\"30\"^^<http://www.w3.org/2001/XMLSchema#integer>");
        assert_eq!(t[2].2, "\"Bob\"@en");
        assert_eq!(t[3].0, "_:b0");
        assert_eq!(t[3].2, "\"plain\"");
    }

    #[test]
    fn literal_with_spaces_and_escaped_quote() {
        let input = r#"<http://ex/s> <http://ex/p> "a \"quoted\" phrase here" ."#;
        let t = parse(input).unwrap();
        assert_eq!(t.len(), 1);
        assert_eq!(t[0].2, r#""a \"quoted\" phrase here""#);
    }

    /// RDF-star: a quoted triple whose object is a **language-tagged literal**
    /// sits directly against the closing `>>` with no separating whitespace
    /// (`"name"@fr>>`). The language-tag scan must stop at `>` — a regression
    /// guard for the greedy scan-to-whitespace that swallowed `@fr>>` as the tag.
    #[test]
    fn quoted_triple_langtagged_object() {
        let s = r#"<<<http://ex/sp> <http://ex/name> "Hirondelle rustique"@fr>>"#;
        let (tok, rest) = take_term(s).unwrap();
        assert_eq!(
            tok,
            r#"<<<http://ex/sp> <http://ex/name> "Hirondelle rustique"@fr>>"#
        );
        assert_eq!(rest, "");
        // and it round-trips through a full annotation line
        let line = r#"<<<http://ex/sp> <http://ex/name> "Oreneta vulgar"@ca>> <http://purl.org/dc/terms/source> "Catalogue of Life" ."#;
        let t = parse(line).unwrap();
        assert_eq!(t.len(), 1);
        assert_eq!(
            t[0].0,
            r#"<<<http://ex/sp> <http://ex/name> "Oreneta vulgar"@ca>>"#
        );
        assert_eq!(t[0].2, r#""Catalogue of Life""#);
        // a plain lang-tagged object still terminates at whitespace
        assert_eq!(take_term(r#""x"@pt-BR ."#).unwrap().0, r#""x"@pt-BR"#);
    }

    /// RDF 1.2 triple-term surface `<<( s p o )>>` parses to the SAME canonical
    /// token as the RDF-star `<< s p o >>`, so the two are interoperable (a query
    /// written either way matches data ingested either way).
    #[test]
    fn rdf12_triple_term_same_token_as_rdf_star() {
        let rdf12 = r#"<http://ex/r> <http://ex/p> <<( <http://ex/s> <http://ex/q> "o" )>> ."#;
        let star = r#"<http://ex/r> <http://ex/p> << <http://ex/s> <http://ex/q> "o" >> ."#;
        let a = parse(rdf12).unwrap();
        let b = parse(star).unwrap();
        assert_eq!(a.len(), 1);
        assert_eq!(a[0].2, r#"<<<http://ex/s> <http://ex/q> "o">>"#);
        assert_eq!(
            a[0].2, b[0].2,
            "RDF 1.2 and RDF-star must dedupe to one token"
        );
        // take_term consumes exactly the triple term (nothing left dangling).
        let (tok, rest) =
            take_term(r#"<<( <http://ex/s> <http://ex/q> <http://ex/o> )>>"#).unwrap();
        assert_eq!(tok, "<<<http://ex/s> <http://ex/q> <http://ex/o>>>");
        assert_eq!(rest, "");
    }

    #[test]
    fn rejects_missing_dot() {
        assert!(parse("<a> <b> <c>").is_err());
    }

    #[test]
    fn parses_quads_with_and_without_graph() {
        let input = "<http://ex/a> <http://ex/p> <http://ex/b> .\n\
                     <http://ex/a> <http://ex/p> <http://ex/c> <http://ex/g> .";
        let q = parse_quads(input).unwrap();
        assert_eq!(q.len(), 2);
        assert_eq!(q[0].3, None); // default graph
        assert_eq!(q[1].3, Some("<http://ex/g>".to_string()));
    }

    #[test]
    fn parse_reader_matches_text_parse() {
        // The streaming reader must produce exactly the same quads as parsing the
        // whole text — including blank/comment skipping, a named graph, and CRLF.
        let nt = "<http://ex/a> <http://ex/p> <http://ex/b> .\r\n\
                  # comment\n\
                  \n\
                  _:b0 <http://ex/q> \"lit\"@en .\n";
        let via_text = parse_statements(nt, "nt").unwrap();
        let via_reader = parse_reader(std::io::Cursor::new(nt), "nt", 0).unwrap();
        assert_eq!(via_text, via_reader);

        let nq = "<http://ex/a> <http://ex/p> <http://ex/b> .\n\
                  <http://ex/a> <http://ex/p> <http://ex/c> <http://ex/g> .\n";
        assert_eq!(
            parse_quads(nq).unwrap(),
            parse_reader(std::io::Cursor::new(nq), "nq", 0).unwrap()
        );
        // Turtle and TriG stream too — that is what lets the external build take a
        // multi-gigabyte `.ttl.gz` without expanding it to N-Triples first. Reading
        // a structured syntax from a reader must agree with parsing its whole text.
        let ttl = "@prefix ex: <http://ex/> .\nex:A ex:knows ex:B , ex:C .\n";
        assert_eq!(
            parse_statements(ttl, "ttl").unwrap(),
            parse_reader(std::io::Cursor::new(ttl), "ttl", 0).unwrap()
        );
        let trig = "@prefix ex: <http://ex/> .\nex:g { ex:A ex:knows ex:B . }\n";
        let streamed = parse_reader(std::io::Cursor::new(trig), "trig", 0).unwrap();
        assert_eq!(parse_statements(trig, "trig").unwrap(), streamed);
        assert_eq!(streamed[0].3.as_deref(), Some("<http://ex/g>"));
        // An unknown format is still a hard error, not a silent guess.
        assert!(parse_reader(std::io::Cursor::new(nt), "jsonld", 0).is_err());
    }

    #[test]
    fn parse_statements_dispatches_by_format() {
        let nt = "<http://ex/a> <http://ex/p> <http://ex/b> .";
        assert_eq!(parse_statements(nt, "nt").unwrap().len(), 1);
        let ttl = "@prefix ex: <http://ex/> .\nex:A ex:knows ex:B , ex:C .";
        assert_eq!(parse_statements(ttl, "ttl").unwrap().len(), 2);
        // N-Triples is a subset of both Turtle and TriG, so the same text parses
        // under all three — and lands in the default graph under each.
        assert_eq!(parse_statements(nt, "trig").unwrap().len(), 1);
        assert!(parse_statements(nt, "trig").unwrap()[0].3.is_none());
        assert!(parse_statements(nt, "jsonld").is_err());
    }

    /// A TriG graph block carries its graph name through as the same canonical
    /// token N-Quads would produce — so the two syntaxes agree on dictionary keys
    /// and a `.trig` shard federates with a `.nq` one.
    #[test]
    fn trig_graph_names_match_nquads() {
        let trig = "@prefix ex: <http://ex/> .\nex:g { ex:a ex:p ex:b . }\nex:a ex:p ex:c .\n";
        let nq = "<http://ex/a> <http://ex/p> <http://ex/b> <http://ex/g> .\n\
                  <http://ex/a> <http://ex/p> <http://ex/c> .\n";
        let mut from_trig = parse_statements(trig, "trig").unwrap();
        let mut from_nq = parse_statements(nq, "nq").unwrap();
        from_trig.sort();
        from_nq.sort();
        assert_eq!(from_trig, from_nq);
    }

    /// RDF/XML (how most OWL ontologies ship) parses to the same canonical tokens,
    /// including the abbreviated typed-node syntax and `rdf:resource` references.
    #[test]
    fn parses_rdfxml_owl() {
        let xml = r#"<?xml version="1.0"?>
            <rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
                     xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
                     xmlns:owl="http://www.w3.org/2002/07/owl#">
              <owl:Class rdf:about="http://ex/Dog">
                <rdfs:subClassOf rdf:resource="http://ex/Animal"/>
                <rdfs:label>Dog</rdfs:label>
              </owl:Class>
            </rdf:RDF>"#;
        let triples = parse_rdfxml(xml).unwrap();
        // rdf:type owl:Class, rdfs:subClassOf, rdfs:label = 3 triples.
        assert_eq!(triples.len(), 3);
        assert!(triples.iter().any(|(s, p, o)| s == "<http://ex/Dog>"
            && p == "<http://www.w3.org/2000/01/rdf-schema#subClassOf>"
            && o == "<http://ex/Animal>"));
        // Same data through the format dispatcher (triples → default graph).
        assert_eq!(parse_statements(xml, "rdfxml").unwrap().len(), 3);
        // Malformed XML is a clear RdfXml error, not a silent empty parse.
        assert!(matches!(
            parse_statements("<rdf:RDF><not closed", "rdfxml"),
            Err(IngestError::RdfXml(_))
        ));
    }

    /// Regression coverage for RUSTSEC-2026-0195: namespace declarations from
    /// an untrusted RDF/XML document must not make parsing panic or exhaust the
    /// bounded test process.
    #[test]
    fn rdfxml_namespace_fanout_is_bounded() {
        let declarations = (0..2_000)
            .map(|i| format!(r#" xmlns:p{i}="http://example.test/{i}/""#))
            .collect::<String>();
        let xml = format!(
            r#"<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"{declarations}/>"#
        );
        let result = parse_statements(&xml, "rdfxml");
        assert!(result.is_ok() || result.unwrap_err().to_string().contains("namespace"));
    }

    /// Regression coverage for RUSTSEC-2026-0194: duplicate-name checking must
    /// complete for a large valid start tag without panicking or exhausting the
    /// bounded test process.
    #[test]
    fn rdfxml_attribute_fanout_completes() {
        let attributes = (0..2_000)
            .map(|i| format!(r#" p:a{i}="{i}""#))
            .collect::<String>();
        let xml = format!(
            r#"<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#" xmlns:p="http://example.test/"><rdf:Description rdf:about="http://example.test/s"{attributes}/></rdf:RDF>"#
        );
        assert!(parse_statements(&xml, "rdfxml").is_ok());
    }

    /// Text in, queryable file image out — the whole in-memory build path the
    /// wasm `build()` binding uses, including a named graph.
    #[test]
    fn assemble_dataset_round_trips() {
        let text = "<http://ex/a> <http://ex/knows> <http://ex/b> .\n\
                    <http://ex/b> <http://ex/knows> <http://ex/c> .\n\
                    <http://ex/a> <http://ex/age> \"30\"^^<http://www.w3.org/2001/XMLSchema#integer> .\n\
                    <http://ex/a> <http://ex/p> <http://ex/d> <http://ex/g1> .";
        let quads = parse_statements(text, "nq").unwrap();
        let (bytes, stats) = assemble_dataset(quads, &[]);
        assert_eq!(stats.statements, 4);
        assert_eq!(stats.default_triples, 3);
        assert_eq!(stats.named_graphs, 1);
        assert!(stats.terms >= 7);

        let rete = Rete::open(&bytes).unwrap();
        assert_eq!(
            rete.query(None, Some("<http://ex/knows>"), None).len(),
            2,
            "default-graph pattern query"
        );
        assert_eq!(rete.graph_names(), vec!["<http://ex/g1>"]);
        let out = crate::eval_query(
            &rete,
            "SELECT ?x WHERE { ?x <http://ex/knows> ?y . ?y <http://ex/knows> ?z }",
        )
        .unwrap();
        match out {
            crate::QueryOutput::Select(_, rows) => assert_eq!(rows.len(), 1),
            other => panic!("expected select result, got {other:?}"),
        }
    }

    /// The exact minimal input the playground Build tab uses: two triples, no
    /// literals, no rdf:type, no named graph — the smallest graph that still
    /// builds a community pyramid. (The wasm `build()` panic this guards was a
    /// `std::time::Instant::now()` in the pyramid timing path; native has a clock
    /// so this passes here, while the playground harness exercises the wasm path.)
    #[test]
    fn assemble_minimal_typeless_graph() {
        let text = "<http://ex/A> <http://ex/knows> <http://ex/B> .\n\
                    <http://ex/B> <http://ex/knows> <http://ex/C> .\n";
        let quads = parse_statements(text, "nt").unwrap();
        let (bytes, stats) = assemble_dataset(quads, &[]);
        assert_eq!(stats.default_triples, 2);
        let rete = Rete::open(&bytes).unwrap();
        assert_eq!(rete.query(None, Some("<http://ex/knows>"), None).len(), 2);
    }
}