From: Vihiga Tyonum Date: Sun, 5 Jul 2026 02:07:10 +0000 (+0100) Subject: ref(payjoin): add payjoin persistence X-Git-Url: http://internal-gitweb-vhost/blockdata/script/encode/-script/engine/struct.GeneralPurposeConfig.html?a=commitdiff_plain;h=c2ca2ddbdfea5bcc7b8aaeed79bfd39798a2165c;p=bdk-cli ref(payjoin): add payjoin persistence - Adds payjoin persistence --- diff --git a/Cargo.lock b/Cargo.lock index 407c962..40d97a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -606,6 +606,12 @@ dependencies = [ "shlex", ] +[[package]] +name = "cesu8" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d43a04d8753f35258c91f8ec639f792891f748a1edbd759cf1dcea3382ad83c" + [[package]] name = "cfg-if" version = "1.0.4" @@ -807,6 +813,16 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "combine" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +dependencies = [ + "bytes", + "memchr", +] + [[package]] name = "core-foundation" version = "0.9.4" @@ -817,6 +833,16 @@ dependencies = [ "libc", ] +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -1617,6 +1643,50 @@ dependencies = [ "syn", ] +[[package]] +name = "jni" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a87aa2bb7d2af34197c04845522473242e1aa17c12f4935d5856491a7fb8c97" +dependencies = [ + "cesu8", + "cfg-if", + "combine", + "jni-sys 0.3.1", + "log", + "thiserror 1.0.69", + "walkdir", + "windows-sys 0.45.0", +] + +[[package]] +name = "jni-sys" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41a652e1f9b6e0275df1f15b32661cf0d4b78d4d87ddec5e0c3c20f097433258" +dependencies = [ + "jni-sys 0.4.1", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn", +] + [[package]] name = "jobserver" version = "0.1.34" @@ -1766,10 +1836,10 @@ dependencies = [ "libc", "log", "openssl", - "openssl-probe", + "openssl-probe 0.1.6", "openssl-sys", "schannel", - "security-framework", + "security-framework 2.11.1", "security-framework-sys", "tempfile", ] @@ -1833,6 +1903,12 @@ version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" +[[package]] +name = "openssl-probe" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" + [[package]] name = "openssl-sys" version = "0.9.111" @@ -1876,9 +1952,9 @@ dependencies = [ [[package]] name = "payjoin" -version = "1.0.0-rc.1" +version = "0.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e20b76ae28f1420a918e8051681fc9669ed7273e542e515baa329be78c3255a" +checksum = "844f2404db4ae16c4062a53da2c73812c7a130abf612149c06650a36e34d0a35" dependencies = [ "bhttp", "bitcoin", @@ -1891,6 +1967,7 @@ dependencies = [ "serde_json", "tracing", "url", + "web-time", ] [[package]] @@ -2025,6 +2102,7 @@ version = "0.11.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" dependencies = [ + "aws-lc-rs", "bytes", "getrandom 0.3.4", "lru-slab", @@ -2240,13 +2318,19 @@ dependencies = [ "http-body", "http-body-util", "hyper", + "hyper-rustls", "hyper-util", "js-sys", "log", "percent-encoding", "pin-project-lite", + "quinn", + "rustls 0.23.36", + "rustls-pki-types", + "rustls-platform-verifier", "sync_wrapper", "tokio", + "tokio-rustls", "tower", "tower-http", "tower-service", @@ -2331,6 +2415,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dab5152771c58876a2146916e53e35057e1a4dfa2b9df0f0305b07f611fdea4d" +dependencies = [ + "openssl-probe 0.2.1", + "rustls-pki-types", + "schannel", + "security-framework 3.5.1", +] + [[package]] name = "rustls-pki-types" version = "1.13.3" @@ -2341,6 +2437,33 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-platform-verifier" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d99feebc72bae7ab76ba994bb5e121b8d83d910ca40b36e0921f53becc41784" +dependencies = [ + "core-foundation 0.10.1", + "core-foundation-sys", + "jni", + "log", + "once_cell", + "rustls 0.23.36", + "rustls-native-certs", + "rustls-platform-verifier-android", + "rustls-webpki 0.103.10", + "security-framework 3.5.1", + "security-framework-sys", + "webpki-root-certs", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls-platform-verifier-android" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" + [[package]] name = "rustls-webpki" version = "0.101.7" @@ -2375,6 +2498,15 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a50f4cf475b65d88e057964e0e9bb1f0aa9bbb2036dc65c64596b42932536984" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "schannel" version = "0.1.28" @@ -2428,7 +2560,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" dependencies = [ "bitflags", - "core-foundation", + "core-foundation 0.9.4", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework" +version = "3.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3297343eaf830f66ede390ea39da1d462b6b0c1b000f420d0a83f898bbbe6ef" +dependencies = [ + "bitflags", + "core-foundation 0.10.1", "core-foundation-sys", "libc", "security-framework-sys", @@ -3024,6 +3169,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" @@ -3127,6 +3282,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-root-certs" +version = "1.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d46a5a140e6f7afeccd8eae97eff335163939eac8b929834875168b29b3d267" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "webpki-roots" version = "0.25.4" @@ -3179,6 +3343,15 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-sys" +version = "0.45.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75283be5efb2831d37ea142365f009c02ec203cd29a3ebecbc093d52315b66d0" +dependencies = [ + "windows-targets 0.42.2", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -3206,6 +3379,21 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-targets" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e5180c00cd44c9b1c88adb3693291f1cd93605ded80c250a75d472756b4d071" +dependencies = [ + "windows_aarch64_gnullvm 0.42.2", + "windows_aarch64_msvc 0.42.2", + "windows_i686_gnu 0.42.2", + "windows_i686_msvc 0.42.2", + "windows_x86_64_gnu 0.42.2", + "windows_x86_64_gnullvm 0.42.2", + "windows_x86_64_msvc 0.42.2", +] + [[package]] name = "windows-targets" version = "0.52.6" @@ -3239,6 +3427,12 @@ dependencies = [ "windows_x86_64_msvc 0.53.1", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "597a5118570b68bc08d8d59125332c54f1ba9d9adeedeef5b99b02ba2b0698f8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -3251,6 +3445,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" +[[package]] +name = "windows_aarch64_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e08e8864a60f06ef0d0ff4ba04124db8b0fb3be5776a5cd47641e942e58c4d43" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" @@ -3263,6 +3463,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" +[[package]] +name = "windows_i686_gnu" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c61d927d8da41da96a81f029489353e68739737d3beca43145c8afec9a31a84f" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -3287,6 +3493,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" +[[package]] +name = "windows_i686_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44d840b6ec649f480a41c8d80f9c65108b92d89345dd94027bfe06ac444d1060" + [[package]] name = "windows_i686_msvc" version = "0.52.6" @@ -3299,6 +3511,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" +[[package]] +name = "windows_x86_64_gnu" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8de912b8b8feb55c064867cf047dda097f92d51efad5b491dfb98f6bbb70cb36" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -3311,6 +3529,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d41b46a36d453748aedef1486d5c7a85db22e56aff34643984ea85514e94a3" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" @@ -3323,6 +3547,12 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" +[[package]] +name = "windows_x86_64_msvc" +version = "0.42.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9aec5da331524158c6d1a4ac0ab1541149c0b9505fde06423b02f5ef0106b9f0" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index 36ff484..0db04f3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,8 +35,8 @@ bdk_kyoto = { version = "0.15.4", optional = true } bdk_redb = { version = "0.1.1", optional = true } bdk_sp = { version = "0.1.0", optional = true, git = "https://github.com/bitcoindevkit/bdk-sp", tag = "v0.1.0" } shlex = { version = "1.3.0", optional = true } -payjoin = { version = "=1.0.0-rc.1", features = ["v1", "v2", "io", "_test-utils"], optional = true} -reqwest = { version = "0.13.2", default-features = false, optional = true } +payjoin = { version = "0.25.0", features = ["v1", "v2", "io", "_test-utils"], optional = true} +reqwest = { version = "0.13.2", default-features = false, features = ["rustls"], optional = true } url = { version = "2.5.8", optional = true } bdk_bip322 = { version = "0.1.0", optional = true } @@ -57,7 +57,7 @@ esplora = ["bdk_esplora", "_payjoin-dependencies"] rpc = ["bdk_bitcoind_rpc", "_payjoin-dependencies"] # Internal features -_payjoin-dependencies = ["payjoin", "reqwest", "url"] +_payjoin-dependencies = ["payjoin", "reqwest", "url", "sqlite"] bip322 = ["bdk_bip322"] # Use this to consensus verify transactions at sync time diff --git a/src/commands.rs b/src/commands.rs index d5ae053..83d92e3 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -38,7 +38,8 @@ use crate::handlers::{descriptor::SilentPaymentCodeCommand, offline::CreateSpTxC use crate::{ client::ClientType, handlers::online::{ - BroadcastCommand, FullScanCommand, ReceivePayjoinCommand, SendPayjoinCommand, SyncCommand, + BroadcastCommand, FullScanCommand, PayjoinHistoryCommand, ReceivePayjoinCommand, + ResumePayjoinCommand, SendPayjoinCommand, SyncCommand, }, }; @@ -381,6 +382,10 @@ pub enum OnlineWalletSubCommand { ReceivePayjoin(ReceivePayjoinCommand), /// Sends an original PSBT to a BIP 21 URI and broadcasts the returned Payjoin PSBT. SendPayjoin(SendPayjoinCommand), + /// Resume pending payjoin sessions. + ResumePayjoin(ResumePayjoinCommand), + /// Show payjoin session history. + PayjoinHistory(PayjoinHistoryCommand), } /// Subcommands for Key operations. diff --git a/src/error.rs b/src/error.rs index 1d5fd82..affd125 100644 --- a/src/error.rs +++ b/src/error.rs @@ -145,6 +145,10 @@ pub enum BDKCliError { #[error("Payjoin create request error: {0}")] PayjoinCreateRequest(#[from] payjoin::send::v2::CreateRequestError), + #[cfg(feature = "payjoin")] + #[error("Payjoin database error: {0}")] + PayjoinDb(#[from] crate::payjoin::db::Error), + #[cfg(feature = "bip322")] #[error("BIP-322 error: {0}")] Bip322Error(#[from] bdk_bip322::error::Error), @@ -176,3 +180,16 @@ impl From for BDKCliError { BDKCliError::RusqliteError(Box::new(err)) } } + +#[cfg(feature = "payjoin")] +impl + From> for BDKCliError +where + ApiErr: std::error::Error, + StorageErr: std::error::Error, + ErrorState: std::fmt::Debug, +{ + fn from(e: payjoin::persist::PersistedError) -> Self { + BDKCliError::Generic(e.to_string()) + } +} diff --git a/src/handlers/mod.rs b/src/handlers/mod.rs index 6e24339..41586ea 100644 --- a/src/handlers/mod.rs +++ b/src/handlers/mod.rs @@ -37,6 +37,7 @@ pub struct OfflineOperations<'a> { pub struct OnlineOperations<'a> { pub wallet: &'a mut Wallet, pub client: &'a BlockchainClient, + pub wallet_name: String, } /// The generic context @@ -79,11 +80,16 @@ impl<'a> AppContext> { datadir: PathBuf, wallet: &'a mut Wallet, client: &'a BlockchainClient, + wallet_name: String, ) -> Self { Self { network, datadir, - state: OnlineOperations { wallet, client }, + state: OnlineOperations { + wallet, + client, + wallet_name, + }, } } } diff --git a/src/handlers/online.rs b/src/handlers/online.rs index 2416e9d..885fd61 100644 --- a/src/handlers/online.rs +++ b/src/handlers/online.rs @@ -26,17 +26,17 @@ use { crate::commands::OnlineWalletSubCommand, crate::error::BDKCliError as Error, crate::handlers::{AppContext, AsyncAppCommand, OnlineOperations}, - crate::payjoin::{PayjoinManager, ohttp::RelayManager}, + crate::payjoin::PayjoinManager, crate::utils::is_final, crate::utils::output::FormatOutput, - crate::utils::print_wallet_events, crate::utils::types::{StatusResult, TransactionResult}, bdk_wallet::bitcoin::{ Psbt, Transaction, Txid, base64::Engine, base64::prelude::BASE64_STANDARD, consensus::Decodable, hex::FromHex, }, - std::sync::{Arc, Mutex}, }; +#[cfg(any(feature = "electrum", feature = "esplora", feature = "rpc"))] +use crate::utils::print_wallet_events; #[cfg(any( feature = "electrum", feature = "esplora", @@ -66,6 +66,14 @@ impl OnlineWalletSubCommand { let response: StatusResult = send_payjoin_command.execute(ctx).await?; response.write_out(std::io::stdout()) } + OnlineWalletSubCommand::ResumePayjoin(resume_payjoin_command) => { + let response: StatusResult = resume_payjoin_command.execute(ctx).await?; + response.write_out(std::io::stdout()) + } + OnlineWalletSubCommand::PayjoinHistory(payjoin_history_command) => { + let response: StatusResult = payjoin_history_command.execute(ctx).await?; + response.write_out(std::io::stdout()) + } } } } @@ -396,11 +404,12 @@ impl AsyncAppCommand>> for ReceivePayjoinCommand &self, ctx: &mut AppContext>, ) -> Result { + let wallet_name = ctx.state.wallet_name.clone(); + let datadir = ctx.datadir.clone(); let wallet = &mut ctx.state.wallet; let client = ctx.state.client; - let relay_manager = Arc::new(Mutex::new(RelayManager::new())); - let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager); + let mut payjoin_manager = PayjoinManager::new(wallet, Some(datadir), &wallet_name)?; let result = payjoin_manager .receive_payjoin( self.amount, @@ -446,11 +455,12 @@ impl AsyncAppCommand>> for SendPayjoinCommand { &self, ctx: &mut AppContext>, ) -> Result { + let wallet_name = ctx.state.wallet_name.clone(); + let datadir = ctx.datadir.clone(); let wallet = &mut ctx.state.wallet; let client = ctx.state.client; - let relay_manager = Arc::new(Mutex::new(RelayManager::new())); - let mut payjoin_manager = PayjoinManager::new(wallet, relay_manager); + let mut payjoin_manager = PayjoinManager::new(wallet, Some(datadir), &wallet_name)?; let result = payjoin_manager .send_payjoin( self.uri.clone(), @@ -463,3 +473,73 @@ impl AsyncAppCommand>> for SendPayjoinCommand { Ok(StatusResult { message: result }) } } + +#[derive(Parser, Debug, Clone, PartialEq, Eq)] +pub struct ResumePayjoinCommand { + /// Payjoin directory for the session + #[arg(env = "PAYJOIN_DIRECTORY", long = "directory", required = true)] + directory: String, + /// URL of the Payjoin OHTTP relay. Can be repeated multiple times. + #[arg(env = "PAYJOIN_OHTTP_RELAY", long = "ohttp_relay", required = true)] + ohttp_relay: Vec, + /// Resume only a specific active session ID (sender and/or receiver). + #[arg(env = "PAYJOIN_SESSION_ID", long = "session_id")] + session_id: Option, +} + +#[cfg(any( + feature = "electrum", + feature = "esplora", + feature = "cbf", + feature = "rpc" +))] +impl AsyncAppCommand>> for ResumePayjoinCommand { + type Output = StatusResult; + + async fn execute( + &self, + ctx: &mut AppContext>, + ) -> Result { + let wallet_name = ctx.state.wallet_name.clone(); + let datadir = ctx.datadir.clone(); + let wallet = &mut ctx.state.wallet; + let client = ctx.state.client; + + let mut payjoin_manager = PayjoinManager::new(wallet, Some(datadir), &wallet_name)?; + let result = payjoin_manager + .resume_payjoins( + self.directory.clone(), + self.ohttp_relay.clone(), + self.session_id, + client, + ) + .await?; + + Ok(StatusResult { message: result }) + } +} + +#[derive(Parser, Debug, Clone, PartialEq, Eq)] +pub struct PayjoinHistoryCommand {} + +#[cfg(any( + feature = "electrum", + feature = "esplora", + feature = "cbf", + feature = "rpc" +))] +impl AsyncAppCommand>> for PayjoinHistoryCommand { + type Output = StatusResult; + + async fn execute( + &self, + ctx: &mut AppContext>, + ) -> Result { + let wallet_name = ctx.state.wallet_name.clone(); + let datadir = ctx.datadir.clone(); + + let result = PayjoinManager::history(Some(datadir), &wallet_name)?; + + Ok(StatusResult { message: result }) + } +} diff --git a/src/handlers/repl.rs b/src/handlers/repl.rs index 73fbc41..a93b2f5 100644 --- a/src/handlers/repl.rs +++ b/src/handlers/repl.rs @@ -30,6 +30,13 @@ pub(crate) async fn respond( client: Option<&BlockchainClient>, line: &str, datadir: std::path::PathBuf, + #[cfg(any( + feature = "electrum", + feature = "esplora", + feature = "rpc", + feature = "cbf" + ))] + wallet_name: &str, ) -> Result { let args = shlex::split(line).ok_or("error: Invalid quoting".to_string())?; @@ -62,7 +69,13 @@ pub(crate) async fn respond( ))] WalletSubCommand::OnlineWalletSubCommand(cmd) => { let client_ref = client.ok_or("Online commands require a client.".to_string())?; - let mut ctx = AppContext::new_online_wallet(network, datadir, wallet, client_ref); + let mut ctx = AppContext::new_online_wallet( + network, + datadir, + wallet, + client_ref, + wallet_name.to_string(), + ); cmd.execute(&mut ctx) .await diff --git a/src/main.rs b/src/main.rs index 3f399dc..6501e70 100644 --- a/src/main.rs +++ b/src/main.rs @@ -80,6 +80,7 @@ async fn run(cli_opts: CliOpts) -> Result<(), Error> { runtime.home_dir.clone(), &mut wallet, &client, + runtime.wallet_name.clone(), ); cmd.execute(&mut ctx).await?; @@ -171,6 +172,13 @@ async fn run(cli_opts: CliOpts) -> Result<(), Error> { client.as_ref(), &line, runtime.home_dir.clone(), + #[cfg(any( + feature = "electrum", + feature = "esplora", + feature = "rpc", + feature = "cbf" + ))] + &wallet_name, ) .await .map_err(Error::Generic)?; diff --git a/src/payjoin/db.rs b/src/payjoin/db.rs new file mode 100644 index 0000000..f21524d --- /dev/null +++ b/src/payjoin/db.rs @@ -0,0 +1,835 @@ +use std::fmt; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, MutexGuard}; + +use bdk_wallet::rusqlite::{Connection, ToSql, params, types::ToSqlOutput}; +use payjoin::HpkePublicKey; +use payjoin::bitcoin::OutPoint; +use payjoin::bitcoin::consensus::encode::serialize; +use payjoin::persist::SessionPersister; +use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent; +use payjoin::send::v2::SessionEvent as SenderSessionEvent; + +use crate::error::BDKCliError; +use crate::utils::prepare_home_dir; + +pub type Result = std::result::Result; + +/// Error type for payjoin database operations +#[derive(Debug)] +pub enum Error { + /// SQLite database error + Rusqlite(bdk_wallet::rusqlite::Error), + /// JSON serialization error + Serialize(serde_json::Error), + /// JSON deserialization error + Deserialize(serde_json::Error), +} + +impl std::fmt::Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Error::Rusqlite(e) => write!(f, "Database operation failed: {e}"), + Error::Serialize(e) => write!(f, "Serialization failed: {e}"), + Error::Deserialize(e) => write!(f, "Deserialization failed: {e}"), + } + } +} + +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Error::Rusqlite(e) => Some(e), + Error::Serialize(e) => Some(e), + Error::Deserialize(e) => Some(e), + } + } +} + +impl From for Error { + fn from(error: bdk_wallet::rusqlite::Error) -> Self { + Error::Rusqlite(error) + } +} + +impl From for payjoin::ImplementationError { + fn from(error: Error) -> Self { + payjoin::ImplementationError::new(error) + } +} + +/// Default filename for the payjoin database +pub const DB_FILENAME: &str = "payjoin.sqlite"; +const SESSION_RETENTION_SECS: i64 = 30 * 24 * 60 * 60; + +pub fn open_payjoin_db( + datadir: Option, + wallet_name: &str, +) -> std::result::Result, BDKCliError> { + let wallet_dir = prepare_home_dir(datadir)?.join(wallet_name); + std::fs::create_dir_all(&wallet_dir).map_err(|e| BDKCliError::Generic(e.to_string()))?; + let db = Arc::new(Database::create(wallet_dir.join(DB_FILENAME))?); + db.prune_expired_sessions()?; + Ok(db) +} + +/// Returns the current Unix timestamp in seconds +#[inline] +fn now() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64 +} + +pub struct Database { + conn: Mutex, +} + +impl Database { + pub fn create(path: impl AsRef) -> Result { + let conn = Connection::open(path.as_ref())?; + Self::init_schema(&conn)?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + fn conn(&self) -> MutexGuard<'_, Connection> { + self.conn + .lock() + .expect("Database mutex should not be poisoned") + } + + fn init_schema(conn: &Connection) -> Result<()> { + conn.execute("PRAGMA foreign_keys = ON", [])?; + + conn.execute( + "CREATE TABLE IF NOT EXISTS send_sessions ( + session_id INTEGER PRIMARY KEY AUTOINCREMENT, + receiver_pubkey BLOB NOT NULL, + completed_at INTEGER + )", + [], + )?; + + conn.execute( + "CREATE TABLE IF NOT EXISTS receive_sessions ( + session_id INTEGER PRIMARY KEY AUTOINCREMENT, + completed_at INTEGER + )", + [], + )?; + + conn.execute( + "CREATE TABLE IF NOT EXISTS send_session_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id INTEGER NOT NULL, + event_data TEXT NOT NULL, + created_at INTEGER NOT NULL, + FOREIGN KEY(session_id) REFERENCES send_sessions(session_id) + )", + [], + )?; + + conn.execute( + "CREATE TABLE IF NOT EXISTS receive_session_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id INTEGER NOT NULL, + event_data TEXT NOT NULL, + created_at INTEGER NOT NULL, + FOREIGN KEY(session_id) REFERENCES receive_sessions(session_id) + )", + [], + )?; + + conn.execute( + "CREATE TABLE IF NOT EXISTS inputs_seen ( + outpoint BLOB PRIMARY KEY, + created_at INTEGER NOT NULL + )", + [], + )?; + + Ok(()) + } + + /// Inserts the input and returns true if the input was seen before, false otherwise. + /// This is used for replay protection to prevent probing attacks. + pub fn insert_input_seen_before(&self, input: OutPoint) -> Result { + let key = serialize(&input); + let was_seen_before = self.conn().execute( + "INSERT OR IGNORE INTO inputs_seen (outpoint, created_at) VALUES (?1, ?2)", + params![key, now()], + )? == 0; + Ok(was_seen_before) + } + + /// Removes old completed sessions and stale incomplete sessions plus their event logs. + pub fn prune_expired_sessions(&self) -> Result<()> { + let cutoff = now() - SESSION_RETENTION_SECS; + let mut conn = self.conn(); + let tx = conn.transaction()?; + let stale_send_session_ids = { + let mut stmt = tx.prepare( + "SELECT session_id FROM send_sessions + WHERE (completed_at IS NOT NULL AND completed_at < ?1) + OR ( + completed_at IS NULL + AND session_id IN ( + SELECT session_id FROM send_session_events + GROUP BY session_id + HAVING MAX(created_at) < ?1 + ) + )", + )?; + let rows = stmt.query_map(params![cutoff], |row| row.get::<_, i64>(0))?; + let mut ids = Vec::new(); + for row in rows { + ids.push(row?); + } + ids + }; + let stale_receive_session_ids = { + let mut stmt = tx.prepare( + "SELECT session_id FROM receive_sessions + WHERE (completed_at IS NOT NULL AND completed_at < ?1) + OR ( + completed_at IS NULL + AND session_id IN ( + SELECT session_id FROM receive_session_events + GROUP BY session_id + HAVING MAX(created_at) < ?1 + ) + )", + )?; + let rows = stmt.query_map(params![cutoff], |row| row.get::<_, i64>(0))?; + let mut ids = Vec::new(); + for row in rows { + ids.push(row?); + } + ids + }; + let deleted_any = + !stale_send_session_ids.is_empty() || !stale_receive_session_ids.is_empty(); + + for session_id in stale_send_session_ids { + tx.execute( + "DELETE FROM send_session_events WHERE session_id = ?1", + params![session_id], + )?; + tx.execute( + "DELETE FROM send_sessions WHERE session_id = ?1", + params![session_id], + )?; + } + + for session_id in stale_receive_session_ids { + tx.execute( + "DELETE FROM receive_session_events WHERE session_id = ?1", + params![session_id], + )?; + tx.execute( + "DELETE FROM receive_sessions WHERE session_id = ?1", + params![session_id], + )?; + } + + tx.commit()?; + if deleted_any { + conn.execute("VACUUM", [])?; + } + Ok(()) + } + + /// Returns IDs of all active (incomplete) receive sessions + pub fn get_recv_session_ids(&self) -> Result> { + let conn = self.conn(); + let mut stmt = + conn.prepare("SELECT session_id FROM receive_sessions WHERE completed_at IS NULL ORDER BY session_id DESC")?; + + let session_rows = stmt.query_map([], |row| { + let session_id: i64 = row.get(0)?; + Ok(SessionId(session_id)) + })?; + + let mut session_ids = Vec::new(); + for session_row in session_rows { + session_ids.push(session_row?); + } + + Ok(session_ids) + } + + /// Returns IDs of all active (incomplete) send sessions + pub fn get_send_session_ids(&self) -> Result> { + let conn = self.conn(); + let mut stmt = + conn.prepare("SELECT session_id FROM send_sessions WHERE completed_at IS NULL ORDER BY session_id DESC")?; + + let session_rows = stmt.query_map([], |row| { + let session_id: i64 = row.get(0)?; + Ok(SessionId(session_id)) + })?; + + let mut session_ids = Vec::new(); + for session_row in session_rows { + session_ids.push(session_row?); + } + + Ok(session_ids) + } + + /// Returns the receiver public key for a send session + pub fn get_send_session_receiver_pk(&self, session_id: &SessionId) -> Result { + let conn = self.conn(); + let mut stmt = + conn.prepare("SELECT receiver_pubkey FROM send_sessions WHERE session_id = ?1")?; + let receiver_pubkey: Vec = stmt.query_row(params![session_id], |row| row.get(0))?; + Ok(HpkePublicKey::from_compressed_bytes(&receiver_pubkey).expect("Valid receiver pubkey")) + } + + /// Returns IDs and completion timestamps of all completed send sessions + pub fn get_inactive_send_session_ids(&self) -> Result> { + let conn = self.conn(); + let mut stmt = conn.prepare( + "SELECT session_id, completed_at FROM send_sessions WHERE completed_at IS NOT NULL", + )?; + let session_rows = stmt.query_map([], |row| { + let session_id: i64 = row.get(0)?; + let completed_at: u64 = row.get(1)?; + Ok((SessionId(session_id), completed_at)) + })?; + + let mut session_ids = Vec::new(); + for session_row in session_rows { + session_ids.push(session_row?); + } + Ok(session_ids) + } + + /// Returns IDs and completion timestamps of all completed receive sessions + pub fn get_inactive_recv_session_ids(&self) -> Result> { + let conn = self.conn(); + let mut stmt = conn.prepare( + "SELECT session_id, completed_at FROM receive_sessions WHERE completed_at IS NOT NULL", + )?; + let session_rows = stmt.query_map([], |row| { + let session_id: i64 = row.get(0)?; + let completed_at: u64 = row.get(1)?; + Ok((SessionId(session_id), completed_at)) + })?; + + let mut session_ids = Vec::new(); + for session_row in session_rows { + session_ids.push(session_row?); + } + Ok(session_ids) + } + + /// Formats a Unix timestamp into local date time text. + pub fn format_unix_timestamp(&self, timestamp: u64) -> Result { + let Ok(timestamp) = i64::try_from(timestamp) else { + return Ok(format!("Invalid timestamp ({timestamp})")); + }; + let conn = self.conn(); + let dt: Option = conn.query_row( + "SELECT datetime(?1, 'unixepoch', 'localtime')", + params![timestamp], + |row| row.get(0), + )?; + Ok(dt.unwrap_or_else(|| format!("Invalid timestamp ({timestamp})"))) + } +} + +/// Wrapper type for session IDs +#[derive(Debug, Clone)] +pub struct SessionId(i64); + +impl ToSql for SessionId { + fn to_sql(&self) -> bdk_wallet::rusqlite::Result> { + self.0.to_sql() + } +} + +impl fmt::Display for SessionId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl SessionId { + pub fn as_i64(&self) -> i64 { + self.0 + } +} + +/// Persister for payjoin v2 send sessions +#[derive(Clone)] +pub struct SenderPersister { + db: Arc, + session_id: SessionId, +} + +impl SenderPersister { + /// Creates a new sender persister, creating a new session in the database + pub fn new(db: Arc, receiver_pubkey: HpkePublicKey) -> Result { + let session_id: i64 = db.conn().query_row( + "INSERT INTO send_sessions (session_id, receiver_pubkey) VALUES (NULL, ?1) RETURNING session_id", + params![receiver_pubkey.to_compressed_bytes()], + |row| row.get(0), + )?; + + Ok(Self { + db, + session_id: SessionId(session_id), + }) + } + + /// Creates a persister from an existing session ID + pub fn from_id(db: Arc, id: SessionId) -> Self { + Self { db, session_id: id } + } +} + +impl SessionPersister for SenderPersister { + type SessionEvent = SenderSessionEvent; + type InternalStorageError = Error; + + fn save_event( + &self, + event: SenderSessionEvent, + ) -> std::result::Result<(), Self::InternalStorageError> { + let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?; + + self.db.conn().execute( + "INSERT INTO send_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)", + params![self.session_id, event_data, now()], + )?; + + Ok(()) + } + + fn load( + &self, + ) -> std::result::Result>, Self::InternalStorageError> + { + let conn = self.db.conn(); + let mut stmt = conn.prepare( + "SELECT event_data FROM send_session_events WHERE session_id = ?1 ORDER BY id ASC", + )?; + + let event_rows = stmt.query_map(params![self.session_id], |row| { + let event_data: String = row.get(0)?; + Ok(event_data) + })?; + + let events: Vec = event_rows + .map(|row| { + let event_data = row.expect("Failed to read event data from database"); + serde_json::from_str::(&event_data) + .expect("Database corruption: failed to deserialize session event") + }) + .collect(); + + Ok(Box::new(events.into_iter())) + } + + fn close(&self) -> std::result::Result<(), Self::InternalStorageError> { + self.db.conn().execute( + "UPDATE send_sessions SET completed_at = ?1 WHERE session_id = ?2", + params![now(), self.session_id], + )?; + + Ok(()) + } +} + +/// Persister for payjoin v2 receive sessions +#[derive(Clone)] +pub struct ReceiverPersister { + db: Arc, + session_id: SessionId, +} + +impl ReceiverPersister { + /// Creates a new receiver persister, creating a new session in the database + pub fn new(db: Arc) -> Result { + let session_id: i64 = db.conn().query_row( + "INSERT INTO receive_sessions (session_id) VALUES (NULL) RETURNING session_id", + [], + |row| row.get(0), + )?; + + Ok(Self { + db, + session_id: SessionId(session_id), + }) + } + + /// Creates a persister from an existing session ID + pub fn from_id(db: Arc, id: SessionId) -> Self { + Self { db, session_id: id } + } +} + +impl SessionPersister for ReceiverPersister { + type SessionEvent = ReceiverSessionEvent; + type InternalStorageError = Error; + + fn save_event( + &self, + event: ReceiverSessionEvent, + ) -> std::result::Result<(), Self::InternalStorageError> { + let event_data = serde_json::to_string(&event).map_err(Error::Serialize)?; + + self.db.conn().execute( + "INSERT INTO receive_session_events (session_id, event_data, created_at) VALUES (?1, ?2, ?3)", + params![self.session_id, event_data, now()], + )?; + + Ok(()) + } + + fn load( + &self, + ) -> std::result::Result< + Box>, + Self::InternalStorageError, + > { + let conn = self.db.conn(); + let mut stmt = conn.prepare( + "SELECT event_data FROM receive_session_events WHERE session_id = ?1 ORDER BY id ASC", + )?; + + let event_rows = stmt.query_map(params![self.session_id], |row| { + let event_data: String = row.get(0)?; + Ok(event_data) + })?; + + let events: Vec = event_rows + .map(|row| { + let event_data = row.expect("Failed to read event data from database"); + serde_json::from_str::(&event_data) + .expect("Database corruption: failed to deserialize session event") + }) + .collect(); + + Ok(Box::new(events.into_iter())) + } + + fn close(&self) -> std::result::Result<(), Self::InternalStorageError> { + self.db.conn().execute( + "UPDATE receive_sessions SET completed_at = ?1 WHERE session_id = ?2", + params![now(), self.session_id], + )?; + + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + use std::time::{SystemTime, UNIX_EPOCH}; + + use payjoin::HpkeKeyPair; + use payjoin::persist::SessionPersister as _; + use payjoin::receive::v2::SessionOutcome as ReceiverSessionOutcome; + use payjoin::send::v2::SessionOutcome as SenderSessionOutcome; + + use super::*; + + fn sample_receiver_pubkey() -> HpkePublicKey { + HpkeKeyPair::gen_keypair().1 + } + + fn unique_test_datadir(label: &str) -> PathBuf { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("clock should be after unix epoch") + .as_nanos(); + std::env::temp_dir().join(format!( + "bdk-cli-payjoin-{label}-{}-{nanos}", + std::process::id() + )) + } + + #[test] + fn insert_input_seen_before_reports_replays() { + let db = Database::create(":memory:").expect("in-memory database should open"); + let input = OutPoint::null(); + + assert!( + !db.insert_input_seen_before(input) + .expect("first insert should succeed"), + "first observation should not be treated as a replay" + ); + assert!( + db.insert_input_seen_before(input) + .expect("second insert should succeed"), + "second observation should be treated as a replay" + ); + } + + #[test] + fn persisters_round_trip_events_and_transition_sessions() { + let db = Arc::new(Database::create(":memory:").expect("in-memory database should open")); + + let sender = SenderPersister::new(db.clone(), sample_receiver_pubkey()) + .expect("sender session should be created"); + sender + .save_event(SenderSessionEvent::PostedOriginalPsbt()) + .expect("sender event should persist"); + sender + .save_event(SenderSessionEvent::Closed(SenderSessionOutcome::Failure)) + .expect("sender close event should persist"); + + let sender_events: Vec<_> = sender.load().expect("sender events should load").collect(); + assert_eq!( + sender_events, + vec![ + SenderSessionEvent::PostedOriginalPsbt(), + SenderSessionEvent::Closed(SenderSessionOutcome::Failure), + ] + ); + + let active_sender_ids = db + .get_send_session_ids() + .expect("active sender ids should load"); + assert_eq!(active_sender_ids.len(), 1); + assert_eq!( + db.get_send_session_receiver_pk(&active_sender_ids[0]) + .expect("receiver pubkey should load") + .to_compressed_bytes() + .len(), + 33 + ); + + sender.close().expect("sender session should close"); + assert!( + db.get_send_session_ids() + .expect("active sender ids should load") + .is_empty() + ); + assert_eq!( + db.get_inactive_send_session_ids() + .expect("inactive sender ids should load") + .len(), + 1 + ); + + let receiver = + ReceiverPersister::new(db.clone()).expect("receiver session should be created"); + receiver + .save_event(ReceiverSessionEvent::CheckedBroadcastSuitability()) + .expect("receiver event should persist"); + receiver + .save_event(ReceiverSessionEvent::Closed(ReceiverSessionOutcome::Cancel)) + .expect("receiver close event should persist"); + + let receiver_events: Vec<_> = receiver + .load() + .expect("receiver events should load") + .collect(); + assert_eq!( + receiver_events, + vec![ + ReceiverSessionEvent::CheckedBroadcastSuitability(), + ReceiverSessionEvent::Closed(ReceiverSessionOutcome::Cancel), + ] + ); + + receiver.close().expect("receiver session should close"); + assert!( + db.get_recv_session_ids() + .expect("active receiver ids should load") + .is_empty() + ); + assert_eq!( + db.get_inactive_recv_session_ids() + .expect("inactive receiver ids should load") + .len(), + 1 + ); + } + + #[test] + fn prune_expired_sessions_drops_stale_send_and_receive_rows() { + let db = Database::create(":memory:").expect("in-memory database should open"); + let stale_timestamp = now() - SESSION_RETENTION_SECS - 1; + let fresh_timestamp = now(); + let receiver_pubkey = sample_receiver_pubkey(); + + db.conn() + .execute( + "INSERT INTO send_sessions (session_id, receiver_pubkey, completed_at) + VALUES (?1, ?2, ?3)", + params![ + 1_i64, + receiver_pubkey.to_compressed_bytes(), + stale_timestamp + ], + ) + .expect("stale completed send session should insert"); + db.conn() + .execute( + "INSERT INTO send_sessions (session_id, receiver_pubkey, completed_at) + VALUES (?1, ?2, NULL)", + params![2_i64, receiver_pubkey.to_compressed_bytes()], + ) + .expect("stale active send session should insert"); + db.conn() + .execute( + "INSERT INTO send_session_events (session_id, event_data, created_at) + VALUES (?1, ?2, ?3)", + params![ + 2_i64, + serde_json::to_string(&SenderSessionEvent::PostedOriginalPsbt()) + .expect("event should serialize"), + stale_timestamp + ], + ) + .expect("stale send event should insert"); + db.conn() + .execute( + "INSERT INTO send_sessions (session_id, receiver_pubkey, completed_at) + VALUES (?1, ?2, NULL)", + params![3_i64, receiver_pubkey.to_compressed_bytes()], + ) + .expect("fresh active send session should insert"); + db.conn() + .execute( + "INSERT INTO send_session_events (session_id, event_data, created_at) + VALUES (?1, ?2, ?3)", + params![ + 3_i64, + serde_json::to_string(&SenderSessionEvent::PostedOriginalPsbt()) + .expect("event should serialize"), + fresh_timestamp + ], + ) + .expect("fresh send event should insert"); + + db.conn() + .execute( + "INSERT INTO receive_sessions (session_id, completed_at) VALUES (?1, ?2)", + params![4_i64, stale_timestamp], + ) + .expect("stale completed receive session should insert"); + db.conn() + .execute( + "INSERT INTO receive_sessions (session_id, completed_at) VALUES (?1, NULL)", + params![5_i64], + ) + .expect("stale active receive session should insert"); + db.conn() + .execute( + "INSERT INTO receive_session_events (session_id, event_data, created_at) + VALUES (?1, ?2, ?3)", + params![ + 5_i64, + serde_json::to_string(&ReceiverSessionEvent::CheckedBroadcastSuitability()) + .expect("event should serialize"), + stale_timestamp + ], + ) + .expect("stale receive event should insert"); + db.conn() + .execute( + "INSERT INTO receive_sessions (session_id, completed_at) VALUES (?1, NULL)", + params![6_i64], + ) + .expect("fresh active receive session should insert"); + db.conn() + .execute( + "INSERT INTO receive_session_events (session_id, event_data, created_at) + VALUES (?1, ?2, ?3)", + params![ + 6_i64, + serde_json::to_string(&ReceiverSessionEvent::CheckedBroadcastSuitability()) + .expect("event should serialize"), + fresh_timestamp + ], + ) + .expect("fresh receive event should insert"); + + db.prune_expired_sessions().expect("pruning should succeed"); + + let remaining_send_ids: Vec = db + .conn() + .prepare("SELECT session_id FROM send_sessions ORDER BY session_id") + .expect("statement should prepare") + .query_map([], |row| row.get(0)) + .expect("query should execute") + .map(|row| row.expect("row should decode")) + .collect(); + assert_eq!(remaining_send_ids, vec![3]); + + let remaining_receive_ids: Vec = db + .conn() + .prepare("SELECT session_id FROM receive_sessions ORDER BY session_id") + .expect("statement should prepare") + .query_map([], |row| row.get(0)) + .expect("query should execute") + .map(|row| row.expect("row should decode")) + .collect(); + assert_eq!(remaining_receive_ids, vec![6]); + } + + #[test] + fn history_lists_active_and_completed_sessions() { + let datadir = unique_test_datadir("history"); + let wallet_name = "history-wallet"; + let db = open_payjoin_db(Some(datadir.clone()), wallet_name) + .expect("database should open in temp directory"); + + let active_sender = SenderPersister::new(db.clone(), sample_receiver_pubkey()) + .expect("active sender should be created"); + let active_receiver = + ReceiverPersister::new(db.clone()).expect("active receiver should be created"); + + let completed_sender = SenderPersister::new(db.clone(), sample_receiver_pubkey()) + .expect("completed sender should be created"); + completed_sender + .close() + .expect("completed sender should close"); + + let completed_receiver = + ReceiverPersister::new(db.clone()).expect("completed receiver should be created"); + completed_receiver + .close() + .expect("completed receiver should close"); + + let active_send_ids = db + .get_send_session_ids() + .expect("active sender ids should load"); + let active_recv_ids = db + .get_recv_session_ids() + .expect("active receiver ids should load"); + let inactive_send_ids = db + .get_inactive_send_session_ids() + .expect("inactive sender ids should load"); + let inactive_recv_ids = db + .get_inactive_recv_session_ids() + .expect("inactive receiver ids should load"); + + let table = crate::payjoin::PayjoinManager::history(Some(datadir.clone()), wallet_name) + .expect("history should render"); + + assert!(table.contains("Sender")); + assert!(table.contains("Receiver")); + assert!(table.contains(&active_send_ids[0].to_string())); + assert!(table.contains(&active_recv_ids[0].to_string())); + assert!(table.contains(&inactive_send_ids[0].0.to_string())); + assert!(table.contains(&inactive_recv_ids[0].0.to_string())); + assert!(table.contains("Not Completed")); + + drop(active_sender); + drop(active_receiver); + drop(completed_sender); + drop(completed_receiver); + drop(db); + let _ = std::fs::remove_dir_all(datadir); + } +} diff --git a/src/payjoin/mod.rs b/src/payjoin/mod.rs index 0eaa059..11e09a0 100644 --- a/src/payjoin/mod.rs +++ b/src/payjoin/mod.rs @@ -1,56 +1,118 @@ -#![cfg(any( - feature = "electrum", - feature = "esplora", - feature = "rpc", - feature = "cbf" -))] - use crate::client::BlockchainClient; +use crate::error::BDKCliError as Error; use bdk_wallet::{ SignOptions, Wallet, bitcoin::{FeeRate, Psbt, Txid, consensus::encode::serialize_hex}, }; +use cli_table::{Cell, CellStruct, Style, Table}; use payjoin::bitcoin::TxIn; use payjoin::persist::{OptionalTransitionOutcome, SessionPersister}; use payjoin::receive::InputPair; use payjoin::receive::v2::{ HasReplyableError, Initialized, MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal, ProvisionalProposal, ReceiveSession, Receiver, - SessionEvent as ReceiverSessionEvent, UncheckedOriginalPayload, WantsFeeRange, WantsInputs, - WantsOutputs, + SessionEvent as ReceiverSessionEvent, SessionOutcome as ReceiverSessionOutcome, + UncheckedOriginalPayload, WantsFeeRange, WantsInputs, WantsOutputs, + replay_event_log as replay_receiver_event_log, }; use payjoin::send::v2::{ PollingForProposal, SendSession, Sender, SessionEvent as SenderSessionEvent, SessionOutcome as SenderSessionOutcome, WithReplyKey, + replay_event_log as replay_sender_event_log, }; -use payjoin::{ImplementationError, UriExt}; +use payjoin::{HpkePublicKey, ImplementationError, UriExt}; use serde_json::{json, to_string_pretty}; -use std::sync::{Arc, Mutex}; +use std::{path::PathBuf, sync::Arc}; -use crate::{ - error::BDKCliError as Error, - payjoin::ohttp::{RelayManager, fetch_ohttp_keys}, -}; +use crate::payjoin::db::{ReceiverPersister, SenderPersister, open_payjoin_db}; +use crate::payjoin::ohttp::RelayManager; +pub mod db; pub mod ohttp; -/// Implements all of the functions required to go through the Payjoin receive and send processes. +/// Coordinates Payjoin receive and send flows. /// -/// TODO: At the time of writing, this struct is written to make a Persister implementation easier -/// but the persister is not implemented yet! For instance [`PayjoinManager::proceed_sender_session`] and -/// [`PayjoinManager::proceed_receiver_session`] are designed such that the manager can enable -/// resuming ongoing payjoins are well. So... this is a TODO for implementing persister. +/// The manager owns the wallet, relay manager, and database handles needed to persist session +/// events and resume ongoing Payjoins through [`PayjoinManager::proceed_sender_session`] and +/// [`PayjoinManager::proceed_receiver_session`]. pub(crate) struct PayjoinManager<'a> { wallet: &'a mut Wallet, - relay_manager: Arc>, + relay_manager: RelayManager, + db: Arc, +} + +trait StatusText { + fn status_text(&self) -> &'static str; +} + +impl StatusText for SendSession { + fn status_text(&self) -> &'static str { + match self { + SendSession::WithReplyKey(_) | SendSession::PollingForProposal(_) => { + "Waiting for proposal" + } + SendSession::Closed(session_outcome) => match session_outcome { + SenderSessionOutcome::Failure => "Session failure", + SenderSessionOutcome::Success(_) => "Session success", + SenderSessionOutcome::Cancel => "Session cancelled", + }, + } + } +} + +impl StatusText for ReceiveSession { + fn status_text(&self) -> &'static str { + match self { + ReceiveSession::Initialized(_) => "Waiting for original proposal", + ReceiveSession::UncheckedOriginalPayload(_) + | ReceiveSession::MaybeInputsOwned(_) + | ReceiveSession::MaybeInputsSeen(_) + | ReceiveSession::OutputsUnknown(_) + | ReceiveSession::WantsOutputs(_) + | ReceiveSession::WantsInputs(_) + | ReceiveSession::WantsFeeRange(_) + | ReceiveSession::ProvisionalProposal(_) => "Processing original proposal", + ReceiveSession::PayjoinProposal(_) => "Payjoin proposal sent", + ReceiveSession::HasReplyableError(_) => { + "Session failure, waiting to post error response" + } + ReceiveSession::Monitor(_) => "Monitoring payjoin proposal", + ReceiveSession::Closed(session_outcome) => match session_outcome { + ReceiverSessionOutcome::Failure => "Session failure", + ReceiverSessionOutcome::Success(_) => { + "Session success, Payjoin proposal was broadcasted" + } + ReceiverSessionOutcome::Cancel => "Session cancelled", + ReceiverSessionOutcome::FallbackBroadcasted => "Fallback broadcasted", + ReceiverSessionOutcome::PayjoinProposalSent => { + "Payjoin proposal sent, skipping monitoring as the sender is spending non-SegWit inputs" + } + }, + } + } +} + +struct SessionHistoryRow { + id: String, + role: &'static str, + status: String, + completed_at: Option, } impl<'a> PayjoinManager<'a> { - pub fn new(wallet: &'a mut Wallet, relay_manager: Arc>) -> Self { - Self { + pub fn new( + wallet: &'a mut Wallet, + datadir: Option, + wallet_name: &str, + ) -> Result { + let db = open_payjoin_db(datadir, wallet_name)?; + let relay_manager = RelayManager::new(); + + Ok(Self { wallet, relay_manager, - } + db, + }) } pub async fn receive_payjoin( @@ -71,35 +133,26 @@ impl<'a> PayjoinManager<'a> { .collect::>() .map_err(|e| Error::Generic(format!("Failed to parse one or more OHTTP URLs: {e}")))?; - if ohttp_relays.is_empty() { - return Err(Error::Generic( - "At least one valid OHTTP relay must be provided.".into(), - )); - } + self.relay_manager.configure(ohttp_relays)?; + let ohttp_keys = self.relay_manager.fetch_ohttp_keys(&directory).await?; - let ohttp_keys = - fetch_ohttp_keys(ohttp_relays, &directory, self.relay_manager.clone()).await?; - // TODO: Implement proper persister. - let persister = payjoin::persist::NoopSessionPersister::::default(); + let persister = crate::payjoin::db::ReceiverPersister::new(self.db.clone())?; let checked_max_fee_rate = max_fee_rate .map(FeeRate::from_sat_per_kwu) .unwrap_or(FeeRate::BROADCAST_MIN); - let receiver = payjoin::receive::v2::ReceiverBuilder::new( - address.address, - directory, - ohttp_keys.ohttp_keys, - )? - .with_amount(payjoin::bitcoin::Amount::from_sat(amount)) - .with_max_fee_rate(checked_max_fee_rate) - .build() - .save(&persister) - .map_err(|e| { - Error::Generic(format!( - "Failed to persister the receiver after initialization: {e}" - )) - })?; + let receiver = + payjoin::receive::v2::ReceiverBuilder::new(address.address, directory, ohttp_keys)? + .with_amount(payjoin::bitcoin::Amount::from_sat(amount)) + .with_max_fee_rate(checked_max_fee_rate) + .build() + .save(&persister) + .map_err(|e| { + Error::Generic(format!( + "Failed to persister the receiver after initialization: {e}" + )) + })?; let pj_uri = receiver.pj_uri(); println!("Request Payjoin by sharing this Payjoin Uri:"); @@ -108,7 +161,6 @@ impl<'a> PayjoinManager<'a> { self.proceed_receiver_session( ReceiveSession::Initialized(receiver.clone()), &persister, - ohttp_keys.relay_url.to_string(), checked_max_fee_rate, blockchain_client, ) @@ -117,12 +169,6 @@ impl<'a> PayjoinManager<'a> { Ok(to_string_pretty(&json!({}))?) } - #[cfg(any( - feature = "electrum", - feature = "esplora", - feature = "rpc", - feature = "cbf" - ))] pub async fn send_payjoin( &mut self, uri: String, @@ -175,7 +221,7 @@ impl<'a> PayjoinManager<'a> { self.process_payjoin_proposal(psbt, blockchain_client) .await? } - payjoin::PjParam::V2(_) => { + payjoin::PjParam::V2(v2_param) => { let ohttp_relays: Vec = ohttp_relays .into_iter() .map(|s| url::Url::parse(&s)) @@ -184,37 +230,53 @@ impl<'a> PayjoinManager<'a> { Error::Generic(format!("Failed to parse one or more OHTTP URLs: {e}")) })?; - if ohttp_relays.is_empty() { - return Err(Error::Generic( - "At least one valid OHTTP relay must be provided.".into(), - )); - } - - // TODO: Implement proper persister. - let persister = - payjoin::persist::NoopSessionPersister::::default(); - - let sender = payjoin::send::v2::SenderBuilder::new(original_psbt.clone(), uri) - .build_recommended(fee_rate)? - .save(&persister) - .map_err(|e| { - Error::Generic(format!( - "Failed to save the Payjoin v2 sender in the persister: {e}" - )) - })?; - - let selected_relay = - fetch_ohttp_keys(ohttp_relays, &sender.endpoint(), self.relay_manager.clone()) - .await? - .relay_url; + self.relay_manager.configure(ohttp_relays)?; + // Check for existing session with the same receiver pubkey + let receiver_pubkey = v2_param.receiver_pubkey(); + let existing_session = + self.db + .get_send_session_ids()? + .into_iter() + .find_map(|session_id| { + let session_receiver_pubkey = self + .db + .get_send_session_receiver_pk(&session_id) + .expect("Receiver pubkey should exist if session id exists"); + if session_receiver_pubkey == *receiver_pubkey { + Some(session_id) + } else { + None + } + }); + + let (sender_state, persister) = if let Some(session_id) = existing_session { + let sender_persister = SenderPersister::from_id(self.db.clone(), session_id); + let (send_session, _) = + replay_sender_event_log(&sender_persister).map_err(|e| { + Error::Generic(format!("Failed to replay sender event log: {e:?}")) + })?; + println!("Resuming existing sender session"); + (send_session, sender_persister) + } else { + let persister = { + let receiver_pubkey: HpkePublicKey = v2_param.receiver_pubkey().clone(); + SenderPersister::new(self.db.clone(), receiver_pubkey)? + }; + + let sender = payjoin::send::v2::SenderBuilder::new(original_psbt.clone(), uri) + .build_recommended(fee_rate)? + .save(&persister) + .map_err(|e| { + Error::Generic(format!( + "Failed to save the Payjoin v2 sender in the persister: {e}" + )) + })?; + + (SendSession::WithReplyKey(sender), persister) + }; - self.proceed_sender_session( - SendSession::WithReplyKey(sender), - &persister, - selected_relay.to_string(), - blockchain_client, - ) - .await? + self.proceed_sender_session(sender_state, &persister, blockchain_client) + .await? } _ => { unimplemented!("Payjoin version not recognized."); @@ -228,20 +290,13 @@ impl<'a> PayjoinManager<'a> { &mut self, session: ReceiveSession, persister: &impl SessionPersister, - relay: impl payjoin::IntoUrl, max_fee_rate: FeeRate, blockchain_client: &BlockchainClient, ) -> Result<(), Error> { match session { ReceiveSession::Initialized(proposal) => { - self.read_from_directory( - proposal, - persister, - relay, - max_fee_rate, - blockchain_client, - ) - .await + self.read_from_directory(proposal, persister, max_fee_rate, blockchain_client) + .await } ReceiveSession::UncheckedOriginalPayload(proposal) => { self.check_proposal(proposal, persister, max_fee_rate, blockchain_client) @@ -297,14 +352,14 @@ impl<'a> PayjoinManager<'a> { &mut self, receiver: Receiver, persister: &impl SessionPersister, - relay: impl payjoin::IntoUrl, max_fee_rate: FeeRate, blockchain_client: &BlockchainClient, ) -> Result<(), Error> { let mut current_receiver_typestate = receiver; let next_receiver_typestate = loop { - let (req, context) = current_receiver_typestate.create_poll_request(relay.as_str())?; - let response = self.send_payjoin_post_request(req).await?; + let (response, context) = self + .post_via_relay(|relay| current_receiver_typestate.create_poll_request(relay)) + .await?; let state_transition = current_receiver_typestate .process_response(response.bytes().await?.to_vec().as_slice(), context) .save(persister); @@ -374,13 +429,8 @@ impl<'a> PayjoinManager<'a> { blockchain_client: &BlockchainClient, ) -> Result<(), Error> { let next_receiver_typestate = receiver - .check_inputs_not_owned(&mut |input| { - Ok(self.wallet.is_mine(input.to_owned())) - }) - .save(persister) - .map_err(|e| { - Error::Generic(format!("Error occurred when saving after checking if inputs in the original proposal are not owned: {e}")) - })?; + .check_inputs_not_owned(&mut |input| Ok(self.wallet.is_mine(input.to_owned()))) + .save(persister)?; self.check_no_inputs_seen_before( next_receiver_typestate, @@ -398,16 +448,11 @@ impl<'a> PayjoinManager<'a> { max_fee_rate: FeeRate, blockchain_client: &BlockchainClient, ) -> Result<(), Error> { - // This is not supported as there is no persistence of previous Payjoin attempts in BDK CLI - // yet. If there is support either in the BDK persister or Payjoin persister, this can be - // implemented, but it is not a concern as the use cases of the CLI does not warrant - // protection against probing attacks. - println!( - "Checking whether the inputs in the proposal were seen before to protect from probing attacks is not supported. Skipping the check..." - ); - let next_receiver_typestate = receiver.check_no_inputs_seen_before(&mut |_| Ok(false)).save(persister).map_err(|e| { - Error::Generic(format!("Error occurred when saving after checking if the inputs in the proposal were seen before: {e}")) - })?; + let db = self.db.clone(); + let next_receiver_typestate = receiver + .check_no_inputs_seen_before(&mut |input| Ok(db.insert_input_seen_before(*input)?)) + .save(persister)?; + self.identify_receiver_outputs( next_receiver_typestate, persister, @@ -424,11 +469,11 @@ impl<'a> PayjoinManager<'a> { max_fee_rate: FeeRate, blockchain_client: &BlockchainClient, ) -> Result<(), Error> { - let next_receiver_typestate = receiver.identify_receiver_outputs(&mut |output_script| { - Ok(self.wallet.is_mine(output_script.to_owned())) - }).save(persister).map_err(|e| { - Error::Generic(format!("Error occurred when saving after checking if the outputs in the original proposal are owned by the receiver: {e}")) - })?; + let next_receiver_typestate = receiver + .identify_receiver_outputs(&mut |output_script| { + Ok(self.wallet.is_mine(output_script.to_owned())) + }) + .save(persister)?; self.commit_outputs( next_receiver_typestate, @@ -473,6 +518,7 @@ impl<'a> PayjoinManager<'a> { let candidate_inputs: Vec = self .wallet .list_unspent() + .filter(|output| output.chain_position.is_confirmed()) .map(|output| { let psbtin = self .wallet @@ -512,9 +558,9 @@ impl<'a> PayjoinManager<'a> { max_fee_rate: FeeRate, blockchain_client: &BlockchainClient, ) -> Result<(), Error> { - let next_receiver_typestate = receiver.apply_fee_range(None, Some(max_fee_rate)).save(persister).map_err(|e| { - Error::Generic(format!("Error occurred when saving after applying the receiver fee range to the transaction: {e}")) - })?; + let next_receiver_typestate = receiver + .apply_fee_range(None, Some(max_fee_rate)) + .save(persister)?; self.finalize_proposal(next_receiver_typestate, persister, blockchain_client) .await } @@ -542,12 +588,7 @@ impl<'a> PayjoinManager<'a> { Ok(psbt_clone) }) - .save(persister) - .map_err(|e| { - Error::Generic(format!( - "Error occurred when saving after signing the Payjoin proposal: {e}" - )) - })?; + .save(persister)?; self.send_payjoin_proposal(next_receiver_typestate, persister, blockchain_client) .await @@ -559,22 +600,13 @@ impl<'a> PayjoinManager<'a> { persister: &impl SessionPersister, blockchain_client: &BlockchainClient, ) -> Result<(), Error> { - let (req, ctx) = receiver.create_post_request( - self.relay_manager - .lock() - .expect("Lock should not be poisoned") - .get_selected_relay() - .expect("A relay should already be selected") - .as_str(), - ).map_err(|e| { - Error::Generic(format!("Error occurred when creating a post request for sending final Payjoin proposal: {e}")) - })?; - - let res = self.send_payjoin_post_request(req).await?; + let (res, ctx) = self + .post_via_relay(|relay| receiver.create_post_request(relay)) + .await?; let payjoin_psbt = receiver.psbt().clone(); - let next_receiver_typestate = receiver.process_response(&res.bytes().await?, ctx).save(persister).map_err(|e| { - Error::Generic(format!("Error occurred when saving after processing the response to the Payjoin proposal send: {e}")) - })?; + let next_receiver_typestate = receiver + .process_response(&res.bytes().await?, ctx) + .save(persister)?; println!( "Response successful. TXID: {}", payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() @@ -588,8 +620,8 @@ impl<'a> PayjoinManager<'a> { /// sender. /// /// This function syncs the wallet at regular intervals and checks for the Payjoin transaction - /// in a loop until it is detected or a timeout is reached. Since [`sync_wallet`] now accepts - /// a reference to the BlockchainClient, we can call it multiple times in a loop. + /// in a loop until it is detected or a timeout is reached. Since [`BlockchainClient::sync_wallet`] + /// accepts a reference to the BlockchainClient, we can call it multiple times in a loop. async fn monitor_payjoin_proposal( &mut self, receiver: Receiver, @@ -628,19 +660,9 @@ impl<'a> PayjoinManager<'a> { return Ok(Some(tx_details.tx.as_ref().clone())); } Err(ImplementationError::from("Cannot find the transaction in the mempool or the blockchain")) - }, - |outpoint| { - let utxo = self.wallet.get_utxo(outpoint); - match utxo { - Some(_) => Ok(false), - None => Ok(true), - } } ) - .save(persister) - .map_err(|e| { - Error::Generic(format!("Error occurred when saving after checking that sender has broadcasted the Payjoin transaction: {e}")) - }); + .save(persister); if let Ok(OptionalTransitionOutcome::Progress(_)) = check_result { println!("Payjoin transaction detected in the mempool!"); @@ -671,31 +693,9 @@ impl<'a> PayjoinManager<'a> { receiver: Receiver, persister: &impl SessionPersister, ) -> Result<(), Error> { - let (err_req, err_ctx) = receiver - .create_error_request( - self.relay_manager - .lock() - .expect("Lock should not be poisoned") - .get_selected_relay() - .expect("A relay should already be selected") - .as_str(), - ) - .map_err(|e| { - Error::Generic(format!( - "Error occurred when creating a receiver error request: {}", - e - )) - })?; - - let err_response = match self.send_payjoin_post_request(err_req).await { - Ok(response) => response, - Err(e) => { - return Err(Error::Generic(format!( - "Failed to post error request: {}", - e - ))); - } - }; + let (err_response, err_ctx) = self + .post_via_relay(|relay| receiver.create_error_request(relay)) + .await?; let err_bytes = match err_response.bytes().await { Ok(bytes) => bytes, @@ -724,16 +724,15 @@ impl<'a> PayjoinManager<'a> { &self, session: SendSession, persister: &impl SessionPersister, - relay: impl payjoin::IntoUrl, blockchain_client: &BlockchainClient, ) -> Result { match session { SendSession::WithReplyKey(context) => { - self.post_original_proposal(context, relay, persister, blockchain_client) + self.post_original_proposal(context, persister, blockchain_client) .await } SendSession::PollingForProposal(context) => { - self.get_proposed_payjoin_proposal(context, relay, persister, blockchain_client) + self.get_proposed_payjoin_proposal(context, persister, blockchain_client) .await } SendSession::Closed(SenderSessionOutcome::Success(psbt)) => { @@ -746,33 +745,30 @@ impl<'a> PayjoinManager<'a> { async fn post_original_proposal( &self, sender: Sender, - relay: impl payjoin::IntoUrl, persister: &impl SessionPersister, blockchain_client: &BlockchainClient, ) -> Result { - let (req, ctx) = sender.create_v2_post_request(relay.as_str())?; - let response = self.send_payjoin_post_request(req).await?; + let (response, ctx) = self + .post_via_relay(|relay| sender.create_v2_post_request(relay)) + .await?; let sender = sender .process_response(&response.bytes().await?, ctx) - .save(persister) - .map_err(|e| { - Error::Generic(format!("Failed to persist the Payjoin send after successfully sending original proposal: {e}")) - })?; - self.get_proposed_payjoin_proposal(sender, relay, persister, blockchain_client) + .save(persister)?; + self.get_proposed_payjoin_proposal(sender, persister, blockchain_client) .await } async fn get_proposed_payjoin_proposal( &self, sender: Sender, - relay: impl payjoin::IntoUrl, persister: &impl SessionPersister, blockchain_client: &BlockchainClient, ) -> Result { let mut sender = sender.clone(); loop { - let (req, ctx) = sender.create_poll_request(relay.as_str())?; - let response = self.send_payjoin_post_request(req).await?; + let (response, ctx) = self + .post_via_relay(|relay| sender.create_poll_request(relay)) + .await?; let processed_response = sender .process_response(&response.bytes().await?, ctx) .save(persister); @@ -821,6 +817,268 @@ impl<'a> PayjoinManager<'a> { .header("Content-Type", req.content_type) .body(req.body) .send() - .await + .await? + .error_for_status() + } + + async fn post_via_relay(&self, mut build: F) -> Result<(reqwest::Response, T), Error> + where + F: FnMut(&str) -> Result<(payjoin::Request, T), E>, + E: std::fmt::Display, + { + loop { + let relay = self.relay_manager.choose_relay()?; + // Build a fresh request for each attempt. Reusing an OHTTP + // ciphertext would let relays correlate retransmissions. + let (req, context) = build(relay.as_str()) + .map_err(|e| Error::Generic(format!("Failed to create OHTTP request: {e}")))?; + + match self.send_payjoin_post_request(req).await { + Ok(response) => return Ok((response, context)), + Err(e) => { + tracing::debug!("Request to OHTTP relay {relay} failed: {e:?}"); + self.relay_manager.add_failed_relay(relay); + } + } + } + } + + /// Resume pending payjoin sessions from the database + pub async fn resume_payjoins( + &mut self, + _directory: String, + ohttp_relays: Vec, + session_id: Option, + blockchain_client: &BlockchainClient, + ) -> Result { + let db = self.db.clone(); + let mut recv_session_ids = db.get_recv_session_ids()?; + let mut send_session_ids = db.get_send_session_ids()?; + + if let Some(session_id) = session_id { + recv_session_ids.retain(|id| id.as_i64() == session_id); + send_session_ids.retain(|id| id.as_i64() == session_id); + + if recv_session_ids.is_empty() && send_session_ids.is_empty() { + return Ok(serde_json::to_string_pretty(&json!({ + "message": format!("No active session found for session_id {}.", session_id) + }))?); + } + } + + if recv_session_ids.is_empty() && send_session_ids.is_empty() { + return Ok(serde_json::to_string_pretty(&json!({ + "message": "No sessions to resume." + }))?); + } + + let ohttp_relays: Vec = ohttp_relays + .into_iter() + .map(|s| url::Url::parse(&s)) + .collect::>() + .map_err(|e| Error::Generic(format!("Failed to parse OHTTP URLs: {e}")))?; + self.relay_manager.configure(ohttp_relays)?; + + let max_fee_rate = FeeRate::BROADCAST_MIN; + let total_sessions = recv_session_ids.len() + send_session_ids.len(); + let mut completed = 0usize; + let mut timed_out = 0usize; + let mut failed = 0usize; + + println!("Resuming {} payjoin session(s)...\n", total_sessions); + + // Resume receiver sessions + for session_id in recv_session_ids { + let persister = ReceiverPersister::from_id(db.clone(), session_id.clone()); + match replay_receiver_event_log(&persister) { + Ok((receiver_state, _)) => { + println!("Resuming receiver session {}", session_id); + match tokio::time::timeout( + std::time::Duration::from_secs(30), + self.proceed_receiver_session( + receiver_state, + &persister, + max_fee_rate, + blockchain_client, + ), + ) + .await + { + Ok(Ok(_)) => { + completed += 1; + } + Ok(Err(e)) => { + failed += 1; + println!("Receiver session {} failed: {}", session_id, e); + } + Err(_) => { + timed_out += 1; + println!("Receiver session {} timed out", session_id); + } + } + } + Err(e) => { + failed += 1; + println!("Failed to replay receiver session {}: {:?}", session_id, e); + } + } + } + + // Resume sender sessions + for session_id in send_session_ids { + let persister = SenderPersister::from_id(db.clone(), session_id.clone()); + match replay_sender_event_log(&persister) { + Ok((sender_state, _)) => { + println!("Resuming sender session {}", session_id); + match tokio::time::timeout( + std::time::Duration::from_secs(30), + self.proceed_sender_session(sender_state, &persister, blockchain_client), + ) + .await + { + Ok(Ok(_)) => { + completed += 1; + } + Ok(Err(e)) => { + failed += 1; + println!("Sender session {} failed: {}", session_id, e); + } + Err(_) => { + timed_out += 1; + println!("Sender session {} timed out", session_id); + } + } + } + Err(e) => { + failed += 1; + println!("Failed to replay sender session {}: {:?}", session_id, e); + } + } + } + + Ok(serde_json::to_string_pretty(&json!({ + "message": format!("Resumed polling for {} session(s).", total_sessions), + "outcome": format!( + "Completed: {}, timed out: {}, failed: {}.", + completed, timed_out, failed + ) + }))?) + } + + /// Show payjoin session history + pub fn history( + datadir: Option, + wallet_name: &str, + ) -> Result { + let db = open_payjoin_db(datadir, wallet_name)?; + let mut send_rows: Vec = Vec::new(); + let mut recv_rows: Vec = Vec::new(); + + // Active send sessions + for session_id in db + .get_send_session_ids() + .map_err(|e| Error::Generic(format!("{e}")))? + { + let persister = SenderPersister::from_id(db.clone(), session_id.clone()); + let status = match replay_sender_event_log(&persister) { + Ok((state, _)) => state.status_text().to_string(), + Err(e) => e.to_string(), + }; + send_rows.push(SessionHistoryRow { + id: session_id.to_string(), + role: "Sender", + status, + completed_at: None, + }); + } + + // Active receive sessions + for session_id in db + .get_recv_session_ids() + .map_err(|e| Error::Generic(format!("{e}")))? + { + let persister = ReceiverPersister::from_id(db.clone(), session_id.clone()); + let status = match replay_receiver_event_log(&persister) { + Ok((state, _)) => state.status_text().to_string(), + Err(e) => e.to_string(), + }; + recv_rows.push(SessionHistoryRow { + id: session_id.to_string(), + role: "Receiver", + status, + completed_at: None, + }); + } + + // Completed send sessions + for (session_id, completed_at) in db + .get_inactive_send_session_ids() + .map_err(|e| Error::Generic(format!("{e}")))? + { + let persister = SenderPersister::from_id(db.clone(), session_id.clone()); + let status = match replay_sender_event_log(&persister) { + Ok((state, _)) => state.status_text().to_string(), + Err(e) => e.to_string(), + }; + let completed_at = db + .format_unix_timestamp(completed_at) + .map_err(|e| Error::Generic(format!("{e}")))?; + send_rows.push(SessionHistoryRow { + id: session_id.to_string(), + role: "Sender", + status, + completed_at: Some(completed_at), + }); + } + + // Completed receive sessions + for (session_id, completed_at) in db + .get_inactive_recv_session_ids() + .map_err(|e| Error::Generic(format!("{e}")))? + { + let persister = ReceiverPersister::from_id(db.clone(), session_id.clone()); + let status = match replay_receiver_event_log(&persister) { + Ok((state, _)) => state.status_text().to_string(), + Err(e) => e.to_string(), + }; + let completed_at = db + .format_unix_timestamp(completed_at) + .map_err(|e| Error::Generic(format!("{e}")))?; + recv_rows.push(SessionHistoryRow { + id: session_id.to_string(), + role: "Receiver", + status, + completed_at: Some(completed_at), + }); + } + + let rows: Vec> = send_rows + .iter() + .chain(recv_rows.iter()) + .map(|row| { + vec![ + row.id.as_str().cell(), + row.role.cell(), + row.completed_at + .clone() + .unwrap_or_else(|| "Not Completed".to_string()) + .cell(), + row.status.as_str().cell(), + ] + }) + .collect(); + + let table = rows + .table() + .title(vec![ + "Session ID".cell().bold(true), + "Sender/Receiver".cell().bold(true), + "Completed At".cell().bold(true), + "Status".cell().bold(true), + ]) + .display() + .map_err(|e| Error::Generic(e.to_string()))?; + + Ok(format!("{table}")) } } diff --git a/src/payjoin/ohttp.rs b/src/payjoin/ohttp.rs index 1cc0935..bc26466 100644 --- a/src/payjoin/ohttp.rs +++ b/src/payjoin/ohttp.rs @@ -3,108 +3,125 @@ use std::sync::{Arc, Mutex}; #[derive(Debug, Clone)] pub(crate) struct RelayManager { - selected_relay: Option, - failed_relays: Vec, + relays: Vec, + failed_relays: Arc>>, } impl RelayManager { - pub fn new() -> Self { - RelayManager { - selected_relay: None, - failed_relays: Vec::new(), + pub(crate) fn new() -> Self { + Self { + relays: Vec::new(), + failed_relays: Arc::new(Mutex::new(Vec::new())), } } - pub fn set_selected_relay(&mut self, relay: url::Url) { - self.selected_relay = Some(relay); - } + pub(crate) fn configure(&mut self, relays: Vec) -> Result<(), Error> { + if relays.is_empty() { + return Err(Error::Generic( + "At least one valid OHTTP relay must be provided.".into(), + )); + } - pub fn get_selected_relay(&self) -> Option { - self.selected_relay.clone() + self.relays = relays; + self.failed_relays + .lock() + .expect("Lock should not be poisoned") + .clear(); + Ok(()) } - pub fn add_failed_relay(&mut self, relay: url::Url) { - self.failed_relays.push(relay); + pub(crate) fn add_failed_relay(&self, relay: url::Url) { + let mut failed_relays = self + .failed_relays + .lock() + .expect("Lock should not be poisoned"); + if !failed_relays.contains(&relay) { + failed_relays.push(relay); + } } - pub fn get_failed_relays(&self) -> Vec { - self.failed_relays.clone() - } -} + pub(crate) fn choose_relay(&self) -> Result { + use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; -pub(crate) struct ValidatedOhttpKeys { - pub(crate) ohttp_keys: payjoin::OhttpKeys, - pub(crate) relay_url: url::Url, -} - -pub(crate) async fn fetch_ohttp_keys( - relays: Vec, - payjoin_directory: impl payjoin::IntoUrl, - relay_manager: Arc>, -) -> Result { - use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; - - loop { - let failed_relays = relay_manager + let failed_relays = self + .failed_relays .lock() - .expect("Lock should not be poisoned") - .get_failed_relays(); - - let remaining_relays: Vec<_> = relays + .expect("Lock should not be poisoned"); + let remaining_relays: Vec<_> = self + .relays .iter() - .filter(|r| !failed_relays.contains(r)) + .filter(|relay| !failed_relays.contains(relay)) .cloned() .collect(); - if remaining_relays.is_empty() { - return Err(Error::Generic( - "No valid OHTTP relays available".to_string(), - )); - } + remaining_relays + .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) + .cloned() + .ok_or_else(|| Error::Generic("No valid OHTTP relays available".to_string())) + } - let selected_relay = - match remaining_relays.choose(&mut payjoin::bitcoin::key::rand::thread_rng()) { - Some(relay) => relay.clone(), - None => { - return Err(Error::Generic( - "Failed to select from remaining relays".to_string(), - )); - } - }; + pub(crate) async fn fetch_ohttp_keys( + &self, + payjoin_directory: impl payjoin::IntoUrl, + ) -> Result { + loop { + let selected_relay = self.choose_relay()?; + let ohttp_keys = + payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), payjoin_directory.as_str()) + .await; - relay_manager - .lock() - .expect("Lock should not be poisoned") - .set_selected_relay(selected_relay.clone()); - - let ohttp_keys = - payjoin::io::fetch_ohttp_keys(selected_relay.as_str(), payjoin_directory.as_str()) - .await; - - match ohttp_keys { - Ok(keys) => { - return Ok(ValidatedOhttpKeys { - ohttp_keys: keys, - relay_url: selected_relay, - }); - } - Err(payjoin::io::Error::UnexpectedStatusCode(e)) => { - return Err(Error::Generic(format!( - "Unexpected error occurred when fetching OHTTP keys: {}", - e - ))); - } - Err(e) => { - tracing::debug!( - "Failed to connect to OHTTP relay: {}, {}", - selected_relay, - e - ); - relay_manager - .lock() - .expect("Lock should not be poisoned") - .add_failed_relay(selected_relay); + match ohttp_keys { + Ok(keys) => return Ok(keys), + Err(payjoin::io::Error::UnexpectedStatusCode(e)) => { + return Err(Error::Generic(format!( + "Unexpected error occurred when fetching OHTTP keys: {e}" + ))); + } + Err(e) => { + tracing::debug!( + "Failed to connect to OHTTP relay: {}, {}", + selected_relay, + e + ); + self.add_failed_relay(selected_relay); + } } } } } + +#[cfg(test)] +mod tests { + use super::RelayManager; + + fn relay(url: &str) -> url::Url { + url::Url::parse(url).expect("valid relay URL") + } + + #[test] + fn choose_relay_excludes_failed_relays() { + let mut manager = RelayManager::new(); + let failed = relay("https://failed.example"); + let available = relay("https://available.example"); + manager + .configure(vec![failed.clone(), available.clone()]) + .expect("relay configuration"); + + manager.add_failed_relay(failed); + + assert_eq!(manager.choose_relay().expect("available relay"), available); + } + + #[test] + fn choose_relay_fails_when_all_relays_failed() { + let mut manager = RelayManager::new(); + let failed = relay("https://failed.example"); + manager + .configure(vec![failed.clone()]) + .expect("relay configuration"); + + manager.add_failed_relay(failed); + + assert!(manager.choose_relay().is_err()); + } +}