diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index be5fdc8084..9135ecbf6b 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -249,6 +249,10 @@ enum NodeError { "InvalidLnurl", "ChainSourceNotSupported", "InvalidPayerProof", + "LiquiditySetWebhookFailed", + "LiquidityRemoveWebhookFailed", + "LiquidityListWebhooksFailed", + "LiquidityNotifyWebhookFailed" }; typedef dictionary NodeStatus; diff --git a/src/builder.rs b/src/builder.rs index dc41aef1a2..5169127dab 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -83,8 +83,8 @@ use crate::runtime::{Runtime, RuntimeSpawner}; use crate::tx_broadcaster::TransactionBroadcaster; use crate::types::{ AsyncPersister, ChainMonitor, ChannelManager, DynStore, DynStoreRef, DynStoreWrapper, - GossipSync, Graph, HRNResolver, KeysManager, MessageRouter, OnionMessenger, PaymentStore, - PeerManager, PendingPaymentStore, + GossipSync, Graph, HRNResolver, KeysManager, LSPS5ServiceConfig, MessageRouter, OnionMessenger, + PaymentStore, PeerManager, PendingPaymentStore, }; use crate::wallet::persist::{read_address_pool, KVStoreWalletPersister}; use crate::wallet::Wallet; @@ -127,10 +127,12 @@ struct PathfindingScoresSyncConfig { #[derive(Debug, Clone, Default)] struct LiquiditySourceConfig { - // Acts for both LSPS1 and LSPS2 clients connecting to the given service. + // Acts for LSPS1, LSPS2 and LSPS5 clients connecting to the given service. lsp_nodes: Vec, // Act as an LSPS2 service. lsps2_service: Option, + // Act as an LSPS5 service. + lsps5_service: Option, } #[derive(Clone)] @@ -507,18 +509,26 @@ impl NodeBuilder { self } - /// Configures the [`Node`] instance to provide an [LSPS2] service, issuing just-in-time - /// channels to clients. + /// Configures the [`Node`] instance to provide [bLIP-52 / LSPS2] and/or [bLIP-55 / LSPS5] + /// services to clients. + /// + /// [bLIP-52 / LSPS2] issues just-in-time channels to clients, [bLIP-55 / LSPS5] allows clients + /// to register webhooks for push notifications. + /// + /// Passing `None` leaves the respective service disabled. /// /// **Caution**: LSP service support is in **alpha** and is considered an experimental feature. /// - /// [LSPS2]: https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md + /// [bLIP-52 / LSPS2]: https://github.com/lightning/blips/blob/master/blip-0052.md + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md pub fn enable_liquidity_provider( - &mut self, lsps2_service_config: LSPS2ServiceConfig, + &mut self, lsps2_service_config: Option, + lsps5_service_config: Option, ) -> &mut Self { let liquidity_source_config = self.liquidity_source_config.get_or_insert(LiquiditySourceConfig::default()); - liquidity_source_config.lsps2_service = Some(lsps2_service_config); + liquidity_source_config.lsps2_service = lsps2_service_config; + liquidity_source_config.lsps5_service = lsps5_service_config; self } @@ -1112,14 +1122,26 @@ impl ArcedNodeBuilder { ); } - /// Configures the [`Node`] instance to provide an [LSPS2] service, issuing just-in-time - /// channels to clients. + /// Configures the [`Node`] instance to provide [bLIP-52 / LSPS2] and/or [bLIP-55 / LSPS5] + /// services to clients. + /// + /// [bLIP-52 / LSPS2] issues just-in-time channels to clients, [bLIP-55 / LSPS5] allows clients + /// to register webhooks for push notifications. + /// + /// Passing `None` leaves the respective service disabled. /// /// **Caution**: LSP service support is in **alpha** and is considered an experimental feature. /// - /// [LSPS2]: https://github.com/BitcoinAndLightningLayerSpecs/lsp/blob/main/LSPS2/README.md - pub fn enable_liquidity_provider(&self, lsps2_service_config: LSPS2ServiceConfig) { - self.inner.write().expect("lock").enable_liquidity_provider(lsps2_service_config); + /// [bLIP-52 / LSPS2]: https://github.com/lightning/blips/blob/master/blip-0052.md + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md + pub fn enable_liquidity_provider( + &self, lsps2_service_config: Option, + lsps5_service_config: Option, + ) { + self.inner + .write() + .expect("lock") + .enable_liquidity_provider(lsps2_service_config, lsps5_service_config); } /// Sets the used storage directory path. @@ -2148,6 +2170,7 @@ fn build_with_store_internal( Arc::clone(&tx_broadcaster), Arc::clone(&kv_store), Arc::clone(&config), + Arc::clone(&runtime), Arc::clone(&logger), ); @@ -2166,6 +2189,10 @@ fn build_with_store_internal( lsc.lsps2_service.as_ref().map(|config| { liquidity_source_builder.lsps2_service(promise_secret, config.clone()) }); + + lsc.lsps5_service + .as_ref() + .map(|config| liquidity_source_builder.lsps5_service(config.clone())); } let liquidity_source = runtime @@ -2225,6 +2252,8 @@ fn build_with_store_internal( liquidity_source.lsps2_service().set_peer_manager(Arc::downgrade(&peer_manager)); + liquidity_source.lsps5_service().set_peer_manager(Arc::downgrade(&peer_manager)); + let connection_manager = Arc::new(ConnectionManager::new( Arc::clone(&peer_manager), config.tor_config.clone(), diff --git a/src/config.rs b/src/config.rs index aa8cc7e615..b7e6c9ec32 100644 --- a/src/config.rs +++ b/src/config.rs @@ -152,6 +152,12 @@ 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 timeout after which we abort a LSPS5 webhook notification operation. +pub(crate) const LSPS5_WEBHOOK_TIMEOUT_SECS: u64 = 30; + +// The maximum size of a response body we'll accept when delivering an LSPS5 webhook notification. +pub(crate) const LSPS5_WEBHOOK_MAX_RESPONSE_SIZE: usize = 64 * 1024; + #[derive(Debug, Clone)] #[cfg_attr(feature = "uniffi", derive(uniffi::Record))] /// Represents the configuration of an [`Node`] instance. diff --git a/src/error.rs b/src/error.rs index 1fbc5066a3..e409a21053 100644 --- a/src/error.rs +++ b/src/error.rs @@ -143,6 +143,14 @@ pub enum Error { ChainSourceNotSupported, /// The provided payer proof is invalid. InvalidPayerProof, + /// Failed to set a webhook with the LSP. + LiquiditySetWebhookFailed, + /// Failed to remove a webhook with the LSP. + LiquidityRemoveWebhookFailed, + /// Failed to list webhooks with the LSP. + LiquidityListWebhooksFailed, + /// Failed to send a webhook notification to a client. + LiquidityNotifyWebhookFailed, } impl fmt::Display for Error { @@ -233,6 +241,18 @@ 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::LiquiditySetWebhookFailed => { + write!(f, "Failed to set a webhook with the LSP.") + }, + Self::LiquidityRemoveWebhookFailed => { + write!(f, "Failed to remove a webhook with the LSP.") + }, + Self::LiquidityListWebhooksFailed => { + write!(f, "Failed to list webhooks with the LSP.") + }, + Self::LiquidityNotifyWebhookFailed => { + write!(f, "Failed to send a webhook notification to a client.") + }, } } } diff --git a/src/event.rs b/src/event.rs index be54969c7a..5ffa459696 100644 --- a/src/event.rs +++ b/src/event.rs @@ -19,10 +19,12 @@ use lightning::events::bump_transaction::BumpTransactionEvent; #[cfg(not(feature = "uniffi"))] use lightning::events::PaidBolt12Invoice; use lightning::events::{ - ClosureReason, Event as LdkEvent, FundingInfo, HTLCLocator as LdkHtlcLocator, - PaymentFailureReason, PaymentPurpose, ReplayEvent, + ClosureReason, Event as LdkEvent, FundingInfo, HTLCHandlingFailureReason, + HTLCHandlingFailureType, HTLCLocator as LdkHtlcLocator, PaymentFailureReason, PaymentPurpose, + ReplayEvent, }; use lightning::ln::channelmanager::{PaymentId, TrustedChannelFeatures}; +use lightning::ln::onion_utils::LocalHTLCFailureReason; use lightning::ln::types::ChannelId; use lightning::routing::gossip::NodeId; use lightning::sign::EntropySource; @@ -1534,11 +1536,29 @@ where prober.handle_background_probe_failed(&path, payment_id); } }, - LdkEvent::HTLCHandlingFailed { failure_type, .. } => { + LdkEvent::HTLCHandlingFailed { failure_type, failure_reason, .. } => { + // Capture the client's node id before `failure_type` is consumed below. A forward + // that failed only because the next-hop peer was offline is our cue to wake an + // LSPS5 client. The HTLC is failed back as `temporary_channel_failure`, which is + // not permanent, so the sender can retry once the client is online. + let offline_node_id = match (&failure_type, &failure_reason) { + ( + HTLCHandlingFailureType::Forward { node_id: Some(node_id), .. }, + Some(HTLCHandlingFailureReason::Local { + reason: LocalHTLCFailureReason::PeerOffline, + }), + ) => Some(*node_id), + _ => None, + }; + self.liquidity_source .lsps2_service() .handle_htlc_handling_failed(failure_type) .await; + + if let Some(node_id) = offline_node_id { + self.liquidity_source.lsps5_service().notify_payment_incoming(node_id); + } }, LdkEvent::SpendableOutputs { outputs, channel_id, counterparty_node_id } => { match self @@ -2022,6 +2042,8 @@ where debug_assert!(false, "We currently don't handle BOLT12 invoices manually, so this event should never be emitted."); }, LdkEvent::ConnectionNeeded { node_id, addresses } => { + self.liquidity_source.lsps5_service().notify_onion_message_incoming(node_id); + let spawn_logger = self.logger.clone(); let spawn_cm = Arc::clone(&self.connection_manager); let future = async move { @@ -2080,6 +2102,9 @@ where "Onion message intercepted, but no onion message mailbox available" ); } + self.liquidity_source + .lsps5_service() + .notify_onion_message_incoming(peer_node_id); } else { log_error!(self.logger, "Onion message intercepted for unknown SCID"); } diff --git a/src/liquidity/client/lsps5.rs b/src/liquidity/client/lsps5.rs new file mode 100644 index 0000000000..de4db8fa73 --- /dev/null +++ b/src/liquidity/client/lsps5.rs @@ -0,0 +1,677 @@ +// 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::HashMap; +use std::ops::Deref; +use std::sync::{Arc, Mutex, RwLock}; +use std::time::Duration; + +use bitcoin::secp256k1::PublicKey; +use lightning::log_debug; +use lightning_liquidity::lsps0::ser::LSPSRequestId; +use lightning_liquidity::lsps5::event::LSPS5ClientEvent; +use lightning_liquidity::lsps5::msgs::{ + LSPS5Error, ListWebhooksResponse, RemoveWebhookResponse, SetWebhookResponse, +}; +use tokio::sync::oneshot; + +use crate::connection::ConnectionManager; +use crate::liquidity::service::lsps5::LSPS5ServiceLiquiditySource; +use crate::liquidity::{ + select_all_lsps_for_protocol, select_lsps_for_protocol, LspConfig, LspNode, + LIQUIDITY_REQUEST_TIMEOUT_SECS, LSPS_DISCOVERY_WAIT_TIMEOUT_SECS, +}; +use crate::logger::{log_error, log_info, LdkLogger, Logger}; +use crate::runtime::Runtime; +use crate::types::LiquidityManager; +use crate::Error; + +pub(crate) struct LSPS5Client +where + L::Target: LdkLogger, +{ + pub(crate) lsp_nodes: Arc>>, + pub(crate) pending_set_webhook_requests: + Mutex>>>, + pub(crate) pending_list_webhooks_requests: + Mutex>>>, + pub(crate) pending_remove_webhook_requests: + Mutex>>>, + pub(crate) discovery_done_rx: tokio::sync::watch::Receiver, + pub(crate) liquidity_manager: Arc, + pub(crate) logger: L, +} + +impl LSPS5Client +where + L::Target: LdkLogger, +{ + pub(crate) async fn lsps5_set_webhook( + &self, app_name: String, webhook_url: String, node_id: Option<&PublicKey>, + ) -> Result { + let lsps5_node = select_lsps_for_protocol(&self.lsp_nodes, 5, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + let client_handler = self.liquidity_manager.lsps5_client_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 liquidity client was not configured."); + Error::LiquiditySourceUnavailable + })?; + + let (sender, receiver) = oneshot::channel(); + { + let mut pending_set_webhook_requests_lock = + self.pending_set_webhook_requests.lock().expect("lock"); + + let request_id = match client_handler.set_webhook( + lsps5_node.node_id, + app_name.clone(), + webhook_url.clone(), + ) { + Ok(request_id) => request_id, + Err(e) => { + log_error!( + self.logger, + "Failed to send set webhook request to liquidity service: {:?}", + e + ); + return Err(Error::LiquiditySetWebhookFailed); + }, + }; + + pending_set_webhook_requests_lock.insert(request_id, sender); + } + + match tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) + .await + { + Ok(Ok(result)) => result.map(LSPS5SetWebhookResponse::from).map_err(|e| { + log_error!(self.logger, "Failed to set webhook: {:?}", e); + Error::LiquiditySetWebhookFailed + }), + Ok(Err(e)) => { + log_error!( + self.logger, + "Failed to handle response from liquidity service: {:?}", + e + ); + Err(Error::LiquidityRequestFailed) + }, + Err(e) => { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Err(Error::LiquidityRequestFailed) + }, + } + } + + pub(crate) async fn lsps5_list_webhooks( + &self, node_id: Option<&PublicKey>, + ) -> Result { + let lsps5_node = select_lsps_for_protocol(&self.lsp_nodes, 5, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + let client_handler = self.liquidity_manager.lsps5_client_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 liquidity client was not configured."); + Error::LiquiditySourceUnavailable + })?; + + let (sender, receiver) = oneshot::channel(); + { + let mut pending_list_webhooks_requests_lock = + self.pending_list_webhooks_requests.lock().expect("lock"); + let request_id = client_handler.list_webhooks(lsps5_node.node_id); + pending_list_webhooks_requests_lock.insert(request_id.clone(), sender); + } + + match tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) + .await + { + Ok(Ok(result)) => result.map(LSPS5ListWebhooksResponse::from).map_err(|e| { + log_error!(self.logger, "Failed to list webhooks: {:?}", e); + Error::LiquidityListWebhooksFailed + }), + Ok(Err(e)) => { + log_error!( + self.logger, + "Failed to handle response from liquidity service: {:?}", + e + ); + Err(Error::LiquidityRequestFailed) + }, + Err(e) => { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Err(Error::LiquidityRequestFailed) + }, + } + } + + pub(crate) async fn lsps5_remove_webhook( + &self, app_name: String, node_id: Option<&PublicKey>, + ) -> Result<(), Error> { + let lsps5_node = select_lsps_for_protocol(&self.lsp_nodes, 5, node_id) + .ok_or(Error::LiquiditySourceUnavailable)?; + let client_handler = self.liquidity_manager.lsps5_client_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 liquidity client was not configured."); + Error::LiquiditySourceUnavailable + })?; + + let (sender, receiver) = oneshot::channel(); + { + let mut pending_remove_webhook_requests_lock = + self.pending_remove_webhook_requests.lock().expect("lock"); + let request_id = + match client_handler.remove_webhook(lsps5_node.node_id, app_name.clone()) { + Ok(request_id) => request_id, + Err(e) => { + log_error!( + self.logger, + "Failed to send remove webhook request to liquidity service: {:?}", + e + ); + return Err(Error::LiquidityRemoveWebhookFailed); + }, + }; + + pending_remove_webhook_requests_lock.insert(request_id.clone(), sender); + } + + match tokio::time::timeout(Duration::from_secs(LIQUIDITY_REQUEST_TIMEOUT_SECS), receiver) + .await + { + Ok(Ok(result)) => result.map(|_| ()).map_err(|e| { + log_error!(self.logger, "Failed to remove webhook: {:?}", e); + Error::LiquidityRemoveWebhookFailed + }), + Ok(Err(e)) => { + log_error!( + self.logger, + "Failed to handle response from liquidity service: {:?}", + e + ); + Err(Error::LiquidityRequestFailed) + }, + Err(e) => { + log_error!(self.logger, "Liquidity request timed out: {}", e); + Err(Error::LiquidityRequestFailed) + }, + } + } + + pub(crate) async fn handle_event(&self, event: LSPS5ClientEvent) { + match event { + LSPS5ClientEvent::WebhookRegistered { + request_id, + counterparty_node_id, + num_webhooks, + max_webhooks, + no_change, + .. + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRegistered".into(), + ) { + return; + } + + let response = Ok(SetWebhookResponse { num_webhooks, max_webhooks, no_change }); + self.deliver_response(&self.pending_set_webhook_requests, &request_id, response); + }, + LSPS5ClientEvent::WebhookRegistrationFailed { + request_id, + counterparty_node_id, + error, + app_name, + url, + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRegistrationFailed".into(), + ) { + return; + } + + log_error!( + self.logger, + "Webhook registration failed for app '{}' with url '{}': {:?}", + app_name.as_str(), + url.as_str(), + error + ); + self.deliver_response(&self.pending_set_webhook_requests, &request_id, Err(error)); + }, + LSPS5ClientEvent::WebhooksListed { + request_id, + counterparty_node_id, + app_names, + max_webhooks, + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhooksListed".into(), + ) { + return; + } + + let response = Ok(ListWebhooksResponse { app_names, max_webhooks }); + self.deliver_response(&self.pending_list_webhooks_requests, &request_id, response); + }, + LSPS5ClientEvent::WebhookRemoved { request_id, counterparty_node_id, .. } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRemoved".into(), + ) { + return; + } + + self.deliver_response( + &self.pending_remove_webhook_requests, + &request_id, + Ok(RemoveWebhookResponse {}), + ); + }, + LSPS5ClientEvent::WebhookRemovalFailed { + request_id, + counterparty_node_id, + error, + app_name, + } => { + if !self.is_expected_counterparty( + &counterparty_node_id, + "LSPS5Client::WebhookRemovalFailed".into(), + ) { + return; + } + + log_error!( + self.logger, + "Webhook removal failed for app '{}': {:?}", + app_name.as_str(), + error + ); + self.deliver_response( + &self.pending_remove_webhook_requests, + &request_id, + Err(error), + ); + }, + } + } + + fn is_expected_counterparty(&self, counterparty_node_id: &PublicKey, event: String) -> bool { + if self.lsp_nodes.read().expect("lock").iter().any(|n| n.node_id == *counterparty_node_id) { + true + } else { + log_error!(self.logger, "Received unexpected {} event!", event); + false + } + } + + fn deliver_response( + &self, pending: &Mutex>>, + request_id: &LSPSRequestId, response: T, + ) { + match pending.lock().expect("lock").remove(request_id) { + Some(sender) => { + if sender.send(response).is_err() { + log_error!( + self.logger, + "Failed to handle response for request {:?} from liquidity service", + request_id + ); + } + }, + None => { + debug_assert!( + false, + "Received response from liquidity service for unknown request." + ); + log_error!( + self.logger, + "Received response from liquidity service for unknown request." + ); + }, + } + } + + async fn get_lsps5_node( + &self, override_node_id: Option<&PublicKey>, + ) -> Result { + if let Some(node) = select_lsps_for_protocol(&self.lsp_nodes, 5, override_node_id) { + return Ok(node); + } + + let has_undiscovered_protocol = + self.lsp_nodes.read().expect("lock").iter().any(|n| n.supported_protocols.is_none()); + + // LSP protocol discovery may still be in flight, we wait briefly for it to finish, then re-check. + if has_undiscovered_protocol && !*self.discovery_done_rx.borrow() { + log_debug!( + self.logger, + "No LSPS5 node available yet, waiting for protocol discovery to complete." + ); + let mut rx = self.discovery_done_rx.clone(); + let _ = tokio::time::timeout( + Duration::from_secs(LSPS_DISCOVERY_WAIT_TIMEOUT_SECS), + rx.wait_for(|done| *done), + ) + .await; + } + + select_lsps_for_protocol(&self.lsp_nodes, 5, override_node_id) + .ok_or(Error::LiquiditySourceUnavailable) + } + + /// Resolves the set of LSPs a webhook operation should target. + /// + /// If `override_node_id` is `Some`, only that LSP is targeted (returned as a single-element + /// list). If it is `None`, every LSP that supports LSPS5 is targeted, waiting briefly for + /// protocol discovery to complete if it is still in flight. + async fn get_lsps5_nodes( + &self, override_node_id: Option<&PublicKey>, + ) -> Result, Error> { + if override_node_id.is_some() { + return self.get_lsps5_node(override_node_id).await.map(|node| vec![node]); + } + + let has_undiscovered_protocol = + self.lsp_nodes.read().expect("lock").iter().any(|n| n.supported_protocols.is_none()); + + // LSP protocol discovery may still be in flight, we wait briefly for it to finish, then re-check. + if has_undiscovered_protocol && !*self.discovery_done_rx.borrow() { + log_debug!( + self.logger, + "No LSPS5 node available yet, waiting for protocol discovery to complete." + ); + let mut rx = self.discovery_done_rx.clone(); + let _ = tokio::time::timeout( + Duration::from_secs(LSPS_DISCOVERY_WAIT_TIMEOUT_SECS), + rx.wait_for(|done| *done), + ) + .await; + } + + let lsps5_nodes = select_all_lsps_for_protocol(&self.lsp_nodes, 5); + if lsps5_nodes.is_empty() { + return Err(Error::LiquiditySourceUnavailable); + } + Ok(lsps5_nodes) + } +} + +/// The response to a [`LSPS5Liquidity::set_webhook`] request. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct LSPS5SetWebhookResponse { + /// The current number of webhooks registered for this client. + pub num_webhooks: u32, + /// The maximum number of webhooks allowed by the LSP. + pub max_webhooks: u32, + /// Whether this was an unchanged registration (same `app_name` and URL). + pub no_change: bool, +} + +impl From for LSPS5SetWebhookResponse { + fn from(response: SetWebhookResponse) -> Self { + Self { + num_webhooks: response.num_webhooks, + max_webhooks: response.max_webhooks, + no_change: response.no_change, + } + } +} + +/// The result of registering a webhook with a single LSP via [`LSPS5Liquidity::set_webhook`]. +/// +/// A call to [`LSPS5Liquidity::set_webhook`] may target multiple LSPs (when no specific `node_id` +/// is given), so one of these is returned per LSP that accepted the registration. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct LSPS5SetWebhookResult { + /// The node ID of the LSP the webhook was registered with. + pub counterparty_node_id: PublicKey, + /// The LSP's response to the registration. + pub response: LSPS5SetWebhookResponse, +} + +/// The response to a [`LSPS5Liquidity::list_webhooks`] request. +#[derive(Debug, Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Record))] +pub struct LSPS5ListWebhooksResponse { + /// The app names with a currently registered webhook. + pub app_names: Vec, + /// The maximum number of webhooks allowed by the LSP. + pub max_webhooks: u32, +} + +impl From for LSPS5ListWebhooksResponse { + fn from(response: ListWebhooksResponse) -> Self { + Self { + app_names: response + .app_names + .into_iter() + .map(|name| name.as_str().to_string()) + .collect(), + max_webhooks: response.max_webhooks, + } + } +} + +/// A liquidity handler for managing LSPS5 webhook notifications. +/// +/// Should be retrieved by calling [`Liquidity::lsps5`]. +/// +/// On the client side, this handler allows registering webhook endpoints with an LSP to receive +/// push notifications for Lightning events while offline. On the service side, it allows notifying +/// clients about such events. +/// +/// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md +/// [`Liquidity::lsps5`]: crate::Liquidity::lsps5 +#[derive(Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct LSPS5Liquidity { + runtime: Arc, + connection_manager: Arc>>, + liquidity_source: Arc>>, + lsps5_service: Arc>>, + logger: Arc, +} + +impl LSPS5Liquidity { + pub(crate) fn new( + runtime: Arc, connection_manager: Arc>>, + liquidity_source: Arc>>, + lsps5_service: Arc>>, logger: Arc, + ) -> Self { + Self { runtime, connection_manager, liquidity_source, lsps5_service, logger } + } +} + +#[cfg_attr(feature = "uniffi", uniffi::export)] +impl LSPS5Liquidity { + /// Connects to one or more LSPs and registers a webhook URL for receiving LSPS5 notifications. + /// + /// The webhook will receive signed push notifications for Lightning events such as incoming + /// payments while the client is offline. Because each LSP stores and calls the webhook + /// independently, the webhook has to be registered with every LSP that should be able to reach + /// the client. + /// + /// If `node_id` is `Some`, the webhook is registered with that single LSP only. If `node_id` + /// is `None`, the webhook is registered with *every* LSP that supports LSPS5 (as added via + /// [`crate::Builder::add_liquidity_source`] or [`crate::Liquidity::add_liquidity_source`]), + /// allowing the webhook to be configured once for all of them. + /// + /// Returns one [`LSPS5SetWebhookResult`] per LSP that accepted the registration. When targeting + /// multiple LSPs, registration is best-effort: failures for individual LSPs are logged and + /// skipped rather than aborting the whole call. An error is only returned if no LSP supporting + /// LSPS5 is available, or if the registration failed for every targeted LSP. + pub fn set_webhook( + &self, app_name: String, webhook_url: String, node_id: Option, + ) -> Result, Error> { + let lsps5_nodes = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_nodes(node_id.as_ref()).await })?; + + let mut results = Vec::with_capacity(lsps5_nodes.len()); + let mut last_error = None; + for lsps5_node in lsps5_nodes { + let counterparty_node_id = lsps5_node.node_id; + + if let Err(e) = self.connect(&lsps5_node) { + log_error!( + self.logger, + "Failed to connect to LSPS5 node {} to set webhook: {:?}", + counterparty_node_id, + e + ); + last_error = Some(e); + continue; + } + + let liquidity_source = Arc::clone(&self.liquidity_source); + let app_name = app_name.clone(); + let webhook_url = webhook_url.clone(); + let result = self.runtime.block_on(async move { + liquidity_source + .lsps5_set_webhook(app_name, webhook_url, Some(&counterparty_node_id)) + .await + }); + + match result { + Ok(response) => { + results.push(LSPS5SetWebhookResult { counterparty_node_id, response }) + }, + Err(e) => { + log_error!( + self.logger, + "Failed to set webhook at LSPS5 node {}: {:?}", + counterparty_node_id, + e + ); + last_error = Some(e); + }, + } + } + + if results.is_empty() { + return Err(last_error.unwrap_or(Error::LiquiditySetWebhookFailed)); + } + + Ok(results) + } + + /// Connects to the configured LSP and lists all currently configured webhooks. + /// + /// If `node_id` is `None` and multiple LSPs support LSPS5, the first one registered + /// via [`crate::Builder::add_liquidity_source`] or [`crate::Liquidity::add_liquidity_source`] is used. + pub fn list_webhooks( + &self, node_id: Option, + ) -> Result { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(node_id.as_ref()).await })?; + + self.connect(&lsps5_node)?; + + let liquidity_source = Arc::clone(&self.liquidity_source); + self.runtime.block_on(async move { + liquidity_source.lsps5_list_webhooks(Some(&lsps5_node.node_id)).await + }) + } + + /// Connects to one or more LSPs and removes a previously-configured webhook. + /// + /// Because each LSP stores the webhook independently, removal has to be performed against every + /// LSP the webhook was registered with. + /// + /// If `node_id` is `Some`, the webhook is removed from that single LSP only. If `node_id` is + /// `None`, the webhook is removed from *every* LSP that supports LSPS5 (as added via + /// [`crate::Builder::add_liquidity_source`] or [`crate::Liquidity::add_liquidity_source`]). + /// + /// Returns the node IDs of the LSPs the webhook was successfully removed from. When targeting + /// multiple LSPs, removal is best-effort: failures for individual LSPs (including attempts to + /// remove a webhook that does not exist at that LSP) are logged and skipped. An error is only + /// returned if no LSP supporting LSPS5 is available, or if removal failed for every targeted LSP. + pub fn remove_webhook( + &self, app_name: String, node_id: Option, + ) -> Result, Error> { + let lsps5_nodes = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_nodes(node_id.as_ref()).await })?; + + let mut removed = Vec::with_capacity(lsps5_nodes.len()); + let mut last_error = None; + for lsps5_node in lsps5_nodes { + let counterparty_node_id = lsps5_node.node_id; + + if let Err(e) = self.connect(&lsps5_node) { + log_error!( + self.logger, + "Failed to connect to LSPS5 node {} to remove webhook: {:?}", + counterparty_node_id, + e + ); + last_error = Some(e); + continue; + } + + let liquidity_source = Arc::clone(&self.liquidity_source); + let app_name = app_name.clone(); + let result = self.runtime.block_on(async move { + liquidity_source.lsps5_remove_webhook(app_name, Some(&counterparty_node_id)).await + }); + + match result { + Ok(()) => removed.push(counterparty_node_id), + Err(e) => { + log_error!( + self.logger, + "Failed to remove webhook at LSPS5 node {}: {:?}", + counterparty_node_id, + e + ); + last_error = Some(e); + }, + } + } + + if removed.is_empty() { + return Err(last_error.unwrap_or(Error::LiquidityRemoveWebhookFailed)); + } + + Ok(removed) + } + + /// Notifies a client that we intend to manage the liquidity on their channels. + /// + /// Should be called by LSP operators when their own policy decides to reclaim or adjust + /// liquidity for `client_node_id` (e.g. before closing or splicing a channel), so the client + /// has a chance to come online and cooperate. Sends a notification to every webhook the + /// client has registered with us. + /// + /// Note that notifications are rate limited per client, so calling this repeatedly in quick + /// succession will fail after the first call. + pub fn notify_liquidity_management_request( + &self, client_node_id: PublicKey, + ) -> Result<(), Error> { + self.lsps5_service.notify_liquidity_management_request(client_node_id) + } +} + +impl LSPS5Liquidity { + fn connect(&self, lsps5_node: &LspConfig) -> Result<(), Error> { + let con_node_id = lsps5_node.node_id; + let con_addr = lsps5_node.address.clone(); + let con_cm = Arc::clone(&self.connection_manager); + + // We need to use our main runtime here as a local runtime might not be around to poll + // connection futures going forward. + self.runtime.block_on(async move { + con_cm.connect_peer_if_necessary(con_node_id, con_addr).await + })?; + + log_info!(self.logger, "Connected to LSP {}@{}. ", lsps5_node.node_id, lsps5_node.address); + Ok(()) + } +} diff --git a/src/liquidity/client/mod.rs b/src/liquidity/client/mod.rs index 52fad2da20..f208eb1932 100644 --- a/src/liquidity/client/mod.rs +++ b/src/liquidity/client/mod.rs @@ -7,5 +7,6 @@ pub(crate) mod lsps1; pub(crate) mod lsps2; +pub(crate) mod lsps5; pub use lsps1::LSPS1OrderStatus; diff --git a/src/liquidity/mod.rs b/src/liquidity/mod.rs index 0eddde1ae8..2083461953 100644 --- a/src/liquidity/mod.rs +++ b/src/liquidity/mod.rs @@ -18,6 +18,7 @@ use std::time::Duration; use bitcoin::secp256k1::PublicKey; pub use client::lsps1::LSPS1Liquidity; +pub use client::lsps5::{LSPS5Liquidity, LSPS5ListWebhooksResponse, LSPS5SetWebhookResponse}; pub use client::LSPS1OrderStatus; use lightning::ln::msgs::SocketAddress; use lightning_liquidity::events::LiquidityEvent; @@ -25,6 +26,8 @@ use lightning_liquidity::lsps0::event::LSPS0ClientEvent; use lightning_liquidity::lsps1::client::LSPS1ClientConfig as LdkLSPS1ClientConfig; use lightning_liquidity::lsps2::client::LSPS2ClientConfig as LdkLSPS2ClientConfig; use lightning_liquidity::lsps2::service::LSPS2ServiceConfig as LdkLSPS2ServiceConfig; +use lightning_liquidity::lsps5::client::LSPS5ClientConfig as LdkLSPS5ClientConfig; +use lightning_liquidity::lsps5::service::LSPS5ServiceConfig as LdkLSPS5ServiceConfig; use lightning_liquidity::{LiquidityClientConfig, LiquidityServiceConfig}; pub use service::lsps2::LSPS2ServiceConfig; use tokio::sync::oneshot; @@ -33,7 +36,9 @@ use crate::builder::BuildError; use crate::connection::ConnectionManager; use crate::liquidity::client::lsps1::LSPS1Client; use crate::liquidity::client::lsps2::LSPS2Client; +use crate::liquidity::client::lsps5::LSPS5Client; use crate::liquidity::service::lsps2::{LSPS2Service, LSPS2ServiceLiquiditySource}; +use crate::liquidity::service::lsps5::LSPS5ServiceLiquiditySource; use crate::logger::{log_debug, log_error, log_info, LdkLogger, Logger}; use crate::runtime::Runtime; use crate::types::{Broadcaster, ChannelManager, DynStore, KeysManager, LiquidityManager, Wallet}; @@ -176,6 +181,23 @@ impl Liquidity { Arc::clone(&self.logger), ) } + + /// Returns a liquidity handler for managing webhook registrations via the [bLIP-55 / LSPS5] + /// protocol. + /// + /// This allows registering, listing, and removing webhook endpoints with an LSP in order to + /// receive push notifications for events such as incoming payments while the client is offline. + /// + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md + pub fn lsps5(&self) -> LSPS5Liquidity { + LSPS5Liquidity::new( + Arc::clone(&self.runtime), + Arc::clone(&self.connection_manager), + self.liquidity_source.lsps5_client(), + self.liquidity_source.lsps5_service(), + Arc::clone(&self.logger), + ) + } } #[derive(Debug, Clone)] @@ -201,12 +223,14 @@ where { lsp_nodes: Vec, lsps2_service: Option, + lsps5_service: Option, wallet: Arc, channel_manager: Arc, keys_manager: Arc, tx_broadcaster: Arc, kv_store: Arc, config: Arc, + runtime: Arc, logger: L, } @@ -216,19 +240,23 @@ where { pub(crate) fn new( wallet: Arc, channel_manager: Arc, keys_manager: Arc, - tx_broadcaster: Arc, kv_store: Arc, config: Arc, logger: L, + tx_broadcaster: Arc, kv_store: Arc, config: Arc, + runtime: Arc, logger: L, ) -> Self { let lsp_nodes = Vec::new(); let lsps2_service = None; + let lsps5_service = None; Self { lsp_nodes, lsps2_service, + lsps5_service, wallet, channel_manager, keys_manager, tx_broadcaster, kv_store, config, + runtime, logger, } } @@ -246,18 +274,32 @@ where self } + pub(crate) fn lsps5_service(&mut self, service_config: LdkLSPS5ServiceConfig) -> &mut Self { + self.lsps5_service = Some(service_config); + self + } + pub(crate) async fn build(self) -> Result, BuildError> { - let liquidity_service_config = self.lsps2_service.as_ref().map(|s| { - let lsps2_service_config = Some(s.ldk_service_config.clone()); - let lsps5_service_config = None; - let advertise_service = s.service_config.advertise_service; - LiquidityServiceConfig { - lsps1_service_config: None, - lsps2_service_config, - lsps5_service_config, - advertise_service, - } - }); + let lsps2_service_config = + self.lsps2_service.as_ref().map(|s| s.ldk_service_config.clone()); + let lsps5_service_config = self.lsps5_service.clone(); + let advertise_service = self + .lsps2_service + .as_ref() + .map(|s| s.service_config.advertise_service) + .unwrap_or(false); + + let liquidity_service_config = + if lsps2_service_config.is_some() || lsps5_service_config.is_some() { + Some(LiquidityServiceConfig { + lsps1_service_config: None, + lsps2_service_config, + lsps5_service_config, + advertise_service, + }) + } else { + None + }; let (discovery_done_tx, discovery_done_rx) = tokio::sync::watch::channel(false); @@ -266,7 +308,7 @@ where let liquidity_client_config = Some(LiquidityClientConfig { lsps1_client_config: Some(LdkLSPS1ClientConfig { max_channel_fees_msat: None }), lsps2_client_config: Some(LdkLSPS2ClientConfig {}), - lsps5_client_config: None, + lsps5_client_config: Some(LdkLSPS5ClientConfig {}), }); let liquidity_manager = Arc::new( @@ -328,6 +370,21 @@ where config: self.config.clone(), logger: self.logger.clone(), }), + lsps5_client: Arc::new(LSPS5Client { + lsp_nodes: Arc::clone(&lsp_nodes), + pending_set_webhook_requests: Mutex::new(HashMap::new()), + pending_list_webhooks_requests: Mutex::new(HashMap::new()), + pending_remove_webhook_requests: Mutex::new(HashMap::new()), + discovery_done_rx: discovery_done_rx.clone(), + liquidity_manager: Arc::clone(&liquidity_manager), + logger: self.logger.clone(), + }), + lsps5_service: Arc::new(LSPS5ServiceLiquiditySource { + liquidity_manager: Arc::clone(&liquidity_manager), + peer_manager: RwLock::new(None), + runtime: Arc::clone(&self.runtime), + logger: self.logger.clone(), + }), pending_lsps0_discovery: Mutex::new(HashMap::new()), discovery_done_tx, discovery_done_rx, @@ -345,6 +402,8 @@ where lsps1_client: Arc>, lsps2_client: Arc>, lsps2_service: Arc>, + lsps5_client: Arc>, + lsps5_service: Arc>, pending_lsps0_discovery: Mutex>>>>, discovery_done_tx: tokio::sync::watch::Sender, discovery_done_rx: tokio::sync::watch::Receiver, @@ -372,11 +431,24 @@ where Arc::clone(&self.lsps2_service) } - pub(crate) async fn handle_next_event(&self) { + pub(crate) fn lsps5_client(&self) -> Arc> { + Arc::clone(&self.lsps5_client) + } + + pub(crate) fn lsps5_service(&self) -> Arc> { + Arc::clone(&self.lsps5_service) + } + + pub(crate) async fn handle_next_event(&self) + where + L: Clone + Send + Sync + 'static, + { match self.liquidity_manager.next_event_async().await { LiquidityEvent::LSPS1Client(event) => self.lsps1_client.handle_event(event).await, LiquidityEvent::LSPS2Client(event) => self.lsps2_client.handle_event(event).await, LiquidityEvent::LSPS2Service(event) => self.lsps2_service.handle_event(event).await, + LiquidityEvent::LSPS5Client(event) => self.lsps5_client.handle_event(event).await, + LiquidityEvent::LSPS5Service(event) => self.lsps5_service.handle_event(event).await, LiquidityEvent::LSPS0Client(LSPS0ClientEvent::ListProtocolsResponse { counterparty_node_id, diff --git a/src/liquidity/service/lsps5.rs b/src/liquidity/service/lsps5.rs new file mode 100644 index 0000000000..fea4925f0c --- /dev/null +++ b/src/liquidity/service/lsps5.rs @@ -0,0 +1,213 @@ +// 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::ops::Deref; +use std::sync::{Arc, RwLock, Weak}; + +use crate::runtime::Runtime; +use bitcoin::secp256k1::PublicKey; +use lightning_liquidity::lsps5::event::LSPS5ServiceEvent; +use lightning_liquidity::lsps5::msgs::LSPS5ProtocolError; + +use crate::config::{LSPS5_WEBHOOK_MAX_RESPONSE_SIZE, LSPS5_WEBHOOK_TIMEOUT_SECS}; +use crate::logger::{log_debug, log_error, log_info, LdkLogger}; +use crate::types::{LiquidityManager, PeerManager}; +use crate::Error; + +pub(crate) struct LSPS5ServiceLiquiditySource +where + L::Target: LdkLogger, +{ + pub(crate) liquidity_manager: Arc, + pub(crate) peer_manager: RwLock>>, + pub(crate) runtime: Arc, + pub(crate) logger: L, +} + +impl LSPS5ServiceLiquiditySource +where + L::Target: LdkLogger, +{ + pub(crate) fn set_peer_manager(&self, peer_manager: Weak) { + *self.peer_manager.write().expect("lock") = Some(peer_manager); + } + + pub(crate) fn notify_payment_incoming(&self, client_id: PublicKey) { + if !self.is_client_offline(&client_id) { + return; + } + + if let Some(handler) = self.liquidity_manager.lsps5_service_handler() { + handler.notify_payment_incoming(client_id).unwrap_or_else(|e| match e { + LSPS5ProtocolError::SlowDownError => log_debug!( + self.logger, + "Skipping notify payment incoming for client {}: rate limited.", + client_id + ), + _ => log_error!( + self.logger, + "Failed to notify payment incoming for client {}: {:?}", + client_id, + e + ), + }) + } else { + log_error!(self.logger, "LSPS5 service handler is not available."); + } + } + + pub(crate) fn notify_expiry_soon(&self, client_id: PublicKey, timeout: u32) { + if !self.is_client_offline(&client_id) { + return; + } + + if let Some(handler) = self.liquidity_manager.lsps5_service_handler() { + handler.notify_expiry_soon(client_id, timeout).unwrap_or_else(|e| match e { + LSPS5ProtocolError::SlowDownError => log_debug!( + self.logger, + "Skipping notify expiry soon for client {}: rate limited.", + client_id + ), + _ => log_error!( + self.logger, + "Failed to notify expiry soon for client {}: {:?}", + client_id, + e + ), + }) + } else { + log_error!(self.logger, "LSPS5 service handler is not available."); + } + } + + pub(crate) fn notify_liquidity_management_request( + &self, client_id: PublicKey, + ) -> Result<(), Error> { + let handler = self.liquidity_manager.lsps5_service_handler().ok_or_else(|| { + log_error!(self.logger, "LSPS5 service handler is not available."); + Error::LiquiditySourceUnavailable + })?; + + handler.notify_liquidity_management_request(client_id).map_err(|e| { + match e { + LSPS5ProtocolError::SlowDownError => log_debug!( + self.logger, + "Skipping liquidity management request notification for client {}: rate limited.", + client_id + ), + _ => log_error!( + self.logger, + "Failed to notify liquidity management request for client {}: {:?}", + client_id, + e + ), + } + Error::LiquidityNotifyWebhookFailed + }) + } + + pub(crate) fn notify_onion_message_incoming(&self, client_id: PublicKey) { + if !self.is_client_offline(&client_id) { + return; + } + + if let Some(handler) = self.liquidity_manager.lsps5_service_handler() { + handler.notify_onion_message_incoming(client_id).unwrap_or_else(|e| match e { + LSPS5ProtocolError::SlowDownError => log_debug!( + self.logger, + "Skipping onion message incoming notification for client {}: rate limited.", + client_id + ), + _ => log_error!( + self.logger, + "Failed to notify onion message incoming for client {}: {:?}", + client_id, + e + ), + }) + } else { + log_error!(self.logger, "LSPS5 service handler is not available."); + } + } + + pub(crate) async fn handle_event(&self, event: LSPS5ServiceEvent) + where + L: Clone + Send + Sync + 'static, + { + match event { + LSPS5ServiceEvent::SendWebhookNotification { + counterparty_node_id: _, + app_name, + url, + notification, + headers, + } => { + if self.liquidity_manager.lsps5_service_handler().is_none() { + log_error!( + self.logger, + "Received unexpected LSPS5ServiceEvent::SendWebhookNotification event!" + ); + return; + } + + log_info!( + self.logger, + "Sending webhook notification for {} to {}: {:?}", + app_name.as_str(), + url.as_str(), + notification + ); + + let notification_body = notification.to_request_body(); + let logger = self.logger.clone(); + + // `url` is client-supplied, so awaiting delivery here would let any client stall + // the liquidity event loop for the full timeout. Deliver out of band instead. + self.runtime.spawn_cancellable_background_task(async move { + let result = bitreq::post(url.as_str()) + .with_headers(headers) + .with_body(notification_body) + .with_timeout(LSPS5_WEBHOOK_TIMEOUT_SECS) + .with_max_redirects(0) + .with_max_body_size(LSPS5_WEBHOOK_MAX_RESPONSE_SIZE) + .send_async() + .await; + + match result { + Ok(response) => { + if !(200..300).contains(&response.status_code) { + log_error!( + logger, + "Webhook call failed with status {} for {} to {}", + response.status_code, + app_name.as_str(), + url.as_str() + ); + } + }, + Err(e) => { + log_error!( + logger, + "Failed to send webhook notification for {} to {}: {}", + app_name.as_str(), + url.as_str(), + e + ); + }, + } + }) + }, + } + } + + fn is_client_offline(&self, client_id: &PublicKey) -> bool { + match self.peer_manager.read().expect("lock").as_ref().and_then(|w| w.upgrade()) { + Some(pm) => pm.peer_by_node_id(client_id).is_none(), + None => false, + } + } +} diff --git a/src/liquidity/service/mod.rs b/src/liquidity/service/mod.rs index cdbaf54265..9547d77fc4 100644 --- a/src/liquidity/service/mod.rs +++ b/src/liquidity/service/mod.rs @@ -6,3 +6,4 @@ // accordance with one or both of these licenses. pub(crate) mod lsps2; +pub(crate) mod lsps5; diff --git a/src/types.rs b/src/types.rs index 22429d980b..522ddb5c1d 100644 --- a/src/types.rs +++ b/src/types.rs @@ -377,6 +377,8 @@ pub(crate) type BumpTransactionEventHandler = pub(crate) type PaymentStore = DataStore>; +pub type LSPS5ServiceConfig = lightning_liquidity::lsps5::service::LSPS5ServiceConfig; + /// 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/tests/integration_tests_rust.rs b/tests/integration_tests_rust.rs index c1e973091d..c2c221e831 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -3449,7 +3449,7 @@ async fn do_lsps2_client_service_integration(client_trusts_lsp: bool) { let service_config = random_config(); setup_builder!(service_builder, service_config.node_config); service_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - service_builder.enable_liquidity_provider(lsps2_service_config); + service_builder.enable_liquidity_provider(Some(lsps2_service_config), None); let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); service_node.start().unwrap(); @@ -3771,7 +3771,7 @@ async fn lsps2_client_trusts_lsp() { let service_config = random_config(); setup_builder!(service_builder, service_config.node_config); service_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - service_builder.enable_liquidity_provider(lsps2_service_config); + service_builder.enable_liquidity_provider(Some(lsps2_service_config), None); let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); service_node.start().unwrap(); let service_node_id = service_node.node_id(); @@ -3948,7 +3948,7 @@ async fn lsps2_lsp_trusts_client_but_client_does_not_claim() { let service_config = random_config(); setup_builder!(service_builder, service_config.node_config); service_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - service_builder.enable_liquidity_provider(lsps2_service_config); + service_builder.enable_liquidity_provider(Some(lsps2_service_config), None); let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); service_node.start().unwrap(); @@ -4851,7 +4851,7 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { let cheap_node_config = random_config(); setup_builder!(cheap_builder, cheap_node_config.node_config); cheap_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - cheap_builder.enable_liquidity_provider(cheap_cfg); + cheap_builder.enable_liquidity_provider(Some(cheap_cfg), None); let cheap = cheap_builder.build(cheap_node_config.node_entropy.into()).unwrap(); cheap.start().unwrap(); let cheap_id = cheap.node_id(); @@ -4874,7 +4874,7 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { let expensive_node_config = random_config(); setup_builder!(expensive_builder, expensive_node_config.node_config); expensive_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); - expensive_builder.enable_liquidity_provider(expensive_cfg); + expensive_builder.enable_liquidity_provider(Some(expensive_cfg), None); let expensive = expensive_builder.build(expensive_node_config.node_entropy.into()).unwrap(); expensive.start().unwrap(); let expensive_id = expensive.node_id(); @@ -4915,3 +4915,186 @@ async fn do_lsps2_multi_lsp_picks_cheapest(reverse_order: bool) { cheap.stop().unwrap(); expensive.stop().unwrap(); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps5_webhook_registration() { + use lightning_liquidity::lsps5::service::LSPS5ServiceConfig; + + let (bitcoind, electrsd) = setup_bitcoind_and_electrsd(); + let esplora_url = format!("http://{}", electrsd.esplora_url.as_ref().unwrap()); + let sync_config = EsploraSyncConfig::default(); + + // Setup LSPS5 service provider node + let service_config = random_config(); + setup_builder!(service_builder, service_config.node_config); + service_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + let lsps5_service_config = LSPS5ServiceConfig { max_webhooks_per_client: 2 }; + service_builder.enable_liquidity_provider(None, Some(lsps5_service_config)); + let service_node = service_builder.build(service_config.node_entropy.into()).unwrap(); + service_node.start().unwrap(); + let service_node_id = service_node.node_id(); + let service_addr = service_node.onchain_payment().new_address().unwrap(); + let service_socket_addr = service_node.listening_addresses().unwrap().first().unwrap().clone(); + + // Setup LSPS5 client node + let client_config = random_config(); + setup_builder!(client_builder, client_config.node_config); + client_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + client_builder.add_liquidity_source(service_node_id, service_socket_addr.clone(), None, false); + let client_node = client_builder.build(client_config.node_entropy.into()).unwrap(); + client_node.start().unwrap(); + let client_node_id = client_node.node_id(); + let client_addr = client_node.onchain_payment().new_address().unwrap(); + + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![service_addr, client_addr], + Amount::from_sat(10_000_000), + ) + .await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + + open_channel(&client_node, &service_node, 5_000_000, false, &electrsd).await; + generate_blocks_and_wait(&bitcoind.client, &electrsd.client, 6).await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + expect_channel_ready_event!(client_node, service_node.node_id()); + expect_channel_ready_event!(service_node, client_node.node_id()); + + // Test webhook registration + let lsps5_client = client_node.liquidity().lsps5(); + let app_name_1 = "test-app".to_string(); + let webhook_url_1 = "https://example.com/webhook".to_string(); + + // Register first webhook + let results = lsps5_client + .set_webhook(app_name_1.clone(), webhook_url_1.clone(), None) + .expect("Failed to register webhook"); + assert_eq!(results.len(), 1, "Expected registration with the single configured LSP"); + assert_eq!( + results[0].counterparty_node_id, service_node_id, + "Expected the result to be attributed to the configured LSP" + ); + let response = &results[0].response; + assert_eq!(response.num_webhooks, 1, "Expected 1 webhook after first registration"); + assert_eq!(response.max_webhooks, 2, "Expected max_webhooks to be 2"); + assert!(!response.no_change, "Expected no_change to be false for new registration"); + + // Register second webhook with different app name + let app_name_2 = "test-app-2".to_string(); + let webhook_url_2 = "https://example.com/webhook-2".to_string(); + let results = lsps5_client + .set_webhook(app_name_2.clone(), webhook_url_2.clone(), Some(service_node_id)) + .expect("Failed to register second webhook"); + assert_eq!(results.len(), 1, "Expected a single result when targeting a single LSP"); + let response = &results[0].response; + assert_eq!(response.num_webhooks, 2, "Expected 2 webhooks after second registration"); + assert_eq!(response.max_webhooks, 2, "Expected max_webhooks to be 2"); + assert!(!response.no_change, "Expected no_change to be false for new registration"); + + // Register the same webhook again - should return no_change=true + let results = lsps5_client + .set_webhook(app_name_2.clone(), webhook_url_2.clone(), None) + .expect("Failed to re-register webhook"); + let response = &results[0].response; + assert_eq!(response.num_webhooks, 2, "Expected 2 webhooks after re-registering same webhook"); + assert_eq!(response.max_webhooks, 2, "Expected max_webhooks to be 2"); + assert!(response.no_change, "Expected no_change to be true for duplicate registration"); + + // Attempt to register a third webhook - should fail due to max_webhooks_per_client=2 + let app_name_3 = "test-app-3".to_string(); + let webhook_url_3 = "https://example.com/webhook-3".to_string(); + let result = lsps5_client.set_webhook(app_name_3.clone(), webhook_url_3.clone(), None); + assert_eq!( + result.err(), + Some(NodeError::LiquiditySetWebhookFailed), + "Expected error when exceeding max webhooks" + ); + + // List registered webhooks + let registered_webhooks = lsps5_client.list_webhooks(None).expect("Failed to list webhooks"); + assert_eq!(registered_webhooks.app_names.len(), 2, "Expected 2 registered webhooks"); + assert!( + registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_1), + "Expected app_name_1 in registered webhooks" + ); + assert!( + registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_2), + "Expected app_name_2 in registered webhooks" + ); + assert_eq!(registered_webhooks.max_webhooks, 2, "Expected max_webhooks to be 2"); + + // Attempt to delete non-existing webhook - should fail + let non_existing_app_name = "non-existing-app".to_string(); + let result = lsps5_client.remove_webhook(non_existing_app_name.clone(), None); + assert_eq!( + result.err(), + Some(NodeError::LiquidityRemoveWebhookFailed), + "Expected error when removing non-existing webhook" + ); + + // Delete a registered webhook + let removed_from = lsps5_client + .remove_webhook(app_name_1.clone(), None) + .expect("Failed to delete first webhook"); + assert_eq!( + removed_from, + vec![service_node_id], + "Expected the webhook to be removed from the single configured LSP" + ); + + // Verify webhook was deleted + let registered_webhooks = + lsps5_client.list_webhooks(None).expect("Failed to list webhooks after deletion"); + assert_eq!(registered_webhooks.app_names.len(), 1, "Expected 1 webhook after deletion"); + assert!( + registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_2), + "Expected app_name_2 to remain after deletion" + ); + assert!( + !registered_webhooks.app_names.iter().any(|name| name.as_str() == app_name_1), + "Expected app_name_1 to be removed" + ); + + // Requests targeting an LSP we don't know about should fail rather than silently fall back to + // another LSP. + let unknown_node_id = client_node_id; + assert_eq!( + lsps5_client.list_webhooks(Some(unknown_node_id)).err(), + Some(NodeError::LiquiditySourceUnavailable), + "Expected an error when targeting a node that is not a configured LSPS5 LSP" + ); + assert_eq!( + lsps5_client + .set_webhook(app_name_1.clone(), webhook_url_1.clone(), Some(unknown_node_id)) + .err(), + Some(NodeError::LiquiditySourceUnavailable), + "Expected an error when targeting a node that is not a configured LSPS5 LSP" + ); + + // Test service-side notifications. + let lsps5_service = service_node.liquidity().lsps5(); + + lsps5_service + .notify_liquidity_management_request(client_node_id) + .expect("notify_liquidity_management_request failed"); + + // Notifications are rate limited per client, so an immediate second notification is rejected. + assert_eq!( + lsps5_service.notify_liquidity_management_request(client_node_id).err(), + Some(NodeError::LiquidityNotifyWebhookFailed), + "Expected the notification to be rate limited" + ); + + // A node that isn't running the LSPS5 service can't notify anyone. + assert_eq!( + client_node.liquidity().lsps5().notify_liquidity_management_request(service_node_id).err(), + Some(NodeError::LiquiditySourceUnavailable), + "Expected an error when notifying without a configured LSPS5 service" + ); + + service_node.stop().unwrap(); + client_node.stop().unwrap(); +}