diff --git a/apps/tauri/src-tauri/Cargo.toml b/apps/tauri/src-tauri/Cargo.toml index f2161fd6..899f8b1a 100644 --- a/apps/tauri/src-tauri/Cargo.toml +++ b/apps/tauri/src-tauri/Cargo.toml @@ -25,6 +25,7 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } sqlx = { version = "0.8", default-features = false, features = [ "runtime-tokio", "postgres", + "mysql", "json", "chrono", "rust_decimal", diff --git a/apps/tauri/src-tauri/src/commands.rs b/apps/tauri/src-tauri/src/commands.rs index a441651d..2a9b9ad0 100644 --- a/apps/tauri/src-tauri/src/commands.rs +++ b/apps/tauri/src-tauri/src/commands.rs @@ -9,7 +9,7 @@ use serde::Deserialize; use tauri::ipc::Channel; use tauri::State; -/// Bundled IPC args for Postgres save+connect (keeps clippy under the limit). +/// Bundled IPC args for Postgres/MySQL save+connect (keeps clippy under the limit). #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ConnectPostgresRequest { @@ -23,6 +23,9 @@ pub struct ConnectPostgresRequest { pub id: Option, } +/// Same shape as Postgres — separate type so Tauri command names stay clear. +pub type ConnectMysqlRequest = ConnectPostgresRequest; + #[tauri::command] pub fn db_connect_sqlite( path: String, @@ -52,6 +55,26 @@ pub fn db_connect_postgres( }) } +#[tauri::command] +pub fn db_connect_mysql( + host: String, + port: u16, + database: String, + user: String, + password: Option, + id: Option, + registry: State<'_, ConnectionRegistry>, +) -> Result { + registry.connect(ConnectionConfig::Mysql { + host, + port, + database, + user, + password, + id, + }) +} + #[tauri::command] pub fn profiles_list( store: State<'_, ProfileStore>, @@ -241,6 +264,103 @@ pub fn profiles_connect_postgres( Ok(profile) } +/// Save MySQL profile metadata (no password) and open a live session. +#[tauri::command] +pub fn profiles_connect_mysql( + request: ConnectMysqlRequest, + store: State<'_, ProfileStore>, + secrets: State<'_, SharedSecretStore>, + registry: State<'_, ConnectionRegistry>, +) -> Result { + let ConnectMysqlRequest { + name, + host, + port, + database, + user, + password, + folder_id, + id, + } = request; + + let display_name = { + let t = name.trim(); + if t.is_empty() { + format!("{user}@{host}/{database}") + } else { + t.to_owned() + } + }; + + let mut profile = match id.filter(|s| !s.trim().is_empty()) { + Some(existing_id) => store + .get(&existing_id) + .map_err(|e| ConnectionError::io(e.to_string()))? + .map(|mut p| { + p.name = display_name.clone(); + p.host = Some(host.clone()); + p.port = Some(i64::from(port)); + p.database_name = Some(database.clone()); + p.username = Some(user.clone()); + p.folder_id = folder_id.clone(); + p.driver = "mysql".into(); + p.path = None; + p + }) + .unwrap_or_else(|| { + let mut p = ConnectionProfile::new_mysql( + display_name.clone(), + host.clone(), + port, + database.clone(), + user.clone(), + ); + p.id = existing_id; + p.folder_id = folder_id.clone(); + p + }), + None => { + let mut p = ConnectionProfile::new_mysql( + display_name, + host.clone(), + port, + database.clone(), + user.clone(), + ); + p.folder_id = folder_id; + p + } + }; + profile.touch(); + store + .upsert(&profile) + .map_err(|e| ConnectionError::io(e.to_string()))?; + + let password = match password { + Some(p) => { + secrets + .set_password(&profile.id, &p) + .map_err(|e| ConnectionError::io(format!("keyring write: {e}")))?; + Some(p) + } + None => secrets + .get_password(&profile.id) + .map_err(|e| ConnectionError::io(format!("keyring read: {e}")))?, + }; + + if !registry.is_connected(&ConnectionId::from(profile.id.as_str())) { + registry.connect(ConnectionConfig::Mysql { + host, + port, + database, + user, + password, + id: Some(profile.id.clone()), + })?; + } + Ok(profile) +} + #[tauri::command] pub fn db_disconnect( id: ConnectionId, diff --git a/apps/tauri/src-tauri/src/connection/manager.rs b/apps/tauri/src-tauri/src/connection/manager.rs index 38577f98..508c04b0 100644 --- a/apps/tauri/src-tauri/src/connection/manager.rs +++ b/apps/tauri/src-tauri/src/connection/manager.rs @@ -1,5 +1,6 @@ use super::error::{ConnectionError, ConnectionResult}; use super::id::{ConnectionId, DriverKind}; +use super::mysql::MysqlSession; use super::postgres::PostgresSession; use super::query::QueryEvent; use super::sqlite::SqliteSession; @@ -31,7 +32,6 @@ pub enum ConnectionConfig { #[serde(default)] id: Option, }, - /// Stub until the MySQL adapter epic wires connect. Mysql { host: String, port: u16, @@ -91,7 +91,7 @@ impl ConnectionRegistry { Self::with_builtin_drivers() } - /// Registry with SQLite + Postgres openers (MySQL still UnsupportedDriver). + /// Registry with SQLite + Postgres + MySQL openers. pub fn with_builtin_drivers() -> Self { Self::with_opener(|config| match config { ConnectionConfig::Sqlite { path, id } => { @@ -121,9 +121,24 @@ impl ConnectionRegistry { })?; Ok(Arc::new(session) as SessionArc) } - other => Err(ConnectionError::UnsupportedDriver { - driver: other.driver_kind(), - }), + ConnectionConfig::Mysql { + host, + port, + database, + user, + password, + id, + } => { + let session = super::mysql::open_mysql(super::mysql::MysqlConnectParams { + host, + port, + database, + user, + password, + id: id.filter(|s| !s.trim().is_empty()).map(ConnectionId::from), + })?; + Ok(Arc::new(session) as SessionArc) + } }) } @@ -195,7 +210,22 @@ impl ConnectionRegistry { f(pg) } - /// Driver-aware query stream (SQLite + Postgres). + pub fn with_mysql( + &self, + id: &ConnectionId, + f: impl FnOnce(&MysqlSession) -> ConnectionResult, + ) -> ConnectionResult { + let session = self.session_arc(id)?; + let mysql = session + .as_any() + .downcast_ref::() + .ok_or(ConnectionError::UnsupportedDriver { + driver: session.driver(), + })?; + f(mysql) + } + + /// Driver-aware query stream (SQLite + Postgres + MySQL). pub fn execute_query( &self, id: &ConnectionId, @@ -212,7 +242,7 @@ impl ConnectionRegistry { DriverKind::Postgres => { self.with_postgres(id, |s| s.execute_streaming(sql, limit, emit)) } - other => Err(ConnectionError::UnsupportedDriver { driver: other }), + DriverKind::Mysql => self.with_mysql(id, |s| s.execute_streaming(sql, limit, emit)), } } @@ -256,7 +286,17 @@ impl ConnectionRegistry { super::postgres_object_tree::PostgresObjectTree::new(pg.clone_handle()), )) } - other => Err(ConnectionError::UnsupportedDriver { driver: other }), + DriverKind::Mysql => { + let mysql = session + .as_any() + .downcast_ref::() + .ok_or(ConnectionError::UnsupportedDriver { + driver: DriverKind::Mysql, + })?; + Ok(Arc::new(super::mysql_object_tree::MysqlObjectTree::new( + mysql.clone_handle(), + ))) + } } } @@ -411,11 +451,11 @@ mod tests { } #[test] - fn registry_mysql_is_unsupported_for_now() { + fn registry_mysql_rejects_empty_host() { let reg = ConnectionRegistry::with_builtin_drivers(); let err = reg .connect(ConnectionConfig::Mysql { - host: "localhost".into(), + host: " ".into(), port: 3306, database: "db".into(), user: "u".into(), @@ -423,12 +463,7 @@ mod tests { id: None, }) .unwrap_err(); - assert!(matches!( - err, - ConnectionError::UnsupportedDriver { - driver: DriverKind::Mysql - } - )); + assert!(matches!(err, ConnectionError::InvalidConfig { .. })); } #[test] diff --git a/apps/tauri/src-tauri/src/connection/mod.rs b/apps/tauri/src-tauri/src/connection/mod.rs index a4853eee..766d2fbc 100644 --- a/apps/tauri/src-tauri/src/connection/mod.rs +++ b/apps/tauri/src-tauri/src/connection/mod.rs @@ -5,6 +5,9 @@ mod error; mod id; mod introspect; mod manager; +mod mysql; +mod mysql_object_tree; +pub(crate) mod mysql_query; mod object_tree; mod postgres; mod postgres_erd; diff --git a/apps/tauri/src-tauri/src/connection/mysql.rs b/apps/tauri/src-tauri/src/connection/mysql.rs new file mode 100644 index 00000000..2a73c697 --- /dev/null +++ b/apps/tauri/src-tauri/src/connection/mysql.rs @@ -0,0 +1,213 @@ +//! MySQL driver via sqlx (connect + disconnect). + +use super::error::{ConnectionError, ConnectionResult}; +use super::id::{ConnectionId, DriverKind}; +use super::manager::ConnectionSession; +use super::runtime; +use sqlx::mysql::{MySqlConnectOptions, MySqlPoolOptions, MySqlSslMode}; +use sqlx::{MySql, Pool}; +use std::str::FromStr; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +pub type MySqlPool = Pool; + +pub struct MysqlSession { + id: ConnectionId, + host: String, + port: u16, + database: String, + user: String, + pool: Arc>>, +} + +impl MysqlSession { + pub fn host(&self) -> &str { + &self.host + } + + pub fn port(&self) -> u16 { + self.port + } + + pub fn database(&self) -> &str { + &self.database + } + + pub fn user(&self) -> &str { + &self.user + } + + pub fn clone_handle(&self) -> Self { + Self { + id: self.id.clone(), + host: self.host.clone(), + port: self.port, + database: self.database.clone(), + user: self.user.clone(), + pool: Arc::clone(&self.pool), + } + } + + pub fn with_pool( + &self, + f: impl FnOnce(&MySqlPool) -> ConnectionResult, + ) -> ConnectionResult { + let guard = self + .pool + .lock() + .map_err(|_| ConnectionError::io("mysql pool lock poisoned"))?; + let pool = guard + .as_ref() + .ok_or_else(|| ConnectionError::Disconnected { + id: self.id.clone(), + })?; + f(pool) + } + + pub fn execute_streaming( + &self, + sql: &str, + limit: Option, + emit: impl FnMut(super::query::QueryEvent) -> ConnectionResult<()>, + ) -> ConnectionResult<()> { + self.with_pool(|pool| { + super::mysql_query::execute_streaming(pool, sql, limit, None, emit) + }) + } +} + +impl ConnectionSession for MysqlSession { + fn id(&self) -> &ConnectionId { + &self.id + } + + fn driver(&self) -> DriverKind { + DriverKind::Mysql + } + + fn disconnect(&self) -> ConnectionResult<()> { + let pool = { + let mut guard = self + .pool + .lock() + .map_err(|_| ConnectionError::io("mysql pool lock poisoned"))?; + guard.take() + }; + if let Some(pool) = pool { + runtime::block_on(pool.close()); + } + Ok(()) + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + +pub struct MysqlConnectParams { + pub host: String, + pub port: u16, + pub database: String, + pub user: String, + pub password: Option, + pub id: Option, +} + +/// Open a MySQL pool and return a live session. +pub fn open_mysql(params: MysqlConnectParams) -> ConnectionResult { + let host = params.host.trim(); + let database = params.database.trim(); + let user = params.user.trim(); + if host.is_empty() { + return Err(ConnectionError::invalid_config("mysql host is empty")); + } + if database.is_empty() { + return Err(ConnectionError::invalid_config("mysql database is empty")); + } + if user.is_empty() { + return Err(ConnectionError::invalid_config("mysql user is empty")); + } + if params.port == 0 { + return Err(ConnectionError::invalid_config("mysql port is 0")); + } + + let opts = if host.contains("://") { + MySqlConnectOptions::from_str(host).map_err(|e| { + ConnectionError::invalid_config(format!("invalid mysql url: {e}")) + })? + } else { + let mut opts = MySqlConnectOptions::new() + .host(host) + .port(params.port) + .username(user) + .database(database) + // No TLS feature wired yet — Disabled matches local docker compose. + .ssl_mode(MySqlSslMode::Disabled); + if let Some(pw) = params.password.as_deref() { + opts = opts.password(pw); + } + opts + }; + + let id = params + .id + .filter(|i| !i.as_str().trim().is_empty()) + .unwrap_or_default(); + + let pool = runtime::block_on(async { + MySqlPoolOptions::new() + .max_connections(4) + .acquire_timeout(Duration::from_secs(8)) + .connect_with(opts) + .await + }) + .map_err(|e| ConnectionError::driver(DriverKind::Mysql, e.to_string()))?; + + runtime::block_on(async { + sqlx::query("SELECT 1") + .execute(&pool) + .await + .map_err(|e| ConnectionError::driver(DriverKind::Mysql, e.to_string())) + })?; + + Ok(MysqlSession { + id, + host: host.to_owned(), + port: params.port, + database: database.to_owned(), + user: user.to_owned(), + pool: Arc::new(Mutex::new(Some(pool))), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_empty_host() { + let err = open_mysql(MysqlConnectParams { + host: " ".into(), + port: 3306, + database: "db".into(), + user: "u".into(), + password: None, + id: None, + }); + assert!(matches!(err, Err(ConnectionError::InvalidConfig { .. }))); + } + + #[test] + fn rejects_zero_port() { + let err = open_mysql(MysqlConnectParams { + host: "localhost".into(), + port: 0, + database: "db".into(), + user: "u".into(), + password: None, + id: None, + }); + assert!(matches!(err, Err(ConnectionError::InvalidConfig { .. }))); + } +} diff --git a/apps/tauri/src-tauri/src/connection/mysql_object_tree.rs b/apps/tauri/src-tauri/src/connection/mysql_object_tree.rs new file mode 100644 index 00000000..8bb63e9a --- /dev/null +++ b/apps/tauri/src-tauri/src/connection/mysql_object_tree.rs @@ -0,0 +1,212 @@ +//! MySQL [`ObjectTreeProvider`] — databases → tables/views → columns. + +use super::error::{ConnectionError, ConnectionResult}; +use super::id::DriverKind; +use super::introspect::{ColumnMeta, TableSchema}; +use super::mysql::MysqlSession; +use super::object_tree::{ + column_id, parse_object_id, relation_id, schema_id, ObjectTreeProvider, ParsedObjectId, + SqlObjectKind, SqlObjectNode, +}; +use super::runtime; +use sqlx::Row; + +const DRIVER: DriverKind = DriverKind::Mysql; + +pub struct MysqlObjectTree { + session: MysqlSession, +} + +impl MysqlObjectTree { + pub fn new(session: MysqlSession) -> Self { + Self { session } + } +} + +impl ObjectTreeProvider for MysqlObjectTree { + fn list_children(&self, parent_id: Option<&str>) -> ConnectionResult> { + match parent_id { + None => self.list_schemas(), + Some(id) => match parse_object_id(id)? { + ParsedObjectId::Schema { schema, .. } => self.list_relations(&schema, id), + ParsedObjectId::Relation { schema, name, .. } => { + self.list_columns(&schema, &name, id) + } + ParsedObjectId::Column { .. } + | ParsedObjectId::Folder { .. } + | ParsedObjectId::Named { .. } + | ParsedObjectId::RelationChild { .. } => Ok(vec![]), + }, + } + } + + fn describe(&self, object_id: &str) -> ConnectionResult { + match parse_object_id(object_id)? { + ParsedObjectId::Relation { schema, name, .. } => self.describe_relation(&schema, &name), + _ => Err(ConnectionError::invalid_config( + "describe requires a table or view id", + )), + } + } +} + +impl MysqlObjectTree { + fn list_schemas(&self) -> ConnectionResult> { + self.session.with_pool(|pool| { + runtime::block_on(async { + let rows = sqlx::query( + "SELECT SCHEMA_NAME + FROM information_schema.SCHEMATA + WHERE SCHEMA_NAME NOT IN ('information_schema', 'mysql', 'performance_schema', 'sys') + ORDER BY SCHEMA_NAME", + ) + .fetch_all(pool) + .await + .map_err(mysql_err)?; + + Ok(rows + .into_iter() + .map(|r| { + let name: String = r.get(0); + SqlObjectNode::new( + schema_id(DRIVER, &name), + name.clone(), + SqlObjectKind::Schema, + Some(name), + None, + ) + }) + .collect()) + }) + }) + } + + fn list_relations( + &self, + schema: &str, + parent_id: &str, + ) -> ConnectionResult> { + let schema = schema.to_owned(); + let parent_id = parent_id.to_owned(); + self.session.with_pool(|pool| { + runtime::block_on(async { + let rows = sqlx::query( + "SELECT TABLE_NAME, TABLE_TYPE + FROM information_schema.TABLES + WHERE TABLE_SCHEMA = ? + ORDER BY TABLE_TYPE, TABLE_NAME", + ) + .bind(&schema) + .fetch_all(pool) + .await + .map_err(mysql_err)?; + + Ok(rows + .into_iter() + .filter_map(|r| { + let name: String = r.get(0); + let table_type: String = r.get(1); + let kind = match table_type.as_str() { + "VIEW" | "SYSTEM VIEW" => SqlObjectKind::View, + "BASE TABLE" => SqlObjectKind::Table, + _ => return None, + }; + Some(SqlObjectNode::new( + relation_id(DRIVER, kind, &schema, &name), + name, + kind, + Some(schema.clone()), + Some(parent_id.clone()), + )) + }) + .collect()) + }) + }) + } + + fn list_columns( + &self, + schema: &str, + table: &str, + parent_id: &str, + ) -> ConnectionResult> { + let desc = self.describe_relation(schema, table)?; + Ok(desc + .columns + .into_iter() + .map(|c| { + SqlObjectNode::new( + column_id(DRIVER, schema, table, &c.name), + c.name, + SqlObjectKind::Column, + Some(schema.to_owned()), + Some(parent_id.to_owned()), + ) + }) + .collect()) + } + + fn describe_relation(&self, schema: &str, table: &str) -> ConnectionResult { + let schema = schema.to_owned(); + let table = table.to_owned(); + self.session.with_pool(|pool| { + runtime::block_on(async { + let rows = sqlx::query( + "SELECT COLUMN_NAME, + COLUMN_TYPE, + IS_NULLABLE, + COLUMN_KEY, + COLUMN_DEFAULT, + ORDINAL_POSITION + FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? + ORDER BY ORDINAL_POSITION", + ) + .bind(&schema) + .bind(&table) + .fetch_all(pool) + .await + .map_err(mysql_err)?; + + if rows.is_empty() { + return Err(mysql_err(format!("relation {schema}.{table} not found"))); + } + + let mut columns = Vec::with_capacity(rows.len()); + let mut primary_keys = Vec::new(); + let mut pk_pos = 0i32; + for r in rows { + let name: String = r.get(0); + let data_type: String = r.get(1); + let nullable: String = r.get(2); + let key: String = r.get(3); + let default_value: Option = r.get(4); + let is_primary_key = key == "PRI"; + if is_primary_key { + pk_pos += 1; + primary_keys.push(name.clone()); + } + columns.push(ColumnMeta { + name, + data_type, + is_nullable: nullable.eq_ignore_ascii_case("YES"), + is_primary_key, + primary_key_position: if is_primary_key { Some(pk_pos) } else { None }, + default_value, + }); + } + + Ok(TableSchema { + table_name: table, + schema: Some(schema), + columns, + primary_keys, + }) + }) + }) + } +} + +fn mysql_err(e: impl ToString) -> ConnectionError { + ConnectionError::driver(DriverKind::Mysql, e.to_string()) +} diff --git a/apps/tauri/src-tauri/src/connection/mysql_query.rs b/apps/tauri/src-tauri/src/connection/mysql_query.rs new file mode 100644 index 00000000..e9e305ad --- /dev/null +++ b/apps/tauri/src-tauri/src/connection/mysql_query.rs @@ -0,0 +1,313 @@ +//! MySQL execute + batched row streaming (sqlx). + +use super::error::{ConnectionError, ConnectionResult}; +use super::id::DriverKind; +use super::mysql::MySqlPool; +use super::query::QueryEvent; +use super::runtime; +use chrono::{NaiveDate, NaiveDateTime, NaiveTime}; +use futures_util::TryStreamExt; +use rust_decimal::Decimal; +use sqlx::mysql::{MySqlRow, MySqlTypeInfo, MySqlValueRef}; +use sqlx::{Column, Decode, Executor, Row, TypeInfo, ValueRef}; +use std::time::Instant; + +const DEFAULT_BATCH: usize = 256; +const DEFAULT_LIMIT: usize = 5_000; + +/// Run SQL and emit column / row-batch / done events. +pub fn execute_streaming( + pool: &MySqlPool, + sql: &str, + limit: Option, + batch_size: Option, + mut emit: impl FnMut(QueryEvent) -> ConnectionResult<()>, +) -> ConnectionResult<()> { + let sql = sql.trim(); + if sql.is_empty() { + return Err(ConnectionError::invalid_config("sql is empty")); + } + + let limit = limit.unwrap_or(DEFAULT_LIMIT); + let batch_size = batch_size.unwrap_or(DEFAULT_BATCH).max(1); + let started = Instant::now(); + + runtime::block_on(async { + let mut conn = pool + .acquire() + .await + .map_err(|e| ConnectionError::driver(DriverKind::Mysql, e.to_string()))?; + + let desc = conn + .describe(sql) + .await + .map_err(|e| ConnectionError::driver(DriverKind::Mysql, e.to_string()))?; + + let columns = desc.columns(); + if columns.is_empty() { + let result = sqlx::query(sql) + .execute(&mut *conn) + .await + .map_err(|e| ConnectionError::driver(DriverKind::Mysql, e.to_string()))?; + emit(QueryEvent::Done { + row_count: 0, + truncated: false, + affected_rows: Some(result.rows_affected()), + elapsed_ms: elapsed_ms(started), + })?; + return Ok(()); + } + + let decoders: Vec = columns + .iter() + .map(|c| CellKind::from_type_info(c.type_info())) + .collect(); + let names: Vec = columns.iter().map(|c| c.name().to_owned()).collect(); + emit(QueryEvent::Columns { names })?; + + let mut stream = sqlx::query(sql).fetch(&mut *conn); + let mut batch: Vec> = Vec::with_capacity(batch_size); + let mut offset: u64 = 0; + let mut row_count: u64 = 0; + let mut truncated = false; + + while let Some(row) = stream + .try_next() + .await + .map_err(|e| ConnectionError::driver(DriverKind::Mysql, e.to_string()))? + { + if row_count as usize >= limit { + truncated = true; + break; + } + batch.push(mysql_row_to_json(&row, &decoders)); + row_count += 1; + if batch.len() >= batch_size { + let rows_out = std::mem::replace(&mut batch, Vec::with_capacity(batch_size)); + let n = rows_out.len() as u64; + emit(QueryEvent::Rows { + rows: rows_out, + offset, + })?; + offset += n; + } + } + + if !batch.is_empty() { + emit(QueryEvent::Rows { + rows: batch, + offset, + })?; + } + + emit(QueryEvent::Done { + row_count, + truncated, + affected_rows: None, + elapsed_ms: elapsed_ms(started), + })?; + Ok(()) + }) +} + +#[derive(Clone, Copy, Debug)] +enum CellKind { + Bool, + I64, + U64, + F64, + Decimal, + Text, + Bytes, + Date, + Time, + DateTime, + Json, + Unknown, +} + +impl CellKind { + fn from_type_info(info: &MySqlTypeInfo) -> Self { + let name = info.name(); + match name { + "TINYINT" | "BOOLEAN" | "BOOL" => Self::Bool, + "SMALLINT" | "MEDIUMINT" | "INT" | "INTEGER" | "BIGINT" | "YEAR" => Self::I64, + "TINYINT UNSIGNED" + | "SMALLINT UNSIGNED" + | "MEDIUMINT UNSIGNED" + | "INT UNSIGNED" + | "INTEGER UNSIGNED" + | "BIGINT UNSIGNED" => Self::U64, + "FLOAT" | "DOUBLE" | "REAL" => Self::F64, + "DECIMAL" | "NUMERIC" | "NEWDECIMAL" => Self::Decimal, + "CHAR" | "VARCHAR" | "TINYTEXT" | "TEXT" | "MEDIUMTEXT" | "LONGTEXT" | "ENUM" + | "SET" => Self::Text, + "BINARY" | "VARBINARY" | "TINYBLOB" | "BLOB" | "MEDIUMBLOB" | "LONGBLOB" | "BIT" => { + Self::Bytes + } + "DATE" => Self::Date, + "TIME" => Self::Time, + "DATETIME" | "TIMESTAMP" => Self::DateTime, + "JSON" => Self::Json, + _ => Self::Unknown, + } + } +} + +fn mysql_row_to_json(row: &MySqlRow, decoders: &[CellKind]) -> Vec { + decoders + .iter() + .enumerate() + .map(|(i, kind)| mysql_cell(row, i, *kind)) + .collect() +} + +fn mysql_cell(row: &MySqlRow, i: usize, kind: CellKind) -> serde_json::Value { + let Ok(raw) = row.try_get_raw(i) else { + return serde_json::Value::Null; + }; + if raw.is_null() { + return serde_json::Value::Null; + } + decode_cell(raw.clone(), kind).unwrap_or(serde_json::Value::Null) +} + +fn decode_cell(raw: MySqlValueRef<'_>, kind: CellKind) -> Option { + match kind { + CellKind::Bool => decode::(raw.clone()) + .or_else(|| decode::(raw).map(i8::from)) + .map(|v| serde_json::Value::Bool(v != 0)), + CellKind::I64 => decode::(raw.clone()) + .or_else(|| decode::(raw.clone()).map(i64::from)) + .or_else(|| decode::(raw).map(i64::from)) + .map(json_i64), + CellKind::U64 => decode::(raw.clone()) + .or_else(|| decode::(raw).map(u64::from)) + .map(|v| serde_json::Number::from(v).into()), + CellKind::F64 => decode::(raw.clone()) + .or_else(|| decode::(raw).map(f64::from)) + .and_then(|v| serde_json::Number::from_f64(v).map(serde_json::Value::Number)), + CellKind::Decimal => decode::(raw) + .map(|v| serde_json::Value::String(v.normalize().to_string())), + CellKind::Text => decode::(raw).map(serde_json::Value::String), + CellKind::Bytes => { + decode::>(raw).map(|v| serde_json::Value::String(bytes_to_hex(&v))) + } + CellKind::Date => { + decode::(raw).map(|v| serde_json::Value::String(v.to_string())) + } + CellKind::Time => { + decode::(raw).map(|v| serde_json::Value::String(v.to_string())) + } + CellKind::DateTime => decode::(raw) + .map(|v| serde_json::Value::String(v.format("%Y-%m-%dT%H:%M:%S%.f").to_string())), + CellKind::Json => decode::(raw.clone()).or_else(|| { + decode::(raw).and_then(|s| serde_json::from_str(&s).ok()) + }), + CellKind::Unknown => decode::(raw.clone()) + .map(serde_json::Value::String) + .or_else(|| { + decode::>(raw).map(|v| serde_json::Value::String(bytes_to_hex(&v))) + }), + } +} + +fn decode<'r, T>(raw: MySqlValueRef<'r>) -> Option +where + T: Decode<'r, sqlx::MySql>, +{ + T::decode(raw).ok() +} + +fn json_i64(v: i64) -> serde_json::Value { + serde_json::Value::Number(v.into()) +} + +fn bytes_to_hex(bytes: &[u8]) -> String { + let mut s = String::with_capacity(2 + bytes.len() * 2); + s.push_str("0x"); + for b in bytes { + use std::fmt::Write; + let _ = write!(s, "{b:02x}"); + } + s +} + +fn elapsed_ms(started: Instant) -> u64 { + started.elapsed().as_millis() as u64 +} + +#[cfg(test)] +mod tests { + use crate::connection::manager::ConnectionSession; + use crate::connection::mysql::{open_mysql, MysqlConnectParams}; + use crate::connection::query::QueryEvent; + + fn live_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, + }) + } + + #[test] + fn live_mysql_streams_select_one() { + let Some(params) = live_params() else { + eprintln!("skip mysql live test (QUERYA_SKIP_MYSQL)"); + return; + }; + let session = match open_mysql(params) { + Ok(s) => s, + Err(e) => { + eprintln!("skip mysql live test: {e}"); + return; + } + }; + + let mut events = Vec::new(); + session + .execute_streaming("SELECT 1 AS n, 'ok' AS label", Some(10), |ev| { + events.push(ev); + Ok(()) + }) + .expect("stream"); + + assert!( + events + .iter() + .any(|e| matches!(e, QueryEvent::Columns { .. })), + "expected Columns" + ); + assert!( + events.iter().any(|e| matches!(e, QueryEvent::Rows { .. })), + "expected Rows" + ); + assert!( + events.iter().any(|e| matches!( + e, + QueryEvent::Done { + row_count: 1, + .. + } + )), + "expected Done row_count=1" + ); + session.disconnect().ok(); + } +} diff --git a/apps/tauri/src-tauri/src/lib.rs b/apps/tauri/src-tauri/src/lib.rs index 2fe373aa..05614962 100644 --- a/apps/tauri/src-tauri/src/lib.rs +++ b/apps/tauri/src-tauri/src/lib.rs @@ -13,10 +13,11 @@ pub use profiles::{ConnectionProfile, ProfileStore, ProfileStoreError}; pub use secrets::{KeyringSecretStore, MemorySecretStore, SecretError, SecretStore, SharedSecretStore}; use commands::{ - db_connect_postgres, db_connect_sqlite, db_describe_object, db_disconnect, db_erd_schema, - db_execute_query, db_is_connected, db_list_children, db_list_columns, db_list_schemas, - db_list_tables, db_postgres_object_detail, db_sqlite_overview, profiles_connect_postgres, - profiles_connect_sqlite, profiles_delete, profiles_list, profiles_upsert, + db_connect_mysql, db_connect_postgres, db_connect_sqlite, db_describe_object, db_disconnect, + db_erd_schema, db_execute_query, db_is_connected, db_list_children, db_list_columns, + db_list_schemas, db_list_tables, db_postgres_object_detail, db_sqlite_overview, + profiles_connect_mysql, profiles_connect_postgres, profiles_connect_sqlite, profiles_delete, + profiles_list, profiles_upsert, }; #[tauri::command] @@ -86,6 +87,7 @@ pub fn run() { host_platform, db_connect_sqlite, db_connect_postgres, + db_connect_mysql, db_disconnect, db_is_connected, db_list_schemas, @@ -102,6 +104,7 @@ pub fn run() { profiles_delete, profiles_connect_sqlite, profiles_connect_postgres, + profiles_connect_mysql, ]) .run(tauri::generate_context!()) .expect("error while running tauri application"); diff --git a/apps/tauri/src-tauri/src/profiles/store.rs b/apps/tauri/src-tauri/src/profiles/store.rs index 40f941a7..0ba1ab1c 100644 --- a/apps/tauri/src-tauri/src/profiles/store.rs +++ b/apps/tauri/src-tauri/src/profiles/store.rs @@ -78,6 +78,30 @@ impl ConnectionProfile { } } + pub fn new_mysql( + name: impl Into, + host: impl Into, + port: u16, + database: impl Into, + user: impl Into, + ) -> Self { + let now = now_rfc3339(); + Self { + id: Uuid::new_v4().to_string(), + driver: "mysql".into(), + name: name.into(), + path: None, + host: Some(host.into()), + port: Some(i64::from(port)), + username: Some(user.into()), + database_name: Some(database.into()), + folder_id: None, + sort_order: 0, + created_at: now.clone(), + updated_at: now, + } + } + pub fn touch(&mut self) { self.updated_at = now_rfc3339(); } diff --git a/apps/tauri/src/App.vue b/apps/tauri/src/App.vue index 26c5db3f..072270f0 100644 --- a/apps/tauri/src/App.vue +++ b/apps/tauri/src/App.vue @@ -27,6 +27,7 @@ import WelcomeTourDialog from "./ui/WelcomeTourDialog.vue"; import AboutDialog from "./ui/AboutDialog.vue"; import NewFolderDialog from "./ui/NewFolderDialog.vue"; import NewConnectionDialog, { + type MysqlConnectPayload, type PostgresConnectPayload, type SqliteConnectPayload, } from "./ui/NewConnectionDialog.vue"; @@ -99,6 +100,21 @@ async function onConnectPostgresFromDialog(payload: PostgresConnectPayload) { } } +async function onConnectMysqlFromDialog(payload: MysqlConnectPayload) { + try { + await connections.connectMysql({ + name: payload.name, + host: payload.host, + port: payload.port, + database: payload.database, + user: payload.user, + password: payload.password, + }); + } catch { + // status already set in store + } +} + function onNewFolder() { newFolderOpen.value = true; } @@ -283,6 +299,7 @@ onUnmounted(() => { v-model:open="newConnectionOpen" @connect-sqlite="onConnectSqliteFromDialog" @connect-postgres="onConnectPostgresFromDialog" + @connect-mysql="onConnectMysqlFromDialog" /> diff --git a/apps/tauri/src/api/connection.ts b/apps/tauri/src/api/connection.ts index e046db37..ac1f66f4 100644 --- a/apps/tauri/src/api/connection.ts +++ b/apps/tauri/src/api/connection.ts @@ -102,6 +102,24 @@ export async function dbConnectPostgres(opts: { }); } +export async function dbConnectMysql(opts: { + host: string; + port: number; + database: string; + user: string; + password?: string | null; + id?: string | null; +}): Promise { + return invoke("db_connect_mysql", { + host: opts.host, + port: opts.port, + database: opts.database, + user: opts.user, + password: opts.password ?? null, + id: opts.id ?? null, + }); +} + export async function profilesList(): Promise { return invoke("profiles_list"); } @@ -156,6 +174,31 @@ export async function profilesConnectPostgres(opts: { }); } +/** Persist MySQL metadata (no password on disk) + open live session. */ +export async function profilesConnectMysql(opts: { + name: string; + host: string; + port: number; + database: string; + user: string; + password?: string | null; + folderId?: string | null; + id?: string | null; +}): Promise { + return invoke("profiles_connect_mysql", { + request: { + name: opts.name, + host: opts.host, + port: opts.port, + database: opts.database, + user: opts.user, + password: opts.password ?? null, + folderId: opts.folderId ?? null, + id: opts.id ?? null, + }, + }); +} + export async function dbDisconnect(id: string): Promise { await invoke("db_disconnect", { id }); } diff --git a/apps/tauri/src/stores/connections.ts b/apps/tauri/src/stores/connections.ts index ca709e32..1ccdf4df 100644 --- a/apps/tauri/src/stores/connections.ts +++ b/apps/tauri/src/stores/connections.ts @@ -3,6 +3,7 @@ import { reactive, ref } from "vue"; import { dbDisconnect, formatConnectionError, + profilesConnectMysql, profilesConnectPostgres, profilesConnectSqlite, profilesDelete, @@ -17,7 +18,7 @@ import { useShellStatusStore } from "./shellStatus"; export type LiveConnection = { id: string; - driver: "sqlite" | "postgres"; + driver: "sqlite" | "postgres" | "mysql"; name: string; /** SQLite file path. */ path?: string; @@ -45,7 +46,8 @@ function endpointLabel(conn: LiveConnection): string { if (conn.mock) return "Demo (offline)"; if (conn.driver === "sqlite") return conn.path ?? ""; const host = conn.host ?? "localhost"; - const port = conn.port ?? 5432; + const port = + conn.port ?? (conn.driver === "mysql" ? 3306 : 5432); const db = conn.database ?? ""; return `${host}:${port}/${db}`; } @@ -63,17 +65,18 @@ function fromProfile(p: ConnectionProfile, connected = false): LiveConnection | connected, }; } - if (p.driver === "postgres") { + if (p.driver === "postgres" || p.driver === "mysql") { const host = p.host?.trim(); const database = p.database_name?.trim(); const user = p.username?.trim(); if (!host || !database || !user) return null; + const defaultPort = p.driver === "mysql" ? 3306 : 5432; return { id: p.id, - driver: "postgres", + driver: p.driver, name: p.name, host, - port: p.port != null ? Number(p.port) : 5432, + port: p.port != null ? Number(p.port) : defaultPort, database, user, folderId: p.folder_id ?? null, @@ -85,13 +88,13 @@ function fromProfile(p: ConnectionProfile, connected = false): LiveConnection | function toProfile(conn: LiveConnection): ConnectionProfile { const now = String(Math.floor(Date.now() / 1000)); - if (conn.driver === "postgres") { + if (conn.driver === "postgres" || conn.driver === "mysql") { return { id: conn.id, - driver: "postgres", + driver: conn.driver, name: conn.name, host: conn.host ?? null, - port: conn.port ?? 5432, + port: conn.port ?? (conn.driver === "mysql" ? 3306 : 5432), username: conn.user ?? null, database_name: conn.database ?? null, folder_id: conn.folderId ?? null, @@ -236,6 +239,35 @@ export const useConnectionsStore = defineStore("connections", () => { folderId?: string | null; id?: string | null; }) { + return connectServerDriver("postgres", opts); + } + + async function connectMysql(opts: { + name: string; + host: string; + port: number; + database: string; + user: string; + password?: string | null; + folderId?: string | null; + id?: string | null; + }) { + return connectServerDriver("mysql", opts); + } + + async function connectServerDriver( + driver: "postgres" | "mysql", + opts: { + name: string; + host: string; + port: number; + database: string; + user: string; + password?: string | null; + folderId?: string | null; + id?: string | null; + }, + ) { const status = useShellStatusStore(); const nav = useShellNavStore(); const name = @@ -246,20 +278,32 @@ export const useConnectionsStore = defineStore("connections", () => { message: `Connecting ${opts.host}:${opts.port}…`, }); try { - const profile = await profilesConnectPostgres({ - name, - host: opts.host, - port: opts.port, - database: opts.database, - user: opts.user, - password: opts.password ?? null, - folderId: opts.folderId ?? null, - id: opts.id ?? null, - }); + const profile = + driver === "postgres" + ? await profilesConnectPostgres({ + name, + host: opts.host, + port: opts.port, + database: opts.database, + user: opts.user, + password: opts.password ?? null, + folderId: opts.folderId ?? null, + id: opts.id ?? null, + }) + : await profilesConnectMysql({ + name, + host: opts.host, + port: opts.port, + database: opts.database, + user: opts.user, + password: opts.password ?? null, + folderId: opts.folderId ?? null, + id: opts.id ?? null, + }); if (opts.password) sessionPasswords.set(profile.id, opts.password); const conn: LiveConnection = { id: profile.id, - driver: "postgres", + driver, name: profile.name, host: profile.host ?? opts.host, port: profile.port != null ? Number(profile.port) : opts.port, @@ -297,16 +341,19 @@ export const useConnectionsStore = defineStore("connections", () => { } // Prefer in-memory cache; otherwise Rust loads from OS keyring (#1878). const password = sessionPasswords.get(id) ?? null; - return connectPostgres({ + const opts = { name: conn.name, host: conn.host ?? "localhost", - port: conn.port ?? 5432, - database: conn.database ?? "postgres", - user: conn.user ?? "postgres", + port: conn.port ?? (conn.driver === "mysql" ? 3306 : 5432), + database: conn.database ?? (conn.driver === "mysql" ? "mysql" : "postgres"), + user: conn.user ?? (conn.driver === "mysql" ? "root" : "postgres"), password, folderId: conn.folderId, id: conn.id, - }); + }; + return conn.driver === "mysql" + ? connectMysql(opts) + : connectPostgres(opts); } async function disconnect(id: string) { @@ -424,6 +471,7 @@ export const useConnectionsStore = defineStore("connections", () => { launchDemo, connectSqlite, connectPostgres, + connectMysql, ensureConnected, disconnect, removeConnection, diff --git a/apps/tauri/src/ui/ConnectionHome.vue b/apps/tauri/src/ui/ConnectionHome.vue index 93e60142..31f92ffd 100644 --- a/apps/tauri/src/ui/ConnectionHome.vue +++ b/apps/tauri/src/ui/ConnectionHome.vue @@ -99,18 +99,23 @@ async function loadOverview() { } const isSqlite = computed(() => conn.value?.driver === "sqlite"); -const isPostgres = computed(() => conn.value?.driver === "postgres"); +const isServerSql = computed( + () => + conn.value?.driver === "postgres" || conn.value?.driver === "mysql", +); const driverLabel = computed(() => { const d = conn.value?.driver; if (d === "sqlite") return "SQLite"; if (d === "postgres") return "PostgreSQL"; + if (d === "mysql") return "MySQL"; return d ?? ""; }); const endpointLabel = computed(() => { const c = conn.value; if (!c) return ""; if (c.driver === "sqlite") return c.path ?? ""; - return `${c.host ?? "localhost"}:${c.port ?? 5432}/${c.database ?? ""}`; + const port = c.port ?? (c.driver === "mysql" ? 3306 : 5432); + return `${c.host ?? "localhost"}:${port}/${c.database ?? ""}`; }); watch( @@ -205,14 +210,16 @@ function openRelation(o: { id: string; name: string; kind: string; schema?: stri -