1use std::{fmt, time::Duration};
2
3use chrono::{DateTime, Utc};
4use sqlx::{PgConnection, PgPool, Row};
5use syrup_rail::{
6 BillingEvent, BillingScopeId, GatewayDiagnostic, GatewayNotSubmittedError, GatewayOrderId,
7 GatewayProviderKey, PaymentAttempt, PaymentAttemptIdentity, PaymentAttemptKind,
8 PaymentAttemptRequest, PaymentAttemptStatus, PaymentMethodId, PaymentResolutionCode, PlanKey,
9 ProcessorChargeProgression, ProcessorEvidence, SubscriberId, Subscription,
10 SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentReservation, SubscriptionId,
11 SubscriptionPaymentMethodReplacement, SubscriptionRecoveryReservation,
12 SubscriptionRenewalReservation,
13};
14use thiserror::Error;
15use uuid::Uuid;
16
17use crate::{
18 BillingTransaction, BillingTransactionError,
19 attempts::{
20 AttemptApproval, AttemptResolutionStatus, AttemptTransition, PaymentAttemptStoreError,
21 find_payment_attempt_by_id_on_connection, lock_payment_attempt_by_id_on_connection,
22 persist_attempt_transition,
23 },
24 processor_charges::{
25 LockFreeApprovedEvidenceOutcome, LockFreeApprovedEvidenceTerms, observe_processor_charge,
26 },
27 renewal_failure::RenewalFailureStoreError,
28 subscription_persistence::{
29 SubscriptionPersistenceCodecError, subscription_from_row as decode_subscription_row,
30 },
31};
32
33mod initial;
34mod payment_method_replacement;
35mod recovery;
36mod renewal;
37
38pub(crate) use initial::resolve_non_approved_outcome;
39pub use initial::{
40 AdmittedSubscriptionEnrollment, SubscriptionEnrollmentAdmissionOutcome,
41 SubscriptionEnrollmentProviderResult, admit_subscription_enrollment_submission,
42 apply_reconciled_subscription_enrollment_gateway_outcome,
43 apply_subscription_enrollment_gateway_outcome, submit_admitted_subscription_enrollment,
44};
45pub(crate) use payment_method_replacement::resolve_payment_method_replacement_non_approved_outcome;
46pub use payment_method_replacement::{
47 AdmittedSubscriptionPaymentMethodReplacement,
48 SubscriptionPaymentMethodReplacementAdmissionOutcome,
49 SubscriptionPaymentMethodReplacementProviderResult,
50 admit_subscription_payment_method_replacement,
51 apply_reconciled_subscription_payment_method_replacement_gateway_outcome,
52 apply_subscription_payment_method_replacement_gateway_outcome,
53 submit_admitted_subscription_payment_method_replacement,
54};
55pub(crate) use recovery::resolve_recovery_non_approved_outcome;
56pub use recovery::{
57 AdmittedSubscriptionRecovery, SubscriptionRecoveryAdmissionOutcome,
58 SubscriptionRecoveryProviderResult, admit_subscription_recovery_submission,
59 apply_reconciled_subscription_recovery_gateway_outcome,
60 apply_subscription_recovery_gateway_outcome, submit_admitted_subscription_recovery,
61};
62pub(crate) use renewal::resolve_renewal_non_approved_outcome;
63pub use renewal::{
64 AdmittedSubscriptionRenewal, SubscriptionRenewalAdmissionOutcome,
65 SubscriptionRenewalProviderResult, admit_subscription_renewal_submission,
66 apply_reconciled_subscription_renewal_gateway_outcome,
67 apply_subscription_renewal_gateway_outcome, submit_admitted_subscription_renewal,
68};
69
70const BILLING_LOCK_TIMEOUT: Duration = Duration::from_millis(250);
71const BILLING_ROW_LOCK_TIMEOUT: &str = "250ms";
72const BILLING_OPERATION_TIMEOUT: &str = "5s";
73const APPROVED_EVIDENCE_WRITE_ATTEMPTS: usize = 3;
74const APPROVED_EVIDENCE_RETRY_DELAY: Duration = Duration::from_millis(50);
75const APPROVED_APPLICATION_ATTEMPTS: usize = 3;
76const PROVIDER_RATE_LIMIT_RETRY_AFTER_SECONDS: i64 = 60;
77const INVALID_APPLICATION_STATE: &str = "canonical initial-enrollment application state is invalid";
78const CURRENT_SUBSCRIPTION_CONFLICT_TEXT: &str =
79 "Approved subscription enrollment conflicts with a current subscription.";
80const CURRENT_GRANT_CONFLICT_TEXT: &str =
81 "Approved subscription enrollment conflicts with an active subscription grant.";
82const INCOMPLETE_APPROVAL_TEXT: &str =
83 "Approved subscription enrollment is missing required processor identity.";
84const APPROVED_STORAGE_FAILURE_TEXT: &str =
85 "Approved subscription enrollment could not be applied; manual review is required.";
86const TERMINAL_APPROVAL_RACE_TEXT: &str =
87 "Approved processor evidence arrived after the enrollment attempt became terminal.";
88const RECOVERY_INCOMPLETE_APPROVAL_TEXT: &str =
89 "Approved subscription recovery is missing required processor identity.";
90const RECOVERY_APPROVED_STORAGE_FAILURE_TEXT: &str =
91 "Approved subscription recovery could not be applied; manual review is required.";
92const RECOVERY_STALE_STATE_TEXT: &str = "Approved subscription recovery could not update billing state because the subscription changed.";
93const RENEWAL_INCOMPLETE_APPROVAL_TEXT: &str =
94 "Approved subscription renewal is missing required processor identity.";
95const RENEWAL_APPROVED_STORAGE_FAILURE_TEXT: &str =
96 "Approved subscription renewal could not be applied; manual review is required.";
97const RENEWAL_STALE_STATE_TEXT: &str = "Approved subscription renewal could not update billing state because the subscription changed.";
98const PAYMENT_METHOD_REPLACEMENT_INCOMPLETE_APPROVAL_TEXT: &str =
99 "Approved payment method replacement is missing required processor identity.";
100const PAYMENT_METHOD_REPLACEMENT_STORAGE_FAILURE_TEXT: &str =
101 "Approved payment method replacement could not be applied; manual review is required.";
102const PAYMENT_METHOD_REPLACEMENT_STALE_STATE_TEXT: &str =
103 "Approved payment method replacement could not attach because the subscription changed.";
104
105#[derive(Error)]
106pub enum SubscriptionEnrollmentApplicationError {
107 #[error("subscription enrollment application storage failed")]
108 Sql(#[from] sqlx::Error),
109 #[error("subscription enrollment attempt storage failed")]
110 Attempt(#[from] PaymentAttemptStoreError),
111 #[error("host billing transaction failed")]
112 Transaction(#[from] BillingTransactionError),
113 #[error("host billing event append failed")]
114 Event(#[from] crate::BillingEventWriteError),
115 #[error("approved subscription enrollment could not be durably applied or parked")]
116 ApprovedEvidenceNotDurable,
117 #[error("admitted subscription enrollment does not match the submission command or gateway")]
118 SubmissionIdentityMismatch,
119 #[error("{0}")]
120 InvalidState(&'static str),
121}
122
123impl From<crate::processor_charges::ProcessorChargeStoreError>
124 for SubscriptionEnrollmentApplicationError
125{
126 fn from(error: crate::processor_charges::ProcessorChargeStoreError) -> Self {
127 match error {
128 crate::processor_charges::ProcessorChargeStoreError::Sql(error) => Self::Sql(error),
129 crate::processor_charges::ProcessorChargeStoreError::Attempt(error) => {
130 Self::Attempt(error)
131 }
132 crate::processor_charges::ProcessorChargeStoreError::InvalidState(message) => {
133 Self::InvalidState(message)
134 }
135 }
136 }
137}
138
139impl From<RenewalFailureStoreError> for SubscriptionEnrollmentApplicationError {
140 fn from(error: RenewalFailureStoreError) -> Self {
141 match error {
142 RenewalFailureStoreError::Sql(error) => Self::Sql(error),
143 RenewalFailureStoreError::Attempt(error) => Self::Attempt(error),
144 RenewalFailureStoreError::InvalidState(message) => Self::InvalidState(message),
145 }
146 }
147}
148
149fn map_subscription_persistence_error(
150 error: SubscriptionPersistenceCodecError,
151) -> SubscriptionEnrollmentApplicationError {
152 match error {
153 SubscriptionPersistenceCodecError::RowRead(error) => {
154 SubscriptionEnrollmentApplicationError::Sql(error)
155 }
156 SubscriptionPersistenceCodecError::InvalidState => {
157 SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
158 }
159 }
160}
161
162pub(crate) fn map_attempt_transition_error(
163 error: PaymentAttemptStoreError,
164) -> SubscriptionEnrollmentApplicationError {
165 match error {
166 PaymentAttemptStoreError::Sql(error) => SubscriptionEnrollmentApplicationError::Sql(error),
167 PaymentAttemptStoreError::InvalidState(_) => {
168 SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE)
169 }
170 }
171}
172
173impl fmt::Debug for SubscriptionEnrollmentApplicationError {
174 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
175 match self {
176 Self::Sql(_) => formatter.write_str("SubscriptionEnrollmentApplicationError::Sql"),
177 Self::Attempt(_) => {
178 formatter.write_str("SubscriptionEnrollmentApplicationError::Attempt")
179 }
180 Self::Transaction(_) => {
181 formatter.write_str("SubscriptionEnrollmentApplicationError::Transaction")
182 }
183 Self::Event(_) => formatter.write_str("SubscriptionEnrollmentApplicationError::Event"),
184 Self::ApprovedEvidenceNotDurable => formatter
185 .write_str("SubscriptionEnrollmentApplicationError::ApprovedEvidenceNotDurable"),
186 Self::SubmissionIdentityMismatch => formatter
187 .write_str("SubscriptionEnrollmentApplicationError::SubmissionIdentityMismatch"),
188 Self::InvalidState(detail) => formatter
189 .debug_tuple("SubscriptionEnrollmentApplicationError::InvalidState")
190 .field(detail)
191 .finish(),
192 }
193 }
194}
195
196async fn finalize_approved_application(
197 mut transaction: Box<dyn BillingTransaction>,
198 application: Result<
199 (SubscriptionEnrollmentPaymentResult, Option<BillingEvent>),
200 SubscriptionEnrollmentApplicationError,
201 >,
202) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
203 let (result, event) = match application {
204 Ok(application) => application,
205 Err(error) => {
206 let _ = transaction.rollback().await;
207 return Err(error);
208 }
209 };
210 if let Some(event) = event.as_ref()
211 && let Err(error) = transaction.append_event(event).await
212 {
213 let _ = transaction.rollback().await;
214 return Err(error.into());
215 }
216 transaction.commit().await?;
217 Ok(result)
218}
219
220async fn lock_expected_reservation_attempt(
221 connection: &mut PgConnection,
222 reservation: OutcomeReservation<'_>,
223) -> Result<PaymentAttempt, SubscriptionEnrollmentApplicationError> {
224 let identity = reservation.identity();
225 let attempt = lock_payment_attempt_by_id_on_connection(
226 connection,
227 identity.billing_scope_id(),
228 identity.attempt_id(),
229 )
230 .await?
231 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
232 INVALID_APPLICATION_STATE,
233 ))?;
234 if !reservation.matches_attempt(&attempt) {
235 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
236 INVALID_APPLICATION_STATE,
237 ));
238 }
239 Ok(attempt)
240}
241
242async fn recovery_subscription_matches(
243 connection: &mut PgConnection,
244 reservation: &SubscriptionRecoveryReservation,
245) -> Result<bool, SubscriptionEnrollmentApplicationError> {
246 let identity = reservation.identity();
247 let expected = reservation.expected_state();
248 let row = sqlx::query(
249 r#"
250 SELECT status, payment_method_id, initial_transaction_id, next_renewal_at
251 FROM billing_subscriptions
252 WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
253 AND gateway_account_id = $4 AND plan_key = $5
254 FOR UPDATE
255 "#,
256 )
257 .bind(reservation.subscription_id().as_uuid())
258 .bind(identity.billing_scope_id().as_uuid())
259 .bind(identity.subscriber_id().as_uuid())
260 .bind(identity.gateway_account_id().as_uuid())
261 .bind(reservation.plan_key().as_str())
262 .fetch_optional(&mut *connection)
263 .await?;
264 let Some(row) = row else {
265 return Ok(false);
266 };
267 let status = row.try_get::<String, _>("status")?;
268 let payment_method_id: Uuid = row.try_get("payment_method_id")?;
269 let initial_transaction_id: String = row.try_get("initial_transaction_id")?;
270 let next_renewal_at: DateTime<Utc> = row.try_get("next_renewal_at")?;
271 Ok(status == expected.status().as_str()
272 && matches!(status.as_str(), "active" | "past_due")
273 && payment_method_id == expected.payment_method_id().into_uuid()
274 && syrup_rail::canonical_gateway_transaction_ids_equal(
275 &initial_transaction_id,
276 expected.initial_transaction_id().expose(),
277 )
278 && next_renewal_at == *reservation.period().start_at())
279}
280
281async fn renewal_subscription_matches(
282 connection: &mut PgConnection,
283 reservation: &SubscriptionRenewalReservation,
284) -> Result<bool, SubscriptionEnrollmentApplicationError> {
285 let identity = reservation.identity();
286 let expected = reservation.expected_state();
287 let row = sqlx::query(
288 r#"
289 SELECT status, payment_method_id, initial_transaction_id,
290 amount_cents, currency, next_renewal_at
291 FROM billing_subscriptions
292 WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
293 AND gateway_account_id = $4 AND plan_key = $5
294 FOR UPDATE
295 "#,
296 )
297 .bind(reservation.subscription_id().as_uuid())
298 .bind(identity.billing_scope_id().as_uuid())
299 .bind(identity.subscriber_id().as_uuid())
300 .bind(identity.gateway_account_id().as_uuid())
301 .bind(reservation.plan_key().as_str())
302 .fetch_optional(&mut *connection)
303 .await?;
304 let Some(row) = row else {
305 return Ok(false);
306 };
307 let status = row.try_get::<String, _>("status")?;
308 let initial_transaction_id: String = row.try_get("initial_transaction_id")?;
309 Ok(status == expected.status().as_str()
310 && matches!(status.as_str(), "active" | "past_due")
311 && row.try_get::<Uuid, _>("payment_method_id")? == expected.payment_method_id().into_uuid()
312 && syrup_rail::canonical_gateway_transaction_ids_equal(
313 &initial_transaction_id,
314 expected.initial_transaction_id().expose(),
315 )
316 && row.try_get::<i32, _>("amount_cents")? == reservation.request().amount().cents()
317 && row.try_get::<String, _>("currency")?
318 == reservation.request().amount().currency().as_str()
319 && row.try_get::<DateTime<Utc>, _>("next_renewal_at")? == *reservation.period().start_at())
320}
321
322async fn disable_payment_method_if_unreferenced(
323 connection: &mut PgConnection,
324 payment_method_id: PaymentMethodId,
325) -> Result<(), sqlx::Error> {
326 sqlx::query(
327 r#"
328 UPDATE billing_payment_methods AS methods
329 SET status = 'disabled', updated_at = clock_timestamp()
330 WHERE methods.id = $1 AND methods.status = 'active'
331 AND NOT EXISTS (
332 SELECT 1 FROM billing_subscriptions AS subscriptions
333 WHERE subscriptions.payment_method_id = methods.id
334 AND subscriptions.status IN ('active', 'past_due')
335 )
336 "#,
337 )
338 .bind(payment_method_id.as_uuid())
339 .execute(connection)
340 .await?;
341 Ok(())
342}
343
344async fn advance_subscription_discount_after_successful_charge(
345 connection: &mut PgConnection,
346 subscription_id: SubscriptionId,
347 plan_key: &PlanKey,
348) -> Result<(), SubscriptionEnrollmentApplicationError> {
349 let row = sqlx::query(
350 r#"
351 SELECT duration, status, periods_total, periods_applied, base_amount_cents
352 FROM billing_subscription_discounts
353 WHERE subscription_id = $1 AND plan_key = $2
354 FOR UPDATE
355 "#,
356 )
357 .bind(subscription_id.as_uuid())
358 .bind(plan_key.as_str())
359 .fetch_optional(&mut *connection)
360 .await?;
361 let Some(row) = row else {
362 return Ok(());
363 };
364 let duration: String = row.try_get("duration")?;
365 let status: String = row.try_get("status")?;
366 if status == "completed" {
367 return Ok(());
368 }
369 let periods_applied: i32 = row.try_get("periods_applied")?;
370 if duration == "indefinite" {
371 match periods_applied {
372 0 => {
373 sqlx::query(
374 r#"
375 UPDATE billing_subscription_discounts
376 SET periods_applied = 1
377 WHERE subscription_id = $1 AND plan_key = $2
378 AND status = 'active' AND periods_applied = 0
379 "#,
380 )
381 .bind(subscription_id.as_uuid())
382 .bind(plan_key.as_str())
383 .execute(&mut *connection)
384 .await?;
385 return Ok(());
386 }
387 1 => return Ok(()),
388 _ => {
389 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
390 INVALID_APPLICATION_STATE,
391 ));
392 }
393 }
394 }
395 if duration != "limited_months" {
396 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
397 INVALID_APPLICATION_STATE,
398 ));
399 }
400 let periods_total: i32 = row.try_get("periods_total")?;
401 if periods_applied >= periods_total {
402 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
403 INVALID_APPLICATION_STATE,
404 ));
405 }
406 let next_periods_applied = periods_applied + 1;
407 let completed = next_periods_applied == periods_total;
408 sqlx::query(
409 r#"
410 UPDATE billing_subscription_discounts
411 SET periods_applied = $3,
412 status = CASE WHEN $4 THEN 'completed' ELSE 'active' END,
413 completed_at = CASE WHEN $4 THEN clock_timestamp() ELSE NULL END
414 WHERE subscription_id = $1 AND plan_key = $2
415 "#,
416 )
417 .bind(subscription_id.as_uuid())
418 .bind(plan_key.as_str())
419 .bind(next_periods_applied)
420 .bind(completed)
421 .execute(&mut *connection)
422 .await?;
423 if completed {
424 let base_amount_cents: i32 = row.try_get("base_amount_cents")?;
425 sqlx::query(
426 "UPDATE billing_subscriptions SET amount_cents = $2, updated_at = clock_timestamp() WHERE id = $1",
427 )
428 .bind(subscription_id.as_uuid())
429 .bind(base_amount_cents)
430 .execute(&mut *connection)
431 .await?;
432 }
433 Ok(())
434}
435
436#[derive(Clone, Copy, Debug, Eq, PartialEq)]
437pub(crate) enum OutcomeResolutionBoundary {
438 Prepared,
439 AdmittedNotSubmitted,
440 Submitted,
441}
442
443#[derive(Clone, Copy)]
444pub(crate) enum RateLimitCooldown {
445 Account,
446 Provider,
447}
448
449#[derive(Clone, Copy, Debug, Eq, PartialEq)]
450enum ReservationOperation {
451 Initial,
452 Recovery,
453 Renewal,
454 PaymentMethodReplacement,
455}
456
457impl ReservationOperation {
458 const fn expected_kind(self) -> PaymentAttemptKind {
459 match self {
460 Self::Initial => PaymentAttemptKind::SubscriptionInitial,
461 Self::Recovery => PaymentAttemptKind::SubscriptionRecovery,
462 Self::Renewal => PaymentAttemptKind::SubscriptionRenewal,
463 Self::PaymentMethodReplacement => PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
464 }
465 }
466
467 const fn preserves_review_required_for_unknown(self) -> bool {
468 matches!(self, Self::PaymentMethodReplacement)
469 }
470}
471
472#[derive(Clone, Copy)]
476enum OutcomeReservation<'a> {
477 Initial(&'a SubscriptionEnrollmentReservation),
478 Recovery(&'a SubscriptionRecoveryReservation),
479 Renewal(&'a SubscriptionRenewalReservation),
480 PaymentMethodReplacement(&'a SubscriptionPaymentMethodReplacement),
481}
482
483impl<'a> OutcomeReservation<'a> {
484 const fn operation(self) -> ReservationOperation {
485 match self {
486 Self::Initial(_) => ReservationOperation::Initial,
487 Self::Recovery(_) => ReservationOperation::Recovery,
488 Self::Renewal(_) => ReservationOperation::Renewal,
489 Self::PaymentMethodReplacement(_) => ReservationOperation::PaymentMethodReplacement,
490 }
491 }
492
493 const fn identity(self) -> PaymentAttemptIdentity {
494 match self {
495 Self::Initial(reservation) => reservation.identity(),
496 Self::Recovery(reservation) => reservation.identity(),
497 Self::Renewal(reservation) => reservation.identity(),
498 Self::PaymentMethodReplacement(reservation) => reservation.identity(),
499 }
500 }
501
502 const fn plan_key(self) -> &'a PlanKey {
503 match self {
504 Self::Initial(reservation) => reservation.plan_key(),
505 Self::Recovery(reservation) => reservation.plan_key(),
506 Self::Renewal(reservation) => reservation.plan_key(),
507 Self::PaymentMethodReplacement(reservation) => reservation.plan_key(),
508 }
509 }
510
511 const fn provider_key(self) -> &'a GatewayProviderKey {
512 match self {
513 Self::Initial(reservation) => reservation.provider_key(),
514 Self::Recovery(reservation) => reservation.provider_key(),
515 Self::Renewal(reservation) => reservation.provider_key(),
516 Self::PaymentMethodReplacement(reservation) => reservation.provider_key(),
517 }
518 }
519
520 const fn expected_kind(self) -> PaymentAttemptKind {
521 self.operation().expected_kind()
522 }
523
524 fn expected_attempt(self) -> ReservationAttemptExpectation<'a> {
525 match self {
526 Self::Initial(reservation) => ReservationAttemptExpectation::Initial {
527 identity: reservation.identity(),
528 plan_key: reservation.plan_key(),
529 gateway_order_id: reservation.gateway_order_id(),
530 },
531 Self::Recovery(reservation) => ReservationAttemptExpectation::Exact {
532 identity: reservation.identity(),
533 kind: self.expected_kind(),
534 request: reservation.request(),
535 },
536 Self::Renewal(reservation) => ReservationAttemptExpectation::Exact {
537 identity: reservation.identity(),
538 kind: self.expected_kind(),
539 request: reservation.request(),
540 },
541 Self::PaymentMethodReplacement(reservation) => ReservationAttemptExpectation::Exact {
542 identity: reservation.identity(),
543 kind: self.expected_kind(),
544 request: reservation.request(),
545 },
546 }
547 }
548
549 fn matches_attempt(self, attempt: &PaymentAttempt) -> bool {
550 self.expected_attempt().matches(attempt)
551 }
552}
553
554enum ReservationAttemptExpectation<'a> {
558 Initial {
559 identity: PaymentAttemptIdentity,
560 plan_key: &'a PlanKey,
561 gateway_order_id: &'a GatewayOrderId,
562 },
563 Exact {
564 identity: PaymentAttemptIdentity,
565 kind: PaymentAttemptKind,
566 request: &'a PaymentAttemptRequest,
567 },
568}
569
570impl ReservationAttemptExpectation<'_> {
571 const fn expected_kind(&self) -> PaymentAttemptKind {
572 match self {
573 Self::Initial { .. } => PaymentAttemptKind::SubscriptionInitial,
574 Self::Exact { kind, .. } => *kind,
575 }
576 }
577
578 fn matches(&self, attempt: &PaymentAttempt) -> bool {
579 match self {
580 Self::Initial {
581 identity,
582 plan_key,
583 gateway_order_id,
584 } => {
585 attempt.identity() == *identity
586 && attempt.kind() == self.expected_kind()
587 && attempt.request().target().plan_key() == Some(*plan_key)
588 && attempt.request().gateway_order_id() == *gateway_order_id
589 }
590 Self::Exact {
591 identity,
592 kind,
593 request,
594 } => {
595 attempt.identity() == *identity
596 && attempt.kind() == *kind
597 && attempt.request() == *request
598 }
599 }
600 }
601}
602
603#[derive(Clone, Copy, Debug, Eq, PartialEq)]
604enum OutcomeResolutionKind {
605 NonApproved,
606 Unknown,
607}
608
609#[derive(Clone, Copy)]
612struct OutcomeResolutionCommand {
613 kind: OutcomeResolutionKind,
614 status: AttemptResolutionStatus,
615 resolution_code: Option<PaymentResolutionCode>,
616 cooldown: Option<RateLimitCooldown>,
617 boundary: OutcomeResolutionBoundary,
618}
619
620impl OutcomeResolutionCommand {
621 const fn non_approved(
622 status: AttemptResolutionStatus,
623 resolution_code: Option<PaymentResolutionCode>,
624 cooldown: Option<RateLimitCooldown>,
625 boundary: OutcomeResolutionBoundary,
626 ) -> Self {
627 Self {
628 kind: OutcomeResolutionKind::NonApproved,
629 status,
630 resolution_code,
631 cooldown,
632 boundary,
633 }
634 }
635
636 const fn unknown(cooldown: Option<RateLimitCooldown>) -> Self {
637 Self {
638 kind: OutcomeResolutionKind::Unknown,
639 status: AttemptResolutionStatus::Unknown,
640 resolution_code: None,
641 cooldown,
642 boundary: OutcomeResolutionBoundary::Submitted,
643 }
644 }
645
646 const fn may_resolve(self, status: PaymentAttemptStatus, submitted: bool) -> bool {
647 status.is_resolvable()
648 && match self.boundary {
649 OutcomeResolutionBoundary::Prepared => !submitted,
650 OutcomeResolutionBoundary::AdmittedNotSubmitted => submitted,
651 OutcomeResolutionBoundary::Submitted => true,
652 }
653 }
654
655 fn resolved_status(
656 self,
657 operation: ReservationOperation,
658 current: PaymentAttemptStatus,
659 ) -> AttemptResolutionStatus {
660 if self.kind == OutcomeResolutionKind::Unknown
661 && operation.preserves_review_required_for_unknown()
662 && current == PaymentAttemptStatus::ReviewRequired
663 {
664 AttemptResolutionStatus::ReviewRequired
665 } else {
666 self.status
667 }
668 }
669
670 fn clears_submitted_at(self) -> bool {
671 self.boundary == OutcomeResolutionBoundary::AdmittedNotSubmitted
672 }
673
674 fn records_pending_evidence(self, status: AttemptResolutionStatus) -> bool {
675 self.kind == OutcomeResolutionKind::Unknown
676 && status != AttemptResolutionStatus::ReviewRequired
677 }
678
679 fn marks_renewal_past_due(self, status: AttemptResolutionStatus) -> bool {
680 self.boundary == OutcomeResolutionBoundary::Submitted
681 && matches!(
682 status,
683 AttemptResolutionStatus::Declined | AttemptResolutionStatus::Failed
684 )
685 }
686}
687
688async fn resolve_pool_outcome(
689 pool: &PgPool,
690 reservation: OutcomeReservation<'_>,
691 evidence: &ProcessorEvidence,
692 resolution: OutcomeResolutionCommand,
693) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
694 let identity = reservation.identity();
695 let mut transaction = pool.begin().await?;
696 set_application_timeouts(&mut transaction).await?;
697 lock_subscription_aggregate(
698 &mut transaction,
699 identity.subscriber_id(),
700 reservation.plan_key(),
701 )
702 .await?;
703 let attempt = lock_expected_reservation_attempt(&mut transaction, reservation).await?;
704 if resolution.may_resolve(
705 attempt.status(),
706 attempt.state().timestamps().submitted_at().is_some(),
707 ) {
708 let status = resolution.resolved_status(reservation.operation(), attempt.status());
709 persist_attempt_transition(
710 &mut transaction,
711 &attempt,
712 evidence,
713 AttemptTransition::Resolved {
714 status,
715 resolution_code: resolution.resolution_code,
716 },
717 )
718 .await
719 .map_err(map_attempt_transition_error)?;
720 if resolution.clears_submitted_at() {
721 clear_attempt_submission(&mut transaction, &attempt).await?;
722 }
723 if resolution.records_pending_evidence(status) && evidence_looks_approved(evidence) {
724 observe_processor_charge(
725 &mut transaction,
726 &attempt,
727 evidence,
728 ProcessorChargeProgression::Pending,
729 )
730 .await?;
731 }
732 }
733 if let Some(cooldown) = resolution.cooldown {
734 extend_rate_limit_cooldown(&mut transaction, reservation, cooldown).await?;
735 }
736 let result = payment_result_for_reservation_attempt(&mut transaction, reservation).await?;
737 transaction.commit().await?;
738 Ok(result)
739}
740
741async fn clear_attempt_submission(
742 connection: &mut PgConnection,
743 attempt: &PaymentAttempt,
744) -> Result<(), SubscriptionEnrollmentApplicationError> {
745 sqlx::query(
746 "UPDATE billing_payment_attempts SET submitted_at = NULL, updated_at = clock_timestamp() WHERE id = $1",
747 )
748 .bind(attempt.identity().attempt_id().as_uuid())
749 .execute(connection)
750 .await?;
751 Ok(())
752}
753
754async fn payment_result_for_reservation_attempt(
755 connection: &mut PgConnection,
756 reservation: OutcomeReservation<'_>,
757) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
758 let identity = reservation.identity();
759 let attempt = find_payment_attempt_by_id_on_connection(
760 connection,
761 identity.billing_scope_id(),
762 identity.attempt_id(),
763 )
764 .await?
765 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
766 INVALID_APPLICATION_STATE,
767 ))?;
768 payment_result_for_attempt(connection, attempt).await
769}
770
771async fn extend_rate_limit_cooldown(
772 connection: &mut PgConnection,
773 reservation: OutcomeReservation<'_>,
774 cooldown: RateLimitCooldown,
775) -> Result<(), SubscriptionEnrollmentApplicationError> {
776 let identity = reservation.identity();
777 let result = match cooldown {
778 RateLimitCooldown::Account => {
779 sqlx::query(
780 r#"
781 UPDATE billing_gateway_accounts
782 SET mutation_rate_limited_until = GREATEST(
783 COALESCE(mutation_rate_limited_until, '-infinity'::timestamptz),
784 clock_timestamp() + make_interval(secs => $4)
785 )
786 WHERE id = $1 AND billing_scope_id = $2
787 AND gateway_configuration_id = $3
788 "#,
789 )
790 .bind(identity.gateway_account_id().as_uuid())
791 .bind(identity.billing_scope_id().as_uuid())
792 .bind(identity.gateway_configuration_id().as_uuid())
793 .bind(PROVIDER_RATE_LIMIT_RETRY_AFTER_SECONDS)
794 .execute(&mut *connection)
795 .await?
796 }
797 RateLimitCooldown::Provider => {
798 sqlx::query(
799 r#"
800 UPDATE billing_gateway_provider_rate_limits
801 SET rate_limited_until = GREATEST(
802 rate_limited_until,
803 clock_timestamp() + make_interval(secs => $2)
804 )
805 WHERE provider_key = $1
806 "#,
807 )
808 .bind(reservation.provider_key().as_str())
809 .bind(PROVIDER_RATE_LIMIT_RETRY_AFTER_SECONDS)
810 .execute(&mut *connection)
811 .await?
812 }
813 };
814 if result.rows_affected() != 1 {
815 return Err(SubscriptionEnrollmentApplicationError::InvalidState(
816 INVALID_APPLICATION_STATE,
817 ));
818 }
819 Ok(())
820}
821
822pub(crate) fn mutation_error_evidence(detail: &GatewayDiagnostic) -> ProcessorEvidence {
823 ProcessorEvidence::new(
824 None,
825 None,
826 None,
827 None,
828 Some(detail.clone()),
829 None,
830 syrup_rail::GatewayPaymentDescriptor::default(),
831 )
832}
833
834pub(crate) const fn not_submitted_resolution_code(
835 error: &GatewayNotSubmittedError,
836) -> PaymentResolutionCode {
837 match error {
838 GatewayNotSubmittedError::RequestRejected(_) => {
839 PaymentResolutionCode::GatewayRequestRejectedBeforeSubmission
840 }
841 GatewayNotSubmittedError::Malformed(_) => {
842 PaymentResolutionCode::GatewayMalformedBeforeSubmission
843 }
844 GatewayNotSubmittedError::Configuration(_) => {
845 PaymentResolutionCode::GatewayConfigurationBeforeSubmission
846 }
847 GatewayNotSubmittedError::Unavailable(_) => {
848 PaymentResolutionCode::GatewayUnavailableBeforeSubmission
849 }
850 GatewayNotSubmittedError::RateLimited(_) => {
851 PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission
852 }
853 }
854}
855
856async fn persist_approved_evidence_without_attempt_lock(
857 pool: &PgPool,
858 terms: LockFreeApprovedEvidenceTerms<'_>,
859 evidence: &ProcessorEvidence,
860) -> Result<(), SubscriptionEnrollmentApplicationError> {
861 match crate::processor_charges::persist_approved_evidence_without_attempt_lock(
862 pool, terms, evidence,
863 )
864 .await?
865 {
866 LockFreeApprovedEvidenceOutcome::Persisted
867 | LockFreeApprovedEvidenceOutcome::ExactReplay
868 | LockFreeApprovedEvidenceOutcome::OwnedByOtherAttempt => Ok(()),
869 LockFreeApprovedEvidenceOutcome::NotDurable => {
870 Err(SubscriptionEnrollmentApplicationError::ApprovedEvidenceNotDurable)
871 }
872 }
873}
874
875fn is_retryable_evidence_error(error: &SubscriptionEnrollmentApplicationError) -> bool {
876 let sqlstate = match error {
877 SubscriptionEnrollmentApplicationError::Sql(sqlx::Error::Database(error)) => error.code(),
878 SubscriptionEnrollmentApplicationError::Attempt(PaymentAttemptStoreError::Sql(
879 sqlx::Error::Database(error),
880 )) => error.code(),
881 _ => None,
882 };
883 matches!(
884 sqlstate.as_deref(),
885 Some("40001" | "40P01" | "55P03" | "57014")
886 )
887}
888
889pub(crate) async fn set_application_timeouts(
890 connection: &mut PgConnection,
891) -> Result<(), sqlx::Error> {
892 sqlx::query(
893 "SELECT set_config('lock_timeout', $1, true), set_config('statement_timeout', $2, true)",
894 )
895 .bind(BILLING_ROW_LOCK_TIMEOUT)
896 .bind(BILLING_OPERATION_TIMEOUT)
897 .execute(connection)
898 .await?;
899 Ok(())
900}
901
902async fn lock_payment_method_domain(
903 connection: &mut PgConnection,
904 subscriber_id: SubscriberId,
905 gateway_account_id: &Uuid,
906) -> Result<(), sqlx::Error> {
907 sqlx::query(
908 "SELECT pg_advisory_xact_lock(hashtextextended($1::uuid::text || ':' || $2::uuid::text, 0))",
909 )
910 .bind(gateway_account_id)
911 .bind(subscriber_id.as_uuid())
912 .execute(connection)
913 .await?;
914 Ok(())
915}
916
917async fn lock_subscription_aggregate(
918 connection: &mut PgConnection,
919 subscriber_id: SubscriberId,
920 plan_key: &PlanKey,
921) -> Result<(), sqlx::Error> {
922 sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1::uuid::text || ':' || $2, 0))")
923 .bind(subscriber_id.as_uuid())
924 .bind(plan_key.as_str())
925 .execute(connection)
926 .await?;
927 Ok(())
928}
929
930async fn upsert_payment_method(
931 connection: &mut PgConnection,
932 attempt: &PaymentAttempt,
933 evidence: &ProcessorEvidence,
934) -> Result<PaymentMethodId, SubscriptionEnrollmentApplicationError> {
935 let identity = attempt.identity();
936 let reference = evidence.payment_method_reference().ok_or(
937 SubscriptionEnrollmentApplicationError::InvalidState(INVALID_APPLICATION_STATE),
938 )?;
939 let descriptor = evidence.descriptor();
940 let row_id: Uuid = sqlx::query_scalar(
941 r#"
942 INSERT INTO billing_payment_methods (
943 id, billing_scope_id, subscriber_id, gateway_account_id,
944 gateway_payment_method_reference, status, payment_type, card_brand,
945 card_last4, card_exp_month, card_exp_year, billing_name, billing_email
946 ) VALUES ($1, $2, $3, $4, $5, 'active', $6, $7, $8, $9, $10, $11, $12)
947 ON CONFLICT (gateway_account_id, subscriber_id, gateway_payment_method_reference)
948 DO UPDATE SET status = 'active', payment_type = EXCLUDED.payment_type,
949 card_brand = EXCLUDED.card_brand, card_last4 = EXCLUDED.card_last4,
950 card_exp_month = EXCLUDED.card_exp_month,
951 card_exp_year = EXCLUDED.card_exp_year,
952 billing_name = EXCLUDED.billing_name,
953 billing_email = EXCLUDED.billing_email,
954 updated_at = clock_timestamp()
955 RETURNING id
956 "#,
957 )
958 .bind(Uuid::now_v7())
959 .bind(identity.billing_scope_id().as_uuid())
960 .bind(identity.subscriber_id().as_uuid())
961 .bind(identity.gateway_account_id().as_uuid())
962 .bind(reference.expose())
963 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
964 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
965 .bind(descriptor.card_last_four().map(|value| value.expose()))
966 .bind(descriptor.card_exp_month())
967 .bind(descriptor.card_exp_year())
968 .bind(attempt.request().billing_contact().name())
969 .bind(attempt.request().billing_contact().email())
970 .fetch_one(connection)
971 .await?;
972 Ok(PaymentMethodId::new(row_id))
973}
974
975async fn mark_attempt_approved(
976 connection: &mut PgConnection,
977 attempt: &PaymentAttempt,
978 evidence: &ProcessorEvidence,
979 subscription_id: SubscriptionId,
980 method_id: PaymentMethodId,
981) -> Result<(), SubscriptionEnrollmentApplicationError> {
982 persist_attempt_transition(
983 connection,
984 attempt,
985 evidence,
986 AttemptTransition::Approved(AttemptApproval::Subscription {
987 subscription_id,
988 payment_method_id: method_id,
989 }),
990 )
991 .await
992 .map_err(map_attempt_transition_error)
993}
994
995pub(crate) async fn park_locked_attempt(
996 connection: &mut PgConnection,
997 attempt: &PaymentAttempt,
998 evidence: &ProcessorEvidence,
999 resolution_code: Option<PaymentResolutionCode>,
1000 message: &'static str,
1001) -> Result<PaymentAttempt, SubscriptionEnrollmentApplicationError> {
1002 persist_attempt_transition(
1003 connection,
1004 attempt,
1005 evidence,
1006 AttemptTransition::LateApprovalReview {
1007 resolution_code,
1008 message,
1009 },
1010 )
1011 .await
1012 .map_err(map_attempt_transition_error)?;
1013 find_payment_attempt_by_id_on_connection(
1014 connection,
1015 attempt.identity().billing_scope_id(),
1016 attempt.identity().attempt_id(),
1017 )
1018 .await?
1019 .ok_or(SubscriptionEnrollmentApplicationError::InvalidState(
1020 INVALID_APPLICATION_STATE,
1021 ))
1022}
1023
1024fn evidence_looks_approved(evidence: &ProcessorEvidence) -> bool {
1025 evidence.transaction_id().is_some()
1026 && (evidence
1027 .response()
1028 .is_some_and(|value| syrup_rail::gateway_response_is_approved(Some(value.expose())))
1029 || evidence
1030 .condition()
1031 .is_some_and(|value| syrup_rail::gateway_state_is_approved(value.expose())))
1032}
1033
1034pub(crate) async fn payment_result_for_attempt(
1035 connection: &mut PgConnection,
1036 attempt: PaymentAttempt,
1037) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
1038 let subscription = match attempt.kind() {
1039 PaymentAttemptKind::SubscriptionRecovery
1040 | PaymentAttemptKind::SubscriptionPaymentMethodUpdate => {
1041 load_applied_subscription(connection, &attempt).await?
1042 }
1043 _ if attempt.status() == PaymentAttemptStatus::Approved => {
1044 load_applied_subscription(connection, &attempt).await?
1045 }
1046 _ => None,
1047 };
1048 Ok(SubscriptionEnrollmentPaymentResult::new(
1049 attempt,
1050 subscription,
1051 ))
1052}
1053
1054async fn load_applied_subscription(
1055 connection: &mut PgConnection,
1056 attempt: &PaymentAttempt,
1057) -> Result<Option<Subscription>, SubscriptionEnrollmentApplicationError> {
1058 let Some(subscription_id) = attempt.request().target().subscription_id() else {
1059 return Ok(None);
1060 };
1061 load_subscription(
1062 connection,
1063 attempt.identity().billing_scope_id(),
1064 subscription_id,
1065 )
1066 .await
1067}
1068
1069async fn load_subscription(
1070 connection: &mut PgConnection,
1071 billing_scope_id: BillingScopeId,
1072 subscription_id: SubscriptionId,
1073) -> Result<Option<Subscription>, SubscriptionEnrollmentApplicationError> {
1074 let row = sqlx::query(
1075 r#"
1076 SELECT id, plan_key, status, payment_method_id, amount_cents, currency,
1077 current_period_start_at, current_period_end_at, next_renewal_at,
1078 phase, recurring_period_kind, recurring_period_count,
1079 dunning_retry_delays_seconds, dunning_exhaustion, past_due_access,
1080 next_payment_attempt_at
1081 FROM billing_subscriptions
1082 WHERE billing_scope_id = $1 AND id = $2
1083 "#,
1084 )
1085 .bind(billing_scope_id.as_uuid())
1086 .bind(subscription_id.as_uuid())
1087 .fetch_optional(connection)
1088 .await?;
1089 row.map(|row| decode_subscription_row(&row).map_err(map_subscription_persistence_error))
1090 .transpose()
1091}
1092
1093#[cfg(test)]
1094mod tests;