syrup_rail_postgres/enrollment_application/
initial.rs

1use std::fmt;
2
3use chrono::{DateTime, Utc};
4use sqlx::{PgConnection, PgPool};
5use syrup_rail::{
6    ApprovedProcessorEvidence, BillingEvent, BillingEventSubject, BillingPeriod, BillingScopeId,
7    EnrollSubscription, GatewayMutationError, GatewayNotSubmittedError, GatewayPaymentOutcome,
8    GatewayPaymentStatus, GatewayProviderKey, GatewaySaleIntent, GatewaySaleRequest,
9    GatewayTransactionId, PaymentAttempt, PaymentAttemptId, PaymentAttemptStatus, PaymentMethodId,
10    PaymentResolutionCode, ProcessorChargeProgression, ProcessorChargeRole, ProcessorEvidence,
11    SubscriptionDiscountDuration, SubscriptionDiscountKind, SubscriptionEnrollmentPaymentResult,
12    SubscriptionEnrollmentReservation, SubscriptionEnrollmentSubmissionOutcome,
13    SubscriptionEnrollmentSubmissionRejection, SubscriptionId, SubscriptionPhase,
14    next_billing_period,
15};
16use uuid::Uuid;
17
18use crate::{
19    BillingTransactionCoordinator, BillingTransactionSubjectState, ModeVerifiedGateway,
20    attempts::find_payment_attempt_by_id_on_connection,
21    processor_charges::{
22        LockFreeApprovedEvidenceTerms, ObservedCharge, observe_processor_charge, transition_charge,
23    },
24};
25
26use super::{
27    APPROVED_APPLICATION_ATTEMPTS, APPROVED_EVIDENCE_RETRY_DELAY, APPROVED_EVIDENCE_WRITE_ATTEMPTS,
28    APPROVED_STORAGE_FAILURE_TEXT, AttemptResolutionStatus, BILLING_LOCK_TIMEOUT,
29    CURRENT_GRANT_CONFLICT_TEXT, CURRENT_SUBSCRIPTION_CONFLICT_TEXT, GatewayNotSubmittedPolicy,
30    INCOMPLETE_APPROVAL_TEXT, INVALID_APPLICATION_STATE, OutcomeApplication, OutcomeReservation,
31    OutcomeResolutionBoundary, OutcomeResolutionCommand, RateLimitCooldown,
32    SubscriptionEnrollmentApplicationError, TERMINAL_APPROVAL_RACE_TEXT,
33    apply_resumable_not_submitted_policy, finalize_approved_application,
34    is_retryable_evidence_error, load_applied_subscription, load_subscription,
35    lock_expected_reservation_attempt, lock_payment_method_domain, lock_subscription_aggregate,
36    mark_attempt_approved, mutation_error_evidence, park_locked_attempt,
37    payment_result_for_attempt, persist_approved_evidence_without_attempt_lock,
38    resolve_pool_outcome, set_application_timeouts, upsert_payment_method,
39};
40
41mod approval;
42
43use approval::apply_approved_outcome;
44
45/// One committed final-admission result that authorizes exactly one immediate
46/// provider submission by consuming this value.
47pub struct AdmittedSubscriptionEnrollment {
48    reservation: SubscriptionEnrollmentReservation,
49    attempt: PaymentAttempt,
50}
51
52impl AdmittedSubscriptionEnrollment {
53    pub const fn attempt(&self) -> &PaymentAttempt {
54        &self.attempt
55    }
56}
57
58impl fmt::Debug for AdmittedSubscriptionEnrollment {
59    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
60        formatter
61            .debug_struct("AdmittedSubscriptionEnrollment")
62            .field("attempt", &self.attempt)
63            .field("has_submission_authority", &true)
64            .finish()
65    }
66}
67
68#[derive(Debug)]
69pub enum SubscriptionEnrollmentAdmissionOutcome {
70    Admitted(Box<AdmittedSubscriptionEnrollment>),
71    AlreadyAdmitted(PaymentAttempt),
72    Rejected {
73        attempt: PaymentAttempt,
74        reason: SubscriptionEnrollmentSubmissionRejection,
75    },
76}
77
78#[derive(Debug)]
79pub enum SubscriptionEnrollmentProviderResult {
80    Payment(SubscriptionEnrollmentPaymentResult),
81    /// The provider mutation was not contacted. A retry-safe readiness failure
82    /// can carry the same pending, prepared payment for same-key replay. A
83    /// concurrent terminal result is returned as `Payment` instead.
84    NotSubmitted {
85        payment: SubscriptionEnrollmentPaymentResult,
86        error: GatewayNotSubmittedError,
87    },
88}
89
90impl SubscriptionEnrollmentProviderResult {
91    pub const fn payment(&self) -> &SubscriptionEnrollmentPaymentResult {
92        match self {
93            Self::Payment(payment) | Self::NotSubmitted { payment, .. } => payment,
94        }
95    }
96
97    pub fn into_payment(self) -> SubscriptionEnrollmentPaymentResult {
98        match self {
99            Self::Payment(payment) | Self::NotSubmitted { payment, .. } => payment,
100        }
101    }
102}
103/// Owns and commits final enrollment admission before exposing a one-shot
104/// submission capability. A rolled-back transaction can never yield the
105/// capability consumed by [`submit_admitted_subscription_enrollment`].
106pub async fn admit_subscription_enrollment_submission(
107    pool: &PgPool,
108    offers: &dyn crate::SubscriptionOfferStore,
109    reservation: &SubscriptionEnrollmentReservation,
110) -> Result<SubscriptionEnrollmentAdmissionOutcome, SubscriptionEnrollmentApplicationError> {
111    let mut transaction = pool.begin().await?;
112    let outcome = crate::admit_subscription_enrollment_submission_in_transaction(
113        &mut transaction,
114        offers,
115        reservation,
116    )
117    .await?;
118    transaction.commit().await?;
119    Ok(match outcome {
120        SubscriptionEnrollmentSubmissionOutcome::Admitted(attempt) => {
121            SubscriptionEnrollmentAdmissionOutcome::Admitted(Box::new(
122                AdmittedSubscriptionEnrollment {
123                    reservation: reservation.clone(),
124                    attempt,
125                },
126            ))
127        }
128        SubscriptionEnrollmentSubmissionOutcome::AlreadyAdmitted(attempt) => {
129            SubscriptionEnrollmentAdmissionOutcome::AlreadyAdmitted(attempt)
130        }
131        SubscriptionEnrollmentSubmissionOutcome::Rejected { attempt, reason } => {
132            SubscriptionEnrollmentAdmissionOutcome::Rejected { attempt, reason }
133        }
134    })
135}
136
137/// Performs the one provider sale authorized by a committed final admission,
138/// then applies or durably parks its result.
139///
140/// This function holds no database transaction or lock across provider I/O.
141pub async fn submit_admitted_subscription_enrollment(
142    pool: &PgPool,
143    coordinator: &dyn BillingTransactionCoordinator,
144    admission: AdmittedSubscriptionEnrollment,
145    command: &EnrollSubscription,
146    gateway: ModeVerifiedGateway<'_>,
147) -> Result<SubscriptionEnrollmentProviderResult, SubscriptionEnrollmentApplicationError> {
148    let resolved_gateway = gateway.resolved_gateway();
149    let reconstructed = SubscriptionEnrollmentReservation::from_command_for_attempt(
150        command,
151        resolved_gateway,
152        admission.attempt.identity().attempt_id(),
153        admission
154            .reservation
155            .identity()
156            .required_gateway_account_mode(),
157    )
158    .map_err(|_| SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch)?;
159    if reconstructed != admission.reservation
160        || admission.attempt.identity() != admission.reservation.identity()
161        || admission.attempt.status() != PaymentAttemptStatus::Pending
162        || admission
163            .attempt
164            .state()
165            .timestamps()
166            .submitted_at()
167            .is_none()
168    {
169        return Err(SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch);
170    }
171    let Some(gateway) = gateway.authorize_attempt(&admission.reservation.identity()) else {
172        return Err(SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch);
173    };
174
175    let charge = syrup_rail::ChargeAmount::new(
176        admission.attempt.request().amount().cents(),
177        admission.attempt.request().amount().currency(),
178    )
179    .map_err(|_| SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE))?;
180    let request = GatewaySaleRequest::new(
181        charge,
182        admission.attempt.request().gateway_order_id().clone(),
183        GatewaySaleIntent::InitialStoredCredential {
184            payment_token: command.payment_token().clone(),
185        },
186        Some(command.billing_contact().clone()),
187    );
188    match gateway.sale(request).await {
189        Ok(outcome) => apply_subscription_enrollment_gateway_outcome(
190            pool,
191            coordinator,
192            &admission.reservation,
193            &outcome,
194        )
195        .await
196        .map(SubscriptionEnrollmentProviderResult::Payment),
197        Err(GatewayMutationError::NotSubmitted(error)) => {
198            let evidence = mutation_error_evidence(error.detail());
199            let policy = GatewayNotSubmittedPolicy::for_error(&error);
200            let application = apply_resumable_not_submitted_policy(
201                pool,
202                OutcomeReservation::Initial(&admission.reservation),
203                &evidence,
204                policy,
205            )
206            .await?;
207            if application.should_surface_not_submitted(policy) {
208                Ok(SubscriptionEnrollmentProviderResult::NotSubmitted {
209                    payment: application.payment,
210                    error,
211                })
212            } else {
213                Ok(SubscriptionEnrollmentProviderResult::Payment(
214                    application.payment,
215                ))
216            }
217        }
218        Err(GatewayMutationError::RateLimitedIndeterminate(detail)) => resolve_unknown_outcome(
219            pool,
220            &admission.reservation,
221            &mutation_error_evidence(&detail),
222            Some(RateLimitCooldown::Provider),
223        )
224        .await
225        .map(SubscriptionEnrollmentProviderResult::Payment),
226        Err(GatewayMutationError::Indeterminate(detail)) => resolve_unknown_outcome(
227            pool,
228            &admission.reservation,
229            &mutation_error_evidence(&detail),
230            None,
231        )
232        .await
233        .map(SubscriptionEnrollmentProviderResult::Payment),
234    }
235}
236
237/// Applies one initial-enrollment gateway outcome to the durable billing ledger.
238///
239/// Approved application begins through the host coordinator so its recipient
240/// authorization lock precedes every shared lock. If the atomic application
241/// fails, this operation returns pending confirmation only after either the
242/// review-required attempt or immutable processor charge has committed.
243pub async fn apply_subscription_enrollment_gateway_outcome(
244    pool: &PgPool,
245    coordinator: &dyn BillingTransactionCoordinator,
246    reservation: &SubscriptionEnrollmentReservation,
247    outcome: &GatewayPaymentOutcome,
248) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
249    apply_subscription_enrollment_gateway_decision(pool, coordinator, reservation, outcome)
250        .await
251        .map(|result| result.with_gateway_diagnostics(outcome.diagnostics().to_vec()))
252}
253
254async fn apply_subscription_enrollment_gateway_decision(
255    pool: &PgPool,
256    coordinator: &dyn BillingTransactionCoordinator,
257    reservation: &SubscriptionEnrollmentReservation,
258    outcome: &GatewayPaymentOutcome,
259) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
260    match outcome.status() {
261        GatewayPaymentStatus::Approved => {
262            let approved_evidence = outcome.approved_evidence().ok_or(
263                SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE),
264            )?;
265            if outcome.transaction_id().is_none() || outcome.payment_method_reference().is_none() {
266                return park_approved_outcome(
267                    pool,
268                    reservation,
269                    &approved_evidence,
270                    INCOMPLETE_APPROVAL_TEXT,
271                )
272                .await;
273            }
274
275            for attempt_index in 0..APPROVED_APPLICATION_ATTEMPTS {
276                match apply_approved_outcome(coordinator, reservation, outcome.evidence()).await {
277                    Ok(result) => return Ok(result),
278                    Err(_) if attempt_index + 1 < APPROVED_APPLICATION_ATTEMPTS => {
279                        tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
280                    }
281                    Err(_) => break,
282                }
283            }
284            park_approved_outcome(
285                pool,
286                reservation,
287                &approved_evidence,
288                APPROVED_STORAGE_FAILURE_TEXT,
289            )
290            .await
291        }
292        GatewayPaymentStatus::Declined => {
293            resolve_non_approved_outcome(
294                pool,
295                reservation,
296                outcome.evidence(),
297                AttemptResolutionStatus::Declined,
298                None,
299                None,
300                OutcomeResolutionBoundary::Submitted,
301            )
302            .await
303        }
304        GatewayPaymentStatus::Failed => {
305            resolve_non_approved_outcome(
306                pool,
307                reservation,
308                outcome.evidence(),
309                AttemptResolutionStatus::Failed,
310                None,
311                None,
312                OutcomeResolutionBoundary::Submitted,
313            )
314            .await
315        }
316        GatewayPaymentStatus::Unknown => {
317            resolve_unknown_outcome(pool, reservation, outcome.evidence(), None).await
318        }
319    }
320}
321
322/// Applies an already-observed provider outcome to an exact durable enrollment attempt.
323///
324/// This is the reconciliation counterpart to foreground enrollment. It rebuilds
325/// the secret-free reservation from immutable attempt state and the canonical
326/// gateway account, then enters the same atomic application path. It never
327/// resolves a live gateway or submits another provider mutation.
328pub async fn apply_reconciled_subscription_enrollment_gateway_outcome(
329    pool: &PgPool,
330    coordinator: &dyn BillingTransactionCoordinator,
331    billing_scope_id: BillingScopeId,
332    attempt_id: PaymentAttemptId,
333    outcome: &GatewayPaymentOutcome,
334) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
335    let mut transaction = pool.begin().await?;
336    let attempt = crate::find_payment_attempt_by_id_in_transaction(
337        &mut transaction,
338        billing_scope_id,
339        attempt_id,
340    )
341    .await?
342    .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
343        "subscription enrollment attempt was not found",
344    ))?;
345    let provider_key = sqlx::query_scalar::<_, String>(
346        r#"
347        SELECT provider_key
348        FROM billing_gateway_accounts
349        WHERE billing_scope_id = $1 AND id = $2
350        "#,
351    )
352    .bind(billing_scope_id.as_uuid())
353    .bind(attempt.identity().gateway_account_id().as_uuid())
354    .fetch_optional(&mut *transaction)
355    .await?
356    .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
357        "subscription enrollment gateway account was not found",
358    ))?;
359    transaction.commit().await?;
360
361    let provider_key = GatewayProviderKey::new(provider_key).map_err(|_| {
362        SubscriptionEnrollmentApplicationError::InvalidState(
363            "subscription enrollment gateway provider key is invalid",
364        )
365    })?;
366    let reservation = SubscriptionEnrollmentReservation::from_attempt(&attempt, provider_key)
367        .map_err(|_| {
368            SubscriptionEnrollmentApplicationError::InvalidState(
369                "reconciled attempt is not a valid subscription enrollment",
370            )
371        })?;
372    apply_subscription_enrollment_gateway_outcome(pool, coordinator, &reservation, outcome).await
373}
374
375pub(crate) async fn resolve_non_approved_outcome(
376    pool: &PgPool,
377    reservation: &SubscriptionEnrollmentReservation,
378    evidence: &ProcessorEvidence,
379    status: AttemptResolutionStatus,
380    resolution_code: Option<PaymentResolutionCode>,
381    cooldown: Option<RateLimitCooldown>,
382    boundary: OutcomeResolutionBoundary,
383) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
384    resolve_pool_outcome(
385        pool,
386        OutcomeReservation::Initial(reservation),
387        evidence,
388        OutcomeResolutionCommand::non_approved(status, resolution_code, cooldown, boundary),
389    )
390    .await
391    .map(OutcomeApplication::into_payment)
392}
393
394async fn resolve_unknown_outcome(
395    pool: &PgPool,
396    reservation: &SubscriptionEnrollmentReservation,
397    evidence: &ProcessorEvidence,
398    cooldown: Option<RateLimitCooldown>,
399) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
400    resolve_pool_outcome(
401        pool,
402        OutcomeReservation::Initial(reservation),
403        evidence,
404        OutcomeResolutionCommand::unknown(cooldown),
405    )
406    .await
407    .map(OutcomeApplication::into_payment)
408}
409
410async fn park_approved_outcome(
411    pool: &PgPool,
412    reservation: &SubscriptionEnrollmentReservation,
413    approved_evidence: &ApprovedProcessorEvidence,
414    message: &'static str,
415) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
416    let evidence = approved_evidence.evidence();
417    match try_park_approved_outcome(pool, reservation, evidence, message).await {
418        Ok(result) => Ok(result),
419        Err(_) => {
420            observe_approved_evidence_with_retry(pool, reservation, evidence).await?;
421            let mut transaction = pool.begin().await?;
422            let attempt = find_payment_attempt_by_id_on_connection(
423                &mut transaction,
424                reservation.identity().billing_scope_id(),
425                reservation.identity().attempt_id(),
426            )
427            .await?
428            .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
429                INVALID_APPLICATION_STATE,
430            ))?;
431            let result = if attempt.status() == PaymentAttemptStatus::Approved {
432                payment_result_for_attempt(&mut transaction, attempt).await?
433            } else {
434                SubscriptionEnrollmentPaymentResult::confirmation_pending(
435                    attempt,
436                    approved_evidence.clone(),
437                )?
438            };
439            transaction.commit().await?;
440            Ok(result)
441        }
442    }
443}
444
445async fn try_park_approved_outcome(
446    pool: &PgPool,
447    reservation: &SubscriptionEnrollmentReservation,
448    evidence: &ProcessorEvidence,
449    message: &'static str,
450) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
451    let mut transaction = pool.begin().await?;
452    set_application_timeouts(&mut transaction).await?;
453    lock_subscription_aggregate(
454        &mut transaction,
455        reservation.identity().subscriber_id(),
456        reservation.plan_key(),
457    )
458    .await?;
459    let attempt = lock_expected_reservation_attempt(
460        &mut transaction,
461        OutcomeReservation::Initial(reservation),
462    )
463    .await?;
464    let attempt = if attempt.status() == PaymentAttemptStatus::Approved {
465        observe_processor_charge(
466            &mut transaction,
467            &attempt,
468            evidence,
469            ProcessorChargeProgression::Applied,
470        )
471        .await?;
472        attempt
473    } else if attempt.status().is_terminal() {
474        let progression =
475            if evidence.transaction_id().is_some() && attempt.request().amount().cents() > 0 {
476                ProcessorChargeProgression::ExternalReversalRequired
477            } else {
478                ProcessorChargeProgression::ReconciliationRequired
479            };
480        observe_processor_charge(&mut transaction, &attempt, evidence, progression).await?;
481        attempt
482    } else {
483        observe_processor_charge(
484            &mut transaction,
485            &attempt,
486            evidence,
487            ProcessorChargeProgression::Pending,
488        )
489        .await?;
490        park_locked_attempt(&mut transaction, &attempt, evidence, None, message).await?
491    };
492    let result = payment_result_for_attempt(&mut transaction, attempt).await?;
493    transaction.commit().await?;
494    Ok(result)
495}
496
497async fn observe_approved_evidence_with_retry(
498    pool: &PgPool,
499    reservation: &SubscriptionEnrollmentReservation,
500    evidence: &ProcessorEvidence,
501) -> Result<(), SubscriptionEnrollmentApplicationError> {
502    let mut last_error = None;
503    for attempt_index in 0..APPROVED_EVIDENCE_WRITE_ATTEMPTS {
504        let result = async {
505            let mut transaction = pool.begin().await?;
506            set_application_timeouts(&mut transaction).await?;
507            let attempt = lock_expected_reservation_attempt(
508                &mut transaction,
509                OutcomeReservation::Initial(reservation),
510            )
511            .await?;
512            observe_processor_charge(
513                &mut transaction,
514                &attempt,
515                evidence,
516                ProcessorChargeProgression::Pending,
517            )
518            .await?;
519            transaction.commit().await?;
520            Ok::<(), SubscriptionEnrollmentApplicationError>(())
521        }
522        .await;
523        match result {
524            Ok(()) => return Ok(()),
525            Err(error) if is_retryable_evidence_error(&error) => {
526                last_error = Some(error);
527                if attempt_index + 1 < APPROVED_EVIDENCE_WRITE_ATTEMPTS {
528                    tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
529                } else {
530                    break;
531                }
532            }
533            Err(error) => return Err(error),
534        }
535    }
536    let _ = last_error;
537    persist_approved_evidence_without_attempt_lock(
538        pool,
539        LockFreeApprovedEvidenceTerms::initial(reservation),
540        evidence,
541    )
542    .await
543}
syrup_rail_postgres/enrollment_application/initial.rs

syrup_rail_postgres/enrollment_application/
initial.rs
