Implement LiquidityManager event handling

Author: tnull

bindings/ldk_node.udl

@@ -131,6 +131,7 @@ enum NodeError {
 	"InvalidNetwork",
 	"DuplicatePayment",
 	"InsufficientFunds",
+	"LiquiditySourceUnavailable",
 };
 
 [Error]

src/error.rs

@@ -63,6 +63,8 @@ pub enum Error {
 	DuplicatePayment,
 	/// The available funds are insufficient to complete the given operation.
 	InsufficientFunds,
+	/// The given operation failed due to the required liquidity source being unavailable.
+	LiquiditySourceUnavailable,
 }
 
 impl fmt::Display for Error {
@@ -106,6 +108,9 @@ impl fmt::Display for Error {
 			Self::InsufficientFunds => {
 				write!(f, "The available funds are insufficient to complete the given operation.")
 			}
+			Self::LiquiditySourceUnavailable => {
+				write!(f, "The given operation failed due to the required liquidity source being unavailable.")
+			}
 		}
 	}
 }

src/liquidity.rs

@@ -1,21 +1,29 @@
+use crate::logger::log_error;
 use crate::types::{ChannelManager, KeysManager, LiquidityManager, PeerManager};
 use crate::Config;
 
 use lightning::ln::msgs::SocketAddress;
 use lightning::util::logger::Logger;
 use lightning::util::persist::KVStore;
 use lightning_liquidity::events::Event;
+use lightning_liquidity::lsps0::msgs::RequestId;
 use lightning_liquidity::lsps2::event::LSPS2ClientEvent;
+use lightning_liquidity::lsps2::msgs::OpeningFeeParams;
 
 use bitcoin::secp256k1::PublicKey;
 
+use tokio::sync::oneshot;
+
+use std::collections::HashMap;
 use std::ops::Deref;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 struct LSPS2Service {
 	address: SocketAddress,
 	node_id: PublicKey,
 	token: Option<String>,
+	pending_fee_requests: Mutex<HashMap<RequestId, oneshot::Sender<LSPS2FeeResponse>>>,
+	pending_buy_requests: Mutex<HashMap<RequestId, oneshot::Sender<LSPS2BuyResponse>>>,
 }
 
 pub(crate) struct LiquiditySource<K: KVStore + Sync + Send + 'static, L: Deref>
@@ -39,7 +47,15 @@ where
 		channel_manager: Arc<ChannelManager<K>>, keys_manager: Arc<KeysManager>,
 		liquidity_manager: Arc<LiquidityManager<K>>, config: Arc<Config>, logger: L,
 	) -> Self {
-		let lsps2_service = Some(LSPS2Service { address, node_id, token });
+		let pending_fee_requests = Mutex::new(HashMap::new());
+		let pending_buy_requests = Mutex::new(HashMap::new());
+		let lsps2_service = Some(LSPS2Service {
+			address,
+			node_id,
+			token,
+			pending_fee_requests,
+			pending_buy_requests,
+		});
 		Self { lsps2_service, channel_manager, keys_manager, liquidity_manager, config, logger }
 	}
 
@@ -55,19 +71,123 @@ where
 	pub(crate) async fn handle_next_event(&self) {
 		match self.liquidity_manager.next_event_async().await {
 			Event::LSPS2Client(LSPS2ClientEvent::OpeningParametersReady {
-				counterparty_node_id: _,
-				opening_fee_params_menu: _,
-				min_payment_size_msat: _,
-				max_payment_size_msat: _,
-			}) => {}
+				request_id,
+				counterparty_node_id,
+				opening_fee_params_menu,
+			}) => {
+				if let Some(lsps2_service) = self.lsps2_service.as_ref() {
+					if counterparty_node_id != lsps2_service.node_id {
+						debug_assert!(
+							false,
+							"Received response from unexpected LSP counterparty. This should never happen."
+							);
+						log_error!(
+							self.logger,
+							"Received response from unexpected LSP counterparty. This should never happen."
+							);
+						return;
+					}
+
+					if let Some(sender) =
+						lsps2_service.pending_fee_requests.lock().unwrap().remove(&request_id)
+					{
+						let response = LSPS2FeeResponse {
+							opening_fee_params_menu,
+						};
+
+						match sender.send(response) {
+							Ok(()) => (),
+							Err(e) => {
+								log_error!(
+									self.logger,
+									"Failed to handle response from liquidity service: {:?}",
+									e
+								);
+							}
+						}
+					} else {
+						debug_assert!(
+							false,
+							"Received response from liquidity service for unknown request."
+						);
+						log_error!(
+							self.logger,
+							"Received response from liquidity service for unknown request."
+						);
+					}
+				} else {
+					log_error!(
+						self.logger,
+						"Received unexpected LSPS2Client::OpeningParametersReady event!"
+					);
+				}
+			}
 			Event::LSPS2Client(LSPS2ClientEvent::InvoiceParametersReady {
-				counterparty_node_id: _,
-				intercept_scid: _,
-				cltv_expiry_delta: _,
-				payment_size_msat: _,
-				user_channel_id: _,
-			}) => {}
-			_ => {}
+				request_id,
+				counterparty_node_id,
+				intercept_scid,
+				cltv_expiry_delta,
+				..
+			}) => {
+				if let Some(lsps2_service) = self.lsps2_service.as_ref() {
+					if counterparty_node_id != lsps2_service.node_id {
+						debug_assert!(
+							false,
+							"Received response from unexpected LSP counterparty. This should never happen."
+						);
+						log_error!(
+							self.logger,
+							"Received response from unexpected LSP counterparty. This should never happen."
+						);
+						return;
+					}
+
+					if let Some(sender) =
+						lsps2_service.pending_buy_requests.lock().unwrap().remove(&request_id)
+					{
+						let response = LSPS2BuyResponse { intercept_scid, cltv_expiry_delta };
+
+						match sender.send(response) {
+							Ok(()) => (),
+							Err(e) => {
+								log_error!(
+									self.logger,
+									"Failed to handle response from liquidity service: {:?}",
+									e
+								);
+							}
+						}
+					} else {
+						debug_assert!(
+							false,
+							"Received response from liquidity service for unknown request."
+						);
+						log_error!(
+							self.logger,
+							"Received response from liquidity service for unknown request."
+						);
+					}
+				} else {
+					log_error!(
+						self.logger,
+						"Received unexpected LSPS2Client::InvoiceParametersReady event!"
+					);
+				}
+			}
+			e => {
+				log_error!(self.logger, "Received unexpected liquidity event: {:?}", e);
+			}
 		}
 	}
 }
+
+#[derive(Debug, Clone)]
+pub(crate) struct LSPS2FeeResponse {
+	opening_fee_params_menu: Vec<OpeningFeeParams>,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) struct LSPS2BuyResponse {
+	intercept_scid: u64,
+	cltv_expiry_delta: u32,
+}