rete-cli 0.4.2

//! `rete serve` — a minimal **SPARQL 1.1 Protocol endpoint over one `.rete`**
//! (or over a **manifest** of segments, see `commands::manifest`), including
//! SPARQL **Update**.
//!
//! The base file is **never mutated** (it stays a content-hashed, CDN-servable
//! artifact). Updates append to a plain-text **journal** next to it
//! (`<file>.changes`: one `+`/`-`-prefixed N-Quads line per change) and mutate
//! an in-memory quad set; the queryable image is rebuilt from that set lazily
//! (dirty flag), so update bursts coalesce into one rebuild. Restart = base +
//! journal replay. `GET /snapshot.rete` serializes the *current* state as a
//! fresh `.rete` — the downloadable companion; publishing an update cycle is
//! "upload the snapshot, delete the journal" — or, when serving a manifest,
//! `rete manifest seal` turns the journal into fresh segments incrementally.
//!
//! This makes the write path a small, single-writer authority while every
//! *read* path stays serverless — and closes the federation loop: any rete
//! client (CLI, browser) can `SERVICE <http://host:port/sparql>` against it.
//!
//! Scale note: until the first update the endpoint answers straight from the
//! base file, opened like every local command opens it (read whole only when
//! it is a small fraction of the free memory, lazily otherwise), so serving a
//! multi-GB catalog file read-only costs what its queries touch. The first
//! update (or a pending journal at startup) materializes the whole graph in
//! memory and rebuilds it on write — right for the living-dataset use case
//! (annotation, curation, small/medium KGs, up to a few million triples), not
//! for the multi-GB catalog files (their story remains offline
//! rebuild/compaction; an engine-level overlay is future work). An update that
//! could not fit in the memory available is refused with `503` rather than
//! letting the server be killed.

use std::collections::BTreeSet;
use std::io::Write as _;

use rete_core::ingest::{assemble_dataset_with_opts, parse_quads, RawQuad};
use rete_core::{
    eval_query_with_warnings, sparql_json_ask, sparql_json_results, QueryOpts, QueryOutput,
    QueryWarning, Rete,
};
use spargebra::term::{
    GraphName, GraphNamePattern, GroundQuadPattern, GroundTermPattern, NamedNodePattern,
    QuadPattern, TermPattern,
};
use spargebra::{GraphUpdateOperation, Query, Update};

use super::service_http::HttpServiceClient;
use crate::commands::range_source::{available_memory, open_local};

/// One quad as term tokens; `None` graph = the default graph. `BTreeSet` gives
/// set semantics (RDF graphs are sets) and a deterministic rebuild order.
type QuadSet = BTreeSet<RawQuad>;

/// The server's whole state: the quad set (the truth), the current queryable
/// image built from it, and the journal that makes the set durable.
///
/// Until the first update there is no quad set: `rete` is the base file
/// itself, `quads` and `image` are `None`, and `/snapshot.rete` streams the
/// base file from disk.
struct Store {
    /// `None` while the server still answers from the unmodified base file.
    quads: Option<QuadSet>,
    /// The serialized `.rete` of the current state (served by
    /// `/snapshot.rete`); `None` while that state is the base file.
    image: Option<Vec<u8>>,
    rete: Rete,
    /// The base `.rete` (not a manifest) the server answers from until the
    /// first update.
    base: Option<std::path::PathBuf>,
    /// Set by updates; the next read rebuilds `image`/`rete` from `quads`.
    dirty: bool,
    journal: std::path::PathBuf,
    /// Counter for minting fresh blank nodes in INSERT templates.
    bnode_seq: u64,
}

/// What materializing a served graph costs per quad at its peak: the quad set
/// of owned term strings plus the rebuild of an image from it. Measured: 646
/// MiB for 1 M OpenCitations quads (127 B per N-Triples line); 1 KiB leaves
/// room for longer terms. Only used to refuse an update that could not fit,
/// never to size anything, and an estimate: a graph of very long literals
/// costs more.
const MATERIALIZE_BYTES_PER_QUAD: u64 = 1024;

/// Why an update could not be taken: the request was bad (`400`), or the
/// server cannot hold the graph it would have to materialize (`503`).
enum UpdateError {
    Bad(anyhow::Error),
    TooLarge(String),
}

