nedb_engine/

pgwire.rs

// SPDX-FileCopyrightText: 2026 INTERCHAINED LLC
// SPDX-License-Identifier: BUSL-1.1
// NEDB · © 2026 INTERCHAINED LLC × Eth-Interchained × Vex (Claude Opus 5)

//! A PostgreSQL wire-protocol endpoint for NEDB — reads **and** writes.
//!
//! # What this is
//!
//! A front door that speaks the PostgreSQL v3 wire protocol well enough that
//! tools built for Postgres — `psql`, DBeaver, Metabase, Grafana, psycopg, any
//! libpq client — can use a NEDB store with ordinary SQL. A documented subset
//! of SQL is translated to NQL and to engine writes; everything else is
//! refused with an error naming exactly what was not understood.
//!
//! It is **not** a claim of Postgres parity. It is a claim that the SQL people
//! actually type works, and that the boundary is stated rather than discovered.
//!
//! # Why writes belong here
//!
//! The first cut of this module was read-only, on the reasoning that a NEDB
//! write carries `caused_by`, valid-time bounds and idempotency, and none of
//! that has a natural SQL spelling. That reasoning was wrong, and looking at
//! the mapping is what made it obvious:
//!
//! | SQL | NEDB | and therefore |
//! |---|---|---|
//! | `INSERT` | a put | — |
//! | `UPDATE … WHERE` | a NEW VERSION of each match | the prior value stays readable |
//! | `DELETE … WHERE` | a tombstone | the deleted row stays in history |
//!
//! NEDB is append-only, so an `UPDATE` is *already* a versioned write and a
//! `DELETE` is *already* a tombstone. Nothing is bent to fit. The consequence
//! is the point of the whole endpoint:
//!
//! ```sql
//! UPDATE orders SET total = 999 WHERE _id = 'o1';
//! SELECT total FROM orders WHERE _id = 'o1';                  -- 999
//! SELECT total FROM orders AS OF SYSTEM TIME 0 WHERE _id = 'o1';  -- 120
//! ```
//!
//! Run the SQL you would run against Postgres, and the tamper-evident history
//! is free. No triggers, no audit table, no application code.
//!
//! Provenance is reachable too: `_caused_by`, `_valid_from` and `_valid_to` are
//! reserved INSERT columns, lifted out of the payload into the write itself.
//!
//! Writes are ON by default — that is the parity position. Set
//! `NEDBD_PG_READ_ONLY=1` for the deployment where this door must never mutate
//! anything.
//!
//! # Supported SQL
//!
//! ```sql
//! SELECT * | col [, col]* | COUNT(*) | <agg>(col)
//!   FROM <collection>
//!   [ AS OF SYSTEM TIME <seq> ]     -- bridges to NQL's AS OF
//!   [ WHERE <predicate> ]           -- the full NQL predicate surface
//!   [ GROUP BY <col> ] [ HAVING <predicate> ]
//!   [ ORDER BY <col> [ASC|DESC] (, ...) ] [ LIMIT <n> ] [ OFFSET <n> ]
//!
//! INSERT INTO <collection> (c1, c2) VALUES (v1, v2), (…) [RETURNING …]
//! UPDATE <collection> SET c = v [, …] [WHERE <predicate>] [RETURNING …]
//! DELETE FROM <collection> [WHERE <predicate>] [RETURNING …]
//! ```
//!
//! Single-quoted SQL literals are rewritten to NQL's double-quoted form and
//! `<>` to `!=`. Column projection is applied here, after NQL returns whole
//! documents, because NQL is FROM-first and has no projection clause.
//!
//! That translation serves user collections. Statements that read the
//! catalogue (`pg_catalog.*`, `information_schema.*`) go instead to the real
//! SQL evaluator in `sqlselect` — joins, subqueries, `EXISTS`, `ARRAY(...)`,
//! `ANY`/`ALL`, `UNION`, derived tables, `LATERAL`, aggregates, `CASE`, scalar
//! functions — because that is what psql's `\d` family is written in. Every
//! psql 17 backslash command that can succeed against an empty-of-features
//! Postgres exits 0 here, verified by driving the real binary
//! (`tests/test_psql_introspection.py`).
//!
//! Not supported on the user-collection path, each refused by name: JOIN,
//! subqueries, CTEs, window functions, DDL, `TRUNCATE`, `GRANT`/`REVOKE`.
//! `INSERT` requires an explicit column list, because NEDB is schemaless and
//! there is no declared column order to infer.
//!
//! # Protocol coverage
//!
//! **Both** protocols are implemented:
//!
//! * the **simple query protocol** (`Q`) — what `psql` and libpq's `PQexec`
//!   use, and therefore psycopg2, which interpolates parameters client-side;
//! * the **extended query protocol** (`Parse`/`Bind`/`Describe`/`Execute`/
//!   `Close`/`Sync`/`Flush`) — what psycopg3, asyncpg and the JDBC driver use
//!   for every parameterised statement. Without it those three could not run a
//!   single query, so "psql works" was a long way from "your framework works".
//!
//! Parameters arrive in text *and* binary format, prepared statements and
//! portals are per-connection, and a row-capped `Execute` suspends its portal
//! (`PortalSuspended`) so a JDBC `setFetchSize` pages instead of stalling.
//!
//! ## Parameter typing in a store with no schema
//!
//! The extended protocol needs types for `$1..$n`, which a relational server
//! reads out of its catalogue. NEDB has none — so the types are sampled from
//! the documents already stored, and the stored data *is* the schema. Where a
//! placeholder sits in a clause rather than beside a column
//! (`AS OF SYSTEM TIME $1`, `LIMIT $1`) the grammar supplies the type instead,
//! and an aggregate column is typed from what the aggregate means: a `COUNT` is
//! an integer, an `AVG` fractional.
//!
//! This is not polish. A client that declares its own parameter types
//! (psycopg3, JDBC) is believed and only its unspecified slots are inferred —
//! but asyncpg declares none, asks, and then **refuses the call client-side**
//! if the answer is wrong. Advertising "text" for everything does not degrade
//! gracefully there; it fails with `expected str, got int` before a query is
//! ever sent.
//!
//! SSL is declined (`N`), so connections are cleartext — hence the loopback
//! default.
//!
//! Authentication mirrors the HTTP surface: with `NEDBD_TOKEN` set the password
//! must equal it; otherwise any connection is accepted.
//!
//! Still outside the boundary, and refused by name: SQL-level cursors
//! (`DECLARE`/`FETCH`), window functions, set operations on the
//! user-collection path, and binary *result* format for a column whose stored
//! values disagree about their type across documents.
//!
//! # What an ORM needs, and what it cost to learn
//!
//! Speaking psql is not speaking to a framework, and the difference was three
//! defects deep. SQLAlchemy could not CONNECT (its dialect opens with
//! `select pg_catalog.version()`, which a table of exact spellings missed); its
//! reflection needed `GROUP BY` and `array_agg(x ORDER BY y)`; and a QUALIFIED
//! column in a `WHERE` clause returned ZERO ROWS — silently — because NQL
//! looks a field up flat and no document has a field named `orders.status`.
//! Every ORM qualifies its predicates, so every filtered query lied.
//!
//! None of that was visible to psql, which is why
//! `tests/pgwire_suite.py` drives asyncpg, SQLAlchemy and node-postgres
//! against a live daemon on every push.

use std::collections::HashMap;
use std::sync::Arc;

use serde_json::Value;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

use crate::db::Db;

// ── Postgres type OIDs we hand out ──────────────────────────────────────────
const OID_BOOL: i32 = 16;
const OID_INT8: i32 = 20;
const OID_FLOAT8: i32 = 701;
const OID_TEXT: i32 = 25;

const PROTO_V3: i32 = 196_608; // 3.0 << 16
const SSL_REQUEST: i32 = 80_877_103;
const GSS_REQUEST: i32 = 80_877_104;
const CANCEL_REQUEST: i32 = 80_877_102;

/// How a caller resolves a database name to an open `Db`.
///
/// A trait object rather than a concrete handle so this module does not depend
/// on `server::Manager` — which keeps the protocol code unit-testable against a
/// plain `Db` with no HTTP stack in the way.
pub trait DbResolver: Send + Sync + 'static {
    /// Look up an open database by the name the client connected with.
    ///
    /// MAY BLOCK. The implementation is allowed to take a lock, so this is
    /// always called from `spawn_blocking` — never on an async worker. Taking
    /// a tokio `RwLock::blocking_read()` on a runtime thread panics outright
    /// ("Cannot block the current thread from within a runtime"), which is
    /// exactly how the first cut of this failed.
    fn resolve(&self, name: &str) -> Option<Arc<Db>>;
    /// The bearer token, when one is configured. `None` = open access.
    fn token(&self) -> Option<String> {
        None
    }
}

// ── wire encoding helpers ───────────────────────────────────────────────────

struct Out(Vec<u8>);

impl Out {
    fn msg(tag: u8) -> Self {
        // Tag, then a 4-byte length placeholder patched in `finish`.
        Out(vec![tag, 0, 0, 0, 0])
    }
    fn i16(&mut self, v: i16) { self.0.extend_from_slice(&v.to_be_bytes()); }
    fn i32(&mut self, v: i32) { self.0.extend_from_slice(&v.to_be_bytes()); }
    fn cstr(&mut self, s: &str) {
        // A NUL inside an identifier would truncate the field and desynchronise
        // the stream, so strip rather than trust.
        self.0.extend_from_slice(s.replace('\0', "").as_bytes());
        self.0.push(0);
    }
    fn bytes(&mut self, b: &[u8]) { self.0.extend_from_slice(b); }
    /// Patch the length prefix (which covers the length field itself, not the tag).
    fn finish(mut self) -> Vec<u8> {
        let len = (self.0.len() - 1) as i32;
        self.0[1..5].copy_from_slice(&len.to_be_bytes());
        self.0
    }
}

fn err_msg(code: &str, message: &str) -> Vec<u8> {
    let mut m = Out::msg(b'E');
    m.bytes(b"S"); m.cstr("ERROR");
    m.bytes(b"C"); m.cstr(code);
    m.bytes(b"M"); m.cstr(message);
    m.0.push(0);
    m.finish()
}

fn ready() -> Vec<u8> {
    let mut m = Out::msg(b'Z');
    m.bytes(b"I"); // idle, not in a transaction
    m.finish()
}

fn command_complete(tag: &str) -> Vec<u8> {
    let mut m = Out::msg(b'C');
    m.cstr(tag);
    m.finish()
}

// ── SQL → NQL translation ───────────────────────────────────────────────────

/// One output column: the key to read from the row, and the name to show.
///
/// The two differ for aggregates. NQL answers `SUM(total)` with a row holding
/// `sum_total` (plus `count` and a legacy `value` alias), while SQL callers
/// expect a single column called `sum`. Carrying both halves keeps NEDB's
/// internal key names off the wire — the first cut leaked `['count','value']`
/// out of a `SELECT COUNT(*)`, which is two columns where SQL promises one.
#[derive(Debug, PartialEq, Clone)]
pub struct Col {
    pub src: String,
    pub out: String,
}

impl Col {
    fn same(name: &str) -> Self {
        Col { src: name.to_string(), out: name.to_string() }
    }
    fn renamed(src: &str, out: &str) -> Self {
        Col { src: src.to_string(), out: out.to_string() }
    }
}

