areev-store 1.10.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.
//!
//! A pool that has gone quiet gives its connections BACK (#229): one reaper
//! thread per process closes a connection idle past `?pool_idle_secs=`
//! (default [`DEFAULT_POOL_IDLE_SECS`]) and then evicts the pool itself once
//! nothing holds it. Without that, "N memories hold at most P connections"
//! bounds nothing on a host that gives every tenant its own ROLE — a
//! different role is a different pool, so a long-lived worker accumulated one
//! pool, and one connection, per tenant it had ever touched.
//!
//! A memory may keep its ENGINE METADATA in a second schema (#353, the
//! **paired layout**): `?meta_schema=<name>` on the DSN sends `meta`,
//! `counters`, `ns_reg` and the telemetry sidecar's `telem_*` tables there,
//! while the grains, every index over them, the dictionary and the CAS blobs
//! stay in the memory schema. [`PgLayout`] is the one routing table —
//! `qualify_tables` consults it for every statement, DDL included, so nothing
//! else in the backend knows which layout it is running under, and a write
//! that touches both schemas is still one transaction on one connection.
//! Without the parameter the layout is the historical single schema, byte
//! for byte.

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))",
    // `ns` (#307) is nullable so an older-build writer against this schema
    // keeps working, and so a tombstone imported from a peer that recorded
    // none is representable rather than mis-attributed.
    "CREATE TABLE IF NOT EXISTS oplog(op_seq bigint PRIMARY KEY, hlc bigint, op bigint, hash bytea, ns bigint)",
    "ALTER TABLE oplog ADD COLUMN IF NOT EXISTS ns bigint",
    // Backfill LIVE rows from the grains they name. Historical tombstones
    // stay NULL: the grain is gone, so the namespace is genuinely not
    // recoverable, and inventing one would be worse than admitting it.
    "UPDATE oplog o SET ns = g.ns FROM grains g WHERE g.hash = o.hash AND o.ns IS NULL",
    "CREATE INDEX IF NOT EXISTS idx_oplog_ns ON oplog(ns, op_seq)",
    "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 = "2";

/// The oldest stamped schema version whose writers keep working during AND
/// after this build's bootstrap — the machine-readable half of
/// `docs/deployment-profile.md`'s "a schema migration is not rolling-deploy
/// safe" (#308).
///
/// Version 2 adds a NULLABLE `oplog.ns` column and an index. An old-build
/// writer inserts without it and keeps working; a new-build reader treats
/// NULL as unattributable. So version 1 writers survive the migration, and
/// this is a rolling-safe change.
///
/// `pg_schema_version_tracks_the_schema` hashes the schema arrays, so a
/// version bump forces this constant to be reconsidered in the same commit.
pub(crate) const PG_ROLLING_SAFE_FROM: &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;
/// `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")
}

/// Our name for the paired layout's parameter (#353), in the places that
/// must agree: the reader, the stripper, the pool key and the TLS splitter.
pub const META_SCHEMA_PARAM: &str = "meta_schema";

/// Read `?meta_schema=` off a DSN (#353): the schema the memory's ENGINE
/// METADATA lives in, or `None` for the single-schema layout.
///
/// Carried on the DSN rather than in `AreevOptions` for the reason
/// `provision` is: every host that already passes a DSN string — the CLI,
/// the console, both bindings, the bench — reaches it without a new
/// parameter, and a host that never heard of it keeps working. It is ours,
/// not the driver's: the TLS splitter (`pgtls::SslRequest::split`) and the
/// pool key both remove it. An empty value is a refusal rather than "single layout",
/// because a DSN that SAYS `meta_schema=` and gets the other layout is the
/// silent misconfiguration this parameter must never produce.
pub fn meta_schema(url: &str) -> Result<Option<String>> {
    let Some((_, query)) = url.split_once('?') else {
        return Ok(None);
    };
    let mut found = None;
    for pair in query.split('&') {
        if let Some((k, v)) = pair.split_once('=') {
            if k == META_SCHEMA_PARAM {
                if v.is_empty() {
                    return Err(AreevError::Validation(format!(
                        "postgres URL: {META_SCHEMA_PARAM}= names no schema — omit the \
                         parameter for the single-schema layout"
                    )));
                }
                found = Some(v.to_string());
            }
        }
    }
    Ok(found)
}

/// Strip `?meta_schema=` from a DSN, for a caller that names the metadata
/// schema another way (`areev provision --meta-schema`).
pub fn strip_meta_schema(url: &str) -> String {
    strip_param(url, META_SCHEMA_PARAM)
}

/// The tables that are ENGINE METADATA rather than memory (#353): in the
/// paired layout these live in the metadata schema, everything else in
/// [`PG_TABLES`] lives in the memory schema. The classification, and why:
///
/// - `meta` — every file-carried declaration and registry row: the
///   `pg_schema`/`link_index`/`ns_registry` stamps, `text_index` and
///   `entity_relations`, embedding provenance, `min_reader_version`, saved
///   queries and templates (`qry:`/`tpl:`), retention policies and floors,
///   anonymization policies and the sealed mapping vault (`anon:`/`vault:`),
///   legal holds (`hold:`) and trigger leases/cursors (`trg:`). The table
///   moves as ONE unit: a key-by-key split would put the choke point every
///   policy read shares (`meta_get`) on two schemas, and the one arguable
///   family — the vault, whose rows are keyed by memory content — is still a
///   mapping the engine keeps about the memory, not a grain of it.
/// - `counters` — write-transaction id allocation.
/// - `ns_reg` — the namespace inventory prefix scoping resolves against.
/// - `telem_*` — the telemetry sidecar. On the embedded backend it is a
///   separate FILE beside the memory; the separate schema is the same rule.
///
/// What stays with the memory: `grains` and every index derived from them
/// (`triples`/`osp`/`entity_latest`/`heads`/`thread_idx`/`prov_idx`/`run_idx`/
/// `corpus_idx`, the BM25 `fts_*` tables, `embeddings`), the `oplog`, the
/// `terms` dictionary (every subject/relation/object string IS memory
/// content) and the CAS `blobs`. That set is exactly what `pg_dump -n
/// <memory schema>` must carry for the grains to be readable, and what a
/// `DROP SCHEMA` erasure must destroy for them to be gone.
pub const META_TABLES: &[&str] = &[
    "counters",
    "meta",
    "ns_reg",
    "telem_budget_stat",
    "telem_grain_access",
    "telem_meta",
    "telem_query_stat",
    "telem_recall_log",
];

