areev-store 1.8.0

//! PostgreSQL backend for the `Db` seam — the server-tier storage transport.
//!
//! One memory = one Postgres schema (the unit of isolation, export
//! `pg_dump -n`, and erasure `DROP SCHEMA … CASCADE`). Unlike the embedded
//! file backend, this one admits MULTIPLE CONCURRENT WRITERS per memory:
//! id allocation is a `counters` row claimed with `UPDATE … RETURNING`
//! inside each write transaction, which both hands out collision-free
//! blocks and briefly serializes concurrent write transactions — so op-log
//! order equals commit order (followers never miss ops) and fork/head
//! semantics match the single-writer model exactly. Reads never block
//! (MVCC). The dictionary and BM25 collection stats are DB-authoritative
//! on cache miss, so one instance immediately sees terms another interned.
//!
//! The store logic in `lib.rs` runs unchanged; this module supplies the
//! transport plus a statement translator for the ~dozen constructs where the
//! dialects genuinely differ (upserts, the vector functions, `PRAGMA`).
//! Unknown divergent constructs FAIL FAST with the offending SQL rather than
//! executing something subtly different.
//!
//! Connections belong to a process-wide pool per DSN (`PgPool`, #181), not
//! to a handle: a handle borrows one per statement, or for the length of a
//! transaction, so N memories in a process hold at most `?pool=` (default
//! [`DEFAULT_POOL_SIZE`]) connections between them and an idle memory holds
//! none. Every statement is schema-qualified and every per-transaction
//! setting is `SET LOCAL`, which is what lets any connection serve any
//! schema.

use std::cell::RefCell;
use std::collections::HashMap;

use areev_core::error::{AreevError, Result};
use tokio_postgres::types::{IsNull, ToSql, Type};
use turso::Value;

use crate::db::{db_err, Db, Row, WriteIds};
use crate::{pb, pi, pt};

/// tokio-postgres's `Display` is just "db error" — the server's message and
/// SQLSTATE live in the source chain. Surface them, or every failure is
/// undiagnosable.
///
/// The no-SQLSTATE arm walks the chain by hand for the same reason: a
/// connect-time failure never reaches a server, so it has no SQLSTATE, and
/// its `Display` alone is "error connecting to server". That is the class a
/// TLS rejection lands in — "invalid peer certificate: UnknownIssuer" is the
/// whole diagnosis, and it sits one level down.
pub(crate) fn pg_err(e: tokio_postgres::Error) -> AreevError {
    match e.as_db_error() {
        Some(db) => AreevError::Storage(format!(
            "postgres error {}: {}",
            db.code().code(),
            db.message()
        )),
        None => {
            let mut msg = e.to_string();
            let mut source = std::error::Error::source(&e);
            // Bounded: a `source()` implementation that cycles would otherwise
            // hang the caller, and no real chain here is deeper than three.
            for _ in 0..8 {
                let Some(cause) = source else { break };
                msg.push_str(": ");
                msg.push_str(&cause.to_string());
                source = cause.source();
            }
            AreevError::Storage(msg)
        }
    }
}

/// Schema DDL — the Turso `SCHEMA` with Postgres types: every integer is
/// `bigint` (an `int4`/`numeric` arriving where the store expects Integer
/// silently zeroes counters), hashes/blobs are `bytea`, and `fts_vocab.id`
/// is IDENTITY because its rowid auto-assign is the one id the store does
/// not allocate itself. `embeddings` is NOT here — its `vector(dim)` column
/// needs the runtime-supplied dim, so `ensure_embeddings` creates it at the
/// first `set_embedder`. `blobs` replaces the `.blobs` filesystem sidecar.
/// No unique constraints on `triples`/`osp`/`fts_post`/`thread_idx`/
/// `prov_idx`/`run_idx`: duplicate rows are reachable, semantics-bearing
/// state (rebuilds delete by full tuple).
pub(crate) const PG_SCHEMA: &[&str] = &[
    "CREATE TABLE IF NOT EXISTS meta(k text PRIMARY KEY, v text)",
    // IDENTITY (unlike the embedded backend's app-assigned ids): with
    // several writers the database owns term-id allocation. GENERATED BY
    // DEFAULT so restores with explicit ids still load.
    // Keyed by the SHA-256 of the term, not the term (#160): a btree entry
    // is capped at ~2704 bytes after pglz compression, so `term text UNIQUE`
    // refused any value that did not compress under that — an entropy
    // dependence, not a length: a 2.7 KB base64 payload failed while an 8 KB
    // run of 'x' passed — and the loop's own recommendation ledger crossed it
    // at roughly eight recommendations. Every subject, relation and object
    // string is interned here, so the dictionary must never bound a value.
    // The unique index and the migration for schemas created before this
    // column existed live in `PG_SEED`, where the column is guaranteed.
    "CREATE TABLE IF NOT EXISTS terms(id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, term text NOT NULL, term_hash bytea NOT NULL)",
    // Write-transaction id blocks (seq / op / hlc): claimed with
    // UPDATE … RETURNING inside each write txn — the row locks are also
    // what serializes concurrent write transactions per memory.
    "CREATE TABLE IF NOT EXISTS counters(name text PRIMARY KEY, v bigint NOT NULL)",
    "CREATE TABLE IF NOT EXISTS grains(
        seq bigint PRIMARY KEY,
        hash bytea NOT NULL,
        ns bigint, gtype bigint, created_at bigint,
        s bigint, p bigint, o bigint,
        vf bigint, vt bigint,
        svf bigint, svt bigint,
        superseded_by bytea, supersedes bytea,
        text text,
        blob bytea NOT NULL)",
    "CREATE UNIQUE INDEX IF NOT EXISTS idx_grains_hash ON grains(hash)",
    // Mirrors the embedded SCHEMA's idx_grains_ns_s, for the same reason and
    // in the same column order: the vector legs join embeddings to grains and
    // filter on (ns, s, p), and without this the planner scans grains and
    // computes a distance per row. This tier has the steeper slope of the two
    // (a round trip per statement on top of the scan), so it needs it more.
    "CREATE INDEX IF NOT EXISTS idx_grains_ns_s ON grains(ns, s, p)",
    // Dimension-less at creation: the `vec vector(dim)` column arrives via
    // `ensure_embeddings` at the first set_embedder (the dim is host-
    // supplied), but the TABLE must exist up front — forget() deletes from
    // it unconditionally.
    "CREATE TABLE IF NOT EXISTS embeddings(seq bigint PRIMARY KEY)",
    "CREATE TABLE IF NOT EXISTS fts_vocab(id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, term text UNIQUE)",
    "CREATE TABLE IF NOT EXISTS fts_post(term bigint, seq bigint, ns bigint, tf bigint)",
    "CREATE INDEX IF NOT EXISTS idx_fts_post ON fts_post(term, ns)",
    "CREATE INDEX IF NOT EXISTS idx_fts_post_seq ON fts_post(seq)",
    "CREATE TABLE IF NOT EXISTS fts_doc(seq bigint PRIMARY KEY, len bigint)",
    "CREATE TABLE IF NOT EXISTS triples(ns bigint, s bigint, p bigint, o bigint, seq bigint, cur bigint)",
    "CREATE INDEX IF NOT EXISTS idx_spo ON triples(ns,s,p,o,seq)",
    "CREATE INDEX IF NOT EXISTS idx_pos ON triples(ns,p,o,s,seq)",
    "CREATE INDEX IF NOT EXISTS idx_triples_seq ON triples(seq)",
    "CREATE TABLE IF NOT EXISTS osp(ns bigint, o bigint, s bigint, p bigint, seq bigint, cur bigint)",
    "CREATE INDEX IF NOT EXISTS idx_osp ON osp(ns,o,s)",
    "CREATE INDEX IF NOT EXISTS idx_osp_seq ON osp(seq)",
    "CREATE TABLE IF NOT EXISTS entity_latest(ns bigint, s bigint, p bigint, o bigint, seq bigint, hash bytea, PRIMARY KEY(ns,s,p))",
    "CREATE TABLE IF NOT EXISTS heads(ns bigint, s bigint, p bigint, seq bigint, hash bytea, created_at bigint, PRIMARY KEY(ns,s,p,seq))",
    "CREATE TABLE IF NOT EXISTS oplog(op_seq bigint PRIMARY KEY, hlc bigint, op bigint, hash bytea)",
    "CREATE TABLE IF NOT EXISTS thread_idx(ns bigint, session bigint, seq bigint)",
    "CREATE INDEX IF NOT EXISTS idx_thread ON thread_idx(ns, session, seq)",
    "CREATE TABLE IF NOT EXISTS prov_idx(ns bigint, parent bytea, seq bigint)",
    "CREATE INDEX IF NOT EXISTS idx_prov ON prov_idx(ns, parent, seq)",
    "CREATE TABLE IF NOT EXISTS run_idx(ns bigint, run bigint, seq bigint)",
    "CREATE INDEX IF NOT EXISTS idx_run ON run_idx(ns, run, seq)",
    "CREATE TABLE IF NOT EXISTS corpus_idx(seq bigint PRIMARY KEY)",
    // Namespace registry (prefix scoping) — see the embedded SCHEMA's ns_reg
    // comment. Same count-maintained contract; the serialized write txns
    // (reserve_write's row lock) make the read-modify-write bumps safe here.
    "CREATE TABLE IF NOT EXISTS ns_reg(ns bigint PRIMARY KEY, n bigint NOT NULL)",
    // Replaces the `.blobs` filesystem sidecar: same CAS semantics, rows
    // keyed by the raw 32-byte content hash.
    "CREATE TABLE IF NOT EXISTS blobs(hash bytea PRIMARY KEY, body bytea NOT NULL)",
];

/// Seed the allocation state from what the schema already holds — safe to
/// run concurrently from several opening instances (GREATEST-merge upserts)
/// and idempotent. Runs after [`PG_SCHEMA`] on every open.
pub(crate) const PG_SEED: &[&str] = &[
    "INSERT INTO counters(name, v) SELECT 'seq', COALESCE(MAX(seq),0) FROM grains \
     ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)",
    "INSERT INTO counters(name, v) SELECT 'op', COALESCE(MAX(op_seq),0) FROM oplog \
     ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)",
    "INSERT INTO counters(name, v) SELECT 'hlc', COALESCE(MAX(hlc),0) FROM oplog \
     ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)",
    // Legacy migration: a schema created by the stage-1 build has
    // terms(id bigint PK) with NO identity — `CREATE TABLE IF NOT EXISTS`
    // never adds one, and the first id-less intern would violate NOT NULL
    // forever. Add the identity and seed its sequence past the explicit
    // ids IN THE SAME step: the seeding is race-free exactly here, because
    // until the identity exists no writer can have drawn from it (and the
    // whole bootstrap runs under the schema's advisory lock). An
    // unconditional setval on every open would instead RACE concurrent
    // interns and rewind the sequence below ids already handed out.
    "DO $$ BEGIN \
       IF NOT EXISTS (SELECT 1 FROM pg_attribute a \
                        JOIN pg_class c ON a.attrelid = c.oid \
                        JOIN pg_namespace n ON c.relnamespace = n.oid \
                       WHERE n.nspname = current_schema() AND c.relname = 'terms' \
                         AND a.attname = 'id' AND a.attidentity <> '') THEN \
         EXECUTE 'ALTER TABLE terms ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY'; \
         PERFORM setval(pg_get_serial_sequence('terms','id'), \
                        GREATEST((SELECT COALESCE(MAX(id),0) FROM terms), 1)); \
       END IF; \
     END $$",
    // Digest-keyed dictionary (#160). Three idempotent steps, each gated on
    // catalog state so every open converges and none repeats work: (1) a
    // schema from before `term_hash` existed gains the column, backfilled
    // with the same SHA-256 over UTF-8 bytes the Rust side computes
    // (`sha256(convert_to(term,'UTF8'))` == `Sha256::digest(term.as_bytes())`
    // — they MUST agree, or an old term is re-interned under a second id);
    // (2) the unique index the insert's `ON CONFLICT (term_hash)` needs;
    // (3) every unique constraint or standalone unique index whose key is
    // exactly `(term)` goes — matched on DEFINITION, not on the default name
    // `terms_term_key`, because a pg_restore'd or hand-repaired schema can
    // carry the same cap under another name and the open would otherwise
    // report itself migrated while still refusing. That drop is the actual
    // fix: it was the ~2704-byte btree entry. Runs under the bootstrap advisory
    // lock like the block above, so two openers cannot race the DDL. A
    // read-only open runs none of this and refuses by name (`STO-E005`) when
    // the column is absent, because reads key on it.
    "DO $$ DECLARE r record; BEGIN \
       IF NOT EXISTS (SELECT 1 FROM information_schema.columns \
                       WHERE table_schema = current_schema() AND table_name = 'terms' \
                         AND column_name = 'term_hash') THEN \
         EXECUTE 'ALTER TABLE terms ADD COLUMN term_hash bytea'; \
         EXECUTE 'UPDATE terms SET term_hash = sha256(convert_to(term, ''UTF8'')) WHERE term_hash IS NULL'; \
         EXECUTE 'ALTER TABLE terms ALTER COLUMN term_hash SET NOT NULL'; \
       END IF; \
       IF NOT EXISTS (SELECT 1 FROM pg_indexes \
                       WHERE schemaname = current_schema() AND tablename = 'terms' \
                         AND indexname = 'idx_terms_hash') THEN \
         EXECUTE 'CREATE UNIQUE INDEX idx_terms_hash ON terms(term_hash)'; \
       END IF; \
       FOR r IN SELECT c.conname FROM pg_constraint c \
                  JOIN pg_class t ON c.conrelid = t.oid \
                  JOIN pg_namespace n ON t.relnamespace = n.oid \
                  JOIN pg_attribute a ON a.attrelid = t.oid AND a.attname = 'term' \
                 WHERE n.nspname = current_schema() AND t.relname = 'terms' \
                   AND c.contype = 'u' AND c.conkey = ARRAY[a.attnum] LOOP \
         EXECUTE format('ALTER TABLE terms DROP CONSTRAINT %I', r.conname); \
       END LOOP; \
       FOR r IN SELECT ic.relname AS idxname FROM pg_index i \
                  JOIN pg_class t ON i.indrelid = t.oid \
                  JOIN pg_class ic ON i.indexrelid = ic.oid \
                  JOIN pg_namespace n ON t.relnamespace = n.oid \
                  JOIN pg_attribute a ON a.attrelid = t.oid AND a.attname = 'term' \
                 WHERE n.nspname = current_schema() AND t.relname = 'terms' \
                   AND i.indisunique AND i.indnkeyatts = 1 AND i.indkey[0] = a.attnum \
                   AND NOT EXISTS (SELECT 1 FROM pg_constraint WHERE conindid = i.indexrelid) LOOP \
         EXECUTE format('DROP INDEX %I', r.idxname); \
       END LOOP; \
     END $$",
];