/// What a translated statement asks for.
///
/// The write variants exist because SQL's write semantics and NEDB's storage
/// model line up almost exactly, which was not obvious until it was written
/// down:
///
/// | SQL | NEDB |
/// |---|---|
/// | `INSERT` | a put |
/// | `UPDATE … WHERE` | a NEW VERSION of each matching document |
/// | `DELETE … WHERE` | a tombstone |
///
/// NEDB is append-only, so an `UPDATE` is *already* a versioned write and a
/// `DELETE` is *already* a tombstone. Nothing is being bent to fit. The
/// consequence is the thing worth selling: run the SQL you would run against
/// Postgres, and the tamper-evident history falls out for free — the prior
/// value is still readable with `AS OF SYSTEM TIME`.
#[derive(Debug, PartialEq)]
pub enum Stmt {
    /// Run this NQL, then project these columns (empty = all).
    Query { nql: String, project: Vec<Col> },
    /// `INSERT INTO coll (cols) VALUES (…), (…) [RETURNING …]`
    Insert { coll: String, rows: Vec<InsertRow>, returning: Vec<Col> },
    /// `UPDATE coll SET … [WHERE …] [RETURNING …]` — a new version per match.
    Update { coll: String, set: Vec<(String, Value)>, nql: String, returning: Vec<Col> },
    /// `DELETE FROM coll [WHERE …] [RETURNING …]` — a tombstone per match.
    Delete { coll: String, nql: String, returning: Vec<Col> },
    /// Answer from a fixed table — the handshake queries clients send on connect.
    Canned { cols: Vec<String>, row: Vec<String> },
    /// Nothing to do (empty statement, or a SET the client does not need honoured).
    Ok(&'static str),
}

/// One row of an `INSERT`: an explicit id when the statement supplied one, the
/// document body, and optional provenance lifted out of reserved columns.
#[derive(Debug, PartialEq, Clone)]
pub struct InsertRow {
    /// From an `_id` or `id` column. `None` means the server assigns one.
    pub id: Option<String>,
    pub doc: serde_json::Map<String, Value>,
    /// From a `_caused_by` column — the causal parents, so provenance is
    /// reachable from SQL rather than only from the HTTP API.
    pub caused_by: Vec<String>,
    pub valid_from: Option<String>,
    pub valid_to: Option<String>,
}

/// Strip SQL comments and collapse whitespace, so the matchers below can be
/// simple without being fragile about formatting.
fn normalise(sql: &str) -> String {
    let mut out = String::with_capacity(sql.len());
    let mut chars = sql.chars().peekable();
    let mut in_s = false;
    while let Some(c) = chars.next() {
        if in_s {
            out.push(c);
            if c == '\'' { in_s = false; }
            continue;
        }
        match c {
            '\'' => { in_s = true; out.push(c); }
            '-' if chars.peek() == Some(&'-') => {
                // line comment
                for n in chars.by_ref() { if n == '\n' { break; } }
                out.push(' ');
            }
            '/' if chars.peek() == Some(&'*') => {
                chars.next();
                let mut prev = ' ';
                while let Some(n) = chars.next() {
                    if prev == '*' && n == '/' { break; }
                    prev = n;
                }
                out.push(' ');
            }
            _ => out.push(c),
        }
    }
    out.split_whitespace().collect::<Vec<_>>().join(" ")
}

/// Rewrite SQL literal/operator spellings into NQL's.
///
/// Only `'…'` → `"…"` and `<>` → `!=`. Done with an explicit scan rather than a
/// regex so a quote inside a string cannot be mistaken for a delimiter: SQL
/// escapes an embedded quote by doubling it (`'it''s'`), and that has to become
/// a single character inside the NQL string rather than terminating it.
/// Drop the table qualifier from every column reference in a clause tail.
///
/// # The silent wrong answer this removes
///
/// NQL has no notion of a qualifier: `field_value` looks a field up FLAT, in
/// one map. So `WHERE orders.status = 'paid'` asked for a field literally
/// named `orders.status`, no document had one, and the query returned ZERO
/// ROWS — with no error and no warning, an empty result that reads exactly
/// like "you have no paid orders".
///
/// Every ORM qualifies its predicates. SQLAlchemy emits
/// `SELECT orders._id FROM orders WHERE orders.status = 'paid'` for the most
/// ordinary filter there is, so EVERY filtered query answered empty, `.get(pk)`
/// answered `None`, and `filter_by` answered nothing. The select list had
/// always stripped qualifiers; the tail was "handed to the NQL parser
/// unchanged", which is right for the clause GRAMMAR and wrong for a name NQL
/// cannot interpret.
///
/// # Why a mismatched qualifier is an ERROR, not a strip
///
/// A qualifier naming something other than this statement's own collection
/// means the query referenced a relation that is not in its FROM clause.
/// Stripping it would answer with rows from the one relation that IS there —
/// a different wrong answer wearing the same empty-looking clothes. Aliases
/// are refused on this path already, so the collection's own name is the only
/// qualifier that can be correct.
///
/// Runs BEFORE `sql_literals_to_nql`, so only SQL's single-quoted strings have
/// to be skipped — the rewrite to NQL's double-quoted form has not happened
/// yet, and a qualifier can never appear inside a literal.
fn strip_column_qualifiers(
    tail: &str,
    coll: &str,
    alias: Option<&str>,
) -> Result<String, String> {
    let bare = coll.rsplit('.').next().unwrap_or(coll);
    let b: Vec<char> = tail.chars().collect();
    let mut out = String::with_capacity(tail.len());
    let mut i = 0usize;
    let ident_start = |c: char| c.is_alphabetic() || c == '_';
    let ident_char = |c: char| c.is_alphanumeric() || c == '_';

    while i < b.len() {
        // A single-quoted literal is copied through untouched.
        if b[i] == '\'' {
            out.push(b[i]);
            i += 1;
            while i < b.len() {
                out.push(b[i]);
                if b[i] == '\'' {
                    // A doubled '' is one literal quote, not a close.
                    if b.get(i + 1) == Some(&'\'') {
                        out.push('\'');
                        i += 2;
                        continue;
                    }
                    i += 1;
                    break;
                }
                i += 1;
            }
            continue;
        }
        // A double-quoted run is copied through too. NQL reads double quotes
        // as a STRING delimiter rather than an identifier one, so a SQL
        // delimited identifier is a genuine divergence — but it already fails
        // LOUDLY in the NQL parser ("expected field name, got Str"), and a
        // loud failure is not this function's problem to solve quietly.
        if b[i] == '"' {
            out.push(b[i]);
            i += 1;
            while i < b.len() {
                out.push(b[i]);
                if b[i] == '"' { i += 1; break; }
                i += 1;
            }
            continue;
        }
        if !ident_start(b[i]) {
            // A number like `1.5` starts with a digit, so it never enters the
            // identifier branch and its dot is never touched.
            out.push(b[i]);
            i += 1;
            continue;
        }

        let start = i;
        while i < b.len() && ident_char(b[i]) {
            i += 1;
        }
        let word: String = b[start..i].iter().collect();

        // `qual.field` — a dot followed immediately by another identifier.
        if b.get(i) == Some(&'.') && b.get(i + 1).is_some_and(|c| ident_start(*c)) {
            let fstart = i + 1;
            let mut j = fstart;
            while j < b.len() && ident_char(b[j]) {
                j += 1;
            }
            let field: String = b[fstart..j].iter().collect();
            // A qualified FUNCTION call (`pg_catalog.something(`) is left
            // exactly as written: this path does not implement functions at
            // all, and NQL's own refusal names the function, which is more use
            // to the reader than a claim about relations.
            let is_call = b[j..].iter().find(|c| !c.is_whitespace()) == Some(&'(');
            if is_call {
                out.push_str(&word);
                out.push('.');
                out.push_str(&field);
                i = j;
                continue;
            }
            let matches_alias = alias.is_some_and(|a| word.eq_ignore_ascii_case(a));
            if matches_alias || word.eq_ignore_ascii_case(bare) || word.eq_ignore_ascii_case(coll) {
                out.push_str(&field);
                i = j;
                continue;
            }
            return Err(format!(
                "no table or alias named {:?} in this query — this statement reads \
                 {:?}{}, and a qualifier naming anything else would have to be \
                 answered from a relation that is not in its FROM clause",
                word,
                bare,
                alias.map(|a| format!(" (aliased {:?})", a)).unwrap_or_default()
            ));
        }
        out.push_str(&word);
    }
    Ok(out)
}

/// Rewrite `SELECT count(*) FROM (<inner>) [AS] alias` into a flat count over
/// the inner query's own collection and predicate — or `None` when the shapes
/// do not permit it.
///
/// `None` is a REFUSAL, never a fallback: every caller reports the boundary
/// rather than trying something else, because the alternative to an exact
/// count is a wrong one.
fn flatten_count_of_subquery(projection: &str, rest: &str) -> Option<String> {
    // The outer select list must be nothing but `count(*)`, optionally
    // aliased. Any other column would have to come from the derived table's
    // output, which a flat count does not produce.
    let (outer_expr, outer_alias) = split_output_alias(projection.trim());
    let ou = outer_expr.to_uppercase().replace(' ', "");
    if ou != "COUNT(*)" {
        return None;
    }

    // Take the balanced parenthesised span, honouring literals so a `)` inside
    // a string cannot close it early.
    let b: Vec<char> = rest.chars().collect();
    let mut depth = 0i32;
    let mut in_s = false;
    let mut end = None;
    for (i, &c) in b.iter().enumerate() {
        match c {
            '\'' => in_s = !in_s,
            '(' if !in_s => depth += 1,
            ')' if !in_s => {
                depth -= 1;
                if depth == 0 {
                    end = Some(i);
                    break;
                }
            }
            _ => {}
        }
    }
    let end = end?;
    let inner = b[1..end].iter().collect::<String>().trim().to_string();

    // Nothing may follow the derived table but its alias — a join or a second
    // FROM item changes what is being counted.
    let trailing = b[end + 1..].iter().collect::<String>();
    let (_alias, after) = split_table_alias(trailing.trim());
    if !after.trim().is_empty() {
        return None;
    }

    let iu = inner.to_uppercase();
    if !iu.starts_with("SELECT") {
        return None;
    }
    // Each of these would make the inner row count differ from the flat one.
    for kw in ["LIMIT", "OFFSET", "GROUP BY", "HAVING", "UNION", "INTERSECT", "EXCEPT", "JOIN"] {
        if find_kw(&iu, kw).is_some() {
            return None;
        }
    }
    if find_kw(&iu, "DISTINCT").is_some() {
        return None;
    }
    // An inner aggregate already reduced the rows to one.
    let inner_from = find_kw(&iu, "FROM")?;
    let inner_list = inner[..inner_from].to_uppercase();
    for agg in ["COUNT(", "SUM(", "AVG(", "MIN(", "MAX(", "ARRAY_AGG(", "STRING_AGG("] {
        if inner_list.contains(agg) {
            return None;
        }
    }
    // A nested derived table is not walked — one level is the claim.
    let inner_rest = inner[inner_from + 4..].trim();
    if inner_rest.starts_with('(') {
        return None;
    }

    // `ORDER BY` cannot change a count, so it is dropped rather than refused.
    let mut tail = inner_rest.to_string();
    let tu = tail.to_uppercase();
    if let Some(ob) = find_kw(&tu, "ORDER BY") {
        tail = tail[..ob].trim_end().to_string();
    }
    Some(format!(
        "SELECT count(*){} FROM {}",
        outer_alias.map(|a| format!(" AS {}", a)).unwrap_or_default(),
        tail
    ))
}

/// Words that begin a clause and can therefore never be a bare table alias.
///
/// `AS` is absent on purpose: it introduces an alias, and `AS OF` is
/// disambiguated by looking at the word after it.
const CLAUSE_WORDS: &[&str] = &[
    "WHERE", "GROUP", "ORDER", "LIMIT", "OFFSET", "HAVING", "FOR", "VALID",
    "TRACE", "TRAVERSE", "SEARCH", "RETURNING", "UNION", "INTERSECT", "EXCEPT",
    "JOIN", "LEFT", "RIGHT", "INNER", "FULL", "CROSS", "ON", "USING", "SET",
];

/// Take a table alias off the front of a clause tail: `FROM orders o WHERE …`.
///
/// Returns the alias and the rest of the tail. The alias is REMOVED because
/// NQL has no table-alias syntax and would report an "unexpected token" on it
/// — which is how `FROM orders o` used to fail. Removing it here and teaching
/// `strip_column_qualifiers` to accept it is what makes `SELECT o.status FROM
/// orders o` work at all.
///
/// `AS OF SYSTEM TIME` also starts with `AS`, so the word AFTER `AS` decides:
/// `AS OF` is a time-travel clause, anything else is an alias.
fn split_table_alias(tail: &str) -> (Option<String>, &str) {
    let t = tail.trim_start();
    let first_end = t.find(char::is_whitespace).unwrap_or(t.len());
    let first = &t[..first_end];
    let fu = first.to_uppercase();

    if fu == "AS" {
        let rest = t[first_end..].trim_start();
        let end = rest.find(char::is_whitespace).unwrap_or(rest.len());
        let word = &rest[..end];
        if word.eq_ignore_ascii_case("OF") {
            return (None, t); // `AS OF …`, not an alias
        }
        if word.is_empty() {
            return (None, t);
        }
        return (Some(word.trim_matches('"').to_string()), rest[end..].trim_start());
    }
    if first.is_empty() || CLAUSE_WORDS.contains(&fu.as_str()) {
        return (None, t);
    }
    // A bare identifier here can only be an alias — the collection name was
    // already consumed by the caller.
    if first.chars().next().is_some_and(|c| c.is_alphabetic() || c == '_' || c == '"') {
        return (Some(first.trim_matches('"').to_string()), t[first_end..].trim_start());
    }
    (None, t)
}

/// Split on a delimiter that is at PAREN DEPTH ZERO and outside a literal.
///
/// `projection.split(',')` cuts `SUM(a, b)` in half; a select list is not a
/// flat comma list once it can contain calls.
fn split_top_level(s: &str, delim: char) -> Vec<String> {
    let mut out = vec![];
    let mut cur = String::new();
    let mut depth = 0i32;
    let mut in_s = false;
    let mut in_d = false;
    for c in s.chars() {
        match c {
            '\'' if !in_d => { in_s = !in_s; cur.push(c); }
            '"' if !in_s => { in_d = !in_d; cur.push(c); }
            '(' if !in_s && !in_d => { depth += 1; cur.push(c); }
            ')' if !in_s && !in_d => { depth -= 1; cur.push(c); }
            c if c == delim && depth == 0 && !in_s && !in_d => {
                out.push(std::mem::take(&mut cur));
            }
            _ => cur.push(c),
        }
    }
    out.push(cur);
    out
}

/// Split `expr AS name` / `expr name` into the expression and its output name.
///
/// The alias is the name the CLIENT will look the column up by — SQLAlchemy
/// reads `count(*) AS count_1` back as `count_1`, so dropping the alias and
/// returning a column called `count` hands it a result it cannot find.
fn split_output_alias(p: &str) -> (&str, Option<&str>) {
    let pu = p.to_uppercase();
    if let Some(at) = find_kw(&pu, "AS") {
        let alias = p[at + 2..].trim().trim_matches('"');
        if !alias.is_empty() {
            return (p[..at].trim(), Some(alias));
        }
    }
    // A bare alias: `count(*) count_1`. Only after a closing paren or a plain
    // identifier, and never when the tail is itself part of the expression —
    // so the split point is the LAST whitespace outside any parenthesis.
    let b: Vec<char> = p.chars().collect();
    let mut depth = 0i32;
    let mut in_s = false;
    let mut cut = None;
    for (i, &c) in b.iter().enumerate() {
        match c {
            '\'' => in_s = !in_s,
            '(' if !in_s => depth += 1,
            ')' if !in_s => depth -= 1,
            c if c.is_whitespace() && depth == 0 && !in_s => cut = Some(i),
            _ => {}
        }
    }
    match cut {
        Some(i) => {
            let alias = p[i..].trim().trim_matches('"');
            if alias.is_empty() { (p, None) } else { (p[..i].trim(), Some(alias)) }
        }
        None => (p, None),
    }
}

/// One string, in NQL's spelling — double-quoted, inner quotes escaped.
///
/// These values arrive already UNQUOTED from the SQL parser, so they cannot be
/// pasted into an NQL query as-is: a value containing `"` would close the
/// literal early and the rest of it would be parsed as grammar. Which is the
/// shape of an injection, not merely a syntax error.
fn nql_string(s: &str) -> String {
    format!("\"{}\"", s.replace('\\', "\\\\").replace('"', "\\\""))
}

fn sql_literals_to_nql(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut it = s.chars().peekable();
    while let Some(c) = it.next() {
        match c {
            '\'' => {
                out.push('"');
                while let Some(ch) = it.next() {
                    if ch == '\'' {
                        if it.peek() == Some(&'\'') {
                            it.next();
                            out.push('\''); // doubled '' is one literal quote
                        } else {
                            break;
                        }
                    } else if ch == '"' {
                        // A double quote inside a SQL literal must be escaped
                        // for NQL, whose lexer collapses \" to a literal quote.
                        out.push('\\');
                        out.push('"');
                    } else {
                        out.push(ch);
                    }
                }
                out.push('"');
            }
            '<' if it.peek() == Some(&'>') => { it.next(); out.push_str("!="); }
            _ => out.push(c),
        }
    }
    out
}

fn strip_prefix_ci(s: &str, prefix: &str) -> Option<String> {
    if s.len() >= prefix.len() && s[..prefix.len()].eq_ignore_ascii_case(prefix) {
        Some(s[prefix.len()..].trim_start().to_string())
    } else {
        None
    }
}

/// Find a top-level keyword (not inside quotes or parentheses), returning its
/// byte offset. Case-insensitive, and only matches on word boundaries.
fn find_kw(s: &str, kw: &str) -> Option<usize> {
    let bytes = s.as_bytes();
    let k = kw.as_bytes();
    let mut depth = 0i32;
    let mut in_s = false;
    let mut in_d = false;
    let mut i = 0usize;
    while i < bytes.len() {
        let c = bytes[i];
        if in_s { if c == b'\'' { in_s = false; } i += 1; continue; }
        if in_d { if c == b'"' { in_d = false; } i += 1; continue; }
        match c {
            b'\'' => { in_s = true; i += 1; continue; }
            b'"' => { in_d = true; i += 1; continue; }
            b'(' => { depth += 1; i += 1; continue; }
            b')' => { depth -= 1; i += 1; continue; }
            _ => {}
        }
        if depth == 0 && i + k.len() <= bytes.len()
            && bytes[i..i + k.len()].eq_ignore_ascii_case(k)
        {
            let before_ok = i == 0 || !(bytes[i - 1] as char).is_alphanumeric() && bytes[i - 1] != b'_';
            let after = i + k.len();
            let after_ok = after >= bytes.len()
                || !(bytes[after] as char).is_alphanumeric() && bytes[after] != b'_';
            if before_ok && after_ok {
                return Some(i);
            }
        }
        i += 1;
    }
    None
}

/// Split a comma-separated list at the TOP level, ignoring commas inside
/// quotes or parentheses — so `VALUES (1, 'a,b'), (2, 'c')` splits into two
/// groups and not four.
fn split_top(s: &str, sep: char) -> Vec<String> {
    let mut out = vec![];
    let mut cur = String::new();
    let mut depth = 0i32;
    let mut in_s = false;
    let mut it = s.chars().peekable();
    while let Some(c) = it.next() {
        if in_s {
            cur.push(c);
            if c == '\'' {
                // A doubled '' is an escaped quote, not the end of the literal.
                if it.peek() == Some(&'\'') { cur.push(it.next().unwrap()); } else { in_s = false; }
            }
            continue;
        }
        match c {
            '\'' => { in_s = true; cur.push(c); }
            '(' => { depth += 1; cur.push(c); }
            ')' => { depth -= 1; cur.push(c); }
            x if x == sep && depth == 0 => { out.push(cur.trim().to_string()); cur.clear(); }
            _ => cur.push(c),
        }
    }
    if !cur.trim().is_empty() { out.push(cur.trim().to_string()); }
    out
}

/// Parse one SQL scalar literal into JSON.
///
/// Deliberately narrow: a string, a number, a boolean, or NULL. Anything else
/// — a function call, an expression, a cast — is refused by name rather than
/// coerced into a string that would silently store the wrong value.
fn sql_value(raw: &str) -> Result<Value, String> {
    let t = raw.trim();
    if t.is_empty() {
        return Err("empty value".into());
    }
    let up = t.to_uppercase();
    if up == "NULL" { return Ok(Value::Null); }
    if up == "TRUE" { return Ok(Value::Bool(true)); }
    if up == "FALSE" { return Ok(Value::Bool(false)); }
    if t.starts_with('\'') && t.ends_with('\'') && t.len() >= 2 {
        // Unwrap, collapsing the SQL '' escape to one quote.
        let inner = &t[1..t.len() - 1];
        return Ok(Value::String(inner.replace("''", "'")));
    }
    if let Ok(i) = t.parse::<i64>() { return Ok(Value::from(i)); }
    if let Ok(f) = t.parse::<f64>() { return Ok(Value::from(f)); }
    Err(format!(
        "cannot use {:?} as a value — this endpoint accepts string literals, \
         numbers, TRUE/FALSE and NULL. Expressions, casts and function calls \
         are not evaluated, because storing an unevaluated expression as text \
         would be worse than refusing it", t))
}

/// Pull a trailing `RETURNING …` off a statement, returning (head, columns).
fn split_returning(tail: &str) -> (String, Vec<Col>) {
    let tu = tail.to_uppercase();
    match find_kw(&tu, "RETURNING") {
        None => (tail.to_string(), vec![]),
        Some(at) => {
            let head = tail[..at].trim().to_string();
            let list = tail[at + "RETURNING".len()..].trim();
            if list == "*" {
                return (head, vec![]);   // empty projection = every column
            }
            let cols = split_top(list, ',')
                .into_iter()
                .map(|p| {
                    let raw = p.split_whitespace().next().unwrap_or(&p).to_string();
                    let name = raw.rsplit('.').next().unwrap_or(&raw).trim_matches('"').to_string();
                    Col::same(&name)
                })
                .collect();
            (head, cols)
        }
    }
}

/// Columns whose names are reserved: they carry provenance rather than data.
fn take_reserved(doc: &mut serde_json::Map<String, Value>) -> (Option<String>, Vec<String>, Option<String>, Option<String>) {
    let id = doc.remove("_id").or_else(|| doc.remove("id"))
        .and_then(|v| match v {
            Value::String(s) => Some(s),
            Value::Null => None,
            other => Some(other.to_string()),   // a numeric key is a fine id
        });
    let caused_by = match doc.remove("_caused_by") {
        Some(Value::String(s)) => vec![s],
        Some(Value::Array(a)) => a.into_iter()
            .filter_map(|v| v.as_str().map(str::to_string)).collect(),
        _ => vec![],
    };
    let vf = doc.remove("_valid_from").and_then(|v| v.as_str().map(str::to_string));
    let vt = doc.remove("_valid_to").and_then(|v| v.as_str().map(str::to_string));
    (id, caused_by, vf, vt)
}

/// `INSERT INTO coll (c1, c2) VALUES (v1, v2), (…) [RETURNING …]`
fn translate_insert(sql: &str) -> Result<Stmt, String> {
    let rest = strip_prefix_ci(sql, "INSERT")
        .and_then(|r| strip_prefix_ci(&r, "INTO"))
        .ok_or("expected INSERT INTO")?;
    // Locate VALUES first. Everything before it is `coll (col, …)`; searching
    // for `(` without that bound finds the VALUES parenthesis instead and
    // swallows the keyword into the collection name.
    let ru = rest.to_uppercase();
    let values_at = find_kw(&ru, "VALUES").ok_or(
        "expected VALUES — `INSERT … SELECT` is not supported on this endpoint")?;
    let head = rest[..values_at].trim().to_string();
    let open = head.find('(').ok_or(
        "INSERT needs an explicit column list — `INSERT INTO t (a, b) VALUES (…)`. \
         NEDB is schemaless, so there is no declared column order to infer from")?;
    let coll = head[..open].trim().trim_matches('"');
    let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
    if coll.is_empty() {
        return Err("expected a collection name after INSERT INTO".into());
    }
    let close = head.rfind(')').ok_or("unterminated column list")?;
    if close < open {
        return Err("malformed column list".into());
    }
    let tail_from_values = rest[values_at..].to_string();
    let cols: Vec<String> = split_top(&head[open + 1..close], ',')
        .into_iter()
        .map(|c| c.trim().trim_matches('"').to_string())
        .collect();
    if cols.is_empty() {
        return Err("the column list is empty".into());
    }

    let after = strip_prefix_ci(&tail_from_values, "VALUES")
        .ok_or("expected VALUES after the column list")?;
    let (values_part, returning) = split_returning(&after);

    let mut rows = vec![];
    for group in split_top(&values_part, ',') {
        let g = group.trim();
        if !(g.starts_with('(') && g.ends_with(')')) {
            return Err(format!("expected a parenthesised row of values, got {:?}", g));
        }
        let vals = split_top(&g[1..g.len() - 1], ',');
        if vals.len() != cols.len() {
            return Err(format!(
                "{} values for {} columns — every row must match the column list",
                vals.len(), cols.len()));
        }
        let mut doc = serde_json::Map::new();
        for (c, v) in cols.iter().zip(vals.iter()) {
            doc.insert(c.clone(), sql_value(v)?);
        }
        let (id, caused_by, valid_from, valid_to) = take_reserved(&mut doc);
        rows.push(InsertRow { id, doc, caused_by, valid_from, valid_to });
    }
    if rows.is_empty() {
        return Err("INSERT with no rows".into());
    }
    Ok(Stmt::Insert { coll, rows, returning })
}

/// `UPDATE coll SET a = 1, b = 'x' [WHERE …] [RETURNING …]`
fn translate_update(sql: &str) -> Result<Stmt, String> {
    let rest = strip_prefix_ci(sql, "UPDATE").ok_or("expected UPDATE")?;
    let ru = rest.to_uppercase();
    let set_at = find_kw(&ru, "SET").ok_or("expected SET in UPDATE")?;
    // `UPDATE orders o SET …` — Postgres allows an alias here, and taking the
    // whole span as the collection name made it part of the name ("orders o").
    let target = rest[..set_at].trim();
    let mut parts = target.split_whitespace();
    let coll = parts.next().unwrap_or("").trim_matches('"');
    let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
    let upd_alias: Option<String> = match parts.next() {
        Some(w) if w.eq_ignore_ascii_case("AS") => {
            parts.next().map(|a| a.trim_matches('"').to_string())
        }
        Some(w) => Some(w.trim_matches('"').to_string()),
        None => None,
    };
    if coll.is_empty() {
        return Err("expected a collection name after UPDATE".into());
    }
    let after_set = rest[set_at + 3..].trim().to_string();
    let (after_set, returning) = split_returning(&after_set);

    // WHERE ends the assignment list; everything after it is a NQL predicate.
    let au = after_set.to_uppercase();
    let (assigns_raw, where_raw) = match find_kw(&au, "WHERE") {
        Some(at) => (after_set[..at].to_string(), after_set[at..].to_string()),
        None => (after_set.clone(), String::new()),
    };

    let mut set = vec![];
    for a in split_top(&assigns_raw, ',') {
        let eq = a.find('=').ok_or(format!("expected `col = value` in SET, got {:?}", a))?;
        let col = a[..eq].trim().trim_matches('"').to_string();
        if col.is_empty() {
            return Err("empty column name in SET".into());
        }
        set.push((col, sql_value(&a[eq + 1..])?));
    }
    if set.is_empty() {
        return Err("UPDATE with no assignments".into());
    }
    // The matching rows are found with an ordinary NQL read, so the whole
    // predicate surface (IN, BETWEEN, LIKE, OR, …) works in an UPDATE too.
    let where_raw = strip_column_qualifiers(where_raw.trim(), &coll, upd_alias.as_deref())?;
    let nql = format!("FROM {} {}", coll, sql_literals_to_nql(&where_raw))
        .trim().to_string();
    Ok(Stmt::Update { coll, set, nql, returning })
}

/// `DELETE FROM coll [WHERE …] [RETURNING …]`
fn translate_delete(sql: &str) -> Result<Stmt, String> {
    let rest = strip_prefix_ci(sql, "DELETE")
        .and_then(|r| strip_prefix_ci(&r, "FROM"))
        .ok_or("expected DELETE FROM")?;
    let (rest, returning) = split_returning(&rest);
    let end = rest.find(' ').unwrap_or(rest.len());
    let coll = rest[..end].trim().trim_matches('"');
    let coll = coll.rsplit('.').next().unwrap_or(coll).to_string();
    if coll.is_empty() {
        return Err("expected a collection name after DELETE FROM".into());
    }
    let (del_alias, where_raw) = split_table_alias(rest[end..].trim());
    let where_raw = strip_column_qualifiers(where_raw, &coll, del_alias.as_deref())?;
    let nql = format!("FROM {} {}", coll, sql_literals_to_nql(&where_raw))
        .trim().to_string();
    Ok(Stmt::Delete { coll, nql, returning })
}

/// Translate one SQL statement into something executable, or explain why not.
pub fn translate(sql_raw: &str) -> Result<Stmt, String> {
    let sql = normalise(sql_raw);
    let sql = sql.trim().trim_end_matches(';').trim();
    if sql.is_empty() {
        return Ok(Stmt::Ok(""));
    }
    let upper = sql.to_uppercase();

    // ── the handshake. Clients issue these before anything useful; answering
    // them with plausible values is the difference between "connects" and
    // "hangs on startup". They are canned on purpose — NEDB has no pg_catalog
    // and pretending otherwise would be worse than a clear boundary.
    if upper.starts_with("SET ") || upper.starts_with("BEGIN") || upper.starts_with("COMMIT")
        || upper.starts_with("ROLLBACK") || upper.starts_with("DISCARD")
        || upper.starts_with("LISTEN ") || upper.starts_with("UNLISTEN ")
    {
        // Accepted and ignored: there is one implicit read-only transaction.
        return Ok(Stmt::Ok(if upper.starts_with("SET") { "SET" } else { "OK" }));
    }
    if upper.starts_with("SHOW ") {
        let name = sql[5..].trim().to_lowercase();
        let val = match name.as_str() {
            "transaction_isolation" | "default_transaction_isolation" => "read committed",
            "server_version" => SERVER_VERSION,
            "server_encoding" | "client_encoding" => "UTF8",
            "standard_conforming_strings" => "on",
            "is_superuser" => "off",
            _ => "",
        };
        return Ok(Stmt::Canned { cols: vec![name], row: vec![val.to_string()] });
    }
    if upper == "SELECT VERSION()" {
        return Ok(Stmt::Canned {
            cols: vec!["version".into()],
            row: vec![full_version_string()],
        });
    }
    if upper == "SELECT 1" || upper == "SELECT 1;" {
        return Ok(Stmt::Canned { cols: vec!["?column?".into()], row: vec!["1".into()] });
    }
    if upper.starts_with("SELECT CURRENT_SCHEMA") {
        return Ok(Stmt::Canned { cols: vec!["current_schema".into()], row: vec!["public".into()] });
    }
    if upper.starts_with("SELECT CURRENT_DATABASE") {
        return Ok(Stmt::Canned { cols: vec!["current_database".into()], row: vec!["nedb".into()] });
    }
    if upper.starts_with("SELECT CURRENT_USER") || upper.starts_with("SELECT USER") {
        return Ok(Stmt::Canned { cols: vec!["current_user".into()], row: vec!["nedb".into()] });
    }

    // ── writes ───────────────────────────────────────────────────────────────
    // SQL's write semantics and NEDB's append-only model line up, so these are
    // first-class rather than refused. See the `Stmt` doc comment.
    if upper.starts_with("INSERT") { return translate_insert(sql); }
    if upper.starts_with("UPDATE") { return translate_update(sql); }
    if upper.starts_with("DELETE") { return translate_delete(sql); }

    // ── the refusals that remain, each naming the boundary ──────────────────
    for (kw, why) in [
        ("CREATE", "DDL is not supported — collections are created implicitly by the first write to them, because NEDB is schemaless"),
        ("ALTER", "DDL is not supported — there is no schema to alter"),
        ("DROP", "DDL is not supported; drop a database with DELETE /v1/databases/<db>"),
        ("TRUNCATE", "not supported, and not an oversight: NEDB is append-only so that history cannot be discarded. That is the product"),
        ("COPY", "not supported; use GET /v1/databases/<db>/since for bulk export"),
        ("GRANT", "there is no SQL-level privilege system; auth is the bearer token"),
        ("REVOKE", "there is no SQL-level privilege system; auth is the bearer token"),
    ] {
        if upper.starts_with(kw) {
            return Err(format!("{} is not supported — {}", kw, why));
        }
    }
    if !upper.starts_with("SELECT") {
        return Err(format!(
            "only SELECT, INSERT, UPDATE and DELETE are supported on the Postgres \
             endpoint (got {:?})",
            sql.split_whitespace().next().unwrap_or("")
        ));
    }
    for (kw, why) in [
        (" JOIN ", "JOIN is not supported — NQL is single-collection; join in your client or model the relation with LINK/TRAVERSE"),
        (" UNION ", "UNION is not supported"),
        (" INTERSECT ", "INTERSECT is not supported"),
        (" EXCEPT ", "EXCEPT is not supported"),
        (" OVER (", "window functions are not supported"),
        ("DISTINCT ", "DISTINCT is not supported — GROUP BY <col> gives the distinct values with counts"),
    ] {
        if upper.contains(kw) {
            return Err(why.to_string());
        }
    }
    if find_kw(&upper, "FROM").is_none() {
        return Err("SELECT without FROM is not supported on this endpoint".into());
    }

    // ── SELECT <projection> FROM <rest> ──────────────────────────────────────
    let after_select = strip_prefix_ci(sql, "SELECT").ok_or("expected SELECT")?;
    let from_at = find_kw(&after_select.to_uppercase(), "FROM")
        .ok_or("expected FROM after the select list")?;
    let projection = after_select[..from_at].trim().to_string();
    let rest = after_select[from_at + 4..].trim().to_string();
    if rest.is_empty() {
        return Err("expected a collection name after FROM".into());
    }
    // ── the one derived table with a provable flat equivalent ───────────────
    //
    // `SELECT count(*) FROM (SELECT … FROM coll WHERE …) AS anon` is what
    // EVERY ORM emits for `.count()` — SQLAlchemy's `Query.count()` wraps the
    // whole query in a subquery unconditionally. Refusing it means "SQLAlchemy
    // works, except counting", which is not a boundary anyone would accept.
    //
    // Counting a derived table whose rows are exactly the inner query's rows
    // is counting the inner query, so the rewrite is an IDENTITY rather than
    // an approximation. Each guard below names a construct that would break
    // that identity, and anything carrying one is still refused:
    //
    //   * `LIMIT` / `OFFSET`   — caps the row count before it is counted
    //   * `DISTINCT`           — collapses duplicates, so the counts differ
    //   * `GROUP BY`           — the inner rows ARE the groups
    //   * an inner aggregate   — already one row, counting it answers 1
    //   * anything but `count(*)` outside — the outer list would need the
    //     inner columns, which a flat count cannot supply
    if rest.starts_with('(') {
        if let Some(flat) = flatten_count_of_subquery(&projection, &rest) {
            // Recurses ONCE at most: the rewrite is only produced when the
            // inner FROM names a real collection, so the flat statement can
            // never re-enter this branch.
            return translate(&flat);
        }
        return Err("subqueries in FROM are not supported — except \
                    `SELECT count(*) FROM (…)`, which is rewritten to a flat \
                    count when the inner query has no LIMIT, OFFSET, DISTINCT, \
                    GROUP BY or aggregate of its own (any of those would make the \
                    two counts different numbers)".into());
    }
    let coll_end = rest.find(' ').unwrap_or(rest.len());
    let coll = &rest[..coll_end];
    if coll.contains(',') {
        return Err("selecting from more than one collection is not supported (no JOIN)".into());
    }
    // Postgres clients often qualify as schema.table; NEDB has one namespace,
    // so the schema is dropped — EXCEPT for `information_schema`, whose table
    // names (`tables`, `columns`) are words a user could plausibly name a
    // collection. Keeping the qualifier there is what stops
    // `SELECT * FROM information_schema.tables` and a real collection called
    // `tables` from resolving to the same thing.
    let bare = coll.rsplit('.').next().unwrap_or(coll).trim_matches('"');
    let qualified = coll
        .split('.')
        .map(|p| p.trim_matches('"'))
        .collect::<Vec<_>>()
        .join(".");
    let coll = if qualified.starts_with("information_schema.") {
        qualified.as_str()
    } else {
        bare
    };
    let tail = rest[coll_end..].trim();

    // ── the select list ──────────────────────────────────────────────────────
    //
    // Parsed ITEM BY ITEM, which is what lets a list MIX plain columns with an
    // aggregate — and that mixture is exactly what a `GROUP BY` query is.
    // SQLAlchemy writes `SELECT orders.status, count(*) AS count_1 FROM orders
    // GROUP BY orders.status` for the most ordinary grouped query there is,
    // and the previous check refused any list containing a parenthesis at all,
    // so the whole shape was unreachable even though NQL expresses it
    // natively.
    //
    // NQL's grouped row carries the group key, `count`, and at most one NAMED
    // aggregate — so `count(*)` is always available and one of SUM/AVG/MIN/MAX
    // may join it. A second named aggregate is refused by name rather than
    // silently dropped.
    let mut agg_clause = String::new();
    let mut agg_srcs: Vec<String> = vec![];
    let mut project: Vec<Col> = vec![];

    if projection == "*" {
        // everything
    } else {
        for part in split_top_level(&projection, ',') {
            let p = part.trim();
            if p.is_empty() {
                return Err("empty column in the select list".into());
            }
            let (expr, alias) = split_output_alias(p);
            let eu = expr.to_uppercase();

            // COUNT(*) and COUNT(col) both become NQL's bare COUNT: NQL counts
            // the group, and a per-column non-null count is not expressible.
            if eu.starts_with("COUNT(") {
                if agg_clause.is_empty() {
                    agg_clause = " COUNT".to_string();
                }
                agg_srcs.push("count".to_string());
                project.push(Col::renamed("count", alias.unwrap_or("count")));
                continue;
            }
            if let Some(agg) = ["SUM", "AVG", "MIN", "MAX"]
                .iter()
                .find(|a| eu.starts_with(&format!("{}(", a)))
            {
                let inner = expr[agg.len() + 1..].trim_end_matches(')').trim();
                if inner.is_empty() || inner == "*" {
                    return Err(format!("{}() needs a column", agg));
                }
                let inner = inner.rsplit('.').next().unwrap_or(inner).trim_matches('"');
                let named = format!("{} {}", agg, inner);
                if !agg_clause.is_empty() && agg_clause.trim() != "COUNT" && agg_clause.trim() != named {
                    return Err(format!(
                        "only one of SUM/AVG/MIN/MAX is supported per statement \
                         (already have {:?}, then {:?}) — NQL's grouped row carries \
                         the group key, `count`, and ONE named aggregate",
                        agg_clause.trim(), named));
                }
                agg_clause = format!(" {}", named);
                // NQL emits `<agg>_<field>`; SQL names the column after the
                // function unless the query aliased it.
                let src = format!("{}_{}", agg.to_lowercase(), inner);
                project.push(Col::renamed(&src, alias.unwrap_or(&agg.to_lowercase())));
                agg_srcs.push(src);
                continue;
            }
            // A paren used to be the whole test for "is this an expression",
            // and it let every paren-free one through: `total * 2` became a
            // FIELD NAME, no document had a field called "total * 2", and the
            // column came back blank for every row with no error. Same silent
            // class as the qualified-WHERE bug -- a wrong answer that looks
            // like data. So the test is now the positive one: what survives
            // has to BE a column reference.
            let bare = expr.rsplit('.').next().unwrap_or(expr).trim_matches('"');
            let is_column = !bare.is_empty()
                && !bare.starts_with(|c: char| c.is_ascii_digit())
                && bare.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '$');
            if !is_column {
                return Err(format!(
                    "expressions in the select list are not supported ({:?}) — \
                     supported: *, a column list, COUNT(*), or SUM/AVG/MIN/MAX(col). \
                     Compute it in your client, or read the column and map it there",
                    p));
            }
            let name = bare;
            project.push(Col::renamed(name, alias.unwrap_or(name)));
        }
    }

    // ── clause tail: AS OF SYSTEM TIME → AS OF, then pass the rest through ──
    //
    // The clause keywords NQL shares with SQL (WHERE, GROUP BY, HAVING,
    // ORDER BY, LIMIT, OFFSET) are deliberately handed to the NQL parser
    // unchanged rather than re-parsed here. NQL is the authority on what is
    // valid; re-implementing its grammar would give two parsers to disagree.
    // `FROM orders o WHERE …` — the alias is taken off the tail (NQL has no
    // alias syntax) and then ACCEPTED as a qualifier on the columns.
    let (alias, tail) = split_table_alias(tail);
    let mut tail = strip_column_qualifiers(tail, coll, alias.as_deref())?;
    let tu = tail.to_uppercase();
    if let Some(at) = find_kw(&tu, "AS OF SYSTEM TIME") {
        let before = tail[..at].to_string();
        let after = tail[at + "AS OF SYSTEM TIME".len()..].trim_start().to_string();
        // Take the sequence token; the rest of the tail follows it.
        let end = after.find(' ').unwrap_or(after.len());
        let seq = after[..end].trim().trim_matches('\'').trim_matches('"').to_string();
        if seq.parse::<u64>().is_err() {
            return Err(format!(
                "AS OF SYSTEM TIME takes a NEDB sequence number here, not a timestamp (got {:?}). \
                 NEDB's history is sequence-addressed and never garbage-collected, so a seq is \
                 exact where a wall-clock time would be approximate", seq));
        }
        tail = format!("{} AS OF {} {}", before.trim(), seq, after[end..].trim())
            .trim()
            .to_string();
    }

