kglite-bolt-server 0.21.4

//! Bolt-layer query intercepts: the recognizers, column contracts, and
//! stream builders for the verbs `kglite-bolt-server` answers itself —
//! `db.checkpoint()`, the server-facts trio (`dbms.components()`,
//! `dbms.showCurrentUser()`, `SHOW DATABASES`) — plus the EXPLAIN-rows →
//! Bolt `summary.plan` conversion. Split from `backend.rs` (which keeps the
//! Backend impl and the verb *handlers* that need server state).

use super::*;

/// Output columns of `db.checkpoint()`, in declaration order — the shape
/// Neo4j's own `db.checkpoint()` yields, so a client can call it the same way.
pub(super) const CHECKPOINT_COLUMNS: [&str; 2] = ["success", "message"];

/// A recognized `CALL db.checkpoint()` invocation and the columns it asked
/// for (all of [`CHECKPOINT_COLUMNS`] when there is no `YIELD`).
#[derive(Debug, PartialEq, Eq)]
pub(super) struct CheckpointCall {
    pub(super) columns: Vec<&'static str>,
}

/// Strip `keyword` from the front of `input`, case-insensitively, requiring a
/// word boundary after it; returns the remainder with leading whitespace
/// trimmed.
///
/// The boundary check is what stops `CALLS ...` or `YIELDing` from being read
/// as the keyword plus a remainder.
pub(super) fn strip_keyword_ci<'a>(input: &'a str, keyword: &str) -> Option<&'a str> {
    if !input.get(..keyword.len())?.eq_ignore_ascii_case(keyword) {
        return None;
    }
    let rest = &input[keyword.len()..];
    match rest.chars().next() {
        Some(c) if c.is_alphanumeric() || c == '_' => None,
        _ => Some(rest.trim_start()),
    }
}

/// The statement without a leading `CYPHER [5|25] [name=value ...]` prefix
/// (Neo4j's version/option selector), so the verb recognizers below see the
/// statement itself. The text is returned unchanged when there is no prefix;
/// a malformed one is left for the engine's parser to reject.
pub(super) fn strip_cypher_prefix(input: &str) -> &str {
    let Some(mut rest) = strip_keyword_ci(input, "cypher") else {
        return input;
    };
    let mut first = true;
    loop {
        let word_end = rest.find(|c: char| c.is_whitespace()).unwrap_or(rest.len());
        let word = &rest[..word_end];
        let is_version = first && matches!(word, "5" | "25");
        let is_option = word.split_once('=').is_some_and(|(name, value)| {
            !name.is_empty()
                && !value.is_empty()
                && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
        });
        if !is_version && !is_option {
            return rest;
        }
        first = false;
        rest = rest[word_end..].trim_start();
    }
}

/// Recognize exactly `CALL db.checkpoint()`, optionally followed by a `YIELD`
/// naming a subset of [`CHECKPOINT_COLUMNS`]; `None` for anything else.
///
/// **Deliberately narrow.** Keywords and the procedure name are matched
/// case-insensitively (Cypher keywords are, and every driver spells the
/// procedure lowercase anyway), surrounding whitespace and one trailing `;`
/// are tolerated — but arguments inside the parentheses, an unknown or
/// repeated `YIELD` column, an alias (`YIELD success AS ok`), or a differently
/// *cased* column name all fall through to the engine, which answers with its
/// standard "Unknown procedure 'db.checkpoint'". That is the intended
/// behaviour, not a gap: YIELD names are Cypher identifiers, and quietly
/// re-casing one would hand the driver a record whose key is not the key the
/// client asked for. Only the exact verb is a bolt-layer verb.
pub(super) fn parse_checkpoint_call(query: &str) -> Option<CheckpointCall> {
    parse_procedure_call(query, "db.checkpoint", &CHECKPOINT_COLUMNS)
        .map(|columns| CheckpointCall { columns })
}