/// The version of [`PG_SCHEMA`] + [`PG_SEED`] that a schema was bootstrapped
/// with, stamped into the memory's own `meta` table as `pg_schema` at the end
/// of a successful bootstrap and read back — before any lock and any DDL — on
/// the next open. A match means "this schema is already at this build's
/// shape", which is what lets a steady-state open issue no DDL at all.
///
/// ⚠️ **BUMP THIS WHENEVER `PG_SCHEMA` OR `PG_SEED` CHANGES — IN THE SAME
/// COMMIT.** ⚠️ The stamp is the *only* thing standing between a schema and a
/// migration it needs: a version left behind after a statement was added makes
/// every already-stamped schema skip that statement forever, and the #160
/// failure mode is what that looks like from outside — the dictionary column a
/// read keys on is simply absent, so every recall answers empty, with no error
/// anywhere. `pg_schema_version_tracks_the_schema` (below) hashes both arrays
/// and fails the build if the digest moves without this constant moving, so
/// forgetting is a red test rather than a silent data bug.
pub(crate) const PG_SCHEMA_VERSION: &str = "1";

/// Where a bootstrap records that it ran, so the next open can skip it.
///
/// Generic over the table/key because two independent bootstraps share one
/// memory schema: the store's own (`meta.pg_schema`) and the telemetry
/// sidecar's (`telem_meta.schema_version`). They must not read each other's
/// stamp — a store stamp is no evidence that `telem_*` exists.
#[derive(Clone, Copy)]
pub(crate) struct PgStamp {
    /// Unqualified table name inside the memory's schema.
    pub(crate) table: &'static str,
    pub(crate) key: &'static str,
    pub(crate) version: &'static str,
}

/// The store's own bootstrap marker.
pub(crate) const STORE_STAMP: PgStamp = PgStamp {
    table: "meta",
    key: "pg_schema",
    version: PG_SCHEMA_VERSION,
};

/// May an open bootstrap the schema it was pointed at?
///
/// Carried on the DSN (`?provision=auto|never`) rather than in
/// `AreevOptions`, so every host that already passes a DSN string — the CLI,
/// the console, both bindings — reaches it without a new parameter.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Default)]
pub enum ProvisionMode {
    /// Bootstrap when the schema is absent or its stamp is stale (the
    /// historical behaviour, and the default).
    #[default]
    Auto,
    /// Never issue DDL on the request path: an absent or stale schema is a
    /// hard refusal (`STO-E008`) with no advisory lock and no DDL attempted.
    /// For deployments whose runtime role holds no `CREATE` — provision with
    /// `areev provision` (or a migration job) instead.
    Never,
}

impl ProvisionMode {
    fn parse(v: &str) -> Result<Self> {
        match v {
            "auto" => Ok(Self::Auto),
            "never" => Ok(Self::Never),
            other => Err(AreevError::Validation(format!(
                "postgres URL: provision={other:?} is not a mode (auto|never)"
            ))),
        }
    }
}

/// Read `?provision=` off a DSN. The parameter is ours, not the driver's;
/// [`crate::pgtls::SslRequest::split`] is what removes it before the DSN
/// reaches `tokio_postgres`, exactly as it does for `sslmode`/`sslrootcert`,
/// so every connect path (open, reconnect, `drop_postgres_schema`, the
/// conformance escape hatches) tolerates it without knowing about it.
pub fn provision_mode(url: &str) -> Result<ProvisionMode> {
    let Some((_, query)) = url.split_once('?') else {
        return Ok(ProvisionMode::Auto);
    };
    let mut mode = ProvisionMode::Auto;
    for pair in query.split('&') {
        if let Some(("provision", v)) = pair.split_once('=') {
            mode = ProvisionMode::parse(v)?;
        }
    }
    Ok(mode)
}

/// Drop one of OUR query parameters from a DSN, leaving every other pair
/// (and the driver's own) untouched.
fn strip_param(url: &str, name: &str) -> String {
    let Some((base, query)) = url.split_once('?') else {
        return url.to_string();
    };
    let rest: Vec<&str> = query
        .split('&')
        .filter(|p| !matches!(p.split_once('='), Some((k, _)) if k == name))
        .collect();
    if rest.is_empty() {
        base.to_string()
    } else {
        format!("{base}?{}", rest.join("&"))
    }
}

/// Strip `?provision=` from a DSN, for the one caller that must ignore it:
/// `areev provision` itself, whose whole job is to bootstrap. Everything else
/// leaves it in place so it rides through to [`provision_mode`].
pub fn strip_provision(url: &str) -> String {
    strip_param(url, "provision")
}

/// Strip `?schema=` from a DSN, for a caller that names the schema another
/// way (`areev provision --schema`). Unlike [`split_schema_url`] it does not
/// require the parameter to be there — the point is to accept a DSN with or
/// without it and let the explicit name win.
pub fn strip_schema(url: &str) -> String {
    strip_param(url, "schema")
}

// ---- the process-wide pool (#181) -------------------------------------

/// Connections one process may hold to one DSN when neither `?pool=` nor
/// `$AREEV_PG_POOL` says otherwise.
pub const DEFAULT_POOL_SIZE: usize = 8;

/// The out-of-band spelling of `?pool=`, for a DSN that is not yours to edit.
pub const POOL_ENV: &str = "AREEV_PG_POOL";

/// The connection cap for `url`: `?pool=N` on the DSN, else `$AREEV_PG_POOL`,
/// else [`DEFAULT_POOL_SIZE`]. The DSN wins, as everywhere.
pub fn pool_size(url: &str) -> Result<usize> {
    if let Some((_, query)) = url.split_once('?') {
        let mut from_dsn = None;
        for pair in query.split('&') {
            if let Some(("pool", v)) = pair.split_once('=') {
                from_dsn = Some(parse_pool_size("postgres URL: pool", v)?);
            }
        }
        if let Some(n) = from_dsn {
            return Ok(n);
        }
    }
    match std::env::var(POOL_ENV) {
        Ok(v) if !v.trim().is_empty() => parse_pool_size(&format!("${POOL_ENV}"), v.trim()),
        _ => Ok(DEFAULT_POOL_SIZE),
    }
}

fn parse_pool_size(what: &str, v: &str) -> Result<usize> {
    match v.parse::<usize>() {
        Ok(n) if n >= 1 => Ok(n),
        _ => Err(AreevError::Validation(format!(
            "{what}={v:?} must be a whole number of connections, at least 1"
        ))),
    }
}

/// Strip `?pool=` from a DSN.
pub fn strip_pool(url: &str) -> String {
    strip_param(url, "pool")
}

/// One connection, with the state that belongs to it rather than to any
/// memory handle: its prepared statements (a `Statement` names a server-side
/// object of ONE session) and the two GUCs a handle may want set.
pub(crate) struct PooledConn {
    client: tokio_postgres::Client,
    /// Keyed by the translated, schema-qualified SQL, so one connection
    /// serves many schemas without one memory's statement answering for
    /// another's. Bounded (see `prepared`).
    cache: HashMap<String, tokio_postgres::Statement>,
    /// What this session currently has applied — reconciled against the
    /// handle's wish before each statement outside a transaction.
    ef_search: Option<usize>,
    exact_scan: bool,
}

/// The per-DSN pool: one runtime driving every connection, a fair semaphore
/// as the cap, and the idle connections. Lives for the process; a handle
/// borrows from it per statement, or per transaction.
pub(crate) struct PgPool {
    rt: tokio::runtime::Runtime,
    url: String,
    cap: usize,
    sem: std::sync::Arc<tokio::sync::Semaphore>,
    idle: std::sync::Mutex<Vec<PooledConn>>,
}

/// A borrowed connection. Returned to the pool on drop unless it broke.
pub(crate) struct Checkout {
    pool: std::sync::Arc<PgPool>,
    conn: Option<PooledConn>,
    _permit: tokio::sync::OwnedSemaphorePermit,
    broken: bool,
}

impl Checkout {
    pub(crate) fn conn(&mut self) -> &mut PooledConn {
        self.conn.as_mut().expect("a checkout holds its connection until drop")
    }
    pub(crate) fn client(&self) -> &tokio_postgres::Client {
        &self.conn.as_ref().expect("a checkout holds its connection until drop").client
    }
    /// Note a failure: a dead socket is discarded rather than returned.
    pub(crate) fn note(&mut self, e: &AreevError) {
        if self.client().is_closed() || PgDb::is_connection_dead(e) {
            self.broken = true;
        }
    }
}

impl Drop for Checkout {
    fn drop(&mut self) {
        if let Some(c) = self.conn.take() {
            if !self.broken && !c.client.is_closed() {
                if let Ok(mut idle) = self.pool.idle.lock() {
                    idle.push(c);
                }
            }
        }
    }
}

static POOLS: std::sync::OnceLock<
    std::sync::Mutex<HashMap<String, std::sync::Arc<PgPool>>>,
> = std::sync::OnceLock::new();

impl PgPool {
    /// The pool for `url`, made on first use. Keyed by the DSN with the
    /// store's own parameters (`schema`, `provision`, `pool`) stripped, so
    /// every memory on one server and role shares one pool; a different
    /// role or TLS setting is a different pool. The first open sizes it;
    /// a later open asking for another size gets the warning back.
    pub(crate) fn for_url(url: &str) -> Result<(std::sync::Arc<PgPool>, Option<String>)> {
        let want = pool_size(url)?;
        let key = strip_param(&strip_param(&strip_param(url, "pool"), "provision"), "schema");
        let mut reg = POOLS
            .get_or_init(Default::default)
            .lock()
            .unwrap_or_else(|p| p.into_inner());
        if let Some(p) = reg.get(&key) {
            let warn = (p.cap != want).then(|| {
                format!(
                    "postgres pool for this DSN was sized at {} connections by an earlier open \
                     in this process; ?pool={want} (or ${POOL_ENV}) does not resize it",
                    p.cap
                )
            });
            return Ok((std::sync::Arc::clone(p), warn));
        }
        // Multi-thread so callers on any thread `block_on` concurrently — a
        // current-thread runtime serializes its `block_on` callers, which
        // would make the pool a process-wide lock. The workers drive the
        // connections' I/O, so there must be enough to pump every borrowed
        // one at once: size them to the cap, bounded (a handful of idle
        // driver threads is cheap; a hundred is not), never below two.
        let workers = want.clamp(2, 8);
        let rt = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(workers)
            .thread_name("areev-pg")
            .enable_all()
            .build()
            .map_err(db_err)?;
        let p = std::sync::Arc::new(PgPool {
            rt,
            url: key.clone(),
            cap: want,
            sem: std::sync::Arc::new(tokio::sync::Semaphore::new(want)),
            idle: std::sync::Mutex::new(Vec::new()),
        });
        reg.insert(key, std::sync::Arc::clone(&p));
        Ok((p, None))
    }

    pub(crate) fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
        self.rt.block_on(f)
    }

    /// Borrow a connection: an idle one, else a new one, waiting (fairly)
    /// while `cap` are out.
    pub(crate) fn checkout(self: &std::sync::Arc<Self>) -> Result<Checkout> {
        let permit = self
            .rt
            .block_on(std::sync::Arc::clone(&self.sem).acquire_owned())
            .map_err(db_err)?;
        let idle = loop {
            let next = self.idle.lock().unwrap_or_else(|p| p.into_inner()).pop();
            match next {
                Some(c) if c.client.is_closed() => continue,
                other => break other,
            }
        };
        let conn = match idle {
            Some(c) => c,
            None => self.dial()?,
        };
        Ok(Checkout { pool: std::sync::Arc::clone(self), conn: Some(conn), _permit: permit, broken: false })
    }

    fn dial(&self) -> Result<PooledConn> {
        let client = crate::pgtls::connect(&self.rt, &self.url)?;
        // Schema-independent baseline: no table resolves through it (every
        // reference is qualified), but pgvector's type and operator do.
        self.rt
            .block_on(client.batch_execute("SET search_path TO public, ext"))
            .map_err(pg_err)?;
        Ok(PooledConn { client, cache: HashMap::new(), ef_search: None, exact_scan: false })
    }
}

// ---- the handle ---------------------------------------------------------