/// Where a memory's tables live (#353): one schema for everything (the
/// historical layout), or a memory schema plus a metadata schema for the
/// [`META_TABLES`]. The one routing table in the backend — `qualify_tables`
/// asks it per table reference, the bootstrap asks it per stamp, and nothing
/// else needs to know which layout is in force.
///
/// Both names are validated to the same identifier rule and quoted on every
/// use; they come from the DSN — host configuration — and from nowhere else,
/// so no message, grain or model output can select a schema.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PgLayout {
    schema: String,
    /// Equal to `schema` in the single-schema layout.
    meta_schema: String,
}

impl PgLayout {
    /// The single-schema layout.
    pub fn single(schema: &str) -> Result<Self> {
        Self::validate_name(schema)?;
        Ok(Self { schema: schema.to_string(), meta_schema: schema.to_string() })
    }

    /// The paired layout. The two names must differ: a pair that names one
    /// schema twice is the single layout spelled confusingly, and refusing
    /// it keeps "paired" meaning "physically separate".
    pub fn paired(schema: &str, meta_schema: &str) -> Result<Self> {
        Self::validate_name(schema)?;
        Self::validate_name(meta_schema)?;
        if schema == meta_schema {
            return Err(AreevError::Validation(format!(
                "postgres URL: {META_SCHEMA_PARAM}={meta_schema:?} names the memory schema \
                 itself — the metadata schema must be a different schema; omit the parameter \
                 for the single-schema layout"
            )));
        }
        Ok(Self { schema: schema.to_string(), meta_schema: meta_schema.to_string() })
    }

    /// The layout a DSN asks for: paired when it carries `?meta_schema=`,
    /// single otherwise. `schema` is the memory schema, already split off the
    /// URL by [`split_schema_url`] (or named by a `--schema` flag).
    pub fn from_url(url: &str, schema: &str) -> Result<Self> {
        match meta_schema(url)? {
            Some(m) => Self::paired(schema, &m),
            None => Self::single(schema),
        }
    }

    /// The memory schema.
    pub fn schema(&self) -> &str {
        &self.schema
    }

    /// The metadata schema, or `None` in the single-schema layout.
    pub fn meta_schema(&self) -> Option<&str> {
        self.is_paired().then_some(self.meta_schema.as_str())
    }

    pub fn is_paired(&self) -> bool {
        self.schema != self.meta_schema
    }

    /// Which schema `table` lives in.
    pub(crate) fn schema_for(&self, table: &str) -> &str {
        if META_TABLES.contains(&table) {
            &self.meta_schema
        } else {
            &self.schema
        }
    }

    /// `"<schema>".<table>`, routed.
    pub(crate) fn qualified(&self, table: &str) -> String {
        format!("\"{}\".{table}", self.schema_for(table))
    }

    /// Every schema the layout occupies: the memory schema first.
    pub(crate) fn schemas(&self) -> Vec<&str> {
        match self.meta_schema() {
            Some(m) => vec![&self.schema, m],
            None => vec![&self.schema],
        }
    }

    /// How the layout reads in an error message.
    fn describe(&self) -> String {
        match self.meta_schema() {
            Some(m) => format!("{:?} (metadata schema {m:?})", self.schema),
            None => format!("{:?}", self.schema),
        }
    }

    fn validate_name(schema: &str) -> Result<()> {
        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:?}"
            )));
        }
        Ok(())
    }
}

// ---- 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")
}

/// How long a pooled connection may sit idle before the reaper closes it,
/// when neither `?pool_idle_secs=` nor `$AREEV_PG_POOL_IDLE_SECS` says
/// otherwise. Five minutes: long enough that a process doing steady work
/// re-dials approximately never, short enough that a host handing every
/// tenant its own role gets its connections back on a human timescale
/// rather than at the next restart (#229).
pub const DEFAULT_POOL_IDLE_SECS: u64 = 300;

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

/// Our name for the parameter, in the two places that must agree: the reader
/// below and the pool key.
const POOL_IDLE_PARAM: &str = "pool_idle_secs";

/// How long `url`'s pool keeps an idle connection: `?pool_idle_secs=N` on the
/// DSN, else `$AREEV_PG_POOL_IDLE_SECS`, else [`DEFAULT_POOL_IDLE_SECS`]. The
/// DSN wins, as everywhere.
///
/// `0` is `None` — never reap, hold every connection dialled until the
/// process ends, which is what every build before #229 did.
pub fn pool_idle_ttl(url: &str) -> Result<Option<std::time::Duration>> {
    if let Some((_, query)) = url.split_once('?') {
        let mut from_dsn = None;
        for pair in query.split('&') {
            if let Some((k, v)) = pair.split_once('=') {
                if k == POOL_IDLE_PARAM {
                    from_dsn = Some(parse_idle_secs(&format!("postgres URL: {POOL_IDLE_PARAM}"), v)?);
                }
            }
        }
        if let Some(t) = from_dsn {
            return Ok(t);
        }
    }
    match std::env::var(POOL_IDLE_ENV) {
        Ok(v) if !v.trim().is_empty() => parse_idle_secs(&format!("${POOL_IDLE_ENV}"), v.trim()),
        _ => Ok(Some(std::time::Duration::from_secs(DEFAULT_POOL_IDLE_SECS))),
    }
}

fn parse_idle_secs(what: &str, v: &str) -> Result<Option<std::time::Duration>> {
    match v.parse::<u64>() {
        Ok(0) => Ok(None),
        Ok(n) => Ok(Some(std::time::Duration::from_secs(n))),
        Err(_) => Err(AreevError::Validation(format!(
            "{what}={v:?} must be a whole number of seconds an idle connection is kept \
             (0 to keep them for the life of the process)"
        ))),
    }
}

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

