Introduce FlowEvents for manual handling of offers messages

Until now, offers messages were processed internally without exposing
intermediate steps. This made it harder for callers to intercept or
analyse offer messages before deciding how to respond to them.

`FlowEvents` provide an optional mechanism to surface these events back
to the user. With events enabled, the caller can manually inspect an
incoming message, choose to construct and sign an invoice, or send back
an InvoiceError. This shifts control to the user where needed, while
keeping the default automatic flow unchanged.

Author: shaavan

lightning/src/ln/channelmanager.rs

@@ -3737,7 +3737,7 @@ where
 		let flow = OffersMessageFlow::new(
 			ChainHash::using_genesis_block(params.network), params.best_block,
 			our_network_pubkey, current_timestamp, expanded_inbound_key,
-			node_signer.get_receive_auth_key(), secp_ctx.clone(), message_router
+			node_signer.get_receive_auth_key(), secp_ctx.clone(), message_router, false,
 		);
 
 		ChannelManager {
@@ -14325,7 +14325,7 @@ where
 					None => return None,
 				};
 
-				let invoice_request = match self.flow.verify_invoice_request(invoice_request, context) {
+				let invoice_request = match self.flow.verify_invoice_request(invoice_request, context, responder.clone()) {
 					Ok(InvreqResponseInstructions::SendInvoice(invoice_request)) => invoice_request,
 					Ok(InvreqResponseInstructions::SendStaticInvoice {
 						recipient_id: _recipient_id, invoice_id: _invoice_id
@@ -14336,6 +14336,7 @@ where
 
 						return None
 					},
+					Ok(InvreqResponseInstructions::AsynchronouslyHandleResponse) => return None,
 					Err(_) => return None,
 				};
 
@@ -16959,6 +16960,7 @@ where
 			args.node_signer.get_receive_auth_key(),
 			secp_ctx.clone(),
 			args.message_router,
+			false,
 		)
 		.with_async_payments_offers_cache(async_receive_offer_cache);
 

lightning/src/offers/flow.rs

@@ -70,6 +70,32 @@ use {
 	crate::onion_message::dns_resolution::{DNSResolverMessage, DNSSECQuery, OMNameResolver},
 };
 
+/// Defines the events that can be optionally triggered when processing offers messages.
+///
+/// Once generated, these events are stored in the [`OffersMessageFlow`], where they can be
+/// manually inspected and responded to.
+pub enum FlowEvents {
+	/// Notifies that an [`InvoiceRequest`] has been received.
+	///
+	/// To respond to this message:
+	/// - Based on the variant of [`InvoiceRequestVerifiedFromOffer`], create the appropriate invoice builder:
+	///   - [`InvoiceRequestVerifiedFromOffer::DerivedKeys`] → use
+	///     [`OffersMessageFlow::create_invoice_builder_from_invoice_request_with_keys`]
+	///   - [`InvoiceRequestVerifiedFromOffer::ExplicitKeys`] → use
+	///     [`OffersMessageFlow::create_invoice_builder_from_invoice_request_without_keys`]
+	/// - After building the invoice, sign it and send it back using the provided reply path via
+	///   [`OffersMessageFlow::enqueue_invoice_using_reply_paths`].
+	///
+	/// If the invoice request is invalid, respond with an [`InvoiceError`] using
+	/// [`OffersMessageFlow::enqueue_invoice_error`].
+	InvoiceRequestReceived {
+		/// The received, verified invoice request.
+		invoice_request: InvoiceRequestVerifiedFromOffer,
+		/// The reply path to use when responding to the invoice request.
+		reply_path: BlindedMessagePath,
+	},
+}
+
 /// A BOLT12 offers code and flow utility provider, which facilitates
 /// BOLT12 builder generation and onion message handling.
 ///
@@ -91,6 +117,8 @@ where
 	secp_ctx: Secp256k1<secp256k1::All>,
 	message_router: MR,
 
+	pub(crate) enable_events: bool,
+
 	#[cfg(not(any(test, feature = "_test_utils")))]
 	pending_offers_messages: Mutex<Vec<(OffersMessage, MessageSendInstructions)>>,
 	#[cfg(any(test, feature = "_test_utils"))]
@@ -103,6 +131,8 @@ where
 	pub(crate) hrn_resolver: OMNameResolver,
 	#[cfg(feature = "dnssec")]
 	pending_dns_onion_messages: Mutex<Vec<(DNSResolverMessage, MessageSendInstructions)>>,
+
+	pending_flow_events: Mutex<Vec<FlowEvents>>,
 }
 
 impl<MR: Deref> OffersMessageFlow<MR>
@@ -114,6 +144,7 @@ where
 		chain_hash: ChainHash, best_block: BestBlock, our_network_pubkey: PublicKey,
 		current_timestamp: u32, inbound_payment_key: inbound_payment::ExpandedKey,
 		receive_auth_key: ReceiveAuthKey, secp_ctx: Secp256k1<secp256k1::All>, message_router: MR,
+		enable_events: bool,
 	) -> Self {
 		Self {
 			chain_hash,
@@ -128,6 +159,8 @@ where
 			secp_ctx,
 			message_router,
 
+			enable_events,
+
 			pending_offers_messages: Mutex::new(Vec::new()),
 			pending_async_payments_messages: Mutex::new(Vec::new()),
 
@@ -137,6 +170,8 @@ where
 			pending_dns_onion_messages: Mutex::new(Vec::new()),
 
 			async_receive_offer_cache: Mutex::new(AsyncReceiveOfferCache::new()),
+
+			pending_flow_events: Mutex::new(Vec::new()),
 		}
 	}
 
