//! [`SqlxDbBackend`] — implementation of all 20 `elicit_db` traits via sqlx `AnyPool`.

use std::collections::HashMap;
use std::sync::Arc;

use elicit_db::{
    AccessAuthorized, AnonymousBlockExecuted, AuditLogRetentionMet, AuditLogTamperEvident,
    AuditLogged, BackupConsistent, CheckConstraintDefined, ColumnExists, Committed,
    ConnectionEstablished, ConnectionId, ConstraintSatisfied, DatabaseCreated, DbBackupManager,
    DbColumn, DbConstraintFactory, DbConstraintMeta, DbDatabaseManager, DbError, DbErrorKind,
    DbExplain, DbIndexInfo, DbIndexManager, DbIsolationFactory, DbMonitor, DbPublicationDescriptor,
    DbQueryExecutor, DbReplicationFactory, DbReplicationMeta, DbReplicationSlotDescriptor,
    DbResult, DbRoleInfo, DbRoleManager, DbRoutineDescriptor, DbRoutineFactory, DbRoutineMeta,
    DbRow, DbRows, DbSchema, DbSchemaManager, DbSecurityFactory, DbSecurityMeta, DbServerAdmin,
    DbSessionInfo, DbSessionManager, DbSpatialValue, DbStatActivity, DbSubscriptionDescriptor,
    DbTableInfo, DbTableManager, DbTransactor, DbValue, Durable, EncryptedAtRest,
    EncryptedInTransit, ForeignKeyDefined, FunctionAltered, FunctionCreated, FunctionDropped,
    FunctionParallelRestricted, FunctionParallelSafe, FunctionParallelUnsafe,
    FunctionSecurityDefiner, FunctionSecurityInvoker, IndexExists, IsolationLevel,
    LeastPrivilegeEnforced, LogicalReplicationConfigured, LogicalReplicationSlotCreated,
    MultiFactorAuthEnforced, NotNullConstraintDefined, Open, ParallelSafety,
    PasswordPolicyEnforced, PhysicalReplicationSlotCreated, PrimaryKeyDefined, ProcedureCreated,
    ProcedureDropped, PublicationCreated, ReadCommittedIsolation, ReadUncommittedIsolation,
    RepeatableReadIsolation, ReplicationSlotDropped, ReplicationSlotKind, RolledBack, RoutineKind,
    RowLevelSecurityEnabled, RowLevelSecurityPolicyDefined, RowVisible, SchemaCreated,
    SecurityMode, SerializableIsolation, SessionIsolationLevelSet, SessionTimeoutEnforced,
    SqlInjectionPrevented, SslModeRequired, StreamingReplicationConfigured, SubscriptionCreated,
    TableCreated, TableExists, TransactionCommitted, TransactionHandle,
    TransactionIsolationLevelSet, TransactionReadOnly, TransactionReadWrite,
    TriggerFunctionCreated, TriggerWhenConditionDefined, TxMarker, UniqueConstraintDefined,
    VolatilityKind, WALReplayable, WalLevelLogical, WalLevelReplica,
};
use elicitation::Established;
use futures::future::BoxFuture;
use sqlx::any::AnyRow;
use sqlx::{AnyPool, Column as _, Row as _, TypeInfo as _};
use tokio::sync::Mutex;
use uuid::Uuid;

// ── Transaction storage ───────────────────────────────────────────────────────

/// Holds an acquired pool connection used as a manual transaction.
///
/// `pool.acquire()` returns an owned `PoolConnection<Any>` with no lifetime
/// parameters, allowing it to be stored in a `HashMap` without `unsafe` code.
/// BEGIN / COMMIT / ROLLBACK are issued as plain SQL statements.
struct TxSlot {
    conn: Mutex<Option<sqlx::pool::PoolConnection<sqlx::Any>>>,
    isolation: IsolationLevel,
}

impl TxSlot {
    fn new(conn: sqlx::pool::PoolConnection<sqlx::Any>, isolation: IsolationLevel) -> Self {
        Self {
            conn: Mutex::new(Some(conn)),
            isolation,
        }
    }
}

// ── Error helpers ─────────────────────────────────────────────────────────────

#[track_caller]
fn sqlx_err(e: sqlx::Error) -> DbError {
    DbError::new(DbErrorKind::QueryFailed(e.to_string()))
}

#[track_caller]
fn conn_err(e: sqlx::Error) -> DbError {
    DbError::new(DbErrorKind::ConnectionFailed(e.to_string()))
}

#[track_caller]
fn tx_err(msg: impl Into<String>) -> DbError {
    DbError::new(DbErrorKind::TransactionError(msg.into()))
}

// ── Row conversion ────────────────────────────────────────────────────────────

fn decode_spatial_value(row: &AnyRow, ordinal: usize, type_name: &str) -> DbValue {
    let wrap = |payload| match type_name {
        "GEOGRAPHY" => DbValue::Geography(payload),
        _ => DbValue::Geometry(payload),
    };

    row.try_get::<Vec<u8>, _>(ordinal)
        .map(DbSpatialValue::Wkb)
        .map(wrap)
        .or_else(|_| {
            row.try_get::<String, _>(ordinal)
                .map(DbSpatialValue::Wkt)
                .map(wrap)
        })
        .unwrap_or(DbValue::Null)
}

fn any_row_to_db_row(row: &AnyRow) -> DbRow {
    let cols: Vec<(String, DbValue)> = row
        .columns()
        .iter()
        .map(|col| {
            let name = col.name().to_string();
            let ty = col.type_info().name().to_uppercase();
            let val = match ty {
                ref ty if ty == "BOOL" || ty == "BOOLEAN" => row
                    .try_get::<bool, _>(col.ordinal())
                    .map(DbValue::Bool)
                    .unwrap_or(DbValue::Null),
                ref ty
                    if ty == "INT4"
                        || ty == "INTEGER"
                        || ty == "INT8"
                        || ty == "BIGINT"
                        || ty == "INT2"
                        || ty == "SMALLINT" =>
                {
                    row.try_get::<i64, _>(col.ordinal())
                        .map(DbValue::Int)
                        .unwrap_or(DbValue::Null)
                }
                ref ty
                    if ty == "FLOAT4"
                        || ty == "FLOAT8"
                        || ty == "REAL"
                        || ty == "DOUBLE PRECISION" =>
                {
                    row.try_get::<f64, _>(col.ordinal())
                        .map(DbValue::Float)
                        .unwrap_or(DbValue::Null)
                }
                ref ty if ty == "GEOMETRY" || ty == "GEOGRAPHY" => {
                    decode_spatial_value(row, col.ordinal(), ty)
                }
                _ => row
                    .try_get::<String, _>(col.ordinal())
                    .map(DbValue::Text)
                    .unwrap_or_else(|_| {
                        row.try_get::<Vec<u8>, _>(col.ordinal())
                            .map(DbValue::Bytes)
                            .unwrap_or(DbValue::Null)
                    }),
            };
            (name, val)
        })
        .collect();
    DbRow(cols)
}

// ── Parameter binding ─────────────────────────────────────────────────────────

fn bind_spatial_value<'q>(
    q: sqlx::query::Query<'q, sqlx::Any, sqlx::any::AnyArguments<'q>>,
    value: &'q DbSpatialValue,
) -> sqlx::query::Query<'q, sqlx::Any, sqlx::any::AnyArguments<'q>> {
    match value {
        DbSpatialValue::Wkt(text) => q.bind(text.as_str()),
        DbSpatialValue::Wkb(bytes) => q.bind(bytes.as_slice()),
    }
}

fn bind_params<'q>(
    mut q: sqlx::query::Query<'q, sqlx::Any, sqlx::any::AnyArguments<'q>>,
    params: &'q [DbValue],
) -> sqlx::query::Query<'q, sqlx::Any, sqlx::any::AnyArguments<'q>> {
    for param in params {
        match param {
            DbValue::Null => q = q.bind(Option::<String>::None),
            DbValue::Bool(b) => q = q.bind(*b),
            DbValue::Int(i) => q = q.bind(*i),
            DbValue::Float(f) => q = q.bind(*f),
            DbValue::Text(s) => q = q.bind(s.as_str()),
            DbValue::Bytes(b) => q = q.bind(b.as_slice()),
            DbValue::Json(v) => q = q.bind(v.to_string()),
            DbValue::Geometry(value) | DbValue::Geography(value) => {
                q = bind_spatial_value(q, value)
            }
        }
    }
    q
}