    // ── ORDER BY <ordinal> → ORDER BY <that select-list column> ─────────────
    //
    // SQL lets a sort key be a POSITION in the select list, and clients write
    // it constantly — `ORDER BY 1, 2` is how psql's own catalogue queries sort,
    // and node-postgres sent `GROUP BY status ORDER BY 1` in the very first
    // run of the driver harness. NQL has no ordinals: it read the `1` as a
    // literal and refused with "expected field name, got Num(1.0)".
    //
    // The projection is already parsed here, so the position resolves to a
    // real field name. An ordinal past the end of the select list, or one used
    // with `SELECT *` where there is no list to index, is refused with the
    // reason — guessing a column would sort by something the query never named.
    let tu_ord = tail.to_uppercase();
    if let Some(ob_at) = find_kw(&tu_ord, "ORDER BY") {
        let start = ob_at + "ORDER BY".len();
        // The clause runs to the next one, or to the end of the tail.
        let end = ["LIMIT", "OFFSET", "GROUP BY", "TRACE", "TRAVERSE", "SEARCH"]
            .iter()
            .filter_map(|k| find_kw(&tu_ord[start..], k).map(|at| start + at))
            .min()
            .unwrap_or(tail.len());
        let mut keys = vec![];
        for item in split_top_level(&tail[start..end], ',') {
            let item = item.trim();
            if item.is_empty() {
                continue;
            }
            let mut parts = item.split_whitespace();
            let first = parts.next().unwrap_or("");
            let rest: Vec<&str> = parts.collect();
            match first.parse::<usize>() {
                Ok(n) if n >= 1 => {
                    let col = project.get(n - 1).ok_or_else(|| {
                        if project.is_empty() {
                            format!(
                                "ORDER BY {} is a select-list POSITION, and `SELECT *` \
                                 has no list to index — name the column instead", n)
                        } else {
                            format!(
                                "ORDER BY {} is out of range: the select list has {} \
                                 column(s)", n, project.len())
                        }
                    })?;
                    keys.push(
                        std::iter::once(col.src.as_str())
                            .chain(rest.iter().copied())
                            .collect::<Vec<_>>()
                            .join(" "),
                    );
                }
                // Not an ordinal — a named column, or `1 + 1`, which NQL will
                // judge for itself.
                _ => keys.push(item.to_string()),
            }
        }
        tail = format!("{} ORDER BY {} {}", &tail[..ob_at], keys.join(", "), &tail[end..])
            .split_whitespace()
            .collect::<Vec<_>>()
            .join(" ");
    }

    // ── GROUP BY: refuse a bare column that SQL would refuse ─────────────────
    //
    // A grouped NQL row holds only the group key, `count` and the aggregate —
    // so projecting `total` from `GROUP BY region` found nothing and rendered
    // NULL. Silently answering NULL for a column the query cannot produce is
    // the exact failure shape this engine keeps getting bitten by, so it is an
    // error, using Postgres's own wording so the message is already familiar.
    let mut gkey: Option<String> = None;
    let tu_all = tail.to_uppercase();
    if let Some(gb_at) = find_kw(&tu_all, "GROUP BY") {
        let head = tail[..gb_at].trim_end().to_string();
        let after = tail[gb_at + "GROUP BY".len()..].trim_start();
        let key_end = after.find(|c: char| c == ' ' || c == ',').unwrap_or(after.len());
        let group_key = after[..key_end].trim().trim_matches('"').to_string();
        let after_key = after[key_end..].trim_start();
        gkey = Some(group_key.clone());

        // NQL groups by ONE field. Taking the first key and leaving the rest
        // in the tail would group by something narrower than the query asked
        // for — more rows than Postgres returns, each aggregating too much.
        if after_key.starts_with(',') {
            return Err(format!(
                "GROUP BY takes one key here (got {:?} and more) — NQL groups by a \
                 single field, and grouping by only the first would aggregate over \
                 rows the query meant to keep apart",
                group_key));
        }

        for c in &project {
            let ok = c.src == group_key
                || c.src == "count"
                || agg_srcs.contains(&c.src);
            if !ok {
                return Err(format!(
                    "column {:?} must appear in the GROUP BY clause or be used in an \
                     aggregate function — a grouped row carries the group key, `count`, \
                     and the aggregate, nothing else",
                    c.src));
            }
        }

        // NQL's aggregate belongs IMMEDIATELY AFTER the group key
        // (`GROUP BY status COUNT`), not after the collection name. Emitting
        // `FROM orders COUNT GROUP BY status` is refused by the NQL parser
        // with "only one aggregate per query" — which is how the most
        // ordinary grouped query an ORM writes still failed even once its
        // select list parsed.
        //
        // `count` rides along free with a named aggregate — an NQL grouped row
        // carries the key, `count` AND the aggregate — so only the named one
        // is emitted when both were asked for.
        tail = format!("{} GROUP BY {}{} {}", head, group_key, agg_clause, after_key)
            .split_whitespace()
            .collect::<Vec<_>>()
            .join(" ");
        agg_clause.clear();
    }

    // ── HAVING <agg> → the spelling NQL's grouped row actually carries ──────
    //
    // NQL's grouped row has fields named `count` and `<agg>_<field>`, and its
    // HAVING matches on those. Every SQL client writes something else:
    //
    //   HAVING count(*) > 1   -> NQL parse error (loud, fine)
    //   HAVING COUNT > 1      -> ZERO ROWS, no error
    //   HAVING n > 1          -> ZERO ROWS, no error  (`n` being the SQL alias)
    //
    // The last two are the dangerous ones: HAVING is advertised as supported,
    // and a filter that silently matches nothing reads as "no groups qualified"
    // rather than "your predicate named a field that does not exist". So the
    // aggregate spellings are translated, and anything left that is not a
    // group-key or aggregate field is refused BY NAME.
    let tu_hav = tail.to_uppercase();
    if let Some(h_at) = find_kw(&tu_hav, "HAVING") {
        let start = h_at + "HAVING".len();
        let end = ["ORDER BY", "LIMIT", "OFFSET"]
            .iter()
            .filter_map(|k| find_kw(&tu_hav[start..], k).map(|at| start + at))
            .min()
            .unwrap_or(tail.len());
        let clause = tail[start..end].to_string();
        // The left-hand side of the first comparison is the key being filtered.
        let lhs_end = clause
            .find(|c: char| "<>=!".contains(c))
            .unwrap_or(clause.len());
        let lhs = clause[..lhs_end].trim();
        if !lhs.is_empty() {
            let lu = lhs.to_uppercase();
            // `count(*)`, `COUNT(*)`, `count`, or the alias the query gave the
            // count -- all mean NQL's `count`.
            // The alias test has to tie THIS column to the count. Asking only
            // "is there a count anywhere in the projection" matched the GROUP
            // BY key too, so `HAVING status > 'a'` -- a perfectly legitimate
            // filter on the group key -- was rewritten into `count > 'a'`.
            let is_count = lu == "COUNT" || lu.replace(' ', "") == "COUNT(*)"
                || project.iter().any(|c| c.out.eq_ignore_ascii_case(lhs) && c.src == "count");
            let mapped = if is_count {
                Some("count".to_string())
            } else {
                // A named aggregate, by its NQL source name or by its alias.
                agg_srcs.iter().find(|s| s.eq_ignore_ascii_case(lhs)).cloned().or_else(|| {
                    project.iter()
                        .find(|c| c.out.eq_ignore_ascii_case(lhs) && agg_srcs.contains(&c.src))
                        .map(|c| c.src.clone())
                })
            };
            match mapped {
                Some(m) => {
                    // The space matters: `count> 1` happens to parse today, but
                    // relying on the tokenizer being forgiving is how a rewrite
                    // breaks the next time the grammar tightens.
                    let rewritten = format!("{} {}", m, clause[lhs_end..].trim());
                    tail = format!("{} HAVING {} {}",
                        tail[..h_at].trim(), rewritten.trim(), tail[end..].trim())
                        .trim().to_string();
                }
                None if gkey.as_deref().map(|g| g.eq_ignore_ascii_case(lhs)) == Some(true) => {}
                None => {
                    return Err(format!(
                        "HAVING names {:?}, which this grouped row does not carry. \
                         It has the group key{}{}. Filtering on anything else would \
                         answer zero rows rather than report a mistake",
                        lhs,
                        gkey.as_deref().map(|g| format!(" ({:?})", g)).unwrap_or_default(),
                        if agg_srcs.is_empty() { String::new() }
                        else { format!(", plus {}", agg_srcs.join(", ")) }));
                }
            }
        }
    }


    let tail = sql_literals_to_nql(&tail);
    let nql = format!("FROM {}{}{}", coll,
                      if agg_clause.is_empty() { String::new() } else { agg_clause },
                      if tail.is_empty() { String::new() } else { format!(" {}", tail) });

    Ok(Stmt::Query { nql: nql.trim().to_string(), project })
}

const SERVER_VERSION: &str = "15.0";

/// The `version()` string, for the SQL engine's `version()` function.
pub fn version_string() -> String {
    full_version_string()
}

fn full_version_string() -> String {
    format!(
        "PostgreSQL {} (NEDB {}) — tamper-evident, append-only, permanent \
         history. SELECT + INSERT/UPDATE/DELETE; an UPDATE is a new version, \
         so prior values stay readable with AS OF SYSTEM TIME.",
        SERVER_VERSION,
        env!("CARGO_PKG_VERSION")
    )
}

// ── result shaping ──────────────────────────────────────────────────────────

/// Pick the column order for a result set.
///
/// With an explicit projection, that order. Otherwise the union of keys across
/// the returned rows — `_`-prefixed provenance columns last, so `psql` shows
/// the user's own fields first and `_hash` does not push `status` off screen.
fn columns_for(rows: &[Value], project: &[Col]) -> Vec<Col> {
    if !project.is_empty() {
        return project.to_vec();
    }
    let mut plain: Vec<String> = vec![];
    let mut meta: Vec<String> = vec![];
    for r in rows {
        if let Value::Object(m) = r {
            for k in m.keys() {
                let target = if k.starts_with('_') { &mut meta } else { &mut plain };
                if !target.contains(k) {
                    target.push(k.clone());
                }
            }
        }
    }
    // The user's own fields keep the DOCUMENT'S order -- `serde_json`'s
    // `preserve_order` is on crate-wide precisely so they can, and Postgres
    // orders `*` by column definition rather than alphabetically. Sorting them
    // here made `SELECT *` answer in a different column order than the SQL
    // evaluator did, so a client reading by POSITION got different columns
    // depending on a deployment flag. Only the provenance block is sorted.
    meta.sort();
    plain.extend(meta);
    plain.into_iter().map(|k| Col::same(&k)).collect()
}

/// The Postgres type of one JSON value.
fn oid_of_value(v: &Value) -> Option<i32> {
    match v {
        Value::Null => None,
        Value::Bool(_) => Some(OID_BOOL),
        Value::Number(n) => Some(if n.is_i64() || n.is_u64() { OID_INT8 } else { OID_FLOAT8 }),
        Value::String(_) => Some(OID_TEXT),
        // Arrays and objects render as their JSON text.
        _ => Some(OID_TEXT),
    }
}

/// Reconcile two observed types for the same column.
///
/// A relational column has one type by construction. A NEDB collection does
/// not: document 1 may hold `qty: 3` and document 2 `qty: "three"`. Widening
/// to `text` on a conflict is the only answer that can carry both, and mixed
/// integers and floats widen to float8 for the same reason.
fn unify_oid(a: i32, b: i32) -> i32 {
    if a == b {
        return a;
    }
    match (a, b) {
        (OID_INT8, OID_FLOAT8) | (OID_FLOAT8, OID_INT8) => OID_FLOAT8,
        _ => OID_TEXT,
    }
}

/// The type of `col` across EVERY row in the result, not just the first.
///
/// Taking the first non-null value's type was a latent wrong answer: a column
/// holding `3` in row one and `"n/a"` in row two was advertised as `int8`, and
/// a client that believes the description then fails parsing `"n/a"` as an
/// integer — or, on the binary path, cannot be sent the value at all.
/// Public alias so `pgcatalog` types a column EXACTLY as the wire does.
///
/// The catalogue reporting `bigint` for a column the protocol then sends as
/// text would be a self-contradiction a client is entitled to trust, so both
/// go through this one function rather than two that agree today.
pub fn oid_for_column(rows: &[Value], col: &str) -> i32 {
    oid_for(rows, col)
}

fn oid_for(rows: &[Value], col: &str) -> i32 {
    let mut acc: Option<i32> = None;
    for r in rows {
        if let Some(o) = r.get(col).and_then(oid_of_value) {
            acc = Some(match acc {
                None => o,
                Some(prev) => unify_oid(prev, o),
            });
            if acc == Some(OID_TEXT) {
                break; // text absorbs everything; no need to look further
            }
        }
    }
    acc.unwrap_or(OID_TEXT)
}

/// Render one cell in the text format Postgres clients expect for format 0.
fn cell(v: Option<&Value>) -> Option<String> {
    match v {
        None | Some(Value::Null) => None, // NULL on the wire
        Some(Value::String(s)) => Some(s.clone()),
        Some(Value::Bool(b)) => Some(if *b { "t".into() } else { "f".into() }),
        Some(other) => Some(other.to_string()),
    }
}

/// Render one cell in binary format for the type the column was advertised as.
///
/// Needed because asyncpg asks for binary results — it is not an optimisation
/// there, it is the only format it requests, so without this it cannot read a
/// single row. Text-format clients never reach this path.
///
/// A value that does not fit the advertised type is an error rather than a
/// coercion. The advertised type comes from sampling stored documents, so a
/// mismatch means the field is genuinely heterogeneous beyond the sample, and
/// quietly sending a zero (or the text bytes under a binary header) would
/// corrupt the value in a way the client cannot detect.
fn cell_binary(v: Option<&Value>, oid: i32) -> Result<Option<Vec<u8>>, String> {
    let v = match v {
        None | Some(Value::Null) => return Ok(None),
        Some(v) => v,
    };
    let as_f64 = |n: &serde_json::Number| n.as_f64()
        .ok_or_else(|| "a number too large to send as float8".to_string());
    Ok(Some(match (oid, v) {
        (OID_BOOL, Value::Bool(b)) => vec![u8::from(*b)],
        (OID_INT2, Value::Number(n)) => {
            let i = n.as_i64().ok_or("not an integer")?;
            i16::try_from(i).map_err(|_| format!("{} does not fit in int2", i))?
                .to_be_bytes().to_vec()
        }
        (OID_INT4, Value::Number(n)) => {
            let i = n.as_i64().ok_or("not an integer")?;
            i32::try_from(i).map_err(|_| format!("{} does not fit in int4", i))?
                .to_be_bytes().to_vec()
        }
        (OID_INT8, Value::Number(n)) => {
            n.as_i64().ok_or("not an integer")?.to_be_bytes().to_vec()
        }
        (OID_FLOAT4, Value::Number(n)) => (as_f64(n)? as f32).to_be_bytes().to_vec(),
        (OID_FLOAT8, Value::Number(n)) => as_f64(n)?.to_be_bytes().to_vec(),
        // For the text family, binary and text are the same bytes.
        (OID_TEXT | OID_VARCHAR | OID_NAME | OID_UNKNOWN | OID_JSON, _) => {
            cell(Some(v)).unwrap_or_default().into_bytes()
        }
        // jsonb is a one-byte version header then the JSON text.
        (OID_JSONB, _) => {
            let mut b = vec![1u8];
            b.extend_from_slice(cell(Some(v)).unwrap_or_default().as_bytes());
            b
        }
        (oid, val) => {
            let kind = match val {
                Value::Bool(_) => "a boolean",
                Value::Number(_) => "a number",
                Value::String(_) => "a string",
                Value::Array(_) => "an array",
                _ => "an object",
            };
            return Err(format!(
                "cannot send {} in binary format as type OID {} — the field holds \
                 more than one type across documents, so it cannot be described \
                 by a single Postgres type. Select it with a text cast, or use a \
                 text-format client",
                kind, oid
            ));
        }
    }))
}

/// A `RowDescription`, with a per-column wire format code.
fn row_description_fmt(cols: &[Col], oids: &[i32], fmts: &[i16]) -> Vec<u8> {
    let mut m = Out::msg(b'T');
    m.i16(cols.len() as i16);
    for (i, c) in cols.iter().enumerate() {
        m.cstr(&c.out);
        m.i32(0); // table OID — unknown
        m.i16((i + 1) as i16); // column attribute number
        m.i32(oids.get(i).copied().unwrap_or(OID_TEXT));
        m.i16(-1); // variable length
        m.i32(-1); // no type modifier
        m.i16(fmts.get(i).copied().unwrap_or(0));
    }
    m.finish()
}

fn row_description(cols: &[Col], oids: &[i32]) -> Vec<u8> {
    row_description_fmt(cols, oids, &[])
}

fn data_row_bytes(vals: &[Option<Vec<u8>>]) -> Vec<u8> {
    let mut m = Out::msg(b'D');
    m.i16(vals.len() as i16);
    for v in vals {
        match v {
            None => m.i32(-1),
            Some(b) => {
                m.i32(b.len() as i32);
                m.bytes(b);
            }
        }
    }
    m.finish()
}

fn data_row(vals: &[Option<String>]) -> Vec<u8> {
    let owned: Vec<Option<Vec<u8>>> =
        vals.iter().map(|v| v.as_ref().map(|s| s.as_bytes().to_vec())).collect();
    data_row_bytes(&owned)
}

/// Encode just the rows: `T` followed by one `D` per row, and NO
/// `CommandComplete`.
///
/// Split out because a write with `RETURNING` must emit `T`/`D`* and then its
/// OWN tag (`INSERT 0 3`, `UPDATE 1`). The first cut called `encode_result`
/// there, which appends `CommandComplete("SELECT n")` — so one statement sent
/// TWO CommandComplete messages. That is a protocol violation, and the visible
/// symptom was `RETURNING` silently yielding no rows at all: the client took
/// the first tag as the end of the statement and discarded the description.
pub fn encode_rows(rows: &[Value], project: &[Col]) -> Vec<u8> {
    let cols = columns_for(rows, project);
    let oids: Vec<i32> = cols.iter().map(|c| oid_for(rows, &c.src)).collect();
    let mut out = row_description(&cols, &oids);
    for r in rows {
        let vals: Vec<Option<String>> = cols.iter().map(|c| cell(r.get(&c.src))).collect();
        out.extend_from_slice(&data_row(&vals));
    }
    out
}

/// A complete SELECT response: rows plus `CommandComplete("SELECT n")`.
pub fn encode_result(rows: &[Value], project: &[Col]) -> Vec<u8> {
    let mut out = encode_rows(rows, project);
    out.extend_from_slice(&command_complete(&format!("SELECT {}", rows.len())));
    out
}

// ── the extended query protocol: Parse / Bind / Describe / Execute ──────────
//
// Why this exists at all: psycopg3, asyncpg and the JDBC driver do not speak
// the simple query protocol for parameterised statements. Without these six
// messages they cannot run a single query — psycopg3 hangs waiting for a
// `ParseComplete`, and asyncpg refuses before it ever sends a `Bind`. "psql
// works" is not the same as "the drivers your evaluators use work".
//
// Two facts about real drivers shaped everything below, and both were read off
// a wire transcript rather than assumed:
//
//   1. psycopg3 sends parameters in a MIXED format — a `str` as OID 0 in text
//      format, but an `int` as int2/int4/int8 in BINARY, a float as float8
//      binary, a bool as a single binary byte. A text-only decoder gets `\x00*`
//      where it expected `42`.
//
//   2. asyncpg declares NO parameter types in `Parse` and then asks
//      `Describe(statement)`, encoding its arguments from whatever OIDs come
//      back. Answering "text" for all of them does not degrade gracefully — it
//      makes asyncpg REFUSE the call client-side ("expected str, got int").
//
// (2) is the reason `infer_param_oids` exists. NEDB is schemaless, so there is
// no catalogue to read a column's type out of — the only honest source of truth
// is the data already stored, so the type is sampled from it.

/// Parameter/result type OIDs handled on the binary path.
const OID_INT2: i32 = 21;
const OID_INT4: i32 = 23;
const OID_OID: i32 = 26;
const OID_FLOAT4: i32 = 700;
const OID_VARCHAR: i32 = 1043;
const OID_NAME: i32 = 19;
const OID_UNKNOWN: i32 = 705;
const OID_JSON: i32 = 114;
const OID_JSONB: i32 = 3802;

/// How many `$n` placeholders a statement carries, and the highest index used.
///
/// Scans outside string literals so a `'$1'` inside a value is not mistaken for
/// a placeholder. Dollar-quoted bodies (`$tag$…$tag$`) are not recognised —
/// they need a procedural language NEDB does not have.
fn param_count(sql: &str) -> usize {
    let b = sql.as_bytes();
    let mut i = 0usize;
    let mut in_s = false;
    let mut max = 0usize;
    while i < b.len() {
        let c = b[i];
        if in_s {
            if c == b'\'' {
                in_s = false;
            }
            i += 1;
            continue;
        }
        if c == b'\'' {
            in_s = true;
            i += 1;
            continue;
        }
        if c == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
            let mut j = i + 1;
            let mut n = 0usize;
            while j < b.len() && b[j].is_ascii_digit() {
                n = n * 10 + (b[j] - b'0') as usize;
                j += 1;
            }
            max = max.max(n);
            i = j;
            continue;
        }
        i += 1;
    }
    max
}

/// The JSON-shaped type of `field` as it is actually stored, sampled from the
/// collection, mapped onto the nearest Postgres OID.
///
/// This is the schemaless answer to "what type is this column?". A relational
/// server reads its catalogue; NEDB has none, so it reads the data. Sampling a
/// bounded number of rows keeps a `Describe` cheap, and the first row that
/// actually carries the field decides — a field missing from row one but
/// present in row nine still types correctly.
fn infer_field_oid(db: Option<&Arc<Db>>, coll: &str, field: &str) -> i32 {
    // `_`-prefixed names are engine metadata, not stored document fields, so
    // they type from the engine's own contract — no sampling, and no database
    // handle needed.
    match field {
        "_seq" => return OID_INT8,
        "_id" | "_hash" | "_prev" | "_collection" | "_valid_from" | "_valid_to" => return OID_TEXT,
        _ => {}
    }
    // A catalogue relation types its own columns. Sampling a USER collection
    // named `pg_type` finds nothing and falls back to text — and asyncpg,
    // which declares parameter types client-side and refuses the call when
    // the server's answer is wrong, then rejected `WHERE oid = $1` with
    // "expected str, got int" before a single byte was sent.
    if !field.is_empty() && crate::pgcatalog::is_catalog(coll) {
        if let Some(rows) = crate::pgcatalog::rows(coll, db) {
            return oid_for(&rows, field);
        }
    }
    let db = match db {
        Some(db) => db,
        None => return OID_TEXT,
    };
    if coll.is_empty() || field.is_empty() {
        return OID_TEXT;
    }
    let rows = match crate::nql::query(db, &format!("FROM {} LIMIT {}", coll, TYPE_SAMPLE)) {
        Ok((rows, _)) => rows,
        Err(_) => return OID_TEXT,
    };
    // Unified over the sample, not taken from the first hit: a field that is a
    // number in one document and a string in another has to be advertised as
    // text or a client cannot decode every row of it.
    oid_for(&rows, field)
}

/// The type of an aggregate output column, which no document holds.
///
/// Sampling stored documents cannot type these: `COUNT(*)` produces a column
/// called `count` that exists in no document, so the sampler finds nothing and
/// falls back to text. A text-format client papers over that, but a binary
/// client is then handed the digits of a number under a text header and
/// `COUNT(*)` comes back as the string `"2"` instead of the integer `2`.
///
/// So aggregates are typed from what the aggregate MEANS: a count is always an
/// integer, an average is always fractional, and min/max/sum inherit the type
/// of the field they were computed over.
fn aggregate_oid(src: &str, db: Option<&Arc<Db>>, coll: &str) -> Option<i32> {
    if src == "count" {
        return Some(OID_INT8);
    }
    for (prefix, fixed) in [
        ("count_", Some(OID_INT8)),
        ("avg_", Some(OID_FLOAT8)),
        ("sum_", None),
        ("min_", None),
        ("max_", None),
    ] {
        if let Some(field) = src.strip_prefix(prefix) {
            return Some(match fixed {
                Some(oid) => oid,
                // SUM/MIN/MAX of an integer field is an integer; of a
                // fractional field, fractional.
                None => match infer_field_oid(db, coll, field) {
                    OID_INT8 => OID_INT8,
                    OID_FLOAT8 => OID_FLOAT8,
                    // Summing or ordering a non-numeric field is not
                    // meaningful; let the row-derived type answer.
                    other => other,
                },
            });
        }
    }
    None
}

/// How many documents to sample when typing a column.
///
/// Bounded so a `Describe` stays cheap. It is a sample, so a field that only
/// turns heterogeneous outside it can still surprise us — which is exactly why
/// `cell_binary` refuses a mismatch loudly instead of coercing.
const TYPE_SAMPLE: usize = 200;

/// The collection a statement reads from or writes to, for type sampling.
fn stmt_collection(sql: &str) -> String {
    let s = normalise(sql);
    let up = s.to_uppercase();
    let after = if let Some(at) = find_kw(&up, "FROM") {
        &s[at + 4..]
    } else if let Some(rest) = strip_prefix_ci(&s, "UPDATE") {
        return rest
            .split_whitespace()
            .next()
            .unwrap_or("")
            .rsplit('.')
            .next()
            .unwrap_or("")
            .trim_matches('"')
            .to_string();
    } else if let Some(rest) = strip_prefix_ci(&s, "INSERT INTO") {
        return rest
            .split(|c: char| c.is_whitespace() || c == '(')
            .find(|t| !t.is_empty())
            .unwrap_or("")
            .rsplit('.')
            .next()
            .unwrap_or("")
            .trim_matches('"')
            .to_string();
    } else {
        return String::new();
    };
    after
        .trim()
        .split(|c: char| c.is_whitespace())
        .find(|t| !t.is_empty())
        .unwrap_or("")
        .rsplit('.')
        .next()
        .unwrap_or("")
        .trim_matches('"')
        .to_string()
}

