1use std::{fmt, time::Duration};
2
3use chrono::{DateTime, Utc};
4use sqlx::{PgConnection, PgPool};
5use syrup_rail::{
6 BillingEvent, BillingEventSubject, ChargeHostTarget, GatewayMutationError,
7 GatewayNotSubmittedError, GatewayPaymentOutcome, GatewayPaymentStatus, GatewaySaleIntent,
8 GatewaySaleRequest, HostChargePaymentResult, HostChargeReservation, HostChargeTargetTransition,
9 HostChargeTargetTransitionKind, HostChargeTargetTransitionOutcome, PaymentAttempt,
10 PaymentAttemptStatus, PaymentResolutionCode, ProcessorChargeProgression, ProcessorChargeRole,
11 ProcessorEvidence, ResolvedGateway,
12};
13use thiserror::Error;
14
15use crate::{
16 BillingTransactionCoordinator, BillingTransactionError, BillingTransactionSubjectState,
17 HostChargeStoreError, HostChargeSubmissionOutcome, HostChargeTargetError,
18 HostChargeTargetStore, ProcessorChargeStoreError, admit_host_charge_submission_in_transaction,
19 attempts::{
20 AttemptApproval, AttemptResolutionStatus, AttemptTransition, PaymentAttemptStoreError,
21 find_payment_attempt_by_id_on_connection, lock_payment_attempt_by_id_on_connection,
22 persist_attempt_transition,
23 },
24 enrollment_application::{
25 OutcomeResolutionBoundary, RateLimitCooldown, SubscriptionEnrollmentApplicationError,
26 map_attempt_transition_error, mutation_error_evidence, not_submitted_resolution_code,
27 park_locked_attempt, set_application_timeouts,
28 },
29 processor_charges::{ObservedCharge, observe_processor_charge, transition_charge},
30};
31
32const BILLING_LOCK_TIMEOUT: Duration = Duration::from_millis(250);
33const APPROVED_APPLICATION_ATTEMPTS: usize = 3;
34const APPROVED_EVIDENCE_RETRY_DELAY: Duration = Duration::from_millis(50);
35const INVALID_HOST_CHARGE_STATE: &str = "canonical host charge application state is invalid";
36const APPROVED_STALE_TARGET_TEXT: &str =
37 "Approved host charge could not update its target because the target changed.";
38const APPROVED_STORAGE_FAILURE_TEXT: &str =
39 "Approved host charge could not be applied; manual review is required.";
40
41#[derive(Debug, Error)]
42pub enum HostChargeApplicationError {
43 #[error("host charge application storage failed")]
44 Sql(#[from] sqlx::Error),
45 #[error("host charge attempt storage failed")]
46 Attempt(#[from] PaymentAttemptStoreError),
47 #[error("host charge target operation failed")]
48 Target(#[from] HostChargeTargetError),
49 #[error("host charge storage operation failed")]
50 Store(#[from] HostChargeStoreError),
51 #[error("host charge processor evidence operation failed")]
52 ProcessorCharge(#[from] ProcessorChargeStoreError),
53 #[error("host billing transaction failed")]
54 Transaction(#[from] BillingTransactionError),
55 #[error("host billing event append failed")]
56 Event(#[from] crate::BillingEventWriteError),
57 #[error("shared payment application failed")]
58 SharedApplication(#[from] SubscriptionEnrollmentApplicationError),
59 #[error("admitted host charge does not match the submission command or gateway")]
60 SubmissionIdentityMismatch,
61 #[error("{0}")]
62 InvalidState(&'static str),
63}
64
65pub struct AdmittedHostCharge {
66 reservation: HostChargeReservation,
67 attempt: PaymentAttempt,
68}
69
70impl AdmittedHostCharge {
71 pub const fn attempt(&self) -> &PaymentAttempt {
72 &self.attempt
73 }
74}
75
76impl fmt::Debug for AdmittedHostCharge {
77 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
78 formatter
79 .debug_struct("AdmittedHostCharge")
80 .field("attempt", &self.attempt)
81 .field("has_submission_authority", &true)
82 .finish()
83 }
84}
85
86#[derive(Debug)]
87pub enum HostChargeAdmissionOutcome {
88 Admitted(Box<AdmittedHostCharge>),
89 AlreadyAdmitted(PaymentAttempt),
90 Rejected {
91 attempt: PaymentAttempt,
92 reason: syrup_rail::HostChargeTargetRejection,
93 },
94}
95
96#[derive(Debug)]
97pub enum HostChargeProviderResult {
98 Payment(HostChargePaymentResult),
99 NotSubmitted {
100 payment: HostChargePaymentResult,
101 error: GatewayNotSubmittedError,
102 },
103}
104
105#[derive(Clone, Copy)]
109pub(crate) struct HostChargeBeforeSubmissionResolution {
110 boundary: OutcomeResolutionBoundary,
111 cooldown: Option<RateLimitCooldown>,
112}
113
114impl HostChargeBeforeSubmissionResolution {
115 pub(crate) const fn prepared() -> Self {
116 Self {
117 boundary: OutcomeResolutionBoundary::Prepared,
118 cooldown: None,
119 }
120 }
121
122 pub(crate) const fn admitted_not_submitted() -> Self {
123 Self {
124 boundary: OutcomeResolutionBoundary::AdmittedNotSubmitted,
125 cooldown: None,
126 }
127 }
128
129 pub(crate) const fn prepared_provider_rate_limited() -> Self {
130 Self {
131 boundary: OutcomeResolutionBoundary::Prepared,
132 cooldown: Some(RateLimitCooldown::Provider),
133 }
134 }
135}
136
137pub async fn admit_host_charge_submission(
138 pool: &PgPool,
139 targets: &dyn HostChargeTargetStore,
140 reservation: &HostChargeReservation,
141) -> Result<HostChargeAdmissionOutcome, HostChargeApplicationError> {
142 let mut transaction = pool.begin().await?;
143 let outcome =
144 admit_host_charge_submission_in_transaction(&mut transaction, targets, reservation).await?;
145 transaction.commit().await?;
146 Ok(match outcome {
147 HostChargeSubmissionOutcome::Admitted(attempt) => {
148 HostChargeAdmissionOutcome::Admitted(Box::new(AdmittedHostCharge {
149 reservation: reservation.clone(),
150 attempt,
151 }))
152 }
153 HostChargeSubmissionOutcome::AlreadyAdmitted(attempt) => {
154 HostChargeAdmissionOutcome::AlreadyAdmitted(attempt)
155 }
156 HostChargeSubmissionOutcome::Rejected { attempt, reason } => {
157 HostChargeAdmissionOutcome::Rejected { attempt, reason }
158 }
159 })
160}
161
162pub async fn submit_admitted_host_charge(
163 pool: &PgPool,
164 coordinator: &dyn BillingTransactionCoordinator,
165 targets: &dyn HostChargeTargetStore,
166 admission: AdmittedHostCharge,
167 command: &ChargeHostTarget,
168 gateway: &ResolvedGateway,
169) -> Result<HostChargeProviderResult, HostChargeApplicationError> {
170 let reconstructed = HostChargeReservation::from_command(
171 command,
172 admission.reservation.snapshot(),
173 gateway,
174 admission.attempt.identity().attempt_id(),
175 )
176 .map_err(|_| HostChargeApplicationError::SubmissionIdentityMismatch)?;
177 if !host_charge_submission_matches_reservation(&reconstructed, &admission.reservation)
178 || admission.attempt.identity() != admission.reservation.identity()
179 || admission.attempt.request() != admission.reservation.request()
180 || admission.attempt.status() != PaymentAttemptStatus::Pending
181 || admission
182 .attempt
183 .state()
184 .timestamps()
185 .submitted_at()
186 .is_none()
187 {
188 return Err(HostChargeApplicationError::SubmissionIdentityMismatch);
189 }
190 let request = GatewaySaleRequest::new(
191 admission.reservation.snapshot().charge(),
192 admission.attempt.request().gateway_order_id().clone(),
193 GatewaySaleIntent::OneTime {
194 payment_token: command.payment_token().clone(),
195 },
196 command.billing_contact().cloned(),
197 );
198 match gateway.sale(request).await {
199 Ok(outcome) => apply_host_charge_gateway_outcome(
200 pool,
201 coordinator,
202 targets,
203 &admission.reservation,
204 &outcome,
205 )
206 .await
207 .map(HostChargeProviderResult::Payment),
208 Err(GatewayMutationError::NotSubmitted(error)) => {
209 let evidence = mutation_error_evidence(error.detail());
210 let release_target = !matches!(error, GatewayNotSubmittedError::RateLimited(_));
211 let payment = resolve_host_charge_non_approved(
212 pool,
213 targets,
214 &admission.reservation,
215 &evidence,
216 AttemptResolutionStatus::Failed,
217 Some(not_submitted_resolution_code(&error)),
218 OutcomeResolutionBoundary::AdmittedNotSubmitted,
219 release_target,
220 None,
221 )
222 .await?;
223 Ok(HostChargeProviderResult::NotSubmitted { payment, error })
224 }
225 Err(GatewayMutationError::RateLimitedIndeterminate(detail)) => resolve_host_charge_unknown(
226 pool,
227 &admission.reservation,
228 &mutation_error_evidence(&detail),
229 Some(RateLimitCooldown::Provider),
230 )
231 .await
232 .map(HostChargeProviderResult::Payment),
233 Err(GatewayMutationError::Indeterminate(detail)) => resolve_host_charge_unknown(
234 pool,
235 &admission.reservation,
236 &mutation_error_evidence(&detail),
237 None,
238 )
239 .await
240 .map(HostChargeProviderResult::Payment),
241 }
242}
243
244fn host_charge_submission_matches_reservation(
245 reconstructed: &HostChargeReservation,
246 durable: &HostChargeReservation,
247) -> bool {
248 let request = reconstructed.request();
249 let durable_request = durable.request();
250 reconstructed.identity() == durable.identity()
251 && reconstructed.snapshot() == durable.snapshot()
252 && request.target() == durable_request.target()
253 && request.idempotency_key() == durable_request.idempotency_key()
254 && request.fingerprint() == durable_request.fingerprint()
255 && request.amount() == durable_request.amount()
256 && request.gateway_order_id() == durable_request.gateway_order_id()
257}
258
259pub async fn apply_host_charge_gateway_outcome(
260 pool: &PgPool,
261 coordinator: &dyn BillingTransactionCoordinator,
262 targets: &dyn HostChargeTargetStore,
263 reservation: &HostChargeReservation,
264 outcome: &GatewayPaymentOutcome,
265) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
266 match outcome.status() {
267 GatewayPaymentStatus::Approved => {
268 if outcome.transaction_id().is_none() {
269 return durably_park_host_charge_approved(pool, reservation, outcome.evidence())
270 .await;
271 }
272 for attempt_index in 0..APPROVED_APPLICATION_ATTEMPTS {
273 match apply_host_charge_approved(
274 coordinator,
275 targets,
276 reservation,
277 outcome.evidence(),
278 )
279 .await
280 {
281 Ok(result) => return Ok(result),
282 Err(_) if attempt_index + 1 < APPROVED_APPLICATION_ATTEMPTS => {
283 tokio::time::sleep(APPROVED_EVIDENCE_RETRY_DELAY).await;
284 }
285 Err(_) => break,
286 }
287 }
288 durably_park_host_charge_approved(pool, reservation, outcome.evidence()).await
289 }
290 GatewayPaymentStatus::Declined => {
291 resolve_host_charge_non_approved(
292 pool,
293 targets,
294 reservation,
295 outcome.evidence(),
296 AttemptResolutionStatus::Declined,
297 None,
298 OutcomeResolutionBoundary::Submitted,
299 true,
300 None,
301 )
302 .await
303 }
304 GatewayPaymentStatus::Failed => {
305 resolve_host_charge_non_approved(
306 pool,
307 targets,
308 reservation,
309 outcome.evidence(),
310 AttemptResolutionStatus::Failed,
311 None,
312 OutcomeResolutionBoundary::Submitted,
313 true,
314 None,
315 )
316 .await
317 }
318 GatewayPaymentStatus::Unknown => {
319 resolve_host_charge_unknown(pool, reservation, outcome.evidence(), None).await
320 }
321 }
322}
323
324pub async fn apply_reconciled_host_charge_gateway_outcome(
325 pool: &PgPool,
326 coordinator: &dyn BillingTransactionCoordinator,
327 targets: &dyn HostChargeTargetStore,
328 billing_scope_id: syrup_rail::BillingScopeId,
329 attempt_id: syrup_rail::PaymentAttemptId,
330 outcome: &GatewayPaymentOutcome,
331) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
332 let mut transaction = pool.begin().await?;
333 let attempt = crate::find_payment_attempt_by_id_in_transaction(
334 &mut transaction,
335 billing_scope_id,
336 attempt_id,
337 )
338 .await?
339 .ok_or(HostChargeApplicationError::InvalidState(
340 INVALID_HOST_CHARGE_STATE,
341 ))?;
342 transaction.commit().await?;
343 let reservation = HostChargeReservation::from_attempt(&attempt)
344 .map_err(|_| HostChargeApplicationError::InvalidState(INVALID_HOST_CHARGE_STATE))?;
345 apply_host_charge_gateway_outcome(pool, coordinator, targets, &reservation, outcome).await
346}
347
348pub(crate) async fn resolve_host_charge_before_submission(
349 pool: &PgPool,
350 targets: &dyn HostChargeTargetStore,
351 reservation: &HostChargeReservation,
352 detail: syrup_rail::GatewayDiagnostic,
353 resolution_code: PaymentResolutionCode,
354 resolution: HostChargeBeforeSubmissionResolution,
355) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
356 let condition = if matches!(
357 resolution_code,
358 PaymentResolutionCode::GatewayAccountMutationCooldownBeforeSubmission
359 | PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission
360 ) {
361 None
362 } else {
363 Some(syrup_rail::GatewayDiagnostic::new("failed"))
364 };
365 let evidence = ProcessorEvidence::new(
366 None,
367 None,
368 None,
369 None,
370 Some(detail),
371 condition,
372 syrup_rail::GatewayPaymentDescriptor::default(),
373 );
374 resolve_host_charge_non_approved(
375 pool,
376 targets,
377 reservation,
378 &evidence,
379 AttemptResolutionStatus::Failed,
380 Some(resolution_code),
381 resolution.boundary,
382 false,
383 resolution.cooldown,
384 )
385 .await
386}
387
388async fn apply_host_charge_approved(
389 coordinator: &dyn BillingTransactionCoordinator,
390 targets: &dyn HostChargeTargetStore,
391 reservation: &HostChargeReservation,
392 evidence: &ProcessorEvidence,
393) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
394 let identity = reservation.identity();
395 let mut transaction = coordinator
396 .begin(
397 BillingEventSubject::new(identity.billing_scope_id(), identity.subscriber_id()),
398 BILLING_LOCK_TIMEOUT,
399 )
400 .await?;
401 if transaction.subject_state() != BillingTransactionSubjectState::LiveRecipient {
402 let _ = transaction.rollback().await;
403 return Err(HostChargeApplicationError::InvalidState(
404 "a host charge payment event requires a live recipient",
405 ));
406 }
407 let connection = transaction.connection();
408 set_application_timeouts(connection).await?;
409 let effective_at: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
410 .fetch_one(&mut *connection)
411 .await?;
412 let target_outcome = targets
413 .apply_transition(
414 connection,
415 HostChargeTargetTransition::new(
416 identity.billing_scope_id(),
417 identity.subscriber_id(),
418 identity.attempt_id(),
419 reservation.snapshot().target_id(),
420 HostChargeTargetTransitionKind::Paid,
421 effective_at,
422 ),
423 )
424 .await?;
425 let attempt = lock_expected_host_charge(connection, reservation).await?;
426 if attempt.status() == PaymentAttemptStatus::Approved {
427 observe_processor_charge(
428 connection,
429 &attempt,
430 evidence,
431 ProcessorChargeProgression::Applied,
432 )
433 .await?;
434 transaction.commit().await?;
435 return Ok(HostChargePaymentResult::new(attempt));
436 }
437 if attempt.status().is_terminal() {
438 transaction.rollback().await?;
439 return observe_terminal_host_charge_approval(coordinator, reservation, &attempt, evidence)
440 .await;
441 }
442 let observation = observe_processor_charge(
443 connection,
444 &attempt,
445 evidence,
446 ProcessorChargeProgression::Pending,
447 )
448 .await?;
449 let ObservedCharge::Owned(charge) = observation else {
450 let _ = transaction.rollback().await;
451 return Err(HostChargeApplicationError::InvalidState(
452 "the approved gateway transaction belongs to another payment attempt",
453 ));
454 };
455 if charge.role == ProcessorChargeRole::Additional {
456 let _ = transaction.rollback().await;
457 return Err(HostChargeApplicationError::InvalidState(
458 "an additional approved host charge requires external reversal",
459 ));
460 }
461 match target_outcome {
462 HostChargeTargetTransitionOutcome::Applied
463 | HostChargeTargetTransitionOutcome::ExactReplay => {
464 persist_attempt_transition(
465 connection,
466 &attempt,
467 evidence,
468 AttemptTransition::Approved(AttemptApproval::HostCharge),
469 )
470 .await
471 .map_err(map_attempt_transition_error)?;
472 transition_charge(
473 connection,
474 charge.id,
475 ProcessorChargeProgression::Applied,
476 None,
477 )
478 .await?;
479 let applied = find_payment_attempt_by_id_on_connection(
480 connection,
481 identity.billing_scope_id(),
482 identity.attempt_id(),
483 )
484 .await?
485 .ok_or(HostChargeApplicationError::InvalidState(
486 INVALID_HOST_CHARGE_STATE,
487 ))?;
488 let event = BillingEvent::HostChargePaid {
489 attempt_id: identity.attempt_id(),
490 target_id: reservation.snapshot().target_id(),
491 charge: reservation.snapshot().charge(),
492 };
493 transaction.append_event(&event).await?;
494 transaction.commit().await?;
495 Ok(HostChargePaymentResult::new(applied))
496 }
497 HostChargeTargetTransitionOutcome::StaleTarget => {
498 transition_charge(
499 connection,
500 charge.id,
501 ProcessorChargeProgression::ExternalReversalRequired,
502 Some(PaymentResolutionCode::HostChargeApprovedStaleState),
503 )
504 .await?;
505 let parked = park_locked_attempt(
506 connection,
507 &attempt,
508 evidence,
509 Some(PaymentResolutionCode::HostChargeApprovedStaleState),
510 APPROVED_STALE_TARGET_TEXT,
511 )
512 .await?;
513 transaction.commit().await?;
514 Ok(HostChargePaymentResult::new(parked))
515 }
516 HostChargeTargetTransitionOutcome::Unchanged { .. } => {
517 let _ = transaction.rollback().await;
518 Err(HostChargeApplicationError::InvalidState(
519 INVALID_HOST_CHARGE_STATE,
520 ))
521 }
522 }
523}
524
525async fn observe_terminal_host_charge_approval(
526 coordinator: &dyn BillingTransactionCoordinator,
527 reservation: &HostChargeReservation,
528 terminal_attempt: &PaymentAttempt,
529 evidence: &ProcessorEvidence,
530) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
531 let identity = reservation.identity();
532 let mut transaction = coordinator
533 .begin(
534 BillingEventSubject::new(identity.billing_scope_id(), identity.subscriber_id()),
535 BILLING_LOCK_TIMEOUT,
536 )
537 .await?;
538 let connection = transaction.connection();
539 set_application_timeouts(connection).await?;
540 if matches!(
541 observe_processor_charge(
542 connection,
543 terminal_attempt,
544 evidence,
545 ProcessorChargeProgression::Pending,
546 )
547 .await?,
548 ObservedCharge::OwnedByOtherAttempt
549 ) {
550 let _ = transaction.rollback().await;
551 return Err(HostChargeApplicationError::InvalidState(
552 "the approved gateway transaction belongs to another payment attempt",
553 ));
554 }
555 let locked = lock_expected_host_charge(connection, reservation).await?;
556 if !locked.status().is_terminal() || locked.status() == PaymentAttemptStatus::Approved {
557 let _ = transaction.rollback().await;
558 return Err(HostChargeApplicationError::InvalidState(
559 INVALID_HOST_CHARGE_STATE,
560 ));
561 }
562 transaction.commit().await?;
563 Ok(HostChargePaymentResult::confirmation_pending(
564 locked,
565 evidence.clone(),
566 ))
567}
568
569#[allow(clippy::too_many_arguments)]
570async fn resolve_host_charge_non_approved(
571 pool: &PgPool,
572 targets: &dyn HostChargeTargetStore,
573 reservation: &HostChargeReservation,
574 evidence: &ProcessorEvidence,
575 status: AttemptResolutionStatus,
576 resolution_code: Option<PaymentResolutionCode>,
577 boundary: OutcomeResolutionBoundary,
578 release_target: bool,
579 cooldown: Option<RateLimitCooldown>,
580) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
581 let identity = reservation.identity();
582 let mut transaction = pool.begin().await?;
583 set_application_timeouts(&mut transaction).await?;
584 if release_target {
585 let effective_at = sqlx::query_scalar("SELECT clock_timestamp()")
586 .fetch_one(&mut *transaction)
587 .await?;
588 let target_outcome = targets
589 .apply_transition(
590 &mut transaction,
591 HostChargeTargetTransition::new(
592 identity.billing_scope_id(),
593 identity.subscriber_id(),
594 identity.attempt_id(),
595 reservation.snapshot().target_id(),
596 HostChargeTargetTransitionKind::PaymentFailed,
597 effective_at,
598 ),
599 )
600 .await?;
601 if target_outcome == HostChargeTargetTransitionOutcome::StaleTarget {
602 return Err(HostChargeApplicationError::InvalidState(
603 INVALID_HOST_CHARGE_STATE,
604 ));
605 }
606 }
607 let attempt = lock_expected_host_charge(&mut transaction, reservation).await?;
608 let may_resolve = attempt.status().is_resolvable()
609 && match boundary {
610 OutcomeResolutionBoundary::Prepared => {
611 attempt.state().timestamps().submitted_at().is_none()
612 }
613 OutcomeResolutionBoundary::AdmittedNotSubmitted => {
614 attempt.state().timestamps().submitted_at().is_some()
615 }
616 OutcomeResolutionBoundary::Submitted => true,
617 };
618 if !may_resolve && release_target {
619 transaction.rollback().await?;
620 let mut reload = pool.begin().await?;
621 let attempt = crate::find_payment_attempt_by_id_in_transaction(
622 &mut reload,
623 identity.billing_scope_id(),
624 identity.attempt_id(),
625 )
626 .await?
627 .ok_or(HostChargeApplicationError::InvalidState(
628 INVALID_HOST_CHARGE_STATE,
629 ))?;
630 reload.commit().await?;
631 return Ok(HostChargePaymentResult::new(attempt));
632 }
633 if may_resolve {
634 persist_attempt_transition(
635 &mut transaction,
636 &attempt,
637 evidence,
638 AttemptTransition::Resolved {
639 status,
640 resolution_code,
641 },
642 )
643 .await
644 .map_err(map_attempt_transition_error)?;
645 if evidence_looks_approved(evidence) {
646 observe_processor_charge(
647 &mut transaction,
648 &attempt,
649 evidence,
650 ProcessorChargeProgression::Pending,
651 )
652 .await?;
653 }
654 if boundary == OutcomeResolutionBoundary::AdmittedNotSubmitted {
655 sqlx::query(
656 "UPDATE billing_payment_attempts SET submitted_at = NULL, updated_at = clock_timestamp() WHERE id = $1 AND status = $2",
657 )
658 .bind(identity.attempt_id().as_uuid())
659 .bind(status.as_str())
660 .execute(&mut *transaction)
661 .await?;
662 }
663 }
664 if host_charge_provider_cooldown_requested(cooldown) {
665 extend_host_charge_provider_cooldown(&mut transaction, reservation).await?;
666 }
667 let attempt = find_payment_attempt_by_id_on_connection(
668 &mut transaction,
669 identity.billing_scope_id(),
670 identity.attempt_id(),
671 )
672 .await?
673 .ok_or(HostChargeApplicationError::InvalidState(
674 INVALID_HOST_CHARGE_STATE,
675 ))?;
676 transaction.commit().await?;
677 Ok(HostChargePaymentResult::new(attempt))
678}
679
680async fn resolve_host_charge_unknown(
681 pool: &PgPool,
682 reservation: &HostChargeReservation,
683 evidence: &ProcessorEvidence,
684 cooldown: Option<RateLimitCooldown>,
685) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
686 let identity = reservation.identity();
687 let mut transaction = pool.begin().await?;
688 set_application_timeouts(&mut transaction).await?;
689 let attempt = lock_expected_host_charge(&mut transaction, reservation).await?;
690 if !attempt.status().is_terminal() {
691 persist_attempt_transition(
692 &mut transaction,
693 &attempt,
694 evidence,
695 AttemptTransition::Resolved {
696 status: AttemptResolutionStatus::Unknown,
697 resolution_code: None,
698 },
699 )
700 .await
701 .map_err(map_attempt_transition_error)?;
702 if evidence_looks_approved(evidence) {
703 observe_processor_charge(
704 &mut transaction,
705 &attempt,
706 evidence,
707 ProcessorChargeProgression::Pending,
708 )
709 .await?;
710 }
711 }
712 if host_charge_provider_cooldown_requested(cooldown) {
713 extend_host_charge_provider_cooldown(&mut transaction, reservation).await?;
714 }
715 let attempt = find_payment_attempt_by_id_on_connection(
716 &mut transaction,
717 identity.billing_scope_id(),
718 identity.attempt_id(),
719 )
720 .await?
721 .ok_or(HostChargeApplicationError::InvalidState(
722 INVALID_HOST_CHARGE_STATE,
723 ))?;
724 transaction.commit().await?;
725 Ok(HostChargePaymentResult::new(attempt))
726}
727
728const fn host_charge_provider_cooldown_requested(cooldown: Option<RateLimitCooldown>) -> bool {
729 matches!(cooldown, Some(RateLimitCooldown::Provider))
730}
731
732async fn extend_host_charge_provider_cooldown(
733 connection: &mut PgConnection,
734 reservation: &HostChargeReservation,
735) -> Result<(), HostChargeApplicationError> {
736 let updated = sqlx::query(
737 r#"
738 UPDATE billing_gateway_provider_rate_limits
739 SET rate_limited_until = GREATEST(
740 rate_limited_until,
741 clock_timestamp() + make_interval(secs => $4)
742 )
743 WHERE provider_key = (
744 SELECT provider_key
745 FROM billing_gateway_accounts
746 WHERE id = $1 AND billing_scope_id = $2
747 AND gateway_configuration_id = $3
748 )
749 "#,
750 )
751 .bind(reservation.identity().gateway_account_id().as_uuid())
752 .bind(reservation.identity().billing_scope_id().as_uuid())
753 .bind(reservation.identity().gateway_configuration_id().as_uuid())
754 .bind(syrup_rail::RENEWAL_PROVIDER_RATE_LIMIT_RETRY_AFTER_SECONDS)
755 .execute(connection)
756 .await?;
757 if updated.rows_affected() != 1 {
758 return Err(HostChargeApplicationError::InvalidState(
759 INVALID_HOST_CHARGE_STATE,
760 ));
761 }
762 Ok(())
763}
764
765async fn park_host_charge_approved(
766 pool: &PgPool,
767 reservation: &HostChargeReservation,
768 evidence: &ProcessorEvidence,
769) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
770 let mut transaction = pool.begin().await?;
771 set_application_timeouts(&mut transaction).await?;
772 let attempt = lock_expected_host_charge(&mut transaction, reservation).await?;
773 let progression = if evidence.transaction_id().is_some() {
774 ProcessorChargeProgression::ExternalReversalRequired
775 } else {
776 ProcessorChargeProgression::ReconciliationRequired
777 };
778 let observation =
779 observe_processor_charge(&mut transaction, &attempt, evidence, progression).await?;
780 if let ObservedCharge::Owned(charge) = observation {
781 transition_charge(&mut transaction, charge.id, progression, None).await?;
782 }
783 let parked = park_locked_attempt(
784 &mut transaction,
785 &attempt,
786 evidence,
787 None,
788 APPROVED_STORAGE_FAILURE_TEXT,
789 )
790 .await?;
791 transaction.commit().await?;
792 Ok(HostChargePaymentResult::new(parked))
793}
794
795async fn durably_park_host_charge_approved(
796 pool: &PgPool,
797 reservation: &HostChargeReservation,
798 evidence: &ProcessorEvidence,
799) -> Result<HostChargePaymentResult, HostChargeApplicationError> {
800 if let Ok(payment) = park_host_charge_approved(pool, reservation, evidence).await {
801 return Ok(payment);
802 }
803 crate::store_compensating_processor_charge(
804 pool,
805 reservation.identity().attempt_id(),
806 reservation.request().gateway_order_id(),
807 evidence,
808 )
809 .await?;
810 let mut transaction = pool.begin().await?;
811 let attempt = crate::find_payment_attempt_by_id_in_transaction(
812 &mut transaction,
813 reservation.identity().billing_scope_id(),
814 reservation.identity().attempt_id(),
815 )
816 .await?
817 .ok_or(HostChargeApplicationError::InvalidState(
818 INVALID_HOST_CHARGE_STATE,
819 ))?;
820 transaction.commit().await?;
821 Ok(HostChargePaymentResult::confirmation_pending(
822 attempt,
823 evidence.clone(),
824 ))
825}
826
827fn evidence_looks_approved(evidence: &ProcessorEvidence) -> bool {
828 evidence.transaction_id().is_some()
829 && (evidence
830 .response()
831 .is_some_and(|value| syrup_rail::gateway_response_is_approved(Some(value.expose())))
832 || evidence
833 .condition()
834 .is_some_and(|value| syrup_rail::gateway_state_is_approved(value.expose())))
835}
836
837async fn lock_expected_host_charge(
838 connection: &mut PgConnection,
839 reservation: &HostChargeReservation,
840) -> Result<PaymentAttempt, HostChargeApplicationError> {
841 let identity = reservation.identity();
842 let attempt = lock_payment_attempt_by_id_on_connection(
843 connection,
844 identity.billing_scope_id(),
845 identity.attempt_id(),
846 )
847 .await?
848 .ok_or(HostChargeApplicationError::InvalidState(
849 INVALID_HOST_CHARGE_STATE,
850 ))?;
851 if attempt.identity() != identity || attempt.request() != reservation.request() {
852 return Err(HostChargeApplicationError::InvalidState(
853 INVALID_HOST_CHARGE_STATE,
854 ));
855 }
856 Ok(attempt)
857}
858
859#[cfg(test)]
860mod tests {
861 use std::{
862 error::Error,
863 sync::{
864 Arc,
865 atomic::{AtomicUsize, Ordering},
866 },
867 };
868
869 use async_trait::async_trait;
870 use chrono::Duration as ChronoDuration;
871 use sqlx::{Postgres, Transaction};
872 use syrup_rail::{
873 BillingContact, BillingEventKey, ChargeAmount, CurrencyCode, EndUserMutationAdmission,
874 EndUserMutationAdmissionResult, EndUserMutationCommand, GatewayAccountId,
875 GatewayAccountMode, GatewayConfigurationId, GatewayDiagnostic, GatewayError,
876 GatewayLifecycleCursorKey, GatewayLifecycleQueryPolicy, GatewayMutationError,
877 GatewayMutationReferenceFactory, GatewayOrderId, GatewayPaymentDescriptor,
878 GatewayProviderKey, GatewayQueryRequest, GatewayResolutionError, GatewayResolver,
879 GatewayStorePaymentMethodRequest, GatewayTransactionId, GatewayTransactionReport,
880 GatewayTransactionReportRequest, HostChargeTargetId, HostChargeTargetNoChange,
881 IdempotencyKey, PaymentAttemptId, PaymentAttemptKind, PaymentGateway, PaymentToken,
882 ResolvedGateway,
883 };
884 use tokio::sync::{Mutex, Notify, oneshot};
885 use uuid::Uuid;
886
887 use super::*;
888 use crate::{
889 BillingEventWriteError, BillingTransaction, HostChargeLedgerAdmission,
890 HostChargeLedgerAdmissionMode, HostChargeLedgerAdmissionQuery,
891 HostChargeReservationDecision, HostChargeReservationOutcome, HostChargeSubmissionAdmission,
892 HostChargeSubmissionDecision, HostChargeTargetReservation, SubscriptionBillingService,
893 SubscriptionOfferStore, host_charge_ledger_admission, reserve_host_charge_in_transaction,
894 test_support::{TestDatabase, create_gateway_account},
895 };
896
897 #[test]
898 fn before_submission_resolution_modes_keep_boundary_and_cooldown_distinct() {
899 let prepared = HostChargeBeforeSubmissionResolution::prepared();
900 assert_eq!(prepared.boundary, OutcomeResolutionBoundary::Prepared);
901 assert!(prepared.cooldown.is_none());
902
903 let admitted = HostChargeBeforeSubmissionResolution::admitted_not_submitted();
904 assert_eq!(
905 admitted.boundary,
906 OutcomeResolutionBoundary::AdmittedNotSubmitted
907 );
908 assert!(admitted.cooldown.is_none());
909
910 let rate_limited = HostChargeBeforeSubmissionResolution::prepared_provider_rate_limited();
911 assert_eq!(rate_limited.boundary, OutcomeResolutionBoundary::Prepared);
912 assert!(matches!(
913 rate_limited.cooldown,
914 Some(RateLimitCooldown::Provider)
915 ));
916 }
917
918 #[test]
919 fn host_charge_cooldown_only_extends_provider_scope() {
920 assert!(!host_charge_provider_cooldown_requested(None));
921 assert!(!host_charge_provider_cooldown_requested(Some(
922 RateLimitCooldown::Account
923 )));
924 assert!(host_charge_provider_cooldown_requested(Some(
925 RateLimitCooldown::Provider
926 )));
927 }
928
929 struct TestReferenceFactory;
930
931 impl GatewayMutationReferenceFactory for TestReferenceFactory {
932 fn for_attempt(
933 &self,
934 _kind: PaymentAttemptKind,
935 attempt_id: PaymentAttemptId,
936 ) -> GatewayOrderId {
937 GatewayOrderId::from_generated_attempt(
938 format!("test_host_{}", attempt_id.as_uuid().simple()),
939 attempt_id,
940 )
941 .expect("valid host test reference")
942 }
943 }
944
945 struct ScriptedGateway {
946 sale_calls: AtomicUsize,
947 outcome: Mutex<Option<GatewayPaymentOutcome>>,
948 }
949
950 #[async_trait]
951 impl PaymentGateway for ScriptedGateway {
952 async fn account_mode(&self) -> Result<GatewayAccountMode, GatewayError> {
953 Ok(GatewayAccountMode::Live)
954 }
955
956 async fn sale(
957 &self,
958 _request: GatewaySaleRequest,
959 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
960 self.sale_calls.fetch_add(1, Ordering::SeqCst);
961 Ok(self
962 .outcome
963 .lock()
964 .await
965 .take()
966 .expect("one sale capability"))
967 }
968
969 async fn store_payment_method(
970 &self,
971 _request: GatewayStorePaymentMethodRequest,
972 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
973 panic!("host charge must not store a payment method")
974 }
975
976 async fn query_transaction(
977 &self,
978 _request: GatewayQueryRequest,
979 ) -> Result<Option<GatewayPaymentOutcome>, GatewayError> {
980 panic!("foreground host charge must not query")
981 }
982
983 async fn query_transaction_reports(
984 &self,
985 _request: GatewayTransactionReportRequest,
986 ) -> Result<Vec<GatewayTransactionReport>, GatewayError> {
987 panic!("foreground host charge must not query reports")
988 }
989 }
990
991 struct RateLimitedAfterReservationGateway {
992 readiness_calls: AtomicUsize,
993 sale_calls: AtomicUsize,
994 }
995
996 struct TerminalRaceGateway {
997 pool: PgPool,
998 sale_calls: AtomicUsize,
999 }
1000
1001 struct RacingPreparedRetryGateway {
1002 readiness_calls: AtomicUsize,
1003 sale_calls: AtomicUsize,
1004 blocked_readiness_started: Mutex<Option<oneshot::Sender<()>>>,
1005 sale_started: Mutex<Option<oneshot::Sender<()>>>,
1006 release_readiness: Notify,
1007 release_sale: Notify,
1008 }
1009
1010 #[async_trait]
1011 impl PaymentGateway for RacingPreparedRetryGateway {
1012 async fn account_mode(&self) -> Result<GatewayAccountMode, GatewayError> {
1013 if self.readiness_calls.fetch_add(1, Ordering::SeqCst) == 1 {
1014 if let Some(started) = self.blocked_readiness_started.lock().await.take() {
1015 let _ = started.send(());
1016 }
1017 self.release_readiness.notified().await;
1018 Ok(GatewayAccountMode::Test)
1019 } else {
1020 Ok(GatewayAccountMode::Live)
1021 }
1022 }
1023
1024 async fn sale(
1025 &self,
1026 _request: GatewaySaleRequest,
1027 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
1028 self.sale_calls.fetch_add(1, Ordering::SeqCst);
1029 if let Some(started) = self.sale_started.lock().await.take() {
1030 let _ = started.send(());
1031 }
1032 self.release_sale.notified().await;
1033 Ok(approved_outcome("host_txn_prepared_retry"))
1034 }
1035
1036 async fn store_payment_method(
1037 &self,
1038 _request: GatewayStorePaymentMethodRequest,
1039 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
1040 panic!("host charge must not store a payment method")
1041 }
1042
1043 async fn query_transaction(
1044 &self,
1045 _request: GatewayQueryRequest,
1046 ) -> Result<Option<GatewayPaymentOutcome>, GatewayError> {
1047 panic!("foreground host charge must not query")
1048 }
1049
1050 async fn query_transaction_reports(
1051 &self,
1052 _request: GatewayTransactionReportRequest,
1053 ) -> Result<Vec<GatewayTransactionReport>, GatewayError> {
1054 panic!("foreground host charge must not query reports")
1055 }
1056 }
1057
1058 #[async_trait]
1059 impl PaymentGateway for TerminalRaceGateway {
1060 async fn account_mode(&self) -> Result<GatewayAccountMode, GatewayError> {
1061 Ok(GatewayAccountMode::Live)
1062 }
1063
1064 async fn sale(
1065 &self,
1066 request: GatewaySaleRequest,
1067 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
1068 self.sale_calls.fetch_add(1, Ordering::SeqCst);
1069 let updated = sqlx::query(
1070 r#"
1071 UPDATE billing_payment_attempts
1072 SET status = 'failed', resolved_at = clock_timestamp(),
1073 gateway_response_text = 'simulated terminal race',
1074 gateway_condition = 'failed', updated_at = clock_timestamp()
1075 WHERE gateway_order_id = $1 AND status = 'pending'
1076 "#,
1077 )
1078 .bind(request.order_id().expose())
1079 .execute(&self.pool)
1080 .await
1081 .expect("simulate a terminal attempt race");
1082 assert_eq!(updated.rows_affected(), 1);
1083 Ok(approved_outcome("host_txn_terminal_race"))
1084 }
1085
1086 async fn store_payment_method(
1087 &self,
1088 _request: GatewayStorePaymentMethodRequest,
1089 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
1090 panic!("host charge must not store a payment method")
1091 }
1092
1093 async fn query_transaction(
1094 &self,
1095 _request: GatewayQueryRequest,
1096 ) -> Result<Option<GatewayPaymentOutcome>, GatewayError> {
1097 panic!("foreground host charge must not query")
1098 }
1099
1100 async fn query_transaction_reports(
1101 &self,
1102 _request: GatewayTransactionReportRequest,
1103 ) -> Result<Vec<GatewayTransactionReport>, GatewayError> {
1104 panic!("foreground host charge must not query reports")
1105 }
1106 }
1107
1108 #[async_trait]
1109 impl PaymentGateway for RateLimitedAfterReservationGateway {
1110 async fn account_mode(&self) -> Result<GatewayAccountMode, GatewayError> {
1111 if self.readiness_calls.fetch_add(1, Ordering::SeqCst) == 0 {
1112 Ok(GatewayAccountMode::Live)
1113 } else {
1114 Err(GatewayError::RateLimited(GatewayDiagnostic::new(
1115 "provider throttled the readiness check",
1116 )))
1117 }
1118 }
1119
1120 async fn sale(
1121 &self,
1122 _request: GatewaySaleRequest,
1123 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
1124 self.sale_calls.fetch_add(1, Ordering::SeqCst);
1125 panic!("rate-limited host charge must not submit a sale")
1126 }
1127
1128 async fn store_payment_method(
1129 &self,
1130 _request: GatewayStorePaymentMethodRequest,
1131 ) -> Result<GatewayPaymentOutcome, GatewayMutationError> {
1132 panic!("host charge must not store a payment method")
1133 }
1134
1135 async fn query_transaction(
1136 &self,
1137 _request: GatewayQueryRequest,
1138 ) -> Result<Option<GatewayPaymentOutcome>, GatewayError> {
1139 panic!("foreground host charge must not query")
1140 }
1141
1142 async fn query_transaction_reports(
1143 &self,
1144 _request: GatewayTransactionReportRequest,
1145 ) -> Result<Vec<GatewayTransactionReport>, GatewayError> {
1146 panic!("foreground host charge must not query reports")
1147 }
1148 }
1149
1150 struct StaticResolver {
1151 gateway: ResolvedGateway,
1152 calls: AtomicUsize,
1153 }
1154
1155 #[async_trait]
1156 impl GatewayResolver for StaticResolver {
1157 async fn resolve(
1158 &self,
1159 billing_scope_id: syrup_rail::BillingScopeId,
1160 gateway_account_id: GatewayAccountId,
1161 gateway_configuration_id: GatewayConfigurationId,
1162 provider_key: GatewayProviderKey,
1163 ) -> Result<ResolvedGateway, GatewayResolutionError> {
1164 self.calls.fetch_add(1, Ordering::SeqCst);
1165 if billing_scope_id != self.gateway.billing_scope_id()
1166 || gateway_account_id != self.gateway.gateway_account_id()
1167 || gateway_configuration_id != self.gateway.gateway_configuration_id()
1168 || provider_key != *self.gateway.provider_key()
1169 {
1170 return Err(GatewayResolutionError::ConfigurationChanged);
1171 }
1172 Ok(self.gateway.clone())
1173 }
1174 }
1175
1176 struct PermitAdmission {
1177 calls: AtomicUsize,
1178 }
1179
1180 #[async_trait]
1181 impl EndUserMutationAdmission for PermitAdmission {
1182 async fn admit(&self, _command: EndUserMutationCommand) -> EndUserMutationAdmissionResult {
1183 self.calls.fetch_add(1, Ordering::SeqCst);
1184 EndUserMutationAdmissionResult::Allowed
1185 }
1186 }
1187
1188 struct UnusedOffers;
1189
1190 #[async_trait]
1191 impl SubscriptionOfferStore for UnusedOffers {
1192 async fn lock_current_offer(
1193 &self,
1194 _connection: &mut PgConnection,
1195 _billing_scope_id: syrup_rail::BillingScopeId,
1196 _plan_key: &syrup_rail::PlanKey,
1197 ) -> Result<Option<syrup_rail::SubscriptionOffer>, sqlx::Error> {
1198 panic!("host charge must not load a subscription offer")
1199 }
1200 }
1201
1202 #[derive(Clone)]
1203 struct TestCoordinator {
1204 pool: PgPool,
1205 events: Arc<Mutex<Vec<BillingEvent>>>,
1206 }
1207
1208 #[async_trait]
1209 impl BillingTransactionCoordinator for TestCoordinator {
1210 async fn begin(
1211 &self,
1212 _subject: BillingEventSubject,
1213 _lock_timeout: Duration,
1214 ) -> Result<Box<dyn BillingTransaction>, BillingTransactionError> {
1215 Ok(Box::new(TestTransaction {
1216 transaction: Some(
1217 self.pool
1218 .begin()
1219 .await
1220 .map_err(BillingTransactionError::new)?,
1221 ),
1222 events: Arc::clone(&self.events),
1223 }))
1224 }
1225 }
1226
1227 struct TestTransaction {
1228 transaction: Option<Transaction<'static, Postgres>>,
1229 events: Arc<Mutex<Vec<BillingEvent>>>,
1230 }
1231
1232 #[async_trait]
1233 impl BillingTransaction for TestTransaction {
1234 fn connection(&mut self) -> &mut PgConnection {
1235 &mut *self.transaction.as_mut().expect("active transaction")
1236 }
1237
1238 fn subject_state(&self) -> BillingTransactionSubjectState {
1239 BillingTransactionSubjectState::LiveRecipient
1240 }
1241
1242 async fn append_event(
1243 &mut self,
1244 event: &BillingEvent,
1245 ) -> Result<(), BillingEventWriteError> {
1246 self.events.lock().await.push(event.clone());
1247 Ok(())
1248 }
1249
1250 async fn commit(mut self: Box<Self>) -> Result<(), BillingTransactionError> {
1251 self.transaction
1252 .take()
1253 .expect("active transaction")
1254 .commit()
1255 .await
1256 .map_err(BillingTransactionError::new)
1257 }
1258
1259 async fn rollback(mut self: Box<Self>) -> Result<(), BillingTransactionError> {
1260 self.transaction
1261 .take()
1262 .expect("active transaction")
1263 .rollback()
1264 .await
1265 .map_err(BillingTransactionError::new)
1266 }
1267 }
1268
1269 struct TestTargets;
1270
1271 #[async_trait]
1272 impl HostChargeTargetStore for TestTargets {
1273 async fn preflight_target(
1274 &self,
1275 connection: &mut PgConnection,
1276 reservation: &HostChargeTargetReservation,
1277 ) -> Result<HostChargeReservationDecision, HostChargeTargetError> {
1278 self.reserve_target(connection, reservation).await
1279 }
1280
1281 async fn reserve_target(
1282 &self,
1283 connection: &mut PgConnection,
1284 reservation: &HostChargeTargetReservation,
1285 ) -> Result<HostChargeReservationDecision, HostChargeTargetError> {
1286 let row = sqlx::query_as::<_, (String, i32, String)>(
1287 r#"
1288 SELECT status, amount_cents, currency
1289 FROM host_charge_targets
1290 WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
1291 FOR UPDATE
1292 "#,
1293 )
1294 .bind(reservation.target_id().as_uuid())
1295 .bind(reservation.billing_scope_id().as_uuid())
1296 .bind(reservation.subscriber_id().as_uuid())
1297 .fetch_optional(&mut *connection)
1298 .await
1299 .map_err(HostChargeTargetError::new)?;
1300 let Some((status, cents, currency)) = row else {
1301 return Ok(HostChargeReservationDecision::Rejected {
1302 reason: syrup_rail::HostChargeTargetRejection::TargetUnavailable,
1303 });
1304 };
1305 let ledger = host_charge_ledger_admission(
1306 connection,
1307 &HostChargeLedgerAdmissionQuery::new(
1308 reservation.billing_scope_id(),
1309 reservation.subscriber_id(),
1310 reservation.target_id(),
1311 HostChargeLedgerAdmissionMode::Reserve {
1312 idempotency_key: reservation.idempotency_key().clone(),
1313 },
1314 ),
1315 )
1316 .await
1317 .map_err(HostChargeTargetError::new)?;
1318 if ledger == HostChargeLedgerAdmission::IdempotentContender {
1319 return Ok(HostChargeReservationDecision::IdempotentContender);
1320 }
1321 if ledger != HostChargeLedgerAdmission::Safe || status != "pending" {
1322 return Ok(HostChargeReservationDecision::Rejected {
1323 reason: syrup_rail::HostChargeTargetRejection::LedgerUnsafe,
1324 });
1325 }
1326 let charge = ChargeAmount::new(cents, CurrencyCode::new(¤cy).unwrap()).unwrap();
1327 Ok(HostChargeReservationDecision::Reserved(
1328 syrup_rail::HostChargeTargetSnapshot::new(reservation.target_id(), charge),
1329 ))
1330 }
1331
1332 async fn admit_submission(
1333 &self,
1334 connection: &mut PgConnection,
1335 admission: &HostChargeSubmissionAdmission,
1336 ) -> Result<HostChargeSubmissionDecision, HostChargeTargetError> {
1337 let row = sqlx::query_as::<_, (String, i32, String)>(
1338 r#"
1339 SELECT status, amount_cents, currency
1340 FROM host_charge_targets
1341 WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
1342 FOR UPDATE
1343 "#,
1344 )
1345 .bind(admission.target_id().as_uuid())
1346 .bind(admission.billing_scope_id().as_uuid())
1347 .bind(admission.subscriber_id().as_uuid())
1348 .fetch_optional(&mut *connection)
1349 .await
1350 .map_err(HostChargeTargetError::new)?;
1351 let Some((status, cents, currency)) = row else {
1352 return Ok(HostChargeSubmissionDecision::Rejected {
1353 reason: syrup_rail::HostChargeTargetRejection::TargetUnavailable,
1354 });
1355 };
1356 let charge = ChargeAmount::new(cents, CurrencyCode::new(¤cy).unwrap()).unwrap();
1357 let ledger = host_charge_ledger_admission(
1358 connection,
1359 &HostChargeLedgerAdmissionQuery::new(
1360 admission.billing_scope_id(),
1361 admission.subscriber_id(),
1362 admission.target_id(),
1363 HostChargeLedgerAdmissionMode::Submit {
1364 attempt_id: admission.attempt_id(),
1365 },
1366 ),
1367 )
1368 .await
1369 .map_err(HostChargeTargetError::new)?;
1370 if ledger == HostChargeLedgerAdmission::Safe
1371 && status == "pending"
1372 && charge == admission.expected_charge()
1373 {
1374 Ok(HostChargeSubmissionDecision::Admitted(
1375 syrup_rail::HostChargeTargetSnapshot::new(admission.target_id(), charge),
1376 ))
1377 } else {
1378 Ok(HostChargeSubmissionDecision::Rejected {
1379 reason: syrup_rail::HostChargeTargetRejection::ChargeChanged,
1380 })
1381 }
1382 }
1383
1384 async fn apply_transition(
1385 &self,
1386 connection: &mut PgConnection,
1387 transition: HostChargeTargetTransition,
1388 ) -> Result<HostChargeTargetTransitionOutcome, HostChargeTargetError> {
1389 let current: Option<String> = sqlx::query_scalar(
1390 r#"
1391 SELECT status FROM host_charge_targets
1392 WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
1393 FOR UPDATE
1394 "#,
1395 )
1396 .bind(transition.target_id().as_uuid())
1397 .bind(transition.billing_scope_id().as_uuid())
1398 .bind(transition.subscriber_id().as_uuid())
1399 .fetch_optional(&mut *connection)
1400 .await
1401 .map_err(HostChargeTargetError::new)?;
1402 let Some(current) = current else {
1403 return Ok(HostChargeTargetTransitionOutcome::StaleTarget);
1404 };
1405 match transition.kind() {
1406 HostChargeTargetTransitionKind::Paid if current == "pending" => {
1407 sqlx::query(
1408 "UPDATE host_charge_targets SET status = 'paid', paid_at = $2 WHERE id = $1",
1409 )
1410 .bind(transition.target_id().as_uuid())
1411 .bind(transition.effective_at())
1412 .execute(connection)
1413 .await
1414 .map_err(HostChargeTargetError::new)?;
1415 Ok(HostChargeTargetTransitionOutcome::Applied)
1416 }
1417 HostChargeTargetTransitionKind::Paid if current == "paid" => {
1418 Ok(HostChargeTargetTransitionOutcome::ExactReplay)
1419 }
1420 HostChargeTargetTransitionKind::Paid => {
1421 Ok(HostChargeTargetTransitionOutcome::StaleTarget)
1422 }
1423 HostChargeTargetTransitionKind::PaymentFailed if current == "pending" => {
1424 Ok(HostChargeTargetTransitionOutcome::Applied)
1425 }
1426 _ => Ok(HostChargeTargetTransitionOutcome::Unchanged {
1427 reason: HostChargeTargetNoChange::InapplicableState,
1428 }),
1429 }
1430 }
1431 }
1432
1433 fn resolved_gateway(
1434 account: crate::test_support::GatewayAccountFixture,
1435 gateway: Arc<dyn PaymentGateway>,
1436 ) -> ResolvedGateway {
1437 ResolvedGateway::new(
1438 syrup_rail::BillingScopeId::new(account.billing_scope_id),
1439 GatewayAccountId::new(account.gateway_account_id),
1440 GatewayConfigurationId::new(account.gateway_configuration_id),
1441 GatewayProviderKey::new("nmi").unwrap(),
1442 GatewayLifecycleQueryPolicy::new(
1443 GatewayLifecycleCursorKey::new("host_test").unwrap(),
1444 ChronoDuration::minutes(1),
1445 10,
1446 2,
1447 2,
1448 20,
1449 )
1450 .unwrap(),
1451 Arc::new(TestReferenceFactory),
1452 gateway,
1453 )
1454 }
1455
1456 fn approved_outcome(transaction_id: &str) -> GatewayPaymentOutcome {
1457 GatewayPaymentOutcome::new(
1458 GatewayPaymentStatus::Approved,
1459 ProcessorEvidence::new(
1460 Some(GatewayTransactionId::new(transaction_id).unwrap()),
1461 None,
1462 Some(GatewayDiagnostic::new("1")),
1463 None,
1464 Some(GatewayDiagnostic::new("approved")),
1465 Some(GatewayDiagnostic::new("complete")),
1466 GatewayPaymentDescriptor::default(),
1467 ),
1468 )
1469 }
1470
1471 #[tokio::test]
1472 async fn foreground_host_charge_is_one_shot_atomic_and_replay_first()
1473 -> Result<(), Box<dyn Error>> {
1474 let database = TestDatabase::start("rail_host_svc").await?;
1475 let result = async {
1476 sqlx::query(
1477 r#"
1478 CREATE TABLE host_charge_targets (
1479 id uuid PRIMARY KEY,
1480 billing_scope_id uuid NOT NULL,
1481 subscriber_id uuid NOT NULL,
1482 status text NOT NULL,
1483 amount_cents integer NOT NULL,
1484 currency text NOT NULL,
1485 paid_at timestamptz
1486 )
1487 "#,
1488 )
1489 .execute(&database.pool)
1490 .await?;
1491 let account = create_gateway_account(&database.pool, "nmi").await?;
1492 let subscriber_id = Uuid::now_v7();
1493 let target_id = Uuid::now_v7();
1494 sqlx::query(
1495 "INSERT INTO host_charge_targets VALUES ($1, $2, $3, 'pending', 1250, 'USD', NULL)",
1496 )
1497 .bind(target_id)
1498 .bind(account.billing_scope_id)
1499 .bind(subscriber_id)
1500 .execute(&database.pool)
1501 .await?;
1502
1503 let gateway = Arc::new(ScriptedGateway {
1504 sale_calls: AtomicUsize::new(0),
1505 outcome: Mutex::new(Some(approved_outcome("host_txn_approved"))),
1506 });
1507 let resolver = Arc::new(StaticResolver {
1508 gateway: resolved_gateway(account, gateway.clone()),
1509 calls: AtomicUsize::new(0),
1510 });
1511 let admission = Arc::new(PermitAdmission {
1512 calls: AtomicUsize::new(0),
1513 });
1514 let events = Arc::new(Mutex::new(Vec::new()));
1515 let coordinator = Arc::new(TestCoordinator {
1516 pool: database.pool.clone(),
1517 events: Arc::clone(&events),
1518 });
1519 let targets = Arc::new(TestTargets);
1520 let service = SubscriptionBillingService::new(
1521 database.pool.clone(),
1522 Arc::new(UnusedOffers),
1523 resolver.clone(),
1524 admission.clone(),
1525 coordinator,
1526 )
1527 .with_host_charge_targets(targets);
1528 let command = ChargeHostTarget::new(
1529 syrup_rail::BillingScopeId::new(account.billing_scope_id),
1530 syrup_rail::SubscriberId::new(subscriber_id),
1531 HostChargeTargetId::new(target_id),
1532 GatewayConfigurationId::new(account.gateway_configuration_id),
1533 PaymentToken::new("tok_host_once")?,
1534 IdempotencyKey::new("host-idempotency")?,
1535 Some(BillingContact::new(
1536 None,
1537 None,
1538 Some("host@example.test".into()),
1539 )?),
1540 );
1541
1542 let first = service.charge_host_target(command.clone()).await?;
1543 assert_eq!(first.status(), PaymentAttemptStatus::Approved);
1544 assert_eq!(gateway.sale_calls.load(Ordering::SeqCst), 1);
1545 assert_eq!(resolver.calls.load(Ordering::SeqCst), 1);
1546 assert_eq!(admission.calls.load(Ordering::SeqCst), 1);
1547 let replay = service.charge_host_target(command).await?;
1548 assert_eq!(replay.attempt(), first.attempt());
1549 assert_eq!(gateway.sale_calls.load(Ordering::SeqCst), 1);
1550 assert_eq!(resolver.calls.load(Ordering::SeqCst), 1);
1551 assert_eq!(admission.calls.load(Ordering::SeqCst), 1);
1552
1553 let target_status: String =
1554 sqlx::query_scalar("SELECT status FROM host_charge_targets WHERE id = $1")
1555 .bind(target_id)
1556 .fetch_one(&database.pool)
1557 .await?;
1558 assert_eq!(target_status, "paid");
1559 let charge_state: String = sqlx::query_scalar(
1560 "SELECT progression_state FROM billing_processor_charges WHERE attempt_id = $1",
1561 )
1562 .bind(first.attempt().identity().attempt_id().as_uuid())
1563 .fetch_one(&database.pool)
1564 .await?;
1565 assert_eq!(charge_state, "applied");
1566 let event_keys = events
1567 .lock()
1568 .await
1569 .iter()
1570 .map(BillingEvent::semantic_key)
1571 .collect::<Vec<_>>();
1572 assert_eq!(
1573 event_keys,
1574 vec![BillingEventKey::HostChargePaid(HostChargeTargetId::new(
1575 target_id
1576 ))]
1577 );
1578 Ok::<_, Box<dyn Error>>(())
1579 }
1580 .await;
1581 let cleanup = database.cleanup().await;
1582 result?;
1583 cleanup?;
1584 Ok(())
1585 }
1586
1587 #[tokio::test]
1588 async fn reservation_race_replays_durable_winner_request() -> Result<(), Box<dyn Error>> {
1589 let database = TestDatabase::start("rail_host_race").await?;
1590 let result = async {
1591 sqlx::query(
1592 r#"
1593 CREATE TABLE host_charge_targets (
1594 id uuid PRIMARY KEY,
1595 billing_scope_id uuid NOT NULL,
1596 subscriber_id uuid NOT NULL,
1597 status text NOT NULL,
1598 amount_cents integer NOT NULL,
1599 currency text NOT NULL,
1600 paid_at timestamptz
1601 )
1602 "#,
1603 )
1604 .execute(&database.pool)
1605 .await?;
1606 let account = create_gateway_account(&database.pool, "nmi").await?;
1607 let subscriber_id = Uuid::now_v7();
1608 let target_id = Uuid::now_v7();
1609 sqlx::query(
1610 "INSERT INTO host_charge_targets VALUES ($1, $2, $3, 'pending', 1250, 'USD', NULL)",
1611 )
1612 .bind(target_id)
1613 .bind(account.billing_scope_id)
1614 .bind(subscriber_id)
1615 .execute(&database.pool)
1616 .await?;
1617 let gateway = resolved_gateway(
1618 account,
1619 Arc::new(ScriptedGateway {
1620 sale_calls: AtomicUsize::new(0),
1621 outcome: Mutex::new(None),
1622 }),
1623 );
1624 let snapshot = syrup_rail::HostChargeTargetSnapshot::new(
1625 HostChargeTargetId::new(target_id),
1626 ChargeAmount::new(1250, CurrencyCode::new("USD")?)?,
1627 );
1628 let command = ChargeHostTarget::new(
1629 syrup_rail::BillingScopeId::new(account.billing_scope_id),
1630 syrup_rail::SubscriberId::new(subscriber_id),
1631 HostChargeTargetId::new(target_id),
1632 GatewayConfigurationId::new(account.gateway_configuration_id),
1633 PaymentToken::new("tok_host_winner")?,
1634 IdempotencyKey::new("host-reservation-race")?,
1635 Some(BillingContact::new(
1636 None,
1637 None,
1638 Some("winner@example.test".into()),
1639 )?),
1640 );
1641 let winner_id = PaymentAttemptId::new(Uuid::now_v7());
1642 let winner =
1643 HostChargeReservation::from_command(&command, snapshot, &gateway, winner_id)?;
1644 let mut transaction = database.pool.begin().await?;
1645 let outcome =
1646 reserve_host_charge_in_transaction(&mut transaction, &TestTargets, &winner).await?;
1647 assert!(matches!(outcome, HostChargeReservationOutcome::Reserved(_)));
1648 transaction.commit().await?;
1649
1650 let retry = ChargeHostTarget::new(
1651 command.billing_scope_id(),
1652 command.subscriber_id(),
1653 command.target_id(),
1654 command.gateway_configuration_id(),
1655 PaymentToken::new("tok_host_retry")?,
1656 command.idempotency_key().clone(),
1657 Some(BillingContact::new(
1658 None,
1659 None,
1660 Some("retry@example.test".into()),
1661 )?),
1662 );
1663 let contender = HostChargeReservation::from_command(
1664 &retry,
1665 snapshot,
1666 &gateway,
1667 PaymentAttemptId::new(Uuid::now_v7()),
1668 )?;
1669 let mut transaction = database.pool.begin().await?;
1670 let outcome =
1671 reserve_host_charge_in_transaction(&mut transaction, &TestTargets, &contender)
1672 .await?;
1673 transaction.commit().await?;
1674 let HostChargeReservationOutcome::Replay(attempt) = outcome else {
1675 panic!("matching contender should replay the durable winner");
1676 };
1677 assert_eq!(attempt.identity().attempt_id(), winner_id);
1678 assert_eq!(
1679 attempt.request().billing_contact().email(),
1680 Some("winner@example.test")
1681 );
1682 Ok::<_, Box<dyn Error>>(())
1683 }
1684 .await;
1685 let cleanup = database.cleanup().await;
1686 result?;
1687 cleanup?;
1688 Ok(())
1689 }
1690
1691 #[tokio::test]
1692 async fn post_reservation_rate_limit_resolves_attempt_and_extends_provider_cooldown()
1693 -> Result<(), Box<dyn Error>> {
1694 let database = TestDatabase::start("rail_host_rate").await?;
1695 let result = async {
1696 sqlx::query(
1697 r#"
1698 CREATE TABLE host_charge_targets (
1699 id uuid PRIMARY KEY,
1700 billing_scope_id uuid NOT NULL,
1701 subscriber_id uuid NOT NULL,
1702 status text NOT NULL,
1703 amount_cents integer NOT NULL,
1704 currency text NOT NULL,
1705 paid_at timestamptz
1706 )
1707 "#,
1708 )
1709 .execute(&database.pool)
1710 .await?;
1711 let account = create_gateway_account(&database.pool, "nmi").await?;
1712 let subscriber_id = Uuid::now_v7();
1713 let target_id = Uuid::now_v7();
1714 sqlx::query(
1715 "INSERT INTO host_charge_targets VALUES ($1, $2, $3, 'pending', 1250, 'USD', NULL)",
1716 )
1717 .bind(target_id)
1718 .bind(account.billing_scope_id)
1719 .bind(subscriber_id)
1720 .execute(&database.pool)
1721 .await?;
1722
1723 let gateway = Arc::new(RateLimitedAfterReservationGateway {
1724 readiness_calls: AtomicUsize::new(0),
1725 sale_calls: AtomicUsize::new(0),
1726 });
1727 let resolver = Arc::new(StaticResolver {
1728 gateway: resolved_gateway(account, gateway.clone()),
1729 calls: AtomicUsize::new(0),
1730 });
1731 let admission = Arc::new(PermitAdmission {
1732 calls: AtomicUsize::new(0),
1733 });
1734 let service = SubscriptionBillingService::new(
1735 database.pool.clone(),
1736 Arc::new(UnusedOffers),
1737 resolver,
1738 admission,
1739 Arc::new(TestCoordinator {
1740 pool: database.pool.clone(),
1741 events: Arc::new(Mutex::new(Vec::new())),
1742 }),
1743 )
1744 .with_host_charge_targets(Arc::new(TestTargets));
1745 let command = ChargeHostTarget::new(
1746 syrup_rail::BillingScopeId::new(account.billing_scope_id),
1747 syrup_rail::SubscriberId::new(subscriber_id),
1748 HostChargeTargetId::new(target_id),
1749 GatewayConfigurationId::new(account.gateway_configuration_id),
1750 PaymentToken::new("tok_host_throttled")?,
1751 IdempotencyKey::new("host-throttled")?,
1752 None,
1753 );
1754
1755 let error = service
1756 .charge_host_target(command)
1757 .await
1758 .expect_err("provider throttle must be reported");
1759 assert!(matches!(
1760 error,
1761 crate::SubscriptionBillingServiceError::GatewayMutationCooldown {
1762 scope: crate::GatewayMutationCooldownScope::Provider
1763 }
1764 ));
1765 assert_eq!(gateway.readiness_calls.load(Ordering::SeqCst), 2);
1766 assert_eq!(gateway.sale_calls.load(Ordering::SeqCst), 0);
1767
1768 let attempt = sqlx::query_as::<_, (String, Option<String>, Option<DateTime<Utc>>)>(
1769 r#"
1770 SELECT status, resolution_code, submitted_at
1771 FROM billing_payment_attempts
1772 WHERE billing_scope_id = $1 AND subscriber_id = $2
1773 "#,
1774 )
1775 .bind(account.billing_scope_id)
1776 .bind(subscriber_id)
1777 .fetch_one(&database.pool)
1778 .await?;
1779 assert_eq!(attempt.0, "failed");
1780 assert_eq!(
1781 attempt.1.as_deref(),
1782 Some("gateway_provider_rate_limited_before_submission")
1783 );
1784 assert!(attempt.2.is_none());
1785 let cooldown_is_active: bool = sqlx::query_scalar(
1786 "SELECT rate_limited_until > clock_timestamp() FROM billing_gateway_provider_rate_limits WHERE provider_key = 'nmi'",
1787 )
1788 .fetch_one(&database.pool)
1789 .await?;
1790 assert!(cooldown_is_active);
1791 let target_status: String =
1792 sqlx::query_scalar("SELECT status FROM host_charge_targets WHERE id = $1")
1793 .bind(target_id)
1794 .fetch_one(&database.pool)
1795 .await?;
1796 assert_eq!(target_status, "pending");
1797 Ok::<_, Box<dyn Error>>(())
1798 }
1799 .await;
1800 let cleanup = database.cleanup().await;
1801 result?;
1802 cleanup?;
1803 Ok(())
1804 }
1805
1806 #[tokio::test]
1807 async fn terminal_approval_race_does_not_mark_host_target_paid() -> Result<(), Box<dyn Error>> {
1808 let database = TestDatabase::start("rail_host_race").await?;
1809 let result = async {
1810 sqlx::query(
1811 r#"
1812 CREATE TABLE host_charge_targets (
1813 id uuid PRIMARY KEY,
1814 billing_scope_id uuid NOT NULL,
1815 subscriber_id uuid NOT NULL,
1816 status text NOT NULL,
1817 amount_cents integer NOT NULL,
1818 currency text NOT NULL,
1819 paid_at timestamptz
1820 )
1821 "#,
1822 )
1823 .execute(&database.pool)
1824 .await?;
1825 let account = create_gateway_account(&database.pool, "nmi").await?;
1826 let subscriber_id = Uuid::now_v7();
1827 let target_id = Uuid::now_v7();
1828 sqlx::query(
1829 "INSERT INTO host_charge_targets VALUES ($1, $2, $3, 'pending', 1250, 'USD', NULL)",
1830 )
1831 .bind(target_id)
1832 .bind(account.billing_scope_id)
1833 .bind(subscriber_id)
1834 .execute(&database.pool)
1835 .await?;
1836
1837 let gateway = Arc::new(TerminalRaceGateway {
1838 pool: database.pool.clone(),
1839 sale_calls: AtomicUsize::new(0),
1840 });
1841 let resolver = Arc::new(StaticResolver {
1842 gateway: resolved_gateway(account, gateway.clone()),
1843 calls: AtomicUsize::new(0),
1844 });
1845 let events = Arc::new(Mutex::new(Vec::new()));
1846 let service = SubscriptionBillingService::new(
1847 database.pool.clone(),
1848 Arc::new(UnusedOffers),
1849 resolver,
1850 Arc::new(PermitAdmission {
1851 calls: AtomicUsize::new(0),
1852 }),
1853 Arc::new(TestCoordinator {
1854 pool: database.pool.clone(),
1855 events: Arc::clone(&events),
1856 }),
1857 )
1858 .with_host_charge_targets(Arc::new(TestTargets));
1859 let payment = service
1860 .charge_host_target(ChargeHostTarget::new(
1861 syrup_rail::BillingScopeId::new(account.billing_scope_id),
1862 syrup_rail::SubscriberId::new(subscriber_id),
1863 HostChargeTargetId::new(target_id),
1864 GatewayConfigurationId::new(account.gateway_configuration_id),
1865 PaymentToken::new("tok_host_race")?,
1866 IdempotencyKey::new("host-race")?,
1867 None,
1868 ))
1869 .await?;
1870
1871 assert_eq!(payment.status(), PaymentAttemptStatus::Unknown);
1872 assert_eq!(payment.attempt().status(), PaymentAttemptStatus::Failed);
1873 assert_eq!(gateway.sale_calls.load(Ordering::SeqCst), 1);
1874 assert!(events.lock().await.is_empty());
1875 let target_status: String =
1876 sqlx::query_scalar("SELECT status FROM host_charge_targets WHERE id = $1")
1877 .bind(target_id)
1878 .fetch_one(&database.pool)
1879 .await?;
1880 assert_eq!(target_status, "pending");
1881 let charge_state: String = sqlx::query_scalar(
1882 "SELECT progression_state FROM billing_processor_charges WHERE attempt_id = $1",
1883 )
1884 .bind(payment.attempt().identity().attempt_id().as_uuid())
1885 .fetch_one(&database.pool)
1886 .await?;
1887 assert_eq!(charge_state, "pending");
1888 Ok::<_, Box<dyn Error>>(())
1889 }
1890 .await;
1891 let cleanup = database.cleanup().await;
1892 result?;
1893 cleanup?;
1894 Ok(())
1895 }
1896
1897 #[tokio::test]
1898 async fn matching_retry_resumes_prepared_attempt_without_readiness_loser_overwrite()
1899 -> Result<(), Box<dyn Error>> {
1900 let database = TestDatabase::start("rail_host_resume").await?;
1901 let result = async {
1902 sqlx::query(
1903 r#"
1904 CREATE TABLE host_charge_targets (
1905 id uuid PRIMARY KEY,
1906 billing_scope_id uuid NOT NULL,
1907 subscriber_id uuid NOT NULL,
1908 status text NOT NULL,
1909 amount_cents integer NOT NULL,
1910 currency text NOT NULL,
1911 paid_at timestamptz
1912 )
1913 "#,
1914 )
1915 .execute(&database.pool)
1916 .await?;
1917 let account = create_gateway_account(&database.pool, "nmi").await?;
1918 let subscriber_id = Uuid::now_v7();
1919 let target_id = Uuid::now_v7();
1920 sqlx::query(
1921 "INSERT INTO host_charge_targets VALUES ($1, $2, $3, 'pending', 1250, 'USD', NULL)",
1922 )
1923 .bind(target_id)
1924 .bind(account.billing_scope_id)
1925 .bind(subscriber_id)
1926 .execute(&database.pool)
1927 .await?;
1928
1929 let (readiness_started_tx, readiness_started_rx) = oneshot::channel();
1930 let (sale_started_tx, sale_started_rx) = oneshot::channel();
1931 let gateway = Arc::new(RacingPreparedRetryGateway {
1932 readiness_calls: AtomicUsize::new(0),
1933 sale_calls: AtomicUsize::new(0),
1934 blocked_readiness_started: Mutex::new(Some(readiness_started_tx)),
1935 sale_started: Mutex::new(Some(sale_started_tx)),
1936 release_readiness: Notify::new(),
1937 release_sale: Notify::new(),
1938 });
1939 let service = SubscriptionBillingService::new(
1940 database.pool.clone(),
1941 Arc::new(UnusedOffers),
1942 Arc::new(StaticResolver {
1943 gateway: resolved_gateway(account, gateway.clone()),
1944 calls: AtomicUsize::new(0),
1945 }),
1946 Arc::new(PermitAdmission {
1947 calls: AtomicUsize::new(0),
1948 }),
1949 Arc::new(TestCoordinator {
1950 pool: database.pool.clone(),
1951 events: Arc::new(Mutex::new(Vec::new())),
1952 }),
1953 )
1954 .with_host_charge_targets(Arc::new(TestTargets));
1955 let command = ChargeHostTarget::new(
1956 syrup_rail::BillingScopeId::new(account.billing_scope_id),
1957 syrup_rail::SubscriberId::new(subscriber_id),
1958 HostChargeTargetId::new(target_id),
1959 GatewayConfigurationId::new(account.gateway_configuration_id),
1960 PaymentToken::new("tok_host_resume")?,
1961 IdempotencyKey::new("host-resume")?,
1962 None,
1963 );
1964
1965 let first_service = service.clone();
1966 let first_command = command.clone();
1967 let first =
1968 tokio::spawn(async move { first_service.charge_host_target(first_command).await });
1969 readiness_started_rx.await?;
1970 let retry = ChargeHostTarget::new(
1971 command.billing_scope_id(),
1972 command.subscriber_id(),
1973 command.target_id(),
1974 command.gateway_configuration_id(),
1975 PaymentToken::new("tok_host_resume_retry")?,
1976 command.idempotency_key().clone(),
1977 Some(BillingContact::new(
1978 None,
1979 None,
1980 Some("retry@example.test".into()),
1981 )?),
1982 );
1983 let second_service = service.clone();
1984 let second =
1985 tokio::spawn(async move { second_service.charge_host_target(retry).await });
1986 sale_started_rx.await?;
1987 gateway.release_readiness.notify_one();
1988 let first = first.await??;
1989 assert_eq!(first.status(), PaymentAttemptStatus::Pending);
1990 assert!(
1991 first
1992 .attempt()
1993 .state()
1994 .timestamps()
1995 .submitted_at()
1996 .is_some()
1997 );
1998 gateway.release_sale.notify_one();
1999 let second = second.await??;
2000 assert_eq!(second.status(), PaymentAttemptStatus::Approved);
2001 assert_eq!(first.attempt().identity(), second.attempt().identity());
2002 assert_eq!(gateway.sale_calls.load(Ordering::SeqCst), 1);
2003 Ok::<_, Box<dyn Error>>(())
2004 }
2005 .await;
2006 let cleanup = database.cleanup().await;
2007 result?;
2008 cleanup?;
2009 Ok(())
2010 }
2011}