feat: direct webhook delivery from LiquiditySource (#9)

Fire HTTP POST directly in handle_next_event() instead of routing
through the event queue (S3 persistence), cutting webhook latency
from 15-23s to <2s.

LiquiditySource gets optional reqwest::Client + webhook_url fields.
When configured, SendWebhook events bypass the event queue entirely.
Falls back to event queue if not configured.

Also bumps rust-lightning to include HTLC expiry threshold increase
(10s -> 45s, moneydevkit/rust-lightning#5, 1432d063c).

Author: martinsaposnic

Cargo.toml

@@ -42,17 +42,17 @@ default = []
 # lightning-macros = { version = "0.2.0" }
 
 # Branch: https://github.com/moneydevkit/rust-lightning/commits/lsp-0.2.0/
-lightning = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["std"] }
-lightning-types = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c" }
-lightning-invoice = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["std"] }
-lightning-net-tokio = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c" }
-lightning-persister = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["tokio"] }
-lightning-background-processor = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c" }
-lightning-rapid-gossip-sync = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c" }
-lightning-block-sync = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["rest-client", "rpc-client", "tokio"] }
-lightning-transaction-sync = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["esplora-async-https", "time", "electrum-rustls-ring"] }
-lightning-liquidity = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["std"] }
-lightning-macros = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c" }
+lightning = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["std"] }
+lightning-types = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85" }
+lightning-invoice = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["std"] }
+lightning-net-tokio = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85" }
+lightning-persister = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["tokio"] }
+lightning-background-processor = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85" }
+lightning-rapid-gossip-sync = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85" }
+lightning-block-sync = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["rest-client", "rpc-client", "tokio"] }
+lightning-transaction-sync = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["esplora-async-https", "time", "electrum-rustls-ring"] }
+lightning-liquidity = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["std"] }
+lightning-macros = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85" }
 
 #lightning = { path = "../rust-lightning/lightning", features = ["std"] }
 #lightning-types = { path = "../rust-lightning/lightning-types" }
@@ -101,7 +101,7 @@ winapi = { version = "0.3", features = ["winbase"] }
 [dev-dependencies]
 # lightning = { version = "0.2.0", features = ["std", "_test_utils"] }
 # Branch: https://github.com/moneydevkit/rust-lightning/commits/lsp-0.2.0/
-lightning = { git = "https://github.com/moneydevkit/rust-lightning", rev = "145438f9fa8e08cddcb2f7fe15c8613cfe464f8c", features = ["std", "_test_utils"] }
+lightning = { git = "https://github.com/moneydevkit/rust-lightning", rev = "1432d063c277f303266a6a5345789617f2e05e85", features = ["std", "_test_utils"] }
 #lightning = { path = "../rust-lightning/lightning", features = ["std", "_test_utils"] }
 proptest = "1.0.0"
 regex = "1.5.6"

src/builder.rs

@@ -136,6 +136,8 @@ struct LiquiditySourceConfig {
 	lsps4_client: Option<LSPS4ClientConfig>,
 	// Act as an LSPS4 service.
 	lsps4_service: Option<LSPS4ServiceConfig>,
+	// URL for direct webhook delivery from LiquiditySource.
+	webhook_url: Option<String>,
 }
 
 #[derive(Clone)]
@@ -525,6 +527,15 @@ impl NodeBuilder {
 		self
 	}
 
+	/// Sets a webhook URL for direct HTTP delivery of payment notifications from the
+	/// liquidity source, bypassing the event queue.
+	pub fn set_webhook_url(&mut self, url: String) -> &mut Self {
+		let liquidity_source_config =
+			self.liquidity_source_config.get_or_insert(LiquiditySourceConfig::default());
+		liquidity_source_config.webhook_url = Some(url);
+		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;
@@ -1782,6 +1793,10 @@ fn build_with_store_internal(
 				.as_ref()
 				.map(|config| liquidity_source_builder.lsps4_service(config.clone()));
 
+			if let Some(ref url) = lsc.webhook_url {
+				liquidity_source_builder.webhook(url.clone());
+			}
+
 			let liquidity_source = runtime
 				.block_on(async move { liquidity_source_builder.build().await.map(Arc::new) })?;
 			let custom_message_handler =

src/liquidity.rs

@@ -208,6 +208,8 @@ where
 	config: Arc<Config>,
 	logger: L,
 	event_queue: Arc<EventQueue<L>>,
+	webhook_client: Option<reqwest::Client>,
+	webhook_url: Option<String>,
 }
 
 impl<L: Deref + Clone> LiquiditySourceBuilder<L>
@@ -239,9 +241,18 @@ where
 			config,
 			logger,
 			event_queue,
+			webhook_client: None,
+			webhook_url: None,
 		}
 	}
 
+	/// Configures a webhook URL for direct HTTP delivery of payment notifications.
+	pub(crate) fn webhook(&mut self, url: String) -> &mut Self {
+		self.webhook_client = Some(reqwest::Client::new());
+		self.webhook_url = Some(url);
+		self
+	}
+
 	pub(crate) fn lsps1_client(
 		&mut self, lsp_node_id: PublicKey, lsp_address: SocketAddress, token: Option<String>,
 	) -> &mut Self {
@@ -366,6 +377,8 @@ where
 			config: self.config,
 			logger: self.logger,
 			event_queue: self.event_queue,
+			webhook_client: self.webhook_client,
+			webhook_url: self.webhook_url,
 		})
 	}
 }
@@ -387,6 +400,8 @@ where
 	config: Arc<Config>,
 	logger: L,
 	event_queue: Arc<EventQueue<L>>,
+	webhook_client: Option<reqwest::Client>,
+	webhook_url: Option<String>,
 }
 
 impl<L: Deref + Clone> LiquiditySource<L>
@@ -1203,11 +1218,35 @@ where
 				}
 			},
 			LiquidityEvent::LSPS4Service(LSPS4ServiceEvent::SendWebhook { counterparty_node_id, payment_hash }) => {
-				if let Err(e) = self
-					.event_queue
-					.add_event(crate::event::Event::SendWebhook { node_id: counterparty_node_id, payment_hash }).await
-				{
-					log_error!(self.logger, "Failed to queue webhook event: {:?}", e);
+				if let (Some(client), Some(url)) = (&self.webhook_client, &self.webhook_url) {
+					let mut json_body = HashMap::new();
+					json_body.insert("nodeId", counterparty_node_id.to_string());
+					json_body.insert("paymentHash", payment_hash.to_string());
+					log_info!(
+						self.logger,
+						"Sending webhook directly for payment_hash={} node={}",
+						payment_hash,
+						counterparty_node_id
+					);
+					if let Err(e) = client.post(url).json(&json_body).send().await {
+						log_error!(self.logger, "Direct webhook POST failed: {:?}", e);
+					}
+				} else {
+					// Fallback: route through event queue (slower path via S3 persistence).
+					log_info!(
+						self.logger,
+						"No direct webhook configured, falling back to event queue"
+					);
+					if let Err(e) = self
+						.event_queue
+						.add_event(crate::event::Event::SendWebhook {
+							node_id: counterparty_node_id,
+							payment_hash,
+						})
+						.await
+					{
+						log_error!(self.logger, "Failed to queue webhook event: {:?}", e);
+					}
 				}
 			},
 			e => {