github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-09 01:05:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Formatted

Browse Source
This commit is contained in:
Agent-43 committed 2026-05-05 13:58:14 +05:30
1 parent 0ee433763a
commit 80e1cf927e
9 files changed
+132 -328
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.husky/pre-commit
+1
View File
@@ -1,2 +1,3 @@
npx lint-staged npx lint-staged
cd src-tauri && cargo fmt && git add -u . cd src-tauri && cargo fmt && git add -u .
cd ../crates/dbx-core && cargo fmt --all && git add -u .
crates/dbx-core/src/db/duckdb_driver.rs
+2 -3
View File
@@ -1,6 +1,6 @@
use super::file_validator::validate_file_path;
use std::sync::Arc; use std::sync::Arc;
use std::sync::Mutex; use std::sync::Mutex;
use super::file_validator::validate_file_path;
/// Connects to a DuckDb database file with file validation. /// Connects to a DuckDb database file with file validation.
/// ///
@@ -14,8 +14,7 @@ pub fn connect_path(path: &str) -> Result<Arc<Mutex<duckdb::Connection>>, String
// Validate file path using universal validator // Validate file path using universal validator
validate_file_path(path, is_network_path)?; validate_file_path(path, is_network_path)?;
let connection = duckdb::Connection::open(path) let connection = duckdb::Connection::open(path).map_err(|e| format!("DuckDb connection failed: {e}"))?;
.map_err(|e| format!("DuckDb connection failed: {e}"))?;
Ok(Arc::new(Mutex::new(connection))) Ok(Arc::new(Mutex::new(connection)))
} }
crates/dbx-core/src/db/file_validator.rs
+3 -12
View File
@@ -35,26 +35,17 @@ where
// For non-network paths, perform file system checks // For non-network paths, perform file system checks
if !is_network_path(path) { if !is_network_path(path) {
if !path_obj.exists() { if !path_obj.exists() {
return Err(format!( return Err(format!("Database file does not exist: {}", path));
"Database file does not exist: {}",
path
));
} }
// Check if path is actually a file, not a directory // Check if path is actually a file, not a directory
if path_obj.is_dir() { if path_obj.is_dir() {
return Err(format!( return Err(format!("Database file path is a directory, not a file: {}", path));
"Database file path is a directory, not a file: {}",
path
));
} }
// Check if path is a valid file // Check if path is a valid file
if !path_obj.is_file() { if !path_obj.is_file() {
return Err(format!( return Err(format!("Database file path is not a valid file: {}", path));
"Database file path is not a valid file: {}",
path
));
} }
} }
crates/dbx-core/src/db/mod.rs
+2 -2
View File
@@ -1,5 +1,7 @@
pub mod clickhouse_driver; pub mod clickhouse_driver;
pub mod duckdb_driver;
pub mod elasticsearch_driver; pub mod elasticsearch_driver;
pub mod file_validator;
pub mod mongo_driver; pub mod mongo_driver;
pub mod mysql; pub mod mysql;
pub mod oracle_driver; pub mod oracle_driver;
@@ -8,8 +10,6 @@ pub mod redis_driver;
pub mod sqlite; pub mod sqlite;
pub mod sqlserver; pub mod sqlserver;
pub mod ssh_tunnel; pub mod ssh_tunnel;
pub mod file_validator;
pub mod duckdb_driver;
use std::future::Future; use std::future::Future;
use std::time::Duration; use std::time::Duration;
crates/dbx-core/src/db/mysql.rs
+29 -71
View File
@@ -23,15 +23,9 @@ fn get_str_by_name(row: &MySqlRow, name: &str) -> String {
} }
fn get_opt_str(row: &MySqlRow, name: &str) -> Option<String> { fn get_opt_str(row: &MySqlRow, name: &str) -> Option<String> {
row.try_get::<Option<String>, _>(name) row.try_get::<Option<String>, _>(name).ok().flatten().or_else(|| {
.ok() row.try_get::<Option<Vec<u8>>, _>(name).ok().flatten().map(|b| String::from_utf8_lossy(&b).to_string())
.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> { fn numeric_metadata_u64_to_i32(value: Option<u64>) -> Option<i32> {
@@ -43,8 +37,7 @@ fn numeric_metadata_i64_to_i32(value: Option<i64>) -> Option<i32> {
} }
fn numeric_metadata_str_to_i32(value: Option<String>) -> Option<i32> { fn numeric_metadata_str_to_i32(value: Option<String>) -> Option<i32> {
value.and_then(|v| v.parse::<i64>().ok()) value.and_then(|v| v.parse::<i64>().ok()).and_then(|v| i32::try_from(v).ok())
.and_then(|v| i32::try_from(v).ok())
} }
fn get_opt_i32(row: &MySqlRow, name: &str) -> Option<i32> { fn get_opt_i32(row: &MySqlRow, name: &str) -> Option<i32> {
@@ -101,20 +94,14 @@ fn mysql_value_to_json(row: &MySqlRow, idx: usize, type_name: &str) -> serde_jso
} }
if upper_type == "BOOLEAN" { if upper_type == "BOOLEAN" {
return row return row.try_get::<bool, _>(idx).map(serde_json::Value::Bool).unwrap_or(serde_json::Value::Null);
.try_get::<bool, _>(idx)
.map(serde_json::Value::Bool)
.unwrap_or(serde_json::Value::Null);
} }
if upper_type.contains("BIGINT") { if upper_type.contains("BIGINT") {
return row return row
.try_get::<i64, _>(idx) .try_get::<i64, _>(idx)
.map(|v| serde_json::Value::String(v.to_string())) .map(|v| serde_json::Value::String(v.to_string()))
.or_else(|_| { .or_else(|_| row.try_get::<u64, _>(idx).map(|v| serde_json::Value::String(v.to_string())))
row.try_get::<u64, _>(idx)
.map(|v| serde_json::Value::String(v.to_string()))
})
.unwrap_or(serde_json::Value::Null); .unwrap_or(serde_json::Value::Null);
} }
@@ -140,15 +127,14 @@ fn mysql_value_to_json(row: &MySqlRow, idx: usize, type_name: &str) -> serde_jso
.map(serde_json::Value::String) .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::<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::<u64, _>(idx).map(|v| serde_json::Value::Number(v.into())))
.or_else(|_| row.try_get::<f64, _>(idx).map(|v| { .or_else(|_| {
serde_json::Number::from_f64(v) row.try_get::<f64, _>(idx).map(|v| {
.map(serde_json::Value::Number) serde_json::Number::from_f64(v).map(serde_json::Value::Number).unwrap_or(serde_json::Value::Null)
.unwrap_or(serde_json::Value::Null) })
})) })
.or_else(|_| row.try_get::<bool, _>(idx).map(serde_json::Value::Bool)) .or_else(|_| row.try_get::<bool, _>(idx).map(serde_json::Value::Bool))
.or_else(|_| { .or_else(|_| {
row.try_get::<Vec<u8>, _>(idx) row.try_get::<Vec<u8>, _>(idx).map(|b| serde_json::Value::String(String::from_utf8_lossy(&b).to_string()))
.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)) .or_else(|e| mysql_temporal_to_json_value(row, idx).ok_or(e))
.unwrap_or(serde_json::Value::Null) .unwrap_or(serde_json::Value::Null)
@@ -168,13 +154,9 @@ pub async fn connect(url: &str) -> Result<MySqlPool, String> {
} }
pub async fn connect_bare(url: &str) -> Result<MySqlPool, String> { pub async fn connect_bare(url: &str) -> Result<MySqlPool, String> {
let options: sqlx::mysql::MySqlConnectOptions = url.parse() let options: sqlx::mysql::MySqlConnectOptions =
.map_err(|e: sqlx::Error| format!("Invalid MySQL URL: {e}"))?; url.parse().map_err(|e: sqlx::Error| format!("Invalid MySQL URL: {e}"))?;
let options = options let options = options.no_engine_substitution(false).set_names(false).pipes_as_concat(false).timezone(None);
.no_engine_substitution(false)
.set_names(false)
.pipes_as_concat(false)
.timezone(None);
super::with_connection_timeout("MySQL", async { super::with_connection_timeout("MySQL", async {
MySqlPoolOptions::new() MySqlPoolOptions::new()
.max_connections(5) .max_connections(5)
@@ -201,10 +183,7 @@ pub async fn list_tables(pool: &MySqlPool, database: &str) -> Result<Vec<TableIn
"SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = {} ORDER BY TABLE_NAME", "SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = {} ORDER BY TABLE_NAME",
quote_value(database), quote_value(database),
); );
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql) let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
@@ -215,11 +194,7 @@ pub async fn list_tables(pool: &MySqlPool, database: &str) -> Result<Vec<TableIn
.collect()) .collect())
} }
pub async fn get_columns( pub async fn get_columns(pool: &MySqlPool, database: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
pool: &MySqlPool,
database: &str,
table: &str,
) -> Result<Vec<ColumnInfo>, String> {
let sql = format!( let sql = format!(
"SELECT c.COLUMN_NAME, c.COLUMN_TYPE, c.IS_NULLABLE, c.COLUMN_DEFAULT, c.EXTRA, c.COLUMN_COMMENT, \ "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, \ CASE WHEN kcu.COLUMN_NAME IS NOT NULL THEN 1 ELSE 0 END AS IS_PK, \
@@ -235,10 +210,7 @@ pub async fn get_columns(
quote_value(database), quote_value(database),
quote_value(table), quote_value(table),
); );
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql) let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
@@ -261,12 +233,13 @@ pub async fn execute_query(pool: &MySqlPool, sql: &str, bare: bool) -> Result<Qu
let start = Instant::now(); let start = Instant::now();
let trimmed = sql.trim().to_uppercase(); let trimmed = sql.trim().to_uppercase();
if trimmed.starts_with("SELECT") || trimmed.starts_with("SHOW") || trimmed.starts_with("DESCRIBE") || trimmed.starts_with("EXPLAIN") { if trimmed.starts_with("SELECT")
|| trimmed.starts_with("SHOW")
|| trimmed.starts_with("DESCRIBE")
|| trimmed.starts_with("EXPLAIN")
{
if bare { if bare {
let rows: Vec<MySqlRow> = sqlx::raw_sql(sql) let rows: Vec<MySqlRow> = sqlx::raw_sql(sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
let (columns, column_types) = if let Some(first) = rows.first() { 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 cols: Vec<String> = first.columns().iter().map(|c| c.name().to_string()).collect();
@@ -297,10 +270,7 @@ pub async fn execute_query(pool: &MySqlPool, sql: &str, bare: bool) -> Result<Qu
let columns: Vec<String> = desc.columns().iter().map(|c| c.name().to_string()).collect(); 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 column_types: Vec<String> = desc.columns().iter().map(|c| c.type_info().name().to_string()).collect();
let rows: Vec<MySqlRow> = sqlx::query(sql) let rows: Vec<MySqlRow> = sqlx::query(sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
let result_rows: Vec<Vec<serde_json::Value>> = rows let result_rows: Vec<Vec<serde_json::Value>> = rows
.iter() .iter()
@@ -320,10 +290,7 @@ pub async fn execute_query(pool: &MySqlPool, sql: &str, bare: bool) -> Result<Qu
}) })
} }
} else { } else {
let result = sqlx::raw_sql(sql) let result = sqlx::raw_sql(sql).execute(pool).await.map_err(|e| e.to_string())?;
.execute(pool)
.await
.map_err(|e| e.to_string())?;
Ok(QueryResult { Ok(QueryResult {
columns: vec![], columns: vec![],
@@ -347,10 +314,7 @@ pub async fn list_indexes(pool: &MySqlPool, database: &str, table: &str) -> Resu
quote_value(database), quote_value(database),
quote_value(table), quote_value(table),
); );
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql) let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
@@ -381,10 +345,7 @@ pub async fn list_foreign_keys(pool: &MySqlPool, database: &str, table: &str) ->
quote_value(database), quote_value(database),
quote_value(table), quote_value(table),
); );
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql) let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
@@ -406,10 +367,7 @@ pub async fn list_triggers(pool: &MySqlPool, database: &str, table: &str) -> Res
quote_value(database), quote_value(database),
quote_value(table), quote_value(table),
); );
let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql) let rows: Vec<MySqlRow> = sqlx::raw_sql(&sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
crates/dbx-core/src/db/postgres.rs
+16 -53
View File
@@ -1,12 +1,12 @@
use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc}; use chrono::{DateTime, NaiveDate, NaiveDateTime, NaiveTime, Utc};
use percent_encoding::percent_decode_str;
use rust_decimal::Decimal; use rust_decimal::Decimal;
use sqlx::postgres::{PgPool, PgPoolOptions, PgRow}; use sqlx::postgres::{PgPool, PgPoolOptions, PgRow};
use sqlx::{Column, Executor, Row, TypeInfo, ValueRef}; use sqlx::{Column, Executor, Row, TypeInfo, ValueRef};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use percent_encoding::percent_decode_str;
use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
use super::file_validator::validate_file_path; use super::file_validator::validate_file_path;
use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
fn pg_temporal_to_json_value(row: &PgRow, idx: usize) -> Option<serde_json::Value> { fn pg_temporal_to_json_value(row: &PgRow, idx: usize) -> Option<serde_json::Value> {
if let Ok(v) = row.try_get::<DateTime<Utc>, _>(idx) { if let Ok(v) = row.try_get::<DateTime<Utc>, _>(idx) {
@@ -42,10 +42,7 @@ fn pg_value_to_json(row: &PgRow, idx: usize, type_name: &str) -> serde_json::Val
} }
if upper == "BOOL" { if upper == "BOOL" {
return row return row.try_get::<bool, _>(idx).map(serde_json::Value::Bool).unwrap_or(serde_json::Value::Null);
.try_get::<bool, _>(idx)
.map(serde_json::Value::Bool)
.unwrap_or(serde_json::Value::Null);
} }
if upper.contains("TIMESTAMP") if upper.contains("TIMESTAMP")
@@ -68,19 +65,11 @@ fn pg_value_to_json(row: &PgRow, idx: usize, type_name: &str) -> serde_json::Val
row.try_get::<String, _>(idx) row.try_get::<String, _>(idx)
.map(serde_json::Value::String) .map(serde_json::Value::String)
.or_else(|_| { .or_else(|_| row.try_get::<i64, _>(idx).map(|v| serde_json::Value::Number(v.into())))
row.try_get::<i64, _>(idx) .or_else(|_| row.try_get::<i32, _>(idx).map(|v| serde_json::Value::Number(v.into())))
.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(|_| { .or_else(|_| {
row.try_get::<f64, _>(idx).map(|v| { row.try_get::<f64, _>(idx).map(|v| {
serde_json::Number::from_f64(v) serde_json::Number::from_f64(v).map(serde_json::Value::Number).unwrap_or(serde_json::Value::Null)
.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::<bool, _>(idx).map(serde_json::Value::Bool))
@@ -137,18 +126,12 @@ fn validate_postgres_ssl_paths(url: &str) -> Result<(), String> {
} }
pub async fn list_databases(pool: &PgPool) -> Result<Vec<DatabaseInfo>, String> { pub async fn list_databases(pool: &PgPool) -> Result<Vec<DatabaseInfo>, String> {
let rows: Vec<PgRow> = let rows: Vec<PgRow> = sqlx::query("SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname")
sqlx::query("SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname") .fetch_all(pool)
.fetch_all(pool) .await
.await .map_err(|e| e.to_string())?;
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows.iter().map(|row| DatabaseInfo { name: row.get::<String, _>("datname") }).collect())
.iter()
.map(|row| DatabaseInfo {
name: row.get::<String, _>("datname"),
})
.collect())
} }
pub async fn list_tables(pool: &PgPool, schema: &str) -> Result<Vec<TableInfo>, String> { pub async fn list_tables(pool: &PgPool, schema: &str) -> Result<Vec<TableInfo>, String> {
@@ -182,17 +165,10 @@ pub async fn list_schemas(pool: &PgPool) -> Result<Vec<String>, String> {
.await .await
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
Ok(rows Ok(rows.iter().map(|row| row.get::<String, _>("schema_name")).collect())
.iter()
.map(|row| row.get::<String, _>("schema_name"))
.collect())
} }
pub async fn get_columns( pub async fn get_columns(pool: &PgPool, schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> {
pool: &PgPool,
schema: &str,
table: &str,
) -> Result<Vec<ColumnInfo>, String> {
let rows: Vec<PgRow> = sqlx::query( let rows: Vec<PgRow> = sqlx::query(
"SELECT a.attname AS column_name, \ "SELECT a.attname AS column_name, \
format_type(a.atttypid, a.atttypmod) AS full_type, \ format_type(a.atttypid, a.atttypmod) AS full_type, \
@@ -254,11 +230,7 @@ pub async fn execute_query(pool: &PgPool, sql: &str) -> Result<QueryResult, Stri
|| trimmed.starts_with("WITH") || trimmed.starts_with("WITH")
|| trimmed.starts_with("TABLE") || trimmed.starts_with("TABLE")
{ {
let rows: Vec<PgRow> = sqlx::query(sql) let rows: Vec<PgRow> = sqlx::query(sql).persistent(false).fetch_all(pool).await.map_err(|e| e.to_string())?;
.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 (columns, column_types): (Vec<String>, Vec<String>) = if let Some(first) = rows.first() {
let cols = first.columns(); let cols = first.columns();
@@ -278,13 +250,7 @@ pub async fn execute_query(pool: &PgPool, sql: &str) -> Result<QueryResult, Stri
.iter() .iter()
.map(|row| { .map(|row| {
(0..row.len()) (0..row.len())
.map(|i| { .map(|i| pg_value_to_json(row, i, column_types.get(i).map(String::as_str).unwrap_or("")))
pg_value_to_json(
row,
i,
column_types.get(i).map(String::as_str).unwrap_or(""),
)
})
.collect() .collect()
}) })
.collect(); .collect();
@@ -297,10 +263,7 @@ pub async fn execute_query(pool: &PgPool, sql: &str) -> Result<QueryResult, Stri
truncated: false, truncated: false,
}) })
} else { } else {
let result = sqlx::query(sql) let result = sqlx::query(sql).execute(pool).await.map_err(|e| e.to_string())?;
.execute(pool)
.await
.map_err(|e| e.to_string())?;
Ok(QueryResult { Ok(QueryResult {
columns: vec![], columns: vec![],
crates/dbx-core/src/db/redis_driver.rs
+42 -150
View File
@@ -27,44 +27,26 @@ pub struct RedisValue {
pub async fn connect(url: &str) -> Result<redis::aio::MultiplexedConnection, String> { 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 client = redis::Client::open(url).map_err(|e| format!("Redis connection failed: {e}"))?;
let mut con = tokio::time::timeout( let mut con = tokio::time::timeout(super::connection_timeout(), client.get_multiplexed_async_connection())
super::connection_timeout(), .await
client.get_multiplexed_async_connection(), .map_err(|_| format!("Redis connection timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
) .map_err(|e| format!("Redis connection failed: {e}"))?;
.await
.map_err(|_| format!("Redis connection timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
.map_err(|e| format!("Redis connection failed: {e}"))?;
tokio::time::timeout( tokio::time::timeout(super::connection_timeout(), redis::cmd("PING").query_async::<String>(&mut con))
super::connection_timeout(), .await
redis::cmd("PING").query_async::<String>(&mut con), .map_err(|_| format!("Redis ping timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
) .map_err(|e| format!("Redis authentication failed or command rejected: {e}"))?;
.await
.map_err(|_| format!("Redis ping timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
.map_err(|e| format!("Redis authentication failed or command rejected: {e}"))?;
Ok(con) Ok(con)
} }
pub async fn list_databases( pub async fn list_databases(con: &mut redis::aio::MultiplexedConnection) -> Result<Vec<u32>, String> {
con: &mut redis::aio::MultiplexedConnection, let configured_count =
) -> Result<Vec<u32>, String> { redis::cmd("CONFIG").arg("GET").arg("databases").query_async(con).await.ok().and_then(parse_database_count);
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 keyspace_dbs = list_keyspace_databases(con).await.unwrap_or_default();
let database_count = configured_count.unwrap_or(DEFAULT_REDIS_DATABASES); let database_count = configured_count.unwrap_or(DEFAULT_REDIS_DATABASES);
let max_db = keyspace_dbs let max_db = keyspace_dbs.iter().copied().max().map(|db| db + 1).unwrap_or(0);
.iter()
.copied()
.max()
.map(|db| db + 1)
.unwrap_or(0);
let visible_count = database_count.max(max_db).max(1); let visible_count = database_count.max(max_db).max(1);
Ok((0..visible_count).collect()) Ok((0..visible_count).collect())
@@ -86,14 +68,8 @@ fn parse_database_count(value: redis::Value) -> Option<u32> {
}) })
} }
async fn list_keyspace_databases( async fn list_keyspace_databases(con: &mut redis::aio::MultiplexedConnection) -> Result<Vec<u32>, String> {
con: &mut redis::aio::MultiplexedConnection, let info: String = redis::cmd("INFO").arg("keyspace").query_async(con).await.map_err(|e| e.to_string())?;
) -> 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(); let mut dbs = Vec::new();
for line in info.lines() { for line in info.lines() {
@@ -109,11 +85,7 @@ async fn list_keyspace_databases(
} }
pub async fn select_db(con: &mut redis::aio::MultiplexedConnection, db: u32) -> Result<(), String> { pub async fn select_db(con: &mut redis::aio::MultiplexedConnection, db: u32) -> Result<(), String> {
redis::cmd("SELECT") redis::cmd("SELECT").arg(db).query_async(con).await.map_err(|e| e.to_string())
.arg(db)
.query_async(con)
.await
.map_err(|e| e.to_string())
} }
pub async fn scan_keys_page( pub async fn scan_keys_page(
@@ -134,35 +106,18 @@ pub async fn scan_keys_page(
let mut result = Vec::new(); let mut result = Vec::new();
for key in &keys { for key in &keys {
let key_type: String = redis::cmd("TYPE") let key_type: String =
.arg(key.as_str()) redis::cmd("TYPE").arg(key.as_str()).query_async(con).await.unwrap_or_else(|_| "unknown".to_string());
.query_async(con)
.await
.unwrap_or_else(|_| "unknown".to_string());
let ttl: i64 = con.ttl(key.as_str()).await.unwrap_or(-1); let ttl: i64 = con.ttl(key.as_str()).await.unwrap_or(-1);
result.push(RedisKeyInfo { result.push(RedisKeyInfo { key: key.clone(), key_type, ttl });
key: key.clone(),
key_type,
ttl,
});
} }
Ok(RedisScanResult { Ok(RedisScanResult { cursor: next_cursor, keys: result })
cursor: next_cursor,
keys: result,
})
} }
pub async fn get_value( pub async fn get_value(con: &mut redis::aio::MultiplexedConnection, key: &str) -> Result<RedisValue, String> {
con: &mut redis::aio::MultiplexedConnection, let key_type: String = redis::cmd("TYPE").arg(key).query_async(con).await.map_err(|e| e.to_string())?;
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 ttl: i64 = con.ttl(key).await.unwrap_or(-1);
@@ -180,33 +135,20 @@ pub async fn get_value(
serde_json::json!(v) serde_json::json!(v)
} }
"zset" => { "zset" => {
let v: Vec<(String, f64)> = con let v: Vec<(String, f64)> = con.zrange_withscores(key, 0, -1).await.map_err(|e| e.to_string())?;
.zrange_withscores(key, 0, -1) serde_json::json!(v.iter().map(|(m, s)| serde_json::json!({"member": m, "score": s})).collect::<Vec<_>>())
.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" => { "hash" => {
let v: Vec<(String, String)> = con.hgetall(key).await.map_err(|e| e.to_string())?; 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 let map: serde_json::Map<String, serde_json::Value> =
.into_iter() v.into_iter().map(|(k, v)| (k, serde_json::Value::String(v))).collect();
.map(|(k, v)| (k, serde_json::Value::String(v)))
.collect();
serde_json::Value::Object(map) serde_json::Value::Object(map)
} }
"stream" => get_stream_entries(con, key).await?, "stream" => get_stream_entries(con, key).await?,
_ => serde_json::Value::Null, _ => serde_json::Value::Null,
}; };
Ok(RedisValue { Ok(RedisValue { key: key.to_string(), key_type, ttl, value })
key: key.to_string(),
key_type,
ttl,
value,
})
} }
async fn get_stream_entries( async fn get_stream_entries(
@@ -284,23 +226,16 @@ pub async fn set_string(
value: &str, value: &str,
ttl: Option<i64>, ttl: Option<i64>,
) -> Result<(), String> { ) -> Result<(), String> {
con.set::<_, _, ()>(key, value) con.set::<_, _, ()>(key, value).await.map_err(|e| e.to_string())?;
.await
.map_err(|e| e.to_string())?;
if let Some(t) = ttl { if let Some(t) = ttl {
if t > 0 { if t > 0 {
con.expire::<_, ()>(key, t) con.expire::<_, ()>(key, t).await.map_err(|e| e.to_string())?;
.await
.map_err(|e| e.to_string())?;
} }
} }
Ok(()) Ok(())
} }
pub async fn delete_key( pub async fn delete_key(con: &mut redis::aio::MultiplexedConnection, key: &str) -> Result<(), String> {
con: &mut redis::aio::MultiplexedConnection,
key: &str,
) -> Result<(), String> {
con.del::<_, ()>(key).await.map_err(|e| e.to_string()) con.del::<_, ()>(key).await.map_err(|e| e.to_string())
} }
@@ -310,67 +245,29 @@ pub async fn hash_set(
field: &str, field: &str,
value: &str, value: &str,
) -> Result<(), String> { ) -> Result<(), String> {
con.hset::<_, _, _, ()>(key, field, value) con.hset::<_, _, _, ()>(key, field, value).await.map_err(|e| e.to_string())
.await
.map_err(|e| e.to_string())
} }
pub async fn hash_del( pub async fn hash_del(con: &mut redis::aio::MultiplexedConnection, key: &str, field: &str) -> Result<(), String> {
con: &mut redis::aio::MultiplexedConnection, con.hdel::<_, _, ()>(key, field).await.map_err(|e| e.to_string())
key: &str,
field: &str,
) -> Result<(), String> {
con.hdel::<_, _, ()>(key, field)
.await
.map_err(|e| e.to_string())
} }
pub async fn list_push( pub async fn list_push(con: &mut redis::aio::MultiplexedConnection, key: &str, value: &str) -> Result<(), String> {
con: &mut redis::aio::MultiplexedConnection, con.rpush::<_, _, ()>(key, value).await.map_err(|e| e.to_string())
key: &str,
value: &str,
) -> Result<(), String> {
con.rpush::<_, _, ()>(key, value)
.await
.map_err(|e| e.to_string())
} }
pub async fn list_remove( pub async fn list_remove(con: &mut redis::aio::MultiplexedConnection, key: &str, index: i64) -> Result<(), String> {
con: &mut redis::aio::MultiplexedConnection,
key: &str,
index: i64,
) -> Result<(), String> {
let placeholder = "__DELETED_PLACEHOLDER__"; let placeholder = "__DELETED_PLACEHOLDER__";
redis::cmd("LSET") redis::cmd("LSET").arg(key).arg(index).arg(placeholder).query_async::<()>(con).await.map_err(|e| e.to_string())?;
.arg(key) con.lrem::<_, _, ()>(key, 1, placeholder).await.map_err(|e| e.to_string())
.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( pub async fn set_add(con: &mut redis::aio::MultiplexedConnection, key: &str, member: &str) -> Result<(), String> {
con: &mut redis::aio::MultiplexedConnection, con.sadd::<_, _, ()>(key, member).await.map_err(|e| e.to_string())
key: &str,
member: &str,
) -> Result<(), String> {
con.sadd::<_, _, ()>(key, member)
.await
.map_err(|e| e.to_string())
} }
pub async fn set_remove( pub async fn set_remove(con: &mut redis::aio::MultiplexedConnection, key: &str, member: &str) -> Result<(), String> {
con: &mut redis::aio::MultiplexedConnection, con.srem::<_, _, ()>(key, member).await.map_err(|e| e.to_string())
key: &str,
member: &str,
) -> Result<(), String> {
con.srem::<_, _, ()>(key, member)
.await
.map_err(|e| e.to_string())
} }
#[cfg(test)] #[cfg(test)]
@@ -385,12 +282,7 @@ mod tests {
fn parses_stream_entries() { fn parses_stream_entries() {
let raw = RedisRawValue::Array(vec![RedisRawValue::Array(vec![ let raw = RedisRawValue::Array(vec![RedisRawValue::Array(vec![
bulk("1714470000000-0"), bulk("1714470000000-0"),
RedisRawValue::Array(vec![ RedisRawValue::Array(vec![bulk("event"), bulk("login"), bulk("user_id"), bulk("42")]),
bulk("event"),
bulk("login"),
bulk("user_id"),
bulk("42"),
]),
])]); ])]);
let parsed = parse_stream_entries(raw); let parsed = parse_stream_entries(raw);
crates/dbx-core/src/db/sqlite.rs
+36 -36
View File
@@ -2,16 +2,14 @@ use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions, SqliteRo
use sqlx::{Column, Executor, Row}; use sqlx::{Column, Executor, Row};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
use super::file_validator::validate_file_path; use super::file_validator::validate_file_path;
use crate::types::{ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo};
pub async fn connect_path(path: &str) -> Result<SqlitePool, String> { pub async fn connect_path(path: &str) -> Result<SqlitePool, String> {
// Validate file path using universal validator // Validate file path using universal validator
validate_file_path(path, is_network_path)?; validate_file_path(path, is_network_path)?;
let mut options = SqliteConnectOptions::new() let mut options = SqliteConnectOptions::new().filename(path).create_if_missing(false);
.filename(path)
.create_if_missing(false);
if is_network_path(path) { if is_network_path(path) {
options = options.vfs("unix-nolock"); options = options.vfs("unix-nolock");
@@ -55,10 +53,8 @@ pub async fn list_tables(pool: &SqlitePool, _schema: &str) -> Result<Vec<TableIn
} }
pub async fn get_columns(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<ColumnInfo>, String> { 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)) let rows: Vec<SqliteRow> =
.fetch_all(pool) sqlx::query(&format!("PRAGMA table_info(\"{}\")", table)).fetch_all(pool).await.map_err(|e| e.to_string())?;
.await
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
@@ -68,7 +64,8 @@ pub async fn get_columns(pool: &SqlitePool, _schema: &str, table: &str) -> Resul
is_nullable: row.get::<i32, _>("notnull") == 0, is_nullable: row.get::<i32, _>("notnull") == 0,
column_default: row.get::<Option<String>, _>("dflt_value"), column_default: row.get::<Option<String>, _>("dflt_value"),
is_primary_key: row.get::<i32, _>("pk") > 0, is_primary_key: row.get::<i32, _>("pk") > 0,
extra: None, comment: None, extra: None,
comment: None,
numeric_precision: None, numeric_precision: None,
numeric_scale: None, numeric_scale: None,
character_maximum_length: None, character_maximum_length: None,
@@ -130,26 +127,33 @@ pub async fn list_foreign_keys(pool: &SqlitePool, _schema: &str, table: &str) ->
} }
pub async fn list_triggers(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> { pub async fn list_triggers(pool: &SqlitePool, _schema: &str, table: &str) -> Result<Vec<TriggerInfo>, String> {
let rows: Vec<SqliteRow> = sqlx::query( let rows: Vec<SqliteRow> =
"SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND tbl_name = ? ORDER BY name", sqlx::query("SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND tbl_name = ? ORDER BY name")
) .bind(table)
.bind(table) .fetch_all(pool)
.fetch_all(pool) .await
.await .map_err(|e| e.to_string())?;
.map_err(|e| e.to_string())?;
Ok(rows Ok(rows
.iter() .iter()
.map(|row| { .map(|row| {
let sql_text: String = row.get::<Option<String>, _>("sql").unwrap_or_default(); let sql_text: String = row.get::<Option<String>, _>("sql").unwrap_or_default();
let upper = sql_text.to_uppercase(); let upper = sql_text.to_uppercase();
let timing = if upper.contains("BEFORE") { "BEFORE" } else if upper.contains("AFTER") { "AFTER" } else { "INSTEAD OF" }; let timing = if upper.contains("BEFORE") {
let event = if upper.contains("INSERT") { "INSERT" } else if upper.contains("UPDATE") { "UPDATE" } else { "DELETE" }; "BEFORE"
TriggerInfo { } else if upper.contains("AFTER") {
name: row.get::<String, _>("name"), "AFTER"
event: event.to_string(), } else {
timing: timing.to_string(), "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()) .collect())
} }
@@ -166,10 +170,7 @@ pub async fn execute_query(pool: &SqlitePool, sql: &str) -> Result<QueryResult,
let desc = pool.describe(sql).await.map_err(|e| e.to_string())?; 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 columns: Vec<String> = desc.columns().iter().map(|c| c.name().to_string()).collect();
let rows: Vec<SqliteRow> = sqlx::query(sql) let rows: Vec<SqliteRow> = sqlx::query(sql).fetch_all(pool).await.map_err(|e| e.to_string())?;
.fetch_all(pool)
.await
.map_err(|e| e.to_string())?;
let result_rows: Vec<Vec<serde_json::Value>> = rows let result_rows: Vec<Vec<serde_json::Value>> = rows
.iter() .iter()
@@ -179,11 +180,13 @@ pub async fn execute_query(pool: &SqlitePool, sql: &str) -> Result<QueryResult,
row.try_get::<String, _>(i) row.try_get::<String, _>(i)
.map(serde_json::Value::String) .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::<i64, _>(i).map(|v| serde_json::Value::Number(v.into())))
.or_else(|_| row.try_get::<f64, _>(i).map(|v| { .or_else(|_| {
serde_json::Number::from_f64(v) row.try_get::<f64, _>(i).map(|v| {
.map(serde_json::Value::Number) serde_json::Number::from_f64(v)
.unwrap_or(serde_json::Value::Null) .map(serde_json::Value::Number)
})) .unwrap_or(serde_json::Value::Null)
})
})
.or_else(|_| row.try_get::<bool, _>(i).map(serde_json::Value::Bool)) .or_else(|_| row.try_get::<bool, _>(i).map(serde_json::Value::Bool))
.unwrap_or(serde_json::Value::Null) .unwrap_or(serde_json::Value::Null)
}) })
@@ -199,10 +202,7 @@ pub async fn execute_query(pool: &SqlitePool, sql: &str) -> Result<QueryResult,
truncated: false, truncated: false,
}) })
} else { } else {
let result = sqlx::query(sql) let result = sqlx::query(sql).execute(pool).await.map_err(|e| e.to_string())?;
.execute(pool)
.await
.map_err(|e| e.to_string())?;
Ok(QueryResult { Ok(QueryResult {
columns: vec![], columns: vec![],
crates/dbx-core/src/db/ssh_tunnel.rs
+1 -1
View File
@@ -9,7 +9,7 @@ use tokio::net::TcpListener;
use tokio::sync::Mutex; use tokio::sync::Mutex;
use tokio::task::JoinHandle; use tokio::task::JoinHandle;
use super::{connection_timeout, CONNECTION_TIMEOUT_SECS, file_validator::validate_file_path}; use super::{connection_timeout, file_validator::validate_file_path, CONNECTION_TIMEOUT_SECS};
struct SshClient; struct SshClient;