syrup_rail_postgres/subscription_billing_service/
renewal.rs

1use super::*;
2
3impl SubscriptionBillingService {
4    /// Runs one complete automatic recurring-renewal boundary.
5    ///
6    /// Stale, future, canceled, paced, and contended work is a successful
7    /// no-op. A dispatch routed to a service with the wrong durable gateway
8    /// account mode fails with `GatewayConfigurationChanged`; callers must
9    /// route it to the matching service rather than silently discard it. The
10    /// operation never invokes end-user admission or the live offer store and
11    /// never holds a database lock across provider I/O.
12    pub async fn renew(
13        &self,
14        command: ChargeRenewal,
15    ) -> Result<SubscriptionRenewalOutcome, SubscriptionBillingServiceError> {
16        let Some(account) = self.renewal_gateway_account(command).await? else {
17            return Ok(SubscriptionRenewalOutcome::Noop);
18        };
19        if self
20            .active_cooldown(&account.as_gateway_snapshot())
21            .await?
22            .is_some()
23        {
24            return Ok(SubscriptionRenewalOutcome::Noop);
25        }
26        let gateway = self
27            .resolver
28            .resolve(
29                command.billing_scope_id(),
30                account.account_id,
31                account.configuration_id,
32                account.provider_key.clone(),
33            )
34            .await?;
35        if gateway.billing_scope_id() != command.billing_scope_id()
36            || gateway.gateway_account_id() != account.account_id
37            || gateway.gateway_configuration_id() != account.configuration_id
38            || gateway.provider_key() != &account.provider_key
39        {
40            return Err(SubscriptionBillingServiceError::ResolvedGatewayIdentityMismatch);
41        }
42        let verified_gateway =
43            match verify_gateway_account_mode(&gateway, self.required_gateway_account_mode).await {
44                Ok(verified_gateway) => verified_gateway,
45                Err(GatewayAccountModeVerificationError::AccountModeMismatch { .. }) => {
46                    return Err(SubscriptionBillingServiceError::GatewayReadiness(
47                        GatewayError::Configuration(gateway_account_mode_mismatch_detail()),
48                    ));
49                }
50                Err(GatewayAccountModeVerificationError::Gateway(GatewayError::RateLimited(_))) => {
51                    self.extend_provider_cooldown(
52                        command.billing_scope_id(),
53                        account.account_id,
54                        &account.provider_key,
55                    )
56                    .await?;
57                    return Ok(SubscriptionRenewalOutcome::Noop);
58                }
59                Err(GatewayAccountModeVerificationError::Gateway(error)) => {
60                    return Err(SubscriptionBillingServiceError::GatewayReadiness(error));
61                }
62            };
63        let (reservation, attempt) = match self.reserve_renewal(command, &gateway).await? {
64            SubscriptionRenewalReservationOutcome::Reserved(reservation, attempt) => {
65                (*reservation, *attempt)
66            }
67            SubscriptionRenewalReservationOutcome::Rejected(
68                SubscriptionRenewalReservationRejection::PaymentMethodUpdateInProgress,
69            ) => {
70                return Err(SubscriptionBillingServiceError::RenewalReservationRejected(
71                    SubscriptionRenewalReservationRejection::PaymentMethodUpdateInProgress,
72                ));
73            }
74            SubscriptionRenewalReservationOutcome::Rejected(
75                SubscriptionRenewalReservationRejection::GatewayAccountModeChanged
76                | SubscriptionRenewalReservationRejection::GatewayConfigurationChanged,
77            ) => {
78                return Err(SubscriptionBillingServiceError::GatewayConfigurationChanged);
79            }
80            SubscriptionRenewalReservationOutcome::Rejected(_) => {
81                return Ok(SubscriptionRenewalOutcome::Noop);
82            }
83        };
84        if attempt.status() != PaymentAttemptStatus::Pending
85            || attempt.state().timestamps().submitted_at().is_some()
86            || attempt.identity() != reservation.identity()
87        {
88            return Err(SubscriptionBillingServiceError::InvalidState(
89                INVALID_SERVICE_STATE,
90            ));
91        }
92        if let Some(scope) = self.active_cooldown(&account.as_gateway_snapshot()).await? {
93            self.resolve_renewal_cooldown(&reservation, scope, OutcomeResolutionBoundary::Prepared)
94                .await?;
95            return Ok(SubscriptionRenewalOutcome::Noop);
96        }
97        let admission = match admit_subscription_renewal_submission(&self.pool, &reservation).await
98        {
99            Err(error) if is_retryable_renewal_admission_error(&error) => {
100                self.resolve_renewal_readiness_failure(
101                    &reservation,
102                    GatewayDiagnostic::new(
103                        "subscription billing state could not be locked for final admission",
104                    ),
105                    PaymentResolutionCode::SubscriptionRenewalRetryStateChangedBeforeCharge,
106                    OutcomeResolutionBoundary::Prepared,
107                )
108                .await?;
109                return Ok(SubscriptionRenewalOutcome::Noop);
110            }
111            Err(error) => return Err(error.into()),
112            Ok(outcome) => match outcome {
113                SubscriptionRenewalAdmissionOutcome::Admitted(admission) => *admission,
114                SubscriptionRenewalAdmissionOutcome::AlreadyAdmitted(attempt) => {
115                    return self
116                        .payment_result(attempt)
117                        .await
118                        .map(Box::new)
119                        .map(SubscriptionRenewalOutcome::Payment);
120                }
121                SubscriptionRenewalAdmissionOutcome::Rejected { attempt, .. } => {
122                    return self
123                        .payment_result(attempt)
124                        .await
125                        .map(Box::new)
126                        .map(SubscriptionRenewalOutcome::Payment);
127                }
128            },
129        };
130        if let Some(scope) = self.active_cooldown(&account.as_gateway_snapshot()).await? {
131            self.resolve_renewal_cooldown(
132                &reservation,
133                scope,
134                OutcomeResolutionBoundary::AdmittedNotSubmitted,
135            )
136            .await?;
137            return Ok(SubscriptionRenewalOutcome::Noop);
138        }
139        match submit_admitted_subscription_renewal(
140            &self.pool,
141            self.coordinator.as_ref(),
142            admission,
143            verified_gateway,
144        )
145        .await?
146        {
147            SubscriptionRenewalProviderResult::Payment(payment) => {
148                Ok(SubscriptionRenewalOutcome::Payment(Box::new(payment)))
149            }
150            SubscriptionRenewalProviderResult::NotSubmitted { payment, error } => {
151                Ok(SubscriptionRenewalOutcome::NotSubmitted {
152                    payment: Box::new(payment),
153                    error,
154                })
155            }
156        }
157    }
158
159    pub(super) async fn reserve_renewal(
160        &self,
161        command: ChargeRenewal,
162        gateway: &syrup_rail::ResolvedGateway,
163    ) -> Result<SubscriptionRenewalReservationOutcome, SubscriptionBillingServiceError> {
164        let mut transaction = self.pool.begin().await?;
165        let outcome = reserve_subscription_renewal_in_transaction(
166            &mut transaction,
167            command,
168            gateway,
169            self.required_gateway_account_mode,
170        )
171        .await?;
172        transaction.commit().await?;
173        Ok(outcome)
174    }
175
176    pub(super) async fn renewal_gateway_account(
177        &self,
178        command: ChargeRenewal,
179    ) -> Result<Option<RenewalGatewayAccountSnapshot>, SubscriptionBillingServiceError> {
180        let mut transaction = self.pool.begin().await?;
181        let row = sqlx::query_as::<_, (uuid::Uuid, uuid::Uuid, String, String)>(
182            r#"
183            SELECT accounts.id, accounts.gateway_configuration_id, accounts.provider_key,
184                subscriptions.required_gateway_account_mode
185            FROM billing_subscriptions AS subscriptions
186            JOIN billing_gateway_accounts AS accounts
187                ON accounts.billing_scope_id = subscriptions.billing_scope_id
188                AND accounts.id = subscriptions.gateway_account_id
189            WHERE subscriptions.billing_scope_id = $1 AND subscriptions.id = $2
190                AND subscriptions.status IN ('active', 'past_due')
191                AND subscriptions.next_renewal_at = $3
192                AND subscriptions.next_payment_attempt_at <= clock_timestamp()
193            "#,
194        )
195        .bind(command.billing_scope_id().as_uuid())
196        .bind(command.subscription_id().as_uuid())
197        .bind(command.period_start_at())
198        .fetch_optional(&mut *transaction)
199        .await?;
200        let Some((account_id, configuration_id, provider_key, required_mode)) = row else {
201            transaction.commit().await?;
202            return Ok(None);
203        };
204        let required_mode = required_mode
205            .parse::<GatewayAccountMode>()
206            .map_err(|_| SubscriptionBillingServiceError::InvalidState(INVALID_SERVICE_STATE))?;
207        if required_mode != self.required_gateway_account_mode {
208            transaction.commit().await?;
209            return Err(SubscriptionBillingServiceError::GatewayConfigurationChanged);
210        }
211        let attempt_state = crate::renewal_attempt_state(
212            &mut transaction,
213            command.subscription_id(),
214            *command.period_start_at(),
215            None,
216        )
217        .await
218        .map_err(|error| match error {
219            crate::RenewalStoreError::Sql(error) => SubscriptionBillingServiceError::Sql(error),
220            crate::RenewalStoreError::MissingProviderCooldown => {
221                SubscriptionBillingServiceError::InvalidState(INVALID_SERVICE_STATE)
222            }
223            crate::RenewalStoreError::CursorModeMismatch => {
224                // `renewal_attempt_state` is an internal cursor-free lookup;
225                // public caller misuse is returned by the paging API itself.
226                SubscriptionBillingServiceError::InvalidState(INVALID_SERVICE_STATE)
227            }
228        })?;
229        let now = sqlx::query_scalar("SELECT clock_timestamp()")
230            .fetch_one(&mut *transaction)
231            .await?;
232        let payment_method_update_policy =
233            LocalAttemptPolicy::for_kind(PaymentAttemptKind::SubscriptionPaymentMethodUpdate);
234        let has_payment_method_update: bool = sqlx::query_scalar(
235            r#"
236            SELECT EXISTS (
237                SELECT 1 FROM billing_payment_attempts
238                WHERE subscription_id = $1
239                    AND attempt_kind = 'subscription_payment_method_update'
240                    AND status IN ('pending', 'unknown', 'review_required')
241                    AND NOT (
242                        status = ANY($2::text[]) AND submitted_at IS NULL
243                        AND created_at <= clock_timestamp()
244                            - ($3::bigint * interval '1 second')
245                    )
246            )
247            "#,
248        )
249        .bind(command.subscription_id().as_uuid())
250        .bind(LocalAttemptPolicy::expirable_status_values())
251        .bind(payment_method_update_policy.stale_after_seconds())
252        .fetch_one(&mut *transaction)
253        .await?;
254        transaction.commit().await?;
255        if attempt_state.blocks_automatic_retry(now) || has_payment_method_update {
256            if has_payment_method_update {
257                return Err(SubscriptionBillingServiceError::RenewalReservationRejected(
258                    SubscriptionRenewalReservationRejection::PaymentMethodUpdateInProgress,
259                ));
260            }
261            return Ok(None);
262        }
263        Some((account_id, configuration_id, provider_key))
264            .map(|(account_id, configuration_id, provider_key)| {
265                Ok(RenewalGatewayAccountSnapshot {
266                    account_id: GatewayAccountId::new(account_id),
267                    configuration_id: syrup_rail::GatewayConfigurationId::new(configuration_id),
268                    provider_key: GatewayProviderKey::new(provider_key).map_err(|_| {
269                        SubscriptionBillingServiceError::InvalidState(INVALID_SERVICE_STATE)
270                    })?,
271                })
272            })
273            .transpose()
274    }
275
276    pub(super) async fn resolve_renewal_cooldown(
277        &self,
278        reservation: &SubscriptionRenewalReservation,
279        scope: GatewayMutationCooldownScope,
280        boundary: OutcomeResolutionBoundary,
281    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionBillingServiceError> {
282        let (message, code) = match scope {
283            GatewayMutationCooldownScope::Account => (
284                "gateway account mutation cooldown is active",
285                PaymentResolutionCode::GatewayAccountMutationCooldownBeforeSubmission,
286            ),
287            GatewayMutationCooldownScope::Provider => (
288                "gateway provider cooldown is active",
289                PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission,
290            ),
291        };
292        self.resolve_renewal_readiness_failure(
293            reservation,
294            GatewayDiagnostic::new(message),
295            code,
296            boundary,
297        )
298        .await
299    }
300
301    pub(super) async fn extend_provider_cooldown(
302        &self,
303        billing_scope_id: BillingScopeId,
304        gateway_account_id: GatewayAccountId,
305        provider_key: &GatewayProviderKey,
306    ) -> Result<(), SubscriptionBillingServiceError> {
307        let mut transaction = self.pool.begin().await?;
308        set_application_timeouts(&mut transaction).await?;
309        match persist_bound_provider_rate_limit_cooldown(
310            &mut transaction,
311            billing_scope_id,
312            gateway_account_id,
313            provider_key,
314        )
315        .await?
316        {
317            RateLimitCooldownPersistence::Applied => transaction.commit().await?,
318            RateLimitCooldownPersistence::IdentityChanged => {
319                transaction.rollback().await?;
320                tracing::warn!(
321                    target: "syrup_rail::gateway_cooldown",
322                    billing_scope_id = %billing_scope_id.as_uuid(),
323                    gateway_account_id = %gateway_account_id.as_uuid(),
324                    provider_key = provider_key.as_str(),
325                    "skipped pre-reservation provider cooldown after the gateway account identity changed"
326                );
327            }
328            RateLimitCooldownPersistence::MissingProviderCooldown => {
329                transaction.rollback().await?;
330                return Err(SubscriptionBillingServiceError::InvalidState(
331                    INVALID_SERVICE_STATE,
332                ));
333            }
334        }
335        Ok(())
336    }
337
338    pub(super) async fn resolve_renewal_readiness_failure(
339        &self,
340        reservation: &SubscriptionRenewalReservation,
341        detail: GatewayDiagnostic,
342        code: PaymentResolutionCode,
343        boundary: OutcomeResolutionBoundary,
344    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionBillingServiceError> {
345        let condition = (code != PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission)
346            .then(|| GatewayDiagnostic::new("failed"));
347        let evidence = ProcessorEvidence::new(
348            None,
349            None,
350            None,
351            None,
352            Some(detail),
353            condition,
354            GatewayPaymentDescriptor::default(),
355        );
356        resolve_renewal_non_approved_outcome(
357            &self.pool,
358            self.coordinator.as_ref(),
359            reservation,
360            &evidence,
361            OutcomeResolutionCommand::non_approved(
362                AttemptResolutionStatus::Failed,
363                Some(code),
364                None,
365                boundary,
366            ),
367        )
368        .await
369        .map(OutcomeApplication::into_payment)
370        .map_err(SubscriptionBillingServiceError::from)
371    }
372}
syrup_rail_postgres/subscription_billing_service/renewal.rs

syrup_rail_postgres/subscription_billing_service/
renewal.rs