/// Build the single-record result a `db.checkpoint()` call answers with,
/// projected onto the columns the client yielded.
///
/// `success` is always `true`: every way a checkpoint can fail — refused,
/// or a save error — returns a Bolt FAILURE instead, so a record that reaches
/// the client is a record about a checkpoint that either wrote or was
/// deliberately skipped. `type` distinguishes the two ("w" wrote, "r" didn't).
pub(super) fn checkpoint_stream(
    call: &CheckpointCall,
    message: String,
    type_str: &str,
    started: Instant,
) -> ResultStream {
    let values: Vec<BoltValue> = call
        .columns
        .iter()
        .map(|column| {
            debug_assert!(
                CHECKPOINT_COLUMNS.contains(column),
                "parse_checkpoint_call yielded an unknown column: {column}"
            );
            if *column == CHECKPOINT_COLUMNS[0] {
                BoltValue::Boolean(true)
            } else {
                BoltValue::String(message.clone())
            }
        })
        .collect();
    ResultStream {
        metadata: ResultMetadata {
            columns: call.columns.iter().map(|c| (*c).to_string()).collect(),
            extra: BoltDict::new(),
        },
        records: vec![BoltRecord { values }],
        summary: BoltDict::from([
            ("type".to_string(), BoltValue::String(type_str.to_string())),
            (
                "t_last".to_string(),
                BoltValue::Integer(started.elapsed().as_millis() as i64),
            ),
        ]),
    }
}

/// Declared output columns of `CALL dbms.components()`, in declaration
/// order — the shape Neo4j yields, which Neo4j Browser reads as its very
/// first call (`serverInfoQuery` in the Browser source) to draw the
/// name/version/edition banner.
pub(super) const COMPONENTS_COLUMNS: [&str; 3] = ["name", "versions", "edition"];

/// Declared output columns of `CALL dbms.showCurrentUser()` (Neo4j shape).
pub(super) const SHOW_CURRENT_USER_COLUMNS: [&str; 3] = ["username", "roles", "flags"];

/// Columns of `SHOW DATABASES`, the Neo4j 5 set. Clients key on `name`,
/// `default`, `home`, and `currentStatus`; the rest are present so a client
/// reading the full row does not find columns missing.
pub(super) const SHOW_DATABASES_COLUMNS: [&str; 13] = [
    "name",
    "type",
    "aliases",
    "access",
    "address",
    "role",
    "writer",
    "requestedStatus",
    "currentStatus",
    "statusMessage",
    "default",
    "home",
    "constituents",
];

/// A bolt-layer *server-facts* verb: identity, user, database roster. Like
/// `db.checkpoint()`, these are answered here because the Cypher engine has
/// none of the state they report — no server identity (`--neo4j-compat`
/// lives in this crate), no auth config, no advertised address. An
/// engine-side implementation would make `graph.cypher("CALL
/// dbms.components()")` in the wheel answer questions about a server that
/// is not running.
#[derive(Debug, PartialEq, Eq)]
pub(super) enum ServerFactsVerb {
    DbmsComponents,
    DbmsShowCurrentUser,
    ShowDatabases,
}

/// A recognized server-facts invocation and the columns it asked for.
#[derive(Debug, PartialEq, Eq)]
pub(super) struct ServerFactsCall {
    pub(super) verb: ServerFactsVerb,
    pub(super) columns: Vec<&'static str>,
}

/// Recognize `CALL <proc>()`, optionally followed by `YIELD` naming a subset
/// of `declared`; `None` for anything else. Shares `parse_checkpoint_call`'s
/// deliberate narrowness: arguments, aliases, unknown or repeated or
/// differently-cased YIELD columns all fall through to the engine and its
/// standard "Unknown procedure" error.
pub(super) fn parse_procedure_call(
    query: &str,
    proc: &str,
    declared: &'static [&'static str],
) -> Option<Vec<&'static str>> {
    let mut body = query.trim();
    if let Some(stripped) = body.strip_suffix(';') {
        body = stripped.trim_end();
    }
    let rest = strip_keyword_ci(body, "call")?;
    let rest = strip_keyword_ci(rest, proc)?;
    let rest = rest.strip_prefix('(')?.trim_start();
    let rest = rest.strip_prefix(')')?.trim_start();
    parse_yield_tail(rest, declared)
}

