persistant-sqlx 0.6.2

// SPDX-License-Identifier: MIT OR Apache-2.0
//! MySQL (and MariaDB) behind an sqlx pool (feature `mysql`).
//!
//! The dialect is read from the server's version string. Placeholders are positional `?`.
//! **Rows affected**: sqlx always connects with `CLIENT_FOUND_ROWS`, so an `UPDATE` reports rows
//! *matched* and an upsert that changes nothing reports 1, the same as an insert (measured on MySQL
//! 8.4; `ROW_COUNT()` follows the same flag). A caller that must tell them apart does not use an
//! upsert's count: `KvOnRelational::claim_two_step` is written around that.
//! TLS policy is not typed here yet: the URL's `ssl-mode` decides (sqlx's default is `PREFERRED`).

use persistant_relational::{Dialect, RelError, RelationalBackend, RelationalTx, Row, Value};
use sqlx::mysql::{MySqlArguments, MySqlPoolOptions, MySqlRow};
use sqlx::query::Query;
use sqlx::{AssertSqlSafe, MySql, Row as SqlxRow, TypeInfo, ValueRef};

/// Classify an sqlx error by MySQL's error number
/// (<https://dev.mysql.com/doc/mysql-errors/8.4/en/server-error-reference.html>).
fn classify(e: sqlx::Error) -> RelError {
    match &e {
        sqlx::Error::Database(db) => {
            let number = db
                .try_downcast_ref::<sqlx::mysql::MySqlDatabaseError>()
                .map(|m| m.number());
            let constraint = db.constraint().map(str::to_string);
            match number {
                // ER_DUP_ENTRY, ER_DUP_ENTRY_WITH_KEY_NAME
                Some(1062) | Some(1586) => RelError::UniqueViolation { constraint },
                // ER_NO_REFERENCED_ROW_2, ER_ROW_IS_REFERENCED_2
                Some(1452) | Some(1451) => RelError::ForeignKeyViolation { constraint },
                // ER_CHECK_CONSTRAINT_VIOLATED, ER_BAD_NULL_ERROR
                Some(3819) | Some(1048) => RelError::CheckViolation { constraint },
                // ER_LOCK_DEADLOCK
                Some(1213) => RelError::Deadlock,
                // ER_LOCK_WAIT_TIMEOUT
                Some(1205) => RelError::Serialization,
                _ => RelError::Backend(e.to_string()),
            }
        }
        sqlx::Error::Io(_) | sqlx::Error::PoolTimedOut | sqlx::Error::PoolClosed => {
            RelError::Unavailable(e.to_string())
        }
        _ => RelError::Backend(e.to_string()),
    }
}

fn bind<'q>(
    mut q: Query<'q, MySql, MySqlArguments>,
    params: &'q [Value],
) -> Result<Query<'q, MySql, MySqlArguments>, RelError> {
    for p in params {
        q = match p {
            Value::Null => q.bind(Option::<String>::None),
            Value::Bool(b) => q.bind(*b),
            Value::Int(n) => q.bind(*n),
            Value::Float(f) => q.bind(*f),
            Value::Text(s) => q.bind(s.as_str()),
            Value::Bytes(b) => q.bind(b.as_slice()),
            other => {
                return Err(RelError::Unsupported(format!(
                    "parameter {other:?} has no MySQL binding in this driver"
                )));
            }
        };
    }
    Ok(q)
}

fn decode(row: &MySqlRow) -> Result<Row, RelError> {
    let mut values = Vec::with_capacity(row.len());
    for i in 0..row.len() {
        let raw = row.try_get_raw(i).map_err(classify)?;
        if raw.is_null() {
            values.push(Value::Null);
            continue;
        }
        let name = raw.type_info().name().to_string();
        let v = match name.as_str() {
            "BOOLEAN" => Value::Bool(row.try_get::<bool, _>(i).map_err(classify)?),
            "TINYINT" => Value::Int(i64::from(row.try_get::<i8, _>(i).map_err(classify)?)),
            "SMALLINT" => Value::Int(i64::from(row.try_get::<i16, _>(i).map_err(classify)?)),
            "INT" | "MEDIUMINT" => {
                Value::Int(i64::from(row.try_get::<i32, _>(i).map_err(classify)?))
            }
            "BIGINT" => Value::Int(row.try_get::<i64, _>(i).map_err(classify)?),
            "FLOAT" => Value::Float(f64::from(row.try_get::<f32, _>(i).map_err(classify)?)),
            "DOUBLE" => Value::Float(row.try_get::<f64, _>(i).map_err(classify)?),
            "CHAR" | "VARCHAR" | "TEXT" | "TINYTEXT" | "MEDIUMTEXT" | "LONGTEXT" => {
                Value::Text(row.try_get::<String, _>(i).map_err(classify)?)
            }
            "BINARY" | "VARBINARY" | "BLOB" | "TINYBLOB" | "MEDIUMBLOB" | "LONGBLOB" => {
                Value::Bytes(row.try_get::<Vec<u8>, _>(i).map_err(classify)?)
            }
            other => {
                return Err(RelError::Unsupported(format!(
                    "column {i} has type {other}, which Value cannot carry; use MysqlBackend::pool"
                )));
            }
        };
        values.push(v);
    }
    Ok(Row::new(values))
}

