1use std::time::Duration;
2
3use sqlx::{PgConnection, PgPool, Row};
4use syrup_rail::{
5 CurrencyCode, GatewayDiagnostic, GatewayOrderId, GatewayTransactionId, PaymentAttempt,
6 PaymentAttemptId, PaymentAttemptIdentity, PaymentAttemptKind, PaymentAttemptStatus,
7 PaymentResolutionCode, PlanKey, ProcessorCharge, ProcessorChargeId, ProcessorChargeProgression,
8 ProcessorChargeRole, ProcessorChargeStateCode, ProcessorEvidence,
9 SubscriptionEnrollmentReservation, SubscriptionPaymentMethodReplacement,
10 SubscriptionRecoveryReservation,
11};
12use thiserror::Error;
13use uuid::Uuid;
14
15use crate::attempts::{
16 PAYMENT_ATTEMPT_SELECT, find_payment_attempt_by_id_on_connection,
17 lock_payment_attempt_by_id_on_connection, lock_subscription_aggregate,
18 payment_attempt_from_row, set_enrollment_timeouts,
19};
20use crate::operator_review::{
21 attestation_by_charge, attestation_matches_source, processor_charge_from_row,
22};
23use crate::{OperatorReviewError, PaymentAttemptStoreError};
24
25const INVALID_CHARGE_STATE: &str = "canonical processor charge state is invalid";
26const STORE_MAX_ATTEMPTS: usize = 3;
27const STORE_RETRY_DELAY: Duration = Duration::from_millis(50);
28
29#[derive(Debug, Error)]
30pub enum ProcessorChargeStoreError {
31 #[error("processor charge storage operation failed")]
32 Sql(#[from] sqlx::Error),
33 #[error("payment attempt storage operation failed")]
34 Attempt(#[from] PaymentAttemptStoreError),
35 #[error("{0}")]
36 InvalidState(&'static str),
37}
38
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub enum CompensatingProcessorChargeOutcome {
41 Observed,
42 ExactReplay,
43 OwnedByOtherAttempt,
44}
45
46#[derive(Clone, Debug, Eq, PartialEq)]
47pub enum ProcessorChargeObservationOutcome {
48 Observed(ProcessorCharge),
49 ExactReplay(ProcessorCharge),
50 OwnedByOtherAttempt,
51}
52
53#[derive(Clone, Copy, Debug)]
54pub(crate) struct ChargeRecord {
55 pub(crate) id: Uuid,
56 pub(crate) role: ProcessorChargeRole,
57 exact_replay: bool,
58}
59
60#[derive(Clone, Copy, Debug)]
61pub(crate) enum ObservedCharge {
62 Owned(ChargeRecord),
63 OwnedByOtherAttempt,
64}
65
66#[derive(Clone, Copy)]
72pub(crate) struct LockFreeApprovedEvidenceTerms<'a> {
73 identity: PaymentAttemptIdentity,
74 attempt_kind: PaymentAttemptKind,
75 plan_key: &'a PlanKey,
76 gateway_order_id: &'a GatewayOrderId,
77 amount_cents: i32,
78 currency: CurrencyCode,
79 host_charge_target_id: Option<Uuid>,
80}
81
82impl<'a> LockFreeApprovedEvidenceTerms<'a> {
83 pub(crate) fn initial(reservation: &'a SubscriptionEnrollmentReservation) -> Self {
84 let charge = reservation.expected_charge().charge();
85 Self::subscription(
86 reservation.identity(),
87 PaymentAttemptKind::SubscriptionInitial,
88 reservation.plan_key(),
89 reservation.gateway_order_id(),
90 charge.cents(),
91 charge.currency(),
92 )
93 }
94
95 pub(crate) fn recovery(reservation: &'a SubscriptionRecoveryReservation) -> Self {
96 let request = reservation.request();
97 let amount = request.amount();
98 Self::subscription(
99 reservation.identity(),
100 PaymentAttemptKind::SubscriptionRecovery,
101 reservation.plan_key(),
102 request.gateway_order_id(),
103 amount.cents(),
104 amount.currency(),
105 )
106 }
107
108 pub(crate) fn payment_method_replacement(
109 reservation: &'a SubscriptionPaymentMethodReplacement,
110 ) -> Self {
111 let request = reservation.request();
112 debug_assert_eq!(request.amount().cents(), 0);
113 Self::subscription(
114 reservation.identity(),
115 PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
116 reservation.plan_key(),
117 request.gateway_order_id(),
118 0,
119 request.amount().currency(),
120 )
121 }
122
123 fn subscription(
124 identity: PaymentAttemptIdentity,
125 attempt_kind: PaymentAttemptKind,
126 plan_key: &'a PlanKey,
127 gateway_order_id: &'a GatewayOrderId,
128 amount_cents: i32,
129 currency: CurrencyCode,
130 ) -> Self {
131 debug_assert!(matches!(
132 attempt_kind,
133 PaymentAttemptKind::SubscriptionInitial
134 | PaymentAttemptKind::SubscriptionRecovery
135 | PaymentAttemptKind::SubscriptionPaymentMethodUpdate
136 ));
137 Self {
138 identity,
139 attempt_kind,
140 plan_key,
141 gateway_order_id,
142 amount_cents,
143 currency,
144 host_charge_target_id: None,
145 }
146 }
147}
148
149#[derive(Clone, Copy, Debug, Eq, PartialEq)]
150pub(crate) enum LockFreeApprovedEvidenceOutcome {
151 Persisted,
152 ExactReplay,
153 OwnedByOtherAttempt,
154 NotDurable,
155}
156
157pub(crate) async fn persist_approved_evidence_without_attempt_lock(
164 pool: &PgPool,
165 terms: LockFreeApprovedEvidenceTerms<'_>,
166 evidence: &ProcessorEvidence,
167) -> Result<LockFreeApprovedEvidenceOutcome, sqlx::Error> {
168 let identity = terms.identity;
169 let descriptor = evidence.descriptor();
170 let transaction_id = evidence.transaction_id().map(GatewayTransactionId::expose);
171 let mut transaction = pool.begin().await?;
172 set_enrollment_timeouts(&mut transaction).await?;
173
174 for _ in 0..2 {
175 let has_existing_charge: bool = sqlx::query_scalar(
176 "SELECT EXISTS (SELECT 1 FROM billing_processor_charges WHERE attempt_id = $1)",
177 )
178 .bind(identity.attempt_id().as_uuid())
179 .fetch_one(&mut *transaction)
180 .await?;
181 let role = if has_existing_charge {
182 "additional"
183 } else {
184 "primary"
185 };
186 let inserted = sqlx::query_scalar::<_, Uuid>(
187 r#"
188 INSERT INTO billing_processor_charges (
189 id, attempt_id, billing_scope_id, gateway_account_id, gateway_order_id,
190 gateway_transaction_id, gateway_payment_method_reference,
191 gateway_response, gateway_response_code, gateway_response_text,
192 gateway_condition, payment_type, card_brand, card_last4,
193 card_exp_month, card_exp_year, charge_role, progression_state,
194 attempt_kind, plan_key, host_charge_target_id, amount_cents, currency
195 ) VALUES (
196 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
197 $14, $15, $16, $17, 'pending', $18, $19, $20, $21, $22
198 )
199 ON CONFLICT DO NOTHING
200 RETURNING id
201 "#,
202 )
203 .bind(Uuid::now_v7())
204 .bind(identity.attempt_id().as_uuid())
205 .bind(identity.billing_scope_id().as_uuid())
206 .bind(identity.gateway_account_id().as_uuid())
207 .bind(terms.gateway_order_id.expose())
208 .bind(transaction_id)
209 .bind(
210 evidence
211 .payment_method_reference()
212 .map(|value| value.expose()),
213 )
214 .bind(evidence.response().map(GatewayDiagnostic::expose))
215 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
216 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
217 .bind(evidence.condition().map(GatewayDiagnostic::expose))
218 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
219 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
220 .bind(descriptor.card_last_four().map(|value| value.expose()))
221 .bind(descriptor.card_exp_month())
222 .bind(descriptor.card_exp_year())
223 .bind(role)
224 .bind(terms.attempt_kind.as_str())
225 .bind(terms.plan_key.as_str())
226 .bind(terms.host_charge_target_id)
227 .bind(terms.amount_cents)
228 .bind(terms.currency.as_str())
229 .fetch_optional(&mut *transaction)
230 .await?;
231 if inserted.is_some() {
232 transaction.commit().await?;
233 return Ok(LockFreeApprovedEvidenceOutcome::Persisted);
234 }
235 }
236
237 let evidence_matches = sqlx::query_scalar::<_, bool>(
238 r#"
239 SELECT gateway_payment_method_reference IS NOT DISTINCT FROM $3
240 AND gateway_response IS NOT DISTINCT FROM $4
241 AND gateway_response_code IS NOT DISTINCT FROM $5
242 AND gateway_response_text IS NOT DISTINCT FROM $6
243 AND gateway_condition IS NOT DISTINCT FROM $7
244 AND payment_type IS NOT DISTINCT FROM $8
245 AND card_brand IS NOT DISTINCT FROM $9
246 AND card_last4 IS NOT DISTINCT FROM $10
247 AND card_exp_month IS NOT DISTINCT FROM $11
248 AND card_exp_year IS NOT DISTINCT FROM $12
249 FROM billing_processor_charges
250 WHERE attempt_id = $1 AND gateway_transaction_id IS NOT DISTINCT FROM $2
251 "#,
252 )
253 .bind(identity.attempt_id().as_uuid())
254 .bind(transaction_id)
255 .bind(
256 evidence
257 .payment_method_reference()
258 .map(|value| value.expose()),
259 )
260 .bind(evidence.response().map(GatewayDiagnostic::expose))
261 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
262 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
263 .bind(evidence.condition().map(GatewayDiagnostic::expose))
264 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
265 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
266 .bind(descriptor.card_last_four().map(|value| value.expose()))
267 .bind(descriptor.card_exp_month())
268 .bind(descriptor.card_exp_year())
269 .fetch_optional(&mut *transaction)
270 .await?;
271 if evidence_matches == Some(true) {
272 transaction.commit().await?;
273 return Ok(LockFreeApprovedEvidenceOutcome::ExactReplay);
274 }
275 if let Some(transaction_id) = transaction_id {
276 let owned_elsewhere: bool = sqlx::query_scalar(
277 r#"
278 SELECT EXISTS (
279 SELECT 1 FROM billing_processor_charges
280 WHERE gateway_account_id = $1 AND gateway_transaction_id = $2
281 AND attempt_id <> $3
282 )
283 "#,
284 )
285 .bind(identity.gateway_account_id().as_uuid())
286 .bind(transaction_id)
287 .bind(identity.attempt_id().as_uuid())
288 .fetch_one(&mut *transaction)
289 .await?;
290 if owned_elsewhere {
291 transaction.commit().await?;
292 return Ok(LockFreeApprovedEvidenceOutcome::OwnedByOtherAttempt);
293 }
294 }
295 Ok(LockFreeApprovedEvidenceOutcome::NotDurable)
296}
297
298pub async fn store_compensating_processor_charge(
299 pool: &PgPool,
300 attempt_id: PaymentAttemptId,
301 gateway_order_id: &GatewayOrderId,
302 evidence: &ProcessorEvidence,
303) -> Result<CompensatingProcessorChargeOutcome, ProcessorChargeStoreError> {
304 let mut last_transient_error = None;
305 for store_attempt in 1..=STORE_MAX_ATTEMPTS {
306 match store_once(pool, attempt_id, gateway_order_id, evidence).await {
307 Ok(outcome) => return Ok(outcome),
308 Err(error) if is_transient(&error) && store_attempt < STORE_MAX_ATTEMPTS => {
309 last_transient_error = Some(error);
310 tokio::time::sleep(STORE_RETRY_DELAY).await;
311 }
312 Err(error) => return Err(error),
313 }
314 }
315 Err(
316 last_transient_error.unwrap_or(ProcessorChargeStoreError::InvalidState(
317 "compensating processor charge retry loop exhausted",
318 )),
319 )
320}
321
322async fn store_once(
323 pool: &PgPool,
324 attempt_id: PaymentAttemptId,
325 gateway_order_id: &GatewayOrderId,
326 evidence: &ProcessorEvidence,
327) -> Result<CompensatingProcessorChargeOutcome, ProcessorChargeStoreError> {
328 let mut transaction = pool.begin().await?;
329 set_enrollment_timeouts(&mut transaction).await?;
330 let query = format!("{PAYMENT_ATTEMPT_SELECT} WHERE id = $1");
331 let row = sqlx::query(&query)
332 .bind(attempt_id.as_uuid())
333 .fetch_optional(&mut *transaction)
334 .await?;
335 let preloaded_attempt = row
336 .as_ref()
337 .map(payment_attempt_from_row)
338 .transpose()?
339 .ok_or(ProcessorChargeStoreError::InvalidState(
340 "compensating processor charge attempt was not found",
341 ))?;
342 if preloaded_attempt.kind() == PaymentAttemptKind::SubscriptionInitial {
343 let plan_key = preloaded_attempt.request().target().plan_key().ok_or(
344 ProcessorChargeStoreError::InvalidState(INVALID_CHARGE_STATE),
345 )?;
346 lock_subscription_aggregate(
347 &mut transaction,
348 preloaded_attempt.identity().subscriber_id(),
349 plan_key,
350 )
351 .await?;
352 }
353 let attempt = if preloaded_attempt.kind() == PaymentAttemptKind::SubscriptionPaymentMethodUpdate
354 {
355 lock_payment_attempt_by_id_on_connection(
356 &mut transaction,
357 preloaded_attempt.identity().billing_scope_id(),
358 attempt_id,
359 )
360 .await?
361 .ok_or(ProcessorChargeStoreError::InvalidState(
362 "compensating processor charge attempt was not found",
363 ))?
364 } else {
365 let reloaded = find_payment_attempt_by_id_on_connection(
366 &mut transaction,
367 preloaded_attempt.identity().billing_scope_id(),
368 attempt_id,
369 )
370 .await?
371 .ok_or(ProcessorChargeStoreError::InvalidState(
372 "compensating processor charge attempt was not found",
373 ))?;
374 if reloaded != preloaded_attempt {
375 return Err(ProcessorChargeStoreError::InvalidState(
376 "compensating processor charge attempt changed while its aggregate was locked",
377 ));
378 }
379 reloaded
380 };
381 if attempt.request().gateway_order_id() != gateway_order_id {
382 return Err(ProcessorChargeStoreError::InvalidState(
383 "compensating processor charge order does not match its attempt",
384 ));
385 }
386 let initial_progression = compensating_progression(&attempt, evidence);
387 let observation =
388 observe_processor_charge(&mut transaction, &attempt, evidence, initial_progression).await?;
389 let outcome = match observation {
390 ObservedCharge::OwnedByOtherAttempt => {
391 CompensatingProcessorChargeOutcome::OwnedByOtherAttempt
392 }
393 ObservedCharge::Owned(charge) => {
394 let mut persisted = charge_by_id(&mut transaction, charge.id).await?;
395 if !charge.exact_replay {
396 let state_code = initial_charge_state_code(
397 persisted.role(),
398 persisted.progression(),
399 evidence.transaction_id().is_some(),
400 );
401 if persisted.state_code() != state_code {
402 persisted = transition_processor_charge(
403 &mut transaction,
404 persisted.id(),
405 &[persisted.progression()],
406 persisted.progression(),
407 state_code,
408 false,
409 )
410 .await?;
411 }
412 }
413 if let Some(attestation) = attestation_by_charge(&mut transaction, charge.id)
414 .await
415 .map_err(map_operator_error)?
416 && (!charge.exact_replay
417 || !attestation_matches_source(&attestation, &attempt, &persisted))
418 {
419 return Err(ProcessorChargeStoreError::InvalidState(
420 "attested processor charge was reobserved with different evidence",
421 ));
422 }
423 if charge.exact_replay {
424 CompensatingProcessorChargeOutcome::ExactReplay
425 } else {
426 CompensatingProcessorChargeOutcome::Observed
427 }
428 }
429 };
430 transaction.commit().await?;
431 Ok(outcome)
432}
433
434pub async fn observe_processor_charge_in_transaction(
435 connection: &mut PgConnection,
436 attempt_id: PaymentAttemptId,
437 gateway_order_id: &GatewayOrderId,
438 evidence: &ProcessorEvidence,
439 initial_progression: ProcessorChargeProgression,
440) -> Result<ProcessorChargeObservationOutcome, ProcessorChargeStoreError> {
441 let query = format!("{PAYMENT_ATTEMPT_SELECT} WHERE id = $1");
442 let row = sqlx::query(&query)
443 .bind(attempt_id.as_uuid())
444 .fetch_optional(&mut *connection)
445 .await?;
446 let attempt = row
447 .as_ref()
448 .map(payment_attempt_from_row)
449 .transpose()?
450 .ok_or(ProcessorChargeStoreError::InvalidState(
451 "processor charge attempt was not found",
452 ))?;
453 if attempt.request().gateway_order_id() != gateway_order_id {
454 return Err(ProcessorChargeStoreError::InvalidState(
455 "processor charge order does not match its attempt",
456 ));
457 }
458 match observe_processor_charge(connection, &attempt, evidence, initial_progression).await? {
459 ObservedCharge::OwnedByOtherAttempt => {
460 Ok(ProcessorChargeObservationOutcome::OwnedByOtherAttempt)
461 }
462 ObservedCharge::Owned(charge) => {
463 if charge.exact_replay {
464 Ok(ProcessorChargeObservationOutcome::ExactReplay(
465 charge_by_id(connection, charge.id).await?,
466 ))
467 } else {
468 let mut persisted = charge_by_id(connection, charge.id).await?;
469 let state_code = initial_charge_state_code(
470 persisted.role(),
471 persisted.progression(),
472 evidence.transaction_id().is_some(),
473 );
474 if persisted.state_code() != state_code {
475 persisted = transition_processor_charge(
476 connection,
477 persisted.id(),
478 &[persisted.progression()],
479 persisted.progression(),
480 state_code,
481 false,
482 )
483 .await?;
484 }
485 Ok(ProcessorChargeObservationOutcome::Observed(persisted))
486 }
487 }
488 }
489}
490
491pub async fn transition_processor_charge_in_transaction(
492 connection: &mut PgConnection,
493 charge_id: ProcessorChargeId,
494 expected_progressions: &[ProcessorChargeProgression],
495 progression: ProcessorChargeProgression,
496 state_code: Option<ProcessorChargeStateCode>,
497) -> Result<ProcessorCharge, ProcessorChargeStoreError> {
498 transition_processor_charge(
499 connection,
500 charge_id,
501 expected_progressions,
502 progression,
503 state_code,
504 false,
505 )
506 .await
507}
508
509pub(crate) async fn observe_processor_charge(
510 connection: &mut PgConnection,
511 attempt: &PaymentAttempt,
512 evidence: &ProcessorEvidence,
513 initial_progression: ProcessorChargeProgression,
514) -> Result<ObservedCharge, ProcessorChargeStoreError> {
515 let identity = attempt.identity();
516 let transaction_id = evidence.transaction_id().map(GatewayTransactionId::expose);
517 let has_existing_charge: bool = sqlx::query_scalar(
518 "SELECT EXISTS (SELECT 1 FROM billing_processor_charges WHERE attempt_id = $1)",
519 )
520 .bind(identity.attempt_id().as_uuid())
521 .fetch_one(&mut *connection)
522 .await?;
523 if let Some(transaction_id) = transaction_id {
524 if owned_by_other_attempt(connection, attempt, transaction_id).await? {
525 return Ok(ObservedCharge::OwnedByOtherAttempt);
526 }
527 if let Some(charge) = identify_transactionless(
528 connection,
529 attempt,
530 evidence,
531 transaction_id,
532 initial_progression,
533 )
534 .await?
535 {
536 return Ok(ObservedCharge::Owned(charge));
537 }
538 }
539 let role = if has_existing_charge {
540 ProcessorChargeRole::Additional
541 } else {
542 ProcessorChargeRole::Primary
543 };
544 let identified = transaction_id.is_some();
545 let progression = initial_charge_progression(
546 role,
547 attempt.request().amount().cents(),
548 identified,
549 initial_progression,
550 );
551 let descriptor = evidence.descriptor();
552 let conflict_clause = if identified {
553 "ON CONFLICT (gateway_account_id, gateway_transaction_id) \
554 WHERE billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NOT NULL \
555 DO NOTHING RETURNING id"
556 } else {
557 "ON CONFLICT (attempt_id) \
558 WHERE billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NULL \
559 DO NOTHING RETURNING id"
560 };
561 let insert = format!(
562 r#"
563 INSERT INTO billing_processor_charges (
564 id, attempt_id, billing_scope_id, gateway_account_id, gateway_order_id,
565 gateway_transaction_id, gateway_payment_method_reference,
566 gateway_response, gateway_response_code, gateway_response_text,
567 gateway_condition, payment_type, card_brand, card_last4,
568 card_exp_month, card_exp_year, charge_role, progression_state, state_code,
569 reconciliation_required_at, external_reversal_required_at, applied_at,
570 attempt_kind, plan_key, host_charge_target_id, amount_cents, currency
571 ) VALUES (
572 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
573 $14, $15, $16, $17, $18, $19,
574 CASE WHEN $18 = 'reconciliation_required' THEN clock_timestamp() END,
575 CASE WHEN $18 = 'external_reversal_required' THEN clock_timestamp() END,
576 CASE WHEN $18 = 'applied' THEN clock_timestamp() END,
577 $20, $21, $22, $23, $24
578 )
579 {conflict_clause}
580 "#
581 );
582 let inserted = sqlx::query_scalar::<_, Uuid>(&insert)
583 .bind(Uuid::now_v7())
584 .bind(identity.attempt_id().as_uuid())
585 .bind(identity.billing_scope_id().as_uuid())
586 .bind(identity.gateway_account_id().as_uuid())
587 .bind(attempt.request().gateway_order_id().expose())
588 .bind(transaction_id)
589 .bind(
590 evidence
591 .payment_method_reference()
592 .map(|value| value.expose()),
593 )
594 .bind(evidence.response().map(GatewayDiagnostic::expose))
595 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
596 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
597 .bind(evidence.condition().map(GatewayDiagnostic::expose))
598 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
599 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
600 .bind(descriptor.card_last_four().map(|value| value.expose()))
601 .bind(descriptor.card_exp_month())
602 .bind(descriptor.card_exp_year())
603 .bind(role.as_str())
604 .bind(progression.as_str())
605 .bind(Option::<&str>::None)
606 .bind(attempt.kind().as_str())
607 .bind(attempt.request().target().plan_key().map(PlanKey::as_str))
608 .bind(
609 attempt
610 .request()
611 .target()
612 .host_charge_target_id()
613 .map(|value| *value.as_uuid()),
614 )
615 .bind(attempt.request().amount().cents())
616 .bind(attempt.request().amount().currency().as_str())
617 .fetch_optional(&mut *connection)
618 .await?;
619 if let Some(id) = inserted {
620 return Ok(ObservedCharge::Owned(ChargeRecord {
621 id,
622 role,
623 exact_replay: false,
624 }));
625 }
626 if let Some(row) = matching_charge(connection, attempt, evidence, transaction_id).await? {
627 if !row.try_get::<bool, _>("evidence_matches")? {
628 return Err(ProcessorChargeStoreError::InvalidState(
629 "processor charge replay evidence changed",
630 ));
631 }
632 return Ok(ObservedCharge::Owned(ChargeRecord {
633 id: row.try_get("id")?,
634 role: parse_role(&row.try_get::<String, _>("charge_role")?)?,
635 exact_replay: true,
636 }));
637 }
638 if let Some(transaction_id) = transaction_id
639 && owned_by_other_attempt(connection, attempt, transaction_id).await?
640 {
641 return Ok(ObservedCharge::OwnedByOtherAttempt);
642 }
643 Err(ProcessorChargeStoreError::InvalidState(
644 INVALID_CHARGE_STATE,
645 ))
646}
647
648pub(crate) async fn transition_charge(
649 connection: &mut PgConnection,
650 charge_id: Uuid,
651 progression: ProcessorChargeProgression,
652 resolution_code: Option<PaymentResolutionCode>,
653) -> Result<(), ProcessorChargeStoreError> {
654 const EXPECTED: &[ProcessorChargeProgression] = &[
655 ProcessorChargeProgression::Pending,
656 ProcessorChargeProgression::ReconciliationRequired,
657 ProcessorChargeProgression::ExternalReversalRequired,
658 ProcessorChargeProgression::Applied,
659 ];
660 transition_processor_charge(
661 connection,
662 ProcessorChargeId::new(charge_id),
663 EXPECTED,
664 progression,
665 resolution_code.map(ProcessorChargeStateCode::PaymentResolution),
666 true,
667 )
668 .await?;
669 Ok(())
670}
671
672async fn transition_processor_charge(
673 connection: &mut PgConnection,
674 charge_id: ProcessorChargeId,
675 expected_progressions: &[ProcessorChargeProgression],
676 progression: ProcessorChargeProgression,
677 state_code: Option<ProcessorChargeStateCode>,
678 preserve_existing_state_code: bool,
679) -> Result<ProcessorCharge, ProcessorChargeStoreError> {
680 if expected_progressions.is_empty() {
681 return Err(ProcessorChargeStoreError::InvalidState(
682 "processor charge transition requires an expected state",
683 ));
684 }
685 let expected_progressions = expected_progressions
686 .iter()
687 .map(|progression| progression.as_str())
688 .collect::<Vec<_>>();
689 let row = sqlx::query(
690 r#"
691 UPDATE billing_processor_charges
692 SET progression_state = $3,
693 state_code = CASE WHEN $5 THEN COALESCE(state_code, $4) ELSE $4 END,
694 reconciliation_required_at = CASE WHEN $3 = 'reconciliation_required'
695 THEN COALESCE(reconciliation_required_at, clock_timestamp())
696 ELSE NULL END,
697 external_reversal_required_at = CASE WHEN $3 = 'external_reversal_required'
698 THEN COALESCE(external_reversal_required_at, clock_timestamp())
699 ELSE NULL END,
700 applied_at = CASE WHEN $3 = 'applied'
701 THEN COALESCE(applied_at, clock_timestamp()) ELSE NULL END,
702 externally_reversed_at = CASE WHEN $3 = 'externally_reversed'
703 THEN COALESCE(externally_reversed_at, clock_timestamp()) ELSE NULL END,
704 updated_at = clock_timestamp()
705 WHERE id = $1 AND progression_state = ANY($2::text[])
706 AND (
707 $3 <> ALL(ARRAY['external_reversal_required'::text, 'externally_reversed'::text])
708 OR (
709 billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NOT NULL
710 AND amount_cents > 0
711 AND attempt_kind <> 'subscription_payment_method_update'
712 )
713 )
714 AND (
715 $3 <> 'applied'
716 OR (
717 charge_role = 'primary'
718 AND billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NOT NULL
719 )
720 )
721 RETURNING id, attempt_id, billing_scope_id, gateway_account_id,
722 gateway_order_id, attempt_kind, amount_cents, currency,
723 charge_role, progression_state, state_code,
724 gateway_transaction_id, gateway_payment_method_reference,
725 gateway_response, gateway_response_code, gateway_response_text,
726 gateway_condition, payment_type, card_brand, card_last4,
727 card_exp_month, card_exp_year, observed_at
728 "#,
729 )
730 .bind(charge_id.as_uuid())
731 .bind(expected_progressions)
732 .bind(progression.as_str())
733 .bind(state_code.map(ProcessorChargeStateCode::as_str))
734 .bind(preserve_existing_state_code)
735 .fetch_optional(&mut *connection)
736 .await?
737 .ok_or(ProcessorChargeStoreError::InvalidState(
738 "processor charge transition did not match its expected state or eligibility",
739 ))?;
740 processor_charge_from_row(&row).map_err(map_operator_error)
741}
742
743async fn matching_charge(
744 connection: &mut PgConnection,
745 attempt: &PaymentAttempt,
746 evidence: &ProcessorEvidence,
747 transaction_id: Option<&str>,
748) -> Result<Option<sqlx::postgres::PgRow>, sqlx::Error> {
749 let descriptor = evidence.descriptor();
750 sqlx::query(
751 r#"
752 SELECT id, charge_role,
753 gateway_payment_method_reference IS NOT DISTINCT FROM $3
754 AND gateway_response IS NOT DISTINCT FROM $4
755 AND gateway_response_code IS NOT DISTINCT FROM $5
756 AND gateway_response_text IS NOT DISTINCT FROM $6
757 AND gateway_condition IS NOT DISTINCT FROM $7
758 AND payment_type IS NOT DISTINCT FROM $8
759 AND card_brand IS NOT DISTINCT FROM $9
760 AND card_last4 IS NOT DISTINCT FROM $10
761 AND card_exp_month IS NOT DISTINCT FROM $11
762 AND card_exp_year IS NOT DISTINCT FROM $12 AS evidence_matches
763 FROM billing_processor_charges
764 WHERE attempt_id = $1 AND gateway_transaction_id IS NOT DISTINCT FROM $2
765 FOR UPDATE
766 "#,
767 )
768 .bind(attempt.identity().attempt_id().as_uuid())
769 .bind(transaction_id)
770 .bind(
771 evidence
772 .payment_method_reference()
773 .map(|value| value.expose()),
774 )
775 .bind(evidence.response().map(GatewayDiagnostic::expose))
776 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
777 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
778 .bind(evidence.condition().map(GatewayDiagnostic::expose))
779 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
780 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
781 .bind(descriptor.card_last_four().map(|value| value.expose()))
782 .bind(descriptor.card_exp_month())
783 .bind(descriptor.card_exp_year())
784 .fetch_optional(connection)
785 .await
786}
787
788async fn owned_by_other_attempt(
789 connection: &mut PgConnection,
790 attempt: &PaymentAttempt,
791 transaction_id: &str,
792) -> Result<bool, sqlx::Error> {
793 sqlx::query_scalar(
794 r#"
795 SELECT EXISTS (
796 SELECT 1 FROM billing_processor_charges
797 WHERE gateway_account_id = $1 AND gateway_transaction_id = $2
798 AND attempt_id <> $3
799 )
800 "#,
801 )
802 .bind(attempt.identity().gateway_account_id().as_uuid())
803 .bind(transaction_id)
804 .bind(attempt.identity().attempt_id().as_uuid())
805 .fetch_one(connection)
806 .await
807}
808
809async fn identify_transactionless(
810 connection: &mut PgConnection,
811 attempt: &PaymentAttempt,
812 evidence: &ProcessorEvidence,
813 transaction_id: &str,
814 requested_progression: ProcessorChargeProgression,
815) -> Result<Option<ChargeRecord>, ProcessorChargeStoreError> {
816 let descriptor = evidence.descriptor();
817 let row = sqlx::query(
818 r#"
819 SELECT id, charge_role,
820 gateway_payment_method_reference IS NOT DISTINCT FROM $2
821 AND gateway_response IS NOT DISTINCT FROM $3
822 AND gateway_response_code IS NOT DISTINCT FROM $4
823 AND gateway_response_text IS NOT DISTINCT FROM $5
824 AND gateway_condition IS NOT DISTINCT FROM $6
825 AND payment_type IS NOT DISTINCT FROM $7
826 AND card_brand IS NOT DISTINCT FROM $8
827 AND card_last4 IS NOT DISTINCT FROM $9
828 AND card_exp_month IS NOT DISTINCT FROM $10
829 AND card_exp_year IS NOT DISTINCT FROM $11 AS evidence_matches
830 FROM billing_processor_charges
831 WHERE attempt_id = $1
832 AND billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NULL
833 FOR UPDATE
834 "#,
835 )
836 .bind(attempt.identity().attempt_id().as_uuid())
837 .bind(
838 evidence
839 .payment_method_reference()
840 .map(|value| value.expose()),
841 )
842 .bind(evidence.response().map(GatewayDiagnostic::expose))
843 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
844 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
845 .bind(evidence.condition().map(GatewayDiagnostic::expose))
846 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
847 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
848 .bind(descriptor.card_last_four().map(|value| value.expose()))
849 .bind(descriptor.card_exp_month())
850 .bind(descriptor.card_exp_year())
851 .fetch_optional(&mut *connection)
852 .await?;
853 let Some(row) = row else {
854 return Ok(None);
855 };
856 if !row.try_get::<bool, _>("evidence_matches")? {
857 return Ok(None);
858 }
859 let id = row.try_get("id")?;
860 let role = parse_role(&row.try_get::<String, _>("charge_role")?)?;
861 sqlx::query(
862 "UPDATE billing_processor_charges SET gateway_transaction_id = $2, updated_at = clock_timestamp() WHERE id = $1",
863 )
864 .bind(id)
865 .bind(transaction_id)
866 .execute(&mut *connection)
867 .await?;
868 let progression = initial_charge_progression(
869 role,
870 attempt.request().amount().cents(),
871 true,
872 requested_progression,
873 );
874 sqlx::query(
875 r#"
876 UPDATE billing_processor_charges
877 SET progression_state = $2, state_code = $3,
878 reconciliation_required_at = CASE WHEN $2 = 'reconciliation_required'
879 THEN COALESCE(reconciliation_required_at, clock_timestamp())
880 ELSE NULL END,
881 external_reversal_required_at = CASE WHEN $2 = 'external_reversal_required'
882 THEN COALESCE(external_reversal_required_at, clock_timestamp())
883 ELSE NULL END,
884 applied_at = CASE WHEN $2 = 'applied'
885 THEN COALESCE(applied_at, clock_timestamp()) ELSE NULL END,
886 updated_at = clock_timestamp()
887 WHERE id = $1
888 "#,
889 )
890 .bind(id)
891 .bind(progression.as_str())
892 .bind(Option::<&str>::None)
893 .execute(&mut *connection)
894 .await?;
895 Ok(Some(ChargeRecord {
896 id,
897 role,
898 exact_replay: false,
899 }))
900}
901
902async fn charge_by_id(
903 connection: &mut PgConnection,
904 charge_id: Uuid,
905) -> Result<ProcessorCharge, ProcessorChargeStoreError> {
906 let row = sqlx::query(
907 r#"
908 SELECT id, attempt_id, billing_scope_id, gateway_account_id,
909 gateway_order_id, attempt_kind, amount_cents, currency,
910 charge_role, progression_state, state_code,
911 gateway_transaction_id, gateway_payment_method_reference,
912 gateway_response, gateway_response_code, gateway_response_text,
913 gateway_condition, payment_type, card_brand, card_last4,
914 card_exp_month, card_exp_year, observed_at
915 FROM billing_processor_charges WHERE id = $1 FOR UPDATE
916 "#,
917 )
918 .bind(charge_id)
919 .fetch_one(connection)
920 .await?;
921 processor_charge_from_row(&row).map_err(map_operator_error)
922}
923
924fn compensating_progression(
925 attempt: &PaymentAttempt,
926 evidence: &ProcessorEvidence,
927) -> ProcessorChargeProgression {
928 if attempt.status() == PaymentAttemptStatus::Approved {
929 ProcessorChargeProgression::Applied
930 } else if attempt.request().amount().cents() > 0
931 && evidence.transaction_id().is_some()
932 && (matches!(
933 attempt.status(),
934 PaymentAttemptStatus::Declined | PaymentAttemptStatus::Failed
935 ) || attempt.state().resolution_code()
936 == Some(PaymentResolutionCode::SubscriptionInitialCurrentGrantConflict)
937 || is_external_reversal_terminal_attempt(attempt))
938 {
939 ProcessorChargeProgression::ExternalReversalRequired
940 } else if matches!(
941 attempt.status(),
942 PaymentAttemptStatus::Declined
943 | PaymentAttemptStatus::Failed
944 | PaymentAttemptStatus::ReviewRequired
945 ) {
946 ProcessorChargeProgression::ReconciliationRequired
947 } else {
948 ProcessorChargeProgression::Pending
949 }
950}
951
952fn is_external_reversal_terminal_attempt(attempt: &PaymentAttempt) -> bool {
953 attempt.status() == PaymentAttemptStatus::Failed
954 && matches!(
955 attempt.state().resolution_code(),
956 Some(
957 PaymentResolutionCode::SubscriptionInitialExternallyRefunded
958 | PaymentResolutionCode::SubscriptionInitialExternallyVoided
959 | PaymentResolutionCode::ProcessorChargeExternallyRefunded
960 | PaymentResolutionCode::ProcessorChargeExternallyVoided
961 )
962 )
963}
964
965fn initial_charge_progression(
966 role: ProcessorChargeRole,
967 amount_cents: i32,
968 identified: bool,
969 requested: ProcessorChargeProgression,
970) -> ProcessorChargeProgression {
971 if !identified {
972 return ProcessorChargeProgression::ReconciliationRequired;
973 }
974 match role {
975 ProcessorChargeRole::Primary => requested,
976 ProcessorChargeRole::Additional if amount_cents > 0 => {
977 ProcessorChargeProgression::ExternalReversalRequired
978 }
979 ProcessorChargeRole::Additional => ProcessorChargeProgression::ReconciliationRequired,
980 }
981}
982
983fn initial_charge_state_code(
984 role: ProcessorChargeRole,
985 progression: ProcessorChargeProgression,
986 identified: bool,
987) -> Option<ProcessorChargeStateCode> {
988 match (role, progression, identified) {
989 (
990 ProcessorChargeRole::Additional,
991 ProcessorChargeProgression::ExternalReversalRequired,
992 true,
993 ) => Some(ProcessorChargeStateCode::AdditionalApprovedChargeIdentified),
994 (_, ProcessorChargeProgression::ReconciliationRequired, false) => {
995 Some(ProcessorChargeStateCode::TransactionIdentityRequired)
996 }
997 (
998 ProcessorChargeRole::Primary,
999 ProcessorChargeProgression::ReconciliationRequired,
1000 true,
1001 ) => Some(ProcessorChargeStateCode::ApprovedChargeWaitingForApplication),
1002 (
1003 ProcessorChargeRole::Additional,
1004 ProcessorChargeProgression::ReconciliationRequired,
1005 true,
1006 ) => Some(ProcessorChargeStateCode::ZeroAmountAdditionalApprovedCharge),
1007 _ => None,
1008 }
1009}
1010
1011fn parse_role(value: &str) -> Result<ProcessorChargeRole, ProcessorChargeStoreError> {
1012 match value {
1013 "primary" => Ok(ProcessorChargeRole::Primary),
1014 "additional" => Ok(ProcessorChargeRole::Additional),
1015 _ => Err(ProcessorChargeStoreError::InvalidState(
1016 INVALID_CHARGE_STATE,
1017 )),
1018 }
1019}
1020
1021fn is_transient(error: &ProcessorChargeStoreError) -> bool {
1022 let database_error = match error {
1023 ProcessorChargeStoreError::Sql(sqlx::Error::Database(error))
1024 | ProcessorChargeStoreError::Attempt(PaymentAttemptStoreError::Sql(
1025 sqlx::Error::Database(error),
1026 )) => error,
1027 _ => return false,
1028 };
1029 matches!(
1030 database_error.code().as_deref(),
1031 Some("40001" | "40P01" | "55P03" | "57014")
1032 )
1033}
1034
1035fn map_operator_error(error: OperatorReviewError) -> ProcessorChargeStoreError {
1036 match error {
1037 OperatorReviewError::Sql(error) => ProcessorChargeStoreError::Sql(error),
1038 OperatorReviewError::InvalidState(message) => {
1039 ProcessorChargeStoreError::InvalidState(message)
1040 }
1041 OperatorReviewError::Host(_)
1042 | OperatorReviewError::ManualFailureHost(_)
1043 | OperatorReviewError::BillingTransaction(_)
1044 | OperatorReviewError::BillingEvent(_) => {
1045 ProcessorChargeStoreError::InvalidState(INVALID_CHARGE_STATE)
1046 }
1047 }
1048}
1049
1050#[cfg(test)]
1051mod tests {
1052 use std::error::Error;
1053
1054 use syrup_rail::{
1055 BillingScopeId, CurrencyCode, GatewayAccountId, GatewayConfigurationId, GatewayDiagnostic,
1056 GatewayPaymentDescriptor, GatewayTransactionId, PlanKey, SubscriberId,
1057 };
1058
1059 use super::*;
1060 use crate::test_support::{GatewayAccountFixture, TestDatabase, create_gateway_account};
1061
1062 async fn insert_host_charge_attempt(
1063 pool: &PgPool,
1064 gateway: GatewayAccountFixture,
1065 gateway_order_id: &str,
1066 ) -> Result<PaymentAttemptId, sqlx::Error> {
1067 let attempt_id = PaymentAttemptId::new(Uuid::now_v7());
1068 let host_charge_target_id = Uuid::now_v7();
1069 sqlx::query(
1070 r#"
1071 INSERT INTO billing_payment_attempts (
1072 id, billing_scope_id, subscriber_id, host_charge_target_id,
1073 attempt_kind, status, idempotency_key, request_fingerprint,
1074 amount_cents, currency, gateway_account_id,
1075 gateway_configuration_id, gateway_order_id
1076 ) VALUES (
1077 $1, $2, $3, $4, 'host_charge', 'pending', $5, $6,
1078 100, 'USD', $7, $8, $9
1079 )
1080 "#,
1081 )
1082 .bind(attempt_id.as_uuid())
1083 .bind(gateway.billing_scope_id)
1084 .bind(Uuid::now_v7())
1085 .bind(host_charge_target_id)
1086 .bind(format!("charge-test-{attempt_id}"))
1087 .bind(format!("host_charge:{host_charge_target_id}:100:USD"))
1088 .bind(gateway.gateway_account_id)
1089 .bind(gateway.gateway_configuration_id)
1090 .bind(gateway_order_id)
1091 .execute(pool)
1092 .await?;
1093 Ok(attempt_id)
1094 }
1095
1096 fn approved_evidence(transaction_id: &str) -> ProcessorEvidence {
1097 ProcessorEvidence::new(
1098 Some(GatewayTransactionId::new(transaction_id).unwrap()),
1099 None,
1100 Some(GatewayDiagnostic::new("1")),
1101 Some(GatewayDiagnostic::new("100")),
1102 Some(GatewayDiagnostic::new("Approved")),
1103 Some(GatewayDiagnostic::new("complete")),
1104 GatewayPaymentDescriptor::default(),
1105 )
1106 }
1107
1108 struct SubscriptionAttemptFixture {
1109 identity: PaymentAttemptIdentity,
1110 plan_key: PlanKey,
1111 gateway_order_id: GatewayOrderId,
1112 }
1113
1114 async fn insert_subscription_attempt(
1115 pool: &PgPool,
1116 gateway: GatewayAccountFixture,
1117 attempt_kind: PaymentAttemptKind,
1118 gateway_order: &str,
1119 ) -> Result<SubscriptionAttemptFixture, Box<dyn Error>> {
1120 assert!(matches!(
1121 attempt_kind,
1122 PaymentAttemptKind::SubscriptionRecovery
1123 | PaymentAttemptKind::SubscriptionPaymentMethodUpdate
1124 ));
1125 let attempt_id = PaymentAttemptId::new(Uuid::now_v7());
1126 let subscriber_id = SubscriberId::new(Uuid::now_v7());
1127 let payment_method_id = Uuid::now_v7();
1128 let subscription_id = Uuid::now_v7();
1129 let plan_key = PlanKey::new("fallback-plan")?;
1130 let gateway_order_id = GatewayOrderId::from_correlation(gateway_order)?;
1131 sqlx::query(
1132 r#"
1133 WITH input AS (
1134 SELECT
1135 $1::uuid AS attempt_id,
1136 $2::uuid AS billing_scope_id,
1137 $3::uuid AS subscriber_id,
1138 $4::uuid AS gateway_account_id,
1139 $5::uuid AS gateway_configuration_id,
1140 $6::text AS plan_key,
1141 $7::text AS gateway_order_id,
1142 $8::text AS attempt_kind,
1143 $9::uuid AS payment_method_id,
1144 $10::uuid AS subscription_id,
1145 clock_timestamp() AS period_start_at
1146 ), payment_method AS (
1147 INSERT INTO billing_payment_methods (
1148 id, billing_scope_id, subscriber_id, gateway_account_id,
1149 gateway_payment_method_reference, status
1150 )
1151 SELECT payment_method_id, billing_scope_id, subscriber_id,
1152 gateway_account_id, 'method-' || payment_method_id::text, 'active'
1153 FROM input
1154 RETURNING id
1155 ), subscription AS (
1156 INSERT INTO billing_subscriptions (
1157 id, billing_scope_id, subscriber_id, plan_key, status,
1158 gateway_account_id, payment_method_id, amount_cents, currency,
1159 current_period_start_at, current_period_end_at, next_renewal_at,
1160 initial_transaction_id
1161 )
1162 SELECT subscription_id, billing_scope_id, subscriber_id, plan_key, 'active',
1163 gateway_account_id, payment_method_id, 100, 'USD', period_start_at,
1164 period_start_at + interval '30 days',
1165 period_start_at + interval '30 days',
1166 'initial-' || subscription_id::text
1167 FROM input CROSS JOIN payment_method
1168 RETURNING id
1169 )
1170 INSERT INTO billing_payment_attempts (
1171 id, billing_scope_id, subscriber_id, plan_key, subscription_id,
1172 payment_method_id, attempt_kind, status, idempotency_key,
1173 request_fingerprint, amount_cents, currency, billing_period_start_at,
1174 billing_period_end_at, gateway_account_id, gateway_configuration_id,
1175 gateway_order_id, payment_method_update_expected_payment_method_id,
1176 payment_method_update_expected_initial_transaction_id,
1177 subscription_expected_payment_method_id,
1178 subscription_expected_initial_transaction_id, subscription_expected_status
1179 )
1180 SELECT attempt_id, billing_scope_id, subscriber_id, plan_key, subscription_id,
1181 payment_method_id, attempt_kind, 'pending', 'fallback-' || attempt_id::text,
1182 attempt_kind || ':' || subscription_id::text,
1183 CASE WHEN attempt_kind = 'subscription_payment_method_update' THEN 0 ELSE 100 END,
1184 'USD',
1185 CASE WHEN attempt_kind = 'subscription_recovery' THEN period_start_at END,
1186 CASE WHEN attempt_kind = 'subscription_recovery'
1187 THEN period_start_at + interval '30 days' END,
1188 gateway_account_id, gateway_configuration_id, gateway_order_id,
1189 CASE WHEN attempt_kind = 'subscription_payment_method_update'
1190 THEN payment_method_id END,
1191 CASE WHEN attempt_kind = 'subscription_payment_method_update'
1192 THEN 'initial-' || subscription_id::text END,
1193 CASE WHEN attempt_kind = 'subscription_recovery' THEN payment_method_id END,
1194 CASE WHEN attempt_kind = 'subscription_recovery'
1195 THEN 'initial-' || subscription_id::text END,
1196 CASE WHEN attempt_kind = 'subscription_recovery' THEN 'active' END
1197 FROM input CROSS JOIN subscription
1198 "#,
1199 )
1200 .bind(attempt_id.as_uuid())
1201 .bind(gateway.billing_scope_id)
1202 .bind(subscriber_id.as_uuid())
1203 .bind(gateway.gateway_account_id)
1204 .bind(gateway.gateway_configuration_id)
1205 .bind(plan_key.as_str())
1206 .bind(gateway_order_id.expose())
1207 .bind(attempt_kind.as_str())
1208 .bind(payment_method_id)
1209 .bind(subscription_id)
1210 .execute(pool)
1211 .await?;
1212
1213 Ok(SubscriptionAttemptFixture {
1214 identity: PaymentAttemptIdentity::new(
1215 attempt_id,
1216 BillingScopeId::new(gateway.billing_scope_id),
1217 subscriber_id,
1218 GatewayAccountId::new(gateway.gateway_account_id),
1219 GatewayConfigurationId::new(gateway.gateway_configuration_id),
1220 ),
1221 plan_key,
1222 gateway_order_id,
1223 })
1224 }
1225
1226 fn lock_free_terms(
1227 fixture: &SubscriptionAttemptFixture,
1228 attempt_kind: PaymentAttemptKind,
1229 ) -> LockFreeApprovedEvidenceTerms<'_> {
1230 let amount_cents = if attempt_kind == PaymentAttemptKind::SubscriptionPaymentMethodUpdate {
1231 0
1232 } else {
1233 100
1234 };
1235 LockFreeApprovedEvidenceTerms::subscription(
1236 fixture.identity,
1237 attempt_kind,
1238 &fixture.plan_key,
1239 &fixture.gateway_order_id,
1240 amount_cents,
1241 CurrencyCode::new("USD").expect("test currency"),
1242 )
1243 }
1244
1245 #[tokio::test]
1246 async fn lock_free_subscription_evidence_preserves_replay_and_ownership()
1247 -> Result<(), Box<dyn Error>> {
1248 let database = TestDatabase::start("rail_lock_fb").await?;
1249 let result = async {
1250 let gateway = create_gateway_account(&database.pool, "test_gateway").await?;
1251 let recovery = insert_subscription_attempt(
1252 &database.pool,
1253 gateway,
1254 PaymentAttemptKind::SubscriptionRecovery,
1255 "recovery-fallback-order",
1256 )
1257 .await?;
1258 let evidence = approved_evidence("txn_recovery_fallback");
1259 assert_eq!(
1260 persist_approved_evidence_without_attempt_lock(
1261 &database.pool,
1262 lock_free_terms(&recovery, PaymentAttemptKind::SubscriptionRecovery),
1263 &evidence,
1264 )
1265 .await?,
1266 LockFreeApprovedEvidenceOutcome::Persisted
1267 );
1268 assert_eq!(
1269 persist_approved_evidence_without_attempt_lock(
1270 &database.pool,
1271 lock_free_terms(&recovery, PaymentAttemptKind::SubscriptionRecovery),
1272 &evidence,
1273 )
1274 .await?,
1275 LockFreeApprovedEvidenceOutcome::ExactReplay
1276 );
1277 let owner = insert_subscription_attempt(
1278 &database.pool,
1279 gateway,
1280 PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
1281 "replacement-owner-order",
1282 )
1283 .await?;
1284 let contender = insert_subscription_attempt(
1285 &database.pool,
1286 gateway,
1287 PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
1288 "replacement-contender-order",
1289 )
1290 .await?;
1291 let evidence = approved_evidence("txn_replacement_owner");
1292 assert_eq!(
1293 persist_approved_evidence_without_attempt_lock(
1294 &database.pool,
1295 lock_free_terms(
1296 &owner,
1297 PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
1298 ),
1299 &evidence,
1300 )
1301 .await?,
1302 LockFreeApprovedEvidenceOutcome::Persisted
1303 );
1304 assert_eq!(
1305 persist_approved_evidence_without_attempt_lock(
1306 &database.pool,
1307 lock_free_terms(
1308 &contender,
1309 PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
1310 ),
1311 &evidence,
1312 )
1313 .await?,
1314 LockFreeApprovedEvidenceOutcome::OwnedByOtherAttempt
1315 );
1316 let dimensions: (String, Option<Uuid>, i32, String, String) = sqlx::query_as(
1317 "SELECT attempt_kind, host_charge_target_id, amount_cents, currency, progression_state FROM billing_processor_charges WHERE attempt_id = $1",
1318 )
1319 .bind(owner.identity.attempt_id().as_uuid())
1320 .fetch_one(&database.pool)
1321 .await?;
1322 assert_eq!(
1323 dimensions,
1324 (
1325 "subscription_payment_method_update".to_owned(),
1326 None,
1327 0,
1328 "USD".to_owned(),
1329 "pending".to_owned(),
1330 )
1331 );
1332 let contender_charges: i64 = sqlx::query_scalar(
1333 "SELECT count(*) FROM billing_processor_charges WHERE attempt_id = $1",
1334 )
1335 .bind(contender.identity.attempt_id().as_uuid())
1336 .fetch_one(&database.pool)
1337 .await?;
1338 assert_eq!(contender_charges, 0);
1339 Ok::<_, Box<dyn Error>>(())
1340 }
1341 .await;
1342 let cleanup = database.cleanup().await;
1343 result?;
1344 cleanup
1345 }
1346
1347 #[tokio::test]
1348 async fn compensating_store_retries_transient_database_failures() -> Result<(), Box<dyn Error>>
1349 {
1350 let database = TestDatabase::start("rail_chg_retry").await?;
1351 let result = async {
1352 let gateway = create_gateway_account(&database.pool, "test_gateway").await?;
1353 let attempt_id =
1354 insert_host_charge_attempt(&database.pool, gateway, "retry-order").await?;
1355 sqlx::raw_sql(
1356 r#"
1357 CREATE SEQUENCE processor_charge_retry_sequence;
1358 CREATE FUNCTION fail_first_processor_charge_writes()
1359 RETURNS trigger LANGUAGE plpgsql AS $$
1360 BEGIN
1361 IF nextval('processor_charge_retry_sequence') <= 2 THEN
1362 RAISE EXCEPTION 'transient test failure' USING ERRCODE = '40001';
1363 END IF;
1364 RETURN NEW;
1365 END;
1366 $$;
1367 CREATE TRIGGER fail_first_processor_charge_writes
1368 BEFORE INSERT ON billing_processor_charges
1369 FOR EACH ROW EXECUTE FUNCTION fail_first_processor_charge_writes();
1370 "#,
1371 )
1372 .execute(&database.pool)
1373 .await?;
1374
1375 let order = GatewayOrderId::from_correlation("retry-order")?;
1376 let outcome = store_compensating_processor_charge(
1377 &database.pool,
1378 attempt_id,
1379 &order,
1380 &approved_evidence("txn_retry"),
1381 )
1382 .await?;
1383 assert_eq!(outcome, CompensatingProcessorChargeOutcome::Observed);
1384 let attempts: i64 = sqlx::query_scalar(
1385 "SELECT last_value::bigint FROM processor_charge_retry_sequence",
1386 )
1387 .fetch_one(&database.pool)
1388 .await?;
1389 assert_eq!(attempts, 3);
1390 let count: i64 = sqlx::query_scalar(
1391 "SELECT count(*) FROM billing_processor_charges WHERE attempt_id = $1",
1392 )
1393 .bind(attempt_id.as_uuid())
1394 .fetch_one(&database.pool)
1395 .await?;
1396 assert_eq!(count, 1);
1397 Ok::<_, Box<dyn Error>>(())
1398 }
1399 .await;
1400 let cleanup = database.cleanup().await;
1401 result?;
1402 cleanup
1403 }
1404
1405 #[tokio::test]
1406 async fn transaction_observation_preserves_exact_replay_and_rejects_drift()
1407 -> Result<(), Box<dyn Error>> {
1408 let database = TestDatabase::start("rail_charge_tx").await?;
1409 let result = async {
1410 let gateway = create_gateway_account(&database.pool, "test_gateway").await?;
1411 let attempt_id =
1412 insert_host_charge_attempt(&database.pool, gateway, "tx-order").await?;
1413 let order = GatewayOrderId::from_correlation("tx-order")?;
1414 let evidence = approved_evidence("txn_transactional");
1415 let mut transaction = database.pool.begin().await?;
1416 let observed = observe_processor_charge_in_transaction(
1417 &mut transaction,
1418 attempt_id,
1419 &order,
1420 &evidence,
1421 ProcessorChargeProgression::ReconciliationRequired,
1422 )
1423 .await?;
1424 assert!(matches!(
1425 observed,
1426 ProcessorChargeObservationOutcome::Observed(_)
1427 ));
1428 let replay = observe_processor_charge_in_transaction(
1429 &mut transaction,
1430 attempt_id,
1431 &order,
1432 &evidence,
1433 ProcessorChargeProgression::ReconciliationRequired,
1434 )
1435 .await?;
1436 assert!(matches!(
1437 replay,
1438 ProcessorChargeObservationOutcome::ExactReplay(_)
1439 ));
1440 let mut changed = approved_evidence("txn_transactional");
1441 changed = ProcessorEvidence::new(
1442 changed.transaction_id().cloned(),
1443 changed.payment_method_reference().cloned(),
1444 changed.response().cloned(),
1445 changed.response_code().cloned(),
1446 Some(GatewayDiagnostic::new("Changed")),
1447 changed.condition().cloned(),
1448 changed.descriptor().clone(),
1449 );
1450 let drift = observe_processor_charge_in_transaction(
1451 &mut transaction,
1452 attempt_id,
1453 &order,
1454 &changed,
1455 ProcessorChargeProgression::ReconciliationRequired,
1456 )
1457 .await;
1458 assert!(matches!(
1459 drift,
1460 Err(ProcessorChargeStoreError::InvalidState(
1461 "processor charge replay evidence changed"
1462 ))
1463 ));
1464 transaction.rollback().await?;
1465 Ok::<_, Box<dyn Error>>(())
1466 }
1467 .await;
1468 let cleanup = database.cleanup().await;
1469 result?;
1470 cleanup
1471 }
1472}