/// Parse what follows a procedure's closing parenthesis: nothing (all
/// `declared` columns) or `YIELD` naming a duplicate-free subset of them.
fn parse_yield_tail(rest: &str, declared: &'static [&'static str]) -> Option<Vec<&'static str>> {
    if rest.is_empty() {
        return Some(declared.to_vec());
    }
    let yielded = strip_keyword_ci(rest, "yield")?;
    let mut columns: Vec<&'static str> = Vec::with_capacity(declared.len());
    for raw in yielded.split(',') {
        let name = raw.trim();
        let canonical = declared.iter().find(|column| **column == name)?;
        if columns.contains(canonical) {
            return None;
        }
        columns.push(*canonical);
    }
    Some(columns)
}

/// Output columns of `db.backup(...)`, in declaration order.
pub(super) const BACKUP_COLUMNS: [&str; 10] = [
    "success",
    "path",
    "lsn",
    "nodes",
    "relationships",
    "bytes",
    "lock_hold_ms",
    "elapsed_ms",
    "graph_version",
    "prepared_copy",
];

/// The single argument of `db.backup(...)`.
#[derive(Debug, PartialEq, Eq)]
pub(super) enum BackupArg {
    /// A quoted string literal, quotes removed.
    Literal(String),
    /// A `$name` parameter reference (name without the `$`).
    Param(String),
}

/// A recognized `CALL db.backup(<name>)` invocation and the columns it asked for.
#[derive(Debug, PartialEq, Eq)]
pub(super) struct BackupCall {
    pub(super) arg: BackupArg,
    pub(super) columns: Vec<&'static str>,
}

/// Recognize `CALL db.backup('<name>')` or `CALL db.backup($param)` with an
/// optional `YIELD` of a subset of [`BACKUP_COLUMNS`]; `None` for anything else.
///
/// Exactly one argument: a string literal or a parameter. A second argument,
/// an expression, a backslash or the other quote character inside a literal
/// all fall through to the engine's "Unknown procedure" — path text is
/// policy-checked later, so the parser keeps no escape rules to get wrong.
pub(super) fn parse_backup_call(query: &str) -> Option<BackupCall> {
    let mut body = query.trim();
    if let Some(stripped) = body.strip_suffix(';') {
        body = stripped.trim_end();
    }
    let rest = strip_keyword_ci(body, "call")?;
    let rest = strip_keyword_ci(rest, "db.backup")?;
    let rest = rest.strip_prefix('(')?.trim_start();
    let (arg, rest) = if let Some(after) = rest.strip_prefix('$') {
        let end = after
            .find(|c: char| !(c.is_alphanumeric() || c == '_'))
            .unwrap_or(after.len());
        if end == 0 {
            return None;
        }
        (BackupArg::Param(after[..end].to_string()), &after[end..])
    } else {
        let quote = rest.chars().next().filter(|c| *c == '\'' || *c == '"')?;
        let after = &rest[1..];
        let end = after.find(quote)?;
        let literal = &after[..end];
        if literal.contains('\\') {
            return None;
        }
        (BackupArg::Literal(literal.to_string()), &after[end + 1..])
    };
    let rest = rest.trim_start().strip_prefix(')')?.trim_start();
    let columns = parse_yield_tail(rest, &BACKUP_COLUMNS)?;
    Some(BackupCall { arg, columns })
}

