aljaz/cdk
1
0
Fork 0
mirror of https://github.com/aljazceru/cdk.git synced 2025-12-19 05:35:18 +01:00
Code Issues Packages Projects Releases Wiki Activity

Working on a better database abstraction (#931)

Browse Source 
* Working on a better database abstraction

After [this question in the chat](https://matrix.to/#/!oJFtttFHGfnTGrIjvD:matrix.cashu.space/$oJFtttFHGfnTGrIjvD:matrix.cashu.space/$I5ZtjJtBM0ctltThDYpoCwClZFlM6PHzf8q2Rjqmso8)
regarding a database transaction within the same function, I realized a few
design flaws in our SQL database abstraction, particularly regarding
transactions.

1. Our upper abstraction got it right, where a transaction is bound with `&mut
   self`, so Rust knows how to handle its lifetime with' async/await'.
2. The raw database does not; instead, it returns &self, and beginning a
   transaction takes &self as well, which is problematic for Rust, but that's not
   all. It is fundamentally wrong. A transaction should take &mut self when
   beginning a transaction, as that connection is bound to a transaction and
   should not be returned to the pool. Currently, that responsibility lies with
   the implementor. If a mistake is made, a transaction could be executed in two
   or more connections.
3. The way a database is bound to our store layer is through a single struct,
   which may or may not internally utilize our connection pool. This is also
   another design flow, in this PR, a connection pool is owned, and to use a
   connection, it should be requested, and that connection is reference with
   mutable when beginning a transaction

* Improve the abstraction with fewer generics

As suggested by @thesimplekid

* Add BEGIN IMMEDIATE for SQLite
This commit is contained in:
C
2025-08-06 03:58:03 -03:00
committed by GitHub
parent cceea654fe
commit ad8f1ece5c
13 changed files with 721 additions and 1122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/cdk-sql-common/Cargo.toml
View File

@@ -23,6 +23,7 @@ cdk-common = { workspace = true, features = ["test"] }
bitcoin.workspace = true
thiserror.workspace = true
tracing.workspace = true
tokio.workspace = true
serde.workspace = true
serde_json.workspace = true
lightning-invoice.workspace = true

7
crates/cdk-sql-common/src/common.rs
View File

@@ -3,11 +3,14 @@ use crate::stmt::query;
/// Migrates the migration generated by `build.rs`
#[inline(always)]
pub async fn migrate<C: DatabaseExecutor>(
pub async fn migrate<C>(
conn: &C,
db_prefix: &str,
migrations: &[(&str, &str, &str)],
) -> Result<(), cdk_common::database::Error> {
) -> Result<(), cdk_common::database::Error>
where
C: DatabaseExecutor,
{
query(
r#"
CREATE TABLE IF NOT EXISTS migrations (

170
crates/cdk-sql-common/src/database.rs
View File

@@ -1,10 +1,12 @@
//! Database traits definition
use std::fmt::Debug;
use std::marker::PhantomData;
use std::ops::{Deref, DerefMut};
use cdk_common::database::Error;
use crate::stmt::{Column, Statement};
use crate::stmt::{query, Column, Statement};
/// Database Executor
///
@@ -32,22 +34,170 @@ pub trait DatabaseExecutor: Debug + Sync + Send {
/// Database transaction trait
#[async_trait::async_trait]
pub trait DatabaseTransaction<'a>: Debug + DatabaseExecutor + Send + Sync {
pub trait DatabaseTransaction<DB>
where
DB: DatabaseExecutor,
{
/// Consumes the current transaction committing the changes
async fn commit(self) -> Result<(), Error>;
async fn commit(conn: &mut DB) -> Result<(), Error>;
/// Begin a transaction
async fn begin(conn: &mut DB) -> Result<(), Error>;
/// Consumes the transaction rolling back all changes
async fn rollback(self) -> Result<(), Error>;
async fn rollback(conn: &mut DB) -> Result<(), Error>;
}
/// Database connection with a transaction
#[derive(Debug)]
pub struct ConnectionWithTransaction<DB, W>
where
DB: DatabaseConnector + 'static,
W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
{
inner: Option<W>,
}
impl<DB, W> ConnectionWithTransaction<DB, W>
where
DB: DatabaseConnector,
W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
{
/// Creates a new transaction
pub async fn new(mut inner: W) -> Result<Self, Error> {
DB::Transaction::begin(inner.deref_mut()).await?;
Ok(Self { inner: Some(inner) })
}
/// Commits the transaction consuming it and releasing the connection back to the pool (or
/// disconnecting)
pub async fn commit(mut self) -> Result<(), Error> {
let mut conn = self
.inner
.take()
.ok_or(Error::Internal("Missing connection".to_owned()))?;
DB::Transaction::commit(&mut conn).await?;
Ok(())
}
/// Rollback the transaction consuming it and releasing the connection back to the pool (or
/// disconnecting)
pub async fn rollback(mut self) -> Result<(), Error> {
let mut conn = self
.inner
.take()
.ok_or(Error::Internal("Missing connection".to_owned()))?;
DB::Transaction::rollback(&mut conn).await?;
Ok(())
}
}
impl<DB, W> Drop for ConnectionWithTransaction<DB, W>
where
DB: DatabaseConnector,
W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
{
fn drop(&mut self) {
if let Some(mut conn) = self.inner.take() {
tokio::spawn(async move {
let _ = DB::Transaction::rollback(conn.deref_mut()).await;
});
}
}
}
#[async_trait::async_trait]
impl<DB, W> DatabaseExecutor for ConnectionWithTransaction<DB, W>
where
DB: DatabaseConnector,
W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
{
fn name() -> &'static str {
"Transaction"
}
/// Executes a query and returns the affected rows
async fn execute(&self, statement: Statement) -> Result<usize, Error> {
self.inner
.as_ref()
.ok_or(Error::Internal("Missing internal connection".to_owned()))?
.execute(statement)
.await
}
/// Runs the query and returns the first row or None
async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
self.inner
.as_ref()
.ok_or(Error::Internal("Missing internal connection".to_owned()))?
.fetch_one(statement)
.await
}
/// Runs the query and returns the first row or None
async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
self.inner
.as_ref()
.ok_or(Error::Internal("Missing internal connection".to_owned()))?
.fetch_all(statement)
.await
}
/// Fetches the first row and column from a query
async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
self.inner
.as_ref()
.ok_or(Error::Internal("Missing internal connection".to_owned()))?
.pluck(statement)
.await
}
/// Batch execution
async fn batch(&self, statement: Statement) -> Result<(), Error> {
self.inner
.as_ref()
.ok_or(Error::Internal("Missing internal connection".to_owned()))?
.batch(statement)
.await
}
}
/// Generic transaction handler for SQLite
pub struct GenericTransactionHandler<W>(PhantomData<W>);
#[async_trait::async_trait]
impl<W> DatabaseTransaction<W> for GenericTransactionHandler<W>
where
W: DatabaseExecutor,
{
/// Consumes the current transaction committing the changes
async fn commit(conn: &mut W) -> Result<(), Error> {
query("COMMIT")?.execute(conn).await?;
Ok(())
}
/// Begin a transaction
async fn begin(conn: &mut W) -> Result<(), Error> {
query("BEGIN")?.execute(conn).await?;
Ok(())
}
/// Consumes the transaction rolling back all changes
async fn rollback(conn: &mut W) -> Result<(), Error> {
query("ROLLBACK")?.execute(conn).await?;
Ok(())
}
}
/// Database connector
#[async_trait::async_trait]
pub trait DatabaseConnector: Debug + DatabaseExecutor + Send + Sync {
/// Transaction type for this database connection
type Transaction<'a>: DatabaseTransaction<'a>
/// Database static trait for the database
type Transaction: DatabaseTransaction<Self>
where
Self: 'a;
/// Begin a new transaction
async fn begin(&self) -> Result<Self::Transaction<'_>, Error>;
Self: Sized;
}

65
crates/cdk-sql-common/src/mint/auth/mod.rs
View File

@@ -1,8 +1,9 @@
//! SQL Mint Auth
use std::collections::HashMap;
use std::marker::PhantomData;
use std::fmt::Debug;
use std::str::FromStr;
use std::sync::Arc;
use async_trait::async_trait;
use cdk_common::database::{self, MintAuthDatabase, MintAuthTransaction};
@@ -15,37 +16,38 @@ use tracing::instrument;
use super::{sql_row_to_blind_signature, sql_row_to_keyset_info, SQLTransaction};
use crate::column_as_string;
use crate::common::migrate;
use crate::database::{DatabaseConnector, DatabaseTransaction};
use crate::database::{ConnectionWithTransaction, DatabaseExecutor};
use crate::mint::Error;
use crate::pool::{DatabasePool, Pool, PooledResource};
use crate::stmt::query;
/// Mint SQL Database
#[derive(Debug, Clone)]
pub struct SQLMintAuthDatabase<DB>
pub struct SQLMintAuthDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
db: DB,
pool: Arc<Pool<RM>>,
}
impl<DB> SQLMintAuthDatabase<DB>
impl<RM> SQLMintAuthDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
/// Creates a new instance
pub async fn new<X>(db: X) -> Result<Self, Error>
where
X: Into<DB>,
X: Into<RM::Config>,
{
let db = db.into();
Self::migrate(&db).await?;
Ok(Self { db })
let pool = Pool::new(db.into());
Self::migrate(pool.get().map_err(|e| Error::Database(Box::new(e)))?).await?;
Ok(Self { pool })
}
/// Migrate
async fn migrate(conn: &DB) -> Result<(), Error> {
let tx = conn.begin().await?;
migrate(&tx, DB::name(), MIGRATIONS).await?;
async fn migrate(conn: PooledResource<RM>) -> Result<(), Error> {
let tx = ConnectionWithTransaction::new(conn).await?;
migrate(&tx, RM::Connection::name(), MIGRATIONS).await?;
tx.commit().await?;
Ok(())
}
@@ -56,9 +58,9 @@ mod migrations;
#[async_trait]
impl<'a, T> MintAuthTransaction<database::Error> for SQLTransaction<'a, T>
impl<RM> MintAuthTransaction<database::Error> for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
#[instrument(skip(self))]
async fn set_active_keyset(&mut self, id: Id) -> Result<(), database::Error> {
@@ -233,9 +235,9 @@ where
}
#[async_trait]
impl<DB> MintAuthDatabase for SQLMintAuthDatabase<DB>
impl<RM> MintAuthDatabase for SQLMintAuthDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
type Err = database::Error;
@@ -244,12 +246,15 @@ where
) -> Result<Box<dyn MintAuthTransaction<database::Error> + Send + Sync + 'a>, database::Error>
{
Ok(Box::new(SQLTransaction {
inner: self.db.begin().await?,
_phantom: PhantomData,
inner: ConnectionWithTransaction::new(
self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
)
.await?,
}))
}
async fn get_active_keyset_id(&self) -> Result<Option<Id>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -260,13 +265,14 @@ where
active = 1;
"#,
)?
.pluck(&self.db)
.pluck(&*conn)
.await?
.map(|id| Ok::<_, Error>(column_as_string!(id, Id::from_str, Id::from_bytes)))
.transpose()?)
}
async fn get_keyset_info(&self, id: &Id) -> Result<Option<MintKeySetInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"SELECT
id,
@@ -283,13 +289,14 @@ where
WHERE id=:id"#,
)?
.bind("id", id.to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_keyset_info)
.transpose()?)
}
async fn get_keyset_infos(&self) -> Result<Vec<MintKeySetInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"SELECT
id,
@@ -305,7 +312,7 @@ where
keyset
WHERE id=:id"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_keyset_info)
@@ -313,9 +320,10 @@ where
}
async fn get_proofs_states(&self, ys: &[PublicKey]) -> Result<Vec<Option<State>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut current_states = query(r#"SELECT y, state FROM proof WHERE y IN (:ys)"#)?
.bind_vec("ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|row| {
@@ -333,6 +341,7 @@ where
&self,
blinded_messages: &[PublicKey],
) -> Result<Vec<Option<BlindSignature>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut blinded_signatures = query(
r#"SELECT
keyset_id,
@@ -353,7 +362,7 @@ where
.map(|y| y.to_bytes().to_vec())
.collect(),
)
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|mut row| {
@@ -377,10 +386,11 @@ where
&self,
protected_endpoint: ProtectedEndpoint,
) -> Result<Option<AuthRequired>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(
query(r#"SELECT auth FROM protected_endpoints WHERE endpoint = :endpoint"#)?
.bind("endpoint", serde_json::to_string(&protected_endpoint)?)
.pluck(&self.db)
.pluck(&*conn)
.await?
.map(|auth| {
Ok::<_, Error>(column_as_string!(
@@ -396,8 +406,9 @@ where
async fn get_auth_for_endpoints(
&self,
) -> Result<HashMap<ProtectedEndpoint, Option<AuthRequired>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(r#"SELECT endpoint, auth FROM protected_endpoints"#)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|row| {

170
crates/cdk-sql-common/src/mint/mod.rs
View File

@@ -9,8 +9,9 @@
//! clients in a pool and expose them to an asynchronous environment, making them compatible with
//! Mint.
use std::collections::HashMap;
use std::marker::PhantomData;
use std::fmt::Debug;
use std::str::FromStr;
use std::sync::Arc;
use async_trait::async_trait;
use bitcoin::bip32::DerivationPath;
@@ -38,7 +39,8 @@ use tracing::instrument;
use uuid::Uuid;
use crate::common::migrate;
use crate::database::{DatabaseConnector, DatabaseExecutor, DatabaseTransaction};
use crate::database::{ConnectionWithTransaction, DatabaseExecutor};
use crate::pool::{DatabasePool, Pool, PooledResource};
use crate::stmt::{query, Column};
use crate::{
column_as_nullable_number, column_as_nullable_string, column_as_number, column_as_string,
@@ -57,20 +59,19 @@ pub use auth::SQLMintAuthDatabase;
/// Mint SQL Database
#[derive(Debug, Clone)]
pub struct SQLMintDatabase<DB>
pub struct SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
db: DB,
pool: Arc<Pool<RM>>,
}
/// SQL Transaction Writer
pub struct SQLTransaction<'a, T>
pub struct SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
inner: T,
_phantom: PhantomData<&'a ()>,
inner: ConnectionWithTransaction<RM::Connection, PooledResource<RM>>,
}
#[inline(always)]
@@ -115,24 +116,26 @@ where
Ok(())
}
impl<DB> SQLMintDatabase<DB>
impl<RM> SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
/// Creates a new instance
pub async fn new<X>(db: X) -> Result<Self, Error>
where
X: Into<DB>,
X: Into<RM::Config>,
{
let db = db.into();
Self::migrate(&db).await?;
Ok(Self { db })
let pool = Pool::new(db.into());
Self::migrate(pool.get().map_err(|e| Error::Database(Box::new(e)))?).await?;
Ok(Self { pool })
}
/// Migrate
async fn migrate(conn: &DB) -> Result<(), Error> {
let tx = conn.begin().await?;
migrate(&tx, DB::name(), MIGRATIONS).await?;
async fn migrate(conn: PooledResource<RM>) -> Result<(), Error> {
let tx = ConnectionWithTransaction::new(conn).await?;
migrate(&tx, RM::Connection::name(), MIGRATIONS).await?;
tx.commit().await?;
Ok(())
}
@@ -142,9 +145,10 @@ where
where
R: serde::de::DeserializeOwned,
{
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let value = column_as_string!(query(r#"SELECT value FROM config WHERE id = :id LIMIT 1"#)?
.bind("id", id.to_owned())
.pluck(&self.db)
.pluck(&*conn)
.await?
.ok_or(Error::UnknownQuoteTTL)?);
@@ -153,9 +157,9 @@ where
}
#[async_trait]
impl<'a, T> database::MintProofsTransaction<'a> for SQLTransaction<'a, T>
impl<RM> database::MintProofsTransaction<'_> for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
type Err = Error;
@@ -267,9 +271,9 @@ where
}
#[async_trait]
impl<'a, T> database::MintTransaction<'a, Error> for SQLTransaction<'a, T>
impl<RM> database::MintTransaction<'_, Error> for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
async fn set_mint_info(&mut self, mint_info: MintInfo) -> Result<(), Error> {
Ok(set_to_config(&self.inner, "mint_info", &mint_info).await?)
@@ -281,18 +285,18 @@ where
}
#[async_trait]
impl<'a, T> MintDbWriterFinalizer for SQLTransaction<'a, T>
impl<RM> MintDbWriterFinalizer for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
type Err = Error;
async fn commit(self: Box<Self>) -> Result<(), Error> {
Ok(self.inner.commit().await?)
self.inner.commit().await
}
async fn rollback(self: Box<Self>) -> Result<(), Error> {
Ok(self.inner.rollback().await?)
self.inner.rollback().await
}
}
@@ -357,9 +361,9 @@ WHERE quote_id=:quote_id
}
#[async_trait]
impl<'a, T> MintKeyDatabaseTransaction<'a, Error> for SQLTransaction<'a, T>
impl<RM> MintKeyDatabaseTransaction<'_, Error> for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
async fn add_keyset_info(&mut self, keyset: MintKeySetInfo) -> Result<(), Error> {
query(
@@ -416,9 +420,9 @@ where
}
#[async_trait]
impl<DB> MintKeysDatabase for SQLMintDatabase<DB>
impl<RM> MintKeysDatabase for SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
type Err = Error;
@@ -426,16 +430,19 @@ where
&'a self,
) -> Result<Box<dyn MintKeyDatabaseTransaction<'a, Error> + Send + Sync + 'a>, Error> {
Ok(Box::new(SQLTransaction {
inner: self.db.begin().await?,
_phantom: PhantomData,
inner: ConnectionWithTransaction::new(
self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
)
.await?,
}))
}
async fn get_active_keyset_id(&self, unit: &CurrencyUnit) -> Result<Option<Id>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(
query(r#" SELECT id FROM keyset WHERE active = 1 AND unit IS :unit"#)?
.bind("unit", unit.to_string())
.pluck(&self.db)
.pluck(&*conn)
.await?
.map(|id| match id {
Column::Text(text) => Ok(Id::from_str(&text)?),
@@ -447,8 +454,9 @@ where
}
async fn get_active_keysets(&self) -> Result<HashMap<CurrencyUnit, Id>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(r#"SELECT id, unit FROM keyset WHERE active = 1"#)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|row| {
@@ -461,6 +469,7 @@ where
}
async fn get_keyset_info(&self, id: &Id) -> Result<Option<MintKeySetInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"SELECT
id,
@@ -477,13 +486,14 @@ where
WHERE id=:id"#,
)?
.bind("id", id.to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_keyset_info)
.transpose()?)
}
async fn get_keyset_infos(&self) -> Result<Vec<MintKeySetInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"SELECT
id,
@@ -499,7 +509,7 @@ where
keyset
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_keyset_info)
@@ -508,9 +518,9 @@ where
}
#[async_trait]
impl<'a, T> MintQuotesTransaction<'a> for SQLTransaction<'a, T>
impl<RM> MintQuotesTransaction<'_> for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
type Err = Error;
@@ -1028,15 +1038,17 @@ VALUES (:quote_id, :amount, :timestamp);
}
#[async_trait]
impl<DB> MintQuotesDatabase for SQLMintDatabase<DB>
impl<RM> MintQuotesDatabase for SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
type Err = Error;
async fn get_mint_quote(&self, quote_id: &Uuid) -> Result<Option<MintQuote>, Self::Err> {
let payments = get_mint_quote_payments(&self.db, quote_id).await?;
let issuance = get_mint_quote_issuance(&self.db, quote_id).await?;
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let payments = get_mint_quote_payments(&*conn, quote_id).await?;
let issuance = get_mint_quote_issuance(&*conn, quote_id).await?;
Ok(query(
r#"
@@ -1058,7 +1070,7 @@ where
WHERE id = :id"#,
)?
.bind("id", quote_id.as_hyphenated().to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(|row| sql_row_to_mint_quote(row, payments, issuance))
.transpose()?)
@@ -1068,6 +1080,7 @@ where
&self,
request: &str,
) -> Result<Option<MintQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut mint_quote = query(
r#"
SELECT
@@ -1088,14 +1101,14 @@ where
WHERE request = :request"#,
)?
.bind("request", request.to_owned())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(|row| sql_row_to_mint_quote(row, vec![], vec![]))
.transpose()?;
if let Some(quote) = mint_quote.as_mut() {
let payments = get_mint_quote_payments(&self.db, &quote.id).await?;
let issuance = get_mint_quote_issuance(&self.db, &quote.id).await?;
let payments = get_mint_quote_payments(&*conn, &quote.id).await?;
let issuance = get_mint_quote_issuance(&*conn, &quote.id).await?;
quote.issuance = issuance;
quote.payments = payments;
}
@@ -1107,6 +1120,7 @@ where
&self,
request_lookup_id: &PaymentIdentifier,
) -> Result<Option<MintQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut mint_quote = query(
r#"
SELECT
@@ -1130,15 +1144,15 @@ where
)?
.bind("request_lookup_id", request_lookup_id.to_string())
.bind("request_lookup_id_kind", request_lookup_id.kind())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(|row| sql_row_to_mint_quote(row, vec![], vec![]))
.transpose()?;
// TODO: these should use an sql join so they can be done in one query
if let Some(quote) = mint_quote.as_mut() {
let payments = get_mint_quote_payments(&self.db, &quote.id).await?;
let issuance = get_mint_quote_issuance(&self.db, &quote.id).await?;
let payments = get_mint_quote_payments(&*conn, &quote.id).await?;
let issuance = get_mint_quote_issuance(&*conn, &quote.id).await?;
quote.issuance = issuance;
quote.payments = payments;
}
@@ -1147,6 +1161,7 @@ where
}
async fn get_mint_quotes(&self) -> Result<Vec<MintQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut mint_quotes = query(
r#"
SELECT
@@ -1166,15 +1181,15 @@ where
mint_quote
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|row| sql_row_to_mint_quote(row, vec![], vec![]))
.collect::<Result<Vec<_>, _>>()?;
for quote in mint_quotes.as_mut_slice() {
let payments = get_mint_quote_payments(&self.db, &quote.id).await?;
let issuance = get_mint_quote_issuance(&self.db, &quote.id).await?;
let payments = get_mint_quote_payments(&*conn, &quote.id).await?;
let issuance = get_mint_quote_issuance(&*conn, &quote.id).await?;
quote.issuance = issuance;
quote.payments = payments;
}
@@ -1183,6 +1198,7 @@ where
}
async fn get_melt_quote(&self, quote_id: &Uuid) -> Result<Option<mint::MeltQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -1207,13 +1223,14 @@ where
"#,
)?
.bind("id", quote_id.as_hyphenated().to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_melt_quote)
.transpose()?)
}
async fn get_melt_quotes(&self) -> Result<Vec<mint::MeltQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -1235,7 +1252,7 @@ where
melt_quote
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_melt_quote)
@@ -1244,13 +1261,14 @@ where
}
#[async_trait]
impl<DB> MintProofsDatabase for SQLMintDatabase<DB>
impl<RM> MintProofsDatabase for SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
type Err = Error;
async fn get_proofs_by_ys(&self, ys: &[PublicKey]) -> Result<Vec<Option<Proof>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut proofs = query(
r#"
SELECT
@@ -1267,7 +1285,7 @@ where
"#,
)?
.bind_vec("ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|mut row| {
@@ -1286,6 +1304,7 @@ where
}
async fn get_proof_ys_by_quote_id(&self, quote_id: &Uuid) -> Result<Vec<PublicKey>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -1301,7 +1320,7 @@ where
"#,
)?
.bind("quote_id", quote_id.as_hyphenated().to_string())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_proof)
@@ -1310,7 +1329,8 @@ where
}
async fn get_proofs_states(&self, ys: &[PublicKey]) -> Result<Vec<Option<State>>, Self::Err> {
let mut current_states = get_current_states(&self.db, ys).await?;
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut current_states = get_current_states(&*conn, ys).await?;
Ok(ys.iter().map(|y| current_states.remove(y)).collect())
}
@@ -1319,6 +1339,7 @@ where
&self,
keyset_id: &Id,
) -> Result<(Proofs, Vec<Option<State>>), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -1335,7 +1356,7 @@ where
"#,
)?
.bind("keyset_id", keyset_id.to_string())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_proof_with_state)
@@ -1346,9 +1367,9 @@ where
}
#[async_trait]
impl<'a, T> MintSignatureTransaction<'a> for SQLTransaction<'a, T>
impl<RM> MintSignatureTransaction<'_> for SQLTransaction<RM>
where
T: DatabaseTransaction<'a>,
RM: DatabasePool + 'static,
{
type Err = Error;
@@ -1436,9 +1457,9 @@ where
}
#[async_trait]
impl<DB> MintSignaturesDatabase for SQLMintDatabase<DB>
impl<RM> MintSignaturesDatabase for SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
type Err = Error;
@@ -1446,6 +1467,7 @@ where
&self,
blinded_messages: &[PublicKey],
) -> Result<Vec<Option<BlindSignature>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut blinded_signatures = query(
r#"SELECT
keyset_id,
@@ -1466,7 +1488,7 @@ where
.map(|b_| b_.to_bytes().to_vec())
.collect(),
)
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|mut row| {
@@ -1490,6 +1512,7 @@ where
&self,
keyset_id: &Id,
) -> Result<Vec<BlindSignature>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -1505,7 +1528,7 @@ where
"#,
)?
.bind("keyset_id", keyset_id.to_string())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_blind_signature)
@@ -1517,6 +1540,7 @@ where
&self,
quote_id: &Uuid,
) -> Result<Vec<BlindSignature>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -1532,7 +1556,7 @@ where
"#,
)?
.bind("quote_id", quote_id.to_string())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_blind_signature)
@@ -1541,16 +1565,18 @@ where
}
#[async_trait]
impl<DB> MintDatabase<Error> for SQLMintDatabase<DB>
impl<RM> MintDatabase<Error> for SQLMintDatabase<RM>
where
DB: DatabaseConnector,
RM: DatabasePool + 'static,
{
async fn begin_transaction<'a>(
&'a self,
) -> Result<Box<dyn database::MintTransaction<'a, Error> + Send + Sync + 'a>, Error> {
Ok(Box::new(SQLTransaction {
inner: self.db.begin().await?,
_phantom: PhantomData,
inner: ConnectionWithTransaction::new(
self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
)
.await?,
}))
}

