From 9d4f022e863b10b3a482afdca461139d4ac79ed7 Mon Sep 17 00:00:00 2001 From: Camillarhi Date: Mon, 6 Jul 2026 14:28:32 +0100 Subject: [PATCH] Add LSPS5 (bLIP-55) webhook notification support Implement the bLIP-55 / LSPS5 webhook registration protocol on top of the multi-LSP liquidity module (src/liquidity/{client,service}). Client side, exposed via Node::liquidity().lsps5(): - set_webhook / list_webhooks / remove_webhook to manage webhook registrations with an LSP. - When no node_id is given, set_webhook and remove_webhook fan out to every LSPS5-capable LSP so a webhook can be configured once across all configured LSPs; set_webhook returns one result per LSP that accepted the registration and remove_webhook returns the LSPs it was removed from. Service side, enabled via Builder::enable_liquidity_provider_lsps5(): - Deliver outgoing webhook notifications over HTTP in response to LSPS5ServiceEvent::SendWebhookNotification. - Automatically send an onion-message-incoming notification when an intercepted onion message targets a client that is currently offline (wired from LdkEvent::OnionMessageIntercepted, gated on peer connectivity). Wires the feature through the UniFFI bindings and adds LSPS5-specific Error variants. --- bindings/ldk_node.udl | 8 + src/builder.rs | 55 +++- src/config.rs | 6 + src/error.rs | 50 +++ src/event.rs | 31 +- src/liquidity/client/lsps5.rs | 559 ++++++++++++++++++++++++++++++++ src/liquidity/client/mod.rs | 1 + src/liquidity/mod.rs | 100 +++++- src/liquidity/service/lsps5.rs | 219 +++++++++++++ src/liquidity/service/mod.rs | 1 + src/types.rs | 2 + tests/integration_tests_rust.rs | 322 +++++++++++++++++- 12 files changed, 1313 insertions(+), 41 deletions(-) create mode 100644 src/liquidity/client/lsps5.rs create mode 100644 src/liquidity/service/lsps5.rs diff --git a/bindings/ldk_node.udl b/bindings/ldk_node.udl index fddf5940ca..8322680da2 100644 --- a/bindings/ldk_node.udl +++ b/bindings/ldk_node.udl @@ -252,6 +252,14 @@ enum NodeError { "InvalidLnurl", "ChainSourceNotSupported", "InvalidPayerProof", + "LiquiditySetWebhookFailed", + "LiquidityRemoveWebhookFailed", + "LiquidityListWebhooksFailed", + "LiquidityNotifyWebhookFailed", + "LiquidityWebhookLimitExceeded", + "LiquidityWebhookNoPriorActivity", + "LiquidityWebhookAppNameNotFound", + "LiquidityWebhookInvalid" }; typedef dictionary NodeStatus; diff --git a/src/builder.rs b/src/builder.rs index f0f38783fb..7e93b7f3a8 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -85,8 +85,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; @@ -129,10 +129,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)] @@ -509,18 +511,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 } @@ -1114,14 +1124,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. @@ -2159,6 +2181,7 @@ fn build_with_store_internal( Arc::clone(&tx_broadcaster), Arc::clone(&kv_store), Arc::clone(&config), + Arc::clone(&runtime), Arc::clone(&logger), ); @@ -2177,6 +2200,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 @@ -2236,6 +2263,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 a409b9e48f..f0d3edb869 100644 --- a/src/config.rs +++ b/src/config.rs @@ -169,6 +169,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 9a03c446fa..77469b8e54 100644 --- a/src/error.rs +++ b/src/error.rs @@ -145,6 +145,29 @@ 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, + /// The LSP rejected a webhook registration because the client has reached the maximum number + /// of webhooks the LSP allows. + LiquidityWebhookLimitExceeded, + /// The LSP rejected a webhook registration because we have no prior activity with it. + /// + /// LSPs typically require an open channel, or an in-flight LSPS1 or LSPS2 flow, before + /// accepting webhook registrations. + LiquidityWebhookNoPriorActivity, + /// No webhook is registered under the given `app_name` with the LSP. + LiquidityWebhookAppNameNotFound, + /// The `app_name` or webhook URL is invalid. + /// + /// The `app_name` may exceed 64 bytes, or the URL may exceed 1024 bytes, fail to parse, or + /// not use the `https` scheme. + LiquidityWebhookInvalid, } impl fmt::Display for Error { @@ -236,6 +259,33 @@ 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.") + }, + Self::LiquidityWebhookLimitExceeded => { + write!( + f, + "The LSP's maximum number of webhooks for this client is already reached." + ) + }, + Self::LiquidityWebhookNoPriorActivity => { + write!(f, "The LSP rejected the webhook registration due to no prior activity.") + }, + Self::LiquidityWebhookAppNameNotFound => { + write!(f, "No webhook is registered under the given app name with this LSP.") + }, + Self::LiquidityWebhookInvalid => { + write!(f, "The given app name or webhook URL is invalid.") + }, } } } diff --git a/src/event.rs b/src/event.rs index 0a35697552..cf3b7182b0 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; @@ -1499,11 +1501,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 @@ -1987,6 +2007,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 { @@ -2045,6 +2067,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..952aca9ff6 --- /dev/null +++ b/src/liquidity/client/lsps5.rs @@ -0,0 +1,559 @@ +// 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, LSPS5ProtocolError, ListWebhooksResponse, RemoveWebhookResponse, SetWebhookResponse, +}; +use tokio::sync::oneshot; + +use crate::connection::ConnectionManager; +use crate::liquidity::service::lsps5::LSPS5ServiceLiquiditySource; +use crate::liquidity::{ + 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(self.map_lsps5_error(&e, 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); + self.map_lsps5_error(&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); + self.map_lsps5_error(&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(self.map_lsps5_error(&e, 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); + self.map_lsps5_error(&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 map_lsps5_error(&self, e: &LSPS5Error, fallback: Error) -> Error { + match e { + LSPS5Error::Protocol(LSPS5ProtocolError::TooManyWebhooks) => { + Error::LiquidityWebhookLimitExceeded + }, + LSPS5Error::Protocol(LSPS5ProtocolError::NoPriorActivityError) => { + Error::LiquidityWebhookNoPriorActivity + }, + LSPS5Error::Protocol(LSPS5ProtocolError::AppNameNotFound) => { + Error::LiquidityWebhookAppNameNotFound + }, + LSPS5Error::Protocol( + LSPS5ProtocolError::AppNameTooLong + | LSPS5ProtocolError::WebhookUrlTooLong + | LSPS5ProtocolError::UrlParse + | LSPS5ProtocolError::UnsupportedProtocol, + ) => Error::LiquidityWebhookInvalid, + _ => fallback, + } + } + + 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) + } +} + +/// 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 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 { + /// Registers a webhook URL with a single LSP for receiving LSPS5 notifications. + /// + /// The webhook will receive signed push notifications for Lightning events such as incoming + /// payments while the client is offline. + /// + /// Webhooks are stored per LSP, so this has to be called once per LSP that should be able to + /// reach the client. Give each LSP a distinct URL: bLIP-55 expects the notification delivery + /// service to verify the `x-lsps5-signature` header against the node ID of the LSP that URL + /// belongs to. + /// + /// Note that LSPs typically reject registrations from clients they have no prior relationship + /// with. LDK Node's own LSPS5 service requires an open channel, or an in-flight LSPS1 or LSPS2 + /// flow, and otherwise fails with [`Error::LiquidityWebhookNoPriorActivity`]. + pub fn set_webhook( + &self, app_name: String, webhook_url: String, node_id: PublicKey, + ) -> Result { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(Some(&node_id)).await })?; + + self.connect(&lsps5_node)?; + + let liquidity_source = Arc::clone(&self.liquidity_source); + self.runtime.block_on(async move { + liquidity_source + .lsps5_set_webhook(app_name, webhook_url, Some(&lsps5_node.node_id)) + .await + }) + } + + /// Lists all webhooks currently registered with the given LSP. + /// + /// Webhooks are stored per LSP, so this returns only the `app_name`s registered with `node_id`. + pub fn list_webhooks(&self, node_id: PublicKey) -> Result { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(Some(&node_id)).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 + }) + } + + /// Removes a previously-configured webhook from a single LSP. + /// + /// Returns an error if `node_id` is not a configured LSP supporting LSPS5, or if the LSP rejected + /// the removal. + pub fn remove_webhook(&self, app_name: String, node_id: PublicKey) -> Result<(), Error> { + let lsps5_node = self + .runtime + .block_on(async { self.liquidity_source.get_lsps5_node(Some(&node_id)).await })?; + + self.connect(&lsps5_node)?; + + let liquidity_source = Arc::clone(&self.liquidity_source); + self.runtime.block_on(async move { + liquidity_source.lsps5_remove_webhook(app_name, Some(&lsps5_node.node_id)).await + }) + } + + /// 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 ffc1f878bc..a1e0fed5d3 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}; @@ -215,6 +220,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)] @@ -240,12 +262,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, } @@ -255,19 +279,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, } } @@ -285,18 +313,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); @@ -305,7 +347,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( @@ -367,6 +409,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, @@ -384,6 +441,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, @@ -411,11 +470,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..9750e90463 --- /dev/null +++ b/src/liquidity/service/lsps5.rs @@ -0,0 +1,219 @@ +// 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) { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return; + }; + + if !self.is_client_offline(&client_id) { + return; + } + + 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 + ), + }) + } + + pub(crate) fn notify_expiry_soon(&self, client_id: PublicKey, timeout: u32) { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return; + }; + + if !self.is_client_offline(&client_id) { + return; + } + + 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 + ), + }) + } + + pub(crate) fn notify_liquidity_management_request( + &self, client_id: PublicKey, + ) -> Result<(), Error> { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return Err(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) { + let Some(handler) = self.liquidity_manager.lsps5_service_handler() else { + return; + }; + + if !self.is_client_offline(&client_id) { + return; + } + + 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 + ), + }) + } + + 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 response.status_code != 200 { + 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 => { + log_debug!( + self.logger, + "No peer manager available, assuming client {} is offline.", + client_id + ); + true + }, + } + } +} 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 65156982eb..78e9b2d87b 100644 --- a/src/types.rs +++ b/src/types.rs @@ -377,6 +377,8 @@ pub(crate) type BumpTransactionEventHandler = pub(crate) type PaymentStore = DataStore, KeepLeastRecentlyUsed>; +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 fd247f74cf..e636c70a3f 100644 --- a/tests/integration_tests_rust.rs +++ b/tests/integration_tests_rust.rs @@ -25,12 +25,13 @@ use common::{ bump_fee_and_broadcast, distribute_funds_unconfirmed, do_channel_full_cycle, expect_channel_pending_event, expect_channel_ready_event, expect_channel_ready_events, expect_event, expect_payment_claimable_event, expect_payment_received_event, - expect_payment_successful_event, expect_splice_negotiated_event, generate_blocks_and_wait, - generate_listening_addresses, invalidate_blocks, open_channel, open_channel_no_wait, - open_channel_push_amt, open_channel_with_all, premine_and_distribute_funds, premine_blocks, - prepare_rbf, random_chain_source, random_config, setup_bitcoind_and_electrsd, setup_builder, - setup_node, setup_two_nodes, splice_in_with_all, wait_for_block, wait_for_tx, InMemoryStore, - NodePaymentExt, TestChainSource, TestConfig, TestStoreType, TestSyncStore, + expect_payment_successful_event, expect_splice_negotiated_event, exponential_backoff_poll, + generate_blocks_and_wait, generate_listening_addresses, invalidate_blocks, open_channel, + open_channel_no_wait, open_channel_push_amt, open_channel_with_all, + premine_and_distribute_funds, premine_blocks, prepare_rbf, random_chain_source, random_config, + setup_bitcoind_and_electrsd, setup_builder, setup_node, setup_two_nodes, splice_in_with_all, + wait_for_block, wait_for_tx, InMemoryStore, NodePaymentExt, TestChainSource, TestConfig, + TestStoreType, TestSyncStore, }; use electrsd::corepc_node::{self, Node as BitcoinD}; use electrsd::ElectrsD; @@ -3450,7 +3451,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(); @@ -3778,7 +3779,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(); @@ -3955,7 +3956,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(); @@ -4861,7 +4862,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(); @@ -4884,7 +4885,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(); @@ -4925,3 +4926,302 @@ 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 response = lsps5_client + .set_webhook(app_name_1.clone(), webhook_url_1.clone(), service_node_id) + .expect("Failed to register webhook"); + 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 response = lsps5_client + .set_webhook(app_name_2.clone(), webhook_url_2.clone(), service_node_id) + .expect("Failed to register second webhook"); + 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 response = lsps5_client + .set_webhook(app_name_2.clone(), webhook_url_2.clone(), service_node_id) + .expect("Failed to re-register webhook"); + 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 response = + lsps5_client.set_webhook(app_name_3.clone(), webhook_url_3.clone(), service_node_id); + assert_eq!( + response.err(), + Some(NodeError::LiquidityWebhookLimitExceeded), + "Expected error when exceeding max webhooks" + ); + + // List registered webhooks + let registered_webhooks = + lsps5_client.list_webhooks(service_node_id).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 response = lsps5_client.remove_webhook(non_existing_app_name.clone(), service_node_id); + assert_eq!( + response.err(), + Some(NodeError::LiquidityWebhookAppNameNotFound), + "Expected error when removing non-existing webhook" + ); + + // Delete a registered webhook + lsps5_client + .remove_webhook(app_name_1.clone(), service_node_id) + .expect("Failed to delete first webhook"); + + // Verify webhook was deleted + let registered_webhooks = lsps5_client + .list_webhooks(service_node_id) + .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(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(), 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(); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn lsps5_payment_incoming_notification() { + 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 mut sync_config = EsploraSyncConfig::default(); + sync_config.background_sync_config = None; + + // The notification leaves no trace in any store, so observe it through the service's logs. + let service_logger = Arc::new(CollectingLogWriter::new()); + 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.set_custom_logger(service_logger.clone()); + + // LSPS2 is enabled so the LSP accepts forwards to the client's unannounced channel. + let lsps2_service_config = LSPS2ServiceConfig { + require_token: None, + advertise_service: false, + channel_opening_fee_ppm: 0, + channel_over_provisioning_ppm: 100_000, + max_payment_size_msat: 1_000_000_000, + min_payment_size_msat: 0, + min_channel_lifetime: 100, + min_channel_opening_fee_msat: 0, + max_client_to_self_delay: 1024, + client_trusts_lsp: false, + disable_client_reserve: false, + }; + + service_builder.enable_liquidity_provider( + Some(lsps2_service_config), + Some(LSPS5ServiceConfig { max_webhooks_per_client: 2 }), + ); + + 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_listening_addr = + service_node.listening_addresses().unwrap().first().unwrap().clone(); + + 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_listening_addr, 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 payer_config = random_config(); + setup_builder!(payer_builder, payer_config.node_config); + payer_builder.set_chain_source_esplora(esplora_url.clone(), Some(sync_config)); + let payer_node = payer_builder.build(payer_config.node_entropy.into()).unwrap(); + payer_node.start().unwrap(); + + let service_addr = service_node.onchain_payment().new_address().unwrap(); + let client_addr = client_node.onchain_payment().new_address().unwrap(); + let payer_addr = payer_node.onchain_payment().new_address().unwrap(); + premine_and_distribute_funds( + &bitcoind.client, + &electrsd.client, + vec![service_addr, client_addr, payer_addr], + Amount::from_sat(10_000_000), + ) + .await; + service_node.sync_wallets().unwrap(); + client_node.sync_wallets().unwrap(); + payer_node.sync_wallets().unwrap(); + + open_channel(&payer_node, &service_node, 5_000_000, false, &electrsd).await; + // Opened service -> client so the LSP has outbound liquidity towards the client, leaving the + // client being offline as the only reason the forward below can fail. + open_channel(&service_node, &client_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(); + payer_node.sync_wallets().unwrap(); + expect_channel_ready_event!(payer_node, service_node_id); + expect_channel_ready_event!(client_node, service_node_id); + expect_channel_ready_events!(service_node, payer_node.node_id(), client_node_id); + + // The LSP only accepts registrations from clients it has prior activity with, which the channel + // opened above satisfies. + client_node + .liquidity() + .lsps5() + .set_webhook( + "test-app".to_string(), + "https://127.0.0.1:1/webhook".to_string(), + service_node_id, + ) + .expect("Failed to register webhook"); + + let invoice_description = + Bolt11InvoiceDescription::Direct(Description::new(String::from("lsps5")).unwrap()).into(); + let invoice = client_node + .bolt11_payment() + .receive(100_000_000, &invoice_description, 1024) + .expect("Failed to create invoice"); + + // Stopped rather than just disconnected so that the client cannot reconnect on its own. + client_node.stop().unwrap(); + + // Wait for the service to observe the disconnect, so the payment does not race against a + // channel that still looks live. + exponential_backoff_poll(|| { + let connected = + service_node.list_peers().iter().any(|p| p.node_id == client_node_id && p.is_connected); + (!connected).then_some(()) + }) + .await; + + // Not waiting on the payment to resolve: the forward failure we trigger on happens on the first + // attempt, while the payer keeps retrying in the background. + let _ = payer_node.bolt11_payment().send(&invoice, None).unwrap(); + + // Matching on the method keeps this distinct from the `lsps5.webhook_registered` notification + // the registration above already sent. + let notification_sent = service_logger.wait_for("LSPS5PaymentIncoming").await; + assert!( + notification_sent, + "Expected the failed forward to an offline client to trigger a payment_incoming notification" + ); + + service_node.stop().unwrap(); + payer_node.stop().unwrap(); +}