Add backend-agnostic Rust-native query helpers over the pool

Introduce a small enum-based layer so simple queries can be written once
regardless of backend: DbPool/DbConn dispatch over each backend's own
pool, and DbConn::query(sql, params) binds positional `?` placeholders
and returns rows as backend-agnostic DbValue cells (read out with
DbRowExt::try_get). No trait/async_trait machinery — a closed enum keeps
the pool.get().await? / conn.query(...).await? surface simple.

The Postgres arm rewrites `?` to `$1, $2, ...` (matching
convert_param_style), reuses PgValue for parameter binding, and adds a
Rust-native DbValueFromSql decoder (the non-Python counterpart of
PythonPgFromSql). The shared column-type list is factored into
accepts_column_type so the three mappings can't drift.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

rust/src/database/postgres/mod.rs

@@ -18,6 +18,7 @@ mod errors;
 mod helpers;
 mod libpq;
 pub mod pool;
+pub(crate) mod query;
 mod value;
 
 /// Register the `postgres` submodule (the `Connection` / `Cursor` classes, the

rust/src/database/postgres/pool.rs

@@ -13,7 +13,7 @@
 //! *same* pool, so both share a single set of connections rather than running
 //! two pools that could exhaust the server's connection limit between them.
 
-use deadpool::managed::{Manager, Metrics, Pool, RecycleError, RecycleResult};
+use deadpool::managed::{Manager, Metrics, Object, Pool, RecycleError, RecycleResult};
 use log::warn;
 use tokio_postgres::{Client, Config, NoTls};
 
