syrup_rail_postgres/subscription_billing_service/
recovery.rs

1use super::*;
2
3impl SubscriptionBillingService {
4    /// Runs one complete subscriber-initiated recovery payment boundary.
5    ///
6    /// The command carries only the owner, requested plan/configuration, and
7    /// memory-only token/contact. Reservation derives the exact due period,
8    /// amount, subscription, and payment-state snapshot under lock.
9    pub async fn recover(
10        &self,
11        command: RecoverSubscriptionPayment,
12    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionBillingServiceError> {
13        let prepared_attempt = match self.preflight_recovery(&command).await? {
14            SubscriptionRecoveryPreflightOutcome::Continue => None,
15            SubscriptionRecoveryPreflightOutcome::Replay(attempt)
16                if attempt_is_prepared(&attempt) =>
17            {
18                Some(*attempt)
19            }
20            SubscriptionRecoveryPreflightOutcome::Replay(attempt) => {
21                return self.payment_result(*attempt).await;
22            }
23            SubscriptionRecoveryPreflightOutcome::IdempotencyConflict => {
24                return Err(SubscriptionBillingServiceError::IdempotencyConflict);
25            }
26        };
27        if prepared_attempt.as_ref().is_some_and(|attempt| {
28            attempt.identity().required_gateway_account_mode() != self.required_gateway_account_mode
29        }) {
30            return Err(SubscriptionBillingServiceError::GatewayConfigurationChanged);
31        }
32
33        self.admit_subscriber_mutation(
34            command.billing_scope_id(),
35            command.subscriber_id(),
36            EndUserMutationOperation::SubscriptionRecovery,
37        )
38        .await?;
39
40        let (account, gateway) = self
41            .resolve_active_gateway(
42                command.billing_scope_id(),
43                command.gateway_configuration_id(),
44            )
45            .await?;
46
47        let (reservation, attempt) = if let Some(attempt) = prepared_attempt {
48            recovery_reservation_from_prepared_attempt(attempt, &gateway)?
49        } else {
50            match self.reserve_recovery(&command, &gateway).await? {
51                SubscriptionRecoveryReservationOutcome::Reserved(reservation, attempt) => {
52                    (*reservation, *attempt)
53                }
54                SubscriptionRecoveryReservationOutcome::Replay(attempt)
55                    if attempt_is_prepared(&attempt) =>
56                {
57                    recovery_reservation_from_prepared_attempt(*attempt, &gateway)?
58                }
59                SubscriptionRecoveryReservationOutcome::Replay(attempt) => {
60                    return self.payment_result(*attempt).await;
61                }
62                SubscriptionRecoveryReservationOutcome::IdempotencyConflict => {
63                    return Err(SubscriptionBillingServiceError::IdempotencyConflict);
64                }
65                SubscriptionRecoveryReservationOutcome::Rejected(
66                    SubscriptionRecoveryReservationRejection::GatewayAccountModeChanged,
67                ) => {
68                    return Err(SubscriptionBillingServiceError::GatewayConfigurationChanged);
69                }
70                SubscriptionRecoveryReservationOutcome::Rejected(reason) => {
71                    return Err(
72                        SubscriptionBillingServiceError::RecoveryReservationRejected(reason),
73                    );
74                }
75            }
76        };
77        if attempt.status() != PaymentAttemptStatus::Pending
78            || attempt.state().timestamps().submitted_at().is_some()
79            || attempt.identity() != reservation.identity()
80        {
81            return Err(SubscriptionBillingServiceError::InvalidState(
82                INVALID_SERVICE_STATE,
83            ));
84        }
85        if reservation.identity().required_gateway_account_mode()
86            != self.required_gateway_account_mode
87        {
88            return Err(SubscriptionBillingServiceError::GatewayConfigurationChanged);
89        }
90
91        if let Some(scope) = self.active_cooldown(&account).await? {
92            return self
93                .resolve_subscriber_readiness_failure(
94                    SubscriberInitiatedReservation::Recovery(&reservation),
95                    SubscriberReadinessFailure::Cooldown(scope),
96                    OutcomeResolutionBoundary::Prepared,
97                )
98                .await;
99        }
100        let verified_gateway = match subscriber_gateway_readiness(
101            &gateway,
102            self.required_gateway_account_mode,
103        )
104        .await
105        {
106            Ok(verified_gateway) => verified_gateway,
107            Err(failure) => {
108                return self
109                    .resolve_subscriber_readiness_failure(
110                        SubscriberInitiatedReservation::Recovery(&reservation),
111                        failure,
112                        OutcomeResolutionBoundary::Prepared,
113                    )
114                    .await;
115            }
116        };
117
118        let admission =
119            match admit_subscription_recovery_submission(&self.pool, &reservation).await? {
120                SubscriptionRecoveryAdmissionOutcome::Admitted(admission) => *admission,
121                SubscriptionRecoveryAdmissionOutcome::AlreadyAdmitted(attempt) => {
122                    return self.payment_result(attempt).await;
123                }
124                SubscriptionRecoveryAdmissionOutcome::Rejected { attempt, .. } => {
125                    return self.payment_result(attempt).await;
126                }
127            };
128        // Admission can race with a cooldown observed by another request;
129        // recheck before the capability performs provider I/O.
130        if let Some(scope) = self.active_cooldown(&account).await? {
131            return self
132                .resolve_subscriber_readiness_failure(
133                    SubscriberInitiatedReservation::Recovery(&reservation),
134                    SubscriberReadinessFailure::Cooldown(scope),
135                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
136                )
137                .await;
138        }
139        match submit_admitted_subscription_recovery(
140            &self.pool,
141            self.coordinator.as_ref(),
142            admission,
143            &command,
144            verified_gateway,
145        )
146        .await?
147        {
148            SubscriptionRecoveryProviderResult::Payment(payment) => Ok(payment),
149            SubscriptionRecoveryProviderResult::NotSubmitted { error, .. } => {
150                Err(SubscriptionBillingServiceError::GatewayNotSubmitted(error))
151            }
152        }
153    }
154
155    pub(super) async fn preflight_recovery(
156        &self,
157        command: &RecoverSubscriptionPayment,
158    ) -> Result<SubscriptionRecoveryPreflightOutcome, SubscriptionBillingServiceError> {
159        let mut transaction = self.pool.begin().await?;
160        let outcome =
161            preflight_subscription_recovery_in_transaction(&mut transaction, command).await?;
162        transaction.commit().await?;
163        Ok(outcome)
164    }
165
166    pub(super) async fn reserve_recovery(
167        &self,
168        command: &RecoverSubscriptionPayment,
169        gateway: &syrup_rail::ResolvedGateway,
170    ) -> Result<SubscriptionRecoveryReservationOutcome, SubscriptionBillingServiceError> {
171        let mut transaction = self.pool.begin().await?;
172        let outcome = reserve_subscription_recovery_in_transaction(
173            &mut transaction,
174            command,
175            gateway,
176            self.required_gateway_account_mode,
177        )
178        .await?;
179        transaction.commit().await?;
180        Ok(outcome)
181    }
182}
183
184fn recovery_reservation_from_prepared_attempt(
185    attempt: PaymentAttempt,
186    gateway: &syrup_rail::ResolvedGateway,
187) -> Result<(SubscriptionRecoveryReservation, PaymentAttempt), SubscriptionBillingServiceError> {
188    if !attempt_is_prepared(&attempt) {
189        return Err(SubscriptionBillingServiceError::InvalidState(
190            INVALID_SERVICE_STATE,
191        ));
192    }
193    if !resolved_gateway_matches_attempt(gateway, &attempt) {
194        return Err(SubscriptionBillingServiceError::GatewayConfigurationChanged);
195    }
196    let reservation =
197        SubscriptionRecoveryReservation::from_attempt(&attempt, gateway.provider_key().clone())
198            .map_err(|_| SubscriptionBillingServiceError::InvalidState(INVALID_SERVICE_STATE))?;
199    Ok((reservation, attempt))
200}
syrup_rail_postgres/subscription_billing_service/recovery.rs

syrup_rail_postgres/subscription_billing_service/
recovery.rs