impl From<anyhow::Error> for UpdateError {
    fn from(e: anyhow::Error) -> Self {
        UpdateError::Bad(e)
    }
}

impl From<std::io::Error> for UpdateError {
    fn from(e: std::io::Error) -> Self {
        UpdateError::Bad(e.into())
    }
}

/// Can `quads` statements be materialized in `available` bytes? `Err` carries
/// the refusal the client gets. Unknown memory (not Linux) never refuses.
fn materialize_fits(quads: u64, available: Option<u64>) -> Result<(), String> {
    let need = quads.saturating_mul(MATERIALIZE_BYTES_PER_QUAD);
    match available {
        Some(avail) if need > avail => Err(format!(
            "this update would load all {quads} quads into memory (about {} MiB) and only \
             {} MiB is available; the endpoint stays read-only for this file. Apply the \
             change offline (rete export, edit, rete build) or serve a smaller file",
            need >> 20,
            avail >> 20
        )),
        _ => Ok(()),
    }
}

/// Every quad of an opened `.rete`, default graph and named graphs.
fn all_quads(rete: &Rete) -> QuadSet {
    let mut quads: QuadSet = rete
        .dump(None)
        .into_iter()
        .map(|(s, p, o)| (s, p, o, None))
        .collect();
    for i in 0..rete.named_graph_count() {
        let Some(g) = rete.named_graph_name_at(i).map(str::to_string) else {
            continue;
        };
        for (s, p, o) in rete.dump(Some(&g)) {
            quads.insert((s, p, o, Some(g.clone())));
        }
    }
    quads
}

/// Render a quad back to one N-Quads line (terms are already canonical tokens).
fn nq_line(q: &RawQuad) -> String {
    match &q.3 {
        Some(g) => format!("{} {} {} {} .", q.0, q.1, q.2, g),
        None => format!("{} {} {} .", q.0, q.1, q.2),
    }
}

/// Assemble a queryable image from a quad set. Pyramid off: the endpoint
/// serves SPARQL, and skipping it makes the after-write rebuild fast.
fn build_image(quads: &QuadSet) -> anyhow::Result<(Vec<u8>, Rete)> {
    let quads: Vec<RawQuad> = quads.iter().cloned().collect();
    let (image, _stats) = assemble_dataset_with_opts(quads, false, false, None, |_, _| Vec::new());
    let mut rete = Rete::open(&image)?;
    // The endpoint itself supports SERVICE blocks in incoming queries.
    rete.set_service_client(Box::new(HttpServiceClient));
    Ok((image, rete))
}

/// Parse one journal payload (an N-Quads line) back to a quad.
fn parse_nq_line(line: &str) -> anyhow::Result<RawQuad> {
    let mut quads = parse_quads(line)?;
    match quads.len() {
        1 => Ok(quads.pop().expect("len checked")),
        n => anyhow::bail!("journal line held {n} statements: {line}"),
    }
}