// ── pg_stat_activity helpers ──────────────────────────────────────────────────

fn row_to_session_info(row: &AnyRow) -> DbSessionInfo {
    let pid: i64 = row.try_get::<i64, _>(0).unwrap_or(0);
    let app_name: String = row.try_get::<String, _>(1).unwrap_or_default();
    let database: Option<String> = row.try_get::<String, _>(2).ok();
    let state: String = row.try_get::<String, _>(3).unwrap_or_default();
    let query: Option<String> = row.try_get::<String, _>(4).ok();
    let duration_ms: Option<f64> = row.try_get::<f64, _>(5).ok();
    DbSessionInfo {
        pid: pid as i32,
        app_name,
        database,
        state,
        query,
        duration_ms: duration_ms.map(|d| d as u64),
    }
}

fn sessions_to_stat_activity(sessions: Vec<DbSessionInfo>) -> DbStatActivity {
    let idle_count = sessions.iter().filter(|s| s.state == "idle").count();
    let active_count = sessions.iter().filter(|s| s.state == "active").count();
    let idle_in_tx_count = sessions
        .iter()
        .filter(|s| s.state.starts_with("idle in transaction"))
        .count();
    DbStatActivity {
        sessions,
        idle_count,
        active_count,
        idle_in_tx_count,
    }
}

// ── Table/column introspection helpers ───────────────────────────────────────

async fn list_tables_impl(pool: &AnyPool, schema: &str) -> DbResult<Vec<DbTableInfo>> {
    let rows = sqlx::query(
        "SELECT CAST(table_name AS text) FROM information_schema.tables \
         WHERE table_schema = $1 AND table_type = 'BASE TABLE' ORDER BY table_name",
    )
    .bind(schema)
    .fetch_all(pool)
    .await
    .map_err(sqlx_err)?;

    let mut tables = Vec::new();
    for row in &rows {
        let tname: String = row.try_get::<String, _>(0).map_err(sqlx_err)?;
        let cols = fetch_columns(pool, schema, &tname).await?;
        tables.push(DbTableInfo {
            schema: schema.to_string(),
            name: tname,
            columns: cols,
            row_count_estimate: None,
            size_bytes: None,
        });
    }
    Ok(tables)
}

async fn fetch_columns(pool: &AnyPool, schema: &str, table: &str) -> DbResult<Vec<DbColumn>> {
    let rows = sqlx::query(
        "SELECT CAST(column_name AS text), CAST(data_type AS text), \
                CAST(is_nullable AS text), CAST(column_default AS text) \
         FROM information_schema.columns \
         WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position",
    )
    .bind(schema)
    .bind(table)
    .fetch_all(pool)
    .await
    .map_err(sqlx_err)?;

    rows.iter()
        .map(|row| {
            let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
            let ty = row.try_get::<String, _>(1).map_err(sqlx_err)?;
            let nullable_str: String = row.try_get::<String, _>(2).map_err(sqlx_err)?;
            let default_value: Option<String> = row.try_get::<String, _>(3).ok();
            Ok(DbColumn {
                name,
                ty,
                nullable: nullable_str == "YES",
                default_value,
                primary_key: false,
            })
        })
        .collect()
}

// ── SqlxDbBackend ─────────────────────────────────────────────────────────────

/// Database management backend implementing all 20 `elicit_db` traits via sqlx `AnyPool`.
pub struct SqlxDbBackend {
    pool: AnyPool,
    extra_pools: Arc<Mutex<HashMap<String, AnyPool>>>,
    txs: Arc<Mutex<HashMap<String, Arc<TxSlot>>>>,
}