@@ -152,6 +187,18 @@ where
 		self
 	}
 
+	/// Enables [`FlowEvents`] for this flow.
+	///
+	/// By default, events are not emitted when processing offers messages. Calling this method
+	/// sets the internal `enable_events` flag to `true`, allowing you to receive [`FlowEvents`]
+	/// such as [`FlowEvents::InvoiceRequestReceived`].
+	///
+	/// This is useful when you want to manually inspect, handle, or respond to incoming
+	/// offers messages rather than having them processed automatically.
+	pub fn enable_events(&mut self) {
+		self.enable_events = true;
+	}
+
 	/// 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.
@@ -407,13 +454,16 @@ pub enum InvreqResponseInstructions {
 		/// An identifier for the specific invoice being requested by the payer.
 		invoice_id: u128,
 	},
+	/// We are recipient of this payment, and should handle the response asynchronously.
+	AsynchronouslyHandleResponse,
 }
 
 impl<MR: Deref> OffersMessageFlow<MR>
 where
 	MR::Target: MessageRouter,
 {
 	/// Verifies an [`InvoiceRequest`] using the provided [`OffersContext`] or the [`InvoiceRequest::metadata`].
+	/// It also helps determine the response instructions, corresponding to the verified invoice request must be taken.
 	///
 	/// - If an [`OffersContext::InvoiceRequest`] with a `nonce` is provided, verification is performed using recipient context data.
 	/// - If no context is provided but the [`InvoiceRequest`] contains [`Offer`] metadata, verification is performed using that metadata.
@@ -426,27 +476,26 @@ where
 	/// - The verification process (via recipient context data or metadata) fails.
 	pub fn verify_invoice_request(
 		&self, invoice_request: InvoiceRequest, context: Option<OffersContext>,
+		responder: Responder,
 	) -> Result<InvreqResponseInstructions, ()> {
 		let secp_ctx = &self.secp_ctx;
 		let expanded_key = &self.inbound_payment_key;
 
+		if let Some(OffersContext::StaticInvoiceRequested {
+			recipient_id,
+			invoice_id,
+			path_absolute_expiry,
+		}) = context
+		{
+			if path_absolute_expiry < self.duration_since_epoch() {
+				return Err(());
+			}
+			return Ok(InvreqResponseInstructions::SendStaticInvoice { recipient_id, invoice_id });
+		}
+
 		let nonce = match context {
-			None if invoice_request.metadata().is_some() => None,
 			Some(OffersContext::InvoiceRequest { nonce }) => Some(nonce),
-			Some(OffersContext::StaticInvoiceRequested {
-				recipient_id,
-				invoice_id,
-				path_absolute_expiry,
-			}) => {
-				if path_absolute_expiry < self.duration_since_epoch() {
-					return Err(());
-				}
-
-				return Ok(InvreqResponseInstructions::SendStaticInvoice {
-					recipient_id,
-					invoice_id,
-				});
-			},
+			None if invoice_request.metadata().is_some() => None,
 			_ => return Err(()),
 		};
 
@@ -457,7 +506,16 @@ where
 			None => invoice_request.verify_using_metadata(expanded_key, secp_ctx),
 		}?;
 
-		Ok(InvreqResponseInstructions::SendInvoice(invoice_request))
+		if self.enable_events {
+			self.pending_flow_events.lock().unwrap().push(FlowEvents::InvoiceRequestReceived {
+				invoice_request,
+				reply_path: responder.into_blinded_path(),
+			});
+
+			Ok(InvreqResponseInstructions::AsynchronouslyHandleResponse)
+		} else {
+			Ok(InvreqResponseInstructions::SendInvoice(invoice_request))
+		}
 	}
 
 	/// Verifies a [`Bolt12Invoice`] using the provided [`OffersContext`] or the invoice's payer metadata,
@@ -1288,6 +1346,11 @@ where
 		Ok(())
 	}
 
+	/// Enqueues the generated [`FlowEvents`] to be processed.
+	pub fn enqueue_flow_event(&self, flow_event: FlowEvents) {
+		self.pending_flow_events.lock().unwrap().push(flow_event);
+	}
+
 	/// Gets the enqueued [`OffersMessage`] with their corresponding [`MessageSendInstructions`].
 	pub fn release_pending_offers_messages(&self) -> Vec<(OffersMessage, MessageSendInstructions)> {
 		core::mem::take(&mut self.pending_offers_messages.lock().unwrap())
@@ -1300,6 +1363,11 @@ where
 		core::mem::take(&mut self.pending_async_payments_messages.lock().unwrap())
 	}
 
+	/// Gets the enqueued [`FlowEvents`] to be processed.
+	pub fn release_pending_flow_events(&self) -> Vec<FlowEvents> {
+		core::mem::take(&mut self.pending_flow_events.lock().unwrap())
+	}
+
 	/// Gets the enqueued [`DNSResolverMessage`] with their corresponding [`MessageSendInstructions`].
 	#[cfg(feature = "dnssec")]
 	pub fn release_pending_dns_messages(