diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index 7c0edc5359..64435821ac 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -98,6 +98,7 @@ interface Node { OnchainPayment onchain_payment(); UnifiedPayment unified_payment(); Liquidity liquidity(); + LSPS5Liquidity lsps5_liquidity(); [Throws=NodeError] void lnurl_auth(string lnurl); [Throws=NodeError] @@ -169,6 +170,8 @@ typedef interface UnifiedPayment; typedef interface Liquidity; +typedef interface LSPS5Liquidity; + [Error] enum NodeError { "AlreadyRunning", @@ -231,6 +234,9 @@ enum NodeError { "LnurlAuthFailed", "LnurlAuthTimeout", "InvalidLnurl", + "LiquiditySetWebhookFailed", + "LiquidityRemoveWebhookFailed", + "LiquidityListWebhooksFailed" }; typedef dictionary NodeStatus; diff --git a/src/builder.rs b/src/builder.rs index 8b575cc3f2..846bc58fc3 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -78,8 +78,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::KVStoreWalletPersister; use crate::wallet::Wallet; @@ -122,10 +122,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)] @@ -514,6 +516,21 @@ impl NodeBuilder { self } + /// Configures the [`Node`] instance to provide an [bLIP-55 / LSPS5] service, enabling clients + /// to register webhooks for push notifications. + /// + /// **Caution**: LSP service support is in **alpha** and is considered an experimental feature. + /// + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md + pub fn enable_liquidity_provider_lsps5( + &mut self, lsps5_service_config: LSPS5ServiceConfig, + ) -> &mut Self { + let liquidity_source_config = + self.liquidity_source_config.get_or_insert(LiquiditySourceConfig::default()); + liquidity_source_config.lsps5_service = Some(lsps5_service_config); + self + } + /// Sets the used storage directory path. pub fn set_storage_dir_path(&mut self, storage_dir_path: String) -> &mut Self { self.config.storage_dir_path = storage_dir_path; @@ -1081,6 +1098,16 @@ impl ArcedNodeBuilder { self.inner.write().expect("lock").enable_liquidity_provider(lsps2_service_config); } + /// Configures the [`Node`] instance to provide an [bLIP-55 / LSPS5] service, enabling clients + /// to register webhooks for push notifications. + /// + /// **Caution**: LSP service support is in **alpha** and is considered an experimental feature. + /// + /// [bLIP-55 / LSPS5]: https://github.com/lightning/blips/blob/master/blip-0055.md + pub fn enable_liquidity_provider_lsps5(&self, lsps5_service_config: LSPS5ServiceConfig) { + self.inner.write().expect("lock").enable_liquidity_provider_lsps5(lsps5_service_config); + } + /// Sets the used storage directory path. pub fn set_storage_dir_path(&self, storage_dir_path: String) { self.inner.write().expect("lock").set_storage_dir_path(storage_dir_path); @@ -2081,6 +2108,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 @@ -2140,6 +2171,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/error.rs b/src/error.rs index d07212b008..b5e1116844 100644 --- a/src/error.rs +++ b/src/error.rs @@ -137,6 +137,12 @@ pub enum Error { LnurlAuthTimeout, /// The provided lnurl is invalid. InvalidLnurl, + /// 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, } impl fmt::Display for Error { @@ -222,6 +228,15 @@ impl fmt::Display for Error { Self::LnurlAuthFailed => write!(f, "LNURL-auth authentication failed."), Self::LnurlAuthTimeout => write!(f, "LNURL-auth authentication timed out."), Self::InvalidLnurl => write!(f, "The provided lnurl 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.") + }, } } } diff --git a/src/event.rs b/src/event.rs index 93d274ff7f..73dc10b173 100644 --- a/src/event.rs +++ b/src/event.rs @@ -1734,6 +1734,8 @@ where "Onion message intercepted, but no onion message mailbox available" ); } + + self.liquidity_source.lsps5_service().notify_onion_message_incoming(peer_node_id); }, LdkEvent::OnionMessagePeerConnected { peer_node_id } => { if let Some(om_mailbox) = self.om_mailbox.as_ref() { diff --git a/src/liquidity/client/lsps5.rs b/src/liquidity/client/lsps5.rs new file mode 100644 index 0000000000..9a27c77cc6 --- /dev/null +++ b/src/liquidity/client/lsps5.rs @@ -0,0 +1,659 @@ +// 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::{ + 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 [`Node::lsps5_liquidity`]. +/// +/// 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 +/// [`Node::lsps5_liquidity`]: crate::Node::lsps5_liquidity +#[derive(Clone)] +#[cfg_attr(feature = "uniffi", derive(uniffi::Object))] +pub struct LSPS5Liquidity { + runtime: Arc, + connection_manager: Arc>>, + liquidity_source: Arc>>, + logger: Arc, +} + +impl LSPS5Liquidity { + pub(crate) fn new( + runtime: Arc, connection_manager: Arc>>, + liquidity_source: Arc>>, logger: Arc, + ) -> Self { + Self { runtime, connection_manager, liquidity_source, 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) + } +} + +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 15ca7e9650..d6848e2574 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 87a0650c83..45527eeb9f 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,22 @@ 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(), + Arc::clone(&self.logger), + ) + } } #[derive(Debug, Clone)] @@ -201,6 +222,7 @@ where { lsp_nodes: Vec, lsps2_service: Option, + lsps5_service: Option, wallet: Arc, channel_manager: Arc, keys_manager: Arc, @@ -220,9 +242,11 @@ where ) -> 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, @@ -246,18 +270,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 +304,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 +366,20 @@ 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), + logger: self.logger.clone(), + }), pending_lsps0_discovery: Mutex::new(HashMap::new()), discovery_done_tx, discovery_done_rx, @@ -345,6 +397,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 +426,21 @@ where Arc::clone(&self.lsps2_service) } + 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) { 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..0ce2858978 --- /dev/null +++ b/src/liquidity/service/lsps5.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::ops::Deref; +use std::sync::{Arc, RwLock, Weak}; + +use bitcoin::secp256k1::PublicKey; +use lightning_liquidity::lsps5::event::LSPS5ServiceEvent; + +use crate::logger::{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) 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| { + 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, + ) -> Result<(), Error> { + let handler = self + .liquidity_manager + .lsps5_service_handler() + .ok_or(Error::LiquiditySourceUnavailable)?; + handler.notify_expiry_soon(client_id, timeout).map_err(|_| Error::LiquidityRequestFailed) + } + + pub(crate) fn notify_liquidity_management_request( + &self, client_id: PublicKey, + ) -> Result<(), Error> { + let handler = self + .liquidity_manager + .lsps5_service_handler() + .ok_or(Error::LiquiditySourceUnavailable)?; + handler + .notify_liquidity_management_request(client_id) + .map_err(|_| Error::LiquidityRequestFailed) + } + + 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| { + log_error!( + self.logger, + "Failed to notify onion message incoming for client {}: {:?}", + client_id, + e + ); + }) + } + } + + pub(crate) async fn handle_event(&self, event: LSPS5ServiceEvent) { + 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 result = bitreq::post(url.as_str()) + .with_headers(headers) + .with_body(notification_body) + .send_async() + .await; + + match result { + Ok(response) => { + if response.status_code != 200 { + log_error!( + self.logger, + "Webhook call failed with status {} for {} to {}", + response.status_code, + app_name.as_str(), + url.as_str() + ); + } + }, + Err(e) => { + log_error!( + self.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 5e3a3b1833..96351efc0b 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 e24db4d253..c331f920d2 100644 --- a/src/types.rs +++ b/src/types.rs @@ -373,6 +373,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 c3c2f4262b..ecfeeb094a 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -4036,3 +4036,142 @@ 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(true); + 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_lsps5(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(true); + 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.lsps5_liquidity(); + 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"); + 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(), None) + .expect("Failed to register second webhook"); + 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 + lsps5_client.remove_webhook(app_name_1.clone(), None).expect("Failed to delete first webhook"); + + // 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" + ); + + // Test service-side notification methods + let lsps5_service = service_node.lsps5_liquidity(); + + lsps5_service.notify_payment_incoming(client_node_id).expect("notify_payment_incoming failed"); + + service_node.stop().unwrap(); + client_node.stop().unwrap(); +}