Transactions for migations

src/database_drivers/libsql.rs

@@ -1,6 +1,7 @@
 use crate::database_drivers::DatabaseDriver;
 use anyhow::{bail, Result};
 use libsql_client::{de, Client, Config};
+use log::error;
 use std::future::Future;
 use std::pin::Pin;
 
@@ -38,8 +39,8 @@ impl DatabaseDriver for LibSQLDriver {
                     tx.commit().await?;
                 }
                 Err(err) => {
+                    error!("Error executing query: {}", err);
                     tx.rollback().await?;
-                    bail!("{:?}", err);
                 }
             }
             Ok(())

src/database_drivers/maria.rs

@@ -1,7 +1,9 @@
 use crate::config;
 use crate::database_drivers::DatabaseDriver;
 use anyhow::{bail, Result};
+use log::error;
 use sqlx::mysql::MySqlRow;
+use sqlx::Executor;
 use sqlx::{Connection, MySqlConnection, Row};
 use std::future::Future;
 use std::pin::Pin;
@@ -60,7 +62,18 @@ impl DatabaseDriver for MariaDBDriver {
         query: &'a str,
     ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + '_>> {
         let fut = async move {
-            sqlx::query(query).execute(&mut self.db).await?;
+            let mut tx = self.db.begin().await?;
+
+            match tx.execute(query).await {
+                Ok(_) => {
+                    tx.commit().await?;
+                }
+                Err(e) => {
+                    error!("Error executing query: {}", e);
+                    tx.rollback().await?;
+                }
+            }
+
             Ok(())
         };
 

src/database_drivers/mysql.rs

@@ -1,7 +1,9 @@
 use crate::config;
 use crate::database_drivers::{utils, DatabaseDriver};
 use anyhow::{bail, Result};
+use log::error;
 use sqlx::mysql::MySqlRow;
+use sqlx::Executor;
 use sqlx::{Connection, MySqlConnection, Row};
 use std::future::Future;
 use std::pin::Pin;
@@ -58,7 +60,18 @@ impl DatabaseDriver for MySQLDriver {
         query: &'a str,
     ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + '_>> {
         let fut = async move {
-            sqlx::query(query).execute(&mut self.db).await?;
+            let mut tx = self.db.begin().await?;
+
+            match tx.execute(query).await {
+                Ok(_) => {
+                    tx.commit().await?;
+                }
+                Err(e) => {
+                    error!("Error executing query: {}", e);
+                    tx.rollback().await?;
+                }
+            }
+
             Ok(())
         };
 

src/database_drivers/postgres.rs

@@ -1,7 +1,9 @@
 use crate::config;
 use crate::database_drivers::DatabaseDriver;
 use anyhow::{bail, Result};
+use log::error;
 use sqlx::postgres::PgRow;
+use sqlx::Executor;
 use sqlx::{Connection, PgConnection, Row};
 use std::future::Future;
 use std::pin::Pin;
@@ -60,7 +62,18 @@ impl DatabaseDriver for PostgresDriver {
         query: &'a str,
     ) -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + '_>> {
         let fut = async move {
-            sqlx::query(query).execute(&mut self.db).await?;
+            let mut tx = self.db.begin().await?;
+
+            match tx.execute(query).await {
+                Ok(_) => {
+                    tx.commit().await?;
+                }
+                Err(e) => {
+                    error!("Error executing query: {}", e);
+                    tx.rollback().await?;
+                }
+            }
+
             Ok(())
         };
 

src/database_drivers/sqlite.rs

@@ -16,7 +16,7 @@ impl<'a> SqliteDriver {
     pub async fn new<'b>(db_url: &str) -> Result<SqliteDriver> {
         let path = std::path::Path::new(db_url.split_once("://").unwrap().1);
 
-        if let Err(_) = File::open(path.to_str().unwrap()) {
+        if File::open(path.to_str().unwrap()).is_err() {
             if let Some(parent) = path.parent() {
                 fs::create_dir_all(parent)?;
             }