areev-store 1.7.2

//! 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.

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::{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.
    "CREATE TABLE IF NOT EXISTS terms(id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, term text UNIQUE)",
    // Write-transaction id blocks (seq / op / hlc): claimed with
    // UPDATE … RETURNING inside each write txn — the row locks are also
    // what serializes concurrent write transactions per memory.
    "CREATE TABLE IF NOT EXISTS counters(name text PRIMARY KEY, v bigint NOT NULL)",
    "CREATE TABLE IF NOT EXISTS grains(
        seq bigint PRIMARY KEY,
        hash bytea NOT NULL,
        ns bigint, gtype bigint, created_at bigint,
        s bigint, p bigint, o bigint,
        vf bigint, vt bigint,
        svf bigint, svt bigint,
        superseded_by bytea, supersedes bytea,
        text text,
        blob bytea NOT NULL)",
    "CREATE UNIQUE INDEX IF NOT EXISTS idx_grains_hash ON grains(hash)",
    // Mirrors the embedded SCHEMA's idx_grains_ns_s, for the same reason and
    // in the same column order: the vector legs join embeddings to grains and
    // filter on (ns, s, p), and without this the planner scans grains and
    // computes a distance per row. This tier has the steeper slope of the two
    // (a round trip per statement on top of the scan), so it needs it more.
    "CREATE INDEX IF NOT EXISTS idx_grains_ns_s ON grains(ns, s, p)",
    // Dimension-less at creation: the `vec vector(dim)` column arrives via
    // `ensure_embeddings` at the first set_embedder (the dim is host-
    // supplied), but the TABLE must exist up front — forget() deletes from
    // it unconditionally.
    "CREATE TABLE IF NOT EXISTS embeddings(seq bigint PRIMARY KEY)",
    "CREATE TABLE IF NOT EXISTS fts_vocab(id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, term text UNIQUE)",
    "CREATE TABLE IF NOT EXISTS fts_post(term bigint, seq bigint, ns bigint, tf bigint)",
    "CREATE INDEX IF NOT EXISTS idx_fts_post ON fts_post(term, ns)",
    "CREATE INDEX IF NOT EXISTS idx_fts_post_seq ON fts_post(seq)",
    "CREATE TABLE IF NOT EXISTS fts_doc(seq bigint PRIMARY KEY, len bigint)",
    "CREATE TABLE IF NOT EXISTS triples(ns bigint, s bigint, p bigint, o bigint, seq bigint, cur bigint)",
    "CREATE INDEX IF NOT EXISTS idx_spo ON triples(ns,s,p,o,seq)",
    "CREATE INDEX IF NOT EXISTS idx_pos ON triples(ns,p,o,s,seq)",
    "CREATE INDEX IF NOT EXISTS idx_triples_seq ON triples(seq)",
    "CREATE TABLE IF NOT EXISTS osp(ns bigint, o bigint, s bigint, p bigint, seq bigint, cur bigint)",
    "CREATE INDEX IF NOT EXISTS idx_osp ON osp(ns,o,s)",
    "CREATE INDEX IF NOT EXISTS idx_osp_seq ON osp(seq)",
    "CREATE TABLE IF NOT EXISTS entity_latest(ns bigint, s bigint, p bigint, o bigint, seq bigint, hash bytea, PRIMARY KEY(ns,s,p))",
    "CREATE TABLE IF NOT EXISTS heads(ns bigint, s bigint, p bigint, seq bigint, hash bytea, created_at bigint, PRIMARY KEY(ns,s,p,seq))",
    "CREATE TABLE IF NOT EXISTS oplog(op_seq bigint PRIMARY KEY, hlc bigint, op bigint, hash bytea)",
    "CREATE TABLE IF NOT EXISTS thread_idx(ns bigint, session bigint, seq bigint)",
    "CREATE INDEX IF NOT EXISTS idx_thread ON thread_idx(ns, session, seq)",
    "CREATE TABLE IF NOT EXISTS prov_idx(ns bigint, parent bytea, seq bigint)",
    "CREATE INDEX IF NOT EXISTS idx_prov ON prov_idx(ns, parent, seq)",
    "CREATE TABLE IF NOT EXISTS run_idx(ns bigint, run bigint, seq bigint)",
    "CREATE INDEX IF NOT EXISTS idx_run ON run_idx(ns, run, seq)",
    "CREATE TABLE IF NOT EXISTS corpus_idx(seq bigint PRIMARY KEY)",
    // Namespace registry (prefix scoping) — see the embedded SCHEMA's ns_reg
    // comment. Same count-maintained contract; the serialized write txns
    // (reserve_write's row lock) make the read-modify-write bumps safe here.
    "CREATE TABLE IF NOT EXISTS ns_reg(ns bigint PRIMARY KEY, n bigint NOT NULL)",
    // Replaces the `.blobs` filesystem sidecar: same CAS semantics, rows
    // keyed by the raw 32-byte content hash.
    "CREATE TABLE IF NOT EXISTS blobs(hash bytea PRIMARY KEY, body bytea NOT NULL)",
];