impl Store {
    /// Open the source — one `.rete` file, or a **manifest** (`.json`, see
    /// `commands::manifest`) whose visible fold becomes the initial state.
    ///
    /// A plain `.rete` with no pending journal is served as it is: opened with
    /// [`open_local`]'s memory-aware rule, no quad set, no rebuild. A manifest,
    /// or a file with journal lines to replay, is materialized: every quad
    /// extracted, the journal replayed, the first image built.
    fn open(base: &str, journal: std::path::PathBuf) -> anyhow::Result<Store> {
        let is_manifest = super::manifest::is_manifest_path(base);
        let pending = match std::fs::read_to_string(&journal) {
            Ok(j) => j.lines().any(|l| !l.trim().is_empty()),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
            Err(e) => anyhow::bail!("{}: {e}", journal.display()),
        };
        if !is_manifest && !pending {
            let mut rete = open_local(base)?;
            rete.set_service_client(Box::new(HttpServiceClient));
            return Ok(Store {
                quads: None,
                image: None,
                rete,
                base: Some(std::path::PathBuf::from(base)),
                dirty: false,
                journal,
                bnode_seq: 0,
            });
        }
        let mut quads: QuadSet = if is_manifest {
            super::manifest::visible_quads(base)?
        } else {
            all_quads(&open_local(base)?)
        };

        // Replay the journal (if any): each line is `+ <nq>` or `- <nq>`.
        let mut replayed = 0usize;
        if journal.exists() {
            for (i, line) in std::fs::read_to_string(&journal)?.lines().enumerate() {
                let line = line.trim();
                if line.is_empty() {
                    continue;
                }
                let (op, payload) = line.split_at(1);
                let quad = parse_nq_line(payload.trim_start())
                    .map_err(|e| anyhow::anyhow!("{}:{}: {e}", journal.display(), i + 1))?;
                match op {
                    "+" => {
                        quads.insert(quad);
                    }
                    "-" => {
                        quads.remove(&quad);
                    }
                    other => anyhow::bail!(
                        "{}:{}: unknown journal op {other:?} (expected + or -)",
                        journal.display(),
                        i + 1
                    ),
                }
                replayed += 1;
            }
        }

        // The base file is already dropped (its quads are in `quads`), so the
        // placeholder is built from an empty image — cheap, and always valid.
        // `ensure_current` below replaces it with the merged one.
        let (empty_image, _) =
            assemble_dataset_with_opts(Vec::new(), false, false, None, |_, _| Vec::new());
        let mut store = Store {
            quads: Some(quads),
            image: None,
            rete: Rete::open(&empty_image)?,
            base: None,
            dirty: true,
            journal,
            bnode_seq: 0,
        };
        store.ensure_current()?;
        if replayed > 0 {
            eprintln!("  replayed {replayed} journal change(s)");
        }
        Ok(store)
    }

    /// How many quads the served state holds.
    fn quad_count(&self) -> u64 {
        match &self.quads {
            Some(q) => q.len() as u64,
            None => self.rete.header().quad_count,
        }
    }

    /// Rebuild the queryable image from the quad set if updates are pending.
    fn ensure_current(&mut self) -> anyhow::Result<()> {
        if !self.dirty {
            return Ok(());
        }
        let Some(quads) = &self.quads else {
            return Ok(()); // still the base file: nothing to rebuild
        };
        let (image, rete) = build_image(quads)?;
        self.image = Some(image);
        self.rete = rete;
        self.dirty = false;
        Ok(())
    }

    /// The first update turns the served base file into an in-memory quad
    /// set. It is refused, before anything is read, when its estimated peak
    /// would not fit in the memory still available.
    fn materialize(&mut self) -> Result<(), UpdateError> {
        if self.quads.is_some() {
            return Ok(());
        }
        let n = self.rete.header().quad_count;
        materialize_fits(n, available_memory()).map_err(UpdateError::TooLarge)?;
        eprintln!("  first update: loading all {n} quads into memory");
        self.quads = Some(all_quads(&self.rete));
        Ok(())
    }

    /// Apply one change to the set; journal it only if it actually changed
    /// something (set semantics: double-inserts and misses are no-ops).
    fn apply(&mut self, add: bool, quad: RawQuad, journal: &mut Vec<String>) {
        let quads = self.quads.as_mut().expect("materialized before any change");
        let changed = if add {
            quads.insert(quad.clone())
        } else {
            quads.remove(&quad)
        };
        if changed {
            journal.push(format!(
                "{} {}",
                if add { "+" } else { "-" },
                nq_line(&quad)
            ));
            self.dirty = true;
        }
    }

    /// Execute a SPARQL UPDATE request: operations run in order, each seeing
    /// the previous one's effects; every net change is journaled. Returns the
    /// number of changes applied.
    fn update(&mut self, text: &str) -> Result<usize, UpdateError> {
        let parsed = Update::parse(text, None).map_err(|e| anyhow::anyhow!("parse: {e}"))?;
        self.materialize()?;
        let mut journal: Vec<String> = Vec::new();
        for op in &parsed.operations {
            self.run_op(op, &mut journal)?;
        }
        if !journal.is_empty() {
            let mut f = std::fs::OpenOptions::new()
                .create(true)
                .append(true)
                .open(&self.journal)?;
            for line in &journal {
                writeln!(f, "{line}")?;
            }
            f.sync_all()?;
        }
        Ok(journal.len())
    }

