From b1f4d4c30042f49f97c026c6c935815129be618b Mon Sep 17 00:00:00 2001 From: skyler <77960507+t8y2@users.noreply.github.com> Date: Thu, 30 Apr 2026 19:40:12 +0800 Subject: [PATCH] feat: add Elasticsearch support with document browser (#38) - New elasticsearch_driver using reqwest HTTP client - Reuse MongoDB document browser (MongoDocBrowser) for ES indices - ES indices shown as collections under "default" database - Document CRUD via ES REST API (_search, _doc) - No extra dependencies (uses existing reqwest) --- src-tauri/src/commands/connection.rs | 32 ++- src-tauri/src/commands/mongo_cmd.rs | 35 ++- src-tauri/src/commands/query.rs | 1 + src-tauri/src/db/elasticsearch_driver.rs | 232 ++++++++++++++++++ src-tauri/src/db/mod.rs | 1 + src-tauri/src/models/connection.rs | 4 + .../connection/ConnectionDialog.vue | 3 + src/components/icons/DatabaseIcon.vue | 3 +- src/components/sidebar/TreeItem.vue | 6 +- src/stores/connectionStore.ts | 1 + src/types/database.ts | 2 +- 11 files changed, 309 insertions(+), 11 deletions(-) create mode 100644 src-tauri/src/db/elasticsearch_driver.rs diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 1b61d2276c..138ef4ecb8 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -17,6 +17,7 @@ pub enum PoolKind { ClickHouse(db::clickhouse_driver::ChClient), SqlServer(std::sync::Arc>), Oracle(std::sync::Arc>), + Elasticsearch(db::elasticsearch_driver::EsClient), } pub struct AppState { @@ -123,6 +124,15 @@ impl AppState { .await?; PoolKind::Oracle(std::sync::Arc::new(tokio::sync::Mutex::new(client))) } + DatabaseType::Elasticsearch => { + let client = db::elasticsearch_driver::EsClient::new( + &url, + Some(&db_config.username), + Some(&db_config.password), + ); + db::elasticsearch_driver::test_connection(&client).await?; + PoolKind::Elasticsearch(client) + } }; self.connections.lock().await.insert(pool_key.clone(), pool); @@ -167,7 +177,7 @@ impl AppState { let is_single_conn = { let configs = self.configs.lock().await; configs.get(connection_id) - .map(|c| c.db_type == DatabaseType::Oracle) + .map(|c| c.db_type == DatabaseType::Oracle || c.db_type == DatabaseType::Elasticsearch) .unwrap_or(false) }; let pool_key = if is_single_conn { @@ -315,6 +325,16 @@ pub async fn test_connection( ) .await .map(|_| "Connection successful".to_string()), + DatabaseType::Elasticsearch => { + let client = db::elasticsearch_driver::EsClient::new( + &url, + Some(&config.username), + Some(&config.password), + ); + db::elasticsearch_driver::test_connection(&client) + .await + .map(|_| "Connection successful".to_string()) + } }; if config.ssh_enabled && !config.ssh_host.is_empty() { @@ -376,6 +396,15 @@ pub async fn connect_db( .await?; PoolKind::Oracle(std::sync::Arc::new(tokio::sync::Mutex::new(client))) } + DatabaseType::Elasticsearch => { + let client = db::elasticsearch_driver::EsClient::new( + &url, + Some(&config.username), + Some(&config.password), + ); + db::elasticsearch_driver::test_connection(&client).await?; + PoolKind::Elasticsearch(client) + } }; state.connections.lock().await.insert(id.clone(), pool); @@ -407,6 +436,7 @@ pub async fn disconnect_db( PoolKind::ClickHouse(_) => {}, PoolKind::SqlServer(_) => {}, PoolKind::Oracle(_) => {}, + PoolKind::Elasticsearch(_) => {}, } } } diff --git a/src-tauri/src/commands/mongo_cmd.rs b/src-tauri/src/commands/mongo_cmd.rs index 9fb84c46c6..9c1c72cab7 100644 --- a/src-tauri/src/commands/mongo_cmd.rs +++ b/src-tauri/src/commands/mongo_cmd.rs @@ -3,6 +3,7 @@ use tauri::State; use crate::commands::connection::{AppState, PoolKind}; use crate::db::mongo_driver::{self, MongoDocumentResult}; +use crate::db::elasticsearch_driver; #[tauri::command] pub async fn mongo_list_databases( @@ -12,7 +13,8 @@ pub async fn mongo_list_databases( let connections = state.connections.lock().await; match connections.get(&connection_id).ok_or("Not found")? { PoolKind::MongoDb(client) => mongo_driver::list_databases(client).await, - _ => Err("Not a MongoDB connection".to_string()), + PoolKind::Elasticsearch(_) => Ok(vec!["default".to_string()]), + _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } @@ -25,7 +27,8 @@ pub async fn mongo_list_collections( let connections = state.connections.lock().await; match connections.get(&connection_id).ok_or("Not found")? { PoolKind::MongoDb(client) => mongo_driver::list_collections(client, &database).await, - _ => Err("Not a MongoDB connection".to_string()), + PoolKind::Elasticsearch(client) => elasticsearch_driver::list_indices(client).await, + _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } @@ -43,7 +46,12 @@ pub async fn mongo_find_documents( PoolKind::MongoDb(client) => { mongo_driver::find_documents(client, &database, &collection, skip, limit).await } - _ => Err("Not a MongoDB connection".to_string()), + PoolKind::Elasticsearch(client) => { + let client = client.clone(); + drop(connections); + elasticsearch_driver::find_documents(&client, &collection, skip, limit).await + } + _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } @@ -60,7 +68,12 @@ pub async fn mongo_insert_document( PoolKind::MongoDb(client) => { mongo_driver::insert_document(client, &database, &collection, &doc_json).await } - _ => Err("Not a MongoDB connection".to_string()), + PoolKind::Elasticsearch(client) => { + let client = client.clone(); + drop(connections); + elasticsearch_driver::insert_document(&client, &collection, &doc_json).await + } + _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } @@ -78,7 +91,12 @@ pub async fn mongo_update_document( PoolKind::MongoDb(client) => { mongo_driver::update_document(client, &database, &collection, &id, &doc_json).await } - _ => Err("Not a MongoDB connection".to_string()), + PoolKind::Elasticsearch(client) => { + let client = client.clone(); + drop(connections); + elasticsearch_driver::update_document(&client, &collection, &id, &doc_json).await + } + _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } @@ -95,6 +113,11 @@ pub async fn mongo_delete_document( PoolKind::MongoDb(client) => { mongo_driver::delete_document(client, &database, &collection, &id).await } - _ => Err("Not a MongoDB connection".to_string()), + PoolKind::Elasticsearch(client) => { + let client = client.clone(); + drop(connections); + elasticsearch_driver::delete_document(&client, &collection, &id).await + } + _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } diff --git a/src-tauri/src/commands/query.rs b/src-tauri/src/commands/query.rs index d00a47b220..d4fcc2046c 100644 --- a/src-tauri/src/commands/query.rs +++ b/src-tauri/src/commands/query.rs @@ -141,6 +141,7 @@ async fn do_execute( .map_err(|_| format!("Query timed out after {} seconds", QUERY_TIMEOUT.as_secs()))? .map(truncate_result) } + PoolKind::Elasticsearch(_) => Err("Use document browser for Elasticsearch".to_string()), PoolKind::Redis(_) => Err("Use Redis-specific commands".to_string()), PoolKind::MongoDb(_) => Err("Use MongoDB-specific commands".to_string()), } diff --git a/src-tauri/src/db/elasticsearch_driver.rs b/src-tauri/src/db/elasticsearch_driver.rs new file mode 100644 index 0000000000..f6094c2d5a --- /dev/null +++ b/src-tauri/src/db/elasticsearch_driver.rs @@ -0,0 +1,232 @@ +use reqwest::Client as HttpClient; +use serde::Deserialize; + +use super::mongo_driver::MongoDocumentResult; + +pub struct EsClient { + http: HttpClient, + base_url: String, + auth: Option<(String, String)>, +} + +impl EsClient { + pub fn new(url: &str, username: Option<&str>, password: Option<&str>) -> Self { + let auth = match (username, password) { + (Some(u), Some(p)) if !u.is_empty() => Some((u.to_string(), p.to_string())), + _ => None, + }; + Self { + http: HttpClient::new(), + base_url: url.trim_end_matches('/').to_string(), + auth, + } + } + + fn get(&self, path: &str) -> reqwest::RequestBuilder { + let req = self.http.get(format!("{}{}", self.base_url, path)); + self.with_auth(req) + } + + fn post(&self, path: &str) -> reqwest::RequestBuilder { + let req = self.http.post(format!("{}{}", self.base_url, path)); + self.with_auth(req) + } + + fn put(&self, path: &str) -> reqwest::RequestBuilder { + let req = self.http.put(format!("{}{}", self.base_url, path)); + self.with_auth(req) + } + + fn delete(&self, path: &str) -> reqwest::RequestBuilder { + let req = self.http.delete(format!("{}{}", self.base_url, path)); + self.with_auth(req) + } + + fn with_auth(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder { + if let Some((ref user, ref pass)) = self.auth { + req.basic_auth(user, Some(pass)) + } else { + req + } + } +} + +impl Clone for EsClient { + fn clone(&self) -> Self { + Self { + http: self.http.clone(), + base_url: self.base_url.clone(), + auth: self.auth.clone(), + } + } +} + +pub async fn test_connection(client: &EsClient) -> Result<(), String> { + let resp = client.get("/") + .send() + .await + .map_err(|e| format!("Elasticsearch connection failed: {e}"))?; + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Elasticsearch error: {body}")); + } + Ok(()) +} + +#[derive(Deserialize)] +struct CatIndex { + index: String, +} + +pub async fn list_indices(client: &EsClient) -> Result, String> { + let resp = client.get("/_cat/indices?format=json&h=index") + .send() + .await + .map_err(|e| format!("Elasticsearch request failed: {e}"))?; + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Elasticsearch error: {body}")); + } + let indices: Vec = resp.json().await.map_err(|e| format!("Elasticsearch parse error: {e}"))?; + let mut names: Vec = indices + .into_iter() + .filter(|i| !i.index.starts_with('.')) + .map(|i| i.index) + .collect(); + names.sort(); + Ok(names) +} + +#[derive(Deserialize)] +struct SearchResponse { + hits: SearchHits, +} + +#[derive(Deserialize)] +struct SearchHits { + total: HitsTotal, + hits: Vec, +} + +#[derive(Deserialize)] +struct HitsTotal { + value: u64, +} + +#[derive(Deserialize)] +struct SearchHit { + #[serde(rename = "_id")] + id: String, + #[serde(rename = "_source")] + source: serde_json::Value, +} + +pub async fn find_documents( + client: &EsClient, + index: &str, + skip: u64, + limit: i64, +) -> Result { + let body = serde_json::json!({ + "from": skip, + "size": limit, + "sort": ["_doc"], + }); + + let path = format!("/{}/_search", index); + let resp = client.post(&path) + .json(&body) + .send() + .await + .map_err(|e| format!("Elasticsearch request failed: {e}"))?; + + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Elasticsearch error: {body}")); + } + + let result: SearchResponse = resp.json().await + .map_err(|e| format!("Elasticsearch parse error: {e}"))?; + + let documents: Vec = result.hits.hits.into_iter().map(|hit| { + let mut doc = match hit.source { + serde_json::Value::Object(map) => map, + _ => serde_json::Map::new(), + }; + doc.insert("_id".to_string(), serde_json::Value::String(hit.id)); + serde_json::Value::Object(doc) + }).collect(); + + Ok(MongoDocumentResult { + documents, + total: result.hits.total.value, + }) +} + +pub async fn insert_document( + client: &EsClient, + index: &str, + doc_json: &str, +) -> Result { + let doc: serde_json::Value = serde_json::from_str(doc_json) + .map_err(|e| format!("Invalid JSON: {e}"))?; + + let path = format!("/{}/_doc?refresh=true", index); + let resp = client.post(&path) + .json(&doc) + .send() + .await + .map_err(|e| format!("Elasticsearch request failed: {e}"))?; + + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Elasticsearch error: {body}")); + } + + let result: serde_json::Value = resp.json().await + .map_err(|e| format!("Elasticsearch parse error: {e}"))?; + Ok(result["_id"].as_str().unwrap_or("").to_string()) +} + +pub async fn update_document( + client: &EsClient, + index: &str, + id: &str, + doc_json: &str, +) -> Result { + let doc: serde_json::Value = serde_json::from_str(doc_json) + .map_err(|e| format!("Invalid JSON: {e}"))?; + + let path = format!("/{}/_doc/{}?refresh=true", index, id); + let resp = client.put(&path) + .json(&doc) + .send() + .await + .map_err(|e| format!("Elasticsearch request failed: {e}"))?; + + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Elasticsearch error: {body}")); + } + + Ok(1) +} + +pub async fn delete_document( + client: &EsClient, + index: &str, + id: &str, +) -> Result { + let path = format!("/{}/_doc/{}?refresh=true", index, id); + let resp = client.delete(&path) + .send() + .await + .map_err(|e| format!("Elasticsearch request failed: {e}"))?; + + if !resp.status().is_success() { + let body = resp.text().await.unwrap_or_default(); + return Err(format!("Elasticsearch error: {body}")); + } + + Ok(1) +} diff --git a/src-tauri/src/db/mod.rs b/src-tauri/src/db/mod.rs index 7e6b95102a..14ce2d436d 100644 --- a/src-tauri/src/db/mod.rs +++ b/src-tauri/src/db/mod.rs @@ -1,4 +1,5 @@ pub mod clickhouse_driver; +pub mod elasticsearch_driver; pub mod mongo_driver; pub mod mysql; pub mod oracle_driver; diff --git a/src-tauri/src/models/connection.rs b/src-tauri/src/models/connection.rs index e07bde928a..4e74573182 100644 --- a/src-tauri/src/models/connection.rs +++ b/src-tauri/src/models/connection.rs @@ -58,6 +58,8 @@ pub enum DatabaseType { MongoDb, #[serde(rename = "oracle")] Oracle, + #[serde(rename = "elasticsearch")] + Elasticsearch, } impl ConnectionConfig { @@ -107,6 +109,7 @@ impl ConnectionConfig { format!("mongodb://{host}:{port}{db_part}") } DatabaseType::Oracle => format!("oracle://{host}:{port}{db_part}"), + DatabaseType::Elasticsearch => format!("http://{host}:{port}"), } } @@ -170,6 +173,7 @@ impl ConnectionConfig { DatabaseType::Oracle => { format!("oracle://{}:{}@{host}:{port}{db_part}", username, password) } + DatabaseType::Elasticsearch => format!("http://{host}:{port}"), } } diff --git a/src/components/connection/ConnectionDialog.vue b/src/components/connection/ConnectionDialog.vue index a74dd6cd64..5d94b03c8c 100644 --- a/src/components/connection/ConnectionDialog.vue +++ b/src/components/connection/ConnectionDialog.vue @@ -82,6 +82,7 @@ const driverProfiles: Record = { mysql: "mysql", postgres: "postgres", sqlite: "sqlite", redis: "redis", mongodb: "mongodb", duckdb: "duckdb", clickhouse: "clickhouse", sqlserver: "sqlserver", oracle: "oracle", + elasticsearch: "elasticsearch", mariadb: "mariadb", tidb: "tidb", oceanbase: "oceanbase", goldendb: "goldendb", opengauss: "opengauss", gaussdb: "gaussdb", kingbase: "kingbase", vastbase: "vastbase", custom_mysql: "mysql", custom_postgres: "postgres", @@ -189,6 +191,7 @@ const dbOptions = [ { value: "clickhouse", label: "ClickHouse" }, { value: "sqlserver", label: "SQL Server" }, { value: "oracle", label: "Oracle" }, + { value: "elasticsearch", label: "Elasticsearch" }, { value: "mariadb", label: "MariaDB" }, ]; diff --git a/src/components/icons/DatabaseIcon.vue b/src/components/icons/DatabaseIcon.vue index 5698989a51..685fa0a8e0 100644 --- a/src/components/icons/DatabaseIcon.vue +++ b/src/components/icons/DatabaseIcon.vue @@ -1,6 +1,6 @@