pub(crate) struct PgDb {
    pool: std::sync::Arc<PgPool>,
    schema: String,
    /// The connection pinned for an open transaction: taken at `begin`,
    /// given back at `commit`/`rollback`. Outside one, every statement
    /// borrows for its own duration, so an idle handle holds nothing.
    pinned: RefCell<Option<Checkout>>,
    /// Whether a transaction is open on this handle. Nothing is replayed while
    /// it is: the transaction died with the connection, so silently re-running
    /// one statement would apply it outside the atomic unit it was written for.
    in_txn: std::cell::Cell<bool>,
    /// What this handle wants set — applied to whichever connection serves
    /// it (`SET LOCAL` inside a transaction, reconciled per connection
    /// outside), never assumed to be on the session.
    ann_ef_search: std::cell::Cell<Option<usize>>,
    exact_scan: std::cell::Cell<bool>,
    /// Hot SQL → its translated, qualified form, so the hot path translates
    /// once per handle.
    xlate: RefCell<HashMap<String, String>>,
    /// Cached BM25 collection stats — a COUNT/SUM over fts_doc is O(corpus)
    /// and would otherwise run on EVERY text query. Invalidated on this
    /// handle's own writes (reserve_write); other writers' documents stay
    /// invisible until then, which BM25 tolerates (N/avgdl drift is
    /// explicitly immaterial to ranking — see search_text_inner).
    stats: RefCell<Option<(i64, i64)>>,
    /// True when this open found the schema already stamped current and ran
    /// NO bootstrap — no advisory lock, no DDL, no seeding. Read by callers
    /// that would otherwise re-do post-bootstrap work of their own (the
    /// telemetry sidecar's v2 migration), so "the schema is current" is
    /// decided once, here, rather than re-derived by each of them.
    bootstrap_skipped: bool,
    /// A pool-size mismatch this open could not honour (see `PgPool::for_url`).
    pool_warning: Option<String>,
}

/// Statements this handle wants applied for the transaction it is opening,
/// where the serving connection's session does not already say so.
fn txn_prelude(want_ef: Option<usize>, want_exact: bool, conn: &PooledConn) -> String {
    let mut sql = String::from("BEGIN");
    match (want_ef, conn.ef_search) {
        (Some(ef), have) if have != Some(ef) => {
            sql.push_str(&format!("; SET LOCAL hnsw.ef_search = {ef}"));
        }
        (None, Some(_)) => sql.push_str("; SET LOCAL hnsw.ef_search TO DEFAULT"),
        _ => {}
    }
    if want_exact && !conn.exact_scan {
        sql.push_str("; SET LOCAL enable_indexscan = off; SET LOCAL enable_bitmapscan = off");
    } else if !want_exact && conn.exact_scan {
        sql.push_str("; SET LOCAL enable_indexscan TO DEFAULT; SET LOCAL enable_bitmapscan TO DEFAULT");
    }
    sql
}

impl PgDb {
    /// Attach to the schema — create-or-bootstrap it when the stamp says so —
    /// through the process's pool for `url`.
    ///
    /// **The bootstrap is the exception, not the rule.** When `stamp` names a
    /// marker row and that row already holds the caller's version, this open
    /// takes the fast path: two SELECTs and nothing else — no advisory lock,
    /// no `CREATE SCHEMA`, no DDL, no seeding. The probe needs nothing beyond
    /// `USAGE` on the schema and `SELECT` on the stamp table, and it uses
    /// `to_regclass` rather than naming the table in a `FROM` clause, so an
    /// absent schema (or absent table) answers NULL instead of raising
    /// `42P01` — a first open costs one query, not one error. (Sibling probe:
    /// [`Self::verify_read_only`] asks the same question for a read-only open,
    /// and the two must stay in step; see the note on each.)
    ///
    /// Only a missing or stale stamp reaches the bootstrap, which then runs
    /// inside ONE explicit transaction holding `pg_advisory_xact_lock`: this
    /// backend admits concurrent openers, and Postgres's `IF NOT EXISTS` DDL
    /// is racy without a lock (the loser gets a spurious 23505 on
    /// pg_namespace/pg_type). Transaction-scoped rather than session-scoped
    /// because Postgres DDL is transactional: the statements become atomic
    /// (a half-applied schema is no longer reachable) and the lock is released
    /// by COMMIT/ROLLBACK. The bare `CREATE TABLE`s in `bootstrap` resolve
    /// through a `SET LOCAL search_path` that ends with the transaction —
    /// the one place the schema is ever on a search path.
    ///
    /// `read_only` skips ALL of that — no advisory lock, no `CREATE SCHEMA`,
    /// no `bootstrap` statements (`PG_SCHEMA`'s DDL + `PG_SEED`'s upserts) —
    /// because a least-privilege SELECT-only role can issue none of them:
    /// Postgres checks the `CREATE` privilege on the database before it
    /// checks whether the schema already exists, and it checks table
    /// OWNERSHIP before it checks whether an index already exists, so even
    /// fully-idempotent `IF NOT EXISTS` DDL 42501s for such a role on a
    /// schema that is already there (issue #127). Instead it VERIFIES with
    /// SELECT-only probes — see [`Self::verify_read_only`].
    ///
    /// The connection borrowed for all of this goes back to the pool before
    /// this returns: an open handle holds none until it begins a transaction.
    pub(crate) fn open(
        url: &str,
        schema: &str,
        bootstrap: &[&str],
        read_only: bool,
        stamp: Option<PgStamp>,
    ) -> Result<Self> {
        if schema.is_empty()
            || schema.len() > 63
            || !schema.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')
            || schema.as_bytes()[0].is_ascii_digit()
        {
            return Err(AreevError::Validation(format!(
                "postgres schema name must be [a-z_][a-z0-9_]* and <= 63 bytes, got {schema:?}"
            )));
        }
        let (pool, pool_warning) = PgPool::for_url(url)?;
        let mut co = pool.checkout()?;
        let opened = Self::open_on(&pool, co.client(), url, schema, bootstrap, read_only, stamp);
        let bootstrap_skipped = match opened {
            Ok(skipped) => skipped,
            Err(e) => {
                co.note(&e);
                return Err(e);
            }
        };
        drop(co);
        Ok(Self {
            pool,
            schema: schema.to_string(),
            pinned: RefCell::new(None),
            in_txn: std::cell::Cell::new(false),
            ann_ef_search: std::cell::Cell::new(None),
            exact_scan: std::cell::Cell::new(false),
            xlate: RefCell::new(HashMap::new()),
            stats: RefCell::new(None),
            bootstrap_skipped,
            pool_warning,
        })
    }

    /// The verify-or-bootstrap half of [`Self::open`], on one borrowed
    /// connection. Returns whether the bootstrap was skipped.
    fn open_on(
        pool: &PgPool,
        client: &tokio_postgres::Client,
        url: &str,
        schema: &str,
        bootstrap: &[&str],
        read_only: bool,
        stamp: Option<PgStamp>,
    ) -> Result<bool> {
        if read_only {
            pool.block_on(Self::verify_read_only(client, schema))?;
            return Ok(false);
        }
        // The fast path (issue #180). Ask the schema what shape it is in
        // BEFORE anything that needs a lock or an ownership grant; a
        // steady-state open ends here, having issued two SELECTs.
        let mut skipped = false;
        if let Some(st) = stamp {
            skipped = pool.block_on(Self::schema_stamp(client, schema, st))?.as_deref()
                == Some(st.version);
        }
        if skipped {
            return Ok(true);
        }
        if provision_mode(url)? == ProvisionMode::Never {
            // `?provision=never`: refuse without touching anything. No
            // advisory lock, no DDL, not even the CREATE SCHEMA — the
            // whole point is a runtime role that holds no CREATE.
            return Err(Self::not_provisioned(client, pool, schema, stamp));
        }
        match Self::bootstrap_once(pool, client, schema, bootstrap, stamp) {
            Ok(()) => Ok(false),
            // A concurrent opener won the race and created the schema first.
            // The advisory lock makes this rare, not impossible: the pool
            // shares one runtime, so two openers' `CREATE SCHEMA` can
            // interleave ahead of the lock under load (Postgres documents
            // IF NOT EXISTS DDL as itself racy). The contract is that both
            // openers succeed — re-read the stamp, and if the winner left it
            // current, this open is done. Otherwise the winner is still
            // mid-bootstrap or failed: one retry, which serialises cleanly
            // now that the racing CREATE has landed. Pinned by #181's
            // `concurrent_first_open_bootstraps_once` under a small pool.
            Err(e) if Self::is_lost_bootstrap_race(&e) => {
                let current = match stamp {
                    Some(st) => {
                        pool.block_on(Self::schema_stamp(client, schema, st))?.as_deref()
                            == Some(st.version)
                    }
                    None => false,
                };
                if current {
                    Ok(true)
                } else {
                    Self::bootstrap_once(pool, client, schema, bootstrap, stamp).map(|()| false)
                }
            }
            Err(e) => Err(e),
        }
    }

    /// A duplicate-object error from the bootstrap: two openers raced the
    /// `IF NOT EXISTS` DDL. `23505` on a catalog index (schema or type name),
    /// `42P06` duplicate_schema, `42710` duplicate_object.
    fn is_lost_bootstrap_race(e: &AreevError) -> bool {
        let AreevError::Storage(msg) = e else { return false };
        msg.contains("postgres error 23505")
            || msg.contains("postgres error 42P06")
            || msg.contains("postgres error 42710")
    }

