Add postgres integration for cdk-cli

Cargo.toml

@@ -56,6 +56,7 @@ cdk-mint-rpc = { path = "./crates/cdk-mint-rpc", version = "=0.11.0" }
 cdk-redb = { path = "./crates/cdk-redb", default-features = true, version = "=0.11.0" }
 cdk-sql-common = { path = "./crates/cdk-sql-common", default-features = true, version = "=0.11.0" }
 cdk-sqlite = { path = "./crates/cdk-sqlite", default-features = true, version = "=0.11.0" }
+cdk-postgres = { path = "./crates/cdk-postgres", default-features = true, version = "=0.11.0" }
 cdk-signatory = { path = "./crates/cdk-signatory", version = "=0.11.0", default-features = false }
 clap = { version = "4.5.31", features = ["derive"] }
 ciborium = { version = "0.2.2", default-features = false, features = ["std"] }

crates/cdk-cli/Cargo.toml

@@ -24,6 +24,7 @@ bitcoin.workspace = true
 cdk = { workspace = true, default-features = false, features = ["wallet", "auth"]}
 cdk-redb = { workspace = true, features = ["wallet"], optional = true }
 cdk-sqlite = { workspace = true, features = ["wallet"] }
+cdk-postgres = { workspace = true, features = ["wallet"] }
 clap.workspace = true
 serde.workspace = true
 serde_json.workspace = true

crates/cdk-cli/src/main.rs

@@ -10,6 +10,7 @@ use cdk::cdk_database;
 use cdk::cdk_database::WalletDatabase;
 use cdk::nuts::CurrencyUnit;
 use cdk::wallet::{HttpClient, MultiMintWallet, Wallet, WalletBuilder};
+use cdk_postgres::WalletPgDatabase;
 #[cfg(feature = "redb")]
 use cdk_redb::WalletRedbDatabase;
 use cdk_sqlite::WalletSqliteDatabase;
@@ -135,6 +136,12 @@ async fn main() -> Result<()> {
 
                 Arc::new(sql)
             }
