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