toasty_driver_mysql/
lib.rs

1#![warn(missing_docs)]
2#![allow(clippy::needless_range_loop)]
3
4//! Toasty driver for [MySQL](https://www.mysql.com/) using
5//! [SQLx](https://docs.rs/sqlx).
6//!
7//! # Examples
8//!
9//! ```no_run
10//! use toasty_driver_mysql::MySQL;
11//!
12//! let driver = MySQL::new("mysql://localhost/mydb").unwrap();
13//! ```
14
15mod value;
16pub(crate) use value::Value;
17
18use async_trait::async_trait;
19use sqlx_core::{
20    connection::{ConnectOptions as _, Connection as SqlxConnection},
21    row::Row,
22    sql_str::AssertSqlSafe,
23};
24use sqlx_mysql::{MySqlArguments, MySqlConnectOptions, MySqlConnection, MySqlDatabaseError};
25use std::{borrow::Cow, cell::Cell, sync::Arc};
26use toasty_core::{
27    Result, Schema,
28    driver::{
29        Capability, ConnectContext, ConnectionUrl, Driver, ExecResponse, Operation, QueryLogConfig,
30        log::QueryLog,
31        operation::{RawSqlRet, Transaction, TransactionMode},
32    },
33    schema::{
34        db::{self, Migration, Table},
35        diff,
36    },
37    stmt::{self, ValueRecord},
38};
39use toasty_sql::{self as sql};
40
41enum SqlReturn {
42    Count,
43    LastInsertId(stmt::Type),
44    Infer,
45    Types(Vec<stmt::Type>),
46}
47
48/// Classifies a SQLx MySQL error into a Toasty error.
49///
50/// Transport and protocol failures become `ConnectionLost`. MySQL
51/// errors with numbers that Toasty understands become typed errors.
52/// Everything else becomes `DriverOperationFailed`.
53fn classify_mysql_error(e: sqlx_core::Error) -> toasty_core::Error {
54    match e {
55        error @ (sqlx_core::Error::Io(_)
56        | sqlx_core::Error::Protocol(_)
57        | sqlx_core::Error::WorkerCrashed) => toasty_core::Error::connection_lost(error),
58        sqlx_core::Error::Database(database_error) => {
59            let mysql_error = database_error
60                .try_downcast_ref::<MySqlDatabaseError>()
61                .expect("SQLx returned a non-MySQL database error from a MySQL connection");
62            let number = mysql_error.number();
63            let message = mysql_error.message().to_owned();
64
65            match number {
66                1213 => toasty_core::Error::serialization_failure(message),
67                1792 => toasty_core::Error::read_only_transaction(message),
68                _ => toasty_core::Error::driver_operation_failed(sqlx_core::Error::Database(
69                    database_error,
70                )),
71            }
72        }
73        other => toasty_core::Error::driver_operation_failed(other),
74    }
75}
76
77/// Classifies a SQLx error and records whether the connection is still usable.
78fn record_mysql_err(valid: &Cell<bool>, e: sqlx_core::Error) -> toasty_core::Error {
79    let err = classify_mysql_error(e);
80    if err.is_connection_lost() {
81        valid.set(false);
82    }
83    err
84}
85
86/// A MySQL [`Driver`] that connects through SQLx.
87///
88/// # Examples
89///
90/// ```no_run
91/// use toasty_driver_mysql::MySQL;
92///
93/// let driver = MySQL::new("mysql://localhost/mydb").unwrap();
94/// ```
95#[derive(Debug)]
96pub struct MySQL {
97    url: String,
98    opts: MySqlConnectOptions,
99}
100
101impl MySQL {
102    /// Creates a MySQL driver from a SQLx connection URL.
103    ///
104    /// The URL must use the `mysql` scheme and include a database path, such as
105    /// `mysql://user:pass@host:3306/dbname`.
106    pub fn new(url: impl Into<String>) -> Result<Self> {
107        let url_str = url.into();
108        let url = ConnectionUrl::parse(&url_str)?;
109
110        if !url.has_scheme("mysql") {
111            return Err(toasty_core::Error::invalid_connection_url(format!(
112                "connection url does not have a `mysql` scheme; url={}",
113                url.as_str()
114            )));
115        }
116
117        url.host()?.ok_or_else(|| {
118            toasty_core::Error::invalid_connection_url(format!(
119                "missing host in connection URL; url={}",
120                url.as_str()
121            ))
122        })?;
123
124        if url.path().is_empty() {
125            return Err(toasty_core::Error::invalid_connection_url(format!(
126                "no database specified - missing path in connection URL; url={}",
127                url.as_str()
128            )));
129        }
130
131        let opts = url
132            .as_str()
133            .parse::<MySqlConnectOptions>()
134            .map_err(toasty_core::Error::driver_operation_failed)?
135            .disable_statement_logging();
136
137        Ok(Self { url: url_str, opts })
138    }
139}
140
141#[async_trait]
142impl Driver for MySQL {
143    fn url(&self) -> Cow<'_, str> {
144        Cow::Borrowed(&self.url)
145    }
146
147    fn capability(&self) -> &'static Capability {
148        &Capability::MYSQL
149    }
150
151    async fn connect(
152        &self,
153        cx: &ConnectContext,
154    ) -> Result<Box<dyn toasty_core::driver::Connection>> {
155        let conn = MySqlConnection::connect_with(&self.opts)
156            .await
157            .map_err(classify_mysql_error)?;
158        let mut connection = Connection::new(conn);
159        connection.query_log = cx.query_log;
160        Ok(Box::new(connection))
161    }
162
163    fn generate_migration(&self, schema_diff: &diff::Schema<'_>) -> Migration {
164        let statements = sql::MigrationStatement::from_diff(schema_diff, &Capability::MYSQL);
165
166        let sql_strings: Vec<String> = statements
167            .iter()
168            .map(|stmt| sql::Serializer::mysql(stmt.schema()).serialize(stmt.statement()))
169            .collect();
170
171        Migration::new_sql_with_breakpoints(&sql_strings)
172    }
173
174    async fn reset_db(&self) -> Result<()> {
175        let mut conn = MySqlConnection::connect_with(&self.opts)
176            .await
177            .map_err(classify_mysql_error)?;
178        let dbname = self.opts.get_database().ok_or_else(|| {
179            toasty_core::Error::invalid_connection_url("no database name configured")
180        })?;
181
182        let dbname = format!("`{}`", dbname.replace('`', "``"));
183
184        sqlx_core::raw_sql::raw_sql(AssertSqlSafe(format!("DROP DATABASE IF EXISTS {dbname}")))
185            .execute(&mut conn)
186            .await
187            .map_err(classify_mysql_error)?;
188        sqlx_core::raw_sql::raw_sql(AssertSqlSafe(format!("CREATE DATABASE {dbname}")))
189            .execute(&mut conn)
190            .await
191            .map_err(classify_mysql_error)?;
192        sqlx_core::raw_sql::raw_sql(AssertSqlSafe(format!("USE {dbname}")))
193            .execute(&mut conn)
194            .await
195            .map_err(classify_mysql_error)?;
196
197        Ok(())
198    }
199}
200
201/// An open connection to a MySQL database.
202#[derive(Debug)]
203pub struct Connection {
204    conn: MySqlConnection,
205    /// Set to `false` after a connection-level failure. SQLx does not expose a
206    /// passive validity flag, so the driver records one for [`is_valid`].
207    valid: Cell<bool>,
208    query_log: QueryLogConfig,
209}
210
211impl Connection {
212    /// Wraps an existing SQLx [`MySqlConnection`] as a Toasty connection.
213    pub fn new(conn: MySqlConnection) -> Self {
214        Self {
215            conn,
216            valid: Cell::new(true),
217            query_log: QueryLogConfig::default(),
218        }
219    }
220
221    async fn exec_sql(
222        &mut self,
223        sql_as_str: &str,
224        args: MySqlArguments,
225        ret: SqlReturn,
226        log: &mut QueryLog<'_>,
227    ) -> Result<ExecResponse> {
228        if matches!(ret, SqlReturn::Count | SqlReturn::LastInsertId(_)) {
229            let result = sqlx_core::query::query_with(AssertSqlSafe(sql_as_str), args)
230                .execute(&mut self.conn)
231                .await
232                .map_err(|e| record_mysql_err(&self.valid, e))?;
233
234            if let SqlReturn::LastInsertId(ty) = ret {
235                let id = ty.cast(&(), stmt::Value::U64(result.last_insert_id()))?;
236                log.rows(1);
237                return Ok(ExecResponse::value_stream(stmt::ValueStream::from_vec(
238                    vec![ValueRecord::from_vec(vec![id]).into()],
239                )));
240            }
241
242            return Ok(ExecResponse::count(result.rows_affected()));
243        }
244
245        let rows = sqlx_core::query::query_with(AssertSqlSafe(sql_as_str), args)
246            .fetch_all(&mut self.conn)
247            .await
248            .map_err(|e| record_mysql_err(&self.valid, e))?;
249
250        log.rows(rows.len() as u64);
251        let mut records = Vec::with_capacity(rows.len());
252
253        for row in rows {
254            let mut values = Vec::with_capacity(row.len());
255
256            match &ret {
257                SqlReturn::Count | SqlReturn::LastInsertId(_) => unreachable!(),
258                SqlReturn::Infer => {
259                    for i in 0..row.len() {
260                        let column = row.column(i);
261                        values.push(
262                            Value::from_sql_infer(i, &row, column)
263                                .map_err(|e| record_mysql_err(&self.valid, e))?
264                                .into_inner(),
265                        );
266                    }
267                }
268                SqlReturn::Types(returning) => {
269                    assert_eq!(
270                        row.len(),
271                        returning.len(),
272                        "row={row:#?}; returning={returning:#?}"
273                    );
274
275                    for i in 0..row.len() {
276                        let column = row.column(i);
277                        values.push(
278                            Value::from_sql(i, &row, column, &returning[i])
279                                .map_err(|e| record_mysql_err(&self.valid, e))?
280                                .into_inner(),
281                        );
282                    }
283                }
284            }
285
286            records.push(Ok(ValueRecord::from_vec(values)));
287        }
288
289        Ok(ExecResponse::value_stream(stmt::ValueStream::from_iter(
290            records.into_iter(),
291        )))
292    }
293
294    /// Creates a table and its indices from a schema definition.
295    pub async fn create_table(&mut self, schema: &db::Schema, table: &Table) -> Result<()> {
296        let serializer = sql::Serializer::mysql(schema);
297        let statement =
298            serializer.serialize(&sql::Statement::create_table(table, &Capability::MYSQL));
299
300        sqlx_core::query::query(AssertSqlSafe(statement))
301            .execute(&mut self.conn)
302            .await
303            .map_err(|e| record_mysql_err(&self.valid, e))?;
304
305        for index in &table.indices {
306            if index.primary_key {
307                continue;
308            }
309
310            let statement = serializer.serialize(&sql::Statement::create_index(index));
311            sqlx_core::query::query(AssertSqlSafe(statement))
312                .execute(&mut self.conn)
313                .await
314                .map_err(|e| record_mysql_err(&self.valid, e))?;
315        }
316
317        Ok(())
318    }
319}
320
321impl From<MySqlConnection> for Connection {
322    fn from(conn: MySqlConnection) -> Self {
323        Self::new(conn)
324    }
325}
326
327#[async_trait]
328impl toasty_core::driver::Connection for Connection {
329    async fn exec(&mut self, schema: &Arc<Schema>, op: Operation) -> Result<ExecResponse> {
330        tracing::trace!(driver = "mysql", op = %op.name(), "driver exec");
331
332        let (sql, typed_params, ret) = match op {
333            Operation::Insert(op) => {
334                let ret = match op.ret {
335                    Some(types) => {
336                        let [ty] = &types[..] else {
337                            return Err(toasty_core::Error::invalid_result(format!(
338                                "MySQL insert ID result requires one type, got {types:?}"
339                            )));
340                        };
341                        SqlReturn::LastInsertId(ty.clone())
342                    }
343                    None => SqlReturn::Count,
344                };
345                (sql::Statement::from(op.stmt), op.params, ret)
346            }
347            Operation::QuerySql(op) => {
348                let ret = match op.ret {
349                    Some(types) => SqlReturn::Types(types),
350                    None => SqlReturn::Count,
351                };
352                (sql::Statement::from(op.stmt), op.params, ret)
353            }
354            Operation::RawSql(op) => {
355                let ret = match op.ret {
356                    RawSqlRet::None => SqlReturn::Count,
357                    RawSqlRet::Infer => SqlReturn::Infer,
358                    RawSqlRet::Types(types) => SqlReturn::Types(types),
359                };
360                let mut log = QueryLog::sql(
361                    &self.query_log,
362                    "mysql",
363                    &op.sql,
364                    op.params.iter().map(|tv| &tv.value),
365                );
366                let mut args = MySqlArguments::default();
367                for param in op.params {
368                    Value::from(param.value)
369                        .add_to(&mut args)
370                        .map_err(|e| record_mysql_err(&self.valid, e))?;
371                }
372                let result = self.exec_sql(&op.sql, args, ret, &mut log).await;
373                log.finish(&result);
374                return result;
375            }
376            Operation::Transaction(op) => {
377                if let Transaction::Start {
378                    mode: mode @ (TransactionMode::Immediate | TransactionMode::Exclusive),
379                    ..
380                } = &op
381                {
382                    return Err(toasty_core::Error::unsupported_feature(format!(
383                        "MySQL does not support TransactionMode::{mode:?}"
384                    )));
385                }
386                let statement = sql::Serializer::mysql(&schema.db).serialize_transaction(&op);
387                sqlx_core::raw_sql::raw_sql(AssertSqlSafe(statement))
388                    .execute(&mut self.conn)
389                    .await
390                    .map_err(|e| record_mysql_err(&self.valid, e))?;
391                return Ok(ExecResponse::count(0));
392            }
393            op => todo!("op={op:#?}"),
394        };
395
396        let (sql_as_str, arg_order) =
397            sql::Serializer::mysql(&schema.db).serialize_with_arg_order(&sql);
398
399        let mut log = QueryLog::sql(
400            &self.query_log,
401            "mysql",
402            &sql_as_str,
403            arg_order.iter().map(|&pos| &typed_params[pos].value),
404        );
405
406        // MySQL uses positional `?` placeholders, so parameters must follow the
407        // order in which their `Expr::Arg(n)` values appear in serialized SQL.
408        let mut remaining = vec![0usize; typed_params.len()];
409        for &pos in &arg_order {
410            remaining[pos] += 1;
411        }
412        let mut values = typed_params
413            .into_iter()
414            .map(|param| Some(param.value))
415            .collect::<Vec<_>>();
416        let mut args = MySqlArguments::default();
417        for pos in arg_order {
418            remaining[pos] -= 1;
419            let value = if remaining[pos] == 0 {
420                values[pos].take().expect("MySQL parameter already moved")
421            } else {
422                values[pos]
423                    .as_ref()
424                    .expect("MySQL parameter missing")
425                    .clone()
426            };
427            Value::from(value)
428                .add_to(&mut args)
429                .map_err(|e| record_mysql_err(&self.valid, e))?;
430        }
431
432        let result = self.exec_sql(&sql_as_str, args, ret, &mut log).await;
433        log.finish(&result);
434        result
435    }
436
437    async fn push_schema(&mut self, schema: &Schema) -> Result<()> {
438        for table in &schema.db.tables {
439            tracing::debug!(table = %table.name, "creating table");
440            self.create_table(&schema.db, table).await?;
441        }
442        Ok(())
443    }
444
445    async fn applied_migrations(
446        &mut self,
447    ) -> Result<Vec<toasty_core::schema::db::AppliedMigration>> {
448        sqlx_core::query::query(
449            "CREATE TABLE IF NOT EXISTS __toasty_migrations (
450                id BIGINT UNSIGNED PRIMARY KEY,
451                name TEXT NOT NULL,
452                applied_at TIMESTAMP NOT NULL
453            )",
454        )
455        .execute(&mut self.conn)
456        .await
457        .map_err(|e| record_mysql_err(&self.valid, e))?;
458
459        let ids = sqlx_core::query_scalar::query_scalar::<sqlx_mysql::MySql, u64>(
460            "SELECT id FROM __toasty_migrations ORDER BY applied_at",
461        )
462        .fetch_all(&mut self.conn)
463        .await
464        .map_err(|e| record_mysql_err(&self.valid, e))?;
465
466        Ok(ids
467            .into_iter()
468            .map(toasty_core::schema::db::AppliedMigration::new)
469            .collect())
470    }
471
472    async fn apply_migration(
473        &mut self,
474        id: u64,
475        name: &str,
476        migration: &toasty_core::schema::db::Migration,
477    ) -> Result<()> {
478        tracing::info!(id, name, "applying migration");
479        sqlx_core::query::query(
480            "CREATE TABLE IF NOT EXISTS __toasty_migrations (
481                id BIGINT UNSIGNED PRIMARY KEY,
482                name TEXT NOT NULL,
483                applied_at TIMESTAMP NOT NULL
484            )",
485        )
486        .execute(&mut self.conn)
487        .await
488        .map_err(|e| record_mysql_err(&self.valid, e))?;
489
490        let mut transaction = self
491            .conn
492            .begin()
493            .await
494            .map_err(|e| record_mysql_err(&self.valid, e))?;
495
496        for statement in migration.statements() {
497            if let Err(error) = sqlx_core::raw_sql::raw_sql(AssertSqlSafe(statement))
498                .execute(&mut *transaction)
499                .await
500            {
501                let error = record_mysql_err(&self.valid, error);
502                transaction
503                    .rollback()
504                    .await
505                    .map_err(|e| record_mysql_err(&self.valid, e))?;
506                return Err(error);
507            }
508        }
509
510        if let Err(error) = sqlx_core::query::query(
511            "INSERT INTO __toasty_migrations (id, name, applied_at) VALUES (?, ?, NOW())",
512        )
513        .bind(id)
514        .bind(name)
515        .execute(&mut *transaction)
516        .await
517        {
518            let error = record_mysql_err(&self.valid, error);
519            transaction
520                .rollback()
521                .await
522                .map_err(|e| record_mysql_err(&self.valid, e))?;
523            return Err(error);
524        }
525
526        transaction
527            .commit()
528            .await
529            .map_err(|e| record_mysql_err(&self.valid, e))?;
530        Ok(())
531    }
532
533    fn is_valid(&self) -> bool {
534        self.valid.get()
535    }
536
537    async fn ping(&mut self) -> Result<()> {
538        match self.conn.ping().await {
539            Ok(()) => Ok(()),
540            Err(error) => {
541                self.valid.set(false);
542                Err(toasty_core::Error::connection_lost(error))
543            }
544        }
545    }
546}
toasty_driver_mysql/lib.rs

toasty_driver_mysql/
lib.rs