    fn run_op(
        &mut self,
        op: &GraphUpdateOperation,
        journal: &mut Vec<String>,
    ) -> anyhow::Result<()> {
        match op {
            GraphUpdateOperation::InsertData { data } => {
                for q in data {
                    let quad = (
                        q.subject.to_string(),
                        q.predicate.to_string(),
                        q.object.to_string(),
                        graph_token(&q.graph_name),
                    );
                    self.apply(true, quad, journal);
                }
                Ok(())
            }
            GraphUpdateOperation::DeleteData { data } => {
                for q in data {
                    let quad = (
                        q.subject.to_string(),
                        q.predicate.to_string(),
                        q.object.to_string(),
                        graph_token(&q.graph_name),
                    );
                    self.apply(false, quad, journal);
                }
                Ok(())
            }
            GraphUpdateOperation::DeleteInsert {
                delete,
                insert,
                using,
                pattern,
            } => {
                if using.is_some() {
                    anyhow::bail!("USING is not supported");
                }
                // The WHERE pattern evaluates against the current state
                // (including earlier operations of this request).
                self.ensure_current()?;
                let query = Query::Select {
                    dataset: None,
                    pattern: (**pattern).clone(),
                    base_iri: None,
                }
                .to_string();
                let (_vars, solutions) = rete_core::eval_sparql(&self.rete, &query)
                    .map_err(|e| anyhow::anyhow!("WHERE evaluation: {e}"))?;
                // Deletes for every solution first, then inserts (per the spec).
                for sol in &solutions {
                    for tpl in delete {
                        if let Some(quad) = instantiate_ground(tpl, sol) {
                            self.apply(false, quad, journal);
                        }
                    }
                }
                for sol in &solutions {
                    // Blank nodes in INSERT templates mint fresh nodes per
                    // solution (never reuse a label from the data).
                    let mut minted: std::collections::HashMap<String, String> =
                        std::collections::HashMap::new();
                    for tpl in insert {
                        if let Some(quad) = self.instantiate(tpl, sol, &mut minted) {
                            self.apply(true, quad, journal);
                        }
                    }
                }
                Ok(())
            }
            GraphUpdateOperation::Clear { graph, .. }
            | GraphUpdateOperation::Drop { graph, .. } => {
                use spargebra::algebra::GraphTarget;
                let doomed: Vec<RawQuad> = self
                    .quads
                    .iter()
                    .flatten()
                    .filter(|q| match graph {
                        GraphTarget::DefaultGraph => q.3.is_none(),
                        GraphTarget::NamedNode(n) => q.3.as_deref() == Some(n.to_string().as_str()),
                        GraphTarget::NamedGraphs => q.3.is_some(),
                        GraphTarget::AllGraphs => true,
                    })
                    .cloned()
                    .collect();
                for quad in doomed {
                    self.apply(false, quad, journal);
                }
                Ok(())
            }
            // Graphs exist implicitly (a graph is its quads), so CREATE is a
            // no-op that succeeds.
            GraphUpdateOperation::Create { .. } => Ok(()),
            GraphUpdateOperation::Load { .. } => {
                anyhow::bail!("LOAD is not supported (ingest with `rete build`)")
            }
        }
    }

    /// Instantiate an INSERT template: like the ground case, plus blank nodes
    /// mint one fresh node per (solution, label).
    fn instantiate(
        &mut self,
        tpl: &QuadPattern,
        sol: &rete_core::Binding,
        minted: &mut std::collections::HashMap<String, String>,
    ) -> Option<RawQuad> {
        let mut term = |t: &TermPattern| -> Option<String> {
            match t {
                TermPattern::NamedNode(n) => Some(n.to_string()),
                TermPattern::Literal(l) => Some(l.to_string()),
                TermPattern::Variable(v) => sol.get(v.as_str()).cloned(),
                TermPattern::BlankNode(b) => Some(
                    minted
                        .entry(b.to_string())
                        .or_insert_with(|| {
                            self.bnode_seq += 1;
                            format!("_:m{}", self.bnode_seq)
                        })
                        .clone(),
                ),
                // RDF-star quoted triple in a template: resolve recursively to the
                // canonical `<< s p o >>` token (blank nodes inside a quoted triple
                // are not minted — skip if present).
                TermPattern::Triple(_) => quoted_term_token(t, sol),
            }
        };
        Some((
            term(&tpl.subject)?,
            named_term(&tpl.predicate, sol)?,
            term(&tpl.object)?,
            graph_pattern_token(&tpl.graph_name, sol)?,
        ))
    }
}

