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//! [`mysql_async`](https://docs.rs/mysql_async).
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 mysql_async::{
20    Conn, OptsBuilder,
21    prelude::{Queryable, ToValue},
22};
23use std::{borrow::Cow, cell::Cell, sync::Arc};
24use toasty_core::{
25    Result, Schema,
26    driver::{
27        Capability, ConnectContext, Driver, ExecResponse, Operation, QueryLogConfig,
28        log::QueryLog,
29        operation::{RawSqlRet, Transaction, TransactionMode},
30    },
31    schema::{
32        db::{self, Migration, Table},
33        diff,
34    },
35    stmt::{self, ValueRecord},
36};
37use toasty_sql::{self as sql};
38use url::Url;
39
40enum SqlReturn {
41    Count {
42        last_insert_id_hack: Option<u64>,
43        sql_is_insert: bool,
44    },
45    Infer,
46    Types(Vec<stmt::Type>),
47}
48
49/// Classifies a `mysql_async::Error` into a Toasty error.
50///
51/// `Error::Io` (any TCP/TLS-level fault) and the IO-shaped `Driver`
52/// variants (`ConnectionClosed`, `PoolDisconnected`) become
53/// `ConnectionLost`. `Server` errors with known SQLSTATE codes are
54/// mapped to typed variants. Everything else is
55/// `DriverOperationFailed`.
56fn classify_mysql_error(e: mysql_async::Error) -> toasty_core::Error {
57    use mysql_async::{DriverError, Error};
58    match e {
59        Error::Io(_) => toasty_core::Error::connection_lost(e),
60        Error::Driver(DriverError::ConnectionClosed | DriverError::PoolDisconnected) => {
61            toasty_core::Error::connection_lost(e)
62        }
63        Error::Server(se) => match se.code {
64            1213 => toasty_core::Error::serialization_failure(se.message),
65            1792 => toasty_core::Error::read_only_transaction(se.message),
66            _ => toasty_core::Error::driver_operation_failed(Error::Server(se)),
67        },
68        other => toasty_core::Error::driver_operation_failed(other),
69    }
70}
71
72/// Classify a `mysql_async::Error`, also flipping the connection's
73/// validity flag if the error indicates the connection is gone.
74fn record_mysql_err(valid: &Cell<bool>, e: mysql_async::Error) -> toasty_core::Error {
75    let err = classify_mysql_error(e);
76    if err.is_connection_lost() {
77        valid.set(false);
78    }
79    err
80}
81
82/// A MySQL [`Driver`] that connects via `mysql_async`.
83///
84/// # Examples
85///
86/// ```no_run
87/// use toasty_driver_mysql::MySQL;
88///
89/// let driver = MySQL::new("mysql://localhost/mydb").unwrap();
90/// ```
91#[derive(Debug)]
92pub struct MySQL {
93    url: String,
94    opts: OptsBuilder,
95}
96
97impl MySQL {
98    /// Create a new MySQL driver from a connection URL.
99    ///
100    /// The URL must use the `mysql` scheme and include a database path, e.g.
101    /// `mysql://user:pass@host:3306/dbname`.
102    pub fn new(url: impl Into<String>) -> Result<Self> {
103        let url_str = url.into();
104        let url = Url::parse(&url_str).map_err(toasty_core::Error::driver_operation_failed)?;
105
106        if url.scheme() != "mysql" {
107            return Err(toasty_core::Error::invalid_connection_url(format!(
108                "connection url does not have a `mysql` scheme; url={}",
109                url
110            )));
111        }
112
113        url.host_str().ok_or_else(|| {
114            toasty_core::Error::invalid_connection_url(format!(
115                "missing host in connection URL; url={}",
116                url
117            ))
118        })?;
119
120        if url.path().is_empty() {
121            return Err(toasty_core::Error::invalid_connection_url(format!(
122                "no database specified - missing path in connection URL; url={}",
123                url
124            )));
125        }
126
127        let opts = mysql_async::Opts::from_url(url.as_ref())
128            .map_err(toasty_core::Error::driver_operation_failed)?;
129        let opts = mysql_async::OptsBuilder::from_opts(opts).client_found_rows(true);
130
131        Ok(Self { url: url_str, opts })
132    }
133}
134
135#[async_trait]
136impl Driver for MySQL {
137    fn url(&self) -> Cow<'_, str> {
138        Cow::Borrowed(&self.url)
139    }
140
141    fn capability(&self) -> &'static Capability {
142        &Capability::MYSQL
143    }
144
145    async fn connect(
146        &self,
147        cx: &ConnectContext,
148    ) -> Result<Box<dyn toasty_core::driver::Connection>> {
149        let conn = Conn::new(self.opts.clone())
150            .await
151            .map_err(classify_mysql_error)?;
152        let mut connection = Connection::new(conn);
153        connection.query_log = cx.query_log;
154        Ok(Box::new(connection))
155    }
156
157    fn generate_migration(&self, schema_diff: &diff::Schema<'_>) -> Migration {
158        let statements = sql::MigrationStatement::from_diff(schema_diff, &Capability::MYSQL);
159
160        let sql_strings: Vec<String> = statements
161            .iter()
162            .map(|stmt| sql::Serializer::mysql(stmt.schema()).serialize(stmt.statement()))
163            .collect();
164
165        Migration::new_sql_with_breakpoints(&sql_strings)
166    }
167
168    async fn reset_db(&self) -> toasty_core::Result<()> {
169        let mut conn = Conn::new(self.opts.clone())
170            .await
171            .map_err(classify_mysql_error)?;
172
173        let dbname = conn
174            .opts()
175            .db_name()
176            .ok_or_else(|| {
177                toasty_core::Error::invalid_connection_url("no database name configured")
178            })?
179            .to_string();
180
181        conn.query_drop(format!("DROP DATABASE IF EXISTS `{}`", dbname))
182            .await
183            .map_err(classify_mysql_error)?;
184
185        conn.query_drop(format!("CREATE DATABASE `{}`", dbname))
186            .await
187            .map_err(classify_mysql_error)?;
188
189        conn.query_drop(format!("USE `{}`", dbname))
190            .await
191            .map_err(classify_mysql_error)?;
192
193        Ok(())
194    }
195}
196
197/// An open connection to a MySQL database.
198#[derive(Debug)]
199pub struct Connection {
200    conn: Conn,
201    /// Set to `false` once `exec` has observed a connection-lost
202    /// error. `mysql_async::Conn` does not expose a passive flag, so
203    /// the driver tracks one itself. Read by [`is_valid`].
204    valid: Cell<bool>,
205    query_log: QueryLogConfig,
206}
207
208impl Connection {
209    /// Wrap an existing [`mysql_async::Conn`] as a Toasty connection.
210    pub fn new(conn: Conn) -> Self {
211        Self {
212            conn,
213            valid: Cell::new(true),
214            query_log: QueryLogConfig::default(),
215        }
216    }
217
218    async fn exec_sql(
219        &mut self,
220        sql_as_str: &str,
221        args: Vec<mysql_async::Value>,
222        ret: SqlReturn,
223        log: &mut QueryLog<'_>,
224    ) -> Result<ExecResponse> {
225        let statement = self
226            .conn
227            .prep(sql_as_str)
228            .await
229            .map_err(|e| record_mysql_err(&self.valid, e))?;
230
231        if let SqlReturn::Count {
232            last_insert_id_hack,
233            sql_is_insert,
234        } = ret
235        {
236            let count = self
237                .conn
238                .exec_iter(&statement, mysql_async::Params::Positional(args))
239                .await
240                .map_err(|e| record_mysql_err(&self.valid, e))?
241                .affected_rows();
242
243            if let Some(num_rows) = last_insert_id_hack {
244                assert!(
245                    sql_is_insert,
246                    "last_insert_id_hack should only be used with INSERT statements"
247                );
248
249                let first_id: u64 = self
250                    .conn
251                    .query_first("SELECT LAST_INSERT_ID()")
252                    .await
253                    .map_err(|e| record_mysql_err(&self.valid, e))?
254                    .ok_or_else(|| {
255                        toasty_core::Error::driver_operation_failed(std::io::Error::other(
256                            "LAST_INSERT_ID() returned no rows",
257                        ))
258                    })?;
259
260                let results = (0..num_rows).map(move |offset| {
261                    let id = first_id + offset;
262                    Ok(ValueRecord::from_vec(vec![stmt::Value::U64(id)]))
263                });
264
265                log.rows(num_rows);
266                return Ok(ExecResponse::value_stream(stmt::ValueStream::from_iter(
267                    results,
268                )));
269            }
270
271            return Ok(ExecResponse::count(count));
272        }
273
274        let rows: Vec<mysql_async::Row> = self
275            .conn
276            .exec(&statement, &args)
277            .await
278            .map_err(|e| record_mysql_err(&self.valid, e))?;
279
280        log.rows(rows.len() as u64);
281
282        let results = rows.into_iter().map(move |mut row| {
283            let mut results = Vec::new();
284
285            match &ret {
286                SqlReturn::Count { .. } => unreachable!(),
287                SqlReturn::Infer => {
288                    for i in 0..row.len() {
289                        let column = row.columns()[i].clone();
290                        results.push(Value::from_sql_infer(i, &mut row, &column).into_inner());
291                    }
292                }
293                SqlReturn::Types(returning) => {
294                    assert_eq!(
295                        row.len(),
296                        returning.len(),
297                        "row={row:#?}; returning={returning:#?}"
298                    );
299
300                    for i in 0..row.len() {
301                        let column = row.columns()[i].clone();
302                        results.push(
303                            Value::from_sql(i, &mut row, &column, &returning[i]).into_inner(),
304                        );
305                    }
306                }
307            }
308
309            Ok(ValueRecord::from_vec(results))
310        });
311
312        Ok(ExecResponse::value_stream(stmt::ValueStream::from_iter(
313            results,
314        )))
315    }
316
317    /// Create a table and its indices from a schema definition.
318    pub async fn create_table(&mut self, schema: &db::Schema, table: &Table) -> Result<()> {
319        let serializer = sql::Serializer::mysql(schema);
320
321        let sql = serializer.serialize(&sql::Statement::create_table(table, &Capability::MYSQL));
322
323        self.conn
324            .exec_drop(&sql, ())
325            .await
326            .map_err(|e| record_mysql_err(&self.valid, e))?;
327
328        for index in &table.indices {
329            if index.primary_key {
330                continue;
331            }
332
333            let sql = serializer.serialize(&sql::Statement::create_index(index));
334
335            self.conn
336                .exec_drop(&sql, ())
337                .await
338                .map_err(|e| record_mysql_err(&self.valid, e))?;
339        }
340
341        Ok(())
342    }
343}
344
345impl From<Conn> for Connection {
346    fn from(conn: Conn) -> Self {
347        Self::new(conn)
348    }
349}
350
351#[async_trait]
352impl toasty_core::driver::Connection for Connection {
353    async fn exec(&mut self, schema: &Arc<Schema>, op: Operation) -> Result<ExecResponse> {
354        tracing::trace!(driver = "mysql", op = %op.name(), "driver exec");
355
356        let (sql, typed_params, ret, last_insert_id_hack) = match op {
357            Operation::QuerySql(op) => (
358                sql::Statement::from(op.stmt),
359                op.params,
360                op.ret,
361                op.last_insert_id_hack,
362            ),
363            Operation::RawSql(op) => {
364                let args = op
365                    .params
366                    .iter()
367                    .map(|tv| Value::from(tv.value.clone()).to_value())
368                    .collect();
369                let ret = match op.ret {
370                    RawSqlRet::None => SqlReturn::Count {
371                        last_insert_id_hack: None,
372                        sql_is_insert: false,
373                    },
374                    RawSqlRet::Infer => SqlReturn::Infer,
375                    RawSqlRet::Types(types) => SqlReturn::Types(types),
376                };
377                let mut log = QueryLog::sql(
378                    &self.query_log,
379                    "mysql",
380                    &op.sql,
381                    op.params.iter().map(|tv| &tv.value),
382                );
383                let result = self.exec_sql(&op.sql, args, ret, &mut log).await;
384                log.finish(&result);
385                return result;
386            }
387            Operation::Transaction(op) => {
388                // MySQL has no `BEGIN IMMEDIATE` / `BEGIN EXCLUSIVE`
389                // analogue; reject non-Default modes loudly rather than
390                // silently dropping them at the serializer.
391                if let Transaction::Start {
392                    mode: mode @ (TransactionMode::Immediate | TransactionMode::Exclusive),
393                    ..
394                } = &op
395                {
396                    return Err(toasty_core::Error::unsupported_feature(format!(
397                        "MySQL does not support TransactionMode::{mode:?}"
398                    )));
399                }
400                let sql = sql::Serializer::mysql(&schema.db).serialize_transaction(&op);
401                self.conn
402                    .query_drop(sql)
403                    .await
404                    .map_err(|e| record_mysql_err(&self.valid, e))?;
405                return Ok(ExecResponse::count(0));
406            }
407            op => todo!("op={:#?}", op),
408        };
409
410        let (sql_as_str, arg_order) =
411            sql::Serializer::mysql(&schema.db).serialize_with_arg_order(&sql);
412
413        // MySQL uses positional `?` without indices, so params must be reordered
414        // to match the order `Expr::Arg(n)` placeholders appear in the SQL.
415        let params: Vec<_> = arg_order
416            .iter()
417            .map(|&pos| Value::from(typed_params[pos].value.clone()))
418            .collect();
419        let args = params
420            .iter()
421            .map(|param| param.to_value())
422            .collect::<Vec<_>>();
423
424        let ret = match ret {
425            Some(types) => SqlReturn::Types(types),
426            None => SqlReturn::Count {
427                last_insert_id_hack,
428                sql_is_insert: matches!(sql, sql::Statement::Insert(_)),
429            },
430        };
431
432        let mut log = QueryLog::sql(
433            &self.query_log,
434            "mysql",
435            &sql_as_str,
436            params.iter().map(|value| value.inner()),
437        );
438        let result = self.exec_sql(&sql_as_str, args, ret, &mut log).await;
439        log.finish(&result);
440        result
441    }
442
443    async fn push_schema(&mut self, schema: &Schema) -> Result<()> {
444        for table in &schema.db.tables {
445            tracing::debug!(table = %table.name, "creating table");
446            self.create_table(&schema.db, table).await?;
447        }
448        Ok(())
449    }
450
451    async fn applied_migrations(
452        &mut self,
453    ) -> Result<Vec<toasty_core::schema::db::AppliedMigration>> {
454        // Ensure the migrations table exists
455        self.conn
456            .exec_drop(
457                "CREATE TABLE IF NOT EXISTS __toasty_migrations (
458                id BIGINT UNSIGNED PRIMARY KEY,
459                name TEXT NOT NULL,
460                applied_at TIMESTAMP NOT NULL
461            )",
462                (),
463            )
464            .await
465            .map_err(|e| record_mysql_err(&self.valid, e))?;
466
467        // Query all applied migrations
468        let rows: Vec<u64> = self
469            .conn
470            .exec("SELECT id FROM __toasty_migrations ORDER BY applied_at", ())
471            .await
472            .map_err(|e| record_mysql_err(&self.valid, e))?;
473
474        Ok(rows
475            .into_iter()
476            .map(toasty_core::schema::db::AppliedMigration::new)
477            .collect())
478    }
479
480    async fn apply_migration(
481        &mut self,
482        id: u64,
483        name: &str,
484        migration: &toasty_core::schema::db::Migration,
485    ) -> Result<()> {
486        tracing::info!(id = id, name = %name, "applying migration");
487        // Ensure the migrations table exists
488        self.conn
489            .exec_drop(
490                "CREATE TABLE IF NOT EXISTS __toasty_migrations (
491                id BIGINT UNSIGNED PRIMARY KEY,
492                name TEXT NOT NULL,
493                applied_at TIMESTAMP NOT NULL
494            )",
495                (),
496            )
497            .await
498            .map_err(|e| record_mysql_err(&self.valid, e))?;
499
500        // Start transaction
501        let mut transaction = self
502            .conn
503            .start_transaction(Default::default())
504            .await
505            .map_err(|e| record_mysql_err(&self.valid, e))?;
506
507        // Execute each migration statement
508        for statement in migration.statements() {
509            if let Err(e) = transaction
510                .query_drop(statement)
511                .await
512                .map_err(|e| record_mysql_err(&self.valid, e))
513            {
514                transaction
515                    .rollback()
516                    .await
517                    .map_err(|e| record_mysql_err(&self.valid, e))?;
518                return Err(e);
519            }
520        }
521
522        // Record the migration
523        if let Err(e) = transaction
524            .exec_drop(
525                "INSERT INTO __toasty_migrations (id, name, applied_at) VALUES (?, ?, NOW())",
526                (id, name),
527            )
528            .await
529            .map_err(|e| record_mysql_err(&self.valid, e))
530        {
531            transaction
532                .rollback()
533                .await
534                .map_err(|e| record_mysql_err(&self.valid, e))?;
535            return Err(e);
536        }
537
538        // Commit transaction
539        transaction
540            .commit()
541            .await
542            .map_err(|e| record_mysql_err(&self.valid, e))?;
543        Ok(())
544    }
545
546    fn is_valid(&self) -> bool {
547        self.valid.get()
548    }
549
550    async fn ping(&mut self) -> Result<()> {
551        // `COM_PING` is the cheapest server round-trip in the MySQL
552        // protocol. Any failure is surfaced as `connection_lost`: the
553        // only meaningful outcome of a ping is "the connection is
554        // alive" or "evict it." Also flip the validity flag so a
555        // subsequent `is_valid` check observes the dead connection.
556        match self.conn.ping().await {
557            Ok(()) => Ok(()),
558            Err(e) => {
559                self.valid.set(false);
560                Err(toasty_core::Error::connection_lost(e))
561            }
562        }
563    }
564}
toasty_driver_mysql/lib.rs

toasty_driver_mysql/
lib.rs