/// Which document field each `$n` is being compared against.
///
/// Three shapes cover essentially all driver-generated SQL:
///   `WHERE qty > $1`        → the identifier immediately left of the operator
///   `SET status = $1`       → same shape, inside the SET list
///   `INSERT INTO t (a,b) VALUES ($1,$2)` → positional against the column list
///
/// Anything it cannot read returns `None`, which types as `text`. Guessing
/// wrong here would make a driver encode a value the engine then fails to
/// match, so an unknown is left unknown on purpose.
fn param_fields(sql: &str, n_params: usize) -> Vec<Option<String>> {
    let s = normalise(sql);
    let mut out = vec![None; n_params];

    // The INSERT column list maps positionally, which is more reliable than
    // scanning leftwards through a VALUES tuple.
    let up = s.to_uppercase();
    if up.starts_with("INSERT") {
        if let (Some(open), Some(vals_at)) = (s.find('('), find_kw(&up, "VALUES")) {
            if open < vals_at {
                if let Some(close) = s[open..vals_at].rfind(')') {
                    let cols: Vec<String> = split_top(&s[open + 1..open + close], ',')
                        .into_iter()
                        .map(|c| c.trim().trim_matches('"').to_string())
                        .collect();
                    // `$1` is the first placeholder in the first tuple, and so on.
                    let tail = &s[vals_at..];
                    let mut seen = 0usize;
                    let b = tail.as_bytes();
                    let mut i = 0usize;
                    let mut in_s = false;
                    while i < b.len() {
                        if in_s {
                            if b[i] == b'\'' { in_s = false; }
                            i += 1;
                            continue;
                        }
                        if b[i] == b'\'' { in_s = true; i += 1; continue; }
                        if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
                            let mut j = i + 1;
                            let mut num = 0usize;
                            while j < b.len() && b[j].is_ascii_digit() {
                                num = num * 10 + (b[j] - b'0') as usize;
                                j += 1;
                            }
                            if num >= 1 && num <= n_params {
                                if let Some(c) = cols.get(seen % cols.len().max(1)) {
                                    out[num - 1] = Some(c.clone());
                                }
                            }
                            seen += 1;
                            i = j;
                            continue;
                        }
                        i += 1;
                    }
                    return out;
                }
            }
        }
    }

    // Otherwise: for each `$n`, walk left past the operator to the identifier.
    let b = s.as_bytes();
    let mut i = 0usize;
    let mut in_s = false;
    while i < b.len() {
        if in_s {
            if b[i] == b'\'' { in_s = false; }
            i += 1;
            continue;
        }
        if b[i] == b'\'' { in_s = true; i += 1; continue; }
        if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
            let mut j = i + 1;
            let mut num = 0usize;
            while j < b.len() && b[j].is_ascii_digit() {
                num = num * 10 + (b[j] - b'0') as usize;
                j += 1;
            }
            if num >= 1 && num <= n_params {
                let left = &s[..i];
                // Skip the operator characters and whitespace sitting between
                // the identifier and the placeholder.
                let trimmed = left.trim_end_matches(|c: char| {
                    c.is_whitespace() || "=<>!+-*/%(,".contains(c)
                });
                // A word operator (`LIKE`, `IN`, `BETWEEN`, `AND`) also sits
                // between them; step over it to reach the real identifier.
                let mut tok = trimmed
                    .rsplit(|c: char| c.is_whitespace() || c == '(' || c == ',')
                    .find(|t| !t.is_empty())
                    .unwrap_or("")
                    .trim_matches('"');
                let mut before = trimmed;
                for _ in 0..4 {
                    let upper_tok = tok.to_uppercase();
                    // `BETWEEN $1 AND $2` puts BOTH a word operator and an
                    // earlier placeholder between `$2` and the column it
                    // constrains, so a placeholder has to be stepped over too —
                    // otherwise the upper bound of every range query types as
                    // text while the lower bound types correctly.
                    if upper_tok.starts_with('$')
                        || matches!(upper_tok.as_str(),
                        "LIKE" | "ILIKE" | "IN" | "BETWEEN" | "AND" | "OR" | "NOT" | "IS") {
                        before = before[..before.len() - tok.len()].trim_end_matches(|c: char| {
                            c.is_whitespace() || "=<>!(,".contains(c)
                        });
                        tok = before
                            .rsplit(|c: char| c.is_whitespace() || c == '(' || c == ',')
                            .find(|t| !t.is_empty())
                            .unwrap_or("")
                            .trim_matches('"');
                    } else {
                        break;
                    }
                }
                if !tok.is_empty()
                    && tok.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '.')
                    && !tok.chars().next().map(|c| c.is_ascii_digit()).unwrap_or(true)
                {
                    out[num - 1] = Some(tok.rsplit('.').next().unwrap_or(tok).to_string());
                }
            }
            i = j;
            continue;
        }
        i += 1;
    }
    out
}

/// The type of a placeholder sitting in a CLAUSE position rather than beside a
/// column.
///
/// `AS OF SYSTEM TIME $1` has no column to sample — the token to its left is
/// the word `TIME`. Its type comes from the grammar instead, which is both
/// cheaper and more certain than any inference: a system-time bound is a
/// sequence number, a valid-time bound is a date string, and a page bound is an
/// integer. Without this, a parameterised time-travel query typed as text and
/// asyncpg refused to send the integer at all.
fn clause_param_oids(sql: &str, n_params: usize) -> Vec<Option<i32>> {
    let s = normalise(sql);
    let mut out = vec![None; n_params];
    let b = s.as_bytes();
    let mut i = 0usize;
    let mut in_s = false;
    while i < b.len() {
        if in_s {
            if b[i] == b'\'' { in_s = false; }
            i += 1;
            continue;
        }
        if b[i] == b'\'' { in_s = true; i += 1; continue; }
        if b[i] == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
            let mut j = i + 1;
            let mut num = 0usize;
            while j < b.len() && b[j].is_ascii_digit() {
                num = num * 10 + (b[j] - b'0') as usize;
                j += 1;
            }
            if num >= 1 && num <= n_params {
                let left = s[..i].trim_end().to_uppercase();
                // VALID AS OF is checked FIRST: it ends with "AS OF" too, and
                // its argument is a DATE STRING, not a sequence number.
                out[num - 1] = if left.ends_with("VALID AS OF") {
                    Some(OID_TEXT)
                } else if left.ends_with("AS OF SYSTEM TIME")
                    || left.ends_with("FOR SYSTEM_TIME AS OF")
                    || left.ends_with("AS OF")
                    || left.ends_with("LIMIT")
                    || left.ends_with("OFFSET")
                {
                    Some(OID_INT8)
                } else {
                    None
                };
            }
            i = j;
            continue;
        }
        i += 1;
    }
    out
}

/// The OIDs to advertise for `$1..$n`, sampled from stored data.
///
/// `declared` is what the client itself put in `Parse`. A client that states a
/// type is believed — it is about to encode its arguments that way, and second
///-guessing it would break the decode. Only the unspecified slots are inferred.
fn infer_param_oids(sql: &str, declared: &[i32], db: Option<&Arc<Db>>) -> Vec<i32> {
    let n = param_count(sql).max(declared.len());
    if n == 0 {
        return vec![];
    }
    let coll = stmt_collection(sql);
    let fields = param_fields(sql, n);
    let clauses = clause_param_oids(sql, n);
    (0..n)
        .map(|i| match declared.get(i) {
            Some(&oid) if oid != 0 => oid,
            // A clause position knows its own type from the grammar, so it
            // outranks sampling a column that is not even there.
            _ => match clauses[i] {
                Some(oid) => oid,
                None => match &fields[i] {
                    Some(f) => infer_field_oid(db, &coll, f),
                    None => OID_TEXT,
                },
            },
        })
        .collect()
}

/// Decode one bound parameter into the SQL literal text to splice into the
/// statement.
///
/// `None` means SQL NULL. Format 1 is binary — see the module note on psycopg3
/// sending small integers as int2.
fn decode_param(raw: Option<&[u8]>, oid: i32, format: i16) -> Result<Option<String>, String> {
    let bytes = match raw {
        None => return Ok(None),
        Some(b) => b,
    };
    let quote = |s: &str| format!("'{}'", s.replace('\'', "''"));

    if format == 0 {
        let s = String::from_utf8_lossy(bytes).to_string();
        return Ok(Some(match oid {
            OID_BOOL => {
                let t = matches!(s.as_str(), "t" | "true" | "TRUE" | "1" | "yes" | "on");
                if t { "TRUE".into() } else { "FALSE".into() }
            }
            OID_INT2 | OID_INT4 | OID_INT8 | OID_OID | OID_FLOAT4 | OID_FLOAT8 => {
                // Validate rather than trust: an unparseable "number" spliced
                // in bare would become a bare identifier in the NQL text and
                // produce a baffling error far from its cause.
                if s.parse::<f64>().is_ok() { s } else { quote(&s) }
            }
            // OID 0 with text format is psycopg3's `str`. Confirmed on the
            // wire: it declares a real numeric OID whenever the value is a
            // number, so an unspecified text parameter is genuinely a string
            // and quoting it is right rather than a guess.
            _ => quote(&s),
        }));
    }
    if format != 1 {
        return Err(format!("unsupported parameter format code {}", format));
    }

    // ── binary ──────────────────────────────────────────────────────────────
    let need = |n: usize| -> Result<(), String> {
        if bytes.len() == n {
            Ok(())
        } else {
            Err(format!(
                "binary parameter of type OID {} should be {} bytes, got {}",
                oid, n, bytes.len()
            ))
        }
    };
    Ok(Some(match oid {
        OID_BOOL => {
            need(1)?;
            if bytes[0] != 0 { "TRUE".into() } else { "FALSE".into() }
        }
        OID_INT2 => {
            need(2)?;
            i16::from_be_bytes([bytes[0], bytes[1]]).to_string()
        }
        OID_INT4 => {
            need(4)?;
            i32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]).to_string()
        }
        OID_OID => {
            need(4)?;
            u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]).to_string()
        }
        OID_INT8 => {
            need(8)?;
            i64::from_be_bytes(bytes[..8].try_into().unwrap()).to_string()
        }
        OID_FLOAT4 => {
            need(4)?;
            let f = f32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
            fmt_float(f as f64)
        }
        OID_FLOAT8 => {
            need(8)?;
            fmt_float(f64::from_be_bytes(bytes[..8].try_into().unwrap()))
        }
        OID_TEXT | OID_VARCHAR | OID_NAME | OID_UNKNOWN | OID_JSON | 0 => {
            quote(&String::from_utf8_lossy(bytes))
        }
        OID_JSONB => {
            // jsonb binary is a 1-byte version header followed by the JSON text.
            let body = if bytes.first() == Some(&1) { &bytes[1..] } else { bytes };
            quote(&String::from_utf8_lossy(body))
        }
        other => {
            return Err(format!(
                "parameter type OID {} is not supported in binary format — \
                 the supported set is bool, int2/int4/int8, float4/float8, \
                 text/varchar/json/jsonb. Send it as text, or cast it in the \
                 statement",
                other
            ))
        }
    }))
}

/// Render a float without Rust's `inf`/`NaN` spellings leaking into SQL text.
fn fmt_float(f: f64) -> String {
    if f.is_nan() {
        "'NaN'".into()
    } else if f.is_infinite() {
        if f > 0.0 { "'Infinity'".into() } else { "'-Infinity'".into() }
    } else if f.fract() == 0.0 && f.abs() < 1e15 {
        format!("{:.0}", f)
    } else {
        f.to_string()
    }
}

/// Splice decoded parameters into the statement text.
///
/// Textual substitution, deliberately: the whole SQL surface is already a text
/// translation into NQL, so one representation is simpler and cannot disagree
/// with itself. Every value arrives already rendered as a SQL literal by
/// `decode_param`, with embedded quotes doubled, so a parameter cannot break
/// out of its literal and alter the statement's shape.
fn substitute_params(sql: &str, params: &[Option<String>]) -> Result<String, String> {
    let b = sql.as_bytes();
    let mut out = String::with_capacity(sql.len() + 16);
    let mut i = 0usize;
    let mut in_s = false;
    while i < b.len() {
        let c = b[i];
        if in_s {
            out.push(c as char);
            if c == b'\'' { in_s = false; }
            i += 1;
            continue;
        }
        if c == b'\'' {
            in_s = true;
            out.push('\'');
            i += 1;
            continue;
        }
        if c == b'$' && i + 1 < b.len() && b[i + 1].is_ascii_digit() {
            let mut j = i + 1;
            let mut n = 0usize;
            while j < b.len() && b[j].is_ascii_digit() {
                n = n * 10 + (b[j] - b'0') as usize;
                j += 1;
            }
            match params.get(n.wrapping_sub(1)) {
                Some(Some(lit)) => out.push_str(lit),
                Some(None) => out.push_str("NULL"),
                None => {
                    return Err(format!(
                        "bind message supplies {} parameter(s) but the statement uses ${}",
                        params.len(), n
                    ))
                }
            }
            i = j;
            continue;
        }
        out.push(c as char);
        i += 1;
    }
    Ok(out)
}

/// A parsed statement, held for the life of the connection (or until `Close`).
struct Prepared {
    sql: String,
    /// OIDs advertised for `$1..$n` — what `ParameterDescription` reports and
    /// what `Bind` values are decoded as.
    param_oids: Vec<i32>,
    /// The advertised output shape, computed on demand and then reused.
    ///
    /// Lazy because working it out samples stored documents, and a text-format
    /// client that never sends `Describe(statement)` should not pay for a scan
    /// on every `Parse` — psycopg3 parses once per query.
    ///
    /// `Some(None)` means "computed, and this statement returns no rows".
    out_shape: Option<Option<(Vec<Col>, Vec<i32>)>>,
}

/// The output columns and types a statement advertises, computed once.
fn prepared_shape<'a>(
    p: &'a mut Prepared,
    db: Option<&Arc<Db>>,
) -> &'a Option<(Vec<Col>, Vec<i32>)> {
    if p.out_shape.is_none() {
        p.out_shape = Some(describe_shape(&p.sql, db, p.param_oids.len()));
    }
    p.out_shape.as_ref().expect("just filled")
}

/// A bound statement: fully substituted SQL plus, once run, its result.
struct Portal {
    sql: String,
    /// Filled by the first `Describe` or `Execute` and reused afterwards.
    ///
    /// Executing once and streaming from the buffer is what makes a suspended
    /// portal safe: a second `Execute` on a partially-drained `INSERT` must
    /// continue the row stream, not perform the insert again.
    result: Option<PortalResult>,
    /// The output shape, frozen at the first `Describe`/`Execute`.
    ///
    /// A schemaless store derives `SELECT *`'s columns from the rows it found,
    /// which would let a `Describe` and a later `Execute` disagree about the
    /// column count — and a driver that was told three fields and handed two
    /// mis-decodes the row rather than failing loudly. Freezing the shape and
    /// projecting every row onto it makes the result set rectangular, as SQL
    /// promises. The simple protocol keeps the dynamic behaviour, where there
    /// is no `Describe` to contradict.
    frozen: Option<Vec<Col>>,
    /// Result-column format codes requested by `Bind`. Empty = all text.
    formats: Vec<i16>,
    /// The shape this portal's statement advertised, carried over from the
    /// prepared statement when any column is to be sent in BINARY.
    ///
    /// It has to be the ADVERTISED shape rather than one derived from the rows
    /// in hand: asyncpg built its decoders from `Describe`, so re-deriving a
    /// different type here would hand it bytes it cannot read.
    declared: Option<(Vec<Col>, Vec<i32>)>,
}

impl Portal {
    /// The format code for column `i`, following the protocol's shorthands:
    /// no codes means all-text, one code applies to every column.
    fn format_of(&self, i: usize) -> i16 {
        match self.formats.len() {
            0 => 0,
            1 => self.formats[0],
            _ => self.formats.get(i).copied().unwrap_or(0),
        }
    }
    /// The columns and types to advertise and encode with.
    fn shape(&self, r: &PortalResult) -> (Vec<Col>, Vec<i32>) {
        match &self.declared {
            Some((cols, oids)) if self.formats.iter().any(|f| *f == 1) => {
                (cols.clone(), oids.clone())
            }
            _ => {
                let cols = columns_for(&r.rows, &r.project);
                let oids = cols.iter().map(|c| oid_for(&r.rows, &c.src)).collect();
                (cols, oids)
            }
        }
    }
}

struct PortalResult {
    rows: Vec<Value>,
    project: Vec<Col>,
    has_rows: bool,
    tag: String,
    tag_counts_rows: bool,
    /// How many rows have gone out across all `Execute`s on this portal.
    sent: usize,
}

fn parse_complete() -> Vec<u8> { Out::msg(b'1').finish() }
fn bind_complete() -> Vec<u8> { Out::msg(b'2').finish() }
fn close_complete() -> Vec<u8> { Out::msg(b'3').finish() }
fn no_data() -> Vec<u8> { Out::msg(b'n').finish() }
fn portal_suspended() -> Vec<u8> { Out::msg(b's').finish() }

fn parameter_description(oids: &[i32]) -> Vec<u8> {
    let mut m = Out::msg(b't');
    m.i16(oids.len() as i16);
    for o in oids {
        m.i32(*o);
    }
    m.finish()
}

/// Split a NUL-terminated string off the front of a message body.
fn take_cstr(body: &[u8], at: &mut usize) -> String {
    let start = *at;
    while *at < body.len() && body[*at] != 0 {
        *at += 1;
    }
    let s = String::from_utf8_lossy(&body[start..*at]).to_string();
    if *at < body.len() {
        *at += 1; // step over the NUL
    }
    s
}

fn take_i16(body: &[u8], at: &mut usize) -> Result<i16, String> {
    if *at + 2 > body.len() {
        return Err("truncated message".into());
    }
    let v = i16::from_be_bytes([body[*at], body[*at + 1]]);
    *at += 2;
    Ok(v)
}

fn take_i32(body: &[u8], at: &mut usize) -> Result<i32, String> {
    if *at + 4 > body.len() {
        return Err("truncated message".into());
    }
    let v = i32::from_be_bytes([body[*at], body[*at + 1], body[*at + 2], body[*at + 3]]);
    *at += 4;
    Ok(v)
}

/// The field names a collection actually holds, sampled from stored documents.
///
/// The answer to `SELECT *` on a store with no schema. Sorted, because
/// `serde_json`'s map is ordered and both this and the row encoder must agree
/// on column order or the values land under the wrong headings.
fn sample_columns(db: Option<&Arc<Db>>, coll: &str) -> Vec<Col> {
    let db = match db {
        Some(db) => db,
        None => return vec![],
    };
    let rows = match crate::nql::query(db, &format!("FROM {} LIMIT 25", coll)) {
        Ok((rows, _)) => rows,
        Err(_) => return vec![],
    };
    let mut names: Vec<String> = vec![];
    for r in &rows {
        if let Value::Object(m) = r {
            for k in m.keys() {
                if !names.iter().any(|n| n == k) {
                    names.push(k.clone());
                }
            }
        }
    }
    names.sort();
    names.iter().map(|n| Col::same(n)).collect()
}

/// The result shape of a statement, worked out WITHOUT running it.
///
/// Needed for `Describe(statement)`, which arrives before any `Bind` — asyncpg
/// builds its row decoders from the answer. Only the select list is read off
/// the result; nothing touches storage except the type sampling.
///
/// Returns `None` when the statement returns no rows at all (`NoData`).
fn describe_shape(
    sql: &str,
    db: Option<&Arc<Db>>,
    n_params: usize,
) -> Option<(Vec<Col>, Vec<i32>)> {
    let probe = probe_sql(sql, n_params);

    // The SQL evaluator describes its own output. It has to: `translate`
    // cannot parse a catalogue join at all, so without this a `Describe`
    // answered `NoData` — and a client told a SELECT has no output never
    // reads its rows.
    //
    // The probe is EXECUTED here, which is affordable precisely because this
    // path only serves catalogue relations and relation-free select lists.
    // Column types come from the values it actually produced, unified across
    // the rows by the same `oid_for` every other path uses — so a column
    // advertised `int8` is one the wire really encodes as int8.
    if sql_engine_owns(&probe) {
        if let Ok(Some((done, _))) = try_catalog_select(&probe, db) {
            if done.project.is_empty() {
                return None;
            }
            let oids = done
                .project
                .iter()
                .map(|c| oid_for(&done.rows, &c.src))
                .collect();
            return Some((done.project, oids));
        }
    }

    let stmt = translate(&probe).ok()?;
    let coll = stmt_collection(sql);

    let cols = match stmt {
        Stmt::Ok(_) => return None,
        Stmt::Canned { cols, .. } => cols.iter().map(|c| Col::same(c)).collect(),
        Stmt::Query { project, .. } => {
            if project.is_empty() { sample_columns(db, &coll) } else { project }
        }
        Stmt::Insert { returning, .. } | Stmt::Update { returning, .. } | Stmt::Delete { returning, .. } => {
            if !wants_returning(sql) {
                return None;
            }
            if returning.is_empty() { sample_columns(db, &coll) } else { returning }
        }
    };
    if cols.is_empty() {
        // Nothing could be determined. `NoData` is a lie for a SELECT, but a
        // RowDescription with zero columns is a worse one — it tells the client
        // the query definitively has no output.
        return None;
    }
    let oids = cols
        .iter()
        .map(|c| {
            aggregate_oid(&c.src, db, &coll)
                .unwrap_or_else(|| infer_field_oid(db, &coll, &c.src))
        })
        .collect();
    Some((cols, oids))
}

/// A parse-only stand-in for a parameterised statement.
///
/// Substituting `NULL` was the obvious choice and the wrong one: a clause that
/// validates its argument rejects it, so `AS OF SYSTEM TIME $1` failed at
/// `Parse` — before the client ever bound a real sequence number. `0` parses
/// everywhere a literal can appear, and since only the SELECT list is read back
/// out, the stub's value never reaches an answer.
fn probe_sql(sql: &str, n_params: usize) -> String {
    let stub: Vec<Option<String>> = vec![Some("0".to_string()); n_params];
    substitute_params(sql, &stub).unwrap_or_else(|_| sql.to_string())
}

/// Run a portal's statement if it has not run yet, then report its shape.
fn ensure_executed(
    portal: &mut Portal,
    db_name: &str,
    db: Option<&Arc<Db>>,
    read_only: bool,
) -> Result<(), Vec<u8>> {
    if portal.result.is_some() {
        return Ok(());
    }
    let ex = execute_stmt(&portal.sql, db_name, db, read_only)?;
    // Freeze the output shape on first sight so `Describe` and every later
    // `Execute` describe the same rectangle.
    let project = if let Some(f) = &portal.frozen {
        f.clone()
    } else {
        let p = if ex.project.is_empty() {
            columns_for(&ex.rows, &[])
        } else {
            ex.project.clone()
        };
        portal.frozen = Some(p.clone());
        p
    };
    portal.result = Some(PortalResult {
        rows: ex.rows,
        project,
        has_rows: ex.has_rows,
        tag: ex.tag,
        tag_counts_rows: ex.tag_counts_rows,
        sent: 0,
    });
    Ok(())
}

// ── connection handling ─────────────────────────────────────────────────────

async fn read_exact(sock: &mut TcpStream, n: usize) -> std::io::Result<Vec<u8>> {
    let mut buf = vec![0u8; n];
    sock.read_exact(&mut buf).await?;
    Ok(buf)
}

async fn read_i32(sock: &mut TcpStream) -> std::io::Result<i32> {
    let b = read_exact(sock, 4).await?;
    Ok(i32::from_be_bytes([b[0], b[1], b[2], b[3]]))
}

fn parse_startup_params(body: &[u8]) -> HashMap<String, String> {
    let mut out = HashMap::new();
    let mut parts = body.split(|b| *b == 0).map(|s| String::from_utf8_lossy(s).to_string());
    while let (Some(k), Some(v)) = (parts.next(), parts.next()) {
        if k.is_empty() {
            break;
        }
        out.insert(k, v);
    }
    out
}