fn graph_token(g: &GraphName) -> Option<String> {
    match g {
        GraphName::DefaultGraph => None,
        GraphName::NamedNode(n) => Some(n.to_string()),
    }
}

/// Instantiate a ground (DELETE) template against one solution; `None` (skip
/// the quad) when a template variable is unbound in this solution.
fn instantiate_ground(tpl: &GroundQuadPattern, sol: &rete_core::Binding) -> Option<RawQuad> {
    Some((
        ground_term(&tpl.subject, sol)?,
        named_term(&tpl.predicate, sol)?,
        ground_term(&tpl.object, sol)?,
        graph_pattern_token(&tpl.graph_name, sol)?,
    ))
}

/// Resolve a template graph name; `Some(None)` = the default graph, outer
/// `None` = an unbound variable (skip the quad).
fn graph_pattern_token(g: &GraphNamePattern, sol: &rete_core::Binding) -> Option<Option<String>> {
    match g {
        GraphNamePattern::DefaultGraph => Some(None),
        GraphNamePattern::NamedNode(n) => Some(Some(n.to_string())),
        GraphNamePattern::Variable(v) => sol.get(v.as_str()).cloned().map(Some),
    }
}

fn ground_term(t: &GroundTermPattern, sol: &rete_core::Binding) -> Option<String> {
    match t {
        GroundTermPattern::NamedNode(n) => Some(n.to_string()),
        GroundTermPattern::Literal(l) => Some(l.to_string()),
        GroundTermPattern::Variable(v) => sol.get(v.as_str()).cloned(),
        // RDF-star quoted triple in a ground (DELETE) template.
        GroundTermPattern::Triple(tp) => {
            let s = ground_term(&tp.subject, sol)?;
            let p = named_term(&tp.predicate, sol)?;
            let o = ground_term(&tp.object, sol)?;
            Some(format!("<<{s} {p} {o}>>"))
        }
    }
}

/// Resolve an RDF-star quoted triple in a CONSTRUCT/INSERT template against a
/// solution → the canonical `<< s p o >>` token. Recurses for nesting; a blank
/// node inside a quoted triple yields `None` (not minted).
fn quoted_term_token(t: &TermPattern, sol: &rete_core::Binding) -> Option<String> {
    match t {
        TermPattern::NamedNode(n) => Some(n.to_string()),
        TermPattern::Literal(l) => Some(l.to_string()),
        TermPattern::Variable(v) => sol.get(v.as_str()).cloned(),
        TermPattern::BlankNode(_) => None,
        TermPattern::Triple(tp) => {
            let s = quoted_term_token(&tp.subject, sol)?;
            let p = named_term(&tp.predicate, sol)?;
            let o = quoted_term_token(&tp.object, sol)?;
            Some(format!("<<{s} {p} {o}>>"))
        }
    }
}

fn named_term(t: &NamedNodePattern, sol: &rete_core::Binding) -> Option<String> {
    match t {
        NamedNodePattern::NamedNode(n) => Some(n.to_string()),
        NamedNodePattern::Variable(v) => sol.get(v.as_str()).cloned(),
    }
}

// --- HTTP ----------------------------------------------------------------------

/// Percent-decode one `application/x-www-form-urlencoded` value (`+` = space).
fn form_unescape(s: &str) -> String {
    let bytes = s.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        match bytes[i] {
            b'+' => out.push(b' '),
            b'%' if i + 2 < bytes.len() => {
                let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).ok();
                match hex.and_then(|h| u8::from_str_radix(h, 16).ok()) {
                    Some(b) => {
                        out.push(b);
                        i += 2;
                    }
                    None => out.push(b'%'),
                }
            }
            b => out.push(b),
        }
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

/// Split a query string / form body into decoded `(key, value)` pairs.
fn form_pairs(s: &str) -> Vec<(String, String)> {
    s.split('&')
        .filter_map(|kv| {
            let (k, v) = kv.split_once('=')?;
            Some((form_unescape(k), form_unescape(v)))
        })
        .collect()
}

fn param(pairs: &[(String, String)], key: &str) -> Option<String> {
    pairs.iter().find(|(k, _)| k == key).map(|(_, v)| v.clone())
}