/// A MySQL or MariaDB database behind an sqlx pool.
#[derive(Debug, Clone)]
pub struct MysqlBackend {
    pool: sqlx::MySqlPool,
    version: String,
    dialect: Dialect,
}

impl MysqlBackend {
    /// Connect with a bounded pool and read the server version.
    pub async fn connect(url: &str) -> Result<Self, RelError> {
        let pool = MySqlPoolOptions::new()
            .max_connections(16)
            .acquire_timeout(std::time::Duration::from_secs(10))
            .connect(url)
            .await
            .map_err(classify)?;
        Self::from_pool(pool).await
    }

    /// Connect with a connection string held as a [`persistant_relational::ConnectionSecret`]:
    /// resolved here, never printed, never part of an error.
    pub async fn connect_secret(
        secret: &persistant_relational::ConnectionSecret,
    ) -> Result<Self, RelError> {
        let url = secret.resolve()?;
        Self::connect(url.expose()).await
    }

    /// Wrap an existing pool. The dialect is `MariaDb` when the server says so, else `MySql`.
    pub async fn from_pool(pool: sqlx::MySqlPool) -> Result<Self, RelError> {
        let full: String = sqlx::query_scalar("SELECT VERSION()")
            .fetch_one(&pool)
            .await
            .map_err(classify)?;
        // "8.4.9" -> MySQL; "11.4.5-MariaDB-ubu2404" -> MariaDB, version "11.4.5".
        let dialect = if full.contains("MariaDB") {
            Dialect::MariaDb
        } else {
            Dialect::MySql
        };
        let version = full.split(['-', ' ']).next().unwrap_or("").to_string();
        Ok(MysqlBackend {
            pool,
            version,
            dialect,
        })
    }

    /// The concrete pool.
    pub fn pool(&self) -> &sqlx::MySqlPool {
        &self.pool
    }
}

impl RelationalBackend for MysqlBackend {
    type Tx<'a> = MysqlTx;

    fn backend(&self) -> &str {
        match self.dialect {
            Dialect::MariaDb => "mariadb",
            _ => "mysql",
        }
    }

    fn version(&self) -> &str {
        &self.version
    }

    fn dialect(&self) -> Dialect {
        self.dialect.clone()
    }

    async fn execute(&self, sql: &str, params: &[Value]) -> Result<u64, RelError> {
        let q = bind(sqlx::query(AssertSqlSafe(sql.to_string())), params)?;
        q.execute(&self.pool)
            .await
            .map(|r| r.rows_affected())
            .map_err(classify)
    }

    async fn query(&self, sql: &str, params: &[Value]) -> Result<Vec<Row>, RelError> {
        let q = bind(sqlx::query(AssertSqlSafe(sql.to_string())), params)?;
        let rows = q.fetch_all(&self.pool).await.map_err(classify)?;
        rows.iter().map(decode).collect()
    }

    async fn begin(&self) -> Result<MysqlTx, RelError> {
        Ok(MysqlTx {
            // READ COMMITTED: under InnoDB's default REPEATABLE READ a locking read or an
            // `UPDATE` that matches no row takes a gap lock, and two transactions that then insert
            // into the same gap deadlock. The statement runs as a text query, so the isolation
            // level applies to exactly this transaction.
            tx: self
                .pool
                .begin_with("SET TRANSACTION ISOLATION LEVEL READ COMMITTED; START TRANSACTION")
                .await
                .map_err(classify)?,
        })
    }
}

/// A MySQL transaction.
#[derive(Debug)]
pub struct MysqlTx {
    tx: sqlx::Transaction<'static, MySql>,
}

impl RelationalTx for MysqlTx {
    async fn execute(&mut self, sql: &str, params: &[Value]) -> Result<u64, RelError> {
        let q = bind(sqlx::query(AssertSqlSafe(sql.to_string())), params)?;
        q.execute(&mut *self.tx)
            .await
            .map(|r| r.rows_affected())
            .map_err(classify)
    }

    async fn query(&mut self, sql: &str, params: &[Value]) -> Result<Vec<Row>, RelError> {
        let q = bind(sqlx::query(AssertSqlSafe(sql.to_string())), params)?;
        let rows = q.fetch_all(&mut *self.tx).await.map_err(classify)?;
        rows.iter().map(decode).collect()
    }

    async fn commit(self) -> Result<(), RelError> {
        self.tx.commit().await.map_err(classify)
    }

    async fn rollback(self) -> Result<(), RelError> {
        self.tx.rollback().await.map_err(classify)
    }
}