Implement Flow for OffersMessageFlow

shaavan · shaavan · commit 136e08f24d0d · 2025-03-04T03:47:13.000+05:30
diff --git a/lightning/src/ln/channel.rs b/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,
 
diff --git a/lightning/src/ln/channelmanager.rs b/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>,
@@ -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(
diff --git a/lightning/src/offers/flow.rs b/lightning/src/offers/flow.rs
@@ -16,7 +16,7 @@ use bitcoin::{secp256k1, Network};
 use types::payment::PaymentHash;
 use crate::blinded_path::message::{BlindedMessagePath, MessageContext, MessageForwardNode, OffersContext};
 use crate::blinded_path::payment::BlindedPaymentPath;
-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::onion_message::dns_resolution::{DNSSECQuery, HumanReadableName};
 use crate::sign::EntropySource;
@@ -142,4 +142,234 @@ where
 	pub fn get_our_node_id(&self) -> PublicKey {
 		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
+	}
+}
+
+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())
+    }
+}