@@ -71,6 +71,10 @@ impl Manager for ConnectionManager {
 /// A pool of [`tokio_postgres`] connections.
 pub type ConnectionPool = Pool<ConnectionManager>;
 
+/// A connection checked out of a [`ConnectionPool`]. Dereferences to the
+/// underlying [`tokio_postgres::Client`]; returned to the pool when dropped.
+pub type PooledConnection = Object<ConnectionManager>;
+
 /// Build a [`ConnectionPool`] from a libpq-style DSN, capped at `max_size`
 /// connections.
 pub fn create_pool(dsn: &str, max_size: usize) -> Result<ConnectionPool, anyhow::Error> {

rust/src/database/postgres/query.rs

@@ -0,0 +1,82 @@
+//! The Rust-native `query` helper for the Postgres backend.
+//!
+//! This is the Postgres arm of [`crate::database::DbConn::query`]: it binds
+//! backend-agnostic [`DbValue`] parameters and decodes rows back into
+//! [`DbRow`]s, so simple queries can be written once regardless of backend.
+
+use tokio_postgres::types::ToSql;
+use tokio_postgres::Client;
+
+use crate::database::postgres::value::{pg_row_to_db_row, PgValue};
+use crate::database::value::{DbRow, DbValue};
+
+/// Run `sql` (with `?` placeholders bound to `params`) and return all rows.
+///
+/// Parameters are bound positionally, in order, to the `?` placeholders — which
+/// are first rewritten to Postgres' `$1, $2, ...` (see [`convert_placeholders`]).
+pub(crate) async fn query(
+    client: &Client,
+    sql: &str,
+    params: &[DbValue],
+) -> Result<Vec<DbRow>, anyhow::Error> {
+    let sql = convert_placeholders(sql);
+
+    // Bind each agnostic `DbValue` as a Postgres parameter. `PgValue` is the
+    // `ToSql` type the rest of the backend already uses.
+    let pg_params: Vec<PgValue> = params.iter().map(PgValue::from_db_value).collect();
+    let params_dyn: Vec<&(dyn ToSql + Sync)> =
+        pg_params.iter().map(|v| v as &(dyn ToSql + Sync)).collect();
+
+    // Passing the SQL as a `&str` lets `tokio_postgres` prepare it (inferring
+    // the parameter types our `ToSql` impl needs) and run it in one call.
+    let rows = client.query(sql.as_str(), &params_dyn).await?;
+
+    rows.iter().map(pg_row_to_db_row).collect()
+}
+
+/// Rewrite `?` placeholders to Postgres' positional `$1, $2, ...`.
+///
+/// A naive left-to-right substitution, matching Synapse's existing
+/// `PostgresEngine.convert_param_style` (`?` -> `%s`): it does not skip a `?`
+/// inside a string literal, so callers should parameterise rather than embed a
+/// literal `?` in the SQL.
+fn convert_placeholders(sql: &str) -> String {
+    let mut out = String::with_capacity(sql.len() + 8);
+    let mut n = 0u32;
+    for ch in sql.chars() {
+        if ch == '?' {
+            n += 1;
+            out.push('$');
+            out.push_str(&n.to_string());
+        } else {
+            out.push(ch);
+        }
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn convert_placeholders_numbers_each_question_mark() {
+        assert_eq!(
+            convert_placeholders("SELECT * FROM t WHERE a = ? AND b = ?"),
+            "SELECT * FROM t WHERE a = $1 AND b = $2"
+        );
+    }
+
+    #[test]
+    fn convert_placeholders_leaves_sql_without_params_untouched() {
+        assert_eq!(convert_placeholders("SELECT 1"), "SELECT 1");
+    }
+
+    #[test]
+    fn convert_placeholders_counts_past_nine() {
+        // The index is decimal, not a single digit.
+        let sql = "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+        let converted = convert_placeholders(sql);
+        assert!(converted.ends_with("$9, $10, $11)"), "{converted}");
+    }
+}

rust/src/database/postgres/value.rs

@@ -28,13 +28,16 @@ use std::error::Error;
 
 use bytes::BytesMut;
 use postgres_protocol::types::{
-    bool_to_sql, bytea_to_sql, float4_to_sql, float8_to_sql, int2_to_sql, int4_to_sql, int8_to_sql,
-    text_to_sql,
+    bool_from_sql, bool_to_sql, bytea_to_sql, float4_from_sql, float4_to_sql, float8_from_sql,
+    float8_to_sql, int2_from_sql, int2_to_sql, int4_from_sql, int4_to_sql, int8_from_sql,
+    int8_to_sql, text_to_sql,
 };
 use pyo3::exceptions::{PyTypeError, PyValueError};
 use pyo3::types::{PyBool, PyBytes, PyFloat, PyInt, PyString, PyTuple};
 use pyo3::{prelude::*, BoundObject};
-use tokio_postgres::types::{to_sql_checked, IsNull, ToSql, Type, WrongType};
+use tokio_postgres::types::{to_sql_checked, FromSql, IsNull, ToSql, Type, WrongType};
+
+use crate::database::value::{DbRow, DbValue};
 
 /// Owned representation of a Python value that we can hand to [`tokio_postgres`]
 /// as a [`ToSql`] parameter.
@@ -86,6 +89,42 @@ impl PgValue {
             obj.get_type().name()?,
         )))
     }
+
+    /// Bind a backend-agnostic [`DbValue`] (from the Rust-native query helpers)
+    /// as a Postgres parameter. This is the non-Python counterpart of
+    /// [`PgValue::from_py`].
+    pub(crate) fn from_db_value(value: &DbValue) -> Self {
+        match value {
+            DbValue::Null => PgValue::Null,
+            DbValue::Bool(b) => PgValue::Bool(*b),
+            DbValue::Int(i) => PgValue::Int(*i),
+            DbValue::Float(f) => PgValue::Float(*f),
+            DbValue::Text(s) => PgValue::Text(s.as_str().into()),
+            DbValue::Bytes(b) => PgValue::Bytea(b.as_slice().into()),
+        }
+    }
+}
+
+/// The Postgres column types the value mapping supports, in both directions.
+///
+/// Shared by every `accepts` implementation here ([`PgValue`]'s `ToSql`,
+/// [`PythonPgFromSql`] and [`DbValueFromSql`]) and kept in sync with their
+/// `match` arms, so the supported set can't silently drift between them.
+pub(crate) fn accepts_column_type(ty: &Type) -> bool {
+    matches!(
+        *ty,
+        Type::BOOL
+            | Type::INT2
+            | Type::INT4
+            | Type::INT8
+            | Type::FLOAT4
+            | Type::FLOAT8
+            | Type::TEXT
+            | Type::VARCHAR
+            | Type::NAME
+            | Type::BPCHAR
+            | Type::BYTEA
+    )
 }
 
 // Lets PyO3 extract a `PgValue` directly from a Python argument, e.g. when a