/// Seed the allocation state from what the schema already holds — safe to
/// run concurrently from several opening instances (GREATEST-merge upserts)
/// and idempotent. Runs after [`PG_SCHEMA`] on every open.
pub(crate) const PG_SEED: &[&str] = &[
    "INSERT INTO counters(name, v) SELECT 'seq', COALESCE(MAX(seq),0) FROM grains \
     ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)",
    "INSERT INTO counters(name, v) SELECT 'op', COALESCE(MAX(op_seq),0) FROM oplog \
     ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)",
    "INSERT INTO counters(name, v) SELECT 'hlc', COALESCE(MAX(hlc),0) FROM oplog \
     ON CONFLICT (name) DO UPDATE SET v = GREATEST(counters.v, EXCLUDED.v)",
    // Legacy migration: a schema created by the stage-1 build has
    // terms(id bigint PK) with NO identity — `CREATE TABLE IF NOT EXISTS`
    // never adds one, and the first id-less intern would violate NOT NULL
    // forever. Add the identity and seed its sequence past the explicit
    // ids IN THE SAME step: the seeding is race-free exactly here, because
    // until the identity exists no writer can have drawn from it (and the
    // whole bootstrap runs under the schema's advisory lock). An
    // unconditional setval on every open would instead RACE concurrent
    // interns and rewind the sequence below ids already handed out.
    "DO $$ BEGIN \
       IF NOT EXISTS (SELECT 1 FROM pg_attribute a \
                        JOIN pg_class c ON a.attrelid = c.oid \
                        JOIN pg_namespace n ON c.relnamespace = n.oid \
                       WHERE n.nspname = current_schema() AND c.relname = 'terms' \
                         AND a.attname = 'id' AND a.attidentity <> '') THEN \
         EXECUTE 'ALTER TABLE terms ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY'; \
         PERFORM setval(pg_get_serial_sequence('terms','id'), \
                        GREATEST((SELECT COALESCE(MAX(id),0) FROM terms), 1)); \
       END IF; \
     END $$",
];

pub(crate) struct PgDb {
    rt: tokio::runtime::Runtime,
    /// Kept so a dead session can be replaced in place. A managed Postgres
    /// restarts, fails over, and drops idle connections as ROUTINE
    /// maintenance, not as an incident — before this, one such event
    /// permanently poisoned a long-lived handle and every later call failed
    /// until the process was recycled.
    url: String,
    schema: String,
    // `Arc` so a caller can clone the handle out of the cell and drop the
    // borrow BEFORE awaiting on it. Holding a `Ref` across an await is the
    // shape that would deadlock against `reconnect`'s `borrow_mut`; taking the
    // clone first makes that impossible by construction rather than by
    // reasoning about which futures run when. `Arc` rather than `Rc` because
    // `Db` is `Send` — the handle moves between threads even though only one
    // uses it at a time.
    client: RefCell<std::sync::Arc<tokio_postgres::Client>>,
    /// Prepared statements keyed by the ORIGINAL (untranslated) SQL, so hot
    /// callers hit the cache without re-translating.
    ///
    /// MUST be cleared on reconnect: a `tokio_postgres::Statement` names a
    /// server-side prepared statement belonging to ONE session, so carrying
    /// the cache across a reconnect makes every hot call fail forever with
    /// `26000 invalid_sql_statement_name` — a reconnect that reconnects and
    /// still cannot serve a query.
    cache: RefCell<HashMap<String, tokio_postgres::Statement>>,
    /// 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>,
    /// 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)>>,
}

