Add async payments

Author: joostjager

src/builder.rs

@@ -1458,7 +1458,7 @@ fn build_with_store_internal(
 		Arc::clone(&channel_manager),
 		message_router,
 		Arc::clone(&channel_manager),
-		IgnoringMessageHandler {},
+		Arc::clone(&channel_manager),
 		IgnoringMessageHandler {},
 		IgnoringMessageHandler {},
 	));

src/event.rs

@@ -5,6 +5,7 @@
 // http://opensource.org/licenses/MIT>, at your option. You may not use this file except in
 // accordance with one or both of these licenses.
 
+use crate::static_invoice_store::StaticInvoiceStore;
 use crate::types::{CustomTlvRecord, DynStore, PaymentStore, Sweeper, Wallet};
 
 use crate::{
@@ -458,6 +459,7 @@ where
 	runtime: Arc<Runtime>,
 	logger: L,
 	config: Arc<Config>,
+	static_invoice_store: StaticInvoiceStore,
 }
 
 impl<L: Deref + Clone + Sync + Send + 'static> EventHandler<L>
@@ -470,8 +472,9 @@ where
 		channel_manager: Arc<ChannelManager>, connection_manager: Arc<ConnectionManager<L>>,
 		output_sweeper: Arc<Sweeper>, network_graph: Arc<Graph>,
 		liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
-		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>, runtime: Arc<Runtime>,
-		logger: L, config: Arc<Config>,
+		payment_store: Arc<PaymentStore>, peer_store: Arc<PeerStore<L>>,
+		static_invoice_store: StaticInvoiceStore, runtime: Arc<Runtime>, logger: L,
+		config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -487,6 +490,7 @@ where
 			logger,
 			runtime,
 			config,
+			static_invoice_store,
 		}
 	}
 
@@ -1494,11 +1498,45 @@ where
 			LdkEvent::OnionMessagePeerConnected { .. } => {
 				debug_assert!(false, "We currently don't support onion message interception, so this event should never be emitted.");
 			},
