diff --git a/Cargo.toml b/Cargo.toml index 453212acd9..aaf65c9d09 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/benthecarman/bitcoin-payment-instructions", rev = "632f2ce8de7ea2035d5c83d7d745a52ea3e1fe70", 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 b2f3d74ac2..abb7a6724d 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -69,6 +69,8 @@ interface Node { Bolt12Payment bolt12_payment(); SpontaneousPayment spontaneous_payment(); OnchainPayment onchain_payment(); + [Throws=NodeError] + PayjoinPayment payjoin_payment(); Liquidity liquidity(); ForwardingAnalytics forwarding_analytics(); [Throws=NodeError] @@ -140,6 +142,8 @@ interface FeeRate { u64 to_sat_per_vb_ceil(); }; +typedef interface PayjoinPayment; + typedef interface Liquidity; typedef interface ForwardingAnalytics; @@ -170,6 +174,8 @@ enum NodeError { "OnchainTxSigningFailed", "TxSyncFailed", "TxSyncTimeout", + "TxLookupFailed", + "TxLookupTimeout", "GossipUpdateFailed", "GossipUpdateTimeout", "LiquidityRequestFailed", @@ -213,6 +219,9 @@ enum NodeError { "InvalidLnurl", "ChainSourceNotSupported", "InvalidPayerProof", + "PayjoinNotConfigured", + "PayjoinSessionCreationFailed", + "PayjoinSessionFailed", }; typedef dictionary NodeStatus; diff --git a/src/builder.rs b/src/builder.rs index 0372e742f6..b9705d0457 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, @@ -83,7 +83,8 @@ use crate::io::utils::{ use crate::io::vss_store::VssStoreBuilder; use crate::io::{ self, CHANNEL_FORWARDING_STATS_PERSISTENCE_SECONDARY_NAMESPACE, - FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, + FORWARDED_PAYMENT_PERSISTENCE_PRIMARY_NAMESPACE, 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, @@ -94,6 +95,7 @@ use crate::logger::{log_error, LdkLogger, LogLevel, LogWriter, Logger}; use crate::message_handler::NodeCustomMessageHandler; use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox; use crate::payment::forwarding_store::ForwardingStore; +use crate::payment::payjoin::manager::PayjoinManager; #[cfg(feature = "unified-payments")] use crate::payment::HRNResolver; use crate::peer_store::PeerStore; @@ -105,8 +107,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; @@ -233,6 +235,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 { @@ -279,6 +283,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.") + }, } } } @@ -675,6 +682,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: @@ -1293,6 +1309,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. @@ -1553,6 +1577,7 @@ fn build_with_store_internal( node_metris_res, pending_payment_store_res, address_pool_res, + payjoin_session_store_res, ) = runtime.block_on(async move { tokio::join!( read_n_objects( @@ -1576,6 +1601,12 @@ fn build_with_store_internal( 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), + ), ) }); @@ -2454,6 +2485,41 @@ 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(&tx_broadcaster), + ))) + } else { + None + }; + let prober = probing_config.map(|probing_cfg| { let strategy: Arc = match &probing_cfg.kind { ProbingStrategyKind::HighDegree { top_node_count } => { @@ -2560,6 +2626,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 ac3b8e6e8d..b39c4e158c 100644 --- a/src/config.rs +++ b/src/config.rs @@ -193,6 +193,18 @@ impl Default for ForwardedPaymentTrackingMode { } } +// 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. @@ -215,8 +227,9 @@ impl Default for ForwardedPaymentTrackingMode { feature = "unified-payments", doc = "| `hrn_config` | HumanReadableNamesConfig::default() |" )] -/// | `manually_handle_unknown_bolt11_payments` | false | +/// | `manually_handle_unknown_bolt11_payments` | false | /// | `forwarded_payment_tracking_mode` | Stats | +/// | `payjoin_config` | None | /// /// See [`AnchorChannelsConfig`], [`RouteParametersConfig`], and /// [`ForwardedPaymentTrackingMode`] for more information regarding their respective default values. @@ -295,6 +308,8 @@ pub struct Config { pub manually_handle_unknown_bolt11_payments: bool, /// The mode used for tracking forwarded payments. pub forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode, + /// Configuration options for PayJoin payments. + pub payjoin_config: Option, } impl Default for Config { @@ -314,6 +329,7 @@ impl Default for Config { hrn_config: HumanReadableNamesConfig::default(), manually_handle_unknown_bolt11_payments: false, forwarded_payment_tracking_mode: ForwardedPaymentTrackingMode::default(), + payjoin_config: None, } } } @@ -857,6 +873,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 d187c68200..8aa1c06722 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. @@ -151,6 +155,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 { @@ -186,6 +196,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."), @@ -249,6 +261,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 b7e4d2131f..a8bed0202c 100644 --- a/src/io/mod.rs +++ b/src/io/mod.rs @@ -102,3 +102,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 6d13141c6a..39cf72c82c 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; @@ -202,6 +202,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; @@ -285,6 +287,7 @@ pub struct Node { prober: Option>, #[cfg(cycle_tests)] _leak_checker: LeakChecker, + payjoin_manager: Option>, } impl Node { @@ -849,6 +852,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(()) @@ -1186,6 +1235,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 e1c60da79a..3bc491286f 100644 --- a/src/payment/mod.rs +++ b/src/payment/mod.rs @@ -15,6 +15,7 @@ pub(crate) mod forwarding_store; #[cfg(feature = "unified-payments")] mod hrn; mod onchain; +pub(crate) mod payjoin; pub(crate) mod pending_payment_store; mod spontaneous; pub(crate) mod store; @@ -32,6 +33,7 @@ pub use forwarding::{ #[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..d7ef167f03 --- /dev/null +++ b/src/payment/payjoin/manager.rs @@ -0,0 +1,1100 @@ +// 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::collections::HashSet; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use bitcoin::{psbt::Input, Amount, FeeRate, OutPoint, Transaction, 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, CreateRequestError, + HasReplyableError, Initialized, MaybeInputsOwned, MaybeInputsSeen, Monitor, OutputsUnknown, + PayjoinProposal, PendingFallback as ReceiverPendingFallback, ProvisionalProposal, + ReceiveSession, Receiver, ReceiverBuilder, SessionOutcome as ReceiverSessionOutcome, + 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::total_anchor_channels_reserve_sats; +use crate::types::{Broadcaster, ChannelManager, PayjoinSessionStore}; +use crate::wallet::Wallet; +use crate::Error; + +#[derive(Clone)] +pub(crate) struct PayjoinManager { + payjoin_session_store: Arc, + inputs_seen_lock: Arc>, + logger: Arc, + config: Arc, + wallet: Arc, + fee_estimator: Arc, + chain_source: Arc, + channel_manager: Arc, + stop_receiver: tokio::sync::watch::Receiver<()>, + 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<()>, broadcaster: Arc, + ) -> Self { + Self { + payjoin_session_store, + inputs_seen_lock: Arc::new(tokio::sync::Mutex::new(())), + logger, + config, + wallet, + fee_estimator, + chain_source, + channel_manager, + stop_receiver, + 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, + Arc::clone(&self.payjoin_session_store), + fee_rate.to_sat_per_kwu(), + 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(|e| { + log_error!(self.logger, "Failed to persist the new payjoin session: {:?}", e); + 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, mut session: ReceiveSession, persister: &KVStorePayjoinReceiverPersister, + ) -> Result<(), Error> { + loop { + session = match session { + ReceiveSession::Initialized(proposal) => { + match self.read_from_directory(proposal, persister).await? { + // No sender yet. Yield so the resume task can service other sessions; + // the next tick re-enters this session where it left off. + ReceiveSession::Initialized(_) => return Ok(()), + next => next, + } + }, + 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) => { + match self.handle_error(error, persister).await? { + // Retry on the next tick instead of hammering the relay + // when a transient failure occurs. + ReceiveSession::HasReplyableError(_) => return Ok(()), + next => next, + } + }, + ReceiveSession::Monitor(proposal) => { + self.monitor_payjoin_proposal(proposal, persister).await?; + return Ok(()); + }, + ReceiveSession::PendingFallback(pending) => { + let fallback_tx = pending.fallback_tx().clone(); + let mut payjoin_session = + persister.get_session().await?.ok_or(Error::InvalidPaymentId)?; + self.close_session_with_fallback(&mut payjoin_session, Some(&fallback_tx)) + .await; + + pending.close().save_async(persister).await?; + return Ok(()); + }, + ReceiveSession::Closed(outcome) => { + return self.handle_closed_session(outcome, persister).await; + }, + } + } + } + + async fn read_from_directory( + &self, session: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + let Some((ohttp_response, context)) = + self.post_via_relay(|relay| session.create_poll_request(relay)).await? + else { + self.cancel_receiver_session(persister.session_id()).await?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + }; + + 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." + ); + Ok(ReceiveSession::UncheckedOriginalPayload(next_state)) + }, + Ok(OptionalTransitionOutcome::Stasis(current_state)) => { + Ok(ReceiveSession::Initialized(current_state)) + }, + Err(e) if e.is_transient() => { + log_debug!(self.logger, "Transient error polling for request, retrying: {e:?}"); + let session = e.transient_state().expect("transient error carries current state"); + Ok(ReceiveSession::Initialized(session)) + }, + Err(e) => { + log_error!( + self.logger, + "Failed to process the sender's directory response: {:?}", + e + ); + 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 { + 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(|e| { + log_error!(self.logger, "Failed to check the sender's original transaction for broadcast suitability: {:?}", e); + 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" + ); + + // Sender inputs with a non-empty script_sig change the txid once the sender signs, so we + // couldn't match the payjoin to its payment record. Cancel and let the fallback + // transaction pay us instead. + if !proposal.proposal_txid_is_stable() { + log_info!( + self.logger, + "Declining payjoin: sender inputs aren't native SegWit. Broadcasting the fallback." + ); + let pending = proposal.cancel().save_async(persister).await.map_err(|e| { + log_error!( + self.logger, + "Failed to cancel the payjoin session after declining: {:?}", + e + ); + Error::PersistenceFailed + })?; + return Ok(ReceiveSession::PendingFallback(pending)); + } + + Ok(ReceiveSession::MaybeInputsOwned(proposal)) + } + + async fn check_inputs_not_owned( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + let proposal = proposal + .check_inputs_not_owned(&mut |outpoint| { + self.wallet + .is_my_outpoint(&outpoint) + .map_err(|e| ImplementationError::from(e.to_string().as_str())) + }) + .save_async(persister) + .await + .map_err(|e| { + log_error!( + self.logger, + "Failed to check whether the sender's inputs belong to us: {:?}", + e + ); + Error::PersistenceFailed + })?; + + Ok(ReceiveSession::MaybeInputsSeen(proposal)) + } + + async fn check_no_inputs_seen_before( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + let session_id = persister.session_id(); + + let _guard = self.inputs_seen_lock.lock().await; + + let inputs_seen_by_other_sessions: HashSet = self + .payjoin_session_store + .list_filter(|s| s.session_id != session_id) + .await + .into_iter() + .flat_map(|s| s.inputs_seen) + .collect(); + + let mut newly_seen = Vec::new(); + let transition = proposal.check_no_inputs_seen_before(&mut |input| { + if inputs_seen_by_other_sessions.contains(input) { + return Ok(true); + } + newly_seen.push(*input); + Ok(false) + }); + + persister.insert_inputs_seen(newly_seen).await?; + + let proposal = transition.save_async(persister).await.map_err(|e| { + log_error!(self.logger, "Failed to check the sender's inputs for reuse: {:?}", e); + Error::PersistenceFailed + })?; + + Ok(ReceiveSession::OutputsUnknown(proposal)) + } + + async fn identify_receiver_outputs( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + 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(|e| { + log_error!( + self.logger, + "Failed to identify our outputs in the sender's proposal: {:?}", + e + ); + Error::PersistenceFailed + })?; + Ok(ReceiveSession::WantsOutputs(proposal)) + } + + async fn commit_outputs( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + let proposal = proposal.commit_outputs().save_async(persister).await.map_err(|e| { + log_error!(self.logger, "Failed to commit the payjoin outputs: {:?}", e); + Error::PersistenceFailed + })?; + Ok(ReceiveSession::WantsInputs(proposal)) + } + + async fn contribute_inputs( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + // 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(|e| { + log_error!(self.logger, "Failed to commit our contributed payjoin inputs: {:?}", e); + Error::PersistenceFailed + })?; + Ok(ReceiveSession::WantsFeeRange(proposal)) + } + + 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 { + 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( + |e| { + log_error!(self.logger, "Failed to apply the payjoin fee range: {:?}", e); + Error::PersistenceFailed + }, + )?; + + Ok(ReceiveSession::ProvisionalProposal(proposal)) + } + + async fn finalize_proposal( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + 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(|e| { + log_error!(self.logger, "Failed to finalize the payjoin proposal: {:?}", e); + Error::PersistenceFailed + })?; + Ok(ReceiveSession::PayjoinProposal(proposal)) + } + + async fn send_payjoin_proposal( + &self, proposal: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + let Some((ohttp_response, context)) = + self.post_via_relay(|relay| proposal.create_post_request(relay)).await? + else { + self.cancel_receiver_session(persister.session_id()).await?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + }; + + let payjoin_psbt = proposal.psbt().clone(); + + match proposal + .process_response(ohttp_response.as_bytes(), context) + .save_async(persister) + .await + { + Ok(session) => { + // At this point we will persist the fee and txid to the session store + let payjoin_tx = payjoin_psbt.extract_tx_unchecked_fee_rate(); + let txid = payjoin_tx.compute_txid(); + + log_info!( + self.logger, + "Response successful. Watch mempool for successful Payjoin. TXID: {}", + txid + ); + Ok(ReceiveSession::Monitor(session)) + }, + Err(e) if e.is_transient() => { + log_debug!( + self.logger, + "Transient error sending payjoin proposal, retrying: {e:?}" + ); + let proposal = e.transient_state().expect("transient error carries current state"); + Ok(ReceiveSession::PayjoinProposal(proposal)) + }, + Err(e) => { + log_error!( + self.logger, + "Failed to process the sender's response to our payjoin proposal: {:?}", + e + ); + Err(Error::PersistenceFailed) + }, + } + } + + async fn monitor_payjoin_proposal( + &self, mut 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 polled = tokio::time::timeout(timeout_duration, async { + loop { + interval.tick().await; + let check_result = proposal + .check_for_transaction(|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(OptionalTransitionOutcome::Progress(())) => { + return Ok(()); + }, + Ok(OptionalTransitionOutcome::Stasis(current_state)) => { + proposal = current_state; + }, + Err(e) if e.is_transient() => { + log_debug!( + self.logger, + "Transient error checking for transaction, retrying: {e:?}" + ); + proposal = + e.transient_state().expect("transient error carries current state"); + }, + Err(_) => return Err(Error::PayjoinSessionFailed), + } + } + }) + .await; + + match polled { + Ok(Ok(())) => { + log_info!(self.logger, "Payjoin transaction detected in the mempool!"); + // Replay the session to recover the outcome and record the + // payment now rather than on the next resume tick. + let (session, _) = + replay_receiver_event_log_async(persister).await.map_err(|e| { + log_error!( + self.logger, + "Failed to replay the closed payjoin session: {:?}", + e + ); + Error::PayjoinSessionFailed + })?; + if let ReceiveSession::Closed(outcome) = session { + return self.handle_closed_session(outcome, persister).await; + } + Ok(()) + }, + Ok(Err(e)) => Err(e), + Err(_) => { + log_debug!( + self.logger, + "Payjoin transaction not yet seen after {:?}. Will retry on next background tick.", + timeout_duration + ); + Ok(()) + }, + } + } + + async fn post_via_relay( + &self, mut build: F, + ) -> Result, Error> + where + F: FnMut(&str) -> Result<(payjoin::Request, payjoin::OhttpResponse), CreateRequestError>, + { + let payjoin_config = + self.config.payjoin_config.as_ref().ok_or(Error::PayjoinNotConfigured)?; + + for relay in self.relay_order(payjoin_config)? { + let (req, ctx) = match build(relay) { + Ok(r) => r, + Err(e) if e.is_expired() => return Ok(None), + Err(e) => { + // Building the request doesn't depend on the relay, so this would fail + // the same way for every remaining relay. + log_error!(self.logger, "Failed to build payjoin request: {}", e); + return Err(Error::PayjoinSessionFailed); + }, + }; + + match self.post_request(req).await { + Ok(resp) => return Ok(Some((resp, ctx))), + Err(e) => { + log_debug!( + self.logger, + "Request via relay {} failed, trying next: {:?}", + relay, + e + ); + }, + } + } + + log_error!(self.logger, "All configured OHTTP relays failed."); + Err(Error::ConnectionFailed) + } + + async fn handle_error( + &self, session: Receiver, persister: &KVStorePayjoinReceiverPersister, + ) -> Result { + let Some((err_response, err_ctx)) = + self.post_via_relay(|relay| session.create_error_request(relay)).await? + else { + self.cancel_receiver_session(persister.session_id()).await?; + return Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)); + }; + + let err_bytes = err_response.as_bytes(); + + match session.process_error_response(err_bytes, err_ctx).save_async(persister).await { + Ok(Some(pending)) => { + log_info!( + self.logger, + "Session delivered error reply. Broadcast the fallback transaction." + ); + Ok(ReceiveSession::PendingFallback(pending)) + }, + Ok(None) => Ok(ReceiveSession::Closed(ReceiverSessionOutcome::Aborted)), + Err(e) if e.is_transient() => { + log_debug!(self.logger, "Transient error posting error response, retrying: {e:?}"); + let session = e.transient_state().expect("transient error carries current state"); + Ok(ReceiveSession::HasReplyableError(session)) + }, + Err(e) => { + if let Some(api_err) = e.api_error_ref() { + log_warn!(self.logger, "Failed to confirm error response delivery: {api_err}"); + } + match e.fatal_state() { + Some(pending) => { + log_error!(self.logger, "Session failed to deliver error reply. Broadcast the fallback transaction"); + Ok(ReceiveSession::PendingFallback(pending)) + }, + None => Err(Error::PayjoinSessionFailed), + } + }, + } + } + + async fn handle_closed_session( + &self, outcome: ReceiverSessionOutcome, 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)?; + session.completed_at.get_or_insert_with(|| { + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs() + }); + + match outcome { + ReceiverSessionOutcome::Success(txid) => { + log_info!( + self.logger, + "Payjoin session detected in the mempool and completed successfully." + ); + // The transaction only supplies figures if wallet sync hasn't recorded it yet. + let tx = match self.chain_source.get_transaction(&txid).await { + Ok(tx) => tx, + Err(e) => { + log_debug!( + self.logger, + "Could not look up payjoin transaction {}: {:?}. Wallet sync will fill in the payment's amount and fee", + txid, + e + ); + None + }, + }; + self.wallet.classify_payjoin(txid, tx.as_ref()).await?; + + session.status = PayjoinStatus::Completed; + + self.payjoin_session_store.insert_or_update(session).await?; + }, + ReceiverSessionOutcome::PayjoinProposalSent => { + log_info!( + self.logger, + "Payjoin proposal sent. Cannot track broadcast due to non-SegWit sender inputs." + ); + session.status = PayjoinStatus::Completed; + + self.payjoin_session_store.insert_or_update(session).await?; + }, + ReceiverSessionOutcome::FallbackBroadcasted => { + log_info!(self.logger, "Payjoin failed. Fallback transaction was broadcasted."); + session.status = PayjoinStatus::Failed; + + self.payjoin_session_store.insert_or_update(session).await?; + }, + ReceiverSessionOutcome::Aborted => { + log_info!(self.logger, "Payjoin session was aborted."); + self.close_session_with_fallback(&mut session, None).await; + }, + ReceiverSessionOutcome::Unrecognized(txid) => { + log_error!( + self.logger, + "Payjoin session {} was settled by an unrecognized transaction {}.", + session_id, + txid + ); + session.status = PayjoinStatus::Failed; + + self.payjoin_session_store.insert_or_update(session).await?; + }, + } + 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) if e.is_expired() => { + if let Err(err) = self.cancel_receiver_session(session_id).await { + log_error!( + self.logger, + "Failed to cancel expired receiver session {session_id}: {err:?}" + ); + } + }, + 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, None).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 task finished."), + 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 cancel_receiver_session(&self, session_id: PaymentId) -> Result<(), Error> { + let mut payjoin_session = + self.payjoin_session_store.get(&session_id).await?.ok_or(Error::InvalidPaymentId)?; + + let persister = KVStorePayjoinReceiverPersister::from_session( + session_id, + Arc::clone(&self.payjoin_session_store), + ) + .await?; + let (session, history) = match replay_receiver_event_log_async(&persister).await { + Ok((session, history)) => (session, history), + Err(e) if e.is_expired() => { + self.close_session_with_fallback(&mut payjoin_session, e.expiry_fallback_tx()) + .await; + return Ok(()); + }, + Err(_) => return Err(Error::PayjoinSessionFailed), + }; + + let pending: Receiver = match session { + ReceiveSession::Initialized(receiver) => { + receiver.cancel().save_async(&persister).await?; + log_info!(self.logger, "Session cancelled. No fallback transaction to broadcast."); + return Ok(()); + }, + ReceiveSession::UncheckedOriginalPayload(receiver) => { + receiver.cancel().save_async(&persister).await?; + log_info!(self.logger, "Session cancelled. No fallback transaction to broadcast."); + return Ok(()); + }, + ReceiveSession::MaybeInputsOwned(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::MaybeInputsSeen(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::OutputsUnknown(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::WantsOutputs(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::WantsInputs(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::WantsFeeRange(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::ProvisionalProposal(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::PayjoinProposal(receiver) => { + receiver.cancel().save_async(&persister).await? + }, + ReceiveSession::Monitor(receiver) => receiver.cancel().save_async(&persister).await?, + ReceiveSession::HasReplyableError(receiver) => { + match receiver.cancel().save_async(&persister).await? { + Some(pending) => pending, + None => { + log_info!( + self.logger, + "Session cancelled. No fallback transaction available." + ); + return Ok(()); + }, + } + }, + ReceiveSession::PendingFallback(receiver) => receiver, + ReceiveSession::Closed( + ReceiverSessionOutcome::Success(_) + | ReceiverSessionOutcome::FallbackBroadcasted + | ReceiverSessionOutcome::PayjoinProposalSent, + ) => { + log_info!(self.logger, "Session already completed successfully. Cannot cancel."); + return Ok(()); + }, + ReceiveSession::Closed(ReceiverSessionOutcome::Aborted) => { + match history.fallback_tx() { + Some(tx) => { + log_info!( + self.logger, + "Session was already cancelled. Broadcast the fallback transaction" + ); + self.close_session_with_fallback(&mut payjoin_session, Some(&tx)).await; + }, + None => log_info!( + self.logger, + "Session is already closed. No fallback transaction available." + ), + } + return Ok(()); + }, + ReceiveSession::Closed(ReceiverSessionOutcome::Unrecognized(_)) => { + log_info!( + self.logger, + "Session was already closed by an unrecognized transaction. Cannot cancel." + ); + + return Ok(()); + }, + }; + + pending.close().save_async(&persister).await?; + Ok(()) + } + + async fn close_session_with_fallback( + &self, session: &mut PayjoinSession, fallback_tx: Option<&Transaction>, + ) { + session.status = PayjoinStatus::Failed; + // Keep an earlier timestamp if payjoin's `close()` already ran + session.completed_at.get_or_insert_with(|| { + SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or(Duration::from_secs(0)).as_secs() + }); + + let fallback_tx = fallback_tx.or_else(|| session.fallback_tx.as_ref()); + + if let Some(fallback_tx) = fallback_tx { + 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 completed_at = s.completed_at.unwrap_or(s.latest_update_timestamp); + is_terminal && now.saturating_sub(completed_at) > 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..56ac8374ec --- /dev/null +++ b/src/payment/payjoin/mod.rs @@ -0,0 +1,75 @@ +// 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, payment::payjoin::manager::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. + /// + /// `fee_rate` caps what we are willing to pay for the input we contribute. It defaults to + /// the current on-chain fee estimate. The cap is fixed when the URI is created and the + /// session may last up to 24 hours, so if fees rise above it before the sender responds, + /// the payjoin is declined and the sender's original transaction pays instead. + 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..5c69cb3013 --- /dev/null +++ b/src/payment/payjoin/payjoin_session.rs @@ -0,0 +1,225 @@ +// 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}; +use lightning::ln::channelmanager::PaymentId; +use lightning::{impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; + +use crate::data_store::{StorableObject, StorableObjectUpdate, UpdatableObject}; + +/// 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 fee rate in satoshis per kilo-weight-unit + pub fee_rate_kwu: u64, + + /// Serialized session events + pub events: Vec, + + /// The fallback transaction (if any) that the sender created for a receive session. + /// This is broadcast if the payjoin transaction fails. + pub fallback_tx: Option, + + /// Inputs seen in this session's original proposal. + /// + /// This is used to detect if the sender is reusing inputs across multiple sessions (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>, + fee_rate_kwu: u64, 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, + fee_rate_kwu, + 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, fee_rate_kwu, required), + (8, events, required_vec), + (10, fallback_tx, option), + (12, inputs_seen, optional_vec), + (14, status, required), + (16, completed_at, option), + (18, 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 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()), + 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; + + fn id(&self) -> Self::Id { + self.session_id + } +} + +impl UpdatableObject for PayjoinSession { + type Update = PayjoinSessionUpdate; + + 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(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..775e96d330 --- /dev/null +++ b/src/payment/payjoin/persist.rs @@ -0,0 +1,158 @@ +// 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}; +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, payjoin_session_store: Arc, fee_rate_kwu: u64, + fallback_tx: Option, + ) -> Result { + let session = PayjoinSession::new( + session_id, + PayjoinDirection::Receive, + None, + fee_rate_kwu, + 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) -> Result, Error> { + self.payjoin_session_store.get(&self.session_id).await + } + + /// 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 41c39045f8..f5f52766c6 100644 --- a/src/payment/store.rs +++ b/src/payment/store.rs @@ -468,6 +468,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, @@ -495,7 +498,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 fd86d1bcd8..78b01e67ef 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::{ ChannelPairForwardingStats, ForwardedPaymentDetails, PaymentDetails, PendingPaymentDetails, }; @@ -342,6 +343,8 @@ pub(crate) type ChannelForwardingStatsStore = pub(crate) type ChannelPairForwardingStatsStore = DataStore, KeepNoEntries>; +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. diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs index 729b2ed63b..149f02e274 100644 --- a/src/wallet/mod.rs +++ b/src/wallet/mod.rs @@ -1755,6 +1755,77 @@ impl Wallet { Ok(()) } + /// Tags an incoming payjoin transaction as [`TransactionType::Payjoin`]. + /// + /// The sender broadcasts a payjoin, so it never passes through [`Self::classify_broadcast`], + /// and we only learn the outcome once monitoring finds the transaction, possibly after wallet + /// sync has already recorded or even graduated it. Like a late funding classification, this + /// merges only the classification into an existing record and leaves its figures, status and + /// confirmation state to wallet sync. If no record exists yet, one is created from `tx` and + /// marked as pending. + pub(crate) async fn classify_payjoin( + &self, txid: Txid, tx: Option<&Transaction>, + ) -> Result<(), Error> { + // Held across both stores, as wallet sync and the other classifiers do, so neither sees + // the record and its pending index out of step. + let _guard = self.funding_payment_update_lock.lock().await; + + let id = PaymentId(txid.to_byte_array()); + let (amount_msat, fee_paid_msat, direction) = match tx { + Some(tx) => self.onchain_payment_fields(tx), + // Wallet sync fills the values in on its next event for this txid. + None => (None, None, PaymentDirection::Inbound), + }; + let details = PaymentDetails::new( + id, + PaymentKind::Onchain { + txid, + status: ConfirmationStatus::Unconfirmed, + tx_type: Some(TransactionType::Payjoin), + }, + amount_msat, + fee_paid_msat, + direction, + PaymentStatus::Pending, + ); + + let mut classification = PaymentDetailsUpdate::new(id); + classification.tx_type = Some(Some(TransactionType::Payjoin)); + + self.payment_store + .mutate(&id, |existing| match existing { + None => Some(details.clone()), + Some(current) => { + let mut updated = current.clone(); + updated.update(classification.clone()).then_some(updated) + }, + }) + .await?; + + let payment_store = Arc::clone(&self.payment_store); + self.pending_payment_store + .mutate_async(&id, move |existing| async move { + let recorded = payment_store.get(&id).await?.unwrap_or(details); + Ok(match existing { + None if recorded.status == PaymentStatus::Pending => { + Some(PendingPaymentDetails::new(recorded, Vec::new(), Vec::new())) + }, + None => None, + Some(mut entry) => { + let pending_update = PendingPaymentDetailsUpdate { + id, + payment_update: Some(classification), + conflicting_txids: None, + candidates: Vec::new(), + }; + entry.update(pending_update).then_some(entry) + }, + }) + }) + .await?; + Ok(()) + } + /// Writes a freshly-classified funding payment to the authoritative payment store and adds a /// pending-store index entry, so wallet sync graduates it through `ANTI_REORG_DELAY`. async fn persist_funding_payment( @@ -2278,6 +2349,46 @@ 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)) + } + + /// Check if an outpoint belongs to this wallet. + pub(crate) fn is_my_outpoint(&self, outpoint: &OutPoint) -> Result { + let locked_wallet = self.inner.lock().expect("lock"); + + let script_pubkey = match locked_wallet.tx_details(outpoint.txid) { + Some(details) => match details.tx.output.get(outpoint.vout as usize) { + Some(txout) => txout.script_pubkey.clone(), + None => return Ok(false), + }, + None => return Ok(false), + }; + + Ok(locked_wallet.is_mine(script_pubkey)) + } + + #[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 {