impl PgDb {
    /// Connect, create-or-attach the schema, pin `search_path`, and run the
    /// caller's `bootstrap` statements (DDL + seeding). The whole bootstrap
    /// runs under a database-wide advisory lock keyed on the schema name:
    /// this backend admits concurrent openers, and Postgres's
    /// `IF NOT EXISTS` DDL is racy without one (the loser gets a spurious
    /// 23505 on pg_namespace/pg_type). The lock is transaction-free and
    /// explicitly released, so it never outlives the bootstrap.
    ///
    /// `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 pins
    /// `search_path` (a session command any role may issue) and VERIFIES
    /// with SELECT-only probes — see [`Self::verify_read_only`].
    pub(crate) fn open(url: &str, schema: &str, bootstrap: &[&str], read_only: bool) -> Result<Self> {
        if schema.is_empty()
            || schema.len() > 63
            || !schema.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')
            || schema.as_bytes()[0].is_ascii_digit()
        {
            return Err(AreevError::Validation(format!(
                "postgres schema name must be [a-z_][a-z0-9_]* and <= 63 bytes, got {schema:?}"
            )));
        }
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(db_err)?;
        // `pgtls::connect` reads the DSN's `sslmode`/`sslrootcert`, spawns the
        // connection future onto this runtime (it must be driven for the
        // client to make progress; the current-thread runtime polls it inside
        // every block_on below), and REFUSES a DSN that asks for encryption a
        // build without `postgres-tls` cannot give it.
        let client = crate::pgtls::connect(&rt, url)?;
        if read_only {
            rt.block_on(Self::verify_read_only(&client, schema))?;
        } else {
            rt.block_on(async {
                client
                    .query_one(
                        "SELECT pg_advisory_lock(hashtext('areev_bootstrap'), hashtext($1))",
                        &[&schema],
                    )
                    .await
                    .map_err(pg_err)?;
                let boot = async {
                    client
                        .batch_execute(&format!("CREATE SCHEMA IF NOT EXISTS \"{schema}\""))
                        .await
                        .map_err(pg_err)?;
                    client
                        .batch_execute(&format!("SET search_path TO \"{schema}\", public, ext"))
                        .await
                        .map_err(pg_err)?;
                    for sql in bootstrap {
                        client.batch_execute(sql).await.map_err(|e| {
                            AreevError::Storage(format!(
                                "bootstrap failed: {} — in: {sql}",
                                pg_err(e)
                            ))
                        })?;
                    }
                    Ok::<_, AreevError>(())
                }
                .await;
                // Release even on failure — the session (and any pooler behind
                // it) may outlive this open attempt.
                let _ = client
                    .query_one(
                        "SELECT pg_advisory_unlock(hashtext('areev_bootstrap'), hashtext($1))",
                        &[&schema],
                    )
                    .await;
                boot
            })?;
        }
        Ok(Self {
            rt,
            url: url.to_string(),
            schema: schema.to_string(),
            client: RefCell::new(std::sync::Arc::new(client)),
            cache: RefCell::new(HashMap::new()),
            in_txn: std::cell::Cell::new(false),
            stats: RefCell::new(None),
        })
    }

