From: Vihiga Tyonum Date: Wed, 8 Jul 2026 15:22:46 +0000 (+0100) Subject: feat(handlers): mv feats into subdirs in handlers X-Git-Url: http://internal-gitweb-vhost/blockdata/script/encode/-script/display/struct.CommandString.html?a=commitdiff_plain;h=5da76608fcd5e0ae340010f19a778750a68a86e1;p=bdk-cli feat(handlers): mv feats into subdirs in handlers --- diff --git a/src/dns_payment_instructions.rs b/src/dns_payment_instructions.rs deleted file mode 100644 index fc7f616..0000000 --- a/src/dns_payment_instructions.rs +++ /dev/null @@ -1,208 +0,0 @@ -use bdk_wallet::bitcoin::{Address, Amount, Network}; -use bitcoin_payment_instructions::{ - FixedAmountPaymentInstructions, ParseError, PaymentInstructions, PaymentMethod, - PaymentMethodType, amount, dns_resolver::DNSHrnResolver, -}; -use core::{net::SocketAddr, str::FromStr}; -use tokio::time::timeout; - -use crate::error::BDKCliError as Error; - -#[derive(Debug)] -pub struct ResolvedPaymentInfo { - pub hrn: String, - pub payment_methods: Vec, - pub description: Option, - pub min_amount: Option, - pub max_amount: Option, - pub notes: String, -} - -impl ResolvedPaymentInfo { - pub fn display(&self) -> Result { - let methods: Vec = self - .payment_methods - .iter() - .map(|pm| match pm { - PaymentMethod::LightningBolt11(bolt11) => { - format!("Bolt 11 invoice ({})", bolt11) - } - PaymentMethod::LightningBolt12(offer) => format!("Bolt 12 invoice ({})", offer), - PaymentMethod::OnChain(address) => format!("On chain ({})", address), - PaymentMethod::Cashu(csh) => format!("Cashu payment ({})", csh), - }) - .collect(); - - Ok(serde_json::to_string_pretty(&serde_json::json!({ - "hrn": self.hrn, - "payment_methods": methods, - "description": self.description, - "min_amount": self.min_amount, - "max_amount": self.max_amount, - "notes": self.notes - }))?) - } -} - -pub(crate) async fn parse_dns_instructions( - hrn: &str, - network: Network, - dns_resolver: &str, -) -> Result<(DNSHrnResolver, PaymentInstructions), ParseError> { - let ip_address = if dns_resolver.contains(':') { - dns_resolver - } else { - &format!("{dns_resolver}:53") - }; - - let sock_addr = SocketAddr::from_str(ip_address).map_err(|_| { - ParseError::HrnResolutionError("Unable to create socket from provided address") - })?; - let resolver = DNSHrnResolver(sock_addr); - let instructions = timeout( - std::time::Duration::from_secs(30), - PaymentInstructions::parse(hrn, network, &resolver, true), - ) - .await - .map_err(|_| ParseError::HrnResolutionError("Resolution request timed out"))??; - Ok((resolver, instructions)) -} - -fn get_onchain_info( - instructions: &FixedAmountPaymentInstructions, -) -> Result<(Address, Amount), Error> { - // Look for on chain payment method as it's the only one we can support - let PaymentMethod::OnChain(addr) = instructions - .methods() - .iter() - .find(|ix| matches!(ix, PaymentMethod::OnChain(_))) - .ok_or(Error::Generic( - "Missing Onchain payment method option.".to_string(), - ))? - else { - return Err(Error::Generic("Unsupported payment method".to_string())); - }; - - let Some(onchain_amount) = instructions.onchain_payment_amount() else { - return Err(Error::Generic( - "On chain amount should be specified".to_string(), - )); - }; - - // We need this conversion since Amount from instructions is different from Amount from bitcoin - Ok((addr.clone(), Amount::from_sat(onchain_amount.milli_sats()))) -} - -pub async fn process_instructions( - amount_to_send: Amount, - payment_instructions: &PaymentInstructions, - resolver: DNSHrnResolver, -) -> Result<(Address, Amount), Error> { - match payment_instructions { - PaymentInstructions::ConfigurableAmount(instructions) => { - // Look for on chain payment method as it's the only one we can support - if !instructions - .methods() - .any(|method| matches!(method.method_type(), PaymentMethodType::OnChain)) - { - return Err(Error::Generic("Unsupported payment method".to_string())); - } - - let min_amount = instructions - .min_amt() - .map(|amnt| Amount::from_sat(amnt.sats_rounding_up())); - - let max_amount = instructions - .max_amt() - .map(|amnt| Amount::from_sat(amnt.sats_rounding_up())); - - if min_amount.is_some_and(|min| amount_to_send < min) { - return Err(Error::Generic( - format!( - "Amount to send should be greater than min {}", - min_amount.unwrap() - ) - .to_string(), - )); - } - - if max_amount.is_some_and(|max| amount_to_send > max) { - return Err(Error::Generic( - format!( - "Amount to send should be lower than max {}", - max_amount.unwrap() - ) - .to_string(), - )); - } - - let fixed_instructions = instructions - .clone() - .set_amount( - amount::Amount::from_sats(amount_to_send.to_sat()).unwrap(), - &resolver, - ) - .await - .map_err(|err| { - Error::Generic(format!("Error occured while parsing instructions {err}")) - })?; - - let onchain_details = get_onchain_info(&fixed_instructions)?; - - Ok((onchain_details.0.clone(), onchain_details.1)) - } - - PaymentInstructions::FixedAmount(instructions) => Ok(get_onchain_info(instructions)?), - } -} - -/// Resolves the dns payment instructions found at the specified Human Readable Name -pub async fn resolve_dns_recipient( - hrn: &str, - network: Network, - dns_resolver: &str, -) -> Result { - let (resolver, instructions) = parse_dns_instructions(hrn, network, dns_resolver).await?; - - match instructions { - PaymentInstructions::ConfigurableAmount(ix) => { - let description = ix.recipient_description().map(|s| s.to_string()); - let min_amount = ix.min_amt().map(|amnt| Amount::from_sat(amnt.milli_sats())); - let max_amount = ix.max_amt().map(|amnt| Amount::from_sat(amnt.milli_sats())); - - // Let's set a dummy amount to resolve the payment methods accepted. - let fixed_instructions = ix - .set_amount(amount::Amount::ZERO, &resolver) - .await - .map_err(ParseError::InvalidInstructions)?; - - let payment = ResolvedPaymentInfo { - min_amount, - max_amount, - payment_methods: fixed_instructions.methods().into(), - description, - hrn: hrn.to_string(), - notes: "This is configurable payment instructions. You must send an amount between min_amount and max_amount if set.".to_string(), - }; - - Ok(payment) - } - - PaymentInstructions::FixedAmount(ix) => { - let max_amount = ix - .max_amount() - .map(|amnt| Amount::from_sat(amnt.milli_sats())); - - let payment = ResolvedPaymentInfo { - min_amount: None, - max_amount, - payment_methods: ix.methods().into(), - description: ix.recipient_description().map(|s| s.to_string()), - hrn: hrn.to_string(), - notes: "This is a fixed payment instructions. You must send exactly the amount specified in max_amount.".to_string(), - }; - - Ok(payment) - } - } -} diff --git a/src/error.rs b/src/error.rs index dbdc7b8..7f7eab0 100644 --- a/src/error.rs +++ b/src/error.rs @@ -154,7 +154,7 @@ pub enum BDKCliError { #[cfg(feature = "payjoin")] #[error("Payjoin database error: {0}")] - PayjoinDb(#[from] crate::payjoin::db::Error), + PayjoinDb(#[from] crate::handlers::payjoin::db::Error), #[cfg(feature = "bip322")] #[error("BIP-322 error: {0}")] diff --git a/src/handlers/dns.rs b/src/handlers/dns.rs deleted file mode 100644 index 18e127b..0000000 --- a/src/handlers/dns.rs +++ /dev/null @@ -1,173 +0,0 @@ -use crate::dns_payment_instructions::{ - parse_dns_instructions, process_instructions, resolve_dns_recipient, -}; -use crate::error::BDKCliError as Error; -use crate::handlers::{AppContext, AsyncAppCommand, Init, OfflineOperations}; -use crate::utils::types::{PsbtResult, StatusResult}; -use crate::utils::{parse_dns_recipient, parse_outpoint, parse_recipient}; -use bdk_wallet::KeychainKind; -use bdk_wallet::bitcoin::base64::Engine; -use bdk_wallet::bitcoin::base64::prelude::BASE64_STANDARD; -use bdk_wallet::bitcoin::script::PushBytesBuf; -use bdk_wallet::bitcoin::{Amount, FeeRate, OutPoint, ScriptBuf, Sequence}; -use clap::Parser; -use std::collections::BTreeMap; - -/// Resolves BIP-353 DNS payment instructions for a human-readable name. -#[derive(Parser, Debug, Clone, PartialEq)] -pub struct ResolveDnsRecipientCommand { - /// Human-readable name (e.g. user@domain.com) - pub hrn: String, - /// DNS resolver address - #[arg(long, default_value = "8.8.8.8")] - pub resolver: String, -} - -impl AsyncAppCommand> for ResolveDnsRecipientCommand { - type Output = StatusResult; - - async fn execute(&self, ctx: &mut AppContext) -> Result { - let resolved = resolve_dns_recipient(&self.hrn, ctx.network, &self.resolver) - .await - .map_err(|e| Error::Generic(format!("{:?}", e)))?; - Ok(StatusResult { - message: resolved.display()?, - }) - } -} - -/// Creates a new unsigned transaction from DNS payment instructions. -#[derive(Parser, Debug, Clone, PartialEq)] -pub struct CreateDnsTxCommand { - #[arg(env = "ADDRESS:SAT", long = "to", value_parser = parse_recipient)] - pub recipients: Vec<(ScriptBuf, u64)>, - #[arg(long = "to_dns", value_parser = parse_dns_recipient)] - pub dns_recipients: Vec<(String, u64)>, - #[arg(long = "dns_resolver", default_value = "8.8.8.8")] - pub dns_resolver: String, - #[arg(long = "send_all", short = 'a')] - pub send_all: bool, - #[arg(long = "enable_rbf", short = 'r', default_value_t = true)] - pub enable_rbf: bool, - #[arg(long = "offline_signer")] - pub offline_signer: bool, - #[arg(env = "MUST_SPEND_TXID:VOUT", long = "utxos", value_parser = parse_outpoint)] - pub utxos: Option>, - #[arg(env = "CANT_SPEND_TXID:VOUT", long = "unspendable", value_parser = parse_outpoint)] - pub unspendable: Option>, - #[arg(env = "SATS_VBYTE", short = 'f', long = "fee_rate")] - pub fee_rate: Option, - #[arg(env = "EXT_POLICY", long = "external_policy")] - pub external_policy: Option, - #[arg(env = "INT_POLICY", long = "internal_policy")] - pub internal_policy: Option, - #[arg( - env = "ADD_STRING", - long = "add_string", - short = 's', - conflicts_with = "add_data" - )] - pub add_string: Option, - #[arg( - env = "ADD_DATA", - long = "add_data", - short = 'o', - conflicts_with = "add_string" - )] - pub add_data: Option, -} - -impl AsyncAppCommand>> for CreateDnsTxCommand { - type Output = PsbtResult; - - async fn execute( - &self, - ctx: &mut AppContext>, - ) -> Result { - let network = ctx.network; - let mut recipients: Vec<(ScriptBuf, u64)> = self.recipients.clone(); - - for (hrn, amount_sat) in &self.dns_recipients { - log::info!("Resolving DNS instructions for recipient {hrn}"); - let amount = Amount::from_sat(*amount_sat); - let (resolver, instructions) = parse_dns_instructions(hrn, network, &self.dns_resolver) - .await - .map_err(|e| Error::Generic(format!("Parsing error occured {e:#?}")))?; - let payment = process_instructions(amount, &instructions, resolver).await?; - recipients.push((payment.0.into(), payment.1.to_sat())); - } - - if recipients.is_empty() { - return Err(Error::Generic( - "Either --to or --to_dns parameters must be specified".to_string(), - )); - } - - let mut tx_builder = ctx.state.wallet.build_tx(); - - if self.send_all { - if recipients.len() == 1 { - tx_builder.drain_wallet().drain_to(recipients[0].0.clone()); - } else { - return Err(Error::Generic( - "Wallet can only be drained to a single output".to_string(), - )); - } - } else { - let recipients = recipients - .into_iter() - .map(|(script, amount)| (script, Amount::from_sat(amount))) - .collect(); - tx_builder.set_recipients(recipients); - } - - if !self.enable_rbf { - tx_builder.set_exact_sequence(Sequence::MAX); - } - if self.offline_signer { - tx_builder.include_output_redeem_witness_script(); - } - if let Some(fee_rate) = self.fee_rate - && let Some(fee_rate) = FeeRate::from_sat_per_vb(fee_rate as u64) - { - tx_builder.fee_rate(fee_rate); - } - if let Some(utxos) = &self.utxos { - tx_builder - .add_utxos(&utxos[..]) - .map_err(|_| bdk_wallet::error::CreateTxError::UnknownUtxo)?; - } - if let Some(unspendable) = &self.unspendable { - tx_builder.unspendable(unspendable.to_vec()); - } - if let Some(base64_data) = &self.add_data { - let op_return_data = BASE64_STANDARD - .decode(base64_data) - .map_err(|e| Error::Generic(e.to_string()))?; - tx_builder.add_data( - &PushBytesBuf::try_from(op_return_data) - .map_err(|e| Error::Generic(e.to_string()))?, - ); - } else if let Some(string_data) = &self.add_string { - let data = PushBytesBuf::try_from(string_data.as_bytes().to_vec()) - .map_err(|e| Error::Generic(e.to_string()))?; - tx_builder.add_data(&data); - } - - let policies = vec![ - self.external_policy - .as_ref() - .map(|p| (p, KeychainKind::External)), - self.internal_policy - .as_ref() - .map(|p| (p, KeychainKind::Internal)), - ]; - for (policy, keychain) in policies.into_iter().flatten() { - let policy = serde_json::from_str::>>(policy)?; - tx_builder.policy_path(policy, keychain); - } - - let psbt = tx_builder.finish()?; - Ok(PsbtResult::new(&psbt, false, Some(false))) - } -} diff --git a/src/handlers/dns/dns_payment_instructions.rs b/src/handlers/dns/dns_payment_instructions.rs new file mode 100644 index 0000000..fc7f616 --- /dev/null +++ b/src/handlers/dns/dns_payment_instructions.rs @@ -0,0 +1,208 @@ +use bdk_wallet::bitcoin::{Address, Amount, Network}; +use bitcoin_payment_instructions::{ + FixedAmountPaymentInstructions, ParseError, PaymentInstructions, PaymentMethod, + PaymentMethodType, amount, dns_resolver::DNSHrnResolver, +}; +use core::{net::SocketAddr, str::FromStr}; +use tokio::time::timeout; + +use crate::error::BDKCliError as Error; + +#[derive(Debug)] +pub struct ResolvedPaymentInfo { + pub hrn: String, + pub payment_methods: Vec, + pub description: Option, + pub min_amount: Option, + pub max_amount: Option, + pub notes: String, +} + +impl ResolvedPaymentInfo { + pub fn display(&self) -> Result { + let methods: Vec = self + .payment_methods + .iter() + .map(|pm| match pm { + PaymentMethod::LightningBolt11(bolt11) => { + format!("Bolt 11 invoice ({})", bolt11) + } + PaymentMethod::LightningBolt12(offer) => format!("Bolt 12 invoice ({})", offer), + PaymentMethod::OnChain(address) => format!("On chain ({})", address), + PaymentMethod::Cashu(csh) => format!("Cashu payment ({})", csh), + }) + .collect(); + + Ok(serde_json::to_string_pretty(&serde_json::json!({ + "hrn": self.hrn, + "payment_methods": methods, + "description": self.description, + "min_amount": self.min_amount, + "max_amount": self.max_amount, + "notes": self.notes + }))?) + } +} + +pub(crate) async fn parse_dns_instructions( + hrn: &str, + network: Network, + dns_resolver: &str, +) -> Result<(DNSHrnResolver, PaymentInstructions), ParseError> { + let ip_address = if dns_resolver.contains(':') { + dns_resolver + } else { + &format!("{dns_resolver}:53") + }; + + let sock_addr = SocketAddr::from_str(ip_address).map_err(|_| { + ParseError::HrnResolutionError("Unable to create socket from provided address") + })?; + let resolver = DNSHrnResolver(sock_addr); + let instructions = timeout( + std::time::Duration::from_secs(30), + PaymentInstructions::parse(hrn, network, &resolver, true), + ) + .await + .map_err(|_| ParseError::HrnResolutionError("Resolution request timed out"))??; + Ok((resolver, instructions)) +} + +fn get_onchain_info( + instructions: &FixedAmountPaymentInstructions, +) -> Result<(Address, Amount), Error> { + // Look for on chain payment method as it's the only one we can support + let PaymentMethod::OnChain(addr) = instructions + .methods() + .iter() + .find(|ix| matches!(ix, PaymentMethod::OnChain(_))) + .ok_or(Error::Generic( + "Missing Onchain payment method option.".to_string(), + ))? + else { + return Err(Error::Generic("Unsupported payment method".to_string())); + }; + + let Some(onchain_amount) = instructions.onchain_payment_amount() else { + return Err(Error::Generic( + "On chain amount should be specified".to_string(), + )); + }; + + // We need this conversion since Amount from instructions is different from Amount from bitcoin + Ok((addr.clone(), Amount::from_sat(onchain_amount.milli_sats()))) +} + +pub async fn process_instructions( + amount_to_send: Amount, + payment_instructions: &PaymentInstructions, + resolver: DNSHrnResolver, +) -> Result<(Address, Amount), Error> { + match payment_instructions { + PaymentInstructions::ConfigurableAmount(instructions) => { + // Look for on chain payment method as it's the only one we can support + if !instructions + .methods() + .any(|method| matches!(method.method_type(), PaymentMethodType::OnChain)) + { + return Err(Error::Generic("Unsupported payment method".to_string())); + } + + let min_amount = instructions + .min_amt() + .map(|amnt| Amount::from_sat(amnt.sats_rounding_up())); + + let max_amount = instructions + .max_amt() + .map(|amnt| Amount::from_sat(amnt.sats_rounding_up())); + + if min_amount.is_some_and(|min| amount_to_send < min) { + return Err(Error::Generic( + format!( + "Amount to send should be greater than min {}", + min_amount.unwrap() + ) + .to_string(), + )); + } + + if max_amount.is_some_and(|max| amount_to_send > max) { + return Err(Error::Generic( + format!( + "Amount to send should be lower than max {}", + max_amount.unwrap() + ) + .to_string(), + )); + } + + let fixed_instructions = instructions + .clone() + .set_amount( + amount::Amount::from_sats(amount_to_send.to_sat()).unwrap(), + &resolver, + ) + .await + .map_err(|err| { + Error::Generic(format!("Error occured while parsing instructions {err}")) + })?; + + let onchain_details = get_onchain_info(&fixed_instructions)?; + + Ok((onchain_details.0.clone(), onchain_details.1)) + } + + PaymentInstructions::FixedAmount(instructions) => Ok(get_onchain_info(instructions)?), + } +} + +/// Resolves the dns payment instructions found at the specified Human Readable Name +pub async fn resolve_dns_recipient( + hrn: &str, + network: Network, + dns_resolver: &str, +) -> Result { + let (resolver, instructions) = parse_dns_instructions(hrn, network, dns_resolver).await?; + + match instructions { + PaymentInstructions::ConfigurableAmount(ix) => { + let description = ix.recipient_description().map(|s| s.to_string()); + let min_amount = ix.min_amt().map(|amnt| Amount::from_sat(amnt.milli_sats())); + let max_amount = ix.max_amt().map(|amnt| Amount::from_sat(amnt.milli_sats())); + + // Let's set a dummy amount to resolve the payment methods accepted. + let fixed_instructions = ix + .set_amount(amount::Amount::ZERO, &resolver) + .await + .map_err(ParseError::InvalidInstructions)?; + + let payment = ResolvedPaymentInfo { + min_amount, + max_amount, + payment_methods: fixed_instructions.methods().into(), + description, + hrn: hrn.to_string(), + notes: "This is configurable payment instructions. You must send an amount between min_amount and max_amount if set.".to_string(), + }; + + Ok(payment) + } + + PaymentInstructions::FixedAmount(ix) => { + let max_amount = ix + .max_amount() + .map(|amnt| Amount::from_sat(amnt.milli_sats())); + + let payment = ResolvedPaymentInfo { + min_amount: None, + max_amount, + payment_methods: ix.methods().into(), + description: ix.recipient_description().map(|s| s.to_string()), + hrn: hrn.to_string(), + notes: "This is a fixed payment instructions. You must send exactly the amount specified in max_amount.".to_string(), + }; + + Ok(payment) + } + } +} diff --git a/src/handlers/dns/mod.rs b/src/handlers/dns/mod.rs new file mode 100644 index 0000000..6e682bc --- /dev/null +++ b/src/handlers/dns/mod.rs @@ -0,0 +1,175 @@ +mod dns_payment_instructions; + +use crate::error::BDKCliError as Error; +use crate::handlers::dns::dns_payment_instructions::{ + parse_dns_instructions, process_instructions, resolve_dns_recipient, +}; +use crate::handlers::{AppContext, AsyncAppCommand, Init, OfflineOperations}; +use crate::utils::types::{PsbtResult, StatusResult}; +use crate::utils::{parse_dns_recipient, parse_outpoint, parse_recipient}; +use bdk_wallet::KeychainKind; +use bdk_wallet::bitcoin::base64::Engine; +use bdk_wallet::bitcoin::base64::prelude::BASE64_STANDARD; +use bdk_wallet::bitcoin::script::PushBytesBuf; +use bdk_wallet::bitcoin::{Amount, FeeRate, OutPoint, ScriptBuf, Sequence}; +use clap::Parser; +use std::collections::BTreeMap; + +/// Resolves BIP-353 DNS payment instructions for a human-readable name. +#[derive(Parser, Debug, Clone, PartialEq)] +pub struct ResolveDnsRecipientCommand { + /// Human-readable name (e.g. user@domain.com) + pub hrn: String, + /// DNS resolver address + #[arg(long, default_value = "8.8.8.8")] + pub resolver: String, +} + +impl AsyncAppCommand> for ResolveDnsRecipientCommand { + type Output = StatusResult; + + async fn execute(&self, ctx: &mut AppContext) -> Result { + let resolved = resolve_dns_recipient(&self.hrn, ctx.network, &self.resolver) + .await + .map_err(|e| Error::Generic(format!("{:?}", e)))?; + Ok(StatusResult { + message: resolved.display()?, + }) + } +} + +/// Creates a new unsigned transaction from DNS payment instructions. +#[derive(Parser, Debug, Clone, PartialEq)] +pub struct CreateDnsTxCommand { + #[arg(env = "ADDRESS:SAT", long = "to", value_parser = parse_recipient)] + pub recipients: Vec<(ScriptBuf, u64)>, + #[arg(long = "to_dns", value_parser = parse_dns_recipient)] + pub dns_recipients: Vec<(String, u64)>, + #[arg(long = "dns_resolver", default_value = "8.8.8.8")] + pub dns_resolver: String, + #[arg(long = "send_all", short = 'a')] + pub send_all: bool, + #[arg(long = "enable_rbf", short = 'r', default_value_t = true)] + pub enable_rbf: bool, + #[arg(long = "offline_signer")] + pub offline_signer: bool, + #[arg(env = "MUST_SPEND_TXID:VOUT", long = "utxos", value_parser = parse_outpoint)] + pub utxos: Option>, + #[arg(env = "CANT_SPEND_TXID:VOUT", long = "unspendable", value_parser = parse_outpoint)] + pub unspendable: Option>, + #[arg(env = "SATS_VBYTE", short = 'f', long = "fee_rate")] + pub fee_rate: Option, + #[arg(env = "EXT_POLICY", long = "external_policy")] + pub external_policy: Option, + #[arg(env = "INT_POLICY", long = "internal_policy")] + pub internal_policy: Option, + #[arg( + env = "ADD_STRING", + long = "add_string", + short = 's', + conflicts_with = "add_data" + )] + pub add_string: Option, + #[arg( + env = "ADD_DATA", + long = "add_data", + short = 'o', + conflicts_with = "add_string" + )] + pub add_data: Option, +} + +impl AsyncAppCommand>> for CreateDnsTxCommand { + type Output = PsbtResult; + + async fn execute( + &self, + ctx: &mut AppContext>, + ) -> Result { + let network = ctx.network; + let mut recipients: Vec<(ScriptBuf, u64)> = self.recipients.clone(); + + for (hrn, amount_sat) in &self.dns_recipients { + log::info!("Resolving DNS instructions for recipient {hrn}"); + let amount = Amount::from_sat(*amount_sat); + let (resolver, instructions) = parse_dns_instructions(hrn, network, &self.dns_resolver) + .await + .map_err(|e| Error::Generic(format!("Parsing error occured {e:#?}")))?; + let payment = process_instructions(amount, &instructions, resolver).await?; + recipients.push((payment.0.into(), payment.1.to_sat())); + } + + if recipients.is_empty() { + return Err(Error::Generic( + "Either --to or --to_dns parameters must be specified".to_string(), + )); + } + + let mut tx_builder = ctx.state.wallet.build_tx(); + + if self.send_all { + if recipients.len() == 1 { + tx_builder.drain_wallet().drain_to(recipients[0].0.clone()); + } else { + return Err(Error::Generic( + "Wallet can only be drained to a single output".to_string(), + )); + } + } else { + let recipients = recipients + .into_iter() + .map(|(script, amount)| (script, Amount::from_sat(amount))) + .collect(); + tx_builder.set_recipients(recipients); + } + + if !self.enable_rbf { + tx_builder.set_exact_sequence(Sequence::MAX); + } + if self.offline_signer { + tx_builder.include_output_redeem_witness_script(); + } + if let Some(fee_rate) = self.fee_rate + && let Some(fee_rate) = FeeRate::from_sat_per_vb(fee_rate as u64) + { + tx_builder.fee_rate(fee_rate); + } + if let Some(utxos) = &self.utxos { + tx_builder + .add_utxos(&utxos[..]) + .map_err(|_| bdk_wallet::error::CreateTxError::UnknownUtxo)?; + } + if let Some(unspendable) = &self.unspendable { + tx_builder.unspendable(unspendable.to_vec()); + } + if let Some(base64_data) = &self.add_data { + let op_return_data = BASE64_STANDARD + .decode(base64_data) + .map_err(|e| Error::Generic(e.to_string()))?; + tx_builder.add_data( + &PushBytesBuf::try_from(op_return_data) + .map_err(|e| Error::Generic(e.to_string()))?, + ); + } else if let Some(string_data) = &self.add_string { + let data = PushBytesBuf::try_from(string_data.as_bytes().to_vec()) + .map_err(|e| Error::Generic(e.to_string()))?; + tx_builder.add_data(&data); + } + + let policies = vec![ + self.external_policy + .as_ref() + .map(|p| (p, KeychainKind::External)), + self.internal_policy + .as_ref() + .map(|p| (p, KeychainKind::Internal)), + ]; + for (policy, keychain) in policies.into_iter().flatten() { + let policy = serde_json::from_str::>>(policy)?; + tx_builder.policy_path(policy, keychain); + } + + let psbt = tx_builder.finish()?; + Ok(PsbtResult::new(&psbt, Some(false))) + } +} diff --git a/src/handlers/key.rs b/src/handlers/key.rs index ce8c844..c1a9e76 100644 --- a/src/handlers/key.rs +++ b/src/handlers/key.rs @@ -1,17 +1,13 @@ use crate::commands::KeySubCommand; use crate::error::BDKCliError as Error; -use crate::handlers::Init; -use crate::handlers::{AppCommand, AppContext}; -use crate::utils::output::FormatOutput; -use crate::utils::types::KeyResult; +use crate::handlers::{AppCommand, AppContext, Init}; +use crate::utils::{output::FormatOutput, types::KeyResult}; use bdk_wallet::bip39::{Language, Mnemonic}; -use bdk_wallet::bitcoin::bip32::DerivationPath; -use bdk_wallet::bitcoin::bip32::KeySource; -use bdk_wallet::bitcoin::bip32::Xpriv; +use bdk_wallet::bitcoin::bip32::{DerivationPath, KeySource, Xpriv}; use bdk_wallet::bitcoin::key::Secp256k1; -use bdk_wallet::keys::bip39::WordCount; -use bdk_wallet::keys::{DerivableKey, GeneratableKey}; -use bdk_wallet::keys::{DescriptorKey, ExtendedKey, GeneratedKey}; +use bdk_wallet::keys::{ + DerivableKey, DescriptorKey, ExtendedKey, GeneratableKey, GeneratedKey, bip39::WordCount, +}; use bdk_wallet::miniscript::{self, Segwitv0}; use clap::Parser; diff --git a/src/handlers/mod.rs b/src/handlers/mod.rs index 113398b..a2c4ad5 100644 --- a/src/handlers/mod.rs +++ b/src/handlers/mod.rs @@ -5,6 +5,13 @@ pub mod dns; pub mod key; pub mod offline; pub mod online; +#[cfg(any( + feature = "electrum", + feature = "esplora", + feature = "cbf", + feature = "rpc" +))] +pub mod payjoin; pub mod repl; #[cfg(any( diff --git a/src/handlers/offline.rs b/src/handlers/offline.rs index d536e18..7490d80 100644 --- a/src/handlers/offline.rs +++ b/src/handlers/offline.rs @@ -84,8 +84,7 @@ impl OfflineWalletSubCommand { .write_out(std::io::stdout()), #[cfg(feature = "dns_payment")] Self::CreateDnsTx(_) => Err(Error::Generic( - "CreateDnsTx is dispatched asynchronously through main" - .to_string(), + "CreateDnsTx is dispatched asynchronously through main".to_string(), )), } } diff --git a/src/handlers/online.rs b/src/handlers/online.rs index f520451..bf84dc0 100644 --- a/src/handlers/online.rs +++ b/src/handlers/online.rs @@ -2,14 +2,13 @@ use clap::Parser; #[cfg(feature = "electrum")] use crate::client::BlockchainClient::Electrum; -#[cfg(feature = "rpc")] -use crate::client::BlockchainClient::RpcClient; #[cfg(feature = "cbf")] use crate::client::{BlockchainClient::KyotoClient, sync_kyoto_client}; #[cfg(feature = "esplora")] use {crate::client::BlockchainClient::Esplora, bdk_esplora::EsploraAsyncExt}; #[cfg(feature = "rpc")] use { + crate::client::BlockchainClient::RpcClient, bdk_bitcoind_rpc::{Emitter, NO_EXPECTED_MEMPOOL_TXS, bitcoincore_rpc::RpcApi}, bdk_wallet::chain::{BlockId, CanonicalizationParams, CheckPoint}, }; @@ -27,11 +26,12 @@ use crate::utils::print_wallet_events; use { crate::commands::OnlineWalletSubCommand, crate::error::BDKCliError as Error, - crate::handlers::{AppContext, AsyncAppCommand, OnlineOperations}, - crate::payjoin::PayjoinManager, - crate::utils::is_final, - crate::utils::output::FormatOutput, - crate::utils::types::{StatusResult, TransactionResult}, + crate::handlers::{AppContext, AsyncAppCommand, OnlineOperations, payjoin::PayjoinManager}, + crate::utils::{ + is_final, + output::FormatOutput, + types::{StatusResult, TransactionResult}, + }, bdk_wallet::bitcoin::{ Psbt, Transaction, Txid, base64::Engine, base64::prelude::BASE64_STANDARD, consensus::Decodable, hex::FromHex, diff --git a/src/handlers/payjoin/db.rs b/src/handlers/payjoin/db.rs new file mode 100644 index 0000000..dd64e4d --- /dev/null +++ b/src/handlers/payjoin/db.rs @@ -0,0 +1,836 @@ +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::handlers::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/handlers/payjoin/mod.rs b/src/handlers/payjoin/mod.rs new file mode 100644 index 0000000..7a2fe36 --- /dev/null +++ b/src/handlers/payjoin/mod.rs @@ -0,0 +1,1084 @@ +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, 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::{HpkePublicKey, ImplementationError, UriExt}; +use serde_json::{json, to_string_pretty}; +use std::{path::PathBuf, sync::Arc}; + +use crate::handlers::payjoin::db::{ReceiverPersister, SenderPersister, open_payjoin_db}; +use crate::handlers::payjoin::ohttp::RelayManager; + +pub mod db; +pub mod ohttp; + +/// Coordinates Payjoin receive and send flows. +/// +/// 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: 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, + 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( + &mut self, + amount: u64, + directory: String, + max_fee_rate: Option, + ohttp_relays: Vec, + blockchain_client: &BlockchainClient, + ) -> Result { + let address = self + .wallet + .next_unused_address(bdk_wallet::KeychainKind::External); + + let ohttp_relays: Vec = ohttp_relays + .into_iter() + .map(|s| url::Url::parse(&s)) + .collect::>() + .map_err(|e| Error::Generic(format!("Failed to parse one or more OHTTP URLs: {e}")))?; + + self.relay_manager.configure(ohttp_relays)?; + let ohttp_keys = self.relay_manager.fetch_ohttp_keys(&directory).await?; + + let persister = crate::handlers::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)? + .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:"); + println!("{pj_uri}"); + + self.proceed_receiver_session( + ReceiveSession::Initialized(receiver.clone()), + &persister, + checked_max_fee_rate, + blockchain_client, + ) + .await?; + + Ok(to_string_pretty(&json!({}))?) + } + + pub async fn send_payjoin( + &mut self, + uri: String, + fee_rate: u64, + ohttp_relays: Vec, + blockchain_client: &BlockchainClient, + ) -> Result { + let uri = payjoin::Uri::try_from(uri) + .map_err(|e| Error::Generic(format!("Failed parsing to Payjoin URI: {}", e)))?; + let uri = uri.require_network(self.wallet.network()).map_err(|e| { + Error::Generic(format!("Failed setting the right network for the URI: {e}")) + })?; + let uri = uri + .check_pj_supported() + .map_err(|e| Error::Generic(format!("URI does not support Payjoin: {}", e)))?; + + let sats = uri + .amount + .ok_or_else(|| Error::Generic("Amount is not specified in the URI.".to_string()))?; + + let fee_rate = FeeRate::from_sat_per_vb(fee_rate).expect("Provided fee rate is not valid."); + + // Build and sign the original PSBT which pays to the receiver. + let mut original_psbt = { + let mut tx_builder = self.wallet.build_tx(); + tx_builder + .add_recipient(uri.address.script_pubkey(), sats) + .fee_rate(fee_rate); + + tx_builder.finish()? + }; + if !self + .wallet + .sign(&mut original_psbt, SignOptions::default())? + { + return Err(Error::Generic( + "Failed to sign and finalize the original PSBT.".to_string(), + )); + } + + let txid = match uri.extras.pj_param() { + payjoin::PjParam::V1(_) => { + let (req, ctx) = payjoin::send::v1::SenderBuilder::new(original_psbt.clone(), uri) + .build_recommended(fee_rate)? + .create_v1_post_request(); + + let response = self.send_payjoin_post_request(req).await?; + let psbt = ctx.process_response(&response.bytes().await?)?; + + self.process_payjoin_proposal(psbt, blockchain_client) + .await? + } + payjoin::PjParam::V2(v2_param) => { + let ohttp_relays: Vec = ohttp_relays + .into_iter() + .map(|s| url::Url::parse(&s)) + .collect::>() + .map_err(|e| { + Error::Generic(format!("Failed to parse one or more OHTTP URLs: {e}")) + })?; + + 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(sender_state, &persister, blockchain_client) + .await? + } + _ => { + unimplemented!("Payjoin version not recognized."); + } + }; + + Ok(to_string_pretty(&json!({ "txid": txid }))?) + } + + async fn proceed_receiver_session( + &mut self, + session: ReceiveSession, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + match session { + ReceiveSession::Initialized(proposal) => { + 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) + .await + } + ReceiveSession::MaybeInputsOwned(proposal) => { + self.check_inputs_not_owned(proposal, persister, max_fee_rate, blockchain_client) + .await + } + ReceiveSession::MaybeInputsSeen(proposal) => { + self.check_no_inputs_seen_before( + proposal, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + ReceiveSession::OutputsUnknown(proposal) => { + self.identify_receiver_outputs(proposal, persister, max_fee_rate, blockchain_client) + .await + } + ReceiveSession::WantsOutputs(proposal) => { + self.commit_outputs(proposal, persister, max_fee_rate, blockchain_client) + .await + } + ReceiveSession::WantsInputs(proposal) => { + self.contribute_inputs(proposal, persister, max_fee_rate, blockchain_client) + .await + } + ReceiveSession::WantsFeeRange(proposal) => { + self.apply_fee_range(proposal, persister, max_fee_rate, blockchain_client) + .await + } + ReceiveSession::ProvisionalProposal(proposal) => { + self.finalize_proposal(proposal, persister, blockchain_client) + .await + } + ReceiveSession::PayjoinProposal(proposal) => { + self.send_payjoin_proposal(proposal, persister, blockchain_client) + .await + } + ReceiveSession::Monitor(proposal) => { + self.monitor_payjoin_proposal(proposal, persister, blockchain_client) + .await + } + ReceiveSession::HasReplyableError(error) => self.handle_error(error, persister).await, + ReceiveSession::Closed(_) => Err(Error::Generic("Session closed".to_string())), + } + } + + async fn read_from_directory( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + let mut current_receiver_typestate = receiver; + let next_receiver_typestate = loop { + 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); + match state_transition { + Ok(OptionalTransitionOutcome::Progress(next_state)) => { + println!("Got a request from the sender. Responding with a Payjoin proposal."); + break next_state; + } + Ok(OptionalTransitionOutcome::Stasis(current_state)) => { + current_receiver_typestate = current_state; + continue; + } + Err(e) => { + return Err(Error::Generic(format!( + "Error occurred when polling for Payjoin proposal from the directory: {e}" + ))); + } + } + }; + self.check_proposal( + next_receiver_typestate, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + + async fn check_proposal( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + let next_receiver_typestate = receiver + .assume_interactive_receiver() + .save(persister) + .map_err(|e| { + Error::Generic(format!( + "Error occurred when saving after assuming interactive receiver and not checking proposal broadcastability: {e}" + )) + })?; + + println!( + "Checking whether the original proposal can be broadcasted itself is not supported. If the Payjoin fails, manually fall back to the transaction below." + ); + println!( + "{}", + serialize_hex(&next_receiver_typestate.extract_tx_to_schedule_broadcast()) + ); + + self.check_inputs_not_owned( + next_receiver_typestate, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + + async fn check_inputs_not_owned( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + 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)?; + + self.check_no_inputs_seen_before( + next_receiver_typestate, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + + async fn check_no_inputs_seen_before( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + 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, + blockchain_client, + ) + .await + } + + async fn identify_receiver_outputs( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + 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)?; + + self.commit_outputs( + next_receiver_typestate, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + + async fn commit_outputs( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + // This is a typestate to modify existing receiver-owned outputs in case the receiver wants + // to do that. This is a very simple implementation of Payjoin so we are just going + // to commit to the existing outputs which the sender included in the original proposal. + let next_receiver_typestate = receiver.commit_outputs().save(persister).map_err(|e| { + Error::Generic(format!( + "Error occurred when saving after committing to the outputs in the proposal: {e}" + )) + })?; + self.contribute_inputs( + next_receiver_typestate, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + + async fn contribute_inputs( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + let candidate_inputs: Vec = self + .wallet + .list_unspent() + .filter(|output| output.chain_position.is_confirmed()) + .map(|output| { + let psbtin = self + .wallet + .get_psbt_input(output.clone(), None, false) + .expect( + "Failed to get the PSBT Input using the output of the unspent transaction", + ); + let txin = TxIn { + previous_output: output.outpoint, + ..Default::default() + }; + InputPair::new(txin, psbtin, None) + .expect("Failed to create InputPair when contributing outputs to the proposal") + }) + .collect(); + let selected_input = receiver.try_preserving_privacy(candidate_inputs)?; + + let next_receiver_typestate = receiver.contribute_inputs(vec![selected_input])? + .commit_inputs().save(persister) + .map_err(|e| { + Error::Generic(format!("Error occurred when saving after committing to the inputs after receiver contribution: {e}")) + })?; + + self.apply_fee_range( + next_receiver_typestate, + persister, + max_fee_rate, + blockchain_client, + ) + .await + } + + async fn apply_fee_range( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + max_fee_rate: FeeRate, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + 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 + } + + async fn finalize_proposal( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + let next_receiver_typestate = receiver + .finalize_proposal(|psbt| { + let mut psbt_clone = psbt.clone(); + + // We cannot finalize the transaction for broadcasting as it does not have the + // sender signatures yet. Hence, we only care about whether this returns an Err. + let _ = !self + .wallet + .sign(&mut psbt_clone, SignOptions::default()) + .map_err(|e| { + ImplementationError::from( + format!("Error occurred when signing the Payjoin PSBT: {e}").as_str(), + ) + })?; + + Ok(psbt_clone) + }) + .save(persister)?; + + self.send_payjoin_proposal(next_receiver_typestate, persister, blockchain_client) + .await + } + + async fn send_payjoin_proposal( + &mut self, + receiver: Receiver, + persister: &impl SessionPersister, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + 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)?; + println!( + "Response successful. TXID: {}", + payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() + ); + return self + .monitor_payjoin_proposal(next_receiver_typestate, persister, blockchain_client) + .await; + } + + /// Polls the blockchain periodically and checks whether the Payjoin was broadcasted by the + /// 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 [`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, + persister: &impl SessionPersister, + blockchain_client: &BlockchainClient, + ) -> Result<(), Error> { + let poll_interval = tokio::time::Duration::from_millis(200); + let sync_interval = tokio::time::Duration::from_secs(3); + let timeout_duration = tokio::time::Duration::from_secs(15); + + println!( + "Polling for Payjoin transaction broadcast. This may take up to {} seconds...", + timeout_duration.as_secs() + ); + let result = tokio::time::timeout(timeout_duration, async { + let mut poll_timer = tokio::time::interval(poll_interval); + let mut sync_timer = tokio::time::interval(sync_interval); + poll_timer.tick().await; + sync_timer.tick().await; + blockchain_client.sync_wallet(self.wallet).await?; + + loop { + tokio::select! { + _ = poll_timer.tick() => { + // Time to check payment + let check_result = receiver + .check_payment( + |txid| { + let Some(tx_details) = self.wallet.tx_details(txid) else { + return Err(ImplementationError::from("Cannot find the transaction in the mempool or the blockchain")); + }; + + let is_seen = matches!(tx_details.chain_position, bdk_wallet::chain::ChainPosition::Confirmed { .. } | bdk_wallet::chain::ChainPosition::Unconfirmed { first_seen: Some(_), .. }); + + if is_seen { + return Ok(Some(tx_details.tx.as_ref().clone())); + } + Err(ImplementationError::from("Cannot find the transaction in the mempool or the blockchain")) + } + ) + .save(persister); + + if let Ok(OptionalTransitionOutcome::Progress(_)) = check_result { + println!("Payjoin transaction detected in the mempool!"); + return Ok(()); + } + // For Stasis or Err, continue polling (implicit - falls through to next loop iteration) + } + _ = sync_timer.tick() => { + // Time to sync wallet + blockchain_client.sync_wallet(self.wallet).await?; + } + } + } + }) + .await; + + match result { + Ok(ok) => ok, + Err(_) => Err(Error::Generic(format!( + "Timeout waiting for Payjoin transaction broadcast after {:?}. Check the state of the transaction manually after running the sync command.", + timeout_duration + ))), + } + } + + async fn handle_error( + &self, + receiver: Receiver, + persister: &impl SessionPersister, + ) -> Result<(), Error> { + 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, + Err(e) => { + return Err(Error::Generic(format!( + "Failed to get error response bytes: {}", + e + ))); + } + }; + + if let Err(e) = receiver + .process_error_response(&err_bytes, err_ctx) + .save(persister) + { + return Err(Error::Generic(format!( + "Failed to process error response: {}", + e + ))); + } + + Ok(()) + } + + async fn proceed_sender_session( + &self, + session: SendSession, + persister: &impl SessionPersister, + blockchain_client: &BlockchainClient, + ) -> Result { + match session { + SendSession::WithReplyKey(context) => { + self.post_original_proposal(context, persister, blockchain_client) + .await + } + SendSession::PollingForProposal(context) => { + self.get_proposed_payjoin_proposal(context, persister, blockchain_client) + .await + } + SendSession::Closed(SenderSessionOutcome::Success(psbt)) => { + self.process_payjoin_proposal(psbt, blockchain_client).await + } + _ => Err(Error::Generic("Unexpected SendSession state!".to_string())), + } + } + + async fn post_original_proposal( + &self, + sender: Sender, + persister: &impl SessionPersister, + blockchain_client: &BlockchainClient, + ) -> Result { + 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)?; + self.get_proposed_payjoin_proposal(sender, persister, blockchain_client) + .await + } + + async fn get_proposed_payjoin_proposal( + &self, + sender: Sender, + persister: &impl SessionPersister, + blockchain_client: &BlockchainClient, + ) -> Result { + let mut sender = sender.clone(); + loop { + 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); + match processed_response { + Ok(OptionalTransitionOutcome::Progress(psbt)) => { + println!("Proposal received. Processing..."); + return self.process_payjoin_proposal(psbt, blockchain_client).await; + } + Ok(OptionalTransitionOutcome::Stasis(current_state)) => { + println!("No response yet. Continuing polling..."); + sender = current_state; + continue; + } + Err(e) => { + break Err(Error::Generic(format!( + "Error occurred when polling for Payjoin v2 proposal: {e}" + ))); + } + } + } + } + + async fn process_payjoin_proposal( + &self, + mut psbt: Psbt, + blockchain_client: &BlockchainClient, + ) -> Result { + if !self.wallet.sign(&mut psbt, SignOptions::default())? { + return Err(Error::Generic( + "Failed to sign and finalize the Payjoin proposal PSBT.".to_string(), + )); + } + + blockchain_client + .broadcast(psbt.extract_tx_fee_rate_limit()?) + .await + } + + async fn send_payjoin_post_request( + &self, + req: payjoin::Request, + ) -> reqwest::Result { + let client = reqwest::Client::new(); + client + .post(req.url) + .header("Content-Type", req.content_type) + .body(req.body) + .send() + .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/handlers/payjoin/ohttp.rs b/src/handlers/payjoin/ohttp.rs new file mode 100644 index 0000000..bc26466 --- /dev/null +++ b/src/handlers/payjoin/ohttp.rs @@ -0,0 +1,127 @@ +use crate::error::BDKCliError as Error; +use std::sync::{Arc, Mutex}; + +#[derive(Debug, Clone)] +pub(crate) struct RelayManager { + relays: Vec, + failed_relays: Arc>>, +} + +impl RelayManager { + pub(crate) fn new() -> Self { + Self { + relays: Vec::new(), + failed_relays: Arc::new(Mutex::new(Vec::new())), + } + } + + 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(), + )); + } + + self.relays = relays; + self.failed_relays + .lock() + .expect("Lock should not be poisoned") + .clear(); + Ok(()) + } + + 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(crate) fn choose_relay(&self) -> Result { + use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; + + let failed_relays = self + .failed_relays + .lock() + .expect("Lock should not be poisoned"); + let remaining_relays: Vec<_> = self + .relays + .iter() + .filter(|relay| !failed_relays.contains(relay)) + .cloned() + .collect(); + + remaining_relays + .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) + .cloned() + .ok_or_else(|| Error::Generic("No valid OHTTP relays available".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; + + 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()); + } +} diff --git a/src/main.rs b/src/main.rs index 8ee07f6..a79492e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,19 +15,9 @@ mod commands; mod config; mod error; mod handlers; -#[cfg(any( - feature = "electrum", - feature = "esplora", - feature = "cbf", - feature = "rpc" -))] -mod payjoin; mod persister; mod utils; -#[cfg(feature = "dns_payment")] -mod dns_payment_instructions; - use bdk_wallet::bitcoin::Network; use log::{debug, warn}; diff --git a/src/payjoin/db.rs b/src/payjoin/db.rs deleted file mode 100644 index f21524d..0000000 --- a/src/payjoin/db.rs +++ /dev/null @@ -1,835 +0,0 @@ -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 deleted file mode 100644 index 11e09a0..0000000 --- a/src/payjoin/mod.rs +++ /dev/null @@ -1,1084 +0,0 @@ -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, 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::{HpkePublicKey, ImplementationError, UriExt}; -use serde_json::{json, to_string_pretty}; -use std::{path::PathBuf, sync::Arc}; - -use crate::payjoin::db::{ReceiverPersister, SenderPersister, open_payjoin_db}; -use crate::payjoin::ohttp::RelayManager; - -pub mod db; -pub mod ohttp; - -/// Coordinates Payjoin receive and send flows. -/// -/// 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: 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, - 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( - &mut self, - amount: u64, - directory: String, - max_fee_rate: Option, - ohttp_relays: Vec, - blockchain_client: &BlockchainClient, - ) -> Result { - let address = self - .wallet - .next_unused_address(bdk_wallet::KeychainKind::External); - - let ohttp_relays: Vec = ohttp_relays - .into_iter() - .map(|s| url::Url::parse(&s)) - .collect::>() - .map_err(|e| Error::Generic(format!("Failed to parse one or more OHTTP URLs: {e}")))?; - - self.relay_manager.configure(ohttp_relays)?; - let ohttp_keys = self.relay_manager.fetch_ohttp_keys(&directory).await?; - - 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)? - .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:"); - println!("{pj_uri}"); - - self.proceed_receiver_session( - ReceiveSession::Initialized(receiver.clone()), - &persister, - checked_max_fee_rate, - blockchain_client, - ) - .await?; - - Ok(to_string_pretty(&json!({}))?) - } - - pub async fn send_payjoin( - &mut self, - uri: String, - fee_rate: u64, - ohttp_relays: Vec, - blockchain_client: &BlockchainClient, - ) -> Result { - let uri = payjoin::Uri::try_from(uri) - .map_err(|e| Error::Generic(format!("Failed parsing to Payjoin URI: {}", e)))?; - let uri = uri.require_network(self.wallet.network()).map_err(|e| { - Error::Generic(format!("Failed setting the right network for the URI: {e}")) - })?; - let uri = uri - .check_pj_supported() - .map_err(|e| Error::Generic(format!("URI does not support Payjoin: {}", e)))?; - - let sats = uri - .amount - .ok_or_else(|| Error::Generic("Amount is not specified in the URI.".to_string()))?; - - let fee_rate = FeeRate::from_sat_per_vb(fee_rate).expect("Provided fee rate is not valid."); - - // Build and sign the original PSBT which pays to the receiver. - let mut original_psbt = { - let mut tx_builder = self.wallet.build_tx(); - tx_builder - .add_recipient(uri.address.script_pubkey(), sats) - .fee_rate(fee_rate); - - tx_builder.finish()? - }; - if !self - .wallet - .sign(&mut original_psbt, SignOptions::default())? - { - return Err(Error::Generic( - "Failed to sign and finalize the original PSBT.".to_string(), - )); - } - - let txid = match uri.extras.pj_param() { - payjoin::PjParam::V1(_) => { - let (req, ctx) = payjoin::send::v1::SenderBuilder::new(original_psbt.clone(), uri) - .build_recommended(fee_rate)? - .create_v1_post_request(); - - let response = self.send_payjoin_post_request(req).await?; - let psbt = ctx.process_response(&response.bytes().await?)?; - - self.process_payjoin_proposal(psbt, blockchain_client) - .await? - } - payjoin::PjParam::V2(v2_param) => { - let ohttp_relays: Vec = ohttp_relays - .into_iter() - .map(|s| url::Url::parse(&s)) - .collect::>() - .map_err(|e| { - Error::Generic(format!("Failed to parse one or more OHTTP URLs: {e}")) - })?; - - 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(sender_state, &persister, blockchain_client) - .await? - } - _ => { - unimplemented!("Payjoin version not recognized."); - } - }; - - Ok(to_string_pretty(&json!({ "txid": txid }))?) - } - - async fn proceed_receiver_session( - &mut self, - session: ReceiveSession, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - match session { - ReceiveSession::Initialized(proposal) => { - 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) - .await - } - ReceiveSession::MaybeInputsOwned(proposal) => { - self.check_inputs_not_owned(proposal, persister, max_fee_rate, blockchain_client) - .await - } - ReceiveSession::MaybeInputsSeen(proposal) => { - self.check_no_inputs_seen_before( - proposal, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - ReceiveSession::OutputsUnknown(proposal) => { - self.identify_receiver_outputs(proposal, persister, max_fee_rate, blockchain_client) - .await - } - ReceiveSession::WantsOutputs(proposal) => { - self.commit_outputs(proposal, persister, max_fee_rate, blockchain_client) - .await - } - ReceiveSession::WantsInputs(proposal) => { - self.contribute_inputs(proposal, persister, max_fee_rate, blockchain_client) - .await - } - ReceiveSession::WantsFeeRange(proposal) => { - self.apply_fee_range(proposal, persister, max_fee_rate, blockchain_client) - .await - } - ReceiveSession::ProvisionalProposal(proposal) => { - self.finalize_proposal(proposal, persister, blockchain_client) - .await - } - ReceiveSession::PayjoinProposal(proposal) => { - self.send_payjoin_proposal(proposal, persister, blockchain_client) - .await - } - ReceiveSession::Monitor(proposal) => { - self.monitor_payjoin_proposal(proposal, persister, blockchain_client) - .await - } - ReceiveSession::HasReplyableError(error) => self.handle_error(error, persister).await, - ReceiveSession::Closed(_) => Err(Error::Generic("Session closed".to_string())), - } - } - - async fn read_from_directory( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - let mut current_receiver_typestate = receiver; - let next_receiver_typestate = loop { - 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); - match state_transition { - Ok(OptionalTransitionOutcome::Progress(next_state)) => { - println!("Got a request from the sender. Responding with a Payjoin proposal."); - break next_state; - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - current_receiver_typestate = current_state; - continue; - } - Err(e) => { - return Err(Error::Generic(format!( - "Error occurred when polling for Payjoin proposal from the directory: {e}" - ))); - } - } - }; - self.check_proposal( - next_receiver_typestate, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - - async fn check_proposal( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - let next_receiver_typestate = receiver - .assume_interactive_receiver() - .save(persister) - .map_err(|e| { - Error::Generic(format!( - "Error occurred when saving after assuming interactive receiver and not checking proposal broadcastability: {e}" - )) - })?; - - println!( - "Checking whether the original proposal can be broadcasted itself is not supported. If the Payjoin fails, manually fall back to the transaction below." - ); - println!( - "{}", - serialize_hex(&next_receiver_typestate.extract_tx_to_schedule_broadcast()) - ); - - self.check_inputs_not_owned( - next_receiver_typestate, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - - async fn check_inputs_not_owned( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - 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)?; - - self.check_no_inputs_seen_before( - next_receiver_typestate, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - - async fn check_no_inputs_seen_before( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - 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, - blockchain_client, - ) - .await - } - - async fn identify_receiver_outputs( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - 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)?; - - self.commit_outputs( - next_receiver_typestate, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - - async fn commit_outputs( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - // This is a typestate to modify existing receiver-owned outputs in case the receiver wants - // to do that. This is a very simple implementation of Payjoin so we are just going - // to commit to the existing outputs which the sender included in the original proposal. - let next_receiver_typestate = receiver.commit_outputs().save(persister).map_err(|e| { - Error::Generic(format!( - "Error occurred when saving after committing to the outputs in the proposal: {e}" - )) - })?; - self.contribute_inputs( - next_receiver_typestate, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - - async fn contribute_inputs( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - let candidate_inputs: Vec = self - .wallet - .list_unspent() - .filter(|output| output.chain_position.is_confirmed()) - .map(|output| { - let psbtin = self - .wallet - .get_psbt_input(output.clone(), None, false) - .expect( - "Failed to get the PSBT Input using the output of the unspent transaction", - ); - let txin = TxIn { - previous_output: output.outpoint, - ..Default::default() - }; - InputPair::new(txin, psbtin, None) - .expect("Failed to create InputPair when contributing outputs to the proposal") - }) - .collect(); - let selected_input = receiver.try_preserving_privacy(candidate_inputs)?; - - let next_receiver_typestate = receiver.contribute_inputs(vec![selected_input])? - .commit_inputs().save(persister) - .map_err(|e| { - Error::Generic(format!("Error occurred when saving after committing to the inputs after receiver contribution: {e}")) - })?; - - self.apply_fee_range( - next_receiver_typestate, - persister, - max_fee_rate, - blockchain_client, - ) - .await - } - - async fn apply_fee_range( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - max_fee_rate: FeeRate, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - 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 - } - - async fn finalize_proposal( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - let next_receiver_typestate = receiver - .finalize_proposal(|psbt| { - let mut psbt_clone = psbt.clone(); - - // We cannot finalize the transaction for broadcasting as it does not have the - // sender signatures yet. Hence, we only care about whether this returns an Err. - let _ = !self - .wallet - .sign(&mut psbt_clone, SignOptions::default()) - .map_err(|e| { - ImplementationError::from( - format!("Error occurred when signing the Payjoin PSBT: {e}").as_str(), - ) - })?; - - Ok(psbt_clone) - }) - .save(persister)?; - - self.send_payjoin_proposal(next_receiver_typestate, persister, blockchain_client) - .await - } - - async fn send_payjoin_proposal( - &mut self, - receiver: Receiver, - persister: &impl SessionPersister, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - 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)?; - println!( - "Response successful. TXID: {}", - payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid() - ); - return self - .monitor_payjoin_proposal(next_receiver_typestate, persister, blockchain_client) - .await; - } - - /// Polls the blockchain periodically and checks whether the Payjoin was broadcasted by the - /// 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 [`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, - persister: &impl SessionPersister, - blockchain_client: &BlockchainClient, - ) -> Result<(), Error> { - let poll_interval = tokio::time::Duration::from_millis(200); - let sync_interval = tokio::time::Duration::from_secs(3); - let timeout_duration = tokio::time::Duration::from_secs(15); - - println!( - "Polling for Payjoin transaction broadcast. This may take up to {} seconds...", - timeout_duration.as_secs() - ); - let result = tokio::time::timeout(timeout_duration, async { - let mut poll_timer = tokio::time::interval(poll_interval); - let mut sync_timer = tokio::time::interval(sync_interval); - poll_timer.tick().await; - sync_timer.tick().await; - blockchain_client.sync_wallet(self.wallet).await?; - - loop { - tokio::select! { - _ = poll_timer.tick() => { - // Time to check payment - let check_result = receiver - .check_payment( - |txid| { - let Some(tx_details) = self.wallet.tx_details(txid) else { - return Err(ImplementationError::from("Cannot find the transaction in the mempool or the blockchain")); - }; - - let is_seen = matches!(tx_details.chain_position, bdk_wallet::chain::ChainPosition::Confirmed { .. } | bdk_wallet::chain::ChainPosition::Unconfirmed { first_seen: Some(_), .. }); - - if is_seen { - return Ok(Some(tx_details.tx.as_ref().clone())); - } - Err(ImplementationError::from("Cannot find the transaction in the mempool or the blockchain")) - } - ) - .save(persister); - - if let Ok(OptionalTransitionOutcome::Progress(_)) = check_result { - println!("Payjoin transaction detected in the mempool!"); - return Ok(()); - } - // For Stasis or Err, continue polling (implicit - falls through to next loop iteration) - } - _ = sync_timer.tick() => { - // Time to sync wallet - blockchain_client.sync_wallet(self.wallet).await?; - } - } - } - }) - .await; - - match result { - Ok(ok) => ok, - Err(_) => Err(Error::Generic(format!( - "Timeout waiting for Payjoin transaction broadcast after {:?}. Check the state of the transaction manually after running the sync command.", - timeout_duration - ))), - } - } - - async fn handle_error( - &self, - receiver: Receiver, - persister: &impl SessionPersister, - ) -> Result<(), Error> { - 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, - Err(e) => { - return Err(Error::Generic(format!( - "Failed to get error response bytes: {}", - e - ))); - } - }; - - if let Err(e) = receiver - .process_error_response(&err_bytes, err_ctx) - .save(persister) - { - return Err(Error::Generic(format!( - "Failed to process error response: {}", - e - ))); - } - - Ok(()) - } - - async fn proceed_sender_session( - &self, - session: SendSession, - persister: &impl SessionPersister, - blockchain_client: &BlockchainClient, - ) -> Result { - match session { - SendSession::WithReplyKey(context) => { - self.post_original_proposal(context, persister, blockchain_client) - .await - } - SendSession::PollingForProposal(context) => { - self.get_proposed_payjoin_proposal(context, persister, blockchain_client) - .await - } - SendSession::Closed(SenderSessionOutcome::Success(psbt)) => { - self.process_payjoin_proposal(psbt, blockchain_client).await - } - _ => Err(Error::Generic("Unexpected SendSession state!".to_string())), - } - } - - async fn post_original_proposal( - &self, - sender: Sender, - persister: &impl SessionPersister, - blockchain_client: &BlockchainClient, - ) -> Result { - 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)?; - self.get_proposed_payjoin_proposal(sender, persister, blockchain_client) - .await - } - - async fn get_proposed_payjoin_proposal( - &self, - sender: Sender, - persister: &impl SessionPersister, - blockchain_client: &BlockchainClient, - ) -> Result { - let mut sender = sender.clone(); - loop { - 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); - match processed_response { - Ok(OptionalTransitionOutcome::Progress(psbt)) => { - println!("Proposal received. Processing..."); - return self.process_payjoin_proposal(psbt, blockchain_client).await; - } - Ok(OptionalTransitionOutcome::Stasis(current_state)) => { - println!("No response yet. Continuing polling..."); - sender = current_state; - continue; - } - Err(e) => { - break Err(Error::Generic(format!( - "Error occurred when polling for Payjoin v2 proposal: {e}" - ))); - } - } - } - } - - async fn process_payjoin_proposal( - &self, - mut psbt: Psbt, - blockchain_client: &BlockchainClient, - ) -> Result { - if !self.wallet.sign(&mut psbt, SignOptions::default())? { - return Err(Error::Generic( - "Failed to sign and finalize the Payjoin proposal PSBT.".to_string(), - )); - } - - blockchain_client - .broadcast(psbt.extract_tx_fee_rate_limit()?) - .await - } - - async fn send_payjoin_post_request( - &self, - req: payjoin::Request, - ) -> reqwest::Result { - let client = reqwest::Client::new(); - client - .post(req.url) - .header("Content-Type", req.content_type) - .body(req.body) - .send() - .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 deleted file mode 100644 index bc26466..0000000 --- a/src/payjoin/ohttp.rs +++ /dev/null @@ -1,127 +0,0 @@ -use crate::error::BDKCliError as Error; -use std::sync::{Arc, Mutex}; - -#[derive(Debug, Clone)] -pub(crate) struct RelayManager { - relays: Vec, - failed_relays: Arc>>, -} - -impl RelayManager { - pub(crate) fn new() -> Self { - Self { - relays: Vec::new(), - failed_relays: Arc::new(Mutex::new(Vec::new())), - } - } - - 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(), - )); - } - - self.relays = relays; - self.failed_relays - .lock() - .expect("Lock should not be poisoned") - .clear(); - Ok(()) - } - - 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(crate) fn choose_relay(&self) -> Result { - use payjoin::bitcoin::secp256k1::rand::prelude::SliceRandom; - - let failed_relays = self - .failed_relays - .lock() - .expect("Lock should not be poisoned"); - let remaining_relays: Vec<_> = self - .relays - .iter() - .filter(|relay| !failed_relays.contains(relay)) - .cloned() - .collect(); - - remaining_relays - .choose(&mut payjoin::bitcoin::key::rand::thread_rng()) - .cloned() - .ok_or_else(|| Error::Generic("No valid OHTTP relays available".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; - - 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()); - } -}