feat(rust): staging apply + e2e for Postgres and MySQL (#1992)

apps/tauri/src-tauri/src/connection/mod.rs

@@ -4,18 +4,18 @@ mod erd;
 mod error;
 mod id;
 mod introspect;
-mod manager;
-mod mysql;
+pub(crate) mod manager;
+pub(crate) mod mysql;
 mod mysql_object_tree;
 pub(crate) mod mysql_query;
 mod object_tree;
-mod postgres;
+pub(crate) mod postgres;
 mod postgres_erd;
 mod postgres_object_detail;
 mod postgres_object_tree;
 pub(crate) mod postgres_query;
 mod query;
-mod runtime;
+pub(crate) mod runtime;
 mod smoke_slice;
 mod sql_guard;
 mod sqlite;

apps/tauri/src-tauri/src/lib.rs

@@ -13,8 +13,9 @@ pub use connection::{
 pub use profiles::{ConnectionProfile, ProfileStore, ProfileStoreError};
 pub use secrets::{KeyringSecretStore, MemorySecretStore, SecretError, SecretStore, SharedSecretStore};
 pub use staging::{
-    apply_plan_sqlite, apply_staging_sqlite, generate_plan, ApplyError, ApplyResult, MutationType,
-    SqlDialect, StagedCellStatus, StagedRowStatus, StagingBuffer, StagingColumnMeta,
+    apply_plan_mysql, apply_plan_postgres, apply_plan_sqlite, apply_staging_mysql,
+    apply_staging_postgres, apply_staging_sqlite, generate_plan, ApplyError, ApplyResult,
+    MutationType, SqlDialect, StagedCellStatus, StagedRowStatus, StagingBuffer, StagingColumnMeta,
     TableMutationPlan, TableMutationStatement, NULL_SENTINEL,
 };
 

apps/tauri/src-tauri/src/staging/apply.rs

@@ -1,10 +1,15 @@
-//! Apply a [`TableMutationPlan`] inside a single DB transaction (#1889).
+//! Apply a [`TableMutationPlan`] inside a single DB transaction (#1889 / #1992).
 //!
 //! On success the caller should clear the staging buffer; on failure the
 //! buffer is left intact and the database is rolled back.
+//!
+//! Drivers: SQLite (`rusqlite`), Postgres / MySQL (`sqlx` via shared runtime).
 
 use super::dml::{SqlDialect, TableMutationPlan};
 use super::StagingBuffer;
+use crate::connection::mysql::MySqlPool;
+use crate::connection::postgres::PgPool;
+use crate::connection::runtime;
 use rusqlite::Connection;
 use serde::Serialize;
 use std::fmt;
@@ -107,17 +112,135 @@ pub fn apply_staging_sqlite(
     table_name: &str,
     schema: Option<&str>,
 ) -> Result<ApplyResult, ApplyError> {
-    let plan = buffer.generate_mutation_plan(
+    apply_staging_with(
+        buffer,
         SqlDialect::Sqlite,
         table_name,
         schema,
-        None,
-        None,
-    );
+        |plan| apply_plan_sqlite(conn, plan),
+    )
+}
+
+/// Execute every statement in `plan` atomically on a Postgres pool.
+pub fn apply_plan_postgres(
+    pool: &PgPool,
+    plan: &TableMutationPlan,
+) -> Result<ApplyResult, ApplyError> {
+    if !matches!(plan.dialect, SqlDialect::Postgres) {
+        return Err(ApplyError::DialectMismatch {
+            expected: "postgres".into(),
+            got: dialect_name(plan.dialect).into(),
+        });
+    }
+    if plan.is_empty() {
+        return Err(ApplyError::EmptyPlan);
+    }
+
+    runtime::block_on(async {
+        let mut tx = pool.begin().await.map_err(|e| ApplyError::Transaction {
+            message: e.to_string(),
+        })?;
+        for (i, stmt) in plan.statements.iter().enumerate() {
+            sqlx::query(&stmt.sql)
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| ApplyError::Execute {
+                    statement_index: i,
+                    sql: stmt.sql.clone(),
+                    message: e.to_string(),
+                })?;
+        }
+        tx.commit().await.map_err(|e| ApplyError::Transaction {
+            message: e.to_string(),
+        })?;
+        Ok(ApplyResult {
+            statements_applied: plan.statements.len(),
+        })
+    })
+}
+
+/// Generate + apply on Postgres; clear staging only on success.
+pub fn apply_staging_postgres(
+    pool: &PgPool,
+    buffer: &mut StagingBuffer,
+    table_name: &str,
+    schema: Option<&str>,
+) -> Result<ApplyResult, ApplyError> {
+    apply_staging_with(
+        buffer,
+        SqlDialect::Postgres,
+        table_name,
+        schema,
+        |plan| apply_plan_postgres(pool, plan),
+    )
+}
+
+/// Execute every statement in `plan` atomically on a MySQL pool.
+pub fn apply_plan_mysql(
+    pool: &MySqlPool,
+    plan: &TableMutationPlan,
+) -> Result<ApplyResult, ApplyError> {
+    if !matches!(plan.dialect, SqlDialect::Mysql) {
+        return Err(ApplyError::DialectMismatch {
+            expected: "mysql".into(),
+            got: dialect_name(plan.dialect).into(),
+        });
+    }
+    if plan.is_empty() {
+        return Err(ApplyError::EmptyPlan);
+    }
+
+    runtime::block_on(async {
+        let mut tx = pool.begin().await.map_err(|e| ApplyError::Transaction {
+            message: e.to_string(),
+        })?;
+        for (i, stmt) in plan.statements.iter().enumerate() {
+            sqlx::query(&stmt.sql)
+                .execute(&mut *tx)
+                .await
+                .map_err(|e| ApplyError::Execute {
+                    statement_index: i,
+                    sql: stmt.sql.clone(),
+                    message: e.to_string(),
+                })?;
+        }
+        tx.commit().await.map_err(|e| ApplyError::Transaction {
+            message: e.to_string(),
+        })?;
+        Ok(ApplyResult {
+            statements_applied: plan.statements.len(),
+        })
+    })
+}
+
+/// Generate + apply on MySQL; clear staging only on success.
+pub fn apply_staging_mysql(
+    pool: &MySqlPool,
+    buffer: &mut StagingBuffer,
+    table_name: &str,
+    schema: Option<&str>,
+) -> Result<ApplyResult, ApplyError> {
+    apply_staging_with(
+        buffer,
+        SqlDialect::Mysql,
+        table_name,
+        schema,
+        |plan| apply_plan_mysql(pool, plan),
+    )
+}
+
+fn apply_staging_with(
+    buffer: &mut StagingBuffer,
+    dialect: SqlDialect,
+    table_name: &str,
+    schema: Option<&str>,
+    apply: impl FnOnce(&TableMutationPlan) -> Result<ApplyResult, ApplyError>,
+) -> Result<ApplyResult, ApplyError> {
+    let plan = buffer.generate_mutation_plan(dialect, table_name, schema, None, None);
     if plan.is_empty() {
         return Err(ApplyError::EmptyPlan);
     }
-    let result = apply_plan_sqlite(conn, &plan)?;
+    let result = apply(&plan)?;
     buffer.revert_all();
     Ok(result)
 }