/// How the two pool knobs read back in a warning — `None` is a setting, not
/// an absence.
fn describe_ttl(ttl: Option<std::time::Duration>) -> String {
    match ttl {
        Some(d) => format!("{}s", d.as_secs()),
        None => "never".to_string(),
    }
}

/// 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 (each stamped with when it went
/// idle, which is what the reaper reads). A handle borrows from it per
/// statement, or per transaction; the pool outlives every handle, but no
/// longer outlives its own usefulness (#229).
pub(crate) struct PgPool {
    rt: tokio::runtime::Runtime,
    url: String,
    cap: usize,
    /// How long an idle connection is kept before the reaper closes it.
    /// `None` = for the life of the process.
    idle_ttl: Option<std::time::Duration>,
    sem: std::sync::Arc<tokio::sync::Semaphore>,
    idle: std::sync::Mutex<Vec<(std::time::Instant, 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() {
                    // Stamped as it goes back, and taken from the END on the
                    // way out: the hot connection stays hot and the cold ones
                    // age out, which is what makes an idle TTL bite.
                    idle.push((std::time::Instant::now(), c));
                }
            }
        }
    }
}

/// The live pools, plus whether the reaper thread is running. One lock over
/// both, deliberately: the reaper decides to stop in the same critical
/// section an opener uses to insert a pool, so a pool can never be left with
/// nobody to reap it.
#[derive(Default)]
struct Registry {
    pools: HashMap<String, std::sync::Arc<PgPool>>,
    reaping: bool,
}

static POOLS: std::sync::OnceLock<std::sync::Mutex<Registry>> = std::sync::OnceLock::new();

fn registry() -> std::sync::MutexGuard<'static, Registry> {
    POOLS.get_or_init(Default::default).lock().unwrap_or_else(|p| p.into_inner())
}

/// What keys a pool: the DSN with the store's own parameters stripped. One
/// server, role and TLS setting, however many memories and whatever each of
/// them asked for in `?pool=` / `?pool_idle_secs=`.
fn pool_key(url: &str) -> String {
    ["pool", POOL_IDLE_PARAM, "provision", "schema", META_SCHEMA_PARAM]
        .iter()
        .fold(url.to_string(), |acc, p| strip_param(&acc, p))
}

/// Does this process still hold a pool for `url`? The reaper's effect on the
/// pool itself — the connection count is the server's side of the same story
/// — for the test that pins #229.
#[cfg(feature = "conformance")]
pub fn pool_is_registered(url: &str) -> bool {
    registry().pools.contains_key(&pool_key(url))
}

/// How often the reaper looks: half the shortest TTL in the process, bounded
/// so a long TTL costs nothing and a short one is still observed promptly.
/// A connection therefore lives at most `ttl + MAX_REAP_TICK`.
const MIN_REAP_TICK: std::time::Duration = std::time::Duration::from_millis(250);
const MAX_REAP_TICK: std::time::Duration = std::time::Duration::from_secs(30);

/// Start the reaper if some pool now wants one. Called with the registry
/// locked, so the flag it sets cannot race the sweep that clears it.
fn start_reaper(reg: &mut Registry) {
    if reg.reaping || !reg.pools.values().any(|p| p.idle_ttl.is_some()) {
        return;
    }
    // ONE thread for every pool in the process — the accumulation this fixes
    // would otherwise be re-created in thread form — and a plain OS thread
    // rather than a task on a pool's own runtime, because it DROPS pools and
    // dropping a `Runtime` from inside that runtime panics.
    let spawned = std::thread::Builder::new().name("areev-pg-reap".into()).spawn(|| {
        while let Some(tick) = sweep() {
            std::thread::sleep(tick);
        }
    });
    // Out of threads: this open still works, unreaped, and the next one tries
    // again. Never a reason to fail an open.
    reg.reaping = spawned.is_ok();
}

/// One pass: close what has been idle too long, then evict a pool that holds
/// nothing and that nothing holds. Returns how long to sleep, or `None` when
/// no pool wants reaping any more — the thread exits and the next open
/// starts a new one.
fn sweep() -> Option<std::time::Duration> {
    let mut evicted: Vec<std::sync::Arc<PgPool>> = Vec::new();
    let tick = {
        let mut reg = registry();
        let now = std::time::Instant::now();
        let mut keep = HashMap::with_capacity(reg.pools.len());
        for (key, p) in std::mem::take(&mut reg.pools) {
            let left = p.reap_idle(now);
            // Nothing left to hand out and nobody holding it: the pool goes
            // too, and its runtime's worker threads with it. Sound under this
            // lock and only under it — every other reference (a handle, a
            // checkout) is an `Arc` clone that can only be made from here, so
            // a count of one means this map is the last owner.
            if left == 0 && p.idle_ttl.is_some() && std::sync::Arc::strong_count(&p) == 1 {
                evicted.push(p);
            } else {
                keep.insert(key, p);
            }
        }
        reg.pools = keep;
        match reg.pools.values().filter_map(|p| p.idle_ttl).min() {
            Some(ttl) => Some((ttl / 2).clamp(MIN_REAP_TICK, MAX_REAP_TICK)),
            None => {
                reg.reaping = false;
                None
            }
        }
    };
    // Off the lock: dropping a pool drops its runtime, which waits for its
    // worker threads — not something to do while every opener in the process
    // is queued behind the registry.
    drop(evicted);
    tick
}

