refactor: move postgres specific code to new module

Author: notmandatory

db/migrations/01_bdk_wallet.sql renamed to migrations/postgres/01_bdk_wallet.sql

@@ -2,37 +2,22 @@
 
 #![warn(missing_docs)]
 
+mod postgres;
+
+#[cfg(test)]
+mod test;
+
 use std::future::Future;
 use std::pin::Pin;
-use std::str::FromStr;
 use std::sync::Arc;
 
-use bdk_chain::{
-    local_chain, miniscript, tx_graph, Anchor, ConfirmationBlockTime, DescriptorExt, DescriptorId,
-    Merge,
-};
-use bdk_wallet::bitcoin::{
-    self,
-    consensus::{self, Decodable},
-    hashes::Hash,
-    Amount, BlockHash, Network, OutPoint, ScriptBuf, TxOut, Txid,
-};
+use bdk_chain::miniscript;
+use bdk_wallet::bitcoin;
 use bdk_wallet::chain as bdk_chain;
-use bdk_wallet::descriptor::{Descriptor, DescriptorPublicKey, ExtendedDescriptor};
-use bdk_wallet::KeychainKind::{External, Internal};
-use bdk_wallet::{AsyncWalletPersister, ChangeSet, KeychainKind};
-use serde_json::json;
-use sqlx::migrate::Migrator;
-use sqlx::postgres::PgRow;
-use sqlx::{
-    postgres::{PgPool, Postgres},
-    FromRow, Pool, Row, Transaction,
-};
-use tokio::sync::Mutex;
-use tracing::info;
 
-#[cfg(test)]
-mod test;
+use sqlx::Database;
+use sqlx::Pool;
+use tokio::sync::Mutex;
 
 /// Crate error
 #[derive(Debug, thiserror::Error)]
