diff --git a/Cargo.toml b/Cargo.toml index ebda4c8a57..2193a9b5c9 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -134,6 +134,8 @@ prost = { version = "0.11.6", default-features = false, optional = true} #bitcoin-payment-instructions = { version = "0.6" } bitcoin-payment-instructions = { git = "https://github.com/jkczyz/bitcoin-payment-instructions", rev = "c359b125e972ff49b5c2e9f6865afb11500286a0", optional = true } +payjoin = { version = "1.0.0", default-features = false, features = ["v2", "io"] } + [target.'cfg(windows)'.dependencies] winapi = { version = "0.3", features = ["winbase"] } diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index 4c4c1a438a..39fee48e09 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -67,6 +67,8 @@ interface Node { Bolt12Payment bolt12_payment(); SpontaneousPayment spontaneous_payment(); OnchainPayment onchain_payment(); + [Throws=NodeError] + PayjoinPayment payjoin_payment(); Liquidity liquidity(); [Throws=NodeError] void lnurl_auth(string lnurl); @@ -137,6 +139,8 @@ interface FeeRate { u64 to_sat_per_vb_ceil(); }; +typedef interface PayjoinPayment; + typedef interface Liquidity; [Error] @@ -165,6 +169,8 @@ enum NodeError { "OnchainTxSigningFailed", "TxSyncFailed", "TxSyncTimeout", + "TxLookupFailed", + "TxLookupTimeout", "GossipUpdateFailed", "GossipUpdateTimeout", "LiquidityRequestFailed", @@ -206,6 +212,9 @@ enum NodeError { "InvalidLnurl", "ChainSourceNotSupported", "InvalidPayerProof", + "PayjoinNotConfigured", + "PayjoinSessionCreationFailed", + "PayjoinSessionFailed", }; typedef dictionary NodeStatus; diff --git a/src/builder.rs b/src/builder.rs index c6b3bd02fc..6266250b56 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -57,7 +57,7 @@ use crate::chain::ChainSource; use crate::config::BitcoindRestClientConfig; use crate::config::{ default_user_config, may_announce_channel, AnnounceError, AsyncPaymentsRole, Config, - ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig, TorConfig, + ElectrumSyncConfig, EsploraSyncConfig, HRNResolverConfig, PayjoinConfig, TorConfig, DEFAULT_ESPLORA_SERVER_URL, DEFAULT_LOG_FILENAME, DEFAULT_LOG_LEVEL, DEFAULT_MAX_PROBE_AMOUNT_MSAT, DEFAULT_MIN_PROBE_AMOUNT_MSAT, PAYMENT_CACHE_CAPACITY, PAYMENT_CACHE_WARMUP_COUNT, @@ -82,7 +82,8 @@ use crate::io::utils::{ #[cfg(feature = "storage-vss")] use crate::io::vss_store::VssStoreBuilder; use crate::io::{ - self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + self, PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE, PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, }; @@ -91,6 +92,7 @@ use crate::lnurl_auth::LnurlAuth; use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger}; use crate::message_handler::NodeCustomMessageHandler; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; +use crate::payment::payjoin::manager::PayjoinManager; #[cfg(feature = "unified-payments")] use crate::payment::HRNResolver; use crate::peer_store::PeerStore; @@ -102,8 +104,8 @@ use crate::runtime::{Runtime, RuntimeSpawner}; use crate::tx_broadcaster::TransactionBroadcaster; use crate::types::{ AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper, - GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PaymentStore, PeerManager, - PendingPaymentStore, + GossipSync, Graph, KeysManager, MessageRouter, OnionMessenger, PayjoinSessionStore, + PaymentStore, PeerManager, PendingPaymentStore, }; use crate::wallet::persist::{read_address_pool, KVStoreWalletPersister}; use crate::wallet::Wallet; @@ -230,6 +232,8 @@ pub enum BuildError { ChainTipFetchFailed, /// The configured wallet rescan height is above the current chain tip. WalletRescanHeightTooHigh, + /// The payjoin configuration requires a Bitcoin Core backend, but a different chain source was configured. + PayjoinConfigMismatch, } impl fmt::Display for BuildError { @@ -276,6 +280,9 @@ impl fmt::Display for BuildError { Self::WalletRescanHeightTooHigh => { write!(f, "Wallet rescan height is above the current chain tip.") }, + Self::PayjoinConfigMismatch => { + write!(f, "Payjoin requires a Bitcoin Core chain source, but a different one was configured.") + }, } } } @@ -664,6 +671,15 @@ impl NodeBuilder { Ok(self) } + /// Configures the [`Node`] instance to enable payjoin payments. + /// + /// The `payjoin_config` specifies the PayJoin directory and OHTTP relay URLs required + /// for payjoin V2 protocol. + pub fn set_payjoin_config(&mut self, payjoin_config: PayjoinConfig) -> &mut Self { + self.config.payjoin_config = Some(payjoin_config); + self + } + /// Sets background probing config. /// /// Use [`ProbingConfigBuilder`] to build the configuration: @@ -1272,6 +1288,14 @@ impl Builder { self.inner.write().expect("lock").set_async_payments_role(role).map(|_| ()) } + /// Configures the [`Node`] instance to enable payjoin payments. + /// + /// The `payjoin_config` specifies the PayJoin directory and OHTTP relay URLs required + /// for payjoin V2 protocol. + pub fn set_payjoin_config(&self, payjoin_config: PayjoinConfig) { + self.inner.write().expect("lock").set_payjoin_config(payjoin_config); + } + /// Configures background probing. /// /// Use [`ProbingConfigBuilder`] to build the configuration. @@ -1526,26 +1550,37 @@ fn build_with_store_internal( let kv_store_ref = Arc::clone(&kv_store); let logger_ref = Arc::clone(&logger); - let (payment_store_res, node_metris_res, pending_payment_store_res, address_pool_res) = runtime - .block_on(async move { - tokio::join!( - read_n_objects( - &*kv_store_ref, - PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - PAYMENT_CACHE_WARMUP_COUNT, - Arc::clone(&logger_ref), - ), - read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), - read_all_objects( - &*kv_store_ref, - PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, - PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, - Arc::clone(&logger_ref), - ), - read_address_pool(&*kv_store_ref, &*logger_ref) - ) - }); + let ( + payment_store_res, + node_metris_res, + pending_payment_store_res, + address_pool_res, + payjoin_session_store_res, + ) = runtime.block_on(async move { + tokio::join!( + read_n_objects( + &*kv_store_ref, + PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + PAYMENT_CACHE_WARMUP_COUNT, + Arc::clone(&logger_ref), + ), + read_node_metrics(&*kv_store_ref, Arc::clone(&logger_ref)), + read_all_objects( + &*kv_store_ref, + PENDING_PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + PENDING_PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + read_address_pool(&*kv_store_ref, &*logger_ref), + read_all_objects( + &*kv_store_ref, + PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE, + PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE, + Arc::clone(&logger_ref), + ), + ) + }); // Initialize the status fields. let node_metrics = match node_metris_res { @@ -2408,6 +2443,43 @@ fn build_with_store_internal( let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone()); + let payjoin_manager = if config.payjoin_config.is_some() { + if !matches!(chain_data_source_config, Some(ChainDataSourceConfig::Bitcoind { .. })) { + return Err(BuildError::PayjoinConfigMismatch); + } + + let payjoin_session_store = match payjoin_session_store_res { + Ok(payjoin_sessions) => Arc::new(PayjoinSessionStore::new( + payjoin_sessions, + KeepAllEntries, + PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE.to_string(), + PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE.to_string(), + Arc::clone(&kv_store), + Arc::clone(&logger), + )), + Err(e) => { + log_error!(logger, "Failed to read payjoin session data from store: {}", e); + return Err(BuildError::ReadFailed); + }, + }; + + Some(Arc::new(PayjoinManager::new( + Arc::clone(&payjoin_session_store), + Arc::clone(&logger), + Arc::clone(&config), + Arc::clone(&wallet), + Arc::clone(&fee_estimator), + Arc::clone(&chain_source), + Arc::clone(&channel_manager), + stop_sender.subscribe(), + Arc::clone(&payment_store), + Arc::clone(&pending_payment_store), + Arc::clone(&tx_broadcaster), + ))) + } else { + None + }; + let prober = probing_config.map(|probing_cfg| { let strategy: Arc = match &probing_cfg.kind { ProbingStrategyKind::HighDegree { top_node_count } => { @@ -2502,6 +2574,7 @@ fn build_with_store_internal( prober, #[cfg(cycle_tests)] _leak_checker, + payjoin_manager, }) } diff --git a/src/chain/bitcoind.rs b/src/chain/bitcoind.rs index 9ed38c2128..41ca367a99 100644 --- a/src/chain/bitcoind.rs +++ b/src/chain/bitcoind.rs @@ -34,7 +34,7 @@ use serde::Serialize; use super::{WalletSyncGuard, WalletSyncStatus}; use crate::config::{ BitcoindRestClientConfig, Config, DEFAULT_FEE_RATE_CACHE_UPDATE_TIMEOUT_SECS, - DEFAULT_TX_BROADCAST_TIMEOUT_SECS, + DEFAULT_TX_BROADCAST_TIMEOUT_SECS, DEFAULT_TX_LOOKUP_TIMEOUT_SECS, }; use crate::fee_estimator::{ apply_post_estimation_adjustments, get_all_conf_targets, get_num_block_defaults_for_target, @@ -719,6 +719,57 @@ impl BitcoindChainSource { }, } } + + pub(crate) async fn can_broadcast_transaction(&self, tx: &Transaction) -> Result { + let timeout_fut = tokio::time::timeout( + Duration::from_secs(DEFAULT_TX_LOOKUP_TIMEOUT_SECS), + self.api_client.test_mempool_accept(tx), + ); + + match timeout_fut.await { + Ok(res) => res.map_err(|e| { + log_error!( + self.logger, + "Failed to test mempool accept for transaction {}: {}", + tx.compute_txid(), + e + ); + Error::TxLookupFailed + }), + Err(e) => { + log_error!( + self.logger, + "Failed to test mempool accept for transaction {} due to timeout: {}", + tx.compute_txid(), + e + ); + log_trace!( + self.logger, + "Failed test mempool accept transaction bytes: {}", + log_bytes!(tx.encode()) + ); + Err(Error::TxLookupTimeout) + }, + } + } + + pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result, Error> { + let timeout_fut = tokio::time::timeout( + Duration::from_secs(DEFAULT_TX_LOOKUP_TIMEOUT_SECS), + self.api_client.get_raw_transaction(txid), + ); + + match timeout_fut.await { + Ok(res) => res.map_err(|e| { + log_error!(self.logger, "Failed to get transaction {}: {}", txid, e); + Error::TxLookupFailed + }), + Err(e) => { + log_error!(self.logger, "Failed to get transaction {} due to timeout: {}", txid, e); + Err(Error::TxLookupTimeout) + }, + } + } } #[derive(Clone)] @@ -1337,6 +1388,34 @@ impl BitcoindClient { .collect(); Ok(evicted_txids) } + + /// Tests whether the provided transaction would be accepted by the mempool. + pub(crate) async fn test_mempool_accept( + &self, tx: &Transaction, + ) -> Result { + match self { + BitcoindClient::Rpc { rpc_client, .. } => { + Self::test_mempool_accept_inner(Arc::clone(rpc_client), tx).await + }, + BitcoindClient::Rest { rpc_client, .. } => { + // We rely on the internal RPC client to make this call, as this + // operation is not supported by Bitcoin Core's REST interface. + Self::test_mempool_accept_inner(Arc::clone(rpc_client), tx).await + }, + } + } + + async fn test_mempool_accept_inner( + rpc_client: Arc, tx: &Transaction, + ) -> Result { + let tx_serialized = bitcoin::consensus::encode::serialize_hex(tx); + let tx_array = serde_json::json!([tx_serialized]); + + rpc_client + .call_method::("testmempoolaccept", &[tx_array]) + .await + .map(|resp| resp.0) + } } impl BlockSource for BitcoindClient { @@ -1517,6 +1596,23 @@ impl TryInto for JsonResponse { } } +pub(crate) struct TestMempoolAcceptResponse(pub bool); + +impl TryInto for JsonResponse { + type Error = String; + fn try_into(self) -> Result { + let array = + self.0.as_array().ok_or("Failed to parse testmempoolaccept response".to_string())?; + let first = + array.first().ok_or("Empty array response from testmempoolaccept".to_string())?; + let allowed = first + .get("allowed") + .and_then(|v| v.as_bool()) + .ok_or("Missing 'allowed' field in testmempoolaccept response".to_string())?; + Ok(TestMempoolAcceptResponse(allowed)) + } +} + #[derive(Debug, Clone)] pub(crate) struct MempoolEntry { /// The transaction id diff --git a/src/chain/electrum.rs b/src/chain/electrum.rs index 86025998e8..49fa398492 100644 --- a/src/chain/electrum.rs +++ b/src/chain/electrum.rs @@ -20,6 +20,7 @@ use bitcoin::transaction::Version; use bitcoin::{FeeRate, Network, Script, ScriptBuf, Transaction, Txid}; use electrum_client::{ Batch, Client as ElectrumClient, ConfigBuilder as ElectrumConfigBuilder, ElectrumApi, + Error as ElectrumError, }; use lightning::chain::{Confirm, Filter, WatchedOutput}; use lightning::util::ser::Writeable; @@ -27,8 +28,8 @@ use lightning_transaction_sync::ElectrumSyncClient; use super::{WalletSyncGuard, WalletSyncStatus}; use crate::config::{ - clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, MAX_FULL_SCAN_STOP_GAP, - MIN_FULL_SCAN_STOP_GAP, + clamp_full_scan_stop_gap, Config, ElectrumSyncConfig, DEFAULT_TX_LOOKUP_TIMEOUT_SECS, + MAX_FULL_SCAN_STOP_GAP, MIN_FULL_SCAN_STOP_GAP, }; use crate::error::Error; use crate::fee_estimator::{ @@ -387,6 +388,22 @@ impl ElectrumChainSource { }, } } + + pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result, Error> { + let electrum_client: Arc = if let Some(client) = + self.electrum_runtime_status.read().expect("lock").client().as_ref() + { + Arc::clone(client) + } else { + debug_assert!( + false, + "We should have started the chain source before getting transactions" + ); + return Err(Error::TxLookupFailed); + }; + + electrum_client.get_transaction(txid).await + } } impl Filter for ElectrumChainSource { @@ -823,6 +840,37 @@ impl ElectrumRuntimeClient { Ok(new_fee_rate_cache) } + + async fn get_transaction(&self, txid: &Txid) -> Result, Error> { + let electrum_client = Arc::clone(&self.electrum_client); + let txid_copy = *txid; + + let spawn_fut = + self.runtime.spawn_blocking(move || electrum_client.transaction_get(&txid_copy)); + let timeout_fut = + tokio::time::timeout(Duration::from_secs(DEFAULT_TX_LOOKUP_TIMEOUT_SECS), spawn_fut); + + match timeout_fut.await { + Ok(res) => match res { + Ok(inner_res) => match inner_res { + Ok(tx) => Ok(Some(tx)), + Err(ElectrumError::Protocol(_)) => Ok(None), + Err(e) => { + log_error!(self.logger, "Failed to get transaction {}: {}", txid, e); + Err(Error::TxLookupFailed) + }, + }, + Err(e) => { + log_error!(self.logger, "Failed to get transaction {}: {}", txid, e); + Err(Error::TxLookupFailed) + }, + }, + Err(e) => { + log_error!(self.logger, "Failed to get transaction {} due to timeout: {}", txid, e); + Err(Error::TxLookupTimeout) + }, + } + } } struct ConfirmGate { diff --git a/src/chain/esplora.rs b/src/chain/esplora.rs index 1c13f141fb..4c72893bd6 100644 --- a/src/chain/esplora.rs +++ b/src/chain/esplora.rs @@ -12,7 +12,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use bdk_esplora::EsploraAsyncExt; use bitcoin::transaction::Version; -use bitcoin::{FeeRate, Network, Script, Txid}; +use bitcoin::{FeeRate, Network, Script, Transaction, Txid}; use esplora_client::AsyncClient as EsploraAsyncClient; use lightning::chain::{Confirm, Filter, WatchedOutput}; use lightning::util::ser::Writeable; @@ -528,6 +528,13 @@ impl EsploraChainSource { }, } } + + pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result, Error> { + self.esplora_client.get_tx(txid).await.map_err(|e| { + log_error!(self.logger, "Failed to get transaction {}: {}", txid, e); + Error::TxLookupFailed + }) + } } impl Filter for EsploraChainSource { diff --git a/src/chain/mod.rs b/src/chain/mod.rs index f01c1c8cb8..b7608cfc1c 100644 --- a/src/chain/mod.rs +++ b/src/chain/mod.rs @@ -18,7 +18,7 @@ use std::collections::HashSet; use std::sync::{Arc, Mutex}; use std::time::Duration; -use bitcoin::{Script, Txid}; +use bitcoin::{Script, Transaction, Txid}; use lightning::chain::{BlockLocator, Filter}; #[cfg(feature = "chain-bitcoind")] @@ -610,6 +610,26 @@ impl ChainSource { } } } + + pub(crate) async fn can_broadcast_transaction(&self, tx: &Transaction) -> Result { + match &self.kind { + ChainSourceKind::Bitcoind(bitcoind_chain_source) => { + bitcoind_chain_source.can_broadcast_transaction(tx).await + }, + ChainSourceKind::Esplora { .. } | ChainSourceKind::Electrum { .. } => { + // Neither supports a `testmempoolaccept` equivalent. + Err(Error::ChainSourceNotSupported) + }, + } + } + + pub(crate) async fn get_transaction(&self, txid: &Txid) -> Result, Error> { + match &self.kind { + ChainSourceKind::Bitcoind(bitcoind) => bitcoind.get_transaction(txid).await, + ChainSourceKind::Esplora(esplora) => esplora.get_transaction(txid).await, + ChainSourceKind::Electrum(electrum) => electrum.get_transaction(txid).await, + } + } } impl Filter for ChainSource { diff --git a/src/config.rs b/src/config.rs index 65e256117a..0aebe11d11 100644 --- a/src/config.rs +++ b/src/config.rs @@ -169,6 +169,18 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY: Duration = Duration::f // thereafter until every configured LSP has been discovered. pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_secs(60 * 60); +// The time interval at which we resume persisted payjoin sessions. +pub(crate) const PAYJOIN_RESUME_INTERVAL: Duration = Duration::from_secs(15); + +// The duration after which completed or failed payjoin sessions are cleaned up (24 hours). +pub(crate) const PAYJOIN_SESSION_CLEANUP_AGE_SECS: u64 = 24 * 60 * 60; + +// The interval at which we check for old payjoin sessions to clean up (1 hour). +pub(crate) const PAYJOIN_SESSION_CLEANUP_INTERVAL: Duration = Duration::from_secs(60 * 60); + +// The default timeout after which we abort a transaction lookup operation. +pub(crate) const DEFAULT_TX_LOOKUP_TIMEOUT_SECS: u64 = 10; + #[derive(Debug, Clone)] #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] /// Represents the configuration of an [`Node`] instance. @@ -191,7 +203,8 @@ pub(crate) const LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY: Duration = Duration::from_ feature = "unified-payments", doc = "| `hrn_config` | HumanReadableNamesConfig::default() |" )] -/// | `manually_handle_unknown_bolt11_payments` | false | +/// | `manually_handle_unknown_bolt11_payments` | false | +/// | `payjoin_config` | None | /// /// See [`AnchorChannelsConfig`] and [`RouteParametersConfig`] for more information regarding their /// respective default values. @@ -268,6 +281,8 @@ pub struct Config { /// /// [`Event::PaymentClaimable`]: crate::Event::PaymentClaimable pub manually_handle_unknown_bolt11_payments: bool, + /// Configuration options for PayJoin payments. + pub payjoin_config: Option, } impl Default for Config { @@ -286,6 +301,7 @@ impl Default for Config { #[cfg(feature = "unified-payments")] hrn_config: HumanReadableNamesConfig::default(), manually_handle_unknown_bolt11_payments: false, + payjoin_config: None, } } } @@ -829,6 +845,16 @@ pub enum AsyncPaymentsRole { Server, } +/// Configuration options for PayJoin payments. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct PayjoinConfig { + /// The URL of the PayJoin directory + pub payjoin_directory: String, + /// The URLs of the OHTTP relays to use for sending OHTTP requests to PayJoin receivers. + pub ohttp_relays: Vec, +} + #[cfg(test)] mod tests { use std::str::FromStr; diff --git a/src/error.rs b/src/error.rs index 485f944c25..aef04b7f06 100644 --- a/src/error.rs +++ b/src/error.rs @@ -65,6 +65,10 @@ pub enum Error { TxSyncFailed, /// A transaction sync operation timed out. TxSyncTimeout, + /// A transaction lookup operation failed. + TxLookupFailed, + /// A transaction lookup operation timed out. + TxLookupTimeout, /// A gossip updating operation failed. GossipUpdateFailed, /// A gossip updating operation timed out. @@ -147,6 +151,12 @@ pub enum Error { ChainSourceNotSupported, /// The provided payer proof is invalid. InvalidPayerProof, + /// Payjoin is not configured. + PayjoinNotConfigured, + /// Payjoin session creation failed. + PayjoinSessionCreationFailed, + /// Payjoin session failed. + PayjoinSessionFailed, } impl fmt::Display for Error { @@ -182,6 +192,8 @@ impl fmt::Display for Error { Self::OnchainTxSigningFailed => write!(f, "Failed to sign given transaction."), Self::TxSyncFailed => write!(f, "Failed to sync transactions."), Self::TxSyncTimeout => write!(f, "Syncing transactions timed out."), + Self::TxLookupFailed => write!(f, "Failed to look up transaction."), + Self::TxLookupTimeout => write!(f, "Transaction lookup timed out."), Self::GossipUpdateFailed => write!(f, "Failed to update gossip data."), Self::GossipUpdateTimeout => write!(f, "Updating gossip data timed out."), Self::LiquidityRequestFailed => write!(f, "Failed to request inbound liquidity."), @@ -239,6 +251,9 @@ impl fmt::Display for Error { write!(f, "The configured chain source is not supported.") }, Self::InvalidPayerProof => write!(f, "The provided payer proof is invalid."), + Self::PayjoinNotConfigured => write!(f, "Payjoin is not configured."), + Self::PayjoinSessionCreationFailed => write!(f, "Payjoin session creation failed."), + Self::PayjoinSessionFailed => write!(f, "Payjoin session failed."), } } } diff --git a/src/io/mod.rs b/src/io/mod.rs index c11475c431..f6c6a96083 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -93,3 +93,7 @@ pub(crate) const BDK_WALLET_ADDRESS_POOL_KEY: &str = "address_pool"; /// /// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice pub(crate) const STATIC_INVOICE_STORE_PRIMARY_NAMESPACE: &str = "static_invoices"; + +/// The payjoin sessions will be persisted under this key. +pub(crate) const PAYJOIN_SESSION_STORE_PRIMARY_NAMESPACE: &str = "payjoin_sessions"; +pub(crate) const PAYJOIN_SESSION_STORE_SECONDARY_NAMESPACE: &str = ""; diff --git a/src/lib.rs b/src/lib.rs index 9ce5a273e7..39487ae29c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -134,8 +134,8 @@ pub use builder::{BuildError, Builder}; use chain::ChainSource; use config::{ default_user_config, may_announce_channel, AsyncPaymentsRole, ChannelConfig, Config, - LNURL_AUTH_TIMEOUT_SECS, NODE_ANN_BCAST_INTERVAL, PEER_RECONNECTION_INTERVAL, - RGS_SYNC_INTERVAL, + LNURL_AUTH_TIMEOUT_SECS, NODE_ANN_BCAST_INTERVAL, PAYJOIN_RESUME_INTERVAL, + PAYJOIN_SESSION_CLEANUP_INTERVAL, PEER_RECONNECTION_INTERVAL, RGS_SYNC_INTERVAL, }; use connection::ConnectionManager; pub use error::Error as NodeError; @@ -200,6 +200,8 @@ pub use vss_client; use crate::config::{LIQUIDITY_DISCOVERY_RETRY_INITIAL_DELAY, LIQUIDITY_DISCOVERY_RETRY_MAX_DELAY}; use crate::ffi::{maybe_deref, maybe_wrap}; use crate::liquidity::Liquidity; +use crate::payment::payjoin::manager::PayjoinManager; +use crate::payment::PayjoinPayment; use crate::scoring::setup_background_pathfinding_scores_sync; use crate::wallet::FundingAmount; @@ -281,6 +283,7 @@ pub struct Node { prober: Option>, #[cfg(cycle_tests)] _leak_checker: LeakChecker, + payjoin_manager: Option>, } impl Node { @@ -836,6 +839,52 @@ impl Node { } }); + if let Some(payjoin_manager) = self.payjoin_manager.as_ref() { + // Periodically resume payjoin sessions. + let resume_payjoin_manager = Arc::clone(payjoin_manager); + let resume_logger = Arc::clone(&self.logger); + let mut stop_resume = self.stop_sender.subscribe(); + self.runtime.spawn_cancellable_background_task(async move { + let mut interval = tokio::time::interval(PAYJOIN_RESUME_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_resume.changed() => { + log_debug!(resume_logger, "Stopping payjoin session resume task."); + return; + } + _ = interval.tick() => { + if let Err(e) = resume_payjoin_manager.resume_payjoin_sessions().await { + log_error!(resume_logger, "Failed to resume payjoin sessions: {:?}", e); + } + } + } + } + }); + + // Periodically clean up old completed/failed payjoin sessions. + let cleanup_payjoin_manager = Arc::clone(payjoin_manager); + let cleanup_logger = Arc::clone(&self.logger); + let mut stop_cleanup = self.stop_sender.subscribe(); + self.runtime.spawn_cancellable_background_task(async move { + let mut interval = tokio::time::interval(PAYJOIN_SESSION_CLEANUP_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _ = stop_cleanup.changed() => { + log_debug!(cleanup_logger, "Stopping payjoin session cleanup task."); + return; + } + _ = interval.tick() => { + if let Err(e) = cleanup_payjoin_manager.cleanup_old_sessions().await { + log_error!(cleanup_logger, "Failed to cleanup old payjoin sessions: {:?}", e); + } + } + } + } + }); + } + log_info!(self.logger, "Startup complete."); *is_running_lock = true; Ok(()) @@ -1173,6 +1222,24 @@ impl Node { self.hrn_resolver.clone(), ) } + + /// Returns a payment handler allowing to send and receive [Payjoin] payments. + /// + /// [Payjoin]: https://payjoin.org + #[cfg(not(feature = "uniffi"))] + pub fn payjoin_payment(&self) -> Result { + let manager = self.payjoin_manager.as_ref().ok_or(Error::PayjoinNotConfigured)?; + Ok(PayjoinPayment::new(Arc::clone(manager), Arc::clone(&self.is_running))) + } + + /// Returns a payment handler allowing to send and receive [Payjoin] payments. + /// + /// [Payjoin]: https://payjoin.org + #[cfg(feature = "uniffi")] + pub fn payjoin_payment(&self) -> Result, Error> { + let manager = self.payjoin_manager.as_ref().ok_or(Error::PayjoinNotConfigured)?; + Ok(Arc::new(PayjoinPayment::new(Arc::clone(manager), Arc::clone(&self.is_running)))) + } } #[cfg(all(feature = "unified-payments", feature = "uniffi"))] diff --git a/src/payment/mod.rs b/src/payment/mod.rs index 13dbe51063..b748135a26 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -13,6 +13,7 @@ mod bolt12; #[cfg(feature = "unified-payments")] mod hrn; mod onchain; +pub(crate) mod payjoin; pub(crate) mod pending_payment_store; mod spontaneous; pub(crate) mod store; @@ -25,6 +26,7 @@ pub use bolt12::{Bolt12Payment, PayerProofOptions}; #[cfg(feature = "unified-payments")] pub(crate) use hrn::HRNResolver; pub use onchain::OnchainPayment; +pub use payjoin::PayjoinPayment; pub(crate) use pending_payment_store::{FundingTxCandidate, PendingPaymentDetails}; pub use spontaneous::SpontaneousPayment; pub use store::{ diff --git a/src/payment/payjoin/manager.rs b/src/payment/payjoin/manager.rs new file mode 100644 index 0000000000..380b4bfab2 --- /dev/null +++ b/src/payment/payjoin/manager.rs @@ -0,0 +1,903 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::{psbt::Input, Amount, FeeRate, TxIn}; +use lightning::ln::channelmanager::PaymentId; +use lightning::log_warn; +use payjoin::io::fetch_ohttp_keys; +use payjoin::persist::OptionalTransitionOutcome; +use payjoin::receive::v2::{ + replay_event_log_async as replay_receiver_event_log_async, HasReplyableError, Initialized, + MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, PayjoinProposal, + ProvisionalProposal, ReceiveSession, Receiver, ReceiverBuilder, SessionOutcome, + UncheckedOriginalPayload, WantsFeeRange, WantsInputs, WantsOutputs, +}; +use payjoin::receive::InputPair; +use payjoin::ImplementationError; + +use crate::chain::ChainSource; +use crate::config::{Config, PayjoinConfig, PAYJOIN_SESSION_CLEANUP_AGE_SECS}; +use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator}; +use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger}; +use crate::payment::payjoin::payjoin_session::{PayjoinDirection, PayjoinSession, PayjoinStatus}; +use crate::payment::payjoin::persist::KVStorePayjoinReceiverPersister; +use crate::payment::{ + ConfirmationStatus, PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, + PendingPaymentDetails, TransactionType, +}; +use crate::total_anchor_channels_reserve_sats; +use crate::types::{ + Broadcaster, ChannelManager, PayjoinSessionStore, PaymentStore, PendingPaymentStore, +}; +use crate::wallet::Wallet; +use crate::Error; + +#[derive(Clone)] +pub(crate) struct PayjoinManager { + payjoin_session_store: Arc, + logger: Arc, + config: Arc, + wallet: Arc, + fee_estimator: Arc, + chain_source: Arc, + channel_manager: Arc, + stop_receiver: tokio::sync::watch::Receiver<()>, + payment_store: Arc, + pending_payment_store: Arc, + broadcaster: Arc, +} + +impl PayjoinManager { + pub(crate) fn new( + payjoin_session_store: Arc, logger: Arc, config: Arc, + wallet: Arc, fee_estimator: Arc, + chain_source: Arc, channel_manager: Arc, + stop_receiver: tokio::sync::watch::Receiver<()>, payment_store: Arc, + pending_payment_store: Arc, broadcaster: Arc, + ) -> Self { + Self { + payjoin_session_store, + logger, + config, + wallet, + fee_estimator, + chain_source, + channel_manager, + stop_receiver, + payment_store, + pending_payment_store, + broadcaster, + } + } + + pub(crate) async fn receive_payjoin( + &self, amount_sats: u64, fee_rate: Option, + ) -> Result { + let payjoin_config = + self.config.payjoin_config.as_ref().ok_or(Error::PayjoinNotConfigured)?; + + if payjoin_config.ohttp_relays.is_empty() { + log_error!(self.logger, "No OHTTP relays configured."); + return Err(Error::PayjoinNotConfigured); + } + + // Generate a new session ID + let mut random_bytes = [0u8; 32]; + getrandom::fill(&mut random_bytes).map_err(|e| { + log_error!(self.logger, "Failed to generate random session ID: {}", e); + Error::PayjoinSessionCreationFailed + })?; + let session_id = PaymentId(random_bytes); + + let confirmation_target = ConfirmationTarget::OnchainPayment; + let fee_rate = + fee_rate.unwrap_or_else(|| self.fee_estimator.estimate_fee_rate(confirmation_target)); + + let address = self.wallet.get_new_address().await?; + let ohttp_keys = { + let mut result = Err(Error::ConnectionFailed); + for relay in self.relay_order(payjoin_config)? { + match fetch_ohttp_keys(relay, payjoin_config.payjoin_directory.as_str()).await { + Ok(keys) => { + result = Ok(keys); + break; + }, + Err(e) => { + log_error!( + self.logger, + "Failed to fetch OHTTP keys via {}: {}. Trying next relay.", + relay, + e + ); + }, + } + } + result + }?; + log_debug!(self.logger, "Fetched OHTTP keys: {:?}", ohttp_keys); + + let amount = Amount::from_sat(amount_sats); + + // Create a new persister for this session + let persister = KVStorePayjoinReceiverPersister::new( + session_id, + Some(amount_sats * 1000), + Arc::clone(&self.payjoin_session_store), + fee_rate.to_sat_per_kwu(), + None, + None, + None, + ) + .await?; + + let session = + ReceiverBuilder::new(address, payjoin_config.payjoin_directory.as_str(), ohttp_keys) + .map_err(|e| { + log_error!(self.logger, "Failed to create receiver builder: {}", e); + Error::PayjoinSessionCreationFailed + })? + .with_amount(amount) + .with_max_fee_rate(fee_rate) + .build() + .save_async(&persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + + log_info!(self.logger, "Receive session established"); + let pj_uri = session.pj_uri(); + log_info!(self.logger, "Request Payjoin by sharing this Payjoin Uri: {}", pj_uri); + + Ok(pj_uri.to_string()) + } + + fn relay_order<'a>(&self, payjoin_config: &'a PayjoinConfig) -> Result, Error> { + let count = payjoin_config.ohttp_relays.len(); + let start = if count > 0 { + let mut bytes = [0u8; 8]; + getrandom::fill(&mut bytes).map_err(|e| { + log_error!(self.logger, "Failed to generate random relay index: {}", e); + Error::PayjoinSessionFailed + })?; + u64::from_ne_bytes(bytes) as usize % count + } else { + 0 + }; + Ok((0..count).map(|i| payjoin_config.ohttp_relays[(start + i) % count].as_str()).collect()) + } + + async fn process_receiver_session( + &self, session: ReceiveSession, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + match session { + ReceiveSession::Initialized(proposal) => { + self.read_from_directory(proposal, persister).await + }, + ReceiveSession::UncheckedOriginalPayload(proposal) => { + self.check_proposal(proposal, persister).await + }, + ReceiveSession::MaybeInputsOwned(proposal) => { + self.check_inputs_not_owned(proposal, persister).await + }, + ReceiveSession::MaybeInputsSeen(proposal) => { + self.check_no_inputs_seen_before(proposal, persister).await + }, + ReceiveSession::OutputsUnknown(proposal) => { + self.identify_receiver_outputs(proposal, persister).await + }, + ReceiveSession::WantsOutputs(proposal) => { + self.commit_outputs(proposal, persister).await + }, + ReceiveSession::WantsInputs(proposal) => { + self.contribute_inputs(proposal, persister).await + }, + ReceiveSession::WantsFeeRange(proposal) => { + self.apply_fee_range(proposal, persister).await + }, + ReceiveSession::ProvisionalProposal(proposal) => { + self.finalize_proposal(proposal, persister).await + }, + ReceiveSession::PayjoinProposal(proposal) => { + self.send_payjoin_proposal(proposal, persister).await + }, + ReceiveSession::HasReplyableError(error) => self.handle_error(error, persister).await, + ReceiveSession::Monitor(proposal) => { + self.monitor_payjoin_proposal(proposal, persister).await + }, + ReceiveSession::PendingFallback(fallback_tx) => { + let payjoin_session = + persister.get_session().await.ok_or(Error::InvalidPaymentId)?; + self.close_session_with_fallback(&mut payjoin_session).await; + Ok(()) + }, + ReceiveSession::Closed(outcome) => self.handle_closed_session(outcome, persister).await, + } + } + + async fn read_from_directory( + &self, session: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let mut interrupt = self.stop_receiver.clone(); + let receiver = tokio::select! { + res = self.long_poll_fallback(session, persister) => res, + _ = interrupt.changed() => { + log_info!(self.logger, "Session interrupted by node shutdown. Will resume on restart."); + return Err(Error::NotRunning); + } + }?; + self.check_proposal(receiver, persister).await + } + + async fn long_poll_fallback( + &self, mut session: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result, Error> { + let payjoin_config = + self.config.payjoin_config.as_ref().ok_or(Error::PayjoinNotConfigured)?; + + loop { + let (ohttp_response, context) = { + let mut result = Err(Error::ConnectionFailed); + for relay in self.relay_order(payjoin_config)? { + let (req, ctx) = match session.create_poll_request(relay) { + Ok(r) => r, + Err(e) => { + // create_poll_request is fatal. No fallback tx exists yet at this stage + // (sender hasn't sent their PSBT), so we just fail the session. + log_error!(self.logger, "Failed to create poll request: {}", e); + result = Err(Error::PayjoinSessionFailed); + self.handle_closed_session(SessionOutcome::Failure, persister).await?; + break; + }, + }; + match self.post_request(req).await { + Ok(resp) => { + result = Ok((resp, ctx)); + break; + }, + Err(e) => { + log_error!(self.logger, "Polling failed via relay, trying next: {}", e); + }, + } + } + result + }?; + log_debug!(self.logger, "Polling receive request..."); + let state_transition = session + .process_response(ohttp_response.as_bytes(), context) + .save_async(persister) + .await; + match state_transition { + Ok(OptionalTransitionOutcome::Progress(next_state)) => { + log_info!( + self.logger, + "Got a request from the sender. Responding with a Payjoin proposal." + ); + return Ok(next_state); + }, + Ok(OptionalTransitionOutcome::Stasis(current_state)) => { + session = current_state; + continue; + }, + Err(_) => return Err(Error::PersistenceFailed), + } + } + } + + async fn post_request(&self, req: payjoin::Request) -> Result { + bitreq::post(req.url) + .with_header("Content-Type", req.content_type) + .with_body(req.body) + .send_async() + .await + .map_err(|e| { + log_error!(self.logger, "HTTP request failed: {}", e); + Error::ConnectionFailed + }) + } + + async fn check_proposal( + &self, proposal: Receiver, + persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + // Note: broadcast suitability can only be verified when using the BitcoindRpc backend. + // For Esplora and Electrum backends this check will return an error as they do not + // support a testmempoolaccept equivalent. + let proposal = proposal + .check_broadcast_suitability(None, |tx| { + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current() + .block_on(self.chain_source.can_broadcast_transaction(tx)) + }) + .map_err(|e| ImplementationError::from(e.to_string().as_str())) + }) + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + + // If the payjoin fails or times out, broadcast this fallback tx to ensure the receiver still gets paid. + let fallback_tx = proposal.extract_tx_to_schedule_broadcast(); + + let session_id = persister.session_id(); + let mut session = + self.payjoin_session_store.get(&session_id).await?.ok_or(Error::InvalidPaymentId)?; + + session.fallback_tx = Some(fallback_tx); + self.payjoin_session_store.insert_or_update(session).await?; + + log_info!( + self.logger, + "Fallback transaction received. This will be broadcast if the Payjoin fails" + ); + self.check_inputs_not_owned(proposal, persister).await + } + + async fn check_inputs_not_owned( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let proposal = proposal + .check_inputs_not_owned(&mut |input| { + self.wallet + .is_mine(input.to_owned()) + .map_err(|e| ImplementationError::from(e.to_string().as_str())) + }) + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + + self.check_no_inputs_seen_before(proposal, persister).await + } + + async fn check_no_inputs_seen_before( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let mut inputs_seen = + persister.get_session().await.ok_or(Error::InvalidPaymentId)?.inputs_seen; + let mut newly_seen = Vec::new(); + + let transition = proposal.check_no_inputs_seen_before(&mut |input| { + if inputs_seen.contains(input) { + return Ok(true); + } + inputs_seen.push(*input); + newly_seen.push(*input); + Ok(false) + }); + + persister.insert_inputs_seen(newly_seen).await?; + + let proposal = + transition.save_async(persister).await.map_err(|_| Error::PersistenceFailed)?; + + self.identify_receiver_outputs(proposal, persister).await + } + + async fn identify_receiver_outputs( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let proposal = proposal + .identify_receiver_outputs(&mut |output_script| { + self.wallet + .is_mine(output_script.to_owned()) + .map_err(|e| ImplementationError::from(e.to_string().as_str())) + }) + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + self.commit_outputs(proposal, persister).await + } + + async fn commit_outputs( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let proposal = proposal + .commit_outputs() + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + self.contribute_inputs(proposal, persister).await + } + + async fn contribute_inputs( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + // Check wallet has spendable funds after accounting for anchor reserve + let cur_anchor_reserve_sats = + total_anchor_channels_reserve_sats(&self.channel_manager, &self.config); + let spendable_amount_sats = + self.wallet.get_spendable_amount_sats(cur_anchor_reserve_sats).unwrap_or(0); + + if spendable_amount_sats == 0 { + log_error!( + self.logger, + "No spendable funds available after anchor reserve. Cannot contribute inputs to payjoin." + ); + return Err(Error::InsufficientFunds); + } + + let candidate_inputs = self.list_input_pairs()?; + + if candidate_inputs.is_empty() { + log_error!( + self.logger, + "No spendable UTXOs available in wallet. Cannot contribute inputs to payjoin." + ); + return Err(Error::InsufficientFunds); + } + + let selected_input = proposal.try_preserving_privacy(candidate_inputs).map_err(|e| { + log_error!(self.logger, "Failed to select input for payjoin contribution: {}", e); + Error::PayjoinSessionFailed + })?; + let proposal = proposal + .contribute_inputs(vec![selected_input]) + .map_err(|e| { + log_error!(self.logger, "Failed to contribute inputs to payjoin: {}", e); + Error::PayjoinSessionFailed + })? + .commit_inputs() + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + self.apply_fee_range(proposal, persister).await + } + + fn list_input_pairs(&self) -> Result, Error> { + let unspent = self.wallet.list_unspent_confirmed_utxos()?; + + let mut input_pairs = Vec::with_capacity(unspent.len()); + + for u in unspent { + let txin = TxIn { previous_output: u.outpoint, ..Default::default() }; + let psbtin = Input { witness_utxo: Some(u.output.clone()), ..Default::default() }; + + let input_pair = InputPair::new(txin, psbtin, None).map_err(|e| { + log_error!(self.logger, "Failed to create InputPair: {}", e); + Error::PayjoinSessionFailed + })?; + + input_pairs.push(input_pair); + } + + Ok(input_pairs) + } + + async fn apply_fee_range( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let session = persister.get_session().await.ok_or(Error::InvalidPaymentId)?; + let fee_rate = FeeRate::from_sat_per_kwu(session.fee_rate_kwu); + + let proposal = proposal + .apply_fee_range(None, Some(fee_rate)) + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + + self.finalize_proposal(proposal, persister).await + } + + async fn finalize_proposal( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let proposal = proposal + .finalize_proposal(|psbt| { + self.wallet + .process_psbt(psbt.clone()) + .map_err(|e| ImplementationError::from(e.to_string().as_str())) + }) + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + self.send_payjoin_proposal(proposal, persister).await + } + + async fn send_payjoin_proposal( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let payjoin_config = + self.config.payjoin_config.as_ref().ok_or(Error::PayjoinNotConfigured)?; + + let (res, ohttp_ctx) = { + let mut result = Err(Error::ConnectionFailed); + for relay in self.relay_order(payjoin_config)? { + let (req, ctx) = match proposal.create_post_request(relay) { + Ok(r) => r, + Err(e) => { + // create_post_request is fatal, so we trigger a session failure + // immediately and broadcast the fallback transaction. + log_error!(self.logger, "v2 req extraction failed {}", e); + result = Err(Error::PayjoinSessionFailed); + self.handle_closed_session(SessionOutcome::Failure, persister).await?; + break; + }, + }; + match self.post_request(req).await { + Ok(resp) => { + result = Ok((resp, ctx)); + break; + }, + Err(e) => { + log_error!( + self.logger, + "Proposal send failed via relay, trying next: {}", + e + ); + }, + } + } + result + }?; + let payjoin_psbt = proposal.psbt().clone(); + let session = proposal + .process_response(res.as_bytes(), ohttp_ctx) + .save_async(persister) + .await + .map_err(|_| Error::PersistenceFailed)?; + + // At this point we will persist the fee and txid to the session store + let fee = payjoin_psbt.fee().unwrap_or(Amount::ZERO); + let fee_sat = fee.to_sat(); + let txid = payjoin_psbt.extract_tx_unchecked_fee_rate().compute_txid(); + + let session_id = persister.session_id(); + let mut payjoin_session = + self.payjoin_session_store.get(&session_id).await?.ok_or(Error::InvalidPaymentId)?; + + payjoin_session.fee_paid_msat = Some(fee_sat * 1000); + payjoin_session.txid = Some(txid); + self.payjoin_session_store.insert_or_update(payjoin_session).await.map_err(|e| { + log_error!( + self.logger, + "Failed to update payjoin session for {:?}: {:?}", + session_id, + e + ); + Error::PersistenceFailed + })?; + + log_info!( + self.logger, + "Response successful. Watch mempool for successful Payjoin. TXID: {}", + txid + ); + + self.monitor_payjoin_proposal(session, persister).await + } + + async fn handle_error( + &self, session: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let payjoin_config = + self.config.payjoin_config.as_ref().ok_or(Error::PayjoinNotConfigured)?; + + let (err_response, err_ctx) = { + let mut result = Err(Error::ConnectionFailed); + for relay in self.relay_order(payjoin_config)? { + let (req, ctx) = match session.create_error_request(relay) { + Ok(r) => r, + Err(_) => { + result = Err(Error::PayjoinSessionFailed); + break; + }, + }; + match self.post_request(req).await { + Ok(resp) => { + result = Ok((resp, ctx)); + break; + }, + Err(e) => { + log_error!( + self.logger, + "Error response send failed via relay, trying next: {}", + e + ); + }, + } + } + result + }?; + + let err_bytes = err_response.as_bytes(); + + if session.process_error_response(err_bytes, err_ctx).save_async(persister).await.is_err() { + return Err(Error::PersistenceFailed); + } + + Ok(()) + } + + async fn handle_closed_session( + &self, outcome: SessionOutcome, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + let session_id = persister.session_id(); + let mut session = + self.payjoin_session_store.get(&session_id).await?.ok_or(Error::InvalidPaymentId)?; + + match outcome { + SessionOutcome::Success(txid) => { + log_info!( + self.logger, + "Payjoin session detected in the mempool and completed successfully." + ); + let kind = PaymentKind::Onchain { + txid, + tx_type: Some(TransactionType::Payjoin), + status: ConfirmationStatus::Unconfirmed, + }; + + let payment = PaymentDetails::new( + session_id, + kind, + session.amount_msat, + session.fee_paid_msat, + PaymentDirection::Inbound, + PaymentStatus::Pending, + ); + + self.payment_store.insert_or_update(payment.clone()).await?; + + let pending_payment = PendingPaymentDetails::new(payment, Vec::new(), Vec::new()); + self.pending_payment_store.insert_or_update(pending_payment).await?; + + session.status = PayjoinStatus::Completed; + session.completed_at = Some( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(), + ); + self.payjoin_session_store.insert_or_update(session).await?; + }, + SessionOutcome::PayjoinProposalSent => { + log_info!( + self.logger, + "Payjoin proposal sent. Cannot track broadcast due to non-SegWit sender inputs." + ); + session.status = PayjoinStatus::Completed; + session.completed_at = Some( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(), + ); + self.payjoin_session_store.insert_or_update(session).await?; + }, + SessionOutcome::FallbackBroadcasted => { + log_info!(self.logger, "Payjoin failed. Fallback transaction was broadcasted."); + session.status = PayjoinStatus::Failed; + self.payjoin_session_store.insert_or_update(session).await?; + }, + SessionOutcome::Cancel => { + log_info!(self.logger, "Payjoin session was cancelled."); + self.close_session_with_fallback(&mut session).await; + }, + SessionOutcome::Failure => { + log_error!(self.logger, "Payjoin session failed due to a protocol error."); + self.close_session_with_fallback(&mut session).await; + }, + } + Ok(()) + } + + async fn monitor_payjoin_proposal( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + // On a session resumption, the receiver will resume again in this state. + let poll_interval = tokio::time::Duration::from_secs(2); + + let timeout_duration = tokio::time::Duration::from_secs(10); + + let mut interval = tokio::time::interval(poll_interval); + interval.tick().await; + + log_debug!(self.logger, "Polling for payjoin transaction in the mempool..."); + + let result = tokio::time::timeout(timeout_duration, async { + loop { + interval.tick().await; + let check_result = proposal + .check_payment(|txid| { + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current() + .block_on(self.chain_source.get_transaction(&txid)) + }) + .map_err(|e| ImplementationError::from(e.to_string().as_str())) + }) + .save_async(persister) + .await; + + match check_result { + Ok(_) => { + log_info!(self.logger, "Payjoin transaction detected in the mempool!"); + return Ok(()); + }, + Err(_) => { + continue; + }, + } + } + }) + .await; + + match result { + Ok(ok) => ok, + Err(_) => { + log_debug!( + self.logger, + "Payjoin transaction not yet seen after {:?}. Will retry on next background tick.", + timeout_duration + ); + Ok(()) + }, + } + } + + pub(crate) async fn resume_payjoin_sessions(&self) -> Result<(), Error> { + let recv_session_ids = self + .payjoin_session_store + .list_filter(|p| { + p.direction == PayjoinDirection::Receive && p.status == PayjoinStatus::Active + }) + .await + .into_iter() + .map(|s| s.session_id) + .collect::>(); + + if recv_session_ids.is_empty() { + log_debug!(self.logger, "No sessions to resume."); + return Ok(()); + } + + let mut join_set: tokio::task::JoinSet> = tokio::task::JoinSet::new(); + + // Process receiver sessions + for session_id in recv_session_ids { + let self_clone = self.clone(); + // Create a persister for this session + let recv_persister = match KVStorePayjoinReceiverPersister::from_session( + session_id, + Arc::clone(&self.payjoin_session_store), + ) + .await + { + Ok(p) => p, + Err(e) => { + log_error!( + self.logger, + "Failed to create persister for session {:?}: {:?}", + session_id, + e + ); + continue; + }, + }; + + match replay_receiver_event_log_async(&recv_persister).await { + Ok((receiver_state, _)) => { + join_set.spawn(async move { + self_clone.process_receiver_session(receiver_state, &recv_persister).await + }); + }, + Err(e) => { + log_error!( + self.logger, + "An error {:?} occurred while replaying receiver session", + e + ); + match self.payjoin_session_store.get(&session_id).await { + Ok(Some(mut session)) => { + self.close_session_with_fallback(&mut session).await; + }, + Ok(None) => { + log_error!( + self.logger, + "Payjoin session {} disappeared before it could be closed.", + session_id + ); + }, + Err(e) => { + log_error!( + self.logger, + "Failed to read payjoin session {} while closing it: {:?}", + session_id, + e + ); + }, + } + }, + } + } + + let mut interrupt = self.stop_receiver.clone(); + tokio::select! { + _ = async { + while let Some(result) = join_set.join_next().await { + match result { + Ok(Ok(())) => log_info!(self.logger, "A payjoin session completed successfully."), + Ok(Err(e)) => log_error!(self.logger, "A payjoin session failed: {:?}", e), + Err(e) => log_error!(self.logger, "A payjoin session task panicked: {:?}", e), + } + } + } => { + log_info!(self.logger, "All payjoin resumed sessions completed."); + } + _ = interrupt.changed() => { + join_set.abort_all(); + log_info!(self.logger, "Resumed payjoin sessions were interrupted."); + } + } + Ok(()) + } + + async fn close_session_with_fallback(&self, session: &mut PayjoinSession) { + session.status = PayjoinStatus::Failed; + + if let Some(fallback_tx) = session.fallback_tx.as_ref() { + self.broadcaster.broadcast_unclassified_transaction(fallback_tx.clone()); + } else { + log_warn!( + self.logger, + "Payjoin session {} missing fallback transaction; closing as Failed without broadcast.", + session.session_id + ); + } + + if let Err(close_err) = self.payjoin_session_store.insert_or_update(session.clone()).await { + log_error!( + self.logger, + "Failed to close receiver session {}: {:?}", + session.session_id, + close_err + ); + } else { + log_info!(self.logger, "Closed failed receiver session: {}", session.session_id); + } + } + + /// Cleans up old payjoin sessions that are completed or failed. + /// Sessions older than `PAYJOIN_SESSION_CLEANUP_AGE_SECS` will be removed. + pub(crate) async fn cleanup_old_sessions(&self) -> Result<(), Error> { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or(std::time::Duration::from_secs(0)) + .as_secs(); + + let sessions_to_remove: Vec = self + .payjoin_session_store + .list_filter(|s| { + let is_terminal = + s.status == PayjoinStatus::Completed || s.status == PayjoinStatus::Failed; + let age = now.saturating_sub(s.latest_update_timestamp); + is_terminal && age > PAYJOIN_SESSION_CLEANUP_AGE_SECS + }) + .await + .into_iter() + .map(|s| s.session_id) + .collect(); + + if sessions_to_remove.is_empty() { + return Ok(()); + } + + log_info!(self.logger, "Cleaning up {} old payjoin sessions", sessions_to_remove.len()); + + for session_id in sessions_to_remove { + if let Err(e) = self.payjoin_session_store.remove(&session_id).await { + log_error!( + self.logger, + "Failed to remove old payjoin session {:?}: {:?}", + session_id, + e + ); + } + } + + Ok(()) + } +} diff --git a/src/payment/payjoin/mod.rs b/src/payment/payjoin/mod.rs new file mode 100644 index 0000000000..b1b8b5455b --- /dev/null +++ b/src/payment/payjoin/mod.rs @@ -0,0 +1,71 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +//! Holds a payment handler for sending and receiving Payjoin payments. + +pub(crate) mod manager; +pub(crate) mod payjoin_session; +pub(crate) mod persist; + +use std::sync::{Arc, RwLock}; + +use crate::error::Error; +use crate::types::PayjoinManager; + +#[cfg(not(feature = "uniffi"))] +type FeeRate = bitcoin::FeeRate; +#[cfg(feature = "uniffi")] +type FeeRate = Arc; + +macro_rules! maybe_map_fee_rate_opt { + ($fee_rate_opt:expr) => {{ + #[cfg(not(feature = "uniffi"))] + { + $fee_rate_opt + } + #[cfg(feature = "uniffi")] + { + $fee_rate_opt.map(|f| *f) + } + }}; +} + +/// A payment handler allowing to receive [Payjoin] payments. +/// +/// Should be retrieved by calling [`Node::payjoin_payment`]. +/// +/// [Payjoin]: https://payjoin.org +/// [`Node::payjoin_payment`]: crate::Node::payjoin_payment +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct PayjoinPayment { + manager: Arc, + is_running: Arc>, +} + +impl PayjoinPayment { + pub(crate) fn new(manager: Arc, is_running: Arc>) -> Self { + Self { manager, is_running } + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl PayjoinPayment { + /// Returns a Payjoin URI that can be shared with a sender to receive a Payjoin payment. + /// + /// The returned string is a BIP 21 URI with Payjoin parameters that the sender can use + /// to initiate the Payjoin flow. + pub async fn receive( + &self, amount_sats: u64, fee_rate: Option, + ) -> Result { + if !*self.is_running.read().expect("lock") { + return Err(Error::NotRunning); + } + + let fee_rate_opt = maybe_map_fee_rate_opt!(fee_rate); + self.manager.receive_payjoin(amount_sats, fee_rate_opt).await + } +} diff --git a/src/payment/payjoin/payjoin_session.rs b/src/payment/payjoin/payjoin_session.rs new file mode 100644 index 0000000000..372f7cffe9 --- /dev/null +++ b/src/payment/payjoin/payjoin_session.rs @@ -0,0 +1,247 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::{OutPoint, Transaction, Txid}; +use lightning::ln::channelmanager::PaymentId; +use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; + +use crate::data_store::{StorableObject, StorableObjectUpdate}; + +/// Represents a payjoin session with persisted events +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PayjoinSession { + /// Session identifier (uses PaymentId from PaymentDetails) + pub session_id: PaymentId, + + /// Direction of the payjoin (Send or Receive) + pub direction: PayjoinDirection, + + /// HPKE public key of receiver (only for sender sessions) + pub receiver_pubkey: Option>, + + /// The amount transferred. + pub amount_msat: Option, + + /// The fee rate in satoshis per kilo-weight-unit + pub fee_rate_kwu: u64, + + /// The fees that were paid for this payment. + pub fee_paid_msat: Option, + + /// The transaction identifier of this payment. + /// Will be None for Receive sessions until the session is completed and the transaction is known. + pub txid: Option, + + /// Serialized session events + pub events: Vec, + + /// The fallback transaction (if any) that the sender created for a receive session. + /// This is broadcasted if the payjoin transacion fails. + pub fallback_tx: Option, + + /// A list of inputs that were seen before in this session. + /// This is used to detect if the sender is reusing inputs across multiple attempts (only for receiver sessions). + pub inputs_seen: Vec, + + /// Current status of the session + pub status: PayjoinStatus, + + /// Unix timestamp of session completion (if completed) + pub completed_at: Option, + + /// The timestamp, in seconds since start of the UNIX epoch, when this entry was last updated. + pub latest_update_timestamp: u64, +} + +impl PayjoinSession { + pub fn new( + session_id: PaymentId, direction: PayjoinDirection, receiver_pubkey: Option>, + amount_msat: Option, fee_rate_kwu: u64, fee_paid_msat: Option, + txid: Option, fallback_tx: Option, + ) -> Self { + let latest_update_timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(); + Self { + session_id, + direction, + receiver_pubkey, + amount_msat, + fee_rate_kwu, + fee_paid_msat, + txid, + events: Vec::new(), + fallback_tx, + inputs_seen: Vec::new(), + status: PayjoinStatus::Active, + completed_at: None, + latest_update_timestamp, + } + } +} + +impl_writeable_tlv_based!(PayjoinSession, { + (0, session_id, required), + (2, direction, required), + (4, receiver_pubkey, option), + (6, amount_msat, option), + (8, fee_rate_kwu, required), + (10, fee_paid_msat, option), + (12, txid, option), + (14, events, required_vec), + (16, fallback_tx, option), + (18, inputs_seen, optional_vec), + (20, status, required), + (22, completed_at, option), + (24, latest_update_timestamp, (default_value, 0u64)), +}); + +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum PayjoinDirection { + /// The session is for sending a payment + Send, + /// The session is for receiving a payment + Receive, +} + +#[derive(Copy, Clone, Debug, PartialEq, Eq)] +pub enum PayjoinStatus { + /// The session is active + Active, + /// The session has completed successfully + Completed, + /// The session has failed + Failed, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SerializedSessionEvent { + /// Serialized event bytes. + pub event_bytes: Vec, + /// Unix timestamp of when the event occurred + pub created_at: u64, +} + +impl_writeable_tlv_based!(SerializedSessionEvent, { + (0, event_bytes, required), + (2, created_at, required), +}); + +impl_writeable_tlv_based_enum!(PayjoinDirection, + (0, Send) => {}, + (2, Receive) => {} +); + +impl_writeable_tlv_based_enum!(PayjoinStatus, + (0, Active) => {}, + (2, Completed) => {}, + (4, Failed) => {} +); + +/// Represents a payjoin session with persisted events +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct PayjoinSessionUpdate { + pub session_id: PaymentId, + pub receiver_pubkey: Option>>, + pub fee_paid_msat: Option>, + pub txid: Option>, + pub events: Option>, + pub fallback_tx: Option>, + pub inputs_seen: Option>, + pub status: Option, + pub completed_at: Option>, +} + +impl From<&PayjoinSession> for PayjoinSessionUpdate { + fn from(value: &PayjoinSession) -> Self { + Self { + session_id: value.session_id, + receiver_pubkey: Some(value.receiver_pubkey.clone()), + fee_paid_msat: Some(value.fee_paid_msat), + txid: Some(value.txid), + events: Some(value.events.clone()), + fallback_tx: Some(value.fallback_tx.clone()), + inputs_seen: Some(value.inputs_seen.clone()), + status: Some(value.status), + completed_at: Some(value.completed_at), + } + } +} + +impl StorableObject for PayjoinSession { + type Id = PaymentId; + type Update = PayjoinSessionUpdate; + + fn id(&self) -> Self::Id { + self.session_id + } + + fn update(&mut self, update: Self::Update) -> bool { + debug_assert_eq!( + self.session_id, update.session_id, + "We should only ever override data for the same id" + ); + + let mut updated = false; + + macro_rules! update_if_necessary { + ($val:expr, $update:expr) => { + if $val != $update { + $val = $update; + updated = true; + } + }; + } + + if let Some(receiver_pubkey_opt) = &update.receiver_pubkey { + update_if_necessary!(self.receiver_pubkey, receiver_pubkey_opt.clone()); + } + if let Some(fee_paid_msat_opt) = update.fee_paid_msat { + update_if_necessary!(self.fee_paid_msat, fee_paid_msat_opt); + } + if let Some(txid_opt) = update.txid { + update_if_necessary!(self.txid, txid_opt); + } + if let Some(events_opt) = &update.events { + update_if_necessary!(self.events, events_opt.clone()); + } + if let Some(fallback_tx_opt) = update.fallback_tx { + update_if_necessary!(self.fallback_tx, fallback_tx_opt); + } + if let Some(txids_input_seen_before_opt) = update.inputs_seen { + update_if_necessary!(self.inputs_seen, txids_input_seen_before_opt); + } + if let Some(status_opt) = update.status { + update_if_necessary!(self.status, status_opt); + } + if let Some(completed_at_opt) = update.completed_at { + update_if_necessary!(self.completed_at, completed_at_opt); + } + + if updated { + self.latest_update_timestamp = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(); + } + + updated + } + + fn to_update(&self) -> Self::Update { + self.into() + } +} + +impl StorableObjectUpdate for PayjoinSessionUpdate { + fn id(&self) -> ::Id { + self.session_id + } +} diff --git a/src/payment/payjoin/persist.rs b/src/payment/payjoin/persist.rs new file mode 100644 index 0000000000..4bb6e9d132 --- /dev/null +++ b/src/payment/payjoin/persist.rs @@ -0,0 +1,162 @@ +// This file is Copyright its original authors, visible in version control history. +// +// This file is licensed under the Apache License, Version 2.0 or the MIT license , at your option. You may not use this file except in +// accordance with one or both of these licenses. + +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::{OutPoint, Transaction, Txid}; +use lightning::ln::channelmanager::PaymentId; +use payjoin::persist::AsyncSessionPersister; +use payjoin::receive::v2::SessionEvent as ReceiverSessionEvent; + +use crate::payment::payjoin::payjoin_session::{ + PayjoinDirection, PayjoinSession, SerializedSessionEvent, +}; +use crate::types::PayjoinSessionStore; +use crate::Error; + +pub(crate) struct KVStorePayjoinReceiverPersister { + session_id: PaymentId, + payjoin_session_store: Arc, +} + +impl KVStorePayjoinReceiverPersister { + pub async fn new( + session_id: PaymentId, amount_msat: Option, + payjoin_session_store: Arc, fee_rate_kwu: u64, + fee_paid_msat: Option, txid: Option, fallback_tx: Option, + ) -> Result { + let session = PayjoinSession::new( + session_id, + PayjoinDirection::Receive, + None, + amount_msat, + fee_rate_kwu, + fee_paid_msat, + txid, + fallback_tx, + ); + + payjoin_session_store.insert(session).await?; + + Ok(Self { session_id, payjoin_session_store }) + } + + pub fn session_id(&self) -> PaymentId { + self.session_id + } + + pub async fn get_session(&self) -> Option { + self.payjoin_session_store.get(&self.session_id).await.ok()? + } + + /// Reconstruct persister from existing session + pub async fn from_session( + session_id: PaymentId, payjoin_session_store: Arc, + ) -> Result { + if payjoin_session_store.get(&session_id).await?.is_none() { + return Err(Error::InvalidPaymentId); + } + + Ok(Self { session_id, payjoin_session_store }) + } + + /// Records the given inputs as seen in this session. + pub async fn insert_inputs_seen(&self, inputs: Vec) -> Result<(), Error> { + if inputs.is_empty() { + return Ok(()); + } + let mut session = self.get_session().await.ok_or(Error::InvalidPaymentId)?; + for input in inputs { + if !session.inputs_seen.contains(&input) { + session.inputs_seen.push(input); + } + } + self.payjoin_session_store.insert_or_update(session).await?; + Ok(()) + } +} + +impl AsyncSessionPersister for KVStorePayjoinReceiverPersister { + type SessionEvent = ReceiverSessionEvent; + type InternalStorageError = Error; + + fn save_event( + &self, event: Self::SessionEvent, + ) -> impl std::future::Future> + Send { + async move { + let mut session = self + .payjoin_session_store + .get(&self.session_id) + .await? + .ok_or(Error::InvalidPaymentId)?; + + let event_bytes = serde_json::to_vec(&event).map_err(|_| Error::PersistenceFailed)?; + + session.events.push(SerializedSessionEvent { + event_bytes, + created_at: SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(), + }); + + self.payjoin_session_store.insert_or_update(session).await?; + + Ok(()) + } + } + + fn load( + &self, + ) -> impl std::future::Future< + Output = Result< + Box + Send>, + Self::InternalStorageError, + >, + > + Send { + async move { + let session = self + .payjoin_session_store + .get(&self.session_id) + .await? + .ok_or(Error::InvalidPaymentId)?; + + let events: Vec = session + .events + .iter() + .map(|e| serde_json::from_slice(&e.event_bytes)) + .collect::, _>>() + .map_err(|_| Error::PersistenceFailed)?; + + Ok(Box::new(events.into_iter()) as Box + Send>) + } + } + + fn close( + &self, + ) -> impl std::future::Future> + Send { + async move { + let mut session = self + .payjoin_session_store + .get(&self.session_id) + .await? + .ok_or(Error::InvalidPaymentId)?; + + session.completed_at = Some( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or(Duration::from_secs(0)) + .as_secs(), + ); + + self.payjoin_session_store.insert_or_update(session).await?; + + Ok(()) + } + } +} diff --git a/src/payment/store.rs b/src/payment/store.rs index 3163ed15b7..4a05ec285e 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -465,6 +465,9 @@ pub enum TransactionType { /// The channels participating in the negotiation. channels: Vec, }, + /// A transaction settling a payjoin, i.e., one to which both we and our counterparty + /// contributed inputs. + Payjoin, } impl_writeable_tlv_based_enum!(TransactionType, @@ -492,7 +495,8 @@ impl_writeable_tlv_based_enum!(TransactionType, }, (12, InteractiveFunding) => { (0, channels, optional_vec), - } + }, + (13, Payjoin) => {} ); impl From for TransactionType { diff --git a/src/types.rs b/src/types.rs index 1a61daa109..301ff34111 100644 --- a/src/types.rs +++ b/src/types.rs @@ -42,6 +42,7 @@ use crate::fee_estimator::OnchainFeeEstimator; use crate::ffi::maybe_wrap; use crate::logger::Logger; use crate::message_handler::NodeCustomMessageHandler; +use crate::payment::payjoin::payjoin_session::PayjoinSession; use crate::payment::{PaymentDetails, PendingPaymentDetails}; use crate::runtime::RuntimeSpawner; @@ -334,6 +335,8 @@ pub(crate) type BumpTransactionEventHandler = pub(crate) type PaymentStore = DataStore, KeepLeastRecentlyUsed>; +pub(crate) type PayjoinSessionStore = DataStore>; + /// A local, potentially user-provided, identifier of a channel. /// /// By default, this will be randomly generated for the user to ensure local uniqueness. @@ -715,3 +718,5 @@ impl From<&(u64, Vec)> for CustomTlvRecord { } pub(crate) type PendingPaymentStore = DataStore, KeepAllEntries>; + +pub(crate) type PayjoinManager = crate::PayjoinManager; diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 4979afff1e..c5560a33ce 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -2281,6 +2281,31 @@ impl Wallet { Ok(new_txid) } + + /// Check if a script belongs to this wallet + pub(crate) fn is_mine(&self, script: ScriptBuf) -> Result { + let locked_wallet = self.inner.lock().expect("lock"); + Ok(locked_wallet.is_mine(script)) + } + + #[allow(deprecated)] + pub(crate) fn process_psbt(&self, mut psbt: Psbt) -> Result { + let locked_wallet = self.inner.lock().expect("lock"); + + let sign_options = SignOptions { trust_witness_utxo: true, ..Default::default() }; + + locked_wallet.sign(&mut psbt, sign_options).map_err(|e| { + log_error!(self.logger, "Failed to sign PSBT: {}", e); + Error::WalletOperationFailed + })?; + + // Return the signed PSBT (not extracted transaction) + Ok(psbt) + } + + pub(crate) fn list_unspent_confirmed_utxos(&self) -> Result, Error> { + self.list_confirmed_utxos_inner().map_err(|()| Error::WalletOperationFailed) + } } struct LocalStakeAggregate {