1use chrono::{DateTime, Utc};
2use sqlx::{PgPool, Postgres, Row, Transaction};
3use syrup_rail::{
4 BillingScopeId, CumulativeRefundCents, GatewayLifecycleAccount, GatewayLifecycleCursorKey,
5 GatewayLifecycleEvidence, GatewayLifecycleQuarantine, GatewayLifecycleQuarantineReason,
6 GatewayLifecycleState, GatewayOrderId, GatewayTransactionId, GatewayTransactionReport,
7 HostChargeTargetId, HostChargeTargetTransition, HostChargeTargetTransitionKind,
8 PaymentAttemptId, PaymentAttemptKind, SubscriberId,
9};
10use thiserror::Error;
11use uuid::Uuid;
12
13const ROW_LOCK_TIMEOUT: &str = "250ms";
14const OPERATION_TIMEOUT: &str = "5s";
15const PENDING_RETENTION_SECONDS: i64 = 7 * 24 * 60 * 60;
16const PENDING_MAX_CHECKS: i32 = 24;
17const PENDING_CLEANUP_BATCH_SIZE: i64 = 500;
18const STAGED_APPLICATION_BATCH_SIZE: i64 = 100;
19const INVALID_STORED_STATE: &str = "canonical gateway lifecycle state is invalid";
20
21use crate::{HostChargeTargetError, HostChargeTargetStore};
22
23#[derive(Debug, Error)]
24pub enum GatewayLifecycleReconciliationError {
25 #[error("gateway lifecycle storage operation failed")]
26 Sql(#[from] sqlx::Error),
27 #[error("gateway lifecycle account was not found")]
28 AccountNotFound,
29 #[error("{0}")]
30 InvalidState(&'static str),
31 #[error(transparent)]
32 HostChargeTarget(#[from] HostChargeTargetError),
33}
34
35#[derive(Clone, Copy, Debug, Eq, PartialEq)]
36pub enum GatewayLifecycleApplyOutcome {
37 Applied,
38 AlreadySuperseded,
39 InvalidRefundEconomics,
40 ConflictingLifecycleEvidence,
41 HostTargetTransitionSkipped,
42 StagedAmbiguous,
43 StagedNoMatch,
44}
45
46impl GatewayLifecycleApplyOutcome {
47 pub const fn applied_count(self) -> u64 {
48 if matches!(self, Self::Applied) { 1 } else { 0 }
49 }
50
51 pub const fn staged_count(self) -> u64 {
57 if matches!(self, Self::StagedAmbiguous | Self::StagedNoMatch) {
58 1
59 } else {
60 0
61 }
62 }
63
64 pub const fn skipped_count(self) -> u64 {
65 if matches!(self, Self::HostTargetTransitionSkipped) {
66 1
67 } else {
68 0
69 }
70 }
71}
72
73#[derive(Clone, Copy, Debug, Eq, PartialEq)]
74struct EvidenceApplication {
75 outcome: GatewayLifecycleApplyOutcome,
76 newly_staged: bool,
77}
78
79#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
80pub struct GatewayLifecycleReconciliationSummary {
81 applied: u64,
82 staged: u64,
83 quarantined: u64,
84 cleaned: u64,
85 skipped: u64,
86}
87
88impl GatewayLifecycleReconciliationSummary {
89 pub const fn applied(self) -> u64 {
90 self.applied
91 }
92
93 pub const fn staged(self) -> u64 {
94 self.staged
95 }
96
97 pub const fn quarantined(self) -> u64 {
98 self.quarantined
99 }
100
101 pub const fn cleaned(self) -> u64 {
102 self.cleaned
103 }
104
105 pub const fn skipped(self) -> u64 {
106 self.skipped
107 }
108
109 fn record_application(&mut self, application: EvidenceApplication) {
110 self.applied += application.outcome.applied_count();
111 self.staged += u64::from(application.newly_staged);
112 self.skipped += application.outcome.skipped_count();
113 }
114}
115
116#[derive(Clone, Debug)]
117struct StoredEvidence {
118 transaction_id: Option<String>,
119 order_id: Option<String>,
120 state: GatewayLifecycleState,
121 condition: Option<String>,
122 action: Option<String>,
123 effective_at: Option<DateTime<Utc>>,
124}
125
126impl From<&GatewayLifecycleEvidence> for StoredEvidence {
127 fn from(evidence: &GatewayLifecycleEvidence) -> Self {
128 Self {
129 transaction_id: evidence
130 .transaction_id()
131 .map(|identifier| identifier.expose().to_owned()),
132 order_id: evidence
133 .order_id()
134 .map(|identifier| identifier.expose().to_owned()),
135 state: evidence.state().clone(),
136 condition: evidence
137 .condition()
138 .map(|diagnostic| diagnostic.expose().to_owned()),
139 action: evidence
140 .action()
141 .map(|diagnostic| diagnostic.expose().to_owned()),
142 effective_at: evidence
143 .effective_at()
144 .copied()
145 .map(postgres_timestamp_precision),
146 }
147 }
148}
149
150fn postgres_timestamp_precision(value: DateTime<Utc>) -> DateTime<Utc> {
151 value - chrono::Duration::nanoseconds(i64::from(value.timestamp_subsec_nanos() % 1_000))
152}
153
154#[derive(Debug)]
155struct AttemptCandidate {
156 id: Uuid,
157 billing_scope_id: BillingScopeId,
158 subscriber_id: SubscriberId,
159 kind: PaymentAttemptKind,
160 host_charge_target_id: Option<HostChargeTargetId>,
161 amount_cents: i32,
162 current_state: GatewayLifecycleState,
163 current_lifecycle_at: Option<DateTime<Utc>>,
164 current_refunded_amount_cents: i32,
165 matched_transaction_id: bool,
166 matched_order_id: bool,
167}
168
169#[derive(Clone, Copy, Debug, Eq, PartialEq)]
170enum LifecycleTransition {
171 Apply { refunded_amount_cents: i32 },
172 AlreadySuperseded,
173 InvalidRefundEconomics,
174 ConflictingLifecycleEvidence,
175}
176
177pub async fn gateway_lifecycle_reconciliation_start(
178 pool: &PgPool,
179 account: &GatewayLifecycleAccount,
180 cursor_key: &GatewayLifecycleCursorKey,
181) -> Result<Option<DateTime<Utc>>, GatewayLifecycleReconciliationError> {
182 let mut transaction = pool.begin().await?;
183 set_timeouts(&mut transaction).await?;
184 ensure_account(&mut transaction, account).await?;
185 sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
186 .bind(format!(
187 "billing_reconciliation_cursor:{}:{}:{}",
188 account.gateway_account_id(),
189 account.provider_key(),
190 cursor_key
191 ))
192 .execute(&mut *transaction)
193 .await?;
194 let row = sqlx::query(
195 r#"
196 WITH cursor_value AS (
197 SELECT (
198 SELECT cursors.last_successful_end_at
199 FROM billing_reconciliation_cursors cursors
200 WHERE cursors.billing_scope_id = $1
201 AND cursors.gateway_account_id = $2
202 AND cursors.provider_key = $3
203 AND cursors.cursor_key = $4
204 FOR UPDATE
205 ) AS cursor_at
206 )
207 SELECT cursor_at,
208 CASE WHEN cursor_at IS NULL THEN (
209 SELECT MIN(COALESCE(attempts.resolved_at, attempts.updated_at))
210 FROM billing_payment_attempts attempts
211 WHERE attempts.billing_scope_id = $1
212 AND attempts.gateway_account_id = $2
213 AND attempts.status = 'approved'
214 AND (
215 public.billing_canonical_gateway_transaction_id(
216 attempts.gateway_transaction_id
217 ) IS NOT NULL
218 OR attempts.gateway_order_id IS NOT NULL
219 )
220 ) END AS first_approved_at
221 FROM cursor_value
222 "#,
223 )
224 .bind(account.billing_scope_id().as_uuid())
225 .bind(account.gateway_account_id().as_uuid())
226 .bind(account.provider_key().as_str())
227 .bind(cursor_key.as_str())
228 .fetch_one(&mut *transaction)
229 .await?;
230 let cursor_at: Option<DateTime<Utc>> = row.try_get("cursor_at")?;
231 let first_approved_at: Option<DateTime<Utc>> = row.try_get("first_approved_at")?;
232 transaction.commit().await?;
233 Ok(cursor_at.or(first_approved_at))
234}
235
236pub async fn save_gateway_lifecycle_reconciliation_cursor(
237 pool: &PgPool,
238 account: &GatewayLifecycleAccount,
239 cursor_key: &GatewayLifecycleCursorKey,
240 last_successful_end_at: DateTime<Utc>,
241) -> Result<(), GatewayLifecycleReconciliationError> {
242 let mut transaction = pool.begin().await?;
243 set_timeouts(&mut transaction).await?;
244 ensure_account(&mut transaction, account).await?;
245 sqlx::query(
246 r#"
247 INSERT INTO billing_reconciliation_cursors (
248 billing_scope_id,
249 gateway_account_id,
250 provider_key,
251 cursor_key,
252 last_successful_end_at
253 )
254 VALUES ($1, $2, $3, $4, $5)
255 ON CONFLICT (gateway_account_id, provider_key, cursor_key) DO UPDATE
256 SET last_successful_end_at = GREATEST(
257 billing_reconciliation_cursors.last_successful_end_at,
258 EXCLUDED.last_successful_end_at
259 ),
260 updated_at = now()
261 "#,
262 )
263 .bind(account.billing_scope_id().as_uuid())
264 .bind(account.gateway_account_id().as_uuid())
265 .bind(account.provider_key().as_str())
266 .bind(cursor_key.as_str())
267 .bind(last_successful_end_at)
268 .execute(&mut *transaction)
269 .await?;
270 transaction.commit().await?;
271 Ok(())
272}
273
274pub async fn reconcile_gateway_transaction_reports(
275 pool: &PgPool,
276 host_charge_targets: &dyn HostChargeTargetStore,
277 account: &GatewayLifecycleAccount,
278 reports: Vec<GatewayTransactionReport>,
279) -> Result<GatewayLifecycleReconciliationSummary, GatewayLifecycleReconciliationError> {
280 let mut summary = GatewayLifecycleReconciliationSummary::default();
281 for report in reports {
282 match report {
283 GatewayTransactionReport::Ignore => {}
284 GatewayTransactionReport::Quarantine(quarantine) => {
285 record_quarantine(pool, account, &quarantine).await?;
286 summary.quarantined += 1;
287 }
288 GatewayTransactionReport::Evidence(evidence) => {
289 let application = apply_or_stage_evidence(
290 pool,
291 host_charge_targets,
292 account,
293 StoredEvidence::from(&evidence),
294 None,
295 )
296 .await?;
297 summary.record_application(application);
298 }
299 }
300 }
301 Ok(summary)
302}
303
304pub async fn apply_gateway_lifecycle_evidence(
310 pool: &PgPool,
311 host_charge_targets: &dyn HostChargeTargetStore,
312 account: &GatewayLifecycleAccount,
313 evidence: &GatewayLifecycleEvidence,
314) -> Result<GatewayLifecycleApplyOutcome, GatewayLifecycleReconciliationError> {
315 Ok(apply_or_stage_evidence(
316 pool,
317 host_charge_targets,
318 account,
319 StoredEvidence::from(evidence),
320 None,
321 )
322 .await?
323 .outcome)
324}
325
326pub async fn stage_gateway_lifecycle_evidence(
327 pool: &PgPool,
328 account: &GatewayLifecycleAccount,
329 evidence: &GatewayLifecycleEvidence,
330) -> Result<(), GatewayLifecycleReconciliationError> {
331 let mut transaction = pool.begin().await?;
332 set_timeouts(&mut transaction).await?;
333 ensure_account(&mut transaction, account).await?;
334 stage_evidence(&mut transaction, account, &StoredEvidence::from(evidence)).await?;
335 transaction.commit().await?;
336 Ok(())
337}
338
339pub async fn record_gateway_lifecycle_quarantines(
340 pool: &PgPool,
341 account: &GatewayLifecycleAccount,
342 quarantines: &[GatewayLifecycleQuarantine],
343) -> Result<(), GatewayLifecycleReconciliationError> {
344 for quarantine in quarantines {
345 record_quarantine(pool, account, quarantine).await?;
346 }
347 Ok(())
348}
349
350pub async fn apply_staged_gateway_lifecycle_evidence(
351 pool: &PgPool,
352 host_charge_targets: &dyn HostChargeTargetStore,
353 account: &GatewayLifecycleAccount,
354) -> Result<GatewayLifecycleReconciliationSummary, GatewayLifecycleReconciliationError> {
355 let mut summary = GatewayLifecycleReconciliationSummary::default();
356 summary.cleaned += cleanup_pending(pool, account).await?;
357 record_unactionable_checks(pool, account).await?;
358 summary.cleaned += cleanup_pending(pool, account).await?;
359
360 for (pending_id, evidence) in actionable_pending(pool, account).await? {
361 let application = apply_or_stage_evidence(
362 pool,
363 host_charge_targets,
364 account,
365 evidence,
366 Some(pending_id),
367 )
368 .await?;
369 summary.record_application(application);
370 }
371 Ok(summary)
372}
373
374async fn apply_or_stage_evidence(
375 pool: &PgPool,
376 host_charge_targets: &dyn HostChargeTargetStore,
377 account: &GatewayLifecycleAccount,
378 evidence: StoredEvidence,
379 pending_id: Option<Uuid>,
380) -> Result<EvidenceApplication, GatewayLifecycleReconciliationError> {
381 let mut transaction = pool.begin().await?;
382 set_timeouts(&mut transaction).await?;
383 ensure_account(&mut transaction, account).await?;
384 let mut candidates = attempt_candidates(&mut transaction, account, &evidence).await?;
385 let candidate_count = candidates.len();
386 let selected_index = candidates
387 .iter()
388 .position(|candidate| candidate.matched_transaction_id)
389 .or_else(|| (candidate_count == 1 && candidates[0].matched_order_id).then_some(0));
390 let Some(selected_index) = selected_index else {
391 let newly_staged = if pending_id.is_none() {
392 stage_evidence(&mut transaction, account, &evidence).await?
393 } else {
394 false
395 };
396 transaction.commit().await?;
397 return Ok(EvidenceApplication {
398 outcome: if candidate_count == 0 {
399 GatewayLifecycleApplyOutcome::StagedNoMatch
400 } else {
401 GatewayLifecycleApplyOutcome::StagedAmbiguous
402 },
403 newly_staged,
404 });
405 };
406 let candidate = candidates.swap_remove(selected_index);
407 let transition = lifecycle_transition(
408 &candidate.current_state,
409 candidate.current_refunded_amount_cents,
410 candidate.current_lifecycle_at,
411 &evidence.state,
412 evidence.effective_at,
413 candidate.amount_cents,
414 )?;
415 let reconciled_at: DateTime<Utc> = sqlx::query_scalar("SELECT now()")
416 .fetch_one(&mut *transaction)
417 .await?;
418 let outcome = match transition {
419 LifecycleTransition::Apply {
420 refunded_amount_cents,
421 } => {
422 let result = sqlx::query(
423 r#"
424 UPDATE billing_payment_attempts
425 SET gateway_condition = COALESCE($2, gateway_condition),
426 gateway_lifecycle_status = $3,
427 gateway_lifecycle_action = $4,
428 gateway_lifecycle_at = GREATEST(gateway_lifecycle_at, $5),
429 refunded_amount_cents = $6,
430 gateway_lifecycle_reconciled_at = $7,
431 updated_at = $7
432 WHERE id = $1
433 AND billing_scope_id = $8
434 AND gateway_account_id = $9
435 "#,
436 )
437 .bind(candidate.id)
438 .bind(evidence.condition.as_deref())
439 .bind(lifecycle_status(&evidence.state))
440 .bind(evidence.action.as_deref())
441 .bind(evidence.effective_at)
442 .bind(refunded_amount_cents)
443 .bind(reconciled_at)
444 .bind(account.billing_scope_id().as_uuid())
445 .bind(account.gateway_account_id().as_uuid())
446 .execute(&mut *transaction)
447 .await?;
448 if result.rows_affected() != 1 {
449 return Err(GatewayLifecycleReconciliationError::InvalidState(
450 INVALID_STORED_STATE,
451 ));
452 }
453 let target_transition_applied = if let Some(kind) = evidence.state.full_reversal_kind()
454 && candidate.kind == PaymentAttemptKind::HostCharge
455 {
456 let target_id = candidate.host_charge_target_id.ok_or(
457 GatewayLifecycleReconciliationError::InvalidState(INVALID_STORED_STATE),
458 )?;
459 let reversed_at =
460 latest_time(candidate.current_lifecycle_at, evidence.effective_at)
461 .unwrap_or(reconciled_at);
462 let target_outcome = host_charge_targets
463 .apply_transition(
464 &mut transaction,
465 HostChargeTargetTransition::new(
466 candidate.billing_scope_id,
467 candidate.subscriber_id,
468 PaymentAttemptId::new(candidate.id),
469 target_id,
470 HostChargeTargetTransitionKind::Reversed { kind },
471 reversed_at,
472 ),
473 )
474 .await?;
475 if !target_outcome.is_applied() {
476 tracing::warn!(
477 target: "syrup_rail::gateway_lifecycle_reconciliation",
478 billing_scope_id = %candidate.billing_scope_id.as_uuid(),
479 subscriber_id = %candidate.subscriber_id.as_uuid(),
480 attempt_id = %candidate.id,
481 target_id = %target_id.as_uuid(),
482 ?target_outcome,
483 "host target refused a full-reversal transition; leaving lifecycle evidence unapplied"
484 );
485 }
486 target_outcome.is_applied()
487 } else {
488 true
489 };
490 if target_transition_applied {
491 GatewayLifecycleApplyOutcome::Applied
492 } else {
493 GatewayLifecycleApplyOutcome::HostTargetTransitionSkipped
494 }
495 }
496 LifecycleTransition::AlreadySuperseded => GatewayLifecycleApplyOutcome::AlreadySuperseded,
497 LifecycleTransition::InvalidRefundEconomics => {
498 GatewayLifecycleApplyOutcome::InvalidRefundEconomics
499 }
500 LifecycleTransition::ConflictingLifecycleEvidence => {
501 GatewayLifecycleApplyOutcome::ConflictingLifecycleEvidence
502 }
503 };
504
505 match outcome {
506 GatewayLifecycleApplyOutcome::Applied | GatewayLifecycleApplyOutcome::AlreadySuperseded => {
507 resolve_matching_quarantines(&mut transaction, account, &evidence).await?;
508 }
509 GatewayLifecycleApplyOutcome::InvalidRefundEconomics
510 | GatewayLifecycleApplyOutcome::ConflictingLifecycleEvidence => {
511 record_quarantine_parts(
512 &mut transaction,
513 account,
514 evidence.transaction_id.as_deref(),
515 evidence.order_id.as_deref(),
516 GatewayLifecycleQuarantineReason::InvalidRefundEconomics,
517 )
518 .await?;
519 }
520 GatewayLifecycleApplyOutcome::HostTargetTransitionSkipped => {
521 transaction.rollback().await?;
522 let newly_staged = if pending_id.is_none() {
526 let mut staging = pool.begin().await?;
527 set_timeouts(&mut staging).await?;
528 ensure_account(&mut staging, account).await?;
529 let newly_staged = stage_evidence(&mut staging, account, &evidence).await?;
530 staging.commit().await?;
531 newly_staged
532 } else {
533 false
534 };
535 return Ok(EvidenceApplication {
536 outcome,
537 newly_staged,
538 });
539 }
540 GatewayLifecycleApplyOutcome::StagedAmbiguous
541 | GatewayLifecycleApplyOutcome::StagedNoMatch => unreachable!("handled before selection"),
542 }
543 if matches!(
544 outcome,
545 GatewayLifecycleApplyOutcome::Applied
546 | GatewayLifecycleApplyOutcome::AlreadySuperseded
547 | GatewayLifecycleApplyOutcome::InvalidRefundEconomics
548 | GatewayLifecycleApplyOutcome::ConflictingLifecycleEvidence
549 ) && let Some(pending_id) = pending_id
550 {
551 delete_pending(&mut transaction, account, pending_id).await?;
552 }
553 transaction.commit().await?;
554 Ok(EvidenceApplication {
555 outcome,
556 newly_staged: false,
557 })
558}
559
560async fn attempt_candidates(
561 transaction: &mut Transaction<'_, Postgres>,
562 account: &GatewayLifecycleAccount,
563 evidence: &StoredEvidence,
564) -> Result<Vec<AttemptCandidate>, GatewayLifecycleReconciliationError> {
565 let rows = sqlx::query(
566 r#"
567 SELECT attempts.id,
568 attempts.billing_scope_id,
569 attempts.subscriber_id,
570 attempts.attempt_kind,
571 attempts.host_charge_target_id,
572 attempts.amount_cents,
573 attempts.gateway_lifecycle_status,
574 attempts.gateway_lifecycle_at,
575 attempts.refunded_amount_cents,
576 COALESCE((
577 $1::text IS NOT NULL
578 AND public.billing_canonical_gateway_transaction_id(
579 attempts.gateway_transaction_id
580 ) = $1
581 ), false) AS matched_transaction_id,
582 COALESCE((
583 $2::text IS NOT NULL
584 AND (
585 $1::text IS NULL
586 OR public.billing_canonical_gateway_transaction_id(
587 attempts.gateway_transaction_id
588 ) IS NULL
589 )
590 AND attempts.gateway_order_id = $2
591 ), false) AS matched_order_id
592 FROM billing_payment_attempts attempts
593 WHERE attempts.billing_scope_id = $3
594 AND attempts.gateway_account_id = $4
595 AND attempts.status = 'approved'
596 AND (
597 (
598 $1::text IS NOT NULL
599 AND public.billing_canonical_gateway_transaction_id(
600 attempts.gateway_transaction_id
601 ) = $1
602 )
603 OR (
604 $2::text IS NOT NULL
605 AND (
606 $1::text IS NULL
607 OR public.billing_canonical_gateway_transaction_id(
608 attempts.gateway_transaction_id
609 ) IS NULL
610 )
611 AND attempts.gateway_order_id = $2
612 )
613 )
614 ORDER BY attempts.created_at, attempts.id
615 FOR UPDATE OF attempts
616 "#,
617 )
618 .bind(evidence.transaction_id.as_deref())
619 .bind(evidence.order_id.as_deref())
620 .bind(account.billing_scope_id().as_uuid())
621 .bind(account.gateway_account_id().as_uuid())
622 .fetch_all(&mut **transaction)
623 .await?;
624 rows.into_iter()
625 .map(|row| {
626 let current_refunded_amount_cents = row.try_get("refunded_amount_cents")?;
627 let current_state = lifecycle_state_from_parts(
628 row.try_get::<String, _>("gateway_lifecycle_status")?
629 .as_str(),
630 Some(current_refunded_amount_cents),
631 true,
632 )?;
633 Ok(AttemptCandidate {
634 id: row.try_get("id")?,
635 billing_scope_id: BillingScopeId::new(row.try_get("billing_scope_id")?),
636 subscriber_id: SubscriberId::new(row.try_get("subscriber_id")?),
637 kind: row
638 .try_get::<String, _>("attempt_kind")?
639 .parse()
640 .map_err(|_| {
641 GatewayLifecycleReconciliationError::InvalidState(INVALID_STORED_STATE)
642 })?,
643 host_charge_target_id: row
644 .try_get::<Option<Uuid>, _>("host_charge_target_id")?
645 .map(HostChargeTargetId::new),
646 amount_cents: row.try_get("amount_cents")?,
647 current_state,
648 current_lifecycle_at: row.try_get("gateway_lifecycle_at")?,
649 current_refunded_amount_cents,
650 matched_transaction_id: row.try_get("matched_transaction_id")?,
651 matched_order_id: row.try_get("matched_order_id")?,
652 })
653 })
654 .collect()
655}
656
657fn lifecycle_transition(
658 current_state: &GatewayLifecycleState,
659 current_refunded_amount_cents: i32,
660 current_lifecycle_at: Option<DateTime<Utc>>,
661 incoming_state: &GatewayLifecycleState,
662 incoming_lifecycle_at: Option<DateTime<Utc>>,
663 captured_amount_cents: i32,
664) -> Result<LifecycleTransition, GatewayLifecycleReconciliationError> {
665 if !stored_amount_is_valid(
666 current_state,
667 current_refunded_amount_cents,
668 captured_amount_cents,
669 ) {
670 return Err(GatewayLifecycleReconciliationError::InvalidState(
671 INVALID_STORED_STATE,
672 ));
673 }
674 let incoming_refunded_amount_cents = lifecycle_refunded_amount(incoming_state);
675 if !incoming_amount_is_valid(incoming_state, captured_amount_cents) {
676 return Ok(LifecycleTransition::InvalidRefundEconomics);
677 }
678 let incoming_rank = lifecycle_rank(incoming_state);
679 let current_rank = lifecycle_rank(current_state);
680 let should_advance = incoming_rank > current_rank
681 || (incoming_rank == current_rank
682 && (current_lifecycle_at.is_none()
683 || incoming_lifecycle_at
684 .zip(current_lifecycle_at)
685 .is_some_and(|(incoming, current)| incoming > current)
686 || incoming_refunded_amount_cents
687 .is_some_and(|incoming| incoming > current_refunded_amount_cents)));
688 if !should_advance {
689 return Ok(LifecycleTransition::AlreadySuperseded);
690 }
691 let next_refunded_amount_cents = current_refunded_amount_cents
692 .max(incoming_refunded_amount_cents.unwrap_or(current_refunded_amount_cents));
693 if !stored_amount_is_valid(
694 incoming_state,
695 next_refunded_amount_cents,
696 captured_amount_cents,
697 ) {
698 return Ok(LifecycleTransition::ConflictingLifecycleEvidence);
699 }
700 Ok(LifecycleTransition::Apply {
701 refunded_amount_cents: next_refunded_amount_cents,
702 })
703}
704
705const fn lifecycle_rank(state: &GatewayLifecycleState) -> i16 {
706 match state {
707 GatewayLifecycleState::Unknown => 0,
708 GatewayLifecycleState::PendingSettlement => 1,
709 GatewayLifecycleState::Settled { .. } => 2,
710 GatewayLifecycleState::Voided => 3,
711 GatewayLifecycleState::Refunded { .. } => 4,
712 GatewayLifecycleState::Chargeback { .. } => 5,
713 }
714}
715
716const fn lifecycle_status(state: &GatewayLifecycleState) -> &'static str {
717 match state {
718 GatewayLifecycleState::Unknown => "unknown",
719 GatewayLifecycleState::PendingSettlement => "pending_settlement",
720 GatewayLifecycleState::Settled { .. } => "settled",
721 GatewayLifecycleState::Voided => "voided",
722 GatewayLifecycleState::Refunded { .. } => "refunded",
723 GatewayLifecycleState::Chargeback { .. } => "chargeback",
724 }
725}
726
727const fn lifecycle_refunded_amount(state: &GatewayLifecycleState) -> Option<i32> {
728 match state {
729 GatewayLifecycleState::Settled {
730 cumulative_refunded_cents,
731 }
732 | GatewayLifecycleState::Chargeback {
733 cumulative_refunded_cents,
734 } => match cumulative_refunded_cents {
735 Some(amount) => Some(amount.get()),
736 None => None,
737 },
738 GatewayLifecycleState::Refunded {
739 cumulative_refunded_cents,
740 } => Some(cumulative_refunded_cents.get()),
741 GatewayLifecycleState::Unknown
742 | GatewayLifecycleState::PendingSettlement
743 | GatewayLifecycleState::Voided => None,
744 }
745}
746
747fn incoming_amount_is_valid(state: &GatewayLifecycleState, captured: i32) -> bool {
748 match state {
749 GatewayLifecycleState::Unknown
750 | GatewayLifecycleState::PendingSettlement
751 | GatewayLifecycleState::Voided => true,
752 GatewayLifecycleState::Settled {
753 cumulative_refunded_cents,
754 } => captured > 0 && cumulative_refunded_cents.is_none_or(|amount| amount.get() < captured),
755 GatewayLifecycleState::Refunded {
756 cumulative_refunded_cents,
757 } => captured > 0 && cumulative_refunded_cents.get() == captured,
758 GatewayLifecycleState::Chargeback {
759 cumulative_refunded_cents,
760 } => {
761 captured > 0 && cumulative_refunded_cents.is_none_or(|amount| amount.get() <= captured)
762 }
763 }
764}
765
766fn stored_amount_is_valid(state: &GatewayLifecycleState, refunded: i32, captured: i32) -> bool {
767 match state {
768 GatewayLifecycleState::Unknown
769 | GatewayLifecycleState::PendingSettlement
770 | GatewayLifecycleState::Voided => refunded == 0,
771 GatewayLifecycleState::Settled { .. } => {
772 captured > 0 && refunded >= 0 && refunded < captured
773 }
774 GatewayLifecycleState::Refunded { .. } => captured > 0 && refunded == captured,
775 GatewayLifecycleState::Chargeback { .. } => {
776 captured > 0 && refunded >= 0 && refunded <= captured
777 }
778 }
779}
780
781fn lifecycle_state_from_parts(
782 status: &str,
783 refunded: Option<i32>,
784 stored_attempt: bool,
785) -> Result<GatewayLifecycleState, GatewayLifecycleReconciliationError> {
786 let state = match (status, refunded) {
787 ("unknown", None | Some(0)) => GatewayLifecycleState::Unknown,
788 ("pending_settlement", None | Some(0)) => GatewayLifecycleState::PendingSettlement,
789 ("voided", None | Some(0)) => GatewayLifecycleState::Voided,
790 ("settled", None | Some(0)) => GatewayLifecycleState::Settled {
791 cumulative_refunded_cents: None,
792 },
793 ("settled", Some(value)) if value > 0 => GatewayLifecycleState::Settled {
794 cumulative_refunded_cents: Some(CumulativeRefundCents::new(value).map_err(|_| {
795 GatewayLifecycleReconciliationError::InvalidState(INVALID_STORED_STATE)
796 })?),
797 },
798 ("refunded", Some(value)) if value > 0 => GatewayLifecycleState::Refunded {
799 cumulative_refunded_cents: CumulativeRefundCents::new(value).map_err(|_| {
800 GatewayLifecycleReconciliationError::InvalidState(INVALID_STORED_STATE)
801 })?,
802 },
803 ("chargeback", None | Some(0)) => GatewayLifecycleState::Chargeback {
804 cumulative_refunded_cents: None,
805 },
806 ("chargeback", Some(value)) if value > 0 => GatewayLifecycleState::Chargeback {
807 cumulative_refunded_cents: Some(CumulativeRefundCents::new(value).map_err(|_| {
808 GatewayLifecycleReconciliationError::InvalidState(INVALID_STORED_STATE)
809 })?),
810 },
811 _ => {
812 return Err(GatewayLifecycleReconciliationError::InvalidState(
813 INVALID_STORED_STATE,
814 ));
815 }
816 };
817 if stored_attempt && refunded.is_none() {
818 return Err(GatewayLifecycleReconciliationError::InvalidState(
819 INVALID_STORED_STATE,
820 ));
821 }
822 Ok(state)
823}
824
825async fn stage_evidence(
826 transaction: &mut Transaction<'_, Postgres>,
827 account: &GatewayLifecycleAccount,
828 evidence: &StoredEvidence,
829) -> Result<bool, GatewayLifecycleReconciliationError> {
830 let candidate_id = Uuid::now_v7();
834 let inserted = sqlx::query_scalar::<_, bool>(
835 r#"
836 INSERT INTO billing_gateway_lifecycle_pending_updates (
837 billing_scope_id,
838 gateway_account_id,
839 gateway_transaction_id,
840 gateway_order_id,
841 gateway_condition,
842 gateway_lifecycle_status,
843 gateway_lifecycle_action,
844 gateway_lifecycle_at,
845 refunded_amount_cents,
846 expires_at,
847 id
848 )
849 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9,
850 now() + ($10::bigint * interval '1 second'), $11)
851 ON CONFLICT (
852 gateway_account_id,
853 COALESCE(gateway_transaction_id, ''),
854 COALESCE(gateway_order_id, ''),
855 COALESCE(gateway_condition, ''),
856 gateway_lifecycle_status,
857 COALESCE(gateway_lifecycle_action, ''),
858 COALESCE(gateway_lifecycle_at, '-infinity'),
859 COALESCE(refunded_amount_cents, -1)
860 ) DO UPDATE SET updated_at = now()
861 RETURNING id = $11
862 "#,
863 )
864 .bind(account.billing_scope_id().as_uuid())
865 .bind(account.gateway_account_id().as_uuid())
866 .bind(evidence.transaction_id.as_deref())
867 .bind(evidence.order_id.as_deref())
868 .bind(evidence.condition.as_deref())
869 .bind(lifecycle_status(&evidence.state))
870 .bind(evidence.action.as_deref())
871 .bind(evidence.effective_at)
872 .bind(lifecycle_refunded_amount(&evidence.state))
873 .bind(PENDING_RETENTION_SECONDS)
874 .bind(candidate_id)
875 .fetch_one(&mut **transaction)
876 .await?;
877 Ok(inserted)
878}
879
880async fn record_quarantine(
881 pool: &PgPool,
882 account: &GatewayLifecycleAccount,
883 quarantine: &GatewayLifecycleQuarantine,
884) -> Result<(), GatewayLifecycleReconciliationError> {
885 let mut transaction = pool.begin().await?;
886 set_timeouts(&mut transaction).await?;
887 ensure_account(&mut transaction, account).await?;
888 record_quarantine_parts(
889 &mut transaction,
890 account,
891 quarantine
892 .transaction_id()
893 .map(GatewayTransactionId::expose),
894 quarantine.order_id().map(GatewayOrderId::expose),
895 quarantine.reason(),
896 )
897 .await?;
898 transaction.commit().await?;
899 Ok(())
900}
901
902async fn record_quarantine_parts(
903 transaction: &mut Transaction<'_, Postgres>,
904 account: &GatewayLifecycleAccount,
905 transaction_id: Option<&str>,
906 order_id: Option<&str>,
907 reason: GatewayLifecycleQuarantineReason,
908) -> Result<(), GatewayLifecycleReconciliationError> {
909 sqlx::query(
910 r#"
911 INSERT INTO billing_gateway_lifecycle_quarantines (
912 billing_scope_id,
913 gateway_account_id,
914 gateway_transaction_id,
915 gateway_order_id,
916 reason_code
917 )
918 VALUES ($1, $2, $3, $4, $5)
919 ON CONFLICT (
920 gateway_account_id,
921 COALESCE(gateway_transaction_id, ''),
922 COALESCE(gateway_order_id, ''),
923 reason_code
924 ) DO UPDATE
925 SET last_seen_at = now(),
926 occurrence_count = billing_gateway_lifecycle_quarantines.occurrence_count + 1,
927 resolved_at = NULL,
928 last_operator_alerted_at = CASE
929 WHEN billing_gateway_lifecycle_quarantines.resolved_at IS NULL
930 THEN billing_gateway_lifecycle_quarantines.last_operator_alerted_at
931 ELSE NULL
932 END
933 "#,
934 )
935 .bind(account.billing_scope_id().as_uuid())
936 .bind(account.gateway_account_id().as_uuid())
937 .bind(transaction_id)
938 .bind(order_id)
939 .bind(reason.as_str())
940 .execute(&mut **transaction)
941 .await?;
942 Ok(())
943}
944
945async fn resolve_matching_quarantines(
946 transaction: &mut Transaction<'_, Postgres>,
947 account: &GatewayLifecycleAccount,
948 evidence: &StoredEvidence,
949) -> Result<(), GatewayLifecycleReconciliationError> {
950 sqlx::query(
951 r#"
952 UPDATE billing_gateway_lifecycle_quarantines
953 SET resolved_at = now()
954 WHERE billing_scope_id = $1
955 AND gateway_account_id = $2
956 AND resolved_at IS NULL
957 AND (
958 (
959 $3::text IS NOT NULL
960 AND public.billing_canonical_gateway_transaction_id(
961 gateway_transaction_id
962 ) = $3
963 )
964 OR (
965 gateway_transaction_id IS NULL
966 AND gateway_order_id IS NOT NULL
967 AND $4::text IS NOT NULL
968 AND gateway_order_id = $4
969 )
970 )
971 "#,
972 )
973 .bind(account.billing_scope_id().as_uuid())
974 .bind(account.gateway_account_id().as_uuid())
975 .bind(evidence.transaction_id.as_deref())
976 .bind(evidence.order_id.as_deref())
977 .execute(&mut **transaction)
978 .await?;
979 Ok(())
980}
981
982async fn delete_pending(
983 transaction: &mut Transaction<'_, Postgres>,
984 account: &GatewayLifecycleAccount,
985 pending_id: Uuid,
986) -> Result<(), GatewayLifecycleReconciliationError> {
987 sqlx::query(
988 r#"
989 DELETE FROM billing_gateway_lifecycle_pending_updates
990 WHERE id = $1
991 AND billing_scope_id = $2
992 AND gateway_account_id = $3
993 "#,
994 )
995 .bind(pending_id)
996 .bind(account.billing_scope_id().as_uuid())
997 .bind(account.gateway_account_id().as_uuid())
998 .execute(&mut **transaction)
999 .await?;
1000 Ok(())
1001}
1002
1003async fn cleanup_pending(
1004 pool: &PgPool,
1005 account: &GatewayLifecycleAccount,
1006) -> Result<u64, GatewayLifecycleReconciliationError> {
1007 let mut transaction = pool.begin().await?;
1008 set_timeouts(&mut transaction).await?;
1009 ensure_account(&mut transaction, account).await?;
1010 let result = sqlx::query(
1011 r#"
1012 WITH stale AS MATERIALIZED (
1013 SELECT pending.id
1014 FROM billing_gateway_lifecycle_pending_updates pending
1015 WHERE pending.billing_scope_id = $1
1016 AND pending.gateway_account_id = $2
1017 AND pending.expires_at <= now()
1018 ORDER BY pending.expires_at, pending.first_seen_at, pending.id
1019 LIMIT $3
1020 FOR UPDATE SKIP LOCKED
1021 )
1022 DELETE FROM billing_gateway_lifecycle_pending_updates pending
1023 USING stale
1024 WHERE pending.id = stale.id
1025 "#,
1026 )
1027 .bind(account.billing_scope_id().as_uuid())
1028 .bind(account.gateway_account_id().as_uuid())
1029 .bind(PENDING_CLEANUP_BATCH_SIZE)
1030 .execute(&mut *transaction)
1031 .await?;
1032 transaction.commit().await?;
1033 Ok(result.rows_affected())
1034}
1035
1036async fn record_unactionable_checks(
1037 pool: &PgPool,
1038 account: &GatewayLifecycleAccount,
1039) -> Result<(), GatewayLifecycleReconciliationError> {
1040 let mut transaction = pool.begin().await?;
1041 set_timeouts(&mut transaction).await?;
1042 ensure_account(&mut transaction, account).await?;
1043 sqlx::query(
1044 r#"
1045 WITH unactionable AS MATERIALIZED (
1046 SELECT pending.id
1047 FROM billing_gateway_lifecycle_pending_updates pending
1048 CROSS JOIN LATERAL (
1049 SELECT COUNT(*)::bigint AS candidate_count,
1050 COALESCE(BOOL_OR(
1051 pending.gateway_transaction_id IS NOT NULL
1052 AND public.billing_canonical_gateway_transaction_id(
1053 attempts.gateway_transaction_id
1054 ) = pending.gateway_transaction_id
1055 ), false) AS has_transaction_match
1056 FROM billing_payment_attempts attempts
1057 WHERE attempts.billing_scope_id = $1
1058 AND attempts.gateway_account_id = $2
1059 AND attempts.status = 'approved'
1060 AND (
1061 (
1062 pending.gateway_transaction_id IS NOT NULL
1063 AND public.billing_canonical_gateway_transaction_id(
1064 attempts.gateway_transaction_id
1065 ) = pending.gateway_transaction_id
1066 )
1067 OR (
1068 pending.gateway_order_id IS NOT NULL
1069 AND (
1070 pending.gateway_transaction_id IS NULL
1071 OR public.billing_canonical_gateway_transaction_id(
1072 attempts.gateway_transaction_id
1073 ) IS NULL
1074 )
1075 AND attempts.gateway_order_id = pending.gateway_order_id
1076 )
1077 )
1078 ) matches
1079 WHERE pending.billing_scope_id = $1
1080 AND pending.gateway_account_id = $2
1081 AND pending.expires_at > now()
1082 AND pending.check_count < $3
1083 AND NOT (matches.has_transaction_match OR matches.candidate_count = 1)
1084 ORDER BY pending.last_checked_at NULLS FIRST,
1085 pending.first_seen_at,
1086 pending.id
1087 LIMIT $4
1088 FOR UPDATE OF pending SKIP LOCKED
1089 )
1090 UPDATE billing_gateway_lifecycle_pending_updates pending
1091 SET last_checked_at = now(),
1092 check_count = pending.check_count + 1
1093 FROM unactionable
1094 WHERE pending.id = unactionable.id
1095 "#,
1096 )
1097 .bind(account.billing_scope_id().as_uuid())
1098 .bind(account.gateway_account_id().as_uuid())
1099 .bind(PENDING_MAX_CHECKS)
1100 .bind(PENDING_CLEANUP_BATCH_SIZE)
1101 .execute(&mut *transaction)
1102 .await?;
1103 transaction.commit().await?;
1104 Ok(())
1105}
1106
1107async fn actionable_pending(
1108 pool: &PgPool,
1109 account: &GatewayLifecycleAccount,
1110) -> Result<Vec<(Uuid, StoredEvidence)>, GatewayLifecycleReconciliationError> {
1111 let mut transaction = pool.begin().await?;
1112 set_timeouts(&mut transaction).await?;
1113 ensure_account(&mut transaction, account).await?;
1114 let rows = sqlx::query(
1115 r#"
1116 SELECT pending.id,
1117 pending.gateway_transaction_id,
1118 pending.gateway_order_id,
1119 pending.gateway_condition,
1120 pending.gateway_lifecycle_status,
1121 pending.gateway_lifecycle_action,
1122 pending.gateway_lifecycle_at,
1123 pending.refunded_amount_cents
1124 FROM billing_gateway_lifecycle_pending_updates pending
1125 CROSS JOIN LATERAL (
1126 SELECT COUNT(*)::bigint AS candidate_count,
1127 COALESCE(BOOL_OR(
1128 pending.gateway_transaction_id IS NOT NULL
1129 AND public.billing_canonical_gateway_transaction_id(
1130 attempts.gateway_transaction_id
1131 ) = pending.gateway_transaction_id
1132 ), false) AS has_transaction_match
1133 FROM billing_payment_attempts attempts
1134 WHERE attempts.billing_scope_id = $1
1135 AND attempts.gateway_account_id = $2
1136 AND attempts.status = 'approved'
1137 AND (
1138 (
1139 pending.gateway_transaction_id IS NOT NULL
1140 AND public.billing_canonical_gateway_transaction_id(
1141 attempts.gateway_transaction_id
1142 ) = pending.gateway_transaction_id
1143 )
1144 OR (
1145 pending.gateway_order_id IS NOT NULL
1146 AND (
1147 pending.gateway_transaction_id IS NULL
1148 OR public.billing_canonical_gateway_transaction_id(
1149 attempts.gateway_transaction_id
1150 ) IS NULL
1151 )
1152 AND attempts.gateway_order_id = pending.gateway_order_id
1153 )
1154 )
1155 ) matches
1156 WHERE pending.billing_scope_id = $1
1157 AND pending.gateway_account_id = $2
1158 AND pending.expires_at > now()
1159 AND (matches.has_transaction_match OR matches.candidate_count = 1)
1160 ORDER BY pending.first_seen_at, pending.id
1161 LIMIT $3
1162 FOR UPDATE OF pending SKIP LOCKED
1163 "#,
1164 )
1165 .bind(account.billing_scope_id().as_uuid())
1166 .bind(account.gateway_account_id().as_uuid())
1167 .bind(STAGED_APPLICATION_BATCH_SIZE)
1168 .fetch_all(&mut *transaction)
1169 .await?;
1170 let pending = rows
1171 .into_iter()
1172 .map(|row| {
1173 let status: String = row.try_get("gateway_lifecycle_status")?;
1174 let refunded = row.try_get("refunded_amount_cents")?;
1175 Ok((
1176 row.try_get("id")?,
1177 StoredEvidence {
1178 transaction_id: row.try_get("gateway_transaction_id")?,
1179 order_id: row.try_get("gateway_order_id")?,
1180 state: lifecycle_state_from_parts(&status, refunded, false)?,
1181 condition: row.try_get("gateway_condition")?,
1182 action: row.try_get("gateway_lifecycle_action")?,
1183 effective_at: row.try_get("gateway_lifecycle_at")?,
1184 },
1185 ))
1186 })
1187 .collect::<Result<Vec<_>, GatewayLifecycleReconciliationError>>()?;
1188 transaction.commit().await?;
1189 Ok(pending)
1190}
1191
1192pub(crate) async fn ensure_account(
1193 transaction: &mut Transaction<'_, Postgres>,
1194 account: &GatewayLifecycleAccount,
1195) -> Result<(), GatewayLifecycleReconciliationError> {
1196 let exists: bool = sqlx::query_scalar(
1197 r#"
1198 SELECT EXISTS (
1199 SELECT 1
1200 FROM billing_gateway_accounts
1201 WHERE billing_scope_id = $1
1202 AND id = $2
1203 AND provider_key = $3
1204 FOR KEY SHARE
1205 )
1206 "#,
1207 )
1208 .bind(account.billing_scope_id().as_uuid())
1209 .bind(account.gateway_account_id().as_uuid())
1210 .bind(account.provider_key().as_str())
1211 .fetch_one(&mut **transaction)
1212 .await?;
1213 if exists {
1214 Ok(())
1215 } else {
1216 Err(GatewayLifecycleReconciliationError::AccountNotFound)
1217 }
1218}
1219
1220pub(crate) async fn set_timeouts(
1221 transaction: &mut Transaction<'_, Postgres>,
1222) -> Result<(), sqlx::Error> {
1223 sqlx::query(
1224 "SELECT set_config('lock_timeout', $1, true), set_config('statement_timeout', $2, true)",
1225 )
1226 .bind(ROW_LOCK_TIMEOUT)
1227 .bind(OPERATION_TIMEOUT)
1228 .execute(&mut **transaction)
1229 .await?;
1230 Ok(())
1231}
1232
1233fn latest_time(left: Option<DateTime<Utc>>, right: Option<DateTime<Utc>>) -> Option<DateTime<Utc>> {
1234 match (left, right) {
1235 (Some(left), Some(right)) => Some(left.max(right)),
1236 (Some(value), None) | (None, Some(value)) => Some(value),
1237 (None, None) => None,
1238 }
1239}
1240
1241#[cfg(test)]
1242mod tests {
1243 use std::{
1244 error::Error,
1245 sync::atomic::{AtomicU64, Ordering},
1246 };
1247
1248 use super::*;
1249 use crate::test_support::{TestDatabase, create_gateway_account};
1250 use async_trait::async_trait;
1251 use sqlx::PgConnection;
1252 use syrup_rail::{
1253 GatewayAccountId, GatewayDiagnostic, GatewayProviderKey, GatewayReferenceValueError,
1254 HostChargeTargetNoChange, HostChargeTargetTransitionOutcome,
1255 };
1256
1257 #[derive(Default)]
1258 struct ExactHostTargets {
1259 calls: AtomicU64,
1260 }
1261
1262 #[async_trait]
1263 impl HostChargeTargetStore for ExactHostTargets {
1264 async fn preflight_target(
1265 &self,
1266 _connection: &mut PgConnection,
1267 _reservation: &crate::HostChargeTargetReservation,
1268 ) -> Result<crate::HostChargeReservationDecision, crate::HostChargeTargetError> {
1269 Ok(crate::HostChargeReservationDecision::Rejected {
1270 reason: syrup_rail::HostChargeTargetRejection::TargetUnavailable,
1271 })
1272 }
1273
1274 async fn reserve_target(
1275 &self,
1276 _connection: &mut PgConnection,
1277 _reservation: &crate::HostChargeTargetReservation,
1278 ) -> Result<crate::HostChargeReservationDecision, crate::HostChargeTargetError> {
1279 Ok(crate::HostChargeReservationDecision::Rejected {
1280 reason: syrup_rail::HostChargeTargetRejection::TargetUnavailable,
1281 })
1282 }
1283
1284 async fn ensure_submission_admitted(
1285 &self,
1286 _connection: &mut PgConnection,
1287 _admission: &crate::HostChargeSubmissionAdmission,
1288 ) -> Result<crate::HostChargeSubmissionDecision, crate::HostChargeTargetError> {
1289 Ok(crate::HostChargeSubmissionDecision::Rejected {
1290 reason: syrup_rail::HostChargeTargetRejection::TargetUnavailable,
1291 })
1292 }
1293
1294 async fn apply_transition(
1295 &self,
1296 connection: &mut PgConnection,
1297 transition: HostChargeTargetTransition,
1298 ) -> Result<HostChargeTargetTransitionOutcome, crate::HostChargeTargetError> {
1299 self.calls.fetch_add(1, Ordering::SeqCst);
1300 let kind = match transition.kind() {
1301 HostChargeTargetTransitionKind::Reversed { kind } => match kind {
1302 syrup_rail::PaymentReversalKind::Refunded => "refunded",
1303 syrup_rail::PaymentReversalKind::Voided => "voided",
1304 syrup_rail::PaymentReversalKind::Chargeback => "chargeback",
1305 },
1306 _ => {
1307 return Ok(HostChargeTargetTransitionOutcome::Unchanged {
1308 reason: HostChargeTargetNoChange::InapplicableState,
1309 });
1310 }
1311 };
1312 let result = sqlx::query(
1313 r#"
1314 UPDATE host_charge_targets
1315 SET status = 'reversed',
1316 reversal_kind = $4,
1317 reversed_at = $5
1318 WHERE id = $1
1319 AND billing_scope_id = $2
1320 AND subscriber_id = $3
1321 AND status = 'paid'
1322 "#,
1323 )
1324 .bind(transition.target_id().as_uuid())
1325 .bind(transition.billing_scope_id().as_uuid())
1326 .bind(transition.subscriber_id().as_uuid())
1327 .bind(kind)
1328 .bind(transition.effective_at())
1329 .execute(connection)
1330 .await
1331 .map_err(crate::HostChargeTargetError::new)?;
1332 Ok(if result.rows_affected() == 1 {
1333 HostChargeTargetTransitionOutcome::Applied
1334 } else {
1335 HostChargeTargetTransitionOutcome::Unchanged {
1336 reason: HostChargeTargetNoChange::InapplicableState,
1337 }
1338 })
1339 }
1340 }
1341
1342 fn lifecycle_account(
1343 fixture: crate::test_support::GatewayAccountFixture,
1344 provider: &str,
1345 ) -> GatewayLifecycleAccount {
1346 GatewayLifecycleAccount::new(
1347 BillingScopeId::new(fixture.billing_scope_id),
1348 GatewayAccountId::new(fixture.gateway_account_id),
1349 GatewayProviderKey::new(provider).unwrap(),
1350 )
1351 }
1352
1353 fn evidence(
1354 transaction_id: &str,
1355 state: GatewayLifecycleState,
1356 effective_at: DateTime<Utc>,
1357 ) -> Result<GatewayTransactionReport, GatewayReferenceValueError> {
1358 Ok(GatewayTransactionReport::Evidence(
1359 GatewayLifecycleEvidence::new(
1360 Some(GatewayTransactionId::new(transaction_id)?),
1361 None,
1362 state,
1363 Some(GatewayDiagnostic::new("condition")),
1364 Some(GatewayDiagnostic::new("diagnostic action")),
1365 Some(effective_at),
1366 )
1367 .unwrap(),
1368 ))
1369 }
1370
1371 async fn insert_host_attempt(
1372 pool: &PgPool,
1373 fixture: crate::test_support::GatewayAccountFixture,
1374 subscriber_id: Uuid,
1375 target_id: Uuid,
1376 transaction_id: &str,
1377 amount_cents: i32,
1378 resolved_at: DateTime<Utc>,
1379 ) -> Result<Uuid, sqlx::Error> {
1380 let attempt_id = Uuid::now_v7();
1381 sqlx::query(
1382 r#"
1383 INSERT INTO billing_payment_attempts (
1384 required_gateway_account_mode,
1385 id,
1386 billing_scope_id,
1387 subscriber_id,
1388 host_charge_target_id,
1389 attempt_kind,
1390 status,
1391 idempotency_key,
1392 request_fingerprint,
1393 amount_cents,
1394 gateway_account_id,
1395 gateway_configuration_id,
1396 gateway_order_id,
1397 gateway_transaction_id,
1398 submitted_at,
1399 resolved_at
1400 ) VALUES (
1401 'live',
1402 $1, $2, $3, $4, 'host_charge', 'approved', $5, $6, $7,
1403 $8, $9, $10, $11, $12, $12
1404 )
1405 "#,
1406 )
1407 .bind(attempt_id)
1408 .bind(fixture.billing_scope_id)
1409 .bind(subscriber_id)
1410 .bind(target_id)
1411 .bind(format!("idempotency-{attempt_id}"))
1412 .bind(format!("fingerprint-{attempt_id}"))
1413 .bind(amount_cents)
1414 .bind(fixture.gateway_account_id)
1415 .bind(fixture.gateway_configuration_id)
1416 .bind(format!("order-{attempt_id}"))
1417 .bind(transaction_id)
1418 .bind(resolved_at)
1419 .execute(pool)
1420 .await?;
1421 Ok(attempt_id)
1422 }
1423
1424 #[tokio::test]
1425 async fn report_lifecycle_is_crash_safe_monotonic_and_exact_targeted()
1426 -> Result<(), Box<dyn Error>> {
1427 let database = TestDatabase::start("rail_lifecycle").await?;
1428 let fixture = create_gateway_account(&database.pool, "nmi").await?;
1429 let account = lifecycle_account(fixture, "nmi");
1430 let host_targets = ExactHostTargets::default();
1431 sqlx::query(
1432 r#"
1433 CREATE TABLE host_charge_targets (
1434 id uuid PRIMARY KEY,
1435 billing_scope_id uuid NOT NULL,
1436 subscriber_id uuid NOT NULL,
1437 status text NOT NULL,
1438 reversal_kind text,
1439 paid_at timestamptz,
1440 reversed_at timestamptz
1441 )
1442 "#,
1443 )
1444 .execute(&database.pool)
1445 .await?;
1446
1447 let staged_at = Utc::now() - chrono::Duration::minutes(3);
1448 let staged = reconcile_gateway_transaction_reports(
1449 &database.pool,
1450 &host_targets,
1451 &account,
1452 vec![evidence(
1453 "txn-late",
1454 GatewayLifecycleState::Settled {
1455 cumulative_refunded_cents: Some(CumulativeRefundCents::new(125)?),
1456 },
1457 staged_at,
1458 )?],
1459 )
1460 .await?;
1461 assert_eq!(staged.staged(), 1);
1462 let pending_state: (String, Option<i32>, String) = sqlx::query_as(
1463 r#"
1464 SELECT gateway_lifecycle_status, refunded_amount_cents,
1465 gateway_lifecycle_action
1466 FROM billing_gateway_lifecycle_pending_updates
1467 WHERE gateway_account_id = $1
1468 "#,
1469 )
1470 .bind(fixture.gateway_account_id)
1471 .fetch_one(&database.pool)
1472 .await?;
1473 assert_eq!(
1474 pending_state,
1475 (
1476 "settled".to_owned(),
1477 Some(125),
1478 "diagnostic action".to_owned()
1479 )
1480 );
1481
1482 let late_subscriber = Uuid::now_v7();
1483 let late_target = Uuid::now_v7();
1484 insert_host_attempt(
1485 &database.pool,
1486 fixture,
1487 late_subscriber,
1488 late_target,
1489 "txn-late",
1490 1_000,
1491 staged_at - chrono::Duration::minutes(1),
1492 )
1493 .await?;
1494 let applied =
1495 apply_staged_gateway_lifecycle_evidence(&database.pool, &host_targets, &account)
1496 .await?;
1497 assert_eq!(applied.applied(), 1);
1498 let stored: (String, i32, String) = sqlx::query_as(
1499 r#"
1500 SELECT gateway_lifecycle_status, refunded_amount_cents,
1501 gateway_lifecycle_action
1502 FROM billing_payment_attempts
1503 WHERE gateway_transaction_id = 'txn-late'
1504 "#,
1505 )
1506 .fetch_one(&database.pool)
1507 .await?;
1508 assert_eq!(
1509 stored,
1510 ("settled".to_owned(), 125, "diagnostic action".to_owned())
1511 );
1512 assert_eq!(host_targets.calls.load(Ordering::SeqCst), 0);
1513
1514 let subscriber_id = Uuid::now_v7();
1515 let target_id = Uuid::now_v7();
1516 sqlx::query(
1517 "INSERT INTO host_charge_targets (id, billing_scope_id, subscriber_id, status) VALUES ($1, $2, $3, 'paid')",
1518 )
1519 .bind(target_id)
1520 .bind(fixture.billing_scope_id)
1521 .bind(subscriber_id)
1522 .execute(&database.pool)
1523 .await?;
1524 insert_host_attempt(
1525 &database.pool,
1526 fixture,
1527 subscriber_id,
1528 target_id,
1529 "txn-refund",
1530 2_500,
1531 staged_at,
1532 )
1533 .await?;
1534 let refunded_at = Utc::now();
1535 let report = evidence(
1536 "txn-refund",
1537 GatewayLifecycleState::Refunded {
1538 cumulative_refunded_cents: CumulativeRefundCents::new(2_500)?,
1539 },
1540 refunded_at,
1541 )?;
1542 let first = reconcile_gateway_transaction_reports(
1543 &database.pool,
1544 &host_targets,
1545 &account,
1546 vec![report.clone()],
1547 )
1548 .await?;
1549 let replay = reconcile_gateway_transaction_reports(
1550 &database.pool,
1551 &host_targets,
1552 &account,
1553 vec![report],
1554 )
1555 .await?;
1556 assert_eq!(first.applied(), 1);
1557 assert_eq!(replay.applied(), 0);
1558 assert_eq!(host_targets.calls.load(Ordering::SeqCst), 1);
1559 let target: (String, String, DateTime<Utc>) = sqlx::query_as(
1560 "SELECT status, reversal_kind, reversed_at FROM host_charge_targets WHERE id = $1",
1561 )
1562 .bind(target_id)
1563 .fetch_one(&database.pool)
1564 .await?;
1565 assert_eq!(target.0, "reversed");
1566 assert_eq!(target.1, "refunded");
1567 assert_eq!(target.2, postgres_timestamp_precision(refunded_at));
1568
1569 let refused_subscriber_id = Uuid::now_v7();
1570 let refused_target_id = Uuid::now_v7();
1571 sqlx::query(
1572 "INSERT INTO host_charge_targets (id, billing_scope_id, subscriber_id, status) VALUES ($1, $2, $3, 'pending')",
1573 )
1574 .bind(refused_target_id)
1575 .bind(fixture.billing_scope_id)
1576 .bind(refused_subscriber_id)
1577 .execute(&database.pool)
1578 .await?;
1579 insert_host_attempt(
1580 &database.pool,
1581 fixture,
1582 refused_subscriber_id,
1583 refused_target_id,
1584 "txn-refund-refused",
1585 800,
1586 staged_at,
1587 )
1588 .await?;
1589 let refused_attempt_before: (Option<String>, Option<i32>) = sqlx::query_as(
1590 "SELECT gateway_lifecycle_status, refunded_amount_cents FROM billing_payment_attempts WHERE gateway_transaction_id = 'txn-refund-refused'",
1591 )
1592 .fetch_one(&database.pool)
1593 .await?;
1594 let following_subscriber_id = Uuid::now_v7();
1595 let following_target_id = Uuid::now_v7();
1596 sqlx::query(
1597 "INSERT INTO host_charge_targets (id, billing_scope_id, subscriber_id, status) VALUES ($1, $2, $3, 'paid')",
1598 )
1599 .bind(following_target_id)
1600 .bind(fixture.billing_scope_id)
1601 .bind(following_subscriber_id)
1602 .execute(&database.pool)
1603 .await?;
1604 insert_host_attempt(
1605 &database.pool,
1606 fixture,
1607 following_subscriber_id,
1608 following_target_id,
1609 "txn-refund-following",
1610 900,
1611 staged_at,
1612 )
1613 .await?;
1614 let refused_evidence = evidence(
1615 "txn-refund-refused",
1616 GatewayLifecycleState::Refunded {
1617 cumulative_refunded_cents: CumulativeRefundCents::new(800)?,
1618 },
1619 Utc::now(),
1620 )?;
1621 let summary = reconcile_gateway_transaction_reports(
1622 &database.pool,
1623 &host_targets,
1624 &account,
1625 vec![
1626 refused_evidence.clone(),
1627 evidence(
1628 "txn-refund-following",
1629 GatewayLifecycleState::Refunded {
1630 cumulative_refunded_cents: CumulativeRefundCents::new(900)?,
1631 },
1632 Utc::now(),
1633 )?,
1634 ],
1635 )
1636 .await?;
1637 assert_eq!(summary.skipped(), 1);
1638 assert_eq!(summary.staged(), 1);
1639 assert_eq!(summary.applied(), 1);
1640 let redelivery = reconcile_gateway_transaction_reports(
1641 &database.pool,
1642 &host_targets,
1643 &account,
1644 vec![refused_evidence],
1645 )
1646 .await?;
1647 assert_eq!(redelivery.skipped(), 1);
1648 assert_eq!(redelivery.staged(), 0);
1649 assert_eq!(redelivery.applied(), 0);
1650 let refused_attempt_state: (Option<String>, Option<i32>) = sqlx::query_as(
1651 "SELECT gateway_lifecycle_status, refunded_amount_cents FROM billing_payment_attempts WHERE gateway_transaction_id = 'txn-refund-refused'",
1652 )
1653 .fetch_one(&database.pool)
1654 .await?;
1655 assert_eq!(refused_attempt_state, refused_attempt_before);
1656 let refused_pending_count: i64 = sqlx::query_scalar(
1657 "SELECT COUNT(*) FROM billing_gateway_lifecycle_pending_updates WHERE gateway_account_id = $1 AND gateway_transaction_id = 'txn-refund-refused'",
1658 )
1659 .bind(fixture.gateway_account_id)
1660 .fetch_one(&database.pool)
1661 .await?;
1662 assert_eq!(refused_pending_count, 1);
1663 let staged_retry =
1664 apply_staged_gateway_lifecycle_evidence(&database.pool, &host_targets, &account)
1665 .await?;
1666 assert_eq!(staged_retry.skipped(), 1);
1667 assert_eq!(staged_retry.staged(), 0);
1668 let following_target_state: String =
1669 sqlx::query_scalar("SELECT status FROM host_charge_targets WHERE id = $1")
1670 .bind(following_target_id)
1671 .fetch_one(&database.pool)
1672 .await?;
1673 assert_eq!(following_target_state, "reversed");
1674
1675 let cursor_key = GatewayLifecycleCursorKey::new("approved_lifecycle")?;
1676 let initial =
1677 gateway_lifecycle_reconciliation_start(&database.pool, &account, &cursor_key).await?;
1678 assert!(initial.is_some());
1679 let later = Utc::now() + chrono::Duration::minutes(5);
1680 save_gateway_lifecycle_reconciliation_cursor(&database.pool, &account, &cursor_key, later)
1681 .await?;
1682 save_gateway_lifecycle_reconciliation_cursor(
1683 &database.pool,
1684 &account,
1685 &cursor_key,
1686 staged_at,
1687 )
1688 .await?;
1689 assert_eq!(
1690 gateway_lifecycle_reconciliation_start(&database.pool, &account, &cursor_key).await?,
1691 Some(postgres_timestamp_precision(later))
1692 );
1693
1694 let wrong_provider = lifecycle_account(fixture, "other");
1695 assert!(matches!(
1696 gateway_lifecycle_reconciliation_start(&database.pool, &wrong_provider, &cursor_key)
1697 .await,
1698 Err(GatewayLifecycleReconciliationError::AccountNotFound)
1699 ));
1700
1701 database.cleanup().await?;
1702 Ok(())
1703 }
1704}