1use std::fmt;
2
3use sqlx::{PgConnection, PgPool};
4use syrup_rail::{
5 ApprovedProcessorEvidence, BillingEvent, BillingEventSubject, BillingScopeId,
6 GatewayMutationError, GatewayNotSubmittedError, GatewayPaymentOutcome, GatewayPaymentStatus,
7 GatewayProviderKey, GatewaySaleIntent, GatewaySaleRequest, PaymentAttempt, PaymentAttemptId,
8 PaymentAttemptStatus, PaymentResolutionCode, ProcessorChargeProgression, ProcessorChargeRole,
9 ProcessorEvidence, RecoverSubscriptionPayment, SubscriptionEnrollmentPaymentResult,
10 SubscriptionRecoveryReservation, SubscriptionRecoverySubmissionOutcome,
11 SubscriptionRecoverySubmissionRejection,
12};
13
14use crate::{
15 BillingTransactionCoordinator, BillingTransactionSubjectState,
16 attempts::find_payment_attempt_by_id_on_connection,
17 processor_charges::{
18 LockFreeApprovedEvidenceTerms, ObservedCharge, observe_processor_charge, transition_charge,
19 },
20};
21
22use super::{
23 APPROVED_APPLICATION_ATTEMPTS, APPROVED_EVIDENCE_RETRY_DELAY, APPROVED_EVIDENCE_WRITE_ATTEMPTS,
24 AttemptResolutionStatus, BILLING_LOCK_TIMEOUT, GatewayNotSubmittedPolicy,
25 INVALID_APPLICATION_STATE, OutcomeApplication, OutcomeReservation, OutcomeResolutionBoundary,
26 OutcomeResolutionCommand, RECOVERY_APPROVED_STORAGE_FAILURE_TEXT,
27 RECOVERY_INCOMPLETE_APPROVAL_TEXT, RECOVERY_STALE_STATE_TEXT, RateLimitCooldown,
28 SubscriptionEnrollmentApplicationError, advance_subscription_discount_after_successful_charge,
29 apply_resumable_not_submitted_policy, disable_payment_method_if_unreferenced,
30 finalize_approved_application, is_retryable_evidence_error, load_applied_subscription,
31 load_subscription, lock_expected_reservation_attempt, lock_payment_method_domain,
32 lock_subscription_aggregate, mark_attempt_approved, mutation_error_evidence,
33 park_locked_attempt, payment_result_for_attempt,
34 persist_approved_evidence_without_attempt_lock, recovery_subscription_matches,
35 resolve_pool_outcome, set_application_timeouts, upsert_payment_method,
36};
37
38pub struct AdmittedSubscriptionRecovery {
41 reservation: SubscriptionRecoveryReservation,
42 attempt: PaymentAttempt,
43}
44
45impl AdmittedSubscriptionRecovery {
46 pub const fn attempt(&self) -> &PaymentAttempt {
47 &self.attempt
48 }
49}
50
51impl fmt::Debug for AdmittedSubscriptionRecovery {
52 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
53 formatter
54 .debug_struct("AdmittedSubscriptionRecovery")
55 .field("attempt", &self.attempt)
56 .field("has_submission_authority", &true)
57 .finish()
58 }
59}
60
61#[derive(Debug)]
62pub enum SubscriptionRecoveryAdmissionOutcome {
63 Admitted(Box<AdmittedSubscriptionRecovery>),
64 AlreadyAdmitted(PaymentAttempt),
65 Rejected {
66 attempt: PaymentAttempt,
67 reason: SubscriptionRecoverySubmissionRejection,
68 },
69}
70
71#[derive(Debug)]
72pub enum SubscriptionRecoveryProviderResult {
73 Payment(SubscriptionEnrollmentPaymentResult),
74 NotSubmitted {
78 payment: SubscriptionEnrollmentPaymentResult,
79 error: GatewayNotSubmittedError,
80 },
81}
82
83impl SubscriptionRecoveryProviderResult {
84 pub const fn payment(&self) -> &SubscriptionEnrollmentPaymentResult {
85 match self {
86 Self::Payment(payment) | Self::NotSubmitted { payment, .. } => payment,
87 }
88 }
89
90 pub fn into_payment(self) -> SubscriptionEnrollmentPaymentResult {
91 match self {
92 Self::Payment(payment) | Self::NotSubmitted { payment, .. } => payment,
93 }
94 }
95}
96
97pub async fn admit_subscription_recovery_submission(
99 pool: &PgPool,
100 reservation: &SubscriptionRecoveryReservation,
101) -> Result<SubscriptionRecoveryAdmissionOutcome, SubscriptionEnrollmentApplicationError> {
102 let mut transaction = pool.begin().await?;
103 let outcome =
104 crate::admit_subscription_recovery_submission_in_transaction(&mut transaction, reservation)
105 .await?;
106 transaction.commit().await?;
107 Ok(match outcome {
108 SubscriptionRecoverySubmissionOutcome::Admitted(attempt) => {
109 SubscriptionRecoveryAdmissionOutcome::Admitted(Box::new(AdmittedSubscriptionRecovery {
110 reservation: reservation.clone(),
111 attempt,
112 }))
113 }
114 SubscriptionRecoverySubmissionOutcome::AlreadyAdmitted(attempt) => {
115 SubscriptionRecoveryAdmissionOutcome::AlreadyAdmitted(attempt)
116 }
117 SubscriptionRecoverySubmissionOutcome::Rejected { attempt, reason } => {
118 SubscriptionRecoveryAdmissionOutcome::Rejected { attempt, reason }
119 }
120 })
121}
122
123pub async fn submit_admitted_subscription_recovery(
127 pool: &PgPool,
128 coordinator: &dyn BillingTransactionCoordinator,
129 admission: AdmittedSubscriptionRecovery,
130 command: &RecoverSubscriptionPayment,
131 gateway: crate::ModeVerifiedGateway<'_>,
132) -> Result<SubscriptionRecoveryProviderResult, SubscriptionEnrollmentApplicationError> {
133 let resolved_gateway = gateway.resolved_gateway();
134 if admission.attempt.identity() != admission.reservation.identity()
135 || admission.attempt.request() != admission.reservation.request()
136 || admission.attempt.status() != PaymentAttemptStatus::Pending
137 || admission
138 .attempt
139 .state()
140 .timestamps()
141 .submitted_at()
142 .is_none()
143 || !admission
144 .reservation
145 .matches_submission(command, resolved_gateway)
146 {
147 return Err(SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch);
148 }
149 let Some(gateway) = gateway.authorize_attempt(&admission.reservation.identity()) else {
150 return Err(SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch);
151 };
152 let charge =
153 syrup_rail::ChargeAmount::try_from(admission.attempt.request().amount()).map_err(|_| {
154 SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
155 })?;
156 let request = GatewaySaleRequest::new(
157 charge,
158 admission.attempt.request().gateway_order_id().clone(),
159 GatewaySaleIntent::InitialStoredCredential {
160 payment_token: command.payment_token().clone(),
161 },
162 Some(command.billing_contact().clone()),
163 );
164 match gateway.sale(request).await {
165 Ok(outcome) => apply_subscription_recovery_gateway_outcome(
166 pool,
167 coordinator,
168 &admission.reservation,
169 &outcome,
170 )
171 .await
172 .map(SubscriptionRecoveryProviderResult::Payment),
173 Err(GatewayMutationError::NotSubmitted(error)) => {
174 let evidence = mutation_error_evidence(error.detail());
175 let policy = GatewayNotSubmittedPolicy::for_error(&error);
176 let application = apply_resumable_not_submitted_policy(
177 pool,
178 OutcomeReservation::Recovery(&admission.reservation),
179 &evidence,
180 policy,
181 )
182 .await?;
183 if application.should_surface_not_submitted(policy) {
184 Ok(SubscriptionRecoveryProviderResult::NotSubmitted {
185 payment: application.payment,
186 error,
187 })
188 } else {
189 Ok(SubscriptionRecoveryProviderResult::Payment(
190 application.payment,
191 ))
192 }
193 }
194 Err(GatewayMutationError::RateLimitedIndeterminate(detail)) => {
195 resolve_recovery_unknown_outcome(
196 pool,
197 &admission.reservation,
198 &mutation_error_evidence(&detail),
199 Some(RateLimitCooldown::Provider),
200 )
201 .await
202 .map(SubscriptionRecoveryProviderResult::Payment)
203 }
204 Err(GatewayMutationError::Indeterminate(detail)) => resolve_recovery_unknown_outcome(
205 pool,
206 &admission.reservation,
207 &mutation_error_evidence(&detail),
208 None,
209 )
210 .await
211 .map(SubscriptionRecoveryProviderResult::Payment),
212 }
213}
214
215pub async fn apply_subscription_recovery_gateway_outcome(
218 pool: &PgPool,
219 coordinator: &dyn BillingTransactionCoordinator,
220 reservation: &SubscriptionRecoveryReservation,
221 outcome: &GatewayPaymentOutcome,
222) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
223 apply_subscription_recovery_gateway_decision(pool, coordinator, reservation, outcome)
224 .await
225 .map(|result| result.with_gateway_diagnostics(outcome.diagnostics().to_vec()))
226}
227
228async fn apply_subscription_recovery_gateway_decision(
229 pool: &PgPool,
230 coordinator: &dyn BillingTransactionCoordinator,
231 reservation: &SubscriptionRecoveryReservation,
232 outcome: &GatewayPaymentOutcome,
233) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
234 match outcome.status() {
235 GatewayPaymentStatus::Approved => {
236 let approved_evidence = outcome.approved_evidence().ok_or(
237 SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE),
238 )?;
239 if outcome.transaction_id().is_none() || outcome.payment_method_reference().is_none() {
240 return park_recovery_approved_outcome(
241 pool,
242 reservation,
243 &approved_evidence,
244 RECOVERY_INCOMPLETE_APPROVAL_TEXT,
245 )
246 .await;
247 }
248 for attempt_index in 0..APPROVED_APPLICATION_ATTEMPTS {
249 match apply_recovery_approved_outcome(coordinator, reservation, &approved_evidence)
250 .await
251 {
252 Ok(result) => return Ok(result),
253 Err(_) if attempt_index + 1 < APPROVED_APPLICATION_ATTEMPTS => {
254 tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
255 }
256 Err(_) => break,
257 }
258 }
259 park_recovery_approved_outcome(
260 pool,
261 reservation,
262 &approved_evidence,
263 RECOVERY_APPROVED_STORAGE_FAILURE_TEXT,
264 )
265 .await
266 }
267 GatewayPaymentStatus::Declined => {
268 resolve_recovery_non_approved_outcome(
269 pool,
270 reservation,
271 outcome.evidence(),
272 AttemptResolutionStatus::Declined,
273 None,
274 None,
275 OutcomeResolutionBoundary::Submitted,
276 )
277 .await
278 }
279 GatewayPaymentStatus::Failed => {
280 resolve_recovery_non_approved_outcome(
281 pool,
282 reservation,
283 outcome.evidence(),
284 AttemptResolutionStatus::Failed,
285 None,
286 None,
287 OutcomeResolutionBoundary::Submitted,
288 )
289 .await
290 }
291 GatewayPaymentStatus::Unknown => {
292 resolve_recovery_unknown_outcome(pool, reservation, outcome.evidence(), None).await
293 }
294 }
295}
296
297pub async fn apply_reconciled_subscription_recovery_gateway_outcome(
300 pool: &PgPool,
301 coordinator: &dyn BillingTransactionCoordinator,
302 billing_scope_id: BillingScopeId,
303 attempt_id: PaymentAttemptId,
304 outcome: &GatewayPaymentOutcome,
305) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
306 let mut transaction = pool.begin().await?;
307 let attempt = crate::find_payment_attempt_by_id_in_transaction(
308 &mut transaction,
309 billing_scope_id,
310 attempt_id,
311 )
312 .await?
313 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
314 "subscription recovery attempt was not found",
315 ))?;
316 let provider_key = sqlx::query_scalar::<_, String>(
317 "SELECT provider_key FROM billing_gateway_accounts WHERE billing_scope_id = $1 AND id = $2",
318 )
319 .bind(billing_scope_id.as_uuid())
320 .bind(attempt.identity().gateway_account_id().as_uuid())
321 .fetch_optional(&mut *transaction)
322 .await?
323 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
324 "subscription recovery gateway account was not found",
325 ))?;
326 transaction.commit().await?;
327 let provider_key = GatewayProviderKey::new(provider_key).map_err(|_| {
328 SubscriptionEnrollmentApplicationError::InvalidState(
329 "subscription recovery gateway provider key is invalid",
330 )
331 })?;
332 let reservation = SubscriptionRecoveryReservation::from_attempt(&attempt, provider_key)
333 .map_err(|_| {
334 SubscriptionEnrollmentApplicationError::InvalidState(
335 "reconciled attempt is not a valid subscription recovery",
336 )
337 })?;
338 apply_subscription_recovery_gateway_outcome(pool, coordinator, &reservation, outcome).await
339}
340
341async fn apply_recovery_approved_outcome(
342 coordinator: &dyn BillingTransactionCoordinator,
343 reservation: &SubscriptionRecoveryReservation,
344 approved_evidence: &ApprovedProcessorEvidence,
345) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
346 let identity = reservation.identity();
347 let mut transaction = coordinator
348 .begin(
349 BillingEventSubject::new(identity.billing_scope_id(), identity.subscriber_id()),
350 BILLING_LOCK_TIMEOUT,
351 )
352 .await?;
353 let subject_state = transaction.subject_state();
354 let application = apply_recovery_approved_on_connection(
355 transaction.connection(),
356 subject_state,
357 reservation,
358 approved_evidence,
359 )
360 .await;
361 finalize_approved_application(transaction, application).await
362}
363
364async fn apply_recovery_approved_on_connection(
365 connection: &mut PgConnection,
366 subject_state: BillingTransactionSubjectState,
367 reservation: &SubscriptionRecoveryReservation,
368 approved_evidence: &ApprovedProcessorEvidence,
369) -> Result<
370 (SubscriptionEnrollmentPaymentResult, Option<BillingEvent>),
371 SubscriptionEnrollmentApplicationError,
372> {
373 let evidence = approved_evidence.evidence();
374 set_application_timeouts(connection).await?;
375 let identity = reservation.identity();
376 lock_payment_method_domain(
377 connection,
378 identity.subscriber_id(),
379 identity.gateway_account_id().as_uuid(),
380 )
381 .await?;
382 lock_subscription_aggregate(connection, identity.subscriber_id(), reservation.plan_key())
383 .await?;
384 let attempt =
385 lock_expected_reservation_attempt(connection, OutcomeReservation::Recovery(reservation))
386 .await?;
387
388 if attempt.status() == PaymentAttemptStatus::Approved {
389 let subscription = load_applied_subscription(connection, &attempt)
390 .await?
391 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
392 INVALID_APPLICATION_STATE,
393 ))?;
394 observe_processor_charge(
395 connection,
396 &attempt,
397 evidence,
398 ProcessorChargeProgression::Applied,
399 )
400 .await?;
401 return Ok((
402 SubscriptionEnrollmentPaymentResult::applied(attempt, subscription)?,
403 None,
404 ));
405 }
406 if subject_state != BillingTransactionSubjectState::LiveRecipient {
407 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
408 "a subscription recovery event requires a live recipient",
409 ));
410 }
411 if attempt.status().is_terminal() {
412 let observation = observe_processor_charge(
413 connection,
414 &attempt,
415 evidence,
416 ProcessorChargeProgression::ExternalReversalRequired,
417 )
418 .await?;
419 if let ObservedCharge::Owned(charge) = observation {
420 transition_charge(
421 connection,
422 charge.id,
423 ProcessorChargeProgression::ExternalReversalRequired,
424 None,
425 )
426 .await?;
427 }
428 return Ok((
429 SubscriptionEnrollmentPaymentResult::confirmation_pending(
430 attempt,
431 approved_evidence.clone(),
432 )?,
433 None,
434 ));
435 }
436
437 let observation = observe_processor_charge(
438 connection,
439 &attempt,
440 evidence,
441 ProcessorChargeProgression::Pending,
442 )
443 .await?;
444 let ObservedCharge::Owned(charge) = observation else {
445 let parked = park_locked_attempt(
446 connection,
447 &attempt,
448 evidence,
449 None,
450 "The approved gateway transaction is already owned by another payment attempt.",
451 )
452 .await?;
453 return Ok((
454 SubscriptionEnrollmentPaymentResult::not_applied(parked)?,
455 None,
456 ));
457 };
458 if charge.role == ProcessorChargeRole::Additional {
459 transition_charge(
460 connection,
461 charge.id,
462 ProcessorChargeProgression::ExternalReversalRequired,
463 None,
464 )
465 .await?;
466 let parked = park_locked_attempt(
467 connection,
468 &attempt,
469 evidence,
470 None,
471 "An additional approved charge requires manual reversal review.",
472 )
473 .await?;
474 return Ok((
475 SubscriptionEnrollmentPaymentResult::not_applied(parked)?,
476 None,
477 ));
478 }
479 if !recovery_subscription_matches(connection, reservation).await? {
480 transition_charge(
481 connection,
482 charge.id,
483 ProcessorChargeProgression::ExternalReversalRequired,
484 Some(PaymentResolutionCode::SubscriptionApprovedRecoveryStaleState),
485 )
486 .await?;
487 let parked = park_locked_attempt(
488 connection,
489 &attempt,
490 evidence,
491 Some(PaymentResolutionCode::SubscriptionApprovedRecoveryStaleState),
492 RECOVERY_STALE_STATE_TEXT,
493 )
494 .await?;
495 return Ok((
496 SubscriptionEnrollmentPaymentResult::not_applied(parked)?,
497 None,
498 ));
499 }
500
501 let transaction_id =
502 evidence
503 .transaction_id()
504 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
505 INVALID_APPLICATION_STATE,
506 ))?;
507 let old_method_id = reservation.expected_state().payment_method_id();
508 let method_id = upsert_payment_method(connection, &attempt, evidence).await?;
509 let updated = sqlx::query(
510 r#"
511 UPDATE billing_subscriptions
512 SET status = 'active', phase = 'recurring', payment_method_id = $2,
513 current_period_start_at = $3, current_period_end_at = $4,
514 next_renewal_at = $4, next_payment_attempt_at = $4,
515 initial_transaction_id = $5,
516 updated_at = clock_timestamp()
517 WHERE id = $1 AND billing_scope_id = $6 AND subscriber_id = $7
518 AND gateway_account_id = $8 AND plan_key = $9
519 -- Reservation of new v2 recoveries requires past_due. Application
520 -- also honors exact active authority snapshotted by durable v1
521 -- attempts that survive the maintenance cutover.
522 AND status = $10 AND status IN ('active', 'past_due')
523 AND payment_method_id = $11 AND initial_transaction_id = $12
524 AND next_renewal_at = $3
525 "#,
526 )
527 .bind(reservation.subscription_id().as_uuid())
528 .bind(method_id.as_uuid())
529 .bind(reservation.period().start_at())
530 .bind(reservation.period().end_at())
531 .bind(transaction_id.expose())
532 .bind(identity.billing_scope_id().as_uuid())
533 .bind(identity.subscriber_id().as_uuid())
534 .bind(identity.gateway_account_id().as_uuid())
535 .bind(reservation.plan_key().as_str())
536 .bind(reservation.expected_state().status().as_str())
537 .bind(old_method_id.as_uuid())
538 .bind(
539 reservation
540 .expected_state()
541 .initial_transaction_id()
542 .expose(),
543 )
544 .execute(&mut *connection)
545 .await?;
546 if updated.rows_affected() != 1 {
547 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
548 INVALID_APPLICATION_STATE,
549 ));
550 }
551 advance_subscription_discount_after_successful_charge(
552 connection,
553 reservation.subscription_id(),
554 reservation.plan_key(),
555 )
556 .await?;
557 disable_payment_method_if_unreferenced(connection, old_method_id).await?;
558 mark_attempt_approved(
559 connection,
560 &attempt,
561 evidence,
562 reservation.subscription_id(),
563 method_id,
564 )
565 .await?;
566 transition_charge(
567 connection,
568 charge.id,
569 ProcessorChargeProgression::Applied,
570 None,
571 )
572 .await?;
573
574 let attempt = find_payment_attempt_by_id_on_connection(
575 connection,
576 identity.billing_scope_id(),
577 identity.attempt_id(),
578 )
579 .await?
580 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
581 INVALID_APPLICATION_STATE,
582 ))?;
583 let subscription = load_subscription(
584 connection,
585 identity.billing_scope_id(),
586 reservation.subscription_id(),
587 )
588 .await?
589 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
590 INVALID_APPLICATION_STATE,
591 ))?;
592 let event = BillingEvent::SubscriptionRenewed {
593 attempt_id: identity.attempt_id(),
594 subscription_id: reservation.subscription_id(),
595 plan_key: reservation.plan_key().clone(),
596 charge: syrup_rail::ChargeAmount::try_from(attempt.request().amount()).map_err(|_| {
597 SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
598 })?,
599 period: reservation.period().clone(),
600 };
601 Ok((
602 SubscriptionEnrollmentPaymentResult::applied(attempt, subscription)?,
603 Some(event),
604 ))
605}
606
607pub(crate) async fn resolve_recovery_non_approved_outcome(
608 pool: &PgPool,
609 reservation: &SubscriptionRecoveryReservation,
610 evidence: &ProcessorEvidence,
611 status: AttemptResolutionStatus,
612 resolution_code: Option<PaymentResolutionCode>,
613 cooldown: Option<RateLimitCooldown>,
614 boundary: OutcomeResolutionBoundary,
615) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
616 resolve_pool_outcome(
617 pool,
618 OutcomeReservation::Recovery(reservation),
619 evidence,
620 OutcomeResolutionCommand::non_approved(status, resolution_code, cooldown, boundary),
621 )
622 .await
623 .map(OutcomeApplication::into_payment)
624}
625
626async fn resolve_recovery_unknown_outcome(
627 pool: &PgPool,
628 reservation: &SubscriptionRecoveryReservation,
629 evidence: &ProcessorEvidence,
630 cooldown: Option<RateLimitCooldown>,
631) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
632 resolve_pool_outcome(
633 pool,
634 OutcomeReservation::Recovery(reservation),
635 evidence,
636 OutcomeResolutionCommand::unknown(cooldown),
637 )
638 .await
639 .map(OutcomeApplication::into_payment)
640}
641
642async fn park_recovery_approved_outcome(
643 pool: &PgPool,
644 reservation: &SubscriptionRecoveryReservation,
645 approved_evidence: &ApprovedProcessorEvidence,
646 message: &'static str,
647) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
648 let evidence = approved_evidence.evidence();
649 match try_park_recovery_approved_outcome(pool, reservation, evidence, message).await {
650 Ok(result) => Ok(result),
651 Err(_) => {
652 observe_recovery_approved_evidence_with_retry(pool, reservation, evidence).await?;
653 let mut transaction = pool.begin().await?;
654 let attempt = find_payment_attempt_by_id_on_connection(
655 &mut transaction,
656 reservation.identity().billing_scope_id(),
657 reservation.identity().attempt_id(),
658 )
659 .await?
660 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
661 INVALID_APPLICATION_STATE,
662 ))?;
663 let result = if attempt.status() == PaymentAttemptStatus::Approved {
664 payment_result_for_attempt(&mut transaction, attempt).await?
665 } else {
666 SubscriptionEnrollmentPaymentResult::confirmation_pending(
667 attempt,
668 approved_evidence.clone(),
669 )?
670 };
671 transaction.commit().await?;
672 Ok(result)
673 }
674 }
675}
676
677async fn try_park_recovery_approved_outcome(
678 pool: &PgPool,
679 reservation: &SubscriptionRecoveryReservation,
680 evidence: &ProcessorEvidence,
681 message: &'static str,
682) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
683 let mut transaction = pool.begin().await?;
684 set_application_timeouts(&mut transaction).await?;
685 lock_subscription_aggregate(
686 &mut transaction,
687 reservation.identity().subscriber_id(),
688 reservation.plan_key(),
689 )
690 .await?;
691 let attempt = lock_expected_reservation_attempt(
692 &mut transaction,
693 OutcomeReservation::Recovery(reservation),
694 )
695 .await?;
696 let attempt = if attempt.status() == PaymentAttemptStatus::Approved {
697 observe_processor_charge(
698 &mut transaction,
699 &attempt,
700 evidence,
701 ProcessorChargeProgression::Applied,
702 )
703 .await?;
704 attempt
705 } else if attempt.status().is_terminal() {
706 let progression =
707 if evidence.transaction_id().is_some() && attempt.request().amount().cents() > 0 {
708 ProcessorChargeProgression::ExternalReversalRequired
709 } else {
710 ProcessorChargeProgression::ReconciliationRequired
711 };
712 observe_processor_charge(&mut transaction, &attempt, evidence, progression).await?;
713 attempt
714 } else {
715 observe_processor_charge(
716 &mut transaction,
717 &attempt,
718 evidence,
719 ProcessorChargeProgression::Pending,
720 )
721 .await?;
722 park_locked_attempt(&mut transaction, &attempt, evidence, None, message).await?
723 };
724 let result = payment_result_for_attempt(&mut transaction, attempt).await?;
725 transaction.commit().await?;
726 Ok(result)
727}
728
729async fn observe_recovery_approved_evidence_with_retry(
730 pool: &PgPool,
731 reservation: &SubscriptionRecoveryReservation,
732 evidence: &ProcessorEvidence,
733) -> Result<(), SubscriptionEnrollmentApplicationError> {
734 for attempt_index in 0..APPROVED_EVIDENCE_WRITE_ATTEMPTS {
735 let result = async {
736 let mut transaction = pool.begin().await?;
737 set_application_timeouts(&mut transaction).await?;
738 let attempt = lock_expected_reservation_attempt(
739 &mut transaction,
740 OutcomeReservation::Recovery(reservation),
741 )
742 .await?;
743 observe_processor_charge(
744 &mut transaction,
745 &attempt,
746 evidence,
747 ProcessorChargeProgression::Pending,
748 )
749 .await?;
750 transaction.commit().await?;
751 Ok::<(), SubscriptionEnrollmentApplicationError>(())
752 }
753 .await;
754 match result {
755 Ok(()) => return Ok(()),
756 Err(error)
757 if is_retryable_evidence_error(&error)
758 && attempt_index + 1 < APPROVED_EVIDENCE_WRITE_ATTEMPTS =>
759 {
760 tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
761 }
762 Err(error) if is_retryable_evidence_error(&error) => break,
763 Err(error) => return Err(error),
764 }
765 }
766 persist_approved_evidence_without_attempt_lock(
767 pool,
768 LockFreeApprovedEvidenceTerms::recovery(reservation),
769 evidence,
770 )
771 .await
772}