/// What one HTTP request asks for, normalized across the SPARQL Protocol's
/// transports (GET query param, form body, raw `application/sparql-*` body).
enum Action {
    Query(String),
    Update(String),
    Snapshot,
    Info,
    Preflight,
    NotFound,
}

fn classify(req: &mut tiny_http::Request) -> anyhow::Result<Action> {
    use tiny_http::Method;
    let url = req.url().to_string();
    let (path, query_string) = match url.split_once('?') {
        Some((p, q)) => (p, Some(q.to_string())),
        None => (url.as_str(), None),
    };
    let path = path.trim_end_matches('/');
    let content_type = req
        .headers()
        .iter()
        .find(|h| h.field.equiv("Content-Type"))
        .map(|h| h.value.as_str().to_ascii_lowercase())
        .unwrap_or_default();

    Ok(match (req.method().clone(), path) {
        (Method::Options, _) => Action::Preflight,
        (Method::Get, "" | "/") => Action::Info,
        (Method::Get, "/snapshot.rete") => Action::Snapshot,
        (Method::Get, "/sparql") => match query_string.as_deref().map(form_pairs) {
            Some(pairs) => match param(&pairs, "query") {
                Some(q) => Action::Query(q),
                None => Action::Info, // a bare GET /sparql: describe the service
            },
            None => Action::Info,
        },
        (Method::Post, "/sparql" | "/update") => {
            let mut body = String::new();
            req.as_reader().read_to_string(&mut body)?;
            if content_type.starts_with("application/sparql-query") {
                Action::Query(body)
            } else if content_type.starts_with("application/sparql-update") {
                Action::Update(body)
            } else {
                // Form-encoded (the protocol's default POST transport).
                let pairs = form_pairs(&body);
                if let Some(u) = param(&pairs, "update") {
                    Action::Update(u)
                } else if let Some(q) = param(&pairs, "query") {
                    Action::Query(q)
                } else {
                    anyhow::bail!("POST body carries neither `query` nor `update`");
                }
            }
        }
        _ => Action::NotFound,
    })
}

/// The query's warnings (expression type errors a FILTER absorbed, see
/// `rete_core::eval_query_with_warnings`) travel in response HEADERS, not in
/// the body: the body stays the standard SPARQL result document byte for byte
/// — JSON results for SELECT/ASK, N-Triples for CONSTRUCT, which has no JSON
/// member to extend — so every SPARQL client parses it exactly as before, and
/// one mechanism covers every result form. `Rete-Warnings` is the JSON array
/// `rete sparql --json` prints (`warnings_json`), escaped to ASCII as a header
/// value must be; `Rete-Warning-Count` is the total. Both are absent when the
/// query raised no warning, so such a response is unchanged.
const WARNINGS_HEADER: &str = "Rete-Warnings";
const WARNING_COUNT_HEADER: &str = "Rete-Warning-Count";

/// Header budget for `Rete-Warnings`. Common servers and proxies cap a header
/// near 8 KiB; whole warnings are dropped from the end until the array fits,
/// and `Rete-Warning-Count` still says how many there were.
const WARNINGS_HEADER_MAX: usize = 6 * 1024;

/// JSON with every non-ASCII character (and DEL) as a `\uXXXX` escape — still
/// the same JSON value, now a legal HTTP header value.
fn ascii_json(json: &str) -> String {
    let mut out = String::with_capacity(json.len());
    for c in json.chars() {
        if c.is_ascii() && c != '\u{7f}' {
            out.push(c);
        } else {
            let mut units = [0u16; 2];
            for u in c.encode_utf16(&mut units) {
                out.push_str(&format!("\\u{u:04x}"));
            }
        }
    }
    out
}

/// The warning headers for a response, or none when there are no warnings.
fn warning_headers(warnings: &[QueryWarning]) -> Vec<(&'static str, String)> {
    if warnings.is_empty() {
        return Vec::new();
    }
    let mut kept = warnings.len();
    let json = loop {
        let json = ascii_json(&rete_core::warnings_json(&warnings[..kept]));
        if json.len() <= WARNINGS_HEADER_MAX || kept == 0 {
            break json;
        }
        kept -= 1;
    };
    vec![
        (WARNINGS_HEADER, json),
        (WARNING_COUNT_HEADER, warnings.len().to_string()),
    ]
}

