1use super::*;
2
3#[derive(Debug)]
5pub struct ExternalReversalHostStoreError {
6 source: RedactedHostErrorSource,
7}
8
9impl ExternalReversalHostStoreError {
10 pub fn new(source: impl Error + Send + Sync + 'static) -> Self {
13 Self {
14 source: RedactedHostErrorSource::new(source),
15 }
16 }
17
18 pub fn into_source(self) -> BoxError {
20 self.source.into_inner()
21 }
22}
23
24impl fmt::Display for ExternalReversalHostStoreError {
25 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
26 formatter.write_str("external reversal host target transition failed")
27 }
28}
29
30impl Error for ExternalReversalHostStoreError {}
31
32#[derive(Clone, Copy, Debug, Eq, PartialEq)]
33pub enum ExternalReversalHostTransitionOutcome {
34 Changed,
35 Unchanged,
36}
37
38#[async_trait]
39pub trait ExternalReversalHostStore: Send + Sync {
40 async fn release(
41 &self,
42 connection: &mut PgConnection,
43 release: ExternalReversalHostChargeRelease,
44 ) -> Result<ExternalReversalHostTransitionOutcome, ExternalReversalHostStoreError>;
45}
46
47#[derive(Clone, Debug, Eq, PartialEq)]
48pub enum ExternalReversalAttestationOutcome {
49 Attested {
50 attempt: PaymentAttempt,
51 attestation: ExternalReversalAttestation,
52 },
53 Replayed {
54 attempt: PaymentAttempt,
55 attestation: ExternalReversalAttestation,
56 },
57 NotFound,
58 Ineligible,
59 ReplayConflict,
60}
61
62#[derive(Clone, Debug)]
63pub(super) struct ChargeLocator {
64 pub(super) charge_id: ProcessorChargeId,
65 pub(super) attempt_id: PaymentAttemptId,
66 pub(super) billing_scope_id: BillingScopeId,
67 pub(super) subscriber_id: SubscriberId,
68 pub(super) kind: PaymentAttemptKind,
69 pub(super) plan_key: Option<PlanKey>,
70 pub(super) host_target_id: Option<HostChargeTargetId>,
71}
72
73pub async fn attest_external_reversal(
74 pool: &PgPool,
75 host: &dyn ExternalReversalHostStore,
76 processor_charge_id: ProcessorChargeId,
77 actor_id: ActorId,
78 kind: ExternalReversalKind,
79 expected_transaction_id: &GatewayTransactionId,
80 reason: &ExternalReversalReason,
81) -> Result<ExternalReversalAttestationOutcome, OperatorReviewError> {
82 let mut transaction = pool.begin().await?;
83 set_enrollment_timeouts(&mut transaction).await?;
84 let Some(locator) = charge_locator(&mut transaction, processor_charge_id).await? else {
85 transaction.commit().await?;
86 return Ok(ExternalReversalAttestationOutcome::NotFound);
87 };
88 if let Some(plan_key) = &locator.plan_key {
89 lock_subscription_aggregate(&mut transaction, locator.subscriber_id, plan_key).await?;
90 }
91 let Some(attempt) = lock_payment_attempt_by_id_on_connection(
92 &mut transaction,
93 locator.billing_scope_id,
94 locator.attempt_id,
95 )
96 .await
97 .map_err(|error| match error {
98 crate::PaymentAttemptStoreError::Sql(error) => OperatorReviewError::Sql(error),
99 crate::PaymentAttemptStoreError::InvalidState(_) => {
100 OperatorReviewError::InvalidState("locked operator review attempt is invalid")
101 }
102 })?
103 else {
104 transaction.commit().await?;
105 return Ok(ExternalReversalAttestationOutcome::NotFound);
106 };
107 let Some(charge) = lock_processor_charge(&mut transaction, processor_charge_id).await? else {
108 return Err(OperatorReviewError::InvalidState(
109 "operator review processor charge disappeared while locked",
110 ));
111 };
112 if !locator_matches(&locator, &attempt, &charge) {
113 return Err(OperatorReviewError::InvalidState(
114 "operator review locator changed while locking",
115 ));
116 }
117
118 if let Some(existing) =
119 attestation_by_charge(&mut transaction, processor_charge_id.into_uuid()).await?
120 {
121 let matches = existing.actor_id() == actor_id
122 && existing.kind() == kind
123 && existing.reason() == reason
124 && existing.gateway_transaction_id() == expected_transaction_id
125 && charge.progression() == ProcessorChargeProgression::ExternallyReversed
126 && attestation_matches_source(&existing, &attempt, &charge);
127 if matches && can_release_host_target(&attempt, &charge) {
128 release_host_target(host, &mut transaction, &attempt).await?;
129 }
130 transaction.commit().await?;
131 return Ok(if matches {
132 ExternalReversalAttestationOutcome::Replayed {
133 attempt,
134 attestation: existing,
135 }
136 } else {
137 ExternalReversalAttestationOutcome::ReplayConflict
138 });
139 }
140
141 if !processor_charge_can_attest_external_reversal(&charge)
142 || charge.evidence().transaction_id() != Some(expected_transaction_id)
143 {
144 transaction.commit().await?;
145 return Ok(ExternalReversalAttestationOutcome::Ineligible);
146 }
147
148 let prior = expected_prior_resolution_code(&attempt, &charge).to_owned();
149 let final_code = expected_final_resolution_code(attempt.kind(), kind);
150 let attested_at: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
151 .fetch_one(&mut *transaction)
152 .await?;
153 insert_attestation(
154 &mut transaction,
155 &attempt,
156 &charge,
157 actor_id,
158 kind,
159 reason,
160 &prior,
161 final_code,
162 attested_at,
163 )
164 .await?;
165 let updated = sqlx::query(
166 r#"
167 UPDATE billing_processor_charges
168 SET progression_state = 'externally_reversed',
169 state_code = $2,
170 externally_reversed_at = COALESCE(externally_reversed_at, clock_timestamp()),
171 updated_at = clock_timestamp()
172 WHERE id = $1 AND progression_state = 'external_reversal_required'
173 "#,
174 )
175 .bind(processor_charge_id.as_uuid())
176 .bind(&prior)
177 .execute(&mut *transaction)
178 .await?;
179 if updated.rows_affected() != 1 {
180 return Err(OperatorReviewError::InvalidState(
181 "eligible processor charge did not accept external reversal",
182 ));
183 }
184 if charge.role() == ProcessorChargeRole::Primary && !attempt.status().is_terminal() {
185 let updated = sqlx::query(
186 r#"
187 UPDATE billing_payment_attempts
188 SET status = 'failed', resolution_code = $2,
189 resolved_at = $3, updated_at = $3
190 WHERE id = $1 AND status = $4
191 "#,
192 )
193 .bind(attempt.identity().attempt_id().as_uuid())
194 .bind(final_code.as_str())
195 .bind(attested_at)
196 .bind(attempt.status().as_str())
197 .execute(&mut *transaction)
198 .await?;
199 if updated.rows_affected() != 1 {
200 return Err(OperatorReviewError::InvalidState(
201 "eligible attempt did not accept external reversal",
202 ));
203 }
204 }
205 if can_release_host_target(&attempt, &charge) {
206 release_host_target(host, &mut transaction, &attempt).await?;
207 }
208 let attempt = load_attempt(
209 &mut transaction,
210 locator.billing_scope_id,
211 locator.attempt_id,
212 )
213 .await?;
214 let attestation = attestation_by_charge(&mut transaction, processor_charge_id.into_uuid())
215 .await?
216 .ok_or(OperatorReviewError::InvalidState(
217 "external reversal attestation disappeared while locked",
218 ))?;
219 transaction.commit().await?;
220 Ok(ExternalReversalAttestationOutcome::Attested {
221 attempt,
222 attestation,
223 })
224}
225
226pub(super) async fn charge_locator(
227 transaction: &mut Transaction<'_, Postgres>,
228 charge_id: ProcessorChargeId,
229) -> Result<Option<ChargeLocator>, OperatorReviewError> {
230 let row = sqlx::query(
231 r#"
232 SELECT charges.id, charges.attempt_id,
233 attempts.billing_scope_id, attempts.subscriber_id,
234 attempts.attempt_kind, attempts.plan_key, attempts.host_charge_target_id
235 FROM billing_processor_charges charges
236 INNER JOIN billing_payment_attempts attempts ON attempts.id = charges.attempt_id
237 WHERE charges.id = $1
238 "#,
239 )
240 .bind(charge_id.as_uuid())
241 .fetch_optional(&mut **transaction)
242 .await?;
243 row.map(|row| {
244 Ok(ChargeLocator {
245 charge_id: ProcessorChargeId::new(row.try_get("id")?),
246 attempt_id: PaymentAttemptId::new(row.try_get("attempt_id")?),
247 billing_scope_id: BillingScopeId::new(row.try_get("billing_scope_id")?),
248 subscriber_id: SubscriberId::new(row.try_get("subscriber_id")?),
249 kind: row
250 .try_get::<String, _>("attempt_kind")?
251 .parse()
252 .map_err(|_| {
253 OperatorReviewError::InvalidState("operator locator attempt kind is invalid")
254 })?,
255 plan_key: row
256 .try_get::<Option<String>, _>("plan_key")?
257 .map(PlanKey::new)
258 .transpose()
259 .map_err(|_| {
260 OperatorReviewError::InvalidState("operator locator plan key is invalid")
261 })?,
262 host_target_id: row
263 .try_get::<Option<Uuid>, _>("host_charge_target_id")?
264 .map(HostChargeTargetId::new),
265 })
266 })
267 .transpose()
268}
269
270pub(super) async fn lock_processor_charge(
271 transaction: &mut Transaction<'_, Postgres>,
272 charge_id: ProcessorChargeId,
273) -> Result<Option<ProcessorCharge>, OperatorReviewError> {
274 let row = sqlx::query(
275 r#"
276 SELECT id, attempt_id, billing_scope_id, gateway_account_id,
277 gateway_order_id, attempt_kind, amount_cents, currency,
278 charge_role, progression_state, state_code,
279 gateway_transaction_id, gateway_payment_method_reference,
280 gateway_response, gateway_response_code, gateway_response_text,
281 gateway_condition, payment_type, card_brand, card_last4,
282 card_exp_month, card_exp_year, observed_at
283 FROM billing_processor_charges WHERE id = $1 FOR UPDATE
284 "#,
285 )
286 .bind(charge_id.as_uuid())
287 .fetch_optional(&mut **transaction)
288 .await?;
289 row.as_ref().map(processor_charge_from_row).transpose()
290}
291
292#[allow(clippy::too_many_arguments)]
293async fn insert_attestation(
294 transaction: &mut Transaction<'_, Postgres>,
295 attempt: &PaymentAttempt,
296 charge: &ProcessorCharge,
297 actor_id: ActorId,
298 kind: ExternalReversalKind,
299 reason: &ExternalReversalReason,
300 prior: &str,
301 final_code: PaymentResolutionCode,
302 attested_at: DateTime<Utc>,
303) -> Result<(), OperatorReviewError> {
304 let identity = attempt.identity();
305 let evidence = charge.evidence();
306 let descriptor = evidence.descriptor();
307 let transaction_id = evidence
308 .transaction_id()
309 .ok_or(OperatorReviewError::InvalidState(
310 "eligible charge is missing transaction identity",
311 ))?;
312 sqlx::query(
313 r#"
314 INSERT INTO billing_external_reversal_attestations (
315 processor_charge_id, attempt_id, actor_id, reversal_kind, reason,
316 prior_resolution_code, final_resolution_code,
317 gateway_account_id, gateway_configuration_id, gateway_order_id,
318 amount_cents, currency, gateway_transaction_id,
319 gateway_payment_method_reference, gateway_response, gateway_response_code,
320 gateway_response_text, gateway_condition, payment_type, card_brand,
321 card_last4, card_exp_month, card_exp_year, attested_at
322 ) VALUES (
323 $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12,
324 $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24
325 )
326 "#,
327 )
328 .bind(charge.id().as_uuid())
329 .bind(identity.attempt_id().as_uuid())
330 .bind(actor_id.as_uuid())
331 .bind(kind.as_str())
332 .bind(reason.expose())
333 .bind(prior)
334 .bind(final_code.as_str())
335 .bind(identity.gateway_account_id().as_uuid())
336 .bind(identity.gateway_configuration_id().as_uuid())
337 .bind(charge.gateway_order_id().expose())
338 .bind(charge.amount().cents())
339 .bind(charge.amount().currency().as_str())
340 .bind(transaction_id.expose())
341 .bind(
342 evidence
343 .payment_method_reference()
344 .map(|value| value.expose()),
345 )
346 .bind(evidence.response().map(|value| value.expose()))
347 .bind(evidence.response_code().map(|value| value.expose()))
348 .bind(evidence.response_text().map(|value| value.expose()))
349 .bind(evidence.condition().map(|value| value.expose()))
350 .bind(descriptor.payment_type().map(|value| value.expose()))
351 .bind(descriptor.card_brand().map(|value| value.expose()))
352 .bind(descriptor.card_last_four().map(|value| value.expose()))
353 .bind(descriptor.card_exp_month())
354 .bind(descriptor.card_exp_year())
355 .bind(attested_at)
356 .execute(&mut **transaction)
357 .await?;
358 Ok(())
359}
360
361fn locator_matches(
362 locator: &ChargeLocator,
363 attempt: &PaymentAttempt,
364 charge: &ProcessorCharge,
365) -> bool {
366 let identity = attempt.identity();
367 locator.charge_id == charge.id()
368 && locator.attempt_id == identity.attempt_id()
369 && locator.billing_scope_id == identity.billing_scope_id()
370 && locator.subscriber_id == identity.subscriber_id()
371 && locator.kind == attempt.kind()
372 && locator.plan_key.as_ref() == attempt.request().target().plan_key()
373 && locator.host_target_id == attempt.request().target().host_charge_target_id()
374 && charge.attempt_id() == identity.attempt_id()
375 && charge.billing_scope_id() == identity.billing_scope_id()
376 && charge.gateway_account_id() == identity.gateway_account_id()
377 && charge.attempt_kind() == attempt.kind()
378 && charge.gateway_order_id() == attempt.request().gateway_order_id()
379 && charge.amount() == attempt.request().amount()
380}
381
382fn processor_charge_can_attest_external_reversal(charge: &ProcessorCharge) -> bool {
383 charge.progression() == ProcessorChargeProgression::ExternalReversalRequired
384 && charge.evidence().transaction_id().is_some()
385 && matches!(
386 charge.attempt_kind(),
387 PaymentAttemptKind::HostCharge
388 | PaymentAttemptKind::SubscriptionInitial
389 | PaymentAttemptKind::SubscriptionRenewal
390 | PaymentAttemptKind::SubscriptionRecovery
391 )
392}
393
394fn can_release_host_target(attempt: &PaymentAttempt, charge: &ProcessorCharge) -> bool {
395 attempt.kind() == PaymentAttemptKind::HostCharge
396 && (matches!(
397 attempt.status(),
398 PaymentAttemptStatus::Declined | PaymentAttemptStatus::Failed
399 ) || (charge.role() == ProcessorChargeRole::Primary && !attempt.status().is_terminal()))
400}
401
402async fn release_host_target(
403 host: &dyn ExternalReversalHostStore,
404 transaction: &mut Transaction<'_, Postgres>,
405 attempt: &PaymentAttempt,
406) -> Result<(), OperatorReviewError> {
407 let target_id = attempt.request().target().host_charge_target_id().ok_or(
408 OperatorReviewError::InvalidState("host charge is missing exact target"),
409 )?;
410 let identity = attempt.identity();
411 let _ = host
412 .release(
413 &mut *transaction,
414 ExternalReversalHostChargeRelease::new(
415 identity.billing_scope_id(),
416 identity.subscriber_id(),
417 target_id,
418 ),
419 )
420 .await?;
421 Ok(())
422}
423
424async fn load_attempt(
425 transaction: &mut Transaction<'_, Postgres>,
426 scope: BillingScopeId,
427 attempt_id: PaymentAttemptId,
428) -> Result<PaymentAttempt, OperatorReviewError> {
429 let query = format!(
430 "{} WHERE billing_scope_id = $1 AND id = $2",
431 crate::attempts::PAYMENT_ATTEMPT_SELECT
432 );
433 let row = sqlx::query(&query)
434 .bind(scope.as_uuid())
435 .bind(attempt_id.as_uuid())
436 .fetch_optional(&mut **transaction)
437 .await?
438 .ok_or(OperatorReviewError::InvalidState(
439 "operator review attempt disappeared",
440 ))?;
441 payment_attempt_from_row(&row).map_err(|_| {
442 OperatorReviewError::InvalidState("post-attestation payment attempt is invalid")
443 })
444}