Integrate LSPS5 with liquidity manager

Fully integrates the LSPS5 webhook components into the
lightning-liquidity framework, enabling usage through the LiquidityManager.

Author: martinsaposnic

fuzz/src/lsps_message.rs

@@ -78,6 +78,7 @@ pub fn do_test(data: &[u8]) {
 	));
 
 	let liquidity_manager = Arc::new(LiquidityManager::new(
+		Arc::clone(&keys_manager),
 		Arc::clone(&keys_manager),
 		Arc::clone(&manager),
 		None::<Arc<dyn Filter + Send + Sync>>,

lightning-background-processor/Cargo.toml

@@ -14,9 +14,9 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [features]
-std = ["lightning/std", "lightning-liquidity/std", "bitcoin-io/std", "bitcoin_hashes/std"]
-
 default = ["std"]
+std = ["lightning/std", "lightning-liquidity/std", "bitcoin-io/std", "bitcoin_hashes/std", "lightning-liquidity/time"]
+
 
 [dependencies]
 bitcoin = { version = "0.32.2", default-features = false }

lightning-background-processor/src/lib.rs

@@ -735,6 +735,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// # use lightning_background_processor::{process_events_async, GossipSync};
 /// # use core::future::Future;
 /// # use core::pin::Pin;
+/// # use lightning_liquidity::lsps5::service::TimeProvider;
 /// # struct Logger {}
 /// # impl lightning::util::logger::Logger for Logger {
 /// #     fn log(&self, _record: lightning::util::logger::Record) {}
@@ -753,6 +754,15 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// #     fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { todo!() }
 /// #     fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> Pin<Box<dyn Future<Output = Result<Vec<String>, io::Error>> + 'static + Send>> { todo!() }
 /// # }
+/// # use core::time::Duration;
+/// # struct DefaultTimeProvider;
+/// #
+/// # impl TimeProvider for DefaultTimeProvider {
+/// #    fn duration_since_epoch(&self) -> Duration {
+/// #        use std::time::{SystemTime, UNIX_EPOCH};
+/// #        SystemTime::now().duration_since(UNIX_EPOCH).expect("system time before Unix epoch")
+/// #    }
+/// # }
 /// # struct EventHandler {}
 /// # impl EventHandler {
 /// #     async fn handle_event(&self, _: lightning::events::Event) -> Result<(), ReplayEvent> { Ok(()) }
@@ -768,7 +778,7 @@ use futures_util::{dummy_waker, OptionalSelector, Selector, SelectorOutput};
 /// # type P2PGossipSync<UL> = lightning::routing::gossip::P2PGossipSync<Arc<NetworkGraph>, Arc<UL>, Arc<Logger>>;
 /// # type ChannelManager<B, F, FE> = lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor<B, F, FE>, B, FE, Logger>;
 /// # type OnionMessenger<B, F, FE> = lightning::onion_message::messenger::OnionMessenger<Arc<lightning::sign::KeysManager>, Arc<lightning::sign::KeysManager>, Arc<Logger>, Arc<ChannelManager<B, F, FE>>, Arc<lightning::onion_message::messenger::DefaultMessageRouter<Arc<NetworkGraph>, Arc<Logger>, Arc<lightning::sign::KeysManager>>>, Arc<ChannelManager<B, F, FE>>, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler, lightning::ln::peer_handler::IgnoringMessageHandler>;
-/// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>>;
+/// # type LiquidityManager<B, F, FE> = lightning_liquidity::LiquidityManager<Arc<lightning::sign::KeysManager>, Arc<lightning::sign::KeysManager>, Arc<ChannelManager<B, F, FE>>, Arc<F>, Arc<DefaultTimeProvider>>;
 /// # type Scorer = RwLock<lightning::routing::scoring::ProbabilisticScorer<Arc<NetworkGraph>, Arc<Logger>>>;
 /// # type PeerManager<B, F, FE, UL> = lightning::ln::peer_handler::SimpleArcPeerManager<SocketDescriptor, ChainMonitor<B, F, FE>, B, FE, Arc<UL>, Logger, F, StoreSync>;
 /// # type OutputSweeper<B, D, FE, F, O> = lightning::util::sweep::OutputSweeper<Arc<B>, Arc<D>, Arc<FE>, Arc<F>, Arc<Store>, Arc<Logger>, Arc<O>>;
@@ -1391,6 +1401,7 @@ mod tests {
 	use lightning::util::sweep::{OutputSpendStatus, OutputSweeperSync, PRUNE_DELAY_BLOCKS};
 	use lightning::util::test_utils;
 	use lightning::{get_event, get_event_msg};
+	use lightning_liquidity::lsps5::service::DefaultTimeProvider;
 	use lightning_liquidity::LiquidityManager;
 	use lightning_persister::fs_store::FilesystemStore;
 	use lightning_rapid_gossip_sync::RapidGossipSync;
@@ -1488,8 +1499,13 @@ mod tests {
 		IgnoringMessageHandler,
 	>;
 
-	type LM =
-		LiquidityManager<Arc<KeysManager>, Arc<ChannelManager>, Arc<dyn Filter + Sync + Send>>;
+	type LM = LiquidityManager<
+		Arc<KeysManager>,
+		Arc<KeysManager>,
+		Arc<ChannelManager>,
+		Arc<dyn Filter + Sync + Send>,
+		Arc<DefaultTimeProvider>,
+	>;
 
 	struct Node {
 		node: Arc<ChannelManager>,
@@ -1936,6 +1952,7 @@ mod tests {
 				Arc::clone(&keys_manager),
 			));
 			let liquidity_manager = Arc::new(LiquidityManager::new(
+				Arc::clone(&keys_manager),
 				Arc::clone(&keys_manager),
 				Arc::clone(&manager),
 				None,

lightning-liquidity/src/events/mod.rs

@@ -23,6 +23,7 @@ pub use event_queue::MAX_EVENT_QUEUE_SIZE;
 use crate::lsps0;
 use crate::lsps1;
 use crate::lsps2;
+use crate::lsps5;
 
 /// An event which you should probably take some action in response to.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -38,6 +39,10 @@ pub enum LiquidityEvent {
 	LSPS2Client(lsps2::event::LSPS2ClientEvent),
 	/// An LSPS2 (JIT Channel) server event.
 	LSPS2Service(lsps2::event::LSPS2ServiceEvent),
+	/// An LSPS5 (Webhook) client event.
+	LSPS5Client(lsps5::event::LSPS5ClientEvent),
+	/// An LSPS5 (Webhook) server event.
+	LSPS5Service(lsps5::event::LSPS5ServiceEvent),
 }
 
 impl From<lsps0::event::LSPS0ClientEvent> for LiquidityEvent {
@@ -70,3 +75,15 @@ impl From<lsps2::event::LSPS2ServiceEvent> for LiquidityEvent {
 		Self::LSPS2Service(event)
 	}
 }
+
+impl From<lsps5::event::LSPS5ClientEvent> for LiquidityEvent {
+	fn from(event: lsps5::event::LSPS5ClientEvent) -> Self {
+		Self::LSPS5Client(event)
+	}
+}
+
+impl From<lsps5::event::LSPS5ServiceEvent> for LiquidityEvent {
+	fn from(event: lsps5::event::LSPS5ServiceEvent) -> Self {
+		Self::LSPS5Service(event)
+	}
+}

lightning-liquidity/src/lib.rs

@@ -23,6 +23,8 @@
 //! an LSP will open a "just-in-time" channel. This is useful for the initial on-boarding of
 //! clients as the channel opening fees are deducted from the incoming payment, i.e., no funds are
 //! required client-side to initiate this flow.
+//! - [bLIP-55 / LSPS5] defines a protocol for sending webhook notifications to clients. This is
+//! useful for notifying clients about incoming payments, channel expiries, etc.
 //!
 //! To get started, you'll want to setup a [`LiquidityManager`] and configure it to be the
 //! [`CustomMessageHandler`] of your LDK node. You can then for example call
@@ -37,6 +39,7 @@
 //! [bLIP-50 / LSPS0]: https://github.com/lightning/blips/blob/master/blip-0050.md
 //! [bLIP-51 / LSPS1]: https://github.com/lightning/blips/blob/master/blip-0051.md
 //! [bLIP-52 / LSPS2]: https://github.com/lightning/blips/blob/master/blip-0052.md
+//! [bLIP-55 / LSPS5]: https://github.com/lightning/blips/pull/55/files
 //! [`CustomMessageHandler`]: lightning::ln::peer_handler::CustomMessageHandler
 //! [`LiquidityManager::next_event`]: crate::LiquidityManager::next_event
 #![deny(missing_docs)]
@@ -59,6 +62,7 @@ pub mod events;
 pub mod lsps0;
 pub mod lsps1;
 pub mod lsps2;
+pub mod lsps5;
 mod manager;
 pub mod message_queue;
 #[allow(dead_code)]

lightning-liquidity/src/lsps0/msgs.rs

@@ -83,6 +83,7 @@ impl TryFrom<LSPSMessage> for LSPS0Message {
 			LSPSMessage::LSPS0(message) => Ok(message),
 			LSPSMessage::LSPS1(_) => Err(()),
 			LSPSMessage::LSPS2(_) => Err(()),
+			LSPSMessage::LSPS5(_) => Err(()),
 		}
 	}
 }

lightning-liquidity/src/lsps0/ser.rs

@@ -21,6 +21,11 @@ use crate::lsps1::msgs::{
 use crate::lsps2::msgs::{
 	LSPS2Message, LSPS2Request, LSPS2Response, LSPS2_BUY_METHOD_NAME, LSPS2_GET_INFO_METHOD_NAME,
 };
+use crate::lsps5::msgs::{
+	LSPS5Message, LSPS5Request, LSPS5Response, LSPS5_LIST_WEBHOOKS_METHOD_NAME,
+	LSPS5_REMOVE_WEBHOOK_METHOD_NAME, LSPS5_SET_WEBHOOK_METHOD_NAME,
+};
+
 use crate::prelude::HashMap;
 
 use lightning::ln::msgs::{DecodeError, LightningError};
@@ -61,6 +66,9 @@ pub(crate) enum LSPSMethod {
 	LSPS1CreateOrder,
 	LSPS2GetInfo,
 	LSPS2Buy,
+	LSPS5SetWebhook,
+	LSPS5ListWebhooks,
+	LSPS5RemoveWebhook,
 }
 
 impl LSPSMethod {
@@ -72,6 +80,9 @@ impl LSPSMethod {
 			Self::LSPS1GetOrder => LSPS1_GET_ORDER_METHOD_NAME,
 			Self::LSPS2GetInfo => LSPS2_GET_INFO_METHOD_NAME,
 			Self::LSPS2Buy => LSPS2_BUY_METHOD_NAME,
+			Self::LSPS5SetWebhook => LSPS5_SET_WEBHOOK_METHOD_NAME,
+			Self::LSPS5ListWebhooks => LSPS5_LIST_WEBHOOKS_METHOD_NAME,
+			Self::LSPS5RemoveWebhook => LSPS5_REMOVE_WEBHOOK_METHOD_NAME,
 		}
 	}
 }
@@ -86,6 +97,9 @@ impl FromStr for LSPSMethod {
 			LSPS1_GET_ORDER_METHOD_NAME => Ok(Self::LSPS1GetOrder),
 			LSPS2_GET_INFO_METHOD_NAME => Ok(Self::LSPS2GetInfo),
 			LSPS2_BUY_METHOD_NAME => Ok(Self::LSPS2Buy),
+			LSPS5_SET_WEBHOOK_METHOD_NAME => Ok(Self::LSPS5SetWebhook),
+			LSPS5_LIST_WEBHOOKS_METHOD_NAME => Ok(Self::LSPS5ListWebhooks),
+			LSPS5_REMOVE_WEBHOOK_METHOD_NAME => Ok(Self::LSPS5RemoveWebhook),
 			_ => Err(&"Unknown method name"),
 		}
 	}
@@ -118,6 +132,16 @@ impl From<&LSPS2Request> for LSPSMethod {
 	}
 }
 
+impl From<&LSPS5Request> for LSPSMethod {
+	fn from(value: &LSPS5Request) -> Self {
+		match value {
+			LSPS5Request::SetWebhook(_) => Self::LSPS5SetWebhook,
+			LSPS5Request::ListWebhooks(_) => Self::LSPS5ListWebhooks,
+			LSPS5Request::RemoveWebhook(_) => Self::LSPS5RemoveWebhook,
+		}
+	}
+}
+
 impl<'de> Deserialize<'de> for LSPSMethod {
 	fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
 	where
@@ -266,6 +290,8 @@ pub enum LSPSMessage {
 	LSPS1(LSPS1Message),
 	/// An LSPS2 message.
 	LSPS2(LSPS2Message),
+	/// An LSPS5 message.
+	LSPS5(LSPS5Message),
 }
 
 impl LSPSMessage {
@@ -293,6 +319,9 @@ impl LSPSMessage {
 			LSPSMessage::LSPS2(LSPS2Message::Request(request_id, request)) => {
 				Some((LSPSRequestId(request_id.0.clone()), request.into()))
 			},
+			LSPSMessage::LSPS5(LSPS5Message::Request(request_id, request)) => {
+				Some((LSPSRequestId(request_id.0.clone()), request.into()))
+			},
 			_ => None,
 		}
 	}
@@ -409,6 +438,44 @@ impl Serialize for LSPSMessage {
 				jsonrpc_object.serialize_field(JSONRPC_ID_FIELD_KEY, &serde_json::Value::Null)?;
 				jsonrpc_object.serialize_field(JSONRPC_ERROR_FIELD_KEY, &error)?;
 			},
+			LSPSMessage::LSPS5(LSPS5Message::Request(request_id, request)) => {
+				jsonrpc_object.serialize_field(JSONRPC_ID_FIELD_KEY, &request_id.0)?;
+				jsonrpc_object
+					.serialize_field(JSONRPC_METHOD_FIELD_KEY, &LSPSMethod::from(request))?;
+
+				match request {
+					LSPS5Request::SetWebhook(params) => {
+						jsonrpc_object.serialize_field(JSONRPC_PARAMS_FIELD_KEY, params)?
+					},
+					LSPS5Request::ListWebhooks(params) => {
+						jsonrpc_object.serialize_field(JSONRPC_PARAMS_FIELD_KEY, params)?
+					},
+					LSPS5Request::RemoveWebhook(params) => {
+						jsonrpc_object.serialize_field(JSONRPC_PARAMS_FIELD_KEY, params)?
+					},
+				}
+			},
+			LSPSMessage::LSPS5(LSPS5Message::Response(request_id, response)) => {
+				jsonrpc_object.serialize_field(JSONRPC_ID_FIELD_KEY, &request_id.0)?;
+
+				match response {
+					LSPS5Response::SetWebhook(result) => {
+						jsonrpc_object.serialize_field(JSONRPC_RESULT_FIELD_KEY, result)?
+					},
+					LSPS5Response::SetWebhookError(error) => {
+						jsonrpc_object.serialize_field(JSONRPC_ERROR_FIELD_KEY, error)?
+					},
+					LSPS5Response::ListWebhooks(result) => {
+						jsonrpc_object.serialize_field(JSONRPC_RESULT_FIELD_KEY, result)?
+					},
+					LSPS5Response::RemoveWebhook(result) => {
+						jsonrpc_object.serialize_field(JSONRPC_RESULT_FIELD_KEY, result)?
+					},
+					LSPS5Response::RemoveWebhookError(error) => {
+						jsonrpc_object.serialize_field(JSONRPC_ERROR_FIELD_KEY, error)?
+					},
+				}
+			},
 		}
 
 		jsonrpc_object.end()
@@ -522,6 +589,30 @@ impl<'de, 'a> Visitor<'de> for LSPSMessageVisitor<'a> {
 						.map_err(de::Error::custom)?;
 					Ok(LSPSMessage::LSPS2(LSPS2Message::Request(id, LSPS2Request::Buy(request))))
 				},
+				LSPSMethod::LSPS5SetWebhook => {
+					let request = serde_json::from_value(params.unwrap_or(json!({})))
+						.map_err(de::Error::custom)?;
+					Ok(LSPSMessage::LSPS5(LSPS5Message::Request(
+						id,
+						LSPS5Request::SetWebhook(request),
+					)))
+				},
+				LSPSMethod::LSPS5ListWebhooks => {
+					let request = serde_json::from_value(params.unwrap_or(json!({})))
+						.map_err(de::Error::custom)?;
+					Ok(LSPSMessage::LSPS5(LSPS5Message::Request(
+						id,
+						LSPS5Request::ListWebhooks(request),
+					)))
+				},
+				LSPSMethod::LSPS5RemoveWebhook => {
+					let request = serde_json::from_value(params.unwrap_or(json!({})))
+						.map_err(de::Error::custom)?;
+					Ok(LSPSMessage::LSPS5(LSPS5Message::Request(
+						id,
+						LSPS5Request::RemoveWebhook(request),
+					)))
+				},
 			},
 			None => match self.request_id_to_method_map.remove(&id) {
 				Some(method) => match method {
@@ -627,6 +718,52 @@ impl<'de, 'a> Visitor<'de> for LSPSMessageVisitor<'a> {
 							Err(de::Error::custom("Received invalid JSON-RPC object: one of method, result, or error required"))
 						}
 					},
+					LSPSMethod::LSPS5SetWebhook => {
+						if let Some(error) = error {
+							Ok(LSPSMessage::LSPS5(LSPS5Message::Response(
+								id,
+								LSPS5Response::SetWebhookError(error.into()),
+							)))
+						} else if let Some(result) = result {
+							let response =
+								serde_json::from_value(result).map_err(de::Error::custom)?;
+							Ok(LSPSMessage::LSPS5(LSPS5Message::Response(
+								id,
+								LSPS5Response::SetWebhook(response),
+							)))
+						} else {
+							Err(de::Error::custom("Received invalid JSON-RPC object: one of method, result, or error required"))
+						}
+					},
+					LSPSMethod::LSPS5ListWebhooks => {
+						if let Some(result) = result {
+							let response =
+								serde_json::from_value(result).map_err(de::Error::custom)?;
+							Ok(LSPSMessage::LSPS5(LSPS5Message::Response(
+								id,
+								LSPS5Response::ListWebhooks(response),
+							)))
+						} else {
+							Err(de::Error::custom("Received invalid JSON-RPC object: one of method, result, or error required"))
+						}
+					},
+					LSPSMethod::LSPS5RemoveWebhook => {
+						if let Some(error) = error {
+							Ok(LSPSMessage::LSPS5(LSPS5Message::Response(
+								id,
+								LSPS5Response::RemoveWebhookError(error.into()),
+							)))
+						} else if let Some(result) = result {
+							let response =
+								serde_json::from_value(result).map_err(de::Error::custom)?;
+							Ok(LSPSMessage::LSPS5(LSPS5Message::Response(
+								id,
+								LSPS5Response::RemoveWebhook(response),
+							)))
+						} else {
+							Err(de::Error::custom("Received invalid JSON-RPC object: one of method, result, or error required"))
+						}
+					},
 				},
 				None => Err(de::Error::custom(format!(
 					"Received response for unknown request id: {}",