diff --git a/apps/desktop/public/icons/database/turso.png b/apps/desktop/public/icons/database/turso.png new file mode 100644 index 0000000000..d78843d483 Binary files /dev/null and b/apps/desktop/public/icons/database/turso.png differ diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index 8a2e2d2346..b935791e62 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -359,6 +359,7 @@ const driverProfiles: Record< redis: { type: "redis", port: 6379, user: "", label: "Redis", icon: "redis" }, sqlite: { type: "sqlite", port: 0, user: "", label: "SQLite", icon: "sqlite" }, rqlite: { type: "rqlite", port: 4001, user: "", label: "RQLite", icon: "rqlite" }, + turso: { type: "turso", port: 443, user: "", label: "Turso", icon: "turso" }, duckdb: { type: "duckdb", port: 0, user: "", label: "DuckDB", icon: "duckdb" }, access: { type: "access", port: 0, user: "", label: "Microsoft Access", icon: "access" }, mongodb: { type: "mongodb", port: 27017, user: "", label: "MongoDB", icon: "mongodb" }, @@ -718,6 +719,7 @@ const iconTypeMap: Record = { postgres: "postgres", sqlite: "sqlite", rqlite: "rqlite", + turso: "turso", access: "access", redis: "redis", mongodb: "mongodb", @@ -779,20 +781,22 @@ const iconTypeMap: Record = { }; const dbOptions = [ - { value: "mysql", label: "MySQL" }, { value: "postgres", label: "PostgreSQL" }, + { value: "mysql", label: "MySQL" }, + { value: "mongodb", label: "MongoDB" }, + { value: "redis", label: "Redis" }, + { value: "oracle", label: "Oracle" }, { value: "sqlite", label: "SQLite" }, + { value: "sqlserver", label: "SQL Server" }, + { value: "elasticsearch", label: "Elasticsearch" }, + { value: "dm", label: "DM (Dameng)" }, + { value: "opengauss", label: "openGauss" }, + { value: "turso", label: "Turso" }, + { value: "duckdb", label: "DuckDB" }, { value: "rqlite", label: "RQLite" }, { value: "access", label: "Microsoft Access" }, - { value: "redis", label: "Redis" }, - { value: "mongodb", label: "MongoDB" }, - { value: "duckdb", label: "DuckDB" }, - { value: "clickhouse", label: "ClickHouse" }, - { value: "sqlserver", label: "SQL Server" }, - { value: "oracle", label: "Oracle" }, - { value: "elasticsearch", label: "Elasticsearch" }, { value: "mariadb", label: "MariaDB" }, - { value: "dm", label: "DM (Dameng)" }, + { value: "clickhouse", label: "ClickHouse" }, { value: "gaussdb", label: "GaussDB" }, { value: "kwdb", label: "KWDB" }, { value: "tidb", label: "TiDB" }, @@ -813,7 +817,6 @@ const dbOptions = [ { value: "firebird", label: "Firebird" }, { value: "exasol", label: "Exasol" }, { value: "gbase", label: "GBase" }, - { value: "opengauss", label: "openGauss" }, { value: "kingbase", label: "KingBase" }, { value: "highgo", label: "瀚高 HighGo" }, { value: "yashandb", label: "崖山 YashanDB" }, diff --git a/apps/desktop/src/components/icons/DatabaseIcon.vue b/apps/desktop/src/components/icons/DatabaseIcon.vue index da46d2e017..f90708b65b 100644 --- a/apps/desktop/src/components/icons/DatabaseIcon.vue +++ b/apps/desktop/src/components/icons/DatabaseIcon.vue @@ -12,6 +12,7 @@ const assetIcons: Record = { postgresql: "postgres", sqlite: "sqlite", rqlite: "rqlite.png", + turso: "turso.png", redis: "redis", mongodb: "mongodb", clickhouse: "clickhouse", diff --git a/apps/desktop/src/components/layout/ContentArea.vue b/apps/desktop/src/components/layout/ContentArea.vue index c978f1c094..6ca42a14b4 100644 --- a/apps/desktop/src/components/layout/ContentArea.vue +++ b/apps/desktop/src/components/layout/ContentArea.vue @@ -172,6 +172,7 @@ const activeSqlFormatDialect = computed(() => { return "postgres"; case "sqlite": case "rqlite": + case "turso": return "sqlite"; case "sqlserver": return "sqlserver"; diff --git a/apps/desktop/src/components/objects/ObjectBrowser.vue b/apps/desktop/src/components/objects/ObjectBrowser.vue index d67d4355c0..f5631eb337 100644 --- a/apps/desktop/src/components/objects/ObjectBrowser.vue +++ b/apps/desktop/src/components/objects/ObjectBrowser.vue @@ -198,6 +198,7 @@ const sourceFormatDialect = computed(() => { case "sqlserver": return effectiveDatabaseType.value; case "rqlite": + case "turso": return "sqlite"; case "gaussdb": case "kwdb": diff --git a/apps/desktop/src/lib/connectionPresentation.ts b/apps/desktop/src/lib/connectionPresentation.ts index 02c70459c4..e65a44dfcb 100644 --- a/apps/desktop/src/lib/connectionPresentation.ts +++ b/apps/desktop/src/lib/connectionPresentation.ts @@ -115,6 +115,9 @@ export function connectionUrlPlaceholder(dbType: DatabaseType): string { case "rqlite": return "http://user:password@host:4001"; + case "turso": + return "https://[your-db]-[org].turso.io"; + case "duckdb": return "duckdb:///absolute/path/to/database.duckdb"; diff --git a/apps/desktop/src/lib/databaseCapabilitySets.ts b/apps/desktop/src/lib/databaseCapabilitySets.ts index ab31625a75..6257fec308 100644 --- a/apps/desktop/src/lib/databaseCapabilitySets.ts +++ b/apps/desktop/src/lib/databaseCapabilitySets.ts @@ -41,6 +41,7 @@ export const DIAGRAM_SUPPORTED_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "sqlserver", "oracle", "redshift", @@ -72,6 +73,7 @@ export const DATABASE_SEARCH_SUPPORTED_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "sqlserver", "oracle", "redshift", @@ -118,6 +120,7 @@ export const TABLE_IMPORT_SUPPORTED_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "duckdb", "clickhouse", "sqlserver", @@ -141,6 +144,7 @@ export const TABLE_STRUCTURE_SUPPORTED_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "duckdb", "clickhouse", "sqlserver", @@ -183,6 +187,7 @@ export const FIELD_LINEAGE_SUPPORTED_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "sqlserver", "oracle", "redshift", @@ -279,6 +284,7 @@ export const TRANSFER_SQL_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "sqlserver", "oracle", "clickhouse", @@ -297,6 +303,7 @@ export const DIAGRAM_SQL_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "sqlserver", "oracle", "redshift", diff --git a/apps/desktop/src/lib/databaseFeatureSupport.ts b/apps/desktop/src/lib/databaseFeatureSupport.ts index f33253eb78..6bac2ba6a7 100644 --- a/apps/desktop/src/lib/databaseFeatureSupport.ts +++ b/apps/desktop/src/lib/databaseFeatureSupport.ts @@ -104,7 +104,7 @@ export function supportsObjectBrowserTreeNode(dbType: DatabaseType | undefined, } export function supportsTableTruncate(dbType?: DatabaseType): boolean { - return !!dbType && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "duckdb"; + return !!dbType && dbType !== "sqlite" && dbType !== "rqlite" && dbType !== "turso" && dbType !== "duckdb"; } export function usesPostgresLikeStructureCopy(dbType?: DatabaseType): boolean { diff --git a/apps/desktop/src/lib/databaseObjectCapabilities.ts b/apps/desktop/src/lib/databaseObjectCapabilities.ts index effba95e2f..18734a7e04 100644 --- a/apps/desktop/src/lib/databaseObjectCapabilities.ts +++ b/apps/desktop/src/lib/databaseObjectCapabilities.ts @@ -16,6 +16,7 @@ const ORACLE_OBJECTS: SidebarObjectKind[] = ["TABLE", "VIEW", "PROCEDURE", "FUNC const TABLE_VIEW_ONLY_TYPES = new Set([ "sqlite", "rqlite", + "turso", "duckdb", "clickhouse", "doris", diff --git a/apps/desktop/src/lib/databaseTableDataCapabilities.ts b/apps/desktop/src/lib/databaseTableDataCapabilities.ts index b2ce37035d..e45fd3f1e6 100644 --- a/apps/desktop/src/lib/databaseTableDataCapabilities.ts +++ b/apps/desktop/src/lib/databaseTableDataCapabilities.ts @@ -50,6 +50,7 @@ const NAVICAT_STYLE_TABLE_DATA_TYPES = new Set([ "postgres", "sqlite", "rqlite", + "turso", "duckdb", "sqlserver", "oracle", diff --git a/apps/desktop/src/lib/objectRenameSql.ts b/apps/desktop/src/lib/objectRenameSql.ts index 233daeb42a..8491d5bb56 100644 --- a/apps/desktop/src/lib/objectRenameSql.ts +++ b/apps/desktop/src/lib/objectRenameSql.ts @@ -32,7 +32,7 @@ export function supportsObjectRename( if (objectType === "PROCEDURE" || objectType === "FUNCTION") { return false; } - if (databaseType === "sqlite" || databaseType === "rqlite") return objectType === "TABLE"; + if (databaseType === "sqlite" || databaseType === "rqlite" || databaseType === "turso") return objectType === "TABLE"; if (databaseType === "mysql" || databaseType === "goldendb") return objectType === "TABLE" || objectType === "VIEW"; if (postgresLikeRenameTypes.has(databaseType)) return objectType === "TABLE" || objectType === "VIEW"; if (oracleLikeRenameTypes.has(databaseType)) return objectType === "TABLE" || objectType === "VIEW"; diff --git a/apps/desktop/src/lib/sqlCompletion.ts b/apps/desktop/src/lib/sqlCompletion.ts index 2f1ca4a25b..3b603a6daa 100644 --- a/apps/desktop/src/lib/sqlCompletion.ts +++ b/apps/desktop/src/lib/sqlCompletion.ts @@ -452,6 +452,7 @@ const DATABASE_SQL_KEYWORDS: Partial> = { postgres: POSTGRES_SQL_KEYWORDS, sqlite: SQLITE_SQL_KEYWORDS, rqlite: SQLITE_SQL_KEYWORDS, + turso: SQLITE_SQL_KEYWORDS, sqlserver: SQLSERVER_SQL_KEYWORDS, }; @@ -850,6 +851,7 @@ const DATABASE_FUNCTION_SIGNATURES: Partial kingbase: postgresCapabilities, sqlite: sqliteCapabilities, rqlite: sqliteCapabilities, + turso: sqliteCapabilities, duckdb: duckdbCapabilities, sqlserver: sqlserverCapabilities, oracle: oracleCapabilities, diff --git a/apps/desktop/src/lib/tableStructureEditorState.ts b/apps/desktop/src/lib/tableStructureEditorState.ts index f1528643e8..0ec8847b26 100644 --- a/apps/desktop/src/lib/tableStructureEditorState.ts +++ b/apps/desktop/src/lib/tableStructureEditorState.ts @@ -115,6 +115,7 @@ export const DATA_TYPE_OPTIONS: Record = { ], sqlite: ["integer", "real", "text", "blob", "numeric"], rqlite: ["integer", "real", "text", "blob", "numeric"], + turso: ["integer", "real", "text", "blob", "numeric"], sqlserver: [ "bit", "tinyint", diff --git a/apps/desktop/src/types/database.ts b/apps/desktop/src/types/database.ts index 5236d289c2..b3fc5deb8e 100644 --- a/apps/desktop/src/types/database.ts +++ b/apps/desktop/src/types/database.ts @@ -3,6 +3,7 @@ export type DatabaseType = | "postgres" | "sqlite" | "rqlite" + | "turso" | "redis" | "duckdb" | "clickhouse" diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index 2632f1db5c..6c28e85c4b 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -40,6 +40,7 @@ pub enum PoolKind { Postgres(deadpool_postgres::Pool), Sqlite(db::sqlite::SqliteHandle), Rqlite(db::rqlite_driver::RqliteClient), + Turso(db::turso_driver::TursoClient), Redis(db::redis_driver::RedisConnection), DuckDb(Arc>), MongoDb(mongodb::Client), @@ -377,6 +378,16 @@ impl AppState { db::rqlite_driver::test_connection(&client, connect_timeout).await?; PoolKind::Rqlite(client) } + DatabaseType::Turso => { + let auth_token = if !db_config.password.is_empty() { + db_config.password.clone() + } else { + db_config.url_params.as_deref().and_then(extract_auth_token_from_params).unwrap_or_default() + }; + let client = db::turso_driver::TursoClient::new(&url, &auth_token, db_config.ssl, connect_timeout)?; + db::turso_driver::test_connection(&client, connect_timeout).await?; + PoolKind::Turso(client) + } DatabaseType::Redis => { let con = if db_config.uses_redis_cluster() { db::redis_driver::RedisConnection::Cluster(db::redis_driver::connect_cluster(&db_config).await?) @@ -939,6 +950,7 @@ pub async fn close_pool_kind(pool: PoolKind) { PoolKind::Postgres(p) => p.close(), PoolKind::Sqlite(_) => {} PoolKind::Rqlite(_) => {} + PoolKind::Turso(_) => {} PoolKind::Redis(_) => {} PoolKind::DuckDb(con) => { crate::db::duckdb_driver::close_connection(con); @@ -956,6 +968,19 @@ pub async fn close_pool_kind(pool: PoolKind) { } } +fn extract_auth_token_from_params(params: &str) -> Option { + params + .trim() + .trim_start_matches('?') + .split('&') + .filter_map(|pair| pair.split_once('=')) + .find(|(key, _)| { + let k = key.trim().to_ascii_lowercase(); + k == "auth_token" || k == "authtoken" || k == "auth-token" + }) + .map(|(_, value)| value.trim().to_string()) +} + fn base_pool_key_for( db_type: Option, connection_id: &str, diff --git a/crates/dbx-core/src/database_capabilities.rs b/crates/dbx-core/src/database_capabilities.rs index b243f2cd69..03a2a87281 100644 --- a/crates/dbx-core/src/database_capabilities.rs +++ b/crates/dbx-core/src/database_capabilities.rs @@ -15,6 +15,7 @@ pub fn is_single_connection_pool(db_type: &DatabaseType) -> bool { DatabaseType::Sqlite | DatabaseType::DuckDb | DatabaseType::Rqlite + | DatabaseType::Turso | DatabaseType::MongoDb | DatabaseType::Oracle | DatabaseType::Dameng @@ -36,5 +37,6 @@ pub fn is_metadata_connection_scoped(db_type: &DatabaseType) -> bool { } pub fn skips_tcp_probe(db_type: &DatabaseType) -> bool { - matches!(db_type, DatabaseType::Sqlite | DatabaseType::DuckDb | DatabaseType::Jdbc) || is_agent_type(db_type) + matches!(db_type, DatabaseType::Sqlite | DatabaseType::DuckDb | DatabaseType::Turso | DatabaseType::Jdbc) + || is_agent_type(db_type) } diff --git a/crates/dbx-core/src/db/mod.rs b/crates/dbx-core/src/db/mod.rs index 91aa497724..e39380560b 100644 --- a/crates/dbx-core/src/db/mod.rs +++ b/crates/dbx-core/src/db/mod.rs @@ -15,6 +15,7 @@ pub mod sqlite; pub mod sqlserver; pub mod ssh_tunnel; pub mod transport_layer_tunnel; +pub mod turso_driver; use reqwest::ClientBuilder; use std::future::Future; diff --git a/crates/dbx-core/src/db/turso_driver.rs b/crates/dbx-core/src/db/turso_driver.rs new file mode 100644 index 0000000000..e69c61454e --- /dev/null +++ b/crates/dbx-core/src/db/turso_driver.rs @@ -0,0 +1,745 @@ +use reqwest::Client as HttpClient; +use serde::Deserialize; +use std::time::{Duration, Instant}; + +use super::{http_client_builder, with_connection_timeout}; +use crate::sql::starts_with_executable_sql_keyword; +use crate::types::{ + ColumnInfo, DatabaseInfo, ForeignKeyInfo, IndexInfo, ObjectSource, ObjectSourceKind, QueryResult, TableInfo, + TriggerInfo, +}; + +#[derive(Clone)] +pub struct TursoClient { + http: HttpClient, + base_url: String, + auth_token: String, +} + +impl TursoClient { + pub fn new(url: &str, auth_token: &str, tls_enabled: bool, timeout: Duration) -> Result { + let mut builder = http_client_builder(timeout); + if turso_accept_invalid_certs(tls_enabled) { + builder = builder.danger_accept_invalid_certs(true); + } + let http = builder.build().map_err(|e| format!("Failed to configure Turso HTTP client: {e}"))?; + Ok(Self { http, base_url: url.trim_end_matches('/').to_string(), auth_token: auth_token.to_string() }) + } + + fn post_pipeline(&self, sql: &str) -> reqwest::RequestBuilder { + let requests = vec![serde_json::json!({ + "type": "execute", + "stmt": { "sql": sql } + })]; + self.post_pipeline_requests(&requests) + } + + fn post_pipeline_batch(&self, statements: &[&str]) -> reqwest::RequestBuilder { + let requests: Vec<_> = statements + .iter() + .map(|sql| { + serde_json::json!({ + "type": "execute", + "stmt": { "sql": sql } + }) + }) + .collect(); + self.post_pipeline_requests(&requests) + } + + fn post_pipeline_requests(&self, requests: &[serde_json::Value]) -> reqwest::RequestBuilder { + self.http + .post(format!("{}/v2/pipeline", self.base_url)) + .header("Authorization", format!("Bearer {}", self.auth_token)) + .json(&serde_json::json!({ "requests": requests })) + } +} + +/// Turso libSQL HTTP API response types + +#[derive(Debug, Deserialize)] +struct TursoPipelineResponse { + #[serde(default)] + results: Vec, + #[serde(default)] + #[allow(dead_code)] + baton: Option, + #[serde(default)] + #[allow(dead_code)] + base_url: Option, +} + +#[derive(Debug, Deserialize)] +struct TursoPipelineResult { + #[serde(rename = "type")] + result_type: String, + /// Present when result_type is "ok" + #[serde(default)] + response: Option, + /// Present when result_type is "error" + #[serde(default)] + error: Option, +} + +#[derive(Debug, Deserialize)] +struct TursoResponseEnvelope { + #[serde(rename = "type")] + #[allow(dead_code)] + response_type: String, + #[serde(default)] + result: Option, +} + +#[derive(Debug, Deserialize)] +struct TursoResultSet { + #[serde(default)] + cols: Vec, + #[serde(default)] + rows: Vec>, + #[serde(default)] + #[allow(dead_code)] + rows_read: Option, + #[serde(default)] + #[allow(dead_code)] + rows_written: Option, + #[serde(default)] + affected_row_count: Option, +} + +#[derive(Debug, Deserialize, Clone)] +struct TursoColumn { + name: String, + #[serde(default)] + #[allow(dead_code)] + decltype: Option, +} + +#[derive(Debug, Deserialize, Clone)] +struct TursoValue { + #[serde(rename = "type", default)] + value_type: String, + #[serde(default)] + value: Option, +} + +#[derive(Debug, Deserialize)] +struct TursoError { + message: String, + #[serde(default)] + #[allow(dead_code)] + code: Option, +} + +// ─── Public API ─────────────────────────────────────────────────────────────── + +pub async fn test_connection(client: &TursoClient, timeout: Duration) -> Result<(), String> { + with_connection_timeout("turso", timeout, async { execute_inner(client, "SELECT 1").await.map(|_| ()) }).await +} + +pub async fn list_databases(_client: &TursoClient) -> Result, String> { + Ok(vec![DatabaseInfo { name: "main".to_string() }]) +} + +pub async fn list_tables(client: &TursoClient, _schema: &str) -> Result, String> { + let result = query_inner( + client, + "SELECT name, type FROM sqlite_master WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY name", + ) + .await?; + Ok(result + .values + .into_iter() + .map(|row| { + let table_type = value_as_string(row.get(1)).unwrap_or_else(|| "table".to_string()); + TableInfo { + name: value_as_string(row.first()).unwrap_or_default(), + table_type: if table_type.eq_ignore_ascii_case("view") { "VIEW" } else { "BASE TABLE" }.to_string(), + comment: None, + parent_schema: None, + parent_name: None, + } + }) + .collect()) +} + +pub async fn get_columns(client: &TursoClient, _schema: &str, table: &str) -> Result, String> { + let result = query_inner(client, &format!("PRAGMA table_info({})", sqlite_ident(table))).await?; + Ok(result + .values + .into_iter() + .map(|row| ColumnInfo { + name: value_by_column(&result.columns, &row, "name").unwrap_or_default(), + data_type: value_by_column(&result.columns, &row, "type").unwrap_or_default(), + is_nullable: value_by_column(&result.columns, &row, "notnull") + .and_then(|value| value.parse::().ok()) + .unwrap_or(0) + == 0, + column_default: value_by_column(&result.columns, &row, "dflt_value"), + is_primary_key: value_by_column(&result.columns, &row, "pk") + .and_then(|value| value.parse::().ok()) + .unwrap_or(0) + > 0, + extra: None, + comment: None, + numeric_precision: None, + numeric_scale: None, + character_maximum_length: None, + }) + .collect()) +} + +pub async fn list_indexes(client: &TursoClient, _schema: &str, table: &str) -> Result, String> { + let result = query_inner(client, &format!("PRAGMA index_list({})", sqlite_ident(table))).await?; + let mut indexes = Vec::new(); + + for row in result.values { + let name = value_by_column(&result.columns, &row, "name").unwrap_or_default(); + if name.is_empty() { + continue; + } + let is_unique = + value_by_column(&result.columns, &row, "unique").and_then(|value| value.parse::().ok()).unwrap_or(0) + != 0; + let origin = value_by_column(&result.columns, &row, "origin").unwrap_or_default(); + let column_result = query_inner(client, &format!("PRAGMA index_info({})", sqlite_ident(&name))).await?; + let columns = column_result + .values + .iter() + .filter_map(|row| value_by_column(&column_result.columns, row, "name")) + .collect(); + indexes.push(IndexInfo { + name, + columns, + is_unique, + is_primary: origin == "pk", + filter: None, + index_type: None, + included_columns: None, + comment: None, + }); + } + + Ok(indexes) +} + +pub async fn list_foreign_keys( + client: &TursoClient, + _schema: &str, + table: &str, +) -> Result, String> { + let result = query_inner(client, &format!("PRAGMA foreign_key_list({})", sqlite_ident(table))).await?; + Ok(result + .values + .into_iter() + .map(|row| ForeignKeyInfo { + name: format!("fk_{}", value_by_column(&result.columns, &row, "id").unwrap_or_else(|| "0".to_string())), + column: value_by_column(&result.columns, &row, "from").unwrap_or_default(), + ref_schema: None, + ref_table: value_by_column(&result.columns, &row, "table").unwrap_or_default(), + ref_column: value_by_column(&result.columns, &row, "to").unwrap_or_default(), + }) + .collect()) +} + +pub async fn list_triggers(client: &TursoClient, _schema: &str, table: &str) -> Result, String> { + let result = query_inner( + client, + &format!( + "SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND tbl_name = {} ORDER BY name", + sqlite_string(table) + ), + ) + .await?; + Ok(result + .values + .into_iter() + .map(|row| { + let sql_text = value_as_string(row.get(1)).unwrap_or_default().to_uppercase(); + let timing = if sql_text.contains("BEFORE") { + "BEFORE" + } else if sql_text.contains("AFTER") { + "AFTER" + } else { + "INSTEAD OF" + }; + let event = if sql_text.contains("INSERT") { + "INSERT" + } else if sql_text.contains("UPDATE") { + "UPDATE" + } else { + "DELETE" + }; + TriggerInfo { + name: value_as_string(row.first()).unwrap_or_default(), + event: event.to_string(), + timing: timing.to_string(), + } + }) + .collect()) +} + +pub async fn table_ddl(client: &TursoClient, table: &str) -> Result { + first_string_cell( + query_inner( + client, + &format!("SELECT sql FROM sqlite_master WHERE type='table' AND name={}", sqlite_string(table)), + ) + .await?, + ) +} + +pub async fn object_source( + client: &TursoClient, + name: &str, + object_type: &ObjectSourceKind, +) -> Result { + let kind = match object_type { + ObjectSourceKind::View => "view", + _ => return Err("Object source is not supported for this Turso object type".to_string()), + }; + let source = first_string_cell( + query_inner( + client, + &format!( + "SELECT sql FROM sqlite_master WHERE type={} AND name={}", + sqlite_string(kind), + sqlite_string(name) + ), + ) + .await?, + )?; + Ok(ObjectSource { name: name.to_string(), object_type: object_type.clone(), schema: None, source }) +} + +pub async fn execute_query(client: &TursoClient, sql: &str) -> Result { + execute_query_with_max_rows(client, sql, None).await +} + +pub async fn execute_query_with_max_rows( + client: &TursoClient, + sql: &str, + max_rows: Option, +) -> Result { + let start = Instant::now(); + + // Split multi-statement SQL: each statement in the pipeline is part of a + // single implicit transaction, so BEGIN/INSERT/COMMIT must travel together. + let statements: Vec<&str> = split_sql_statements(sql); + let is_reader = statements.len() == 1 + && starts_with_executable_sql_keyword(statements[0], &["SELECT", "PRAGMA", "EXPLAIN", "WITH"]); + + // Single-statement transaction control: Turso pipelines are implicitly + // auto-committed, so standalone BEGIN/COMMIT/ROLLBACK are no-ops. + let is_tx_control = statements.len() == 1 + && starts_with_executable_sql_keyword(statements[0], &["BEGIN", "COMMIT", "ROLLBACK", "START", "END"]); + + if is_tx_control { + return Ok(QueryResult { + columns: vec![], + column_types: Vec::new(), + column_sortables: vec![], + rows: vec![], + affected_rows: 0, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + session_id: None, + has_more: false, + }); + } + + if is_reader { + let result = query_inner(client, statements[0]).await?; + Ok(query_result_from_turso_result(result, start.elapsed().as_millis(), max_rows)) + } else if statements.len() == 1 { + let affected_rows = execute_inner(client, statements[0]).await?; + Ok(QueryResult { + columns: vec![], + column_types: Vec::new(), + column_sortables: vec![], + rows: vec![], + affected_rows, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + session_id: None, + has_more: false, + }) + } else { + // Batch multiple statements into a single pipeline for transactional integrity + let result_set = send_pipeline_batch(client, &statements).await?; + let affected = result_set.affected_row_count.unwrap_or(0); + Ok(QueryResult { + columns: vec![], + column_types: Vec::new(), + column_sortables: vec![], + rows: vec![], + affected_rows: affected, + execution_time_ms: start.elapsed().as_millis(), + truncated: false, + session_id: None, + has_more: false, + }) + } +} + +/// Split SQL text on statement boundaries (naive split on `;` that respects +/// basic quoting; sufficient for the multi-statement batching use case). +fn split_sql_statements(sql: &str) -> Vec<&str> { + let trimmed = sql.trim(); + if trimmed.is_empty() { + return vec![]; + } + let mut stmts: Vec<&str> = Vec::new(); + let mut start = 0; + let bytes = trimmed.as_bytes(); + let mut in_single = false; + let mut in_double = false; + for (i, &b) in bytes.iter().enumerate() { + match b { + b'\'' if !in_double => in_single = !in_single, + b'"' if !in_single => in_double = !in_double, + b';' if !in_single && !in_double => { + let stmt = trimmed[start..i].trim(); + if !stmt.is_empty() { + stmts.push(stmt); + } + start = i + 1; + } + _ => {} + } + } + let last = trimmed[start..].trim(); + if !last.is_empty() && last != ";" { + stmts.push(last); + } + stmts +} + +// ─── Internal helpers ───────────────────────────────────────────────────────── + +async fn query_inner(client: &TursoClient, sql: &str) -> Result { + let result_set = send_pipeline_for_result_set(client, sql).await?; + Ok(extract_result(result_set)) +} + +async fn execute_inner(client: &TursoClient, sql: &str) -> Result { + let result_set = send_pipeline_for_result_set(client, sql).await?; + Ok(result_set.affected_row_count.unwrap_or(0)) +} + +async fn send_pipeline_for_result_set(client: &TursoClient, sql: &str) -> Result { + let result = send_pipeline(client, sql).await?; + match result { + TursoPipelineResult { result_type, response: Some(resp), error: None } if result_type == "ok" => { + match resp.result { + Some(rs) => Ok(rs), + None => Err("Turso response missing result".to_string()), + } + } + TursoPipelineResult { error: Some(err), .. } => Err(format!("Turso error: {}", err.message)), + other => Err(format!("Unexpected Turso result type: {}", other.result_type)), + } +} + +async fn send_pipeline(client: &TursoClient, sql: &str) -> Result { + let resp = client.post_pipeline(sql).send().await.map_err(|e| format!("Turso request failed: {e}"))?; + let status = resp.status(); + let body = resp.text().await.map_err(|e| format!("Turso response read failed: {e}"))?; + if !status.is_success() { + return Err(format!("Turso error ({}): {}", status.as_u16(), body)); + } + let response: TursoPipelineResponse = + serde_json::from_str(&body).map_err(|e| format!("Turso parse error: {e}; body: {body}"))?; + response.results.into_iter().next().ok_or_else(|| "Turso returned no result".to_string()) +} + +async fn send_pipeline_batch(client: &TursoClient, statements: &[&str]) -> Result { + let resp = client.post_pipeline_batch(statements).send().await.map_err(|e| format!("Turso request failed: {e}"))?; + let status = resp.status(); + let body = resp.text().await.map_err(|e| format!("Turso response read failed: {e}"))?; + if !status.is_success() { + return Err(format!("Turso error ({}): {}", status.as_u16(), body)); + } + let response: TursoPipelineResponse = + serde_json::from_str(&body).map_err(|e| format!("Turso parse error: {e}; body: {body}"))?; + + // For batch, take the last result (COMMIT result for BEGIN/INSERT/COMMIT patterns) + let mut last_result_set = + TursoResultSet { cols: vec![], rows: vec![], rows_read: None, rows_written: None, affected_row_count: Some(0) }; + for result in &response.results { + match result { + TursoPipelineResult { result_type, error: Some(err), .. } if result_type == "error" => { + return Err(format!("Turso error: {}", err.message)); + } + TursoPipelineResult { response: Some(resp), .. } => { + if let Some(rs) = &resp.result { + last_result_set = TursoResultSet { + cols: rs.cols.clone(), + rows: rs.rows.clone(), + rows_read: rs.rows_read, + rows_written: rs.rows_written, + affected_row_count: Some( + last_result_set.affected_row_count.unwrap_or(0) + rs.affected_row_count.unwrap_or(0), + ), + }; + } + } + _ => {} + } + } + Ok(last_result_set) +} + +/// Internal result representation with decoded columns and rows +struct TursoExtractedResult { + columns: Vec, + values: Vec>, + #[allow(dead_code)] + rows_written: Option, +} + +fn extract_result(result_set: TursoResultSet) -> TursoExtractedResult { + let columns: Vec = result_set.cols.iter().map(|c| c.name.clone()).collect(); + let values: Vec> = + result_set.rows.into_iter().map(|row| row.into_iter().map(turso_value_to_json).collect()).collect(); + TursoExtractedResult { columns, values, rows_written: result_set.rows_written } +} + +fn turso_value_to_json(v: TursoValue) -> serde_json::Value { + match v.value { + Some(val) => match v.value_type.as_str() { + "integer" => val + .as_str() + .and_then(|s| s.parse::().ok()) + .map(serde_json::Value::from) + .unwrap_or(serde_json::Value::Null), + "float" => val + .as_str() + .and_then(|s| s.parse::().ok()) + .map(|f| serde_json::json!(f)) + .unwrap_or(serde_json::Value::Null), + "text" => serde_json::Value::String(val.as_str().unwrap_or("").to_string()), + "null" => serde_json::Value::Null, + "blob" => { + // Blobs in Turso are base64-encoded strings + let hex = val.as_str().map(base64_to_hex).unwrap_or_else(|| val.to_string()); + serde_json::Value::String(format!("0x{}", hex)) + } + _ => { + // Fallback: try to use the raw value + val + } + }, + None => serde_json::Value::Null, + } +} + +fn base64_to_hex(_b64: &str) -> String { + // For blob values, we pass through as-is if not parseable + // A full base64→hex implementation can be added later + _b64.to_string() +} + +fn query_result_from_turso_result( + mut result: TursoExtractedResult, + execution_time_ms: u128, + max_rows: Option, +) -> QueryResult { + let row_limit = max_rows.unwrap_or(crate::query::MAX_ROWS).max(1); + let truncated = result.values.len() > row_limit; + if truncated { + result.values.truncate(row_limit); + } + QueryResult { + columns: result.columns, + column_types: Vec::new(), + column_sortables: vec![], + rows: result.values, + affected_rows: 0, + execution_time_ms, + truncated, + session_id: None, + has_more: false, + } +} + +fn first_string_cell(result: TursoExtractedResult) -> Result { + result + .values + .first() + .and_then(|row| row.first()) + .and_then(|value| value_as_string(Some(value))) + .ok_or_else(|| "Object not found".to_string()) +} + +fn value_by_column(columns: &[String], row: &[serde_json::Value], name: &str) -> Option { + columns + .iter() + .position(|column| column.eq_ignore_ascii_case(name)) + .and_then(|index| row.get(index)) + .and_then(|value| value_as_string(Some(value))) +} + +fn value_as_string(value: Option<&serde_json::Value>) -> Option { + match value? { + serde_json::Value::Null => None, + serde_json::Value::String(value) => Some(value.clone()), + serde_json::Value::Number(value) => Some(value.to_string()), + serde_json::Value::Bool(value) => Some(value.to_string()), + other => Some(other.to_string()), + } +} + +fn sqlite_ident(value: &str) -> String { + format!("\"{}\"", value.replace('"', "\"\"")) +} + +fn sqlite_string(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + +fn turso_accept_invalid_certs(tls_enabled: bool) -> bool { + tls_enabled +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn converts_query_result_and_truncates_rows() { + let result = TursoExtractedResult { + columns: vec!["id".to_string(), "name".to_string()], + values: vec![ + vec![serde_json::json!(1), serde_json::json!("Ada")], + vec![serde_json::json!(2), serde_json::json!("Linus")], + ], + rows_written: None, + }; + + let result = query_result_from_turso_result(result, 8, Some(1)); + + assert_eq!(result.columns, vec!["id", "name"]); + assert_eq!(result.rows, vec![vec![serde_json::json!(1), serde_json::json!("Ada")]]); + assert_eq!(result.execution_time_ms, 8); + assert!(result.truncated); + } + + #[test] + fn maps_column_values_by_name_case_insensitive() { + let columns = vec!["name".to_string(), "notnull".to_string(), "pk".to_string()]; + let row = vec![serde_json::json!("id"), serde_json::json!(1), serde_json::json!(1)]; + + assert_eq!(value_by_column(&columns, &row, "NAME").as_deref(), Some("id")); + assert_eq!(value_by_column(&columns, &row, "notnull").as_deref(), Some("1")); + } + + #[test] + #[test] + fn split_sql_statements_splits_on_semicolons() { + assert_eq!(split_sql_statements("SELECT 1"), vec!["SELECT 1"]); + assert_eq!( + split_sql_statements("BEGIN; INSERT INTO t VALUES (1); COMMIT"), + vec!["BEGIN", "INSERT INTO t VALUES (1)", "COMMIT"] + ); + assert_eq!(split_sql_statements("SELECT 1; "), vec!["SELECT 1"]); + assert_eq!(split_sql_statements(" ; "), Vec::<&str>::new()); + assert_eq!(split_sql_statements("SELECT 'a;b' AS x"), vec!["SELECT 'a;b' AS x"]); + assert_eq!(split_sql_statements("SELECT \"a;b\" AS x; SELECT 2"), vec!["SELECT \"a;b\" AS x", "SELECT 2"]); + } + + #[test] + fn value_as_string_converts_types() { + assert_eq!(value_as_string(Some(&serde_json::json!("hello"))), Some("hello".to_string())); + assert_eq!(value_as_string(Some(&serde_json::json!(42))), Some("42".to_string())); + assert_eq!(value_as_string(Some(&serde_json::json!(true))), Some("true".to_string())); + assert_eq!(value_as_string(Some(&serde_json::json!(null))), None); + } + + #[test] + fn turso_value_integer_parsed() { + let v = TursoValue { value_type: "integer".to_string(), value: Some(serde_json::json!("42")) }; + assert_eq!(turso_value_to_json(v), serde_json::json!(42)); + } + + #[test] + fn turso_value_text_parsed() { + let v = TursoValue { value_type: "text".to_string(), value: Some(serde_json::json!("hello")) }; + assert_eq!(turso_value_to_json(v), serde_json::json!("hello")); + } + + #[test] + fn turso_value_null_parsed() { + let v = TursoValue { value_type: "null".to_string(), value: None }; + assert_eq!(turso_value_to_json(v), serde_json::Value::Null); + } + + #[test] + fn turso_value_float_parsed() { + let v = TursoValue { value_type: "float".to_string(), value: Some(serde_json::json!("3.14")) }; + assert_eq!(turso_value_to_json(v), serde_json::json!(3.14)); + } + + #[test] + fn extract_result_decodes_rows_and_columns() { + let rs = TursoResultSet { + cols: vec![ + TursoColumn { name: "id".to_string(), decltype: None }, + TursoColumn { name: "name".to_string(), decltype: None }, + ], + rows: vec![vec![ + TursoValue { value_type: "integer".to_string(), value: Some(serde_json::json!("1")) }, + TursoValue { value_type: "text".to_string(), value: Some(serde_json::json!("Ada")) }, + ]], + rows_read: Some(1), + rows_written: Some(0), + affected_row_count: Some(0), + }; + + let result = extract_result(rs); + assert_eq!(result.columns, vec!["id", "name"]); + assert_eq!(result.values.len(), 1); + assert_eq!(result.values[0][0], serde_json::json!(1)); + assert_eq!(result.values[0][1], serde_json::json!("Ada")); + } + + #[test] + fn first_string_cell_returns_value() { + let result = TursoExtractedResult { + columns: vec!["sql".to_string()], + values: vec![vec![serde_json::json!("CREATE TABLE t (id INTEGER)")]], + rows_written: None, + }; + assert_eq!(first_string_cell(result).unwrap(), "CREATE TABLE t (id INTEGER)"); + } + + #[test] + fn first_string_cell_empty_returns_error() { + let result = TursoExtractedResult { columns: vec!["sql".to_string()], values: vec![], rows_written: None }; + assert!(first_string_cell(result).is_err()); + } + + #[test] + fn sqlite_ident_quotes_properly() { + assert_eq!(sqlite_ident("my_table"), "\"my_table\""); + assert_eq!(sqlite_ident("my\"table"), "\"my\"\"table\""); + } + + #[test] + fn sqlite_string_quotes_properly() { + assert_eq!(sqlite_string("hello"), "'hello'"); + assert_eq!(sqlite_string("it's"), "'it''s'"); + } + + #[test] + fn query_result_no_truncation_when_within_limit() { + let result = TursoExtractedResult { + columns: vec!["x".to_string()], + values: vec![vec![serde_json::json!(1)], vec![serde_json::json!(2)]], + rows_written: None, + }; + + let qr = query_result_from_turso_result(result, 5, Some(100)); + assert_eq!(qr.rows.len(), 2); + assert!(!qr.truncated); + } +} diff --git a/crates/dbx-core/src/models/connection.rs b/crates/dbx-core/src/models/connection.rs index 9d7c9ff068..b047d418db 100644 --- a/crates/dbx-core/src/models/connection.rs +++ b/crates/dbx-core/src/models/connection.rs @@ -270,6 +270,8 @@ pub enum DatabaseType { Etcd, #[serde(rename = "iris")] Iris, + #[serde(rename = "turso")] + Turso, Jdbc, } @@ -554,7 +556,7 @@ impl ConnectionConfig { }, DatabaseType::Redshift => Some("dev"), DatabaseType::ClickHouse => Some("default"), - DatabaseType::Rqlite => Some("main"), + DatabaseType::Rqlite | DatabaseType::Turso => Some("main"), DatabaseType::Gaussdb | DatabaseType::OpenGauss => Some("postgres"), DatabaseType::Kwdb => Some("defaultdb"), DatabaseType::Kingbase | DatabaseType::Vastbase => Some("postgres"), @@ -642,6 +644,7 @@ impl ConnectionConfig { } DatabaseType::ClickHouse => clickhouse_http_url(self, raw_host, port), DatabaseType::Rqlite => rqlite_http_url(self, raw_host, port), + DatabaseType::Turso => turso_http_url(self, raw_host, port), DatabaseType::SqlServer => { format!("server=tcp:{host},{port};database={}", self.database.as_deref().unwrap_or("master")) } @@ -758,6 +761,7 @@ impl ConnectionConfig { } DatabaseType::ClickHouse => clickhouse_http_url(self, raw_host, port), DatabaseType::Rqlite => rqlite_http_url(self, raw_host, port), + DatabaseType::Turso => turso_http_url(self, raw_host, port), DatabaseType::SqlServer => format!( "server=tcp:{host},{port};user={};password={};database={}", self.username, @@ -1135,6 +1139,19 @@ fn rqlite_http_url(config: &ConnectionConfig, host: &str, port: u16) -> String { format!("{scheme}://{}:{port}", bracket_ipv6(trimmed)) } +fn turso_http_url(config: &ConnectionConfig, host: &str, port: u16) -> String { + let trimmed = host.trim(); + if let Some(rest) = trimmed.strip_prefix("https://").or_else(|| trimmed.strip_prefix("libsql://")) { + return format!("https://{}", trim_http_host_port(rest, port)); + } + if let Some(rest) = trimmed.strip_prefix("http://") { + let scheme = if config.ssl { "https" } else { "http" }; + return format!("{scheme}://{}", trim_http_host_port(rest, port)); + } + let scheme = if config.ssl { "https" } else { "http" }; + format!("{scheme}://{}:{port}", bracket_ipv6(trimmed)) +} + fn trim_http_host_port(value: &str, default_port: u16) -> String { let authority = value.trim_end_matches('/').split('/').next().unwrap_or(value).split('?').next().unwrap_or(value); if authority.starts_with('[') && !authority.contains("]:") { diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index f0bf774f31..87d5a15a80 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -648,6 +648,17 @@ pub async fn do_execute( ) .await } + PoolKind::Turso(client) => { + let client = client.clone(); + let max_rows = options.max_rows; + drop(connections); + wait_for_query_opt( + cancel_token, + query_timeout, + db::turso_driver::execute_query_with_max_rows(&client, sql, max_rows), + ) + .await + } PoolKind::ClickHouse(client) => { let client = client.clone(); let database = pool_key.split(':').nth(1).unwrap_or("default").to_string(); @@ -927,6 +938,19 @@ pub async fn execute_multi_core_with_options( return execute_multi_sqlserver(state, &pool_key, sql, cancel_token, options).await; } + let is_turso = { + let configs = state.configs.read().await; + configs.get(connection_id).is_some_and(|c| c.db_type == DatabaseType::Turso) + }; + + // Turso sends all statements in one HTTP pipeline for transactional integrity. + if is_turso { + let result = + execute_sql_statement_with_options(state, connection_id, database, sql, schema, cancel_token, options) + .await?; + return Ok(vec![result]); + } + let db_type = connection_database_type(state, connection_id).await; let statements = db_type.map_or_else( || split_sql_statements(sql), @@ -1210,9 +1234,11 @@ pub async fn execute_statements_in_transaction( PoolKind::Postgres(pg) => TxPath::Pg(pg.clone()), PoolKind::Mysql(mp, _mode) => TxPath::Mysql(mp.clone(), false), PoolKind::Sqlite(sq) => TxPath::Sqlite(sq.clone()), - PoolKind::ClickHouse(_) | PoolKind::Rqlite(_) | PoolKind::SqlServer(_) | PoolKind::Agent(_) => { - TxPath::Explicit - } + PoolKind::ClickHouse(_) + | PoolKind::Rqlite(_) + | PoolKind::Turso(_) + | PoolKind::SqlServer(_) + | PoolKind::Agent(_) => TxPath::Explicit, PoolKind::DuckDb(_) | PoolKind::Redis(_) | PoolKind::MongoDb(_) diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 6e3cce50e3..ef45bbf77b 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -303,6 +303,7 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result db::postgres::list_databases(p).await, PoolKind::Sqlite(p) => db::sqlite::list_databases(p).await, PoolKind::Rqlite(client) => db::rqlite_driver::list_databases(client).await, + PoolKind::Turso(client) => db::turso_driver::list_databases(client).await, PoolKind::DuckDb(con) => { let con = con.lock().map_err(|e| e.to_string())?; duckdb_list_databases_with_attached(&con, &duckdb_attached_names) @@ -523,6 +524,9 @@ async fn list_tables_once( PoolKind::Rqlite(client) => db::rqlite_driver::list_tables(client, schema) .await .map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())), + PoolKind::Turso(client) => db::turso_driver::list_tables(client, schema) + .await + .map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())), PoolKind::MongoDb(client) => db::mongo_driver::list_collections(client, database) .await .map(|names| collection_names_to_tables(names, "COLLECTION")) @@ -1298,6 +1302,9 @@ pub async fn get_columns_core( PoolKind::Rqlite(client) => { db::rqlite_driver::get_columns(client, schema, table).await.map(deduplicate_column_infos) } + PoolKind::Turso(client) => { + db::turso_driver::get_columns(client, schema, table).await.map(deduplicate_column_infos) + } PoolKind::Elasticsearch(client) => { db::elasticsearch_driver::get_columns(client, table).await.map(deduplicate_column_infos) } @@ -1373,6 +1380,7 @@ pub async fn list_indexes_core( PoolKind::Postgres(p) => db::postgres::list_indexes(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::list_indexes(p, schema, table).await, PoolKind::Rqlite(client) => db::rqlite_driver::list_indexes(client, schema, table).await, + PoolKind::Turso(client) => db::turso_driver::list_indexes(client, schema, table).await, PoolKind::MongoDb(client) => db::mongo_driver::list_indexes(client, database, table).await, _ => Ok(vec![]), } @@ -1403,6 +1411,7 @@ pub async fn list_foreign_keys_core( PoolKind::Postgres(p) => db::postgres::list_foreign_keys(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::list_foreign_keys(p, schema, table).await, PoolKind::Rqlite(client) => db::rqlite_driver::list_foreign_keys(client, schema, table).await, + PoolKind::Turso(client) => db::turso_driver::list_foreign_keys(client, schema, table).await, _ => Ok(vec![]), } } @@ -1432,6 +1441,7 @@ pub async fn list_triggers_core( PoolKind::Postgres(p) => db::postgres::list_triggers(p, schema, table).await, PoolKind::Sqlite(p) => db::sqlite::list_triggers(p, schema, table).await, PoolKind::Rqlite(client) => db::rqlite_driver::list_triggers(client, schema, table).await, + PoolKind::Turso(client) => db::turso_driver::list_triggers(client, schema, table).await, _ => Ok(vec![]), } } @@ -1500,6 +1510,7 @@ pub async fn get_table_ddl_core( PoolKind::Postgres(p) => pg_ddl(p, schema, table).await, PoolKind::Sqlite(p) => sqlite_ddl(p, table).await, PoolKind::Rqlite(client) => db::rqlite_driver::table_ddl(client, table).await, + PoolKind::Turso(client) => db::turso_driver::table_ddl(client, table).await, _ => Err("DDL not supported for this database type".to_string()), } } @@ -1756,6 +1767,9 @@ pub async fn get_object_source_core( PoolKind::Rqlite(client) => { return db::rqlite_driver::object_source(client, name, &object_type).await; } + PoolKind::Turso(client) => { + return db::turso_driver::object_source(client, name, &object_type).await; + } PoolKind::ClickHouse(client) if matches!(object_type, db::ObjectSourceKind::View) => { let result = db::clickhouse_driver::execute_query( client, diff --git a/crates/dbx-core/tests/turso_integration.rs b/crates/dbx-core/tests/turso_integration.rs new file mode 100644 index 0000000000..bbdccaf358 --- /dev/null +++ b/crates/dbx-core/tests/turso_integration.rs @@ -0,0 +1,190 @@ +use dbx_core::db::turso_driver::TursoClient; +use std::time::Duration; + +const TURSO_URL: &str = "http://172.20.66.143:20007"; +const TURSO_TOKEN: &str = ""; + +fn client() -> TursoClient { + TursoClient::new(TURSO_URL, TURSO_TOKEN, false, Duration::from_secs(10)).expect("create TursoClient") +} + +#[tokio::test] +async fn test_connection_works() { + let c = client(); + dbx_core::db::turso_driver::test_connection(&c, Duration::from_secs(10)).await.expect("connection should work"); +} + +#[tokio::test] +async fn list_databases_returns_main() { + let c = client(); + let dbs = dbx_core::db::turso_driver::list_databases(&c).await.expect("list databases"); + assert_eq!(dbs.len(), 1); + assert_eq!(dbs[0].name, "main"); +} + +#[tokio::test] +async fn execute_query_select_one() { + let c = client(); + let result = dbx_core::db::turso_driver::execute_query(&c, "SELECT 1 AS num").await.expect("query should work"); + assert_eq!(result.columns, vec!["num"]); + assert_eq!(result.rows.len(), 1); +} + +#[tokio::test] +async fn execute_query_with_table() { + let c = client(); + // Create table + dbx_core::db::turso_driver::execute_query( + &c, + "CREATE TABLE IF NOT EXISTS test_people (id INTEGER PRIMARY KEY, name TEXT)", + ) + .await + .expect("create table"); + + // Insert + dbx_core::db::turso_driver::execute_query(&c, "INSERT INTO test_people VALUES (1, 'Alice')").await.expect("insert"); + + // Select + let result = + dbx_core::db::turso_driver::execute_query(&c, "SELECT * FROM test_people ORDER BY id").await.expect("select"); + + assert_eq!(result.columns, vec!["id", "name"]); + assert_eq!(result.rows.len(), 1); + + // Clean up + dbx_core::db::turso_driver::execute_query(&c, "DROP TABLE test_people").await.expect("drop"); +} + +#[tokio::test] +async fn list_tables_works() { + let c = client(); + // Create a test table first + dbx_core::db::turso_driver::execute_query(&c, "CREATE TABLE IF NOT EXISTS test_meta (x INTEGER)") + .await + .expect("create"); + + let tables = dbx_core::db::turso_driver::list_tables(&c, "").await.expect("list tables"); + + let has_test = tables.iter().any(|t| t.name == "test_meta"); + assert!(has_test, "should find test_meta table"); + + // Clean up + dbx_core::db::turso_driver::execute_query(&c, "DROP TABLE test_meta").await.expect("drop"); +} + +#[tokio::test] +async fn get_columns_works() { + let c = client(); + dbx_core::db::turso_driver::execute_query( + &c, + "CREATE TABLE IF NOT EXISTS test_cols (id INTEGER PRIMARY KEY, name TEXT NOT NULL, age INTEGER)", + ) + .await + .expect("create"); + + let cols = dbx_core::db::turso_driver::get_columns(&c, "", "test_cols").await.expect("get columns"); + + assert_eq!(cols.len(), 3); + assert_eq!(cols[0].name, "id"); + assert_eq!(cols[1].name, "name"); + assert_eq!(cols[2].name, "age"); + assert!(cols[0].is_primary_key); + assert!(!cols[1].is_nullable); + + dbx_core::db::turso_driver::execute_query(&c, "DROP TABLE test_cols").await.expect("drop"); +} + +#[tokio::test] +async fn list_indexes_works() { + let c = client(); + dbx_core::db::turso_driver::execute_query( + &c, + "CREATE TABLE IF NOT EXISTS test_idx (id INTEGER PRIMARY KEY, email TEXT)", + ) + .await + .expect("create"); + dbx_core::db::turso_driver::execute_query(&c, "CREATE UNIQUE INDEX IF NOT EXISTS idx_email ON test_idx(email)") + .await + .expect("create index"); + + let indexes = dbx_core::db::turso_driver::list_indexes(&c, "", "test_idx").await.expect("list indexes"); + + let email_idx = indexes.iter().find(|i| i.name == "idx_email"); + assert!(email_idx.is_some(), "should find email index: {:?}", indexes); + assert!(email_idx.unwrap().is_unique); + + // Note: libsql-server may not expose auto-generated primary key indexes + // via PRAGMA index_list, unlike standard SQLite. + + dbx_core::db::turso_driver::execute_query(&c, "DROP TABLE test_idx").await.expect("drop"); +} + +#[tokio::test] +async fn table_ddl_works() { + let c = client(); + dbx_core::db::turso_driver::execute_query(&c, "CREATE TABLE IF NOT EXISTS test_ddl (id INTEGER PRIMARY KEY)") + .await + .expect("create"); + + let ddl = dbx_core::db::turso_driver::table_ddl(&c, "test_ddl").await.expect("get ddl"); + + assert!(ddl.to_uppercase().contains("CREATE TABLE"), "DDL should contain CREATE TABLE: {ddl}"); + + dbx_core::db::turso_driver::execute_query(&c, "DROP TABLE test_ddl").await.expect("drop"); +} + +#[tokio::test] +async fn multi_statement_transaction_works() { + let c = client(); + // Create table first (single statement) + dbx_core::db::turso_driver::execute_query( + &c, + "CREATE TABLE IF NOT EXISTS test_batch (id INTEGER PRIMARY KEY, val TEXT)", + ) + .await + .expect("create table"); + + // Multi-statement: BEGIN + INSERT + INSERT + COMMIT in one call + dbx_core::db::turso_driver::execute_query( + &c, + "BEGIN; INSERT INTO test_batch VALUES (1, 'a'); INSERT INTO test_batch VALUES (2, 'b'); COMMIT", + ) + .await + .expect("batch transaction"); + + // Verify rows exist (proving COMMIT worked) + let result = + dbx_core::db::turso_driver::execute_query(&c, "SELECT * FROM test_batch ORDER BY id").await.expect("select"); + assert_eq!(result.rows.len(), 2, "should have 2 rows after commit"); + + // Clean up + dbx_core::db::turso_driver::execute_query(&c, "DROP TABLE test_batch").await.expect("drop"); +} + +#[tokio::test] +async fn standalone_commit_is_noop() { + let c = client(); + // Standalone COMMIT should succeed (no-op since pipeline is auto-committed) + let result = + dbx_core::db::turso_driver::execute_query(&c, "COMMIT").await.expect("standalone COMMIT should not fail"); + assert_eq!(result.affected_rows, 0); + assert!(result.columns.is_empty()); +} + +#[tokio::test] +async fn standalone_begin_is_noop() { + let c = client(); + let result = + dbx_core::db::turso_driver::execute_query(&c, "BEGIN").await.expect("standalone BEGIN should not fail"); + assert_eq!(result.affected_rows, 0); +} + +#[tokio::test] +async fn error_on_invalid_sql() { + let c = client(); + let result = dbx_core::db::turso_driver::execute_query(&c, "SELECT * FROM nonexistent_table_xyz").await; + match result { + Err(e) => assert!(e.contains("no such table"), "error should mention missing table: {e}"), + Ok(_) => panic!("should have failed"), + } +} diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index 50fd951530..e31fd3cecc 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -400,6 +400,31 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection .await .map(|_| "Connection successful".to_string()) } + DatabaseType::Turso => { + let auth_token = if !config.password.is_empty() { + config.password.clone() + } else { + config + .url_params + .as_deref() + .and_then(|p| { + p.trim() + .trim_start_matches('?') + .split('&') + .filter_map(|pair| pair.split_once('=')) + .find(|(key, _)| { + let k = key.trim().to_ascii_lowercase(); + k == "auth_token" || k == "authtoken" || k == "auth-token" + }) + .map(|(_, value)| value.trim().to_string()) + }) + .unwrap_or_default() + }; + let client = db::turso_driver::TursoClient::new(&url, &auth_token, config.ssl, connect_timeout)?; + db::turso_driver::test_connection(&client, connect_timeout) + .await + .map(|_| "Connection successful".to_string()) + } db_type if database_capabilities::is_agent_type(&db_type) => { test_agent_connection(state.inner(), &config, &host, port).await } @@ -567,6 +592,30 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi db::rqlite_driver::test_connection(&client, connect_timeout).await?; PoolKind::Rqlite(client) } + DatabaseType::Turso => { + let auth_token = if !db_config.password.is_empty() { + db_config.password.clone() + } else { + db_config + .url_params + .as_deref() + .and_then(|p| { + p.trim() + .trim_start_matches('?') + .split('&') + .filter_map(|pair| pair.split_once('=')) + .find(|(key, _)| { + let k = key.trim().to_ascii_lowercase(); + k == "auth_token" || k == "authtoken" || k == "auth-token" + }) + .map(|(_, value)| value.trim().to_string()) + }) + .unwrap_or_default() + }; + let client = db::turso_driver::TursoClient::new(&url, &auth_token, db_config.ssl, connect_timeout)?; + db::turso_driver::test_connection(&client, connect_timeout).await?; + PoolKind::Turso(client) + } db_type if database_capabilities::is_agent_type(&db_type) => { connect_agent_pool(state.inner(), &db_config, &host, port).await? }