mirror of
https://github.com/t8y2/dbx.git
synced 2026-11-02 00:29:30 +00:00
feat: extract dbx-core crate for web deployment support
Refactor all database drivers, query execution, schema operations, and business logic into a shared `dbx-core` crate. The Tauri desktop app becomes a thin shell delegating to dbx-core, enabling future web and CLI frontends to reuse the same core.
This commit is contained in:
t8y2
@@ -1,175 +0,0 @@
|
||||
use reqwest::Client as HttpClient;
|
||||
use serde::Deserialize;
|
||||
use std::time::Instant;
|
||||
|
||||
use super::{ColumnInfo, DatabaseInfo, QueryResult, TableInfo};
|
||||
|
||||
pub struct ChClient {
|
||||
http: HttpClient,
|
||||
base_url: String,
|
||||
username: Option<String>,
|
||||
password: Option<String>,
|
||||
}
|
||||
|
||||
impl ChClient {
|
||||
pub fn new(url: &str, username: Option<String>, password: Option<String>) -> Self {
|
||||
Self {
|
||||
http: HttpClient::new(),
|
||||
base_url: url.trim_end_matches('/').to_string(),
|
||||
username,
|
||||
password,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Clone for ChClient {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
http: self.http.clone(),
|
||||
base_url: self.base_url.clone(),
|
||||
username: self.username.clone(),
|
||||
password: self.password.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ChJsonResult {
|
||||
meta: Vec<ChColumn>,
|
||||
data: Vec<Vec<serde_json::Value>>,
|
||||
#[serde(default)]
|
||||
#[allow(dead_code)]
|
||||
rows: usize,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ChColumn {
|
||||
name: String,
|
||||
#[serde(rename = "type")]
|
||||
_type: String,
|
||||
}
|
||||
|
||||
fn build_request(client: &ChClient, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
|
||||
match (&client.username, &client.password) {
|
||||
(Some(u), Some(p)) if !u.is_empty() => req.basic_auth(u, Some(p)),
|
||||
(Some(u), None) if !u.is_empty() => req.basic_auth(u, None::<&str>),
|
||||
_ => req,
|
||||
}
|
||||
}
|
||||
|
||||
async fn ch_query(client: &ChClient, sql: &str, database: Option<&str>) -> Result<ChJsonResult, String> {
|
||||
let mut url = format!("{}/?default_format=JSONCompact", client.base_url);
|
||||
if let Some(db) = database {
|
||||
url.push_str(&format!("&database={}", db));
|
||||
}
|
||||
let req = build_request(client, client.http.post(&url).body(sql.to_string()));
|
||||
let resp = req
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| format!("ClickHouse request failed: {e}"))?;
|
||||
if !resp.status().is_success() {
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
return Err(format!("ClickHouse error: {body}"));
|
||||
}
|
||||
resp.json::<ChJsonResult>().await.map_err(|e| format!("ClickHouse parse error: {e}"))
|
||||
}
|
||||
|
||||
pub async fn test_connection(client: &ChClient) -> Result<(), String> {
|
||||
let url = format!("{}/ping", client.base_url);
|
||||
let req = build_request(client, client.http.get(&url));
|
||||
req.send().await
|
||||
.map_err(|e| format!("ClickHouse connection failed: {e}"))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn list_databases(client: &ChClient) -> Result<Vec<DatabaseInfo>, String> {
|
||||
let result = ch_query(client, "SELECT name FROM system.databases ORDER BY name", None).await?;
|
||||
Ok(result.data.iter().map(|row| {
|
||||
DatabaseInfo { name: row[0].as_str().unwrap_or("").to_string() }
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_tables(client: &ChClient, database: &str) -> Result<Vec<TableInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT name, engine FROM system.tables WHERE database = '{}' ORDER BY name",
|
||||
database.replace('\'', "\\'")
|
||||
);
|
||||
let result = ch_query(client, &sql, Some(database)).await?;
|
||||
Ok(result.data.iter().map(|row| {
|
||||
let engine = row.get(1).and_then(|v| v.as_str()).unwrap_or("");
|
||||
let table_type = if engine.contains("View") { "VIEW" } else { "BASE TABLE" };
|
||||
TableInfo {
|
||||
name: row[0].as_str().unwrap_or("").to_string(),
|
||||
table_type: table_type.to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(client: &ChClient, database: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT name, type, default_kind, default_expression, is_in_primary_key \
|
||||
FROM system.columns WHERE database = '{}' AND table = '{}' ORDER BY position",
|
||||
database.replace('\'', "\\'"),
|
||||
table.replace('\'', "\\'")
|
||||
);
|
||||
let result = ch_query(client, &sql, Some(database)).await?;
|
||||
Ok(result.data.iter().map(|row| {
|
||||
let data_type = row.get(1).and_then(|v| v.as_str()).unwrap_or("").to_string();
|
||||
let is_nullable = data_type.starts_with("Nullable");
|
||||
let is_pk = row.get(4).and_then(|v| v.as_u64()).unwrap_or(0) == 1;
|
||||
let default_kind = row.get(2).and_then(|v| v.as_str()).unwrap_or("");
|
||||
let default_expr = row.get(3).and_then(|v| v.as_str()).unwrap_or("");
|
||||
let column_default = if default_kind.is_empty() { None } else { Some(default_expr.to_string()) };
|
||||
ColumnInfo {
|
||||
name: row[0].as_str().unwrap_or("").to_string(),
|
||||
data_type,
|
||||
is_nullable,
|
||||
column_default,
|
||||
is_primary_key: is_pk,
|
||||
extra: None, comment: None,
|
||||
numeric_precision: None,
|
||||
numeric_scale: None,
|
||||
character_maximum_length: None,
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn execute_query(client: &ChClient, database: &str, sql: &str) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
let trimmed = sql.trim().to_uppercase();
|
||||
|
||||
if trimmed.starts_with("SELECT")
|
||||
|| trimmed.starts_with("SHOW")
|
||||
|| trimmed.starts_with("DESCRIBE")
|
||||
|| trimmed.starts_with("EXPLAIN")
|
||||
|| trimmed.starts_with("WITH")
|
||||
{
|
||||
let result = ch_query(client, sql, Some(database)).await?;
|
||||
let columns: Vec<String> = result.meta.iter().map(|c| c.name.clone()).collect();
|
||||
Ok(QueryResult {
|
||||
columns,
|
||||
rows: result.data,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
} else {
|
||||
let url = format!("{}/?default_format=JSONCompact&database={}", client.base_url, database);
|
||||
let req = build_request(client, client.http.post(&url).body(sql.to_string()));
|
||||
let resp = req
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| format!("ClickHouse request failed: {e}"))?;
|
||||
if !resp.status().is_success() {
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
return Err(format!("ClickHouse error: {body}"));
|
||||
}
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,232 +0,0 @@
|
||||
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<Vec<String>, 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<CatIndex> = resp.json().await.map_err(|e| format!("Elasticsearch parse error: {e}"))?;
|
||||
let mut names: Vec<String> = 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<SearchHit>,
|
||||
}
|
||||
|
||||
#[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<MongoDocumentResult, String> {
|
||||
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<serde_json::Value> = 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<String, String> {
|
||||
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<u64, String> {
|
||||
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<u64, String> {
|
||||
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)
|
||||
}
|
||||
+4
-74
@@ -1,74 +1,4 @@
|
||||
pub mod clickhouse_driver;
|
||||
pub mod elasticsearch_driver;
|
||||
pub mod mongo_driver;
|
||||
pub mod mysql;
|
||||
pub mod oracle_driver;
|
||||
pub mod postgres;
|
||||
pub mod redis_driver;
|
||||
pub mod sqlite;
|
||||
pub mod sqlserver;
|
||||
pub mod ssh_tunnel;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DatabaseInfo {
|
||||
pub name: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct TableInfo {
|
||||
pub name: String,
|
||||
pub table_type: String, // "TABLE" or "VIEW"
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ColumnInfo {
|
||||
pub name: String,
|
||||
pub data_type: String,
|
||||
pub is_nullable: bool,
|
||||
pub column_default: Option<String>,
|
||||
pub is_primary_key: bool,
|
||||
pub extra: Option<String>,
|
||||
pub comment: Option<String>,
|
||||
pub numeric_precision: Option<i32>,
|
||||
pub numeric_scale: Option<i32>,
|
||||
pub character_maximum_length: Option<i32>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct QueryResult {
|
||||
pub columns: Vec<String>,
|
||||
pub rows: Vec<Vec<serde_json::Value>>,
|
||||
pub affected_rows: u64,
|
||||
pub execution_time_ms: u128,
|
||||
#[serde(default)]
|
||||
pub truncated: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct IndexInfo {
|
||||
pub name: String,
|
||||
pub columns: Vec<String>,
|
||||
pub is_unique: bool,
|
||||
pub is_primary: bool,
|
||||
pub filter: Option<String>,
|
||||
pub index_type: Option<String>,
|
||||
pub included_columns: Option<Vec<String>>,
|
||||
pub comment: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ForeignKeyInfo {
|
||||
pub name: String,
|
||||
pub column: String,
|
||||
pub ref_table: String,
|
||||
pub ref_column: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct TriggerInfo {
|
||||
pub name: String,
|
||||
pub event: String,
|
||||
pub timing: String,
|
||||
}
|
||||
#[allow(unused_imports)]
|
||||
pub use dbx_core::db::*;
|
||||
#[allow(unused_imports)]
|
||||
pub use dbx_core::types::*;
|
||||
|
||||
@@ -1,121 +0,0 @@
|
||||
use mongodb::{bson::{doc, Document, Bson}, Client};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MongoDocumentResult {
|
||||
pub documents: Vec<serde_json::Value>,
|
||||
pub total: u64,
|
||||
}
|
||||
|
||||
pub async fn list_databases(client: &Client) -> Result<Vec<String>, String> {
|
||||
client
|
||||
.list_database_names()
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn list_collections(client: &Client, database: &str) -> Result<Vec<String>, String> {
|
||||
client
|
||||
.database(database)
|
||||
.list_collection_names()
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn find_documents(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
|
||||
let total = col.count_documents(doc! {}).await.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut cursor = col
|
||||
.find(doc! {})
|
||||
.skip(skip)
|
||||
.limit(limit)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut documents = Vec::new();
|
||||
while cursor.advance().await.map_err(|e| e.to_string())? {
|
||||
let doc = cursor.deserialize_current().map_err(|e| e.to_string())?;
|
||||
let json = bson_to_json(&Bson::Document(doc));
|
||||
documents.push(json);
|
||||
}
|
||||
|
||||
Ok(MongoDocumentResult { documents, total })
|
||||
}
|
||||
|
||||
pub async fn insert_document(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
doc_json: &str,
|
||||
) -> Result<String, String> {
|
||||
let doc: Document = serde_json::from_str(doc_json)
|
||||
.map_err(|e| format!("Invalid JSON: {e}"))?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let result = col.insert_one(doc).await.map_err(|e| e.to_string())?;
|
||||
Ok(format!("{}", result.inserted_id))
|
||||
}
|
||||
|
||||
pub async fn update_document(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
id: &str,
|
||||
doc_json: &str,
|
||||
) -> Result<u64, String> {
|
||||
let oid = mongodb::bson::oid::ObjectId::parse_str(id)
|
||||
.map_err(|e| format!("Invalid ObjectId: {e}"))?;
|
||||
let new_doc: Document = serde_json::from_str(doc_json)
|
||||
.map_err(|e| format!("Invalid JSON: {e}"))?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let result = col
|
||||
.replace_one(doc! { "_id": oid }, new_doc)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(result.modified_count)
|
||||
}
|
||||
|
||||
pub async fn delete_document(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
id: &str,
|
||||
) -> Result<u64, String> {
|
||||
let oid = mongodb::bson::oid::ObjectId::parse_str(id)
|
||||
.map_err(|e| format!("Invalid ObjectId: {e}"))?;
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let result = col
|
||||
.delete_one(doc! { "_id": oid })
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(result.deleted_count)
|
||||
}
|
||||
|
||||
fn bson_to_json(bson: &Bson) -> serde_json::Value {
|
||||
match bson {
|
||||
Bson::Double(v) => serde_json::json!(v),
|
||||
Bson::String(v) => serde_json::Value::String(v.clone()),
|
||||
Bson::Boolean(v) => serde_json::Value::Bool(*v),
|
||||
Bson::Null => serde_json::Value::Null,
|
||||
Bson::Int32(v) => serde_json::json!(v),
|
||||
Bson::Int64(v) => serde_json::json!(v),
|
||||
Bson::ObjectId(oid) => serde_json::Value::String(oid.to_hex()),
|
||||
Bson::DateTime(dt) => serde_json::Value::String(dt.to_string()),
|
||||
Bson::Array(arr) => serde_json::Value::Array(arr.iter().map(bson_to_json).collect()),
|
||||
Bson::Document(doc) => {
|
||||
let mut map = serde_json::Map::new();
|
||||
for (k, v) in doc {
|
||||
map.insert(k.clone(), bson_to_json(v));
|
||||
}
|
||||
serde_json::Value::Object(map)
|
||||
}
|
||||
_ => serde_json::Value::String(format!("{bson}")),
|
||||
}
|
||||
}
|
||||
@@ -1,432 +0,0 @@
|
||||
use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
|
||||
use rust_decimal::Decimal;
|
||||
use sqlx::mysql::{MySqlPool, MySqlPoolOptions, MySqlRow};
|
||||
use sqlx::{Column, Executor, Row, TypeInfo, ValueRef};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use super::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
|
||||
|
||||
fn quote_value(s: &str) -> String {
|
||||
format!("'{}'", s.replace('\\', "\\\\").replace('\'', "\\'"))
|
||||
}
|
||||
|
||||
fn get_str(row: &MySqlRow, idx: usize) -> String {
|
||||
row.try_get::<String, _>(idx)
|
||||
.or_else(|_| row.try_get::<Vec<u8>, _>(idx).map(|b| String::from_utf8_lossy(&b).to_string()))
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn get_str_by_name(row: &MySqlRow, name: &str) -> String {
|
||||
row.try_get::<String, _>(name)
|
||||
.or_else(|_| row.try_get::<Vec<u8>, _>(name).map(|b| String::from_utf8_lossy(&b).to_string()))
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn get_opt_str(row: &MySqlRow, name: &str) -> Option<String> {
|
||||
row.try_get::<Option<String>, _>(name)
|
||||
.ok()
|
||||
.flatten()
|
||||
.or_else(|| {
|
||||
row.try_get::<Option<Vec<u8>>, _>(name)
|
||||
.ok()
|
||||
.flatten()
|
||||
.map(|b| String::from_utf8_lossy(&b).to_string())
|
||||
})
|
||||
}
|
||||
|
||||
fn numeric_metadata_u64_to_i32(value: Option<u64>) -> Option<i32> {
|
||||
value.and_then(|v| i32::try_from(v).ok())
|
||||
}
|
||||
|
||||
fn numeric_metadata_i64_to_i32(value: Option<i64>) -> Option<i32> {
|
||||
value.and_then(|v| i32::try_from(v).ok())
|
||||
}
|
||||
|
||||
fn numeric_metadata_str_to_i32(value: Option<String>) -> Option<i32> {
|
||||
value.and_then(|v| v.parse::<i64>().ok())
|
||||
.and_then(|v| i32::try_from(v).ok())
|
||||
}
|
||||
|
||||
fn get_opt_i32(row: &MySqlRow, name: &str) -> Option<i32> {
|
||||
if row.try_get_raw(name).map(|v| v.is_null()).unwrap_or(true) {
|
||||
return None;
|
||||
}
|
||||
|
||||
row.try_get::<Option<i32>, _>(name)
|
||||
.ok()
|
||||
.flatten()
|
||||
.or_else(|| numeric_metadata_i64_to_i32(row.try_get::<Option<i64>, _>(name).ok().flatten()))
|
||||
.or_else(|| numeric_metadata_u64_to_i32(row.try_get::<Option<u64>, _>(name).ok().flatten()))
|
||||
.or_else(|| numeric_metadata_str_to_i32(row.try_get::<Option<String>, _>(name).ok().flatten()))
|
||||
.or_else(|| {
|
||||
row.try_get::<Option<Vec<u8>>, _>(name)
|
||||
.ok()
|
||||
.flatten()
|
||||
.and_then(|b| String::from_utf8(b).ok())
|
||||
.and_then(|v| numeric_metadata_str_to_i32(Some(v)))
|
||||
})
|
||||
}
|
||||
|
||||
fn mysql_temporal_to_json_value(row: &MySqlRow, idx: usize) -> Option<serde_json::Value> {
|
||||
if let Ok(v) = row.try_get::<NaiveDateTime, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_string()));
|
||||
}
|
||||
if let Ok(v) = row.try_get::<DateTime<Utc>, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_rfc3339()));
|
||||
}
|
||||
if let Ok(v) = row.try_get::<NaiveDate, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_string()));
|
||||
}
|
||||
if let Ok(v) = row.try_get::<NaiveTime, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_string()));
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn mysql_value_to_json(row: &MySqlRow, idx: usize, type_name: &str) -> serde_json::Value {
|
||||
if row.try_get_raw(idx).map(|v| v.is_null()).unwrap_or(true) {
|
||||
return serde_json::Value::Null;
|
||||
}
|
||||
|
||||
let upper_type = type_name.to_uppercase();
|
||||
|
||||
if upper_type == "JSON" {
|
||||
if let Ok(v) = row.try_get::<serde_json::Value, _>(idx) {
|
||||
return v;
|
||||
}
|
||||
if let Ok(v) = row.try_get::<String, _>(idx) {
|
||||
return serde_json::from_str::<serde_json::Value>(&v).unwrap_or(serde_json::Value::String(v));
|
||||
}
|
||||
return serde_json::Value::Null;
|
||||
}
|
||||
|
||||
if upper_type == "BOOLEAN" {
|
||||
return row
|
||||
.try_get::<bool, _>(idx)
|
||||
.map(serde_json::Value::Bool)
|
||||
.unwrap_or(serde_json::Value::Null);
|
||||
}
|
||||
|
||||
if upper_type.contains("BIGINT") {
|
||||
return row
|
||||
.try_get::<i64, _>(idx)
|
||||
.map(|v| serde_json::Value::String(v.to_string()))
|
||||
.or_else(|_| {
|
||||
row.try_get::<u64, _>(idx)
|
||||
.map(|v| serde_json::Value::String(v.to_string()))
|
||||
})
|
||||
.unwrap_or(serde_json::Value::Null);
|
||||
}
|
||||
|
||||
if upper_type == "DECIMAL" {
|
||||
return row
|
||||
.try_get::<Decimal, _>(idx)
|
||||
.map(|v: Decimal| serde_json::Value::String(v.to_string()))
|
||||
.unwrap_or(serde_json::Value::Null);
|
||||
}
|
||||
|
||||
if upper_type.starts_with("DATETIME")
|
||||
|| upper_type.starts_with("TIMESTAMP")
|
||||
|| upper_type == "DATE"
|
||||
|| upper_type == "TIME"
|
||||
|| upper_type.starts_with("TIME(")
|
||||
{
|
||||
if let Some(v) = mysql_temporal_to_json_value(row, idx) {
|
||||
return v;
|
||||
}
|
||||
}
|
||||
|
||||
row.try_get::<String, _>(idx)
|
||||
.map(serde_json::Value::String)
|
||||
.or_else(|_| row.try_get::<i64, _>(idx).map(|v| serde_json::Value::Number(v.into())))
|
||||
.or_else(|_| row.try_get::<u64, _>(idx).map(|v| serde_json::Value::Number(v.into())))
|
||||
.or_else(|_| row.try_get::<f64, _>(idx).map(|v| {
|
||||
serde_json::Number::from_f64(v)
|
||||
.map(serde_json::Value::Number)
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
}))
|
||||
.or_else(|_| row.try_get::<bool, _>(idx).map(serde_json::Value::Bool))
|
||||
.or_else(|_| {
|
||||
row.try_get::<Vec<u8>, _>(idx)
|
||||
.map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string()))
|
||||
})
|
||||
.or_else(|e| mysql_temporal_to_json_value(row, idx).ok_or(e))
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
}
|
||||
|
||||
pub async fn connect(url: &str) -> Result<MySqlPool, String> {
|
||||
MySqlPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.acquire_timeout(Duration::from_secs(10))
|
||||
.idle_timeout(Duration::from_secs(300))
|
||||
.connect(url)
|
||||
.await
|
||||
.map_err(|e| format!("MySQL connection failed: {e}"))
|
||||
}
|
||||
|
||||
pub async fn connect_bare(url: &str) -> Result<MySqlPool, String> {
|
||||
let options: sqlx::mysql::MySqlConnectOptions = url.parse()
|
||||
.map_err(|e: sqlx::Error| format!("Invalid MySQL URL: {e}"))?;
|
||||
let options = options
|
||||
.no_engine_substitution(false)
|
||||
.set_names(false)
|
||||
.pipes_as_concat(false)
|
||||
.timezone(None);
|
||||
MySqlPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.acquire_timeout(Duration::from_secs(10))
|
||||
.idle_timeout(Duration::from_secs(300))
|
||||
.connect_with(options)
|
||||
.await
|
||||
.map_err(|e| format!("MySQL connection failed: {e}"))
|
||||
}
|
||||
|
||||
pub async fn list_databases(pool: &MySqlPool) -> Result<Vec<DatabaseInfo>, String> {
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql("SELECT SCHEMA_NAME FROM information_schema.SCHEMATA ORDER BY SCHEMA_NAME")
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows.iter().map(|row| DatabaseInfo { name: get_str(row, 0) }).collect())
|
||||
}
|
||||
|
||||
pub async fn list_tables(pool: &MySqlPool, database: &str) -> Result<Vec<TableInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = {} ORDER BY TABLE_NAME",
|
||||
quote_value(database),
|
||||
);
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| TableInfo {
|
||||
name: get_str_by_name(row, "TABLE_NAME"),
|
||||
table_type: get_str_by_name(row, "TABLE_TYPE"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(
|
||||
pool: &MySqlPool,
|
||||
database: &str,
|
||||
table: &str,
|
||||
) -> Result<Vec<ColumnInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT c.COLUMN_NAME, c.COLUMN_TYPE, c.IS_NULLABLE, c.COLUMN_DEFAULT, c.EXTRA, c.COLUMN_COMMENT, \
|
||||
CASE WHEN kcu.COLUMN_NAME IS NOT NULL THEN 1 ELSE 0 END AS IS_PK, \
|
||||
c.NUMERIC_PRECISION, c.NUMERIC_SCALE, c.CHARACTER_MAXIMUM_LENGTH \
|
||||
FROM information_schema.COLUMNS c \
|
||||
LEFT JOIN information_schema.KEY_COLUMN_USAGE kcu \
|
||||
ON c.TABLE_SCHEMA = kcu.TABLE_SCHEMA \
|
||||
AND c.TABLE_NAME = kcu.TABLE_NAME \
|
||||
AND c.COLUMN_NAME = kcu.COLUMN_NAME \
|
||||
AND kcu.CONSTRAINT_NAME = 'PRIMARY' \
|
||||
WHERE c.TABLE_SCHEMA = {} AND c.TABLE_NAME = {} \
|
||||
ORDER BY c.ORDINAL_POSITION",
|
||||
quote_value(database),
|
||||
quote_value(table),
|
||||
);
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| ColumnInfo {
|
||||
name: get_str_by_name(row, "COLUMN_NAME"),
|
||||
data_type: get_str_by_name(row, "COLUMN_TYPE"),
|
||||
is_nullable: get_str_by_name(row, "IS_NULLABLE") == "YES",
|
||||
column_default: get_opt_str(row, "COLUMN_DEFAULT"),
|
||||
is_primary_key: row.get::<i32, _>("IS_PK") == 1,
|
||||
extra: get_opt_str(row, "EXTRA"),
|
||||
comment: get_opt_str(row, "COLUMN_COMMENT").filter(|s| !s.is_empty()),
|
||||
numeric_precision: get_opt_i32(row, "NUMERIC_PRECISION"),
|
||||
numeric_scale: get_opt_i32(row, "NUMERIC_SCALE"),
|
||||
character_maximum_length: get_opt_i32(row, "CHARACTER_MAXIMUM_LENGTH"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn execute_query(pool: &MySqlPool, sql: &str, bare: bool) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
let trimmed = sql.trim().to_uppercase();
|
||||
|
||||
if trimmed.starts_with("SELECT") || trimmed.starts_with("SHOW") || trimmed.starts_with("DESCRIBE") || trimmed.starts_with("EXPLAIN") {
|
||||
if bare {
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql(sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let (columns, column_types) = if let Some(first) = rows.first() {
|
||||
let cols: Vec<String> = first.columns().iter().map(|c| c.name().to_string()).collect();
|
||||
let types: Vec<String> = first.columns().iter().map(|c| c.type_info().name().to_string()).collect();
|
||||
(cols, types)
|
||||
} else {
|
||||
(vec![], vec![])
|
||||
};
|
||||
|
||||
let result_rows: Vec<Vec<serde_json::Value>> = rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
(0..row.len())
|
||||
.map(|i| mysql_value_to_json(row, i, column_types.get(i).map(String::as_str).unwrap_or("")))
|
||||
.collect()
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(QueryResult {
|
||||
columns,
|
||||
rows: result_rows,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
} else {
|
||||
let desc = pool.describe(sql).await.map_err(|e| e.to_string())?;
|
||||
let columns: Vec<String> = desc.columns().iter().map(|c| c.name().to_string()).collect();
|
||||
let column_types: Vec<String> = desc.columns().iter().map(|c| c.type_info().name().to_string()).collect();
|
||||
|
||||
let rows: Vec<MySqlRow> = sqlx::query(sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let result_rows: Vec<Vec<serde_json::Value>> = rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
(0..row.len())
|
||||
.map(|i| mysql_value_to_json(row, i, column_types.get(i).map(String::as_str).unwrap_or("")))
|
||||
.collect()
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(QueryResult {
|
||||
columns,
|
||||
rows: result_rows,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
} else {
|
||||
let result = sqlx::raw_sql(sql)
|
||||
.execute(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows: result.rows_affected(),
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT INDEX_NAME, GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX) AS columns, \
|
||||
MIN(NON_UNIQUE) = 0 AS is_unique, INDEX_NAME = 'PRIMARY' AS is_primary, \
|
||||
INDEX_TYPE \
|
||||
FROM information_schema.STATISTICS \
|
||||
WHERE TABLE_SCHEMA = {} AND TABLE_NAME = {} \
|
||||
GROUP BY INDEX_NAME, INDEX_TYPE \
|
||||
ORDER BY INDEX_NAME",
|
||||
quote_value(database),
|
||||
quote_value(table),
|
||||
);
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
let cols_str = get_str_by_name(row, "columns");
|
||||
IndexInfo {
|
||||
name: get_str_by_name(row, "INDEX_NAME"),
|
||||
columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(),
|
||||
is_unique: row.get::<bool, _>("is_unique"),
|
||||
is_primary: row.get::<bool, _>("is_primary"),
|
||||
filter: None,
|
||||
index_type: Some(get_str_by_name(row, "INDEX_TYPE")),
|
||||
included_columns: None,
|
||||
comment: None,
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_foreign_keys(pool: &MySqlPool, database: &str, table: &str) -> Result<Vec<ForeignKeyInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT kcu.CONSTRAINT_NAME, kcu.COLUMN_NAME, \
|
||||
kcu.REFERENCED_TABLE_NAME, kcu.REFERENCED_COLUMN_NAME \
|
||||
FROM information_schema.KEY_COLUMN_USAGE kcu \
|
||||
WHERE kcu.TABLE_SCHEMA = {} AND kcu.TABLE_NAME = {} \
|
||||
AND kcu.REFERENCED_TABLE_NAME IS NOT NULL \
|
||||
ORDER BY kcu.CONSTRAINT_NAME",
|
||||
quote_value(database),
|
||||
quote_value(table),
|
||||
);
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| ForeignKeyInfo {
|
||||
name: get_str_by_name(row, "CONSTRAINT_NAME"),
|
||||
column: get_str_by_name(row, "COLUMN_NAME"),
|
||||
ref_table: get_str_by_name(row, "REFERENCED_TABLE_NAME"),
|
||||
ref_column: get_str_by_name(row, "REFERENCED_COLUMN_NAME"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_triggers(pool: &MySqlPool, database: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT TRIGGER_NAME, EVENT_MANIPULATION, ACTION_TIMING \
|
||||
FROM information_schema.TRIGGERS \
|
||||
WHERE TRIGGER_SCHEMA = {} AND EVENT_OBJECT_TABLE = {} \
|
||||
ORDER BY TRIGGER_NAME",
|
||||
quote_value(database),
|
||||
quote_value(table),
|
||||
);
|
||||
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| TriggerInfo {
|
||||
name: get_str_by_name(row, "TRIGGER_NAME"),
|
||||
event: get_str_by_name(row, "EVENT_MANIPULATION"),
|
||||
timing: get_str_by_name(row, "ACTION_TIMING"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn numeric_metadata_accepts_unsigned_information_schema_values() {
|
||||
assert_eq!(numeric_metadata_u64_to_i32(Some(65)), Some(65));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn numeric_metadata_ignores_values_outside_frontend_range() {
|
||||
assert_eq!(numeric_metadata_u64_to_i32(Some(i32::MAX as u64 + 1)), None);
|
||||
assert_eq!(numeric_metadata_u64_to_i32(None), None);
|
||||
}
|
||||
}
|
||||
@@ -1,254 +0,0 @@
|
||||
use oracle_rs::{Config, Connection};
|
||||
use std::time::Instant;
|
||||
|
||||
use super::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
|
||||
|
||||
pub type OracleClient = Connection;
|
||||
|
||||
pub async fn connect(host: &str, port: u16, service: &str, user: &str, pass: &str) -> Result<OracleClient, String> {
|
||||
let config = Config::new(host, port, service, user, pass);
|
||||
Connection::connect_with_config(config)
|
||||
.await
|
||||
.map_err(|e| format!("Oracle connection failed: {e}"))
|
||||
}
|
||||
|
||||
fn value_to_json(val: &oracle_rs::Value) -> serde_json::Value {
|
||||
match val {
|
||||
oracle_rs::Value::Null => serde_json::Value::Null,
|
||||
oracle_rs::Value::String(s) => serde_json::Value::String(s.clone()),
|
||||
oracle_rs::Value::Integer(n) => serde_json::Value::Number((*n).into()),
|
||||
oracle_rs::Value::Float(f) => serde_json::Number::from_f64(*f)
|
||||
.map(serde_json::Value::Number)
|
||||
.unwrap_or(serde_json::Value::Null),
|
||||
oracle_rs::Value::Boolean(b) => serde_json::Value::Bool(*b),
|
||||
oracle_rs::Value::Json(v) => v.clone(),
|
||||
_ => serde_json::Value::String(format!("{val:?}")),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn list_databases(conn: &OracleClient) -> Result<Vec<DatabaseInfo>, String> {
|
||||
let result = conn.query(
|
||||
"SELECT username FROM all_users ORDER BY username",
|
||||
&[],
|
||||
).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.rows.iter().map(|row| {
|
||||
DatabaseInfo { name: row.get_string(0).unwrap_or("").to_string() }
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_schemas(conn: &OracleClient) -> Result<Vec<String>, String> {
|
||||
let result = conn.query(
|
||||
"SELECT username FROM all_users ORDER BY username",
|
||||
&[],
|
||||
).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.rows.iter().map(|row| {
|
||||
row.get_string(0).unwrap_or("").to_string()
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_tables(conn: &OracleClient, schema: &str) -> Result<Vec<TableInfo>, String> {
|
||||
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",
|
||||
s = schema.replace('\'', "''")
|
||||
);
|
||||
let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.rows.iter().map(|row| {
|
||||
TableInfo {
|
||||
name: row.get_string(0).unwrap_or("").to_string(),
|
||||
table_type: row.get_string(1).unwrap_or("TABLE").to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(conn: &OracleClient, schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
|
||||
let s = schema.replace('\'', "''");
|
||||
let t = table.replace('\'', "''");
|
||||
|
||||
let pk_result = conn.query(
|
||||
&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}'"
|
||||
),
|
||||
&[],
|
||||
).await.map_err(|e| e.to_string())?;
|
||||
let pk_names: std::collections::HashSet<String> = pk_result.rows.iter()
|
||||
.filter_map(|row| row.get_string(0).map(|s| s.to_string()))
|
||||
.collect();
|
||||
|
||||
let col_result = conn.query(
|
||||
&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"
|
||||
),
|
||||
&[],
|
||||
).await.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(col_result.rows.iter().map(|row| {
|
||||
let name = row.get_string(0).unwrap_or("").to_string();
|
||||
let base = row.get_string(1).unwrap_or("").to_string();
|
||||
let data_len = row.get_i64(5).map(|v| v as i32);
|
||||
let char_len = row.get_i64(6).map(|v| v as i32);
|
||||
let num_prec = row.get_i64(3).map(|v| v as i32);
|
||||
let num_scale = row.get_i64(4).map(|v| v as i32);
|
||||
let data_type = match base.to_uppercase().as_str() {
|
||||
"VARCHAR2" | "NVARCHAR2" | "CHAR" | "NCHAR" => {
|
||||
let len = char_len.or(data_len);
|
||||
match len {
|
||||
Some(n) => format!("{base}({n})"),
|
||||
None => base,
|
||||
}
|
||||
}
|
||||
"NUMBER" => match (num_prec, num_scale) {
|
||||
(Some(p), Some(s)) if s > 0 => format!("NUMBER({p},{s})"),
|
||||
(Some(p), _) if p > 0 => format!("NUMBER({p})"),
|
||||
_ => "NUMBER".to_string(),
|
||||
},
|
||||
"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: row.get_string(2).unwrap_or("N") == "Y",
|
||||
column_default: None,
|
||||
extra: None, comment: None,
|
||||
numeric_precision: num_prec,
|
||||
numeric_scale: num_scale,
|
||||
character_maximum_length: char_len,
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_indexes(conn: &OracleClient, schema: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
|
||||
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",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.rows.iter().map(|row| {
|
||||
let cols_str = row.get_string(1).unwrap_or("");
|
||||
IndexInfo {
|
||||
name: row.get_string(0).unwrap_or("").to_string(),
|
||||
columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(),
|
||||
is_unique: row.get_string(2).unwrap_or("") == "UNIQUE",
|
||||
is_primary: row.get_i64(3).unwrap_or(0) == 1,
|
||||
filter: None,
|
||||
index_type: row.get_string(4).map(|s| s.to_string()),
|
||||
included_columns: None,
|
||||
comment: None,
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_foreign_keys(conn: &OracleClient, schema: &str, table: &str) -> Result<Vec<ForeignKeyInfo>, String> {
|
||||
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",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.rows.iter().map(|row| {
|
||||
ForeignKeyInfo {
|
||||
name: row.get_string(0).unwrap_or("").to_string(),
|
||||
column: row.get_string(1).unwrap_or("").to_string(),
|
||||
ref_table: row.get_string(2).unwrap_or("").to_string(),
|
||||
ref_column: row.get_string(3).unwrap_or("").to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_triggers(conn: &OracleClient, schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT TRIGGER_NAME, TRIGGERING_EVENT, TRIGGER_TYPE \
|
||||
FROM ALL_TRIGGERS \
|
||||
WHERE OWNER = '{s}' AND TABLE_NAME = '{t}' \
|
||||
ORDER BY TRIGGER_NAME",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let result = conn.query(&sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.rows.iter().map(|row| {
|
||||
TriggerInfo {
|
||||
name: row.get_string(0).unwrap_or("").to_string(),
|
||||
event: row.get_string(1).unwrap_or("").to_string(),
|
||||
timing: row.get_string(2).unwrap_or("").to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn execute_query(conn: &OracleClient, sql: &str) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
let sql = sql.trim().trim_end_matches(';');
|
||||
let trimmed = sql.to_uppercase();
|
||||
|
||||
if trimmed.starts_with("SELECT")
|
||||
|| trimmed.starts_with("WITH")
|
||||
|| trimmed.starts_with("SHOW")
|
||||
|| trimmed.starts_with("DESCRIBE")
|
||||
|| trimmed.starts_with("EXPLAIN")
|
||||
{
|
||||
let result = conn.query(sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let columns: Vec<String> = result.columns.iter().map(|c| c.name.clone()).collect();
|
||||
let rows: Vec<Vec<serde_json::Value>> = result.rows.iter().map(|row| {
|
||||
(0..columns.len()).map(|i| {
|
||||
row.get(i)
|
||||
.map(|v| value_to_json(v))
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
}).collect()
|
||||
}).collect();
|
||||
|
||||
Ok(QueryResult {
|
||||
columns,
|
||||
rows,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
} else {
|
||||
match conn.execute(sql, &[]).await {
|
||||
Ok(result) => {
|
||||
let _ = conn.commit().await;
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows: result.rows_affected,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
Err(e) => {
|
||||
let msg = e.to_string();
|
||||
if msg.contains("Server rejected") || msg.contains("closed the connection") {
|
||||
Err("Operation failed (connection closed) — possibly a constraint violation (foreign key, unique, or check constraint).".to_string())
|
||||
} else {
|
||||
Err(msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,376 +0,0 @@
|
||||
use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
|
||||
use rust_decimal::Decimal;
|
||||
use sqlx::postgres::{PgPool, PgPoolOptions, PgRow};
|
||||
use sqlx::{Column, Executor, Row, TypeInfo, ValueRef};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use super::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
|
||||
|
||||
fn pg_temporal_to_json_value(row: &PgRow, idx: usize) -> Option<serde_json::Value> {
|
||||
if let Ok(v) = row.try_get::<DateTime<Utc>, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_rfc3339()));
|
||||
}
|
||||
if let Ok(v) = row.try_get::<NaiveDateTime, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_string()));
|
||||
}
|
||||
if let Ok(v) = row.try_get::<NaiveDate, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_string()));
|
||||
}
|
||||
if let Ok(v) = row.try_get::<NaiveTime, _>(idx) {
|
||||
return Some(serde_json::Value::String(v.to_string()));
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn pg_value_to_json(row: &PgRow, idx: usize, type_name: &str) -> serde_json::Value {
|
||||
if row.try_get_raw(idx).map(|v| v.is_null()).unwrap_or(true) {
|
||||
return serde_json::Value::Null;
|
||||
}
|
||||
|
||||
let upper = type_name.to_uppercase();
|
||||
|
||||
if upper == "JSON" || upper == "JSONB" {
|
||||
if let Ok(v) = row.try_get::<serde_json::Value, _>(idx) {
|
||||
return v;
|
||||
}
|
||||
if let Ok(v) = row.try_get::<String, _>(idx) {
|
||||
return serde_json::from_str::<serde_json::Value>(&v).unwrap_or(serde_json::Value::String(v));
|
||||
}
|
||||
return serde_json::Value::Null;
|
||||
}
|
||||
|
||||
if upper == "BOOL" {
|
||||
return row
|
||||
.try_get::<bool, _>(idx)
|
||||
.map(serde_json::Value::Bool)
|
||||
.unwrap_or(serde_json::Value::Null);
|
||||
}
|
||||
|
||||
if upper.contains("TIMESTAMP")
|
||||
|| upper == "DATE"
|
||||
|| upper == "TIME"
|
||||
|| upper == "TIMETZ"
|
||||
|| upper.contains("INTERVAL")
|
||||
{
|
||||
if let Some(v) = pg_temporal_to_json_value(row, idx) {
|
||||
return v;
|
||||
}
|
||||
}
|
||||
|
||||
if upper == "NUMERIC" || upper == "DECIMAL" || upper == "MONEY" {
|
||||
return row
|
||||
.try_get::<Decimal, _>(idx)
|
||||
.map(|v: Decimal| serde_json::Value::String(v.to_string()))
|
||||
.unwrap_or(serde_json::Value::Null);
|
||||
}
|
||||
|
||||
row.try_get::<String, _>(idx)
|
||||
.map(serde_json::Value::String)
|
||||
.or_else(|_| {
|
||||
row.try_get::<i64, _>(idx)
|
||||
.map(|v| serde_json::Value::Number(v.into()))
|
||||
})
|
||||
.or_else(|_| {
|
||||
row.try_get::<i32, _>(idx)
|
||||
.map(|v| serde_json::Value::Number(v.into()))
|
||||
})
|
||||
.or_else(|_| {
|
||||
row.try_get::<f64, _>(idx).map(|v| {
|
||||
serde_json::Number::from_f64(v)
|
||||
.map(serde_json::Value::Number)
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
})
|
||||
})
|
||||
.or_else(|_| row.try_get::<bool, _>(idx).map(serde_json::Value::Bool))
|
||||
.or_else(|e| pg_temporal_to_json_value(row, idx).ok_or(e))
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
}
|
||||
|
||||
pub async fn connect(url: &str) -> Result<PgPool, String> {
|
||||
PgPoolOptions::new()
|
||||
.max_connections(5)
|
||||
.acquire_timeout(Duration::from_secs(10))
|
||||
.idle_timeout(Duration::from_secs(300))
|
||||
.connect(url)
|
||||
.await
|
||||
.map_err(|e| format!("PostgreSQL connection failed: {e}"))
|
||||
}
|
||||
|
||||
pub async fn list_databases(pool: &PgPool) -> Result<Vec<DatabaseInfo>, String> {
|
||||
let rows: Vec<PgRow> =
|
||||
sqlx::query("SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname")
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| DatabaseInfo {
|
||||
name: row.get::<String, _>("datname"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_tables(pool: &PgPool, schema: &str) -> Result<Vec<TableInfo>, String> {
|
||||
let rows: Vec<PgRow> = sqlx::query(
|
||||
"SELECT table_name, table_type FROM information_schema.tables \
|
||||
WHERE table_schema = $1 ORDER BY table_name",
|
||||
)
|
||||
.bind(schema)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| TableInfo {
|
||||
name: row.get::<String, _>("table_name"),
|
||||
table_type: row.get::<String, _>("table_type"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_schemas(pool: &PgPool) -> Result<Vec<String>, String> {
|
||||
let rows: Vec<PgRow> = sqlx::query(
|
||||
"SELECT schema_name FROM information_schema.schemata \
|
||||
WHERE schema_name NOT IN ('information_schema', 'pg_catalog', 'pg_toast') \
|
||||
AND schema_name NOT LIKE 'pg_toast_temp_%' \
|
||||
AND schema_name NOT LIKE 'pg_temp_%' \
|
||||
ORDER BY schema_name",
|
||||
)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| row.get::<String, _>("schema_name"))
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(
|
||||
pool: &PgPool,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
) -> Result<Vec<ColumnInfo>, String> {
|
||||
let rows: Vec<PgRow> = sqlx::query(
|
||||
"SELECT a.attname AS column_name, \
|
||||
format_type(a.atttypid, a.atttypmod) AS full_type, \
|
||||
NOT a.attnotnull AS is_nullable, \
|
||||
pg_get_expr(ad.adbin, ad.adrelid) AS column_default, \
|
||||
EXISTS ( \
|
||||
SELECT 1 FROM pg_constraint co \
|
||||
JOIN pg_index i ON i.indrelid = co.conrelid AND co.conindid = i.indexrelid \
|
||||
WHERE co.conrelid = a.attrelid AND co.contype = 'p' \
|
||||
AND a.attnum = ANY(i.indkey) \
|
||||
) AS is_pk, \
|
||||
col_description(a.attrelid, a.attnum) AS column_comment, \
|
||||
CASE WHEN t.typname = 'numeric' AND a.atttypmod > 0 \
|
||||
THEN ((a.atttypmod - 4) >> 16) & 65535 ELSE NULL END AS numeric_precision, \
|
||||
CASE WHEN t.typname = 'numeric' AND a.atttypmod > 0 \
|
||||
THEN (a.atttypmod - 4) & 65535 ELSE NULL END AS numeric_scale, \
|
||||
CASE WHEN t.typname IN ('varchar', 'bpchar') AND a.atttypmod > 0 \
|
||||
THEN a.atttypmod - 4 ELSE NULL END AS character_maximum_length \
|
||||
FROM pg_attribute a \
|
||||
JOIN pg_type t ON t.oid = a.atttypid \
|
||||
LEFT JOIN pg_attrdef ad ON ad.adrelid = a.attrelid AND ad.adnum = a.attnum \
|
||||
WHERE a.attrelid = ($1 || '.' || $2)::regclass \
|
||||
AND a.attnum > 0 AND NOT a.attisdropped \
|
||||
ORDER BY a.attnum",
|
||||
)
|
||||
.bind(schema)
|
||||
.bind(table)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
let full_type = row.get::<Option<String>, _>("full_type").unwrap_or_default();
|
||||
ColumnInfo {
|
||||
name: row.get::<String, _>("column_name"),
|
||||
data_type: full_type,
|
||||
is_nullable: row.get::<bool, _>("is_nullable"),
|
||||
column_default: row.get::<Option<String>, _>("column_default"),
|
||||
is_primary_key: row.get::<bool, _>("is_pk"),
|
||||
extra: None,
|
||||
comment: row.get::<Option<String>, _>("column_comment"),
|
||||
numeric_precision: row.get::<Option<i32>, _>("numeric_precision"),
|
||||
numeric_scale: row.get::<Option<i32>, _>("numeric_scale"),
|
||||
character_maximum_length: row.get::<Option<i32>, _>("character_maximum_length"),
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn execute_query(pool: &PgPool, sql: &str) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
let trimmed = sql.trim().to_uppercase();
|
||||
|
||||
if trimmed.starts_with("SELECT")
|
||||
|| trimmed.starts_with("SHOW")
|
||||
|| trimmed.starts_with("EXPLAIN")
|
||||
|| trimmed.starts_with("WITH")
|
||||
|| trimmed.starts_with("TABLE")
|
||||
{
|
||||
let rows: Vec<PgRow> = sqlx::query(sql)
|
||||
.persistent(false)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let (columns, column_types): (Vec<String>, Vec<String>) = if let Some(first) = rows.first() {
|
||||
let cols = first.columns();
|
||||
(
|
||||
cols.iter().map(|c| c.name().to_string()).collect(),
|
||||
cols.iter().map(|c| c.type_info().name().to_string()).collect(),
|
||||
)
|
||||
} else {
|
||||
let desc = pool.describe(sql).await.map_err(|e| e.to_string())?;
|
||||
(
|
||||
desc.columns().iter().map(|c| c.name().to_string()).collect(),
|
||||
desc.columns().iter().map(|c| c.type_info().name().to_string()).collect(),
|
||||
)
|
||||
};
|
||||
|
||||
let result_rows: Vec<Vec<serde_json::Value>> = rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
(0..row.len())
|
||||
.map(|i| {
|
||||
pg_value_to_json(
|
||||
row,
|
||||
i,
|
||||
column_types.get(i).map(String::as_str).unwrap_or(""),
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(QueryResult {
|
||||
columns,
|
||||
rows: result_rows,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
} else {
|
||||
let result = sqlx::query(sql)
|
||||
.execute(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows: result.rows_affected(),
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn list_indexes(pool: &PgPool, schema: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
|
||||
let rows: Vec<PgRow> = sqlx::query(
|
||||
"SELECT i.relname AS index_name, \
|
||||
array_agg(COALESCE(a.attname, pg_get_indexdef(ix.indexrelid, k.n::int, true)) ORDER BY k.n) AS columns, \
|
||||
ix.indisunique AS is_unique, \
|
||||
ix.indisprimary AS is_primary, \
|
||||
pg_get_expr(ix.indpred, ix.indrelid) AS filter_expr, \
|
||||
am.amname AS index_type, \
|
||||
ix.indnkeyatts AS nkeyatts, \
|
||||
ix.indkey AS indkey, \
|
||||
obj_description(i.oid, 'pg_class') AS index_comment \
|
||||
FROM pg_index ix \
|
||||
JOIN pg_class t ON t.oid = ix.indrelid \
|
||||
JOIN pg_class i ON i.oid = ix.indexrelid \
|
||||
JOIN pg_namespace n ON n.oid = t.relnamespace \
|
||||
JOIN pg_am am ON am.oid = i.relam \
|
||||
JOIN LATERAL unnest(ix.indkey) WITH ORDINALITY AS k(attnum, n) ON true \
|
||||
LEFT JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = k.attnum AND k.attnum > 0 \
|
||||
WHERE n.nspname = $1 AND t.relname = $2 \
|
||||
GROUP BY i.relname, i.oid, ix.indisunique, ix.indisprimary, ix.indpred, ix.indrelid, am.amname, ix.indnkeyatts, ix.indkey \
|
||||
ORDER BY i.relname",
|
||||
)
|
||||
.bind(schema)
|
||||
.bind(table)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
let all_cols: Vec<String> = row.get::<Vec<String>, _>("columns");
|
||||
let nkeyatts = row.get::<Option<i16>, _>("nkeyatts").unwrap_or(all_cols.len() as i16) as usize;
|
||||
let key_cols = all_cols[..nkeyatts].to_vec();
|
||||
let included = if nkeyatts < all_cols.len() { all_cols[nkeyatts..].to_vec() } else { vec![] };
|
||||
IndexInfo {
|
||||
name: row.get::<String, _>("index_name"),
|
||||
columns: key_cols,
|
||||
is_unique: row.get::<bool, _>("is_unique"),
|
||||
is_primary: row.get::<bool, _>("is_primary"),
|
||||
filter: row.get::<Option<String>, _>("filter_expr"),
|
||||
index_type: row.get::<Option<String>, _>("index_type"),
|
||||
included_columns: if included.is_empty() { None } else { Some(included) },
|
||||
comment: row.get::<Option<String>, _>("index_comment"),
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_foreign_keys(pool: &PgPool, schema: &str, table: &str) -> Result<Vec<ForeignKeyInfo>, String> {
|
||||
let rows: Vec<PgRow> = sqlx::query(
|
||||
"SELECT kcu.constraint_name, kcu.column_name, \
|
||||
ccu.table_name AS ref_table, ccu.column_name AS ref_column \
|
||||
FROM information_schema.key_column_usage kcu \
|
||||
JOIN information_schema.referential_constraints rc \
|
||||
ON kcu.constraint_name = rc.constraint_name \
|
||||
AND kcu.constraint_schema = rc.constraint_schema \
|
||||
JOIN information_schema.constraint_column_usage ccu \
|
||||
ON rc.unique_constraint_name = ccu.constraint_name \
|
||||
AND rc.unique_constraint_schema = ccu.constraint_schema \
|
||||
WHERE kcu.table_schema = $1 AND kcu.table_name = $2 \
|
||||
ORDER BY kcu.constraint_name",
|
||||
)
|
||||
.bind(schema)
|
||||
.bind(table)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| ForeignKeyInfo {
|
||||
name: row.get::<String, _>("constraint_name"),
|
||||
column: row.get::<String, _>("column_name"),
|
||||
ref_table: row.get::<String, _>("ref_table"),
|
||||
ref_column: row.get::<String, _>("ref_column"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_triggers(pool: &PgPool, schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
|
||||
let rows: Vec<PgRow> = sqlx::query(
|
||||
"SELECT trigger_name, event_manipulation, action_timing \
|
||||
FROM information_schema.triggers \
|
||||
WHERE trigger_schema = $1 AND event_object_table = $2 \
|
||||
ORDER BY trigger_name",
|
||||
)
|
||||
.bind(schema)
|
||||
.bind(table)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| TriggerInfo {
|
||||
name: row.get::<String, _>("trigger_name"),
|
||||
event: row.get::<String, _>("event_manipulation"),
|
||||
timing: row.get::<String, _>("action_timing"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
@@ -1,443 +0,0 @@
|
||||
use redis::{AsyncCommands, FromRedisValue, Value as RedisRawValue};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
const STREAM_ENTRY_LIMIT: usize = 100;
|
||||
const DEFAULT_REDIS_DATABASES: u32 = 16;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RedisKeyInfo {
|
||||
pub key: String,
|
||||
pub key_type: String,
|
||||
pub ttl: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RedisScanResult {
|
||||
pub cursor: u64,
|
||||
pub keys: Vec<RedisKeyInfo>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RedisValue {
|
||||
pub key: String,
|
||||
pub key_type: String,
|
||||
pub ttl: i64,
|
||||
pub value: serde_json::Value,
|
||||
}
|
||||
|
||||
pub async fn connect(url: &str) -> Result<redis::aio::MultiplexedConnection, String> {
|
||||
let client = redis::Client::open(url).map_err(|e| format!("Redis connection failed: {e}"))?;
|
||||
let mut con = tokio::time::timeout(
|
||||
std::time::Duration::from_secs(10),
|
||||
client.get_multiplexed_async_connection(),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| "Redis connection timed out (10s)".to_string())?
|
||||
.map_err(|e| format!("Redis connection failed: {e}"))?;
|
||||
|
||||
redis::cmd("PING")
|
||||
.query_async::<String>(&mut con)
|
||||
.await
|
||||
.map_err(|e| format!("Redis authentication failed or command rejected: {e}"))?;
|
||||
|
||||
Ok(con)
|
||||
}
|
||||
|
||||
pub async fn list_databases(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
) -> Result<Vec<u32>, String> {
|
||||
let configured_count = redis::cmd("CONFIG")
|
||||
.arg("GET")
|
||||
.arg("databases")
|
||||
.query_async(con)
|
||||
.await
|
||||
.ok()
|
||||
.and_then(parse_database_count);
|
||||
|
||||
let keyspace_dbs = list_keyspace_databases(con).await.unwrap_or_default();
|
||||
let database_count = configured_count.unwrap_or(DEFAULT_REDIS_DATABASES);
|
||||
let max_db = keyspace_dbs
|
||||
.iter()
|
||||
.copied()
|
||||
.max()
|
||||
.map(|db| db + 1)
|
||||
.unwrap_or(0);
|
||||
let visible_count = database_count.max(max_db).max(1);
|
||||
|
||||
Ok((0..visible_count).collect())
|
||||
}
|
||||
|
||||
fn parse_database_count(value: redis::Value) -> Option<u32> {
|
||||
let values = match value {
|
||||
redis::Value::Array(values) => values,
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
values.windows(2).find_map(|pair| {
|
||||
let key = String::from_redis_value(&pair[0]).ok()?;
|
||||
if key.eq_ignore_ascii_case("databases") {
|
||||
String::from_redis_value(&pair[1]).ok()?.parse().ok()
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async fn list_keyspace_databases(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
) -> Result<Vec<u32>, String> {
|
||||
let info: String = redis::cmd("INFO")
|
||||
.arg("keyspace")
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut dbs = Vec::new();
|
||||
for line in info.lines() {
|
||||
if line.starts_with("db") {
|
||||
if let Some(num) = line.strip_prefix("db").and_then(|s| s.split(':').next()) {
|
||||
if let Ok(n) = num.parse::<u32>() {
|
||||
dbs.push(n);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(dbs)
|
||||
}
|
||||
|
||||
pub async fn select_db(con: &mut redis::aio::MultiplexedConnection, db: u32) -> Result<(), String> {
|
||||
redis::cmd("SELECT")
|
||||
.arg(db)
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn scan_keys_page(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
cursor: u64,
|
||||
pattern: &str,
|
||||
count: usize,
|
||||
) -> Result<RedisScanResult, String> {
|
||||
let (next_cursor, keys): (u64, Vec<String>) = redis::cmd("SCAN")
|
||||
.arg(cursor)
|
||||
.arg("MATCH")
|
||||
.arg(pattern)
|
||||
.arg("COUNT")
|
||||
.arg(count)
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut result = Vec::new();
|
||||
for key in &keys {
|
||||
let key_type: String = redis::cmd("TYPE")
|
||||
.arg(key.as_str())
|
||||
.query_async(con)
|
||||
.await
|
||||
.unwrap_or_else(|_| "unknown".to_string());
|
||||
|
||||
let ttl: i64 = con.ttl(key.as_str()).await.unwrap_or(-1);
|
||||
|
||||
result.push(RedisKeyInfo {
|
||||
key: key.clone(),
|
||||
key_type,
|
||||
ttl,
|
||||
});
|
||||
}
|
||||
Ok(RedisScanResult {
|
||||
cursor: next_cursor,
|
||||
keys: result,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn get_value(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
) -> Result<RedisValue, String> {
|
||||
let key_type: String = redis::cmd("TYPE")
|
||||
.arg(key)
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let ttl: i64 = con.ttl(key).await.unwrap_or(-1);
|
||||
|
||||
let value = match key_type.as_str() {
|
||||
"string" => {
|
||||
let v: String = con.get(key).await.map_err(|e| e.to_string())?;
|
||||
serde_json::Value::String(v)
|
||||
}
|
||||
"list" => {
|
||||
let v: Vec<String> = con.lrange(key, 0, -1).await.map_err(|e| e.to_string())?;
|
||||
serde_json::json!(v)
|
||||
}
|
||||
"set" => {
|
||||
let v: Vec<String> = con.smembers(key).await.map_err(|e| e.to_string())?;
|
||||
serde_json::json!(v)
|
||||
}
|
||||
"zset" => {
|
||||
let v: Vec<(String, f64)> = con
|
||||
.zrange_withscores(key, 0, -1)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
serde_json::json!(v
|
||||
.iter()
|
||||
.map(|(m, s)| serde_json::json!({"member": m, "score": s}))
|
||||
.collect::<Vec<_>>())
|
||||
}
|
||||
"hash" => {
|
||||
let v: Vec<(String, String)> = con.hgetall(key).await.map_err(|e| e.to_string())?;
|
||||
let map: serde_json::Map<String, serde_json::Value> = v
|
||||
.into_iter()
|
||||
.map(|(k, v)| (k, serde_json::Value::String(v)))
|
||||
.collect();
|
||||
serde_json::Value::Object(map)
|
||||
}
|
||||
"stream" => get_stream_entries(con, key).await?,
|
||||
_ => serde_json::Value::Null,
|
||||
};
|
||||
|
||||
Ok(RedisValue {
|
||||
key: key.to_string(),
|
||||
key_type,
|
||||
ttl,
|
||||
value,
|
||||
})
|
||||
}
|
||||
|
||||
async fn get_stream_entries(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
) -> Result<serde_json::Value, String> {
|
||||
let raw: RedisRawValue = redis::cmd("XRANGE")
|
||||
.arg(key)
|
||||
.arg("-")
|
||||
.arg("+")
|
||||
.arg("COUNT")
|
||||
.arg(STREAM_ENTRY_LIMIT)
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(parse_stream_entries(raw))
|
||||
}
|
||||
|
||||
fn parse_stream_entries(raw: RedisRawValue) -> serde_json::Value {
|
||||
match raw {
|
||||
RedisRawValue::Array(entries) => {
|
||||
serde_json::Value::Array(entries.into_iter().filter_map(parse_stream_entry).collect())
|
||||
}
|
||||
_ => serde_json::Value::Null,
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_stream_entry(entry: RedisRawValue) -> Option<serde_json::Value> {
|
||||
let mut parts = match entry {
|
||||
RedisRawValue::Array(parts) if parts.len() == 2 => parts.into_iter(),
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
let id = redis_value_to_string(parts.next()?)?;
|
||||
let fields = match parts.next()? {
|
||||
RedisRawValue::Array(fields) => fields,
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
let mut field_map = serde_json::Map::new();
|
||||
let mut fields = fields.into_iter();
|
||||
while let Some(field) = fields.next() {
|
||||
let Some(value) = fields.next() else {
|
||||
break;
|
||||
};
|
||||
if let Some(field_name) = redis_value_to_string(field) {
|
||||
let value = redis_value_to_string(value).unwrap_or_default();
|
||||
field_map.insert(field_name, serde_json::Value::String(value));
|
||||
}
|
||||
}
|
||||
|
||||
Some(serde_json::json!({
|
||||
"id": id,
|
||||
"fields": field_map,
|
||||
}))
|
||||
}
|
||||
|
||||
fn redis_value_to_string(value: RedisRawValue) -> Option<String> {
|
||||
match value {
|
||||
RedisRawValue::BulkString(bytes) => Some(String::from_utf8_lossy(&bytes).to_string()),
|
||||
RedisRawValue::SimpleString(value) => Some(value),
|
||||
RedisRawValue::Int(value) => Some(value.to_string()),
|
||||
RedisRawValue::Double(value) => Some(value.to_string()),
|
||||
RedisRawValue::Boolean(value) => Some(value.to_string()),
|
||||
RedisRawValue::VerbatimString { text, .. } => Some(text),
|
||||
RedisRawValue::Okay => Some("OK".to_string()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn set_string(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
value: &str,
|
||||
ttl: Option<i64>,
|
||||
) -> Result<(), String> {
|
||||
con.set::<_, _, ()>(key, value)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
if let Some(t) = ttl {
|
||||
if t > 0 {
|
||||
con.expire::<_, ()>(key, t)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn delete_key(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
) -> Result<(), String> {
|
||||
con.del::<_, ()>(key).await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn hash_set(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
field: &str,
|
||||
value: &str,
|
||||
) -> Result<(), String> {
|
||||
con.hset::<_, _, _, ()>(key, field, value)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn hash_del(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
field: &str,
|
||||
) -> Result<(), String> {
|
||||
con.hdel::<_, _, ()>(key, field)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn list_push(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
value: &str,
|
||||
) -> Result<(), String> {
|
||||
con.rpush::<_, _, ()>(key, value)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn list_remove(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
index: i64,
|
||||
) -> Result<(), String> {
|
||||
let placeholder = "__DELETED_PLACEHOLDER__";
|
||||
redis::cmd("LSET")
|
||||
.arg(key)
|
||||
.arg(index)
|
||||
.arg(placeholder)
|
||||
.query_async::<()>(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
con.lrem::<_, _, ()>(key, 1, placeholder)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn set_add(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
member: &str,
|
||||
) -> Result<(), String> {
|
||||
con.sadd::<_, _, ()>(key, member)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn set_remove(
|
||||
con: &mut redis::aio::MultiplexedConnection,
|
||||
key: &str,
|
||||
member: &str,
|
||||
) -> Result<(), String> {
|
||||
con.srem::<_, _, ()>(key, member)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{parse_database_count, parse_stream_entries, RedisRawValue};
|
||||
|
||||
fn bulk(value: &str) -> RedisRawValue {
|
||||
RedisRawValue::BulkString(value.as_bytes().to_vec())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_stream_entries() {
|
||||
let raw = RedisRawValue::Array(vec![RedisRawValue::Array(vec![
|
||||
bulk("1714470000000-0"),
|
||||
RedisRawValue::Array(vec![
|
||||
bulk("event"),
|
||||
bulk("login"),
|
||||
bulk("user_id"),
|
||||
bulk("42"),
|
||||
]),
|
||||
])]);
|
||||
|
||||
let parsed = parse_stream_entries(raw);
|
||||
|
||||
assert_eq!(
|
||||
parsed,
|
||||
serde_json::json!([
|
||||
{
|
||||
"id": "1714470000000-0",
|
||||
"fields": {
|
||||
"event": "login",
|
||||
"user_id": "42"
|
||||
}
|
||||
}
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn skips_malformed_stream_entries() {
|
||||
let raw = RedisRawValue::Array(vec![
|
||||
RedisRawValue::Array(vec![bulk("1714470000000-0")]),
|
||||
RedisRawValue::Array(vec![
|
||||
bulk("1714470000001-0"),
|
||||
RedisRawValue::Array(vec![bulk("event"), bulk("logout")]),
|
||||
]),
|
||||
]);
|
||||
|
||||
let parsed = parse_stream_entries(raw);
|
||||
|
||||
assert_eq!(
|
||||
parsed,
|
||||
serde_json::json!([
|
||||
{
|
||||
"id": "1714470000001-0",
|
||||
"fields": {
|
||||
"event": "logout"
|
||||
}
|
||||
}
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_configured_database_count() {
|
||||
let value = RedisRawValue::Array(vec![
|
||||
RedisRawValue::BulkString(b"databases".to_vec()),
|
||||
RedisRawValue::BulkString(b"32".to_vec()),
|
||||
]);
|
||||
|
||||
assert_eq!(parse_database_count(value), Some(32));
|
||||
}
|
||||
}
|
||||
@@ -1,211 +0,0 @@
|
||||
use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions, SqliteRow};
|
||||
use sqlx::{Column, Executor, Row};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use super::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
|
||||
|
||||
pub async fn connect_path(path: &str) -> Result<SqlitePool, String> {
|
||||
let mut options = SqliteConnectOptions::new()
|
||||
.filename(path)
|
||||
.create_if_missing(true);
|
||||
|
||||
if is_network_path(path) {
|
||||
options = options.vfs("unix-nolock");
|
||||
}
|
||||
|
||||
SqlitePoolOptions::new()
|
||||
.max_connections(5)
|
||||
.acquire_timeout(Duration::from_secs(10))
|
||||
.idle_timeout(Duration::from_secs(300))
|
||||
.connect_with(options)
|
||||
.await
|
||||
.map_err(|e| format!("SQLite connection failed: {e}"))
|
||||
}
|
||||
|
||||
fn is_network_path(path: &str) -> bool {
|
||||
path.starts_with("\\\\") || path.starts_with("//") || path.contains("wsl.localhost") || path.contains("wsl$")
|
||||
}
|
||||
|
||||
pub async fn list_databases(_pool: &SqlitePool) -> Result<Vec<DatabaseInfo>, String> {
|
||||
Ok(vec![DatabaseInfo { name: "main".to_string() }])
|
||||
}
|
||||
|
||||
pub async fn list_tables(pool: &SqlitePool, _schema: &str) -> Result<Vec<TableInfo>, String> {
|
||||
let rows: Vec<SqliteRow> = sqlx::query(
|
||||
"SELECT name, type FROM sqlite_master WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY name",
|
||||
)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
let t: String = row.get("type");
|
||||
TableInfo {
|
||||
name: row.get::<String, _>("name"),
|
||||
table_type: if t == "view" { "VIEW".to_string() } else { "BASE TABLE".to_string() },
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
|
||||
let rows: Vec<SqliteRow> = sqlx::query(&format!("PRAGMA table_info(\"{}\")", table))
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| ColumnInfo {
|
||||
name: row.get::<String, _>("name"),
|
||||
data_type: row.get::<String, _>("type"),
|
||||
is_nullable: row.get::<i32, _>("notnull") == 0,
|
||||
column_default: row.get::<Option<String>, _>("dflt_value"),
|
||||
is_primary_key: row.get::<i32, _>("pk") > 0,
|
||||
extra: None, comment: None,
|
||||
numeric_precision: None,
|
||||
numeric_scale: None,
|
||||
character_maximum_length: None,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_indexes(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
|
||||
let safe_table = table.replace('"', "\"\"");
|
||||
let idx_rows: Vec<SqliteRow> = sqlx::query(&format!("PRAGMA index_list(\"{safe_table}\")"))
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut indexes = Vec::new();
|
||||
for idx_row in &idx_rows {
|
||||
let name: String = idx_row.get("name");
|
||||
let is_unique: bool = idx_row.get::<i32, _>("unique") != 0;
|
||||
let origin: String = idx_row.get::<String, _>("origin");
|
||||
let is_primary = origin == "pk";
|
||||
|
||||
let safe_name = name.replace('"', "\"\"");
|
||||
let col_rows: Vec<SqliteRow> = sqlx::query(&format!("PRAGMA index_info(\"{safe_name}\")"))
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let columns: Vec<String> = col_rows.iter().map(|r| r.get::<String, _>("name")).collect();
|
||||
|
||||
indexes.push(IndexInfo {
|
||||
name,
|
||||
columns,
|
||||
is_unique,
|
||||
is_primary,
|
||||
filter: None,
|
||||
index_type: None,
|
||||
included_columns: None,
|
||||
comment: None,
|
||||
});
|
||||
}
|
||||
Ok(indexes)
|
||||
}
|
||||
|
||||
pub async fn list_foreign_keys(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<ForeignKeyInfo>, String> {
|
||||
let rows: Vec<SqliteRow> = sqlx::query(&format!("PRAGMA foreign_key_list(\"{}\")", table))
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| ForeignKeyInfo {
|
||||
name: format!("fk_{}", row.get::<i32, _>("id")),
|
||||
column: row.get::<String, _>("from"),
|
||||
ref_table: row.get::<String, _>("table"),
|
||||
ref_column: row.get::<String, _>("to"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_triggers(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
|
||||
let rows: Vec<SqliteRow> = sqlx::query(
|
||||
"SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND tbl_name = ? ORDER BY name",
|
||||
)
|
||||
.bind(table)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
let sql_text: String = row.get::<Option<String>, _>("sql").unwrap_or_default();
|
||||
let upper = sql_text.to_uppercase();
|
||||
let timing = if upper.contains("BEFORE") { "BEFORE" } else if upper.contains("AFTER") { "AFTER" } else { "INSTEAD OF" };
|
||||
let event = if upper.contains("INSERT") { "INSERT" } else if upper.contains("UPDATE") { "UPDATE" } else { "DELETE" };
|
||||
TriggerInfo {
|
||||
name: row.get::<String, _>("name"),
|
||||
event: event.to_string(),
|
||||
timing: timing.to_string(),
|
||||
}
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn execute_query(pool: &SqlitePool, sql: &str) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
let trimmed = sql.trim().to_uppercase();
|
||||
|
||||
if trimmed.starts_with("SELECT")
|
||||
|| trimmed.starts_with("PRAGMA")
|
||||
|| trimmed.starts_with("EXPLAIN")
|
||||
|| trimmed.starts_with("WITH")
|
||||
{
|
||||
let desc = pool.describe(sql).await.map_err(|e| e.to_string())?;
|
||||
let columns: Vec<String> = desc.columns().iter().map(|c| c.name().to_string()).collect();
|
||||
|
||||
let rows: Vec<SqliteRow> = sqlx::query(sql)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let result_rows: Vec<Vec<serde_json::Value>> = rows
|
||||
.iter()
|
||||
.map(|row| {
|
||||
(0..row.len())
|
||||
.map(|i| {
|
||||
row.try_get::<String, _>(i)
|
||||
.map(serde_json::Value::String)
|
||||
.or_else(|_| row.try_get::<i64, _>(i).map(|v| serde_json::Value::Number(v.into())))
|
||||
.or_else(|_| row.try_get::<f64, _>(i).map(|v| {
|
||||
serde_json::Number::from_f64(v)
|
||||
.map(serde_json::Value::Number)
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
}))
|
||||
.or_else(|_| row.try_get::<bool, _>(i).map(serde_json::Value::Bool))
|
||||
.unwrap_or(serde_json::Value::Null)
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(QueryResult {
|
||||
columns,
|
||||
rows: result_rows,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
} else {
|
||||
let result = sqlx::query(sql)
|
||||
.execute(pool)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows: result.rows_affected(),
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,272 +0,0 @@
|
||||
use rust_decimal::Decimal;
|
||||
use tiberius::{AuthMethod, Client, Config};
|
||||
use tokio::net::TcpStream;
|
||||
use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt};
|
||||
use std::time::Instant;
|
||||
|
||||
use super::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
|
||||
|
||||
pub type SqlServerClient = Client<Compat<TcpStream>>;
|
||||
|
||||
pub async fn connect(host: &str, port: u16, user: &str, pass: &str, database: Option<&str>) -> Result<SqlServerClient, String> {
|
||||
match try_connect(host, port, user, pass, database, true).await {
|
||||
Ok(client) => Ok(client),
|
||||
Err(_) => try_connect(host, port, user, pass, database, false).await,
|
||||
}
|
||||
}
|
||||
|
||||
async fn try_connect(host: &str, port: u16, user: &str, pass: &str, database: Option<&str>, use_encryption: bool) -> Result<SqlServerClient, String> {
|
||||
let mut config = Config::new();
|
||||
config.host(host);
|
||||
config.port(port);
|
||||
config.authentication(AuthMethod::sql_server(user, pass));
|
||||
if let Some(db) = database {
|
||||
config.database(db);
|
||||
}
|
||||
config.trust_cert();
|
||||
if !use_encryption {
|
||||
config.encryption(tiberius::EncryptionLevel::NotSupported);
|
||||
}
|
||||
|
||||
let tcp = TcpStream::connect(config.get_addr())
|
||||
.await
|
||||
.map_err(|e| format!("SQL Server connection failed: {e}"))?;
|
||||
Client::connect(config, tcp.compat_write())
|
||||
.await
|
||||
.map_err(|e| format!("SQL Server connection failed: {e}"))
|
||||
}
|
||||
|
||||
fn row_to_json(row: &tiberius::Row) -> Vec<serde_json::Value> {
|
||||
(0..row.len()).map(|i| {
|
||||
if let Some(v) = row.try_get::<&str, _>(i).ok().flatten() {
|
||||
serde_json::Value::String(v.to_string())
|
||||
} else if let Some(v) = row.try_get::<Decimal, _>(i).ok().flatten() {
|
||||
serde_json::Value::String(v.to_string())
|
||||
} else if let Some(v) = row.try_get::<i32, _>(i).ok().flatten() {
|
||||
serde_json::Value::Number(v.into())
|
||||
} else if let Some(v) = row.try_get::<i64, _>(i).ok().flatten() {
|
||||
serde_json::Value::Number(v.into())
|
||||
} else if let Some(v) = row.try_get::<f64, _>(i).ok().flatten() {
|
||||
serde_json::Number::from_f64(v).map(serde_json::Value::Number).unwrap_or(serde_json::Value::Null)
|
||||
} else if let Some(v) = row.try_get::<bool, _>(i).ok().flatten() {
|
||||
serde_json::Value::Bool(v)
|
||||
} else {
|
||||
serde_json::Value::Null
|
||||
}
|
||||
}).collect()
|
||||
}
|
||||
|
||||
pub async fn list_databases(client: &mut SqlServerClient) -> Result<Vec<DatabaseInfo>, String> {
|
||||
let stream = client.query("SELECT name FROM sys.databases ORDER BY name", &[])
|
||||
.await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
DatabaseInfo { name: row.get::<&str, _>(0).unwrap_or("").to_string() }
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_schemas(client: &mut SqlServerClient) -> Result<Vec<String>, String> {
|
||||
let stream = client.query(
|
||||
"SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA \
|
||||
WHERE SCHEMA_NAME NOT IN ('guest','INFORMATION_SCHEMA','sys') \
|
||||
ORDER BY SCHEMA_NAME",
|
||||
&[],
|
||||
).await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
row.get::<&str, _>(0).unwrap_or("").to_string()
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_tables(client: &mut SqlServerClient, schema: &str) -> Result<Vec<TableInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT TABLE_NAME, TABLE_TYPE FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '{}' ORDER BY TABLE_NAME",
|
||||
schema.replace('\'', "''")
|
||||
);
|
||||
let stream = client.query(&*sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
TableInfo {
|
||||
name: row.get::<&str, _>(0).unwrap_or("").to_string(),
|
||||
table_type: row.get::<&str, _>(1).unwrap_or("BASE TABLE").to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn get_columns(client: &mut SqlServerClient, schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT c.COLUMN_NAME, c.DATA_TYPE, c.IS_NULLABLE, c.COLUMN_DEFAULT, \
|
||||
CASE WHEN kcu.COLUMN_NAME IS NOT NULL THEN 1 ELSE 0 END AS IS_PK, \
|
||||
c.NUMERIC_PRECISION, c.NUMERIC_SCALE, c.CHARACTER_MAXIMUM_LENGTH, c.DATETIME_PRECISION \
|
||||
FROM INFORMATION_SCHEMA.COLUMNS c \
|
||||
LEFT JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE kcu \
|
||||
ON c.TABLE_SCHEMA = kcu.TABLE_SCHEMA AND c.TABLE_NAME = kcu.TABLE_NAME AND c.COLUMN_NAME = kcu.COLUMN_NAME \
|
||||
AND kcu.CONSTRAINT_NAME IN (SELECT CONSTRAINT_NAME FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS WHERE CONSTRAINT_TYPE = 'PRIMARY KEY' AND TABLE_SCHEMA = '{s}' AND TABLE_NAME = '{t}') \
|
||||
WHERE c.TABLE_SCHEMA = '{s}' AND c.TABLE_NAME = '{t}' \
|
||||
ORDER BY c.ORDINAL_POSITION",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let stream = client.query(&*sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
let base = row.get::<&str, _>(1).unwrap_or("").to_string();
|
||||
let max_len = row.get::<i32, _>(7);
|
||||
let dt_prec = row.get::<i32, _>(8);
|
||||
let num_prec = row.get::<i32, _>(5);
|
||||
let num_scale = row.get::<i32, _>(6);
|
||||
let data_type = match base.to_lowercase().as_str() {
|
||||
"varchar" => match max_len {
|
||||
Some(-1) => "varchar(max)".to_string(),
|
||||
Some(n) => format!("varchar({n})"),
|
||||
None => "varchar".to_string(),
|
||||
},
|
||||
"nvarchar" => match max_len {
|
||||
Some(-1) => "nvarchar(max)".to_string(),
|
||||
Some(n) => format!("nvarchar({n})"),
|
||||
None => "nvarchar".to_string(),
|
||||
},
|
||||
"varbinary" => match max_len {
|
||||
Some(-1) => "varbinary(max)".to_string(),
|
||||
Some(n) if n > 0 => format!("varbinary({n})"),
|
||||
_ => "varbinary".to_string(),
|
||||
},
|
||||
"char" | "nchar" | "binary" => match max_len {
|
||||
Some(n) if n > 0 => format!("{base}({n})"),
|
||||
_ => base,
|
||||
}
|
||||
"decimal" | "numeric" => match (num_prec, num_scale) {
|
||||
(Some(p), Some(s)) => format!("{base}({p},{s})"),
|
||||
_ => base,
|
||||
},
|
||||
"datetime2" | "datetimeoffset" | "time" => match dt_prec {
|
||||
Some(p) => format!("{base}({p})"),
|
||||
_ => base,
|
||||
},
|
||||
_ => base,
|
||||
};
|
||||
ColumnInfo {
|
||||
name: row.get::<&str, _>(0).unwrap_or("").to_string(),
|
||||
data_type,
|
||||
is_nullable: row.get::<&str, _>(2).unwrap_or("NO") == "YES",
|
||||
column_default: row.get::<&str, _>(3).map(|s| s.to_string()),
|
||||
is_primary_key: row.get::<i32, _>(4).unwrap_or(0) == 1,
|
||||
extra: None, comment: None,
|
||||
numeric_precision: num_prec,
|
||||
numeric_scale: num_scale,
|
||||
character_maximum_length: max_len,
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_indexes(client: &mut SqlServerClient, schema: &str, table: &str) -> Result<Vec<IndexInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT i.name, \
|
||||
STRING_AGG(CASE WHEN ic.is_included_column = 0 THEN c.name END, ',') WITHIN GROUP (ORDER BY ic.key_ordinal) AS columns, \
|
||||
i.is_unique, i.is_primary_key, i.type_desc, \
|
||||
STRING_AGG(CASE WHEN ic.is_included_column = 1 THEN c.name END, ',') AS included_cols, \
|
||||
i.filter_definition \
|
||||
FROM sys.indexes i \
|
||||
JOIN sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id \
|
||||
JOIN sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id \
|
||||
WHERE i.object_id = OBJECT_ID('{s}.{t}') AND i.name IS NOT NULL \
|
||||
GROUP BY i.name, i.is_unique, i.is_primary_key, i.type_desc, i.filter_definition \
|
||||
ORDER BY i.name",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let stream = client.query(&*sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
let cols_str = row.get::<&str, _>(1).unwrap_or("");
|
||||
let inc_str = row.get::<&str, _>(5).unwrap_or("");
|
||||
IndexInfo {
|
||||
name: row.get::<&str, _>(0).unwrap_or("").to_string(),
|
||||
columns: cols_str.split(',').filter(|s| !s.is_empty()).map(|s| s.to_string()).collect(),
|
||||
is_unique: row.get::<bool, _>(2).unwrap_or(false),
|
||||
is_primary: row.get::<bool, _>(3).unwrap_or(false),
|
||||
filter: row.get::<&str, _>(6).map(|s| s.to_string()),
|
||||
index_type: row.get::<&str, _>(4).map(|s| s.to_string()),
|
||||
included_columns: if inc_str.is_empty() { None } else { Some(inc_str.split(',').map(|s| s.to_string()).collect()) },
|
||||
comment: None,
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_foreign_keys(client: &mut SqlServerClient, schema: &str, table: &str) -> Result<Vec<ForeignKeyInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT fk.name, c.name, rt.name, rc.name \
|
||||
FROM sys.foreign_keys fk \
|
||||
JOIN sys.foreign_key_columns fkc ON fk.object_id = fkc.constraint_object_id \
|
||||
JOIN sys.columns c ON fkc.parent_object_id = c.object_id AND fkc.parent_column_id = c.column_id \
|
||||
JOIN sys.tables rt ON fkc.referenced_object_id = rt.object_id \
|
||||
JOIN sys.columns rc ON fkc.referenced_object_id = rc.object_id AND fkc.referenced_column_id = rc.column_id \
|
||||
WHERE fk.parent_object_id = OBJECT_ID('{s}.{t}') \
|
||||
ORDER BY fk.name",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let stream = client.query(&*sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
ForeignKeyInfo {
|
||||
name: row.get::<&str, _>(0).unwrap_or("").to_string(),
|
||||
column: row.get::<&str, _>(1).unwrap_or("").to_string(),
|
||||
ref_table: row.get::<&str, _>(2).unwrap_or("").to_string(),
|
||||
ref_column: row.get::<&str, _>(3).unwrap_or("").to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn list_triggers(client: &mut SqlServerClient, schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
|
||||
let sql = format!(
|
||||
"SELECT t.name, te.type_desc, CASE WHEN t.is_instead_of_trigger = 1 THEN 'INSTEAD OF' ELSE 'AFTER' END \
|
||||
FROM sys.triggers t \
|
||||
JOIN sys.trigger_events te ON t.object_id = te.object_id \
|
||||
WHERE t.parent_id = OBJECT_ID('{s}.{t}') \
|
||||
ORDER BY t.name",
|
||||
s = schema.replace('\'', "''"), t = table.replace('\'', "''")
|
||||
);
|
||||
let stream = client.query(&*sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| {
|
||||
TriggerInfo {
|
||||
name: row.get::<&str, _>(0).unwrap_or("").to_string(),
|
||||
event: row.get::<&str, _>(1).unwrap_or("").to_string(),
|
||||
timing: row.get::<&str, _>(2).unwrap_or("AFTER").to_string(),
|
||||
}
|
||||
}).collect())
|
||||
}
|
||||
|
||||
pub async fn execute_query(client: &mut SqlServerClient, sql: &str) -> Result<QueryResult, String> {
|
||||
let start = Instant::now();
|
||||
let trimmed = sql.trim().to_uppercase();
|
||||
|
||||
if trimmed.starts_with("SELECT")
|
||||
|| trimmed.starts_with("EXEC")
|
||||
|| trimmed.starts_with("WITH")
|
||||
|| trimmed.starts_with("TABLE")
|
||||
{
|
||||
let mut stream = client.query(sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
let columns_meta = stream.columns().await.map_err(|e| e.to_string())?
|
||||
.map(|cols| cols.iter().map(|c| c.name().to_string()).collect::<Vec<_>>())
|
||||
.unwrap_or_default();
|
||||
|
||||
let rows = stream.into_first_result().await.map_err(|e| e.to_string())?;
|
||||
let result_rows: Vec<Vec<serde_json::Value>> = rows.iter().map(|row| row_to_json(row)).collect();
|
||||
|
||||
Ok(QueryResult {
|
||||
columns: columns_meta,
|
||||
rows: result_rows,
|
||||
affected_rows: 0,
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
} else {
|
||||
let result = client.execute(sql, &[]).await.map_err(|e| e.to_string())?;
|
||||
Ok(QueryResult {
|
||||
columns: vec![],
|
||||
rows: vec![],
|
||||
affected_rows: result.rows_affected().iter().sum::<u64>(),
|
||||
execution_time_ms: start.elapsed().as_millis(),
|
||||
truncated: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,203 +0,0 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use russh::client::{self, Config, Handle};
|
||||
use russh::keys::{key::PrivateKeyWithHashAlg, load_secret_key};
|
||||
use russh::ChannelMsg;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::Mutex;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
struct SshClient;
|
||||
|
||||
impl client::Handler for SshClient {
|
||||
type Error = russh::Error;
|
||||
|
||||
async fn check_server_key(
|
||||
&mut self,
|
||||
_server_public_key: &russh::keys::ssh_key::PublicKey,
|
||||
) -> Result<bool, Self::Error> {
|
||||
Ok(true)
|
||||
}
|
||||
}
|
||||
|
||||
async fn connect_and_authenticate(
|
||||
ssh_host: &str,
|
||||
ssh_port: u16,
|
||||
ssh_user: &str,
|
||||
ssh_password: &str,
|
||||
ssh_key_path: &str,
|
||||
ssh_key_passphrase: &str,
|
||||
) -> Result<Handle<SshClient>, String> {
|
||||
let config = Arc::new(Config {
|
||||
nodelay: true,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let mut session = client::connect(config, (ssh_host, ssh_port), SshClient {})
|
||||
.await
|
||||
.map_err(|e| format!("SSH connection failed: {e}"))?;
|
||||
|
||||
if !ssh_key_path.is_empty() {
|
||||
let passphrase = if ssh_key_passphrase.is_empty() { None } else { Some(ssh_key_passphrase) };
|
||||
let key_pair = load_secret_key(ssh_key_path, passphrase)
|
||||
.map_err(|e| format!("Failed to load SSH key: {e}"))?;
|
||||
let auth_res = session
|
||||
.authenticate_publickey(
|
||||
ssh_user,
|
||||
PrivateKeyWithHashAlg::new(
|
||||
Arc::new(key_pair),
|
||||
session
|
||||
.best_supported_rsa_hash()
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.flatten(),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| format!("SSH key auth failed: {e}"))?;
|
||||
if !auth_res.success() {
|
||||
return Err("SSH public key authentication failed".to_string());
|
||||
}
|
||||
} else if !ssh_password.is_empty() {
|
||||
let auth_res = session
|
||||
.authenticate_password(ssh_user, ssh_password)
|
||||
.await
|
||||
.map_err(|e| format!("SSH password auth failed: {e}"))?;
|
||||
if !auth_res.success() {
|
||||
return Err("SSH password authentication failed".to_string());
|
||||
}
|
||||
} else {
|
||||
return Err("No SSH password or key provided".to_string());
|
||||
}
|
||||
|
||||
Ok(session)
|
||||
}
|
||||
|
||||
async fn forward_loop(
|
||||
session: Handle<SshClient>,
|
||||
listener: TcpListener,
|
||||
remote_host: String,
|
||||
remote_port: u16,
|
||||
) {
|
||||
loop {
|
||||
let (mut stream, peer_addr) = match listener.accept().await {
|
||||
Ok(v) => v,
|
||||
Err(_) => break,
|
||||
};
|
||||
|
||||
let mut channel = match session
|
||||
.channel_open_direct_tcpip(
|
||||
&remote_host,
|
||||
remote_port.into(),
|
||||
peer_addr.ip().to_string(),
|
||||
peer_addr.port().into(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
log::error!("SSH direct-tcpip failed: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut buf = vec![0u8; 65536];
|
||||
let mut stream_closed = false;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
r = stream.read(&mut buf), if !stream_closed => {
|
||||
match r {
|
||||
Ok(0) => {
|
||||
stream_closed = true;
|
||||
let _ = channel.eof().await;
|
||||
}
|
||||
Ok(n) => {
|
||||
if channel.data(&buf[..n]).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
msg = channel.wait() => {
|
||||
match msg {
|
||||
Some(ChannelMsg::Data { ref data }) => {
|
||||
if stream.write_all(data).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Some(ChannelMsg::Eof) | None => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
pub struct TunnelManager {
|
||||
tunnels: Mutex<HashMap<String, (JoinHandle<()>, u16)>>,
|
||||
}
|
||||
|
||||
impl TunnelManager {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
tunnels: Mutex::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn start_tunnel(
|
||||
&self,
|
||||
connection_id: &str,
|
||||
ssh_host: &str,
|
||||
ssh_port: u16,
|
||||
ssh_user: &str,
|
||||
ssh_password: &str,
|
||||
ssh_key_path: &str,
|
||||
ssh_key_passphrase: &str,
|
||||
remote_host: &str,
|
||||
remote_port: u16,
|
||||
expose_to_lan: bool,
|
||||
) -> Result<u16, String> {
|
||||
let local_port = portpicker::pick_unused_port().ok_or("No available port")?;
|
||||
|
||||
let session =
|
||||
connect_and_authenticate(ssh_host, ssh_port, ssh_user, ssh_password, ssh_key_path, ssh_key_passphrase)
|
||||
.await?;
|
||||
|
||||
let bind_addr = if expose_to_lan { "0.0.0.0" } else { "127.0.0.1" };
|
||||
let listener = TcpListener::bind((bind_addr, local_port))
|
||||
.await
|
||||
.map_err(|e| format!("Failed to bind local port: {e}"))?;
|
||||
|
||||
let remote_host = remote_host.to_string();
|
||||
let handle = tokio::spawn(forward_loop(session, listener, remote_host, remote_port));
|
||||
|
||||
self.tunnels
|
||||
.lock()
|
||||
.await
|
||||
.insert(connection_id.to_string(), (handle, local_port));
|
||||
|
||||
Ok(local_port)
|
||||
}
|
||||
|
||||
pub async fn local_port(&self, connection_id: &str) -> Option<u16> {
|
||||
self.tunnels
|
||||
.lock()
|
||||
.await
|
||||
.get(connection_id)
|
||||
.map(|(_, port)| *port)
|
||||
}
|
||||
|
||||
pub async fn stop_tunnel(&self, connection_id: &str) {
|
||||
if let Some((handle, _)) = self.tunnels.lock().await.remove(connection_id) {
|
||||
handle.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user