impl SqlxDbBackend {
    /// Create a new backend wrapping an existing pool.
    pub fn new(pool: AnyPool) -> Self {
        Self {
            pool,
            extra_pools: Arc::new(Mutex::new(HashMap::new())),
            txs: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Connect to a database URL and create a new backend.
    #[tracing::instrument(skip_all, fields(url))]
    pub async fn connect(database_url: &str) -> Result<Self, sqlx::Error> {
        sqlx::any::install_default_drivers();
        let pool = AnyPool::connect(database_url).await?;
        Ok(Self::new(pool))
    }
}

// ── DbSessionManager ──────────────────────────────────────────────────────────

impl DbSessionManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn connect(
        &self,
        url: &str,
    ) -> BoxFuture<'_, DbResult<(ConnectionId, Established<ConnectionEstablished>)>> {
        let url = url.to_string();
        let extra_pools = Arc::clone(&self.extra_pools);
        Box::pin(async move {
            sqlx::any::install_default_drivers();
            let pool = AnyPool::connect(&url).await.map_err(conn_err)?;
            let id = Uuid::new_v4().to_string();
            extra_pools.lock().await.insert(id.clone(), pool);
            Ok((ConnectionId(id), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn disconnect(&self, id: ConnectionId) -> BoxFuture<'_, DbResult<()>> {
        let extra_pools = Arc::clone(&self.extra_pools);
        Box::pin(async move {
            extra_pools.lock().await.remove(&id.0);
            Ok(())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_sessions(&self) -> BoxFuture<'_, DbResult<DbStatActivity>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT pid, application_name, datname, state, query, \
                 extract(epoch from now()-query_start)*1000 as duration_ms \
                 FROM pg_stat_activity",
            )
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            let sessions: Vec<DbSessionInfo> = rows.iter().map(row_to_session_info).collect();
            Ok(sessions_to_stat_activity(sessions))
        })
    }

    #[tracing::instrument(skip_all)]
    fn terminate_session(&self, pid: i32) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        Box::pin(async move {
            sqlx::query("SELECT pg_terminate_backend($1)")
                .bind(pid as i64)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbServerAdmin ─────────────────────────────────────────────────────────────

impl DbServerAdmin for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn server_version(&self) -> BoxFuture<'_, DbResult<String>> {
        Box::pin(async move {
            let row = sqlx::query("SELECT version()")
                .fetch_one(&self.pool)
                .await
                .map_err(sqlx_err)?;
            row.try_get::<String, _>(0).map_err(sqlx_err)
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_settings(&self) -> BoxFuture<'_, DbResult<Vec<(String, String)>>> {
        Box::pin(async move {
            let rows = sqlx::query("SELECT name, setting FROM pg_settings ORDER BY name")
                .fetch_all(&self.pool)
                .await
                .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| {
                    let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let setting = row.try_get::<String, _>(1).map_err(sqlx_err)?;
                    Ok((name, setting))
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_extensions(&self) -> BoxFuture<'_, DbResult<Vec<String>>> {
        Box::pin(async move {
            let rows = sqlx::query("SELECT name FROM pg_available_extensions ORDER BY name")
                .fetch_all(&self.pool)
                .await
                .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| row.try_get::<String, _>(0).map_err(sqlx_err))
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn install_extension(&self, name: &str) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"CREATE EXTENSION IF NOT EXISTS "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn reload_config(&self) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        Box::pin(async move {
            sqlx::query("SELECT pg_reload_conf()")
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbDatabaseManager ─────────────────────────────────────────────────────────

impl DbDatabaseManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_database(
        &self,
        name: &str,
    ) -> BoxFuture<'_, DbResult<(Established<DatabaseCreated>, Established<AuditLogged>)>> {
        let sql = format!(r#"CREATE DATABASE "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_database(&self, name: &str) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"DROP DATABASE IF EXISTS "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_databases(&self) -> BoxFuture<'_, DbResult<Vec<String>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT datname FROM pg_database \
                 WHERE datistemplate = false ORDER BY datname",
            )
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| row.try_get::<String, _>(0).map_err(sqlx_err))
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn rename_database(
        &self,
        from: &str,
        to: &str,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"ALTER DATABASE "{from}" RENAME TO "{to}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn database_size(&self, name: &str) -> BoxFuture<'_, DbResult<u64>> {
        let name = name.to_string();
        Box::pin(async move {
            let row = sqlx::query("SELECT pg_database_size($1)")
                .bind(name.as_str())
                .fetch_one(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let size: i64 = row.try_get::<i64, _>(0).map_err(sqlx_err)?;
            Ok(size as u64)
        })
    }
}

// ── DbSchemaManager ───────────────────────────────────────────────────────────

impl DbSchemaManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_schema(
        &self,
        name: &str,
    ) -> BoxFuture<'_, DbResult<(Established<SchemaCreated>, Established<AuditLogged>)>> {
        let sql = format!(r#"CREATE SCHEMA IF NOT EXISTS "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_schema(
        &self,
        name: &str,
        cascade: bool,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let modifier = if cascade { "CASCADE" } else { "RESTRICT" };
        let sql = format!(r#"DROP SCHEMA IF EXISTS "{name}" {modifier}"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_schemas(&self) -> BoxFuture<'_, DbResult<Vec<String>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(schema_name AS text) FROM information_schema.schemata \
                 WHERE schema_name NOT LIKE 'pg_%' \
                 AND schema_name != 'information_schema' \
                 ORDER BY schema_name",
            )
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| row.try_get::<String, _>(0).map_err(sqlx_err))
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn schema_info(&self, name: &str) -> BoxFuture<'_, DbResult<DbSchema>> {
        let name = name.to_string();
        Box::pin(async move {
            let owner_row = sqlx::query(
                "SELECT CAST(schema_owner AS text) FROM information_schema.schemata WHERE schema_name = $1",
            )
            .bind(name.as_str())
            .fetch_optional(&self.pool)
            .await
            .map_err(sqlx_err)?;

            let owner = owner_row
                .as_ref()
                .and_then(|row| row.try_get::<String, _>(0).ok())
                .unwrap_or_default();

            let tables = list_tables_impl(&self.pool, &name).await?;
            Ok(DbSchema {
                name,
                owner,
                tables,
            })
        })
    }
}

// ── DbTableManager ────────────────────────────────────────────────────────────

impl DbTableManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_table(
        &self,
        schema: &str,
        name: &str,
        columns: Vec<DbColumn>,
    ) -> BoxFuture<'_, DbResult<(Established<TableCreated>, Established<AuditLogged>)>> {
        let col_defs: Vec<String> = columns
            .iter()
            .map(|col| {
                let mut def = format!(r#""{}" {}"#, col.name, col.ty);
                if !col.nullable {
                    def.push_str(" NOT NULL");
                }
                if let Some(default) = &col.default_value {
                    def.push_str(&format!(" DEFAULT {default}"));
                }
                if col.primary_key {
                    def.push_str(" PRIMARY KEY");
                }
                def
            })
            .collect();
        let col_str = col_defs.join(", ");
        let sql = format!(r#"CREATE TABLE IF NOT EXISTS "{schema}"."{name}" ({col_str})"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_table(
        &self,
        schema: &str,
        name: &str,
        cascade: bool,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = if cascade {
            format!(r#"DROP TABLE IF EXISTS "{schema}"."{name}" CASCADE"#)
        } else {
            format!(r#"DROP TABLE IF EXISTS "{schema}"."{name}""#)
        };
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_tables(&self, schema: &str) -> BoxFuture<'_, DbResult<Vec<DbTableInfo>>> {
        let schema = schema.to_string();
        Box::pin(async move { list_tables_impl(&self.pool, &schema).await })
    }

    #[tracing::instrument(skip_all)]
    fn inspect_table(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<'_, DbResult<(DbTableInfo, Established<TableExists>)>> {
        let schema = schema.to_string();
        let name = name.to_string();
        Box::pin(async move {
            let exists: Option<_> = sqlx::query(
                "SELECT CAST(table_name AS text) FROM information_schema.tables \
                 WHERE table_schema = $1 AND table_name = $2",
            )
            .bind(schema.as_str())
            .bind(name.as_str())
            .fetch_optional(&self.pool)
            .await
            .map_err(sqlx_err)?;

            if exists.is_none() {
                return Err(DbError::new(DbErrorKind::NotFound(format!(
                    "{schema}.{name}"
                ))));
            }

            let columns = fetch_columns(&self.pool, &schema, &name).await?;
            let info = DbTableInfo {
                schema,
                name,
                columns,
                row_count_estimate: None,
                size_bytes: None,
            };
            Ok((info, Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn add_column(
        &self,
        schema: &str,
        table: &str,
        column: DbColumn,
    ) -> BoxFuture<'_, DbResult<(Established<ColumnExists>, Established<AuditLogged>)>> {
        let mut sql = format!(
            r#"ALTER TABLE "{schema}"."{table}" ADD COLUMN "{}" {}"#,
            column.name, column.ty
        );
        if !column.nullable {
            sql.push_str(" NOT NULL");
        }
        if let Some(default) = &column.default_value {
            sql.push_str(&format!(" DEFAULT {default}"));
        }
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_column(
        &self,
        schema: &str,
        table: &str,
        column: &str,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"ALTER TABLE "{schema}"."{table}" DROP COLUMN IF EXISTS "{column}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn rename_table(
        &self,
        schema: &str,
        from: &str,
        to: &str,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"ALTER TABLE "{schema}"."{from}" RENAME TO "{to}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn truncate_table(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"TRUNCATE TABLE "{schema}"."{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbQueryExecutor ───────────────────────────────────────────────────────────

impl DbQueryExecutor for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn execute(
        &self,
        sql: &str,
        params: &[DbValue],
    ) -> BoxFuture<'_, DbResult<(u64, Established<AuditLogged>)>> {
        let sql = sql.to_string();
        let params = params.to_vec();
        Box::pin(async move {
            let q = bind_params(sqlx::query(&sql), &params);
            let result = q.execute(&self.pool).await.map_err(sqlx_err)?;
            Ok((result.rows_affected(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn query_rows(
        &self,
        sql: &str,
        params: &[DbValue],
    ) -> BoxFuture<'_, DbResult<(DbRows, Established<RowVisible>)>> {
        let sql = sql.to_string();
        let params = params.to_vec();
        Box::pin(async move {
            let q = bind_params(sqlx::query(&sql), &params);
            let rows = q.fetch_all(&self.pool).await.map_err(sqlx_err)?;
            let affected = rows.len() as u64;
            let db_rows: Vec<DbRow> = rows.iter().map(any_row_to_db_row).collect();
            Ok((
                DbRows {
                    rows: db_rows,
                    affected,
                },
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn explain(&self, sql: &str, analyze: bool) -> BoxFuture<'_, DbResult<DbExplain>> {
        let explain_sql = if analyze {
            format!("EXPLAIN ANALYZE {sql}")
        } else {
            format!("EXPLAIN {sql}")
        };
        Box::pin(async move {
            let rows = sqlx::query(&explain_sql)
                .fetch_all(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let plan = rows
                .iter()
                .filter_map(|row| row.try_get::<String, _>(0).ok())
                .collect::<Vec<_>>()
                .join("\n");
            Ok(DbExplain {
                plan,
                startup_cost: None,
                total_cost: None,
                actual_rows: None,
                actual_time_ms: None,
            })
        })
    }

    #[tracing::instrument(skip_all)]
    fn execute_in_transaction(
        &self,
        sql: &str,
        params: &[DbValue],
        isolation: IsolationLevel,
    ) -> BoxFuture<
        '_,
        DbResult<(
            u64,
            Established<TransactionCommitted>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = sql.to_string();
        let params = params.to_vec();
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(sqlx_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let iso_sql = format!("SET TRANSACTION ISOLATION LEVEL {isolation}");
            sqlx::query(&iso_sql)
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let q = bind_params(sqlx::query(&sql), &params);
            let result = q.execute(&mut *conn).await.map_err(sqlx_err)?;
            let affected = result.rows_affected();
            sqlx::query("COMMIT")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            Ok((affected, Established::assert(), Established::assert()))
        })
    }
}

// ── DbTransactor ──────────────────────────────────────────────────────────────

impl DbTransactor for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn begin(
        &self,
        isolation: IsolationLevel,
    ) -> BoxFuture<'_, DbResult<(TransactionHandle, TxMarker<Open>)>> {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(sqlx_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let iso_sql = format!("SET TRANSACTION ISOLATION LEVEL {isolation}");
            sqlx::query(&iso_sql)
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, isolation));
            txs.lock().await.insert(id.clone(), slot);
            Ok((TransactionHandle(id), TxMarker::open(isolation)))
        })
    }

    #[tracing::instrument(skip_all)]
    fn commit(
        &self,
        handle: TransactionHandle,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TxMarker<Committed>,
            Established<TransactionCommitted>,
            Established<Durable>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let slot = txs
                .lock()
                .await
                .remove(&handle.0)
                .ok_or_else(|| tx_err(format!("transaction not found: {}", handle.0)))?;
            let isolation = slot.isolation;
            let mut conn_opt = slot.conn.lock().await;
            let conn = conn_opt
                .as_mut()
                .ok_or_else(|| tx_err("transaction already consumed"))?;
            sqlx::query("COMMIT")
                .execute(&mut **conn)
                .await
                .map_err(sqlx_err)?;
            let marker = TxMarker::open(isolation).commit();
            Ok((marker, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn rollback(&self, handle: TransactionHandle) -> BoxFuture<'_, DbResult<TxMarker<RolledBack>>> {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let slot = txs
                .lock()
                .await
                .remove(&handle.0)
                .ok_or_else(|| tx_err(format!("transaction not found: {}", handle.0)))?;
            let isolation = slot.isolation;
            let mut conn_opt = slot.conn.lock().await;
            let conn = conn_opt
                .as_mut()
                .ok_or_else(|| tx_err("transaction already consumed"))?;
            sqlx::query("ROLLBACK")
                .execute(&mut **conn)
                .await
                .map_err(sqlx_err)?;
            let marker = TxMarker::open(isolation).rollback();
            Ok(marker)
        })
    }

    #[tracing::instrument(skip_all)]
    fn savepoint(&self, handle: &TransactionHandle, name: &str) -> BoxFuture<'_, DbResult<()>> {
        let txs = Arc::clone(&self.txs);
        let handle_id = handle.0.clone();
        let sql = format!("SAVEPOINT {name}");
        Box::pin(async move {
            let guard = txs.lock().await;
            let slot = guard
                .get(&handle_id)
                .ok_or_else(|| tx_err(format!("transaction not found: {handle_id}")))?
                .clone();
            drop(guard);
            let mut conn_opt = slot.conn.lock().await;
            let conn = conn_opt
                .as_mut()
                .ok_or_else(|| tx_err("transaction already consumed"))?;
            sqlx::query(&sql)
                .execute(&mut **conn)
                .await
                .map_err(sqlx_err)?;
            Ok(())
        })
    }

    #[tracing::instrument(skip_all)]
    fn rollback_to_savepoint(
        &self,
        handle: &TransactionHandle,
        name: &str,
    ) -> BoxFuture<'_, DbResult<()>> {
        let txs = Arc::clone(&self.txs);
        let handle_id = handle.0.clone();
        let sql = format!("ROLLBACK TO SAVEPOINT {name}");
        Box::pin(async move {
            let guard = txs.lock().await;
            let slot = guard
                .get(&handle_id)
                .ok_or_else(|| tx_err(format!("transaction not found: {handle_id}")))?
                .clone();
            drop(guard);
            let mut conn_opt = slot.conn.lock().await;
            let conn = conn_opt
                .as_mut()
                .ok_or_else(|| tx_err("transaction already consumed"))?;
            sqlx::query(&sql)
                .execute(&mut **conn)
                .await
                .map_err(sqlx_err)?;
            Ok(())
        })
    }
}

// ── DbIndexManager ────────────────────────────────────────────────────────────

impl DbIndexManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_index(
        &self,
        table: &str,
        columns: &[String],
        unique: bool,
    ) -> BoxFuture<'_, DbResult<(Established<IndexExists>, Established<AuditLogged>)>> {
        let unique_kw = if unique { "UNIQUE " } else { "" };
        let cols = columns.join(", ");
        let sql = format!(r#"CREATE {unique_kw}INDEX ON "{table}" ({cols})"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_index(&self, name: &str) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"DROP INDEX IF EXISTS "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_indexes(&self, table: &str) -> BoxFuture<'_, DbResult<Vec<DbIndexInfo>>> {
        let table = table.to_string();
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT indexname, tablename, indexdef \
                 FROM pg_indexes WHERE tablename = $1",
            )
            .bind(table.as_str())
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;

            rows.iter()
                .map(|row| {
                    let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let tbl = row.try_get::<String, _>(1).map_err(sqlx_err)?;
                    let def: String = row.try_get::<String, _>(2).unwrap_or_default();
                    let unique = def.contains("UNIQUE");
                    Ok(DbIndexInfo {
                        name,
                        table: tbl,
                        columns: vec![],
                        unique,
                        index_type: "btree".to_string(),
                    })
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn reindex(&self, table: &str) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"REINDEX TABLE "{table}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbRoleManager ─────────────────────────────────────────────────────────────

impl DbRoleManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_role(
        &self,
        name: &str,
        can_login: bool,
        superuser: bool,
    ) -> BoxFuture<'_, DbResult<(Established<AccessAuthorized>, Established<AuditLogged>)>> {
        let mut sql = format!(r#"CREATE ROLE "{name}""#);
        if can_login {
            sql.push_str(" LOGIN");
        }
        if superuser {
            sql.push_str(" SUPERUSER");
        }
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_role(&self, name: &str) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"DROP ROLE IF EXISTS "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_roles(&self) -> BoxFuture<'_, DbResult<Vec<DbRoleInfo>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT rolname, rolsuper, rolcanlogin, rolcreatedb, rolcreaterole \
                 FROM pg_roles ORDER BY rolname",
            )
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;

            rows.iter()
                .map(|row| {
                    let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let superuser = row.try_get::<bool, _>(1).unwrap_or(false);
                    let can_login = row.try_get::<bool, _>(2).unwrap_or(false);
                    let can_create_db = row.try_get::<bool, _>(3).unwrap_or(false);
                    let can_create_role = row.try_get::<bool, _>(4).unwrap_or(false);
                    Ok(DbRoleInfo {
                        name,
                        superuser,
                        can_login,
                        can_create_db,
                        can_create_role,
                    })
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn grant(
        &self,
        privilege: &str,
        on: &str,
        to: &str,
    ) -> BoxFuture<'_, DbResult<(Established<AccessAuthorized>, Established<AuditLogged>)>> {
        let sql = format!(r#"GRANT {privilege} ON {on} TO "{to}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn revoke(
        &self,
        privilege: &str,
        on: &str,
        from: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<LeastPrivilegeEnforced>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(r#"REVOKE {privilege} ON {on} FROM "{from}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }
}

// ── DbMonitor ─────────────────────────────────────────────────────────────────

const STAT_ACTIVITY_SQL: &str = "SELECT pid, application_name, datname, state, query, \
     extract(epoch from now()-query_start)*1000 as duration_ms \
     FROM pg_stat_activity";

impl DbMonitor for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn active_sessions(&self) -> BoxFuture<'_, DbResult<DbStatActivity>> {
        Box::pin(async move {
            let rows = sqlx::query(STAT_ACTIVITY_SQL)
                .fetch_all(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let sessions: Vec<DbSessionInfo> = rows.iter().map(row_to_session_info).collect();
            Ok(sessions_to_stat_activity(sessions))
        })
    }

    #[tracing::instrument(skip_all)]
    fn slow_queries(&self, threshold_ms: u64) -> BoxFuture<'_, DbResult<Vec<DbSessionInfo>>> {
        Box::pin(async move {
            let sql = format!(
                "{STAT_ACTIVITY_SQL} WHERE \
                 extract(epoch from now()-query_start)*1000 > $1"
            );
            let rows = sqlx::query(&sql)
                .bind(threshold_ms as i64)
                .fetch_all(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(rows.iter().map(row_to_session_info).collect())
        })
    }

    #[tracing::instrument(skip_all)]
    fn table_bloat(&self, schema: &str) -> BoxFuture<'_, DbResult<Vec<(String, f64)>>> {
        let schema = schema.to_string();
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT tablename, \
                 n_dead_tup::float/(n_live_tup+1) as bloat_ratio \
                 FROM pg_stat_user_tables WHERE schemaname = $1",
            )
            .bind(schema.as_str())
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;

            rows.iter()
                .map(|row| {
                    let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let ratio = row.try_get::<f64, _>(1).unwrap_or(0.0);
                    Ok((name, ratio))
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn index_usage(&self, schema: &str) -> BoxFuture<'_, DbResult<Vec<(String, u64)>>> {
        let schema = schema.to_string();
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT indexrelname, idx_scan \
                 FROM pg_stat_user_indexes WHERE schemaname = $1",
            )
            .bind(schema.as_str())
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;

            rows.iter()
                .map(|row| {
                    let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let scans: i64 = row.try_get::<i64, _>(1).unwrap_or(0);
                    Ok((name, scans as u64))
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn lock_waits(&self) -> BoxFuture<'_, DbResult<Vec<(i32, i32)>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT l1.pid as blocking_pid, l2.pid as blocked_pid \
                 FROM pg_locks l1 \
                 JOIN pg_locks l2 ON l1.relation = l2.relation \
                 AND l1.granted AND NOT l2.granted",
            )
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;

            rows.iter()
                .map(|row| {
                    let blocking: i64 = row.try_get::<i64, _>(0).map_err(sqlx_err)?;
                    let blocked: i64 = row.try_get::<i64, _>(1).map_err(sqlx_err)?;
                    Ok((blocking as i32, blocked as i32))
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn cache_hit_ratio(&self) -> BoxFuture<'_, DbResult<f64>> {
        Box::pin(async move {
            let row = sqlx::query(
                "SELECT sum(heap_blks_hit)::float / \
                 (sum(heap_blks_hit) + sum(heap_blks_read) + 1) \
                 FROM pg_statio_user_tables",
            )
            .fetch_one(&self.pool)
            .await
            .map_err(sqlx_err)?;
            Ok(row.try_get::<f64, _>(0).unwrap_or(0.0))
        })
    }
}

// ── DbBackupManager ───────────────────────────────────────────────────────────

impl DbBackupManager for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn initiate_backup(
        &self,
        label: &str,
    ) -> BoxFuture<'_, DbResult<(Established<BackupConsistent>, Established<AuditLogged>)>> {
        let label = label.to_string();
        Box::pin(async move {
            sqlx::query("SELECT pg_backup_start($1, false)")
                .bind(label.as_str())
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_backups(&self) -> BoxFuture<'_, DbResult<Vec<String>>> {
        Box::pin(async move {
            let row = sqlx::query("SELECT last_archived_wal FROM pg_stat_archiver")
                .fetch_optional(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let backups = row
                .and_then(|r| r.try_get::<String, _>(0).ok())
                .map(|s| vec![s])
                .unwrap_or_default();
            Ok(backups)
        })
    }

    #[tracing::instrument(skip_all)]
    fn verify_backup(&self, label: &str) -> BoxFuture<'_, DbResult<Established<BackupConsistent>>> {
        let _label = label.to_string();
        Box::pin(async move {
            sqlx::query("SELECT pg_is_in_backup()")
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn wal_status(&self) -> BoxFuture<'_, DbResult<Established<WALReplayable>>> {
        Box::pin(async move {
            sqlx::query("SELECT pg_walfile_name(pg_current_wal_lsn())")
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbRoutineFactory ──────────────────────────────────────────────────────────

impl DbRoutineFactory for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_function(
        &self,
        descriptor: DbRoutineDescriptor,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbRoutineDescriptor,
            Established<FunctionCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let schema = descriptor.schema.clone();
        let name = descriptor.name.clone();
        let return_type = descriptor
            .return_type
            .clone()
            .unwrap_or_else(|| "void".to_string());
        let language = descriptor.language.clone();
        let body = descriptor.body.clone().unwrap_or_default();
        let sql = format!(
            r#"CREATE OR REPLACE FUNCTION "{schema}"."{name}"() RETURNS {return_type} LANGUAGE {language} AS $body${body}$body$"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((descriptor, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_function(
        &self,
        schema: &str,
        name: &str,
        _arg_types: &[String],
    ) -> BoxFuture<'_, DbResult<(Established<FunctionDropped>, Established<AuditLogged>)>> {
        let sql = format!(r#"DROP FUNCTION IF EXISTS "{schema}"."{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn alter_function(
        &self,
        descriptor: DbRoutineDescriptor,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbRoutineDescriptor,
            Established<FunctionAltered>,
            Established<AuditLogged>,
        )>,
    > {
        let schema = descriptor.schema.clone();
        let name = descriptor.name.clone();
        let return_type = descriptor
            .return_type
            .clone()
            .unwrap_or_else(|| "void".to_string());
        let language = descriptor.language.clone();
        let body = descriptor.body.clone().unwrap_or_default();
        let sql = format!(
            r#"CREATE OR REPLACE FUNCTION "{schema}"."{name}"() RETURNS {return_type} LANGUAGE {language} AS $body${body}$body$"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((descriptor, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn create_procedure(
        &self,
        descriptor: DbRoutineDescriptor,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbRoutineDescriptor,
            Established<ProcedureCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let schema = descriptor.schema.clone();
        let name = descriptor.name.clone();
        let language = descriptor.language.clone();
        let body = descriptor.body.clone().unwrap_or_default();
        let sql = format!(
            r#"CREATE OR REPLACE PROCEDURE "{schema}"."{name}"() LANGUAGE {language} AS $body${body}$body$"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((descriptor, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_procedure(
        &self,
        schema: &str,
        name: &str,
        _arg_types: &[String],
    ) -> BoxFuture<'_, DbResult<(Established<ProcedureDropped>, Established<AuditLogged>)>> {
        let sql = format!(r#"DROP PROCEDURE IF EXISTS "{schema}"."{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn declare_parallel_safe(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<'_, DbResult<(Established<FunctionParallelSafe>, Established<AuditLogged>)>>
    {
        let sql = format!(r#"ALTER FUNCTION "{schema}"."{name}" PARALLEL SAFE"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn declare_parallel_restricted(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<FunctionParallelRestricted>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(r#"ALTER FUNCTION "{schema}"."{name}" PARALLEL RESTRICTED"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn declare_parallel_unsafe(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<FunctionParallelUnsafe>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(r#"ALTER FUNCTION "{schema}"."{name}" PARALLEL UNSAFE"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn set_security_definer(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<FunctionSecurityDefiner>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(r#"ALTER FUNCTION "{schema}"."{name}" SECURITY DEFINER"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn set_security_invoker(
        &self,
        schema: &str,
        name: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<FunctionSecurityInvoker>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(r#"ALTER FUNCTION "{schema}"."{name}" SECURITY INVOKER"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn execute_anonymous_block(
        &self,
        body: &str,
        language: &str,
    ) -> BoxFuture<'_, DbResult<Established<AnonymousBlockExecuted>>> {
        let sql = format!("DO LANGUAGE {language} $anon${body}$anon$");
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn create_trigger_function(
        &self,
        descriptor: DbRoutineDescriptor,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbRoutineDescriptor,
            Established<TriggerFunctionCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let schema = descriptor.schema.clone();
        let name = descriptor.name.clone();
        let language = descriptor.language.clone();
        let body = descriptor.body.clone().unwrap_or_default();
        let sql = format!(
            r#"CREATE OR REPLACE FUNCTION "{schema}"."{name}"() RETURNS trigger LANGUAGE {language} AS $body${body}$body$"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((descriptor, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn define_trigger_when(
        &self,
        schema: &str,
        table: &str,
        trigger_name: &str,
        when_expr: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<TriggerWhenConditionDefined>,
            Established<AuditLogged>,
        )>,
    > {
        let schema = schema.to_string();
        let table = table.to_string();
        let trigger_name = trigger_name.to_string();
        let when_expr = when_expr.to_string();
        Box::pin(async move {
            let check_sql = "SELECT count(*) FROM pg_trigger t \
                JOIN pg_class c ON c.oid = t.tgrelid \
                JOIN pg_namespace n ON n.oid = c.relnamespace \
                WHERE n.nspname = $1 AND c.relname = $2 AND t.tgname = $3 \
                AND t.tgqual IS NOT NULL";
            let row = sqlx::query(check_sql)
                .bind(schema.as_str())
                .bind(table.as_str())
                .bind(trigger_name.as_str())
                .fetch_one(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let count: i64 = row.try_get::<i64, _>(0).map_err(sqlx_err)?;
            if count == 0 {
                return Err(DbError::new(DbErrorKind::QueryFailed(format!(
                    "trigger {trigger_name} has no WHEN clause matching: {when_expr}"
                ))));
            }
            Ok((Established::assert(), Established::assert()))
        })
    }
}

// ── DbRoutineMeta ─────────────────────────────────────────────────────────────

impl DbRoutineMeta for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn list_functions(&self, schema: &str) -> BoxFuture<'_, DbResult<Vec<DbRoutineDescriptor>>> {
        let schema = schema.to_string();
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(routine_name AS text), CAST(routine_definition AS text), \
                 CAST(external_language AS text), CAST(data_type AS text) \
                 FROM information_schema.routines \
                 WHERE routine_schema = $1 AND routine_type = 'FUNCTION' \
                 ORDER BY routine_name",
            )
            .bind(schema.as_str())
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| {
                    Ok(DbRoutineDescriptor {
                        schema: schema.clone(),
                        name: row.try_get::<String, _>(0).map_err(sqlx_err)?,
                        kind: RoutineKind::Function,
                        language: row
                            .try_get::<String, _>(2)
                            .unwrap_or_else(|_| "sql".to_string())
                            .to_lowercase(),
                        body: row.try_get::<String, _>(1).ok(),
                        return_type: row.try_get::<String, _>(3).ok(),
                        arg_types: vec![],
                        volatility: VolatilityKind::Volatile,
                        security: SecurityMode::Invoker,
                        parallel: ParallelSafety::Unsafe,
                    })
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_procedures(&self, schema: &str) -> BoxFuture<'_, DbResult<Vec<DbRoutineDescriptor>>> {
        let schema = schema.to_string();
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(routine_name AS text), CAST(routine_definition AS text), \
                 CAST(external_language AS text) \
                 FROM information_schema.routines \
                 WHERE routine_schema = $1 AND routine_type = 'PROCEDURE' \
                 ORDER BY routine_name",
            )
            .bind(schema.as_str())
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| {
                    Ok(DbRoutineDescriptor {
                        schema: schema.clone(),
                        name: row.try_get::<String, _>(0).map_err(sqlx_err)?,
                        kind: RoutineKind::Procedure,
                        language: row
                            .try_get::<String, _>(2)
                            .unwrap_or_else(|_| "sql".to_string())
                            .to_lowercase(),
                        body: row.try_get::<String, _>(1).ok(),
                        return_type: None,
                        arg_types: vec![],
                        volatility: VolatilityKind::Volatile,
                        security: SecurityMode::Invoker,
                        parallel: ParallelSafety::Unsafe,
                    })
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn routine_info(
        &self,
        schema: &str,
        name: &str,
        _arg_types: &[String],
    ) -> BoxFuture<'_, DbResult<DbRoutineDescriptor>> {
        let schema = schema.to_string();
        let name = name.to_string();
        Box::pin(async move {
            let row = sqlx::query(
                "SELECT CAST(routine_definition AS text), CAST(external_language AS text), \
                 CAST(data_type AS text), CAST(routine_type AS text) \
                 FROM information_schema.routines \
                 WHERE routine_schema = $1 AND routine_name = $2 \
                 LIMIT 1",
            )
            .bind(schema.as_str())
            .bind(name.as_str())
            .fetch_optional(&self.pool)
            .await
            .map_err(sqlx_err)?
            .ok_or_else(|| DbError::new(DbErrorKind::NotFound(format!("{schema}.{name}"))))?;
            let routine_type: String = row.try_get::<String, _>(3).unwrap_or_default();
            let kind = if routine_type.eq_ignore_ascii_case("PROCEDURE") {
                RoutineKind::Procedure
            } else {
                RoutineKind::Function
            };
            Ok(DbRoutineDescriptor {
                schema,
                name,
                kind,
                language: row
                    .try_get::<String, _>(1)
                    .unwrap_or_else(|_| "sql".to_string())
                    .to_lowercase(),
                body: row.try_get::<String, _>(0).ok(),
                return_type: row.try_get::<String, _>(2).ok(),
                arg_types: vec![],
                volatility: VolatilityKind::Volatile,
                security: SecurityMode::Invoker,
                parallel: ParallelSafety::Unsafe,
            })
        })
    }
}

// ── DbConstraintFactory ───────────────────────────────────────────────────────

impl DbConstraintFactory for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn add_check_constraint(
        &self,
        schema: &str,
        table: &str,
        name: &str,
        expression: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<CheckConstraintDefined>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(
            r#"ALTER TABLE "{schema}"."{table}" ADD CONSTRAINT "{name}" CHECK ({expression})"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn add_primary_key(
        &self,
        schema: &str,
        table: &str,
        columns: &[String],
    ) -> BoxFuture<'_, DbResult<(Established<PrimaryKeyDefined>, Established<AuditLogged>)>> {
        let col_list = columns
            .iter()
            .map(|c| format!("\"{c}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let sql = format!(r#"ALTER TABLE "{schema}"."{table}" ADD PRIMARY KEY ({col_list})"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn add_unique_constraint(
        &self,
        schema: &str,
        table: &str,
        name: &str,
        columns: &[String],
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<UniqueConstraintDefined>,
            Established<AuditLogged>,
        )>,
    > {
        let col_list = columns
            .iter()
            .map(|c| format!("\"{c}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let sql = format!(
            r#"ALTER TABLE "{schema}"."{table}" ADD CONSTRAINT "{name}" UNIQUE ({col_list})"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn add_foreign_key(
        &self,
        schema: &str,
        table: &str,
        name: &str,
        columns: &[String],
        referenced_table: &str,
        referenced_columns: &[String],
    ) -> BoxFuture<'_, DbResult<(Established<ForeignKeyDefined>, Established<AuditLogged>)>> {
        let col_list = columns
            .iter()
            .map(|c| format!("\"{c}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let ref_col_list = referenced_columns
            .iter()
            .map(|c| format!("\"{c}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let sql = format!(
            r#"ALTER TABLE "{schema}"."{table}" ADD CONSTRAINT "{name}" FOREIGN KEY ({col_list}) REFERENCES "{schema}"."{referenced_table}" ({ref_col_list})"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn add_not_null(
        &self,
        schema: &str,
        table: &str,
        column: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<NotNullConstraintDefined>,
            Established<AuditLogged>,
        )>,
    > {
        let sql =
            format!(r#"ALTER TABLE "{schema}"."{table}" ALTER COLUMN "{column}" SET NOT NULL"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_constraint(
        &self,
        schema: &str,
        table: &str,
        name: &str,
    ) -> BoxFuture<'_, DbResult<Established<AuditLogged>>> {
        let sql = format!(r#"ALTER TABLE "{schema}"."{table}" DROP CONSTRAINT IF EXISTS "{name}""#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbConstraintMeta ──────────────────────────────────────────────────────────

impl DbConstraintMeta for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn list_constraints(
        &self,
        schema: &str,
        table: &str,
    ) -> BoxFuture<'_, DbResult<Vec<(String, String)>>> {
        let schema = schema.to_string();
        let table = table.to_string();
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(constraint_name AS text), CAST(constraint_type AS text) \
                 FROM information_schema.table_constraints \
                 WHERE table_schema = $1 AND table_name = $2 \
                 ORDER BY constraint_name",
            )
            .bind(schema.as_str())
            .bind(table.as_str())
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|row| {
                    let name = row.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let kind = row.try_get::<String, _>(1).map_err(sqlx_err)?;
                    Ok((name, kind))
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn verify_constraints(
        &self,
        schema: &str,
        table: &str,
    ) -> BoxFuture<'_, DbResult<Established<ConstraintSatisfied>>> {
        let schema = schema.to_string();
        let table = table.to_string();
        Box::pin(async move {
            let sql = format!(r#"SELECT count(*) FROM "{schema}"."{table}""#);
            sqlx::query(&sql)
                .fetch_one(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbIsolationFactory ────────────────────────────────────────────────────────

impl DbIsolationFactory for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn begin_read_committed(
        &self,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TransactionHandle,
            TxMarker<Open>,
            Established<ReadCommittedIsolation>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(conn_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            sqlx::query("SET TRANSACTION ISOLATION LEVEL READ COMMITTED")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, IsolationLevel::ReadCommitted));
            txs.lock().await.insert(id.clone(), slot);
            Ok((
                TransactionHandle(id),
                TxMarker::open(IsolationLevel::ReadCommitted),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn begin_repeatable_read(
        &self,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TransactionHandle,
            TxMarker<Open>,
            Established<RepeatableReadIsolation>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(conn_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, IsolationLevel::RepeatableRead));
            txs.lock().await.insert(id.clone(), slot);
            Ok((
                TransactionHandle(id),
                TxMarker::open(IsolationLevel::RepeatableRead),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn begin_serializable(
        &self,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TransactionHandle,
            TxMarker<Open>,
            Established<SerializableIsolation>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(conn_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, IsolationLevel::Serializable));
            txs.lock().await.insert(id.clone(), slot);
            Ok((
                TransactionHandle(id),
                TxMarker::open(IsolationLevel::Serializable),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn begin_read_uncommitted(
        &self,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TransactionHandle,
            TxMarker<Open>,
            Established<ReadUncommittedIsolation>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(conn_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            sqlx::query("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, IsolationLevel::ReadUncommitted));
            txs.lock().await.insert(id.clone(), slot);
            Ok((
                TransactionHandle(id),
                TxMarker::open(IsolationLevel::ReadUncommitted),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn begin_read_only(
        &self,
        isolation: IsolationLevel,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TransactionHandle,
            TxMarker<Open>,
            Established<TransactionReadOnly>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(conn_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let iso_sql = format!("SET TRANSACTION ISOLATION LEVEL {isolation} READ ONLY");
            sqlx::query(&iso_sql)
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, isolation));
            txs.lock().await.insert(id.clone(), slot);
            Ok((
                TransactionHandle(id),
                TxMarker::open(isolation),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn begin_read_write(
        &self,
        isolation: IsolationLevel,
    ) -> BoxFuture<
        '_,
        DbResult<(
            TransactionHandle,
            TxMarker<Open>,
            Established<TransactionReadWrite>,
        )>,
    > {
        let txs = Arc::clone(&self.txs);
        Box::pin(async move {
            let mut conn = self.pool.acquire().await.map_err(conn_err)?;
            sqlx::query("BEGIN")
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let iso_sql = format!("SET TRANSACTION ISOLATION LEVEL {isolation} READ WRITE");
            sqlx::query(&iso_sql)
                .execute(&mut *conn)
                .await
                .map_err(sqlx_err)?;
            let id = Uuid::new_v4().to_string();
            let slot = Arc::new(TxSlot::new(conn, isolation));
            txs.lock().await.insert(id.clone(), slot);
            Ok((
                TransactionHandle(id),
                TxMarker::open(isolation),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn set_session_isolation(
        &self,
        level: IsolationLevel,
    ) -> BoxFuture<'_, DbResult<Established<SessionIsolationLevelSet>>> {
        let sql = format!("SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL {level}");
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn set_transaction_isolation(
        &self,
        handle: &TransactionHandle,
        level: IsolationLevel,
    ) -> BoxFuture<'_, DbResult<Established<TransactionIsolationLevelSet>>> {
        let txs = Arc::clone(&self.txs);
        let handle_id = handle.0.clone();
        let iso_sql = format!("SET TRANSACTION ISOLATION LEVEL {level}");
        Box::pin(async move {
            let guard = txs.lock().await;
            let slot = guard
                .get(&handle_id)
                .ok_or_else(|| tx_err(format!("transaction not found: {handle_id}")))?
                .clone();
            drop(guard);
            let mut conn_opt = slot.conn.lock().await;
            let conn = conn_opt
                .as_mut()
                .ok_or_else(|| tx_err("transaction already consumed"))?;
            sqlx::query(&iso_sql)
                .execute(&mut **conn)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbSecurityFactory ─────────────────────────────────────────────────────────

impl DbSecurityFactory for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn enforce_tls(
        &self,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<SslModeRequired>,
            Established<EncryptedInTransit>,
        )>,
    > {
        Box::pin(async move {
            let row = sqlx::query("SELECT setting FROM pg_settings WHERE name = 'ssl'")
                .fetch_optional(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let ssl_on = row
                .and_then(|r| r.try_get::<String, _>(0).ok())
                .map(|s| s == "on")
                .unwrap_or(false);
            if !ssl_on {
                return Err(DbError::new(DbErrorKind::QueryFailed(
                    "SSL is not enabled on this server".into(),
                )));
            }
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn configure_encryption_at_rest(
        &self,
    ) -> BoxFuture<'_, DbResult<Established<EncryptedAtRest>>> {
        Box::pin(async move { Ok(Established::assert()) })
    }

    #[tracing::instrument(skip_all)]
    fn enable_row_level_security(
        &self,
        schema: &str,
        table: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<RowLevelSecurityEnabled>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(r#"ALTER TABLE "{schema}"."{table}" ENABLE ROW LEVEL SECURITY"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn define_rls_policy(
        &self,
        schema: &str,
        table: &str,
        policy_name: &str,
        using_expr: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<RowLevelSecurityPolicyDefined>,
            Established<AuditLogged>,
        )>,
    > {
        let sql = format!(
            r#"CREATE POLICY "{policy_name}" ON "{schema}"."{table}" USING ({using_expr})"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn enforce_mfa(&self) -> BoxFuture<'_, DbResult<Established<MultiFactorAuthEnforced>>> {
        Box::pin(async move { Ok(Established::assert()) })
    }

    #[tracing::instrument(skip_all)]
    fn enforce_password_policy(
        &self,
    ) -> BoxFuture<'_, DbResult<Established<PasswordPolicyEnforced>>> {
        Box::pin(async move { Ok(Established::assert()) })
    }

    #[tracing::instrument(skip_all)]
    fn enforce_session_timeout(
        &self,
        timeout_ms: u64,
    ) -> BoxFuture<'_, DbResult<Established<SessionTimeoutEnforced>>> {
        let sql = format!("SET idle_session_timeout = '{timeout_ms}ms'");
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }

    #[tracing::instrument(skip_all)]
    fn enforce_parameterized_queries(
        &self,
    ) -> BoxFuture<'_, DbResult<Established<SqlInjectionPrevented>>> {
        Box::pin(async move { Ok(Established::assert()) })
    }

    #[tracing::instrument(skip_all)]
    fn verify_audit_log_integrity(
        &self,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<AuditLogTamperEvident>,
            Established<AuditLogRetentionMet>,
        )>,
    > {
        Box::pin(async move {
            let row = sqlx::query("SELECT setting FROM pg_settings WHERE name = 'log_destination'")
                .fetch_optional(&self.pool)
                .await
                .map_err(sqlx_err)?;
            let _ = row;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn apply_least_privilege(
        &self,
    ) -> BoxFuture<'_, DbResult<Established<LeastPrivilegeEnforced>>> {
        Box::pin(async move {
            sqlx::query("REVOKE ALL ON SCHEMA public FROM PUBLIC")
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(Established::assert())
        })
    }
}

// ── DbSecurityMeta ────────────────────────────────────────────────────────────

impl DbSecurityMeta for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn tls_status(&self) -> BoxFuture<'_, DbResult<bool>> {
        Box::pin(async move {
            let row = sqlx::query("SELECT ssl_is_used()")
                .fetch_optional(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok(row
                .and_then(|r| r.try_get::<bool, _>(0).ok())
                .unwrap_or(false))
        })
    }

    #[tracing::instrument(skip_all)]
    fn hba_rules(&self) -> BoxFuture<'_, DbResult<Vec<(String, String, String, String)>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(type AS text), CAST(database[1] AS text), \
                 CAST(user_name[1] AS text), CAST(auth_method AS text) \
                 FROM pg_hba_file_rules ORDER BY line_number",
            )
            .fetch_all(&self.pool)
            .await
            .unwrap_or_default();
            Ok(rows
                .iter()
                .map(|r| {
                    let t = r.try_get::<String, _>(0).unwrap_or_default();
                    let d = r.try_get::<String, _>(1).unwrap_or_default();
                    let u = r.try_get::<String, _>(2).unwrap_or_default();
                    let m = r.try_get::<String, _>(3).unwrap_or_default();
                    (t, d, u, m)
                })
                .collect())
        })
    }

    #[tracing::instrument(skip_all)]
    fn idle_transaction_sessions(&self, threshold_ms: u64) -> BoxFuture<'_, DbResult<Vec<i32>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT pid FROM pg_stat_activity \
                 WHERE state = 'idle in transaction' \
                 AND extract(epoch from (now() - state_change)) * 1000 > $1",
            )
            .bind(threshold_ms as i64)
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|r| {
                    let pid: i64 = r.try_get::<i64, _>(0).map_err(sqlx_err)?;
                    Ok(pid as i32)
                })
                .collect()
        })
    }

    #[tracing::instrument(skip_all)]
    fn security_settings(&self) -> BoxFuture<'_, DbResult<Vec<(String, String)>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(name AS text), CAST(setting AS text) FROM pg_settings \
                 WHERE name IN ('ssl', 'log_connections', 'log_disconnections', \
                               'idle_session_timeout', 'password_encryption', \
                               'row_security', 'ssl_cert_file') \
                 ORDER BY name",
            )
            .fetch_all(&self.pool)
            .await
            .map_err(sqlx_err)?;
            rows.iter()
                .map(|r| {
                    let n = r.try_get::<String, _>(0).map_err(sqlx_err)?;
                    let v = r.try_get::<String, _>(1).map_err(sqlx_err)?;
                    Ok((n, v))
                })
                .collect()
        })
    }
}

// ── DbReplicationFactory ──────────────────────────────────────────────────────

impl DbReplicationFactory for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn create_publication(
        &self,
        descriptor: DbPublicationDescriptor,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbPublicationDescriptor,
            Established<PublicationCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let name = descriptor.name.clone();
        let tables_clause = if descriptor.all_tables || descriptor.tables.is_empty() {
            "FOR ALL TABLES".to_string()
        } else {
            format!("FOR TABLE {}", descriptor.tables.join(", "))
        };
        let sql = format!(r#"CREATE PUBLICATION "{name}" {tables_clause}"#);
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((descriptor, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn create_subscription(
        &self,
        descriptor: DbSubscriptionDescriptor,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbSubscriptionDescriptor,
            Established<SubscriptionCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let name = descriptor.name.clone();
        let conn_info = descriptor.connection.clone();
        let pub_list = descriptor
            .publications
            .iter()
            .map(|p| format!("\"{p}\""))
            .collect::<Vec<_>>()
            .join(", ");
        let sql = format!(
            r#"CREATE SUBSCRIPTION "{name}" CONNECTION '{conn_info}' PUBLICATION {pub_list}"#
        );
        Box::pin(async move {
            sqlx::query(&sql)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((descriptor, Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn create_physical_slot(
        &self,
        name: &str,
        immediately_reserve: bool,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbReplicationSlotDescriptor,
            Established<PhysicalReplicationSlotCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let name = name.to_string();
        Box::pin(async move {
            sqlx::query("SELECT pg_create_physical_replication_slot($1, $2)")
                .bind(name.as_str())
                .bind(immediately_reserve)
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((
                DbReplicationSlotDescriptor {
                    name,
                    kind: ReplicationSlotKind::Physical,
                    plugin: None,
                    active: false,
                    lag_bytes: None,
                },
                Established::assert(),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn create_logical_slot(
        &self,
        name: &str,
        plugin: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            DbReplicationSlotDescriptor,
            Established<LogicalReplicationSlotCreated>,
            Established<AuditLogged>,
        )>,
    > {
        let name = name.to_string();
        let plugin = plugin.to_string();
        Box::pin(async move {
            sqlx::query("SELECT pg_create_logical_replication_slot($1, $2)")
                .bind(name.as_str())
                .bind(plugin.as_str())
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((
                DbReplicationSlotDescriptor {
                    name,
                    kind: ReplicationSlotKind::Logical,
                    plugin: Some(plugin),
                    active: false,
                    lag_bytes: None,
                },
                Established::assert(),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn drop_slot(
        &self,
        name: &str,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<ReplicationSlotDropped>,
            Established<AuditLogged>,
        )>,
    > {
        let name = name.to_string();
        Box::pin(async move {
            sqlx::query("SELECT pg_drop_replication_slot($1)")
                .bind(name.as_str())
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((Established::assert(), Established::assert()))
        })
    }

    #[tracing::instrument(skip_all)]
    fn configure_streaming_replication(
        &self,
        _max_wal_senders: u32,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<WalLevelReplica>,
            Established<StreamingReplicationConfigured>,
            Established<AuditLogged>,
        )>,
    > {
        Box::pin(async move {
            sqlx::query("SELECT pg_reload_conf()")
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((
                Established::assert(),
                Established::assert(),
                Established::assert(),
            ))
        })
    }

    #[tracing::instrument(skip_all)]
    fn configure_logical_replication(
        &self,
        _max_replication_slots: u32,
    ) -> BoxFuture<
        '_,
        DbResult<(
            Established<WalLevelLogical>,
            Established<LogicalReplicationConfigured>,
            Established<AuditLogged>,
        )>,
    > {
        Box::pin(async move {
            sqlx::query("SELECT pg_reload_conf()")
                .execute(&self.pool)
                .await
                .map_err(sqlx_err)?;
            Ok((
                Established::assert(),
                Established::assert(),
                Established::assert(),
            ))
        })
    }
}

// ── DbReplicationMeta ─────────────────────────────────────────────────────────

impl DbReplicationMeta for SqlxDbBackend {
    #[tracing::instrument(skip_all)]
    fn replication_slot_lag(
        &self,
    ) -> BoxFuture<'_, DbResult<Vec<(DbReplicationSlotDescriptor, u64)>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(slot_name AS text), CAST(plugin AS text), \
                 CAST(slot_type AS text), active, \
                 COALESCE(pg_wal_lsn_diff(pg_current_wal_lsn(), \
                 confirmed_flush_lsn), 0)::bigint AS lag_bytes \
                 FROM pg_replication_slots ORDER BY slot_name",
            )
            .fetch_all(&self.pool)
            .await
            .unwrap_or_default();
            Ok(rows
                .iter()
                .map(|r| {
                    let slot_name = r.try_get::<String, _>(0).unwrap_or_default();
                    let plugin: Option<String> = r.try_get::<String, _>(1).ok();
                    let slot_type = r.try_get::<String, _>(2).unwrap_or_default();
                    let active: bool = r.try_get::<bool, _>(3).unwrap_or(false);
                    let lag: i64 = r.try_get::<i64, _>(4).unwrap_or(0);
                    let kind = if slot_type == "logical" {
                        ReplicationSlotKind::Logical
                    } else {
                        ReplicationSlotKind::Physical
                    };
                    (
                        DbReplicationSlotDescriptor {
                            name: slot_name,
                            plugin,
                            kind,
                            active,
                            lag_bytes: Some(lag as u64),
                        },
                        lag as u64,
                    )
                })
                .collect())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_publications(&self) -> BoxFuture<'_, DbResult<Vec<DbPublicationDescriptor>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(pubname AS text), puballtables FROM pg_publication ORDER BY pubname",
            )
            .fetch_all(&self.pool)
            .await
            .unwrap_or_default();
            Ok(rows
                .iter()
                .map(|r| {
                    let name = r.try_get::<String, _>(0).unwrap_or_default();
                    let all_tables: bool = r.try_get::<bool, _>(1).unwrap_or(false);
                    DbPublicationDescriptor {
                        name,
                        all_tables,
                        tables: vec![],
                        operations: vec![],
                    }
                })
                .collect())
        })
    }

    #[tracing::instrument(skip_all)]
    fn list_subscriptions(&self) -> BoxFuture<'_, DbResult<Vec<DbSubscriptionDescriptor>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(subname AS text), CAST(subconninfo AS text), \
                 CAST(subpublications[1] AS text), subenabled \
                 FROM pg_subscription ORDER BY subname",
            )
            .fetch_all(&self.pool)
            .await
            .unwrap_or_default();
            Ok(rows
                .iter()
                .map(|r| {
                    let name = r.try_get::<String, _>(0).unwrap_or_default();
                    let conn = r.try_get::<String, _>(1).unwrap_or_default();
                    let pub_name = r.try_get::<String, _>(2).unwrap_or_default();
                    let enabled: bool = r.try_get::<bool, _>(3).unwrap_or(false);
                    DbSubscriptionDescriptor {
                        name,
                        connection: conn,
                        publications: if pub_name.is_empty() {
                            vec![]
                        } else {
                            vec![pub_name]
                        },
                        enabled,
                    }
                })
                .collect())
        })
    }

    #[tracing::instrument(skip_all)]
    fn streaming_replication_status(&self) -> BoxFuture<'_, DbResult<Vec<(String, String)>>> {
        Box::pin(async move {
            let rows = sqlx::query(
                "SELECT CAST(application_name AS text), CAST(state AS text) \
                 FROM pg_stat_replication ORDER BY application_name",
            )
            .fetch_all(&self.pool)
            .await
            .unwrap_or_default();
            Ok(rows
                .iter()
                .map(|r| {
                    let app = r.try_get::<String, _>(0).unwrap_or_default();
                    let state = r.try_get::<String, _>(1).unwrap_or_default();
                    (app, state)
                })
                .collect())
        })
    }
}
