From c1b4499f5ccfa038ad6bd65baa4d284985199f5a Mon Sep 17 00:00:00 2001 From: ZhuchkaTriplesix Date: Sun, 15 Nov 2026 17:17:47 +0300 Subject: [PATCH] feat(rust): staging apply + live e2e for Postgres and MySQL Transactional apply via sqlx pools; temp-table e2e covers success and rollback on both drivers (skips when live DB unavailable). Closes #1992 --- apps/tauri/src-tauri/src/connection/mod.rs | 8 +- apps/tauri/src-tauri/src/lib.rs | 5 +- apps/tauri/src-tauri/src/staging/apply.rs | 135 +++++- .../tauri/src-tauri/src/staging/apply_live.rs | 407 ++++++++++++++++++ apps/tauri/src-tauri/src/staging/mod.rs | 7 +- 5 files changed, 549 insertions(+), 13 deletions(-) create mode 100644 apps/tauri/src-tauri/src/staging/apply_live.rs diff --git a/apps/tauri/src-tauri/src/connection/mod.rs b/apps/tauri/src-tauri/src/connection/mod.rs index 766d2fbc..fe3224e2 100644 --- a/apps/tauri/src-tauri/src/connection/mod.rs +++ b/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; diff --git a/apps/tauri/src-tauri/src/lib.rs b/apps/tauri/src-tauri/src/lib.rs index 008379c0..d513e58c 100644 --- a/apps/tauri/src-tauri/src/lib.rs +++ b/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, }; diff --git a/apps/tauri/src-tauri/src/staging/apply.rs b/apps/tauri/src-tauri/src/staging/apply.rs index d7aaa2ca..fb81e158 100644 --- a/apps/tauri/src-tauri/src/staging/apply.rs +++ b/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 { - 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 { + 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 { + 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 { + 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 { + 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, +) -> Result { + 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) } diff --git a/apps/tauri/src-tauri/src/staging/apply_live.rs b/apps/tauri/src-tauri/src/staging/apply_live.rs new file mode 100644 index 00000000..00ac0950 --- /dev/null +++ b/apps/tauri/src-tauri/src/staging/apply_live.rs @@ -0,0 +1,407 @@ +//! Live e2e: staging → generate → apply on Postgres / MySQL (#1992). +//! +//! Skips when `QUERYA_SKIP_PG` / `QUERYA_SKIP_MYSQL` is set, or when the +//! server is unreachable (same convention as query live tests). + +#![cfg(test)] + +use super::{ + apply_staging_mysql, apply_staging_postgres, ApplyError, StagingBuffer, +}; +use crate::connection::manager::ConnectionSession; +use crate::connection::mysql::{open_mysql, MysqlConnectParams, MysqlSession, MySqlPool}; +use crate::connection::postgres::{ + open_postgres, PgPool, PostgresConnectParams, PostgresSession, +}; +use crate::connection::runtime; +use sqlx::Row; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn unique_table(prefix: &str) -> String { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + format!("{prefix}_{nanos}") +} + +fn pg_params() -> Option { + if std::env::var_os("QUERYA_SKIP_PG").is_some() { + return None; + } + let host = std::env::var("QUERYA_PG_HOST").unwrap_or_else(|_| "127.0.0.1".into()); + let port: u16 = std::env::var("QUERYA_PG_PORT") + .ok() + .and_then(|p| p.parse().ok()) + .unwrap_or(5432); + let database = std::env::var("QUERYA_PG_DB").unwrap_or_else(|_| "querya".into()); + let user = std::env::var("QUERYA_PG_USER").unwrap_or_else(|_| "querya".into()); + let password = std::env::var("QUERYA_PG_PASSWORD").unwrap_or_else(|_| "querya".into()); + Some(PostgresConnectParams { + host, + port, + database, + user, + password: Some(password), + id: None, + }) +} + +fn mysql_params() -> Option { + if std::env::var_os("QUERYA_SKIP_MYSQL").is_some() { + return None; + } + let host = std::env::var("QUERYA_MYSQL_HOST").unwrap_or_else(|_| "127.0.0.1".into()); + let port: u16 = std::env::var("QUERYA_MYSQL_PORT") + .ok() + .and_then(|p| p.parse().ok()) + .unwrap_or(3306); + let database = std::env::var("QUERYA_MYSQL_DB").unwrap_or_else(|_| "querya".into()); + let user = std::env::var("QUERYA_MYSQL_USER").unwrap_or_else(|_| "querya".into()); + let password = std::env::var("QUERYA_MYSQL_PASSWORD").unwrap_or_else(|_| "querya".into()); + Some(MysqlConnectParams { + host, + port, + database, + user, + password: Some(password), + id: None, + }) +} + +struct PgHarness { + session: PostgresSession, + table: String, +} + +impl PgHarness { + fn try_new() -> Option { + let params = pg_params()?; + let session = match open_postgres(params) { + Ok(s) => s, + Err(e) => { + eprintln!("skip postgres staging e2e: {e}"); + return None; + } + }; + let table = unique_table("querya_stg_pg"); + let stmts = [ + format!( + "CREATE TABLE public.{table} ( + id INTEGER PRIMARY KEY, + name TEXT NOT NULL, + email TEXT + )" + ), + format!( + "INSERT INTO public.{table} (id, name, email) VALUES + (1, 'Alice', 'alice@test.com'), + (2, 'Bob', 'bob@test.com'), + (3, 'Charlie', 'charlie@test.com')" + ), + ]; + if let Err(e) = session.with_pool(|pool| { + runtime::block_on(async { + for stmt in &stmts { + sqlx::query(stmt) + .execute(pool) + .await + .map_err(|e| { + crate::connection::ConnectionError::driver( + crate::connection::DriverKind::Postgres, + e.to_string(), + ) + })?; + } + Ok(()) + }) + }) { + eprintln!("skip postgres staging e2e (create): {e}"); + let _ = session.disconnect(); + return None; + } + Some(Self { session, table }) + } + + fn pool(&self) -> PgPool { + self.session + .with_pool(|p| Ok(p.clone())) + .expect("pool") + } + + fn load_buffer(&self) -> StagingBuffer { + let sql = format!( + "SELECT id, name, email FROM public.{} ORDER BY id", + self.table + ); + let pool = self.pool(); + let rows = runtime::block_on(async { + let rows = sqlx::query(&sql).fetch_all(&pool).await.expect("select"); + rows.iter() + .map(|r| { + vec![ + r.get::(0).to_string(), + r.get::(1), + r.try_get::(2) + .unwrap_or_else(|_| "NULL".into()), + ] + }) + .collect::>() + }); + StagingBuffer::new(["id", "name", "email"], rows, ["id"]) + } + + fn read_users(&self) -> Vec<(i32, String, Option)> { + let sql = format!( + "SELECT id, name, email FROM public.{} ORDER BY id", + self.table + ); + let pool = self.pool(); + runtime::block_on(async { + let rows = sqlx::query(&sql).fetch_all(&pool).await.expect("select"); + rows.iter() + .map(|r| { + ( + r.get::(0), + r.get::(1), + r.try_get::(2).ok(), + ) + }) + .collect() + }) + } +} + +impl Drop for PgHarness { + fn drop(&mut self) { + let drop = format!("DROP TABLE IF EXISTS public.{}", self.table); + let _ = self.session.with_pool(|pool| { + runtime::block_on(async { + let _ = sqlx::query(&drop).execute(pool).await; + Ok(()) + }) + }); + let _ = self.session.disconnect(); + } +} + +struct MysqlHarness { + session: MysqlSession, + table: String, +} + +impl MysqlHarness { + fn try_new() -> Option { + let params = mysql_params()?; + let session = match open_mysql(params) { + Ok(s) => s, + Err(e) => { + eprintln!("skip mysql staging e2e: {e}"); + return None; + } + }; + let table = unique_table("querya_stg_my"); + let create = format!( + r#" + CREATE TABLE `{table}` ( + id INT PRIMARY KEY, + name VARCHAR(64) NOT NULL, + email VARCHAR(128) NULL + ); + INSERT INTO `{table}` (id, name, email) VALUES + (1, 'Alice', 'alice@test.com'), + (2, 'Bob', 'bob@test.com'), + (3, 'Charlie', 'charlie@test.com'); + "# + ); + if let Err(e) = session.with_pool(|pool| { + runtime::block_on(async { + // MySQL may need separate statements. + for stmt in create.split(';').map(str::trim).filter(|s| !s.is_empty()) { + sqlx::query(stmt) + .execute(pool) + .await + .map_err(|e| crate::connection::ConnectionError::driver( + crate::connection::DriverKind::Mysql, + e.to_string(), + ))?; + } + Ok(()) + }) + }) { + eprintln!("skip mysql staging e2e (create): {e}"); + let _ = session.disconnect(); + return None; + } + Some(Self { session, table }) + } + + fn pool(&self) -> MySqlPool { + self.session + .with_pool(|p| Ok(p.clone())) + .expect("pool") + } + + fn load_buffer(&self) -> StagingBuffer { + let sql = format!( + "SELECT id, name, email FROM `{}` ORDER BY id", + self.table + ); + let pool = self.pool(); + let rows = runtime::block_on(async { + let rows = sqlx::query(&sql).fetch_all(&pool).await.expect("select"); + rows.iter() + .map(|r| { + vec![ + r.get::(0).to_string(), + r.get::(1), + r.try_get::(2) + .unwrap_or_else(|_| "NULL".into()), + ] + }) + .collect::>() + }); + StagingBuffer::new(["id", "name", "email"], rows, ["id"]) + } + + fn read_users(&self) -> Vec<(i32, String, Option)> { + let sql = format!( + "SELECT id, name, email FROM `{}` ORDER BY id", + self.table + ); + let pool = self.pool(); + runtime::block_on(async { + let rows = sqlx::query(&sql).fetch_all(&pool).await.expect("select"); + rows.iter() + .map(|r| { + ( + r.get::(0), + r.get::(1), + r.try_get::(2).ok(), + ) + }) + .collect() + }) + } +} + +impl Drop for MysqlHarness { + fn drop(&mut self) { + let drop = format!("DROP TABLE IF EXISTS `{}`", self.table); + let _ = self.session.with_pool(|pool| { + runtime::block_on(async { + let _ = sqlx::query(&drop).execute(pool).await; + Ok(()) + }) + }); + let _ = self.session.disconnect(); + } +} + +#[test] +fn e2e_postgres_stage_generate_apply_success() { + let Some(h) = PgHarness::try_new() else { + return; + }; + let mut buf = h.load_buffer(); + buf.set_cell(0, 1, "Alice Cooper"); + buf.toggle_delete_row(1); + buf.add_row(Some(vec![ + "4".into(), + "Diana".into(), + "diana@test.com".into(), + ])); + + let result = + apply_staging_postgres(&h.pool(), &mut buf, &h.table, Some("public")).unwrap(); + assert_eq!(result.statements_applied, 3); + assert!(!buf.is_dirty()); + assert_eq!( + h.read_users(), + vec![ + (1, "Alice Cooper".into(), Some("alice@test.com".into())), + (3, "Charlie".into(), Some("charlie@test.com".into())), + (4, "Diana".into(), Some("diana@test.com".into())), + ] + ); +} + +#[test] +fn e2e_postgres_apply_failure_keeps_staging() { + let Some(h) = PgHarness::try_new() else { + return; + }; + let mut buf = h.load_buffer(); + buf.set_cell(0, 1, "Alice Updated"); + buf.add_row(Some(vec![ + "1".into(), + "Dup".into(), + "dup@test.com".into(), + ])); + + let err = apply_staging_postgres(&h.pool(), &mut buf, &h.table, Some("public")).unwrap_err(); + assert!(matches!(err, ApplyError::Execute { .. }), "{err:?}"); + assert!(buf.is_dirty()); + assert_eq!(buf.get_cell_value(0, 1), "Alice Updated"); + assert_eq!( + h.read_users(), + vec![ + (1, "Alice".into(), Some("alice@test.com".into())), + (2, "Bob".into(), Some("bob@test.com".into())), + (3, "Charlie".into(), Some("charlie@test.com".into())), + ] + ); +} + +#[test] +fn e2e_mysql_stage_generate_apply_success() { + let Some(h) = MysqlHarness::try_new() else { + return; + }; + let mut buf = h.load_buffer(); + buf.set_cell(0, 1, "Alice Cooper"); + buf.toggle_delete_row(1); + buf.add_row(Some(vec![ + "4".into(), + "Diana".into(), + "diana@test.com".into(), + ])); + + let result = apply_staging_mysql(&h.pool(), &mut buf, &h.table, None).unwrap(); + assert_eq!(result.statements_applied, 3); + assert!(!buf.is_dirty()); + assert_eq!( + h.read_users(), + vec![ + (1, "Alice Cooper".into(), Some("alice@test.com".into())), + (3, "Charlie".into(), Some("charlie@test.com".into())), + (4, "Diana".into(), Some("diana@test.com".into())), + ] + ); +} + +#[test] +fn e2e_mysql_apply_failure_keeps_staging() { + let Some(h) = MysqlHarness::try_new() else { + return; + }; + let mut buf = h.load_buffer(); + buf.set_cell(0, 1, "Alice Updated"); + buf.add_row(Some(vec![ + "1".into(), + "Dup".into(), + "dup@test.com".into(), + ])); + + let err = apply_staging_mysql(&h.pool(), &mut buf, &h.table, None).unwrap_err(); + assert!(matches!(err, ApplyError::Execute { .. }), "{err:?}"); + assert!(buf.is_dirty()); + assert_eq!( + h.read_users(), + vec![ + (1, "Alice".into(), Some("alice@test.com".into())), + (2, "Bob".into(), Some("bob@test.com".into())), + (3, "Charlie".into(), Some("charlie@test.com".into())), + ] + ); +} diff --git a/apps/tauri/src-tauri/src/staging/mod.rs b/apps/tauri/src-tauri/src/staging/mod.rs index 6187c207..25e76d8e 100644 --- a/apps/tauri/src-tauri/src/staging/mod.rs +++ b/apps/tauri/src-tauri/src/staging/mod.rs @@ -4,10 +4,15 @@ //! baseline. DML generation lives in [`dml`]; transactional apply in [`apply`]. mod apply; +#[cfg(test)] +mod apply_live; mod buffer; mod dml; -pub use apply::{apply_plan_sqlite, apply_staging_sqlite, ApplyError, ApplyResult}; +pub use apply::{ + apply_plan_mysql, apply_plan_postgres, apply_plan_sqlite, apply_staging_mysql, + apply_staging_postgres, apply_staging_sqlite, ApplyError, ApplyResult, +}; pub use buffer::{ StagedCellStatus, StagedRowStatus, StagingBuffer, NULL_SENTINEL, };