65
crates/cdk-sql-common/src/pool.rs
View File

@@ -8,9 +8,14 @@ use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;
use crate::database::DatabaseConnector;
/// Pool error
#[derive(thiserror::Error, Debug)]
pub enum Error<E> {
#[derive(Debug, thiserror::Error)]
pub enum Error<E>
where
E: std::error::Error + Send + Sync + 'static,
{
/// Mutex Poison Error
#[error("Internal: PoisonError")]
Poison,
@@ -24,16 +29,25 @@ pub enum Error<E> {
Resource(#[from] E),
}
/// Configuration
pub trait DatabaseConfig: Clone + Debug + Send + Sync {
/// Max resource sizes
fn max_size(&self) -> usize;
/// Default timeout
fn default_timeout(&self) -> Duration;
}
/// Trait to manage resources
pub trait ResourceManager: Debug {
pub trait DatabasePool: Debug {
/// The resource to be pooled
type Resource: Debug;
type Connection: DatabaseConnector;
/// The configuration that is needed in order to create the resource
type Config: Clone + Debug;
type Config: DatabaseConfig;
/// The error the resource may return when creating a new instance
type Error: Debug;
type Error: Debug + std::error::Error + Send + Sync + 'static;
/// Creates a new resource with a given config.
///
@@ -43,20 +57,20 @@ pub trait ResourceManager: Debug {
config: &Self::Config,
stale: Arc<AtomicBool>,
timeout: Duration,
) -> Result<Self::Resource, Error<Self::Error>>;
) -> Result<Self::Connection, Error<Self::Error>>;
/// The object is dropped
fn drop(_resource: Self::Resource) {}
fn drop(_resource: Self::Connection) {}
}
/// Generic connection pool of resources R
#[derive(Debug)]
pub struct Pool<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
config: RM::Config,
queue: Mutex<Vec<(Arc<AtomicBool>, RM::Resource)>>,
queue: Mutex<Vec<(Arc<AtomicBool>, RM::Connection)>>,
in_use: AtomicUsize,
max_size: usize,
default_timeout: Duration,
@@ -66,15 +80,24 @@ where
/// The pooled resource
pub struct PooledResource<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
resource: Option<(Arc<AtomicBool>, RM::Resource)>,
resource: Option<(Arc<AtomicBool>, RM::Connection)>,
pool: Arc<Pool<RM>>,
}
impl<RM> Debug for PooledResource<RM>
where
RM: DatabasePool,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "Resource: {:?}", self.resource)
}
}
impl<RM> Drop for PooledResource<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
fn drop(&mut self) {
if let Some(resource) = self.resource.take() {
@@ -90,9 +113,9 @@ where
impl<RM> Deref for PooledResource<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
type Target = RM::Resource;
type Target = RM::Connection;
fn deref(&self) -> &Self::Target {
&self.resource.as_ref().expect("resource already dropped").1
@@ -101,7 +124,7 @@ where
impl<RM> DerefMut for PooledResource<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.resource.as_mut().expect("resource already dropped").1
@@ -110,17 +133,17 @@ where
impl<RM> Pool<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
/// Creates a new pool
pub fn new(config: RM::Config, max_size: usize, default_timeout: Duration) -> Arc<Self> {
pub fn new(config: RM::Config) -> Arc<Self> {
Arc::new(Self {
default_timeout: config.default_timeout(),
max_size: config.max_size(),
config,
queue: Default::default(),
in_use: Default::default(),
waiter: Default::default(),
default_timeout,
max_size,
})
}
@@ -185,7 +208,7 @@ where
impl<RM> Drop for Pool<RM>
where
RM: ResourceManager,
RM: DatabasePool,
{
fn drop(&mut self) {
if let Ok(mut resources) = self.queue.lock() {

141
crates/cdk-sql-common/src/wallet/mod.rs
View File

@@ -1,7 +1,9 @@
//! SQLite Wallet Database
use std::collections::HashMap;
use std::fmt::Debug;
use std::str::FromStr;
use std::sync::Arc;
use async_trait::async_trait;
use cdk_common::common::ProofInfo;
@@ -17,7 +19,8 @@ use cdk_common::{
use tracing::instrument;
use crate::common::migrate;
use crate::database::DatabaseExecutor;
use crate::database::{ConnectionWithTransaction, DatabaseExecutor};
use crate::pool::{DatabasePool, Pool, PooledResource};
use crate::stmt::{query, Column};
use crate::{
column_as_binary, column_as_nullable_binary, column_as_nullable_number,
@@ -29,36 +32,43 @@ mod migrations;
/// Wallet SQLite Database
#[derive(Debug, Clone)]
pub struct SQLWalletDatabase<T>
pub struct SQLWalletDatabase<RM>
where
T: DatabaseExecutor,
RM: DatabasePool + 'static,
{
db: T,
pool: Arc<Pool<RM>>,
}
impl<DB> SQLWalletDatabase<DB>
impl<RM> SQLWalletDatabase<RM>
where
DB: DatabaseExecutor,
RM: DatabasePool + 'static,
{
/// Creates a new instance
pub async fn new<X>(db: X) -> Result<Self, Error>
where
X: Into<DB>,
X: Into<RM::Config>,
{
let db = db.into();
Self::migrate(&db).await?;
Ok(Self { db })
let pool = Pool::new(db.into());
Self::migrate(pool.get().map_err(|e| Error::Database(Box::new(e)))?).await?;
Ok(Self { pool })
}
/// Migrate [`WalletSqliteDatabase`]
async fn migrate(conn: &DB) -> Result<(), Error> {
migrate(conn, DB::name(), migrations::MIGRATIONS).await?;
async fn migrate(conn: PooledResource<RM>) -> Result<(), Error> {
let tx = ConnectionWithTransaction::new(conn).await?;
migrate(&tx, RM::Connection::name(), migrations::MIGRATIONS).await?;
// Update any existing keys with missing keyset_u32 values
Self::add_keyset_u32(conn).await?;
Self::add_keyset_u32(&tx).await?;
tx.commit().await?;
Ok(())
}
async fn add_keyset_u32(conn: &DB) -> Result<(), Error> {
async fn add_keyset_u32<T>(conn: &T) -> Result<(), Error>
where
T: DatabaseExecutor,
{
// First get the keysets where keyset_u32 on key is null
let keys_without_u32: Vec<Vec<Column>> = query(
r#"
@@ -126,14 +136,16 @@ where
}
#[async_trait]
impl<T> WalletDatabase for SQLWalletDatabase<T>
impl<RM> WalletDatabase for SQLWalletDatabase<RM>
where
T: DatabaseExecutor,
RM: DatabasePool + 'static,
{
type Err = database::Error;
#[instrument(skip(self))]
async fn get_melt_quotes(&self) -> Result<Vec<wallet::MeltQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -149,7 +161,7 @@ where
melt_quote
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_melt_quote)
@@ -212,6 +224,8 @@ where
),
};
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(
r#"
INSERT INTO mint
@@ -253,7 +267,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
.bind("motd", motd)
.bind("mint_time", time.map(|v| v as i64))
.bind("tos_url", tos_url)
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -261,9 +275,11 @@ ON CONFLICT(mint_url) DO UPDATE SET
#[instrument(skip(self))]
async fn remove_mint(&self, mint_url: MintUrl) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(r#"DELETE FROM mint WHERE mint_url=:mint_url"#)?
.bind("mint_url", mint_url.to_string())
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -271,6 +287,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
#[instrument(skip(self))]
async fn get_mint(&self, mint_url: MintUrl) -> Result<Option<MintInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -292,7 +309,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
"#,
)?
.bind("mint_url", mint_url.to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_mint_info)
.transpose()?)
@@ -300,6 +317,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
#[instrument(skip(self))]
async fn get_mints(&self) -> Result<HashMap<MintUrl, Option<MintInfo>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -320,7 +338,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
mint
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(|mut row| {
@@ -340,6 +358,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
old_mint_url: MintUrl,
new_mint_url: MintUrl,
) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let tables = ["mint_quote", "proof"];
for table in &tables {
@@ -352,7 +371,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
))?
.bind("new_mint_url", new_mint_url.to_string())
.bind("old_mint_url", old_mint_url.to_string())
.execute(&self.db)
.execute(&*conn)
.await?;
}
@@ -365,6 +384,8 @@ ON CONFLICT(mint_url) DO UPDATE SET
mint_url: MintUrl,
keysets: Vec<KeySetInfo>,
) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
for keyset in keysets {
query(
r#"
@@ -384,7 +405,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
.bind("input_fee_ppk", keyset.input_fee_ppk as i64)
.bind("final_expiry", keyset.final_expiry.map(|v| v as i64))
.bind("keyset_u32", u32::from(keyset.id))
.execute(&self.db)
.execute(&*conn)
.await?;
}
@@ -396,6 +417,8 @@ ON CONFLICT(mint_url) DO UPDATE SET
&self,
mint_url: MintUrl,
) -> Result<Option<Vec<KeySetInfo>>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let keysets = query(
r#"
SELECT
@@ -410,7 +433,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
"#,
)?
.bind("mint_url", mint_url.to_string())
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_keyset)
@@ -424,6 +447,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
#[instrument(skip(self), fields(keyset_id = %keyset_id))]
async fn get_keyset_by_id(&self, keyset_id: &Id) -> Result<Option<KeySetInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -438,7 +462,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
"#,
)?
.bind("id", keyset_id.to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_keyset)
.transpose()?)
@@ -446,6 +470,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
#[instrument(skip_all)]
async fn add_mint_quote(&self, quote: MintQuote) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(
r#"
INSERT INTO mint_quote
@@ -477,13 +502,14 @@ ON CONFLICT(id) DO UPDATE SET
.bind("payment_method", quote.payment_method.to_string())
.bind("amount_issued", quote.amount_issued.to_i64())
.bind("amount_paid", quote.amount_paid.to_i64())
.execute(&self.db).await?;
.execute(&*conn).await?;
Ok(())
}
#[instrument(skip(self))]
async fn get_mint_quote(&self, quote_id: &str) -> Result<Option<MintQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -505,7 +531,7 @@ ON CONFLICT(id) DO UPDATE SET
"#,
)?
.bind("id", quote_id.to_string())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_mint_quote)
.transpose()?)
@@ -513,6 +539,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn get_mint_quotes(&self) -> Result<Vec<MintQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -528,7 +555,7 @@ ON CONFLICT(id) DO UPDATE SET
mint_quote
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.map(sql_row_to_mint_quote)
@@ -537,9 +564,10 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn remove_mint_quote(&self, quote_id: &str) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(r#"DELETE FROM mint_quote WHERE id=:id"#)?
.bind("id", quote_id.to_string())
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -547,6 +575,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip_all)]
async fn add_melt_quote(&self, quote: wallet::MeltQuote) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(
r#"
INSERT INTO melt_quote
@@ -570,7 +599,7 @@ ON CONFLICT(id) DO UPDATE SET
.bind("fee_reserve", u64::from(quote.fee_reserve) as i64)
.bind("state", quote.state.to_string())
.bind("expiry", quote.expiry as i64)
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -578,6 +607,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn get_melt_quote(&self, quote_id: &str) -> Result<Option<wallet::MeltQuote>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -596,7 +626,7 @@ ON CONFLICT(id) DO UPDATE SET
"#,
)?
.bind("id", quote_id.to_owned())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_melt_quote)
.transpose()?)
@@ -604,9 +634,10 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn remove_melt_quote(&self, quote_id: &str) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(r#"DELETE FROM melt_quote WHERE id=:id"#)?
.bind("id", quote_id.to_owned())
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -614,6 +645,8 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip_all)]
async fn add_keys(&self, keyset: KeySet) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
// Recompute ID for verification
keyset.verify_id()?;
@@ -631,7 +664,7 @@ ON CONFLICT(id) DO UPDATE SET
serde_json::to_string(&keyset.keys).map_err(Error::from)?,
)
.bind("keyset_u32", u32::from(keyset.id))
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -639,6 +672,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self), fields(keyset_id = %keyset_id))]
async fn get_keys(&self, keyset_id: &Id) -> Result<Option<Keys>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -648,7 +682,7 @@ ON CONFLICT(id) DO UPDATE SET
"#,
)?
.bind("id", keyset_id.to_string())
.pluck(&self.db)
.pluck(&*conn)
.await?
.map(|keys| {
let keys = column_as_string!(keys);
@@ -659,9 +693,10 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn remove_keys(&self, id: &Id) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(r#"DELETE FROM key WHERE id = :id"#)?
.bind("id", id.to_string())
.pluck(&self.db)
.pluck(&*conn)
.await?;
Ok(())
@@ -672,6 +707,10 @@ ON CONFLICT(id) DO UPDATE SET
added: Vec<ProofInfo>,
removed_ys: Vec<PublicKey>,
) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let tx = ConnectionWithTransaction::new(conn).await?;
// TODO: Use a transaction for all these operations
for proof in added {
query(
@@ -729,7 +768,7 @@ ON CONFLICT(id) DO UPDATE SET
"dleq_r",
proof.proof.dleq.as_ref().map(|dleq| dleq.r.to_secret_bytes().to_vec()),
)
.execute(&self.db).await?;
.execute(&tx).await?;
}
query(r#"DELETE FROM proof WHERE y IN (:ys)"#)?
@@ -737,9 +776,11 @@ ON CONFLICT(id) DO UPDATE SET
"ys",
removed_ys.iter().map(|y| y.to_bytes().to_vec()).collect(),
)
.execute(&self.db)
.execute(&tx)
.await?;
tx.commit().await?;
Ok(())
}
@@ -751,6 +792,7 @@ ON CONFLICT(id) DO UPDATE SET
state: Option<Vec<State>>,
spending_conditions: Option<Vec<SpendingConditions>>,
) -> Result<Vec<ProofInfo>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -770,7 +812,7 @@ ON CONFLICT(id) DO UPDATE SET
FROM proof
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.filter_map(|row| {
@@ -786,10 +828,11 @@ ON CONFLICT(id) DO UPDATE SET
}
async fn update_proofs_state(&self, ys: Vec<PublicKey>, state: State) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query("UPDATE proof SET state = :state WHERE y IN (:ys)")?
.bind_vec("ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
.bind("state", state.to_string())
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -797,6 +840,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self), fields(keyset_id = %keyset_id))]
async fn increment_keyset_counter(&self, keyset_id: &Id, count: u32) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(
r#"
UPDATE keyset
@@ -806,7 +850,7 @@ ON CONFLICT(id) DO UPDATE SET
)?
.bind("count", count)
.bind("id", keyset_id.to_string())
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -814,6 +858,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self), fields(keyset_id = %keyset_id))]
async fn get_keyset_counter(&self, keyset_id: &Id) -> Result<Option<u32>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -825,7 +870,7 @@ ON CONFLICT(id) DO UPDATE SET
"#,
)?
.bind("id", keyset_id.to_string())
.pluck(&self.db)
.pluck(&*conn)
.await?
.map(|n| Ok::<_, Error>(column_as_number!(n)))
.transpose()?)
@@ -833,6 +878,7 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn add_transaction(&self, transaction: Transaction) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
let mint_url = transaction.mint_url.to_string();
let direction = transaction.direction.to_string();
let unit = transaction.unit.to_string();
@@ -876,7 +922,7 @@ ON CONFLICT(id) DO UPDATE SET
"metadata",
serde_json::to_string(&transaction.metadata).map_err(Error::from)?,
)
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())
@@ -887,6 +933,7 @@ ON CONFLICT(id) DO UPDATE SET
&self,
transaction_id: TransactionId,
) -> Result<Option<Transaction>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -906,7 +953,7 @@ ON CONFLICT(id) DO UPDATE SET
"#,
)?
.bind("id", transaction_id.as_slice().to_vec())
.fetch_one(&self.db)
.fetch_one(&*conn)
.await?
.map(sql_row_to_transaction)
.transpose()?)
@@ -919,6 +966,8 @@ ON CONFLICT(id) DO UPDATE SET
direction: Option<TransactionDirection>,
unit: Option<CurrencyUnit>,
) -> Result<Vec<Transaction>, Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
Ok(query(
r#"
SELECT
@@ -935,7 +984,7 @@ ON CONFLICT(id) DO UPDATE SET
transactions
"#,
)?
.fetch_all(&self.db)
.fetch_all(&*conn)
.await?
.into_iter()
.filter_map(|row| {
@@ -952,9 +1001,11 @@ ON CONFLICT(id) DO UPDATE SET
#[instrument(skip(self))]
async fn remove_transaction(&self, transaction_id: TransactionId) -> Result<(), Self::Err> {
let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
query(r#"DELETE FROM transactions WHERE id=:id"#)?
.bind("id", transaction_id.as_slice().to_vec())
.execute(&self.db)
.execute(&*conn)
.await?;
Ok(())

197
crates/cdk-sqlite/src/async_sqlite.rs Normal file
View File

@@ -0,0 +1,197 @@
//! Simple SQLite
use cdk_common::database::Error;
use cdk_sql_common::database::{DatabaseConnector, DatabaseExecutor, DatabaseTransaction};
use cdk_sql_common::stmt::{query, Column, SqlPart, Statement};
use rusqlite::{ffi, CachedStatement, Connection, Error as SqliteError, ErrorCode};
use tokio::sync::Mutex;
use crate::common::{from_sqlite, to_sqlite};
/// Async Sqlite wrapper
#[derive(Debug)]
pub struct AsyncSqlite {
inner: Mutex<Connection>,
}
impl AsyncSqlite {
pub fn new(inner: Connection) -> Self {
Self {
inner: inner.into(),
}
}
}
impl AsyncSqlite {
fn get_stmt<'a>(
&self,
conn: &'a Connection,
statement: Statement,
) -> Result<CachedStatement<'a>, Error> {
let (sql, placeholder_values) = statement.to_sql()?;
let new_sql = sql.trim().trim_end_matches("FOR UPDATE");
let mut stmt = conn
.prepare_cached(new_sql)
.map_err(|e| Error::Database(Box::new(e)))?;
for (i, value) in placeholder_values.into_iter().enumerate() {
stmt.raw_bind_parameter(i + 1, to_sqlite(value))
.map_err(|e| Error::Database(Box::new(e)))?;
}
Ok(stmt)
}
}
#[inline(always)]
fn to_sqlite_error(err: SqliteError) -> Error {
tracing::error!("Failed query with error {:?}", err);
if let rusqlite::Error::SqliteFailure(
ffi::Error {
code,
extended_code,
},
_,
) = err
{
if code == ErrorCode::ConstraintViolation
&& (extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
|| extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
{
Error::Duplicate
} else {
Error::Database(Box::new(err))
}
} else {
Error::Database(Box::new(err))
}
}
/// SQLite trasanction handler
pub struct SQLiteTransactionHandler;
#[async_trait::async_trait]
impl DatabaseTransaction<AsyncSqlite> for SQLiteTransactionHandler {
/// Consumes the current transaction committing the changes
async fn commit(conn: &mut AsyncSqlite) -> Result<(), Error> {
query("COMMIT")?.execute(conn).await?;
Ok(())
}
/// Begin a transaction
async fn begin(conn: &mut AsyncSqlite) -> Result<(), Error> {
query("BEGIN IMMEDIATE")?.execute(conn).await?;
Ok(())
}
/// Consumes the transaction rolling back all changes
async fn rollback(conn: &mut AsyncSqlite) -> Result<(), Error> {
query("ROLLBACK")?.execute(conn).await?;
Ok(())
}
}
impl DatabaseConnector for AsyncSqlite {
type Transaction = SQLiteTransactionHandler;
}
#[async_trait::async_trait]
impl DatabaseExecutor for AsyncSqlite {
fn name() -> &'static str {
"sqlite"
}
async fn execute(&self, statement: Statement) -> Result<usize, Error> {
let conn = self.inner.lock().await;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
Ok(stmt.raw_execute().map_err(to_sqlite_error)?)
}
async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
let conn = self.inner.lock().await;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
let columns = stmt.column_count();
let mut rows = stmt.raw_query();
rows.next()
.map_err(to_sqlite_error)?
.map(|row| {
(0..columns)
.map(|i| row.get(i).map(from_sqlite))
.collect::<Result<Vec<_>, _>>()
})
.transpose()
.map_err(to_sqlite_error)
}
async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
let conn = self.inner.lock().await;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
let columns = stmt.column_count();
let mut rows = stmt.raw_query();
let mut results = vec![];
while let Some(row) = rows.next().map_err(to_sqlite_error)? {
results.push(
(0..columns)
.map(|i| row.get(i).map(from_sqlite))
.collect::<Result<Vec<_>, _>>()
.map_err(to_sqlite_error)?,
)
}
Ok(results)
}
async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
let conn = self.inner.lock().await;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
let mut rows = stmt.raw_query();
rows.next()
.map_err(to_sqlite_error)?
.map(|row| row.get(0usize).map(from_sqlite))
.transpose()
.map_err(to_sqlite_error)
}
async fn batch(&self, mut statement: Statement) -> Result<(), Error> {
let sql = {
let part = statement
.parts
.pop()
.ok_or(Error::Internal("Empty SQL".to_owned()))?;
if !statement.parts.is_empty() || matches!(part, SqlPart::Placeholder(_, _)) {
return Err(Error::Internal(
"Invalid usage, batch does not support placeholders".to_owned(),
));
}
if let SqlPart::Raw(sql) = part {
sql
} else {
unreachable!()
}
};
self.inner
.lock()
.await
.execute_batch(&sql)
.map_err(to_sqlite_error)
}
}

