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::processor_charge_persistence::{
21 attestation_by_charge, attestation_matches_source, processor_charge_from_row,
22};
23use crate::{OperatorReviewError, PaymentAttemptStoreError};
24
25use storage::{
26 charge_by_id, compensating_progression, identify_transactionless, initial_charge_progression,
27 initial_charge_state_code, is_transient, map_operator_error, matching_charge,
28 owned_by_other_attempt, parse_role,
29};
30
31mod storage;
32
33const INVALID_CHARGE_STATE: &str = "canonical processor charge state is invalid";
34const STORE_MAX_ATTEMPTS: usize = 3;
35const STORE_RETRY_DELAY: Duration = Duration::from_millis(50);
36
37#[derive(Debug, Error)]
38pub enum ProcessorChargeStoreError {
39 #[error("processor charge storage operation failed")]
40 Sql(#[from] sqlx::Error),
41 #[error("payment attempt storage operation failed")]
42 Attempt(#[from] PaymentAttemptStoreError),
43 #[error("{0}")]
44 InvalidState(&'static str),
45}
46
47#[derive(Clone, Copy, Debug, Eq, PartialEq)]
48pub enum CompensatingProcessorChargeOutcome {
49 Observed,
50 ExactReplay,
51 OwnedByOtherAttempt,
52}
53
54#[derive(Clone, Debug, Eq, PartialEq)]
55pub enum ProcessorChargeObservationOutcome {
56 Observed(ProcessorCharge),
57 ExactReplay(ProcessorCharge),
58 OwnedByOtherAttempt,
59}
60
61#[derive(Clone, Copy, Debug)]
62pub(crate) struct ChargeRecord {
63 pub(crate) id: Uuid,
64 pub(crate) role: ProcessorChargeRole,
65 exact_replay: bool,
66}
67
68#[derive(Clone, Copy, Debug)]
69pub(crate) enum ObservedCharge {
70 Owned(ChargeRecord),
71 OwnedByOtherAttempt,
72}
73
74#[derive(Clone, Copy)]
80pub(crate) struct LockFreeApprovedEvidenceTerms<'a> {
81 identity: PaymentAttemptIdentity,
82 attempt_kind: PaymentAttemptKind,
83 plan_key: &'a PlanKey,
84 gateway_order_id: &'a GatewayOrderId,
85 amount_cents: i32,
86 currency: CurrencyCode,
87 host_charge_target_id: Option<Uuid>,
88}
89
90impl<'a> LockFreeApprovedEvidenceTerms<'a> {
91 pub(crate) fn initial(reservation: &'a SubscriptionEnrollmentReservation) -> Self {
92 let charge = reservation.expected_terms().initial_charge();
93 Self::subscription(
94 reservation.identity(),
95 PaymentAttemptKind::SubscriptionInitial,
96 reservation.plan_key(),
97 reservation.gateway_order_id(),
98 charge.cents(),
99 charge.currency(),
100 )
101 }
102
103 pub(crate) fn recovery(reservation: &'a SubscriptionRecoveryReservation) -> Self {
104 let request = reservation.request();
105 let amount = request.amount();
106 Self::subscription(
107 reservation.identity(),
108 PaymentAttemptKind::SubscriptionRecovery,
109 reservation.plan_key(),
110 request.gateway_order_id(),
111 amount.cents(),
112 amount.currency(),
113 )
114 }
115
116 pub(crate) fn payment_method_replacement(
117 reservation: &'a SubscriptionPaymentMethodReplacement,
118 ) -> Self {
119 let request = reservation.request();
120 debug_assert_eq!(request.amount().cents(), 0);
121 Self::subscription(
122 reservation.identity(),
123 PaymentAttemptKind::SubscriptionPaymentMethodUpdate,
124 reservation.plan_key(),
125 request.gateway_order_id(),
126 0,
127 request.amount().currency(),
128 )
129 }
130
131 fn subscription(
132 identity: PaymentAttemptIdentity,
133 attempt_kind: PaymentAttemptKind,
134 plan_key: &'a PlanKey,
135 gateway_order_id: &'a GatewayOrderId,
136 amount_cents: i32,
137 currency: CurrencyCode,
138 ) -> Self {
139 debug_assert!(matches!(
140 attempt_kind,
141 PaymentAttemptKind::SubscriptionInitial
142 | PaymentAttemptKind::SubscriptionRecovery
143 | PaymentAttemptKind::SubscriptionPaymentMethodUpdate
144 ));
145 Self {
146 identity,
147 attempt_kind,
148 plan_key,
149 gateway_order_id,
150 amount_cents,
151 currency,
152 host_charge_target_id: None,
153 }
154 }
155}
156
157#[derive(Clone, Copy, Debug, Eq, PartialEq)]
158pub(crate) enum LockFreeApprovedEvidenceOutcome {
159 Persisted,
160 ExactReplay,
161 OwnedByOtherAttempt,
162 NotDurable,
163}
164
165pub(crate) async fn persist_approved_evidence_without_attempt_lock(
172 pool: &PgPool,
173 terms: LockFreeApprovedEvidenceTerms<'_>,
174 evidence: &ProcessorEvidence,
175) -> Result<LockFreeApprovedEvidenceOutcome, sqlx::Error> {
176 let identity = terms.identity;
177 let descriptor = evidence.descriptor();
178 let transaction_id = evidence.transaction_id().map(GatewayTransactionId::expose);
179 let mut transaction = pool.begin().await?;
180 set_enrollment_timeouts(&mut transaction).await?;
181
182 for _ in 0..2 {
183 let has_existing_charge: bool = sqlx::query_scalar(
184 "SELECT EXISTS (SELECT 1 FROM billing_processor_charges WHERE attempt_id = $1)",
185 )
186 .bind(identity.attempt_id().as_uuid())
187 .fetch_one(&mut *transaction)
188 .await?;
189 let role = if has_existing_charge {
190 "additional"
191 } else {
192 "primary"
193 };
194 let inserted = sqlx::query_scalar::<_, Uuid>(
195 r#"
196 INSERT INTO billing_processor_charges (
197 id, attempt_id, billing_scope_id, gateway_account_id, gateway_order_id,
198 gateway_transaction_id, gateway_payment_method_reference,
199 gateway_response, gateway_response_code, gateway_response_text,
200 gateway_condition, payment_type, card_brand, card_last4,
201 card_exp_month, card_exp_year, charge_role, progression_state,
202 attempt_kind, plan_key, host_charge_target_id, amount_cents, currency
203 ) VALUES (
204 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
205 $14, $15, $16, $17, 'pending', $18, $19, $20, $21, $22
206 )
207 ON CONFLICT DO NOTHING
208 RETURNING id
209 "#,
210 )
211 .bind(Uuid::now_v7())
212 .bind(identity.attempt_id().as_uuid())
213 .bind(identity.billing_scope_id().as_uuid())
214 .bind(identity.gateway_account_id().as_uuid())
215 .bind(terms.gateway_order_id.expose())
216 .bind(transaction_id)
217 .bind(
218 evidence
219 .payment_method_reference()
220 .map(|value| value.expose()),
221 )
222 .bind(evidence.response().map(GatewayDiagnostic::expose))
223 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
224 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
225 .bind(evidence.condition().map(GatewayDiagnostic::expose))
226 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
227 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
228 .bind(descriptor.card_last_four().map(|value| value.expose()))
229 .bind(descriptor.card_exp_month())
230 .bind(descriptor.card_exp_year())
231 .bind(role)
232 .bind(terms.attempt_kind.as_str())
233 .bind(terms.plan_key.as_str())
234 .bind(terms.host_charge_target_id)
235 .bind(terms.amount_cents)
236 .bind(terms.currency.as_str())
237 .fetch_optional(&mut *transaction)
238 .await?;
239 if inserted.is_some() {
240 transaction.commit().await?;
241 return Ok(LockFreeApprovedEvidenceOutcome::Persisted);
242 }
243 }
244
245 let evidence_matches = sqlx::query_scalar::<_, bool>(
246 r#"
247 SELECT gateway_payment_method_reference IS NOT DISTINCT FROM $3
248 AND gateway_response IS NOT DISTINCT FROM $4
249 AND gateway_response_code IS NOT DISTINCT FROM $5
250 AND gateway_response_text IS NOT DISTINCT FROM $6
251 AND gateway_condition IS NOT DISTINCT FROM $7
252 AND payment_type IS NOT DISTINCT FROM $8
253 AND card_brand IS NOT DISTINCT FROM $9
254 AND card_last4 IS NOT DISTINCT FROM $10
255 AND card_exp_month IS NOT DISTINCT FROM $11
256 AND card_exp_year IS NOT DISTINCT FROM $12
257 FROM billing_processor_charges
258 WHERE attempt_id = $1 AND gateway_transaction_id IS NOT DISTINCT FROM $2
259 "#,
260 )
261 .bind(identity.attempt_id().as_uuid())
262 .bind(transaction_id)
263 .bind(
264 evidence
265 .payment_method_reference()
266 .map(|value| value.expose()),
267 )
268 .bind(evidence.response().map(GatewayDiagnostic::expose))
269 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
270 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
271 .bind(evidence.condition().map(GatewayDiagnostic::expose))
272 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
273 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
274 .bind(descriptor.card_last_four().map(|value| value.expose()))
275 .bind(descriptor.card_exp_month())
276 .bind(descriptor.card_exp_year())
277 .fetch_optional(&mut *transaction)
278 .await?;
279 if evidence_matches == Some(true) {
280 transaction.commit().await?;
281 return Ok(LockFreeApprovedEvidenceOutcome::ExactReplay);
282 }
283 if let Some(transaction_id) = transaction_id {
284 let owned_elsewhere: bool = sqlx::query_scalar(
285 r#"
286 SELECT EXISTS (
287 SELECT 1 FROM billing_processor_charges
288 WHERE gateway_account_id = $1 AND gateway_transaction_id = $2
289 AND attempt_id <> $3
290 )
291 "#,
292 )
293 .bind(identity.gateway_account_id().as_uuid())
294 .bind(transaction_id)
295 .bind(identity.attempt_id().as_uuid())
296 .fetch_one(&mut *transaction)
297 .await?;
298 if owned_elsewhere {
299 transaction.commit().await?;
300 return Ok(LockFreeApprovedEvidenceOutcome::OwnedByOtherAttempt);
301 }
302 }
303 Ok(LockFreeApprovedEvidenceOutcome::NotDurable)
304}
305
306pub async fn store_compensating_processor_charge(
307 pool: &PgPool,
308 attempt_id: PaymentAttemptId,
309 gateway_order_id: &GatewayOrderId,
310 evidence: &ProcessorEvidence,
311) -> Result<CompensatingProcessorChargeOutcome, ProcessorChargeStoreError> {
312 let mut last_transient_error = None;
313 for store_attempt in 1..=STORE_MAX_ATTEMPTS {
314 match store_once(pool, attempt_id, gateway_order_id, evidence).await {
315 Ok(outcome) => return Ok(outcome),
316 Err(error) if is_transient(&error) && store_attempt < STORE_MAX_ATTEMPTS => {
317 last_transient_error = Some(error);
318 tokio::time::sleep(STORE_RETRY_DELAY).await;
319 }
320 Err(error) => return Err(error),
321 }
322 }
323 Err(
324 last_transient_error.unwrap_or(ProcessorChargeStoreError::InvalidState(
325 "compensating processor charge retry loop exhausted",
326 )),
327 )
328}
329
330async fn store_once(
331 pool: &PgPool,
332 attempt_id: PaymentAttemptId,
333 gateway_order_id: &GatewayOrderId,
334 evidence: &ProcessorEvidence,
335) -> Result<CompensatingProcessorChargeOutcome, ProcessorChargeStoreError> {
336 let mut transaction = pool.begin().await?;
337 set_enrollment_timeouts(&mut transaction).await?;
338 let query = format!("{PAYMENT_ATTEMPT_SELECT} WHERE id = $1");
339 let row = sqlx::query(&query)
340 .bind(attempt_id.as_uuid())
341 .fetch_optional(&mut *transaction)
342 .await?;
343 let preloaded_attempt = row
344 .as_ref()
345 .map(payment_attempt_from_row)
346 .transpose()?
347 .ok_or(ProcessorChargeStoreError::InvalidState(
348 "compensating processor charge attempt was not found",
349 ))?;
350 if preloaded_attempt.kind() == PaymentAttemptKind::SubscriptionInitial {
351 let plan_key = preloaded_attempt.request().target().plan_key().ok_or(
352 ProcessorChargeStoreError::InvalidState(INVALID_CHARGE_STATE),
353 )?;
354 lock_subscription_aggregate(
355 &mut transaction,
356 preloaded_attempt.identity().subscriber_id(),
357 plan_key,
358 )
359 .await?;
360 }
361 let attempt = if preloaded_attempt.kind() == PaymentAttemptKind::SubscriptionPaymentMethodUpdate
362 {
363 lock_payment_attempt_by_id_on_connection(
364 &mut transaction,
365 preloaded_attempt.identity().billing_scope_id(),
366 attempt_id,
367 )
368 .await?
369 .ok_or(ProcessorChargeStoreError::InvalidState(
370 "compensating processor charge attempt was not found",
371 ))?
372 } else {
373 let reloaded = find_payment_attempt_by_id_on_connection(
374 &mut transaction,
375 preloaded_attempt.identity().billing_scope_id(),
376 attempt_id,
377 )
378 .await?
379 .ok_or(ProcessorChargeStoreError::InvalidState(
380 "compensating processor charge attempt was not found",
381 ))?;
382 if reloaded != preloaded_attempt {
383 return Err(ProcessorChargeStoreError::InvalidState(
384 "compensating processor charge attempt changed while its aggregate was locked",
385 ));
386 }
387 reloaded
388 };
389 if attempt.request().gateway_order_id() != gateway_order_id {
390 return Err(ProcessorChargeStoreError::InvalidState(
391 "compensating processor charge order does not match its attempt",
392 ));
393 }
394 let initial_progression = compensating_progression(&attempt, evidence);
395 let observation =
396 observe_processor_charge(&mut transaction, &attempt, evidence, initial_progression).await?;
397 let outcome = match observation {
398 ObservedCharge::OwnedByOtherAttempt => {
399 CompensatingProcessorChargeOutcome::OwnedByOtherAttempt
400 }
401 ObservedCharge::Owned(charge) => {
402 let mut persisted = charge_by_id(&mut transaction, charge.id).await?;
403 if !charge.exact_replay {
404 let state_code = initial_charge_state_code(
405 persisted.role(),
406 persisted.progression(),
407 evidence.transaction_id().is_some(),
408 );
409 if persisted.state_code() != state_code {
410 persisted = transition_processor_charge(
411 &mut transaction,
412 persisted.id(),
413 &[persisted.progression()],
414 persisted.progression(),
415 state_code,
416 false,
417 )
418 .await?;
419 }
420 }
421 if let Some(attestation) = attestation_by_charge(&mut transaction, charge.id)
422 .await
423 .map_err(map_operator_error)?
424 && (!charge.exact_replay
425 || !attestation_matches_source(&attestation, &attempt, &persisted))
426 {
427 return Err(ProcessorChargeStoreError::InvalidState(
428 "attested processor charge was reobserved with different evidence",
429 ));
430 }
431 if charge.exact_replay {
432 CompensatingProcessorChargeOutcome::ExactReplay
433 } else {
434 CompensatingProcessorChargeOutcome::Observed
435 }
436 }
437 };
438 transaction.commit().await?;
439 Ok(outcome)
440}
441
442pub async fn observe_processor_charge_in_transaction(
443 connection: &mut PgConnection,
444 attempt_id: PaymentAttemptId,
445 gateway_order_id: &GatewayOrderId,
446 evidence: &ProcessorEvidence,
447 initial_progression: ProcessorChargeProgression,
448) -> Result<ProcessorChargeObservationOutcome, ProcessorChargeStoreError> {
449 let query = format!("{PAYMENT_ATTEMPT_SELECT} WHERE id = $1");
450 let row = sqlx::query(&query)
451 .bind(attempt_id.as_uuid())
452 .fetch_optional(&mut *connection)
453 .await?;
454 let attempt = row
455 .as_ref()
456 .map(payment_attempt_from_row)
457 .transpose()?
458 .ok_or(ProcessorChargeStoreError::InvalidState(
459 "processor charge attempt was not found",
460 ))?;
461 if attempt.request().gateway_order_id() != gateway_order_id {
462 return Err(ProcessorChargeStoreError::InvalidState(
463 "processor charge order does not match its attempt",
464 ));
465 }
466 match observe_processor_charge(connection, &attempt, evidence, initial_progression).await? {
467 ObservedCharge::OwnedByOtherAttempt => {
468 Ok(ProcessorChargeObservationOutcome::OwnedByOtherAttempt)
469 }
470 ObservedCharge::Owned(charge) => {
471 if charge.exact_replay {
472 Ok(ProcessorChargeObservationOutcome::ExactReplay(
473 charge_by_id(connection, charge.id).await?,
474 ))
475 } else {
476 let mut persisted = charge_by_id(connection, charge.id).await?;
477 let state_code = initial_charge_state_code(
478 persisted.role(),
479 persisted.progression(),
480 evidence.transaction_id().is_some(),
481 );
482 if persisted.state_code() != state_code {
483 persisted = transition_processor_charge(
484 connection,
485 persisted.id(),
486 &[persisted.progression()],
487 persisted.progression(),
488 state_code,
489 false,
490 )
491 .await?;
492 }
493 Ok(ProcessorChargeObservationOutcome::Observed(persisted))
494 }
495 }
496 }
497}
498
499pub async fn transition_processor_charge_in_transaction(
500 connection: &mut PgConnection,
501 charge_id: ProcessorChargeId,
502 expected_progressions: &[ProcessorChargeProgression],
503 progression: ProcessorChargeProgression,
504 state_code: Option<ProcessorChargeStateCode>,
505) -> Result<ProcessorCharge, ProcessorChargeStoreError> {
506 transition_processor_charge(
507 connection,
508 charge_id,
509 expected_progressions,
510 progression,
511 state_code,
512 false,
513 )
514 .await
515}
516
517pub(crate) async fn observe_processor_charge(
518 connection: &mut PgConnection,
519 attempt: &PaymentAttempt,
520 evidence: &ProcessorEvidence,
521 initial_progression: ProcessorChargeProgression,
522) -> Result<ObservedCharge, ProcessorChargeStoreError> {
523 let identity = attempt.identity();
524 let transaction_id = evidence.transaction_id().map(GatewayTransactionId::expose);
525 let has_existing_charge: bool = sqlx::query_scalar(
526 "SELECT EXISTS (SELECT 1 FROM billing_processor_charges WHERE attempt_id = $1)",
527 )
528 .bind(identity.attempt_id().as_uuid())
529 .fetch_one(&mut *connection)
530 .await?;
531 if let Some(transaction_id) = transaction_id {
532 if owned_by_other_attempt(connection, attempt, transaction_id).await? {
533 return Ok(ObservedCharge::OwnedByOtherAttempt);
534 }
535 if let Some(charge) = identify_transactionless(
536 connection,
537 attempt,
538 evidence,
539 transaction_id,
540 initial_progression,
541 )
542 .await?
543 {
544 return Ok(ObservedCharge::Owned(charge));
545 }
546 }
547 let role = if has_existing_charge {
548 ProcessorChargeRole::Additional
549 } else {
550 ProcessorChargeRole::Primary
551 };
552 let identified = transaction_id.is_some();
553 let progression = initial_charge_progression(
554 role,
555 attempt.request().amount().cents(),
556 identified,
557 initial_progression,
558 );
559 let descriptor = evidence.descriptor();
560 let conflict_clause = if identified {
561 "ON CONFLICT (gateway_account_id, gateway_transaction_id) \
562 WHERE billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NOT NULL \
563 DO NOTHING RETURNING id"
564 } else {
565 "ON CONFLICT (attempt_id) \
566 WHERE billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NULL \
567 DO NOTHING RETURNING id"
568 };
569 let insert = format!(
570 r#"
571 INSERT INTO billing_processor_charges (
572 id, attempt_id, billing_scope_id, gateway_account_id, gateway_order_id,
573 gateway_transaction_id, gateway_payment_method_reference,
574 gateway_response, gateway_response_code, gateway_response_text,
575 gateway_condition, payment_type, card_brand, card_last4,
576 card_exp_month, card_exp_year, charge_role, progression_state, state_code,
577 reconciliation_required_at, external_reversal_required_at, applied_at,
578 attempt_kind, plan_key, host_charge_target_id, amount_cents, currency
579 ) VALUES (
580 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13,
581 $14, $15, $16, $17, $18, $19,
582 CASE WHEN $18 = 'reconciliation_required' THEN clock_timestamp() END,
583 CASE WHEN $18 = 'external_reversal_required' THEN clock_timestamp() END,
584 CASE WHEN $18 = 'applied' THEN clock_timestamp() END,
585 $20, $21, $22, $23, $24
586 )
587 {conflict_clause}
588 "#
589 );
590 let inserted = sqlx::query_scalar::<_, Uuid>(&insert)
591 .bind(Uuid::now_v7())
592 .bind(identity.attempt_id().as_uuid())
593 .bind(identity.billing_scope_id().as_uuid())
594 .bind(identity.gateway_account_id().as_uuid())
595 .bind(attempt.request().gateway_order_id().expose())
596 .bind(transaction_id)
597 .bind(
598 evidence
599 .payment_method_reference()
600 .map(|value| value.expose()),
601 )
602 .bind(evidence.response().map(GatewayDiagnostic::expose))
603 .bind(evidence.response_code().map(GatewayDiagnostic::expose))
604 .bind(evidence.response_text().map(GatewayDiagnostic::expose))
605 .bind(evidence.condition().map(GatewayDiagnostic::expose))
606 .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
607 .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
608 .bind(descriptor.card_last_four().map(|value| value.expose()))
609 .bind(descriptor.card_exp_month())
610 .bind(descriptor.card_exp_year())
611 .bind(role.as_str())
612 .bind(progression.as_str())
613 .bind(Option::<&str>::None)
614 .bind(attempt.kind().as_str())
615 .bind(attempt.request().target().plan_key().map(PlanKey::as_str))
616 .bind(
617 attempt
618 .request()
619 .target()
620 .host_charge_target_id()
621 .map(|value| *value.as_uuid()),
622 )
623 .bind(attempt.request().amount().cents())
624 .bind(attempt.request().amount().currency().as_str())
625 .fetch_optional(&mut *connection)
626 .await?;
627 if let Some(id) = inserted {
628 return Ok(ObservedCharge::Owned(ChargeRecord {
629 id,
630 role,
631 exact_replay: false,
632 }));
633 }
634 if let Some(row) = matching_charge(connection, attempt, evidence, transaction_id).await? {
635 if !row.try_get::<bool, _>("evidence_matches")? {
636 return Err(ProcessorChargeStoreError::InvalidState(
637 "processor charge replay evidence changed",
638 ));
639 }
640 return Ok(ObservedCharge::Owned(ChargeRecord {
641 id: row.try_get("id")?,
642 role: parse_role(&row.try_get::<String, _>("charge_role")?)?,
643 exact_replay: true,
644 }));
645 }
646 if let Some(transaction_id) = transaction_id
647 && owned_by_other_attempt(connection, attempt, transaction_id).await?
648 {
649 return Ok(ObservedCharge::OwnedByOtherAttempt);
650 }
651 Err(ProcessorChargeStoreError::InvalidState(
652 INVALID_CHARGE_STATE,
653 ))
654}
655
656pub(crate) async fn transition_charge(
657 connection: &mut PgConnection,
658 charge_id: Uuid,
659 progression: ProcessorChargeProgression,
660 resolution_code: Option<PaymentResolutionCode>,
661) -> Result<(), ProcessorChargeStoreError> {
662 const EXPECTED: &[ProcessorChargeProgression] = &[
663 ProcessorChargeProgression::Pending,
664 ProcessorChargeProgression::ReconciliationRequired,
665 ProcessorChargeProgression::ExternalReversalRequired,
666 ProcessorChargeProgression::Applied,
667 ];
668 transition_processor_charge(
669 connection,
670 ProcessorChargeId::new(charge_id),
671 EXPECTED,
672 progression,
673 resolution_code.map(ProcessorChargeStateCode::PaymentResolution),
674 true,
675 )
676 .await?;
677 Ok(())
678}
679
680async fn transition_processor_charge(
681 connection: &mut PgConnection,
682 charge_id: ProcessorChargeId,
683 expected_progressions: &[ProcessorChargeProgression],
684 progression: ProcessorChargeProgression,
685 state_code: Option<ProcessorChargeStateCode>,
686 preserve_existing_state_code: bool,
687) -> Result<ProcessorCharge, ProcessorChargeStoreError> {
688 if expected_progressions.is_empty() {
689 return Err(ProcessorChargeStoreError::InvalidState(
690 "processor charge transition requires an expected state",
691 ));
692 }
693 let expected_progressions = expected_progressions
694 .iter()
695 .map(|progression| progression.as_str())
696 .collect::<Vec<_>>();
697 let row = sqlx::query(
698 r#"
699 UPDATE billing_processor_charges
700 SET progression_state = $3,
701 state_code = CASE WHEN $5 THEN COALESCE(state_code, $4) ELSE $4 END,
702 reconciliation_required_at = CASE WHEN $3 = 'reconciliation_required'
703 THEN COALESCE(reconciliation_required_at, clock_timestamp())
704 ELSE NULL END,
705 external_reversal_required_at = CASE WHEN $3 = 'external_reversal_required'
706 THEN COALESCE(external_reversal_required_at, clock_timestamp())
707 ELSE NULL END,
708 applied_at = CASE WHEN $3 = 'applied'
709 THEN COALESCE(applied_at, clock_timestamp()) ELSE NULL END,
710 externally_reversed_at = CASE WHEN $3 = 'externally_reversed'
711 THEN COALESCE(externally_reversed_at, clock_timestamp()) ELSE NULL END,
712 updated_at = clock_timestamp()
713 WHERE id = $1 AND progression_state = ANY($2::text[])
714 AND (
715 $3 <> ALL(ARRAY['external_reversal_required'::text, 'externally_reversed'::text])
716 OR (
717 billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NOT NULL
718 AND amount_cents > 0
719 AND attempt_kind <> 'subscription_payment_method_update'
720 )
721 )
722 AND (
723 $3 <> 'applied'
724 OR (
725 charge_role = 'primary'
726 AND billing_canonical_gateway_transaction_id(gateway_transaction_id) IS NOT NULL
727 )
728 )
729 RETURNING id, attempt_id, billing_scope_id, gateway_account_id,
730 gateway_order_id, attempt_kind, amount_cents, currency,
731 charge_role, progression_state, state_code,
732 gateway_transaction_id, gateway_payment_method_reference,
733 gateway_response, gateway_response_code, gateway_response_text,
734 gateway_condition, payment_type, card_brand, card_last4,
735 card_exp_month, card_exp_year, observed_at
736 "#,
737 )
738 .bind(charge_id.as_uuid())
739 .bind(expected_progressions)
740 .bind(progression.as_str())
741 .bind(state_code.map(ProcessorChargeStateCode::as_str))
742 .bind(preserve_existing_state_code)
743 .fetch_optional(&mut *connection)
744 .await?
745 .ok_or(ProcessorChargeStoreError::InvalidState(
746 "processor charge transition did not match its expected state or eligibility",
747 ))?;
748 processor_charge_from_row(&row).map_err(map_operator_error)
749}
750
751#[cfg(test)]
752mod tests;