/// Serve one client connection to completion.
async fn handle(mut sock: TcpStream, resolver: Arc<dyn DbResolver>, read_only: bool) -> std::io::Result<()> {
    // ── startup, including the SSL negotiation clients try first ────────────
    let params = loop {
        let len = read_i32(&mut sock).await?;
        if len < 8 || len > 1 << 20 {
            return Ok(()); // nonsense framing — drop the connection
        }
        let code = read_i32(&mut sock).await?;
        let body = read_exact(&mut sock, (len - 8) as usize).await?;
        match code {
            SSL_REQUEST | GSS_REQUEST => {
                // Decline and let the client retry in the clear.
                sock.write_all(b"N").await?;
                continue;
            }
            CANCEL_REQUEST => return Ok(()), // nothing cancellable: reads are synchronous
            PROTO_V3 => break parse_startup_params(&body),
            other => {
                let major = other >> 16;
                sock.write_all(&err_msg(
                    "0A000",
                    &format!("unsupported frontend protocol {}.{} — this endpoint speaks 3.0",
                             major, other & 0xffff),
                )).await?;
                return Ok(());
            }
        }
    };

    let db_name = params.get("database").cloned().unwrap_or_default();

    // Resolve the database ONCE, here, on a blocking thread.
    //
    // A Postgres connection is bound to one database for its whole life, so
    // per-connection resolution is both correct and simpler than resolving per
    // statement — and it keeps the lock acquisition off the async worker.
    let resolved: Option<Arc<Db>> = {
        let r = Arc::clone(&resolver);
        let name = db_name.clone();
        tokio::task::spawn_blocking(move || r.resolve(&name))
            .await
            .unwrap_or(None)
    };

    // ── auth: mirror the HTTP surface ───────────────────────────────────────
    if let Some(expected) = resolver.token() {
        // AuthenticationCleartextPassword (3)
        let mut m = Out::msg(b'R');
        m.i32(3);
        sock.write_all(&m.finish()).await?;

        let tag = read_exact(&mut sock, 1).await?;
        if tag[0] != b'p' {
            sock.write_all(&err_msg("28000", "expected a password message")).await?;
            return Ok(());
        }
        let len = read_i32(&mut sock).await?;
        if len < 4 || len > 1 << 16 {
            return Ok(());
        }
        let body = read_exact(&mut sock, (len - 4) as usize).await?;
        let supplied = String::from_utf8_lossy(&body).trim_end_matches('\0').to_string();
        // Constant-time-ish: compare lengths and bytes without early return.
        let ok = supplied.len() == expected.len()
            && supplied.bytes().zip(expected.bytes()).fold(0u8, |a, (x, y)| a | (x ^ y)) == 0;
        if !ok {
            sock.write_all(&err_msg("28P01", "password authentication failed")).await?;
            return Ok(());
        }
    }

    let mut m = Out::msg(b'R');
    m.i32(0); // AuthenticationOk
    sock.write_all(&m.finish()).await?;

    for (k, v) in [
        ("server_version", SERVER_VERSION),
        ("server_encoding", "UTF8"),
        ("client_encoding", "UTF8"),
        ("DateStyle", "ISO, MDY"),
        ("integer_datetimes", "on"),
        ("standard_conforming_strings", "on"),
        ("application_name", "nedbd"),
    ] {
        let mut p = Out::msg(b'S');
        p.cstr(k);
        p.cstr(v);
        sock.write_all(&p.finish()).await?;
    }
    let mut k = Out::msg(b'K');
    k.i32(std::process::id() as i32);
    k.i32(0);
    sock.write_all(&k.finish()).await?;
    sock.write_all(&ready()).await?;

    // ── message loop ────────────────────────────────────────────────────────
    //
    // Prepared statements and portals live for the connection. `""` is the
    // unnamed statement/portal, which every driver reuses constantly — it is an
    // ordinary entry in the map rather than a special case.
    let mut prepared: HashMap<String, Prepared> = HashMap::new();
    let mut portals: HashMap<String, Portal> = HashMap::new();
    // After an error inside an extended-protocol sequence, everything up to the
    // next `Sync` is discarded. Skipping this is how a server ends up answering
    // a Bind the client has already abandoned, and the stream desynchronises.
    let mut failed = false;

    loop {
        let mut tag = [0u8; 1];
        if sock.read_exact(&mut tag).await.is_err() {
            return Ok(()); // client hung up
        }
        let len = read_i32(&mut sock).await?;
        if len < 4 || len > 64 << 20 {
            return Ok(());
        }
        let body = read_exact(&mut sock, (len - 4) as usize).await?;

        // `Sync` always clears the error state; `Terminate` always applies.
        if failed && tag[0] != b'S' && tag[0] != b'X' {
            continue;
        }

        match tag[0] {
            b'X' => return Ok(()), // Terminate

            b'Q' => {
                let sql = String::from_utf8_lossy(&body).trim_end_matches('\0').to_string();
                let out = run_simple_query(&sql, &db_name, resolved.as_ref(), read_only);
                sock.write_all(&out).await?;
                sock.write_all(&ready()).await?;
                // A simple query closes the unnamed portal, per the protocol.
                portals.remove("");
            }

            // ── Parse: name, SQL, declared parameter type OIDs ─────────────
            b'P' => {
                let mut at = 0usize;
                let name = take_cstr(&body, &mut at);
                let sql = take_cstr(&body, &mut at);
                let n = take_i16(&body, &mut at).unwrap_or(0).max(0) as usize;
                let mut declared = Vec::with_capacity(n);
                let mut bad = false;
                for _ in 0..n {
                    match take_i32(&body, &mut at) {
                        Ok(o) => declared.push(o),
                        Err(_) => { bad = true; break; }
                    }
                }
                if bad {
                    sock.write_all(&err_msg("08P01", "malformed Parse message")).await?;
                    failed = true;
                    continue;
                }
                // Reject unsupported SQL here rather than at Execute, so the
                // client learns at the point it asked — which is also where
                // Postgres reports it.
                //
                // The SQL evaluator gets asked first, or a catalogue query
                // would be refused at `Parse` by the NQL path that was never
                // going to run it — and the extended protocol is where every
                // ORM and async driver lives, so refusing here refuses them
                // all.
                let probe = probe_sql(&sql, param_count(&sql));
                if !sql_engine_owns(&probe) {
                    if let Err(why) = translate(&probe) {
                        sock.write_all(&err_msg("0A000", &why)).await?;
                        failed = true;
                        continue;
                    }
                }
                let param_oids = infer_param_oids(&sql, &declared, resolved.as_ref());
                prepared.insert(name, Prepared { sql, param_oids, out_shape: None });
                sock.write_all(&parse_complete()).await?;
            }

            // ── Bind: portal, statement, formats, values, result formats ───
            b'B' => {
                let mut at = 0usize;
                let portal_name = take_cstr(&body, &mut at);
                let stmt_name = take_cstr(&body, &mut at);
                if !prepared.contains_key(&stmt_name) {
                    sock.write_all(&err_msg("26000", &format!(
                        "prepared statement {:?} does not exist", stmt_name))).await?;
                    failed = true;
                    continue;
                }
                let p = &prepared[&stmt_name];
                let mut want_formats: Vec<i16> = vec![];
                let res: Result<String, String> = (|| {
                    let nfmt = take_i16(&body, &mut at)? .max(0) as usize;
                    let mut fmts = Vec::with_capacity(nfmt);
                    for _ in 0..nfmt {
                        fmts.push(take_i16(&body, &mut at)?);
                    }
                    let nparam = take_i16(&body, &mut at)?.max(0) as usize;
                    let mut vals: Vec<Option<String>> = Vec::with_capacity(nparam);
                    for i in 0..nparam {
                        let l = take_i32(&body, &mut at)?;
                        let raw: Option<Vec<u8>> = if l < 0 {
                            None
                        } else {
                            let l = l as usize;
                            if at + l > body.len() {
                                return Err("truncated Bind parameter".into());
                            }
                            let v = body[at..at + l].to_vec();
                            at += l;
                            Some(v)
                        };
                        // Zero format codes means "all text"; one means "this
                        // format for every parameter"; otherwise one per value.
                        let f = match fmts.len() {
                            0 => 0,
                            1 => fmts[0],
                            _ => *fmts.get(i).unwrap_or(&0),
                        };
                        let oid = *p.param_oids.get(i).unwrap_or(&OID_TEXT);
                        vals.push(decode_param(raw.as_deref(), oid, f)?);
                    }
                    // Result format codes. asyncpg asks for binary on every
                    // column, so honouring these is not an optimisation — it
                    // is the difference between asyncpg reading rows and
                    // refusing the result outright.
                    let nres = take_i16(&body, &mut at)?.max(0) as usize;
                    for _ in 0..nres {
                        let f = take_i16(&body, &mut at)?;
                        if f != 0 && f != 1 {
                            return Err(format!("unknown result format code {}", f));
                        }
                        want_formats.push(f);
                    }
                    substitute_params(&p.sql, &vals)
                })();
                match res {
                    Ok(sql) => {
                        // Binary encoding must use the types the client was
                        // TOLD about, so pull the advertised shape across.
                        let declared = if want_formats.iter().any(|f| *f == 1) {
                            let p = prepared.get_mut(&stmt_name).expect("checked above");
                            prepared_shape(p, resolved.as_ref()).clone()
                        } else {
                            None
                        };
                        portals.insert(portal_name, Portal {
                            sql, result: None, frozen: None,
                            formats: want_formats, declared,
                        });
                        sock.write_all(&bind_complete()).await?;
                    }
                    Err(why) => {
                        sock.write_all(&err_msg("08P01", &why)).await?;
                        failed = true;
                    }
                }
            }

            // ── Describe: 'S' statement, or 'P' portal ─────────────────────
            b'D' => {
                let kind = body.first().copied().unwrap_or(b'S');
                let mut at = 1usize;
                let name = take_cstr(&body, &mut at);
                if kind == b'S' {
                    if !prepared.contains_key(&name) {
                        sock.write_all(&err_msg("26000", &format!(
                            "prepared statement {:?} does not exist", name))).await?;
                        failed = true;
                        continue;
                    }
                    let p = prepared.get_mut(&name).expect("checked above");
                    let oids = p.param_oids.clone();
                    // asyncpg encodes its arguments from this, so the count has
                    // to be right or it refuses the call before sending a Bind.
                    sock.write_all(&parameter_description(&oids)).await?;
                    // Describe(statement) happens before Bind, so the requested
                    // result format is not known yet; Postgres reports text
                    // here too and the client's own Bind decides the encoding.
                    let out = match prepared_shape(p, resolved.as_ref()) {
                        Some((cols, col_oids)) => row_description(cols, col_oids),
                        None => no_data(),
                    };
                    sock.write_all(&out).await?;
                } else {
                    let portal = match portals.get_mut(&name) {
                        Some(p) => p,
                        None => {
                            sock.write_all(&err_msg("34000", &format!(
                                "portal {:?} does not exist", name))).await?;
                            failed = true;
                            continue;
                        }
                    };
                    // A bound portal can be run: doing it here means the
                    // RowDescription reports the columns and types actually
                    // present, which is strictly better than a guess. psycopg3
                    // takes this path on every query.
                    match ensure_executed(portal, &db_name, resolved.as_ref(), read_only) {
                        Err(encoded) => {
                            sock.write_all(&encoded).await?;
                            failed = true;
                        }
                        Ok(()) => {
                            let r = portal.result.as_ref().expect("just executed");
                            if !r.has_rows {
                                sock.write_all(&no_data()).await?;
                            } else {
                                let (cols, oids) = portal.shape(r);
                                let fmts: Vec<i16> =
                                    (0..cols.len()).map(|i| portal.format_of(i)).collect();
                                sock.write_all(&row_description_fmt(&cols, &oids, &fmts)).await?;
                            }
                        }
                    }
                }
            }

            // ── Execute: portal, maximum rows (0 = all) ────────────────────
            b'E' => {
                let mut at = 0usize;
                let name = take_cstr(&body, &mut at);
                let max_rows = take_i32(&body, &mut at).unwrap_or(0);
                let portal = match portals.get_mut(&name) {
                    Some(p) => p,
                    None => {
                        sock.write_all(&err_msg("34000", &format!(
                            "portal {:?} does not exist", name))).await?;
                        failed = true;
                        continue;
                    }
                };
                if let Err(encoded) = ensure_executed(portal, &db_name, resolved.as_ref(), read_only) {
                    sock.write_all(&encoded).await?;
                    failed = true;
                    continue;
                }
                let r = portal.result.as_ref().expect("just executed");
                if !r.has_rows {
                    let tag = r.tag.clone();
                    sock.write_all(&command_complete(&tag)).await?;
                    continue;
                }
                let (cols, oids) = portal.shape(r);
                let limit = if max_rows > 0 {
                    (r.sent + max_rows as usize).min(r.rows.len())
                } else {
                    r.rows.len()
                };
                // Encode the whole batch BEFORE writing any of it. A value that
                // cannot be sent in the advertised binary type has to become an
                // error instead of a truncated row stream — half a result set
                // followed by an error is far harder to diagnose than an error.
                let mut encoded: Vec<Vec<u8>> = Vec::with_capacity(limit - r.sent);
                let mut fail: Option<String> = None;
                for row in &r.rows[r.sent..limit] {
                    let mut vals: Vec<Option<Vec<u8>>> = Vec::with_capacity(cols.len());
                    for (i, c) in cols.iter().enumerate() {
                        let v = row.get(&c.src);
                        let got = if portal.format_of(i) == 1 {
                            cell_binary(v, oids.get(i).copied().unwrap_or(OID_TEXT))
                                .map_err(|e| format!("column {:?}: {}", c.out, e))
                        } else {
                            Ok(cell(v).map(|s| s.into_bytes()))
                        };
                        match got {
                            Ok(b) => vals.push(b),
                            Err(e) => { fail = Some(e); break; }
                        }
                    }
                    if fail.is_some() {
                        break;
                    }
                    encoded.push(data_row_bytes(&vals));
                }
                if let Some(why) = fail {
                    sock.write_all(&err_msg("22P03", &why)).await?;
                    failed = true;
                    continue;
                }
                let mut out = vec![];
                for e in &encoded {
                    out.extend_from_slice(e);
                }
                let r = portal.result.as_mut().expect("just executed");
                r.sent = limit;
                // More rows left and the client capped the batch: suspend the
                // portal instead of completing it. This is what a JDBC
                // `setFetchSize` and a psycopg3 server-side cursor rely on.
                if max_rows > 0 && r.sent < r.rows.len() {
                    out.extend_from_slice(&portal_suspended());
                } else {
                    let tag = if r.tag_counts_rows {
                        format!("{} {}", r.tag, r.sent)
                    } else {
                        r.tag.clone()
                    };
                    out.extend_from_slice(&command_complete(&tag));
                }
                sock.write_all(&out).await?;
            }

            // ── Close: 'S' statement, or 'P' portal ───────────────────────
            b'C' => {
                let kind = body.first().copied().unwrap_or(b'S');
                let mut at = 1usize;
                let name = take_cstr(&body, &mut at);
                if kind == b'S' {
                    prepared.remove(&name);
                } else {
                    portals.remove(&name);
                }
                // Closing something that was never open is explicitly not an
                // error in the protocol.
                sock.write_all(&close_complete()).await?;
            }

            // Flush: everything is written unbuffered already, so this is a
            // no-op — but it must NOT produce a ReadyForQuery, or a client that
            // flushes mid-sequence (asyncpg does, after Describe) loses sync.
            b'H' => {}

            b'S' => {
                failed = false;
                sock.write_all(&ready()).await?;
            }

            other => {
                sock.write_all(&err_msg(
                    "08P01",
                    &format!("unexpected frontend message {:?}", other as char),
                )).await?;
                failed = true;
            }
        }
    }
}

const READ_ONLY_MSG: &str =
    "this endpoint is running read-only (NEDBD_PG_READ_ONLY=1). Writes are \
     implemented but disabled on this server — unset the flag to allow them.";

fn no_db(db_name: &str) -> Vec<u8> {
    err_msg("3D000", &format!(
        "database {:?} is not open on this server — create it first \
         (POST /v1/databases), or connect with -d <name>", db_name))
}

/// `pg_catalog.pg_class` → `pg_class`, but `information_schema.tables` keeps
/// its qualifier, because `tables` is a plausible collection name and the
/// catalogue must never shadow a user's own data.
fn catalog_name(n: &str) -> String {
    let joined: Vec<&str> = n.split('.').collect();
    if joined.len() >= 2 && joined[joined.len() - 2] == "information_schema" {
        format!("information_schema.{}", joined[joined.len() - 1])
    } else {
        joined[joined.len() - 1].to_string()
    }
}

/// Does the SQL evaluator own this statement?
///
/// Two ways in. The first is obvious: it reads a catalogue relation.
///
/// The second is a statement with NO relation at all — a select list of
/// literals and scalar function calls, which is exactly what this evaluator
/// does and which the SQL→NQL path cannot express (NQL is FROM-first). That
/// path answers a handful of EXACT spellings from a canned table
/// (`SELECT 1`, `SELECT VERSION()`, `SELECT CURRENT_SCHEMA`), and those
/// answers are what existing clients already see — so this predicate rescues
/// only what it REFUSES, leaving every spelling it does handle alone.
///
/// That gap was not hypothetical. SQLAlchemy's PostgreSQL dialect opens every
/// connection with `select pg_catalog.version()`, which is one character of
/// qualification away from the canned `SELECT VERSION()` and therefore missed
/// it — so the engine refused the first statement of dialect initialisation
/// and NO SQLAlchemy application could connect at all. A canned list of
/// spellings is the same brittleness `pgcatalog` exists to avoid; the fix is
/// to let the evaluator answer, because it has `version()`,
/// `current_setting()` and the rest as real functions.
///
/// Cheap: one parse, no execution, no storage access.
/// Opt-in: route USER-collection `SELECT`s through the SQL evaluator too.
///
/// `NEDBD_SQL_ENGINE=1`. Default OFF, and the default is the point — this
/// changes which engine answers ordinary queries, and the two engines have to
/// be shown to agree before anyone's production reads move. Flipping it is a
/// deployment decision, not a build one, so it is read from the environment
/// once rather than compiled in.
///
/// What it unlocks is everything the translator refuses because NQL cannot
/// express it: joins, subqueries, `EXISTS`, `UNION`/`INTERSECT`/`EXCEPT`,
/// several named aggregates in one grouped row, `array_agg(x ORDER BY y)`.
/// What it must not lose is what only the translator has — and a statement the
/// evaluator's grammar cannot parse (`TRACE`, `SEARCH`, `VALID AS OF`,
/// `TRAVERSE`, every write) still falls through to the translator on its own,
/// because `parse` fails and this function is never consulted.
fn sql_engine_for_collections() -> bool {
    use std::sync::OnceLock;
    static ON: OnceLock<bool> = OnceLock::new();
    *ON.get_or_init(|| {
        matches!(std::env::var("NEDBD_SQL_ENGINE").as_deref(), Ok("1") | Ok("true") | Ok("on"))
    })
}

fn sql_engine_owns(sql: &str) -> bool {
    let Ok(sel) = crate::sqlselect::parse(sql) else { return false };
    let touched = sel.base_relations();
    if touched.is_empty() {
        return translate(sql).is_err();
    }
    if touched.iter().any(|t| crate::pgcatalog::is_catalog(&catalog_name(t))) {
        return true;
    }
    // A user collection reaches the evaluator only when asked for. Note the
    // asymmetry with the line above: a catalogue relation has always been the
    // evaluator's because the translator cannot serve it at all, whereas a
    // collection has a working answer on both paths — so the choice between
    // them is a judgement about parity, not about capability.
    sql_engine_for_collections()
}

/// Run a `SELECT` through the full SQL engine when it touches the catalogue.
///
/// The gate is deliberately narrow: a statement goes to `sqlselect` only when
/// one of its tables is a catalogue relation. Everything else keeps the
/// SQL→NQL path, which has the index pushdown, `AS OF`, `TRACE` and the
/// bounded scans — and whose join story is a real planning question rather
/// than a nested loop. Routing a large collection through a nested-loop join
/// would be a promise this engine cannot keep.
///
/// `None` means "not mine": the caller falls through to the ordinary path, so
/// the error the client sees is the ordinary path's error rather than a
/// confusing one from a parser that was never meant to handle the statement.
fn try_catalog_select(
    sql: &str,
    db: Option<&Arc<Db>>,
) -> Result<Option<(Executed, crate::sqlplan::Plan)>, Vec<u8>> {
    let sel = match crate::sqlselect::parse(sql) {
        Ok(sel) => sel,
        Err(why) => {
            // A statement that plainly reads the catalogue but that this
            // engine cannot parse gets the PARSE error, not the NQL path's.
            //
            // Falling through unconditionally produced an actively false
            // message: `\d` and `\dp` were told "JOIN is not supported",
            // which stopped being true the moment joins started working — and
            // a wrong explanation is worse than a blunt one, because it sends
            // the reader to fix the wrong thing.
            if mentions_catalog(sql) {
                return Err(err_msg("0A000", &format!(
                    "this catalogue query uses SQL this endpoint does not \
                     implement: {}", why)));
            }
            return Ok(None);
        }
    };

    // Which relations does it read — at ANY depth? `\dd` names its catalogue
    // relations only inside a derived table, and `\dT` only inside two
    // subqueries; a walk over the top-level FROM list alone would route both
    // to the NQL path, which cannot parse them and would report an error that
    // sends the reader to fix the wrong thing.
    if !sql_engine_owns(sql) {
        return Ok(None);
    }

    // The storage pre-filter, resolved per relation NAME and computed once.
    //
    // The resolver is handed a name (`orders`) but the WHERE clause qualifies
    // by BINDING (`o.status` for `FROM orders o`), so the predicate has to be
    // looked up by name and rendered against that relation's binding. Getting
    // this wrong is silent: the pre-filter simply never matches and the scan
    // quietly reads the whole collection, which is exactly what EXPLAIN caught
    // the first time round — `Seq Scan on orders o (actual rows=3)` when the
    // query wanted two.
    //
    // A name appearing TWICE (a self-join, `FROM t a JOIN t b`) maps to two
    // different bindings with different predicates, and one scan cannot serve
    // both. Those are dropped rather than guessed at.
    // `AS OF SYSTEM TIME <seq>`, per relation name.
    //
    // The resolver is keyed by NAME, so one collection named twice gets ONE
    // scan. `FROM orders AS OF 1 o JOIN orders n` asks for that collection at
    // two different sequences at once, and a single scan cannot serve both.
    //
    // This is REFUSED rather than resolved to one of them, and the reason is
    // worth keeping: the first version dropped the qualifier when a name was
    // ambiguous — the same "don't guess" instinct that is right for a
    // pre-filter. It is wrong here. Dropping a pre-filter costs a wasted row;
    // dropping an AS OF answers a question about the past with data from the
    // present, and it does it silently. The query `... orders AS OF 1 o JOIN
    // orders n ...` returned the CURRENT value for both sides and looked fine.
    let temporal: std::collections::HashMap<String, u64> = {
        // Gather every sequence each name is read at first, INCLUDING the
        // absent one, then judge. Deciding as we walk got this wrong: the
        // first arm of a self-join was judged before it had been recorded, so
        // a legitimate pair reported the wrong reason.
        let mut seen: std::collections::HashMap<String, Vec<Option<u64>>> =
            std::collections::HashMap::new();
        for t in sel.from.iter().chain(sel.joins.iter().map(|j| &j.table)) {
            seen.entry(catalog_name(&t.name).to_ascii_lowercase())
                .or_default()
                .push(t.as_of);
        }
        let mut out: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
        for (key, ats) in &seen {
            let mut distinct: Vec<Option<u64>> = ats.clone();
            distinct.sort();
            distinct.dedup();
            match distinct.as_slice() {
                // One sequence for this name, however many times it appears.
                [Some(seq)] => {
                    out.insert(key.clone(), *seq);
                }
                [None] => {}
                // More than one. Say WHICH disagreement it is, because the two
                // read very differently to whoever wrote the query.
                _ => {
                    let mixed_tip = distinct.contains(&None);
                    let seqs: Vec<String> =
                        distinct.iter().flatten().map(|s| s.to_string()).collect();
                    let detail = if mixed_tip {
                        format!(
                            "at the tip and AS OF {}",
                            seqs.join(" and "))
                    } else {
                        format!("AS OF {}", seqs.join(" and "))
                    };
                    return Err(err_msg("0A000", &format!(
                        "{:?} is read {} in one statement. This endpoint reads each \
                         collection once per statement, so it cannot serve both — and \
                         answering from either one would silently return the same rows for \
                         both arms, which is the comparison failing to be a comparison. Ask \
                         the two questions separately.",
                        key, detail)));
                }
            }
        }
        out
    };

    // NQL's own verbs, per relation name: `(VALID AS OF, SEARCH)`.
    //
    // Same one-scan-per-name constraint as the temporal map, and the same
    // verdict for the same reason: two different values for one scan is
    // REFUSED, because silently picking one would answer a different question
    // than the one asked and look like it worked.
    let nql_verbs: std::collections::HashMap<String, (Option<String>, Option<String>)> = {
        let mut out: std::collections::HashMap<String, (Option<String>, Option<String>)> =
            std::collections::HashMap::new();
        for t in sel.from.iter().chain(sel.joins.iter().map(|j| &j.table)) {
            let k = catalog_name(&t.name).to_ascii_lowercase();
            let e = out.entry(k.clone()).or_default();
            for (slot, incoming, verb) in [
                (&mut e.0, &t.valid_as_of, "VALID AS OF"),
                (&mut e.1, &t.search, "SEARCH"),
            ] {
                match (slot.as_deref(), incoming.as_deref()) {
                    (Some(a), Some(b)) if a != b => {
                        return Err(err_msg("0A000", &format!(
                            "{:?} is read with two different {} arguments in one statement \
                             ({:?} and {:?}). This endpoint reads each collection once, so \
                             it cannot serve both. Ask the two questions separately.",
                            k, verb, a, b)));
                    }
                    (None, Some(b)) => *slot = Some(b.to_string()),
                    _ => {}
                }
            }
        }
        out
    };

    let pushdown_prefilters: std::collections::HashMap<String, String> = {
        let refs: Vec<&crate::sqlselect::TableRef> = sel
            .from
            .iter()
            .chain(sel.joins.iter().map(|j| &j.table))
            .collect();
        let bindings: Vec<String> = refs.iter().map(|t| t.binding()).collect();
        let nullable = crate::sqlpush::nullable_bindings(&sel);
        let mut out = std::collections::HashMap::new();
        let mut ambiguous: Vec<String> = vec![];
        for t in &refs {
            let key = catalog_name(&t.name).to_ascii_lowercase();
            if out.contains_key(&key) || ambiguous.contains(&key) {
                out.remove(&key);
                ambiguous.push(key);
                continue;
            }
            if let Some(p) = crate::sqlpush::nql_prefilter(
                sel.where_.as_ref(), &t.binding(), &bindings, &nullable) {
                out.insert(key, p);
            }
        }
        out
    };

    let resolve = |name: &str| -> anyhow::Result<Option<Box<dyn crate::sqlselect::Relation>>> {
        let cname = catalog_name(name);
        // A catalogue relation is SYNTHESISED from the current shape of the
        // store: it has no log, so it has no history, and there is nothing for
        // a temporal or full-text qualifier to mean.
        //
        // Refused rather than ignored, and the difference is the entire point.
        // Ignoring `AS OF SYSTEM TIME 0` answers a question about the past with
        // present-day rows and looks like it worked — and that is exactly what
        // started happening here the moment the SQL parser learned `AS OF`:
        // before, the statement failed to parse and fell through to the
        // translator, which refused it properly. Teaching one layer a clause
        // silently un-taught another layer's refusal, and a test written long
        // before this change is what caught it.
        {
            let k = cname.to_ascii_lowercase();
            let bad = if temporal.contains_key(&k) {
                Some("AS OF SYSTEM TIME")
            } else {
                match nql_verbs.get(&k) {
                    Some((Some(_), _)) => Some("VALID AS OF"),
                    Some((_, Some(_))) => Some("SEARCH"),
                    _ => None,
                }
            };
            if let Some(clause) = bad {
                if crate::pgcatalog::is_catalog(&cname) {
                    anyhow::bail!(
                        "{} is not supported on the catalogue relation {:?} — a catalogue is \
                         synthesised from the store's current shape rather than read from the \
                         log, so it has no history to reach and no document text to search. \
                         Ignoring the clause would answer your question with present-day rows \
                         and look like it worked",
                        clause, cname);
                }
            }
        }
        if let Some(rows) = crate::pgcatalog::rows(&cname, db) {
            // A synthesised catalogue relation is small and built eagerly;
            // wrapping it satisfies the streaming contract without pretending
            // it is lazy.
            return Ok(Some(crate::sqlselect::from_vec(rows)));
        }
        // A join between a catalogue relation and a real collection is
        // legitimate, so a user table still resolves.
        //
        // `nql::query` materialises whatever it is asked for, so what it is
        // ASKED for is the whole cost of this line. It used to be
        // `FROM <collection>` — every document, unconditionally, before a
        // single predicate ran. Free on a catalogue relation of a few dozen
        // synthesised rows; on a user collection it is the difference between
        // reading one document and reading all of them.
        //
        // `sqlpush::nql_prefilter` renders the part of the WHERE that NQL is
        // known to evaluate identically, and the full WHERE still runs above
        // this — so the pre-filter can only ever cost a wasted row, never an
        // answer. See the module note in `sqlpush` for why each refused
        // construct is refused.
        //
        // Still eager, and deliberately not claimed otherwise: this narrows
        // WHAT is materialised, not WHETHER it is. A lazy storage scan is the
        // other half and is tracked in HANDOFF.
        let key = cname.to_ascii_lowercase();
        let pre = pushdown_prefilters.get(&key);
        // Composed in NQL'S OWN CLAUSE ORDER, which its grammar fixes as
        //
        //     FROM coll [AS OF seq] [VALID AS OF "date"] [WHERE p] [SEARCH "t"]
        //
        // and which is not negotiable: emit `AS OF` after `WHERE` and the NQL
        // parser reads it as part of the predicate expression. This is the
        // whole mechanism behind "NQL folded into neSQL" — the SQL side parses
        // the verbs and composes joins and subqueries around them, while the
        // NQL engine remains the one implementation that executes them.
        let nql = {
            let mut q = format!("FROM {}", cname);
            if let Some(seq) = temporal.get(&key) {
                q.push_str(&format!(" AS OF {}", seq));
            }
            if let Some(d) = nql_verbs.get(&key).and_then(|v| v.0.as_deref()) {
                q.push_str(&format!(" VALID AS OF {}", nql_string(d)));
            }
            if let Some(p) = pre {
                q.push_str(&format!(" WHERE {}", p));
            }
            if let Some(t) = nql_verbs.get(&key).and_then(|v| v.1.as_deref()) {
                q.push_str(&format!(" SEARCH {}", nql_string(t)));
            }
            q
        };
        match db {
            Some(db) => match crate::nql::query(db, &nql) {
                Ok((rows, _)) => Ok(Some(crate::sqlselect::from_vec(rows))),
                // A pre-filter that NQL refuses must not fail the statement:
                // it is an optimisation, so the honest fallback is the
                // unfiltered scan the query would have done anyway. Silently
                // returning None here would turn a slow-but-correct query into
                // "relation does not exist".
                Err(_) if pre.is_some() => {
                    // The fallback drops the PRE-FILTER, which is free, and
                    // must keep the AS OF, which is not: falling back to the
                    // tip would answer a historical question with current
                    // data. That is the silent-wrong-answer shape this engine
                    // keeps getting bitten by, so the sequence travels with
                    // the retry.
                    let bare = match temporal.get(&key) {
                        Some(seq) => format!("FROM {} AS OF {}", cname, seq),
                        None => format!("FROM {}", cname),
                    };
                    match crate::nql::query(db, &bare) {
                        Ok((rows, _)) => Ok(Some(crate::sqlselect::from_vec(rows))),
                        Err(_) => Ok(None),
                    }
                }
                Err(_) => Ok(None),
            },
            None => Ok(None),
        }
    };

    let (cols, rows, plan) = crate::sqlselect::execute_explain(
        &sel,
        &resolve,
        crate::sqljoin::JoinExec::Auto,
    )
    .map_err(|e| err_msg("42601", &e.to_string()))?;

    Ok(Some((
        Executed {
            rows,
            // The KEY is what the row is stored under; the NAME is what the
            // client sees. They differ when a select list has duplicate output
            // names, which PostgreSQL permits and generated SQL relies on.
            project: cols
                .iter()
                .map(|c| Col::renamed(&c.key, &c.name))
                .collect(),
            has_rows: true,
            tag: "SELECT".into(),
            tag_counts_rows: true,
        },
        plan,
    )))
}