/// Project a finished backup onto the yielded columns as a one-record stream.
/// `success` is always `true`: every failure is a Bolt FAILURE instead.
pub(super) fn backup_stream(
    call: &BackupCall,
    report: &kglite::api::session::BackupReport,
    started: Instant,
) -> ResultStream {
    let values: Vec<BoltValue> = call
        .columns
        .iter()
        .map(|column| match *column {
            "success" => BoltValue::Boolean(true),
            "path" => BoltValue::String(report.path.display().to_string()),
            "lsn" => report
                .lsn
                .map_or(BoltValue::Null, |lsn| BoltValue::Integer(lsn as i64)),
            "nodes" => BoltValue::Integer(report.nodes as i64),
            "relationships" => BoltValue::Integer(report.relationships as i64),
            "bytes" => BoltValue::Integer(report.bytes as i64),
            "lock_hold_ms" => BoltValue::Integer(report.lock_hold.as_millis() as i64),
            "elapsed_ms" => BoltValue::Integer(report.elapsed.as_millis() as i64),
            "graph_version" => BoltValue::Integer(report.graph_version as i64),
            "prepared_copy" => BoltValue::Boolean(report.prepared_copy),
            other => unreachable!("parse_backup_call yielded an unknown column: {other}"),
        })
        .collect();
    ResultStream {
        metadata: ResultMetadata {
            columns: call.columns.iter().map(|c| (*c).to_string()).collect(),
            extra: BoltDict::new(),
        },
        records: vec![BoltRecord { values }],
        summary: BoltDict::from([
            ("type".to_string(), BoltValue::String("r".to_string())),
            (
                "t_last".to_string(),
                BoltValue::Integer(started.elapsed().as_millis() as i64),
            ),
        ]),
    }
}

/// Recognize the server-facts verbs. `SHOW DATABASES` is matched in its
/// plain statement form only — a `YIELD`/`WHERE` tail falls through to the
/// engine (and its parse error), which is the honest answer until the
/// engine grows SHOW projections.
pub(super) fn parse_server_facts_call(query: &str) -> Option<ServerFactsCall> {
    if let Some(columns) = parse_procedure_call(query, "dbms.components", &COMPONENTS_COLUMNS) {
        return Some(ServerFactsCall {
            verb: ServerFactsVerb::DbmsComponents,
            columns,
        });
    }
    if let Some(columns) =
        parse_procedure_call(query, "dbms.showcurrentuser", &SHOW_CURRENT_USER_COLUMNS)
    {
        return Some(ServerFactsCall {
            verb: ServerFactsVerb::DbmsShowCurrentUser,
            columns,
        });
    }
    let mut body = query.trim();
    if let Some(stripped) = body.strip_suffix(';') {
        body = stripped.trim_end();
    }
    let rest = strip_keyword_ci(body, "show")?;
    let rest = strip_keyword_ci(rest, "databases")?;
    if rest.is_empty() {
        return Some(ServerFactsCall {
            verb: ServerFactsVerb::ShowDatabases,
            columns: SHOW_DATABASES_COLUMNS.to_vec(),
        });
    }
    None
}

/// Rewrite `CALL dbms.components() [YIELD cols] <WHERE|RETURN|WITH ...>` into
/// a query the engine answers: the one row `dbms.components()` reports becomes
/// a literal `UNWIND`, and the YIELD columns (with their aliases) a `WITH`, so
/// filtering, projection, ordering and paging reuse the engine's pipeline.
/// `None` for the bare call and the YIELD-only forms (answered directly by
/// [`parse_server_facts_call`]) and for anything this does not recognise,
/// which falls through to the engine unchanged.
pub(super) fn components_pipeline_rewrite(
    query: &str,
    rows: &[(String, Vec<String>)],
    edition: &str,
) -> Option<String> {
    let mut body = query.trim();
    if let Some(stripped) = body.strip_suffix(';') {
        body = stripped.trim_end();
    }
    let rest = strip_keyword_ci(body, "call")?;
    let rest = strip_keyword_ci(rest, "dbms.components")?;
    let rest = rest.strip_prefix('(')?.trim_start();
    let rest = rest.strip_prefix(')')?.trim_start();
    let (yielded, tail) = match strip_keyword_ci(rest, "yield") {
        Some(after) => {
            let at = first_clause_keyword(after)?;
            (Some(after[..at].trim()), &after[at..])
        }
        None => (None, rest),
    };
    let first_word = tail.split(|c: char| !c.is_ascii_alphabetic()).next()?;
    if !["where", "return", "with"]
        .iter()
        .any(|k| first_word.eq_ignore_ascii_case(k))
    {
        return None;
    }
    let mut projected: Vec<(&str, &str)> = Vec::new();
    match yielded {
        None | Some("*") => {
            projected.extend(COMPONENTS_COLUMNS.iter().map(|c| (*c, *c)));
        }
        Some(list) => {
            for raw in list.split(',') {
                let mut words = raw.split_whitespace();
                let column = words.next()?;
                COMPONENTS_COLUMNS.iter().find(|c| **c == column)?;
                let alias = match (words.next(), words.next(), words.next()) {
                    (None, ..) => column,
                    (Some(kw), Some(alias), None) if kw.eq_ignore_ascii_case("as") => alias,
                    _ => return None,
                };
                if projected.iter().any(|(_, a)| *a == alias) {
                    return None;
                }
                projected.push((column, alias));
            }
        }
    }
    let quote = |text: &str| format!("'{}'", text.replace('\\', "\\\\").replace('\'', "\\'"));
    let row_literals: Vec<String> = rows
        .iter()
        .map(|(name, versions)| {
            let versions: Vec<String> = versions.iter().map(|v| quote(v)).collect();
            format!(
                "{{name: {}, versions: [{}], edition: {}}}",
                quote(name),
                versions.join(", "),
                quote(edition)
            )
        })
        .collect();
    let with: Vec<String> = projected
        .iter()
        .map(|(column, alias)| format!("__components.{column} AS {alias}"))
        .collect();
    Some(format!(
        "UNWIND [{}] AS __components WITH {} {}",
        row_literals.join(", "),
        with.join(", "),
        tail
    ))
}