impl PgPool {
    /// The pool for `url`, made on first use. Keyed by the DSN with the
    /// store's own parameters (`schema`, `provision`, `pool`,
    /// `pool_idle_secs`) 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 and sets its idle TTL; a later open asking
    /// for another size or TTL gets the warning back.
    pub(crate) fn for_url(url: &str) -> Result<(std::sync::Arc<PgPool>, Option<String>)> {
        let want = pool_size(url)?;
        let want_ttl = pool_idle_ttl(url)?;
        let key = pool_key(url);
        let mut reg = registry();
        if let Some(p) = reg.pools.get(&key) {
            let mut warn: Vec<String> = Vec::new();
            if p.cap != want {
                warn.push(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
                ));
            }
            if p.idle_ttl != want_ttl {
                warn.push(format!(
                    "postgres pool for this DSN reaps idle connections after {} by an earlier \
                     open in this process; ?{POOL_IDLE_PARAM}={} (or ${POOL_IDLE_ENV}) does not \
                     change it",
                    describe_ttl(p.idle_ttl),
                    describe_ttl(want_ttl)
                ));
            }
            let p = std::sync::Arc::clone(p);
            return Ok((p, (!warn.is_empty()).then(|| warn.join("; "))));
        }
        // 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,
            idle_ttl: want_ttl,
            sem: std::sync::Arc::new(tokio::sync::Semaphore::new(want)),
            idle: std::sync::Mutex::new(Vec::new()),
        });
        reg.pools.insert(key, std::sync::Arc::clone(&p));
        start_reaper(&mut reg);
        Ok((p, None))
    }

    /// Close every connection idle longer than this pool's TTL — and any the
    /// server has already hung up on, whatever the TTL. Returns how many are
    /// left, which is what tells the caller whether the pool itself is now
    /// holding nothing.
    fn reap_idle(&self, now: std::time::Instant) -> usize {
        let mut idle = self.idle.lock().unwrap_or_else(|p| p.into_inner());
        let Some(ttl) = self.idle_ttl else {
            return idle.len();
        };
        let mut kept = Vec::with_capacity(idle.len());
        let mut doomed = Vec::new();
        for (since, c) in idle.drain(..) {
            if now.duration_since(since) < ttl && !c.client.is_closed() {
                kept.push((since, c));
            } else {
                doomed.push(c);
            }
        }
        *idle = kept;
        let left = idle.len();
        drop(idle);
        // The sockets close here, off the pool's lock: dropping the `Client`
        // is what ends its connection task.
        drop(doomed);
        left
    }

    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.map(|(_, c)| c),
            }
        };
        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>,
    /// Which schema each table resolves to (#353). Every statement is routed
    /// through it, so the handle itself never asks which layout it is under.
    layout: PgLayout,
    /// 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.
    ///
    /// The layout comes off `url` (#353): `?meta_schema=` makes this a paired
    /// open, and every path below — the stamp probe, the read-only
    /// verification, the `provision=never` refusal and the bootstrap — routes
    /// through the resulting [`PgLayout`]. An open whose DSN disagrees with
    /// the layout the memory already has is refused (`STO-E011`) before any
    /// DDL, in both directions; see [`Self::check_layout`].
    pub(crate) fn open(
        url: &str,
        schema: &str,
        bootstrap: &[&str],
        read_only: bool,
        stamp: Option<PgStamp>,
    ) -> Result<Self> {
        let layout = PgLayout::from_url(url, schema)?;
        let (pool, pool_warning) = PgPool::for_url(url)?;
        let mut co = pool.checkout()?;
        let opened = Self::open_on(&pool, co.client(), url, &layout, 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,
            layout,
            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,
        layout: &PgLayout,
        bootstrap: &[&str],
        read_only: bool,
        stamp: Option<PgStamp>,
    ) -> Result<bool> {
        if read_only {
            pool.block_on(Self::verify_read_only(client, layout))?;
            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, layout, st))?.as_deref()
                == Some(st.version);
        }
        if skipped {
            return Ok(true);
        }
        // Off the fast path, and before either refusal below or any DDL: a
        // DSN whose layout disagrees with the memory's is refused outright
        // (#353). SELECT-only, two `to_regclass` probes, first open only.
        pool.block_on(Self::check_layout(client, layout))?;
        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, layout, stamp));
        }
        match Self::bootstrap_once(pool, client, layout, 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, layout, st))?.as_deref()
                            == Some(st.version)
                    }
                    None => false,
                };
                if current {
                    Ok(true)
                } else {
                    Self::bootstrap_once(pool, client, layout, 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.
    ///
    /// Every statement goes through `qualify_tables` (#353), so a paired
    /// layout's `meta`/`counters`/`ns_reg`/`telem_*` DDL lands in the
    /// metadata schema and everything else in the memory schema — one
    /// transaction, both schemas. The `search_path` is still set, for the
    /// one thing qualification cannot reach: the `DO $$ … $$` migration
    /// blocks in `PG_SEED` resolve `terms` (a memory table) through
    /// `current_schema()` and `pg_get_serial_sequence`.
    fn bootstrap_once(
        pool: &PgPool,
        client: &tokio_postgres::Client,
        layout: &PgLayout,
        bootstrap: &[&str],
        stamp: Option<PgStamp>,
    ) -> Result<()> {
        let schema = layout.schema();
        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 {
                // Keyed on the MEMORY schema: a pair is one memory, and two
                // openers of one pair must serialise on one lock.
                client
                    .query_one(
                        "SELECT pg_advisory_xact_lock(hashtext('areev_bootstrap'), hashtext($1))",
                        &[&schema],
                    )
                    .await
                    .map_err(pg_err)?;
                for s in layout.schemas() {
                    client
                        .batch_execute(&format!("CREATE SCHEMA IF NOT EXISTS \"{s}\""))
                        .await
                        .map_err(pg_err)?;
                }
                // LOCAL: the DO-block migrations below need the memory
                // 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 {
                    let routed = qualify_tables(sql, layout);
                    client.batch_execute(&routed).await.map_err(|e| {
                        AreevError::Storage(format!(
                            "bootstrap failed: {} — in: {routed}",
                            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 {}(k, v) VALUES ($1, $2) \
                                 ON CONFLICT (k) DO UPDATE SET v = EXCLUDED.v",
                                layout.qualified(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,
        layout: &PgLayout,
        stamp: PgStamp,
    ) -> Result<Option<String>> {
        let table = layout.qualified(stamp.table);
        if !Self::relation_present(client, &table).await? {
            return Ok(None);
        }
        let row = client
            .query_opt(&format!("SELECT v FROM {table} WHERE k = $1"), &[&stamp.key])
            .await
            .map_err(pg_err)?;
        Ok(row.and_then(|r| r.get::<_, Option<String>>(0)))
    }

    /// `to_regclass` on an already-qualified relation name: NULL — never an
    /// error — when the schema or the table is absent. Needs no privilege
    /// beyond `USAGE` on the schema.
    async fn relation_present(client: &tokio_postgres::Client, qualified: &str) -> Result<bool> {
        Ok(client
            .query_one("SELECT to_regclass($1) IS NOT NULL", &[&qualified])
            .await
            .map_err(pg_err)?
            .get::<_, bool>(0))
    }

    /// Does every schema the layout occupies exist?
    async fn schemas_exist(client: &tokio_postgres::Client, layout: &PgLayout) -> Result<bool> {
        for s in layout.schemas() {
            let exists = client
                .query_one(
                    "SELECT EXISTS(SELECT 1 FROM information_schema.schemata WHERE schema_name = $1)",
                    &[&s],
                )
                .await
                .map_err(pg_err)?
                .get::<_, bool>(0);
            if !exists {
                return Ok(false);
            }
        }
        Ok(true)
    }

    /// Refuse an open whose DSN disagrees with the layout the memory already
    /// has (#353) — `STO-E011`, in BOTH directions, SELECT-only, before any
    /// lock or DDL:
    ///
    /// - a paired DSN over a memory whose own schema carries `meta` (the
    ///   single layout) would bootstrap a second, empty `meta`/`counters`/
    ///   `ns_reg` in the metadata schema and run against those — every hold,
    ///   policy and saved query silently gone, and two writers allocating ids
    ///   from two counter rows;
    /// - a single DSN over a memory whose schema holds `grains` but no `meta`
    ///   — which is what a paired memory looks like from its memory schema,
    ///   and nothing else does, since `meta` is the first table the bootstrap
    ///   creates and the bootstrap is one transaction — would do the same in
    ///   the other direction.
    ///
    /// Only the slow path pays for it: a stamped-current memory has already
    /// proven its layout by where the stamp was found. A layout change is an
    /// operator's explicit migration (`docs/deployment-profile.md`, "Paired
    /// layout"), never something an open does by itself.
    async fn check_layout(client: &tokio_postgres::Client, layout: &PgLayout) -> Result<()> {
        let schema = layout.schema();
        let in_schema_meta = Self::relation_present(client, &format!("\"{schema}\".meta")).await?;
        match layout.meta_schema() {
            Some(m) if in_schema_meta => Err(AreevError::LayoutMismatch(format!(
                "postgres schema {schema:?} carries its engine metadata in-schema (the \
                 single-schema layout), but this DSN names {META_SCHEMA_PARAM}={m:?} — \
                 opening it paired would split counters, the namespace registry, legal holds, \
                 retention policies and saved queries across two schemas. Either drop \
                 {META_SCHEMA_PARAM} from the DSN, or move the metadata tables explicitly \
                 (`ALTER TABLE … SET SCHEMA`, see docs/deployment-profile.md \"Paired \
                 layout\") and retry"
            ))),
            None if !in_schema_meta
                && Self::relation_present(client, &format!("\"{schema}\".grains")).await? =>
            {
                Err(AreevError::LayoutMismatch(format!(
                    "postgres schema {schema:?} holds memory tables but no meta table — the \
                     shape of a paired-layout memory whose engine metadata lives in a separate \
                     schema — and this DSN names no {META_SCHEMA_PARAM}. Add \
                     ?{META_SCHEMA_PARAM}=<its metadata schema> to the DSN; a single-schema \
                     open would bootstrap a second, empty set of engine metadata beside the \
                     real one"
                )))
            }
            _ => Ok(()),
        }
    }

    /// 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,
        layout: &PgLayout,
        stamp: Option<PgStamp>,
    ) -> AreevError {
        let version = stamp.map(|s| s.version).unwrap_or(PG_SCHEMA_VERSION);
        let exists = pool.block_on(Self::schemas_exist(client, layout)).unwrap_or(false);
        let what = layout.describe();
        let schema = layout.schema();
        // The DSN handed to `areev provision` must carry the same layout, so
        // the pair — not just the memory schema — gets provisioned.
        let hint = match layout.meta_schema() {
            Some(m) => format!("`areev provision --db <dsn> --schema {schema} --meta-schema {m}`"),
            None => format!("`areev provision --db <dsn> --schema {schema}`"),
        };
        AreevError::SchemaNotProvisioned(if exists {
            format!(
                "postgres schema {what} 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 {hint} (or your migration job) with a role that owns the schema, then retry"
            )
        } else {
            format!(
                "postgres schema {what} does not exist, and this DSN says provision=never \
                 — so no CREATE SCHEMA was attempted. Either a schema name is wrong, or the \
                 memory has never been created: run {hint} 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,
        layout: &PgLayout,
    ) -> Result<()> {
        let schema = layout.schema();
        let what = layout.describe();
        // Layout first (#353): a paired DSN over a single-schema memory has
        // no metadata schema to find, and "does not exist" would send the
        // operator to provision one — the wrong action. A read-only open of
        // the wrong layout would otherwise not fail at all: it would read an
        // empty `meta` and answer as if no hold, policy or query existed.
        Self::check_layout(client, layout).await?;
        if !Self::schemas_exist(client, layout).await? {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {what} does not exist — a read-only open never runs \
                 bootstrap DDL (CREATE SCHEMA included), so it cannot create it. Either a \
                 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<String> = Vec::new();
        for table in CORE_TABLES {
            let qualified = layout.qualified(table);
            if !Self::relation_present(client, &qualified).await? {
                missing.push(qualified);
            }
        }
        if !missing.is_empty() {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {what} 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.layout);
        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.layout.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.layout.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.layout.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.layout.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. The subset in
/// [`META_TABLES`] routes to the metadata schema under a paired layout.
pub 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.
///
/// The schema each reference gets comes from the `layout` (#353): a paired
/// layout sends the [`META_TABLES`] to the metadata schema and everything
/// else to the memory schema, so one statement may legitimately name both
/// (`INSERT INTO "m".counters … SELECT … FROM "s".grains`). `ON` is a
/// trigger for the sake of `CREATE INDEX … ON <table>` — the bootstrap DDL
/// goes through here too, which is what puts a metadata table's index in the
/// metadata schema; an `ON <alias>.<col>` join condition names no table and
/// is left alone, and `ON <table>.<col>` becomes a three-part name Postgres
/// accepts.
pub(crate) fn qualify_tables(sql: &str, layout: &PgLayout) -> String {
    const TRIGGERS: &[&str] = &["FROM", "JOIN", "INTO", "UPDATE", "TABLE", "EXISTS", "ON"];
    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(layout.schema_for(&word));
                    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
/// (`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
    }
}

/// One stamp's found-vs-wanted state, as `provision --check` reports it.
#[derive(Debug, Clone, serde::Serialize)]
pub struct StampState {
    pub name: String,
    pub found: Option<String>,
    pub wanted: String,
}

impl StampState {
    pub fn is_current(&self) -> bool {
        self.found.as_deref() == Some(self.wanted.as_str())
    }
}

/// What a read-only provisioning probe found (#308).
#[derive(Debug, Clone, serde::Serialize)]
pub struct ProvisionReport {
    pub schema: String,
    /// The metadata schema under a paired layout (#353); `None` for the
    /// single-schema layout.
    pub meta_schema: Option<String>,
    /// Every schema the layout occupies exists — the memory schema, and the
    /// metadata schema too when there is one.
    pub exists: bool,
    pub stamps: Vec<StampState>,
    /// Names of the stamps that are absent or behind — empty means current.
    pub pending: Vec<String>,
    /// `"safe"`, `"drain_writers_first"`, or `"unknown"` (no stamp to
    /// compare against).
    pub rolling_deploy: String,
}

impl ProvisionReport {
    /// Exit-code shape: 0 current, 2 pending or absent. Matches the
    /// `loop list --fail-on` convention.
    pub fn is_current(&self) -> bool {
        self.exists && self.pending.is_empty()
    }
}

/// Read-only probe of a memory schema's migration state (#308).
///
/// SELECTs only: no advisory lock, no DDL, no `meta` write, so it runs under
/// the documented least-privilege read-only role. The engine already knew all
/// of this — `schema_stamp` is two SELECTs — but the only way to ask was to
/// OPEN with `provision=never` and watch for `STO-E008`: an error path that
/// covered one stamp and said "stale" without saying what was pending.
///
/// `telemetry` selects which telemetry stamp is expected; a schema
/// provisioned with telemetry off and checked with the default correctly
/// reports telemetry pending, because for that deployment it is.
///
/// The layout comes off `url` (#353): with `?meta_schema=` the stamps are
/// read where a paired memory keeps them, and `exists` covers both schemas.
pub fn check_provision(
    url: &str,
    schema: &str,
    telemetry: crate::TelemetryMode,
) -> Result<ProvisionReport> {
    let layout = PgLayout::from_url(url, schema)?;
    let (pool, _) = PgPool::for_url(url)?;
    let co = pool.checkout()?;
    let exists = pool.block_on(PgDb::schemas_exist(co.client(), &layout))?;

    let mut stamps = Vec::new();
    let mut wanted: Vec<(&str, &str, &str, &str)> = vec![
        ("pg_schema", "meta", "pg_schema", PG_SCHEMA_VERSION),
        (
            "link_index",
            "meta",
            crate::LINK_INDEX_KEY,
            crate::LINK_INDEX_VERSION,
        ),
        (
            "ns_registry",
            "meta",
            crate::NS_REGISTRY_KEY,
            crate::NS_REGISTRY_VERSION,
        ),
    ];
    if !matches!(telemetry, crate::TelemetryMode::Off) {
        wanted.push((
            "telem_schema",
            "telem_meta",
            "schema_version",
            crate::telemetry::TELEM_SCHEMA_VERSION,
        ));
    }
    for (name, table, key, want) in wanted {
        let found = if exists {
            pool.block_on(PgDb::schema_stamp(
                co.client(),
                &layout,
                PgStamp { table, key, version: want },
            ))?
        } else {
            None
        };
        stamps.push(StampState {
            name: name.to_string(),
            found,
            wanted: want.to_string(),
        });
    }
    let pending: Vec<String> = stamps
        .iter()
        .filter(|s| !s.is_current())
        .map(|s| s.name.clone())
        .collect();
    let store_found = stamps
        .iter()
        .find(|s| s.name == "pg_schema")
        .and_then(|s| s.found.clone());
    let rolling_deploy = match store_found.as_deref() {
        None => "unknown",
        Some(v) if version_lt(v, PG_ROLLING_SAFE_FROM) => "drain_writers_first",
        Some(_) => "safe",
    }
    .to_string();
    Ok(ProvisionReport {
        schema: schema.to_string(),
        meta_schema: layout.meta_schema().map(str::to_string),
        exists,
        stamps,
        pending,
        rolling_deploy,
    })
}

/// Numeric `a < b` over the schema-version strings (plain integers today).
fn version_lt(a: &str, b: &str) -> bool {
    match (a.parse::<u64>(), b.parse::<u64>()) {
        (Ok(x), Ok(y)) => x < y,
        // An unparseable stamp is not evidence of safety.
        _ => a != b,
    }
}

/// 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.
///
/// Under a paired layout (`?meta_schema=` on `url`, #353) the metadata
/// schema is dropped with the memory schema, in one transaction: a pair is
/// one memory, and an erasure that left its holds, counters and registry
/// behind would be neither complete nor safe to re-provision over.
pub fn drop_postgres_schema(url: &str, schema: &str) -> Result<()> {
    drop_postgres_schema_with(url, schema, &DropOptions::default())
}

/// Options for [`drop_postgres_schema_with`].
#[derive(Debug, Clone, Default)]
pub struct DropOptions {
    /// Drop the schema even though it carries live legal holds (#278).
    /// Explicit and never the default: `DROP SCHEMA … CASCADE` is the widest
    /// destruction the backend has, and it never opens the memory — so
    /// without this check it was the one path a hold could not reach at all.
    pub override_hold: Option<crate::HoldOverride>,
}

/// [`drop_postgres_schema`] with an explicit override of any legal holds the
/// schema carries.
///
/// Reads the memory's `meta` table — wherever the layout keeps it — for
/// `hold:%` rows first. A schema that does not exist, or holds none, drops
/// as before.
pub fn drop_postgres_schema_with(
    url: &str,
    schema: &str,
    opts: &DropOptions,
) -> Result<()> {
    let layout = PgLayout::from_url(url, schema)?;
    let (pool, _) = PgPool::for_url(url)?;
    let mut co = pool.checkout()?;
    if opts.override_hold.is_none() {
        // SELECT-only probe. A missing schema or meta table is not a hold;
        // the drop below is `IF EXISTS` and stays the authority on existence.
        let probe = pool.block_on(co.client().query(
            &format!(
                "SELECT k FROM {} WHERE k LIKE 'hold:%' ORDER BY k",
                layout.qualified("meta")
            ),
            &[],
        ));
        if let Ok(rows) = probe {
            if !rows.is_empty() {
                let held: Vec<String> = rows
                    .iter()
                    .filter_map(|r| r.try_get::<_, String>(0).ok())
                    .map(|k| k.trim_start_matches("hold:").to_string())
                    .collect();
                return Err(AreevError::LegalHold(format!(
                    "schema {schema:?} carries live legal holds on {} — dropping it \
                     would destroy held records; release the holds or drop with an \
                     explicit override",
                    held.join(", ")
                )));
            }
        }
    }
    // One simple-query batch = one implicit transaction, so a pair goes
    // together or not at all.
    let sql = layout
        .schemas()
        .iter()
        .map(|s| format!("DROP SCHEMA IF EXISTS \"{s}\" CASCADE"))
        .collect::<Vec<_>>()
        .join("; ");
    let r = pool.block_on(co.client().batch_execute(&sql)).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 single = PgLayout::single("s1").unwrap();
        let q = |sql: &str| qualify_tables(sql, &single);
        // 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");
        // `ON` (#353): index DDL names its table there, and nowhere else does
        // a table follow it — a join condition's `ON g.seq` names an alias.
        assert_eq!(
            q("CREATE INDEX IF NOT EXISTS idx_grains_hash ON grains(hash)"),
            "CREATE INDEX IF NOT EXISTS idx_grains_hash ON \"s1\".grains(hash)"
        );
        assert_eq!(
            q("SELECT 1 FROM heads h LEFT JOIN grains g ON g.seq = h.seq"),
            "SELECT 1 FROM \"s1\".heads h LEFT JOIN \"s1\".grains g ON g.seq = h.seq"
        );
    }

    /// The paired layout (#353): the metadata tables route to the metadata
    /// schema, everything else to the memory schema, within one statement.
    #[test]
    fn qualifier_routes_metadata_tables_to_the_metadata_schema() {
        let pair = PgLayout::paired("s1", "m1").unwrap();
        let q = |sql: &str| qualify_tables(sql, &pair);
        assert_eq!(q("SELECT v FROM meta WHERE k = $1"), "SELECT v FROM \"m1\".meta WHERE k = $1");
        assert_eq!(
            q("UPDATE counters SET v = v + $1 WHERE name = 'seq' RETURNING name, v"),
            "UPDATE \"m1\".counters SET v = v + $1 WHERE name = 'seq' RETURNING name, v"
        );
        assert_eq!(q("DELETE FROM ns_reg WHERE ns = $1"), "DELETE FROM \"m1\".ns_reg WHERE ns = $1");
        // PG_SEED's counter seed reads memory and writes metadata.
        assert_eq!(
            q("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 \"m1\".counters(name, v) SELECT 'seq', COALESCE(MAX(seq),0) FROM \"s1\".grains \
             ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)"
        );
        // The sidecar's index DDL is the reason `ON` triggers: without it the
        // index would resolve through the memory schema's search_path and
        // fail to find a table that lives in the other one.
        assert_eq!(
            q("CREATE INDEX IF NOT EXISTS idx_telem_recall_ts ON telem_recall_log(ts_ms)"),
            "CREATE INDEX IF NOT EXISTS idx_telem_recall_ts ON \"m1\".telem_recall_log(ts_ms)"
        );
        assert_eq!(
            q("CREATE TABLE IF NOT EXISTS telem_meta(k text PRIMARY KEY, v text)"),
            "CREATE TABLE IF NOT EXISTS \"m1\".telem_meta(k text PRIMARY KEY, v text)"
        );
        // Memory tables are untouched by the pairing.
        assert_eq!(
            q("SELECT body FROM blobs WHERE hash = $1"),
            "SELECT body FROM \"s1\".blobs WHERE hash = $1"
        );
        // Every bootstrap statement routes somewhere the qualifier
        // recognises, and every META table's DDL lands in the metadata schema
        // — a statement that created one of them in the memory schema would
        // be the "classified metadata left behind" the pair exists to rule out.
        for sql in PG_SCHEMA.iter().chain(PG_SEED.iter()).chain(crate::telemetry::TELEM_SCHEMA_PG) {
            let routed = q(sql);
            for t in META_TABLES {
                assert!(
                    !routed.contains(&format!("\"s1\".{t}")),
                    "metadata table {t} routed to the memory schema in: {routed}"
                );
            }
        }
    }

    #[test]
    fn layout_reads_the_dsn_and_refuses_a_degenerate_pair() {
        let single = PgLayout::from_url("postgres://h/db?pool=2", "s").unwrap();
        assert!(!single.is_paired());
        assert_eq!(single.meta_schema(), None);
        assert_eq!(single.schemas(), vec!["s"]);
        assert_eq!(single.schema_for("meta"), "s");

        let pair = PgLayout::from_url("postgres://h/db?meta_schema=s_meta&sslmode=require", "s").unwrap();
        assert!(pair.is_paired());
        assert_eq!(pair.meta_schema(), Some("s_meta"));
        assert_eq!(pair.schemas(), vec!["s", "s_meta"]);
        for t in META_TABLES {
            assert_eq!(pair.schema_for(t), "s_meta", "{t}");
        }
        for t in PG_TABLES.iter().filter(|t| !META_TABLES.contains(t)) {
            assert_eq!(pair.schema_for(t), "s", "{t}");
        }
        assert_eq!(pair.qualified("counters"), "\"s_meta\".counters");
        assert_eq!(pair.qualified("grains"), "\"s\".grains");

        // The same name twice is not a pair.
        let e = PgLayout::from_url("postgres://h/db?meta_schema=s", "s").unwrap_err();
        assert!(e.to_string().contains("names the memory schema itself"), "{e}");
        // An empty value is a refusal, never a silent fallback to single.
        let e = PgLayout::from_url("postgres://h/db?meta_schema=", "s").unwrap_err();
        assert!(e.to_string().contains("names no schema"), "{e}");
        // Both names obey the identifier rule.
        assert!(PgLayout::paired("s", "has space").is_err());
        assert!(PgLayout::paired("9lead", "m").is_err());
        assert_eq!(
            strip_meta_schema("postgres://h/db?schema=s&meta_schema=m&provision=never"),
            "postgres://h/db?schema=s&provision=never"
        );
        // The driver never sees it, and neither does the pool key.
        let req = crate::pgtls::SslRequest::split("postgres://u@h/db?meta_schema=m&sslmode=disable")
            .unwrap();
        assert_eq!(req.dsn, "postgres://u@h/db");
        assert_eq!(pool_key("postgres://u@h/db?schema=s&meta_schema=m&pool=3"), "postgres://u@h/db");
    }

    #[test]
    fn qualifier_leaves_data_catalogs_and_target_references_alone() {
        let single = PgLayout::single("s1").unwrap();
        let q = |sql: &str| qualify_tables(sql, &single);
        // 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 single = PgLayout::single("s1").unwrap();
        let out = qualify_tables(&translate("PRAGMA integrity_check").unwrap(), &single);
        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 {
    #[test]
    fn rolling_deploy_verdict_reads_the_found_stamp() {
        // #308: `unknown` when there is no stamp to compare against —
        // absence of evidence is never reported as safety.
        assert!(version_lt("1", "2"));
        assert!(!version_lt("2", "2"));
        assert!(!version_lt("3", "2"));
        // An unparseable stamp is not evidence of safety either.
        assert!(version_lt("wat", "2"));
    }

    #[test]
    fn this_builds_migration_is_rolling_safe_from_the_stated_version() {
        // Version 2 adds a NULLABLE column and an index, so a version-1
        // writer survives the migration. If a later bump is NOT rolling
        // safe, PG_ROLLING_SAFE_FROM must move with it — this test is the
        // reminder, and the schema-digest test is what forces the review.
        assert!(!version_lt(PG_ROLLING_SAFE_FROM, PG_SCHEMA_VERSION)
            || PG_ROLLING_SAFE_FROM == "1");
    }

    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 =
        "6e0120dfa4c54bdbc8e1beaaf402719df728d0bda1fffb628881458cc39f541f";

    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 pool_idle_ttl_reads_the_dsn_then_the_environment() {
        let secs = |u: &str| pool_idle_ttl(u).unwrap().map(|d| d.as_secs());
        assert_eq!(secs("postgres://u@h/db?schema=x"), Some(DEFAULT_POOL_IDLE_SECS));
        assert_eq!(secs("postgres://u@h/db?pool_idle_secs=30&schema=x"), Some(30));
        // 0 is a setting, not an absence: keep every connection dialled for
        // the life of the process, which is what every build before #229 did.
        assert_eq!(secs("postgres://u@h/db?pool_idle_secs=0"), None);
        assert!(pool_idle_ttl("postgres://u@h/db?pool_idle_secs=soon").is_err());
        std::env::set_var(POOL_IDLE_ENV, "45");
        assert_eq!(secs("postgres://u@h/db"), Some(45));
        assert_eq!(secs("postgres://u@h/db?pool_idle_secs=10"), Some(10), "the DSN wins");
        std::env::set_var(POOL_IDLE_ENV, "whenever");
        assert!(pool_idle_ttl("postgres://u@h/db").is_err());
        std::env::remove_var(POOL_IDLE_ENV);
        assert_eq!(
            strip_pool_idle("postgres://u@h/db?pool_idle_secs=30&schema=x"),
            "postgres://u@h/db?schema=x"
        );
        // The driver never sees it — `tokio_postgres` rejects unknown options.
        let req =
            crate::pgtls::SslRequest::split("postgres://u@h/db?pool_idle_secs=30&sslmode=disable")
                .unwrap();
        assert_eq!(req.dsn, "postgres://u@h/db");
    }

    /// No server needed: a pool is a runtime, a semaphore and an empty idle
    /// list until somebody checks out, so the registry's own rules are
    /// testable against a DSN that resolves nowhere.
    #[test]
    fn the_pool_key_ignores_the_idle_parameter_and_a_later_open_is_told_not_obeyed() {
        let host = "pool-key.invalid";
        let (first, w) =
            PgPool::for_url(&format!("postgres://u@{host}/db?pool=4&pool_idle_secs=0&schema=one"))
                .unwrap();
        assert!(w.is_none(), "{w:?}");
        let (second, w) =
            PgPool::for_url(&format!("postgres://u@{host}/db?pool=4&pool_idle_secs=0&schema=two"))
                .unwrap();
        assert!(w.is_none(), "{w:?}");
        assert!(std::sync::Arc::ptr_eq(&first, &second), "neither schema nor TTL keys the pool");

        let (third, w) =
            PgPool::for_url(&format!("postgres://u@{host}/db?pool=9&pool_idle_secs=30&schema=c"))
                .unwrap();
        assert!(std::sync::Arc::ptr_eq(&first, &third));
        assert_eq!(third.idle_ttl, None, "the first open set the TTL");
        let w = w.unwrap();
        assert!(w.contains("sized at 4"), "{w}");
        assert!(w.contains("after never"), "{w}");
    }

    /// The #229 lifecycle, minus the connections: a pool nothing holds, with
    /// nothing idle in it, does not outlive the sweep.
    #[test]
    fn a_pool_that_holds_nothing_and_that_nothing_holds_is_evicted() {
        let host = "pool-evict.invalid";
        let (p, _) =
            PgPool::for_url(&format!("postgres://u@{host}/db?pool=2&pool_idle_secs=1")).unwrap();
        assert!(registry().pools.keys().any(|k| k.contains(host)));
        drop(p);
        sweep();
        assert!(
            !registry().pools.keys().any(|k| k.contains(host)),
            "an unreferenced, empty pool must not survive a sweep"
        );
    }

    #[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");
    }
}