+            "postgres" => {
+                let db_url = std::env::var("DATABASE_URL").unwrap_or(
+                    "host=localhost user=test password=test dbname=cdk_wallet port=5433".to_owned(),
+                );
+                Arc::new(WalletPgDatabase::new(db_url.as_str()).await?)
+            }
             "redb" => {
                 #[cfg(feature = "redb")]
                 {

crates/cdk-integration-tests/Cargo.toml

@@ -25,6 +25,7 @@ cdk-cln = { workspace = true }
 cdk-lnd = { workspace = true }
 cdk-axum = { workspace = true }
 cdk-sqlite = { workspace = true }
+cdk-postgres = { workspace = true }
 cdk-redb = { workspace = true }
 cdk-fake-wallet = { workspace = true }
 futures = { workspace = true, default-features = false, features = [

crates/cdk-mintd/Cargo.toml

@@ -35,6 +35,9 @@ cdk = { workspace = true, features = [
 cdk-sqlite = { workspace = true, features = [
     "mint",
 ] }
+cdk-postgres = { workspace = true, features = [
+    "mint",
+] }
 cdk-cln = { workspace = true, optional = true }
 cdk-lnbits = { workspace = true, optional = true }
 cdk-lnd = { workspace = true, optional = true }

crates/cdk-mintd/src/config.rs

@@ -190,6 +190,7 @@ pub struct GrpcProcessor {
 pub enum DatabaseEngine {
     #[default]
     Sqlite,
+    Postgres,
 }
 
 impl std::str::FromStr for DatabaseEngine {
@@ -198,6 +199,7 @@ impl std::str::FromStr for DatabaseEngine {
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_lowercase().as_str() {
             "sqlite" => Ok(DatabaseEngine::Sqlite),
+            "postgres" => Ok(DatabaseEngine::Postgres),
             _ => Err(format!("Unknown database engine: {s}")),
         }
     }

crates/cdk-mintd/src/main.rs

@@ -198,6 +198,7 @@ async fn setup_database(
             let keystore: Arc<dyn MintKeysDatabase<Err = cdk_database::Error> + Send + Sync> = db;
             Ok((localstore, keystore))
         }
+        DatabaseEngine::Postgres => todo!(),
     }
 }
 

crates/cdk-postgres/src/lib.rs

@@ -63,7 +63,7 @@ pub struct FutureConnect {
 
 impl FutureConnect {
     /// Creates a new instance
-    pub fn new(config: PgConfig, timeout: Duration, still_valid: Arc<AtomicBool>) -> Self {
+    pub fn new(config: PgConfig, timeout: Duration, stale: Arc<AtomicBool>) -> Self {
         let failed = Arc::new(Mutex::new(None));
         let result = Arc::new(OnceLock::new());
         let notify = Arc::new(Notify::new());
@@ -79,7 +79,7 @@ impl FutureConnect {
                         Err(err) => {
                             *error_clone.lock().await =
                                 Some(cdk_common::database::Error::Database(Box::new(err)));
-                            still_valid.store(false, std::sync::atomic::Ordering::Release);
+                            stale.store(true, std::sync::atomic::Ordering::Release);
                             notify_clone.notify_waiters();
                             return;
                         }
@@ -87,7 +87,7 @@ impl FutureConnect {
 
                     tokio::spawn(async move {
                         let _ = connection.await;
-                        still_valid.store(false, std::sync::atomic::Ordering::Release);
+                        stale.store(true, std::sync::atomic::Ordering::Release);
                     });
 
                     let _ = result_clone.set(client);
@@ -99,7 +99,7 @@ impl FutureConnect {
                         Err(err) => {
                             *error_clone.lock().await =
                                 Some(cdk_common::database::Error::Database(Box::new(err)));
-                            still_valid.store(false, std::sync::atomic::Ordering::Release);
+                            stale.store(true, std::sync::atomic::Ordering::Release);
                             notify_clone.notify_waiters();
                             return;
                         }
@@ -107,7 +107,7 @@ impl FutureConnect {
 
                     tokio::spawn(async move {
                         let _ = connection.await;
-                        still_valid.store(false, std::sync::atomic::Ordering::Release);
+                        stale.store(true, std::sync::atomic::Ordering::Release);
                     });
 
                     let _ = result_clone.set(client);
@@ -161,10 +161,10 @@ impl ResourceManager for PgConnectionPool {
 
     fn new_resource(
         config: &Self::Config,
-        still_valid: Arc<AtomicBool>,
+        stale: Arc<AtomicBool>,
         timeout: Duration,
     ) -> Result<Self::Resource, cdk_sql_common::pool::Error<Self::Error>> {
-        Ok(FutureConnect::new(config.to_owned(), timeout, still_valid))
+        Ok(FutureConnect::new(config.to_owned(), timeout, stale))
     }
 }
 

crates/cdk-postgres/start_db_for_test.sh

@@ -22,5 +22,11 @@ until docker exec -e PGPASSWORD="${DB_PASS}" "${CONTAINER_NAME}" \
   sleep 0.5
 done
 
+echo "Creating additional databases 'testdb2' and 'testdb3'..."
+docker exec -e PGPASSWORD="${DB_PASS}" "${CONTAINER_NAME}" \
+  psql -U "${DB_USER}" -d "${DB_NAME}" -c "CREATE DATABASE cdk_mintd;"
+docker exec -e PGPASSWORD="${DB_PASS}" "${CONTAINER_NAME}" \
+  psql -U "${DB_USER}" -d "${DB_NAME}" -c "CREATE DATABASE cdk_wallet;"
+
 export DATABASE_URL="host=localhost user=${DB_USER} password=${DB_PASS} dbname=${DB_NAME} port=${DB_PORT}"
 

crates/cdk-sql-common/src/wallet/migrations.rs

@@ -1,6 +1,7 @@
 /// @generated
 /// Auto-generated by build.rs
 pub static MIGRATIONS: &[(&str, &str, &str)] = &[
+    ("postgres", "1_initial.sql", include_str!(r#"./migrations/postgres/1_initial.sql"#)),
     ("sqlite", "1_fix_sqlx_migration.sql", include_str!(r#"./migrations/sqlite/1_fix_sqlx_migration.sql"#)),
     ("sqlite", "20240612132920_init.sql", include_str!(r#"./migrations/sqlite/20240612132920_init.sql"#)),
     ("sqlite", "20240618200350_quote_state.sql", include_str!(r#"./migrations/sqlite/20240618200350_quote_state.sql"#)),

crates/cdk-sql-common/src/wallet/migrations/postgres/1_initial.sql

@@ -0,0 +1,73 @@
+CREATE TABLE mint (
+    mint_url TEXT PRIMARY KEY,
+    name TEXT,
+    pubkey BYTEA,
+    version TEXT,
+    description TEXT,
+    description_long TEXT,
+    contact TEXT,
+    nuts TEXT,
+    motd TEXT
+, icon_url TEXT, mint_time INTEGER, urls TEXT, tos_url TEXT);
+CREATE TABLE keyset (
+    id TEXT PRIMARY KEY,
+    mint_url TEXT NOT NULL,
+    unit TEXT NOT NULL,
+    active BOOL NOT NULL,
+    counter INTEGER NOT NULL DEFAULT 0, input_fee_ppk INTEGER, final_expiry INTEGER DEFAULT NULL,
+    FOREIGN KEY(mint_url) REFERENCES mint(mint_url) ON UPDATE CASCADE ON DELETE CASCADE
+);
+CREATE TABLE melt_quote (
+    id TEXT PRIMARY KEY,
+    unit TEXT NOT NULL,
+    amount INTEGER NOT NULL,
+    request TEXT NOT NULL,
+    fee_reserve INTEGER NOT NULL,
+    expiry INTEGER NOT NULL
+, state TEXT CHECK ( state IN ('UNPAID', 'PENDING', 'PAID' ) ) NOT NULL DEFAULT 'UNPAID', payment_preimage TEXT);
+CREATE TABLE key (
+    id TEXT PRIMARY KEY,
+    keys TEXT NOT NULL
+);
+CREATE INDEX melt_quote_state_index ON melt_quote(state);
+CREATE TABLE IF NOT EXISTS "proof" (
+y BYTEA PRIMARY KEY,
+mint_url TEXT NOT NULL,
+state TEXT CHECK ( state IN ('SPENT', 'UNSPENT', 'PENDING', 'RESERVED', 'PENDING_SPENT' ) ) NOT NULL,
+spending_condition TEXT,
+unit TEXT NOT NULL,
+amount INTEGER NOT NULL,
+keyset_id TEXT NOT NULL,
+secret TEXT NOT NULL,
+c BYTEA NOT NULL,
+witness TEXT
+, dleq_e BYTEA, dleq_s BYTEA, dleq_r BYTEA);
+CREATE TABLE transactions (
+    id BYTEA PRIMARY KEY,
+    mint_url TEXT NOT NULL,
+    direction TEXT CHECK (direction IN ('Incoming', 'Outgoing')) NOT NULL,
+    amount INTEGER NOT NULL,
+    fee INTEGER NOT NULL,
+    unit TEXT NOT NULL,
+    ys BYTEA NOT NULL,
+    timestamp INTEGER NOT NULL,
+    memo TEXT,
+    metadata TEXT
+);
+CREATE INDEX mint_url_index ON transactions(mint_url);
+CREATE INDEX direction_index ON transactions(direction);
+CREATE INDEX unit_index ON transactions(unit);
+CREATE INDEX timestamp_index ON transactions(timestamp);
+CREATE TABLE IF NOT EXISTS "mint_quote" (
+    id TEXT PRIMARY KEY,
+    mint_url TEXT NOT NULL,
+    payment_method TEXT NOT NULL DEFAULT 'bolt11',
+    amount INTEGER,
+    unit TEXT NOT NULL,
+    request TEXT NOT NULL,
+    state TEXT NOT NULL,
+    expiry INTEGER NOT NULL,
+    amount_paid INTEGER NOT NULL DEFAULT 0,
+    amount_issued INTEGER NOT NULL DEFAULT 0,
+    secret_key TEXT
+);