Implement Flow for OffersMessageFlow

Author: shaavan

lightning/src/ln/channel.rs

@@ -1807,7 +1807,7 @@ pub(super) struct ChannelContext<SP: Deref> where SP::Target: SignerProvider {
 	/// Either the height at which this channel was created or the height at which it was last
 	/// serialized if it was serialized by versions prior to 0.0.103.
 	/// We use this to close if funding is never broadcasted.
-	pub(super) channel_creation_height: u32,
+	pub(crate) channel_creation_height: u32,
 
 	counterparty_dust_limit_satoshis: u64,
 

lightning/src/ln/channelmanager.rs

@@ -1349,11 +1349,11 @@ impl Readable for Option<RAAMonitorUpdateBlockingAction> {
 }
 
 /// State we hold per-peer.
-pub(super) struct PeerState<SP: Deref> where SP::Target: SignerProvider {
+pub(crate) struct PeerState<SP: Deref> where SP::Target: SignerProvider {
 	/// `channel_id` -> `Channel`
 	///
 	/// Holds all channels where the peer is the counterparty.
-	pub(super) channel_by_id: HashMap<ChannelId, Channel<SP>>,
+	pub(crate) channel_by_id: HashMap<ChannelId, Channel<SP>>,
 	/// `temporary_channel_id` -> `InboundChannelRequest`.
 	///
 	/// When manual channel acceptance is enabled, this holds all unaccepted inbound channels where
@@ -1362,7 +1362,7 @@ pub(super) struct PeerState<SP: Deref> where SP::Target: SignerProvider {
 	/// the channel is rejected, then the entry is simply removed.
 	pub(super) inbound_channel_request_by_id: HashMap<ChannelId, InboundChannelRequest>,
 	/// The latest `InitFeatures` we heard from the peer.
-	latest_features: InitFeatures,
+	pub(crate) latest_features: InitFeatures,
 	/// Messages to send to the peer - pushed to in the same lock that they are generated in (except
 	/// for broadcast messages, where ordering isn't as strict).
 	pub(super) pending_msg_events: Vec<MessageSendEvent>,
@@ -8035,7 +8035,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/
 
 		let per_peer_state = self.per_peer_state.read().unwrap();
 		let peer_state_mutex = per_peer_state.get(counterparty_node_id)
-		    .ok_or_else(|| {
+			.ok_or_else(|| {
 				debug_assert!(false);
 				MsgHandleErrInternal::send_err_msg_no_close(
 					format!("Can't find a peer matching the passed counterparty node_id {}", counterparty_node_id),
@@ -10202,7 +10202,7 @@ macro_rules! create_refund_builder { ($self: ident, $builder: ty) => {
 /// Sending multiple requests increases the chances of successful delivery in case some
 /// paths are unavailable. However, only one invoice for a given [`PaymentId`] will be paid,
 /// even if multiple invoices are received.
-const OFFERS_MESSAGE_REQUEST_LIMIT: usize = 10;
+pub(crate) const OFFERS_MESSAGE_REQUEST_LIMIT: usize = 10;
 
 impl<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref> ChannelManager<M, T, ES, NS, SP, F, R, MR, L>
 where
@@ -10812,6 +10812,23 @@ where
 			.and_then(|paths| (!paths.is_empty()).then(|| paths).ok_or(()))
 	}
 
+	fn peer_for_blinded_path(&self) -> Vec<MessageForwardNode> {
+		self.per_peer_state.read().unwrap()
+			.iter()
+			.map(|(node_id, peer_state)| (node_id, peer_state.lock().unwrap()))
+			.filter(|(_, peer)| peer.is_connected)
+			.filter(|(_, peer)| peer.latest_features.supports_onion_messages())
+			.map(|(node_id, peer)| MessageForwardNode {
+				node_id: *node_id,
+				short_channel_id: peer.channel_by_id
+					.iter()
+					.filter(|(_, channel)| channel.context().is_usable())
+					.min_by_key(|(_, channel)| channel.context().channel_creation_height)
+					.and_then(|(_, channel)| channel.context().get_short_channel_id()),
+			})
+			.collect::<Vec<_>>()
+	}
+
 	/// Creates multi-hop blinded payment paths for the given `amount_msats` by delegating to
 	/// [`Router::create_blinded_payment_paths`].
 	fn create_blinded_payment_paths(

lightning/src/offers/flow.rs

@@ -16,7 +16,9 @@ use crate::blinded_path::message::{
 use crate::blinded_path::payment::BlindedPaymentPath;
 use crate::chain;
 use crate::chain::transaction::TransactionData;
-use crate::ln::channelmanager::PaymentId;
+use crate::ln::channelmanager::{
+	PaymentId, MAX_SHORT_LIVED_RELATIVE_EXPIRY, OFFERS_MESSAGE_REQUEST_LIMIT,
+};
 use crate::ln::inbound_payment;
 use crate::offers::invoice::{Bolt12Invoice, DerivedSigningPubkey, InvoiceBuilder};
 use crate::offers::invoice_request::{InvoiceRequest, InvoiceRequestBuilder};
@@ -167,6 +169,17 @@ where
 		self.our_network_pubkey
 	}
 
+	fn duration_since_epoch(&self) -> Duration {
+		#[cfg(not(feature = "std"))]
+		let now = Duration::from_secs(self.highest_seen_timestamp.load(Ordering::Acquire) as u64);
+		#[cfg(feature = "std")]
+		let now = std::time::SystemTime::now()
+			.duration_since(std::time::SystemTime::UNIX_EPOCH)
+			.expect("SystemTime::now() should come after SystemTime::UNIX_EPOCH");
+
+		now
+	}
+
 	fn best_block_updated(&self, header: &Header) {
 		macro_rules! max_time {
 			($timestamp: expr) => {
@@ -189,6 +202,257 @@ where
 	}
 }
 
+impl<ES: Deref, MR: Deref> Flow for OffersMessageFlow<ES, MR>
+where
+	ES::Target: EntropySource,
+	MR::Target: MessageRouter,
+{
+	fn create_offer_builder(
+		&self, nonce: Nonce,
+	) -> Result<OfferBuilder<DerivedMetadata, secp256k1::All>, Bolt12SemanticError> {
+		let node_id = self.get_our_node_id();
+		let expanded_key = &self.inbound_payment_key;
+		let secp_ctx = &self.secp_ctx;
+
+		let builder = OfferBuilder::deriving_signing_pubkey(node_id, expanded_key, nonce, secp_ctx)
+			.chain_hash(self.chain_hash);
+
+		Ok(builder)
+	}
+
+	fn create_refund_builder(
+		&self, amount_msats: u64, absolute_expiry: Duration, payment_id: PaymentId, nonce: Nonce,
+	) -> Result<RefundBuilder<secp256k1::All>, Bolt12SemanticError> {
+		let node_id = self.get_our_node_id();
+		let expanded_key = &self.inbound_payment_key;
+		let secp_ctx = &self.secp_ctx;
+
+		let builder = RefundBuilder::deriving_signing_pubkey(
+			node_id,
+			expanded_key,
+			nonce,
+			secp_ctx,
+			amount_msats,
+			payment_id,
+		)?
+		.chain_hash(self.chain_hash)
+		.absolute_expiry(absolute_expiry);
+
+		Ok(builder)
+	}
+
+	fn create_invoice_request_builder<'a>(
+		&'a self, offer: &'a Offer, nonce: Nonce, quantity: Option<u64>, amount_msats: Option<u64>,
+		payer_note: Option<String>, human_readable_name: Option<HumanReadableName>,
+		payment_id: PaymentId,
+	) -> Result<InvoiceRequestBuilder<'a, 'a, secp256k1::All>, Bolt12SemanticError> {
+		let expanded_key = &self.inbound_payment_key;
+		let secp_ctx = &self.secp_ctx;
+
+		let builder = offer
+			.request_invoice(expanded_key, nonce, secp_ctx, payment_id)?
+			.chain_hash(self.chain_hash)?;
+
+		let builder = match quantity {
+			None => builder,
+			Some(quantity) => builder.quantity(quantity)?,
+		};
+		let builder = match amount_msats {
+			None => builder,
+			Some(amount_msats) => builder.amount_msats(amount_msats)?,
+		};
+		let builder = match payer_note {
+			None => builder,
+			Some(payer_note) => builder.payer_note(payer_note),
+		};
+		let builder = match human_readable_name {
+			None => builder,
+			Some(hrn) => builder.sourced_from_human_readable_name(hrn),
+		};
+
+		Ok(builder.into())
+	}
+
+	fn create_invoice_builder<'a>(
+		&'a self, refund: &'a Refund, payment_paths: Vec<BlindedPaymentPath>,
+		payment_hash: PaymentHash,
+	) -> Result<InvoiceBuilder<'a, DerivedSigningPubkey>, Bolt12SemanticError> {
+		let expanded_key = &self.inbound_payment_key;
+		let entropy = &*self.entropy_source;
+
+		#[cfg(feature = "std")]
+		let builder = refund.respond_using_derived_keys(
+			payment_paths,
+			payment_hash,
+			expanded_key,
+			entropy,
+		)?;
+		#[cfg(not(feature = "std"))]
+		let created_at = Duration::from_secs(self.highest_seen_timestamp.load(Ordering::Acquire) as u64);
+		#[cfg(not(feature = "std"))]
+		let builder = refund.respond_using_derived_keys_no_std(
+			payment_paths,
+			payment_hash,
+			created_at,
+			expanded_key,
+			entropy,
+		)?;
+		let builder: InvoiceBuilder<DerivedSigningPubkey> = builder.into();
+
+		Ok(builder)
+	}
+
+	fn create_blinded_paths(
+		&self, peers: Vec<MessageForwardNode>, context: MessageContext,
+	) -> Result<Vec<BlindedMessagePath>, ()> {
+		let recipient = self.get_our_node_id();
+		let secp_ctx = &self.secp_ctx;
+
+		let peers = peers.iter().map(|node| node.node_id).collect::<Vec<_>>();
+
+		self.message_router
+			.create_blinded_paths(recipient, context, peers, secp_ctx)
+			.and_then(|paths| (!paths.is_empty()).then(|| paths).ok_or(()))
+	}
+
+	fn create_compact_blinded_paths(
+		&self, peers: Vec<MessageForwardNode>, context: OffersContext,
+	) -> Result<Vec<BlindedMessagePath>, ()> {
+		let recipient = self.get_our_node_id();
+		let secp_ctx = &self.secp_ctx;
+
+		self.message_router
+			.create_compact_blinded_paths(
+				recipient,
+				MessageContext::Offers(context),
+				peers,
+				secp_ctx,
+			)
+			.and_then(|paths| (!paths.is_empty()).then(|| paths).ok_or(()))
+	}
+
+	fn create_blinded_paths_using_absolute_expiry(
+		&self, peers: Vec<MessageForwardNode>, context: OffersContext,
+		absolute_expiry: Option<Duration>,
+	) -> Result<Vec<BlindedMessagePath>, ()> {
+		let now = self.duration_since_epoch();
+		let max_short_lived_absolute_expiry = now.saturating_add(MAX_SHORT_LIVED_RELATIVE_EXPIRY);
+
+		if absolute_expiry.unwrap_or(Duration::MAX) <= max_short_lived_absolute_expiry {
+			self.create_compact_blinded_paths(peers, context)
+		} else {
+			self.create_blinded_paths(peers, MessageContext::Offers(context))
+		}
+	}
+
+	fn enqueue_invoice_request(
+		&self, invoice_request: InvoiceRequest, reply_paths: Vec<BlindedMessagePath>,
+	) -> Result<(), Bolt12SemanticError> {
+		let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
+		if !invoice_request.paths().is_empty() {
+			reply_paths
+				.iter()
+				.flat_map(|reply_path| {
+					invoice_request.paths().iter().map(move |path| (path, reply_path))
+				})
+				.take(OFFERS_MESSAGE_REQUEST_LIMIT)
+				.for_each(|(path, reply_path)| {
+					let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+						destination: Destination::BlindedPath(path.clone()),
+						reply_path: reply_path.clone(),
+					};
+					let message = OffersMessage::InvoiceRequest(invoice_request.clone());
+					pending_offers_messages.push((message, instructions));
+				});
+		} else if let Some(node_id) = invoice_request.issuer_signing_pubkey() {
+			for reply_path in reply_paths {
+				let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+					destination: Destination::Node(node_id),
+					reply_path,
+				};
+				let message = OffersMessage::InvoiceRequest(invoice_request.clone());
+				pending_offers_messages.push((message, instructions));
+			}
+		} else {
+			debug_assert!(false);
+			return Err(Bolt12SemanticError::MissingIssuerSigningPubkey);
+		}
+
+		Ok(())
+	}
+
+	fn enqueue_invoice(
+		&self, invoice: Bolt12Invoice, refund: &Refund, reply_paths: Vec<BlindedMessagePath>,
+	) -> Result<(), Bolt12SemanticError> {
+		let mut pending_offers_messages = self.pending_offers_messages.lock().unwrap();
+		if refund.paths().is_empty() {
+			for reply_path in reply_paths {
+				let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+					destination: Destination::Node(refund.payer_signing_pubkey()),
+					reply_path,
+				};
+				let message = OffersMessage::Invoice(invoice.clone());
+				pending_offers_messages.push((message, instructions));
+			}
+		} else {
+			reply_paths
+				.iter()
+				.flat_map(|reply_path| refund.paths().iter().map(move |path| (path, reply_path)))
+				.take(OFFERS_MESSAGE_REQUEST_LIMIT)
+				.for_each(|(path, reply_path)| {
+					let instructions = MessageSendInstructions::WithSpecifiedReplyPath {
+						destination: Destination::BlindedPath(path.clone()),
+						reply_path: reply_path.clone(),
+					};
+					let message = OffersMessage::Invoice(invoice.clone());
+					pending_offers_messages.push((message, instructions));
+				});
+		}
+
+		Ok(())
+	}
+
+	fn enqueue_dns_onion_message(
+		&self, message: DNSSECQuery, dns_resolvers: Vec<Destination>,
+		reply_paths: Vec<BlindedMessagePath>,
+	) -> Result<(), Bolt12SemanticError> {
+		let message_params = dns_resolvers
+			.iter()
+			.flat_map(|destination| reply_paths.iter().map(move |path| (path, destination)))
+			.take(OFFERS_MESSAGE_REQUEST_LIMIT);
+		for (reply_path, destination) in message_params {
+			self.pending_dns_onion_messages.lock().unwrap().push((
+				DNSResolverMessage::DNSSECQuery(message.clone()),
+				MessageSendInstructions::WithSpecifiedReplyPath {
+					destination: destination.clone(),
+					reply_path: reply_path.clone(),
+				},
+			));
+		}
+
+		Ok(())
+	}
+
+	fn get_and_clear_pending_offers_messages(
+		&self,
+	) -> Vec<(OffersMessage, MessageSendInstructions)> {
+		core::mem::take(&mut self.pending_offers_messages.lock().unwrap())
+	}
+
+	fn get_and_clear_pending_async_messages(
+		&self,
+	) -> Vec<(AsyncPaymentsMessage, MessageSendInstructions)> {
+		core::mem::take(&mut self.pending_async_payments_messages.lock().unwrap())
+	}
+
+	#[cfg(feature = "dnssec")]
+	fn get_and_clear_pending_dns_messages(
+		&self,
+	) -> Vec<(DNSResolverMessage, MessageSendInstructions)> {
+		core::mem::take(&mut self.pending_dns_onion_messages.lock().unwrap())
+	}
+}
+
 impl<ES: Deref, MR: Deref> chain::Listen for OffersMessageFlow<ES, MR>
 where
 	ES::Target: EntropySource,