"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"
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"
"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"
"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"
"libc",
"log",
"openssl",
- "openssl-probe",
+ "openssl-probe 0.1.6",
"openssl-sys",
"schannel",
- "security-framework",
+ "security-framework 2.11.1",
"security-framework-sys",
"tempfile",
]
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"
[[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",
"serde_json",
"tracing",
"url",
+ "web-time",
]
[[package]]
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
dependencies = [
+ "aws-lc-rs",
"bytes",
"getrandom 0.3.4",
"lru-slab",
"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",
"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"
"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"
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"
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",
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"
"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"
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"
"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"
"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"
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"
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"
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"
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"
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"
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"
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 }
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
use crate::{
client::ClientType,
handlers::online::{
- BroadcastCommand, FullScanCommand, ReceivePayjoinCommand, SendPayjoinCommand, SyncCommand,
+ BroadcastCommand, FullScanCommand, PayjoinHistoryCommand, ReceivePayjoinCommand,
+ ResumePayjoinCommand, SendPayjoinCommand, SyncCommand,
},
};
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.
#[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),
BDKCliError::RusqliteError(Box::new(err))
}
}
+
+#[cfg(feature = "payjoin")]
+impl<ApiErr, StorageErr, ErrorState>
+ From<payjoin::persist::PersistedError<ApiErr, StorageErr, ErrorState>> for BDKCliError
+where
+ ApiErr: std::error::Error,
+ StorageErr: std::error::Error,
+ ErrorState: std::fmt::Debug,
+{
+ fn from(e: payjoin::persist::PersistedError<ApiErr, StorageErr, ErrorState>) -> Self {
+ BDKCliError::Generic(e.to_string())
+ }
+}
pub struct OnlineOperations<'a> {
pub wallet: &'a mut Wallet,
pub client: &'a BlockchainClient,
+ pub wallet_name: String,
}
/// The generic context
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,
+ },
}
}
}
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",
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())
+ }
}
}
}
&self,
ctx: &mut AppContext<OnlineOperations<'_>>,
) -> Result<Self::Output, Error> {
+ 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,
&self,
ctx: &mut AppContext<OnlineOperations<'_>>,
) -> Result<Self::Output, Error> {
+ 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(),
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<String>,
+ /// Resume only a specific active session ID (sender and/or receiver).
+ #[arg(env = "PAYJOIN_SESSION_ID", long = "session_id")]
+ session_id: Option<i64>,
+}
+
+#[cfg(any(
+ feature = "electrum",
+ feature = "esplora",
+ feature = "cbf",
+ feature = "rpc"
+))]
+impl AsyncAppCommand<AppContext<OnlineOperations<'_>>> for ResumePayjoinCommand {
+ type Output = StatusResult;
+
+ async fn execute(
+ &self,
+ ctx: &mut AppContext<OnlineOperations<'_>>,
+ ) -> Result<Self::Output, Error> {
+ 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<AppContext<OnlineOperations<'_>>> for PayjoinHistoryCommand {
+ type Output = StatusResult;
+
+ async fn execute(
+ &self,
+ ctx: &mut AppContext<OnlineOperations<'_>>,
+ ) -> Result<Self::Output, Error> {
+ 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 })
+ }
+}
client: Option<&BlockchainClient>,
line: &str,
datadir: std::path::PathBuf,
+ #[cfg(any(
+ feature = "electrum",
+ feature = "esplora",
+ feature = "rpc",
+ feature = "cbf"
+ ))]
+ wallet_name: &str,
) -> Result<bool, String> {
let args = shlex::split(line).ok_or("error: Invalid quoting".to_string())?;
))]
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
runtime.home_dir.clone(),
&mut wallet,
&client,
+ runtime.wallet_name.clone(),
);
cmd.execute(&mut ctx).await?;
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)?;
--- /dev/null
+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<T> = std::result::Result<T, Error>;
+
+/// 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<bdk_wallet::rusqlite::Error> for Error {
+ fn from(error: bdk_wallet::rusqlite::Error) -> Self {
+ Error::Rusqlite(error)
+ }
+}
+
+impl From<Error> 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<PathBuf>,
+ wallet_name: &str,
+) -> std::result::Result<Arc<Database>, 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<Connection>,
+}
+
+impl Database {
+ pub fn create(path: impl AsRef<Path>) -> Result<Self> {
+ 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<bool> {
+ 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<Vec<SessionId>> {
+ 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<Vec<SessionId>> {
+ 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<HpkePublicKey> {
+ let conn = self.conn();
+ let mut stmt =
+ conn.prepare("SELECT receiver_pubkey FROM send_sessions WHERE session_id = ?1")?;
+ let receiver_pubkey: Vec<u8> = 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<Vec<(SessionId, u64)>> {
+ 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<Vec<(SessionId, u64)>> {
+ 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<String> {
+ let Ok(timestamp) = i64::try_from(timestamp) else {
+ return Ok(format!("Invalid timestamp ({timestamp})"));
+ };
+ let conn = self.conn();
+ let dt: Option<String> = 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<ToSqlOutput<'_>> {
+ 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<Database>,
+ session_id: SessionId,
+}
+
+impl SenderPersister {
+ /// Creates a new sender persister, creating a new session in the database
+ pub fn new(db: Arc<Database>, receiver_pubkey: HpkePublicKey) -> Result<Self> {
+ 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<Database>, 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<Box<dyn Iterator<Item = SenderSessionEvent>>, 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<SenderSessionEvent> = event_rows
+ .map(|row| {
+ let event_data = row.expect("Failed to read event data from database");
+ serde_json::from_str::<SenderSessionEvent>(&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<Database>,
+ session_id: SessionId,
+}
+
+impl ReceiverPersister {
+ /// Creates a new receiver persister, creating a new session in the database
+ pub fn new(db: Arc<Database>) -> Result<Self> {
+ 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<Database>, 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<dyn Iterator<Item = ReceiverSessionEvent>>,
+ 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<ReceiverSessionEvent> = event_rows
+ .map(|row| {
+ let event_data = row.expect("Failed to read event data from database");
+ serde_json::from_str::<ReceiverSessionEvent>(&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<i64> = 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<i64> = 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);
+ }
+}
-#![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<Mutex<RelayManager>>,
+ relay_manager: RelayManager,
+ db: Arc<crate::payjoin::db::Database>,
+}
+
+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<String>,
}
impl<'a> PayjoinManager<'a> {
- pub fn new(wallet: &'a mut Wallet, relay_manager: Arc<Mutex<RelayManager>>) -> Self {
- Self {
+ pub fn new(
+ wallet: &'a mut Wallet,
+ datadir: Option<PathBuf>,
+ wallet_name: &str,
+ ) -> Result<Self, Error> {
+ let db = open_payjoin_db(datadir, wallet_name)?;
+ let relay_manager = RelayManager::new();
+
+ Ok(Self {
wallet,
relay_manager,
- }
+ db,
+ })
}
pub async fn receive_payjoin(
.collect::<Result<_, _>>()
.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::<ReceiverSessionEvent>::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:");
self.proceed_receiver_session(
ReceiveSession::Initialized(receiver.clone()),
&persister,
- ohttp_keys.relay_url.to_string(),
checked_max_fee_rate,
blockchain_client,
)
Ok(to_string_pretty(&json!({}))?)
}
- #[cfg(any(
- feature = "electrum",
- feature = "esplora",
- feature = "rpc",
- feature = "cbf"
- ))]
pub async fn send_payjoin(
&mut self,
uri: String,
self.process_payjoin_proposal(psbt, blockchain_client)
.await?
}
- payjoin::PjParam::V2(_) => {
+ payjoin::PjParam::V2(v2_param) => {
let ohttp_relays: Vec<url::Url> = ohttp_relays
.into_iter()
.map(|s| url::Url::parse(&s))
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::<SenderSessionEvent>::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.");
&mut self,
session: ReceiveSession,
persister: &impl SessionPersister<SessionEvent = ReceiverSessionEvent>,
- 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)
&mut self,
receiver: Receiver<Initialized>,
persister: &impl SessionPersister<SessionEvent = ReceiverSessionEvent>,
- 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);
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,
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,
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,
let candidate_inputs: Vec<InputPair> = self
.wallet
.list_unspent()
+ .filter(|output| output.chain_position.is_confirmed())
.map(|output| {
let psbtin = self
.wallet
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
}
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
persister: &impl SessionPersister<SessionEvent = ReceiverSessionEvent>,
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()
/// 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<Monitor>,
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!");
receiver: Receiver<HasReplyableError>,
persister: &impl SessionPersister<SessionEvent = ReceiverSessionEvent>,
) -> 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,
&self,
session: SendSession,
persister: &impl SessionPersister<SessionEvent = SenderSessionEvent>,
- relay: impl payjoin::IntoUrl,
blockchain_client: &BlockchainClient,
) -> Result<Txid, Error> {
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)) => {
async fn post_original_proposal(
&self,
sender: Sender<WithReplyKey>,
- relay: impl payjoin::IntoUrl,
persister: &impl SessionPersister<SessionEvent = SenderSessionEvent>,
blockchain_client: &BlockchainClient,
) -> Result<Txid, Error> {
- 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<PollingForProposal>,
- relay: impl payjoin::IntoUrl,
persister: &impl SessionPersister<SessionEvent = SenderSessionEvent>,
blockchain_client: &BlockchainClient,
) -> Result<Txid, Error> {
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);
.header("Content-Type", req.content_type)
.body(req.body)
.send()
- .await
+ .await?
+ .error_for_status()
+ }
+
+ async fn post_via_relay<F, T, E>(&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<String>,
+ session_id: Option<i64>,
+ blockchain_client: &BlockchainClient,
+ ) -> Result<String, Error> {
+ 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<url::Url> = ohttp_relays
+ .into_iter()
+ .map(|s| url::Url::parse(&s))
+ .collect::<Result<_, _>>()
+ .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<std::path::PathBuf>,
+ wallet_name: &str,
+ ) -> Result<String, Error> {
+ let db = open_payjoin_db(datadir, wallet_name)?;
+ let mut send_rows: Vec<SessionHistoryRow> = Vec::new();
+ let mut recv_rows: Vec<SessionHistoryRow> = 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<Vec<CellStruct>> = 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}"))
}
}
#[derive(Debug, Clone)]
pub(crate) struct RelayManager {
- selected_relay: Option<url::Url>,
- failed_relays: Vec<url::Url>,
+ relays: Vec<url::Url>,
+ failed_relays: Arc<Mutex<Vec<url::Url>>>,
}
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<url::Url>) -> 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<url::Url> {
- 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<url::Url> {
- self.failed_relays.clone()
- }
-}
+ pub(crate) fn choose_relay(&self) -> Result<url::Url, Error> {
+ 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<url::Url>,
- payjoin_directory: impl payjoin::IntoUrl,
- relay_manager: Arc<Mutex<RelayManager>>,
-) -> Result<ValidatedOhttpKeys, Error> {
- 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<payjoin::OhttpKeys, Error> {
+ 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());
+ }
+}
projects / bdk-cli / commitdiff