    /// One bootstrap attempt: DDL + seeding in ONE transaction holding the
    /// advisory lock.
    fn bootstrap_once(
        pool: &PgPool,
        client: &tokio_postgres::Client,
        schema: &str,
        bootstrap: &[&str],
        stamp: Option<PgStamp>,
    ) -> Result<()> {
        pool.block_on(async {
            // ONE transaction around the whole bootstrap. Postgres DDL
            // is transactional, so the statements are atomic, and
            // `pg_advisory_xact_lock` is released by COMMIT/ROLLBACK —
            // there is no unlock to forget on a failure path.
            client.batch_execute("BEGIN").await.map_err(pg_err)?;
            let boot = async {
                client
                    .query_one(
                        "SELECT pg_advisory_xact_lock(hashtext('areev_bootstrap'), hashtext($1))",
                        &[&schema],
                    )
                    .await
                    .map_err(pg_err)?;
                client
                    .batch_execute(&format!("CREATE SCHEMA IF NOT EXISTS \"{schema}\""))
                    .await
                    .map_err(pg_err)?;
                // LOCAL: the bare DDL below needs the schema on the path,
                // and the pooled connection must not keep it afterwards.
                client
                    .batch_execute(&format!(
                        "SET LOCAL search_path TO \"{schema}\", public, ext"
                    ))
                    .await
                    .map_err(pg_err)?;
                for sql in bootstrap {
                    client.batch_execute(sql).await.map_err(|e| {
                        AreevError::Storage(format!(
                            "bootstrap failed: {} — in: {sql}",
                            pg_err(e)
                        ))
                    })?;
                }
                // LAST, inside the same transaction: the stamp is only
                // true if every statement above committed with it.
                if let Some(st) = stamp {
                    client
                        .execute(
                            &format!(
                                "INSERT INTO \"{schema}\".{}(k, v) VALUES ($1, $2) \
                                 ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v",
                                st.table
                            ),
                            &[&st.key, &st.version],
                        )
                        .await
                        .map_err(pg_err)?;
                }
                Ok::<_, AreevError>(())
            }
            .await;
            match boot {
                Ok(()) => client.batch_execute("COMMIT").await.map_err(pg_err),
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    Err(e)
                }
            }
        })
    }

    /// True when this open found the schema already at the caller's stamped
    /// version and therefore ran no bootstrap at all.
    pub(crate) fn bootstrap_skipped(&self) -> bool {
        self.bootstrap_skipped
    }

    /// The pool-size mismatch this open could not honour, if any.
    pub(crate) fn pool_warning(&self) -> Option<&str> {
        self.pool_warning.as_deref()
    }

    /// Read the bootstrap marker without naming the stamp table in a `FROM`
    /// clause until it is known to exist.
    ///
    /// `to_regclass` takes the relation name as *text*, so a schema (or table)
    /// that is not there answers NULL rather than aborting the statement with
    /// `42P01`/`3F000` — which matters because the absent case is the FIRST
    /// open of every new memory, and an error there would be a permanent line
    /// of noise in the server log for a wholly expected condition. Needs no
    /// privilege beyond `USAGE` on the schema and `SELECT` on the table.
    ///
    /// Kept deliberately adjacent to [`Self::verify_read_only`], which asks
    /// the same question for a read-only open and by a different method (it
    /// probes the five tables `finish_open` reads, because a read-only open
    /// must also work against a schema stamped by an OLDER build, which has no
    /// `pg_schema` row at all). Change one and re-read the other.
    async fn schema_stamp(
        client: &tokio_postgres::Client,
        schema: &str,
        stamp: PgStamp,
    ) -> Result<Option<String>> {
        let present = client
            .query_one(
                "SELECT to_regclass($1) IS NOT NULL",
                &[&format!("\"{schema}\".{}", stamp.table)],
            )
            .await
            .map_err(pg_err)?
            .get::<_, bool>(0);
        if !present {
            return Ok(None);
        }
        let row = client
            .query_opt(
                &format!("SELECT v FROM \"{schema}\".{} WHERE k = $1", stamp.table),
                &[&stamp.key],
            )
            .await
            .map_err(pg_err)?;
        Ok(row.and_then(|r| r.get::<_, Option<String>>(0)))
    }

    /// The `?provision=never` refusal. Names which of the two operator
    /// actions is needed — create the memory, or migrate it — the same way
    /// [`Self::verify_read_only`] does, because they are different jobs and a
    /// single "not provisioned" would send an operator hunting for the wrong
    /// one.
    fn not_provisioned(
        client: &tokio_postgres::Client,
        pool: &PgPool,
        schema: &str,
        stamp: Option<PgStamp>,
    ) -> AreevError {
        let version = stamp.map(|s| s.version).unwrap_or(PG_SCHEMA_VERSION);
        let exists = pool
            .block_on(client.query_one(
                "SELECT EXISTS(SELECT 1 FROM information_schema.schemata WHERE schema_name = $1)",
                &[&schema],
            ))
            .map(|r| r.get::<_, bool>(0))
            .unwrap_or(false);
        AreevError::SchemaNotProvisioned(if exists {
            format!(
                "postgres schema {schema:?} exists but is not stamped at schema version \
                 {version}, and this DSN says provision=never — so no bootstrap DDL was \
                 attempted. It was created by an older build, or a migration did not finish: \
                 run `areev provision --db <dsn> --schema {schema}` (or your migration job) \
                 with a role that owns the schema, then retry"
            )
        } else {
            format!(
                "postgres schema {schema:?} does not exist, and this DSN says provision=never \
                 — so no CREATE SCHEMA was attempted. Either the schema name is wrong, or the \
                 memory has never been created: run `areev provision --db <dsn> --schema \
                 {schema}` with a role that may create it, then retry"
            )
        })
    }

    /// The read-only counterpart to the bootstrap block above: confirm with
    /// SELECT-only probes that there is something real to open — never
    /// attempt to create anything. Distinguishes the two failure modes an
    /// operator must act on differently: a schema that was never
    /// created/migrated at all, versus one that exists but is missing tables
    /// this build's bootstrap would have added (a partial or outdated
    /// migration) — both need an owning role to run bootstrap read-write, but
    /// "absent" also means the schema name itself may be wrong.
    ///
    /// The table list mirrors exactly what `finish_open` reads
    /// unconditionally right after `open()` returns (`meta`, `terms`,
    /// `grains`, `oplog`, `fts_doc`) — not the full `PG_SCHEMA` list — so a
    /// pass here is precisely "the read path this crate always takes on
    /// open will not immediately fail with `undefined table`".
    ///
    /// **Sibling of [`Self::schema_stamp`]** — the two are the only places
    /// that answer "is this schema usable?", and they answer it differently
    /// ON PURPOSE. The stamp is an equality test against THIS build's
    /// `PG_SCHEMA_VERSION` and gates whether to run the bootstrap; this one is
    /// a capability test ("are the tables a read needs present?") and gates
    /// whether to refuse the open. A read-only open must keep working against
    /// a schema bootstrapped by an older build — which carries no `pg_schema`
    /// row at all — so it deliberately does NOT consult the stamp. If you add
    /// a table to `PG_SCHEMA` that `finish_open` reads unconditionally, both
    /// need editing: the list below, and `PG_SCHEMA_VERSION`.
    async fn verify_read_only(
        client: &tokio_postgres::Client,
        schema: &str,
    ) -> Result<()> {
        let exists = client
            .query_one(
                "SELECT EXISTS(SELECT 1 FROM information_schema.schemata WHERE schema_name = $1)",
                &[&schema],
            )
            .await
            .map_err(pg_err)?
            .get::<_, bool>(0);
        if !exists {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {schema:?} does not exist — a read-only open never runs \
                 bootstrap DDL (CREATE SCHEMA included), so it cannot create it. Either the \
                 schema name is wrong, or the memory has never been created: have the owning \
                 role open it read-write once (or run the migration), then retry read-only"
            )));
        }
        const CORE_TABLES: &[&str] = &["meta", "terms", "grains", "oplog", "fts_doc"];
        let mut missing: Vec<&str> = Vec::new();
        for table in CORE_TABLES {
            let qualified = format!("\"{schema}\".{table}");
            let present = client
                .query_one("SELECT to_regclass($1) IS NOT NULL", &[&qualified])
                .await
                .map_err(pg_err)?
                .get::<_, bool>(0);
            if !present {
                missing.push(table);
            }
        }
        if !missing.is_empty() {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {schema:?} exists but is not fully initialized — missing \
                 table(s): {} — a read-only open never runs bootstrap DDL to add them. Have the \
                 owning role open this memory read-write once to finish migrating it, then \
                 retry read-only",
                missing.join(", ")
            )));
        }
        // Every dictionary lookup keys on `terms.term_hash` (#160). A schema
        // from before that column existed would not fail — it would answer
        // every recall empty, which is worse — so it is refused by name.
        let keyed = client
            .query_one(
                "SELECT EXISTS(SELECT 1 FROM information_schema.columns \
                   WHERE table_schema = $1 AND table_name = 'terms' AND column_name = 'term_hash')",
                &[&schema],
            )
            .await
            .map_err(pg_err)?
            .get::<_, bool>(0);
        if !keyed {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {schema:?} predates the digest-keyed dictionary (terms.term_hash \
                 is missing) — a read-only open never runs the migration that adds it. Have the \
                 owning role open this memory read-write once, then retry read-only"
            )));
        }
        Ok(())
    }

    /// Is this error the CONNECTION dying, as opposed to the statement failing?
    ///
    /// Only these justify discarding the connection: an ordinary constraint
    /// violation must stay an ordinary error. Covers SQLSTATE class 08
    /// (connection exception) and the three 57Pnn shutdown codes a managed
    /// Postgres sends on restart/failover — `57P01 terminating connection due
    /// to administrator command` is the one a Cloud SQL maintenance window
    /// produces. The client noticing the socket is gone (`is_closed`) is the
    /// other signal, checked where the checkout is at hand.
    fn is_connection_dead(e: &AreevError) -> bool {
        let AreevError::Storage(msg) = e else {
            return false;
        };
        msg.contains("57P01")
            || msg.contains("57P02")
            || msg.contains("57P03")
            || msg.contains("postgres error 08")
            || msg.contains("connection closed")
            || msg.contains("connection unexpectedly closed")
    }

    /// Run `f` on a connection: the pinned one inside a transaction, else a
    /// borrow for this call. A connection that died under `f` is discarded
    /// rather than returned.
    fn with_conn<T>(&self, f: impl FnOnce(&PgPool, &mut PooledConn) -> Result<T>) -> Result<T> {
        let mut pinned = self.pinned.borrow_mut();
        if let Some(co) = pinned.as_mut() {
            let r = f(&self.pool, co.conn());
            if let Err(e) = &r {
                co.note(e);
            }
            return r;
        }
        drop(pinned);
        let mut co = self.pool.checkout()?;
        let r = f(&self.pool, co.conn());
        if let Err(e) = &r {
            co.note(e);
        }
        r
    }

    /// Bring the connection's session in line with what this handle wants,
    /// outside a transaction (inside one, `begin` said it with `SET LOCAL`).
    /// A `RESET` is only ever issued to a session that SET the parameter,
    /// so an unloaded `hnsw.*` placeholder is never named.
    fn align_session(&self, pool: &PgPool, conn: &mut PooledConn) -> Result<()> {
        if self.in_txn.get() {
            return Ok(());
        }
        let want_ef = self.ann_ef_search.get();
        let want_exact = self.exact_scan.get();
        let mut sql = String::new();
        if conn.ef_search != want_ef {
            match want_ef {
                Some(ef) => sql.push_str(&format!("SET hnsw.ef_search = {ef}; ")),
                None => sql.push_str("RESET hnsw.ef_search; "),
            }
        }
        if conn.exact_scan != want_exact {
            sql.push_str(if want_exact {
                "SET enable_indexscan = off; SET enable_bitmapscan = off; "
            } else {
                "RESET enable_indexscan; RESET enable_bitmapscan; "
            });
        }
        if sql.is_empty() {
            return Ok(());
        }
        pool.block_on(conn.client.batch_execute(&sql)).map_err(pg_err)?;
        conn.ef_search = want_ef;
        conn.exact_scan = want_exact;
        Ok(())
    }

    /// The translated, schema-qualified form of `sql` — memoised for the
    /// `_hot` literals, which are the fixed set worth remembering.
    fn translated(&self, sql: &str, hot: bool) -> Result<String> {
        if hot {
            if let Some(t) = self.xlate.borrow().get(sql) {
                return Ok(t.clone());
            }
        }
        let t = qualify_tables(&translate(sql)?, &self.schema);
        if hot {
            self.xlate.borrow_mut().insert(sql.to_string(), t.clone());
        }
        Ok(t)
    }

    /// Prepare-and-cache on THIS connection. ONLY the `_hot` calls come
    /// through here: their SQL is a fixed set of literals per schema, and the
    /// cache is emptied when it grows past what one connection should hold
    /// server-side (a thousand schemas through one connection would
    /// otherwise pin a thousand copies of every hot statement).
    fn prepared(
        pool: &PgPool,
        conn: &mut PooledConn,
        translated: &str,
    ) -> Result<tokio_postgres::Statement> {
        if let Some(st) = conn.cache.get(translated) {
            return Ok(st.clone());
        }
        let st = pool.block_on(conn.client.prepare(translated)).map_err(|e| {
            // Keep the SQLSTATE and server message (`pg_err`), not the
            // driver's bare "db error", so a failing prepare names its cause.
            AreevError::Storage(format!(
                "prepare failed: {} — translated SQL: {translated}",
                pg_err(e)
            ))
        })?;
        if conn.cache.len() >= 512 {
            conn.cache.clear();
        }
        conn.cache.insert(translated.to_string(), st.clone());
        Ok(st)
    }

    /// Conformance-only: reset the session before a statement, the way a
    /// transaction-mode pooler's backend switch does between transactions.
    /// `RESET ALL` drops `search_path` and every GUC, so a suite run under it
    /// proves no statement resolves through the session. `deallocate` adds
    /// `DEALLOCATE ALL`, the prepared-statement half, for the one probe that
    /// exercises the `26000` retry on its own. Never inside a transaction — a
    /// pooler keeps one backend for its length.
    fn maybe_discard_session(&self, pool: &PgPool, conn: &mut PooledConn) -> Result<()> {
        if self.in_txn.get() {
            return Ok(());
        }
        // Read per statement, so a probe can turn the reset on and off
        // between its own calls; a shipped build compiles this to `None`.
        let sql = match session_chaos_mode().as_deref() {
            None => return Ok(()),
            Some("deallocate") => "RESET ALL; DEALLOCATE ALL",
            Some(_) => "RESET ALL",
        };
        pool.block_on(conn.client.batch_execute(sql)).map_err(pg_err)?;
        // The session no longer carries what this process set on it; say so,
        // and re-apply below rather than believing the old bookkeeping.
        conn.ef_search = None;
        conn.exact_scan = false;
        if sql.contains("DEALLOCATE") {
            conn.cache.clear();
        }
        Ok(())
    }

    fn query_once(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<Vec<Row>> {
        let translated = self.translated(sql, hot)?;
        self.with_conn(|pool, conn| {
            self.maybe_discard_session(pool, conn)?;
            self.align_session(pool, conn)?;
            let vals: Vec<PgVal> = params.into_iter().map(PgVal).collect();
            let refs: Vec<&(dyn ToSql + Sync)> =
                vals.iter().map(|v| v as &(dyn ToSql + Sync)).collect();
            let rows = if hot {
                let st = Self::prepared(pool, conn, &translated)?;
                pool.block_on(conn.client.query(&st, &refs))
            } else {
                // Uncached path: the temporary statement is closed on drop, so
                // dynamic SQL leaves nothing behind on either side.
                pool.block_on(conn.client.query(translated.as_str(), &refs))
            }
            .map_err(pg_err)
            .map_err(|e| Self::forget_stale(conn, e))?;
            rows.iter().map(row_to_row).collect()
        })
    }

    fn execute_once(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<u64> {
        let translated = self.translated(sql, hot)?;
        self.with_conn(|pool, conn| {
            self.maybe_discard_session(pool, conn)?;
            self.align_session(pool, conn)?;
            let vals: Vec<PgVal> = params.into_iter().map(PgVal).collect();
            let refs: Vec<&(dyn ToSql + Sync)> =
                vals.iter().map(|v| v as &(dyn ToSql + Sync)).collect();
            if hot {
                let st = Self::prepared(pool, conn, &translated)?;
                pool.block_on(conn.client.execute(&st, &refs))
            } else {
                pool.block_on(conn.client.execute(translated.as_str(), &refs))
            }
            .map_err(pg_err)
            .map_err(|e| Self::forget_stale(conn, e))
        })
    }

    /// A `26000` means this session lost its named statements wholesale (a
    /// reset, or a backend switch behind an external pooler), so every
    /// cached name is suspect: drop them all while the connection is still
    /// at hand, and let the caller decide whether to retry.
    fn forget_stale(conn: &mut PooledConn, e: AreevError) -> AreevError {
        if matches!(&e, AreevError::Storage(m) if m.starts_with("postgres error 26000")) {
            conn.cache.clear();
        }
        e
    }

    /// A read, replayed once if the connection died under it or the backend
    /// no longer knew a cached statement.
    ///
    /// Replay is safe here precisely because it is a read: no effect to
    /// duplicate. This is what turns a routine managed-Postgres restart from
    /// "this handle is dead until the process is recycled" into one slow call.
    fn run_query(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<Vec<Row>> {
        match self.query_once(sql, params.clone(), hot) {
            Ok(rows) => Ok(rows),
            Err(e) => {
                if self.stale_statement(&e, hot) || self.recover(&e, true) {
                    self.query_once(sql, params, hot)
                } else {
                    Err(Self::explain_stale(e))
                }
            }
        }
    }

    /// Whether a `26000` may be retried: only a hot statement (the cache was
    /// cleared where it failed) and only outside a transaction — inside one
    /// the failed Bind has already aborted it, and a retry would only add
    /// `25P02`. The caller gets the 26000, with the remedy named
    /// (see `explain_stale`).
    fn stale_statement(&self, e: &AreevError, hot: bool) -> bool {
        if !hot {
            return false;
        }
        let AreevError::Storage(msg) = e else { return false };
        msg.starts_with("postgres error 26000") && !self.in_txn.get()
    }

    /// After a connection-level failure, whether the failed call may be
    /// replayed. The dead connection was already discarded by the checkout
    /// that noticed; the next borrow dials a fresh one.
    ///
    /// `replayable` is the whole safety argument. A READ outside a transaction
    /// can be re-run: it had no effect, so running it twice is running it
    /// once. A WRITE cannot — the connection may have died AFTER the server
    /// committed it, and nothing on this side can tell the difference, so
    /// replaying risks applying it twice. Those calls get this error now,
    /// which is the honest outcome: the host learns its unit of work did not
    /// complete.
    fn recover(&self, e: &AreevError, replayable: bool) -> bool {
        if self.in_txn.get() {
            return false;
        }
        Self::is_connection_dead(e) && replayable
    }

    /// A `26000` that could not be retried (it happened mid-transaction) is
    /// a pooler handing transactions to backends that never saw the named
    /// statement. Say so, and name what fixes it: the driver names every
    /// parameterized statement, so the pooler must track them or run in
    /// session mode — no store-side switch can substitute.
    fn explain_stale(e: AreevError) -> AreevError {
        match e {
            AreevError::Storage(msg) if msg.starts_with("postgres error 26000") => {
                AreevError::Storage(format!(
                    "{msg} — a prepared statement was unknown to this backend \
                     mid-transaction, which is what a transaction-mode pooler that does \
                     not track prepared statements does; use PgBouncer 1.21+ with \
                     max_prepared_statements > 0 (or another pooler that tracks them), \
                     or session mode"
                ))
            }
            other => other,
        }
    }

    /// A write. Never replayed — see [`recover`](Self::recover) for why a
    /// possibly-committed write must not be re-run; a stale statement is the
    /// one exception, since the server refused at Bind, before anything ran.
    fn run_execute(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<u64> {
        match self.execute_once(sql, params.clone(), hot) {
            Ok(n) => Ok(n),
            Err(e) => {
                if self.stale_statement(&e, hot) {
                    return self.execute_once(sql, params, hot);
                }
                Err(Self::explain_stale(e))
            }
        }
    }

    /// End the pinned transaction with `verb`, giving the connection back.
    fn end_txn(&self, verb: &str) -> Result<()> {
        let co = self.pinned.borrow_mut().take();
        // Cleared either way: a failed COMMIT ends the transaction just as
        // surely as a successful one.
        self.in_txn.set(false);
        let Some(mut co) = co else {
            return Ok(());
        };
        let r = self.pool.block_on(co.client().batch_execute(verb)).map_err(pg_err);
        if let Err(e) = &r {
            co.note(e);
        }
        r
    }
}

impl Drop for PgDb {
    fn drop(&mut self) {
        // A transaction left open (a panic mid-write, a caller that never
        // committed) must not travel back to the pool with the connection.
        if self.pinned.borrow().is_some() {
            let _ = self.end_txn("ROLLBACK");
        }
    }
}

impl Db for PgDb {
    fn execute(&self, sql: &str, params: Vec<Value>) -> Result<u64> {
        self.run_execute(sql, params, false)
    }

    fn query(&self, sql: &str, params: Vec<Value>) -> Result<Vec<Row>> {
        self.run_query(sql, params, false)
    }

    fn execute_hot(&self, sql: &'static str, params: Vec<Value>) -> Result<u64> {
        self.run_execute(sql, params, true)
    }

    fn query_hot(&self, sql: &'static str, params: Vec<Value>) -> Result<Vec<Row>> {
        self.run_query(sql, params, true)
    }

    fn begin(&self) -> Result<()> {
        if self.pinned.borrow().is_some() {
            // Postgres itself only warns on a nested BEGIN; keep the one
            // connection and the one transaction.
            return Ok(());
        }
        let mut co = self.pool.checkout()?;
        let r = {
            let conn = co.conn();
            // Conformance-only `deallocate-txn`: a backend switch AT
            // transaction start, which is what a non-tracking pooler does —
            // the statements cached so far are unknown to the backend this
            // transaction lands on.
            if session_chaos_mode().as_deref() == Some("deallocate-txn") {
                self.pool
                    .block_on(conn.client.batch_execute("DEALLOCATE ALL"))
                    .map_err(pg_err)?;
                conn.cache.clear();
            }
            // BEGIN and, in the same round trip, what this handle wants for
            // the transaction's length — LOCAL, so the connection returns to
            // the pool as it came.
            let sql = txn_prelude(self.ann_ef_search.get(), self.exact_scan.get(), conn);
            self.pool.block_on(conn.client.batch_execute(&sql)).map_err(pg_err)
        };
        match r {
            Ok(()) => {
                *self.pinned.borrow_mut() = Some(co);
                self.in_txn.set(true);
                Ok(())
            }
            Err(e) => {
                // Set only on success, so a failed BEGIN does not wedge the
                // handle into "a transaction is open".
                co.note(&e);
                Err(e)
            }
        }
    }

    fn commit(&self) -> Result<()> {
        self.end_txn("COMMIT")
    }

    fn rollback(&self) -> Result<()> {
        self.end_txn("ROLLBACK")
    }

    fn prefers_batched_reads(&self) -> bool {
        true // a round trip dominates: one batch beats k point probes
    }

    fn reserve_write(
        &self,
        n_seq: i64,
        n_op: i64,
        n_hlc: i64,
        wall_hlc: i64,
        hlc_floor: i64,
    ) -> Result<Option<WriteIds>> {
        // One statement, three counter rows, locked in a consistent order
        // (identical statement text for every writer -> identical plan) —
        // held until the enclosing transaction commits, which is what
        // serializes concurrent write transactions per memory.
        //
        // Counters hold the LAST USED id. New hlc last =
        // max(old + n, wall + n - 1, floor); the first of the block is
        // last - (n - 1).
        // Any local write may change the BM25 collection stats.
        *self.stats.borrow_mut() = None;
        let rows = self.run_query(
            "UPDATE counters SET v = CASE name \
               WHEN 'seq' THEN v + ?1 \
               WHEN 'op' THEN v + ?2 \
               ELSE GREATEST(v + ?3, ?4 + ?3 - 1, ?5) END \
             WHERE name IN ('seq','op','hlc') RETURNING name, v",
            vec![pi(n_seq), pi(n_op), pi(n_hlc), pi(wall_hlc), pi(hlc_floor)],
            true,
        )?;
        let (mut seq_last, mut op_last, mut hlc_last) = (0i64, 0i64, 0i64);
        for row in &rows {
            match (row.text(0), row.i64(1)) {
                (Some("seq"), Some(v)) => seq_last = v,
                (Some("op"), Some(v)) => op_last = v,
                (Some("hlc"), Some(v)) => hlc_last = v,
                _ => {}
            }
        }
        if rows.len() != 3 {
            return Err(AreevError::Storage(format!(
                "counters rows missing ({} of 3) — schema not seeded?",
                rows.len()
            )));
        }
        Ok(Some(WriteIds {
            seq0: seq_last - n_seq + 1,
            op0: op_last - n_op + 1,
            hlc0: hlc_last - (n_hlc - 1).max(0),
        }))
    }

    fn intern_term(&self, term: &str) -> Result<Option<i64>> {
        // Keyed by digest (#160): the index entry is 32 bytes whatever the
        // term's size or entropy. The conflict target is the unique index
        // `PG_SEED` creates on `term_hash`.
        let hash = term_hash(term);
        let inserted = self.run_query(
            "INSERT INTO terms(term, term_hash) VALUES (?1, ?2) ON CONFLICT (term_hash) DO NOTHING RETURNING id",
            vec![pt(term), pb(hash.clone())],
            true,
        )?;
        if let Some(id) = inserted.first().and_then(|r| r.i64(0)) {
            return Ok(Some(id));
        }
        let existing =
            self.run_query("SELECT id FROM terms WHERE term_hash = ?1", vec![pb(hash)], true)?;
        match existing.first().and_then(|r| r.i64(0)) {
            Some(id) => Ok(Some(id)),
            None => Err(AreevError::Storage("terms insert vanished".into())),
        }
    }

    fn lookup_term(&self, term: &str) -> Result<Option<i64>> {
        Ok(self
            .run_query(
                "SELECT id FROM terms WHERE term_hash = ?1",
                vec![pb(term_hash(term))],
                true,
            )?
            .first()
            .and_then(|r| r.i64(0)))
    }

    fn scrub_term(&self, id: i64, placeholder: &str) -> Result<u64> {
        // The digest must follow the string: an erased term left keyed by
        // its old hash would hand the scrubbed id back to the next intern of
        // the erased string, and a fresh grain would point at the placeholder.
        self.execute(
            "UPDATE terms SET term = ?2, term_hash = ?4 WHERE id = ?1 \
             AND NOT EXISTS (SELECT 1 FROM triples WHERE s=?1 OR p=?1 OR o=?1) \
             AND NOT EXISTS (SELECT 1 FROM thread_idx WHERE session=?1) \
             AND NOT EXISTS (SELECT 1 FROM run_idx WHERE run=?1) \
             AND NOT EXISTS (SELECT 1 FROM grains WHERE ns=?1) \
             AND term NOT LIKE ?3",
            vec![pi(id), pt(placeholder), pt("\u{1}erased:%"), pb(term_hash(placeholder))],
        )
    }

    fn lookup_term_str(&self, id: i64) -> Result<Option<String>> {
        Ok(self
            .run_query("SELECT term FROM terms WHERE id = ?1", vec![pi(id)], true)?
            .first()
            .and_then(|r| r.text(0).map(str::to_string)))
    }

    fn collection_stats(&self) -> Result<Option<(i64, i64)>> {
        if let Some(s) = *self.stats.borrow() {
            return Ok(Some(s));
        }
        let rows = self.run_query(
            "SELECT COUNT(*), COALESCE(SUM(len),0)::bigint FROM fts_doc",
            vec![],
            true,
        )?;
        let s = rows.first().map(|r| (r.i64(0).unwrap_or(0), r.i64(1).unwrap_or(0)));
        *self.stats.borrow_mut() = s;
        Ok(s)
    }

    fn for_update(&self) -> &'static str {
        " FOR UPDATE"
    }

    fn terms_with_prefix(&self, prefix: &str) -> Result<Option<Vec<(i64, String)>>> {
        // `LIKE 'prefix%'`, with the pattern metacharacters in the caller's
        // prefix escaped — the step-action relation prefix carries none today,
        // but a dictionary prefix is caller data and a stray `_` would quietly
        // widen the scan.
        let mut pat = String::with_capacity(prefix.len() + 1);
        for c in prefix.chars() {
            if matches!(c, '%' | '_' | '\\') {
                pat.push('\\');
            }
            pat.push(c);
        }
        pat.push('%');
        let rows = self.run_query(
            "SELECT id, term FROM terms WHERE term LIKE ?1 ESCAPE '\\'",
            vec![pt(&pat)],
            true,
        )?;
        Ok(Some(
            rows.iter()
                .filter_map(|r| Some((r.i64(0)?, r.text(1)?.to_string())))
                .collect(),
        ))
    }

    fn seeds_state_at_open(&self) -> bool {
        // Everything `finish_open` would seed is already DB-authoritative
        // here: `reserve_write` owns seq/op/HLC, `intern_term`/`lookup_term*`/
        // `terms_with_prefix` own the dictionary, `collection_stats` owns the
        // BM25 counters. Seeding them would be a full dictionary transfer plus
        // four `COUNT(*)`/`MAX()` scans per open, over the network, for values
        // this handle never reads.
        false
    }

    fn may_have_legacy_fts_index(&self) -> bool {
        // `idx_fts` is a Turso-era artifact. This backend never created it.
        false
    }

    fn ensure_embeddings(&self, dim: usize) -> Result<()> {
        let schema = self.schema.clone();
        self.with_conn(|pool, conn| {
            pool.block_on(async {
                let client = &conn.client;
                // Best-effort: only required if the extension is not installed
                // yet, and refused on unprivileged roles where an admin already
                // installed it — the ALTER below gives the real error.
                // WITH SCHEMA public is load-bearing: without it pgvector's
                // objects would install into whatever schema is first on the
                // path — invisible to sibling memories, and destroyed
                // database-wide by that one memory's DROP SCHEMA erasure.
                let _ = client
                    .batch_execute("CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public")
                    .await;
                client
                    .batch_execute(&format!(
                        "ALTER TABLE \"{schema}\".embeddings ADD COLUMN IF NOT EXISTS vec vector({dim})"
                    ))
                    .await
                    .map_err(|e| {
                        AreevError::Storage(format!(
                            "cannot add vector({dim}) column (is the pgvector extension available?): {e}"
                        ))
                    })?;
                // A pre-existing column with a different dim would make every add
                // fail mid-transaction — refuse the embedder up front instead.
                let row = client
                    .query_one(
                        "SELECT a.atttypmod FROM pg_attribute a
                          JOIN pg_class c ON a.attrelid = c.oid
                          JOIN pg_namespace n ON c.relnamespace = n.oid
                         WHERE n.nspname = $1 AND c.relname = 'embeddings'
                           AND a.attname = 'vec'",
                        &[&schema],
                    )
                    .await
                    .map_err(pg_err)?;
                let existing = row.get::<_, i32>(0);
                if existing > 0 && existing as usize != dim {
                    return Err(AreevError::Validation(format!(
                        "embeddings column holds vector({existing}) but the embedder produces dim {dim} \
                         — refusing to mix vector spaces (this is a hard error on the postgres backend)"
                    )));
                }
                Ok(())
            })
        })
    }

    /// pgvector HNSW over `embeddings.vec`, cosine ops to match the `<=>`
    /// that `vector_distance_cos` translates into.
    ///
    /// Three things a caller must know, all of them consequences rather than
    /// opinions:
    ///
    /// 1. **Recall becomes approximate.** `ef_search` is the knob: raise it
    ///    for accuracy, lower it for latency. This is why the index is opt-in
    ///    and not in `PG_SCHEMA` — no one should acquire approximate results
    ///    by upgrading a binary.
    /// 2. **The build takes an exclusive lock** on `embeddings` and is O(rows)
    ///    — minutes at a million vectors. Not `CONCURRENTLY`: a failed
    ///    concurrent build leaves an INVALID index behind that still has to be
    ///    dropped by hand, and a store method that can leave the schema in a
    ///    state only a DBA can exit is worse than one that asks for a
    ///    maintenance window.
    /// 3. **It does not help a filtered query.** With `WHERE g.ns = ?` the
    ///    planner drives from `idx_grains_ns_s` and computes exact distances
    ///    over the survivors, which is both faster and exact. HNSW is for the
    ///    whole-corpus query that has no filter to narrow it.
    fn ensure_ann_index(&self, m: usize, ef_construction: usize, ef_search: usize) -> Result<()> {
        // pgvector's own bounds; rejected here so the failure names the knob
        // rather than surfacing a raw SQLSTATE from deep in the build.
        if !(2..=100).contains(&m) {
            return Err(AreevError::Validation(format!(
                "hnsw m must be 2..=100, got {m}"
            )));
        }
        if !(4..=1000).contains(&ef_construction) {
            return Err(AreevError::Validation(format!(
                "hnsw ef_construction must be 4..=1000, got {ef_construction}"
            )));
        }
        let schema = self.schema.clone();
        self.with_conn(|pool, conn| {
            pool.block_on(async {
                let client = &conn.client;
                // The column arrives with the first embedder (`ensure_embeddings`).
                // Without it there is nothing to index, and `CREATE INDEX` would
                // fail with an undefined-column error that reads like a bug.
                let has_vec = client
                    .query_one(
                        "SELECT count(*) FROM pg_attribute a
                           JOIN pg_class c ON a.attrelid = c.oid
                           JOIN pg_namespace n ON c.relnamespace = n.oid
                          WHERE n.nspname = $1 AND c.relname = 'embeddings'
                            AND a.attname = 'vec' AND NOT a.attisdropped",
                        &[&schema],
                    )
                    .await
                    .map_err(pg_err)?
                    .get::<_, i64>(0);
                if has_vec == 0 {
                    return Err(AreevError::AnnIndexUnsupported(
                        "this memory has no embeddings yet — install an embedder and store at least \
                         one vector before building an index over them"
                            .into(),
                    ));
                }
                client
                    .batch_execute(&format!(
                        "CREATE INDEX IF NOT EXISTS idx_embeddings_hnsw ON \"{schema}\".embeddings \
                         USING hnsw (vec vector_cosine_ops) WITH (m = {m}, ef_construction = {ef_construction})"
                    ))
                    .await
                    .map_err(|e| {
                        AreevError::Storage(format!("cannot build the HNSW index over embeddings: {e}"))
                    })
            })
        })?;
        // A wish of this HANDLE, applied to whichever connection serves it.
        // Not a file truth: a different host opening the same schema picks
        // its own accuracy/latency point, which is right — that trade
        // belongs to the caller, not to the data.
        self.ann_ef_search.set(Some(ef_search));
        Ok(())
    }

    fn set_ann_ef_search(&self, ef_search: usize) -> Result<()> {
        if !(1..=1000).contains(&ef_search) {
            return Err(AreevError::Validation(format!(
                "hnsw ef_search must be 1..=1000, got {ef_search}"
            )));
        }
        self.ann_ef_search.set(Some(ef_search));
        Ok(())
    }

    fn ann_ef_search(&self) -> Result<Option<usize>> {
        if let Some(ef) = self.ann_ef_search.get() {
            return Ok(Some(ef));
        }
        // Nothing asked for on this handle: report what the server would
        // use (a database- or role-level `ALTER … SET`, or the default).
        // `current_setting(_, true)` is NULL rather than an error when the
        // extension has never been loaded in this session.
        self.with_conn(|pool, conn| {
            let row = pool
                .block_on(conn.client.query_one("SELECT current_setting('hnsw.ef_search', true)", &[]))
                .map_err(pg_err)?;
            Ok(row.get::<_, Option<String>>(0).and_then(|v| v.parse().ok()))
        })
    }

    fn drop_ann_index(&self) -> Result<()> {
        let sql = format!("DROP INDEX IF EXISTS \"{}\".idx_embeddings_hnsw", self.schema);
        self.with_conn(|pool, conn| pool.block_on(conn.client.batch_execute(&sql)).map_err(pg_err))
    }

    fn set_exact_vector_scan(&self, on: bool) -> Result<()> {
        // pgvector serves an approximate k-NN through an index scan over the
        // HNSW graph; with index scans disabled the planner falls back to the
        // sequential scan that computes every distance, which is the exact
        // answer. A wish of this handle, like `ef_search`, applied to the
        // connection that serves each statement — and only until switched back.
        self.exact_scan.set(on);
        Ok(())
    }

    fn ann_index_name(&self) -> Result<Option<String>> {
        let schema = self.schema.clone();
        self.with_conn(|pool, conn| {
            // Straight off the catalogs, by access method. `pg_indexes` would
            // do, but its `indexdef` column is `pg_get_indexdef()` over every
            // index row in the snapshot, and a sibling schema dropped between
            // the snapshot and that call is an XX000 for a question about
            // THIS schema's index.
            let rows = pool
                .block_on(conn.client.query(
                    "SELECT c.relname FROM pg_index i
                       JOIN pg_class c ON c.oid = i.indexrelid
                       JOIN pg_class t ON t.oid = i.indrelid
                       JOIN pg_namespace n ON n.oid = t.relnamespace
                       JOIN pg_am am ON am.oid = c.relam
                      WHERE n.nspname = $1 AND t.relname = 'embeddings'
                        AND am.amname = 'hnsw'",
                    &[&schema],
                ))
                .map_err(pg_err)?;
            Ok(rows.first().map(|r| r.get::<_, String>(0)))
        })
    }
}

/// Bind-side bridge: `turso::Value` → Postgres, letting the statement's
/// declared parameter type drive the encoding (all-bigint DDL keeps this a
/// small matrix).
#[derive(Debug)]
struct PgVal(Value);

impl ToSql for PgVal {
    fn to_sql(
        &self,
        ty: &Type,
        out: &mut tokio_postgres::types::private::BytesMut,
    ) -> std::result::Result<IsNull, Box<dyn std::error::Error + Sync + Send>> {
        match &self.0 {
            Value::Null => Ok(IsNull::Yes),
            Value::Integer(i) => match *ty {
                Type::INT8 => i.to_sql(ty, out),
                Type::INT4 => (*i as i32).to_sql(&Type::INT4, out),
                Type::FLOAT8 => (*i as f64).to_sql(&Type::FLOAT8, out),
                Type::TEXT | Type::VARCHAR => i.to_string().to_sql(&Type::TEXT, out),
                _ => Err(format!("cannot bind Integer to {ty}").into()),
            },
            Value::Real(r) => match *ty {
                Type::FLOAT8 => r.to_sql(ty, out),
                Type::FLOAT4 => (*r as f32).to_sql(&Type::FLOAT4, out),
                _ => Err(format!("cannot bind Real to {ty}").into()),
            },
            Value::Text(s) => match *ty {
                Type::TEXT | Type::VARCHAR | Type::NAME => s.to_sql(&Type::TEXT, out),
                _ => Err(format!("cannot bind Text to {ty}").into()),
            },
            Value::Blob(b) => match *ty {
                Type::BYTEA => b.to_sql(ty, out),
                _ => Err(format!("cannot bind Blob to {ty}").into()),
            },
        }
    }

    fn accepts(_ty: &Type) -> bool {
        true // checked per value in to_sql
    }

    tokio_postgres::types::to_sql_checked!();
}

/// Result-side bridge: decode per the column's declared type into the
/// `Value` coercion contract the store expects.
fn row_to_row(row: &tokio_postgres::Row) -> Result<Row> {
    let mut vals = Vec::with_capacity(row.len());
    for (i, col) in row.columns().iter().enumerate() {
        let v = match *col.type_() {
            Type::INT8 => row
                .try_get::<_, Option<i64>>(i)
                .map(|o| o.map_or(Value::Null, Value::Integer)),
            Type::INT4 => row
                .try_get::<_, Option<i32>>(i)
                .map(|o| o.map_or(Value::Null, |x| Value::Integer(x as i64))),
            Type::INT2 => row
                .try_get::<_, Option<i16>>(i)
                .map(|o| o.map_or(Value::Null, |x| Value::Integer(x as i64))),
            Type::FLOAT8 => row
                .try_get::<_, Option<f64>>(i)
                .map(|o| o.map_or(Value::Null, Value::Real)),
            Type::FLOAT4 => row
                .try_get::<_, Option<f32>>(i)
                .map(|o| o.map_or(Value::Null, |x| Value::Real(x as f64))),
            Type::TEXT | Type::VARCHAR | Type::NAME | Type::UNKNOWN => row
                .try_get::<_, Option<String>>(i)
                .map(|o| o.map_or(Value::Null, Value::Text)),
            Type::BYTEA => row
                .try_get::<_, Option<Vec<u8>>>(i)
                .map(|o| o.map_or(Value::Null, Value::Blob)),
            Type::BOOL => row
                .try_get::<_, Option<bool>>(i)
                .map(|o| o.map_or(Value::Null, |b| Value::Integer(b as i64))),
            ref other => {
                return Err(AreevError::Storage(format!(
                    "unmapped postgres column type {other} at index {i} — extend row_to_row"
                )))
            }
        }
        .map_err(pg_err)?;
        vals.push(v);
    }
    Ok(Row(vals))
}

/// Upsert conflict targets per table — SQLite `INSERT OR REPLACE` becomes
/// `ON CONFLICT (<pk>) DO UPDATE` of the non-key columns.
fn replace_suffix(table: &str) -> Option<(&'static str, &'static str)> {
    match table {
        "meta" => Some(("(k)", "v = EXCLUDED.v")),
        "fts_doc" => Some(("(seq)", "len = EXCLUDED.len")),
        "entity_latest" => {
            Some(("(ns,s,p)", "o = EXCLUDED.o, seq = EXCLUDED.seq, hash = EXCLUDED.hash"))
        }
        "heads" => Some(("(ns,s,p,seq)", "hash = EXCLUDED.hash, created_at = EXCLUDED.created_at")),
        "telem_meta" => Some(("(k)", "v = EXCLUDED.v")),
        // telem_grain_access / telem_query_stat / telem_budget_stat are NOT
        // here: their SQLite form is read-modify-write (correlated COALESCE
        // subqueries), which under CONCURRENT flushes into one shared Pg
        // sidecar loses increments. translate() rewrites them to atomic
        // accumulating upserts instead — see the prefix dispatch there.
        _ => None,
    }
}

/// Translate one statement from the store's SQLite dialect to Postgres.
/// Every divergent construct is handled explicitly or the translation FAILS
/// with the offending SQL — never execute something subtly different.
/// Every table the Postgres backend creates (`PG_SCHEMA`, `ensure_embeddings`,
/// and the telemetry sidecar's `TELEM_SCHEMA_PG`). The qualifier rewrites a
/// reference to any of these; catalog tables (`pg_class`, `information_schema`)
/// are deliberately absent, since they must resolve globally.
pub(crate) const PG_TABLES: &[&str] = &[
    "blobs", "corpus_idx", "counters", "embeddings", "entity_latest", "fts_doc", "fts_post",
    "fts_vocab", "grains", "heads", "meta", "ns_reg", "oplog", "osp", "prov_idx", "run_idx",
    "telem_budget_stat", "telem_grain_access", "telem_meta", "telem_query_stat",
    "telem_recall_log", "terms", "thread_idx", "triples",
];

/// Schema-qualify every table reference in `sql` (#181): `FROM grains g` →
/// `FROM "s".grains g`. This is what frees the store from the session's
/// `search_path` — a transaction-mode pooler hands each transaction to
/// whichever backend is free, and a statement that relied on `search_path`
/// would then read another schema's tables or none. Quote-aware (a table name
/// inside a string literal is data) and idempotent (an already-qualified name
/// is left alone). Only the keywords a table name can follow trigger it, so
/// `counters.v` in an `ON CONFLICT … DO UPDATE` stays the target's own name,
/// which Postgres resolves without a schema.
pub(crate) fn qualify_tables(sql: &str, schema: &str) -> String {
    const TRIGGERS: &[&str] = &["FROM", "JOIN", "INTO", "UPDATE", "TABLE", "EXISTS"];
    let mut out = String::with_capacity(sql.len() + 64);
    let chars: Vec<char> = sql.chars().collect();
    let n = chars.len();
    let mut i = 0;
    let mut in_str = false;
    let mut in_ident = false;
    let mut prev_word_is_trigger = false;
    let is_word = |c: char| c.is_ascii_alphanumeric() || c == '_';
    while i < n {
        let c = chars[i];
        if in_str {
            out.push(c);
            if c == '\'' {
                in_str = false;
            }
            i += 1;
            continue;
        }
        if in_ident {
            out.push(c);
            if c == '"' {
                in_ident = false;
            }
            i += 1;
            continue;
        }
        match c {
            '\'' => {
                in_str = true;
                prev_word_is_trigger = false;
                out.push(c);
                i += 1;
            }
            '"' => {
                in_ident = true;
                prev_word_is_trigger = false;
                out.push(c);
                i += 1;
            }
            c if is_word(c) => {
                let start = i;
                while i < n && is_word(chars[i]) {
                    i += 1;
                }
                let word: String = chars[start..i].iter().collect();
                let preceded_by_dot = start > 0 && chars[start - 1] == '.';
                if prev_word_is_trigger && !preceded_by_dot && PG_TABLES.contains(&word.as_str()) {
                    out.push('"');
                    out.push_str(schema);
                    out.push_str("\".");
                }
                out.push_str(&word);
                prev_word_is_trigger = TRIGGERS.contains(&word.to_ascii_uppercase().as_str());
            }
            c if c.is_whitespace() => {
                out.push(c);
                i += 1;
            }
            _ => {
                // Any other punctuation ends the "next word may be a table"
                // window: `EXISTS (SELECT`, `FOR UPDATE)`, `DO UPDATE SET`.
                prev_word_is_trigger = false;
                out.push(c);
                i += 1;
            }
        }
    }
    out
}

pub(crate) fn translate(sql: &str) -> Result<String> {
    let trimmed = sql.trim();
    // Exact-match specials first.
    if trimmed == "PRAGMA integrity_check" {
        // No engine-level analogue, so the check becomes the index-layer
        // referential invariants (every index row must point at a grains
        // row — forget deletes them all precisely because seq is reused).
        // Zero rows = clean; verify() classifies any returned line as a
        // real integrity problem. The content-address re-hash remains the
        // portable tamper check.
        return Ok("SELECT * FROM (\
            SELECT 'heads rows without a grain: ' || count(*)::text AS line \
              FROM heads h LEFT JOIN grains g ON g.seq = h.seq \
             WHERE g.seq IS NULL HAVING count(*) > 0 \
            UNION ALL \
            SELECT 'entity_latest rows without a grain: ' || count(*)::text \
              FROM entity_latest e LEFT JOIN grains g ON g.seq = e.seq \
             WHERE g.seq IS NULL HAVING count(*) > 0 \
            UNION ALL \
            SELECT 'triples rows without a grain: ' || count(*)::text \
              FROM triples t LEFT JOIN grains g ON g.seq = t.seq \
             WHERE g.seq IS NULL HAVING count(*) > 0 \
            UNION ALL \
            SELECT 'osp rows without a grain: ' || count(*)::text \
              FROM osp o LEFT JOIN grains g ON g.seq = o.seq \
             WHERE g.seq IS NULL HAVING count(*) > 0 \
            ) checks"
            .into());
    }
    if trimmed == "DROP INDEX idx_fts" {
        return Ok("DROP INDEX IF EXISTS idx_fts".into());
    }
    if trimmed == "SELECT COALESCE(SUM(len),0) FROM fts_doc" {
        // SUM(bigint) is numeric in Pg; the store's coercion contract wants
        // Integer.
        return Ok("SELECT COALESCE(SUM(len),0)::bigint FROM fts_doc".into());
    }

    // Telemetry rollups: the SQLite form is read-modify-write (correlated
    // COALESCE subqueries + REPLACE), which loses increments when several
    // instances flush into ONE shared Pg sidecar concurrently. Rewrite them
    // to atomic accumulating upserts keyed on the same bind order the
    // caller uses. Autocommit per statement (flush skips its transaction on
    // this backend), so concurrent flushes can't deadlock either.
    if trimmed.starts_with("INSERT OR REPLACE INTO telem_grain_access") {
        // binds: $1 hash, $2 ns, $3 ts_ms
        return Ok("INSERT INTO telem_grain_access(hash, ns, recall_count, first_ms, last_ms) \
                   VALUES ($1, $2, 1, $3, $3) \
                   ON CONFLICT (hash) DO UPDATE SET \
                     recall_count = telem_grain_access.recall_count + 1, \
                     last_ms = EXCLUDED.last_ms, ns = EXCLUDED.ns"
            .into());
    }
    if trimmed.starts_with("INSERT OR REPLACE INTO telem_query_stat") {
        // binds: $1 qkey, $2 ns, $3 sample, $4 ts_ms, $5 empty_delta, $6 n_results
        return Ok("INSERT INTO telem_query_stat(qkey, ns, sample, run_count, last_ms, empty_count, sum_results) \
                   VALUES ($1, $2, $3, 1, $4, $5, $6) \
                   ON CONFLICT (qkey) DO UPDATE SET \
                     run_count = telem_query_stat.run_count + 1, \
                     last_ms = EXCLUDED.last_ms, \
                     empty_count = telem_query_stat.empty_count + EXCLUDED.empty_count, \
                     sum_results = telem_query_stat.sum_results + EXCLUDED.sum_results, \
                     sample = EXCLUDED.sample, ns = EXCLUDED.ns"
            .into());
    }
    if trimmed.starts_with("INSERT OR REPLACE INTO telem_budget_stat") {
        // binds: $1 overflow_delta
        return Ok("INSERT INTO telem_budget_stat(id, sample_count, overflow_count) \
                   VALUES (1, 1, $1) \
                   ON CONFLICT (id) DO UPDATE SET \
                     sample_count = telem_budget_stat.sample_count + 1, \
                     overflow_count = telem_budget_stat.overflow_count + EXCLUDED.overflow_count"
            .into());
    }

    let mut s = sql.to_string();

    // INSERT OR REPLACE INTO <table> …  →  INSERT INTO <table> … ON CONFLICT …
    if let Some(rest) = s.trim_start().strip_prefix("INSERT OR REPLACE INTO ") {
        let table: String =
            rest.chars().take_while(|c| c.is_ascii_alphanumeric() || *c == '_').collect();
        let (target, set) = replace_suffix(&table).ok_or_else(|| {
            AreevError::Storage(format!("no upsert mapping for table {table:?} in: {sql}"))
        })?;
        s = format!(
            "INSERT INTO {rest} ON CONFLICT {target} DO UPDATE SET {set}",
            rest = rest.trim_end()
        );
    } else if let Some(rest) = s.trim_start().strip_prefix("INSERT OR IGNORE INTO ") {
        s = format!("INSERT INTO {} ON CONFLICT DO NOTHING", rest.trim_end());
    }

    // vector_distance_cos(a, b) → (a <=> b) — before the vector32 pass so
    // nested vector32 args are still in their source spelling when we scan
    // for the top-level comma.
    while let Some(pos) = s.find("vector_distance_cos(") {
        let args_start = pos + "vector_distance_cos(".len();
        let (a, b, end) = split_two_args(&s[args_start..]).ok_or_else(|| {
            AreevError::Storage(format!("unparseable vector_distance_cos in: {sql}"))
        })?;
        let repl = format!("({} <=> {})", a.trim(), b.trim());
        s = format!("{}{}{}", &s[..pos], repl, &s[args_start + end + 1..]);
    }

    // vector32(X) → ((X)::text::vector): the param stays a plain text bind
    // (the JSON-array rendering) and the server casts it, so no pgvector
    // binary wire format is needed.
    while let Some(pos) = s.find("vector32(") {
        let args_start = pos + "vector32(".len();
        let end = s[args_start..]
            .find(')')
            .ok_or_else(|| AreevError::Storage(format!("unparseable vector32 in: {sql}")))?;
        let inner = s[args_start..args_start + end].to_string();
        s = format!("{}(({})::text::vector){}", &s[..pos], inner, &s[args_start + end + 1..]);
    }

    // Placeholders: ?N → $N. Char-based (a byte scan would re-encode
    // multi-byte UTF-8 one byte per char) and quote-aware (a '?1' inside a
    // single-quoted literal is data, not a placeholder; the '' escape
    // toggles the state twice, which nets out correct).
    let mut out = String::with_capacity(s.len());
    let mut in_str = false;
    let mut chars = s.chars().peekable();
    while let Some(c) = chars.next() {
        match c {
            '\'' => {
                in_str = !in_str;
                out.push(c);
            }
            '?' if !in_str && chars.peek().is_some_and(|n| n.is_ascii_digit()) => out.push('$'),
            _ => out.push(c),
        }
    }

    // Fail fast on anything divergent that slipped through.
    for marker in ["INSERT OR REPLACE", "INSERT OR IGNORE", "PRAGMA", "vector32", "vector_distance_cos"] {
        if out.contains(marker) {
            return Err(AreevError::Storage(format!(
                "untranslated {marker} construct in: {sql}"
            )));
        }
    }
    Ok(out)
}

/// Split `"a, b)…"` at the top-level comma and matching close paren.
/// Returns (a, b, index_of_close_paren_relative_to_input).
fn split_two_args(s: &str) -> Option<(&str, &str, usize)> {
    let mut depth = 0usize;
    let mut comma = None;
    for (i, c) in s.char_indices() {
        match c {
            '(' => depth += 1,
            ')' => {
                if depth == 0 {
                    let comma = comma?;
                    return Some((&s[..comma], &s[comma + 1..i], i));
                }
                depth -= 1;
            }
            ',' if depth == 0 && comma.is_none() => comma = Some(i),
            _ => {}
        }
    }
    None
}

/// Split a `postgres://…?schema=<name>` URL into (connection URL, schema).
/// The `schema` query parameter is ours, not the driver's, so it is removed
/// from the URL handed to the connector. Hosts use this to accept one DSN
/// string wherever a file path is accepted today.
///
/// Everything else rides through untouched — including `sslmode`/`sslrootcert`
/// and `provision`, which are read further down the stack
/// ([`crate::pgtls::SslRequest::split`], [`provision_mode`]) off the URL this
/// returns. That is deliberate: every host that already calls this — the CLI,
/// the console, both bindings — then gets those parameters for free, without
/// this function's signature having to grow one member per option.
pub fn split_schema_url(url: &str) -> Result<(String, String)> {
    let Some((base, query)) = url.split_once('?') else {
        return Err(AreevError::Validation(
            "postgres URL needs ?schema=<name> — one memory = one schema, so the schema \
             must be named explicitly"
                .into(),
        ));
    };
    let mut schema = None;
    let mut rest: Vec<&str> = Vec::new();
    for pair in query.split('&') {
        match pair.split_once('=') {
            Some(("schema", v)) if !v.is_empty() => schema = Some(v.to_string()),
            _ => rest.push(pair),
        }
    }
    let schema = schema.ok_or_else(|| {
        AreevError::Validation("postgres URL needs ?schema=<name>".into())
    })?;
    let url = if rest.is_empty() {
        base.to_string()
    } else {
        format!("{base}?{}", rest.join("&"))
    };
    Ok((url, schema))
}

/// SHA-256 over the term's UTF-8 bytes: the dictionary key on this backend
/// (#160). `PG_SEED`'s backfill computes the same digest in SQL.
fn term_hash(term: &str) -> Vec<u8> {
    use sha2::{Digest, Sha256};
    Sha256::digest(term.as_bytes()).to_vec()
}

/// Run raw SQL against a Postgres URL, outside any memory handle. For the
/// conformance suite only — it needs to stage a schema in the shape an OLDER
/// build wrote (a `terms` table without `term_hash`) to prove the open-time
/// migration converges it.
///
/// Behind the non-default `conformance` feature, and deliberately: this is
/// arbitrary SQL against an arbitrary DSN, with none of the schema-name
/// validation [`drop_postgres_schema`] performs, so it must not exist at all
/// on a shipped build. `#[doc(hidden)]` would have hidden it from rustdoc
/// while leaving it callable.
#[cfg(feature = "conformance")]
pub fn execute_raw(url: &str, sql: &str) -> Result<()> {
    let (pool, _) = PgPool::for_url(url)?;
    let mut co = pool.checkout()?;
    let r = pool.block_on(co.client().batch_execute(sql)).map_err(pg_err);
    if let Err(e) = &r {
        co.note(e);
    }
    r
}

#[cfg(feature = "conformance")]
pub fn query_raw_i64(url: &str, sql: &str) -> Result<i64> {
    let (pool, _) = PgPool::for_url(url)?;
    let mut co = pool.checkout()?;
    let r = pool.block_on(co.client().query_one(sql, &[])).map_err(pg_err);
    if let Err(e) = &r {
        co.note(e);
    }
    Ok(r?.get::<_, i64>(0))
}

fn session_chaos_mode() -> Option<String> {
    #[cfg(feature = "conformance")]
    {
        std::env::var("AREEV_PG_SESSION_CHAOS").ok().filter(|v| !v.is_empty() && v != "0")
    }
    #[cfg(not(feature = "conformance"))]
    {
        None
    }
}

/// Drop a memory schema entirely — the Postgres backend's memory-level
/// erasure primitive (`DROP SCHEMA … CASCADE`), the analogue of deleting a
/// memory file. Admin-surface only: not reachable from CAL, and hosts must
/// gate it like any destructive operation.
pub fn drop_postgres_schema(url: &str, schema: &str) -> Result<()> {
    if !schema.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_') {
        return Err(AreevError::Validation(format!("invalid schema name {schema:?}")));
    }
    let (pool, _) = PgPool::for_url(url)?;
    let mut co = pool.checkout()?;
    let r = pool
        .block_on(co.client().batch_execute(&format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE")))
        .map_err(pg_err);
    if let Err(e) = &r {
        co.note(e);
    }
    r
}

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

    #[test]
    fn translates_placeholders() {
        assert_eq!(
            translate("SELECT seq FROM triples WHERE ns=?1 AND s=?2 LIMIT ?3").unwrap(),
            "SELECT seq FROM triples WHERE ns=$1 AND s=$2 LIMIT $3"
        );
        // a bare ? (not a placeholder) is left alone
        assert_eq!(translate("SELECT '?' FROM meta").unwrap(), "SELECT '?' FROM meta");
        // a '?1' inside a string literal is DATA, not a placeholder
        assert_eq!(
            translate("SELECT k FROM meta WHERE v = '?1' AND k = ?1").unwrap(),
            "SELECT k FROM meta WHERE v = '?1' AND k = $1"
        );
        // multi-byte characters survive intact ('' is the literal escape)
        assert_eq!(
            translate("SELECT 'café ''?2'' naïve' FROM meta WHERE k = ?1").unwrap(),
            "SELECT 'café ''?2'' naïve' FROM meta WHERE k = $1"
        );
    }

    #[test]
    fn integrity_check_maps_to_referential_invariants() {
        let q = translate("PRAGMA integrity_check").unwrap();
        for frag in ["heads rows without a grain", "entity_latest rows", "triples rows", "osp rows"] {
            assert!(q.contains(frag), "{q}");
        }
    }

    #[test]
    fn translates_upserts_per_table() {
        assert_eq!(
            translate("INSERT OR REPLACE INTO meta(k, v) VALUES (?1, ?2)").unwrap(),
            "INSERT INTO meta(k, v) VALUES ($1, $2) ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v"
        );
        let el = translate(
            "INSERT OR REPLACE INTO entity_latest(ns,s,p,o,seq,hash) VALUES (?1,?2,?3,?4,?5,?6)",
        )
        .unwrap();
        assert!(el.ends_with(
            "ON CONFLICT (ns,s,p) DO UPDATE SET o = EXCLUDED.o, seq = EXCLUDED.seq, hash = EXCLUDED.hash"
        ), "{el}");
        let heads = translate(
            "INSERT OR REPLACE INTO heads(ns,s,p,seq,hash,created_at) VALUES (?1,?2,?3,?4,?5,?6)",
        )
        .unwrap();
        assert!(heads.contains("ON CONFLICT (ns,s,p,seq) DO UPDATE"), "{heads}");
        assert!(translate("INSERT OR REPLACE INTO nope(x) VALUES (?1)").is_err());
    }

    #[test]
    fn translates_or_ignore_and_pragma() {
        assert_eq!(
            translate("INSERT OR IGNORE INTO fts_vocab(term) VALUES (?1)").unwrap(),
            "INSERT INTO fts_vocab(term) VALUES ($1) ON CONFLICT DO NOTHING"
        );
        assert!(translate("PRAGMA integrity_check").unwrap().starts_with("SELECT * FROM ("));
        assert_eq!(
            translate("SELECT COALESCE(SUM(len),0) FROM fts_doc").unwrap(),
            "SELECT COALESCE(SUM(len),0)::bigint FROM fts_doc"
        );
        // The namespace-registry statements ride the generic paths — pinned
        // here so a future per-table tightening cannot silently strand them.
        assert_eq!(
            translate("INSERT OR IGNORE INTO ns_reg(ns, n) VALUES (?1, 0)").unwrap(),
            "INSERT INTO ns_reg(ns, n) VALUES ($1, 0) ON CONFLICT DO NOTHING"
        );
        assert_eq!(
            translate("UPDATE ns_reg SET n = n + ?2 WHERE ns = ?1").unwrap(),
            "UPDATE ns_reg SET n = n + $2 WHERE ns = $1"
        );
        assert_eq!(
            translate("DELETE FROM ns_reg WHERE ns = ?1 AND n <= 0").unwrap(),
            "DELETE FROM ns_reg WHERE ns = $1 AND n <= 0"
        );
    }

    #[test]
    fn translates_vector_functions() {
        assert_eq!(
            translate("INSERT INTO embeddings(seq, vec) VALUES (?1, vector32(?2))").unwrap(),
            "INSERT INTO embeddings(seq, vec) VALUES ($1, (($2)::text::vector))"
        );
        let q = translate(
            "SELECT e.seq FROM embeddings e JOIN grains g ON g.seq=e.seq WHERE g.ns=?1 AND g.svt IS NULL ORDER BY vector_distance_cos(e.vec, vector32(?2)) LIMIT ?3",
        )
        .unwrap();
        assert!(
            q.contains("ORDER BY (e.vec <=> (($2)::text::vector)) LIMIT $3"),
            "{q}"
        );
        // pairwise self-join form
        let p = translate(
            "SELECT a.seq, b.seq, vector_distance_cos(a.vec, b.vec) FROM embeddings a JOIN embeddings b ON a.seq < b.seq",
        )
        .unwrap();
        assert!(p.contains("(a.vec <=> b.vec)"), "{p}");
        // AS dist form used by nearest_semantic
        let n = translate(
            "SELECT g.hash, vector_distance_cos(e.vec, vector32(?2)) AS dist FROM embeddings e JOIN grains g ON g.seq=e.seq WHERE g.ns=?1 ORDER BY dist LIMIT ?3",
        )
        .unwrap();
        assert!(n.contains("(e.vec <=> (($2)::text::vector)) AS dist"), "{n}");
    }

    #[test]
    fn a_stale_statement_error_names_both_remedies() {
        let e = PgDb::explain_stale(AreevError::Storage(
            "postgres error 26000: prepared statement \"s7\" does not exist".into(),
        ));
        let msg = e.to_string();
        assert!(msg.contains("26000") && msg.contains("max_prepared_statements"), "{msg}");
        assert!(msg.contains("session mode"), "{msg}");
        // Anything else passes through untouched.
        let other = PgDb::explain_stale(AreevError::Storage("postgres error 42P01: relation".into()));
        assert_eq!(other.to_string(), "STO-E001: storage error: postgres error 42P01: relation");
    }

    #[test]
    fn qualifies_every_table_reference_and_nothing_else() {
        let q = |sql: &str| qualify_tables(sql, "s1");
        // Aliases, joins, subqueries, and INSERT's column list.
        assert_eq!(
            q("SELECT g.hash FROM embeddings e JOIN grains g ON g.seq = e.seq WHERE g.ns = $1"),
            "SELECT g.hash FROM \"s1\".embeddings e JOIN \"s1\".grains g ON g.seq = e.seq WHERE g.ns = $1"
        );
        assert_eq!(
            q("INSERT INTO terms(term, term_hash) VALUES ($1, $2) ON CONFLICT (term_hash) DO NOTHING RETURNING id"),
            "INSERT INTO \"s1\".terms(term, term_hash) VALUES ($1, $2) ON CONFLICT (term_hash) DO NOTHING RETURNING id"
        );
        assert_eq!(
            q("DELETE FROM run_idx WHERE seq=$1 AND NOT EXISTS (SELECT 1 FROM triples WHERE s=$1)"),
            "DELETE FROM \"s1\".run_idx WHERE seq=$1 AND NOT EXISTS (SELECT 1 FROM \"s1\".triples WHERE s=$1)"
        );
        assert_eq!(q("UPDATE terms SET term = $2 WHERE id = $1"), "UPDATE \"s1\".terms SET term = $2 WHERE id = $1");
        assert_eq!(
            q("ALTER TABLE telem_recall_log ADD COLUMN run_id TEXT"),
            "ALTER TABLE \"s1\".telem_recall_log ADD COLUMN run_id TEXT"
        );
        assert_eq!(q("DROP TABLE IF EXISTS meta"), "DROP TABLE IF EXISTS \"s1\".meta");
    }

    #[test]
    fn qualifier_leaves_data_catalogs_and_target_references_alone() {
        let q = |sql: &str| qualify_tables(sql, "s1");
        // A table name inside a string literal is data.
        assert_eq!(q("SELECT k FROM meta WHERE v = 'FROM grains'"), "SELECT k FROM \"s1\".meta WHERE v = 'FROM grains'");
        // Catalog tables resolve globally and are not ours to move.
        let cat = "SELECT 1 FROM pg_class c JOIN pg_namespace n ON c.relnamespace = n.oid";
        assert_eq!(q(cat), cat);
        let info = "SELECT 1 FROM information_schema.columns WHERE table_name = 'terms'";
        assert_eq!(q(info), info);
        // The upsert's target reference and FOR UPDATE / DO UPDATE are not
        // table positions.
        assert_eq!(
            q("INSERT INTO counters(name, v) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)"),
            "INSERT INTO \"s1\".counters(name, v) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)"
        );
        assert_eq!(q("SELECT seq FROM grains WHERE hash=$1 FOR UPDATE"), "SELECT seq FROM \"s1\".grains WHERE hash=$1 FOR UPDATE");
        // Idempotent: an already-qualified reference is left as it is, and a
        // column that happens to share a table's name is not a table.
        let done = "SELECT 1 FROM \"s1\".grains g WHERE g.meta = 1";
        assert_eq!(q(done), done);
        assert_eq!(q(&q("SELECT 1 FROM heads")), "SELECT 1 FROM \"s1\".heads");
    }

    #[test]
    fn every_translated_special_qualifies_cleanly() {
        // The integrity probe is emitted by translate itself; it must name
        // its tables in positions the qualifier recognises, or it would read
        // another schema's index rows under a pooler.
        let out = qualify_tables(&translate("PRAGMA integrity_check").unwrap(), "s1");
        for t in ["heads", "grains", "entity_latest", "triples"] {
            assert!(out.contains(&format!("\"s1\".{t}")), "{t} unqualified in: {out}");
        }
        assert!(!out.contains(" heads h") || out.contains("\"s1\".heads h"), "{out}");
    }

    #[test]
    fn every_pg_schema_statement_is_plain_ddl() {
        for sql in PG_SCHEMA {
            // The DDL is written natively for Pg; translate must pass it
            // through untouched (no placeholders, no divergent constructs).
            assert_eq!(&translate(sql).unwrap(), sql);
        }
    }

    #[test]
    fn splits_schema_urls() {
        let (u, s) =
            split_schema_url("postgres://u:p@h:5432/db?schema=org_7").unwrap();
        assert_eq!(u, "postgres://u:p@h:5432/db");
        assert_eq!(s, "org_7");
        let (u, s) = split_schema_url(
            "postgres://u:p@h/db?sslmode=disable&schema=m1&connect_timeout=5",
        )
        .unwrap();
        assert_eq!(u, "postgres://u:p@h/db?sslmode=disable&connect_timeout=5");
        assert_eq!(s, "m1");
        assert!(split_schema_url("postgres://u:p@h/db").is_err());
        assert!(split_schema_url("postgres://u:p@h/db?schema=").is_err());
    }

    #[test]
    fn schema_name_validation() {
        for bad in ["", "has space", "semi;colon", "quote\"", "9leading", &"x".repeat(64)] {
            assert!(
                PgDb::open("postgres://ignored", bad, &[], false, None).is_err(),
                "{bad:?} must be rejected"
            );
        }
    }
}

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

    /// The digest the current [`PG_SCHEMA`] + [`PG_SEED`] hash to.
    ///
    /// If this test fails you changed the schema. That is fine — but the
    /// change is only half done: **bump [`PG_SCHEMA_VERSION`] and then paste
    /// the new digest here, in the same commit.**
    ///
    /// Why a test and not a code review: the stamp is what tells the next open
    /// "this schema is already the right shape, skip the bootstrap". Add a
    /// statement without bumping the version and every schema already stamped
    /// skips it *forever*, with no error at any layer — which is precisely the
    /// #160 shape (a dictionary keyed on a column that was never added, so
    /// every recall answers empty). A stale stamp is worse than a missing one.
    const EXPECTED_DIGEST: &str =
        "3c4625610533e1882f956dc6f40ae85bfc4515f3ead701687bf05173b5b36034";

    fn digest() -> String {
        use sha2::{Digest, Sha256};
        let mut h = Sha256::new();
        for sql in PG_SCHEMA.iter().chain(PG_SEED.iter()) {
            h.update(sql.as_bytes());
            // Length-delimited, so splitting one statement into two (or
            // merging two into one) cannot hash to the same value.
            h.update([0u8]);
        }
        hex::encode(h.finalize())
    }

    #[test]
    fn pg_schema_version_tracks_the_schema() {
        assert_eq!(
            digest(),
            EXPECTED_DIGEST,
            "PG_SCHEMA/PG_SEED changed. Bump PG_SCHEMA_VERSION (currently {PG_SCHEMA_VERSION:?}) \
             and update EXPECTED_DIGEST to the value above — a schema stamped with an unbumped \
             version silently skips the migration it needs"
        );
    }

    #[test]
    fn pool_size_reads_the_dsn_then_the_environment() {
        assert_eq!(pool_size("postgres://u@h/db?schema=x").unwrap(), DEFAULT_POOL_SIZE);
        assert_eq!(pool_size("postgres://u@h/db?pool=3&schema=x").unwrap(), 3);
        assert!(pool_size("postgres://u@h/db?pool=0").is_err());
        assert!(pool_size("postgres://u@h/db?pool=many").is_err());
        std::env::set_var(POOL_ENV, "5");
        assert_eq!(pool_size("postgres://u@h/db").unwrap(), 5);
        assert_eq!(pool_size("postgres://u@h/db?pool=2").unwrap(), 2, "the DSN wins");
        std::env::set_var(POOL_ENV, "x");
        assert!(pool_size("postgres://u@h/db").is_err());
        std::env::remove_var(POOL_ENV);
        assert_eq!(strip_pool("postgres://u@h/db?pool=3&schema=x"), "postgres://u@h/db?schema=x");
        // The driver never sees it, whatever else rides along.
        let req = crate::pgtls::SslRequest::split("postgres://u@h/db?pool=3&sslmode=disable").unwrap();
        assert_eq!(req.dsn, "postgres://u@h/db");
    }

    #[test]
    fn provision_mode_reads_the_dsn() {
        assert_eq!(provision_mode("postgres://h/db").unwrap(), ProvisionMode::Auto);
        assert_eq!(
            provision_mode("postgres://h/db?sslmode=require").unwrap(),
            ProvisionMode::Auto
        );
        assert_eq!(provision_mode("postgres://h/db?provision=auto").unwrap(), ProvisionMode::Auto);
        assert_eq!(
            provision_mode("postgres://h/db?schema=s&provision=never").unwrap(),
            ProvisionMode::Never
        );
        let e = provision_mode("postgres://h/db?provision=maybe").unwrap_err();
        assert!(e.to_string().contains("auto|never"), "{e}");
    }

    #[test]
    fn strip_provision_leaves_every_other_parameter() {
        assert_eq!(
            strip_provision("postgres://h/db?schema=s&provision=never&sslmode=require"),
            "postgres://h/db?schema=s&sslmode=require"
        );
        assert_eq!(strip_provision("postgres://h/db?provision=never"), "postgres://h/db");
        assert_eq!(strip_provision("postgres://h/db"), "postgres://h/db");
    }
}