/// Byte offset of the first `WHERE`, `RETURN` or `WITH` word in `text`.
fn first_clause_keyword(text: &str) -> Option<usize> {
    let bytes = text.as_bytes();
    let mut start = 0;
    while start < bytes.len() {
        if bytes[start].is_ascii_alphabetic() && (start == 0 || !is_word_byte(bytes[start - 1])) {
            let end = text[start..]
                .find(|c: char| !(c.is_ascii_alphanumeric() || c == '_'))
                .map_or(text.len(), |n| start + n);
            let word = &text[start..end];
            if ["where", "return", "with"]
                .iter()
                .any(|k| word.eq_ignore_ascii_case(k))
            {
                return Some(start);
            }
            start = end;
        } else {
            start += 1;
        }
    }
    None
}

fn is_word_byte(b: u8) -> bool {
    b.is_ascii_alphanumeric() || b == b'_'
}

/// Convert the engine's EXPLAIN step rows into Neo4j's nested Bolt plan
/// shape: each node `{operatorType, args, identifiers, children}`, with the
/// FINAL pipeline step as the root (Neo4j's root is the last operator) and
/// each earlier step nested as its single child. `OptimizerPass <name>` rows
/// are not operators — they are collected into the root's
/// `args["optimizer-passes"]` list. Returns `None` when the shape is not the
/// engine's EXPLAIN output (leaving the stream untouched).
pub(super) fn plan_from_explain_rows(
    columns: &[String],
    rows: &[Vec<kglite::datatypes::values::Value>],
) -> Option<BoltValue> {
    use kglite::datatypes::values::Value;

    let step_idx = columns.iter().position(|c| c == "step")?;
    let op_idx = columns.iter().position(|c| c == "operation")?;
    let est_idx = columns.iter().position(|c| c == "estimated_rows")?;

    let mut steps: Vec<(i64, String, Option<i64>)> = Vec::new();
    let mut passes: Vec<BoltValue> = Vec::new();
    for row in rows {
        let step = match row.get(step_idx)? {
            Value::Int64(i) => *i,
            _ => return None,
        };
        let op = match row.get(op_idx)? {
            Value::String(s) => s.clone(),
            _ => return None,
        };
        let est = match row.get(est_idx) {
            Some(Value::Int64(i)) => Some(*i),
            _ => None,
        };
        if let Some(name) = op.strip_prefix("OptimizerPass ") {
            passes.push(BoltValue::String(name.to_string()));
        } else {
            steps.push((step, op, est));
        }
    }
    if steps.is_empty() {
        return None;
    }
    steps.sort_by_key(|(step, _, _)| *step);

    // Text rendering of the plan for `args["string-representation"]` on the
    // root — Neo4j puts its ASCII plan there, and G.V()'s plan tab reads the
    // key UNCONDITIONALLY (`summary.plan().arguments().get("string-
    // representation").asString()`, verified by decompilation after its tab
    // rendered an NPE against a plan without it). Root operator first,
    // matching the tree orientation.
    let text: String = {
        let mut lines = Vec::with_capacity(steps.len() + passes.len());
        for (_, op, est) in steps.iter().rev() {
            match est {
                Some(est) => lines.push(format!("+ {op}  (estimated rows: {est})")),
                None => lines.push(format!("+ {op}")),
            }
        }
        if !passes.is_empty() {
            let names: Vec<&str> = passes
                .iter()
                .filter_map(|p| match p {
                    BoltValue::String(s) => Some(s.as_str()),
                    _ => None,
                })
                .collect();
            lines.push(format!("optimizer passes: {}", names.join(", ")));
        }
        lines.join("\n")
    };

    // Fold from the first step outward: the last operator becomes the root.
    let mut node: Option<BoltValue> = None;
    let last = steps.len() - 1;
    for (i, (_, op, est)) in steps.into_iter().enumerate() {
        let mut args = BoltDict::new();
        if let Some(est) = est {
            args.insert("EstimatedRows".to_string(), BoltValue::Float(est as f64));
        }
        if i == last {
            args.insert(
                "string-representation".to_string(),
                BoltValue::String(text.clone()),
            );
            args.insert(
                "runtime".to_string(),
                BoltValue::String("kglite".to_string()),
            );
            if !passes.is_empty() {
                args.insert(
                    "optimizer-passes".to_string(),
                    BoltValue::List(std::mem::take(&mut passes)),
                );
            }
        }
        let children = match node.take() {
            Some(child) => vec![child],
            None => Vec::new(),
        };
        node = Some(BoltValue::Dict(BoltDict::from([
            ("operatorType".to_string(), BoltValue::String(op)),
            ("args".to_string(), BoltValue::Dict(args)),
            ("identifiers".to_string(), BoltValue::List(Vec::new())),
            ("children".to_string(), BoltValue::List(children)),
        ])));
    }
    node
}

