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