-			LdkEvent::PersistStaticInvoice { .. } => {
-				debug_assert!(false, "We currently don't support static invoice persistence, so this event should never be emitted.");
+
+			LdkEvent::PersistStaticInvoice {
+				invoice,
+				invoice_slot,
+				recipient_id,
+				invoice_persisted_path,
+			} => {
+				match self
+					.static_invoice_store
+					.handle_persist_static_invoice(invoice, invoice_slot, recipient_id)
+					.await
+				{
+					Ok(_) => {},
+					Err(e) => {
+						log_error!(self.logger, "Failed to persist static invoice: {}", e);
+					},
+				};
+
+				self.channel_manager.static_invoice_persisted(invoice_persisted_path);
 			},
-			LdkEvent::StaticInvoiceRequested { .. } => {
-				debug_assert!(false, "We currently don't support static invoice persistence, so this event should never be emitted.");
+			LdkEvent::StaticInvoiceRequested { recipient_id, invoice_slot, reply_path } => {
+				let invoice = self
+					.static_invoice_store
+					.handle_static_invoice_requested(recipient_id, invoice_slot)
+					.await;
+
+				match invoice {
+					Ok(Some(invoice)) => {
+						if let Err(_) =
+							self.channel_manager.send_static_invoice(invoice, reply_path)
+						{
+							log_error!(self.logger, "Failed to send static invoice");
+						}
+					},
+					Ok(None) => {},
+					Err(e) => {
+						log_error!(self.logger, "Failed to retrieve static invoice: {}", e);
+					},
+				}
 			},
 			LdkEvent::FundingTransactionReadyForSigning { .. } => {
 				debug_assert!(false, "We currently don't support interactive-tx, so this event should never be emitted.");

src/lib.rs

@@ -94,7 +94,9 @@ pub mod logger;
 mod message_handler;
 pub mod payment;
 mod peer_store;
+mod rate_limiter;
 mod runtime;
+mod static_invoice_store;
 mod tx_broadcaster;
 mod types;
 mod wallet;
@@ -150,6 +152,7 @@ pub use types::{ChannelDetails, CustomTlvRecord, PeerDetails, UserChannelId};
 
 use logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 
+use lightning::blinded_path::message::BlindedMessagePath;
 use lightning::chain::BestBlock;
 use lightning::events::bump_transaction::Wallet as LdkWallet;
 use lightning::impl_writeable_tlv_based;
@@ -170,6 +173,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
+use crate::static_invoice_store::StaticInvoiceStore;
+
 #[cfg(feature = "uniffi")]
 uniffi::include_scaffolding!("ldk_node");
 
@@ -497,6 +502,8 @@ impl Node {
 			Arc::clone(&self.logger),
 		));
 
+		let static_invoice_store = StaticInvoiceStore::new(Arc::clone(&self.kv_store));
+
 		let event_handler = Arc::new(EventHandler::new(
 			Arc::clone(&self.event_queue),
 			Arc::clone(&self.wallet),
@@ -508,6 +515,7 @@ impl Node {
 			self.liquidity_source.clone(),
 			Arc::clone(&self.payment_store),
 			Arc::clone(&self.peer_store),
+			static_invoice_store,
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
@@ -1461,6 +1469,22 @@ impl Node {
 				Error::PersistenceFailed
 			})
 	}
+
+	/// Sets the [`BlindedMessagePath`]s that we will use as an async recipient to interactively build [`Offer`]s with a
+	/// static invoice server, so the server can serve [`StaticInvoice`]s to payers on our behalf when we're offline.
+	pub fn set_paths_to_static_invoice_server(
+		&self, paths: Vec<BlindedMessagePath>,
+	) -> Result<(), ()> {
+		self.channel_manager.set_paths_to_static_invoice_server(paths)
+	}
+
+	/// [`BlindedMessagePath`]s for an async recipient to communicate with this node and interactively
+	/// build [`Offer`]s and [`StaticInvoice`]s for receiving async payments.
+	pub fn blinded_paths_for_async_recipient(
+		&self, recipient_id: Vec<u8>, relative_expiry: Option<Duration>,
+	) -> Result<Vec<BlindedMessagePath>, ()> {
+		self.channel_manager.blinded_paths_for_async_recipient(recipient_id, relative_expiry)
+	}
 }
 
 impl Drop for Node {

src/payment/bolt12.rs

@@ -450,4 +450,13 @@ impl Bolt12Payment {
 
 		Ok(maybe_wrap(refund))
 	}
+
+	/// Retrieve an [`Offer`] for receiving async payments as an often-offline recipient. Will only return an offer if
+	/// [`Builder::set_paths_to_static_invoice_server`] was called and we succeeded in interactively building a
+	/// [`StaticInvoice`] with the static invoice server.
+	///
+	/// Useful for posting offers to receive payments later, such as posting an offer on a website.
+	pub fn get_async_receive_offer(&self) -> Result<Offer, ()> {
+		self.channel_manager.get_async_receive_offer()
+	}
 }

src/rate_limiter.rs

@@ -0,0 +1,34 @@
+use std::collections::HashMap;
+use std::time::{Duration, Instant};
+
+pub(crate) struct RateLimiter {
+	users: HashMap<Vec<u8>, Instant>,
+	min_interval: Duration,
+}
+
+impl RateLimiter {
+	pub(crate) fn new(min_interval: Duration) -> Self {
+		Self { users: HashMap::new(), min_interval }
+	}
+
+	pub(crate) fn allow(&mut self, user_id: &[u8]) -> bool {
+		let now = Instant::now();
+		match self.users.get(user_id) {
+			Some(&last) if now.duration_since(last) < self.min_interval => false,
+			_ => {
+				let new_user = self.users.insert(user_id.to_vec(), now).is_none();
+
+				// Each time a new user is added, we take the opportunity to clean up old rate limits.
+				if new_user {
+					self.garbage_collect();
+				}
+				true
+			},
+		}
+	}
+
+	fn garbage_collect(&mut self) {
+		let now = Instant::now();
+		self.users.retain(|_, &mut last| now.duration_since(last) < self.min_interval);
+	}
+}

src/static_invoice_store.rs

@@ -0,0 +1,78 @@
+use crate::{hex_utils, rate_limiter::RateLimiter, types::DynStore};
+use bitcoin::hashes::{sha256, Hash};
+use lightning::{offers::static_invoice::StaticInvoice, util::ser::Writeable};
+use std::{
+	sync::{Arc, Mutex},
+	time::Duration,
+};
+
+pub(crate) struct StaticInvoiceStore {
+	kv_store: Arc<DynStore>,
+	request_rate_limiter: Mutex<RateLimiter>,
+	persist_rate_limiter: Mutex<RateLimiter>,
+}
+
+impl StaticInvoiceStore {
+	// Static invoices are stored at "static_invoices/<sha256(recipient_id)>/<invoice_slot>".
+	//
+	// Example:
+	// static_invoices/039058c6f2c0cb492c533b0a4d14ef77cc0f78abccced5287d84a1a2011cfb81/001f
+	const PRIMARY_NAMESPACE: &str = "static_invoices";
+
+	pub(crate) fn new(kv_store: Arc<DynStore>) -> Self {
+		Self {
+			kv_store,
+			request_rate_limiter: Mutex::new(RateLimiter::new(Duration::from_millis(100))),
+			persist_rate_limiter: Mutex::new(RateLimiter::new(Duration::from_millis(100))),
+		}
+	}
+
+	fn check_rate_limit(
+		limiter: &Mutex<RateLimiter>, recipient_id: &[u8],
+	) -> Result<(), lightning::io::Error> {
+		let mut limiter = limiter.lock().unwrap();
+		if !limiter.allow(recipient_id) {
+			Err(lightning::io::Error::new(lightning::io::ErrorKind::Other, "Rate limit exceeded"))
+		} else {
+			Ok(())
+		}
+	}
+
+	pub(crate) async fn handle_static_invoice_requested(
+		&self, recipient_id: Vec<u8>, invoice_slot: u16,
+	) -> Result<Option<StaticInvoice>, lightning::io::Error> {
+		Self::check_rate_limit(&self.request_rate_limiter, &recipient_id)?;
+
+		let (secondary_namespace, key) = Self::get_storage_location(invoice_slot, recipient_id);
+
+		self.kv_store.read(Self::PRIMARY_NAMESPACE, &secondary_namespace, &key).and_then(|data| {
+			data.try_into().map(Some).map_err(|e| {
+				lightning::io::Error::new(
+					lightning::io::ErrorKind::InvalidData,
+					format!("Failed to parse static invoice: {:?}", e),
+				)
+			})
+		})
+	}
+
+	pub(crate) async fn handle_persist_static_invoice(
+		&self, invoice: StaticInvoice, invoice_slot: u16, recipient_id: Vec<u8>,
+	) -> Result<(), lightning::io::Error> {
+		Self::check_rate_limit(&self.persist_rate_limiter, &recipient_id)?;
+
+		let (secondary_namespace, key) = Self::get_storage_location(invoice_slot, recipient_id);
+
+		let mut buf = Vec::new();
+		invoice.write(&mut buf)?;
+
+		self.kv_store.write(Self::PRIMARY_NAMESPACE, &secondary_namespace, &key, buf)
+	}
+
+	fn get_storage_location(invoice_slot: u16, recipient_id: Vec<u8>) -> (String, String) {
+		let hash = sha256::Hash::hash(&recipient_id).to_byte_array();
+		let secondary_namespace = hex_utils::to_string(&hash);
+
+		let key = hex_utils::to_string(&invoice_slot.to_be_bytes());
+		(secondary_namespace, key)
+	}
+}

src/types.rs

@@ -123,7 +123,7 @@ pub(crate) type OnionMessenger = lightning::onion_message::messenger::OnionMesse
 	Arc<ChannelManager>,
 	Arc<MessageRouter>,
 	Arc<ChannelManager>,
-	IgnoringMessageHandler,
+	Arc<ChannelManager>,
 	IgnoringMessageHandler,
 	IgnoringMessageHandler,
 >;

tests/common/mod.rs

@@ -288,10 +288,13 @@ pub(crate) fn setup_two_nodes(
 ) -> (TestNode, TestNode) {
 	println!("== Node A ==");
 	let config_a = random_config(anchor_channels);
+	println!("Node A storage path: {}", config_a.node_config.storage_dir_path.clone());
+
 	let node_a = setup_node(chain_source, config_a, None);
 
 	println!("\n== Node B ==");
 	let mut config_b = random_config(anchor_channels);
+	println!("Node B storage path: {}", config_b.node_config.storage_dir_path.clone());
 	if allow_0conf {
 		config_b.node_config.trusted_peers_0conf.push(node_a.node_id());
 	}