/// Project `(column, value)` pairs onto the client's yielded columns and wrap
/// them as a single-record read stream.
pub(super) fn server_facts_stream(
    columns: &[&'static str],
    values: &[(&'static str, BoltValue)],
    started: Instant,
) -> ResultStream {
    server_facts_rows_stream(columns, &[values.to_vec()], started)
}

/// The `dbms.components()` rows, projected onto the yielded columns.
pub(super) fn components_stream(
    columns: &[&'static str],
    rows: &[(String, Vec<String>)],
    edition: &str,
    started: Instant,
) -> ResultStream {
    let rows: Vec<Vec<(&'static str, BoltValue)>> = rows
        .iter()
        .map(|(name, versions)| {
            vec![
                ("name", BoltValue::String(name.clone())),
                (
                    "versions",
                    BoltValue::List(versions.iter().cloned().map(BoltValue::String).collect()),
                ),
                ("edition", BoltValue::String(edition.to_string())),
            ]
        })
        .collect();
    server_facts_rows_stream(columns, &rows, started)
}

/// [`server_facts_stream`] for a verb that reports several rows.
fn server_facts_rows_stream(
    columns: &[&'static str],
    rows: &[Vec<(&'static str, BoltValue)>],
    started: Instant,
) -> ResultStream {
    let records: Vec<BoltRecord> = rows
        .iter()
        .map(|values| BoltRecord {
            values: columns
                .iter()
                .map(|column| {
                    values
                        .iter()
                        .find(|(name, _)| name == column)
                        .map(|(_, value)| value.clone())
                        .unwrap_or(BoltValue::Null)
                })
                .collect(),
        })
        .collect();
    ResultStream {
        metadata: ResultMetadata {
            columns: columns.iter().map(|c| (*c).to_string()).collect(),
            extra: BoltDict::new(),
        },
        records,
        summary: BoltDict::from([
            ("type".to_string(), BoltValue::String("r".to_string())),
            (
                "t_last".to_string(),
                BoltValue::Integer(started.elapsed().as_millis() as i64),
            ),
        ]),
    }
}