    /// The read-only counterpart to the bootstrap block above: pin
    /// `search_path` (harmless for any role), then confirm with SELECT-only
    /// probes that there is something real to point it at — 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`".
    async fn verify_read_only(
        client: &tokio_postgres::Client,
        schema: &str,
    ) -> Result<()> {
        client
            .batch_execute(&format!("SET search_path TO \"{schema}\", public, ext"))
            .await
            .map_err(pg_err)?;
        let exists = client
            .query_one(
                "SELECT EXISTS(SELECT 1 FROM information_schema.schemata WHERE schema_name = $1)",
                &[&schema],
            )
            .await
            .map_err(pg_err)?
            .get::<_, bool>(0);
        if !exists {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {schema:?} does not exist — a read-only open never runs \
                 bootstrap DDL (CREATE SCHEMA included), so it cannot create it. Either the \
                 schema name is wrong, or the memory has never been created: have the owning \
                 role open it read-write once (or run the migration), then retry read-only"
            )));
        }
        const CORE_TABLES: &[&str] = &["meta", "terms", "grains", "oplog", "fts_doc"];
        let mut missing: Vec<&str> = Vec::new();
        for table in CORE_TABLES {
            let qualified = format!("\"{schema}\".{table}");
            let present = client
                .query_one("SELECT to_regclass($1) IS NOT NULL", &[&qualified])
                .await
                .map_err(pg_err)?
                .get::<_, bool>(0);
            if !present {
                missing.push(table);
            }
        }
        if !missing.is_empty() {
            return Err(AreevError::ReadOnlyOpenFailed(format!(
                "postgres schema {schema:?} exists but is not fully initialized — missing \
                 table(s): {} — a read-only open never runs bootstrap DDL to add them. Have the \
                 owning role open this memory read-write once to finish migrating it, then \
                 retry read-only",
                missing.join(", ")
            )));
        }
        Ok(())
    }

    /// Is this error the CONNECTION dying, as opposed to the statement failing?
    ///
    /// Only these justify replacing the client: an ordinary constraint
    /// violation must stay an ordinary error. Covers the client noticing the
    /// socket is gone (`is_closed`), 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.
    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")
    }

    /// Replace the dead session with a fresh one.
    ///
    /// Deliberately does NOT re-run the bootstrap DDL: the schema already
    /// exists, and re-taking the bootstrap advisory lock on every blip would
    /// serialise recovery across every handle on the database. Only the
    /// session-local state has to be restored — `search_path` — and the two
    /// caches that belonged to the old session have to go.
    fn reconnect(&self) -> Result<()> {
        let client = crate::pgtls::connect(&self.rt, &self.url)?;
        self.rt.block_on(client.batch_execute(&format!(
            "SET search_path TO \"{}\", public, ext",
            self.schema
        )))
        .map_err(pg_err)?;
        // Order matters only in that both must happen before the next call:
        // a Statement from the old session is invalid, and the BM25 collection
        // stats were cached against a connection that can no longer confirm them.
        self.cache.borrow_mut().clear();
        *self.stats.borrow_mut() = None;
        *self.client.borrow_mut() = std::sync::Arc::new(client);
        Ok(())
    }

    /// Reconnect after a connection-level failure, and report whether the
    /// failed call may be replayed.
    ///
    /// `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 the fresh connection
    /// for NEXT time and 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;
        }
        // Two independent signals, because neither alone is complete. The
        // SQLSTATE match reads the message `pg_err` formatted, which only
        // exists when the SERVER got far enough to send an error; when the
        // socket simply vanished there is no SQLSTATE to match and the client
        // is the only witness. Asking the client directly also means a change
        // to `pg_err`'s wording degrades recovery rather than silently
        // disabling it. The borrow is dropped before `reconnect` takes its
        // `borrow_mut`.
        let socket_gone = self.client.borrow().is_closed();
        if !socket_gone && !Self::is_connection_dead(e) {
            return false;
        }
        // A failed reconnect leaves the handle dead-but-retryable rather than
        // erroring here: the caller's original error is the more useful one,
        // and the next call tries again.
        if self.reconnect().is_err() {
            return false;
        }
        replayable
    }

    /// Prepare-and-cache. ONLY the `_hot` calls come through here: their SQL
    /// is a fixed set of `&'static` literals, so the cache (and the server-
    /// side named statements behind it) is bounded. Plain `query`/`execute`
    /// must NOT cache — the recall path builds per-call-unique IN-list SQL,
    /// which would grow both sides without bound.
    fn prepared(&self, sql: &str) -> Result<tokio_postgres::Statement> {
        if let Some(st) = self.cache.borrow().get(sql) {
            return Ok(st.clone());
        }
        let translated = translate(sql)?;
        let client = self.client.borrow().clone();
        let st = self.rt.block_on(client.prepare(&translated)).map_err(|e| {
            AreevError::Storage(format!("prepare failed: {e} — translated SQL: {translated}"))
        })?;
        self.cache.borrow_mut().insert(sql.to_string(), st.clone());
        Ok(st)
    }

    fn query_once(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<Vec<Row>> {
        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(sql)?;
            let client = self.client.borrow().clone();
            self.rt.block_on(client.query(&st, &refs))
        } else {
            // Uncached path: the temporary statement is closed on drop, so
            // dynamic SQL leaves nothing behind on either side.
            let translated = translate(sql)?;
            let client = self.client.borrow().clone();
            self.rt.block_on(client.query(translated.as_str(), &refs))
        }
        .map_err(pg_err)?;
        rows.iter().map(row_to_row).collect()
    }

    /// A read, replayed once if the connection died under it.
    ///
    /// 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.recover(&e, true) {
                    self.query_once(sql, params, hot)
                } else {
                    Err(e)
                }
            }
        }
    }

    /// A write. Reconnects for the NEXT call but never replays this one — see
    /// [`recover`](Self::recover) for why a possibly-committed write must not
    /// be re-run.
    fn run_execute(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<u64> {
        match self.execute_once(sql, params, hot) {
            Ok(n) => Ok(n),
            Err(e) => {
                self.recover(&e, false);
                Err(e)
            }
        }
    }

    fn execute_once(&self, sql: &str, params: Vec<Value>, hot: bool) -> Result<u64> {
        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(sql)?;
            let client = self.client.borrow().clone();
            self.rt.block_on(client.execute(&st, &refs)).map_err(pg_err)
        } else {
            let translated = translate(sql)?;
            let client = self.client.borrow().clone();
            self.rt
                .block_on(client.execute(translated.as_str(), &refs))
                .map_err(pg_err)
        }
    }
}

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<()> {
        let client = self.client.borrow().clone();
        let r = self.rt.block_on(client.batch_execute("BEGIN")).map_err(pg_err);
        // Set only on success, so a failed BEGIN does not wedge the handle
        // into "a transaction is open" and block every later recovery.
        match &r {
            Ok(_) => self.in_txn.set(true),
            Err(e) => {
                self.recover(e, false);
            }
        }
        r
    }

    fn commit(&self) -> Result<()> {
        let client = self.client.borrow().clone();
        let r = self.rt.block_on(client.batch_execute("COMMIT")).map_err(pg_err);
        // Cleared either way: a failed COMMIT ends the transaction just as
        // surely as a successful one, and leaving the flag set would suppress
        // reconnects forever after a single outage mid-transaction.
        self.in_txn.set(false);
        if let Err(ref e) = r {
            self.recover(e, false);
        }
        r
    }

    fn rollback(&self) -> Result<()> {
        let client = self.client.borrow().clone();
        let r = self.rt.block_on(client.batch_execute("ROLLBACK")).map_err(pg_err);
        self.in_txn.set(false);
        if let Err(ref e) = r {
            self.recover(e, false);
        }
        r
    }

    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>> {
        let inserted = self.run_query(
            "INSERT INTO terms(term) VALUES (?1) ON CONFLICT (term) DO NOTHING RETURNING id",
            vec![pt(term)],
            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 = ?1", vec![pt(term)], 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 = ?1", vec![pt(term)], true)?
            .first()
            .and_then(|r| r.i64(0)))
    }

    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 ensure_embeddings(&self, dim: usize) -> Result<()> {
        self.rt.block_on(async {
            // 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: search_path is pinned to
            // the MEMORY's schema, and without it pgvector's objects would
            // install into that schema — invisible to sibling memories, and
            // destroyed database-wide by that one memory's DROP SCHEMA
            // erasure.
            // Cloned out of the cell up front, so no borrow is held across an
            // await (see the `client` field).
            let client = self.client.borrow().clone();
            let _ = client
                .batch_execute("CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public")
                .await;
            client
                .batch_execute(&format!(
                    "ALTER TABLE 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 = current_schema() AND c.relname = 'embeddings'
                       AND a.attname = 'vec'",
                    &[],
                )
                .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}"
            )));
        }
        self.rt.block_on(async {
            let client = self.client.borrow().clone();
            // 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 = current_schema() AND c.relname = 'embeddings'
                        AND a.attname = 'vec' AND NOT a.attisdropped",
                    &[],
                )
                .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 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}"))
                })?;
            // Session-scoped, and this backend holds one connection per store
            // handle, so it lasts as long as the handle does. It is 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.
            client
                .batch_execute(&format!("SET hnsw.ef_search = {ef_search}"))
                .await
                .map_err(pg_err)?;
            Ok(())
        })
    }

    fn drop_ann_index(&self) -> Result<()> {
        self.rt.block_on(async {
            let client = self.client.borrow().clone();
            client
                .batch_execute("DROP INDEX IF EXISTS idx_embeddings_hnsw")
                .await
                .map_err(pg_err)?;
            Ok(())
        })
    }

    fn ann_index_name(&self) -> Result<Option<String>> {
        self.rt.block_on(async {
            let client = self.client.borrow().clone();
            let rows = client
                .query(
                    "SELECT indexname FROM pg_indexes
                      WHERE schemaname = current_schema() AND tablename = 'embeddings'
                        AND indexdef LIKE '%hnsw%'",
                    &[],
                )
                .await
                .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.
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.
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))
}

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

#[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 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).is_err(),
                "{bad:?} must be rejected"
            );
        }
    }
}