/// Build a response with the CORS headers every browser client needs.
fn respond(
    req: tiny_http::Request,
    status: u16,
    content_type: &str,
    body: Vec<u8>,
) -> anyhow::Result<()> {
    respond_with(req, status, content_type, body, &[])
}

/// [`respond`] plus `extra` headers.
fn respond_with(
    req: tiny_http::Request,
    status: u16,
    content_type: &str,
    body: Vec<u8>,
    extra: &[(&str, String)],
) -> anyhow::Result<()> {
    send(
        req,
        tiny_http::Response::from_data(body).with_status_code(status),
        content_type,
        extra,
    )
}

/// Send `resp` with the CORS headers, the content type and `extra`.
fn send<R: std::io::Read>(
    req: tiny_http::Request,
    mut resp: tiny_http::Response<R>,
    content_type: &str,
    extra: &[(&str, String)],
) -> anyhow::Result<()> {
    let fixed = [
        ("Content-Type", content_type),
        ("Access-Control-Allow-Origin", "*"),
        ("Access-Control-Allow-Methods", "GET, POST, OPTIONS"),
        (
            "Access-Control-Allow-Headers",
            "Content-Type, Authorization",
        ),
        // A browser script may read only the response headers named here.
        (
            "Access-Control-Expose-Headers",
            "Rete-Warnings, Rete-Warning-Count",
        ),
    ];
    for (k, v) in fixed
        .into_iter()
        .chain(extra.iter().map(|(k, v)| (*k, v.as_str())))
    {
        resp.add_header(
            tiny_http::Header::from_bytes(k.as_bytes(), v.as_bytes())
                .map_err(|_| anyhow::anyhow!("bad header"))?,
        );
    }
    req.respond(resp)?;
    Ok(())
}

/// `", 2 warning(s)"` for the request log, or nothing.
fn warned(warnings: &[QueryWarning]) -> String {
    if warnings.is_empty() {
        String::new()
    } else {
        format!(", {} warning(s)", warnings.len())
    }
}

