[feat] Advance recurrence after successful payment

On successful completion of a payment that corresponds to a recurrence
we introduce logic to update the details after validating the invoice
parameters.

Co-authored-by: GLaDOS

Author: shaavan

src/event.rs

@@ -24,6 +24,7 @@ use lightning::events::{
 };
 use lightning::ln::channelmanager::{PaymentId, TrustedChannelFeatures};
 use lightning::ln::types::ChannelId;
+use lightning::offers::offer::Offer;
 use lightning::routing::gossip::NodeId;
 use lightning::sign::EntropySource;
 use lightning::util::config::{ChannelConfigOverrides, ChannelConfigUpdate};
@@ -36,8 +37,9 @@ use lightning_types::payment::{PaymentHash, PaymentPreimage};
 
 use crate::config::{may_announce_channel, Config, PEER_RECONNECTION_INTERVAL};
 use crate::connection::ConnectionManager;
-use crate::data_store::DataStoreUpdateResult;
+use crate::data_store::{DataStoreUpdateResult, StorableObject};
 use crate::fee_estimator::ConfirmationTarget;
+use crate::ffi::Bolt12Invoice;
 #[cfg(feature = "uniffi")]
 use crate::ffi::PaidBolt12Invoice;
 use crate::io::{
@@ -48,6 +50,7 @@ use crate::liquidity::LiquiditySource;
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::asynchronous::om_mailbox::OnionMessageMailbox;
 use crate::payment::asynchronous::static_invoice_store::StaticInvoiceStore;
+use crate::payment::recurrence::RecurrenceStatus;
 use crate::payment::store::{
 	PaymentDetails, PaymentDetailsUpdate, PaymentDirection, PaymentKind, PaymentStatus,
 };
@@ -694,6 +697,82 @@ where
 		})
 	}
 
+	/// Processes a successful payment event for an outbound recurrence.
+	async fn advance_recurrence_on_payment_sent(
+		&self, payment_id: PaymentId, invoice: &Bolt12Invoice,
+	) -> Result<(), ReplayEvent> {
+		let mut matches = self
+			.recurrence_store
+			.list_filter(|details| details.active_payment_id == Some(payment_id))
+			.await
+			.into_iter();
+
+		let Some(mut details) = matches.next() else {
+			// Ordinary payments do not have recurrence state and need no extra handling.
+			return Ok(());
+		};
+
+		debug_assert!(
+			matches.next().is_none(),
+			"multiple recurrences have the same active payment ID?"
+		);
+
+		if details.last_successful_payment_id == Some(payment_id) {
+			// Event replay after a completed transition must be idempotent.
+			return Ok(());
+		}
+
+		let offer = Offer::try_from(details.original_offer.clone()).map_err(|_| ReplayEvent())?;
+
+		if invoice.offer_id() != Some(offer.id()) {
+			debug_assert!(false, "invoice offer ID does not match recurrence offer ID");
+			return Ok(());
+		}
+
+		let Some(invoice_recurrence) = invoice.invoice_recurrence() else {
+			debug_assert!(false, "Invoice that should be corresponding to recurrence according to our recurrence store doesn't?");
+			return Ok(());
+		};
+
+		if let Some(recurrence_basetime) = details.basetime {
+			if recurrence_basetime != invoice_recurrence.recurrence_basetime() {
+				debug_assert!(false, "Recurrence basetime info mismatch");
+				return Ok(());
+			}
+		}
+
+		details.paid_count = details.paid_count.saturating_add(1);
+		details.basetime.get_or_insert(invoice_recurrence.recurrence_basetime());
+		details.opaque_state =
+			invoice_recurrence.recurrence_next_state().map(|state| state.to_vec());
+		details.last_successful_payment_id = Some(payment_id);
+		details.active_payment_id = None;
+
+		let recurrence = offer.offer_recurrence().ok_or(ReplayEvent())?;
+		let period_index = recurrence
+			.period_index(details.paid_count as u32, details.initial_start)
+			.map_err(|_| ReplayEvent())?;
+
+		if recurrence.recurrence_limit.map(|limit| period_index >= limit.0).unwrap_or(false) {
+			details.status = RecurrenceStatus::Completed;
+		}
+
+		match self.recurrence_store.update(details.to_update()).await {
+			Ok(DataStoreUpdateResult::Updated | DataStoreUpdateResult::Unchanged) => Ok(()),
+			Ok(DataStoreUpdateResult::NotFound) => {
+				log_error!(
+					self.logger,
+					"Recurrence disappeared before its payment update was stored"
+				);
+				Err(ReplayEvent())
+			},
+			Err(e) => {
+				log_error!(self.logger, "Failed to store updated recurrence: {}", e);
+				Err(ReplayEvent())
+			},
+		}
+	}
+
 	async fn resolve_inbound_payment_id(
 		&self, event_payment_id: Option<PaymentId>, payment_hash: &PaymentHash,
 	) -> Result<(PaymentId, Option<PaymentDetails>), ReplayEvent> {
@@ -1413,6 +1492,16 @@ where
 					..PaymentDetailsUpdate::new(payment_id)
 				};
 
+				if let Some(invoice) = bolt12_invoice
+					.as_ref()
+					.and_then(|paid_invoice| paid_invoice.bolt12_invoice())
+				{
+					if let Err(e) = self.advance_recurrence_on_payment_sent(payment_id, invoice.as_ref()).await {
+						log_error!(self.logger, "Failed to advance recurrence: replaying event");
+						return Err(e);
+					}
+				}
+
 				match self.payment_store.update(update).await {
 					Ok(_) => {},
 					Err(e) => {