fraiseql_db/postgres/adapter/

database.rs

//! `DatabaseAdapter` and `Writer` implementations for `PostgresAdapter`.

use std::sync::Arc;

use async_trait::async_trait;
use bytes::BufMut as _;
use fraiseql_error::{FraiseQLError, Result};
use tokio_postgres::{
    Row,
    types::{FromSql, Kind, Type},
};

use super::{
    PostgresAdapter, build_projection_select_sql, build_where_select_sql,
    build_where_select_sql_ordered, composed::build_composed_select_sql, jsonb_cell,
    numeric::PgNumericText,
};
use crate::{
    identifier::quote_postgres_identifier,
    postgres::pg_detail,
    traits::{
        ColumnRowStream, ComposedLevel, DatabaseAdapter, JsonbRowStream, ProjectionRequest, Writer,
    },
    types::{
        DatabaseType, JsonbValue, PoolMetrics, QueryParam, ReadRouting,
        sql_hints::{OrderByClause, SqlProjectionHint},
    },
    where_clause::WhereClause,
};

/// The one write (ruling AA 6): run a write function inside one explicit transaction,
/// with the change-log outbox row in the same statement when the request carries one, and
/// let the gate decide the transaction's end.
///
/// `gate` runs after the function has executed and before the transaction ends, on the
/// rows it returned. An `Err` from it rolls back and is returned to the caller verbatim, so
/// a refusal that could only be decided from the written row still refuses the write. In
/// [`WriteMode::DryRun`] everything else happens exactly as in a commit — session
/// variables, CDC marking, the timing stamp, the outbox row, the gate — and the transaction
/// then rolls back: a dry run exercises the write it stands in for, and nothing persists.
async fn run_function_in_txn(
    adapter: &PostgresAdapter,
    request: &crate::traits::WriteRequest<'_>,
    gate: crate::traits::MutationRowGate<'_>,
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
    let crate::traits::WriteRequest {
        function: function_name,
        args,
        session_vars,
        changelog,
        mode,
        ..
    } = *request;
    let commit = matches!(mode, crate::traits::WriteMode::Commit);
    // Arm the read-your-writes pin (entry + post-commit) — #407. A dry run writes nothing
    // for a later read to miss.
    if commit {
        adapter.mark_write();
    }
    let quoted_fn = quote_postgres_identifier(function_name);

    // One statement when there is an outbox row: run the function once and INSERT
    // its outbox row in the same txn, atomically, with no extra connection acquire
    // (Change Spine). Otherwise the plain call.
    let sql = if let Some(cl) = changelog {
        build_changelog_cte_sql(&quoted_fn, args.len(), cl.pre_image)
    } else {
        let placeholders: Vec<String> = (1..=args.len()).map(|i| format!("${i}")).collect();
        format!("SELECT * FROM {quoted_fn}({})", placeholders.join(", "))
    };

    // Arguments bind as FlexParam: serde_json::Value only accepts JSON/JSONB types, and
    // Option<String> only text-family types, and a function signature can mix JSONB, UUID,
    // INT4 and TEXT. FlexParam accepts every PostgreSQL type and serialises each value in the
    // binary wire format of the server-resolved parameter type.
    let mut flex_args: Vec<FlexParam> = args
        .iter()
        .map(|v| match v {
            serde_json::Value::Null => FlexParam::Null,
            serde_json::Value::String(s) => FlexParam::Text(s.clone()),
            _ => FlexParam::Text(v.to_string()),
        })
        .collect();
    if let Some(changelog) = changelog {
        // Function args first; then the threaded change-log envelope params —
        // object_type fallback ($n+1), modification_type verb ($n+2), the
        // tenant_id stamp ($n+3, bound against `::uuid`), the trace_id
        // ($n+4, plain text), the schema_version ($n+5, plain text), the
        // trace_context ($n+6, bound against `::jsonb`), the actor_type ($n+7,
        // plain text), the acting_for ($n+8, bound against `::uuid`) and the
        // transport ($n+9, plain text merged into extra_metadata — #376). Order
        // matches build_changelog_cte_sql's positional contract; appending the
        // envelope params keeps the SQL text stable for prepare_cached.
        flex_args.push(FlexParam::Text(changelog.object_type.to_string()));
        flex_args.push(FlexParam::Text(changelog.modification_type.to_string()));
        // tenant_id: bound as text and serialised by FlexParam's UUID branch
        // (the `::uuid` cast pins the param type); None → SQL NULL.
        flex_args
            .push(changelog.tenant_id.map_or(FlexParam::Null, |t| FlexParam::Text(t.to_string())));
        // trace_id ($n+4): plain text, None → SQL NULL.
        flex_args
            .push(changelog.trace_id.map_or(FlexParam::Null, |t| FlexParam::Text(t.to_string())));
        // schema_version ($n+5): plain text, None → SQL NULL.
        flex_args.push(
            changelog
                .schema_version
                .map_or(FlexParam::Null, |s| FlexParam::Text(s.to_string())),
        );
        // trace_context ($n+6): JSON text bound against `::jsonb`, None → SQL NULL.
        flex_args.push(
            changelog
                .trace_context
                .map_or(FlexParam::Null, |s| FlexParam::Text(s.to_string())),
        );
        // actor_type ($n+7): plain text, None → SQL NULL.
        flex_args
            .push(changelog.actor_type.map_or(FlexParam::Null, |s| FlexParam::Text(s.to_string())));
        // acting_for ($n+8): bound as text + serialised by FlexParam's UUID branch
        // (the `::uuid` cast pins the param type); None → SQL NULL.
        flex_args
            .push(changelog.acting_for.map_or(FlexParam::Null, |u| FlexParam::Text(u.to_string())));
        // transport ($n+9): plain text merged into extra_metadata.transport by the
        // CTE's deterministic CASE; None → SQL NULL → key omitted (#376).
        flex_args
            .push(changelog.transport.map_or(FlexParam::Null, |s| FlexParam::Text(s.to_string())));
    }
    let params: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = flex_args
        .iter()
        .map(|v| v as &(dyn tokio_postgres::types::ToSql + Sync))
        .collect();

    let mut client = adapter.acquire_connection_with_retry().await?;
    // Parse/plan the (complex) CTE once per connection (statement cache) — this
    // is the dominant outbox hot-path cost; cached, the in-txn write is ~free.
    let stmt = prepare_cached_stmt(&client, sql.as_str()).await?;
    // The messages below name the outbox when the outbox is what ran. They are the
    // strings the error sanitizer's corpus is reduced from, and an operator reading a
    // failed write wants to know whether the CTE was in it.
    let mechanism = if changelog.is_some() {
        " (with change-log outbox)"
    } else {
        ""
    };
    let txn = client.build_transaction().start().await.map_err(|e| FraiseQLError::Database {
        message:   format!("Failed to start mutation{mechanism} transaction: {}", pg_detail(&e)),
        sql_state: e.code().map(|c| c.code().to_string()),
    })?;

    // Apply session variables FIRST so the function body sees them (and so
    // the started_at directive stamps the DB clock on this very txn).
    apply_session_vars(&txn, session_vars).await?;

    // Mark the txn FraiseQL-mediated so the #366 fallback-capture trigger
    // suppresses its row — this write is already logged by the outbox below,
    // and on the no-outbox path the opt-out means "no change-log row," not
    // "let the trigger write a degraded one."
    mark_cdc_mediated(&txn).await?;

    // If mutation timing is on, stamp the timing variable in the same txn.
    if adapter.mutation_timing_enabled {
        txn.execute(
            "SELECT set_config($1, clock_timestamp()::text, true)",
            &[&adapter.timing_variable_name],
        )
        .await
        .map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to set mutation timing variable: {}", pg_detail(&e)),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;
    }

    // The outbox INSERT reads `fraiseql.started_at` with current_setting()
    // (no missing_ok), so the GUC MUST exist on this txn. The mutation runner
    // injects it via session_vars on the authenticated path; guarantee it on
    // every other path (e.g. an unauthenticated mutation that resolves no
    // session vars) so the duration computation never hits an unset
    // parameter and aborts the mutation. Only the outbox reads it, so the
    // no-changelog path is left exactly as a plain call
    // leaves it.
    if changelog.is_some() {
        let started_at_set =
            session_vars.iter().any(|(name, _)| *name == crate::changelog::STARTED_AT_VAR)
                || (adapter.mutation_timing_enabled
                    && adapter.timing_variable_name == crate::changelog::STARTED_AT_VAR);
        if !started_at_set {
            txn.execute(
                "SELECT set_config($1, clock_timestamp()::text, true)",
                &[&crate::changelog::STARTED_AT_VAR],
            )
            .await
            .map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to stamp change-log started_at: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;
        }
    }

    let rows: Vec<Row> =
        txn.query(&stmt, params.as_slice()).await.map_err(|e| FraiseQLError::Database {
            message:   format!(
                "Function call {function_name}{mechanism} failed: {}",
                pg_detail(&e)
            ),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

    let results: Vec<std::collections::HashMap<String, serde_json::Value>> =
        rows.iter().map(row_to_map).collect();

    // The gate decides the commit. It sees the rows the function returned — the
    // only thing a per-row decision can be keyed on — while they are still
    // uncommitted, so a refusal takes the write with it rather than leaving the
    // caller refused and the side effect standing (#1353).
    if let Err(refusal) = gate(&results) {
        txn.rollback().await.map_err(|e| FraiseQLError::Database {
            message:   format!(
                "Failed to roll back refused mutation transaction: {}",
                pg_detail(&e)
            ),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;
        return Err(refusal);
    }

    if !commit {
        // The whole point of a dry run: discard every write, the outbox row included.
        txn.rollback().await.map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to roll back dry-run transaction: {}", pg_detail(&e)),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;
        return Ok(results);
    }

    txn.commit().await.map_err(|e| FraiseQLError::Database {
        message:   format!("Failed to commit mutation{mechanism} transaction: {}", pg_detail(&e)),
        sql_state: e.code().map(|c| c.code().to_string()),
    })?;
    adapter.mark_write();

    Ok(results)
}

/// PostgreSQL SQLSTATE 42703: undefined column.
const PG_UNDEFINED_COLUMN: &str = "42703";

/// A flexible SQL parameter that binds to any PostgreSQL type.
///
/// Solves the impedance mismatch between `serde_json::Value` (only accepts JSON/JSONB)
/// and `Option<String>` (only accepts text-family types) when binding function-call
/// arguments whose types are resolved at runtime from the function signature.
///
/// Serialisation strategy (binary wire format):
/// - `JSONB`: 1-byte version header (1) + UTF-8 JSON bytes
/// - `JSON`: UTF-8 JSON bytes
/// - `UUID`: 16-byte big-endian UUID
/// - `INT4`: 4-byte big-endian i32
/// - `INT8`: 8-byte big-endian i64
/// - `BOOL`: 1-byte (0 or 1)
/// - All other types: UTF-8 bytes (PostgreSQL text binary = raw UTF-8)
#[derive(Debug)]
enum FlexParam {
    /// SQL NULL — accepted by any PostgreSQL type.
    Null,
    /// A text-encoded value; binary-serialised according to the server-resolved type.
    Text(String),
}

impl tokio_postgres::types::ToSql for FlexParam {
    fn to_sql(
        &self,
        ty: &tokio_postgres::types::Type,
        out: &mut bytes::BytesMut,
    ) -> std::result::Result<tokio_postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>>
    {
        use tokio_postgres::types::{IsNull, Type};
        match self {
            Self::Null => Ok(IsNull::Yes),
            Self::Text(s) => {
                if *ty == Type::JSONB {
                    // JSONB binary wire format: 1-byte version (1) + JSON bytes
                    out.put_u8(1);
                    out.extend_from_slice(s.as_bytes());
                } else if *ty == Type::JSON {
                    out.extend_from_slice(s.as_bytes());
                } else if *ty == Type::UUID {
                    let uuid = uuid::Uuid::parse_str(s)?;
                    out.extend_from_slice(uuid.as_bytes());
                } else if *ty == Type::INT4 {
                    let n: i32 = s.parse()?;
                    out.put_i32(n);
                } else if *ty == Type::INT8 {
                    let n: i64 = s.parse()?;
                    out.put_i64(n);
                } else if *ty == Type::BOOL {
                    let b: bool = s.parse()?;
                    out.put_u8(u8::from(b));
                } else {
                    // TEXT, VARCHAR, BPCHAR, NAME, UNKNOWN, and any user-defined type:
                    // UTF-8 bytes are the binary wire representation for text-family types.
                    out.extend_from_slice(s.as_bytes());
                }
                Ok(IsNull::No)
            },
        }
    }

    fn accepts(_ty: &tokio_postgres::types::Type) -> bool {
        // Accepts all types; per-type serialisation is handled in `to_sql`.
        true
    }

    fn to_sql_checked(
        &self,
        ty: &tokio_postgres::types::Type,
        out: &mut bytes::BytesMut,
    ) -> std::result::Result<tokio_postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>>
    {
        // `accepts()` returns true for all types, so the standard WrongType check is
        // unnecessary.  Delegate directly to `to_sql`.
        self.to_sql(ty, out)
    }
}

/// Enrich a `FraiseQLError::Database` error for PostgreSQL SQLSTATE 42703 (undefined column)
/// when the WHERE clause contains `NativeField` conditions.
///
/// Native columns may be inferred automatically at compile time from `ID`/`UUID`-typed
/// arguments.  If the column does not exist on the target table at runtime, the raw
/// PostgreSQL error is replaced with a diagnostic message that names the native columns
/// involved and explains how to fix the schema.
fn enrich_undefined_column_error(
    err: FraiseQLError,
    view: &str,
    where_clause: Option<&WhereClause>,
) -> FraiseQLError {
    let FraiseQLError::Database { ref sql_state, .. } = err else {
        return err;
    };
    if sql_state.as_deref() != Some(PG_UNDEFINED_COLUMN) {
        return err;
    }
    let native_cols: Vec<&str> =
        where_clause.map(|wc| wc.native_column_names()).unwrap_or_default();
    if native_cols.is_empty() {
        return err;
    }
    FraiseQLError::Database {
        message:   format!(
            "Column(s) {:?} referenced as native column(s) on `{view}` do not exist. \
             These columns were auto-inferred from ID/UUID-typed query arguments. \
             Either add the column(s) to the table/view, or set \
             `native_columns = {{}}` explicitly in your schema to disable inference.",
            native_cols,
        ),
        sql_state: Some(PG_UNDEFINED_COLUMN.to_string()),
    }
}

/// Decodes any PostgreSQL `ENUM` value as its text label.
///
/// `String`'s `FromSql` accepts only the text family (TEXT/VARCHAR/BPCHAR/NAME), not
/// a custom enum OID, so an enum column otherwise falls through [`row_to_map`]'s
/// decode ladder to `Null`. An enum's binary wire representation *is* its label as
/// UTF-8 text, so this reads it directly (#472).
struct PgEnumLabel(String);

impl<'a> FromSql<'a> for PgEnumLabel {
    fn from_sql(
        _ty: &Type,
        raw: &'a [u8],
    ) -> std::result::Result<Self, Box<dyn std::error::Error + Sync + Send>> {
        Ok(Self(std::str::from_utf8(raw)?.to_owned()))
    }

    fn accepts(ty: &Type) -> bool {
        matches!(ty.kind(), Kind::Enum(_))
    }
}

/// Convert a single `tokio_postgres::Row` into a `HashMap<String, serde_json::Value>`.
///
/// Tries each PostgreSQL type in priority order; falls back to `Null` for
/// types that cannot be represented as JSON.
fn row_to_map(row: &Row) -> std::collections::HashMap<String, serde_json::Value> {
    let mut map = std::collections::HashMap::new();
    for (idx, column) in row.columns().iter().enumerate() {
        map.insert(column.name().to_string(), decode_cell(row, idx));
    }
    map
}

/// Decode one cell of a `tokio_postgres::Row` to JSON.
///
/// The decode ladder itself, extracted from [`row_to_map`] so the positional
/// reader the admin SQL console needs (#962) and the name-keyed map every other
/// caller wants cannot disagree about what a `numeric`, an enum or a `timestamptz`
/// becomes. A console keyed by column *name* would also collapse `SELECT 1 AS a,
/// 2 AS a` to one column and silently drop the other.
///
/// Tries each PostgreSQL type in priority order; falls back to `Null` for types
/// that cannot be represented as JSON.
pub(super) fn decode_cell(row: &Row, idx: usize) -> serde_json::Value {
    let column = &row.columns()[idx];
    if let Ok(v) = row.try_get::<_, i16>(idx) {
        // SMALLINT/int2 columns (e.g. `app.mutation_response.http_status`).
        // Without this branch a non-null int2 fell through to `Null` because
        // `FromSql for i32` rejects int2 — which dropped `MutationError`'s
        // `httpStatus` to absent. Order among i16/i32/i64 is correctness-
        // neutral (each `FromSql` accepts only its own PG int width), so this
        // sits first for readability.
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, i32>(idx) {
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, i64>(idx) {
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, f64>(idx) {
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, String>(idx) {
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, bool>(idx) {
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, PgNumericText>(idx) {
        // NUMERIC/DECIMAL — decoded to PostgreSQL's exact text rendering
        // (#980), then re-parsed as a JSON number so the value isn't forced
        // through f64 at construction. Without this branch a `SUM(revenue)`
        // aggregate (and every raw NUMERIC column) fell through to `Null`
        // (H35). Falls back to a JSON string for the values whose text is
        // not JSON numeric syntax: NaN, Infinity and -Infinity.
        let s = v.0;
        serde_json::from_str::<serde_json::Value>(&s).unwrap_or(serde_json::Value::String(s))
    } else if let Ok(v) = row.try_get::<_, uuid::Uuid>(idx) {
        // UUID columns (e.g. `app.mutation_response.entity_id`) render as the
        // canonical hyphenated string instead of silently nulling (H35).
        serde_json::json!(v.to_string())
    } else if let Ok(v) = row.try_get::<_, chrono::DateTime<chrono::Utc>>(idx) {
        // `timestamptz` → RFC 3339 / ISO 8601 text (H35).
        serde_json::json!(v.to_rfc3339())
    } else if let Ok(v) = row.try_get::<_, chrono::NaiveDateTime>(idx) {
        // `timestamp` (no tz) → ISO 8601 text (H35).
        serde_json::json!(v.to_string())
    } else if let Ok(v) = row.try_get::<_, chrono::NaiveDate>(idx) {
        // `date` → ISO 8601 date text (H35).
        serde_json::json!(v.to_string())
    } else if let Ok(v) = row.try_get::<_, Vec<String>>(idx) {
        // TEXT[] columns (e.g. `app.mutation_response.updated_fields`) — without
        // this branch a non-null text array falls through to Null and a parser
        // expecting a sequence fails. Must precede the jsonb branch (a jsonb
        // column never deserializes as Vec<String>, so ordering is safe).
        serde_json::json!(v)
    } else if let Ok(v) = row.try_get::<_, PgEnumLabel>(idx) {
        // ENUM columns (e.g. `app.mutation_response.error_class`, the
        // `app.mutation_error_class` enum) render as their text label. Without
        // this branch a non-null enum fell through to `Null` (`String: FromSql`
        // rejects a custom enum OID), which dropped `error_class` to absent on a
        // failed mutation — and the parser then rejected the row with
        // "succeeded=false requires error_class", so the typed error arm was
        // never reached (#472). `PgEnumLabel` accepts only `Kind::Enum`, so this
        // never shadows another branch regardless of position.
        serde_json::json!(v.0)
    } else if let Ok(v) = row.try_get::<_, serde_json::Value>(idx) {
        v
    } else {
        // A column whose PostgreSQL type none of the branches above can
        // decode. Name the column and its type so the next drift is visible
        // in logs instead of a silent null (H35).
        tracing::warn!(
            column = %column.name(),
            pg_type = %column.type_(),
            "decode_cell: column type not representable as JSON; returning null"
        );
        serde_json::Value::Null
    }
}

/// Apply transaction-local session variables on an in-progress transaction.
///
/// Each `(name, value)` pair is set with `SELECT set_config($1, $2, true)`, so
/// the values are scoped to `txn` and visible to every statement run on it
/// (fixes #329). The caller is responsible for committing or rolling back.
pub(super) async fn apply_session_vars(
    txn: &tokio_postgres::Transaction<'_>,
    session_vars: &[(&str, &str)],
) -> Result<()> {
    for (name, value) in session_vars {
        // A var carrying the clock-timestamp directive (e.g. fraiseql.started_at)
        // is stamped with the DB clock at apply time, on this very transaction, so
        // the start timestamp and the close-of-interval at the outbox write share
        // one clock (no app↔DB skew). All other vars bind their literal value.
        if *value == crate::changelog::CLOCK_TIMESTAMP_DIRECTIVE {
            txn.execute("SELECT set_config($1, clock_timestamp()::text, true)", &[name])
                .await
                .map_err(|e| FraiseQLError::Database {
                    message:   format!(
                        "set_config({name:?}, clock_timestamp()) failed: {}",
                        pg_detail(&e)
                    ),
                    sql_state: e.code().map(|c| c.code().to_string()),
                })?;
        } else {
            txn.execute("SELECT set_config($1, $2, true)", &[name, value])
                .await
                .map_err(|e| FraiseQLError::Database {
                    message:   format!("set_config({name:?}) failed: {}", pg_detail(&e)),
                    sql_state: e.code().map(|c| c.code().to_string()),
                })?;
        }
    }
    Ok(())
}

/// Mark the in-progress mutation transaction as **FraiseQL-mediated** (#366).
///
/// Sets [`crate::changelog::CDC_MEDIATED_VAR`] to
/// [`crate::changelog::CDC_MEDIATED_ON`] transaction-locally
/// (`set_config(..., true)`), so the value auto-resets on commit and is invisible
/// to other connections. The shipped fallback-capture trigger reads it and
/// suppresses its own change-log row for this write, so an app-path mutation —
/// already logged by the in-transaction outbox — is never double-captured. Set
/// on every mutation transaction (independent of whether the outbox row is
/// written), so an opted-out mutation (`changelog=false`) also suppresses the
/// trigger rather than leaving a degraded fallback row.
pub(super) async fn mark_cdc_mediated(txn: &tokio_postgres::Transaction<'_>) -> Result<()> {
    txn.execute(
        "SELECT set_config($1, $2, true)",
        &[
            &crate::changelog::CDC_MEDIATED_VAR,
            &crate::changelog::CDC_MEDIATED_ON,
        ],
    )
    .await
    .map_err(|e| FraiseQLError::Database {
        message:   format!(
            "Failed to set {} marker: {}",
            crate::changelog::CDC_MEDIATED_VAR,
            pg_detail(&e)
        ),
        sql_state: e.code().map(|c| c.code().to_string()),
    })?;
    Ok(())
}

/// Build the single statement that runs a mutation function and writes its
/// `core.tb_entity_change_log` outbox row in the same transaction (the Change
/// Spine transactional outbox).
///
/// The function call is materialised once (`WITH r AS MATERIALIZED`, so a
/// volatile mutation function executes **exactly once** even though two CTEs
/// read it), a data-modifying CTE INSERTs the change-log row for an effective
/// change (`succeeded AND state_changed`), and the primary `SELECT * FROM r`
/// returns the function's row unchanged to the caller. The changed-entity
/// columns (`object_id`, `object_data`, `updated_fields`, `cascade`) come from
/// the function's own `app.mutation_response` row; `$<n+1>` is the NOT-NULL
/// `object_type` fallback, `$<n+2>` is the `modification_type` verb,
/// `$<n+3>` is the envelope `tenant_id` (the Trinity public-facing UUID, NULL
/// for system / unauthenticated / non-UUID-tenant rows), `$<n+4>` is the
/// `trace_id` (the request's W3C trace id, plain text), `$<n+5>` is the
/// `schema_version` (the compiled schema's content hash, plain text), `$<n+6>`
/// is the `trace_context` (the full W3C trace context, cast `::jsonb`), and
/// `$<n+9>` is the ingress `transport` (#376), merged into `extra_metadata` as
/// `{"transport": …}` by a deterministic `CASE` (NULL omits the key, so the SQL
/// text never depends on the value), where `n` = `n_args` (the number of
/// function arguments). The envelope params are
/// appended **after** the function args so the SQL text stays deterministic per
/// `(function, n_args)` — `prepare_cached` keys the statement by text, so the
/// column list and placeholder positions must never depend on the param *values*.
///
/// `started_at`/`duration_ms` read `fraiseql.started_at` (set txn-locally by the
/// caller) on the DB clock — the canonical computation from
/// [`crate::changelog::duration_ms_sql`], stamped with
/// [`crate::changelog::DURATION_CALC_VERSION`]. `commit_time` is the DB clock at
/// INSERT; `seq` is omitted so the column's `SEQUENCE` default fires (so any
/// INSERTer, incl. cooperative external producers, gets a monotonic value).
pub(super) fn build_changelog_cte_sql(quoted_fn: &str, n_args: usize, pre_image: bool) -> String {
    let placeholders: Vec<String> = (1..=n_args).map(|i| format!("${i}")).collect();
    let object_type_idx = n_args + 1;
    let modification_type_idx = n_args + 2;
    let tenant_id_idx = n_args + 3;
    let trace_id_idx = n_args + 4;
    let schema_version_idx = n_args + 5;
    let trace_context_idx = n_args + 6;
    let actor_type_idx = n_args + 7;
    let acting_for_idx = n_args + 8;
    let transport_idx = n_args + 9;
    let started_var = crate::changelog::STARTED_AT_VAR;
    let duration = crate::changelog::duration_ms_sql(started_var);
    let calc_version = crate::changelog::DURATION_CALC_VERSION;
    // Pre-image (opt-in, #changelog_pre_image): also record the changed entity's
    // before-state from the function's own `entity_before` column, alongside the
    // after-image `entity`. Adds `object_data_before` to the INSERT and
    // `r.entity_before` to the SELECT ONLY when requested — `object_data` stays
    // the after-image for every consumer; an absent flag emits today's SQL
    // byte-for-byte. No new placeholder: the before-image is a column on `r`, like
    // `r.entity`. The two SQL forms differ in text, so `prepare_cached` keys them
    // as separate statements automatically (no extra cache discriminator needed).
    let after_before_cols = if pre_image {
        "object_data, object_data_before"
    } else {
        "object_data"
    };
    let after_before_vals = if pre_image {
        "r.entity, r.entity_before"
    } else {
        "r.entity"
    };
    format!(
        "WITH r AS MATERIALIZED (SELECT * FROM {quoted_fn}({args})), \
         _changelog AS ( \
           INSERT INTO core.tb_entity_change_log \
             (object_type, modification_type, object_id, {after_before_cols}, \
              updated_fields, cascade, started_at, duration_ms, extra_metadata, \
              tenant_id, trace_id, schema_version, trace_context, \
              actor_type, acting_for, commit_time) \
           SELECT \
             COALESCE(r.entity_type, ${object_type_idx}), \
             ${modification_type_idx}, \
             r.entity_id, {after_before_vals}, r.updated_fields, r.cascade, \
             current_setting('{started_var}')::timestamptz, \
             {duration}, \
             jsonb_build_object('duration_calc_version', {calc_version}::int) \
               || CASE WHEN ${transport_idx}::text IS NULL THEN '{{}}'::jsonb \
                       ELSE jsonb_build_object('transport', ${transport_idx}::text) END, \
             ${tenant_id_idx}::uuid, \
             ${trace_id_idx}, \
             ${schema_version_idx}, \
             ${trace_context_idx}::jsonb, \
             ${actor_type_idx}, \
             ${acting_for_idx}::uuid, \
             clock_timestamp() \
           FROM r \
           WHERE r.succeeded AND r.state_changed \
           RETURNING 1 \
         ) \
         SELECT * FROM r",
        args = placeholders.join(", "),
    )
}

/// Prepare `sql` on `client` using deadpool's **per-connection statement cache**.
///
/// The mutation function-call path sends the same statement (a fixed
/// `SELECT * FROM fn(...)` or the change-log CTE) on every call, so without
/// caching PostgreSQL re-parses and re-plans it for each mutation — the dominant
/// hot-path cost (the complex outbox CTE in particular). `prepare_cached` keys
/// the parsed/planned [`tokio_postgres::Statement`] by SQL text per connection
/// and reuses it across calls, so the parse/plan happens once per connection.
///
/// Returns an owned `Statement` (releasing the `&client` borrow), so it can be
/// prepared before `client.build_transaction()` and used inside that transaction.
async fn prepare_cached_stmt(
    client: &deadpool_postgres::Client,
    sql: &str,
) -> Result<tokio_postgres::Statement> {
    client.prepare_cached(sql).await.map_err(|e| {
        // Surface the underlying PostgreSQL diagnostic (SQLSTATE message, e.g.
        // `function "foo" does not exist`) rather than the opaque `db error` the
        // top-level `Display` renders (#451, generalised to every query path by #888).
        let detail = pg_detail(&e);
        let hint = changelog_prepare_hint(&detail);
        FraiseQLError::Database {
            message:   format!("Failed to prepare statement: {detail}{hint}"),
            sql_state: e.code().map(|c| c.code().to_string()),
        }
    })
}

/// Actionable operator guidance for the two prepare failures every freshly
/// authored stack hits (#569). The bare PostgreSQL diagnostics — `relation
/// "core.tb_entity_change_log" does not exist` and `column r.entity_type does
/// not exist` — point at neither of the actual requirements: the change-log
/// outbox table must be installed (`fraiseql setup`), and every mutation
/// function must return the 13-column v2.2 `mutation_response` row the outbox
/// CTE reads (`r.` is that CTE's alias for the function result, so a missing
/// `r.*` column means the function returns some other shape — classically
/// `RETURNS SETOF v_*`).
fn changelog_prepare_hint(detail: &str) -> &'static str {
    if detail.contains("core.tb_entity_change_log") {
        "\nhint: the mutation change-log outbox table is not installed in this database — \
         run `fraiseql setup` (or apply the fraiseql-observers change-log contract \
         migration) before executing mutations"
    } else if detail.contains("column r.") {
        "\nhint: the mutation function does not return the v2.2 mutation_response row \
         (13 columns: succeeded, state_changed, …) that the change-log outbox reads. \
         `RETURNS SETOF v_*` and bare scalar returns are not supported — build the row \
         with fraiseql.mutation_ok()/fraiseql.mutation_err() (installed by `fraiseql \
         setup`); see docs/architecture/mutation-response.md"
    } else {
        ""
    }
}

// Reason: DatabaseAdapter is defined with #[async_trait]; all implementations must match
// its transformed method signatures to satisfy the trait contract
// async_trait: dyn-dispatch required; remove when RTN + Send is stable (RFC 3425)
#[async_trait]
impl DatabaseAdapter for PostgresAdapter {
    fn serves_reads_from_standbys(&self) -> bool {
        self.read_replicas.is_some()
    }

    async fn execute_with_projection(
        &self,
        view: &str,
        projection: Option<&SqlProjectionHint>,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
        order_by: Option<&[OrderByClause]>,
    ) -> Result<Vec<JsonbValue>> {
        self.execute_with_projection_impl(view, projection, where_clause, limit, offset, order_by)
            .await
    }

    async fn execute_where_query(
        &self,
        view: &str,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
        order_by: Option<&[OrderByClause]>,
    ) -> Result<Vec<JsonbValue>> {
        let (sql, typed_params) =
            build_where_select_sql_ordered(view, where_clause, limit, offset, order_by)?;

        let param_refs = crate::types::as_sql_param_refs(&typed_params);

        self.execute_raw(&sql, &param_refs, ReadRouting::Any)
            .await
            .map_err(|e| enrich_undefined_column_error(e, view, where_clause))
    }

    async fn explain_where_query(
        &self,
        view: &str,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
    ) -> Result<serde_json::Value> {
        let (select_sql, typed_params) = build_where_select_sql(view, where_clause, limit, offset)?;
        // Defense-in-depth: compiler-generated SQL should never contain a
        // semicolon, but guard against it to prevent statement injection.
        if select_sql.contains(';') {
            return Err(FraiseQLError::Validation {
                message: "EXPLAIN SQL must be a single statement".into(),
                path:    None,
            });
        }
        let explain_sql = format!("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {select_sql}");

        let param_refs = crate::types::as_sql_param_refs(&typed_params);

        // EXPLAIN ANALYZE of a compiled SELECT: read-only, and most truthful when
        // it runs where the reads it describes run — the replica route (#407).
        let client = self.acquire_read_connection_with_retry(ReadRouting::Any).await?;
        let rows = client.query(explain_sql.as_str(), &param_refs).await.map_err(|e| {
            FraiseQLError::Database {
                message:   format!("EXPLAIN ANALYZE failed: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            }
        })?;

        if let Some(row) = rows.first() {
            let plan: serde_json::Value = row.try_get(0).map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to parse EXPLAIN output: {e}"),
                sql_state: None,
            })?;
            Ok(plan)
        } else {
            Ok(serde_json::Value::Null)
        }
    }

    fn database_type(&self) -> DatabaseType {
        DatabaseType::PostgreSQL
    }

    /// `execute_composed_with_session` is implemented below.
    fn supports_composed_reads(&self) -> bool {
        true
    }

    async fn health_check(&self) -> Result<()> {
        // Use retry logic for health check to avoid false negatives during pool exhaustion
        let client = self.acquire_connection_with_retry().await?;

        client.query("SELECT 1", &[]).await.map_err(|e| FraiseQLError::Database {
            message:   format!("Health check failed: {}", pg_detail(&e)),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        Ok(())
    }

    #[allow(clippy::cast_possible_truncation)] // Reason: value is bounded; truncation cannot occur in practice
    fn pool_metrics(&self) -> PoolMetrics {
        let status = self.pool.status();

        PoolMetrics {
            total_connections:  status.size as u32,
            idle_connections:   status.available as u32,
            active_connections: (status.size - status.available) as u32,
            waiting_requests:   status.waiting as u32,
        }
    }

    /// # Security
    ///
    /// `sql` **must** be compiler-generated. Never pass user-supplied strings
    /// directly — doing so would open SQL-injection vulnerabilities.
    async fn execute_raw_query(
        &self,
        sql: &str,
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        // Use retry logic for connection acquisition
        let client = self.acquire_connection_with_retry().await?;

        let rows: Vec<Row> = client.query(sql, &[]).await.map_err(|e| FraiseQLError::Database {
            message:   format!("Query execution failed: {}", pg_detail(&e)),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        // Convert each row to HashMap<String, Value>
        let results: Vec<std::collections::HashMap<String, serde_json::Value>> =
            rows.iter().map(row_to_map).collect();

        Ok(results)
    }

    async fn execute_parameterized_aggregate(
        &self,
        sql: &str,
        params: &[serde_json::Value],
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        // Convert serde_json::Value params to QueryParam so that strings are bound
        // as TEXT (not JSONB), which is required for correct WHERE comparisons against
        // data->>'field' expressions that return TEXT.
        let typed: Vec<QueryParam> = params.iter().cloned().map(QueryParam::from).collect();
        let param_refs = crate::types::as_sql_param_refs(&typed);

        // Compiled aggregate SELECT: read-only, replica-eligible (#407).
        let client = self.acquire_read_connection_with_retry(ReadRouting::Any).await?;
        let rows: Vec<Row> =
            client.query(sql, &param_refs).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Parameterized aggregate query failed: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;

        let results: Vec<std::collections::HashMap<String, serde_json::Value>> =
            rows.iter().map(row_to_map).collect();

        Ok(results)
    }

    async fn execute_parameterized_aggregate_with_session(
        &self,
        sql: &str,
        params: &[serde_json::Value],
        session_vars: &[(&str, &str)],
        routing: ReadRouting,
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        if session_vars.is_empty() {
            return self.execute_parameterized_aggregate(sql, params).await;
        }

        let typed: Vec<QueryParam> = params.iter().cloned().map(QueryParam::from).collect();
        let param_refs = crate::types::as_sql_param_refs(&typed);

        // Read-only aggregate with session vars; standby-safe → replica-eligible.
        let mut client = self.acquire_read_connection_with_retry(routing).await?;
        let txn =
            client.build_transaction().start().await.map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to start session-var transaction: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;
        apply_session_vars(&txn, session_vars).await?;
        let rows: Vec<Row> =
            txn.query(sql, &param_refs).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Parameterized aggregate query failed: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;
        txn.commit().await.map_err(|e| FraiseQLError::Database {
            message:   format!("Failed to commit session-var transaction: {}", pg_detail(&e)),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        Ok(rows.iter().map(row_to_map).collect())
    }

    // PostgreSQL session variables are applied connection-affinely by the
    // `*_with_session` methods below: `set_config(..., true)` and the operation
    // share one transaction on one connection, so transaction-local GUCs are
    // visible to the function / view (fixes #329).

    async fn execute_where_query_arc_with_session(
        &self,
        view: &str,
        where_clause: Option<&WhereClause>,
        limit: Option<u32>,
        offset: Option<u32>,
        order_by: Option<&[OrderByClause]>,
        session_vars: &[(&str, &str)],
        routing: ReadRouting,
    ) -> Result<Arc<Vec<JsonbValue>>> {
        let (sql, typed_params) =
            build_where_select_sql_ordered(view, where_clause, limit, offset, order_by)?;
        let param_refs = crate::types::as_sql_param_refs(&typed_params);

        // No session variables => no transaction wrapper, but still this query's
        // routing. The fast path used to delegate to `execute_where_query_arc`,
        // which has no routing parameter, so every read that happened not to carry
        // session variables silently reverted to server policy — `read_routing`
        // accepted and dropped on the most common shape there is.
        if session_vars.is_empty() {
            return self
                .execute_raw(&sql, &param_refs, routing)
                .await
                .map(Arc::new)
                .map_err(|e| enrich_undefined_column_error(e, view, where_clause));
        }

        self.execute_raw_with_session(&sql, &param_refs, session_vars, routing)
            .await
            .map(Arc::new)
            .map_err(|e| enrich_undefined_column_error(e, view, where_clause))
    }

    /// `SELECT COUNT(*)`, pushed into the database rather than counted in Rust.
    ///
    /// The trait's default fetches every matching row and takes `.len()`, which
    /// is correct but materialises the whole filtered set — on an unfiltered
    /// list that is the entire table in memory to answer a single integer.
    async fn count_where_query(
        &self,
        view: &str,
        where_clause: Option<&WhereClause>,
        session_vars: &[(&str, &str)],
        routing: ReadRouting,
    ) -> Result<u64> {
        let (sql, typed_params) = super::build_count_sql(view, where_clause)?;
        let param_refs = crate::types::as_sql_param_refs(&typed_params);

        // Structurally read-only: replica-eligible, and `set_config` + SELECT are
        // both fine inside a hot-standby read-only transaction (#407).
        let row = if session_vars.is_empty() {
            let client = self.acquire_read_connection_with_retry(routing).await?;
            client.query_one(&sql, &param_refs).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Count query failed: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?
        } else {
            // Session variables must land on the *same* connection as the count,
            // or an RLS-filtered page gets an unfiltered total beside it (#329).
            let mut client = self.acquire_read_connection_with_retry(routing).await?;
            let txn =
                client.build_transaction().start().await.map_err(|e| FraiseQLError::Database {
                    message:   format!(
                        "Failed to start count session-var transaction: {}",
                        pg_detail(&e)
                    ),
                    sql_state: e.code().map(|c| c.code().to_string()),
                })?;
            super::database::apply_session_vars(&txn, session_vars).await?;
            let row =
                txn.query_one(&sql, &param_refs).await.map_err(|e| FraiseQLError::Database {
                    message:   format!("Count query failed: {}", pg_detail(&e)),
                    sql_state: e.code().map(|c| c.code().to_string()),
                })?;
            txn.commit().await.map_err(|e| FraiseQLError::Database {
                message:   format!(
                    "Failed to commit count session-var transaction: {}",
                    pg_detail(&e)
                ),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;
            row
        };

        let total: i64 = row.get(0);
        // COUNT(*) is never negative, so the bit pattern is the value; `cast_unsigned`
        // states that intent where `as u64` would only imply it.
        Ok(total.cast_unsigned())
    }

    async fn execute_with_projection_arc_with_session(
        &self,
        request: &ProjectionRequest<'_>,
        session_vars: &[(&str, &str)],
        routing: ReadRouting,
    ) -> Result<Arc<Vec<JsonbValue>>> {
        // No projection => behave like a plain WHERE query, matching
        // execute_with_projection_impl's fallback.
        let Some(projection) = request.projection else {
            return self
                .execute_where_query_arc_with_session(
                    request.view,
                    request.where_clause,
                    request.limit,
                    request.offset,
                    request.order_by,
                    session_vars,
                    routing,
                )
                .await;
        };

        let (sql, typed_params) = build_projection_select_sql(
            projection,
            request.view,
            request.where_clause,
            request.limit,
            request.offset,
            request.order_by,
        )?;
        let param_refs = crate::types::as_sql_param_refs(&typed_params);

        // See `execute_where_query_arc_with_session`: the session-free path has to
        // carry the routing too.
        if session_vars.is_empty() {
            return self.execute_raw(&sql, &param_refs, routing).await.map(Arc::new);
        }

        self.execute_raw_with_session(&sql, &param_refs, session_vars, routing)
            .await
            .map(Arc::new)
    }

    /// One statement, one snapshot, one connection held for the delivery (#958).
    ///
    /// The SQL is built by the same two builders the collecting read uses, so a
    /// streamed export and a buffered read of the same query cannot select
    /// different rows — the drift that made `execute_query_direct` disagree with
    /// `count_rows` about a tenant filter (#739).
    async fn execute_composed_with_session(
        &self,
        read: &ComposedLevel,
        session_vars: &[(&str, &str)],
        routing: ReadRouting,
    ) -> Result<Arc<Vec<JsonbValue>>> {
        let (sql, typed_params) = build_composed_select_sql(read)?;
        let param_refs = crate::types::as_sql_param_refs(&typed_params);

        // See `execute_where_query_arc_with_session`: the session-free path has to
        // carry the routing too.
        if session_vars.is_empty() {
            return self.execute_raw(&sql, &param_refs, routing).await.map(Arc::new);
        }

        self.execute_raw_with_session(&sql, &param_refs, session_vars, routing)
            .await
            .map(Arc::new)
    }

    async fn stream_with_projection(
        &self,
        request: &ProjectionRequest<'_>,
        session_vars: &[(&str, &str)],
        routing: ReadRouting,
    ) -> Result<JsonbRowStream> {
        let (sql, typed_params) = match request.projection {
            Some(projection) => build_projection_select_sql(
                projection,
                request.view,
                request.where_clause,
                request.limit,
                request.offset,
                request.order_by,
            )?,
            None => build_where_select_sql_ordered(
                request.view,
                request.where_clause,
                request.limit,
                request.offset,
                request.order_by,
            )?,
        };

        super::streaming::stream_rows(self, sql, typed_params, session_vars, routing, |row, sql| {
            jsonb_cell(row, 0, sql)
        })
        .await
        .map_err(|e| enrich_undefined_column_error(e, request.view, request.where_clause))
    }

    /// The column-shaped streaming read behind gRPC server-streaming (#958).
    ///
    /// Routed to the primary like its collecting sibling: `execute_row_query`
    /// reaches `execute_raw_query`, and a surface that executes composed SQL is
    /// never sent to a replica.
    async fn stream_row_query(
        &self,
        view_name: &str,
        columns: &[crate::types::ColumnSpec],
        where_sql: Option<&str>,
        order_by: Option<&str>,
        limit: Option<u32>,
        offset: Option<u32>,
    ) -> Result<ColumnRowStream> {
        let sql = crate::traits::build_row_query_sql(view_name, where_sql, order_by, limit, offset);
        let columns = columns.to_vec();

        super::streaming::stream_rows(
            self,
            sql,
            Vec::new(),
            &[],
            ReadRouting::Primary,
            move |row, _sql| Ok(crate::traits::row_to_column_values(&row_to_map(row), &columns)),
        )
        .await
    }

    /// The session-pinned collecting row read (#1351).
    ///
    /// Same shape as [`count_where_query`](Self::count_where_query): with no session
    /// variables it is a plain read; with them it is one transaction, so `set_config`
    /// and the SELECT land on the same connection and an RLS policy backed by
    /// `current_setting()` sees them (#329).
    async fn execute_row_query_with_session(
        &self,
        view_name: &str,
        columns: &[crate::types::ColumnSpec],
        where_sql: Option<&str>,
        order_by: Option<&str>,
        limit: Option<u32>,
        offset: Option<u32>,
        session_vars: &[(&str, &str)],
    ) -> Result<Vec<Vec<crate::types::ColumnValue>>> {
        if session_vars.is_empty() {
            return self
                .execute_row_query(view_name, columns, where_sql, order_by, limit, offset)
                .await;
        }

        let sql = crate::traits::build_row_query_sql(view_name, where_sql, order_by, limit, offset);

        let mut client = self.acquire_connection_with_retry().await?;
        let txn =
            client.build_transaction().start().await.map_err(|e| FraiseQLError::Database {
                message:   format!(
                    "Failed to start row-read session-var transaction: {}",
                    pg_detail(&e)
                ),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;
        apply_session_vars(&txn, session_vars).await?;
        let rows: Vec<Row> =
            txn.query(sql.as_str(), &[]).await.map_err(|e| FraiseQLError::Database {
                message:   format!("Row query execution failed: {}", pg_detail(&e)),
                sql_state: e.code().map(|c| c.code().to_string()),
            })?;
        txn.commit().await.map_err(|e| FraiseQLError::Database {
            message:   format!(
                "Failed to commit row-read session-var transaction: {}",
                pg_detail(&e)
            ),
            sql_state: e.code().map(|c| c.code().to_string()),
        })?;

        Ok(rows
            .iter()
            .map(|row| crate::traits::row_to_column_values(&row_to_map(row), columns))
            .collect())
    }

    /// The session-pinned streaming row read (#1351).
    ///
    /// `stream_rows` already pins session variables for the life of the portal —
    /// the non-session arm above passes `&[]` to the very same call.
    async fn stream_row_query_with_session(
        &self,
        view_name: &str,
        columns: &[crate::types::ColumnSpec],
        where_sql: Option<&str>,
        order_by: Option<&str>,
        limit: Option<u32>,
        offset: Option<u32>,
        session_vars: &[(&str, &str)],
    ) -> Result<ColumnRowStream> {
        let sql = crate::traits::build_row_query_sql(view_name, where_sql, order_by, limit, offset);
        let columns = columns.to_vec();

        super::streaming::stream_rows(
            self,
            sql,
            Vec::new(),
            session_vars,
            ReadRouting::Primary,
            move |row, _sql| Ok(crate::traits::row_to_column_values(&row_to_map(row), &columns)),
        )
        .await
    }

    async fn explain_query(
        &self,
        sql: &str,
        _params: &[serde_json::Value],
    ) -> Result<serde_json::Value> {
        // Defense-in-depth: reject multi-statement input even though this SQL is
        // compiler-generated. A semicolon would allow a second statement to be
        // appended to the EXPLAIN prefix.
        if sql.contains(';') {
            return Err(FraiseQLError::Validation {
                message: "EXPLAIN SQL must be a single statement".into(),
                path:    None,
            });
        }
        let explain_sql = format!("EXPLAIN (ANALYZE false, FORMAT JSON) {sql}");
        // Read-only (plan only, ANALYZE false): replica-eligible (#407).
        let client = self.acquire_read_connection_with_retry(ReadRouting::Any).await?;
        let rows: Vec<Row> =
            client
                .query(explain_sql.as_str(), &[])
                .await
                .map_err(|e| FraiseQLError::Database {
                    message:   format!("EXPLAIN failed: {}", pg_detail(&e)),
                    sql_state: e.code().map(|c| c.code().to_string()),
                })?;

        if let Some(row) = rows.first() {
            let plan: serde_json::Value = row.try_get(0).map_err(|e| FraiseQLError::Database {
                message:   format!("Failed to parse EXPLAIN output: {e}"),
                sql_state: None,
            })?;
            Ok(plan)
        } else {
            Ok(serde_json::Value::Null)
        }
    }

    async fn query_stats(&self, limit: u32) -> Result<Vec<crate::types::QueryStatEntry>> {
        self.pg_query_stats(limit).await
    }

    async fn query_stats_by_id(&self, id: &str) -> Result<Option<crate::types::QueryStatEntry>> {
        self.pg_query_stats_by_id(id).await
    }

    async fn reset_query_stats(&self) -> Result<()> {
        self.pg_reset_query_stats().await
    }

    /// The operator SQL console (#962). The containment lives in the private
    /// `postgres::adapter::admin_sql` module; see [`AdminSqlRequest`] for what
    /// each bound is and how PostgreSQL — not a parse of the statement text —
    /// enforces it.
    ///
    /// [`AdminSqlRequest`]: crate::traits::AdminSqlRequest
    async fn execute_admin_sql(
        &self,
        request: &crate::traits::AdminSqlRequest,
    ) -> Result<crate::traits::AdminSqlOutcome> {
        super::admin_sql::execute_admin_sql_impl(self, request).await
    }
}

// async_trait: dyn-dispatch required; remove when RTN + Send is stable (RFC 3425)
#[async_trait]
impl Writer for PostgresAdapter {
    async fn execute_write(
        &self,
        request: &crate::traits::WriteRequest<'_>,
        gate: crate::traits::MutationRowGate<'_>,
    ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
        run_function_in_txn(self, request, gate).await
    }
}