/// `rete serve`: run the endpoint until interrupted.
pub(crate) fn serve(
    file: &str,
    bind: &str,
    token: Option<&str>,
    journal: Option<&str>,
) -> anyhow::Result<()> {
    let journal_path = std::path::PathBuf::from(
        journal
            .map(String::from)
            .unwrap_or_else(|| format!("{file}.changes")),
    );
    let mut store = Store::open(file, journal_path.clone())?;
    let server = tiny_http::Server::http(bind).map_err(|e| anyhow::anyhow!("bind {bind}: {e}"))?;
    eprintln!(
        "serving {file} ({} quads{}) at http://{bind}/sparql\n  updates: POST query=/update= (SPARQL 1.1 Protocol){}\n  journal: {}\n  snapshot: GET /snapshot.rete",
        store.quad_count(),
        if store.quads.is_none() {
            ", answered from the file until the first update"
        } else {
            ""
        },
        if token.is_some() { " — Bearer token required" } else { "" },
        journal_path.display(),
    );

    let running = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
    let signal_running = running.clone();
    ctrlc::set_handler(move || {
        signal_running.store(false, std::sync::atomic::Ordering::SeqCst);
    })
    .map_err(|error| anyhow::anyhow!("installing Ctrl-C handler: {error}"))?;

    while running.load(std::sync::atomic::Ordering::SeqCst) {
        let Some(mut req) = server.recv_timeout(std::time::Duration::from_millis(100))? else {
            continue;
        };
        let action = match classify(&mut req) {
            Ok(a) => a,
            Err(e) => {
                let _ = respond(
                    req,
                    400,
                    "text/plain",
                    format!("bad request: {e}\n").into_bytes(),
                );
                continue;
            }
        };
        let result = match action {
            Action::Preflight => respond(req, 204, "text/plain", Vec::new()),
            Action::Info => {
                let body = format!(
                    "rete SPARQL endpoint\n  {} quads ({} pending journal file: {})\n  GET/POST /sparql   query= (SPARQL 1.1 Protocol; results JSON)\n  POST /sparql|/update  update= or application/sparql-update\n  GET /snapshot.rete    the current state as a .rete file\n",
                    store.quad_count(),
                    if store.dirty { "dirty;" } else { "clean;" },
                    store.journal.display(),
                );
                respond(req, 200, "text/plain; charset=utf-8", body.into_bytes())
            }
            Action::Snapshot => match store.ensure_current() {
                Ok(()) => match (&store.image, &store.base) {
                    (Some(image), _) => {
                        respond(req, 200, "application/octet-stream", image.clone())
                    }
                    // Unmodified: the base file is the current state, streamed
                    // from disk rather than read into memory.
                    (None, Some(base)) => match std::fs::File::open(base) {
                        Ok(f) => send(
                            req,
                            tiny_http::Response::from_file(f),
                            "application/octet-stream",
                            &[],
                        ),
                        Err(e) => respond(req, 500, "text/plain", format!("{e}\n").into_bytes()),
                    },
                    (None, None) => {
                        respond(req, 500, "text/plain", b"no snapshot available\n".to_vec())
                    }
                },
                Err(e) => respond(req, 500, "text/plain", format!("{e}\n").into_bytes()),
            },
            Action::Query(q) => match store.ensure_current() {
                Ok(()) => match eval_query_with_warnings(&store.rete, &q, QueryOpts::default()) {
                    Ok((QueryOutput::Select(vars, rows), warnings)) => {
                        eprintln!("  query: {} row(s){}", rows.len(), warned(&warnings));
                        respond_with(
                            req,
                            200,
                            "application/sparql-results+json",
                            sparql_json_results(&vars, &rows).into_bytes(),
                            &warning_headers(&warnings),
                        )
                    }
                    Ok((QueryOutput::Ask(b), warnings)) => respond_with(
                        req,
                        200,
                        "application/sparql-results+json",
                        sparql_json_ask(b).into_bytes(),
                        &warning_headers(&warnings),
                    ),
                    Ok((QueryOutput::Construct(triples), warnings)) => {
                        let mut body = String::new();
                        for (s, p, o) in &triples {
                            body.push_str(&format!("{s} {p} {o} .\n"));
                        }
                        respond_with(
                            req,
                            200,
                            "application/n-triples",
                            body.into_bytes(),
                            &warning_headers(&warnings),
                        )
                    }
                    Ok(_) => respond(
                        req,
                        501,
                        "text/plain",
                        b"query result kind is not supported by this server build\n".to_vec(),
                    ),
                    Err(e) => respond(req, 400, "text/plain", format!("{e}\n").into_bytes()),
                },
                Err(e) => respond(req, 500, "text/plain", format!("{e}\n").into_bytes()),
            },
            Action::Update(u) => {
                // Updates may be token-guarded; queries stay open.
                let authorized = match token {
                    None => true,
                    Some(t) => req
                        .headers()
                        .iter()
                        .find(|h| h.field.equiv("Authorization"))
                        .map(|h| h.value.as_str() == format!("Bearer {t}"))
                        .unwrap_or(false),
                };
                if !authorized {
                    respond(
                        req,
                        401,
                        "text/plain",
                        b"update requires Bearer token\n".to_vec(),
                    )
                } else {
                    match store.update(&u) {
                        Ok(n) => {
                            eprintln!("  update: {n} change(s)");
                            respond(req, 204, "text/plain", Vec::new())
                        }
                        Err(UpdateError::Bad(e)) => {
                            respond(req, 400, "text/plain", format!("{e}\n").into_bytes())
                        }
                        Err(UpdateError::TooLarge(msg)) => {
                            eprintln!("  update refused: {msg}");
                            respond(req, 503, "text/plain", format!("{msg}\n").into_bytes())
                        }
                    }
                }
            }
            Action::NotFound => respond(req, 404, "text/plain", b"not found\n".to_vec()),
        };
        if let Err(e) = result {
            eprintln!("  response error: {e}");
        }
    }
    Ok(())
}

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

    #[test]
    fn materializing_is_refused_only_when_it_cannot_fit() {
        let gib = 1u64 << 30;
        assert!(materialize_fits(1_000, Some(gib)).is_ok());
        // The OpenCitations sample (18.9 M quads) in a 4 GiB container.
        let err = materialize_fits(18_869_242, Some(3 * gib)).unwrap_err();
        assert!(err.contains("18869242 quads"), "{err}");
        assert!(err.contains("read-only"), "{err}");
        assert!(
            materialize_fits(u64::MAX, None).is_ok(),
            "unknown memory never refuses"
        );
    }
}