88
crates/cdk-sqlite/src/common.rs
View File

@@ -1,11 +1,14 @@
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::time::Duration;
use cdk_sql_common::pool::{self, Pool, ResourceManager};
use cdk_sql_common::pool::{self, DatabasePool};
use cdk_sql_common::value::Value;
use rusqlite::Connection;
use crate::async_sqlite;
/// The config need to create a new SQLite connection
#[derive(Clone, Debug)]
pub struct Config {
@@ -13,14 +16,28 @@ pub struct Config {
password: Option<String>,
}
impl pool::DatabaseConfig for Config {
fn default_timeout(&self) -> Duration {
Duration::from_secs(5)
}
fn max_size(&self) -> usize {
if self.password.is_none() {
1
} else {
20
}
}
}
/// Sqlite connection manager
#[derive(Debug)]
pub struct SqliteConnectionManager;
impl ResourceManager for SqliteConnectionManager {
impl DatabasePool for SqliteConnectionManager {
type Config = Config;
type Resource = Connection;
type Connection = async_sqlite::AsyncSqlite;
type Error = rusqlite::Error;
@@ -28,7 +45,7 @@ impl ResourceManager for SqliteConnectionManager {
config: &Self::Config,
_stale: Arc<AtomicBool>,
_timeout: Duration,
) -> Result<Self::Resource, pool::Error<Self::Error>> {
) -> Result<Self::Connection, pool::Error<Self::Error>> {
let conn = if let Some(path) = config.path.as_ref() {
Connection::open(path)?
} else {
@@ -52,35 +69,58 @@ impl ResourceManager for SqliteConnectionManager {
conn.busy_timeout(Duration::from_secs(10))?;
Ok(conn)
Ok(async_sqlite::AsyncSqlite::new(conn))
}
}
/// Create a configured rusqlite connection to a SQLite database.
/// For SQLCipher support, enable the "sqlcipher" feature and pass a password.
pub fn create_sqlite_pool(
path: &str,
password: Option<String>,
) -> Arc<Pool<SqliteConnectionManager>> {
let (config, max_size) = if path.contains(":memory:") {
(
impl From<PathBuf> for Config {
fn from(path: PathBuf) -> Self {
path.to_str().unwrap_or_default().into()
}
}
impl From<(PathBuf, String)> for Config {
fn from((path, password): (PathBuf, String)) -> Self {
(path.to_str().unwrap_or_default(), password.as_str()).into()
}
}
impl From<&PathBuf> for Config {
fn from(path: &PathBuf) -> Self {
path.to_str().unwrap_or_default().into()
}
}
impl From<&str> for Config {
fn from(path: &str) -> Self {
if path.contains(":memory:") {
Config {
path: None,
password,
},
1,
)
password: None,
}
} else {
(
Config {
path: Some(path.to_owned()),
password,
},
20,
)
};
password: None,
}
}
}
}
Pool::new(config, max_size, Duration::from_secs(10))
impl From<(&str, &str)> for Config {
fn from((path, pass): (&str, &str)) -> Self {
if path.contains(":memory:") {
Config {
path: None,
password: Some(pass.to_owned()),
}
} else {
Config {
path: Some(path.to_owned()),
password: Some(pass.to_owned()),
}
}
}
}
/// Convert cdk_sql_common::value::Value to rusqlite Value

1
crates/cdk-sqlite/src/lib.rs
View File

@@ -3,6 +3,7 @@
#![warn(missing_docs)]
#![warn(rustdoc::bare_urls)]
mod async_sqlite;
mod common;
#[cfg(feature = "mint")]

727
crates/cdk-sqlite/src/mint/async_rusqlite.rs
View File

@@ -1,727 +0,0 @@
//! Async, pipelined rusqlite client
use std::marker::PhantomData;
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{mpsc as std_mpsc, Arc, Mutex};
use std::thread::spawn;
use std::time::Instant;
use cdk_common::database::Error;
use cdk_sql_common::database::{DatabaseConnector, DatabaseExecutor, DatabaseTransaction};
use cdk_sql_common::pool::{self, Pool, PooledResource};
use cdk_sql_common::stmt::{Column, ExpectedSqlResponse, Statement as InnerStatement};
use cdk_sql_common::ConversionError;
use rusqlite::{ffi, Connection, ErrorCode, TransactionBehavior};
use tokio::sync::{mpsc, oneshot};
use crate::common::{create_sqlite_pool, from_sqlite, to_sqlite, SqliteConnectionManager};
/// The number of queued SQL statements before it start failing
const SQL_QUEUE_SIZE: usize = 10_000;
/// How many ms is considered a slow query, and it'd be logged for further debugging
const SLOW_QUERY_THRESHOLD_MS: u128 = 20;
/// How many SQLite parallel connections can be used to read things in parallel
const WORKING_THREAD_POOL_SIZE: usize = 5;
#[derive(Debug, Clone)]
pub struct AsyncRusqlite {
sender: mpsc::Sender<DbRequest>,
inflight_requests: Arc<AtomicUsize>,
}
impl From<PathBuf> for AsyncRusqlite {
fn from(value: PathBuf) -> Self {
AsyncRusqlite::new(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
}
}
impl From<&str> for AsyncRusqlite {
fn from(value: &str) -> Self {
AsyncRusqlite::new(create_sqlite_pool(value, None))
}
}
impl From<(&str, &str)> for AsyncRusqlite {
fn from((value, pass): (&str, &str)) -> Self {
AsyncRusqlite::new(create_sqlite_pool(value, Some(pass.to_owned())))
}
}
impl From<(PathBuf, &str)> for AsyncRusqlite {
fn from((value, pass): (PathBuf, &str)) -> Self {
AsyncRusqlite::new(create_sqlite_pool(
value.to_str().unwrap_or_default(),
Some(pass.to_owned()),
))
}
}
impl From<(&str, String)> for AsyncRusqlite {
fn from((value, pass): (&str, String)) -> Self {
AsyncRusqlite::new(create_sqlite_pool(value, Some(pass)))
}
}
impl From<(PathBuf, String)> for AsyncRusqlite {
fn from((value, pass): (PathBuf, String)) -> Self {
AsyncRusqlite::new(create_sqlite_pool(
value.to_str().unwrap_or_default(),
Some(pass),
))
}
}
impl From<&PathBuf> for AsyncRusqlite {
fn from(value: &PathBuf) -> Self {
AsyncRusqlite::new(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
}
}
/// Internal request for the database thread
#[derive(Debug)]
enum DbRequest {
Sql(InnerStatement, oneshot::Sender<DbResponse>),
Begin(oneshot::Sender<DbResponse>),
Commit(oneshot::Sender<DbResponse>),
Rollback(oneshot::Sender<DbResponse>),
}
#[derive(Debug)]
enum DbResponse {
Transaction(mpsc::Sender<DbRequest>),
AffectedRows(usize),
Pluck(Option<Column>),
Row(Option<Vec<Column>>),
Rows(Vec<Vec<Column>>),
Error(SqliteError),
Unexpected,
Ok,
}
#[derive(thiserror::Error, Debug)]
enum SqliteError {
#[error(transparent)]
Sqlite(#[from] rusqlite::Error),
#[error(transparent)]
Inner(#[from] Error),
#[error(transparent)]
Pool(#[from] pool::Error<rusqlite::Error>),
/// Duplicate entry
#[error("Duplicate")]
Duplicate,
#[error(transparent)]
Conversion(#[from] ConversionError),
}
impl From<SqliteError> for Error {
fn from(val: SqliteError) -> Self {
match val {
SqliteError::Duplicate => Error::Duplicate,
SqliteError::Conversion(e) => e.into(),
o => Error::Internal(o.to_string()),
}
}
}
/// Process a query
#[inline(always)]
fn process_query(conn: &Connection, statement: InnerStatement) -> Result<DbResponse, SqliteError> {
let start = Instant::now();
let expected_response = statement.expected_response;
let (sql, placeholder_values) = statement.to_sql()?;
let sql = sql.trim_end_matches("FOR UPDATE");
let mut stmt = conn.prepare_cached(sql)?;
for (i, value) in placeholder_values.into_iter().enumerate() {
stmt.raw_bind_parameter(i + 1, to_sqlite(value))?;
}
let columns = stmt.column_count();
let to_return = match expected_response {
ExpectedSqlResponse::AffectedRows => DbResponse::AffectedRows(stmt.raw_execute()?),
ExpectedSqlResponse::Batch => {
conn.execute_batch(sql)?;
DbResponse::Ok
}
ExpectedSqlResponse::ManyRows => {
let mut rows = stmt.raw_query();
let mut results = vec![];
while let Some(row) = rows.next()? {
results.push(
(0..columns)
.map(|i| row.get(i).map(from_sqlite))
.collect::<Result<Vec<_>, _>>()?,
)
}
DbResponse::Rows(results)
}
ExpectedSqlResponse::Pluck => {
let mut rows = stmt.raw_query();
DbResponse::Pluck(
rows.next()?
.map(|row| row.get(0usize).map(from_sqlite))
.transpose()?,
)
}
ExpectedSqlResponse::SingleRow => {
let mut rows = stmt.raw_query();
let row = rows
.next()?
.map(|row| {
(0..columns)
.map(|i| row.get(i).map(from_sqlite))
.collect::<Result<Vec<_>, _>>()
})
.transpose()?;
DbResponse::Row(row)
}
};
let duration = start.elapsed();
if duration.as_millis() > SLOW_QUERY_THRESHOLD_MS {
tracing::warn!("[SLOW QUERY] Took {} ms: {}", duration.as_millis(), sql);
}
Ok(to_return)
}
/// Spawns N number of threads to execute SQL statements
///
/// Enable parallelism with a pool of threads.
///
/// There is a main thread, which receives SQL requests and routes them to a worker thread from a
/// fixed-size pool.
///
/// By doing so, SQLite does synchronization, and Rust will only intervene when a transaction is
/// executed. Transactions are executed in the main thread.
fn rusqlite_spawn_worker_threads(
inflight_requests: Arc<AtomicUsize>,
threads: usize,
) -> std_mpsc::Sender<(
PooledResource<SqliteConnectionManager>,
InnerStatement,
oneshot::Sender<DbResponse>,
)> {
let (sender, receiver) = std_mpsc::channel::<(
PooledResource<SqliteConnectionManager>,
InnerStatement,
oneshot::Sender<DbResponse>,
)>();
let receiver = Arc::new(Mutex::new(receiver));
for _ in 0..threads {
let rx = receiver.clone();
let inflight_requests = inflight_requests.clone();
spawn(move || loop {
while let Ok((conn, sql, reply_to)) = rx.lock().expect("failed to acquire").recv() {
let result = process_query(&conn, sql);
let _ = match result {
Ok(ok) => reply_to.send(ok),
Err(err) => {
tracing::error!("Failed query with error {:?}", err);
let err = if let SqliteError::Sqlite(rusqlite::Error::SqliteFailure(
ffi::Error {
code,
extended_code,
},
_,
)) = &err
{
if *code == ErrorCode::ConstraintViolation
&& (*extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
|| *extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
{
SqliteError::Duplicate
} else {
err
}
} else {
err
};
reply_to.send(DbResponse::Error(err))
}
};
drop(conn);
inflight_requests.fetch_sub(1, Ordering::Relaxed);
}
});
}
sender
}
/// # Rusqlite main worker
///
/// This function takes ownership of a pool of connections to SQLite, executes SQL statements, and
/// returns the results or number of affected rows to the caller. All communications are done
/// through channels. This function is synchronous, but a thread pool exists to execute queries, and
/// SQLite will coordinate data access. Transactions are executed in the main and it takes ownership
/// of the main thread until it is finalized
///
/// This is meant to be called in their thread, as it will not exit the loop until the communication
/// channel is closed.
fn rusqlite_worker_manager(
mut receiver: mpsc::Receiver<DbRequest>,
pool: Arc<Pool<SqliteConnectionManager>>,
inflight_requests: Arc<AtomicUsize>,
) {
let send_sql_to_thread =
rusqlite_spawn_worker_threads(inflight_requests.clone(), WORKING_THREAD_POOL_SIZE);
let mut tx_id: usize = 0;
while let Some(request) = receiver.blocking_recv() {
inflight_requests.fetch_add(1, Ordering::Relaxed);
match request {
DbRequest::Sql(statement, reply_to) => {
let conn = match pool.get() {
Ok(conn) => conn,
Err(err) => {
tracing::error!("Failed to acquire a pool connection: {:?}", err);
inflight_requests.fetch_sub(1, Ordering::Relaxed);
let _ = reply_to.send(DbResponse::Error(err.into()));
continue;
}
};
let _ = send_sql_to_thread.send((conn, statement, reply_to));
continue;
}
DbRequest::Begin(reply_to) => {
let (sender, mut receiver) = mpsc::channel(SQL_QUEUE_SIZE);
let mut conn = match pool.get() {
Ok(conn) => conn,
Err(err) => {
tracing::error!("Failed to acquire a pool connection: {:?}", err);
inflight_requests.fetch_sub(1, Ordering::Relaxed);
let _ = reply_to.send(DbResponse::Error(err.into()));
continue;
}
};
let tx = match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
Ok(tx) => tx,
Err(err) => {
tracing::error!("Failed to begin a transaction: {:?}", err);
inflight_requests.fetch_sub(1, Ordering::Relaxed);
let _ = reply_to.send(DbResponse::Error(err.into()));
continue;
}
};
// Transaction has begun successfully, send the `sender` back to the caller
// and wait for statements to execute. On `Drop` the wrapper transaction
// should send a `rollback`.
let _ = reply_to.send(DbResponse::Transaction(sender));
tx_id += 1;
// We intentionally handle the transaction hijacking the main loop, there is
// no point is queueing more operations for SQLite, since transaction have
// exclusive access. In other database implementation this block of code
// should be sent to their own thread to allow concurrency
loop {
let request = if let Some(request) = receiver.blocking_recv() {
request
} else {
// If the receiver loop is broken (i.e no more `senders` are active) and no
// `Commit` statement has been sent, this will trigger a `Rollback`
// automatically
tracing::trace!("Tx {}: Transaction rollback on drop", tx_id);
let _ = tx.rollback();
break;
};
match request {
DbRequest::Commit(reply_to) => {
tracing::trace!("Tx {}: Commit", tx_id);
let _ = reply_to.send(match tx.commit() {
Ok(()) => DbResponse::Ok,
Err(err) => {
tracing::error!("Failed commit {:?}", err);
DbResponse::Error(err.into())
}
});
break;
}
DbRequest::Rollback(reply_to) => {
tracing::trace!("Tx {}: Rollback", tx_id);
let _ = reply_to.send(match tx.rollback() {
Ok(()) => DbResponse::Ok,
Err(err) => {
tracing::error!("Failed rollback {:?}", err);
DbResponse::Error(err.into())
}
});
break;
}
DbRequest::Begin(reply_to) => {
let _ = reply_to.send(DbResponse::Unexpected);
}
DbRequest::Sql(statement, reply_to) => {
tracing::trace!("Tx {}: SQL {:?}", tx_id, statement);
let _ = match process_query(&tx, statement) {
Ok(ok) => reply_to.send(ok),
Err(err) => {
tracing::error!(
"Tx {}: Failed query with error {:?}",
tx_id,
err
);
let err = if let SqliteError::Sqlite(
rusqlite::Error::SqliteFailure(
ffi::Error {
code,
extended_code,
},
_,
),
) = &err
{
if *code == ErrorCode::ConstraintViolation
&& (*extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
|| *extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
{
SqliteError::Duplicate
} else {
err
}
} else {
err
};
reply_to.send(DbResponse::Error(err))
}
};
}
}
}
drop(conn);
}
DbRequest::Commit(reply_to) => {
let _ = reply_to.send(DbResponse::Unexpected);
}
DbRequest::Rollback(reply_to) => {
let _ = reply_to.send(DbResponse::Unexpected);
}
}
// If wasn't a `continue` the transaction is done by reaching this code, and we should
// decrease the inflight_request counter
inflight_requests.fetch_sub(1, Ordering::Relaxed);
}
}
impl AsyncRusqlite {
/// Creates a new Async Rusqlite wrapper.
pub fn new(pool: Arc<Pool<SqliteConnectionManager>>) -> Self {
let (sender, receiver) = mpsc::channel(SQL_QUEUE_SIZE);
let inflight_requests = Arc::new(AtomicUsize::new(0));
let inflight_requests_for_thread = inflight_requests.clone();
spawn(move || {
rusqlite_worker_manager(receiver, pool, inflight_requests_for_thread);
});
Self {
sender,
inflight_requests,
}
}
fn get_queue_sender(&self) -> &mpsc::Sender<DbRequest> {
&self.sender
}
/// Show how many inflight requests
#[allow(dead_code)]
pub fn inflight_requests(&self) -> usize {
self.inflight_requests.load(Ordering::Relaxed)
}
}
#[async_trait::async_trait]
impl DatabaseConnector for AsyncRusqlite {
type Transaction<'a> = Transaction<'a>;
/// Begins a transaction
///
/// If the transaction is Drop it will trigger a rollback operation
async fn begin(&self) -> Result<Self::Transaction<'_>, Error> {
let (sender, receiver) = oneshot::channel();
self.sender
.send(DbRequest::Begin(sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Transaction(db_sender) => Ok(Transaction {
db_sender,
_marker: PhantomData,
}),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
}
#[async_trait::async_trait]
impl DatabaseExecutor for AsyncRusqlite {
fn name() -> &'static str {
"sqlite"
}
async fn fetch_one(&self, mut statement: InnerStatement) -> Result<Option<Vec<Column>>, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::SingleRow;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Row(row) => Ok(row),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn batch(&self, mut statement: InnerStatement) -> Result<(), Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::Batch;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Ok => Ok(()),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn fetch_all(&self, mut statement: InnerStatement) -> Result<Vec<Vec<Column>>, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::ManyRows;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Rows(row) => Ok(row),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn execute(&self, mut statement: InnerStatement) -> Result<usize, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::AffectedRows;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::AffectedRows(total) => Ok(total),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn pluck(&self, mut statement: InnerStatement) -> Result<Option<Column>, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::Pluck;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Pluck(value) => Ok(value),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
}
/// Database transaction
#[derive(Debug)]
pub struct Transaction<'conn> {
db_sender: mpsc::Sender<DbRequest>,
_marker: PhantomData<&'conn ()>,
}
impl Transaction<'_> {
fn get_queue_sender(&self) -> &mpsc::Sender<DbRequest> {
&self.db_sender
}
}
impl Drop for Transaction<'_> {
fn drop(&mut self) {
let (sender, _) = oneshot::channel();
let _ = self.db_sender.try_send(DbRequest::Rollback(sender));
}
}
#[async_trait::async_trait]
impl<'a> DatabaseTransaction<'a> for Transaction<'a> {
async fn commit(self) -> Result<(), Error> {
let (sender, receiver) = oneshot::channel();
self.db_sender
.send(DbRequest::Commit(sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Ok => Ok(()),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn rollback(self) -> Result<(), Error> {
let (sender, receiver) = oneshot::channel();
self.db_sender
.send(DbRequest::Rollback(sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Ok => Ok(()),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
}
#[async_trait::async_trait]
impl DatabaseExecutor for Transaction<'_> {
fn name() -> &'static str {
"sqlite"
}
async fn fetch_one(&self, mut statement: InnerStatement) -> Result<Option<Vec<Column>>, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::SingleRow;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Row(row) => Ok(row),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn batch(&self, mut statement: InnerStatement) -> Result<(), Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::Batch;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Ok => Ok(()),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn fetch_all(&self, mut statement: InnerStatement) -> Result<Vec<Vec<Column>>, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::ManyRows;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Rows(row) => Ok(row),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn execute(&self, mut statement: InnerStatement) -> Result<usize, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::AffectedRows;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::AffectedRows(total) => Ok(total),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
async fn pluck(&self, mut statement: InnerStatement) -> Result<Option<Column>, Error> {
let (sender, receiver) = oneshot::channel();
statement.expected_response = ExpectedSqlResponse::Pluck;
self.get_queue_sender()
.send(DbRequest::Sql(statement, sender))
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?;
match receiver
.await
.map_err(|_| Error::Internal("Communication".to_owned()))?
{
DbResponse::Pluck(value) => Ok(value),
DbResponse::Error(err) => Err(err.into()),
_ => Err(Error::InvalidDbResponse),
}
}
}

21
crates/cdk-sqlite/src/mint/mod.rs
View File

@@ -3,26 +3,27 @@
use cdk_sql_common::mint::SQLMintAuthDatabase;
use cdk_sql_common::SQLMintDatabase;
mod async_rusqlite;
use crate::common::SqliteConnectionManager;
pub mod memory;
/// Mint SQLite implementation with rusqlite
pub type MintSqliteDatabase = SQLMintDatabase<async_rusqlite::AsyncRusqlite>;
pub type MintSqliteDatabase = SQLMintDatabase<SqliteConnectionManager>;
/// Mint Auth database with rusqlite
#[cfg(feature = "auth")]
pub type MintSqliteAuthDatabase = SQLMintAuthDatabase<async_rusqlite::AsyncRusqlite>;
pub type MintSqliteAuthDatabase = SQLMintAuthDatabase<SqliteConnectionManager>;
#[cfg(test)]
mod test {
use std::fs::remove_file;
use cdk_common::mint_db_test;
use cdk_sql_common::pool::Pool;
use cdk_sql_common::stmt::query;
use super::*;
use crate::mint::async_rusqlite::AsyncRusqlite;
use crate::common::Config;
async fn provide_db() -> MintSqliteDatabase {
memory::empty().await.unwrap()
@@ -40,13 +41,17 @@ mod test {
{
let _ = remove_file(&file);
#[cfg(not(feature = "sqlcipher"))]
let conn: AsyncRusqlite = file.as_str().into();
let config: Config = file.as_str().into();
#[cfg(feature = "sqlcipher")]
let conn: AsyncRusqlite = (file.as_str(), "test".to_owned()).into();
let config: Config = (file.as_str(), "test").into();
let pool = Pool::<SqliteConnectionManager>::new(config);
let conn = pool.get().expect("valid connection");
query(include_str!("../../tests/legacy-sqlx.sql"))
.expect("query")
.execute(&conn)
.execute(&*conn)
.await
.expect("create former db failed");
}
@@ -55,7 +60,7 @@ mod test {
let conn = MintSqliteDatabase::new(file.as_str()).await;
#[cfg(feature = "sqlcipher")]
let conn = MintSqliteDatabase::new((file.as_str(), "test".to_owned())).await;
let conn = MintSqliteDatabase::new((file.as_str(), "test")).await;
assert!(conn.is_ok(), "Failed with {:?}", conn.unwrap_err());

188
crates/cdk-sqlite/src/wallet/mod.rs
View File

@@ -1,195 +1,13 @@
//! SQLite Wallet Database
use std::path::PathBuf;
use std::sync::Arc;
use cdk_common::database::Error;
use cdk_sql_common::database::DatabaseExecutor;
use cdk_sql_common::pool::{Pool, PooledResource};
use cdk_sql_common::stmt::{Column, SqlPart, Statement};
use cdk_sql_common::SQLWalletDatabase;
use rusqlite::CachedStatement;
use crate::common::{create_sqlite_pool, from_sqlite, to_sqlite, SqliteConnectionManager};
use crate::common::SqliteConnectionManager;
pub mod memory;
/// Simple Sqlite wapper, since the wallet may not need rusqlite with concurrency, a shared instance
/// may be enough
#[derive(Debug)]
pub struct SimpleAsyncRusqlite(Arc<Pool<SqliteConnectionManager>>);
impl SimpleAsyncRusqlite {
fn get_stmt<'a>(
&self,
conn: &'a PooledResource<SqliteConnectionManager>,
statement: Statement,
) -> Result<CachedStatement<'a>, Error> {
let (sql, placeholder_values) = statement.to_sql()?;
let mut stmt = conn
.prepare_cached(&sql)
.map_err(|e| Error::Database(Box::new(e)))?;
for (i, value) in placeholder_values.into_iter().enumerate() {
stmt.raw_bind_parameter(i + 1, to_sqlite(value))
.map_err(|e| Error::Database(Box::new(e)))?;
}
Ok(stmt)
}
}
#[async_trait::async_trait]
impl DatabaseExecutor for SimpleAsyncRusqlite {
fn name() -> &'static str {
"sqlite"
}
async fn execute(&self, statement: Statement) -> Result<usize, Error> {
let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
Ok(stmt
.raw_execute()
.map_err(|e| Error::Database(Box::new(e)))?)
}
async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
let columns = stmt.column_count();
let mut rows = stmt.raw_query();
rows.next()
.map_err(|e| Error::Database(Box::new(e)))?
.map(|row| {
(0..columns)
.map(|i| row.get(i).map(from_sqlite))
.collect::<Result<Vec<_>, _>>()
})
.transpose()
.map_err(|e| Error::Database(Box::new(e)))
}
async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
let columns = stmt.column_count();
let mut rows = stmt.raw_query();
let mut results = vec![];
while let Some(row) = rows.next().map_err(|e| Error::Database(Box::new(e)))? {
results.push(
(0..columns)
.map(|i| row.get(i).map(from_sqlite))
.collect::<Result<Vec<_>, _>>()
.map_err(|e| Error::Database(Box::new(e)))?,
)
}
Ok(results)
}
async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
let mut stmt = self
.get_stmt(&conn, statement)
.map_err(|e| Error::Database(Box::new(e)))?;
let mut rows = stmt.raw_query();
rows.next()
.map_err(|e| Error::Database(Box::new(e)))?
.map(|row| row.get(0usize).map(from_sqlite))
.transpose()
.map_err(|e| Error::Database(Box::new(e)))
}
async fn batch(&self, mut statement: Statement) -> Result<(), Error> {
let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
let sql = {
let part = statement
.parts
.pop()
.ok_or(Error::Internal("Empty SQL".to_owned()))?;
if !statement.parts.is_empty() || matches!(part, SqlPart::Placeholder(_, _)) {
return Err(Error::Internal(
"Invalid usage, batch does not support placeholders".to_owned(),
));
}
if let SqlPart::Raw(sql) = part {
sql
} else {
unreachable!()
}
};
conn.execute_batch(&sql)
.map_err(|e| Error::Database(Box::new(e)))
}
}
impl From<PathBuf> for SimpleAsyncRusqlite {
fn from(value: PathBuf) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
}
}
impl From<&str> for SimpleAsyncRusqlite {
fn from(value: &str) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(value, None))
}
}
impl From<(&str, &str)> for SimpleAsyncRusqlite {
fn from((value, pass): (&str, &str)) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(value, Some(pass.to_owned())))
}
}
impl From<(PathBuf, &str)> for SimpleAsyncRusqlite {
fn from((value, pass): (PathBuf, &str)) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(
value.to_str().unwrap_or_default(),
Some(pass.to_owned()),
))
}
}
impl From<(&str, String)> for SimpleAsyncRusqlite {
fn from((value, pass): (&str, String)) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(value, Some(pass)))
}
}
impl From<(PathBuf, String)> for SimpleAsyncRusqlite {
fn from((value, pass): (PathBuf, String)) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(
value.to_str().unwrap_or_default(),
Some(pass),
))
}
}
impl From<&PathBuf> for SimpleAsyncRusqlite {
fn from(value: &PathBuf) -> Self {
SimpleAsyncRusqlite(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
}
}
/// Mint SQLite implementation with rusqlite
pub type WalletSqliteDatabase = SQLWalletDatabase<SimpleAsyncRusqlite>;
pub type WalletSqliteDatabase = SQLWalletDatabase<SqliteConnectionManager>;
#[cfg(test)]
mod tests {
@@ -330,7 +148,7 @@ mod tests {
// Test PaymentMethod variants
let mint_url = MintUrl::from_str("https://example.com").unwrap();
let payment_methods = vec![
let payment_methods = [
PaymentMethod::Bolt11,
PaymentMethod::Bolt12,
PaymentMethod::Custom("custom".to_string()),