Add native database abstraction

Author: lovasoa

Cargo.lock

@@ -18,20 +18,13 @@ panic = "abort"
 codegen-units = 2
 
 [dependencies]
-sqlx = { version = "0.9.0", default-features = false, features = [
-    "runtime-tokio",
-    "tls-rustls",
-    "migrate",
-    "sqlite",
-    "postgres",
-    "mysql",
-    "chrono",
-    "bigdecimal",
-    "json",
-    "uuid",
-] }
-sqlx-sqlserver = { version = "0.0.2", features = ["migrate"] }
-sqlx-odbc = { version = "0.0.1", features = ["runtime-tokio"] }
+tokio-postgres = { version = "0.7.17", features = ["with-chrono-0_4", "with-serde_json-1", "with-uuid-1"] }
+mysql_async = { version = "0.37.0", default-features = false, features = ["default-rustls-ring", "chrono", "bigdecimal"] }
+tiberius = { version = "0.12.3", default-features = false, features = ["tds73", "tokio", "tokio-util", "rustls", "chrono", "bigdecimal"] }
+tokio-rusqlite = { version = "0.7.0", features = ["bundled", "functions", "load_extension", "collation", "chrono", "blob"] }
+odbc-api = { version = "28.0.0", default-features = false, features = ["odbc_version_3_80"] }
+deadpool = { version = "0.13.0", features = ["rt_tokio_1"] }
+uuid = "1"
 chrono = "0.4.23"
 actix-web = { version = "4", features = ["rustls-0_23", "cookies"] }
 percent-encoding = "2.2.0"
@@ -70,7 +63,7 @@ sha2 = "0.11"
 rustls-acme = "0.15"
 dotenvy = "0.15.7"
 csv-async = { version = "1.2.6", features = ["tokio"] }
-rustls = { version = "0.23" } # keep in sync with actix-web, awc, rustls-acme, and sqlx
+rustls = { version = "0.23" } # 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"] }
@@ -94,7 +87,7 @@ opentelemetry-semantic-conventions = { version = "0.32", features = ["semconv_ex
 
 [features]
 default = []
-odbc-static = ["sqlx-odbc/vendored-unix-odbc"]
+odbc-static = ["odbc-api/vendored-unix-odbc"]
 lambda-web = ["dep:lambda-web", "odbc-static"]
 
 [dev-dependencies]

Cargo.toml

@@ -1,12 +1,10 @@
 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::{AppState, TEMPLATES_DIR};
 use anyhow::Context;
 use chrono::{DateTime, Utc};
-use sqlx::any::{AnyStatement, AnyTypeInfo};
-use sqlx::postgres::types::PgTimeTz;
-use sqlx::{Executor, Postgres, Statement, Type};
 use std::fmt::Write;
 use std::io::ErrorKind;
 use std::path::{Component, Path, PathBuf};
@@ -241,9 +239,9 @@ async fn file_modified_since_local(path: &Path, since: DateTime<Utc>) -> tokio::
 }
 
 pub struct DbFsQueries {
-    was_modified: AnyStatement<'static>,
-    read_file: AnyStatement<'static>,
-    exists: AnyStatement<'static>,
+    was_modified: String,
+    read_file: String,
+    exists: String,
 }
 
 impl DbFsQueries {
@@ -277,44 +275,40 @@ impl DbFsQueries {
 
     async fn check_table_available(db: &Database) -> anyhow::Result<()> {
         db.connection
-            .execute("SELECT 1 FROM sqlpage_files WHERE 1 = 0")
+            .acquire()
+            .await
+            .context("Unable to acquire database connection")?
+            .execute_command("SELECT 1 FROM sqlpage_files WHERE 1 = 0", &[])
             .await
-            .map(|_| ())
             .context("Unable to access sqlpage_files")?;
         Ok(())
     }
 
-    async fn make_was_modified_query(db: &Database) -> anyhow::Result<AnyStatement<'static>> {
+    async fn make_was_modified_query(db: &Database) -> anyhow::Result<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)
         );
-        let param_types: &[AnyTypeInfo; 2] = &[
-            PgTimeTz::type_info().into(),
-            <str as Type<Postgres>>::type_info().into(),
-        ];
         log::debug!("Preparing the database filesystem was_modified_query: {was_modified_query}");
-        db.prepare_with(&was_modified_query, param_types).await
+        Ok(was_modified_query)
     }
 
-    async fn make_read_file_query(db: &Database) -> anyhow::Result<AnyStatement<'static>> {
+    async fn make_read_file_query(db: &Database) -> anyhow::Result<String> {
         let read_file_query = format!(
             "SELECT contents from sqlpage_files WHERE path = {}",
             make_placeholder(db.info.kind, 1),
         );
-        let param_types: &[AnyTypeInfo; 1] = &[<str as Type<Postgres>>::type_info().into()];
         log::debug!("Preparing the database filesystem read_file_query: {read_file_query}");
-        db.prepare_with(&read_file_query, param_types).await
+        Ok(read_file_query)
     }
 
