diff --git a/Cargo.lock b/Cargo.lock index 6363492c..271e7e89 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -617,28 +617,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "aws-lc-rs" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00" -dependencies = [ - "aws-lc-sys", - "zeroize", -] - -[[package]] -name = "aws-lc-sys" -version = "0.41.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4" -dependencies = [ - "cc", - "cmake", - "dunce", - "fs_extra", -] - [[package]] name = "base16ct" version = "0.2.0" @@ -914,15 +892,6 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" -[[package]] -name = "cmake" -version = "0.1.58" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" -dependencies = [ - "cc", -] - [[package]] name = "cmov" version = "0.5.4" @@ -1465,12 +1434,6 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" -[[package]] -name = "dunce" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" - [[package]] name = "dyn-clone" version = "1.0.20" @@ -1702,12 +1665,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "fs_extra" -version = "1.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" - [[package]] name = "futures" version = "0.3.32" @@ -3525,8 +3482,8 @@ version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2" dependencies = [ - "aws-lc-rs", "pem", + "ring", "rustls-pki-types", "time", "yasna", @@ -3734,7 +3691,6 @@ version = "0.23.40" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" dependencies = [ - "aws-lc-rs", "log", "once_cell", "ring", @@ -3753,7 +3709,6 @@ dependencies = [ "async-io", "async-trait", "async-web-client", - "aws-lc-rs", "base64 0.22.1", "blocking", "chrono", @@ -3763,6 +3718,7 @@ dependencies = [ "log", "pem", "rcgen", + "ring", "serde", "serde_json", "thiserror 2.0.18", @@ -3828,7 +3784,6 @@ version = "0.103.13" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" dependencies = [ - "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", @@ -4252,6 +4207,7 @@ dependencies = [ "awc", "base64 0.22.1", "bigdecimal 0.4.10", + "bytes", "chrono", "clap", "config", diff --git a/Cargo.toml b/Cargo.toml index 8c41eea6..a5dcd060 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,16 +58,17 @@ actix-web-httpauth = "0.8.0" rand = "0.10.0" actix-multipart = "0.7.2" base64 = "0.22" +bytes = "1" hmac = "0.13" sha2 = "0.11" -rustls-acme = "0.15" +rustls-acme = { version = "0.15", default-features = false, features = ["ring", "tls12", "webpki-roots"] } dotenvy = "0.15.7" csv-async = { version = "1.2.6", features = ["tokio"] } -rustls = { version = "0.23" } # keep in sync with actix-web, awc, and rustls-acme +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } # keep in sync with actix-web, awc, and rustls-acme rustls-native-certs = "0.8.1" awc = { version = "3", features = ["rustls-0_23-webpki-roots"] } clap = { version = "4.5.17", features = ["derive"] } -tokio-util = "0.7.12" +tokio-util = { version = "0.7.12", features = ["compat"] } openidconnect = { version = "4.0.0", default-features = false, features = ["accept-rfc3339-timestamps"] } encoding_rs = "0.8.35" regex = "1" @@ -96,7 +97,7 @@ tokio = { version = "1", features = ["rt", "time", "test-util"] } [build-dependencies] awc = { version = "3", features = ["rustls-0_23-webpki-roots"] } -rustls = "0.23" +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } actix-rt = "2.8" libflate = "2" futures-util = "0.3.21" diff --git a/build.rs b/build.rs index 6f9f4a96..804cdb7f 100644 --- a/build.rs +++ b/build.rs @@ -12,7 +12,7 @@ use std::time::Duration; #[actix_rt::main] async fn main() { - rustls::crypto::aws_lc_rs::default_provider() + rustls::crypto::ring::default_provider() .install_default() .unwrap(); diff --git a/src/filesystem.rs b/src/filesystem.rs index 92a2410a..5fe82a72 100644 --- a/src/filesystem.rs +++ b/src/filesystem.rs @@ -1,7 +1,7 @@ use crate::webserver::ErrorWithStatus; use crate::webserver::database::SupportedDatabase; -use crate::webserver::{Database, StatusCodeResultExt, make_placeholder}; use crate::webserver::database::{DbParam, driver::DbValue}; +use crate::webserver::{Database, StatusCodeResultExt, make_placeholder}; use crate::{AppState, TEMPLATES_DIR}; use anyhow::Context; use chrono::{DateTime, Utc}; @@ -267,9 +267,9 @@ impl DbFsQueries { log::debug!("Initializing database filesystem queries"); Self::check_table_available(db).await?; Ok(Self { - was_modified: Self::make_was_modified_query(db).await?, - read_file: Self::make_read_file_query(db).await?, - exists: Self::make_exists_query(db).await?, + was_modified: Self::make_was_modified_query(db), + read_file: Self::make_read_file_query(db), + exists: Self::make_exists_query(db), }) } @@ -284,31 +284,31 @@ impl DbFsQueries { Ok(()) } - async fn make_was_modified_query(db: &Database) -> anyhow::Result { + fn make_was_modified_query(db: &Database) -> String { let was_modified_query = format!( "SELECT 1 from sqlpage_files WHERE last_modified >= {} AND path = {}", make_placeholder(db.info.kind, 1), make_placeholder(db.info.kind, 2) ); log::debug!("Preparing the database filesystem was_modified_query: {was_modified_query}"); - Ok(was_modified_query) + was_modified_query } - async fn make_read_file_query(db: &Database) -> anyhow::Result { + fn make_read_file_query(db: &Database) -> String { let read_file_query = format!( "SELECT contents from sqlpage_files WHERE path = {}", make_placeholder(db.info.kind, 1), ); log::debug!("Preparing the database filesystem read_file_query: {read_file_query}"); - Ok(read_file_query) + read_file_query } - async fn make_exists_query(db: &Database) -> anyhow::Result { + fn make_exists_query(db: &Database) -> String { let exists_query = format!( "SELECT 1 from sqlpage_files WHERE path = {}", make_placeholder(db.info.kind, 1), ); - Ok(exists_query) + exists_query } async fn file_modified_since_in_db( @@ -351,26 +351,26 @@ impl DbFsQueries { log::debug!("Reading file {} from the database", path.display()); let mut conn = app_state.db.connection.acquire().await?; conn.fetch_optional( - &self.read_file, - &[DbParam::Text(path.display().to_string())], - ) - .await - .map_err(anyhow::Error::from) - .and_then(|row| { - if let Some(row) = row { - match row.values.first() { - Some(DbValue::Bytes(bytes)) => Ok(bytes.clone()), - Some(DbValue::Text(text)) => Ok(text.as_bytes().to_vec()), - _ => Ok(Vec::new()), - } - } else { - Err(ErrorWithStatus { - status: actix_web::http::StatusCode::NOT_FOUND, - } - .into()) + &self.read_file, + &[DbParam::Text(path.display().to_string())], + ) + .await + .map_err(anyhow::Error::from) + .and_then(|row| { + if let Some(row) = row { + match row.values.first() { + Some(DbValue::Bytes(bytes)) => Ok(bytes.clone()), + Some(DbValue::Text(text)) => Ok(text.as_bytes().to_vec()), + _ => Ok(Vec::new()), } - }) - .with_context(|| format!("Unable to read {} from the database", path.display())) + } else { + Err(ErrorWithStatus { + status: actix_web::http::StatusCode::NOT_FOUND, + } + .into()) + } + }) + .with_context(|| format!("Unable to read {} from the database", path.display())) } async fn file_exists(&self, app_state: &AppState, path: &Path) -> anyhow::Result { @@ -412,9 +412,9 @@ async fn test_sql_file_read_utf8() -> anyhow::Result<()> { let create_table_sql = DbFsQueries::get_create_table_sql(state.db.info.database_type); let db = &state.db; - let conn = &db.connection; let mut conn = db.connection.acquire().await?; - conn.execute_command("DROP TABLE IF EXISTS sqlpage_files", &[]).await?; + conn.execute_command("DROP TABLE IF EXISTS sqlpage_files", &[]) + .await?; log::debug!("Creating table sqlpage_files: {create_table_sql}"); conn.execute_command(create_table_sql, &[]).await?; @@ -425,13 +425,13 @@ async fn test_sql_file_read_utf8() -> anyhow::Result<()> { make_placeholder(dbms, 2) ); conn.execute_command( - &insert_sql, - &[ - DbParam::Text("unit test file.txt".into()), - DbParam::Bytes("Héllö world! 😀".as_bytes().to_vec()), - ], - ) - .await?; + &insert_sql, + &[ + DbParam::Text("unit test file.txt".into()), + DbParam::Bytes("Héllö world! 😀".as_bytes().to_vec()), + ], + ) + .await?; let fs = FileSystem::init("/", db).await; let actual = fs diff --git a/src/lib.rs b/src/lib.rs index d0c13b3f..7600f74b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -37,7 +37,7 @@ //! When processing a request, `SQLPage`: //! //! 1. Parses the SQL using sqlparser-rs. Once a SQL file is parsed, it is cached for later reuse. -//! 2. Executes queries through sqlx. +//! 2. Executes queries through native database drivers. //! 3. Finds the requested component's handlebars template in the database or in the filesystem. //! 4. Maps results to the component template, using handlebars-rs. //! 5. Streams rendered HTML to the client. diff --git a/src/telemetry_metrics.rs b/src/telemetry_metrics.rs index 56d7993b..7748aaac 100644 --- a/src/telemetry_metrics.rs +++ b/src/telemetry_metrics.rs @@ -1,8 +1,8 @@ +use crate::webserver::database::DbPool; use opentelemetry::global; use opentelemetry::metrics::{Histogram, ObservableGauge}; use opentelemetry_semantic_conventions::attribute as otel; use opentelemetry_semantic_conventions::metric as otel_metric; -use crate::webserver::database::DbPool; pub struct TelemetryMetrics { pub http_request_duration: Histogram, diff --git a/src/webserver/database/connect.rs b/src/webserver/database/connect.rs index 8ee8d046..92093551 100644 --- a/src/webserver/database/connect.rs +++ b/src/webserver/database/connect.rs @@ -23,8 +23,8 @@ impl Database { } log::debug!("Connecting to a {db_kind:?} database on {database_url}"); let on_connect_sql = read_connection_handler(config, ON_CONNECT_FILE); - let _on_reset_sql = read_connection_handler(config, ON_RESET_FILE); - if _on_reset_sql.is_some() { + let on_reset_sql = read_connection_handler(config, ON_RESET_FILE); + if on_reset_sql.is_some() { log::warn!( "{ON_RESET_FILE} is currently ignored by the native driver pool because connections are not reused yet" ); @@ -111,7 +111,10 @@ fn read_connection_handler(config: &AppConfig, file_name: &str) -> Option { log::trace!("The custom SQL connection handler is:\n{sql}"); diff --git a/src/webserver/database/csv_import.rs b/src/webserver/database/csv_import.rs index 78f3b93c..66d047a4 100644 --- a/src/webserver/database/csv_import.rs +++ b/src/webserver/database/csv_import.rs @@ -324,7 +324,9 @@ async fn test_end_to_end() { uploaded_file: "my_file.csv".into(), } ); - let db = crate::webserver::Database::init(&test_config()).await.unwrap(); + let db = crate::webserver::Database::init(&test_config()) + .await + .unwrap(); let mut conn = db.connection.acquire().await.unwrap(); conn.execute_command("CREATE TABLE my_table (col1 TEXT, col2 TEXT)", &[]) .await @@ -339,8 +341,14 @@ async fn test_end_to_end() { .into_iter() .filter_map(|item| match item { super::driver::DbStatementResult::Row(row) => Some(( - match &row.values[0] { super::driver::DbValue::Text(s) => s.clone(), other => format!("{other:?}") }, - match &row.values[1] { super::driver::DbValue::Text(s) => s.clone(), other => format!("{other:?}") }, + match &row.values[0] { + super::driver::DbValue::Text(s) => s.clone(), + other => format!("{other:?}"), + }, + match &row.values[1] { + super::driver::DbValue::Text(s) => s.clone(), + other => format!("{other:?}"), + }, )), super::driver::DbStatementResult::Finished => None, }) diff --git a/src/webserver/database/driver.rs b/src/webserver/database/driver.rs index 763e008f..258c9f08 100644 --- a/src/webserver/database/driver.rs +++ b/src/webserver/database/driver.rs @@ -1,5 +1,8 @@ +use std::borrow::Cow; use std::fmt; use std::path::PathBuf; +use std::pin::Pin; +use std::sync::Mutex; use std::sync::{ Arc, atomic::{AtomicU32, Ordering}, @@ -7,8 +10,14 @@ use std::sync::{ use std::time::Duration; use anyhow::Context; +use chrono::{Datelike, Timelike}; +use futures_util::stream::Stream; +use mysql_async::prelude::Queryable; +use odbc_api::parameter::{InputParameter, VarCharBox, WithDataType}; +use odbc_api::{ConnectionOptions, Cursor, IntoParameter, ResultSetMetadata}; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tokio_rusqlite::rusqlite; +use tokio_util::compat::{Compat, TokioAsyncWriteCompatExt}; use crate::app_config::AppConfig; @@ -24,6 +33,8 @@ pub enum DbKind { #[derive(Debug, Clone)] pub enum DbParam { Null, + Bool(bool), + Integer(i64), Text(String), Bytes(Vec), Timestamp(chrono::DateTime), @@ -38,6 +49,7 @@ impl From> for DbParam { #[derive(Debug, Clone)] pub enum DbValue { Null, + Bool(bool), Integer(i64), Real(f64), Text(String), @@ -105,6 +117,31 @@ impl From for DbError { } } +impl From for DbError { + fn from(error: tokio_postgres::Error) -> Self { + db_error(error) + } +} + +impl From for DbError { + fn from(error: mysql_async::Error) -> Self { + db_error(error) + } +} + +impl From for DbError { + fn from(error: tiberius::error::Error) -> Self { + db_error(error) + } +} + +fn db_error(error: impl std::error::Error) -> DbError { + DbError::Database { + message: error.to_string(), + offset: None, + } +} + #[derive(Clone)] pub struct DbPool { inner: Arc, @@ -119,6 +156,7 @@ struct DbPoolInner { active: AtomicU32, on_connect_sql: Option>, sqlite_extensions: Vec, + idle_sqlite: Mutex>, } pub struct DbConnection { @@ -129,10 +167,22 @@ pub struct DbConnection { enum NativeConnection { Sqlite(tokio_rusqlite::Connection), + Postgres(tokio_postgres::Client), + MySql(mysql_async::Conn), + Mssql(Box>>), + Odbc(odbc_api::Connection<'static>), + Closed, } impl Drop for DbConnection { fn drop(&mut self) { + let inner = std::mem::replace(&mut self.inner, NativeConnection::Closed); + if let NativeConnection::Sqlite(conn) = inner + && let Ok(mut idle) = self.pool.idle_sqlite.lock() + && idle.is_none() + { + *idle = Some(conn); + } self.pool.active.fetch_sub(1, Ordering::Relaxed); } } @@ -144,18 +194,26 @@ impl DbPool { max_size: u32, on_connect_sql: Option, ) -> Self { + let url = if kind == DbKind::Sqlite { + normalize_sqlite_url(&config.database_url) + } else { + config.database_url.clone() + }; Self { inner: Arc::new(DbPoolInner { - url: config.database_url.clone(), + url, kind, max_size, acquire_timeout: Duration::from_secs_f64( config.database_connection_acquire_timeout_seconds, ), - semaphore: Arc::new(Semaphore::new(usize::try_from(max_size).unwrap_or(usize::MAX))), + semaphore: Arc::new(Semaphore::new( + usize::try_from(max_size).unwrap_or(usize::MAX), + )), active: AtomicU32::new(0), on_connect_sql: on_connect_sql.map(Arc::new), sqlite_extensions: config.sqlite_extensions.clone(), + idle_sqlite: Mutex::new(None), }), } } @@ -180,7 +238,7 @@ impl DbPool { self.inner.max_size } - pub async fn close(&self) {} + pub fn close(&self) {} pub async fn acquire(&self) -> Result { let permit = tokio::time::timeout( @@ -191,6 +249,16 @@ impl DbPool { .map_err(|_| DbError::PoolTimedOut)? .map_err(|_| DbError::PoolTimedOut)?; self.inner.active.fetch_add(1, Ordering::Relaxed); + if self.inner.kind == DbKind::Sqlite + && let Ok(mut idle) = self.inner.idle_sqlite.lock() + && let Some(conn) = idle.take() + { + return Ok(DbConnection { + inner: NativeConnection::Sqlite(conn), + pool: self.inner.clone(), + _permit: permit, + }); + } match self.inner.connect().await { Ok(inner) => Ok(DbConnection { inner, @@ -209,7 +277,10 @@ impl DbPoolInner { async fn connect(&self) -> Result { let mut conn = match self.kind { DbKind::Sqlite => NativeConnection::Sqlite(open_sqlite(&self.url).await?), - other => return Err(DbError::UnsupportedBackend(other)), + DbKind::Postgres => NativeConnection::Postgres(open_postgres(&self.url).await?), + DbKind::MySql => NativeConnection::MySql(open_mysql(&self.url).await?), + DbKind::Mssql => NativeConnection::Mssql(Box::new(open_mssql(&self.url).await?)), + DbKind::Odbc => NativeConnection::Odbc(open_odbc(&self.url)?), }; conn.configure(self).await?; if let Some(sql) = &self.on_connect_sql { @@ -228,6 +299,24 @@ impl DbConnection { pub async fn dbms_name(&mut self) -> Result { match &mut self.inner { NativeConnection::Sqlite(_) => Ok("SQLite".to_string()), + NativeConnection::Postgres(client) => { + let row = client.query_one("SELECT version()", &[]).await?; + let version: String = row.try_get(0)?; + if version.starts_with("PostgreSQL") { + Ok("PostgreSQL".to_string()) + } else { + Ok(version) + } + } + NativeConnection::MySql(_) => Ok("MySQL".to_string()), + NativeConnection::Mssql(_) => Ok("Microsoft SQL Server".to_string()), + NativeConnection::Odbc(conn) => { + conn.database_management_system_name().map_err(db_error) + } + NativeConnection::Closed => Err(DbError::Database { + message: "database connection is closed".to_string(), + offset: None, + }), } } @@ -239,15 +328,23 @@ impl DbConnection { self.inner.execute(sql, params).await } - pub async fn execute_command( - &mut self, - sql: &str, - params: &[DbParam], - ) -> Result<(), DbError> { + pub fn execute_stream<'a>( + &'a mut self, + sql: &'a str, + params: &'a [DbParam], + ) -> Pin> + 'a>> { + self.inner.execute_stream(sql, params) + } + + pub async fn execute_command(&mut self, sql: &str, params: &[DbParam]) -> Result<(), DbError> { let _ = self.execute(sql, params).await?; Ok(()) } + pub async fn execute_batch(&mut self, sql: &str) -> Result<(), DbError> { + self.inner.execute_batch(sql).await + } + pub async fn fetch_optional( &mut self, sql: &str, @@ -265,6 +362,9 @@ impl NativeConnection { async fn configure(&mut self, pool: &DbPoolInner) -> Result<(), DbError> { match self { Self::Sqlite(conn) => configure_sqlite(conn, &pool.sqlite_extensions).await, + Self::Postgres(_) | Self::MySql(_) | Self::Mssql(_) | Self::Odbc(_) | Self::Closed => { + Ok(()) + } } } @@ -275,6 +375,52 @@ impl NativeConnection { ) -> Result, DbError> { match self { Self::Sqlite(conn) => execute_sqlite(conn, sql, params).await, + Self::Postgres(client) => execute_postgres(client, sql, params).await, + Self::MySql(conn) => execute_mysql(conn, sql, params).await, + Self::Mssql(client) => execute_mssql(client, sql, params).await, + Self::Odbc(conn) => execute_odbc(conn, sql, params), + Self::Closed => Err(DbError::Database { + message: "database connection is closed".to_string(), + offset: None, + }), + } + } + + fn execute_stream<'a>( + &'a mut self, + sql: &'a str, + params: &'a [DbParam], + ) -> Pin> + 'a>> { + match self { + Self::Sqlite(conn) => stream_sqlite(conn, sql, params), + _ => Box::pin(async_stream::try_stream! { + for item in self.execute(sql, params).await? { + yield item; + } + }), + } + } + + async fn execute_batch(&mut self, sql: &str) -> Result<(), DbError> { + match self { + Self::Sqlite(conn) => execute_sqlite_batch(conn, sql).await, + Self::Postgres(client) => client.batch_execute(sql).await.map_err(Into::into), + Self::MySql(conn) => { + conn.query_drop(sql).await?; + Ok(()) + } + Self::Mssql(client) => { + client.simple_query(sql).await?.into_results().await?; + Ok(()) + } + Self::Odbc(conn) => { + let _ = conn.execute(sql, (), None).map_err(db_error)?; + Ok(()) + } + Self::Closed => Err(DbError::Database { + message: "database connection is closed".to_string(), + offset: None, + }), } } } @@ -298,6 +444,111 @@ async fn open_sqlite(url: &str) -> Result { } } +async fn open_postgres(url: &str) -> Result { + let (client, connection) = tokio_postgres::connect(url, tokio_postgres::NoTls).await?; + tokio::spawn(async move { + if let Err(error) = connection.await { + log::debug!("PostgreSQL connection task finished with error: {error}"); + } + }); + Ok(client) +} + +async fn open_mysql(url: &str) -> Result { + mysql_async::Conn::from_url(url).await.map_err(Into::into) +} + +async fn open_mssql(url: &str) -> Result>, DbError> { + let mut config = mssql_config_from_url(url)?; + config.trust_cert(); + let tcp = tokio::net::TcpStream::connect(config.get_addr()) + .await + .map_err(db_error)?; + tcp.set_nodelay(true).map_err(db_error)?; + tiberius::Client::connect(config, tcp.compat_write()) + .await + .map_err(Into::into) +} + +fn open_odbc(url: &str) -> Result, DbError> { + let conn_str = odbc_connection_string(url); + odbc_api::environment() + .map_err(db_error)? + .connect_with_connection_string(&conn_str, ConnectionOptions::default()) + .map_err(db_error) +} + +fn odbc_connection_string(url: &str) -> String { + let trimmed = url.trim().strip_prefix("odbc:").unwrap_or(url.trim()); + if trimmed.contains('=') { + trimmed.to_string() + } else { + format!("DSN={trimmed}") + } +} + +fn mssql_config_from_url(url: &str) -> Result { + if url.starts_with("jdbc:") { + return tiberius::Config::from_jdbc_string(url).map_err(Into::into); + } + if url.contains('=') { + return tiberius::Config::from_ado_string(url).map_err(Into::into); + } + + let without_scheme = url + .strip_prefix("mssql://") + .or_else(|| url.strip_prefix("sqlserver://")) + .ok_or_else(|| DbError::Database { + message: format!("not a SQL Server URL: {url}"), + offset: None, + })?; + let (authority, database) = without_scheme + .split_once('/') + .map_or((without_scheme, ""), |(authority, rest)| { + (authority, rest.split(['?', '#']).next().unwrap_or("")) + }); + let (credentials, host_port) = authority + .rsplit_once('@') + .map_or(("", authority), |(credentials, host_port)| { + (credentials, host_port) + }); + let (user, password) = credentials.split_once(':').unwrap_or((credentials, "")); + let (host, port) = parse_host_port(host_port); + + let mut config = tiberius::Config::new(); + config.host(decode_url_part(host)?); + if let Some(port) = port { + config.port(port); + } + if !database.is_empty() { + config.database(decode_url_part(database)?); + } + if !user.is_empty() { + config.authentication(tiberius::AuthMethod::sql_server( + decode_url_part(user)?, + decode_url_part(password)?, + )); + } + Ok(config) +} + +fn parse_host_port(host_port: &str) -> (&str, Option) { + let Some((host, port)) = host_port.rsplit_once(':') else { + return (host_port, None); + }; + match port.parse::() { + Ok(port) => (host, Some(port)), + Err(_) => (host_port, None), + } +} + +fn decode_url_part(value: &str) -> Result { + percent_encoding::percent_decode_str(value) + .decode_utf8() + .map(std::borrow::Cow::into_owned) + .map_err(db_error) +} + fn sqlite_path_from_url(url: &str) -> anyhow::Result { let Some(rest) = url.strip_prefix("sqlite:") else { anyhow::bail!("not a sqlite URL: {url}"); @@ -306,7 +557,25 @@ fn sqlite_path_from_url(url: &str) -> anyhow::Result { let decoded = percent_encoding::percent_decode_str(rest) .decode_utf8() .with_context(|| format!("invalid percent encoding in sqlite URL {url:?}"))?; - Ok(decoded.into_owned()) + let decoded = decoded.into_owned(); + if decoded.contains('?') && !decoded.starts_with("file:") { + Ok(format!("file:{decoded}")) + } else { + Ok(decoded) + } +} + +fn normalize_sqlite_url(url: &str) -> String { + static MEMORY_DB_ID: AtomicU32 = AtomicU32::new(0); + let Ok(path) = sqlite_path_from_url(url) else { + return url.to_string(); + }; + if path == ":memory:" || path.starts_with("file::memory:") { + let id = MEMORY_DB_ID.fetch_add(1, Ordering::Relaxed); + format!("sqlite://file:sqlpage_memory_{id}?mode=memory&cache=shared") + } else { + url.to_string() + } } async fn configure_sqlite( @@ -315,9 +584,7 @@ async fn configure_sqlite( ) -> Result<(), DbError> { let extensions = extensions.to_vec(); conn.call(move |conn| { - conn.create_collation("NOCASE", |a, b| { - a.to_lowercase().cmp(&b.to_lowercase()) - })?; + conn.create_collation("NOCASE", |a, b| a.to_lowercase().cmp(&b.to_lowercase()))?; conn.create_scalar_function( "upper", 1, @@ -366,7 +633,12 @@ async fn execute_sqlite( .collect::>(); let column_count = stmt.column_count(); if column_count == 0 { - stmt.execute(rusqlite::params_from_iter(values))?; + if values.is_empty() { + drop(stmt); + conn.execute_batch(&sql)?; + } else { + stmt.execute(rusqlite::params_from_iter(values))?; + } return Ok(vec![DbStatementResult::Finished]); } let columns = (0..column_count) @@ -394,12 +666,589 @@ async fn execute_sqlite( .map_err(Into::into) } +fn stream_sqlite<'a>( + conn: &'a tokio_rusqlite::Connection, + sql: &str, + params: &[DbParam], +) -> Pin> + 'a>> { + let sql = sql.to_string(); + let params = params.to_vec(); + Box::pin(async_stream::try_stream! { + let (tx, mut rx) = tokio::sync::mpsc::channel(32); + let call = conn.call(move |conn| { + let mut stmt = conn.prepare(&sql).map_err(DbError::from)?; + let values = params + .into_iter() + .map(sqlite_value_from_param) + .collect::>(); + let column_count = stmt.column_count(); + if column_count == 0 { + if values.is_empty() { + drop(stmt); + conn.execute_batch(&sql).map_err(DbError::from)?; + } else { + stmt.execute(rusqlite::params_from_iter(values)) + .map_err(DbError::from)?; + } + let _ = tx.blocking_send(DbStatementResult::Finished); + return Ok(()); + } + let columns = (0..column_count) + .map(|idx| DbColumn { + name: stmt.column_name(idx).unwrap_or("").to_string(), + type_name: None, + }) + .collect::>(); + let mut rows = stmt + .query(rusqlite::params_from_iter(values)) + .map_err(DbError::from)?; + while let Some(row) = rows.next().map_err(DbError::from)? { + let mut values = Vec::with_capacity(column_count); + for idx in 0..column_count { + values.push(sqlite_value(row.get_ref(idx).map_err(DbError::from)?)); + } + if tx + .blocking_send(DbStatementResult::Row(DbRow { + columns: columns.clone(), + values, + kind: DbKind::Sqlite, + })) + .is_err() + { + return Ok(()); + } + } + Ok(()) + }); + tokio::pin!(call); + let mut call_finished = false; + let mut call_result = None; + loop { + tokio::select! { + result = &mut call, if !call_finished => { + call_finished = true; + call_result = Some(sqlite_call_result(result)); + } + maybe_item = rx.recv() => { + match maybe_item { + Some(item) => yield item, + None => break, + } + } + } + } + if !call_finished { + call_result = Some(sqlite_call_result(call.await)); + } + if let Some(result) = call_result { + result?; + } + }) +} + +fn sqlite_call_result(result: Result<(), tokio_rusqlite::Error>) -> Result<(), DbError> { + match result { + Ok(()) => Ok(()), + Err(tokio_rusqlite::Error::Error(error)) => Err(error), + Err(error) => Err(DbError::Database { + message: error.to_string(), + offset: None, + }), + } +} + +async fn execute_sqlite_batch(conn: &tokio_rusqlite::Connection, sql: &str) -> Result<(), DbError> { + let sql = sql.to_string(); + conn.call(move |conn| conn.execute_batch(&sql)) + .await + .map_err(Into::into) +} + +async fn execute_postgres( + client: &tokio_postgres::Client, + sql: &str, + params: &[DbParam], +) -> Result, DbError> { + let params = params.iter().map(PgParam::from).collect::>(); + let param_refs = params + .iter() + .map(|param| param as &(dyn tokio_postgres::types::ToSql + Sync)) + .collect::>(); + let rows = client.query(sql, ¶m_refs).await?; + if rows.is_empty() { + return Ok(vec![DbStatementResult::Finished]); + } + Ok(rows + .into_iter() + .map(|row| DbStatementResult::Row(postgres_row(&row))) + .collect()) +} + +#[derive(Debug)] +enum PgParam { + Null(Option), + Bool(bool), + Integer(i64), + Text(String), + Bytes(Vec), + Timestamp(chrono::DateTime), +} + +impl From<&DbParam> for PgParam { + fn from(value: &DbParam) -> Self { + match value { + DbParam::Null => Self::Null(None), + DbParam::Bool(value) => Self::Bool(*value), + DbParam::Integer(value) => Self::Integer(*value), + DbParam::Text(value) => Self::Text(value.clone()), + DbParam::Bytes(value) => Self::Bytes(value.clone()), + DbParam::Timestamp(value) => Self::Timestamp(*value), + } + } +} + +impl tokio_postgres::types::ToSql for PgParam { + fn to_sql( + &self, + ty: &tokio_postgres::types::Type, + out: &mut bytes::BytesMut, + ) -> Result> + where + Self: Sized, + { + match self { + Self::Null(value) => value.to_sql(ty, out), + Self::Bool(value) => value.to_sql(ty, out), + Self::Integer(value) => value.to_sql(ty, out), + Self::Text(value) => value.to_sql(ty, out), + Self::Bytes(value) => value.to_sql(ty, out), + Self::Timestamp(value) => value.to_sql(ty, out), + } + } + + fn accepts(_ty: &tokio_postgres::types::Type) -> bool + where + Self: Sized, + { + true + } + + tokio_postgres::types::to_sql_checked!(); +} + +fn postgres_row(row: &tokio_postgres::Row) -> DbRow { + let columns = row + .columns() + .iter() + .map(|column| DbColumn { + name: column.name().to_string(), + type_name: Some(column.type_().name().to_string()), + }) + .collect::>(); + let values = row + .columns() + .iter() + .enumerate() + .map(|(idx, column)| postgres_value(row, idx, column.type_())) + .collect::>(); + DbRow { + columns, + values, + kind: DbKind::Postgres, + } +} + +fn postgres_value( + row: &tokio_postgres::Row, + idx: usize, + ty: &tokio_postgres::types::Type, +) -> DbValue { + use tokio_postgres::types::Type; + if row.try_get::<_, Option>(idx).ok() == Some(None) { + return DbValue::Null; + } + match *ty { + Type::BOOL => row + .try_get::<_, bool>(idx) + .map_or(DbValue::Null, DbValue::Bool), + Type::INT2 => row + .try_get::<_, i16>(idx) + .map_or(DbValue::Null, |value| DbValue::Integer(i64::from(value))), + Type::INT4 => row + .try_get::<_, i32>(idx) + .map_or(DbValue::Null, |value| DbValue::Integer(i64::from(value))), + Type::INT8 => row + .try_get::<_, i64>(idx) + .map_or(DbValue::Null, DbValue::Integer), + Type::FLOAT4 => row + .try_get::<_, f32>(idx) + .map_or(DbValue::Null, |value| DbValue::Real(f64::from(value))), + Type::FLOAT8 => row + .try_get::<_, f64>(idx) + .map_or(DbValue::Null, DbValue::Real), + Type::BYTEA => row + .try_get::<_, Vec>(idx) + .map_or(DbValue::Null, DbValue::Bytes), + Type::TIMESTAMPTZ => row + .try_get::<_, chrono::DateTime>(idx) + .map_or(DbValue::Null, |value| DbValue::Text(value.to_rfc3339())), + Type::TIMESTAMP => row + .try_get::<_, chrono::NaiveDateTime>(idx) + .map_or(DbValue::Null, |value| DbValue::Text(value.to_string())), + Type::DATE => row + .try_get::<_, chrono::NaiveDate>(idx) + .map_or(DbValue::Null, |value| DbValue::Text(value.to_string())), + Type::JSON | Type::JSONB => row + .try_get::<_, serde_json::Value>(idx) + .map_or(DbValue::Null, |value| DbValue::Text(value.to_string())), + Type::UUID => row + .try_get::<_, uuid::Uuid>(idx) + .map_or(DbValue::Null, |value| DbValue::Text(value.to_string())), + _ => row + .try_get::<_, String>(idx) + .map_or(DbValue::Null, DbValue::Text), + } +} + +async fn execute_mysql( + conn: &mut mysql_async::Conn, + sql: &str, + params: &[DbParam], +) -> Result, DbError> { + let values = params + .iter() + .map(mysql_value_from_param) + .collect::>(); + if values.is_empty() { + collect_mysql_results(conn.query_iter(sql).await?).await + } else { + collect_mysql_results( + conn.exec_iter(sql, mysql_async::Params::Positional(values)) + .await?, + ) + .await + } +} + +async fn collect_mysql_results

( + mut query_result: mysql_async::QueryResult<'_, 'static, P>, +) -> Result, DbError> +where + P: mysql_async::prelude::Protocol, +{ + let mut result = Vec::new(); + loop { + let columns = query_result + .columns_ref() + .iter() + .map(|column| DbColumn { + name: column.name_str().into_owned(), + type_name: Some(format!("{:?}", column.column_type())), + }) + .collect::>(); + while let Some(row) = query_result.next().await? { + let values = row + .unwrap_raw() + .into_iter() + .map(mysql_value) + .collect::>(); + result.push(DbStatementResult::Row(DbRow { + columns: columns.clone(), + values, + kind: DbKind::MySql, + })); + } + if query_result.is_empty() { + break; + } + } + query_result.drop_result().await?; + if result.is_empty() { + result.push(DbStatementResult::Finished); + } + Ok(result) +} + +fn mysql_value_from_param(param: &DbParam) -> mysql_async::Value { + match param { + DbParam::Null => mysql_async::Value::NULL, + DbParam::Bool(value) => mysql_async::Value::Int(i64::from(*value)), + DbParam::Integer(value) => mysql_async::Value::Int(*value), + DbParam::Text(value) => mysql_async::Value::Bytes(value.clone().into_bytes()), + DbParam::Bytes(value) => mysql_async::Value::Bytes(value.clone()), + DbParam::Timestamp(value) => { + let value = value.naive_utc(); + mysql_async::Value::Date( + u16::try_from(value.year()).unwrap_or_default(), + u8::try_from(value.month()).unwrap_or_default(), + u8::try_from(value.day()).unwrap_or_default(), + u8::try_from(value.hour()).unwrap_or_default(), + u8::try_from(value.minute()).unwrap_or_default(), + u8::try_from(value.second()).unwrap_or_default(), + value.nanosecond() / 1000, + ) + } + } +} + +fn mysql_value(value: Option) -> DbValue { + use mysql_async::Value; + match value { + None | Some(Value::NULL) => DbValue::Null, + Some(Value::Int(value)) => DbValue::Integer(value), + Some(Value::UInt(value)) => { + i64::try_from(value).map_or_else(|_| DbValue::Text(value.to_string()), DbValue::Integer) + } + Some(Value::Float(value)) => DbValue::Real(f64::from(value)), + Some(Value::Double(value)) => DbValue::Real(value), + Some(Value::Bytes(bytes)) => String::from_utf8(bytes) + .map_or_else(|err| DbValue::Bytes(err.into_bytes()), DbValue::Text), + Some(Value::Date(year, month, day, hour, minute, second, micros)) => DbValue::Text( + format!("{year:04}-{month:02}-{day:02} {hour:02}:{minute:02}:{second:02}.{micros:06}"), + ), + Some(Value::Time(negative, days, hours, minutes, seconds, micros)) => { + DbValue::Text(format!( + "{}{:03}:{minutes:02}:{seconds:02}.{micros:06}", + if negative { "-" } else { "" }, + days * 24 + u32::from(hours) + )) + } + } +} + +async fn execute_mssql( + client: &mut tiberius::Client>, + sql: &str, + params: &[DbParam], +) -> Result, DbError> { + let params = params.iter().map(MssqlParam::from).collect::>(); + let param_refs = params + .iter() + .map(|param| param as &dyn tiberius::ToSql) + .collect::>(); + let rows = client.query(sql, ¶m_refs).await?.into_results().await?; + let mut result = Vec::new(); + for set in rows { + for row in set { + result.push(DbStatementResult::Row(mssql_row(row))); + } + } + if result.is_empty() { + result.push(DbStatementResult::Finished); + } + Ok(result) +} + +enum MssqlParam { + Text(Option), + Bool(bool), + Integer(i64), + Bytes(Vec), + Timestamp(chrono::NaiveDateTime), +} + +impl From<&DbParam> for MssqlParam { + fn from(value: &DbParam) -> Self { + match value { + DbParam::Null => Self::Text(None), + DbParam::Bool(value) => Self::Bool(*value), + DbParam::Integer(value) => Self::Integer(*value), + DbParam::Text(value) => Self::Text(Some(value.clone())), + DbParam::Bytes(value) => Self::Bytes(value.clone()), + DbParam::Timestamp(value) => Self::Timestamp(value.naive_utc()), + } + } +} + +impl tiberius::ToSql for MssqlParam { + fn to_sql(&self) -> tiberius::ColumnData<'_> { + match self { + Self::Text(value) => tiberius::ColumnData::String(value.as_ref().map(Cow::from)), + Self::Bool(value) => tiberius::ColumnData::Bit(Some(*value)), + Self::Integer(value) => tiberius::ColumnData::I64(Some(*value)), + Self::Bytes(value) => tiberius::ColumnData::Binary(Some(Cow::from(value))), + Self::Timestamp(value) => value.to_sql(), + } + } +} + +fn mssql_row(row: tiberius::Row) -> DbRow { + let columns = row + .columns() + .iter() + .map(|column| DbColumn { + name: column.name().to_string(), + type_name: Some(format!("{:?}", column.column_type())), + }) + .collect::>(); + let values = row.into_iter().map(mssql_value).collect::>(); + DbRow { + columns, + values, + kind: DbKind::Mssql, + } +} + +fn mssql_value(value: tiberius::ColumnData<'static>) -> DbValue { + use tiberius::ColumnData; + match value { + ColumnData::U8(value) => { + value.map_or(DbValue::Null, |value| DbValue::Integer(i64::from(value))) + } + ColumnData::I16(value) => { + value.map_or(DbValue::Null, |value| DbValue::Integer(i64::from(value))) + } + ColumnData::I32(value) => { + value.map_or(DbValue::Null, |value| DbValue::Integer(i64::from(value))) + } + ColumnData::I64(value) => value.map_or(DbValue::Null, DbValue::Integer), + ColumnData::F32(value) => { + value.map_or(DbValue::Null, |value| DbValue::Real(f64::from(value))) + } + ColumnData::F64(value) => value.map_or(DbValue::Null, DbValue::Real), + ColumnData::Bit(value) => value.map_or(DbValue::Null, DbValue::Bool), + ColumnData::String(value) => { + value.map_or(DbValue::Null, |value| DbValue::Text(value.into_owned())) + } + ColumnData::Binary(value) => { + value.map_or(DbValue::Null, |value| DbValue::Bytes(value.into_owned())) + } + ColumnData::Guid(value) => { + value.map_or(DbValue::Null, |value| DbValue::Text(value.to_string())) + } + ColumnData::Numeric(value) => { + value.map_or(DbValue::Null, |value| DbValue::Text(value.to_string())) + } + other => DbValue::Text(format!("{other:?}")), + } +} + +fn execute_odbc( + conn: &mut odbc_api::Connection<'static>, + sql: &str, + params: &[DbParam], +) -> Result, DbError> { + let parameters = OdbcParameters::from_params(params); + let cursor = if parameters.is_empty() { + conn.execute(sql, (), None).map_err(db_error)? + } else { + conn.execute(sql, parameters.as_slice(), None) + .map_err(db_error)? + }; + let Some(cursor) = cursor else { + return Ok(vec![DbStatementResult::Finished]); + }; + collect_odbc_rows(cursor) +} + +struct OdbcParameters { + values: Vec>, +} + +impl OdbcParameters { + fn from_params(params: &[DbParam]) -> Self { + Self { + values: params.iter().map(odbc_parameter).collect(), + } + } + + fn is_empty(&self) -> bool { + self.values.is_empty() + } + + fn as_slice(&self) -> &[Box] { + &self.values + } +} + +fn odbc_parameter(param: &DbParam) -> Box { + match param { + DbParam::Null => Box::new(WithDataType::new( + VarCharBox::null(), + odbc_api::DataType::Varchar { length: None }, + )), + DbParam::Text(value) => Box::new(value.clone().into_parameter()), + DbParam::Bool(value) => Box::new(i32::from(*value).into_parameter()), + DbParam::Integer(value) => Box::new((*value).into_parameter()), + DbParam::Bytes(value) => Box::new(value.clone().into_parameter()), + DbParam::Timestamp(value) => Box::new( + WithDataType::new( + odbc_api::Nullable::new(odbc_api::sys::Timestamp { + year: i16::try_from(value.year()).unwrap_or_default(), + month: u16::try_from(value.month()).unwrap_or_default(), + day: u16::try_from(value.day()).unwrap_or_default(), + hour: u16::try_from(value.hour()).unwrap_or_default(), + minute: u16::try_from(value.minute()).unwrap_or_default(), + second: u16::try_from(value.second()).unwrap_or_default(), + fraction: value.nanosecond(), + }), + odbc_api::DataType::Timestamp { precision: 6 }, + ) + .into_parameter(), + ), + } +} + +fn collect_odbc_rows(mut cursor: C) -> Result, DbError> +where + C: Cursor + ResultSetMetadata, +{ + let column_count = cursor.num_result_cols().map_err(db_error)?; + let column_count = usize::try_from(column_count).map_err(db_error)?; + let mut columns = Vec::with_capacity(column_count); + for idx in 0..column_count { + let mut description = odbc_api::ColumnDescription::default(); + let column_number = u16::try_from(idx + 1).map_err(db_error)?; + cursor + .describe_col(column_number, &mut description) + .map_err(db_error)?; + columns.push(DbColumn { + name: description + .name_to_string() + .unwrap_or_else(|_| format!("col{idx}")), + type_name: Some(format!("{:?}", description.data_type)), + }); + } + let mut result = Vec::new(); + while let Some(mut row) = cursor.next_row().map_err(db_error)? { + let mut values = Vec::with_capacity(column_count); + for idx in 0..column_count { + let column_number = u16::try_from(idx + 1).map_err(db_error)?; + let mut value = Vec::new(); + if row + .get_binary(column_number, &mut value) + .map_err(db_error)? + { + values.push( + String::from_utf8(value) + .map_or_else(|err| DbValue::Bytes(err.into_bytes()), DbValue::Text), + ); + } else { + values.push(DbValue::Null); + } + } + result.push(DbStatementResult::Row(DbRow { + columns: columns.clone(), + values, + kind: DbKind::Odbc, + })); + } + if result.is_empty() { + result.push(DbStatementResult::Finished); + } + Ok(result) +} + fn sqlite_value_from_param(param: DbParam) -> rusqlite::types::Value { match param { DbParam::Null => rusqlite::types::Value::Null, + DbParam::Bool(value) => rusqlite::types::Value::Integer(i64::from(value)), + DbParam::Integer(value) => rusqlite::types::Value::Integer(value), DbParam::Text(s) => rusqlite::types::Value::Text(s), DbParam::Bytes(bytes) => rusqlite::types::Value::Blob(bytes), - DbParam::Timestamp(ts) => rusqlite::types::Value::Text(ts.to_rfc3339()), + DbParam::Timestamp(ts) => { + rusqlite::types::Value::Text(ts.naive_utc().format("%F %T").to_string()) + } } } @@ -414,3 +1263,41 @@ fn sqlite_value(value: rusqlite::types::ValueRef<'_>) -> DbValue { rusqlite::types::ValueRef::Blob(b) => DbValue::Bytes(b.to_vec()), } } + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::StreamExt; + use std::time::Duration; + + #[tokio::test] + async fn sqlite_stream_yields_first_row_before_statement_finishes() { + let conn = tokio_rusqlite::Connection::open_in_memory().await.unwrap(); + conn.call(|conn| { + conn.create_scalar_function( + "wait_ms", + 1, + rusqlite::functions::FunctionFlags::SQLITE_UTF8, + |ctx| { + let millis = ctx.get::(0)?; + std::thread::sleep(Duration::from_millis(millis.cast_unsigned())); + Ok(millis) + }, + ) + }) + .await + .unwrap(); + + let mut stream = stream_sqlite(&conn, "SELECT 1 AS n UNION ALL SELECT wait_ms(500)", &[]); + + let first = tokio::time::timeout(Duration::from_millis(100), stream.next()) + .await + .expect("SQLite should yield the first row before the second row has been computed") + .unwrap() + .unwrap(); + let DbStatementResult::Row(row) = first else { + panic!("expected first SQLite stream item to be a row"); + }; + assert!(matches!(row.values.first(), Some(DbValue::Integer(1)))); + } +} diff --git a/src/webserver/database/error_highlighting.rs b/src/webserver/database/error_highlighting.rs index 95045c58..463d1572 100644 --- a/src/webserver/database/error_highlighting.rs +++ b/src/webserver/database/error_highlighting.rs @@ -3,8 +3,8 @@ use std::{ path::{Path, PathBuf}, }; -use super::sql::{SourceSpan, StmtWithParams}; use super::DbError; +use super::sql::{SourceSpan, StmtWithParams}; #[derive(Debug)] struct NiceDatabaseError { @@ -44,7 +44,11 @@ impl std::fmt::Display for NiceDatabaseError { self.source_file.display(), self.db_err )?; - if let DbError::Database { offset: Some(offset), .. } = &self.db_err { + if let DbError::Database { + offset: Some(offset), + .. + } = &self.db_err + { let mut offset = *offset; for line in self.query.lines() { if offset > line.len() { @@ -94,11 +98,7 @@ impl std::error::Error for NicePositionedError { /// Display a database error without any position information #[must_use] -pub fn display_db_error( - source_file: &Path, - query: &str, - db_err: DbError, -) -> anyhow::Error { +pub fn display_db_error(source_file: &Path, query: &str, db_err: DbError) -> anyhow::Error { anyhow::Error::new(NiceDatabaseError { source_file: source_file.to_path_buf(), db_err, diff --git a/src/webserver/database/execute_queries.rs b/src/webserver/database/execute_queries.rs index 9fd48428..260e61cc 100644 --- a/src/webserver/database/execute_queries.rs +++ b/src/webserver/database/execute_queries.rs @@ -20,8 +20,8 @@ use crate::webserver::http_request_info::ExecutionContext; use crate::webserver::request_variables::SetVariablesMap; use crate::webserver::single_or_vec::SingleOrVec; +use super::driver::{DbParam, DbStatementResult}; use super::syntax_tree::{StmtParam, extract_req_param}; -use super::driver::{DbStatementResult, DbParam}; use super::{Database, DbConnection, DbItem}; pub type DbConn = Option; @@ -129,6 +129,7 @@ fn create_db_query_span( (span, operation_name) } +#[allow(clippy::too_many_lines)] pub fn stream_query_results_with_conn<'a>( sql_file: &'a ParsedSqlFile, request: &'a ExecutionContext, @@ -164,17 +165,24 @@ pub fn stream_query_results_with_conn<'a>( &request.app_state.telemetry_metrics, ); record_query_params(&query_metrics.span, &query.param_values); - let results = connection - .execute(query.sql, &query.arguments) - .instrument(query_span.clone()) - .await; let mut error = None; let mut returned_rows: i64 = 0; - let start_next = std::time::Instant::now(); - match results { - Ok(results) => { + { + let mut results = connection.execute_stream(query.sql, &query.arguments); + loop { + let start_next = std::time::Instant::now(); + let elem = results.next().instrument(query_span.clone()).await; query_metrics.add_duration(start_next.elapsed()); - for elem in results { + let Some(elem) = elem else { + break; + }; + let elem = match elem { + Ok(elem) => elem, + Err(e) => { + error = Some(display_stmt_db_error(source_file, stmt, e)); + break; + } + }; let mut query_result = parse_single_sql_result(source_file, stmt, elem); if let DbItem::Error(e) = query_result { error = Some(e); @@ -196,11 +204,6 @@ pub fn stream_query_results_with_conn<'a>( } } } - Err(e) => { - query_metrics.add_duration(start_next.elapsed()); - error = Some(display_stmt_db_error(source_file, stmt, e)); - } - } if let Some(error) = error { query_metrics.record_error(returned_rows, &error); try_rollback_transaction(connection).await; @@ -293,7 +296,7 @@ async fn exec_static_simple_select( async fn try_rollback_transaction(db_connection: &mut DbConnection) { log::debug!("Attempting to rollback transaction"); match db_connection.execute_command("ROLLBACK", &[]).await { - Ok(_) => log::debug!("Rolled back transaction"), + Ok(()) => log::debug!("Rolled back transaction"), Err(e) => { log::debug!("There was probably no transaction in progress when this happened: {e:?}"); } @@ -531,15 +534,19 @@ fn debug_row(r: &super::driver::DbRow) { use std::fmt::Write; let mut row_str = String::new(); for (col, value) in r.columns.iter().zip(r.values.iter()) { - write!( - &mut row_str, - "[{:?} ({}): {:?}: {:?}]", - col.name, - if matches!(value, super::driver::DbValue::Null) { "NULL" } else { "NOT NULL" }, - col, - value - ) - .unwrap(); + write!( + &mut row_str, + "[{:?} ({}): {:?}: {:?}]", + col.name, + if matches!(value, super::driver::DbValue::Null) { + "NULL" + } else { + "NOT NULL" + }, + col, + value + ) + .unwrap(); } log::trace!("Received db row: {row_str}"); } diff --git a/src/webserver/database/migrations.rs b/src/webserver/database/migrations.rs index 72879aa0..fcb91b6d 100644 --- a/src/webserver/database/migrations.rs +++ b/src/webserver/database/migrations.rs @@ -5,7 +5,7 @@ use anyhow::Context; use sha2::{Digest, Sha384}; use super::error_highlighting::display_db_error; -use super::{Database, DbParam, make_placeholder}; +use super::{Database, DbKind, DbParam, make_placeholder}; use crate::MIGRATIONS_DIR; #[derive(Debug)] @@ -45,7 +45,7 @@ pub async fn apply(config: &crate::app_config::AppConfig, db: &Database) -> anyh } let mut conn = db.connection.acquire().await?; - ensure_migrations_table(&mut conn).await?; + ensure_migrations_table(&mut conn, db.info.kind).await?; for migration in migrations { let applied = migration_row(&mut conn, db, migration.version).await?; if let Some(applied_checksum) = applied { @@ -58,12 +58,14 @@ pub async fn apply(config: &crate::app_config::AppConfig, db: &Database) -> anyh } let start = Instant::now(); - if let Err(err) = conn.execute_command(&migration.sql, &[]).await { - return Err(display_db_error(&migration.path, &migration.sql, err).context(format!( - "Failed to apply {} migration {}", - db, - DisplayMigration(&migration) - ))); + if let Err(err) = conn.execute_batch(&migration.sql).await { + return Err( + display_db_error(&migration.path, &migration.sql, err).context(format!( + "Failed to apply {} migration {}", + db, + DisplayMigration(&migration) + )), + ); } let execution_time = i64::try_from(start.elapsed().as_millis()).unwrap_or(i64::MAX); record_migration(&mut conn, db, &migration, execution_time).await?; @@ -104,19 +106,54 @@ fn load_migrations(migrations_dir: &Path) -> anyhow::Result> { Ok(migrations) } -async fn ensure_migrations_table(conn: &mut super::DbConnection) -> anyhow::Result<()> { - conn.execute_command( - "CREATE TABLE IF NOT EXISTS _sqlx_migrations ( +async fn ensure_migrations_table( + conn: &mut super::DbConnection, + kind: DbKind, +) -> anyhow::Result<()> { + let sql = match kind { + DbKind::Sqlite => { + "CREATE TABLE IF NOT EXISTS _sqlx_migrations ( version BIGINT PRIMARY KEY, description TEXT NOT NULL, installed_on TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, success BOOLEAN NOT NULL, checksum BLOB NOT NULL, execution_time BIGINT NOT NULL - )", - &[], - ) - .await?; + )" + } + DbKind::Postgres => { + "CREATE TABLE IF NOT EXISTS _sqlx_migrations ( + version BIGINT PRIMARY KEY, + description TEXT NOT NULL, + installed_on TIMESTAMPTZ NOT NULL DEFAULT now(), + success BOOLEAN NOT NULL, + checksum BYTEA NOT NULL, + execution_time BIGINT NOT NULL + )" + } + DbKind::MySql | DbKind::Odbc => { + "CREATE TABLE IF NOT EXISTS _sqlx_migrations ( + version BIGINT PRIMARY KEY, + description VARCHAR(255) NOT NULL, + installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + success BOOLEAN NOT NULL, + checksum BLOB NOT NULL, + execution_time BIGINT NOT NULL + )" + } + DbKind::Mssql => { + "IF OBJECT_ID(N'_sqlx_migrations', N'U') IS NULL + CREATE TABLE _sqlx_migrations ( + version BIGINT PRIMARY KEY, + description NVARCHAR(255) NOT NULL, + installed_on DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(), + success BIT NOT NULL, + checksum VARBINARY(MAX) NOT NULL, + execution_time BIGINT NOT NULL + )" + } + }; + conn.execute_command(sql, &[]).await?; Ok(()) } @@ -130,7 +167,7 @@ async fn migration_row( make_placeholder(db.info.kind, 1) ); let row = conn - .fetch_optional(&sql, &[DbParam::Text(version.to_string())]) + .fetch_optional(&sql, &[DbParam::Integer(version)]) .await?; Ok(row.and_then(|row| match row.values.first() { Some(super::driver::DbValue::Bytes(bytes)) => Some(bytes.clone()), @@ -156,11 +193,11 @@ async fn record_migration( conn.execute_command( &sql, &[ - DbParam::Text(migration.version.to_string()), + DbParam::Integer(migration.version), DbParam::Text(migration.description.clone()), - DbParam::Text("true".to_string()), + DbParam::Bool(true), DbParam::Bytes(migration.checksum.clone()), - DbParam::Text(execution_time.to_string()), + DbParam::Integer(execution_time), ], ) .await?; diff --git a/src/webserver/database/mod.rs b/src/webserver/database/mod.rs index f46370cb..049f8d19 100644 --- a/src/webserver/database/mod.rs +++ b/src/webserver/database/mod.rs @@ -1,22 +1,22 @@ pub mod blob_to_data_url; mod connect; mod csv_import; +pub mod driver; pub mod execute_queries; pub mod migrations; mod sql; mod sqlpage_functions; mod syntax_tree; -pub mod driver; mod error_highlighting; mod sql_to_json; +pub use driver::{DbConnection, DbError, DbKind, DbParam, DbPool}; pub use sql::ParsedSqlFile; use sql::{DB_PLACEHOLDERS, DbPlaceHolder}; -pub use driver::{DbConnection, DbError, DbKind, DbParam, DbPool}; // SupportedDatabase is defined in this module -/// Supported database types in `SQLPage`. Represents an actual DBMS, not a sqlx backend kind (like "Odbc") +/// Supported database types in `SQLPage`. Represents an actual DBMS, not a driver kind like ODBC. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum SupportedDatabase { Sqlite, @@ -109,9 +109,9 @@ pub struct DbInfo { } impl Database { - pub async fn close(&self) -> anyhow::Result<()> { + pub fn close(&self) -> anyhow::Result<()> { log::info!("Closing all database connections..."); - self.connection.close().await; + self.connection.close(); Ok(()) } } diff --git a/src/webserver/database/sql.rs b/src/webserver/database/sql.rs index 62b9646a..941ef485 100644 --- a/src/webserver/database/sql.rs +++ b/src/webserver/database/sql.rs @@ -783,11 +783,11 @@ mod test { fn create_test_db_info(database_type: SupportedDatabase) -> DbInfo { let kind = match database_type { - SupportedDatabase::Postgres => super::DbKind::Postgres, - SupportedDatabase::Mssql => super::DbKind::Mssql, - SupportedDatabase::MySql => super::DbKind::MySql, - SupportedDatabase::Sqlite => super::DbKind::Sqlite, - _ => super::DbKind::Odbc, + SupportedDatabase::Postgres => crate::webserver::database::DbKind::Postgres, + SupportedDatabase::Mssql => crate::webserver::database::DbKind::Mssql, + SupportedDatabase::MySql => crate::webserver::database::DbKind::MySql, + SupportedDatabase::Sqlite => crate::webserver::database::DbKind::Sqlite, + _ => crate::webserver::database::DbKind::Odbc, }; DbInfo { dbms_name: database_type.display_name().to_string(), @@ -1019,6 +1019,18 @@ mod test { assert_eq!(parameters, [StmtParam::PostOrGet("1".to_string()),]); } + #[test] + fn test_mysql_statement_rewrite() { + let mut ast = parse_stmt("select '' || $1 || 'x'", &MySqlDialect {}); + let db_info = create_test_db_info(SupportedDatabase::MySql); + let parameters = ParameterExtractor::extract_parameters(&mut ast, db_info).unwrap(); + assert_eq!( + ast.to_string(), + "SELECT CONCAT(CONCAT('', CAST(@SQLPAGE_TEMP1 AS CHAR)), 'x')" + ); + assert_eq!(parameters, [StmtParam::PostOrGet("1".to_string()),]); + } + #[test] fn test_static_extract() { use SimpleSelectValue::Static; @@ -1316,7 +1328,7 @@ mod test { delayed_functions: vec![], json_columns: vec![], }; - transform_to_positional_placeholders(&mut stmt, super::DbKind::MySql); + transform_to_positional_placeholders(&mut stmt, crate::webserver::database::DbKind::MySql); assert_eq!( stmt.query, "select \ diff --git a/src/webserver/database/sql/parameter_extraction.rs b/src/webserver/database/sql/parameter_extraction.rs index 3d17e047..d4b122d2 100644 --- a/src/webserver/database/sql/parameter_extraction.rs +++ b/src/webserver/database/sql/parameter_extraction.rs @@ -1,8 +1,8 @@ use super::super::{DbInfo, SupportedDatabase}; use super::{is_sqlpage_func, sqlpage_func_name}; +use crate::webserver::database::DbKind; use crate::webserver::database::sqlpage_functions::func_call_to_param; use crate::webserver::database::syntax_tree::StmtParam; -use crate::webserver::database::DbKind; use sqlparser::ast::{ BinaryOperator, CastKind, CharacterLength, DataType, Expr, Function, FunctionArg, FunctionArgExpr, FunctionArgumentList, FunctionArguments, Ident, ObjectName, ObjectNamePart, @@ -39,10 +39,7 @@ pub(crate) const DB_PLACEHOLDERS: [(DbKind, DbPlaceHolder); 5] = [ DbKind::Mssql, DbPlaceHolder::PrefixedNumber { prefix: "@p" }, ), - ( - DbKind::Odbc, - DbPlaceHolder::Positional { placeholder: "?" }, - ), + (DbKind::Odbc, DbPlaceHolder::Positional { placeholder: "?" }), ]; /// For positional parameters, we use a temporary placeholder during parameter extraction, @@ -541,12 +538,16 @@ impl VisitorMut for ParameterExtractor { } self.replace_with_placeholder(value, param); } - // Replace 'str1' || 'str2' with CONCAT('str1', 'str2') for MSSQL + // Replace 'str1' || 'str2' with CONCAT('str1', 'str2') where pipes are not string concatenation. Expr::BinaryOp { left, op: BinaryOperator::StringConcat, right, - } if self.db_info.database_type == SupportedDatabase::Mssql => { + } if matches!( + self.db_info.database_type, + SupportedDatabase::Mssql | SupportedDatabase::MySql + ) => + { let left = std::mem::replace(left.as_mut(), Expr::value(Value::Null)); let right = std::mem::replace(right.as_mut(), Expr::value(Value::Null)); *value = Expr::Function(Function { diff --git a/src/webserver/database/sql_to_json.rs b/src/webserver/database/sql_to_json.rs index a60bb52e..e9de2932 100644 --- a/src/webserver/database/sql_to_json.rs +++ b/src/webserver/database/sql_to_json.rs @@ -14,11 +14,7 @@ pub fn row_to_json(row: &DbRow) -> Value { } fn canonical_col_name(col: &DbColumn, kind: DbKind) -> String { - if matches!(kind, DbKind::Odbc) - && col - .name - .chars() - .all(|c| c.is_ascii_uppercase() || c == '_') + if matches!(kind, DbKind::Odbc) && col.name.chars().all(|c| c.is_ascii_uppercase() || c == '_') { col.name.to_ascii_lowercase() } else { @@ -29,6 +25,7 @@ fn canonical_col_name(col: &DbColumn, kind: DbKind) -> String { pub fn sql_value_to_json(value: &DbValue) -> Value { match value { DbValue::Null => Value::Null, + DbValue::Bool(b) => (*b).into(), DbValue::Integer(i) => (*i).into(), DbValue::Real(f) => (*f).into(), DbValue::Text(s) => Value::String(s.clone()), @@ -78,6 +75,9 @@ mod tests { values: vec![DbValue::Text("hello".into())], kind: DbKind::Odbc, }; - assert_eq!(row_to_json(&row), serde_json::json!({"title_text": "hello"})); + assert_eq!( + row_to_json(&row), + serde_json::json!({"title_text": "hello"}) + ); } } diff --git a/src/webserver/database/sqlpage_functions/functions.rs b/src/webserver/database/sqlpage_functions/functions.rs index bf669470..f1076ec9 100644 --- a/src/webserver/database/sqlpage_functions/functions.rs +++ b/src/webserver/database/sqlpage_functions/functions.rs @@ -7,7 +7,6 @@ use crate::webserver::{ sqlpage_functions::{http_fetch_request::HttpFetchRequest, url_parameters::URLParameters}, }, http_client::make_http_client, - request_variables::SetVariablesMap, single_or_vec::SingleOrVec, }; use anyhow::{Context, anyhow}; @@ -722,6 +721,17 @@ async fn run_sql<'a>( log::debug!("run_sql: first argument is NULL, returning NULL"); return Ok(None); }; + if request + .included_sql_files + .iter() + .any(|path| path == sql_file_path.as_ref()) + { + anyhow::bail!( + "Too many nested inclusions. run_sql cannot include a file that is already being executed in the same inclusion chain. \ + Executing sqlpage.run_sql('{sql_file_path}') would create a loop. \ + This is to prevent infinite loops and stack overflows." + ); + } let run_sql_span = tracing::info_span!( "sqlpage.file", otel.name = format!("SQL {sql_file_path}"), @@ -738,14 +748,14 @@ async fn run_sql<'a>( .instrument(run_sql_span.clone()) .await .with_context(|| format!("run_sql: invalid path {sql_file_path:?}"))?; - let tmp_req = if let Some(variables) = variables { - let variables: SetVariablesMap = serde_json::from_str(&variables).with_context(|| { + let variables = if let Some(variables) = variables { + serde_json::from_str(&variables).with_context(|| { format!("run_sql(\'{sql_file_path}\', \'{variables}\'): the second argument should be a JSON object with string keys and values") - })?; - request.fork_with_variables(variables) + })? } else { - request.fork() + request.set_variables.borrow().clone() }; + let tmp_req = request.fork_for_run_sql(sql_file_path.as_ref(), variables); let max_recursion_depth = app_state.config.max_recursion_depth; if tmp_req.clone_depth > max_recursion_depth { anyhow::bail!( diff --git a/src/webserver/http.rs b/src/webserver/http.rs index 242d9613..2884b98b 100644 --- a/src/webserver/http.rs +++ b/src/webserver/http.rs @@ -275,11 +275,11 @@ async fn render_sql( let database_entries_stream = stream_query_results_with_conn(&sql_file, &exec_ctx, &mut conn); let database_entries_stream = stop_at_first_error(database_entries_stream); - let response_with_writer = build_response_header_and_stream( + let response_with_writer = Box::pin(build_response_header_and_stream( Arc::clone(&app_state), database_entries_stream, request_context, - ) + )) .await; match response_with_writer { Ok(ResponseWithWriter::RenderStream { @@ -678,7 +678,7 @@ pub async fn run_server(config: &AppConfig, state: AppState) -> anyhow::Result<( .with_context(|| "Unable to start the application")?; // We are done, we can close the database connection - final_state.db.close().await?; + final_state.db.close()?; Ok(()) } diff --git a/src/webserver/http_request_info.rs b/src/webserver/http_request_info.rs index da628736..6cba2e32 100644 --- a/src/webserver/http_request_info.rs +++ b/src/webserver/http_request_info.rs @@ -53,6 +53,7 @@ pub struct ExecutionContext { pub request: Rc, pub set_variables: RefCell, pub clone_depth: u8, + pub included_sql_files: Rc>, } impl ExecutionContext { @@ -62,6 +63,7 @@ impl ExecutionContext { request: Rc::new(request), set_variables: RefCell::new(SetVariablesMap::new()), clone_depth: 0, + included_sql_files: Rc::new(Vec::new()), } } @@ -71,6 +73,7 @@ impl ExecutionContext { request: Rc::clone(&self.request), set_variables: RefCell::new(self.set_variables.borrow().clone()), clone_depth: self.clone_depth + 1, + included_sql_files: Rc::clone(&self.included_sql_files), } } @@ -80,6 +83,19 @@ impl ExecutionContext { request: Rc::clone(&self.request), set_variables: RefCell::new(variables), clone_depth: self.clone_depth + 1, + included_sql_files: Rc::clone(&self.included_sql_files), + } + } + + #[must_use] + pub fn fork_for_run_sql(&self, sql_file_path: &str, variables: SetVariablesMap) -> Self { + let mut included_sql_files = self.included_sql_files.as_ref().clone(); + included_sql_files.push(sql_file_path.to_string()); + Self { + request: Rc::clone(&self.request), + set_variables: RefCell::new(variables), + clone_depth: self.clone_depth + 1, + included_sql_files: Rc::new(included_sql_files), } } diff --git a/tests/core/mod.rs b/tests/core/mod.rs index 229aeac2..159cabd0 100644 --- a/tests/core/mod.rs +++ b/tests/core/mod.rs @@ -1,9 +1,9 @@ use actix_web::{http::StatusCode, test}; use sqlpage::{ AppState, + webserver::database::DbParam, webserver::{self, make_placeholder}, }; -use sqlx::Executor as _; use crate::common::{make_app_data_from_config, req_path, req_path_with_app_data, test_config}; @@ -58,19 +58,25 @@ async fn test_routing_with_db_fs() { } let drop_sql = "DROP TABLE IF EXISTS sqlpage_files"; - state.db.connection.execute(drop_sql).await.unwrap(); + let mut conn = state.db.connection.acquire().await.unwrap(); + conn.execute_command(drop_sql, &[]).await.unwrap(); let create_table_sql = sqlpage::filesystem::DbFsQueries::get_create_table_sql(state.db.info.database_type); - state.db.connection.execute(create_table_sql).await.unwrap(); + conn.execute_command(create_table_sql, &[]).await.unwrap(); let insert_sql = format!( "INSERT INTO sqlpage_files(path, contents) VALUES ('on_db.sql', {})", make_placeholder(state.db.info.kind, 1) ); - sqlx::query(&insert_sql) - .bind("select ''text'' as component, ''Hi from db !'' AS contents;".as_bytes()) - .execute(&state.db.connection) - .await - .unwrap(); + conn.execute_command( + &insert_sql, + &[DbParam::Bytes( + "select ''text'' as component, ''Hi from db !'' AS contents;" + .as_bytes() + .to_vec(), + )], + ) + .await + .unwrap(); let state = AppState::init(&config).await.unwrap(); let app_data = actix_web::web::Data::new(state); @@ -101,23 +107,28 @@ async fn test_non_unicode_static_path_returns_bad_request_with_db_fs() { let expected_db_path = "\u{FFFD}.txt"; let mut conn = state.db.connection.acquire().await.unwrap(); - (&mut *conn) - .execute(sqlpage::filesystem::DbFsQueries::get_create_table_sql( + conn.execute_command( + sqlpage::filesystem::DbFsQueries::get_create_table_sql( sqlpage::webserver::database::SupportedDatabase::Sqlite, - )) - .await - .unwrap(); + ), + &[], + ) + .await + .unwrap(); let insert_sql = format!( "INSERT INTO sqlpage_files(path, contents) VALUES ({}, {})", make_placeholder(state.db.info.kind, 1), make_placeholder(state.db.info.kind, 2) ); - sqlx::query(&insert_sql) - .bind(expected_db_path) - .bind("file from db fs".as_bytes()) - .execute(&mut *conn) - .await - .unwrap(); + conn.execute_command( + &insert_sql, + &[ + DbParam::Text(expected_db_path.into()), + DbParam::Bytes("file from db fs".as_bytes().to_vec()), + ], + ) + .await + .unwrap(); drop(conn); let state = AppState::init(&config).await.unwrap();