/// Strip a leading `EXPLAIN`, returning the statement it wraps.
///
/// `ANALYZE` and `VERBOSE` are accepted and ignored: this endpoint always
/// executes and always reports actual rows, so `EXPLAIN` and
/// `EXPLAIN ANALYZE` genuinely do the same thing here. Accepting the keyword
/// and silently doing the honest thing beats refusing a client's spelling.
fn strip_explain(sql: &str) -> Option<&str> {
    let t = sql.trim().trim_end_matches(';').trim();
    let mut rest = t.strip_prefix("EXPLAIN").or_else(|| t.strip_prefix("explain"))?;
    // Require a word boundary so `EXPLAINED` is not mistaken for a keyword.
    if !rest.starts_with(char::is_whitespace) {
        return None;
    }
    rest = rest.trim_start();
    loop {
        let low = rest.to_lowercase();
        if let Some(r) = low.strip_prefix("analyze").or_else(|| low.strip_prefix("analyse")) {
            if r.starts_with(char::is_whitespace) || r.is_empty() {
                rest = rest[rest.len() - r.len()..].trim_start();
                continue;
            }
        }
        if let Some(r) = low.strip_prefix("verbose") {
            if r.starts_with(char::is_whitespace) || r.is_empty() {
                rest = rest[rest.len() - r.len()..].trim_start();
                continue;
            }
        }
        break;
    }
    Some(rest)
}

/// One text column named `QUERY PLAN`, which is exactly the shape PostgreSQL
/// returns — so `psql` prints it without special handling.
fn plan_result(lines: Vec<String>) -> Executed {
    Executed {
        rows: lines
            .into_iter()
            .map(|l| serde_json::json!({ "QUERY PLAN": l }))
            .collect(),
        project: vec![Col::same("QUERY PLAN")],
        has_rows: true,
        tag: "EXPLAIN".into(),
        tag_counts_rows: false,
    }
}

/// Does the raw SQL plainly read a catalogue relation?
///
/// A cheap text check, used only to decide WHICH error to report when the
/// statement cannot be parsed — never to decide what a parsable statement
/// means. `pg_` is the giveaway: every catalogue relation is prefixed, and so
/// is the `pg_catalog` schema qualifier.
fn mentions_catalog(sql: &str) -> bool {
    let low = sql.to_lowercase();
    low.contains("pg_catalog.")
        || low.contains("information_schema.")
        || low.contains("from pg_")
        || low.contains("join pg_")
}

/// The catalogue relation a translated query reads from, if any.
///
/// Reads the collection straight off the parsed NQL rather than re-parsing the
/// SQL, so it cannot disagree with what the executor is about to run.
fn catalog_target(nql: &str) -> Option<String> {
    let coll = crate::nql::parse(nql).ok()?.coll;
    if crate::pgcatalog::is_catalog(&coll) {
        Some(coll)
    } else {
        None
    }
}

/// True when the statement carried a RETURNING clause. Checked against the raw
/// SQL because `RETURNING *` yields an EMPTY projection, which is otherwise
/// indistinguishable from "no RETURNING at all".
fn wants_returning(sql: &str) -> bool {
    find_kw(&sql.to_uppercase(), "RETURNING").is_some()
}

/// A unique key for a server-assigned INSERT id.
fn next_row_id() -> String {
    use std::sync::atomic::{AtomicU64, Ordering};
    static N: AtomicU64 = AtomicU64::new(0);
    let n = N.fetch_add(1, Ordering::Relaxed);
    let ts = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_micros())
        .unwrap_or(0);
    format!("r{}{}", ts, n)
}

/// One executed statement, held apart from any wire encoding.
///
/// This type is why the simple and extended protocols share an execution path
/// rather than growing two copies of the SQL→NEDB semantics. The simple path
/// encodes it immediately; the extended path parks it in a portal and dribbles
/// the rows out across successive `Execute` messages. Both get identical
/// answers because both call `execute_stmt`.
pub struct Executed {
    /// The rows the client gets — a SELECT's result, or a write's `RETURNING`.
    pub rows: Vec<Value>,
    /// How to project them (empty = every key in the row).
    pub project: Vec<Col>,
    /// Whether the client asked for rows at all. Distinct from `rows.is_empty()`:
    /// a `SELECT` matching nothing still owes a `RowDescription`, while an
    /// `UPDATE` without `RETURNING` owes `NoData`.
    pub has_rows: bool,
    /// The command tag, already rendered — except for a SELECT, where the row
    /// count is only known once the rows have actually been sent.
    pub tag: String,
    /// True when `tag` is a SELECT-shaped tag whose count is the rows sent.
    pub tag_counts_rows: bool,
}

impl Executed {
    fn nothing(tag: &str) -> Self {
        Executed { rows: vec![], project: vec![], has_rows: false, tag: tag.to_string(), tag_counts_rows: false }
    }
    /// Render the final `CommandComplete` given how many rows went out.
    fn tag_for(&self, sent: usize) -> String {
        if self.tag_counts_rows { format!("{} {}", self.tag, sent) } else { self.tag.clone() }
    }
}

/// Run ONE statement. `Err` carries an already-encoded `ErrorResponse`.
///
/// Every SQL→NEDB decision lives here, which is the point: the extended query
/// protocol added below is then purely a matter of message framing, and cannot
/// drift from the simple path's semantics.
/// Run one neQL statement against a database, in process.
///
/// # Why this exists
///
/// Until this, the engine had exactly one SQL execution path and it was welded
/// to the wire protocol: `execute_stmt` is private, takes the connection's
/// read-only flag, and reports failure as ALREADY-ENCODED Postgres error bytes.
/// Nothing outside a pgwire session could run SQL against a `Db`.
///
/// That was survivable while the only SQL client was a socket. It stopped being
/// survivable when neSQL — which owns the language — needed to run the language
/// from a CLI, because the alternatives were a CLI that opens a TCP connection
/// to its own process, or a second SQL front end living in the CLI. The second
/// one is worse than it sounds: it makes the CLI a quieter second authority on
/// what the language accepts, and the first divergence between them would be
/// discovered by a user, not by us.
///
/// So the path the wire already takes is exposed, with the error decoded into
/// text. Same parser, same translator, same evaluator, same decision about
/// which engine runs a statement — one authority.
pub fn execute_sql(db: &Arc<Db>, sql: &str, read_only: bool)
    -> std::result::Result<Executed, String>
{
    execute_stmt(sql, "", Some(db), read_only).map_err(|wire| decode_wire_error(&wire))
}

/// Pull the human-readable message out of an encoded ErrorResponse.
///
/// The wire format is a sequence of NUL-terminated `field-code || text` runs
/// terminated by an empty field. `M` is the primary message and `C` the
/// SQLSTATE; both are reported, because a caller who loses the SQLSTATE loses
/// the only machine-stable part of the error.
fn decode_wire_error(buf: &[u8]) -> String {
    let mut code: Option<String> = None;
    let mut msg: Option<String> = None;
    // Skip the 1-byte tag and 4-byte length when they are present.
    let body = if buf.len() > 5 { &buf[5..] } else { buf };
    let mut i = 0usize;
    while i < body.len() && body[i] != 0 {
        let field = body[i];
        i += 1;
        let start = i;
        while i < body.len() && body[i] != 0 { i += 1; }
        let text = String::from_utf8_lossy(&body[start..i]).into_owned();
        i += 1; // the NUL
        match field {
            b'C' => code = Some(text),
            b'M' => msg = Some(text),
            _ => {}
        }
    }
    match (code, msg) {
        (Some(c), Some(m)) => format!("{} ({})", m, c),
        (None, Some(m)) => m,
        // Never silently produce an empty error. A failure we cannot read is
        // still a failure, and saying so beats returning "".
        _ => format!(
            "the engine refused the statement and the error could not be decoded              ({} bytes of wire response)", buf.len()
        ),
    }
}

fn execute_stmt(
    stmt_sql: &str,
    db_name: &str,
    db: Option<&Arc<Db>>,
    read_only: bool,
) -> Result<Executed, Vec<u8>> {
    // The full SQL engine gets first refusal, but ONLY for statements that
    // touch the catalogue — see `try_catalog_select`. It has to run before
    // `translate`, because `translate` targets NQL and NQL cannot express a
    // join, a CASE or a scalar function at all.
    // EXPLAIN reports which engine would run the statement, and a plan only
    // when the SQL evaluator is the engine that actually runs it. Describing a
    // pipeline the statement would not take is the one thing an EXPLAIN must
    // never do.
    if let Some(inner) = strip_explain(stmt_sql) {
        if let Some((_, plan)) = try_catalog_select(inner, db)? {
            return Ok(plan_result(plan.render()));
        }
        let mut lines = vec![];
        match translate(inner) {
            Ok(_) => {
                lines.push(
                    "NQL path — this statement is translated to NQL and \
                     executed by the storage engine, not by the SQL evaluator."
                        .to_string(),
                );
                lines.push(
                    "No plan is reported, because the SQL evaluator is not \
                     what runs it. Reporting one would describe a pipeline \
                     that never executed."
                        .to_string(),
                );
                lines.push(
                    "The SQL evaluator (joins, CASE, scalar functions, a \
                     hash-join planner) currently serves catalogue queries."
                        .to_string(),
                );
            }
            Err(why) => lines.push(format!("cannot be executed: {why}")),
        }
        return Ok(plan_result(lines));
    }

    if let Some((done, _plan)) = try_catalog_select(stmt_sql, db)? {
        return Ok(done);
    }

    let stmt = translate(stmt_sql).map_err(|why| err_msg("0A000", &why))?;

    // Every arm below that touches storage needs a database; resolve the
    // "no such database" answer once instead of at each use.
    macro_rules! need_db {
        () => {
            match db {
                Some(db) => db,
                None => return Err(no_db(db_name)),
            }
        };
    }
    macro_rules! need_write {
        () => {
            if read_only {
                return Err(err_msg("25006", READ_ONLY_MSG));
            }
        };
    }

    match stmt {
        Stmt::Ok(tag) => Ok(Executed::nothing(if tag.is_empty() { "SELECT 0" } else { tag })),

        Stmt::Canned { cols, row } => {
            // Fold the canned answer into an ordinary row so the encoders,
            // the portal machinery and `Describe` all see one shape.
            let mut obj = serde_json::Map::new();
            for (c, v) in cols.iter().zip(row.iter()) {
                obj.insert(c.clone(), Value::String(v.clone()));
            }
            Ok(Executed {
                rows: vec![Value::Object(obj)],
                project: cols.iter().map(|c| Col::same(c)).collect(),
                has_rows: true,
                tag: "SELECT".into(),
                tag_counts_rows: true,
            })
        }

        Stmt::Query { nql, project } => {
            // A catalogue relation is synthesised from the live database
            // rather than read from it — but it is still queried with the
            // ORDINARY predicate path, so WHERE / ORDER BY / LIMIT and the
            // `~` operators work on it because they are the same operators.
            //
            // Checked BEFORE `need_db!()`: `SELECT * FROM pg_namespace` has to
            // answer even when the client connected without naming a database,
            // which is exactly what psql does on startup. Refusing there is
            // how "psql cannot connect" starts.
            if let Some(coll) = catalog_target(&nql) {
                let rows = crate::pgcatalog::rows(&coll, db)
                    .expect("catalog_target only returns names pgcatalog serves");
                let rows = crate::nql::query_rows(rows, &nql)
                    .map_err(|e| err_msg("42601", &e.to_string()))?;
                return Ok(Executed {
                    rows, project, has_rows: true,
                    tag: "SELECT".into(), tag_counts_rows: true,
                });
            }
            let db = need_db!();
            let (rows, _) = crate::nql::query(db, &nql).map_err(|e| {
                err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
            })?;
            Ok(Executed { rows, project, has_rows: true, tag: "SELECT".into(), tag_counts_rows: true })
        }

        Stmt::Insert { coll, rows, returning } => {
            let db = need_db!();
            need_write!();
            let mut written: Vec<Value> = vec![];
            for (i, r) in rows.iter().enumerate() {
                // The engine requires an id. When the statement did not supply
                // one, mint a unique key rather than silently overwriting a
                // shared default.
                let id = match &r.id {
                    Some(id) => id.clone(),
                    None => format!("{}-{}", next_row_id(), i),
                };
                let node = db
                    .put(&coll, &id, Value::Object(r.doc.clone()),
                         r.caused_by.clone(), r.valid_from.clone(), r.valid_to.clone())
                    .map_err(|e| err_msg("XX000", &format!("INSERT failed: {}", e)))?;
                written.push(crate::nql::node_to_json(&node));
            }
            let n = written.len();
            let has_rows = wants_returning(stmt_sql);
            Ok(Executed {
                rows: if has_rows { written } else { vec![] },
                project: returning,
                has_rows,
                // Postgres reports `INSERT <oid> <rows>`; the oid is always 0.
                tag: format!("INSERT 0 {}", n),
                tag_counts_rows: false,
            })
        }

        Stmt::Update { coll, set, nql, returning } => {
            let db = need_db!();
            need_write!();
            // Matching rows come from an ordinary NQL read, so the whole
            // predicate surface works inside an UPDATE.
            let (matched, _) = crate::nql::query(db, &nql).map_err(|e| {
                err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
            })?;
            let mut written: Vec<Value> = vec![];
            for row in &matched {
                let id = match row.get("_id").and_then(|v| v.as_str()) {
                    Some(id) => id.to_string(),
                    None => continue,
                };
                // Merge onto the CURRENT stored document, not onto the query
                // row: a query row carries injected `_`-prefixed metadata that
                // must never be written back into the payload.
                let mut doc = match db.get(&coll, &id) {
                    Some(n) => match n.data {
                        Value::Object(m) => m,
                        _ => serde_json::Map::new(),
                    },
                    None => continue,
                };
                for (k, v) in &set {
                    doc.insert(k.clone(), v.clone());
                }
                // An UPDATE is a NEW VERSION — the prior value stays readable
                // with AS OF SYSTEM TIME. That is the whole point.
                let node = db
                    .put(&coll, &id, Value::Object(doc), vec![], None, None)
                    .map_err(|e| err_msg("XX000", &format!("UPDATE failed: {}", e)))?;
                written.push(crate::nql::node_to_json(&node));
            }
            let n = written.len();
            let has_rows = wants_returning(stmt_sql);
            Ok(Executed {
                rows: if has_rows { written } else { vec![] },
                project: returning,
                has_rows,
                tag: format!("UPDATE {}", n),
                tag_counts_rows: false,
            })
        }

        Stmt::Delete { coll, nql, returning } => {
            let db = need_db!();
            need_write!();
            let (matched, _) = crate::nql::query(db, &nql).map_err(|e| {
                err_msg("42601", &format!("{} (translated to NQL: {})", e, nql))
            })?;
            // RETURNING must be captured BEFORE the delete: after the tombstone
            // the row is no longer readable by id.
            let returned = matched.clone();
            let mut n = 0usize;
            for row in &matched {
                if let Some(id) = row.get("_id").and_then(|v| v.as_str()) {
                    match db.delete(&coll, id) {
                        Ok(true) => n += 1,
                        Ok(false) => {}
                        Err(e) => return Err(err_msg("XX000", &format!("DELETE failed: {}", e))),
                    }
                }
            }
            let has_rows = wants_returning(stmt_sql);
            Ok(Executed {
                rows: if has_rows { returned } else { vec![] },
                project: returning,
                has_rows,
                tag: format!("DELETE {}", n),
                tag_counts_rows: false,
            })
        }
    }
}

/// Execute a simple-query payload, which may hold several `;`-separated statements.
fn run_simple_query(sql: &str, db_name: &str, db: Option<&Arc<Db>>, read_only: bool) -> Vec<u8> {
    let mut out = vec![];
    let statements = split_statements(sql);
    if statements.is_empty() {
        // EmptyQueryResponse
        return Out::msg(b'I').finish();
    }
    for stmt_sql in statements {
        match execute_stmt(&stmt_sql, db_name, db, read_only) {
            // Abandon the rest of the batch on the first error, as Postgres does.
            Err(encoded) => {
                out.extend_from_slice(&encoded);
                return out;
            }
            Ok(ex) => {
                if ex.has_rows {
                    out.extend_from_slice(&encode_rows(&ex.rows, &ex.project));
                }
                out.extend_from_slice(&command_complete(&ex.tag_for(ex.rows.len())));
            }
        }
    }
    out
}

/// Split on `;` at the top level, ignoring separators inside string literals.
fn split_statements(sql: &str) -> Vec<String> {
    let mut out = vec![];
    let mut cur = String::new();
    let mut in_s = false;
    for c in sql.chars() {
        match c {
            '\'' => { in_s = !in_s; cur.push(c); }
            ';' if !in_s => {
                if !cur.trim().is_empty() { out.push(cur.clone()); }
                cur.clear();
            }
            _ => cur.push(c),
        }
    }
    if !cur.trim().is_empty() {
        out.push(cur);
    }
    out
}

/// Bind and serve the Postgres read endpoint until the process exits.
pub async fn run(host: &str, port: u16, resolver: Arc<dyn DbResolver>) -> anyhow::Result<()> {
    // Writes are ON by default — that is the parity position. An operator who
    // wants the "system of proof beside your database" deployment, where this
    // door must never mutate anything, sets NEDBD_PG_READ_ONLY=1.
    let read_only = std::env::var("NEDBD_PG_READ_ONLY")
        .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
        .unwrap_or(false);
    let listener = TcpListener::bind((host, port)).await?;
    println!("  pgwire   postgres endpoint on {}:{} — psql / DBeaver / psycopg ({})",
             host, port,
             if read_only { "SELECT only — read-only mode" } else { "SELECT + INSERT/UPDATE/DELETE" });
    loop {
        let (sock, _peer) = match listener.accept().await {
            Ok(v) => v,
            Err(e) => {
                eprintln!("  [pgwire] accept failed: {}", e);
                continue;
            }
        };
        let r = Arc::clone(&resolver);
        tokio::spawn(async move {
            let _ = sock.set_nodelay(true);
            if let Err(e) = handle(sock, r, read_only).await {
                // A client disconnecting mid-message is routine, not an incident.
                if e.kind() != std::io::ErrorKind::UnexpectedEof
                    && e.kind() != std::io::ErrorKind::ConnectionReset
                {
                    eprintln!("  [pgwire] connection error: {}", e);
                }
            }
        });
    }
}