-    async fn make_exists_query(db: &Database) -> anyhow::Result<AnyStatement<'static>> {
+    async fn make_exists_query(db: &Database) -> anyhow::Result<String> {
         let exists_query = format!(
             "SELECT 1 from sqlpage_files WHERE path = {}",
             make_placeholder(db.info.kind, 1),
         );
-        let param_types: &[AnyTypeInfo; 1] = &[<str as Type<Postgres>>::type_info().into()];
-        db.prepare_with(&exists_query, param_types).await
+        Ok(exists_query)
     }
 
     async fn file_modified_since_in_db(
@@ -323,22 +317,22 @@ impl DbFsQueries {
         path: &Path,
         since: DateTime<Utc>,
     ) -> anyhow::Result<bool> {
-        let query = self
-            .was_modified
-            .query_as::<(i32,)>()
-            .bind(since)
-            .bind(path.display().to_string());
+        let params = [
+            DbParam::Timestamp(since),
+            DbParam::Text(path.display().to_string()),
+        ];
         log::trace!(
             "Checking if file {} was modified since {} by executing query: \n\
             {}\n\
             with parameters: {:?}",
             path.display(),
             since,
-            self.was_modified.sql(),
+            self.was_modified,
             (since, path)
         );
-        let was_modified_i32 = query
-            .fetch_optional(&app_state.db.connection)
+        let mut conn = app_state.db.connection.acquire().await?;
+        let was_modified_i32 = conn
+            .fetch_optional(&self.was_modified, &params)
             .await
             .with_context(|| {
                 format!(
@@ -350,20 +344,25 @@ impl DbFsQueries {
             "DB File {} was modified result: {was_modified_i32:?}",
             path.display()
         );
-        Ok(was_modified_i32 == Some((1,)))
+        Ok(was_modified_i32.is_some())
     }
 
     async fn read_file(&self, app_state: &AppState, path: &Path) -> anyhow::Result<Vec<u8>> {
         log::debug!("Reading file {} from the database", path.display());
-        self.read_file
-            .query_as::<(Vec<u8>,)>()
-            .bind(path.display().to_string())
-            .fetch_optional(&app_state.db.connection)
+        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(|modified| {
-                if let Some((modified,)) = modified {
-                    Ok(modified)
+            .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,
@@ -375,19 +374,17 @@ impl DbFsQueries {
     }
 
     async fn file_exists(&self, app_state: &AppState, path: &Path) -> anyhow::Result<bool> {
-        let query = self
-            .exists
-            .query_as::<(i32,)>()
-            .bind(path.display().to_string());
+        let params = [DbParam::Text(path.display().to_string())];
         log::trace!(
             "Checking if file {} exists by executing query: \n\
             {}\n\
             with parameters: {:?}",
             path.display(),
-            self.exists.sql(),
+            self.exists,
             (path,)
         );
-        let result = query.fetch_optional(&app_state.db.connection).await;
+        let mut conn = app_state.db.connection.acquire().await?;
+        let result = conn.fetch_optional(&self.exists, &params).await;
         log::debug!("DB File exists result: {result:?}");
         result.map(|result| result.is_some()).with_context(|| {
             format!(
@@ -401,7 +398,6 @@ impl DbFsQueries {
 #[actix_web::test]
 async fn test_sql_file_read_utf8() -> anyhow::Result<()> {
     use crate::app_config;
-    use sqlx::Executor;
     let config = app_config::tests::test_config();
     let state = AppState::init(&config).await?;
 
@@ -417,20 +413,24 @@ 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;
-    conn.execute("DROP TABLE IF EXISTS sqlpage_files").await?;
+    let mut conn = db.connection.acquire().await?;
+    conn.execute_command("DROP TABLE IF EXISTS sqlpage_files", &[]).await?;
     log::debug!("Creating table sqlpage_files: {create_table_sql}");
-    conn.execute(create_table_sql).await?;
+    conn.execute_command(create_table_sql, &[]).await?;
 
     let dbms = db.info.kind;
     let insert_sql = format!(
         "INSERT INTO sqlpage_files(path, contents) VALUES ({}, {})",
         make_placeholder(dbms, 1),
         make_placeholder(dbms, 2)
     );
-    sqlx::query(&insert_sql)
-        .bind("unit test file.txt")
-        .bind("Héllö world! 😀".as_bytes())
-        .execute(conn)
+    conn.execute_command(
+            &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;

src/filesystem.rs

@@ -2,7 +2,7 @@ use opentelemetry::global;
 use opentelemetry::metrics::{Histogram, ObservableGauge};
 use opentelemetry_semantic_conventions::attribute as otel;
 use opentelemetry_semantic_conventions::metric as otel_metric;
-use sqlx::AnyPool;
+use crate::webserver::database::DbPool;
 
 pub struct TelemetryMetrics {
     pub http_request_duration: Histogram<f64>,
@@ -41,7 +41,7 @@ impl Default for TelemetryMetrics {
 
 impl TelemetryMetrics {
     #[must_use]
-    pub fn new(pool: &AnyPool, db_system_name: &'static str) -> Self {
+    pub fn new(pool: &DbPool, db_system_name: &'static str) -> Self {
         let meter = global::meter("sqlpage");
         let http_request_duration = meter
             .f64_histogram(otel_metric::HTTP_SERVER_REQUEST_DURATION)
@@ -60,9 +60,8 @@ impl TelemetryMetrics {
             .with_description("Number of connections in the database pool.")
             .with_callback(move |observer| {
                 let size = pool_ref.size();
-                let idle_u32 = u32::try_from(pool_ref.num_idle()).unwrap_or(u32::MAX);
-                let used = i64::from(size.saturating_sub(idle_u32));
-                let idle = i64::from(idle_u32);
+                let idle = i64::from(pool_ref.num_idle());
+                let used = i64::from(size);
                 observer.observe(
                     used,
                     &[