refactor(sql): switch execute to sqlx

src/sql/mod.rs

@@ -1,13 +1,11 @@
 //! # SQLite wrapper
 
-use async_std::prelude::*;
-use async_std::sync::RwLock;
-
 use std::collections::HashSet;
 use std::path::Path;
-use std::time::Duration;
 
-use rusqlite::{Connection, Error as SqlError, OpenFlags};
+use async_std::prelude::*;
+use async_std::sync::RwLock;
+use sqlx::sqlite::*;
 
 use crate::chat::{update_device_icon, update_saved_messages_icon};
 use crate::constants::DC_CHAT_ID_TRASH;
@@ -16,24 +14,14 @@ use crate::dc_tools::*;
 use crate::param::*;
 use crate::peerstate::*;
 
-#[macro_export]
-macro_rules! paramsv {
-    () => {
-        Vec::new()
-    };
-    ($($param:expr),+ $(,)?) => {
-        vec![$(&$param as &dyn $crate::ToSql),+]
-    };
-}
-
+#[macro_use]
+mod macros;
 mod migrations;
 
+pub use macros::*;
+
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
-    #[error("Sqlite Error: {0:?}")]
-    Sql(#[from] rusqlite::Error),
-    #[error("Sqlite Connection Pool Error: {0:?}")]
-    ConnectionPool(#[from] r2d2::Error),
     #[error("Sqlite: Connection closed")]
     SqlNoConnection,
     #[error("Sqlite: Already open")]
@@ -48,6 +36,8 @@ pub enum Error {
     Other(#[from] crate::error::Error),
     #[error("{0}")]
     Sqlx(#[from] sqlx::Error),
+    #[error("{0}: {1}")]
+    SqlxWithContext(String, #[source] sqlx::Error),
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -55,14 +45,12 @@ pub type Result<T> = std::result::Result<T, Error>;
 /// A wrapper around the underlying Sqlite3 object.
 #[derive(Debug)]
 pub struct Sql {
-    pool: RwLock<Option<r2d2::Pool<r2d2_sqlite::SqliteConnectionManager>>>,
     xpool: RwLock<Option<sqlx::SqlitePool>>,
 }
 
 impl Default for Sql {
     fn default() -> Self {
         Self {
-            pool: RwLock::new(None),
             xpool: RwLock::new(None),
         }
     }
@@ -73,14 +61,17 @@ impl Sql {
         Self::default()
     }
 
+    /// Returns `true` if there is a working sqlite connection, `false` otherwise.
     pub async fn is_open(&self) -> bool {
-        self.pool.read().await.is_some() && self.xpool.read().await.is_some()
+        let pool = self.xpool.read().await;
+        pool.is_some() && !pool.as_ref().unwrap().is_closed()
     }
 
+    /// Shuts down all sqlite connections.
     pub async fn close(&self) {
-        let _ = self.pool.write().await.take();
-        let _ = self.xpool.write().await.take();
-        // drop closes the connection
+        if let Some(pool) = self.xpool.write().await.take() {
+            pool.close().await;
+        }
     }
 
     pub async fn open<T: AsRef<Path>>(
@@ -102,37 +93,21 @@ impl Sql {
         Ok(())
     }
 
-    pub async fn execute<S: AsRef<str>>(
-        &self,
-        sql: S,
-        params: Vec<&dyn crate::ToSql>,
-    ) -> Result<usize> {
-        let res = {
-            let conn = self.get_conn().await?;
-            conn.execute(sql.as_ref(), params)
-        };
-
-        res.map_err(Into::into)
-    }
-
-    pub async fn xexecute<S: AsRef<str>>(
-        &self,
-        statement: S,
-        params: sqlx::sqlite::SqliteArguments,
-    ) -> Result<()> {
+    /// Execute a single query.
+    pub async fn execute<'a, P>(&self, statement: &'a str, params: P) -> Result<usize>
+    where
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+    {
         let lock = self.xpool.read().await;
         let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
 
-        sqlx::query(statement.as_ref())
-            .bind_all(params)
-            .execute(xpool)
-            .await?;
+        let count = sqlx::query_with(statement, params).execute(xpool).await?;
 
-        Ok(())
+        Ok(count as usize)
     }
 
     /// Execute a list of statements, without any bindings
-    pub async fn xexecute_batch<S: AsRef<str>>(&self, statement: S) -> Result<()> {
+    pub async fn execute_batch(&self, statement: &str) -> Result<()> {
         let lock = self.xpool.read().await;
         let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
 
@@ -141,173 +116,151 @@ impl Sql {
         Ok(())
     }
 
-    pub async fn execute_batch<S: AsRef<str>>(&self, sql: S) -> Result<()> {
-        let res = {
-            let conn = self.get_conn().await?;
-            conn.execute_batch(sql.as_ref())
-        };
-
-        res.map_err(Into::into)
+    pub async fn get_pool(&self) -> Result<SqlitePool> {
+        let lock = self.xpool.read().await;
+        lock.as_ref().cloned().ok_or_else(|| Error::SqlNoConnection)
     }
-    /// Prepares and executes the statement and maps a function over the resulting rows.
-    /// Then executes the second function over the returned iterator and returns the
-    /// result of that function.
-    pub async fn query_map<T, F, G, H>(
+
+    /// Starts a new transaction.
+    pub async fn begin(
         &self,
-        sql: impl AsRef<str>,
-        params: Vec<&dyn crate::ToSql>,
-        f: F,
-        mut g: G,
-    ) -> Result<H>
-    where
-        F: FnMut(&rusqlite::Row) -> rusqlite::Result<T>,
-        G: FnMut(rusqlite::MappedRows<F>) -> Result<H>,
-    {
-        let sql = sql.as_ref();
-
-        let conn = self.get_conn().await?;
-        let mut stmt = conn.prepare(sql)?;
-        let res = stmt.query_map(&params, f)?;
-        g(res)
-    }
-
-    pub async fn get_conn(
-        &self,
-    ) -> Result<r2d2::PooledConnection<r2d2_sqlite::SqliteConnectionManager>> {
-        let lock = self.pool.read().await;
-        let pool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
-        let conn = pool.get()?;
-
-        Ok(conn)
-    }
-
-    pub async fn with_conn<G, H>(&self, g: G) -> Result<H>
-    where
-        H: Send + 'static,
-        G: Send
-            + 'static
-            + FnOnce(r2d2::PooledConnection<r2d2_sqlite::SqliteConnectionManager>) -> Result<H>,
-    {
-        let lock = self.pool.read().await;
-        let pool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
-        let conn = pool.get()?;
-
-        g(conn)
-    }
-
-    pub async fn with_conn_async<G, H, Fut>(&self, mut g: G) -> Result<H>
-    where
-        G: FnMut(r2d2::PooledConnection<r2d2_sqlite::SqliteConnectionManager>) -> Fut,
-        Fut: Future<Output = Result<H>> + Send,
-    {
-        let lock = self.pool.read().await;
+    ) -> Result<sqlx::Transaction<'static, Sqlite, sqlx::pool::PoolConnection<Sqlite>>> {
+        let lock = self.xpool.read().await;
         let pool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
 
-        let conn = pool.get()?;
-        g(conn).await
+        let tx = pool.begin().await?;
+        Ok(tx)
+    }
+
+    /// Execute a query which is expected to return zero or more rows.
+    pub async fn query_rows<'a, T, P>(&self, statement: &'a str, params: P) -> Result<Vec<T>>
+    where
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+        T: for<'b> sqlx::FromRow<'b, SqliteRow> + Unpin + Send,
+    {
+        let lock = self.xpool.read().await;
+        let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
+        let rows = sqlx::query_with(statement.as_ref(), params)
+            .try_map(|row: SqliteRow| sqlx::FromRow::from_row(&row))
+            .fetch_all(xpool)
+            .await?;
+
+        Ok(rows)
+    }
+
+    /// Execute a query which is expected to return zero or more rows.
+    pub async fn query_values<'a, T, P>(&self, statement: &'a str, params: P) -> Result<Vec<T>>
+    where
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+        T: for<'b> sqlx::decode::Decode<'b, Sqlite>,
+        T: sqlx::Type<Sqlite>,
+        T: 'static + Unpin + Send,
+    {
+        let lock = self.xpool.read().await;
+        let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
+        let rows = sqlx::query_with(statement.as_ref(), params)
+            .try_map(|row: SqliteRow| {
+                let (val,): (T,) = sqlx::FromRow::from_row(&row)?;
+                Ok(val)
+            })
+            .fetch_all(xpool)
+            .await?;
+
+        Ok(rows)
     }
 
     /// Return `true` if a query in the SQL statement it executes returns one or more
     /// rows and false if the SQL returns an empty set.
-    pub async fn exists(&self, sql: &str, params: Vec<&dyn crate::ToSql>) -> Result<bool> {
-        let res = {
-            let conn = self.get_conn().await?;
-            let mut stmt = conn.prepare(sql)?;
-            stmt.exists(&params)
-        };
+    pub async fn exists<'a, P>(&self, statement: &'a str, params: P) -> Result<bool>
+    where
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+    {
+        let lock = self.xpool.read().await;
+        let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
 
-        res.map_err(Into::into)
+        let mut rows = sqlx::query_with(statement, params).fetch(xpool);
+
+        match rows.next().await {
+            Some(Ok(_)) => Ok(true),
+            None => Ok(false),
+            Some(Err(sqlx::Error::RowNotFound)) => Ok(false),
+            Some(Err(err)) => Err(Error::SqlxWithContext(
+                format!("exists: '{}'", statement),
+                err,
+            )),
+        }
     }
 
     /// Execute a query which is expected to return one row.
-    pub async fn query_row<T, F>(
-        &self,
-        sql: impl AsRef<str>,
-        params: Vec<&dyn crate::ToSql>,
-        f: F,
-    ) -> Result<T>
+    pub async fn query_row<'a, T, P>(&self, statement: &'a str, params: P) -> Result<T>
     where
-        F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+        T: for<'b> sqlx::FromRow<'b, SqliteRow> + Unpin + Send,
     {
-        let sql = sql.as_ref();
-        let res = {
-            let conn = self.get_conn().await?;
-            conn.query_row(sql, params, f)
-        };
+        let lock = self.xpool.read().await;
+        let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
+        let row = sqlx::query_with(statement.as_ref(), params)
+            .try_map(|row: SqliteRow| sqlx::FromRow::from_row(&row))
+            .fetch_one(xpool)
+            .await?;
 
-        res.map_err(Into::into)
-    }
-
-    pub async fn table_exists(&self, name: impl AsRef<str>) -> Result<bool> {
-        let name = name.as_ref().to_string();
-        self.with_conn(move |conn| {
-            let mut exists = false;
-            conn.pragma(None, "table_info", &name, |_row| {
-                // will only be executed if the info was found
-                exists = true;
-                Ok(())
-            })?;
-
-            Ok(exists)
-        })
-        .await
+        Ok(row)
     }
 
     /// Execute a query which is expected to return zero or one row.
-    pub async fn query_row_optional<T, F>(
+    pub async fn query_row_optional<'a, T, P>(
         &self,
-        sql: impl AsRef<str>,
-        params: Vec<&dyn crate::ToSql>,
-        f: F,
+        statement: &'a str,
+        params: P,
     ) -> Result<Option<T>>
     where
-        F: FnOnce(&rusqlite::Row) -> rusqlite::Result<T>,
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+        T: for<'b> sqlx::FromRow<'b, SqliteRow> + Unpin + Send,
     {
-        match self.query_row(sql, params, f).await {
-            Ok(res) => Ok(Some(res)),
-            Err(Error::Sql(rusqlite::Error::QueryReturnedNoRows)) => Ok(None),
-            Err(Error::Sql(rusqlite::Error::InvalidColumnType(
-                _,
-                _,
-                rusqlite::types::Type::Null,
-            ))) => Ok(None),
-            Err(err) => Err(err),
+        let lock = self.xpool.read().await;
+        let xpool = lock.as_ref().ok_or_else(|| Error::SqlNoConnection)?;
+        let row = sqlx::query_with(statement.as_ref(), params)
+            .try_map(|row: SqliteRow| sqlx::FromRow::from_row(&row))
+            .fetch_optional(xpool)
+            .await?;
+
+        Ok(row)
+    }
+
+    pub async fn query_value_optional<'a, T, P>(
+        &self,
+        statement: &'a str,
+        params: P,
+    ) -> Result<Option<T>>
+    where
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+        T: for<'b> sqlx::decode::Decode<'b, Sqlite>,
+        T: sqlx::Type<Sqlite>,
+        T: 'static + Unpin + Send,
+    {
+        match self.query_row_optional(statement, params).await? {
+            Some((val,)) => Ok(Some(val)),
+            None => Ok(None),
         }
     }
 
-    /// Executes a query which is expected to return one row and one
-    /// column. If the query does not return a value or returns SQL
-    /// `NULL`, returns `Ok(None)`.
-    pub async fn query_get_value_result<T>(
-        &self,
-        query: &str,
-        params: Vec<&dyn crate::ToSql>,
-    ) -> Result<Option<T>>
+    pub async fn query_value<'a, T, P>(&self, statement: &'a str, params: P) -> Result<T>
     where
-        T: rusqlite::types::FromSql,
+        P: sqlx::IntoArguments<'a, Sqlite> + 'a,
+        T: for<'b> sqlx::decode::Decode<'b, Sqlite>,
+        T: sqlx::Type<Sqlite>,
+        T: 'static + Unpin + Send,
     {
-        self.query_row_optional(query, params, |row| row.get::<_, T>(0))
-            .await
+        let (val,): (T,) = self.query_row(statement, params).await?;
+        Ok(val)
     }
 
-    /// Not resultified version of `query_get_value_result`. Returns
-    /// `None` on error.
-    pub async fn query_get_value<T>(
-        &self,
-        context: &Context,
-        query: &str,
-        params: Vec<&dyn crate::ToSql>,
-    ) -> Option<T>
-    where
-        T: rusqlite::types::FromSql,
-    {
-        match self.query_get_value_result(query, params).await {
-            Ok(res) => res,
-            Err(err) => {
-                warn!(context, "sql: Failed query_row: {}", err);
-                None
-            }
-        }
+    pub async fn table_exists(&self, name: impl AsRef<str>) -> Result<bool> {
+        self.exists(
+            "SELECT name FROM sqlite_master WHERE type = 'table' AND name=?",
+            paramsx![name.as_ref()],
+        )
+        .await
     }
 
     /// Set private configuration options.
@@ -316,58 +269,49 @@ impl Sql {
     /// will already have been logged.
     pub async fn set_raw_config(
         &self,
-        context: &Context,
+        _context: &Context,
         key: impl AsRef<str>,
         value: Option<&str>,
     ) -> Result<()> {
-        if !self.is_open().await {
-            error!(context, "set_raw_config(): Database not ready.");
-            return Err(Error::SqlNoConnection);
-        }
-
         let key = key.as_ref();
-        let res = if let Some(ref value) = value {
+
+        if let Some(ref value) = value {
             let exists = self
-                .exists("SELECT value FROM config WHERE keyname=?;", paramsv![key])
+                .exists("SELECT value FROM config WHERE keyname=?;", paramsx![key])
                 .await?;
             if exists {
                 self.execute(
                     "UPDATE config SET value=? WHERE keyname=?;",
-                    paramsv![(*value).to_string(), key.to_string()],
+                    paramsx![value, key],
                 )
-                .await
+                .await?;
             } else {
                 self.execute(
                     "INSERT INTO config (keyname, value) VALUES (?, ?);",
-                    paramsv![key.to_string(), (*value).to_string()],
+                    paramsx![key, value],
                 )
-                .await
+                .await?;
             }
         } else {
-            self.execute("DELETE FROM config WHERE keyname=?;", paramsv![key])
-                .await
-        };
-
-        match res {
-            Ok(_) => Ok(()),
-            Err(err) => {
-                error!(context, "set_raw_config(): Cannot change value. {:?}", &err);
-                Err(err)
-            }
+            self.execute("DELETE FROM config WHERE keyname=?;", paramsx![key])
+                .await?;
         }
+
+        Ok(())
     }
 
     /// Get configuration options from the database.
-    pub async fn get_raw_config(&self, context: &Context, key: impl AsRef<str>) -> Option<String> {
+    pub async fn get_raw_config(&self, key: impl AsRef<str>) -> Option<String> {
         if !self.is_open().await || key.as_ref().is_empty() {
             return None;
         }
-        self.query_get_value(
-            context,
+        self.query_row(
             "SELECT value FROM config WHERE keyname=?;",
-            paramsv![key.as_ref().to_string()],
+            paramsx![key.as_ref().to_string()],
         )
         .await
+        .ok()
+        .map(|(res,)| res)
     }
 
     pub async fn set_raw_config_int(
@@ -380,16 +324,14 @@ impl Sql {
             .await
     }
 
-    pub async fn get_raw_config_int(&self, context: &Context, key: impl AsRef<str>) -> Option<i32> {
-        self.get_raw_config(context, key)
-            .await
-            .and_then(|s| s.parse().ok())
+    pub async fn get_raw_config_int(&self, key: impl AsRef<str>) -> Option<i32> {
+        self.get_raw_config(key).await.and_then(|s| s.parse().ok())
     }
 
-    pub async fn get_raw_config_bool(&self, context: &Context, key: impl AsRef<str>) -> bool {
+    pub async fn get_raw_config_bool(&self, key: impl AsRef<str>) -> bool {
         // Not the most obvious way to encode bool as string, but it is matter
         // of backward compatibility.
-        let res = self.get_raw_config_int(context, key).await;
+        let res = self.get_raw_config_int(key).await;
         res.unwrap_or_default() > 0
     }
 
@@ -411,14 +353,8 @@ impl Sql {
             .await
     }
 
-    pub async fn get_raw_config_int64(
-        &self,
-        context: &Context,
-        key: impl AsRef<str>,
-    ) -> Option<i64> {
-        self.get_raw_config(context, key)
-            .await
-            .and_then(|r| r.parse().ok())
+    pub async fn get_raw_config_int64(&self, key: impl AsRef<str>) -> Option<i64> {
+        self.get_raw_config(key).await.and_then(|r| r.parse().ok())
     }
 
     /// Alternative to sqlite3_last_insert_rowid() which MUST NOT be used due to race conditions, see comment above.
@@ -426,128 +362,88 @@ impl Sql {
     /// eg. if a Message-ID is split into different messages.
     pub async fn get_rowid(
         &self,
-        _context: &Context,
         table: impl AsRef<str>,
         field: impl AsRef<str>,
         value: impl AsRef<str>,
     ) -> Result<u32> {
-        let res = {
-            let mut conn = self.get_conn().await?;
-            get_rowid(&mut conn, table, field, value)
-        };
+        // alternative to sqlite3_last_insert_rowid() which MUST NOT be used due to race conditions, see comment above.
+        // the ORDER BY ensures, this function always returns the most recent id,
+        // eg. if a Message-ID is split into different messages.
+        let query = format!(
+            "SELECT id FROM {} WHERE {}=? ORDER BY id DESC",
+            table.as_ref(),
+            field.as_ref(),
+        );
 
-        res.map_err(Into::into)
+        let res: i64 = self.query_value(&query, paramsx![value.as_ref()]).await?;
+
+        Ok(res as u32)
     }
 
     pub async fn get_rowid2(
         &self,
-        _context: &Context,
         table: impl AsRef<str>,
         field: impl AsRef<str>,
         value: i64,
         field2: impl AsRef<str>,
         value2: i32,
     ) -> Result<u32> {
-        let res = {
-            let mut conn = self.get_conn().await?;
-            get_rowid2(&mut conn, table, field, value, field2, value2)
-        };
+        let query = format!(
+            "SELECT id FROM {} WHERE {}=? AND {}=? ORDER BY id DESC",
+            table.as_ref(),
+            field.as_ref(),
+            field2.as_ref(),
+        );
 
-        res.map_err(Into::into)
+        let res: i64 = self
+            .query_value(query.as_ref(), paramsx![value, value2])
+            .await?;
+
+        Ok(res as u32)
     }
 }
 
-pub fn get_rowid(
-    conn: &mut Connection,
-    table: impl AsRef<str>,
-    field: impl AsRef<str>,
-    value: impl AsRef<str>,
-) -> std::result::Result<u32, SqlError> {
-    // alternative to sqlite3_last_insert_rowid() which MUST NOT be used due to race conditions, see comment above.
-    // the ORDER BY ensures, this function always returns the most recent id,
-    // eg. if a Message-ID is split into different messages.
-    let query = format!(
-        "SELECT id FROM {} WHERE {}=? ORDER BY id DESC",
-        table.as_ref(),
-        field.as_ref(),
-    );
-
-    conn.query_row(&query, params![value.as_ref()], |row| row.get::<_, u32>(0))
-}
-
-pub fn get_rowid2(
-    conn: &mut Connection,
-    table: impl AsRef<str>,
-    field: impl AsRef<str>,
-    value: i64,
-    field2: impl AsRef<str>,
-    value2: i32,
-) -> std::result::Result<u32, SqlError> {
-    conn.query_row(
-        &format!(
-            "SELECT id FROM {} WHERE {}={} AND {}={} ORDER BY id DESC",
-            table.as_ref(),
-            field.as_ref(),
-            value,
-            field2.as_ref(),
-            value2,
-        ),
-        params![],
-        |row| row.get::<_, u32>(0),
-    )
-}
-
-pub async fn housekeeping(context: &Context) {
+pub async fn housekeeping(context: &Context) -> Result<()> {
     let mut files_in_use = HashSet::new();
-    let mut unreferenced_count = 0;
+    let mut unreferenced_count = 0usize;
 
     info!(context, "Start housekeeping...");
     maybe_add_from_param(
         context,
         &mut files_in_use,
-        "SELECT param FROM msgs  WHERE chat_id!=3   AND type!=10;",
+        "SELECT param FROM msgs WHERE chat_id!=3 AND type!=10;",
         Param::File,
     )
-    .await;
+    .await?;
     maybe_add_from_param(
         context,
         &mut files_in_use,
         "SELECT param FROM jobs;",
         Param::File,
     )
-    .await;
+    .await?;
     maybe_add_from_param(
         context,
         &mut files_in_use,
         "SELECT param FROM chats;",
         Param::ProfileImage,
     )
-    .await;
+    .await?;
     maybe_add_from_param(
         context,
         &mut files_in_use,
         "SELECT param FROM contacts;",
         Param::ProfileImage,
     )
-    .await;
+    .await?;
 
-    context
-        .sql
-        .query_map(
-            "SELECT value FROM config;",
-            paramsv![],
-            |row| row.get::<_, String>(0),
-            |rows| {
-                for row in rows {
-                    maybe_add_file(&mut files_in_use, row?);
-                }
-                Ok(())
-            },
-        )
-        .await
-        .unwrap_or_else(|err| {
-            warn!(context, "sql: failed query: {}", err);
-        });
+    let pool = context.sql.get_pool().await?;
+    let mut rows = sqlx::query_as("SELECT value FROM config;").fetch(&pool);
+
+    while let Some(row) = rows.next().await {
+        let (row,): (String,) = row?;
+        maybe_add_file(&mut files_in_use, row);
+    }
 
     info!(context, "{} files in use.", files_in_use.len(),);
     /* go through directory and delete unused files */
@@ -622,6 +518,8 @@ pub async fn housekeeping(context: &Context) {
     }
 
     info!(context, "Housekeeping done.",);
+
+    Ok(())
 }
 
 fn is_file_in_use(files_in_use: &HashSet<String>, namespc_opt: Option<&str>, name: &str) -> bool {
@@ -651,27 +549,23 @@ async fn maybe_add_from_param(
     files_in_use: &mut HashSet<String>,
     query: &str,
     param_id: Param,
-) {
-    context
-        .sql
-        .query_map(
-            query,
-            paramsv![],
-            |row| row.get::<_, String>(0),
-            |rows| {
-                for row in rows {
-                    let param: Params = row?.parse().unwrap_or_default();
-                    if let Some(file) = param.get(param_id) {
-                        maybe_add_file(files_in_use, file);
-                    }
-                }
-                Ok(())
-            },
-        )
-        .await
-        .unwrap_or_else(|err| {
-            warn!(context, "sql: failed to add_from_param: {}", err);
-        });
+) -> Result<()> {
+    info!(context, "maybe_add_from_param: {}", query);
+    let pool = context.sql.get_pool().await?;
+    let mut rows = sqlx::query_as(query).fetch(&pool);
+
+    while let Some(row) = rows.next().await {
+        let (row,): (String,) = row?;
+        info!(context, "param: {}", &row);
+        let param: Params = row.parse().unwrap_or_default();
+
+        if let Some(file) = param.get(param_id) {
+            info!(context, "got file: {:?} {}", param_id, file);
+            maybe_add_file(files_in_use, file);
+        }
+    }
+
+    Ok(())
 }
 
 #[allow(clippy::cognitive_complexity)]
@@ -690,40 +584,13 @@ async fn open(
         return Err(Error::SqlAlreadyOpen.into());
     }
 
-    let mut open_flags = OpenFlags::SQLITE_OPEN_NO_MUTEX;
     if readonly {
-        open_flags.insert(OpenFlags::SQLITE_OPEN_READ_ONLY);
-    } else {
-        open_flags.insert(OpenFlags::SQLITE_OPEN_READ_WRITE);
-        open_flags.insert(OpenFlags::SQLITE_OPEN_CREATE);
-    }
-
-    // this actually creates min_idle database handles just now.
-    // therefore, with_init() must not try to modify the database as otherwise
-    // we easily get busy-errors (eg. table-creation, journal_mode etc. should be done on only one handle)
-    let mgr = r2d2_sqlite::SqliteConnectionManager::file(dbfile.as_ref())
-        .with_flags(open_flags)
-        .with_init(|c| {
-            c.execute_batch(&format!(
-                "PRAGMA secure_delete=on; PRAGMA busy_timeout = {};",
-                Duration::from_secs(10).as_millis()
-            ))?;
-            Ok(())
-        });
-    let pool = r2d2::Pool::builder()
-        .min_idle(Some(2))
-        .max_size(10)
-        .connection_timeout(Duration::from_secs(60))
-        .build(mgr)
-        .map_err(Error::ConnectionPool)?;
-
-    {
-        *sql.pool.write().await = Some(pool);
+        // TODO: readonly mode
     }
 
     let xpool = sqlx::SqlitePool::builder()
         .min_size(1)
-        .max_size(1)
+        .max_size(4)
         .build(&format!("sqlite://{}", dbfile.as_ref().to_string_lossy()))
         .await?;
 
@@ -732,24 +599,14 @@ async fn open(
     }
 
     if !readonly {
-        // journal_mode is persisted, it is sufficient to change it only for one handle.
-        // (nb: execute() always returns errors for this PRAGMA call, just discard it.
-        // but even if execute() would handle errors more gracefully, we should continue on errors -
-        // systems might not be able to handle WAL, in which case the standard-journal is used.
-        // that may be not optimal, but better than not working at all :)
-        sql.execute("PRAGMA journal_mode=WAL;", paramsv![])
-            .await
-            .ok();
-
         let mut exists_before_update = false;
         let mut dbversion_before_update: i32 = -1;
 
         if sql.table_exists("config").await? {
             exists_before_update = true;
-            dbversion_before_update = sql
-                .get_raw_config_int(context, "dbversion")
-                .await
-                .unwrap_or_default();
+            if let Some(version) = sql.get_raw_config_int("dbversion").await {
+                dbversion_before_update = version;
+            }
         }
 
         // (1) update low-level database structure.
@@ -776,25 +633,19 @@ async fn open(
 
         if recalc_fingerprints {
             info!(context, "[migration] recalc fingerprints");
-            let addrs = sql
-                .query_map(
-                    "SELECT addr FROM acpeerstates;",
-                    paramsv![],
-                    |row| row.get::<_, String>(0),
-                    |addrs| {
-                        addrs
-                            .collect::<std::result::Result<Vec<_>, _>>()
-                            .map_err(Into::into)
-                    },
-                )
-                .await?;
-            for addr in &addrs {
-                if let Some(ref mut peerstate) = Peerstate::from_addr(context, addr).await {
+            let pool = context.sql.get_pool().await?;
+            let mut rows = sqlx::query_as("SELECT addr FROM acpeerstates;").fetch(&pool);
+
+            while let Some(addr) = rows.next().await {
+                let (addr,): (String,) = addr?;
+
+                if let Ok(ref mut peerstate) = Peerstate::from_addr(context, &addr).await {
                     peerstate.recalc_fingerprint();
                     peerstate.save_to_db(sql, false).await?;
                 }
             }
         }
+
         if update_icons {
             update_saved_messages_icon(context).await?;
             update_device_icon(context).await?;
@@ -812,10 +663,12 @@ async fn prune_tombstones(context: &Context) -> Result<()> {
     context
         .sql
         .execute(
-            "DELETE FROM msgs \
-         WHERE (chat_id = ? OR hidden) \
-         AND server_uid = 0",
-            paramsv![DC_CHAT_ID_TRASH],
+            r#"
+DELETE FROM msgs
+  WHERE (chat_id = ? OR hidden)
+  AND server_uid = 0
+"#,
+            paramsx![DC_CHAT_ID_TRASH as i32],
         )
         .await?;
     Ok(())