@@ -53,76 +38,12 @@ pub enum BdkSqlxError {
 
 /// Manages a pool of database connections.
 #[derive(Debug)]
-pub struct Store {
-    pub(crate) pool: Arc<Mutex<Pool<Postgres>>>,
+pub struct Store<DB: Database> {
+    pub(crate) pool: Arc<Mutex<Pool<DB>>>,
     wallet_name: String,
     migration: bool,
 }
 
-impl Store {
-    /// Construct a new [`Store`] with an existing pg connection.
-    #[tracing::instrument]
-    pub async fn new(
-        pool: Arc<Mutex<Pool<Postgres>>>,
-        wallet_name: Option<String>,
-        migration: bool,
-    ) -> Result<Self, BdkSqlxError> {
-        info!("new store");
-
-        let wallet_name = wallet_name.unwrap_or_else(|| "bdk_pg_wallet".to_string());
-
-        Ok(Self {
-            pool,
-            wallet_name,
-            migration,
-        })
-    }
-
-    /// Construct a new [`Store`] without an existing pg connection.
-    #[tracing::instrument]
-    pub async fn new_with_url(
-        url: String,
-        wallet_name: Option<String>,
-    ) -> Result<Store, BdkSqlxError> {
-        info!("new store with url");
-
-        let pool = PgPool::connect(url.as_str()).await?;
-        let pool = Arc::new(Mutex::new(pool));
-        let wallet_name = wallet_name.unwrap_or_else(|| "bdk_pg_wallet".to_string());
-
-        Ok(Self {
-            pool,
-            wallet_name,
-            migration: true,
-        })
-    }
-}
-
-impl AsyncWalletPersister for Store {
-    type Error = BdkSqlxError;
-
-    #[tracing::instrument]
-    fn initialize<'a>(store: &'a mut Self) -> FutureResult<'a, ChangeSet, Self::Error>
-    where
-        Self: 'a,
-    {
-        info!("initialize store");
-        Box::pin(store.migrate_and_read())
-    }
-
-    #[tracing::instrument]
-    fn persist<'a>(
-        store: &'a mut Self,
-        changeset: &'a ChangeSet,
-    ) -> FutureResult<'a, (), Self::Error>
-    where
-        Self: 'a,
-    {
-        info!("persist store");
-        Box::pin(store.write(changeset))
-    }
-}
-
 type FutureResult<'a, T, E> = Pin<Box<dyn Future<Output = Result<T, E>> + Send + 'a>>;
 
 impl Store {
@@ -159,7 +80,7 @@ impl Store {
         dbg!(&row);
 
         if let Some(row) = row {
-            Self::changeset_from_row(&mut tx, &mut changeset, row, &self.wallet_name).await?;
+            Self::changeset_from_row(&mut tx, &mut changeset, row).await?;
         }
 
         Ok(changeset)
@@ -170,7 +91,6 @@ impl Store {
         tx: &mut Transaction<'_, Postgres>,
         changeset: &mut ChangeSet,
         row: PgRow,
-        wallet_name: &str,
     ) -> Result<(), BdkSqlxError> {
         info!("changeset from row");
 
@@ -200,8 +120,8 @@ impl Store {
             }
         }
 
-        changeset.tx_graph = tx_graph_changeset_from_postgres(tx, wallet_name).await?;
-        changeset.local_chain = local_chain_changeset_from_postgres(tx, wallet_name).await?;
+        changeset.tx_graph = tx_graph_changeset_from_postgres(tx).await?;
+        changeset.local_chain = local_chain_changeset_from_postgres(tx).await?;
         Ok(())
     }
 
@@ -318,14 +238,12 @@ async fn update_last_revealed(
 #[tracing::instrument]
 pub async fn tx_graph_changeset_from_postgres(
     db_tx: &mut Transaction<'_, Postgres>,
-    wallet_name: &str,
 ) -> Result<tx_graph::ChangeSet<ConfirmationBlockTime>, BdkSqlxError> {
     info!("tx graph changeset from postgres");
     let mut changeset = tx_graph::ChangeSet::default();
 
     // Fetch transactions
-    let rows = sqlx::query("SELECT txid, whole_tx, last_seen FROM tx WHERE wallet_name = $1")
-        .bind(wallet_name)
+    let rows = sqlx::query("SELECT txid, whole_tx, last_seen FROM tx")
         .fetch_all(&mut **db_tx)
         .await?;
 
@@ -346,8 +264,7 @@ pub async fn tx_graph_changeset_from_postgres(
     }
 
     // Fetch txouts
-    let rows = sqlx::query("SELECT txid, vout, value, script FROM txout  WHERE wallet_name = $1")
-        .bind(wallet_name)
+    let rows = sqlx::query("SELECT txid, vout, value, script FROM txout")
         .fetch_all(&mut **db_tx)
         .await?;
 
@@ -371,8 +288,7 @@ pub async fn tx_graph_changeset_from_postgres(
     }
 
     // Fetch anchors
-    let rows = sqlx::query("SELECT anchor, txid FROM anchor_tx WHERE wallet_name = $1")
-        .bind(wallet_name)
+    let rows = sqlx::query("SELECT anchor, txid FROM anchor_tx")
         .fetch_all(&mut **db_tx)
         .await?;
 
@@ -454,13 +370,11 @@ pub async fn tx_graph_changeset_persist_to_postgres(
 #[tracing::instrument]
 pub async fn local_chain_changeset_from_postgres(
     db_tx: &mut Transaction<'_, Postgres>,
-    wallet_name: &str,
 ) -> Result<local_chain::ChangeSet, BdkSqlxError> {
     info!("local chain changeset from postgres");
     let mut changeset = local_chain::ChangeSet::default();
 
-    let rows = sqlx::query("SELECT hash, height FROM block WHERE wallet_name = $1")
-        .bind(wallet_name)
+    let rows = sqlx::query("SELECT hash, height FROM block")
         .fetch_all(&mut **db_tx)
         .await?;
 

src/lib.rs

@@ -9,6 +9,7 @@ use bdk_wallet::bitcoin::Network;
 use bdk_wallet::{KeychainKind, PersistedWallet, Wallet};
 use better_panic::Settings;
 use rustls::crypto::ring::default_provider;
+use sqlx::Postgres;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
 use tracing_subscriber::EnvFilter;
@@ -59,7 +60,8 @@ async fn main() -> anyhow::Result<()> {
         NETWORK,
         &secp,
     )?;
-    let mut store = bdk_sqlx::Store::new_with_url(url.clone(), Some(wallet_name)).await?;
+    let mut store =
+        bdk_sqlx::Store::<Postgres>::new_with_url(url.clone(), Some(wallet_name)).await?;
 
     let mut wallet = match Wallet::load().load_wallet_async(&mut store).await? {
         Some(wallet) => wallet,
@@ -114,7 +116,7 @@ async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
-fn electrum(wallet: &mut PersistedWallet<Store>) -> anyhow::Result<()> {
+fn electrum(wallet: &mut PersistedWallet<Store<Postgres>>) -> anyhow::Result<()> {
     let client = BdkElectrumClient::new(electrum_client::Client::new(ELECTRUM_URL)?);
 
     // Populate the electrum client's transaction cache so it doesn't redownload transaction we