areev-store 1.2.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.
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 => db_err(e),
    }
}

/// 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)",
    // 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,
    client: tokio_postgres::Client,
    /// Prepared statements keyed by the ORIGINAL (untranslated) SQL, so hot
    /// callers hit the cache without re-translating.
    cache: RefCell<HashMap<String, tokio_postgres::Statement>>,
    /// 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.
    pub(crate) fn open(url: &str, schema: &str, bootstrap: &[&str]) -> 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)?;
        let (client, connection) = rt
            .block_on(tokio_postgres::connect(url, tokio_postgres::NoTls))
            .map_err(pg_err)?;
        // The connection future must be driven for the client to make
        // progress; the current-thread runtime polls it inside every
        // block_on below.
        rt.spawn(async move {
            let _ = connection.await;
        });
        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, client, cache: RefCell::new(HashMap::new()), stats: RefCell::new(None) })
    }

    /// 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 st = self.rt.block_on(self.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 run_query(&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)?;
            self.rt.block_on(self.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)?;
            self.rt.block_on(self.client.query(translated.as_str(), &refs))
        }
        .map_err(pg_err)?;
        rows.iter().map(row_to_row).collect()
    }

    fn run_execute(&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)?;
            self.rt.block_on(self.client.execute(&st, &refs)).map_err(pg_err)
        } else {
            let translated = translate(sql)?;
            self.rt
                .block_on(self.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<()> {
        self.rt.block_on(self.client.batch_execute("BEGIN")).map_err(pg_err)
    }

    fn commit(&self) -> Result<()> {
        self.rt.block_on(self.client.batch_execute("COMMIT")).map_err(pg_err)
    }

    fn rollback(&self) -> Result<()> {
        self.rt.block_on(self.client.batch_execute("ROLLBACK")).map_err(pg_err)
    }

    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.
            let _ = self
                .client
                .batch_execute("CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public")
                .await;
            self.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 = self
                .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(())
        })
    }
}

/// 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, connection) =
        rt.block_on(tokio_postgres::connect(url, tokio_postgres::NoTls)).map_err(pg_err)?;
    rt.spawn(async move {
        let _ = connection.await;
    });
    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, &[]).is_err(), "{bad:?} must be rejected");
        }
    }
}