Add async payments

Author: joostjager

bindings/ldk_node.udl

@@ -157,6 +157,10 @@ interface Node {
 	boolean verify_signature([ByRef]sequence<u8> msg, [ByRef]string sig, [ByRef]PublicKey pkey);
 	[Throws=NodeError]
 	bytes export_pathfinding_scores();
+	[Throws=NodeError]
+	void set_paths_to_static_invoice_server(sequence<u8> paths);
+	[Throws=NodeError]
+	sequence<u8> blinded_paths_for_async_recipient(sequence<u8> recipient_id);
 };
 
 [Enum]
@@ -209,6 +213,8 @@ interface Bolt12Payment {
 	Bolt12Invoice request_refund_payment([ByRef]Refund refund);
 	[Throws=NodeError]
 	Refund initiate_refund(u64 amount_msat, u32 expiry_secs, u64? quantity, string? payer_note);
+	[Throws=NodeError]
+	Offer get_async_receive_offer();
 };
 
 interface SpontaneousPayment {
@@ -311,6 +317,8 @@ enum NodeError {
 	"InsufficientFunds",
 	"LiquiditySourceUnavailable",
 	"LiquidityFeeTooHigh",
+	"InvalidBlindedPaths",
+	"OperationFailed",
 };
 
 dictionary NodeStatus {

src/builder.rs

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

src/error.rs

@@ -120,6 +120,10 @@ pub enum Error {
 	LiquiditySourceUnavailable,
 	/// The given operation failed due to the LSP's required opening fee being too high.
 	LiquidityFeeTooHigh,
+	/// The given blinded paths are invalid.
+	InvalidBlindedPaths,
+	/// The requested operation failed.
+	OperationFailed,
 }
 
 impl fmt::Display for Error {
@@ -193,6 +197,8 @@ impl fmt::Display for Error {
 			Self::LiquidityFeeTooHigh => {
 				write!(f, "The given operation failed due to the LSP's required opening fee being too high.")
 			},
+			Self::InvalidBlindedPaths => write!(f, "The given blinded paths are invalid."),
+			Self::OperationFailed => write!(f, "The requested operation failed."),
 		}
 	}
 }

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,13 +152,16 @@ 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;
 use lightning::ln::channel_state::ChannelShutdownState;
 use lightning::ln::channelmanager::PaymentId;
 use lightning::ln::msgs::SocketAddress;
 use lightning::routing::gossip::NodeAlias;
+use lightning::util::ser::Readable;
+use lightning::util::ser::Writeable;
 
 use lightning_background_processor::process_events_async_with_kv_store_sync;
 
@@ -170,6 +175,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");
 
@@ -498,6 +505,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),
@@ -509,6 +518,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),
@@ -1476,6 +1486,39 @@ 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.
+	///
+	/// [`Offer`]: lightning::offers::offer::Offer
+	/// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice
+	pub fn set_paths_to_static_invoice_server(&self, paths: Vec<u8>) -> Result<(), Error> {
+		let decoded_paths = <Vec<BlindedMessagePath> as Readable>::read(&mut &paths[..])
+			.or(Err(Error::InvalidBlindedPaths))?;
+
+		self.channel_manager
+			.set_paths_to_static_invoice_server(decoded_paths)
+			.or(Err(Error::InvalidBlindedPaths))
+	}
+
+	/// [`BlindedMessagePath`]s for an async recipient to communicate with this node and interactively
+	/// build [`Offer`]s and [`StaticInvoice`]s for receiving async payments.
+	///
+	/// [`Offer`]: lightning::offers::offer::Offer
+	/// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice
+	pub fn blinded_paths_for_async_recipient(
+		&self, recipient_id: Vec<u8>,
+	) -> Result<Vec<u8>, Error> {
+		let paths = self
+			.channel_manager
+			.blinded_paths_for_async_recipient(recipient_id, None)
+			.or(Err(Error::OperationFailed))?;
+
+		let mut bytes = Vec::new();
+		paths.write(&mut bytes).or(Err(Error::OperationFailed))?;
+
+		Ok(bytes)
+	}
 }
 
 impl Drop for Node {

src/payment/bolt12.rs

@@ -450,4 +450,19 @@ impl Bolt12Payment {
 
 		Ok(maybe_wrap(refund))
 	}
+
+	/// Retrieve an [`Offer`] for receiving async payments as an often-offline recipient. Will only return an offer if
+	/// [`Node::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.
+	///
+	/// [`Node::set_paths_to_static_invoice_server`]: crate::Node::set_paths_to_static_invoice_server
+	/// [`StaticInvoice`]: lightning::offers::static_invoice::StaticInvoice
+	pub fn get_async_receive_offer(&self) -> Result<Offer, Error> {
+		self.channel_manager
+			.get_async_receive_offer()
+			.map(maybe_wrap)
+			.or(Err(Error::OfferCreationFailed))
+	}
 }

src/rate_limiter.rs

@@ -0,0 +1,60 @@
+use std::collections::HashMap;
+use std::time::{Duration, Instant};
+
+/// Implements a leaky-bucket style rate limiter parameterized by the max capacity of the bucket, the refill interval,
+/// and the max idle duration. For every passing of the refill interval, one token is added to the bucket, up to the
+/// maximum capacity. When the bucket has remained at the maximum capacity for longer than the max idle duration, it is
+/// removed to prevent memory leakage.
+pub(crate) struct RateLimiter {
+	users: HashMap<Vec<u8>, Bucket>,
+	capacity: u32,
+	refill_interval: Duration,
+	max_idle: Duration,
+}
+
+struct Bucket {
+	tokens: u32,
+	last_refill: Instant,
+}
+
+impl RateLimiter {
+	pub(crate) fn new(capacity: u32, refill_interval: Duration, max_idle: Duration) -> Self {
+		Self { users: HashMap::new(), capacity, refill_interval, max_idle }
+	}
+
+	pub(crate) fn allow(&mut self, user_id: &[u8]) -> bool {
+		let now = Instant::now();
+
+		let entry = self.users.entry(user_id.to_vec());
+		let is_new_user = matches!(entry, std::collections::hash_map::Entry::Vacant(_));
+
+		let bucket = entry.or_insert(Bucket { tokens: self.capacity, last_refill: now });
+
+		let elapsed = now.duration_since(bucket.last_refill);
+		let tokens_to_add = (elapsed.as_secs_f64() / self.refill_interval.as_secs_f64()) as u32;
+
+		if tokens_to_add > 0 {
+			bucket.tokens = (bucket.tokens + tokens_to_add).min(self.capacity);
+			bucket.last_refill = now;
+		}
+
+		let allow = if bucket.tokens > 0 {
+			bucket.tokens -= 1;
+			true
+		} else {
+			false
+		};
+
+		// Each time a new user is added, we take the opportunity to clean up old rate limits.
+		if is_new_user {
+			self.garbage_collect(self.max_idle);
+		}
+
+		allow
+	}
+
+	fn garbage_collect(&mut self, max_idle: Duration) {
+		let now = Instant::now();
+		self.users.retain(|_, bucket| now.duration_since(bucket.last_refill) < max_idle);
+	}
+}