From e85f9df1fe87e919c7603feddfd01a9690ee3e38 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Wed, 13 May 2026 04:44:52 +0800 Subject: [PATCH] feat(agent): pluggable database drivers via JDBC agent subprocess Migrate domestic database support (DaMeng, KingBase, Vastbase, GoldenDB) from built-in ODBC/protocol drivers to standalone Java agent processes communicating via stdin/stdout JSON-RPC 2.0. Agents are distributed as fat JARs with embedded JDBC drivers, downloaded on-demand from GitHub Releases with proxy fallback. - Add AgentDriverClient for subprocess lifecycle + JSON-RPC communication - Add AgentManager for JRE/JAR install state management - Add Tauri commands for driver install/uninstall/list - Add DriverManager UI panel in settings dialog - Extend DatabaseType with Kingbase, Vastbase, Goldendb variants - Migrate Dameng from ODBC (dm_driver.rs) to JDBC agent - Delete dm_driver.rs (343 lines) - Add GitHub proxy fallback for downloads (update.hwdns.net, gh-proxy.org) Agent code lives at https://github.com/t8y2/dbx-agents --- .gitignore | 3 + crates/dbx-core/src/agent_manager.rs | 172 +++++++++ crates/dbx-core/src/connection.rs | 47 ++- crates/dbx-core/src/db/agent_driver.rs | 162 +++++++++ crates/dbx-core/src/db/dm_driver.rs | 343 ------------------ crates/dbx-core/src/db/mod.rs | 2 +- crates/dbx-core/src/lib.rs | 1 + crates/dbx-core/src/models/connection.rs | 16 + crates/dbx-core/src/query.rs | 21 +- crates/dbx-core/src/schema.rs | 101 ++---- src-tauri/src/commands/agents.rs | 171 +++++++++ src-tauri/src/commands/connection.rs | 52 ++- src-tauri/src/commands/mod.rs | 1 + src-tauri/src/lib.rs | 4 + src/components/config/DriverManager.vue | 94 +++++ .../connection/ConnectionDialog.vue | 18 +- .../editor/EditorSettingsDialog.vue | 6 + src/lib/databaseCapabilities.ts | 13 + src/stores/connectionStore.ts | 9 + src/types/database.ts | 3 + 20 files changed, 769 insertions(+), 470 deletions(-) create mode 100644 crates/dbx-core/src/agent_manager.rs create mode 100644 crates/dbx-core/src/db/agent_driver.rs delete mode 100644 crates/dbx-core/src/db/dm_driver.rs create mode 100644 src-tauri/src/commands/agents.rs create mode 100644 src/components/config/DriverManager.vue diff --git a/.gitignore b/.gitignore index 0cd799edb..d59ce464c 100644 --- a/.gitignore +++ b/.gitignore @@ -43,3 +43,6 @@ tmp/ # Logs *.log + +# Agent JDBC driver JARs +agents/*/libs/*.jar diff --git a/crates/dbx-core/src/agent_manager.rs b/crates/dbx-core/src/agent_manager.rs new file mode 100644 index 000000000..09e650466 --- /dev/null +++ b/crates/dbx-core/src/agent_manager.rs @@ -0,0 +1,172 @@ +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use crate::db::agent_driver::AgentDriverClient; +use crate::models::connection::DatabaseType; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentRegistry { + pub jre: JreInfo, + pub drivers: std::collections::HashMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JreInfo { + pub version: String, + pub platforms: std::collections::HashMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DriverInfo { + pub version: String, + pub label: String, + pub min_app_version: String, + pub jar: ArtifactInfo, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ArtifactInfo { + pub url: String, + pub sha256: String, + pub size: u64, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentState { + #[serde(default)] + pub jre_version: Option, + #[serde(default)] + pub installed_drivers: std::collections::HashMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InstalledDriver { + pub version: String, + pub installed_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentDriverInfo { + pub db_type: String, + pub label: String, + pub version: String, + pub size: u64, + pub installed: bool, + pub installed_version: Option, + pub update_available: bool, +} + +pub struct AgentManager { + base_dir: PathBuf, +} + +impl AgentManager { + pub fn new() -> Self { + let home = + std::env::var(if cfg!(windows) { "USERPROFILE" } else { "HOME" }).unwrap_or_else(|_| ".".to_string()); + Self { base_dir: PathBuf::from(home).join(".dbx").join("agents") } + } + + pub fn base_dir(&self) -> &PathBuf { + &self.base_dir + } + + pub fn jre_java_path(&self) -> PathBuf { + if cfg!(windows) { + self.base_dir.join("jre").join("bin").join("java.exe") + } else { + self.base_dir.join("jre").join("bin").join("java") + } + } + + pub fn driver_jar_path(&self, db_type: &str) -> PathBuf { + self.base_dir.join("drivers").join(db_type).join("agent.jar") + } + + fn state_path(&self) -> PathBuf { + self.base_dir.join("state.json") + } + + pub fn load_state(&self) -> AgentState { + std::fs::read_to_string(self.state_path()) + .ok() + .and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or(AgentState { jre_version: None, installed_drivers: Default::default() }) + } + + pub fn save_state(&self, state: &AgentState) -> Result<(), String> { + let dir = self.base_dir.clone(); + std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?; + let json = serde_json::to_string_pretty(state).map_err(|e| e.to_string())?; + std::fs::write(self.state_path(), json).map_err(|e| e.to_string()) + } + + pub fn is_jre_installed(&self) -> bool { + self.jre_java_path().exists() + } + + pub fn is_driver_installed(&self, db_type: &str) -> bool { + self.driver_jar_path(db_type).exists() + } + + pub fn db_type_to_agent_key(db_type: &DatabaseType) -> Option<&'static str> { + match db_type { + DatabaseType::Dameng => Some("dameng"), + DatabaseType::Kingbase => Some("kingbase"), + DatabaseType::Vastbase => Some("vastbase"), + DatabaseType::Goldendb => Some("goldendb"), + _ => None, + } + } + + pub fn is_agent_type(db_type: &DatabaseType) -> bool { + Self::db_type_to_agent_key(db_type).is_some() + } + + pub async fn spawn(&self, db_type: &DatabaseType) -> Result { + let key = Self::db_type_to_agent_key(db_type) + .ok_or_else(|| format!("{:?} is not an agent-driven database type", db_type))?; + + if !self.is_jre_installed() { + return Err("JRE runtime is not installed. Please install it from the Driver Manager.".to_string()); + } + if !self.is_driver_installed(key) { + return Err(format!("{key} driver is not installed. Please install it from the Driver Manager.")); + } + + let java = self.jre_java_path().to_string_lossy().to_string(); + let jar = self.driver_jar_path(key).to_string_lossy().to_string(); + AgentDriverClient::spawn(&java, &jar).await + } + + pub async fn download_file(url: &str, dest: &Path) -> Result<(), String> { + if let Some(parent) = dest.parent() { + std::fs::create_dir_all(parent).map_err(|e| e.to_string())?; + } + let resp = reqwest::get(url).await.map_err(|e| format!("Download failed: {e}"))?; + if !resp.status().is_success() { + return Err(format!("Download failed with status: {}", resp.status())); + } + let bytes = resp.bytes().await.map_err(|e| format!("Download read failed: {e}"))?; + std::fs::write(dest, &bytes).map_err(|e| format!("Failed to write file: {e}")) + } + + pub fn current_platform() -> &'static str { + if cfg!(target_os = "macos") && cfg!(target_arch = "aarch64") { + "macos-aarch64" + } else if cfg!(target_os = "macos") && cfg!(target_arch = "x86_64") { + "macos-x64" + } else if cfg!(target_os = "linux") && cfg!(target_arch = "aarch64") { + "linux-aarch64" + } else if cfg!(target_os = "linux") && cfg!(target_arch = "x86_64") { + "linux-x64" + } else if cfg!(target_os = "windows") && cfg!(target_arch = "aarch64") { + "windows-aarch64" + } else if cfg!(target_os = "windows") && cfg!(target_arch = "x86_64") { + "windows-x64" + } else { + "unknown" + } + } +} diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index f8adcf8d5..c8d498b2d 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -45,7 +45,7 @@ pub enum PoolKind { SqlServer(Arc>), Oracle(Arc), Elasticsearch(db::elasticsearch_driver::EsClient), - Dameng(Arc>), + Agent(Arc>), Gaussdb(Arc>), ExternalTabular(Arc), ExternalDriver { driver_id: String, config: ConnectionConfig, session: Arc }, @@ -99,6 +99,7 @@ pub struct AppState { pub proxy_tunnels: ProxyTunnelManager, pub storage: Storage, pub plugins: PluginRegistry, + pub agent_manager: crate::agent_manager::AgentManager, } pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig { @@ -112,7 +113,7 @@ pub fn metadata_connection_config(config: &ConnectionConfig) -> ConnectionConfig pub fn database_connection_config(config: &ConnectionConfig, database: Option<&str>) -> ConnectionConfig { let mut db_config = if database.is_some() { config.clone() } else { metadata_connection_config(config) }; if let Some(db) = database { - if db_config.db_type != DatabaseType::Oracle && db_config.db_type != DatabaseType::Dameng { + if !matches!(db_config.db_type, DatabaseType::Oracle | DatabaseType::Dameng) { db_config.database = Some(db.to_string()); } } @@ -133,6 +134,7 @@ impl AppState { proxy_tunnels: ProxyTunnelManager::new(), storage, plugins: PluginRegistry::new(plugin_dir), + agent_manager: crate::agent_manager::AgentManager::new(), } } @@ -169,6 +171,9 @@ impl AppState { | Some(DatabaseType::DuckDb) | Some(DatabaseType::Oracle) | Some(DatabaseType::Dameng) + | Some(DatabaseType::Kingbase) + | Some(DatabaseType::Vastbase) + | Some(DatabaseType::Goldendb) | Some(DatabaseType::Jdbc) ); let pool_key = if is_single_conn { @@ -272,16 +277,21 @@ impl AppState { db::elasticsearch_driver::test_connection(&client).await?; PoolKind::Elasticsearch(client) } - DatabaseType::Dameng => { - let client = db::dm_driver::connect( - &host, - port, - db_config.database.as_deref().unwrap_or(""), - &db_config.username, - &db_config.password, - ) - .await?; - PoolKind::Dameng(Arc::new(std::sync::Mutex::new(client))) + DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => { + let mut client = self.agent_manager.spawn(&db_config.db_type).await?; + client + .call::( + "connect", + serde_json::json!({ + "host": host, + "port": port, + "database": db_config.effective_database().unwrap_or(""), + "username": db_config.username, + "password": db_config.password, + }), + ) + .await?; + PoolKind::Agent(Arc::new(tokio::sync::Mutex::new(client))) } DatabaseType::Gaussdb => { let client = db::gaussdb_driver::connect( @@ -404,9 +414,15 @@ impl AppState { configs .get(connection_id) .map(|c| { - c.db_type == DatabaseType::Oracle - || c.db_type == DatabaseType::Elasticsearch - || c.db_type == DatabaseType::Dameng + matches!( + c.db_type, + DatabaseType::Oracle + | DatabaseType::Elasticsearch + | DatabaseType::Dameng + | DatabaseType::Kingbase + | DatabaseType::Vastbase + | DatabaseType::Goldendb + ) }) .unwrap_or(false) }; @@ -449,6 +465,7 @@ pub async fn probe_connection_endpoint(config: &ConnectionConfig, host: &str, po DatabaseType::Sqlite | DatabaseType::DuckDb => Ok(()), DatabaseType::MongoDb if config.connection_string.as_deref().is_some_and(|value| !value.is_empty()) => Ok(()), DatabaseType::Jdbc => Ok(()), + DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => Ok(()), _ => db::probe_tcp_endpoint(&format!("{:?}", config.db_type), host, port).await, } } diff --git a/crates/dbx-core/src/db/agent_driver.rs b/crates/dbx-core/src/db/agent_driver.rs new file mode 100644 index 000000000..2743769a8 --- /dev/null +++ b/crates/dbx-core/src/db/agent_driver.rs @@ -0,0 +1,162 @@ +use std::io::{BufRead, BufReader, BufWriter, Write}; +use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; +use std::time::Duration; + +use serde::de::DeserializeOwned; +use serde_json::Value; + +const RPC_TIMEOUT_SECS: u64 = 30; +const STARTUP_TIMEOUT_SECS: u64 = 15; + +pub struct AgentDriverClient { + child: Child, + stdin: Option>, + stdout: Option>, + next_id: u64, +} + +impl AgentDriverClient { + /// Spawn a Java agent process and wait for it to signal readiness. + /// + /// The agent is started via `java -jar ` with stdin/stdout piped. + /// Blocks (async) until the agent writes `{"ready":true}` to stdout. + pub async fn spawn(java_path: &str, jar_path: &str) -> Result { + let mut child = Command::new(java_path) + .arg("-jar") + .arg(jar_path) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .map_err(|e| format!("Failed to spawn agent process: {e}"))?; + + let child_stdin = child.stdin.take().ok_or("Failed to capture agent stdin")?; + let child_stdout = child.stdout.take().ok_or("Failed to capture agent stdout")?; + + let stdin = BufWriter::new(child_stdin); + let mut stdout = BufReader::new(child_stdout); + + // Wait for the agent to signal readiness with {"ready":true} + let ready_stdout = tokio::time::timeout( + Duration::from_secs(STARTUP_TIMEOUT_SECS), + tokio::task::spawn_blocking(move || { + let mut line = String::new(); + stdout.read_line(&mut line).map_err(|e| format!("Failed to read startup line from agent: {e}"))?; + let v: Value = serde_json::from_str(line.trim()) + .map_err(|e| format!("Invalid JSON from agent during startup: {e}"))?; + if v.get("ready") != Some(&Value::Bool(true)) { + return Err(format!("Agent did not send ready signal, got: {line}")); + } + Ok(stdout) + }), + ) + .await + .map_err(|_| format!("Agent startup timed out ({STARTUP_TIMEOUT_SECS}s)"))? + .map_err(|e| format!("Agent startup task failed: {e}"))??; + + Ok(Self { child, stdin: Some(stdin), stdout: Some(ready_stdout), next_id: 0 }) + } + + /// Send a JSON-RPC 2.0 request and wait for the response. + pub async fn call( + &mut self, + method: &str, + params: Value, + ) -> Result { + self.next_id += 1; + let id = self.next_id; + + let request = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + }); + let request_line = + serde_json::to_string(&request).map_err(|e| format!("Failed to serialize JSON-RPC request: {e}"))?; + + // Write request to stdin + { + let writer = self.stdin.as_mut().ok_or("Agent stdin not available")?; + writer.write_all(request_line.as_bytes()).map_err(|e| format!("Failed to write to agent stdin: {e}"))?; + writer.write_all(b"\n").map_err(|e| format!("Failed to write newline to agent stdin: {e}"))?; + writer.flush().map_err(|e| format!("Failed to flush agent stdin: {e}"))?; + } + + // Read response from stdout (blocking, with timeout) + let mut reader = self.stdout.take().ok_or("Agent stdout not available")?; + + let (returned_reader, result) = tokio::time::timeout( + Duration::from_secs(RPC_TIMEOUT_SECS), + tokio::task::spawn_blocking(move || { + let mut line = String::new(); + let read_result = + reader.read_line(&mut line).map_err(|e| format!("Failed to read response from agent: {e}")); + if let Err(e) = read_result { + return (reader, Err(e)); + } + + let resp: Value = match serde_json::from_str(line.trim()) { + Ok(v) => v, + Err(e) => { + return (reader, Err(format!("Invalid JSON response from agent: {e}"))); + } + }; + + let result = if let Some(err) = resp.get("error") { + let msg = err.get("message").and_then(|m| m.as_str()).unwrap_or("Unknown agent error"); + let code = err.get("code").and_then(|c| c.as_i64()).unwrap_or(-1); + Err(format!("Agent RPC error ({code}): {msg}")) + } else if let Some(result_val) = resp.get("result") { + serde_json::from_value::(result_val.clone()) + .map_err(|e| format!("Failed to deserialize agent result: {e}")) + } else { + Err(format!("Agent response missing both 'result' and 'error': {line}")) + }; + + (reader, result) + }), + ) + .await + .map_err(|_| format!("Agent RPC call timed out ({RPC_TIMEOUT_SECS}s)"))? + .map_err(|e| format!("Agent RPC task failed: {e}"))?; + + let _ = self.stdout.insert(returned_reader); + result + } + + /// Send a shutdown message to the agent and wait for the process to exit. + pub async fn shutdown(&mut self) { + // Try to send a shutdown RPC; ignore errors if the agent is already gone + let shutdown_result: Result = self.call("shutdown", Value::Null).await; + if let Err(e) = &shutdown_result { + log::warn!("Agent shutdown RPC failed: {e}"); + } + + // Drop stdin to signal EOF + self.stdin.take(); + + // Wait for the child to exit + match self.child.wait() { + Ok(status) => log::info!("Agent process exited with {status}"), + Err(e) => log::warn!("Failed to wait for agent process: {e}"), + } + } + + /// Forcefully kill the agent process. + pub fn kill(&mut self) { + self.stdin.take(); + self.stdout.take(); + if let Err(e) = self.child.kill() { + log::warn!("Failed to kill agent process: {e}"); + } + // Reap the child to avoid zombie processes + let _ = self.child.wait(); + } +} + +impl Drop for AgentDriverClient { + fn drop(&mut self) { + self.kill(); + } +} diff --git a/crates/dbx-core/src/db/dm_driver.rs b/crates/dbx-core/src/db/dm_driver.rs deleted file mode 100644 index a3cd8b5ea..000000000 --- a/crates/dbx-core/src/db/dm_driver.rs +++ /dev/null @@ -1,343 +0,0 @@ -use std::time::Instant; - -use odbc_api::{buffers::TextRowSet, ConnectionOptions, Cursor, ResultSetMetadata}; - -use crate::sql::starts_with_executable_sql_keyword; -use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo}; - -use super::CONNECTION_TIMEOUT_SECS; - -pub struct DmClient { - conn: odbc_api::Connection<'static>, -} - -unsafe impl Send for DmClient {} - -impl DmClient { - pub fn query_rows(&self, sql: &str) -> Result>, String> { - match self.conn.execute(sql, (), None).map_err(|e| e.to_string())? { - Some(cursor) => read_cursor(cursor), - None => Ok(vec![]), - } - } - - fn query_single_column(&self, sql: &str) -> Result, String> { - Ok(self.query_rows(sql)?.into_iter().filter_map(|r| r.into_iter().next()).collect()) - } -} - -fn read_cursor(cursor: impl Cursor) -> Result>, String> { - let mut cursor = cursor; - let col_count = cursor.num_result_cols().map_err(|e| e.to_string())? as u16; - let buffer = TextRowSet::for_cursor(1000, &mut cursor, Some(8192)).map_err(|e| e.to_string())?; - let mut row_cursor = cursor.bind_buffer(buffer).map_err(|e| e.to_string())?; - let mut rows = Vec::new(); - while let Some(batch) = row_cursor.fetch().map_err(|e| e.to_string())? { - for row_idx in 0..batch.num_rows() { - let vals: Vec = (0..col_count as usize) - .map(|col| { - batch.at(col, row_idx).and_then(|bytes| std::str::from_utf8(bytes).ok()).unwrap_or("").to_string() - }) - .collect(); - rows.push(vals); - } - } - Ok(rows) -} - -pub async fn connect(host: &str, port: u16, database: &str, user: &str, pass: &str) -> Result { - let conn_str = format!( - "Driver={{DM8 ODBC DRIVER}};Server={host};TCP_PORT={port};DATABASE={db};UID={user};PWD={pass}", - host = host, - port = port, - db = database, - user = user, - pass = pass, - ); - - let result = tokio::time::timeout( - std::time::Duration::from_secs(CONNECTION_TIMEOUT_SECS), - tokio::task::spawn_blocking(move || { - super::ODBC_ENV.connect_with_connection_string(&conn_str, ConnectionOptions::default()) - .map_err(|e| { - let msg = e.to_string(); - if msg.contains("Data source name not found") || msg.contains("Can't open lib") { - format!( - "DM8 ODBC driver not found. Please install the DM8 ODBC driver \ - and register it in odbcinst.ini (Linux/macOS) or the ODBC Data Source Administrator (Windows). \ - Original error: {msg}" - ) - } else { - format!("DM connection failed: {msg}") - } - }) - .map(|conn| DmClient { conn }) - }), - ) - .await - .map_err(|_| format!("DM connection timed out ({CONNECTION_TIMEOUT_SECS}s)"))? - .map_err(|e| format!("DM connection task failed: {e}"))?; - - result -} - -pub fn list_databases(client: &DmClient) -> Result, String> { - let rows = client.query_single_column( - "SELECT USERNAME FROM ALL_USERS \ - WHERE USERNAME NOT IN (\ - 'SYS','SYSDBA','SYSAUDITOR','SYSSSO','CTISYS',\ - 'SYS_DBA','_SYS_STATISTICS','SYS_PHM'\ - ) ORDER BY USERNAME", - )?; - Ok(rows.into_iter().map(|name| DatabaseInfo { name }).collect()) -} - -pub fn list_schemas(client: &DmClient) -> Result, String> { - let dbs = list_databases(client)?; - Ok(dbs.into_iter().map(|d| d.name).collect()) -} - -pub fn list_tables(client: &DmClient, schema: &str) -> Result, String> { - let s = schema.replace('\'', "''"); - let sql = format!( - "SELECT TABLE_NAME, 'TABLE' AS TABLE_TYPE FROM ALL_TABLES WHERE OWNER = '{s}' \ - UNION ALL \ - SELECT VIEW_NAME, 'VIEW' FROM ALL_VIEWS WHERE OWNER = '{s}' \ - ORDER BY 1" - ); - let rows = client.query_rows(&sql)?; - Ok(rows - .into_iter() - .map(|r| TableInfo { - name: r.first().cloned().unwrap_or_default(), - table_type: r.get(1).cloned().unwrap_or_else(|| "TABLE".to_string()), - comment: None, - }) - .collect()) -} - -pub fn get_columns(client: &DmClient, schema: &str, table: &str) -> Result, String> { - let s = schema.replace('\'', "''"); - let t = table.replace('\'', "''"); - - let pk_rows = client.query_single_column(&format!( - "SELECT cols.COLUMN_NAME FROM ALL_CONS_COLUMNS cols \ - JOIN ALL_CONSTRAINTS cons ON cols.CONSTRAINT_NAME = cons.CONSTRAINT_NAME AND cols.OWNER = cons.OWNER \ - WHERE cons.CONSTRAINT_TYPE = 'P' AND cons.OWNER = '{s}' AND cons.TABLE_NAME = '{t}'" - ))?; - let pk_names: std::collections::HashSet = pk_rows.into_iter().collect(); - - let col_rows = client.query_rows(&format!( - "SELECT COLUMN_NAME, DATA_TYPE, NULLABLE, DATA_PRECISION, DATA_SCALE, DATA_LENGTH, CHAR_LENGTH \ - FROM ALL_TAB_COLUMNS \ - WHERE OWNER = '{s}' AND TABLE_NAME = '{t}' \ - ORDER BY COLUMN_ID" - ))?; - - Ok(col_rows - .into_iter() - .map(|r| { - let name = r.first().cloned().unwrap_or_default(); - let base = r.get(1).cloned().unwrap_or_default(); - let num_prec = r.get(3).and_then(|v| v.parse::().ok()); - let num_scale = r.get(4).and_then(|v| v.parse::().ok()); - let data_len = r.get(5).and_then(|v| v.parse::().ok()); - let char_len = r.get(6).and_then(|v| v.parse::().ok()); - let data_type = match base.to_uppercase().as_str() { - "VARCHAR2" | "NVARCHAR2" | "VARCHAR" | "CHAR" | "NCHAR" => { - let len = char_len.or(data_len); - match len { - Some(n) => format!("{base}({n})"), - None => base, - } - } - "NUMBER" | "NUMERIC" | "DECIMAL" => match (num_prec, num_scale) { - (Some(p), Some(s)) if s > 0 => format!("{base}({p},{s})"), - (Some(p), _) if p > 0 => format!("{base}({p})"), - _ => base, - }, - "RAW" => match data_len { - Some(n) => format!("RAW({n})"), - None => "RAW".to_string(), - }, - _ => base, - }; - ColumnInfo { - is_primary_key: pk_names.contains(&name), - name, - data_type, - is_nullable: r.get(2).map(|v| v == "Y").unwrap_or(false), - column_default: None, - extra: None, - comment: None, - numeric_precision: num_prec, - numeric_scale: num_scale, - character_maximum_length: char_len, - } - }) - .collect()) -} - -pub fn list_indexes(client: &DmClient, schema: &str, table: &str) -> Result, String> { - let s = schema.replace('\'', "''"); - let t = table.replace('\'', "''"); - let sql = format!( - "SELECT i.INDEX_NAME, \ - LISTAGG(ic.COLUMN_NAME, ',') WITHIN GROUP (ORDER BY ic.COLUMN_POSITION) AS COLUMNS, \ - i.UNIQUENESS, \ - CASE WHEN c.CONSTRAINT_TYPE = 'P' THEN 1 ELSE 0 END AS IS_PK, \ - i.INDEX_TYPE \ - FROM ALL_INDEXES i \ - JOIN ALL_IND_COLUMNS ic ON i.INDEX_NAME = ic.INDEX_NAME AND i.OWNER = ic.INDEX_OWNER AND i.TABLE_OWNER = ic.TABLE_OWNER \ - LEFT JOIN ALL_CONSTRAINTS c ON i.INDEX_NAME = c.INDEX_NAME AND i.TABLE_OWNER = c.OWNER \ - AND c.CONSTRAINT_TYPE = 'P' \ - WHERE i.TABLE_OWNER = '{s}' AND i.TABLE_NAME = '{t}' \ - GROUP BY i.INDEX_NAME, i.UNIQUENESS, c.CONSTRAINT_TYPE, i.INDEX_TYPE \ - ORDER BY i.INDEX_NAME" - ); - let rows = client.query_rows(&sql)?; - Ok(rows - .into_iter() - .map(|r| { - let cols_str = r.get(1).cloned().unwrap_or_default(); - IndexInfo { - name: r.first().cloned().unwrap_or_default(), - columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(), - is_unique: r.get(2).map(|v| v == "UNIQUE").unwrap_or(false), - is_primary: r.get(3).map(|v| v == "1").unwrap_or(false), - filter: None, - index_type: r.get(4).cloned(), - included_columns: None, - comment: None, - } - }) - .collect()) -} - -pub fn list_foreign_keys(client: &DmClient, schema: &str, table: &str) -> Result, String> { - let s = schema.replace('\'', "''"); - let t = table.replace('\'', "''"); - let sql = format!( - "SELECT c.CONSTRAINT_NAME, cc.COLUMN_NAME, rc.TABLE_NAME, rcc.COLUMN_NAME \ - FROM ALL_CONSTRAINTS c \ - JOIN ALL_CONS_COLUMNS cc ON c.CONSTRAINT_NAME = cc.CONSTRAINT_NAME AND c.OWNER = cc.OWNER \ - JOIN ALL_CONSTRAINTS rc ON c.R_CONSTRAINT_NAME = rc.CONSTRAINT_NAME AND c.R_OWNER = rc.OWNER \ - JOIN ALL_CONS_COLUMNS rcc ON rc.CONSTRAINT_NAME = rcc.CONSTRAINT_NAME AND rc.OWNER = rcc.OWNER \ - WHERE c.CONSTRAINT_TYPE = 'R' AND c.OWNER = '{s}' AND c.TABLE_NAME = '{t}' \ - ORDER BY c.CONSTRAINT_NAME" - ); - let rows = client.query_rows(&sql)?; - Ok(rows - .into_iter() - .map(|r| ForeignKeyInfo { - name: r.first().cloned().unwrap_or_default(), - column: r.get(1).cloned().unwrap_or_default(), - ref_table: r.get(2).cloned().unwrap_or_default(), - ref_column: r.get(3).cloned().unwrap_or_default(), - }) - .collect()) -} - -pub fn list_triggers(client: &DmClient, schema: &str, table: &str) -> Result, String> { - let s = schema.replace('\'', "''"); - let t = table.replace('\'', "''"); - let sql = format!( - "SELECT TRIGGER_NAME, TRIGGERING_EVENT, TRIGGER_TYPE \ - FROM ALL_TRIGGERS \ - WHERE OWNER = '{s}' AND TABLE_NAME = '{t}' \ - ORDER BY TRIGGER_NAME" - ); - let rows = client.query_rows(&sql)?; - Ok(rows - .into_iter() - .map(|r| TriggerInfo { - name: r.first().cloned().unwrap_or_default(), - event: r.get(1).cloned().unwrap_or_default(), - timing: r.get(2).cloned().unwrap_or_default(), - }) - .collect()) -} - -pub fn execute_query_with_schema_sync(client: &DmClient, schema: &str, sql: &str) -> Result { - let set_schema = format!("SET SCHEMA \"{}\"", schema); - client.conn.execute(&set_schema, (), None).map_err(|e| { - log::error!("[dameng] set schema failed: {e}"); - e.to_string() - })?; - execute_query_sync(client, sql) -} - -pub fn execute_query_sync(client: &DmClient, sql: &str) -> Result { - let start = Instant::now(); - let sql = sql.trim().trim_end_matches(';'); - - if starts_with_executable_sql_keyword(sql, &["SELECT", "WITH", "SHOW", "DESCRIBE", "EXPLAIN"]) { - match client.conn.execute(sql, (), None).map_err(|e| e.to_string())? { - Some(mut cursor) => { - let col_count = cursor.num_result_cols().map_err(|e| e.to_string())? as u16; - let columns: Vec = (1..=col_count) - .map(|i| cursor.col_name(i).map_err(|e| e.to_string()).unwrap_or_else(|_| format!("col{i}"))) - .collect(); - - let buffer = TextRowSet::for_cursor(1000, &mut cursor, Some(8192)).map_err(|e| e.to_string())?; - let mut row_cursor = cursor.bind_buffer(buffer).map_err(|e| e.to_string())?; - - let mut rows = Vec::new(); - while let Some(batch) = row_cursor.fetch().map_err(|e| e.to_string())? { - for row_idx in 0..batch.num_rows() { - let vals: Vec = (0..col_count as usize) - .map(|col| { - batch - .at(col, row_idx) - .and_then(|bytes| std::str::from_utf8(bytes).ok()) - .map(|s| serde_json::Value::String(s.to_string())) - .unwrap_or(serde_json::Value::Null) - }) - .collect(); - rows.push(vals); - if rows.len() >= crate::query::MAX_ROWS { - break; - } - } - if rows.len() >= crate::query::MAX_ROWS { - break; - } - } - - let truncated = rows.len() >= crate::query::MAX_ROWS; - Ok(QueryResult { - columns, - rows, - affected_rows: 0, - execution_time_ms: start.elapsed().as_millis(), - truncated, - }) - } - None => Ok(QueryResult { - columns: vec![], - rows: vec![], - affected_rows: 0, - execution_time_ms: start.elapsed().as_millis(), - truncated: false, - }), - } - } else { - match client.conn.execute(sql, (), None) { - Ok(Some(_cursor)) => Ok(QueryResult { - columns: vec![], - rows: vec![], - affected_rows: 0, - execution_time_ms: start.elapsed().as_millis(), - truncated: false, - }), - Ok(None) => Ok(QueryResult { - columns: vec![], - rows: vec![], - affected_rows: 0, - execution_time_ms: start.elapsed().as_millis(), - truncated: false, - }), - Err(e) => Err(e.to_string()), - } - } -} diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index 2ef320001..17d718342 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/crates/dbx-core/src/db/mod.rs @@ -1,5 +1,5 @@ +pub mod agent_driver; pub mod clickhouse_driver; -pub mod dm_driver; pub mod duckdb_driver; pub mod elasticsearch_driver; pub mod file_validator; diff --git a/crates/dbx-core/src/lib.rs b/crates/dbx-core/src/lib.rs index d561104d4..59c8d5d3b 100644 --- a/crates/dbx-core/src/lib.rs +++ b/crates/dbx-core/src/lib.rs @@ -1,3 +1,4 @@ +pub mod agent_manager; pub mod ai; pub mod connection; pub mod connection_secrets; diff --git a/crates/dbx-core/src/models/connection.rs b/crates/dbx-core/src/models/connection.rs index f47e90f2d..af7442659 100644 --- a/crates/dbx-core/src/models/connection.rs +++ b/crates/dbx-core/src/models/connection.rs @@ -113,6 +113,9 @@ pub enum DatabaseType { StarRocks, Redshift, Dameng, + Kingbase, + Vastbase, + Goldendb, Gaussdb, Jdbc, } @@ -142,6 +145,7 @@ impl ConnectionConfig { }, DatabaseType::Redshift => Some("dev"), DatabaseType::Gaussdb => Some("postgres"), + DatabaseType::Kingbase | DatabaseType::Vastbase => Some("postgres"), _ => None, } } @@ -208,6 +212,9 @@ impl ConnectionConfig { DatabaseType::Oracle => format!("oracle://{host}:{port}{db_part}"), DatabaseType::Elasticsearch => format!("http://{host}:{port}"), DatabaseType::Dameng => format!("dm://{host}:{port}{db_part}"), + DatabaseType::Kingbase => format!("kingbase://{host}:{port}{db_part}"), + DatabaseType::Vastbase => format!("vastbase://{host}:{port}{db_part}"), + DatabaseType::Goldendb => format!("goldendb://{host}:{port}{db_part}"), DatabaseType::Gaussdb => format!("gaussdb://{host}:{port}{db_part}"), DatabaseType::Jdbc => "jdbc:".to_string(), } @@ -277,6 +284,15 @@ impl ConnectionConfig { DatabaseType::Dameng => { format!("dm://{}:{}@{host}:{port}{db_part}", username, password) } + DatabaseType::Kingbase => { + format!("kingbase://{}:{}@{host}:{port}{db_part}", username, password) + } + DatabaseType::Vastbase => { + format!("vastbase://{}:{}@{host}:{port}{db_part}", username, password) + } + DatabaseType::Goldendb => { + format!("goldendb://{}:{}@{host}:{port}{db_part}", username, password) + } DatabaseType::Gaussdb => { format!("gaussdb://{}:{}@{host}:{port}{db_part}", username, password) } diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 56f9c9a28..9d7ac67c1 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -233,21 +233,18 @@ pub async fn do_execute( } PoolKind::Redis(_) => Err("Use Redis-specific commands".to_string()), PoolKind::MongoDb(_) => Err("Use MongoDB-specific commands".to_string()), - PoolKind::Dameng(client) => { + PoolKind::Agent(client) => { let client = client.clone(); let sql = sql.to_string(); let schema = schema.map(|s| s.to_string()); drop(connections); wait_for_query(cancel_token, async move { - let task = tokio::task::spawn_blocking(move || { - let client = client.lock().map_err(|e| e.to_string())?; - if let Some(schema) = schema { - db::dm_driver::execute_query_with_schema_sync(&client, &schema, &sql) - } else { - db::dm_driver::execute_query_sync(&client, &sql) - } - }); - task.await.map_err(|e| e.to_string())? + let mut client = client.lock().await; + let params = match schema { + Some(s) => serde_json::json!({"sql": sql, "schema": s}), + None => serde_json::json!({"sql": sql}), + }; + client.call("execute_query", params).await }) .await .map(truncate_result) @@ -501,7 +498,7 @@ pub async fn execute_statements( /// Execute multiple SQL statements within a single transaction. /// For sqlx-based pools (Postgres/MySQL/SQLite), uses the Transaction API to /// guarantee all statements run on the same physical connection. -/// For custom drivers (ClickHouse/SqlServer/Dameng/Gaussdb), uses explicit +/// For custom drivers (ClickHouse/SqlServer/Agent/Gaussdb), uses explicit /// BEGIN/COMMIT/ROLLBACK on the already-single-connection client. /// For databases that don't support explicit transactions (Redis, MongoDB, Oracle), /// executes statements sequentially without transaction. @@ -528,7 +525,7 @@ pub async fn execute_statements_in_transaction( PoolKind::Postgres(pg) => TxPath::Pg(pg.clone()), PoolKind::Mysql(mp, _mode) => TxPath::Mysql(mp.clone(), false), PoolKind::Sqlite(sq) => TxPath::Sqlite(sq.clone()), - PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Dameng(_) | PoolKind::Gaussdb(_) => { + PoolKind::ClickHouse(_) | PoolKind::SqlServer(_) | PoolKind::Agent(_) | PoolKind::Gaussdb(_) => { TxPath::Explicit } PoolKind::DuckDb(_) diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 05005063e..10c7d6779 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -109,12 +109,12 @@ pub fn extract_oracle(connections: &HashMap, key: &str) -> Opt } } -pub fn extract_dameng( +pub fn extract_agent( connections: &HashMap, key: &str, -) -> Option>> { +) -> Option>> { match connections.get(key)? { - PoolKind::Dameng(client) => Some(client.clone()), + PoolKind::Agent(client) => Some(client.clone()), _ => None, } } @@ -158,10 +158,10 @@ pub async fn list_databases_core(state: &AppState, connection_id: &str) -> Resul let client = client.lock().await; return db::oracle_driver::list_databases(&*client).await; } - if let Some(client) = extract_dameng(&connections, connection_id) { + if let Some(client) = extract_agent(&connections, connection_id) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::list_databases(&client); + let mut client = client.lock().await; + return client.call("list_databases", serde_json::json!({})).await; } if let Some(client) = extract_gaussdb(&connections, connection_id) { drop(connections); @@ -212,10 +212,10 @@ pub async fn list_schemas_core(state: &AppState, connection_id: &str, database: let client = client.lock().await; return db::oracle_driver::list_schemas(&*client).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::list_schemas(&client); + let mut client = client.lock().await; + return client.call("list_schemas", serde_json::json!({"database": database})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -284,10 +284,10 @@ pub async fn list_tables_core( let client = client.lock().await; return db::oracle_driver::list_tables(&*client, schema).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::list_tables(&client, schema); + let mut client = client.lock().await; + return client.call("list_tables", serde_json::json!({"schema": schema})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -424,10 +424,10 @@ pub async fn get_columns_core( let client = client.lock().await; return db::oracle_driver::get_columns(&*client, schema, table).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::get_columns(&client, schema, table); + let mut client = client.lock().await; + return client.call("get_columns", serde_json::json!({"schema": schema, "table": table})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -475,10 +475,10 @@ pub async fn list_indexes_core( let client = client.lock().await; return db::oracle_driver::list_indexes(&*client, schema, table).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::list_indexes(&client, schema, table); + let mut client = client.lock().await; + return client.call("list_indexes", serde_json::json!({"schema": schema, "table": table})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -526,10 +526,10 @@ pub async fn list_foreign_keys_core( let client = client.lock().await; return db::oracle_driver::list_foreign_keys(&*client, schema, table).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::list_foreign_keys(&client, schema, table); + let mut client = client.lock().await; + return client.call("list_foreign_keys", serde_json::json!({"schema": schema, "table": table})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -577,10 +577,10 @@ pub async fn list_triggers_core( let client = client.lock().await; return db::oracle_driver::list_triggers(&*client, schema, table).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return db::dm_driver::list_triggers(&client, schema, table); + let mut client = client.lock().await; + return client.call("list_triggers", serde_json::json!({"schema": schema, "table": table})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -654,10 +654,10 @@ pub async fn get_table_ddl_core( let client = client.lock().await; return build_oracle_ddl(&*client, schema, table).await; } - if let Some(client) = extract_dameng(&connections, &pool_key) { + if let Some(client) = extract_agent(&connections, &pool_key) { drop(connections); - let client = client.lock().map_err(|e| e.to_string())?; - return build_dameng_ddl(&client, schema, table); + let mut client = client.lock().await; + return client.call("get_table_ddl", serde_json::json!({"schema": schema, "table": table})).await; } if let Some(client) = extract_gaussdb(&connections, &pool_key) { drop(connections); @@ -1099,53 +1099,6 @@ pub async fn build_oracle_ddl( Ok(ddl) } -pub fn build_dameng_ddl(client: &db::dm_driver::DmClient, schema: &str, table: &str) -> Result { - let columns = db::dm_driver::get_columns(client, schema, table)?; - let indexes = db::dm_driver::list_indexes(client, schema, table)?; - let fkeys = db::dm_driver::list_foreign_keys(client, schema, table)?; - - let mut ddl = format!("CREATE TABLE \"{schema}\".\"{table}\" (\n"); - let col_lines: Vec = columns - .iter() - .map(|c| { - let mut line = format!(" \"{}\" {}", c.name, c.data_type); - if !c.is_nullable { - line.push_str(" NOT NULL"); - } - if let Some(ref def) = c.column_default { - line.push_str(&format!(" DEFAULT {def}")); - } - line - }) - .collect(); - ddl.push_str(&col_lines.join(",\n")); - - let pks: Vec<&str> = columns.iter().filter(|c| c.is_primary_key).map(|c| c.name.as_str()).collect(); - if !pks.is_empty() { - ddl.push_str(&format!( - ",\n PRIMARY KEY ({})", - pks.iter().map(|k| format!("\"{k}\"")).collect::>().join(", ") - )); - } - for fk in &fkeys { - ddl.push_str(&format!( - ",\n CONSTRAINT \"{}\" FOREIGN KEY (\"{}\") REFERENCES \"{}\"(\"{}\")", - fk.name, fk.column, fk.ref_table, fk.ref_column - )); - } - ddl.push_str("\n);\n"); - - for idx in &indexes { - if idx.is_primary { - continue; - } - let unique = if idx.is_unique { "UNIQUE " } else { "" }; - let cols = idx.columns.iter().map(|c| format!("\"{c}\"")).collect::>().join(", "); - ddl.push_str(&format!("\nCREATE {unique}INDEX \"{}\" ON \"{schema}\".\"{table}\" ({cols});", idx.name)); - } - Ok(ddl) -} - pub async fn build_gaussdb_ddl( client: &mut db::gaussdb_driver::GaussdbClient, schema: &str, diff --git a/src-tauri/src/commands/agents.rs b/src-tauri/src/commands/agents.rs new file mode 100644 index 000000000..7d4d03fc3 --- /dev/null +++ b/src-tauri/src/commands/agents.rs @@ -0,0 +1,171 @@ +use std::sync::Arc; + +use tauri::State; + +use dbx_core::agent_manager::{AgentDriverInfo, AgentManager, AgentRegistry, InstalledDriver}; +use dbx_core::connection::AppState; + +const REGISTRY_URLS: &[&str] = &[ + "https://update.hwdns.net/https://github.com/t8y2/dbx-agents/releases/latest/download/agent-registry.json", + "https://gh-proxy.org/https://github.com/t8y2/dbx-agents/releases/latest/download/agent-registry.json", + "https://github.com/t8y2/dbx-agents/releases/latest/download/agent-registry.json", +]; + +const DOWNLOAD_PROXIES: &[&str] = &["https://update.hwdns.net/", "https://gh-proxy.org/", ""]; + +#[tauri::command] +pub async fn list_installed_agents(state: State<'_, Arc>) -> Result, String> { + let am = &state.agent_manager; + let local_state = am.load_state(); + let registry = fetch_registry().await.ok(); + + let agent_types = [ + ("dameng", "达梦 DM8"), + ("kingbase", "人大金仓 KingbaseES"), + ("vastbase", "Vastbase"), + ("goldendb", "GoldenDB"), + ]; + + Ok(agent_types + .iter() + .map(|(key, label)| { + let installed = am.is_driver_installed(key); + let local = local_state.installed_drivers.get(*key); + let remote = registry.as_ref().and_then(|r| r.drivers.get(*key)); + AgentDriverInfo { + db_type: key.to_string(), + label: label.to_string(), + version: remote.map(|r| r.version.clone()).unwrap_or_default(), + size: remote.map(|r| r.jar.size).unwrap_or(0), + installed, + installed_version: local.map(|l| l.version.clone()), + update_available: match (local, remote) { + (Some(l), Some(r)) => l.version != r.version, + _ => false, + }, + } + }) + .collect()) +} + +#[tauri::command] +pub async fn install_agent(state: State<'_, Arc>, db_type: String) -> Result<(), String> { + let am = &state.agent_manager; + let registry = fetch_registry().await?; + + if !am.is_jre_installed() { + let platform = AgentManager::current_platform(); + let jre_info = + registry.jre.platforms.get(platform).ok_or_else(|| format!("No JRE available for platform: {platform}"))?; + let jre_archive = am.base_dir().join("jre-download.tar.gz"); + download_with_proxy(&jre_info.url, &jre_archive).await?; + extract_archive(&jre_archive, &am.base_dir().join("jre"))?; + std::fs::remove_file(&jre_archive).ok(); + } + + let driver = registry.drivers.get(&db_type).ok_or_else(|| format!("Unknown driver type: {db_type}"))?; + let jar_path = am.driver_jar_path(&db_type); + download_with_proxy(&driver.jar.url, &jar_path).await?; + + let mut local_state = am.load_state(); + local_state.jre_version = Some(registry.jre.version.clone()); + local_state.installed_drivers.insert( + db_type, + InstalledDriver { version: driver.version.clone(), installed_at: chrono::Utc::now().to_rfc3339() }, + ); + am.save_state(&local_state)?; + Ok(()) +} + +#[tauri::command] +pub async fn uninstall_agent(state: State<'_, Arc>, db_type: String) -> Result<(), String> { + let am = &state.agent_manager; + let jar_path = am.driver_jar_path(&db_type); + if jar_path.exists() { + std::fs::remove_file(&jar_path).map_err(|e| e.to_string())?; + } + let driver_dir = jar_path.parent().unwrap(); + if driver_dir.exists() { + std::fs::remove_dir_all(driver_dir).map_err(|e| e.to_string())?; + } + let mut local_state = am.load_state(); + local_state.installed_drivers.remove(&db_type); + am.save_state(&local_state)?; + Ok(()) +} + +#[tauri::command] +pub async fn check_jre_installed(state: State<'_, Arc>) -> Result { + Ok(state.agent_manager.is_jre_installed()) +} + +async fn fetch_registry() -> Result { + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .build() + .map_err(|e| format!("Failed to create HTTP client: {e}"))?; + let mut last_err = String::new(); + for url in REGISTRY_URLS { + match client + .get(*url) + .header(reqwest::header::USER_AGENT, "dbx-agent-manager") + .send() + .await + .and_then(|r| r.error_for_status()) + { + Ok(resp) => { + return resp.json().await.map_err(|e| format!("Failed to parse registry: {e}")); + } + Err(e) => { + last_err = format!("{e}"); + } + } + } + Err(format!("Failed to fetch agent registry: {last_err}")) +} + +async fn download_with_proxy(url: &str, dest: &std::path::Path) -> Result<(), String> { + if let Some(parent) = dest.parent() { + std::fs::create_dir_all(parent).map_err(|e| e.to_string())?; + } + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(300)) + .build() + .map_err(|e| format!("Failed to create HTTP client: {e}"))?; + let mut last_err = String::new(); + for proxy in DOWNLOAD_PROXIES { + let full_url = format!("{proxy}{url}"); + log::info!("[agent] downloading from {full_url}"); + match client + .get(&full_url) + .header(reqwest::header::USER_AGENT, "dbx-agent-manager") + .send() + .await + .and_then(|r| r.error_for_status()) + { + Ok(resp) => { + let bytes = resp.bytes().await.map_err(|e| format!("Download read failed: {e}"))?; + std::fs::write(dest, &bytes).map_err(|e| format!("Failed to write file: {e}"))?; + return Ok(()); + } + Err(e) => { + last_err = format!("{e}"); + log::warn!("[agent] download failed from {full_url}: {last_err}"); + } + } + } + Err(format!("Failed to download {url}: {last_err}")) +} + +fn extract_archive(archive: &std::path::Path, dest: &std::path::Path) -> Result<(), String> { + use std::process::Command; + std::fs::create_dir_all(dest).map_err(|e| e.to_string())?; + let status = Command::new("tar") + .args(["xzf", &archive.to_string_lossy(), "-C", &dest.to_string_lossy(), "--strip-components=1"]) + .status() + .map_err(|e| format!("Failed to extract archive: {e}"))?; + if !status.success() { + return Err("Failed to extract JRE archive".to_string()); + } + Ok(()) +} diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index e074cc767..77d01f3e4 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -112,15 +112,22 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection db::elasticsearch_driver::EsClient::new(&url, Some(&config.username), Some(&config.password)); db::elasticsearch_driver::test_connection(&client).await.map(|_| "Connection successful".to_string()) } - DatabaseType::Dameng => db::dm_driver::connect( - &host, - port, - config.database.as_deref().unwrap_or(""), - &config.username, - &config.password, - ) - .await - .map(|_| "Connection successful".to_string()), + DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => { + let mut client = state.agent_manager.spawn(&config.db_type).await?; + client + .call::( + "test_connection", + serde_json::json!({ + "host": host, + "port": port, + "database": config.database.as_deref().unwrap_or(""), + "username": config.username, + "password": config.password, + }), + ) + .await?; + Ok("Connection successful".to_string()) + } DatabaseType::Gaussdb => db::gaussdb_driver::connect( &host, port, @@ -220,16 +227,21 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi db::elasticsearch_driver::test_connection(&client).await?; PoolKind::Elasticsearch(client) } - DatabaseType::Dameng => { - let client = db::dm_driver::connect( - &host, - port, - db_config.database.as_deref().unwrap_or(""), - &db_config.username, - &db_config.password, - ) - .await?; - PoolKind::Dameng(std::sync::Arc::new(std::sync::Mutex::new(client))) + DatabaseType::Dameng | DatabaseType::Kingbase | DatabaseType::Vastbase | DatabaseType::Goldendb => { + let mut client = state.agent_manager.spawn(&db_config.db_type).await?; + client + .call::( + "connect", + serde_json::json!({ + "host": host, + "port": port, + "database": db_config.effective_database().unwrap_or(""), + "username": db_config.username, + "password": db_config.password, + }), + ) + .await?; + PoolKind::Agent(std::sync::Arc::new(tokio::sync::Mutex::new(client))) } DatabaseType::Gaussdb => { let client = db::gaussdb_driver::connect( @@ -269,7 +281,7 @@ pub async fn disconnect_db(state: State<'_, Arc>, connection_id: Strin PoolKind::SqlServer(_) => {} PoolKind::Oracle(_) => {} PoolKind::Elasticsearch(_) => {} - PoolKind::Dameng(_) => {} + PoolKind::Agent(_) => {} PoolKind::Gaussdb(_) => {} PoolKind::ExternalTabular(_) => {} PoolKind::ExternalDriver { .. } => {} diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index ce374a731..3e3841a3c 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -1,3 +1,4 @@ +pub mod agents; pub mod ai; pub mod connection; #[allow(dead_code, unused_imports)] diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 169998049..a431131d4 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -143,6 +143,10 @@ pub fn run() { commands::transfer::cancel_transfer, commands::database_export::export_database_sql, commands::database_export::cancel_database_export, + commands::agents::list_installed_agents, + commands::agents::install_agent, + commands::agents::uninstall_agent, + commands::agents::check_jre_installed, ]) .build(tauri::generate_context!()) .expect("error while building tauri application") diff --git a/src/components/config/DriverManager.vue b/src/components/config/DriverManager.vue new file mode 100644 index 000000000..30c6dd303 --- /dev/null +++ b/src/components/config/DriverManager.vue @@ -0,0 +1,94 @@ + + + diff --git a/src/components/connection/ConnectionDialog.vue b/src/components/connection/ConnectionDialog.vue index 7b71c77e1..3d87c7cf8 100644 --- a/src/components/connection/ConnectionDialog.vue +++ b/src/components/connection/ConnectionDialog.vue @@ -146,7 +146,7 @@ const driverProfiles: Record< mariadb: { type: "mysql", port: 3306, user: "root", label: "MariaDB", icon: "mariadb" }, tidb: { type: "mysql", port: 4000, user: "root", label: "TiDB", icon: "tidb" }, oceanbase: { type: "mysql", port: 2881, user: "root", label: "OceanBase", icon: "oceanbase" }, - goldendb: { type: "mysql", port: 3306, user: "root", label: "GoldenDB", icon: "goldendb" }, + goldendb: { type: "goldendb", port: 3306, user: "root", label: "GoldenDB", icon: "goldendb" }, opengauss: { type: "gaussdb", port: 5432, @@ -155,8 +155,8 @@ const driverProfiles: Record< icon: "opengauss", }, gaussdb: { type: "gaussdb", port: 5432, user: "gaussdb", label: "GaussDB", icon: "gaussdb" }, - kingbase: { type: "postgres", port: 54321, user: "system", label: "KingBase", icon: "kingbase" }, - vastbase: { type: "postgres", port: 5432, user: "vastbase", label: "Vastbase", icon: "vastbase" }, + kingbase: { type: "kingbase", port: 54321, user: "system", label: "KingBase", icon: "kingbase" }, + vastbase: { type: "vastbase", port: 5432, user: "vastbase", label: "Vastbase", icon: "vastbase" }, doris: { type: "mysql", port: 9030, user: "root", label: "Doris", icon: "doris", urlParams: "" }, selectdb: { type: "mysql", @@ -322,7 +322,8 @@ function defaultDatabaseForProfile() { if (form.value.db_type === "redshift") return "dev"; if (form.value.db_type === "gaussdb") return "postgres"; if (selectedType.value === "cockroachdb") return "defaultdb"; - if (form.value.db_type === "postgres") return "postgres"; + if (form.value.db_type === "postgres" || form.value.db_type === "kingbase" || form.value.db_type === "vastbase") + return "postgres"; if (form.value.db_type === "sqlserver") return "master"; if (form.value.db_type === "oracle") return "ORCL"; return ""; @@ -1116,7 +1117,14 @@ function openExternalUrl(url: string) {
diff --git a/src/components/editor/EditorSettingsDialog.vue b/src/components/editor/EditorSettingsDialog.vue index e0cb19f6a..5081df957 100644 --- a/src/components/editor/EditorSettingsDialog.vue +++ b/src/components/editor/EditorSettingsDialog.vue @@ -25,6 +25,7 @@ import { isTauriRuntime } from "@/lib/tauriRuntime"; import type { JdbcDriverInfo, JdbcPluginStatus } from "@/types/database"; import * as api from "@/lib/api"; import { aiTestConnection } from "@/lib/api"; +import DriverManager from "@/components/config/DriverManager.vue"; import { useToast } from "@/composables/useToast"; const { t } = useI18n(); @@ -489,6 +490,7 @@ watch( {{ t("settings.appearanceTab") }} {{ t("settings.aiTab") }} {{ t("settings.jdbcTab") }} + 驱动管理 {{ t("settings.securityTab") }} {{ t("settings.aboutTab") }} @@ -928,6 +930,10 @@ watch( + + + +
diff --git a/src/lib/databaseCapabilities.ts b/src/lib/databaseCapabilities.ts index c485f4ffc..47e36d5d5 100644 --- a/src/lib/databaseCapabilities.ts +++ b/src/lib/databaseCapabilities.ts @@ -7,6 +7,8 @@ export const SCHEMA_AWARE_TYPES = new Set([ "redshift", "dameng", "gaussdb", + "kingbase", + "vastbase", "jdbc", ]); @@ -21,6 +23,9 @@ export const DIAGRAM_SUPPORTED_TYPES = new Set([ "redshift", "dameng", "gaussdb", + "kingbase", + "vastbase", + "goldendb", ]); export const DATABASE_SEARCH_SUPPORTED_TYPES = new Set([ @@ -34,6 +39,9 @@ export const DATABASE_SEARCH_SUPPORTED_TYPES = new Set([ "clickhouse", "dameng", "gaussdb", + "kingbase", + "vastbase", + "goldendb", ]); export const TABLE_IMPORT_SUPPORTED_TYPES = new Set([ @@ -49,6 +57,9 @@ export const TABLE_IMPORT_SUPPORTED_TYPES = new Set([ "redshift", "dameng", "gaussdb", + "kingbase", + "vastbase", + "goldendb", ]); export const TABLE_STRUCTURE_SUPPORTED_TYPES = new Set(["mysql", "postgres", "sqlite", "sqlserver"]); @@ -85,6 +96,8 @@ export const TREE_SCHEMA_TYPES = new Set(["postgres", "redshift", export const PG_LIKE_STRUCTURE_TYPES = new Set(["postgres", "redshift", "gaussdb"]); +export const AGENT_DRIVER_TYPES = new Set(["dameng", "kingbase", "vastbase", "goldendb"]); + export const TRANSFER_SQL_TYPES = new Set([ "mysql", "postgres", diff --git a/src/stores/connectionStore.ts b/src/stores/connectionStore.ts index 5843c97b7..3ec27afa3 100644 --- a/src/stores/connectionStore.ts +++ b/src/stores/connectionStore.ts @@ -159,6 +159,9 @@ export const useConnectionStore = defineStore("connection", () => { redshift: "Redshift", dameng: "DM (Dameng)", gaussdb: "GaussDB", + kingbase: "KingBase", + vastbase: "Vastbase", + goldendb: "GoldenDB", }; const profile = config.driver_profile || config.db_type; @@ -167,6 +170,12 @@ export const useConnectionStore = defineStore("connection", () => { dbType = "gaussdb" as ConnectionConfig["db_type"]; } else if (profile === "redshift" && dbType === "postgres") { dbType = "redshift" as ConnectionConfig["db_type"]; + } else if (profile === "kingbase" && dbType === "postgres") { + dbType = "kingbase" as ConnectionConfig["db_type"]; + } else if (profile === "vastbase" && dbType === "postgres") { + dbType = "vastbase" as ConnectionConfig["db_type"]; + } else if (profile === "goldendb" && dbType === "mysql") { + dbType = "goldendb" as ConnectionConfig["db_type"]; } return { diff --git a/src/types/database.ts b/src/types/database.ts index e5c23c758..6da9b8f80 100644 --- a/src/types/database.ts +++ b/src/types/database.ts @@ -14,6 +14,9 @@ export type DatabaseType = | "redshift" | "dameng" | "gaussdb" + | "kingbase" + | "vastbase" + | "goldendb" | "jdbc"; export interface ConnectionConfig {