// ─────────────────────────────────────────────────────────────────────────────

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

    #[test]
    fn a_bare_explain_is_stripped() {
        assert_eq!(strip_explain("EXPLAIN SELECT 1"), Some("SELECT 1"));
        assert_eq!(strip_explain("explain select 1"), Some("select 1"));
        assert_eq!(strip_explain("  EXPLAIN   SELECT 1 ;  "), Some("SELECT 1"));
    }

    #[test]
    fn analyze_and_verbose_are_accepted_and_ignored() {
        // This endpoint always executes and always reports actual rows, so
        // EXPLAIN and EXPLAIN ANALYZE genuinely do the same thing. Accepting
        // the client's spelling beats refusing it.
        assert_eq!(strip_explain("EXPLAIN ANALYZE SELECT 1"), Some("SELECT 1"));
        assert_eq!(strip_explain("EXPLAIN ANALYSE SELECT 1"), Some("SELECT 1"));
        assert_eq!(strip_explain("EXPLAIN VERBOSE SELECT 1"), Some("SELECT 1"));
        assert_eq!(strip_explain("EXPLAIN ANALYZE VERBOSE SELECT 1"), Some("SELECT 1"));
        assert_eq!(strip_explain("explain analyze verbose select 1"), Some("select 1"));
    }

    #[test]
    fn a_word_merely_starting_with_explain_is_not_a_keyword() {
        assert_eq!(strip_explain("EXPLAINED SELECT 1"), None);
        assert_eq!(strip_explain("SELECT 1"), None);
        assert_eq!(strip_explain("SELECT explain FROM t"), None);
    }

    #[test]
    fn a_column_named_analyze_is_not_eaten() {
        // `analyzed` merely starts with the keyword; the word boundary check
        // is what stops it being consumed as an option.
        assert_eq!(strip_explain("EXPLAIN analyzed_view"), Some("analyzed_view"));
    }

    #[test]
    fn the_plan_result_has_postgres_shape() {
        let e = plan_result(vec!["Seq Scan on t".into(), "note".into()]);
        assert_eq!(e.project.len(), 1);
        assert_eq!(e.project[0].out, "QUERY PLAN");
        assert_eq!(e.rows.len(), 2);
        assert_eq!(e.rows[0]["QUERY PLAN"], "Seq Scan on t");
        assert_eq!(e.tag, "EXPLAIN");
        // EXPLAIN's tag carries no row count in PostgreSQL.
        assert!(!e.tag_counts_rows);
    }
}

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

    fn q(sql: &str) -> String {
        match translate(sql) {
            Ok(Stmt::Query { nql, .. }) => nql,
            other => panic!("expected a query for {:?}, got {:?}", sql, other),
        }
    }
    /// Output column names, in order.
    fn proj(sql: &str) -> Vec<String> {
        match translate(sql) {
            Ok(Stmt::Query { project, .. }) => project.iter().map(|c| c.out.clone()).collect(),
            other => panic!("expected a query for {:?}, got {:?}", sql, other),
        }
    }
    /// (source key, output name) pairs, for the aggregate renaming.
    fn proj_pairs(sql: &str) -> Vec<(String, String)> {
        match translate(sql) {
            Ok(Stmt::Query { project, .. }) =>
                project.iter().map(|c| (c.src.clone(), c.out.clone())).collect(),
            other => panic!("expected a query for {:?}, got {:?}", sql, other),
        }
    }
    fn names(cols: &[Col]) -> Vec<String> { cols.iter().map(|c| c.out.clone()).collect() }

    /// The full projection, so a test can assert the SRC and the OUT
    /// separately — they are different jobs and conflating them is how an
    /// alias got lost.
    fn cols_of(sql: &str) -> Vec<Col> {
        match translate(sql).unwrap() {
            Stmt::Query { project, .. } => project,
            other => panic!("{:?}", other),
        }
    }

    #[test]
    fn select_star_becomes_bare_from() {
        assert_eq!(q("SELECT * FROM orders"), "FROM orders");
        assert_eq!(q("select * from orders;"), "FROM orders");
        assert_eq!(proj("SELECT * FROM orders"), Vec::<String>::new());
    }

    #[test]
    fn a_column_list_becomes_a_projection_not_a_clause() {
        // NQL has no projection, so the column list is carried separately and
        // applied to the returned rows.
        assert_eq!(q("SELECT status, total FROM orders"), "FROM orders");
        assert_eq!(proj("SELECT status, total FROM orders"), vec!["status", "total"]);
    }

    #[test]
    fn a_qualifier_reduces_to_the_field_while_an_ALIAS_is_the_name_the_client_sees() {
        // Two different jobs, and they used to be conflated. The SRC is what
        // NEDB reads out of the row, so a qualifier must be stripped from it.
        // The OUT is the name the CLIENT looks the column up by, so an alias
        // must be KEPT in it — `SELECT status AS s` returns a column called
        // `s`, and answering with one called `status` hands a client a result
        // it cannot find. SQLAlchemy writes `count(*) AS count_1` and then
        // reads `count_1`.
        let cols = cols_of("SELECT o.status AS s, o.total total, o.region FROM orders o");
        assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
                   vec!["status", "total", "region"]);
        assert_eq!(cols.iter().map(|c| c.out.clone()).collect::<Vec<_>>(),
                   vec!["s", "total", "region"]);
        assert_eq!(q("SELECT * FROM public.orders"), "FROM orders");
        assert_eq!(q("SELECT * FROM \"orders\""), "FROM orders");
    }

    #[test]
    fn a_select_list_may_MIX_columns_with_an_aggregate() {
        // What a GROUP BY query actually looks like. The previous parser
        // refused any list containing a parenthesis, so this whole shape was
        // unreachable even though NQL expresses it natively — and it is the
        // single most common grouped query an ORM emits.
        // The aggregate sits IMMEDIATELY AFTER the group key — verified
        // against the running engine, which refuses the other order with
        // "only one aggregate per query".
        assert_eq!(q("SELECT status, count(*) AS count_1 FROM orders GROUP BY status"),
                   "FROM orders GROUP BY status COUNT");
        // SQL puts GROUP BY before ORDER BY / LIMIT; the aggregate still lands
        // on the key, and the rest of the tail follows.
        assert_eq!(q("SELECT status, count(*) FROM orders WHERE total > 1 GROUP BY status ORDER BY status LIMIT 5"),
                   "FROM orders WHERE total > 1 GROUP BY status COUNT ORDER BY status LIMIT 5");
        // A bare aggregate with NO grouping still goes after the collection.
        assert_eq!(q("SELECT count(*) FROM orders"), "FROM orders COUNT");
        assert_eq!(q("SELECT sum(total) FROM orders"), "FROM orders SUM total");
        // More than one group key is refused by name: NQL groups by a single
        // field, and using only the first would aggregate over rows the query
        // meant to keep apart.
        let e = translate("SELECT status, count(*) FROM orders GROUP BY status, region").unwrap_err();
        assert!(e.contains("GROUP BY takes one key"), "{}", e);
        let cols = cols_of("SELECT status, count(*) AS count_1 FROM orders GROUP BY status");
        assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
                   vec!["status", "count"]);
        assert_eq!(cols.iter().map(|c| c.out.clone()).collect::<Vec<_>>(),
                   vec!["status", "count_1"]);

        // A named aggregate rides along with `count`, because an NQL grouped
        // row carries both.
        let cols = cols_of("SELECT status, count(*), sum(total) FROM orders GROUP BY status");
        assert_eq!(cols.iter().map(|c| c.src.clone()).collect::<Vec<_>>(),
                   vec!["status", "count", "sum_total"]);
        assert_eq!(q("SELECT status, count(*), sum(total) FROM orders GROUP BY status"),
                   "FROM orders GROUP BY status SUM total");

        // A qualifier on the aggregate's column is stripped like any other.
        assert_eq!(q("SELECT o.status, sum(o.total) FROM orders o GROUP BY o.status"),
                   "FROM orders GROUP BY status SUM total");

        // Two NAMED aggregates cannot both be carried, and that is refused by
        // name rather than silently dropping one.
        let e = translate("SELECT status, sum(total), avg(total) FROM orders GROUP BY status")
            .unwrap_err();
        assert!(e.contains("only one of SUM/AVG/MIN/MAX"), "{}", e);

        // A column that is neither a key nor an aggregate is still refused.
        let e = translate("SELECT status, total, count(*) FROM orders GROUP BY status")
            .unwrap_err();
        assert!(e.contains("must appear in the GROUP BY clause"), "{}", e);
    }

    #[test]
    fn ORDER_BY_an_ordinal_resolves_to_that_select_list_column() {
        // SQL lets a sort key be a POSITION, and clients write it constantly.
        // NQL has no ordinals — it read the `1` as a literal and refused with
        // "expected field name, got Num(1.0)". node-postgres sent
        // `GROUP BY status ORDER BY 1` in the harness's first run.
        assert_eq!(q("SELECT status, total FROM orders ORDER BY 1"),
                   "FROM orders ORDER BY status");
        assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 DESC"),
                   "FROM orders ORDER BY total DESC");
        // Several keys, mixing ordinals with names, and a direction on each.
        assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 DESC, 1"),
                   "FROM orders ORDER BY total DESC, status");
        assert_eq!(q("SELECT status, total FROM orders ORDER BY 1, total DESC"),
                   "FROM orders ORDER BY status, total DESC");
        // An ordinal survives the GROUP BY splice, and resolves to the group
        // key rather than to the literal 1 — which is the exact shape that
        // failed in CI.
        assert_eq!(q("SELECT status, count(*) AS n FROM orders GROUP BY status ORDER BY 1"),
                   "FROM orders GROUP BY status COUNT ORDER BY status");
        // An ordinal may name the AGGREGATE column too.
        assert_eq!(q("SELECT status, count(*) AS n FROM orders GROUP BY status ORDER BY 2 DESC"),
                   "FROM orders GROUP BY status COUNT ORDER BY count DESC");
        // The clause boundary is respected: a following LIMIT is not swallowed
        // into the sort list, and `LIMIT 1` is not mistaken for an ordinal.
        assert_eq!(q("SELECT status, total FROM orders ORDER BY 2 LIMIT 1"),
                   "FROM orders ORDER BY total LIMIT 1");
        // A `1` anywhere else stays a literal.
        assert_eq!(q("SELECT status FROM orders WHERE total > 1 ORDER BY 1"),
                   "FROM orders WHERE total > 1 ORDER BY status");

        // Out of range, and `SELECT *` where there is no list to index, are
        // both refused with the reason — guessing a column would sort by
        // something the query never named.
        let e = translate("SELECT status FROM orders ORDER BY 4").unwrap_err();
        assert!(e.contains("out of range") && e.contains("1 column"), "{}", e);
        let e = translate("SELECT * FROM orders ORDER BY 1").unwrap_err();
        assert!(e.contains("no list to index"), "{}", e);
    }

    #[test]
    fn count_of_a_subquery_flattens_only_when_the_two_counts_MUST_agree() {
        // `.count()` in every ORM wraps the whole query in a derived table.
        // Counting rows that ARE the inner query's rows is counting the inner
        // query, so this is an identity, not an approximation.
        assert_eq!(
            q("SELECT count(*) AS count_1 FROM (SELECT orders._id AS a, orders.status AS b \
               FROM orders WHERE orders.status = 'paid') AS anon_1"),
            // Verified against the running engine: with no GROUP BY the
            // aggregate may sit either side of WHERE and answers identically.
            r#"FROM orders COUNT WHERE status = "paid""#);
        // No predicate at all.
        assert_eq!(q("SELECT count(*) FROM (SELECT orders._id FROM orders) AS anon_1"),
                   "FROM orders COUNT");
        // ORDER BY cannot change a count, so it is dropped rather than refused.
        assert_eq!(q("SELECT count(*) FROM (SELECT _id FROM orders ORDER BY total DESC) AS a"),
                   "FROM orders COUNT");
        // The outer alias is the name the client reads the column back by.
        let cols = cols_of("SELECT count(*) AS count_1 FROM (SELECT _id FROM orders) AS a");
        assert_eq!(cols[0].src, "count");
        assert_eq!(cols[0].out, "count_1");

        // Each guard is a construct that would make the two counts DIFFERENT
        // numbers, so each is refused rather than silently flattened.
        for sql in [
            // LIMIT / OFFSET cap the rows before they are counted
            "SELECT count(*) FROM (SELECT _id FROM orders LIMIT 1) AS a",
            "SELECT count(*) FROM (SELECT _id FROM orders OFFSET 1) AS a",
            // the inner rows ARE the groups
            "SELECT count(*) FROM (SELECT status FROM orders GROUP BY status) AS a",
            // an inner aggregate already reduced the rows to one
            "SELECT count(*) FROM (SELECT count(*) FROM orders) AS a",
            "SELECT count(*) FROM (SELECT sum(total) FROM orders) AS a",
            // the outer list would need the derived table's own columns
            "SELECT count(*), status FROM (SELECT status FROM orders) AS a",
            "SELECT status FROM (SELECT status FROM orders) AS a",
            // one level is the claim
            "SELECT count(*) FROM (SELECT x FROM (SELECT _id AS x FROM orders) AS b) AS a",
        ] {
            let e = translate(sql).unwrap_err();
            assert!(e.contains("subqueries in FROM"), "{} -> {}", sql, e);
        }

        // DISTINCT and the set operators are caught EARLIER, by their own
        // rules, which scan the whole statement before the FROM list is even
        // read. Asserted separately so the test records which check owns each
        // refusal rather than implying one catch-all does.
        for (sql, needle) in [
            ("SELECT count(*) FROM (SELECT DISTINCT status FROM orders) AS a", "DISTINCT"),
            ("SELECT count(*) FROM (SELECT a FROM t UNION SELECT b FROM u) AS x", "UNION"),
        ] {
            let e = translate(sql).unwrap_err();
            assert!(e.contains(needle), "{} -> {}", sql, e);
        }
    }

    #[test]
    fn a_QUALIFIED_column_in_WHERE_finds_its_field_instead_of_ZERO_ROWS() {
        // THE silent wrong answer. NQL looks a field up FLAT, so
        // `WHERE orders.status = 'paid'` asked for a field literally named
        // "orders.status", no document had one, and the query returned ZERO
        // ROWS with no error — an empty result that reads exactly like "you
        // have no paid orders". Every ORM qualifies its predicates, so every
        // filtered SQLAlchemy query answered empty and `.get(pk)` answered
        // None.
        assert_eq!(q("SELECT _id FROM orders WHERE orders.status = 'paid'"),
                   r#"FROM orders WHERE status = "paid""#);
        assert_eq!(q("SELECT _id FROM orders WHERE orders.total > 50"),
                   "FROM orders WHERE total > 50");
        // Every clause in the tail, not just WHERE.
        assert_eq!(q("SELECT _id FROM orders ORDER BY orders.total DESC LIMIT 2"),
                   "FROM orders ORDER BY total DESC LIMIT 2");
        assert_eq!(q("SELECT status, count(*) FROM orders GROUP BY orders.status"),
                   "FROM orders GROUP BY status COUNT");

        // An alias is a legal qualifier and is accepted as one. It is also
        // REMOVED from the tail, because NQL has no alias syntax and reported
        // an "unexpected token" on it.
        assert_eq!(q("SELECT o.status FROM orders o WHERE o.status = 'paid'"),
                   r#"FROM orders WHERE status = "paid""#);
        assert_eq!(q("SELECT o.status FROM orders AS o WHERE o.total > 1"),
                   "FROM orders WHERE total > 1");

        // A qualifier naming NEITHER the collection nor its alias is an
        // ERROR, not a strip. Stripping it would answer from the one relation
        // that IS present, which is a different wrong answer in the same
        // empty-looking clothes.
        let e = translate("SELECT _id FROM orders WHERE nosuch.status = 'paid'").unwrap_err();
        assert!(e.contains("no table or alias named \"nosuch\""), "{}", e);
        let e = translate("SELECT _id FROM orders o WHERE p.status = 'paid'").unwrap_err();
        assert!(e.contains("aliased \"o\""), "the message names the alias in scope: {}", e);

        // A dot INSIDE a literal is data, not a qualifier.
        assert_eq!(q("SELECT _id FROM orders WHERE status = 'pa.id'"),
                   r#"FROM orders WHERE status = "pa.id""#);
        // ...and a decimal point is not one either.
        assert_eq!(q("SELECT _id FROM orders WHERE total > 1.5"),
                   "FROM orders WHERE total > 1.5");

        // UPDATE and DELETE carry the same tail, and had the same bug.
        match translate("UPDATE orders o SET status = 'x' WHERE o.total > 5").unwrap() {
            Stmt::Update { coll, nql, .. } => {
                assert_eq!(coll, "orders", "the alias is not part of the collection name");
                assert_eq!(nql, "FROM orders WHERE total > 5");
            }
            other => panic!("{:?}", other),
        }
        match translate("DELETE FROM orders o WHERE o.status = 'paid'").unwrap() {
            Stmt::Delete { coll, nql, .. } => {
                assert_eq!(coll, "orders");
                assert_eq!(nql, r#"FROM orders WHERE status = "paid""#);
            }
            other => panic!("{:?}", other),
        }

        // `AS OF SYSTEM TIME` also begins with AS and is NOT an alias.
        assert_eq!(q("SELECT _id FROM orders AS OF SYSTEM TIME 3 WHERE orders.total > 1"),
                   "FROM orders AS OF 3 WHERE total > 1");
    }

    #[test]
    fn where_clauses_pass_through_with_sql_literals_rewritten() {
        assert_eq!(q("SELECT * FROM orders WHERE status = 'paid'"),
                   r#"FROM orders WHERE status = "paid""#);
        assert_eq!(q("SELECT * FROM orders WHERE status <> 'paid'"),
                   r#"FROM orders WHERE status != "paid""#);
        assert_eq!(q("SELECT * FROM orders WHERE status IN ('paid','open')"),
                   r#"FROM orders WHERE status IN ("paid","open")"#);
    }

    /// SQL escapes an embedded quote by doubling it. That must become ONE
    /// character inside the NQL string, not terminate it.
    #[test]
    fn a_doubled_sql_quote_is_one_literal_character() {
        assert_eq!(q("SELECT * FROM t WHERE name = 'it''s'"),
                   r#"FROM t WHERE name = "it's""#);
    }

    /// A double quote inside a SQL literal has to be escaped for NQL, whose
    /// lexer collapses \" — otherwise it would close the string early.
    #[test]
    fn a_double_quote_inside_a_sql_literal_is_escaped_for_nql() {
        assert_eq!(q(r#"SELECT * FROM t WHERE name = 'say "hi"'"#),
                   r#"FROM t WHERE name = "say \"hi\"""#);
    }

    #[test]
    fn the_shared_clauses_are_handed_to_nql_unchanged() {
        assert_eq!(q("SELECT * FROM orders ORDER BY total DESC LIMIT 10 OFFSET 5"),
                   "FROM orders ORDER BY total DESC LIMIT 10 OFFSET 5");
        assert_eq!(q("SELECT * FROM orders GROUP BY region"), "FROM orders GROUP BY region");
        assert_eq!(q("SELECT * FROM o WHERE total BETWEEN 1 AND 9 ORDER BY a, b DESC"),
                   "FROM o WHERE total BETWEEN 1 AND 9 ORDER BY a, b DESC");
    }

    /// An aggregate must surface as ONE column, named as SQL names it.
    ///
    /// NQL answers `SUM(total)` with `{count, sum_total, value}` — `value`
    /// being a back-compat alias. Passing that straight through gave
    /// `SELECT COUNT(*)` two columns (`count`, `value`) where SQL promises
    /// one, and leaked an internal key name onto the wire.
    #[test]
    fn an_aggregate_is_one_column_named_as_sql_names_it() {
        assert_eq!(proj_pairs("SELECT COUNT(*) FROM orders"),
                   vec![("count".to_string(), "count".to_string())]);
        assert_eq!(proj_pairs("SELECT SUM(total) FROM orders"),
                   vec![("sum_total".to_string(), "sum".to_string())]);
        assert_eq!(proj_pairs("SELECT avg(total) FROM orders"),
                   vec![("avg_total".to_string(), "avg".to_string())]);
        assert_eq!(proj_pairs("SELECT MIN(total) FROM orders"),
                   vec![("min_total".to_string(), "min".to_string())]);
        // And the encoded result really is one column with that name.
        let rows = vec![json!({"count": 4, "sum_total": 420, "value": 420})];
        let p = vec![Col::renamed("sum_total", "sum")];
        let cols = columns_for(&rows, &p);
        assert_eq!(names(&cols), vec!["sum"], "one column, SQL's name");
        assert_eq!(cell(rows[0].get(&cols[0].src)), Some("420".to_string()));
    }

    /// A grouped NQL row holds the group key, `count` and the aggregate —
    /// nothing else. Projecting another column found nothing and rendered
    /// NULL, which is a silent wrong answer. Postgres errors; so do we, in
    /// Postgres's own words.
    #[test]
    fn a_bare_column_with_group_by_is_refused_not_nulled() {
        let e = translate("SELECT region, total FROM orders GROUP BY region").unwrap_err();
        assert!(e.contains("must appear in the GROUP BY clause"), "{}", e);
        assert!(e.contains("total"), "the message names the offending column: {}", e);

        // The group key itself, and `count`, are both legitimate.
        assert!(translate("SELECT region FROM orders GROUP BY region").is_ok());
        assert!(translate("SELECT region, count FROM orders GROUP BY region").is_ok());
        // As is an aggregate over the grouped set.
        assert!(translate("SELECT SUM(total) FROM orders GROUP BY region").is_ok());
        // And `*` is unaffected — it returns whatever the grouped row holds.
        assert!(translate("SELECT * FROM orders GROUP BY region").is_ok());
    }

    #[test]
    fn count_star_becomes_nql_count() {
        assert_eq!(q("SELECT COUNT(*) FROM orders"), "FROM orders COUNT");
        assert_eq!(q("SELECT count(*) FROM orders WHERE total > 5"),
                   "FROM orders COUNT WHERE total > 5");
    }

    #[test]
    fn aggregates_carry_their_target_column() {
        assert_eq!(q("SELECT SUM(total) FROM orders"), "FROM orders SUM total");
        assert_eq!(q("SELECT avg(total) FROM orders WHERE region = 'eu'"),
                   r#"FROM orders AVG total WHERE region = "eu""#);
        assert!(translate("SELECT SUM(*) FROM orders").is_err());
    }

    /// The bridge worth having: Postgres spells time travel
    /// `AS OF SYSTEM TIME`, and NEDB's is sequence-addressed and permanent.
    #[test]
    fn as_of_system_time_bridges_to_nql_as_of() {
        assert_eq!(q("SELECT * FROM orders AS OF SYSTEM TIME 42"),
                   "FROM orders AS OF 42");
        assert_eq!(q("SELECT * FROM orders AS OF SYSTEM TIME 42 WHERE total > 1"),
                   "FROM orders AS OF 42 WHERE total > 1");
        // A wall-clock timestamp is refused with the reason, not silently ignored.
        let e = translate("SELECT * FROM orders AS OF SYSTEM TIME '2026-01-01'").unwrap_err();
        assert!(e.contains("sequence number"), "{}", e);
    }

    /// A select-list item that is not a column reference must be REFUSED, not
    /// turned into a field name.
    ///
    /// The guard used to be `expr.contains('(')`, which only catches expressions
    /// that happen to have a paren. `total * 2` sailed through, became the field
    /// name "total * 2", matched no document, and the column came back EMPTY for
    /// every row with no error. Same silent class as the qualified-WHERE bug: a
    /// wrong answer wearing the shape of data.
    #[test]
    fn a_select_list_expression_is_refused_rather_than_answered_blank() {
        for sql in [
            "SELECT total * 2 FROM orders",
            "SELECT total, total*2 AS doubled FROM orders",
            "SELECT total + 1 FROM orders",
            "SELECT status || 'x' FROM orders",
            "SELECT -total FROM orders",
            "SELECT lower(status) FROM orders",
        ] {
            let e = translate(sql).unwrap_err();
            assert!(e.contains("expressions in the select list"), "{} -> {}", sql, e);
        }
        // ...and the things that ARE column references still pass, or the fix
        // would have bought correctness by refusing everything.
        assert_eq!(q("SELECT _id, status FROM orders"), "FROM orders");
        assert_eq!(q("SELECT \"status\" FROM orders"), "FROM orders");
        assert_eq!(q("SELECT orders.status FROM orders"), "FROM orders");
        assert_eq!(q("SELECT o.status FROM orders o"), "FROM orders");
        assert_eq!(q("SELECT total AS t FROM orders"), "FROM orders");
        assert!(translate("SELECT count(*) FROM orders").is_ok());
        assert!(translate("SELECT sum(total) FROM orders").is_ok());
    }

    /// HAVING has to reach NQL in the spelling NQL's grouped row actually uses.
    ///
    /// An NQL grouped row carries `count` and `<agg>_<field>`. SQL clients write
    /// `count(*)`, or the alias they gave it. `count(*)` failed LOUDLY (fine),
    /// but `COUNT` and an alias both passed through verbatim and answered ZERO
    /// ROWS — which reads as "no groups qualified" rather than "your predicate
    /// named a field that does not exist".
    #[test]
    fn having_is_translated_to_nqls_spelling_and_refuses_an_unknown_key() {
        // Every spelling a client might send for the count.
        for sql in [
            "SELECT status, count(*) AS n FROM orders GROUP BY status HAVING count(*) > 1",
            "SELECT status, count(*) AS n FROM orders GROUP BY status HAVING n > 1",
            "SELECT status, count(*) FROM orders GROUP BY status HAVING COUNT > 1",
            "SELECT status, count(*) FROM orders GROUP BY status HAVING count > 1",
        ] {
            let got = q(sql);
            assert_eq!(got, "FROM orders GROUP BY status COUNT HAVING count > 1",
                       "{} -> {}", sql, got);
        }
        // A named aggregate, by its alias -- NQL calls the field `sum_total`.
        assert_eq!(q("SELECT status, sum(total) AS s FROM orders GROUP BY status HAVING s > 100"),
                   "FROM orders GROUP BY status SUM total HAVING sum_total > 100");
        // ...and by NQL's own name for it, which must not be rewritten twice.
        assert_eq!(q("SELECT status, sum(total) FROM orders GROUP BY status HAVING sum_total > 100"),
                   "FROM orders GROUP BY status SUM total HAVING sum_total > 100");
        // Filtering on the group key itself is legitimate and passes through
        // untouched -- the SQL literal becomes an NQL one, as everywhere else.
        assert_eq!(q("SELECT status, count(*) FROM orders GROUP BY status HAVING status > 'a'"),
                   "FROM orders GROUP BY status COUNT HAVING status > \"a\"");
        // A key the grouped row cannot carry is an ERROR, not zero rows.
        let e = translate(
            "SELECT status, count(*) FROM orders GROUP BY status HAVING nosuch > 1").unwrap_err();
        assert!(e.contains("HAVING names") && e.contains("nosuch"), "{}", e);
        assert!(e.contains("zero rows"), "the message must say what it prevented: {}", e);
    }

    #[test]
    fn handshake_queries_are_answered_so_clients_can_connect() {
        assert!(matches!(translate("SELECT version()"), Ok(Stmt::Canned { .. })));
        assert!(matches!(translate("SHOW transaction_isolation"), Ok(Stmt::Canned { .. })));
        assert!(matches!(translate("SELECT current_schema()"), Ok(Stmt::Canned { .. })));
        assert!(matches!(translate("SET extra_float_digits = 3"), Ok(Stmt::Ok(_))));
        assert!(matches!(translate("BEGIN"), Ok(Stmt::Ok(_))));
        assert!(matches!(translate(""), Ok(Stmt::Ok(_))));
    }

    /// Every refusal has to name the boundary. "Syntax error" would send a
    /// developer hunting for a typo that is not there.
    #[test]
    fn unsupported_sql_is_refused_with_a_reason() {
        for (sql, expect) in [
            ("INSERT INTO t VALUES (1)", "explicit column list"),
            ("CREATE TABLE t (a int)", "DDL"),
            ("TRUNCATE t", "append-only"),
            ("GRANT ALL ON t TO x", "privilege system"),
            ("SELECT * FROM a JOIN b ON a.x = b.x", "JOIN is not supported"),
            ("SELECT * FROM a UNION SELECT * FROM b", "UNION"),
            ("SELECT DISTINCT region FROM orders", "GROUP BY"),
            ("SELECT * FROM (SELECT 1) x", "subqueries in FROM"),
            ("SELECT * FROM a, b", "more than one collection"),
            ("SELECT lower(status) FROM orders", "expressions in the select list"),
            ("VACUUM", "only SELECT"),
        ] {
            let e = translate(sql).unwrap_err();
            assert!(e.contains(expect), "for {:?} expected {:?} in {:?}", sql, expect, e);
        }
    }

    // ── writes ───────────────────────────────────────────────────────────────
    //
    // SQL's write semantics and NEDB's append-only model line up: INSERT is a
    // put, UPDATE is a new version, DELETE is a tombstone. These tests pin the
    // parse; tests/test_pgwire.py proves the behaviour against a live server,
    // including that the PRIOR value is still readable afterwards.

    fn ins(sql: &str) -> (String, Vec<InsertRow>, Vec<Col>) {
        match translate(sql) {
            Ok(Stmt::Insert { coll, rows, returning }) => (coll, rows, returning),
            other => panic!("expected INSERT for {:?}, got {:?}", sql, other),
        }
    }

    #[test]
    fn insert_becomes_a_put_per_row() {
        let (coll, rows, ret) = ins("INSERT INTO orders (_id, status, total) VALUES ('o1', 'paid', 120)");
        assert_eq!(coll, "orders");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].id.as_deref(), Some("o1"));
        assert_eq!(rows[0].doc.get("status"), Some(&json!("paid")));
        assert_eq!(rows[0].doc.get("total"), Some(&json!(120)));
        // `_id` is the key, not a payload field.
        assert!(!rows[0].doc.contains_key("_id"));
        assert!(ret.is_empty());
    }

    #[test]
    fn a_multi_row_insert_yields_one_row_each() {
        let (_, rows, _) = ins(
            "INSERT INTO t (id, n) VALUES ('a', 1), ('b', 2), ('c', 3)");
        assert_eq!(rows.len(), 3);
        assert_eq!(rows[1].id.as_deref(), Some("b"));
        assert_eq!(rows[2].doc.get("n"), Some(&json!(3)));
    }

    #[test]
    fn an_insert_without_an_id_column_lets_the_server_assign_one() {
        let (_, rows, _) = ins("INSERT INTO t (n) VALUES (1)");
        assert_eq!(rows[0].id, None, "the executor mints a unique key");
        assert_eq!(rows[0].doc.get("n"), Some(&json!(1)));
    }

    /// Provenance is reachable from SQL, not only from the HTTP API — which is
    /// the point of having writes here at all.
    #[test]
    fn insert_lifts_provenance_out_of_reserved_columns() {
        let (_, rows, _) = ins(
            "INSERT INTO audit (_id, _caused_by, _valid_from, kind) \
             VALUES ('e1', 'abc123', '2026-01-01', 'reprice')");
        assert_eq!(rows[0].caused_by, vec!["abc123".to_string()]);
        assert_eq!(rows[0].valid_from.as_deref(), Some("2026-01-01"));
        assert_eq!(rows[0].doc.get("kind"), Some(&json!("reprice")));
        // None of the reserved names leak into the stored payload.
        for k in ["_id", "_caused_by", "_valid_from"] {
            assert!(!rows[0].doc.contains_key(k), "{} leaked into the doc", k);
        }
    }

    #[test]
    fn insert_values_cover_the_scalar_types() {
        let (_, rows, _) = ins(
            "INSERT INTO t (s, i, f, b, n) VALUES ('x', 42, 1.5, TRUE, NULL)");
        assert_eq!(rows[0].doc.get("s"), Some(&json!("x")));
        assert_eq!(rows[0].doc.get("i"), Some(&json!(42)));
        assert_eq!(rows[0].doc.get("f"), Some(&json!(1.5)));
        assert_eq!(rows[0].doc.get("b"), Some(&json!(true)));
        assert_eq!(rows[0].doc.get("n"), Some(&Value::Null));
    }

    /// A doubled '' is one literal quote, and a comma inside a string is not a
    /// value separator.
    #[test]
    fn insert_literals_survive_quotes_and_commas() {
        let (_, rows, _) = ins("INSERT INTO t (a, b) VALUES ('it''s', 'x,y')");
        assert_eq!(rows[0].doc.get("a"), Some(&json!("it's")));
        assert_eq!(rows[0].doc.get("b"), Some(&json!("x,y")));
    }

    #[test]
    fn insert_refuses_what_it_cannot_store_faithfully() {
        // An unevaluated expression stored as text would be a wrong value.
        assert!(translate("INSERT INTO t (a) VALUES (1 + 1)").is_err());
        assert!(translate("INSERT INTO t (a) VALUES (now())").is_err());
        // Column/value count mismatch.
        let e = translate("INSERT INTO t (a, b) VALUES (1)").unwrap_err();
        assert!(e.contains("values for"), "{}", e);
        // No column list at all.
        let e2 = translate("INSERT INTO t VALUES (1)").unwrap_err();
        assert!(e2.contains("explicit column list"), "{}", e2);
    }

    #[test]
    fn update_finds_rows_with_the_full_predicate_surface() {
        match translate("UPDATE orders SET status = 'void' WHERE total < 50 AND region IN ('eu')") {
            Ok(Stmt::Update { coll, set, nql, .. }) => {
                assert_eq!(coll, "orders");
                assert_eq!(set, vec![("status".to_string(), json!("void"))]);
                // The WHERE became ordinary NQL, so IN/BETWEEN/LIKE all work.
                assert_eq!(nql, r#"FROM orders WHERE total < 50 AND region IN ("eu")"#);
            }
            other => panic!("expected UPDATE, got {:?}", other),
        }
    }

    #[test]
    fn update_without_where_targets_the_whole_collection() {
        // Postgres allows it, so parity allows it.
        match translate("UPDATE t SET a = 1") {
            Ok(Stmt::Update { nql, .. }) => assert_eq!(nql, "FROM t"),
            other => panic!("expected UPDATE, got {:?}", other),
        }
    }

    #[test]
    fn update_handles_several_assignments() {
        match translate("UPDATE t SET a = 1, b = 'x,y', c = NULL WHERE id = 'k'") {
            Ok(Stmt::Update { set, .. }) => {
                assert_eq!(set.len(), 3);
                assert_eq!(set[1], ("b".to_string(), json!("x,y")));
                assert_eq!(set[2], ("c".to_string(), Value::Null));
            }
            other => panic!("expected UPDATE, got {:?}", other),
        }
        assert!(translate("UPDATE t SET").is_err());
        assert!(translate("UPDATE t SET a").is_err());
    }

    #[test]
    fn delete_becomes_a_predicate_over_the_collection() {
        match translate("DELETE FROM orders WHERE status = 'void'") {
            Ok(Stmt::Delete { coll, nql, .. }) => {
                assert_eq!(coll, "orders");
                assert_eq!(nql, r#"FROM orders WHERE status = "void""#);
            }
            other => panic!("expected DELETE, got {:?}", other),
        }
        match translate("DELETE FROM t") {
            Ok(Stmt::Delete { nql, .. }) => assert_eq!(nql, "FROM t"),
            other => panic!("expected DELETE, got {:?}", other),
        }
    }

    #[test]
    fn returning_is_parsed_off_every_write() {
        let (_, _, ret) = ins("INSERT INTO t (a) VALUES (1) RETURNING a, _id");
        assert_eq!(ret.iter().map(|c| c.out.clone()).collect::<Vec<_>>(), vec!["a", "_id"]);
        // `RETURNING *` is an empty projection — every column — which is why
        // the executor checks the raw SQL for the keyword instead.
        let (_, _, star) = ins("INSERT INTO t (a) VALUES (1) RETURNING *");
        assert!(star.is_empty());
        assert!(wants_returning("INSERT INTO t (a) VALUES (1) RETURNING *"));
        assert!(!wants_returning("INSERT INTO t (a) VALUES (1)"));

        match translate("UPDATE t SET a = 1 WHERE id = 'k' RETURNING a") {
            Ok(Stmt::Update { nql, returning, .. }) => {
                assert_eq!(returning.len(), 1);
                // RETURNING must NOT leak into the predicate.
                assert!(!nql.to_uppercase().contains("RETURNING"), "{}", nql);
            }
            other => panic!("expected UPDATE, got {:?}", other),
        }
        match translate("DELETE FROM t WHERE id = 'k' RETURNING *") {
            Ok(Stmt::Delete { nql, .. }) =>
                assert!(!nql.to_uppercase().contains("RETURNING"), "{}", nql),
            other => panic!("expected DELETE, got {:?}", other),
        }
    }

    #[test]
    fn a_keyword_inside_a_value_is_not_a_clause() {
        match translate("UPDATE t SET note = 'where returning from' WHERE id = 'k'") {
            Ok(Stmt::Update { set, nql, .. }) => {
                assert_eq!(set[0].1, json!("where returning from"));
                assert_eq!(nql, r#"FROM t WHERE id = "k""#);
            }
            other => panic!("expected UPDATE, got {:?}", other),
        }
    }

    #[test]
    fn split_top_respects_quotes_and_nesting() {
        assert_eq!(split_top("a, b, c", ',').len(), 3);
        assert_eq!(split_top("(1, 2), (3, 4)", ',').len(), 2);
        assert_eq!(split_top("'a,b', c", ',').len(), 2);
        assert_eq!(split_top("'it''s, fine', c", ',').len(), 2);
    }

    #[test]
    fn comments_and_whitespace_do_not_confuse_the_translator() {
        assert_eq!(q("SELECT *\n  FROM orders  -- trailing note\n"), "FROM orders");
        assert_eq!(q("SELECT /* inline */ * FROM orders"), "FROM orders");
        // A keyword inside a string literal must not be treated as a clause.
        assert_eq!(q("SELECT * FROM t WHERE note = 'from here to JOIN'"),
                   r#"FROM t WHERE note = "from here to JOIN""#);
    }

    #[test]
    fn find_kw_ignores_quotes_parens_and_substrings() {
        assert_eq!(find_kw("SELECT A FROM B", "FROM"), Some(9));
        assert_eq!(find_kw("SELECT 'FROM' FROM B", "FROM"), Some(14));
        assert_eq!(find_kw("SELECT F(x FROM y) FROM B", "FROM"), Some(19));
        assert_eq!(find_kw("SELECT FROMAGE", "FROM"), None);
        assert_eq!(find_kw("SELECT X_FROM", "FROM"), None);
    }

    // ── result encoding ──────────────────────────────────────────────────────

    #[test]
    fn provenance_columns_sort_after_the_users_own_fields() {
        let rows = vec![json!({"_id":"1","_hash":"ab","status":"paid","total":9})];
        assert_eq!(names(&columns_for(&rows, &[])),
                   vec!["status", "total", "_hash", "_id"]);
    }

    #[test]
    fn an_explicit_projection_sets_the_column_order() {
        let rows = vec![json!({"a":1,"b":2})];
        let p = vec![Col::same("b"), Col::same("a")];
        assert_eq!(names(&columns_for(&rows, &p)), vec!["b", "a"]);
    }

    #[test]
    fn columns_are_the_union_across_sparse_rows() {
        // A document store has no schema, so row 2 may carry a field row 1 lacks.
        let rows = vec![json!({"a":1}), json!({"b":2})];
        assert_eq!(names(&columns_for(&rows, &[])), vec!["a", "b"]);
    }

    #[test]
    fn type_oids_follow_the_first_non_null_value() {
        let rows = vec![json!({"i":1,"f":1.5,"b":true,"s":"x","n":null})];
        assert_eq!(oid_for(&rows, "i"), OID_INT8);
        assert_eq!(oid_for(&rows, "f"), OID_FLOAT8);
        assert_eq!(oid_for(&rows, "b"), OID_BOOL);
        assert_eq!(oid_for(&rows, "s"), OID_TEXT);
        // All-null and absent columns fall back to text rather than guessing.
        assert_eq!(oid_for(&rows, "n"), OID_TEXT);
        assert_eq!(oid_for(&rows, "absent"), OID_TEXT);
    }

    #[test]
    fn a_column_that_is_null_in_the_first_row_still_gets_its_type() {
        let rows = vec![json!({"v": null}), json!({"v": 7})];
        assert_eq!(oid_for(&rows, "v"), OID_INT8);
    }

    #[test]
    fn cells_render_in_postgres_text_format() {
        assert_eq!(cell(Some(&json!("x"))), Some("x".to_string()));
        assert_eq!(cell(Some(&json!(true))), Some("t".to_string()));
        assert_eq!(cell(Some(&json!(false))), Some("f".to_string()));
        assert_eq!(cell(Some(&json!(42))), Some("42".to_string()));
        assert_eq!(cell(Some(&json!(null))), None);
        assert_eq!(cell(None), None);
        // Nested values render as JSON text rather than being dropped.
        assert_eq!(cell(Some(&json!({"a":1}))), Some("{\"a\":1}".to_string()));
    }

    /// The framing has to be exact or the client desynchronises and hangs.
    /// Length covers the length field itself but not the tag byte.
    #[test]
    fn message_framing_length_excludes_the_tag() {
        let mut m = Out::msg(b'Z');
        m.bytes(b"I");
        let bytes = m.finish();
        assert_eq!(bytes[0], b'Z');
        assert_eq!(i32::from_be_bytes([bytes[1], bytes[2], bytes[3], bytes[4]]), 5);
        assert_eq!(bytes.len(), 6);
    }

    #[test]
    fn a_result_set_encodes_as_description_then_rows_then_complete() {
        let rows = vec![json!({"a": 1}), json!({"a": 2})];
        let out = encode_result(&rows, &[]);
        assert_eq!(out[0], b'T');
        let tags: Vec<u8> = {
            // Walk the message stream by its own length prefixes.
            let mut t = vec![];
            let mut i = 0usize;
            while i < out.len() {
                t.push(out[i]);
                let len = i32::from_be_bytes([out[i+1], out[i+2], out[i+3], out[i+4]]) as usize;
                i += 1 + len;
            }
            t
        };
        assert_eq!(tags, vec![b'T', b'D', b'D', b'C'],
                   "one description, one row each, one completion");
    }

    /// A statement must emit EXACTLY ONE CommandComplete. A write with
    /// RETURNING that reused the SELECT encoder sent two, and the visible
    /// symptom was RETURNING yielding no rows: the client took the first tag
    /// as the end of the statement and threw the description away.
    #[test]
    fn a_write_with_returning_emits_exactly_one_command_complete() {
        let rows = vec![json!({"_id": "o1", "total": 9})];
        let mut out = encode_rows(&rows, &[Col::same("_id")]);
        out.extend_from_slice(&command_complete("INSERT 0 1"));
        let mut tags = vec![];
        let mut i = 0usize;
        while i < out.len() {
            tags.push(out[i]);
            let len = i32::from_be_bytes([out[i+1], out[i+2], out[i+3], out[i+4]]) as usize;
            i += 1 + len;
        }
        assert_eq!(tags, vec![b'T', b'D', b'C'], "one description, one row, ONE tag");
        assert_eq!(tags.iter().filter(|t| **t == b'C').count(), 1);
        // encode_rows alone must not carry a tag at all.
        assert!(!encode_rows(&rows, &[]).contains(&b'C')
                || encode_rows(&rows, &[]).iter().filter(|b| **b == b'C').count() > 0);
        let bare = encode_rows(&rows, &[Col::same("_id")]);
        let mut bare_tags = vec![];
        let mut j = 0usize;
        while j < bare.len() {
            bare_tags.push(bare[j]);
            let len = i32::from_be_bytes([bare[j+1], bare[j+2], bare[j+3], bare[j+4]]) as usize;
            j += 1 + len;
        }
        assert_eq!(bare_tags, vec![b'T', b'D'], "encode_rows never appends a tag");
    }

    #[test]
    fn an_empty_result_still_sends_a_description() {
        let out = encode_result(&[], &[Col::same("a")]);
        assert_eq!(out[0], b'T', "clients need the shape even with no rows");
    }

    #[test]
    fn statements_split_on_top_level_semicolons_only() {
        assert_eq!(split_statements("SELECT 1; SELECT 2").len(), 2);
        assert_eq!(split_statements("SELECT ';'").len(), 1);
        assert_eq!(split_statements("SELECT 1;").len(), 1);
        assert_eq!(split_statements("   ").len(), 0);
    }

    #[test]
    fn an_error_names_its_sqlstate() {
        let e = String::from_utf8_lossy(&err_msg("0A000", "x")).to_string();
        assert!(e.contains("ERROR"));
        assert!(e.contains("0A000"));
    }

    // ── the extended query protocol ─────────────────────────────────────────

    #[test]
    fn placeholders_are_counted_outside_string_literals() {
        assert_eq!(param_count("SELECT a FROM t WHERE b = $1 AND c = $2"), 2);
        assert_eq!(param_count("SELECT a FROM t"), 0);
        // The highest index wins, because a parameter may be reused.
        assert_eq!(param_count("WHERE a = $2 OR b = $2 OR c = $1"), 2);
        assert_eq!(param_count("SELECT a FROM t WHERE b = '$1'"), 0,
                   "a placeholder inside a literal is data, not a parameter");
        assert_eq!(param_count("WHERE a = $10 AND b = $1"), 10,
                   "two-digit indexes must not be read as $1 followed by 0");
    }

    #[test]
    fn parameters_are_spliced_as_literals() {
        let out = substitute_params("WHERE a = $1 AND b = $2 AND c = $3",
            &[Some("'x'".into()), Some("42".into()), None]).unwrap();
        assert_eq!(out, "WHERE a = 'x' AND b = 42 AND c = NULL");
    }

    #[test]
    fn substitution_leaves_string_literals_alone() {
        let out = substitute_params("WHERE a = '$1' AND b = $1", &[Some("9".into())]).unwrap();
        assert_eq!(out, "WHERE a = '$1' AND b = 9");
    }

    #[test]
    fn too_few_parameters_is_an_error_not_a_silent_null() {
        // The alternative — treating a missing parameter as NULL — turns a
        // client bug into a wrong answer with a 200-shaped response.
        let e = substitute_params("WHERE a = $2", &[Some("1".into())]).unwrap_err();
        assert!(e.contains("$2"), "{}", e);
    }

    #[test]
    fn a_quote_in_a_parameter_cannot_escape_its_literal() {
        let lit = decode_param(Some(b"it's"), OID_TEXT, 0).unwrap().unwrap();
        assert_eq!(lit, "'it''s'");
        // And it survives a round trip through the splice unchanged.
        let out = substitute_params("WHERE a = $1", &[Some(lit)]).unwrap();
        assert_eq!(out, "WHERE a = 'it''s'");
    }

    #[test]
    fn binary_parameters_decode_in_every_width_psycopg_sends() {
        // These are the exact encodings read off a psycopg3 wire transcript:
        // a small int arrives as int2, a float as float8, a bool as one byte.
        assert_eq!(decode_param(Some(&[0x00, 0x2a]), OID_INT2, 1).unwrap().unwrap(), "42");
        assert_eq!(decode_param(Some(&[0, 0, 0, 7]), OID_INT4, 1).unwrap().unwrap(), "7");
        assert_eq!(
            decode_param(Some(&[0, 0, 0, 0, 0, 0, 0, 9]), OID_INT8, 1).unwrap().unwrap(), "9");
        assert_eq!(
            decode_param(Some(&0x400c_0000_0000_0000u64.to_be_bytes()), OID_FLOAT8, 1)
                .unwrap().unwrap(), "3.5");
        assert_eq!(decode_param(Some(&[1]), OID_BOOL, 1).unwrap().unwrap(), "TRUE");
        assert_eq!(decode_param(Some(&[0]), OID_BOOL, 1).unwrap().unwrap(), "FALSE");
    }

    #[test]
    fn a_negative_binary_integer_keeps_its_sign() {
        assert_eq!(decode_param(Some(&(-5i32).to_be_bytes()), OID_INT4, 1).unwrap().unwrap(), "-5");
        assert_eq!(decode_param(Some(&(-5i16).to_be_bytes()), OID_INT2, 1).unwrap().unwrap(), "-5");
    }

    #[test]
    fn a_binary_parameter_of_the_wrong_width_is_refused() {
        // Truncating or zero-extending would produce a plausible wrong number,
        // which is the failure mode worth engineering against.
        let e = decode_param(Some(&[0x2a]), OID_INT4, 1).unwrap_err();
        assert!(e.contains("4 bytes"), "{}", e);
    }

    #[test]
    fn an_unspecified_text_parameter_is_treated_as_a_string() {
        // psycopg3 declares OID 0 only for `str`; every number it sends carries
        // a real numeric OID. So quoting here is grounded, not a guess.
        assert_eq!(decode_param(Some(b"hello"), 0, 0).unwrap().unwrap(), "'hello'");
    }

    #[test]
    fn a_null_parameter_decodes_to_none_in_every_format() {
        assert_eq!(decode_param(None, OID_TEXT, 0).unwrap(), None);
        assert_eq!(decode_param(None, OID_INT8, 1).unwrap(), None);
    }

    #[test]
    fn an_unsupported_binary_type_says_so_by_name() {
        let e = decode_param(Some(&[0u8; 8]), 1114, 1).unwrap_err();
        assert!(e.contains("1114"), "{}", e);
        assert!(e.contains("text"), "the error should point at the way out: {}", e);
    }

    #[test]
    fn a_text_number_that_is_not_a_number_gets_quoted() {
        // Splicing it in bare would emit a naked identifier into the NQL text
        // and fail somewhere far away from the cause.
        assert_eq!(decode_param(Some(b"oops"), OID_INT8, 0).unwrap().unwrap(), "'oops'");
    }

    #[test]
    fn a_client_declared_type_is_believed_over_inference() {
        // The client is about to encode its argument that way; overriding it
        // would break the decode.
        let oids = infer_param_oids("SELECT a FROM t WHERE b = $1 AND c = $2", &[OID_INT4, 0], None);
        assert_eq!(oids, vec![OID_INT4, OID_TEXT]);
    }

    #[test]
    fn parameter_arity_is_taken_from_the_sql_when_the_client_declares_none() {
        // asyncpg declares nothing and then refuses the call if the count that
        // comes back is wrong, so this is the load-bearing path for it.
        let oids = infer_param_oids("SELECT a FROM t WHERE b = $1 AND c = $2", &[], None);
        assert_eq!(oids.len(), 2);
    }

    #[test]
    fn the_field_behind_each_placeholder_is_identified() {
        assert_eq!(
            param_fields("SELECT a FROM t WHERE qty > $1 AND status = $2", 2),
            vec![Some("qty".to_string()), Some("status".to_string())]);
    }

    #[test]
    fn word_operators_do_not_hide_the_field() {
        assert_eq!(param_fields("SELECT a FROM t WHERE name LIKE $1", 1),
                   vec![Some("name".to_string())]);
        assert_eq!(param_fields("SELECT a FROM t WHERE qty BETWEEN $1 AND $2", 2),
                   vec![Some("qty".to_string()), Some("qty".to_string())]);
        assert_eq!(param_fields("SELECT a FROM t WHERE region IN ($1, $2)", 2),
                   vec![Some("region".to_string()), Some("region".to_string())]);
    }

    #[test]
    fn a_clause_position_types_from_the_grammar_not_from_a_column() {
        // `AS OF SYSTEM TIME $1` has no column beside it — the token to its
        // left is the word TIME. Typing it text made asyncpg refuse to send
        // the sequence number at all.
        assert_eq!(
            infer_param_oids("SELECT a FROM t AS OF SYSTEM TIME $1 WHERE b = $2", &[], None),
            vec![OID_INT8, OID_TEXT]);
        assert_eq!(infer_param_oids("SELECT a FROM t AS OF $1", &[], None), vec![OID_INT8]);
        // VALID AS OF also ends with "AS OF", but its argument is a DATE
        // STRING. Checking the longer clause first is load-bearing.
        assert_eq!(
            infer_param_oids("SELECT a FROM t VALID AS OF $1", &[], None), vec![OID_TEXT]);
        assert_eq!(
            infer_param_oids("SELECT a FROM t LIMIT $1 OFFSET $2", &[], None),
            vec![OID_INT8, OID_INT8]);
    }

    #[test]
    fn an_aggregate_column_types_from_what_the_aggregate_means() {
        // No document holds a field called `count`, so sampling stored data
        // finds nothing and falls back to text — which hands a binary client
        // the string "2" for COUNT(*).
        assert_eq!(aggregate_oid("count", None, "t"), Some(OID_INT8));
        assert_eq!(aggregate_oid("avg_fee", None, "t"), Some(OID_FLOAT8),
                   "an average is fractional even over integers");
        // SUM/MIN/MAX inherit the field's type; with no database to sample,
        // that resolves to text, and `_seq` is known from the engine contract.
        assert_eq!(aggregate_oid("max__seq", None, "t"), Some(OID_INT8));
        assert_eq!(aggregate_oid("total", None, "t"), None, "not an aggregate");
    }

    #[test]
    fn the_parse_probe_uses_a_literal_that_every_clause_accepts() {
        // Stubbing with NULL was the obvious choice and the wrong one: clauses
        // that validate their argument rejected it, so `AS OF SYSTEM TIME $1`
        // failed at Parse before a real sequence was ever bound.
        let probe = probe_sql("SELECT a FROM t AS OF SYSTEM TIME $1 WHERE b = $2", 2);
        assert!(!probe.contains("NULL"), "{}", probe);
        assert!(translate(&probe).is_ok(), "the probe must parse: {}", probe);
    }

    #[test]
    fn a_column_with_mixed_types_across_documents_is_advertised_as_text() {
        // Taking the first non-null value's type told the client `int8` and
        // then sent it "n/a" — which fails to parse client-side, and on the
        // binary path cannot be encoded at all.
        let rows = vec![json!({"x": 3}), json!({"x": "n/a"})];
        assert_eq!(oid_for(&rows, "x"), OID_TEXT);
        // Integers and floats in one column widen rather than conflict.
        let rows = vec![json!({"x": 3}), json!({"x": 1.5})];
        assert_eq!(oid_for(&rows, "x"), OID_FLOAT8);
        // A leading null must not decide the type.
        let rows = vec![json!({"x": Value::Null}), json!({"x": 7})];
        assert_eq!(oid_for(&rows, "x"), OID_INT8);
    }

    #[test]
    fn binary_output_encodes_each_advertised_type() {
        assert_eq!(cell_binary(Some(&json!(true)), OID_BOOL).unwrap().unwrap(), vec![1]);
        assert_eq!(cell_binary(Some(&json!(42)), OID_INT8).unwrap().unwrap(),
                   42i64.to_be_bytes().to_vec());
        assert_eq!(cell_binary(Some(&json!(3.5)), OID_FLOAT8).unwrap().unwrap(),
                   3.5f64.to_be_bytes().to_vec());
        // For the text family, binary and text are the same bytes.
        assert_eq!(cell_binary(Some(&json!("hi")), OID_TEXT).unwrap().unwrap(), b"hi".to_vec());
        assert_eq!(cell_binary(Some(&Value::Null), OID_INT8).unwrap(), None);
        // A boolean renders as `t`/`f` in text but one byte in binary.
        assert_eq!(cell(Some(&json!(true))).unwrap(), "t");
    }

    #[test]
    fn a_value_that_does_not_fit_its_advertised_binary_type_is_refused() {
        // Advertised types come from a bounded sample, so a field that only
        // turns heterogeneous outside it lands here. Sending a zero, or the
        // text bytes under a binary header, would corrupt the value in a way
        // the client cannot detect — so it is an error instead.
        let e = cell_binary(Some(&json!("nope")), OID_INT8).unwrap_err();
        assert!(e.contains("a string"), "{}", e);
        assert!(e.contains("more than one type"), "the error should explain WHY: {}", e);
    }

    #[test]
    fn a_row_description_carries_the_requested_format_per_column() {
        let cols = [Col::same("a"), Col::same("b")];
        let m = row_description_fmt(&cols, &[OID_INT8, OID_TEXT], &[1, 0]);
        assert_eq!(m[0], b'T');
        // The trailing i16 of each field entry is its format code.
        assert_eq!(m[m.len() - 1], 0, "the last column was requested as text");
    }

    #[test]
    fn a_qualified_column_resolves_to_its_bare_name() {
        assert_eq!(param_fields("SELECT a FROM t WHERE t.qty = $1", 1),
                   vec![Some("qty".to_string())]);
    }

    #[test]
    fn insert_placeholders_map_positionally_to_the_column_list() {
        assert_eq!(
            param_fields("INSERT INTO t (_id, qty, status) VALUES ($1, $2, $3)", 3),
            vec![Some("_id".to_string()), Some("qty".to_string()), Some("status".to_string())]);
    }

    #[test]
    fn a_set_clause_placeholder_finds_its_column() {
        assert_eq!(param_fields("UPDATE t SET status = $1 WHERE _id = $2", 2),
                   vec![Some("status".to_string()), Some("_id".to_string())]);
    }

    #[test]
    fn the_target_collection_is_found_for_every_statement_kind() {
        assert_eq!(stmt_collection("SELECT a FROM inv WHERE b = $1"), "inv");
        assert_eq!(stmt_collection("UPDATE inv SET a = $1"), "inv");
        assert_eq!(stmt_collection("DELETE FROM inv WHERE a = $1"), "inv");
        assert_eq!(stmt_collection("INSERT INTO inv (a) VALUES ($1)"), "inv");
        // Clients qualify as schema.table; NEDB has one namespace.
        assert_eq!(stmt_collection("SELECT a FROM public.inv"), "inv");
        assert_eq!(stmt_collection("INSERT INTO inv(a) VALUES ($1)"), "inv");
    }

    #[test]
    fn engine_metadata_fields_type_without_touching_storage() {
        assert_eq!(infer_field_oid(None, "t", "_seq"), OID_INT8);
        assert_eq!(infer_field_oid(None, "t", "_id"), OID_TEXT);
    }

    #[test]
    fn the_protocol_acknowledgements_are_single_empty_messages() {
        // Each is a tag plus a 4-byte length of exactly 4.
        for (m, tag) in [
            (parse_complete(), b'1'), (bind_complete(), b'2'),
            (close_complete(), b'3'), (no_data(), b'n'), (portal_suspended(), b's'),
        ] {
            assert_eq!(m.len(), 5, "{:?}", tag as char);
            assert_eq!(m[0], tag);
            assert_eq!(i32::from_be_bytes([m[1], m[2], m[3], m[4]]), 4);
        }
    }

    #[test]
    fn parameter_description_reports_its_arity_and_types() {
        let m = parameter_description(&[OID_TEXT, OID_INT8]);
        assert_eq!(m[0], b't');
        assert_eq!(i16::from_be_bytes([m[5], m[6]]), 2);
        assert_eq!(i32::from_be_bytes([m[7], m[8], m[9], m[10]]), OID_TEXT);
        assert_eq!(i32::from_be_bytes([m[11], m[12], m[13], m[14]]), OID_INT8);
    }

    #[test]
    fn a_cstring_is_taken_without_its_terminator() {
        let body = b"one\0two\0".to_vec();
        let mut at = 0usize;
        assert_eq!(take_cstr(&body, &mut at), "one");
        assert_eq!(take_cstr(&body, &mut at), "two");
        assert_eq!(at, body.len());
    }

    #[test]
    fn truncated_integers_are_reported_rather_than_read_past_the_end() {
        let body = vec![0u8, 1];
        let mut at = 0usize;
        assert!(take_i32(&body, &mut at).is_err());
        let mut at = 0usize;
        assert!(take_i16(&body, &mut at).is_ok());
    }

    #[test]
    fn a_binary_result_format_request_is_refused_rather_than_faked() {
        // Sending text under a binary header corrupts every value silently,
        // which is far worse than an error naming the limitation.
        let out = encode_rows(&[], &[Col::same("a")]);
        let desc_format = &out[out.len() - 2..];
        assert_eq!(i16::from_be_bytes([desc_format[0], desc_format[1]]), 0,
                   "every column is advertised as text format");
    }

    #[test]
    fn a_float_parameter_does_not_render_as_rust_infinity() {
        assert_eq!(fmt_float(f64::INFINITY), "'Infinity'");
        assert_eq!(fmt_float(f64::NEG_INFINITY), "'-Infinity'");
        assert_eq!(fmt_float(f64::NAN), "'NaN'");
        assert_eq!(fmt_float(3.0), "3", "a whole float should not gain a .0 tail");
        assert_eq!(fmt_float(3.5), "3.5");
    }
}