syrup_rail_postgres/enrollment_application/
renewal.rs

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, SubscriptionEnrollmentPaymentResult, SubscriptionRenewalReservation,
10    SubscriptionRenewalSubmissionOutcome, SubscriptionRenewalSubmissionRejection,
11};
12
13use crate::{
14    BillingTransactionCoordinator, BillingTransactionSubjectState,
15    attempts::find_payment_attempt_by_id_on_connection,
16    processor_charges::{ObservedCharge, observe_processor_charge, transition_charge},
17    renewal_failure::{RenewalFailureApplication, apply_resolved_automatic_renewal_failure},
18};
19
20use super::{
21    APPROVED_APPLICATION_ATTEMPTS, APPROVED_EVIDENCE_RETRY_DELAY, APPROVED_EVIDENCE_WRITE_ATTEMPTS,
22    AttemptResolutionStatus, AttemptTransition, BILLING_LOCK_TIMEOUT, GatewayNotSubmittedPolicy,
23    INVALID_APPLICATION_STATE, OutcomeApplication, OutcomeReservation, OutcomeResolutionBoundary,
24    OutcomeResolutionCommand, PreparedAttemptReplay, RENEWAL_APPROVED_STORAGE_FAILURE_TEXT,
25    RENEWAL_INCOMPLETE_APPROVAL_TEXT, RENEWAL_STALE_STATE_TEXT, RateLimitCooldown,
26    SubscriptionEnrollmentApplicationError, advance_subscription_discount_after_successful_charge,
27    clear_resolved_attempt_submission, commit_rate_limit_cooldown, finalize_approved_application,
28    is_retryable_evidence_error, load_applied_subscription, load_subscription,
29    lock_expected_reservation_attempt, lock_payment_method_domain, lock_subscription_aggregate,
30    map_attempt_transition_error, mark_attempt_approved, mutation_error_evidence,
31    park_locked_attempt, payment_result_for_attempt, payment_result_for_reservation_attempt,
32    persist_attempt_transition, renewal_subscription_matches, resolve_pool_outcome,
33    set_application_timeouts,
34};
35
36/// One committed final-admission result authorizing exactly one immediate
37/// automatic recurring charge.
38pub struct AdmittedSubscriptionRenewal {
39    reservation: SubscriptionRenewalReservation,
40    attempt: PaymentAttempt,
41    payment_method_reference: syrup_rail::GatewayPaymentMethodReference,
42}
43
44impl AdmittedSubscriptionRenewal {
45    pub const fn attempt(&self) -> &PaymentAttempt {
46        &self.attempt
47    }
48}
49
50impl fmt::Debug for AdmittedSubscriptionRenewal {
51    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
52        formatter
53            .debug_struct("AdmittedSubscriptionRenewal")
54            .field("attempt", &self.attempt)
55            .field("has_submission_authority", &true)
56            .field("has_payment_method_reference", &true)
57            .finish()
58    }
59}
60
61#[derive(Debug)]
62pub enum SubscriptionRenewalAdmissionOutcome {
63    Admitted(Box<AdmittedSubscriptionRenewal>),
64    AlreadyAdmitted(PaymentAttempt),
65    Rejected {
66        attempt: PaymentAttempt,
67        reason: SubscriptionRenewalSubmissionRejection,
68    },
69}
70
71#[derive(Debug)]
72pub enum SubscriptionRenewalProviderResult {
73    Payment(SubscriptionEnrollmentPaymentResult),
74    /// The provider mutation was not contacted. Renewal has no prepared replay
75    /// path, so its returned payment remains a terminal canonical result. A
76    /// concurrent terminal result is returned as `Payment` instead.
77    NotSubmitted {
78        payment: SubscriptionEnrollmentPaymentResult,
79        error: GatewayNotSubmittedError,
80    },
81}
82
83/// Commits final automatic-renewal admission and captures the exact stored credential.
84pub async fn admit_subscription_renewal_submission(
85    pool: &PgPool,
86    reservation: &SubscriptionRenewalReservation,
87) -> Result<SubscriptionRenewalAdmissionOutcome, SubscriptionEnrollmentApplicationError> {
88    let mut transaction = pool.begin().await?;
89    let outcome =
90        crate::admit_subscription_renewal_submission_in_transaction(&mut transaction, reservation)
91            .await?;
92    let outcome = match outcome {
93        SubscriptionRenewalSubmissionOutcome::Admitted(attempt) => {
94            let identity = reservation.identity();
95            let reference = sqlx::query_scalar::<_, String>(
96                r#"
97                SELECT gateway_payment_method_reference
98                FROM billing_payment_methods
99                WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
100                    AND gateway_account_id = $4 AND status = 'active'
101                FOR SHARE
102                "#,
103            )
104            .bind(reservation.expected_state().payment_method_id().as_uuid())
105            .bind(identity.billing_scope_id().as_uuid())
106            .bind(identity.subscriber_id().as_uuid())
107            .bind(identity.gateway_account_id().as_uuid())
108            .fetch_optional(&mut *transaction)
109            .await?
110            .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
111                INVALID_APPLICATION_STATE,
112            ))?;
113            let payment_method_reference =
114                syrup_rail::GatewayPaymentMethodReference::new(reference).map_err(|_| {
115                    SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
116                })?;
117            SubscriptionRenewalAdmissionOutcome::Admitted(Box::new(AdmittedSubscriptionRenewal {
118                reservation: reservation.clone(),
119                attempt,
120                payment_method_reference,
121            }))
122        }
123        SubscriptionRenewalSubmissionOutcome::AlreadyAdmitted(attempt) => {
124            SubscriptionRenewalAdmissionOutcome::AlreadyAdmitted(attempt)
125        }
126        SubscriptionRenewalSubmissionOutcome::Rejected { attempt, reason } => {
127            SubscriptionRenewalAdmissionOutcome::Rejected { attempt, reason }
128        }
129    };
130    transaction.commit().await?;
131    Ok(outcome)
132}
133
134/// Performs the one recurring sale authorized by committed renewal admission.
135pub async fn submit_admitted_subscription_renewal(
136    pool: &PgPool,
137    coordinator: &dyn BillingTransactionCoordinator,
138    admission: AdmittedSubscriptionRenewal,
139    gateway: crate::ModeVerifiedGateway<'_>,
140) -> Result<SubscriptionRenewalProviderResult, SubscriptionEnrollmentApplicationError> {
141    let identity = admission.reservation.identity();
142    let resolved_gateway = gateway.resolved_gateway();
143    if admission.attempt.identity() != identity
144        || admission.attempt.request() != admission.reservation.request()
145        || admission.attempt.status() != PaymentAttemptStatus::Pending
146        || admission
147            .attempt
148            .state()
149            .timestamps()
150            .submitted_at()
151            .is_none()
152        || resolved_gateway.provider_key() != admission.reservation.provider_key()
153    {
154        return Err(SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch);
155    }
156    let Some(gateway) = gateway.authorize_attempt(&identity) else {
157        return Err(SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch);
158    };
159    let charge =
160        syrup_rail::ChargeAmount::try_from(admission.attempt.request().amount()).map_err(|_| {
161            SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
162        })?;
163    let request = GatewaySaleRequest::new(
164        charge,
165        admission.attempt.request().gateway_order_id().clone(),
166        GatewaySaleIntent::RecurringStoredCredential {
167            payment_method_reference: admission.payment_method_reference,
168            initial_transaction_id: admission
169                .reservation
170                .expected_state()
171                .initial_transaction_id()
172                .clone(),
173        },
174        None,
175    );
176    match gateway.sale(request).await {
177        Ok(outcome) => apply_subscription_renewal_gateway_outcome(
178            pool,
179            coordinator,
180            &admission.reservation,
181            &outcome,
182        )
183        .await
184        .map(SubscriptionRenewalProviderResult::Payment),
185        Err(GatewayMutationError::NotSubmitted(error)) => {
186            let evidence = mutation_error_evidence(error.detail());
187            let policy = GatewayNotSubmittedPolicy::for_error(&error);
188            let application = resolve_renewal_non_approved_outcome(
189                pool,
190                coordinator,
191                &admission.reservation,
192                &evidence,
193                OutcomeResolutionCommand::non_approved(
194                    AttemptResolutionStatus::Failed,
195                    Some(policy.resolution_code()),
196                    policy.cooldown(),
197                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
198                ),
199            )
200            .await?;
201            if application.should_surface_not_submitted(policy) {
202                Ok(SubscriptionRenewalProviderResult::NotSubmitted {
203                    payment: application.payment,
204                    error,
205                })
206            } else {
207                Ok(SubscriptionRenewalProviderResult::Payment(
208                    application.payment,
209                ))
210            }
211        }
212        Err(GatewayMutationError::RateLimitedIndeterminate(detail)) => {
213            resolve_renewal_unknown_outcome(
214                pool,
215                &admission.reservation,
216                &mutation_error_evidence(&detail),
217                Some(RateLimitCooldown::Provider),
218            )
219            .await
220            .map(SubscriptionRenewalProviderResult::Payment)
221        }
222        Err(GatewayMutationError::Indeterminate(detail)) => resolve_renewal_unknown_outcome(
223            pool,
224            &admission.reservation,
225            &mutation_error_evidence(&detail),
226            None,
227        )
228        .await
229        .map(SubscriptionRenewalProviderResult::Payment),
230    }
231}
232
233pub async fn apply_subscription_renewal_gateway_outcome(
234    pool: &PgPool,
235    coordinator: &dyn BillingTransactionCoordinator,
236    reservation: &SubscriptionRenewalReservation,
237    outcome: &GatewayPaymentOutcome,
238) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
239    apply_subscription_renewal_gateway_decision(pool, coordinator, reservation, outcome)
240        .await
241        .map(|result| result.with_gateway_diagnostics(outcome.diagnostics().to_vec()))
242}
243
244async fn apply_subscription_renewal_gateway_decision(
245    pool: &PgPool,
246    coordinator: &dyn BillingTransactionCoordinator,
247    reservation: &SubscriptionRenewalReservation,
248    outcome: &GatewayPaymentOutcome,
249) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
250    match outcome.status() {
251        GatewayPaymentStatus::Approved => {
252            let approved_evidence = outcome.approved_evidence().ok_or(
253                SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE),
254            )?;
255            if outcome.transaction_id().is_none() {
256                return park_renewal_approved_outcome(
257                    pool,
258                    reservation,
259                    &approved_evidence,
260                    RENEWAL_INCOMPLETE_APPROVAL_TEXT,
261                )
262                .await;
263            }
264            for attempt_index in 0..APPROVED_APPLICATION_ATTEMPTS {
265                match apply_renewal_approved_outcome(coordinator, reservation, &approved_evidence)
266                    .await
267                {
268                    Ok(result) => return Ok(result),
269                    Err(_) if attempt_index + 1 < APPROVED_APPLICATION_ATTEMPTS => {
270                        tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
271                    }
272                    Err(_) => break,
273                }
274            }
275            park_renewal_approved_outcome(
276                pool,
277                reservation,
278                &approved_evidence,
279                RENEWAL_APPROVED_STORAGE_FAILURE_TEXT,
280            )
281            .await
282        }
283        GatewayPaymentStatus::Declined => resolve_renewal_non_approved_outcome(
284            pool,
285            coordinator,
286            reservation,
287            outcome.evidence(),
288            OutcomeResolutionCommand::non_approved(
289                AttemptResolutionStatus::Declined,
290                None,
291                None,
292                OutcomeResolutionBoundary::Submitted,
293            ),
294        )
295        .await
296        .map(OutcomeApplication::into_payment),
297        GatewayPaymentStatus::Failed => resolve_renewal_non_approved_outcome(
298            pool,
299            coordinator,
300            reservation,
301            outcome.evidence(),
302            OutcomeResolutionCommand::non_approved(
303                AttemptResolutionStatus::Failed,
304                None,
305                None,
306                OutcomeResolutionBoundary::Submitted,
307            ),
308        )
309        .await
310        .map(OutcomeApplication::into_payment),
311        GatewayPaymentStatus::Unknown => {
312            resolve_renewal_unknown_outcome(pool, reservation, outcome.evidence(), None).await
313        }
314    }
315}
316
317pub async fn apply_reconciled_subscription_renewal_gateway_outcome(
318    pool: &PgPool,
319    coordinator: &dyn BillingTransactionCoordinator,
320    billing_scope_id: BillingScopeId,
321    attempt_id: PaymentAttemptId,
322    outcome: &GatewayPaymentOutcome,
323) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
324    let mut transaction = pool.begin().await?;
325    let attempt = crate::find_payment_attempt_by_id_in_transaction(
326        &mut transaction,
327        billing_scope_id,
328        attempt_id,
329    )
330    .await?
331    .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
332        "subscription renewal attempt was not found",
333    ))?;
334    let provider_key = sqlx::query_scalar::<_, String>(
335        "SELECT provider_key FROM billing_gateway_accounts WHERE billing_scope_id = $1 AND id = $2",
336    )
337    .bind(billing_scope_id.as_uuid())
338    .bind(attempt.identity().gateway_account_id().as_uuid())
339    .fetch_optional(&mut *transaction)
340    .await?
341    .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
342        "subscription renewal gateway account was not found",
343    ))?;
344    transaction.commit().await?;
345    let provider_key = GatewayProviderKey::new(provider_key).map_err(|_| {
346        SubscriptionEnrollmentApplicationError::InvalidState(
347            "subscription renewal gateway provider key is invalid",
348        )
349    })?;
350    let reservation = SubscriptionRenewalReservation::from_attempt(&attempt, provider_key)
351        .map_err(|_| {
352            SubscriptionEnrollmentApplicationError::InvalidState(
353                "reconciled attempt is not a valid subscription renewal",
354            )
355        })?;
356    apply_subscription_renewal_gateway_outcome(pool, coordinator, &reservation, outcome).await
357}
358
359async fn apply_renewal_approved_outcome(
360    coordinator: &dyn BillingTransactionCoordinator,
361    reservation: &SubscriptionRenewalReservation,
362    approved_evidence: &ApprovedProcessorEvidence,
363) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
364    let identity = reservation.identity();
365    let mut transaction = coordinator
366        .begin(
367            BillingEventSubject::new(identity.billing_scope_id(), identity.subscriber_id()),
368            BILLING_LOCK_TIMEOUT,
369        )
370        .await?;
371    let subject_state = transaction.subject_state();
372    let application = apply_renewal_approved_on_connection(
373        transaction.connection(),
374        subject_state,
375        reservation,
376        approved_evidence,
377    )
378    .await;
379    finalize_approved_application(transaction, application).await
380}
381
382async fn apply_renewal_approved_on_connection(
383    connection: &mut PgConnection,
384    subject_state: BillingTransactionSubjectState,
385    reservation: &SubscriptionRenewalReservation,
386    approved_evidence: &ApprovedProcessorEvidence,
387) -> Result<
388    (SubscriptionEnrollmentPaymentResult, Option<BillingEvent>),
389    SubscriptionEnrollmentApplicationError,
390> {
391    let evidence = approved_evidence.evidence();
392    set_application_timeouts(connection).await?;
393    let identity = reservation.identity();
394    lock_payment_method_domain(
395        connection,
396        identity.subscriber_id(),
397        identity.gateway_account_id().as_uuid(),
398    )
399    .await?;
400    lock_subscription_aggregate(connection, identity.subscriber_id(), reservation.plan_key())
401        .await?;
402    let attempt =
403        lock_expected_reservation_attempt(connection, OutcomeReservation::Renewal(reservation))
404            .await?;
405    if attempt.status() == PaymentAttemptStatus::Approved {
406        let subscription = load_applied_subscription(connection, &attempt)
407            .await?
408            .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
409                INVALID_APPLICATION_STATE,
410            ))?;
411        observe_processor_charge(
412            connection,
413            &attempt,
414            evidence,
415            ProcessorChargeProgression::Applied,
416        )
417        .await?;
418        return Ok((
419            SubscriptionEnrollmentPaymentResult::applied(attempt, subscription)?,
420            None,
421        ));
422    }
423    if subject_state != BillingTransactionSubjectState::LiveRecipient {
424        return Err(SubscriptionEnrollmentApplicationError::InvalidState(
425            "a subscription renewal event requires a live recipient",
426        ));
427    }
428    if attempt.status().is_terminal() {
429        let observation = observe_processor_charge(
430            connection,
431            &attempt,
432            evidence,
433            ProcessorChargeProgression::ExternalReversalRequired,
434        )
435        .await?;
436        if let ObservedCharge::Owned(charge) = observation {
437            transition_charge(
438                connection,
439                charge.id,
440                ProcessorChargeProgression::ExternalReversalRequired,
441                None,
442            )
443            .await?;
444        }
445        return Ok((
446            SubscriptionEnrollmentPaymentResult::confirmation_pending(
447                attempt,
448                approved_evidence.clone(),
449            )?,
450            None,
451        ));
452    }
453    let observation = observe_processor_charge(
454        connection,
455        &attempt,
456        evidence,
457        ProcessorChargeProgression::Pending,
458    )
459    .await?;
460    let ObservedCharge::Owned(charge) = observation else {
461        let parked = park_locked_attempt(
462            connection,
463            &attempt,
464            evidence,
465            None,
466            "The approved gateway transaction is already owned by another payment attempt.",
467        )
468        .await?;
469        return Ok((
470            SubscriptionEnrollmentPaymentResult::not_applied(parked)?,
471            None,
472        ));
473    };
474    if charge.role == ProcessorChargeRole::Additional {
475        transition_charge(
476            connection,
477            charge.id,
478            ProcessorChargeProgression::ExternalReversalRequired,
479            None,
480        )
481        .await?;
482        let parked = park_locked_attempt(
483            connection,
484            &attempt,
485            evidence,
486            None,
487            "An additional approved charge requires manual reversal review.",
488        )
489        .await?;
490        return Ok((
491            SubscriptionEnrollmentPaymentResult::not_applied(parked)?,
492            None,
493        ));
494    }
495    if !renewal_subscription_matches(connection, reservation).await? {
496        transition_charge(
497            connection,
498            charge.id,
499            ProcessorChargeProgression::ExternalReversalRequired,
500            Some(PaymentResolutionCode::SubscriptionApprovedRenewalStaleState),
501        )
502        .await?;
503        let parked = park_locked_attempt(
504            connection,
505            &attempt,
506            evidence,
507            Some(PaymentResolutionCode::SubscriptionApprovedRenewalStaleState),
508            RENEWAL_STALE_STATE_TEXT,
509        )
510        .await?;
511        return Ok((
512            SubscriptionEnrollmentPaymentResult::not_applied(parked)?,
513            None,
514        ));
515    }
516    let expected = reservation.expected_state();
517    let updated = sqlx::query(
518        r#"
519        UPDATE billing_subscriptions
520        SET status = 'active', phase = 'recurring', current_period_start_at = $2,
521            current_period_end_at = $3, next_renewal_at = $3,
522            next_payment_attempt_at = $3,
523            updated_at = clock_timestamp()
524        WHERE id = $1 AND billing_scope_id = $4 AND subscriber_id = $5
525            AND gateway_account_id = $6 AND plan_key = $7
526            AND status = $8 AND status IN ('active', 'past_due')
527            AND payment_method_id = $9 AND initial_transaction_id = $10
528            AND amount_cents = $11 AND currency = $12
529            AND next_renewal_at = $2
530        "#,
531    )
532    .bind(reservation.subscription_id().as_uuid())
533    .bind(reservation.period().start_at())
534    .bind(reservation.period().end_at())
535    .bind(identity.billing_scope_id().as_uuid())
536    .bind(identity.subscriber_id().as_uuid())
537    .bind(identity.gateway_account_id().as_uuid())
538    .bind(reservation.plan_key().as_str())
539    .bind(expected.status().as_str())
540    .bind(expected.payment_method_id().as_uuid())
541    .bind(expected.initial_transaction_id().expose())
542    .bind(reservation.request().amount().cents())
543    .bind(reservation.request().amount().currency().as_str())
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    mark_attempt_approved(
558        connection,
559        &attempt,
560        evidence,
561        reservation.subscription_id(),
562        expected.payment_method_id(),
563    )
564    .await?;
565    transition_charge(
566        connection,
567        charge.id,
568        ProcessorChargeProgression::Applied,
569        None,
570    )
571    .await?;
572    let attempt = find_payment_attempt_by_id_on_connection(
573        connection,
574        identity.billing_scope_id(),
575        identity.attempt_id(),
576    )
577    .await?
578    .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
579        INVALID_APPLICATION_STATE,
580    ))?;
581    let subscription = load_subscription(
582        connection,
583        identity.billing_scope_id(),
584        reservation.subscription_id(),
585    )
586    .await?
587    .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
588        INVALID_APPLICATION_STATE,
589    ))?;
590    let event = BillingEvent::SubscriptionRenewed {
591        attempt_id: identity.attempt_id(),
592        subscription_id: reservation.subscription_id(),
593        plan_key: reservation.plan_key().clone(),
594        charge: syrup_rail::ChargeAmount::try_from(attempt.request().amount()).map_err(|_| {
595            SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
596        })?,
597        period: reservation.period().clone(),
598    };
599    Ok((
600        SubscriptionEnrollmentPaymentResult::applied(attempt, subscription)?,
601        Some(event),
602    ))
603}
604
605pub(crate) async fn resolve_renewal_non_approved_outcome(
606    pool: &PgPool,
607    coordinator: &dyn BillingTransactionCoordinator,
608    reservation: &SubscriptionRenewalReservation,
609    evidence: &ProcessorEvidence,
610    resolution: OutcomeResolutionCommand,
611) -> Result<OutcomeApplication, SubscriptionEnrollmentApplicationError> {
612    let renewal = reservation;
613    let reservation = OutcomeReservation::Renewal(renewal);
614    let identity = reservation.identity();
615    if let Some(cooldown) = resolution.cooldown {
616        commit_rate_limit_cooldown(pool, reservation, cooldown).await?;
617    }
618    let mut transaction = coordinator
619        .begin(
620            BillingEventSubject::new(identity.billing_scope_id(), identity.subscriber_id()),
621            BILLING_LOCK_TIMEOUT,
622        )
623        .await?;
624    let result = async {
625        let connection = transaction.connection();
626        set_application_timeouts(connection).await?;
627        lock_subscription_aggregate(connection, identity.subscriber_id(), reservation.plan_key())
628            .await?;
629        let attempt = lock_expected_reservation_attempt(connection, reservation).await?;
630        let may_resolve = resolution.may_resolve(
631            attempt.status(),
632            attempt.state().timestamps().submitted_at().is_some(),
633        );
634        let mut events = Vec::new();
635        // Renewal has no prepared-attempt replay entrypoint. Even when the
636        // shared policy identifies a not-submitted failure as retry-safe for
637        // resumable flows, renewal preserves its established terminal result.
638        if may_resolve {
639            let status = resolution.resolved_status(reservation.operation(), attempt.status());
640            persist_attempt_transition(
641                connection,
642                &attempt,
643                evidence,
644                AttemptTransition::Resolved {
645                    status,
646                    resolution_code: resolution.resolution_code,
647                },
648            )
649            .await
650            .map_err(map_attempt_transition_error)?;
651            if resolution.clears_submitted_at() {
652                clear_resolved_attempt_submission(connection, &attempt).await?;
653            } else if resolution.marks_renewal_past_due(status)
654                && resolution.resolution_code.is_none()
655            {
656                let resolved_attempt = find_payment_attempt_by_id_on_connection(
657                    connection,
658                    identity.billing_scope_id(),
659                    identity.attempt_id(),
660                )
661                .await?
662                .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
663                    INVALID_APPLICATION_STATE,
664                ))?;
665                if let RenewalFailureApplication::Applied {
666                    events: applied_events,
667                    ..
668                } =
669                    apply_resolved_automatic_renewal_failure(connection, &resolved_attempt).await?
670                {
671                    events = applied_events;
672                }
673            }
674        }
675        let result = payment_result_for_reservation_attempt(connection, reservation).await?;
676        Ok::<_, SubscriptionEnrollmentApplicationError>((result, events, may_resolve))
677    }
678    .await;
679    let (result, events, applied) = match result {
680        Ok(result) => result,
681        Err(error) => {
682            let _ = transaction.rollback().await;
683            return Err(error);
684        }
685    };
686    for event in &events {
687        if let Err(error) = transaction.append_event(event).await {
688            let _ = transaction.rollback().await;
689            return Err(error.into());
690        }
691    }
692    transaction.commit().await?;
693    Ok(OutcomeApplication {
694        payment: result,
695        applied,
696        prepared_attempt_replay: PreparedAttemptReplay::Unsupported,
697    })
698}
699
700async fn resolve_renewal_unknown_outcome(
701    pool: &PgPool,
702    reservation: &SubscriptionRenewalReservation,
703    evidence: &ProcessorEvidence,
704    cooldown: Option<RateLimitCooldown>,
705) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
706    resolve_pool_outcome(
707        pool,
708        OutcomeReservation::Renewal(reservation),
709        evidence,
710        OutcomeResolutionCommand::unknown(cooldown),
711    )
712    .await
713    .map(OutcomeApplication::into_payment)
714}
715
716async fn park_renewal_approved_outcome(
717    pool: &PgPool,
718    reservation: &SubscriptionRenewalReservation,
719    approved_evidence: &ApprovedProcessorEvidence,
720    message: &'static str,
721) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
722    let evidence = approved_evidence.evidence();
723    match try_park_renewal_approved_outcome(pool, reservation, evidence, message).await {
724        Ok(result) => Ok(result),
725        Err(_) => {
726            observe_renewal_approved_evidence_with_retry(pool, reservation, evidence).await?;
727            let mut transaction = pool.begin().await?;
728            let attempt = find_payment_attempt_by_id_on_connection(
729                &mut transaction,
730                reservation.identity().billing_scope_id(),
731                reservation.identity().attempt_id(),
732            )
733            .await?
734            .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
735                INVALID_APPLICATION_STATE,
736            ))?;
737            let result = if attempt.status() == PaymentAttemptStatus::Approved {
738                payment_result_for_attempt(&mut transaction, attempt).await?
739            } else {
740                SubscriptionEnrollmentPaymentResult::confirmation_pending(
741                    attempt,
742                    approved_evidence.clone(),
743                )?
744            };
745            transaction.commit().await?;
746            Ok(result)
747        }
748    }
749}
750
751async fn try_park_renewal_approved_outcome(
752    pool: &PgPool,
753    reservation: &SubscriptionRenewalReservation,
754    evidence: &ProcessorEvidence,
755    message: &'static str,
756) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
757    let mut transaction = pool.begin().await?;
758    set_application_timeouts(&mut transaction).await?;
759    lock_subscription_aggregate(
760        &mut transaction,
761        reservation.identity().subscriber_id(),
762        reservation.plan_key(),
763    )
764    .await?;
765    let attempt = lock_expected_reservation_attempt(
766        &mut transaction,
767        OutcomeReservation::Renewal(reservation),
768    )
769    .await?;
770    let attempt = if attempt.status() == PaymentAttemptStatus::Approved {
771        observe_processor_charge(
772            &mut transaction,
773            &attempt,
774            evidence,
775            ProcessorChargeProgression::Applied,
776        )
777        .await?;
778        attempt
779    } else if attempt.status().is_terminal() {
780        let progression =
781            if evidence.transaction_id().is_some() && attempt.request().amount().cents() > 0 {
782                ProcessorChargeProgression::ExternalReversalRequired
783            } else {
784                ProcessorChargeProgression::ReconciliationRequired
785            };
786        observe_processor_charge(&mut transaction, &attempt, evidence, progression).await?;
787        attempt
788    } else {
789        observe_processor_charge(
790            &mut transaction,
791            &attempt,
792            evidence,
793            ProcessorChargeProgression::Pending,
794        )
795        .await?;
796        park_locked_attempt(&mut transaction, &attempt, evidence, None, message).await?
797    };
798    let result = payment_result_for_attempt(&mut transaction, attempt).await?;
799    transaction.commit().await?;
800    Ok(result)
801}
802
803async fn observe_renewal_approved_evidence_with_retry(
804    pool: &PgPool,
805    reservation: &SubscriptionRenewalReservation,
806    evidence: &ProcessorEvidence,
807) -> Result<(), SubscriptionEnrollmentApplicationError> {
808    for attempt_index in 0..APPROVED_EVIDENCE_WRITE_ATTEMPTS {
809        let result = async {
810            let mut transaction = pool.begin().await?;
811            set_application_timeouts(&mut transaction).await?;
812            let attempt = lock_expected_reservation_attempt(
813                &mut transaction,
814                OutcomeReservation::Renewal(reservation),
815            )
816            .await?;
817            observe_processor_charge(
818                &mut transaction,
819                &attempt,
820                evidence,
821                ProcessorChargeProgression::Pending,
822            )
823            .await?;
824            transaction.commit().await?;
825            Ok::<(), SubscriptionEnrollmentApplicationError>(())
826        }
827        .await;
828        match result {
829            Ok(()) => return Ok(()),
830            Err(error)
831                if is_retryable_evidence_error(&error)
832                    && attempt_index + 1 < APPROVED_EVIDENCE_WRITE_ATTEMPTS =>
833            {
834                tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
835            }
836            Err(error) if is_retryable_evidence_error(&error) => break,
837            Err(error) => return Err(error),
838        }
839    }
840    Err(SubscriptionEnrollmentApplicationError::ApprovedEvidenceNotDurable)
841}
syrup_rail_postgres/enrollment_application/renewal.rs

syrup_rail_postgres/enrollment_application/
renewal.rs