@@ -168,20 +207,7 @@ impl ToSql for PgValue {
     }
 
     fn accepts(ty: &Type) -> bool {
-        matches!(
-            *ty,
-            Type::BOOL
-                | Type::INT2
-                | Type::INT4
-                | Type::INT8
-                | Type::FLOAT4
-                | Type::FLOAT8
-                | Type::TEXT
-                | Type::VARCHAR
-                | Type::NAME
-                | Type::BPCHAR
-                | Type::BYTEA
-        )
+        accepts_column_type(ty)
     }
 
     to_sql_checked!();
@@ -229,20 +255,7 @@ impl<'a> tokio_postgres::types::FromSql<'a> for PythonPgFromSql {
     }
 
     fn accepts(ty: &Type) -> bool {
-        matches!(
-            *ty,
-            Type::BOOL
-                | Type::INT2
-                | Type::INT4
-                | Type::INT8
-                | Type::FLOAT4
-                | Type::FLOAT8
-                | Type::TEXT
-                | Type::VARCHAR
-                | Type::NAME
-                | Type::BPCHAR
-                | Type::BYTEA
-        )
+        accepts_column_type(ty)
     }
 }
 
@@ -294,6 +307,63 @@ impl PythonPgFromSql {
     }
 }
 
+/// Convert a Postgres row into a backend-agnostic [`DbRow`] for the Rust-native
+/// query helpers, one [`DbValue`] per column.
+///
+/// The non-Python counterpart of [`pg_row_to_py`]; decodes each column via
+/// [`DbValueFromSql`]. Errors (with the column index and type) if a column can't
+/// be decoded.
+pub(crate) fn pg_row_to_db_row(row: &tokio_postgres::Row) -> Result<DbRow, anyhow::Error> {
+    let mut out = Vec::with_capacity(row.len());
+    for idx in 0..row.len() {
+        let value: DbValueFromSql = row.try_get(idx).map_err(|e| {
+            anyhow::anyhow!(
+                "failed to decode column {idx} (type {}): {e}",
+                row.columns()[idx].type_()
+            )
+        })?;
+        out.push(value.0);
+    }
+    Ok(out)
+}
+
+/// A column value decoded into a backend-agnostic [`DbValue`].
+///
+/// The non-Python counterpart of [`PythonPgFromSql`]: the same supported types
+/// (see [`accepts_column_type`]), but with no GIL and no Python objects — just
+/// the plain Rust value.
+pub(crate) struct DbValueFromSql(pub DbValue);
+
+impl<'a> FromSql<'a> for DbValueFromSql {
+    fn from_sql(ty: &Type, raw: &'a [u8]) -> Result<Self, Box<dyn Error + Sync + Send>> {
+        let value = match *ty {
+            Type::BOOL => DbValue::Bool(bool_from_sql(raw)?),
+            Type::INT2 => DbValue::Int(int2_from_sql(raw)?.into()),
+            Type::INT4 => DbValue::Int(int4_from_sql(raw)?.into()),
+            Type::INT8 => DbValue::Int(int8_from_sql(raw)?),
+            Type::FLOAT4 => DbValue::Float(float4_from_sql(raw)?.into()),
+            Type::FLOAT8 => DbValue::Float(float8_from_sql(raw)?),
+            Type::TEXT | Type::VARCHAR | Type::NAME | Type::BPCHAR => {
+                DbValue::Text(std::str::from_utf8(raw)?.to_owned())
+            }
+            Type::BYTEA => DbValue::Bytes(raw.to_vec()),
+            _ => {
+                // Unreachable unless `accepts` drifts out of sync with this match.
+                return Err(format!("unsupported column type for postgres: {ty}").into());
+            }
+        };
+        Ok(DbValueFromSql(value))
+    }
+
+    fn from_sql_null(_ty: &Type) -> Result<Self, Box<dyn Error + Sync + Send>> {
+        Ok(DbValueFromSql(DbValue::Null))
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        accepts_column_type(ty)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     //! These tests exercise the value mapping in isolation — no live Postgres