syrup_rail_postgres/

subscription_billing_service.rs

use std::{fmt, sync::Arc, time::Duration};

use sqlx::PgPool;
use syrup_rail::{
    BillingScopeId, ChargeHostTarget, ChargeRenewal, EndUserMutationAdmission,
    EndUserMutationAdmissionResult, EndUserMutationCommand, EndUserMutationOperation,
    EnrollSubscription, GatewayAccountId, GatewayAccountMode, GatewayDiagnostic, GatewayError,
    GatewayNotSubmittedError, GatewayPaymentDescriptor, GatewayPaymentOutcome, GatewayProviderKey,
    GatewayResolutionError, GatewayResolver, HostChargePaymentResult, HostChargeReservation,
    HostChargeTargetRejection, PaymentAttempt, PaymentAttemptId, PaymentAttemptKind,
    PaymentAttemptStatus, PaymentResolutionCode, ProcessorEvidence, RecoverSubscriptionPayment,
    ReplaceSubscriptionPaymentMethod, SubscriptionEnrollmentPaymentResult,
    SubscriptionEnrollmentPreflightOutcome, SubscriptionEnrollmentReservation,
    SubscriptionEnrollmentReservationBuildError, SubscriptionEnrollmentReservationOutcome,
    SubscriptionEnrollmentReservationRejection, SubscriptionEnrollmentSubmissionRejection,
    SubscriptionPaymentMethodReplacement, SubscriptionPaymentMethodReplacementPreflightOutcome,
    SubscriptionPaymentMethodReplacementRejection,
    SubscriptionPaymentMethodReplacementReservationOutcome,
    SubscriptionPaymentMethodReplacementSubmissionRejection, SubscriptionRecoveryPreflightOutcome,
    SubscriptionRecoveryReservation, SubscriptionRecoveryReservationOutcome,
    SubscriptionRecoveryReservationRejection, SubscriptionRecoverySubmissionRejection,
    SubscriptionRenewalOutcome, SubscriptionRenewalReservation,
    SubscriptionRenewalReservationOutcome, SubscriptionRenewalReservationRejection,
};
use thiserror::Error;

use crate::host_charge_application::{
    HostChargeBeforeSubmissionResolution, resolve_host_charge_before_submission,
};
use crate::{
    BillingTransactionCoordinator, HostChargeAdmissionOutcome, HostChargeApplicationError,
    HostChargePreflightOutcome, HostChargeProviderResult, HostChargeReservationOutcome,
    HostChargeStoreError, HostChargeTargetStore, PaymentAttemptStoreError,
    SubscriptionEnrollmentAdmissionOutcome, SubscriptionEnrollmentApplicationError,
    SubscriptionEnrollmentProviderResult, SubscriptionOfferStore,
    SubscriptionPaymentMethodReplacementAdmissionOutcome,
    SubscriptionPaymentMethodReplacementProviderResult, SubscriptionRecoveryAdmissionOutcome,
    SubscriptionRecoveryProviderResult, SubscriptionRenewalAdmissionOutcome,
    SubscriptionRenewalProviderResult, admit_host_charge_submission,
    admit_subscription_enrollment_submission, admit_subscription_payment_method_replacement,
    admit_subscription_recovery_submission, admit_subscription_renewal_submission,
    apply_reconciled_host_charge_gateway_outcome,
    apply_reconciled_subscription_enrollment_gateway_outcome,
    apply_reconciled_subscription_payment_method_replacement_gateway_outcome,
    apply_reconciled_subscription_recovery_gateway_outcome,
    apply_reconciled_subscription_renewal_gateway_outcome,
    enrollment_application::{
        OutcomeResolutionBoundary, RateLimitCooldown, payment_result_for_attempt,
        resolve_non_approved_outcome, resolve_payment_method_replacement_non_approved_outcome,
        resolve_recovery_non_approved_outcome, resolve_renewal_non_approved_outcome,
    },
    preflight_host_charge_in_transaction, preflight_subscription_enrollment_in_transaction,
    preflight_subscription_payment_method_replacement_in_transaction,
    preflight_subscription_recovery_in_transaction, reserve_host_charge_in_transaction,
    reserve_subscription_enrollment_in_transaction,
    reserve_subscription_payment_method_replacement_in_transaction,
    reserve_subscription_recovery_in_transaction, reserve_subscription_renewal_in_transaction,
    submit_admitted_host_charge, submit_admitted_subscription_enrollment,
    submit_admitted_subscription_payment_method_replacement, submit_admitted_subscription_recovery,
    submit_admitted_subscription_renewal,
};

const INVALID_SERVICE_STATE: &str = "canonical subscription enrollment service state is invalid";
const LIVE_READINESS_FAILED_TEXT: &str =
    "Payment was not submitted because the payment processor was not ready for live transactions.";

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum GatewayMutationCooldownScope {
    Account,
    Provider,
}

#[derive(Error)]
pub enum SubscriptionEnrollmentServiceError {
    #[error("subscription enrollment storage failed")]
    Sql(#[from] sqlx::Error),
    #[error("subscription enrollment attempt storage failed")]
    Attempt(#[from] PaymentAttemptStoreError),
    #[error("subscription enrollment application failed")]
    Application(#[from] SubscriptionEnrollmentApplicationError),
    #[error("host charge application failed")]
    HostChargeApplication(#[from] HostChargeApplicationError),
    #[error("host charge storage failed")]
    HostChargeStore(#[from] HostChargeStoreError),
    #[error("host charge capability is not configured")]
    HostChargeUnavailable,
    #[error("the idempotency key belongs to a different payment request")]
    IdempotencyConflict,
    #[error("end-user mutation admission was denied")]
    AdmissionDenied { retry_after: Duration },
    #[error("end-user mutation admission timed out")]
    AdmissionTimeout,
    #[error("end-user mutation admission is unavailable")]
    AdmissionUnavailable,
    #[error("gateway account or configuration changed")]
    GatewayConfigurationChanged,
    #[error("gateway resolution failed")]
    GatewayResolution(#[from] GatewayResolutionError),
    #[error("gateway resolver returned a different canonical identity")]
    ResolvedGatewayIdentityMismatch,
    #[error("gateway mutation cooldown is active")]
    GatewayMutationCooldown { scope: GatewayMutationCooldownScope },
    #[error("subscription enrollment reservation was rejected")]
    ReservationRejected(SubscriptionEnrollmentReservationRejection),
    #[error("subscription enrollment submission was rejected")]
    SubmissionRejected(SubscriptionEnrollmentSubmissionRejection),
    #[error("host charge reservation was rejected")]
    HostChargeReservationRejected(HostChargeTargetRejection),
    #[error("host charge submission was rejected")]
    HostChargeSubmissionRejected(HostChargeTargetRejection),
    #[error("subscription recovery reservation was rejected")]
    RecoveryReservationRejected(SubscriptionRecoveryReservationRejection),
    #[error("subscription recovery submission was rejected")]
    RecoverySubmissionRejected(SubscriptionRecoverySubmissionRejection),
    #[error("subscription renewal reservation was rejected")]
    RenewalReservationRejected(SubscriptionRenewalReservationRejection),
    #[error("subscription payment method replacement reservation was rejected")]
    PaymentMethodReplacementReservationRejected(SubscriptionPaymentMethodReplacementRejection),
    #[error("subscription payment method replacement submission was rejected")]
    PaymentMethodReplacementSubmissionRejected(
        SubscriptionPaymentMethodReplacementSubmissionRejection,
    ),
    #[error("gateway mutation was not submitted")]
    GatewayNotSubmitted(#[source] GatewayNotSubmittedError),
    #[error("gateway readiness check failed")]
    GatewayReadiness(#[source] GatewayError),
    #[error("{0}")]
    InvalidState(&'static str),
}

impl fmt::Debug for SubscriptionEnrollmentServiceError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Sql(_) => formatter.write_str("SubscriptionEnrollmentServiceError::Sql"),
            Self::Attempt(_) => formatter.write_str("SubscriptionEnrollmentServiceError::Attempt"),
            Self::Application(_) => {
                formatter.write_str("SubscriptionEnrollmentServiceError::Application")
            }
            Self::HostChargeApplication(_) => {
                formatter.write_str("SubscriptionEnrollmentServiceError::HostChargeApplication")
            }
            Self::HostChargeStore(_) => {
                formatter.write_str("SubscriptionEnrollmentServiceError::HostChargeStore")
            }
            Self::HostChargeUnavailable => formatter
                .write_str("SubscriptionEnrollmentServiceError::HostChargeUnavailable"),
            Self::IdempotencyConflict => {
                formatter.write_str("SubscriptionEnrollmentServiceError::IdempotencyConflict")
            }
            Self::AdmissionDenied { retry_after } => formatter
                .debug_struct("SubscriptionEnrollmentServiceError::AdmissionDenied")
                .field("retry_after", retry_after)
                .finish(),
            Self::AdmissionTimeout => {
                formatter.write_str("SubscriptionEnrollmentServiceError::AdmissionTimeout")
            }
            Self::AdmissionUnavailable => {
                formatter.write_str("SubscriptionEnrollmentServiceError::AdmissionUnavailable")
            }
            Self::GatewayConfigurationChanged => formatter
                .write_str("SubscriptionEnrollmentServiceError::GatewayConfigurationChanged"),
            Self::GatewayResolution(error) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::GatewayResolution")
                .field(error)
                .finish(),
            Self::ResolvedGatewayIdentityMismatch => formatter
                .write_str("SubscriptionEnrollmentServiceError::ResolvedGatewayIdentityMismatch"),
            Self::GatewayMutationCooldown { scope } => formatter
                .debug_struct("SubscriptionEnrollmentServiceError::GatewayMutationCooldown")
                .field("scope", scope)
                .finish(),
            Self::ReservationRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::ReservationRejected")
                .field(reason)
                .finish(),
            Self::SubmissionRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::SubmissionRejected")
                .field(reason)
                .finish(),
            Self::HostChargeReservationRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::HostChargeReservationRejected")
                .field(reason)
                .finish(),
            Self::HostChargeSubmissionRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::HostChargeSubmissionRejected")
                .field(reason)
                .finish(),
            Self::RecoveryReservationRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::RecoveryReservationRejected")
                .field(reason)
                .finish(),
            Self::RecoverySubmissionRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::RecoverySubmissionRejected")
                .field(reason)
                .finish(),
            Self::RenewalReservationRejected(reason) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::RenewalReservationRejected")
                .field(reason)
                .finish(),
            Self::PaymentMethodReplacementReservationRejected(reason) => formatter
                .debug_tuple(
                    "SubscriptionEnrollmentServiceError::PaymentMethodReplacementReservationRejected",
                )
                .field(reason)
                .finish(),
            Self::PaymentMethodReplacementSubmissionRejected(reason) => formatter
                .debug_tuple(
                    "SubscriptionEnrollmentServiceError::PaymentMethodReplacementSubmissionRejected",
                )
                .field(reason)
                .finish(),
            Self::GatewayNotSubmitted(error) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::GatewayNotSubmitted")
                .field(error)
                .finish(),
            Self::GatewayReadiness(error) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::GatewayReadiness")
                .field(error)
                .finish(),
            Self::InvalidState(detail) => formatter
                .debug_tuple("SubscriptionEnrollmentServiceError::InvalidState")
                .field(detail)
                .finish(),
        }
    }
}

#[derive(Clone)]
pub struct SubscriptionBillingService {
    pool: PgPool,
    offers: Arc<dyn SubscriptionOfferStore>,
    resolver: Arc<dyn GatewayResolver>,
    admission: Arc<dyn EndUserMutationAdmission>,
    coordinator: Arc<dyn BillingTransactionCoordinator>,
    host_charge_targets: Option<Arc<dyn HostChargeTargetStore>>,
}

impl SubscriptionBillingService {
    pub fn new(
        pool: PgPool,
        offers: Arc<dyn SubscriptionOfferStore>,
        resolver: Arc<dyn GatewayResolver>,
        admission: Arc<dyn EndUserMutationAdmission>,
        coordinator: Arc<dyn BillingTransactionCoordinator>,
    ) -> Self {
        Self {
            pool,
            offers,
            resolver,
            admission,
            coordinator,
            host_charge_targets: None,
        }
    }

    pub fn with_host_charge_targets(mut self, targets: Arc<dyn HostChargeTargetStore>) -> Self {
        self.host_charge_targets = Some(targets);
        self
    }

    /// Charges one host-owned target through the canonical attempt ledger.
    ///
    /// Target eligibility and economics are supplied by the configured host
    /// extension. No transaction or target lock spans gateway I/O.
    pub async fn charge_host_target(
        &self,
        command: ChargeHostTarget,
    ) -> Result<HostChargePaymentResult, SubscriptionEnrollmentServiceError> {
        let targets = self
            .host_charge_targets
            .as_deref()
            .ok_or(SubscriptionEnrollmentServiceError::HostChargeUnavailable)?;
        let (snapshot, prepared_reservation) =
            match self.preflight_host_charge(targets, &command).await? {
                HostChargePreflightOutcome::Continue(snapshot) => (snapshot, None),
                HostChargePreflightOutcome::Replay(attempt)
                    if attempt.status() == PaymentAttemptStatus::Pending
                        && attempt.state().timestamps().submitted_at().is_none() =>
                {
                    let reservation =
                        HostChargeReservation::from_attempt(&attempt).map_err(|_| {
                            SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE)
                        })?;
                    (reservation.snapshot(), Some(reservation))
                }
                HostChargePreflightOutcome::Replay(attempt) => {
                    return Ok(HostChargePaymentResult::new(*attempt));
                }
                HostChargePreflightOutcome::IdempotencyConflict => {
                    return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
                }
                HostChargePreflightOutcome::Rejected { reason } => {
                    return Err(
                        SubscriptionEnrollmentServiceError::HostChargeReservationRejected(reason),
                    );
                }
            };

        let account = self
            .gateway_account(
                command.billing_scope_id(),
                command.gateway_configuration_id(),
            )
            .await?;
        if prepared_reservation.is_none()
            && let Some(scope) = self.active_cooldown(&account).await?
        {
            return Err(SubscriptionEnrollmentServiceError::GatewayMutationCooldown { scope });
        }
        let gateway = self
            .resolver
            .resolve(
                command.billing_scope_id(),
                account.account_id,
                command.gateway_configuration_id(),
                account.provider_key.clone(),
            )
            .await?;
        if gateway.billing_scope_id() != command.billing_scope_id()
            || gateway.gateway_account_id() != account.account_id
            || gateway.gateway_configuration_id() != command.gateway_configuration_id()
            || gateway.provider_key() != &account.provider_key
        {
            return Err(SubscriptionEnrollmentServiceError::ResolvedGatewayIdentityMismatch);
        }
        if prepared_reservation.as_ref().is_some_and(|reservation| {
            reservation.identity().gateway_account_id() != gateway.gateway_account_id()
                || reservation.request().gateway_order_id()
                    != &gateway.mutation_reference_factory().for_attempt(
                        PaymentAttemptKind::HostCharge,
                        reservation.identity().attempt_id(),
                    )
        }) {
            return Err(SubscriptionEnrollmentServiceError::InvalidState(
                INVALID_SERVICE_STATE,
            ));
        }
        match gateway.account_mode().await {
            Ok(GatewayAccountMode::Live) => {}
            Ok(GatewayAccountMode::Test) => {
                return Err(SubscriptionEnrollmentServiceError::GatewayReadiness(
                    GatewayError::Configuration(GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT)),
                ));
            }
            Err(GatewayError::RateLimited(_)) => {
                self.extend_provider_cooldown(&account.provider_key).await?;
                return Err(
                    SubscriptionEnrollmentServiceError::GatewayMutationCooldown {
                        scope: GatewayMutationCooldownScope::Provider,
                    },
                );
            }
            Err(error) => return Err(SubscriptionEnrollmentServiceError::GatewayReadiness(error)),
        }

        if prepared_reservation.is_none() {
            self.admit_subscriber_mutation(
                command.billing_scope_id(),
                command.subscriber_id(),
                EndUserMutationOperation::HostCharge,
            )
            .await?;
        }

        let candidate_id = prepared_reservation.as_ref().map_or_else(
            || PaymentAttemptId::new(uuid::Uuid::now_v7()),
            |reservation| reservation.identity().attempt_id(),
        );
        let mut reservation = match prepared_reservation {
            Some(reservation) => reservation,
            None => HostChargeReservation::from_command(&command, snapshot, &gateway, candidate_id)
                .map_err(|_| {
                    SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE)
                })?,
        };
        let attempt = match self.reserve_host_charge(targets, &reservation).await? {
            HostChargeReservationOutcome::Reserved(attempt)
            | HostChargeReservationOutcome::Replay(attempt)
                if attempt.status() == PaymentAttemptStatus::Pending
                    && attempt.state().timestamps().submitted_at().is_none() =>
            {
                attempt
            }
            HostChargeReservationOutcome::Replay(attempt) => {
                return Ok(HostChargePaymentResult::new(attempt));
            }
            HostChargeReservationOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
            HostChargeReservationOutcome::Rejected { reason } => {
                return Err(
                    SubscriptionEnrollmentServiceError::HostChargeReservationRejected(reason),
                );
            }
            HostChargeReservationOutcome::Reserved(_) => {
                return Err(SubscriptionEnrollmentServiceError::InvalidState(
                    INVALID_SERVICE_STATE,
                ));
            }
        };
        if attempt.identity().attempt_id() != candidate_id {
            reservation = HostChargeReservation::from_attempt(&attempt).map_err(|_| {
                SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE)
            })?;
            if reservation.identity().gateway_account_id() != gateway.gateway_account_id()
                || reservation.request().gateway_order_id()
                    != &gateway.mutation_reference_factory().for_attempt(
                        PaymentAttemptKind::HostCharge,
                        reservation.identity().attempt_id(),
                    )
            {
                return Err(SubscriptionEnrollmentServiceError::InvalidState(
                    INVALID_SERVICE_STATE,
                ));
            }
        }

        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_host_charge_cooldown(
                    targets,
                    &reservation,
                    &account.provider_key,
                    scope,
                    HostChargeBeforeSubmissionResolution::prepared(),
                )
                .await;
        }
        match gateway.account_mode().await {
            Ok(GatewayAccountMode::Live) => {}
            Ok(GatewayAccountMode::Test) => {
                return self
                    .resolve_host_charge_readiness(
                        targets,
                        &reservation,
                        GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT),
                        PaymentResolutionCode::GatewayLiveReadinessFailedBeforeSubmission,
                        HostChargeBeforeSubmissionResolution::prepared(),
                    )
                    .await;
            }
            Err(GatewayError::RateLimited(detail)) => {
                return self
                    .resolve_host_charge_cooldown_with_detail(
                        targets,
                        &reservation,
                        GatewayMutationCooldownScope::Provider,
                        detail,
                        HostChargeBeforeSubmissionResolution::prepared_provider_rate_limited(),
                    )
                    .await;
            }
            Err(_) => {
                return self
                    .resolve_host_charge_readiness(
                        targets,
                        &reservation,
                        GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT),
                        PaymentResolutionCode::GatewayLiveReadinessFailedBeforeSubmission,
                        HostChargeBeforeSubmissionResolution::prepared(),
                    )
                    .await;
            }
        }

        let admission =
            match admit_host_charge_submission(&self.pool, targets, &reservation).await? {
                HostChargeAdmissionOutcome::Admitted(admission) => *admission,
                HostChargeAdmissionOutcome::AlreadyAdmitted(attempt) => {
                    return Ok(HostChargePaymentResult::new(attempt));
                }
                HostChargeAdmissionOutcome::Rejected { attempt, .. } => {
                    return Ok(HostChargePaymentResult::new(attempt));
                }
            };
        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_host_charge_cooldown(
                    targets,
                    &reservation,
                    &account.provider_key,
                    scope,
                    HostChargeBeforeSubmissionResolution::admitted_not_submitted(),
                )
                .await;
        }
        match submit_admitted_host_charge(
            &self.pool,
            self.coordinator.as_ref(),
            targets,
            admission,
            &command,
            &gateway,
        )
        .await?
        {
            HostChargeProviderResult::Payment(payment) => Ok(payment),
            HostChargeProviderResult::NotSubmitted { payment, error } => {
                if payment.attempt().state().resolution_code()
                    == Some(crate::enrollment_application::not_submitted_resolution_code(&error))
                {
                    Err(SubscriptionEnrollmentServiceError::GatewayNotSubmitted(
                        error,
                    ))
                } else {
                    Ok(payment)
                }
            }
        }
    }

    /// Runs one complete initial-subscription payment boundary.
    ///
    /// Matching replay and conflict are resolved before host admission. No
    /// database transaction or lock is held across host admission, gateway
    /// resolution, readiness I/O, or the one provider mutation.
    pub async fn enroll(
        &self,
        command: EnrollSubscription,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        match self.preflight(&command).await? {
            SubscriptionEnrollmentPreflightOutcome::Continue => {}
            SubscriptionEnrollmentPreflightOutcome::Replay(attempt) => {
                return self.payment_result(*attempt).await;
            }
            SubscriptionEnrollmentPreflightOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
        }

        self.admit_subscriber_mutation(
            command.billing_scope_id(),
            command.subscriber_id(),
            EndUserMutationOperation::SubscriptionInitial,
        )
        .await?;

        let (account, gateway) = self
            .resolve_active_gateway(
                command.billing_scope_id(),
                command.gateway_configuration_id(),
            )
            .await?;
        let mut reservation = SubscriptionEnrollmentReservation::from_command(&command, &gateway)
            .map_err(map_reservation_build_error)?;

        let attempt = match self.reserve(&reservation).await? {
            SubscriptionEnrollmentReservationOutcome::Reserved(attempt)
            | SubscriptionEnrollmentReservationOutcome::Replay(attempt)
                if attempt.status() == PaymentAttemptStatus::Pending
                    && attempt.state().timestamps().submitted_at().is_none() =>
            {
                attempt
            }
            SubscriptionEnrollmentReservationOutcome::Replay(attempt) => {
                return self.payment_result(attempt).await;
            }
            SubscriptionEnrollmentReservationOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
            SubscriptionEnrollmentReservationOutcome::Rejected(reason) => {
                return Err(SubscriptionEnrollmentServiceError::ReservationRejected(
                    reason,
                ));
            }
            SubscriptionEnrollmentReservationOutcome::Reserved(_) => {
                return Err(SubscriptionEnrollmentServiceError::InvalidState(
                    INVALID_SERVICE_STATE,
                ));
            }
        };
        if reservation.identity().attempt_id() != attempt.identity().attempt_id() {
            reservation = SubscriptionEnrollmentReservation::from_command_for_attempt(
                &command,
                &gateway,
                attempt.identity().attempt_id(),
            )
            .map_err(map_reservation_build_error)?;
        }

        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Initial(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::Prepared,
                )
                .await;
        }
        if let Some(failure) = subscriber_gateway_readiness_failure(&gateway).await {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Initial(&reservation),
                    failure,
                    OutcomeResolutionBoundary::Prepared,
                )
                .await;
        }

        let admission = match admit_subscription_enrollment_submission(
            &self.pool,
            self.offers.as_ref(),
            &reservation,
        )
        .await?
        {
            SubscriptionEnrollmentAdmissionOutcome::Admitted(admission) => *admission,
            SubscriptionEnrollmentAdmissionOutcome::AlreadyAdmitted(attempt) => {
                return self.payment_result(attempt).await;
            }
            SubscriptionEnrollmentAdmissionOutcome::Rejected { reason, .. } => {
                return Err(SubscriptionEnrollmentServiceError::SubmissionRejected(
                    reason,
                ));
            }
        };

        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Initial(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        match submit_admitted_subscription_enrollment(
            &self.pool,
            self.coordinator.as_ref(),
            admission,
            &command,
            &gateway,
        )
        .await?
        {
            SubscriptionEnrollmentProviderResult::Payment(payment) => Ok(payment),
            SubscriptionEnrollmentProviderResult::NotSubmitted { payment, error } => {
                preserve_concurrent_terminal_payment(payment, error)
            }
        }
    }

    /// Runs one complete automatic recurring-renewal boundary.
    ///
    /// Stale, future, canceled, paced, and contended work is a successful
    /// no-op. The operation never invokes end-user admission or the live offer
    /// store and never holds a database lock across provider I/O.
    pub async fn renew(
        &self,
        command: ChargeRenewal,
    ) -> Result<SubscriptionRenewalOutcome, SubscriptionEnrollmentServiceError> {
        let Some(account) = self.renewal_gateway_account(command).await? else {
            return Ok(SubscriptionRenewalOutcome::Noop);
        };
        if self
            .active_cooldown(&account.as_gateway_snapshot())
            .await?
            .is_some()
        {
            return Ok(SubscriptionRenewalOutcome::Noop);
        }
        let gateway = self
            .resolver
            .resolve(
                command.billing_scope_id(),
                account.account_id,
                account.configuration_id,
                account.provider_key.clone(),
            )
            .await?;
        if gateway.billing_scope_id() != command.billing_scope_id()
            || gateway.gateway_account_id() != account.account_id
            || gateway.gateway_configuration_id() != account.configuration_id
            || gateway.provider_key() != &account.provider_key
        {
            return Err(SubscriptionEnrollmentServiceError::ResolvedGatewayIdentityMismatch);
        }
        match gateway.account_mode().await {
            Ok(GatewayAccountMode::Live) => {}
            Ok(GatewayAccountMode::Test) => {
                return Err(SubscriptionEnrollmentServiceError::GatewayReadiness(
                    GatewayError::Configuration(GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT)),
                ));
            }
            Err(GatewayError::RateLimited(_)) => {
                self.extend_provider_cooldown(&account.provider_key).await?;
                return Ok(SubscriptionRenewalOutcome::Noop);
            }
            Err(error) => return Err(SubscriptionEnrollmentServiceError::GatewayReadiness(error)),
        }

        let (reservation, attempt) = match self.reserve_renewal(command, &gateway).await? {
            SubscriptionRenewalReservationOutcome::Reserved(reservation, attempt) => {
                (*reservation, *attempt)
            }
            SubscriptionRenewalReservationOutcome::Rejected(
                SubscriptionRenewalReservationRejection::PaymentMethodUpdateInProgress,
            ) => {
                return Err(
                    SubscriptionEnrollmentServiceError::RenewalReservationRejected(
                        SubscriptionRenewalReservationRejection::PaymentMethodUpdateInProgress,
                    ),
                );
            }
            SubscriptionRenewalReservationOutcome::Rejected(
                SubscriptionRenewalReservationRejection::GatewayConfigurationChanged,
            ) => {
                return Err(SubscriptionEnrollmentServiceError::GatewayConfigurationChanged);
            }
            SubscriptionRenewalReservationOutcome::Rejected(_) => {
                return Ok(SubscriptionRenewalOutcome::Noop);
            }
        };
        if attempt.status() != PaymentAttemptStatus::Pending
            || attempt.state().timestamps().submitted_at().is_some()
            || attempt.identity() != reservation.identity()
        {
            return Err(SubscriptionEnrollmentServiceError::InvalidState(
                INVALID_SERVICE_STATE,
            ));
        }
        if let Some(scope) = self.active_cooldown(&account.as_gateway_snapshot()).await? {
            self.resolve_renewal_cooldown(&reservation, scope, OutcomeResolutionBoundary::Prepared)
                .await?;
            return Ok(SubscriptionRenewalOutcome::Noop);
        }
        if !self
            .renewal_readiness_open(&reservation, &gateway, OutcomeResolutionBoundary::Prepared)
            .await?
        {
            return Ok(SubscriptionRenewalOutcome::Noop);
        }

        let admission = match admit_subscription_renewal_submission(&self.pool, &reservation).await
        {
            Err(error) if is_retryable_renewal_admission_error(&error) => {
                self.resolve_renewal_readiness_failure(
                    &reservation,
                    GatewayDiagnostic::new(
                        "subscription billing state could not be locked for final admission",
                    ),
                    PaymentResolutionCode::SubscriptionRenewalRetryStateChangedBeforeCharge,
                    None,
                    OutcomeResolutionBoundary::Prepared,
                )
                .await?;
                return Ok(SubscriptionRenewalOutcome::Noop);
            }
            Err(error) => return Err(error.into()),
            Ok(outcome) => match outcome {
                SubscriptionRenewalAdmissionOutcome::Admitted(admission) => *admission,
                SubscriptionRenewalAdmissionOutcome::AlreadyAdmitted(attempt) => {
                    return self
                        .payment_result(attempt)
                        .await
                        .map(Box::new)
                        .map(SubscriptionRenewalOutcome::Payment);
                }
                SubscriptionRenewalAdmissionOutcome::Rejected { attempt, .. } => {
                    return self
                        .payment_result(attempt)
                        .await
                        .map(Box::new)
                        .map(SubscriptionRenewalOutcome::Payment);
                }
            },
        };
        if let Some(scope) = self.active_cooldown(&account.as_gateway_snapshot()).await? {
            self.resolve_renewal_cooldown(
                &reservation,
                scope,
                OutcomeResolutionBoundary::AdmittedNotSubmitted,
            )
            .await?;
            return Ok(SubscriptionRenewalOutcome::Noop);
        }
        match submit_admitted_subscription_renewal(
            &self.pool,
            self.coordinator.as_ref(),
            admission,
            &gateway,
        )
        .await?
        {
            SubscriptionRenewalProviderResult::Payment(payment) => {
                Ok(SubscriptionRenewalOutcome::Payment(Box::new(payment)))
            }
            SubscriptionRenewalProviderResult::NotSubmitted { payment, error } => {
                Ok(SubscriptionRenewalOutcome::NotSubmitted {
                    payment: Box::new(payment),
                    error,
                })
            }
        }
    }

    /// Runs one complete subscriber-initiated recovery payment boundary.
    ///
    /// The command carries only the owner, requested plan/configuration, and
    /// memory-only token/contact. Reservation derives the exact due period,
    /// amount, subscription, and payment-state snapshot under lock.
    pub async fn recover(
        &self,
        command: RecoverSubscriptionPayment,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        match self.preflight_recovery(&command).await? {
            SubscriptionRecoveryPreflightOutcome::Continue => {}
            SubscriptionRecoveryPreflightOutcome::Replay(attempt) => {
                return self.payment_result(*attempt).await;
            }
            SubscriptionRecoveryPreflightOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
        }

        self.admit_subscriber_mutation(
            command.billing_scope_id(),
            command.subscriber_id(),
            EndUserMutationOperation::SubscriptionRecovery,
        )
        .await?;

        let (account, gateway) = self
            .resolve_active_gateway(
                command.billing_scope_id(),
                command.gateway_configuration_id(),
            )
            .await?;

        let (reservation, attempt) = match self.reserve_recovery(&command, &gateway).await? {
            SubscriptionRecoveryReservationOutcome::Reserved(reservation, attempt) => {
                (*reservation, *attempt)
            }
            SubscriptionRecoveryReservationOutcome::Replay(attempt) => {
                return self.payment_result(*attempt).await;
            }
            SubscriptionRecoveryReservationOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
            SubscriptionRecoveryReservationOutcome::Rejected(reason) => {
                return Err(
                    SubscriptionEnrollmentServiceError::RecoveryReservationRejected(reason),
                );
            }
        };
        if attempt.status() != PaymentAttemptStatus::Pending
            || attempt.state().timestamps().submitted_at().is_some()
            || attempt.identity() != reservation.identity()
        {
            return Err(SubscriptionEnrollmentServiceError::InvalidState(
                INVALID_SERVICE_STATE,
            ));
        }

        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Recovery(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::Prepared,
                )
                .await;
        }
        if let Some(failure) = subscriber_gateway_readiness_failure(&gateway).await {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Recovery(&reservation),
                    failure,
                    OutcomeResolutionBoundary::Prepared,
                )
                .await;
        }

        let admission =
            match admit_subscription_recovery_submission(&self.pool, &reservation).await? {
                SubscriptionRecoveryAdmissionOutcome::Admitted(admission) => *admission,
                SubscriptionRecoveryAdmissionOutcome::AlreadyAdmitted(attempt) => {
                    return self.payment_result(attempt).await;
                }
                SubscriptionRecoveryAdmissionOutcome::Rejected { attempt, .. } => {
                    return self.payment_result(attempt).await;
                }
            };
        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Recovery(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        if let Some(failure) = subscriber_gateway_readiness_failure(&gateway).await {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Recovery(&reservation),
                    failure,
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::Recovery(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        match submit_admitted_subscription_recovery(
            &self.pool,
            self.coordinator.as_ref(),
            admission,
            &command,
            &gateway,
        )
        .await?
        {
            SubscriptionRecoveryProviderResult::Payment(payment) => Ok(payment),
            SubscriptionRecoveryProviderResult::NotSubmitted { payment, error } => {
                preserve_concurrent_terminal_payment(payment, error)
            }
        }
    }

    /// Runs one complete stored payment-method replacement boundary.
    pub async fn replace_payment_method(
        &self,
        command: ReplaceSubscriptionPaymentMethod,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        match self.preflight_payment_method_replacement(&command).await? {
            SubscriptionPaymentMethodReplacementPreflightOutcome::Continue => {}
            SubscriptionPaymentMethodReplacementPreflightOutcome::Replay(attempt) => {
                return self.payment_result(*attempt).await;
            }
            SubscriptionPaymentMethodReplacementPreflightOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
        }
        self.admit_subscriber_mutation(
            command.billing_scope_id(),
            command.subscriber_id(),
            EndUserMutationOperation::SubscriptionPaymentMethodUpdate,
        )
        .await?;
        let (account, gateway) = self
            .resolve_active_gateway(
                command.billing_scope_id(),
                command.gateway_configuration_id(),
            )
            .await?;
        let (reservation, attempt) = match self
            .reserve_payment_method_replacement(&command, &gateway)
            .await?
        {
            SubscriptionPaymentMethodReplacementReservationOutcome::Reserved(
                reservation,
                attempt,
            ) => (*reservation, *attempt),
            SubscriptionPaymentMethodReplacementReservationOutcome::Replay(attempt) => {
                return self.payment_result(*attempt).await;
            }
            SubscriptionPaymentMethodReplacementReservationOutcome::IdempotencyConflict => {
                return Err(SubscriptionEnrollmentServiceError::IdempotencyConflict);
            }
            SubscriptionPaymentMethodReplacementReservationOutcome::Rejected(reason) => {
                return Err(
                    SubscriptionEnrollmentServiceError::PaymentMethodReplacementReservationRejected(
                        reason,
                    ),
                );
            }
        };
        if attempt.status() != PaymentAttemptStatus::Pending
            || attempt.state().timestamps().submitted_at().is_some()
            || attempt.identity() != reservation.identity()
        {
            return Err(SubscriptionEnrollmentServiceError::InvalidState(
                INVALID_SERVICE_STATE,
            ));
        }
        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::PaymentMethodReplacement(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::Prepared,
                )
                .await;
        }
        if let Some(failure) = subscriber_gateway_readiness_failure(&gateway).await {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::PaymentMethodReplacement(&reservation),
                    failure,
                    OutcomeResolutionBoundary::Prepared,
                )
                .await;
        }
        let admission =
            match admit_subscription_payment_method_replacement(&self.pool, &reservation).await? {
                SubscriptionPaymentMethodReplacementAdmissionOutcome::Admitted(admission) => {
                    *admission
                }
                SubscriptionPaymentMethodReplacementAdmissionOutcome::AlreadyAdmitted(attempt) => {
                    return self.payment_result(attempt).await;
                }
                SubscriptionPaymentMethodReplacementAdmissionOutcome::Rejected {
                    attempt, ..
                } => {
                    return self.payment_result(attempt).await;
                }
            };
        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::PaymentMethodReplacement(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        if let Some(failure) = subscriber_gateway_readiness_failure(&gateway).await {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::PaymentMethodReplacement(&reservation),
                    failure,
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        if let Some(scope) = self.active_cooldown(&account).await? {
            return self
                .resolve_subscriber_readiness_failure(
                    SubscriberInitiatedReservation::PaymentMethodReplacement(&reservation),
                    SubscriberReadinessFailure::Cooldown(scope),
                    OutcomeResolutionBoundary::AdmittedNotSubmitted,
                )
                .await;
        }
        match submit_admitted_subscription_payment_method_replacement(
            &self.pool,
            self.coordinator.as_ref(),
            admission,
            &command,
            &gateway,
        )
        .await?
        {
            SubscriptionPaymentMethodReplacementProviderResult::Payment(payment) => Ok(payment),
            SubscriptionPaymentMethodReplacementProviderResult::NotSubmitted { payment, error } => {
                preserve_concurrent_terminal_payment(payment, error)
            }
        }
    }

    /// Applies an already-observed provider outcome without another submission.
    ///
    /// Reconciliation enters the same application authority as foreground
    /// enrollment but reconstructs its secret-free reservation from the exact
    /// durable attempt and canonical gateway account.
    pub async fn apply_reconciled_outcome(
        &self,
        billing_scope_id: BillingScopeId,
        attempt_id: PaymentAttemptId,
        outcome: &GatewayPaymentOutcome,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let attempt = crate::find_payment_attempt_by_id_in_transaction(
            &mut transaction,
            billing_scope_id,
            attempt_id,
        )
        .await?
        .ok_or(SubscriptionEnrollmentServiceError::InvalidState(
            "reconciled subscription payment attempt was not found",
        ))?;
        transaction.commit().await?;
        match attempt.kind() {
            PaymentAttemptKind::SubscriptionInitial => {
                apply_reconciled_subscription_enrollment_gateway_outcome(
                    &self.pool,
                    self.coordinator.as_ref(),
                    billing_scope_id,
                    attempt_id,
                    outcome,
                )
                .await
            }
            PaymentAttemptKind::SubscriptionRecovery => {
                apply_reconciled_subscription_recovery_gateway_outcome(
                    &self.pool,
                    self.coordinator.as_ref(),
                    billing_scope_id,
                    attempt_id,
                    outcome,
                )
                .await
            }
            PaymentAttemptKind::SubscriptionRenewal => {
                apply_reconciled_subscription_renewal_gateway_outcome(
                    &self.pool,
                    self.coordinator.as_ref(),
                    billing_scope_id,
                    attempt_id,
                    outcome,
                )
                .await
            }
            PaymentAttemptKind::SubscriptionPaymentMethodUpdate => {
                apply_reconciled_subscription_payment_method_replacement_gateway_outcome(
                    &self.pool,
                    self.coordinator.as_ref(),
                    billing_scope_id,
                    attempt_id,
                    outcome,
                )
                .await
            }
            _ => Err(SubscriptionEnrollmentApplicationError::InvalidState(
                "attempt kind is not owned by the subscription billing service",
            )),
        }
        .map_err(Into::into)
    }

    /// Applies an exact-query outcome to one host charge without resubmission.
    pub async fn apply_reconciled_host_charge_outcome(
        &self,
        billing_scope_id: BillingScopeId,
        attempt_id: PaymentAttemptId,
        outcome: &GatewayPaymentOutcome,
    ) -> Result<HostChargePaymentResult, SubscriptionEnrollmentServiceError> {
        let targets = self
            .host_charge_targets
            .as_deref()
            .ok_or(SubscriptionEnrollmentServiceError::HostChargeUnavailable)?;
        apply_reconciled_host_charge_gateway_outcome(
            &self.pool,
            self.coordinator.as_ref(),
            targets,
            billing_scope_id,
            attempt_id,
            outcome,
        )
        .await
        .map_err(Into::into)
    }

    async fn preflight(
        &self,
        command: &EnrollSubscription,
    ) -> Result<SubscriptionEnrollmentPreflightOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome =
            preflight_subscription_enrollment_in_transaction(&mut transaction, command).await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn preflight_host_charge(
        &self,
        targets: &dyn HostChargeTargetStore,
        command: &ChargeHostTarget,
    ) -> Result<HostChargePreflightOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome =
            preflight_host_charge_in_transaction(&mut transaction, targets, command).await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn reserve_host_charge(
        &self,
        targets: &dyn HostChargeTargetStore,
        reservation: &HostChargeReservation,
    ) -> Result<HostChargeReservationOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome =
            reserve_host_charge_in_transaction(&mut transaction, targets, reservation).await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn resolve_host_charge_readiness(
        &self,
        targets: &dyn HostChargeTargetStore,
        reservation: &HostChargeReservation,
        detail: GatewayDiagnostic,
        code: PaymentResolutionCode,
        resolution: HostChargeBeforeSubmissionResolution,
    ) -> Result<HostChargePaymentResult, SubscriptionEnrollmentServiceError> {
        resolve_host_charge_before_submission(
            &self.pool,
            targets,
            reservation,
            detail,
            code,
            resolution,
        )
        .await
        .map_err(Into::into)
    }

    async fn resolve_host_charge_cooldown(
        &self,
        targets: &dyn HostChargeTargetStore,
        reservation: &HostChargeReservation,
        provider_key: &GatewayProviderKey,
        scope: GatewayMutationCooldownScope,
        resolution: HostChargeBeforeSubmissionResolution,
    ) -> Result<HostChargePaymentResult, SubscriptionEnrollmentServiceError> {
        let provider_name = provider_key.as_str().to_ascii_uppercase();
        let detail = match scope {
            GatewayMutationCooldownScope::Account => GatewayDiagnostic::new(&format!(
                "{provider_name} account mutation cooldown is active."
            )),
            GatewayMutationCooldownScope::Provider => GatewayDiagnostic::new(&format!(
                "{provider_name} system provider cooldown is active."
            )),
        };
        self.resolve_host_charge_cooldown_with_detail(
            targets,
            reservation,
            scope,
            detail,
            resolution,
        )
        .await
    }

    async fn resolve_host_charge_cooldown_with_detail(
        &self,
        targets: &dyn HostChargeTargetStore,
        reservation: &HostChargeReservation,
        scope: GatewayMutationCooldownScope,
        detail: GatewayDiagnostic,
        resolution: HostChargeBeforeSubmissionResolution,
    ) -> Result<HostChargePaymentResult, SubscriptionEnrollmentServiceError> {
        let code = match scope {
            GatewayMutationCooldownScope::Account => {
                PaymentResolutionCode::GatewayAccountMutationCooldownBeforeSubmission
            }
            GatewayMutationCooldownScope::Provider => {
                PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission
            }
        };
        let payment = self
            .resolve_host_charge_readiness(targets, reservation, detail, code, resolution)
            .await?;
        if payment.attempt().state().resolution_code() == Some(code) {
            Err(SubscriptionEnrollmentServiceError::GatewayMutationCooldown { scope })
        } else {
            Ok(payment)
        }
    }

    async fn preflight_recovery(
        &self,
        command: &RecoverSubscriptionPayment,
    ) -> Result<SubscriptionRecoveryPreflightOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome =
            preflight_subscription_recovery_in_transaction(&mut transaction, command).await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn preflight_payment_method_replacement(
        &self,
        command: &ReplaceSubscriptionPaymentMethod,
    ) -> Result<
        SubscriptionPaymentMethodReplacementPreflightOutcome,
        SubscriptionEnrollmentServiceError,
    > {
        let mut transaction = self.pool.begin().await?;
        let outcome = preflight_subscription_payment_method_replacement_in_transaction(
            &mut transaction,
            command,
        )
        .await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn reserve(
        &self,
        reservation: &SubscriptionEnrollmentReservation,
    ) -> Result<SubscriptionEnrollmentReservationOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome = reserve_subscription_enrollment_in_transaction(
            &mut transaction,
            self.offers.as_ref(),
            reservation,
        )
        .await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn reserve_recovery(
        &self,
        command: &RecoverSubscriptionPayment,
        gateway: &syrup_rail::ResolvedGateway,
    ) -> Result<SubscriptionRecoveryReservationOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome =
            reserve_subscription_recovery_in_transaction(&mut transaction, command, gateway)
                .await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn reserve_renewal(
        &self,
        command: ChargeRenewal,
        gateway: &syrup_rail::ResolvedGateway,
    ) -> Result<SubscriptionRenewalReservationOutcome, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let outcome =
            reserve_subscription_renewal_in_transaction(&mut transaction, command, gateway).await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn reserve_payment_method_replacement(
        &self,
        command: &ReplaceSubscriptionPaymentMethod,
        gateway: &syrup_rail::ResolvedGateway,
    ) -> Result<
        SubscriptionPaymentMethodReplacementReservationOutcome,
        SubscriptionEnrollmentServiceError,
    > {
        let mut transaction = self.pool.begin().await?;
        let outcome = reserve_subscription_payment_method_replacement_in_transaction(
            &mut transaction,
            command,
            gateway,
        )
        .await?;
        transaction.commit().await?;
        Ok(outcome)
    }

    async fn payment_result(
        &self,
        attempt: PaymentAttempt,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let result = payment_result_for_attempt(&mut transaction, attempt).await?;
        transaction.commit().await?;
        Ok(result)
    }

    async fn admit_subscriber_mutation(
        &self,
        billing_scope_id: BillingScopeId,
        subscriber_id: syrup_rail::SubscriberId,
        operation: EndUserMutationOperation,
    ) -> Result<(), SubscriptionEnrollmentServiceError> {
        let result = self
            .admission
            .admit(EndUserMutationCommand::new(
                billing_scope_id,
                subscriber_id,
                operation,
            ))
            .await;
        map_subscriber_mutation_admission(result)
    }

    /// Resolves the canonical account after the caller's operation-specific
    /// preflight and admission phases. The three subscriber-initiated paths
    /// share the same cooldown and exact resolver-identity contract.
    async fn resolve_active_gateway(
        &self,
        billing_scope_id: BillingScopeId,
        gateway_configuration_id: syrup_rail::GatewayConfigurationId,
    ) -> Result<
        (GatewayAccountSnapshot, syrup_rail::ResolvedGateway),
        SubscriptionEnrollmentServiceError,
    > {
        let account = self
            .gateway_account(billing_scope_id, gateway_configuration_id)
            .await?;
        if let Some(scope) = self.active_cooldown(&account).await? {
            return Err(SubscriptionEnrollmentServiceError::GatewayMutationCooldown { scope });
        }
        let expected = ExpectedGatewayIdentity::for_account(
            billing_scope_id,
            gateway_configuration_id,
            &account,
        );
        let gateway = self
            .resolver
            .resolve(
                expected.billing_scope_id,
                expected.gateway_account_id,
                expected.gateway_configuration_id,
                expected.provider_key.clone(),
            )
            .await?;
        if !expected.matches(&gateway) {
            return Err(SubscriptionEnrollmentServiceError::ResolvedGatewayIdentityMismatch);
        }
        Ok((account, gateway))
    }

    async fn gateway_account(
        &self,
        billing_scope_id: BillingScopeId,
        gateway_configuration_id: syrup_rail::GatewayConfigurationId,
    ) -> Result<GatewayAccountSnapshot, SubscriptionEnrollmentServiceError> {
        let row = sqlx::query_as::<_, (uuid::Uuid, String)>(
            r#"
            SELECT id, provider_key
            FROM billing_gateway_accounts
            WHERE billing_scope_id = $1 AND gateway_configuration_id = $2
            "#,
        )
        .bind(billing_scope_id.as_uuid())
        .bind(gateway_configuration_id.as_uuid())
        .fetch_optional(&self.pool)
        .await?
        .ok_or(SubscriptionEnrollmentServiceError::GatewayConfigurationChanged)?;
        let provider_key = GatewayProviderKey::new(&row.1)
            .map_err(|_| SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE))?;
        Ok(GatewayAccountSnapshot {
            account_id: GatewayAccountId::new(row.0),
            provider_key,
        })
    }

    async fn active_cooldown(
        &self,
        account: &GatewayAccountSnapshot,
    ) -> Result<Option<GatewayMutationCooldownScope>, SubscriptionEnrollmentServiceError> {
        let row = sqlx::query_as::<_, (bool, bool)>(
            r#"
            SELECT
                COALESCE(accounts.mutation_rate_limited_until > clock_timestamp(), false),
                provider.rate_limited_until > clock_timestamp()
            FROM billing_gateway_accounts AS accounts
            INNER JOIN billing_gateway_provider_rate_limits AS provider
                ON provider.provider_key = accounts.provider_key
            WHERE accounts.id = $1 AND accounts.provider_key = $2
            "#,
        )
        .bind(account.account_id.as_uuid())
        .bind(account.provider_key.as_str())
        .fetch_optional(&self.pool)
        .await?
        .ok_or(SubscriptionEnrollmentServiceError::GatewayConfigurationChanged)?;
        Ok(if row.1 {
            Some(GatewayMutationCooldownScope::Provider)
        } else if row.0 {
            Some(GatewayMutationCooldownScope::Account)
        } else {
            None
        })
    }

    async fn resolve_subscriber_readiness_failure(
        &self,
        reservation: SubscriberInitiatedReservation<'_>,
        failure: SubscriberReadinessFailure,
        boundary: OutcomeResolutionBoundary,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        let code = failure.resolution_code();
        let cooldown = failure.cooldown();
        let cooldown_error_scope = failure.cooldown_error_scope();
        let detail = failure.into_detail();
        let evidence = ProcessorEvidence::new(
            None,
            None,
            None,
            None,
            Some(detail),
            Some(GatewayDiagnostic::new("failed")),
            GatewayPaymentDescriptor::default(),
        );
        let payment = reservation
            .resolve_non_approved(&self.pool, &evidence, code, cooldown, boundary)
            .await
            .map_err(SubscriptionEnrollmentServiceError::from)?;
        if let Some(scope) = cooldown_error_scope
            && payment.attempt().state().resolution_code() == Some(code)
        {
            return Err(SubscriptionEnrollmentServiceError::GatewayMutationCooldown { scope });
        }
        Ok(payment)
    }

    async fn renewal_gateway_account(
        &self,
        command: ChargeRenewal,
    ) -> Result<Option<RenewalGatewayAccountSnapshot>, SubscriptionEnrollmentServiceError> {
        let mut transaction = self.pool.begin().await?;
        let row = sqlx::query_as::<_, (uuid::Uuid, uuid::Uuid, String)>(
            r#"
            SELECT accounts.id, accounts.gateway_configuration_id, accounts.provider_key
            FROM billing_subscriptions AS subscriptions
            JOIN billing_gateway_accounts AS accounts
                ON accounts.billing_scope_id = subscriptions.billing_scope_id
                AND accounts.id = subscriptions.gateway_account_id
            WHERE subscriptions.billing_scope_id = $1 AND subscriptions.id = $2
                AND subscriptions.status IN ('active', 'past_due')
                AND subscriptions.next_renewal_at = $3
                AND subscriptions.next_renewal_at <= clock_timestamp()
            "#,
        )
        .bind(command.billing_scope_id().as_uuid())
        .bind(command.subscription_id().as_uuid())
        .bind(command.period_start_at())
        .fetch_optional(&mut *transaction)
        .await?;
        let Some((account_id, configuration_id, provider_key)) = row else {
            transaction.commit().await?;
            return Ok(None);
        };
        let attempt_state = crate::renewal_attempt_state(
            &mut transaction,
            command.subscription_id(),
            *command.period_start_at(),
            None,
        )
        .await
        .map_err(|error| match error {
            crate::RenewalStoreError::Sql(error) => SubscriptionEnrollmentServiceError::Sql(error),
            crate::RenewalStoreError::MissingProviderCooldown => {
                SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE)
            }
        })?;
        let now = sqlx::query_scalar("SELECT clock_timestamp()")
            .fetch_one(&mut *transaction)
            .await?;
        let has_payment_method_update: bool = sqlx::query_scalar(
            r#"
            SELECT EXISTS (
                SELECT 1 FROM billing_payment_attempts
                WHERE subscription_id = $1
                    AND attempt_kind = 'subscription_payment_method_update'
                    AND status IN ('pending', 'unknown', 'review_required')
                    AND NOT (
                        status = 'pending' AND submitted_at IS NULL
                        AND created_at <= clock_timestamp() - interval '3 minutes'
                    )
            )
            "#,
        )
        .bind(command.subscription_id().as_uuid())
        .fetch_one(&mut *transaction)
        .await?;
        transaction.commit().await?;
        if attempt_state.blocks_automatic_retry(now) || has_payment_method_update {
            if has_payment_method_update {
                return Err(
                    SubscriptionEnrollmentServiceError::RenewalReservationRejected(
                        SubscriptionRenewalReservationRejection::PaymentMethodUpdateInProgress,
                    ),
                );
            }
            return Ok(None);
        }
        Some((account_id, configuration_id, provider_key))
            .map(|(account_id, configuration_id, provider_key)| {
                Ok(RenewalGatewayAccountSnapshot {
                    account_id: GatewayAccountId::new(account_id),
                    configuration_id: syrup_rail::GatewayConfigurationId::new(configuration_id),
                    provider_key: GatewayProviderKey::new(provider_key).map_err(|_| {
                        SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE)
                    })?,
                })
            })
            .transpose()
    }

    async fn extend_provider_cooldown(
        &self,
        provider_key: &GatewayProviderKey,
    ) -> Result<(), SubscriptionEnrollmentServiceError> {
        let result = sqlx::query(
            r#"
            UPDATE billing_gateway_provider_rate_limits
            SET rate_limited_until = GREATEST(
                    rate_limited_until,
                    clock_timestamp() + make_interval(secs => $2)
                )
            WHERE provider_key = $1
            "#,
        )
        .bind(provider_key.as_str())
        .bind(syrup_rail::RENEWAL_PROVIDER_RATE_LIMIT_RETRY_AFTER_SECONDS)
        .execute(&self.pool)
        .await?;
        if result.rows_affected() != 1 {
            return Err(SubscriptionEnrollmentServiceError::InvalidState(
                INVALID_SERVICE_STATE,
            ));
        }
        Ok(())
    }

    async fn resolve_renewal_cooldown(
        &self,
        reservation: &SubscriptionRenewalReservation,
        scope: GatewayMutationCooldownScope,
        boundary: OutcomeResolutionBoundary,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        let (message, code) = match scope {
            GatewayMutationCooldownScope::Account => (
                "gateway account mutation cooldown is active",
                PaymentResolutionCode::GatewayAccountMutationCooldownBeforeSubmission,
            ),
            GatewayMutationCooldownScope::Provider => (
                "gateway provider cooldown is active",
                PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission,
            ),
        };
        self.resolve_renewal_readiness_failure(
            reservation,
            GatewayDiagnostic::new(message),
            code,
            None,
            boundary,
        )
        .await
    }

    async fn renewal_readiness_open(
        &self,
        reservation: &SubscriptionRenewalReservation,
        gateway: &syrup_rail::ResolvedGateway,
        boundary: OutcomeResolutionBoundary,
    ) -> Result<bool, SubscriptionEnrollmentServiceError> {
        match gateway.account_mode().await {
            Ok(GatewayAccountMode::Live) => Ok(true),
            Ok(GatewayAccountMode::Test) => {
                self.resolve_renewal_readiness_failure(
                    reservation,
                    GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT),
                    PaymentResolutionCode::GatewayLiveReadinessFailedBeforeSubmission,
                    None,
                    boundary,
                )
                .await?;
                Ok(false)
            }
            Err(GatewayError::RateLimited(detail)) => {
                self.resolve_renewal_readiness_failure(
                    reservation,
                    detail,
                    PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission,
                    Some(RateLimitCooldown::Provider),
                    boundary,
                )
                .await?;
                Ok(false)
            }
            Err(_) => {
                self.resolve_renewal_readiness_failure(
                    reservation,
                    GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT),
                    PaymentResolutionCode::GatewayLiveReadinessFailedBeforeSubmission,
                    None,
                    boundary,
                )
                .await?;
                Ok(false)
            }
        }
    }

    async fn resolve_renewal_readiness_failure(
        &self,
        reservation: &SubscriptionRenewalReservation,
        detail: GatewayDiagnostic,
        code: PaymentResolutionCode,
        cooldown: Option<RateLimitCooldown>,
        boundary: OutcomeResolutionBoundary,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
        let condition = (code != PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission)
            .then(|| GatewayDiagnostic::new("failed"));
        let evidence = ProcessorEvidence::new(
            None,
            None,
            None,
            None,
            Some(detail),
            condition,
            GatewayPaymentDescriptor::default(),
        );
        resolve_renewal_non_approved_outcome(
            self.coordinator.as_ref(),
            reservation,
            &evidence,
            PaymentAttemptStatus::Failed,
            Some(code),
            cooldown,
            boundary,
        )
        .await
        .map_err(SubscriptionEnrollmentServiceError::from)
    }
}

/// The subscriber-initiated reservation families share readiness resolution,
/// while keeping their operation-specific durable application functions
/// explicit.
enum SubscriberInitiatedReservation<'a> {
    Initial(&'a SubscriptionEnrollmentReservation),
    Recovery(&'a SubscriptionRecoveryReservation),
    PaymentMethodReplacement(&'a SubscriptionPaymentMethodReplacement),
}

impl SubscriberInitiatedReservation<'_> {
    async fn resolve_non_approved(
        self,
        pool: &PgPool,
        evidence: &ProcessorEvidence,
        code: PaymentResolutionCode,
        cooldown: Option<RateLimitCooldown>,
        boundary: OutcomeResolutionBoundary,
    ) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentApplicationError> {
        match self {
            Self::Initial(reservation) => {
                resolve_non_approved_outcome(
                    pool,
                    reservation,
                    evidence,
                    PaymentAttemptStatus::Failed,
                    Some(code),
                    cooldown,
                    boundary,
                )
                .await
            }
            Self::Recovery(reservation) => {
                resolve_recovery_non_approved_outcome(
                    pool,
                    reservation,
                    evidence,
                    PaymentAttemptStatus::Failed,
                    Some(code),
                    cooldown,
                    boundary,
                )
                .await
            }
            Self::PaymentMethodReplacement(reservation) => {
                resolve_payment_method_replacement_non_approved_outcome(
                    pool,
                    reservation,
                    evidence,
                    PaymentAttemptStatus::Failed,
                    Some(code),
                    cooldown,
                    boundary,
                )
                .await
            }
        }
    }
}

/// A closed representation of the only pre-submission readiness failures
/// shared by subscriber-initiated mutations.
enum SubscriberReadinessFailure {
    Cooldown(GatewayMutationCooldownScope),
    ProviderRateLimited(GatewayDiagnostic),
    LiveModeUnavailable,
}

impl SubscriberReadinessFailure {
    fn into_detail(self) -> GatewayDiagnostic {
        match self {
            Self::Cooldown(GatewayMutationCooldownScope::Account) => {
                GatewayDiagnostic::new("gateway account mutation cooldown is active")
            }
            Self::Cooldown(GatewayMutationCooldownScope::Provider) => {
                GatewayDiagnostic::new("gateway provider cooldown is active")
            }
            Self::ProviderRateLimited(detail) => detail,
            Self::LiveModeUnavailable => GatewayDiagnostic::new(LIVE_READINESS_FAILED_TEXT),
        }
    }

    const fn resolution_code(&self) -> PaymentResolutionCode {
        match self {
            Self::Cooldown(GatewayMutationCooldownScope::Account) => {
                PaymentResolutionCode::GatewayAccountMutationCooldownBeforeSubmission
            }
            Self::Cooldown(GatewayMutationCooldownScope::Provider)
            | Self::ProviderRateLimited(_) => {
                PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission
            }
            Self::LiveModeUnavailable => {
                PaymentResolutionCode::GatewayLiveReadinessFailedBeforeSubmission
            }
        }
    }

    const fn cooldown(&self) -> Option<RateLimitCooldown> {
        match self {
            Self::ProviderRateLimited(_) => Some(RateLimitCooldown::Provider),
            Self::Cooldown(_) | Self::LiveModeUnavailable => None,
        }
    }

    const fn cooldown_error_scope(&self) -> Option<GatewayMutationCooldownScope> {
        match self {
            Self::Cooldown(scope) => Some(*scope),
            Self::ProviderRateLimited(_) => Some(GatewayMutationCooldownScope::Provider),
            Self::LiveModeUnavailable => None,
        }
    }
}

fn map_subscriber_mutation_admission(
    result: EndUserMutationAdmissionResult,
) -> Result<(), SubscriptionEnrollmentServiceError> {
    match result {
        EndUserMutationAdmissionResult::Allowed => Ok(()),
        EndUserMutationAdmissionResult::Denied { retry_after } => {
            Err(SubscriptionEnrollmentServiceError::AdmissionDenied {
                retry_after: retry_after.get(),
            })
        }
        EndUserMutationAdmissionResult::Timeout => {
            Err(SubscriptionEnrollmentServiceError::AdmissionTimeout)
        }
        EndUserMutationAdmissionResult::Unavailable => {
            Err(SubscriptionEnrollmentServiceError::AdmissionUnavailable)
        }
    }
}

async fn subscriber_gateway_readiness_failure(
    gateway: &syrup_rail::ResolvedGateway,
) -> Option<SubscriberReadinessFailure> {
    match gateway.account_mode().await {
        Ok(GatewayAccountMode::Live) => None,
        Ok(GatewayAccountMode::Test) => Some(SubscriberReadinessFailure::LiveModeUnavailable),
        Err(GatewayError::RateLimited(detail)) => {
            Some(SubscriberReadinessFailure::ProviderRateLimited(detail))
        }
        Err(_) => Some(SubscriberReadinessFailure::LiveModeUnavailable),
    }
}

fn preserve_concurrent_terminal_payment(
    payment: SubscriptionEnrollmentPaymentResult,
    error: GatewayNotSubmittedError,
) -> Result<SubscriptionEnrollmentPaymentResult, SubscriptionEnrollmentServiceError> {
    if payment.attempt().state().resolution_code()
        == Some(crate::enrollment_application::not_submitted_resolution_code(&error))
    {
        Err(SubscriptionEnrollmentServiceError::GatewayNotSubmitted(
            error,
        ))
    } else {
        Ok(payment)
    }
}

fn is_retryable_renewal_admission_error(error: &SubscriptionEnrollmentApplicationError) -> bool {
    let sqlstate = match error {
        SubscriptionEnrollmentApplicationError::Sql(sqlx::Error::Database(error)) => error.code(),
        SubscriptionEnrollmentApplicationError::Attempt(PaymentAttemptStoreError::Sql(
            sqlx::Error::Database(error),
        )) => error.code(),
        _ => None,
    };
    matches!(
        sqlstate.as_deref(),
        Some("40001" | "40P01" | "55P03" | "57014")
    )
}

struct GatewayAccountSnapshot {
    account_id: GatewayAccountId,
    provider_key: GatewayProviderKey,
}

/// Canonical identity the resolver must return for subscriber-initiated
/// gateway mutations.
struct ExpectedGatewayIdentity<'a> {
    billing_scope_id: BillingScopeId,
    gateway_account_id: GatewayAccountId,
    gateway_configuration_id: syrup_rail::GatewayConfigurationId,
    provider_key: &'a GatewayProviderKey,
}

impl<'a> ExpectedGatewayIdentity<'a> {
    const fn for_account(
        billing_scope_id: BillingScopeId,
        gateway_configuration_id: syrup_rail::GatewayConfigurationId,
        account: &'a GatewayAccountSnapshot,
    ) -> Self {
        Self {
            billing_scope_id,
            gateway_account_id: account.account_id,
            gateway_configuration_id,
            provider_key: &account.provider_key,
        }
    }

    fn matches(&self, gateway: &syrup_rail::ResolvedGateway) -> bool {
        self.matches_components(
            gateway.billing_scope_id(),
            gateway.gateway_account_id(),
            gateway.gateway_configuration_id(),
            gateway.provider_key(),
        )
    }

    fn matches_components(
        &self,
        billing_scope_id: BillingScopeId,
        gateway_account_id: GatewayAccountId,
        gateway_configuration_id: syrup_rail::GatewayConfigurationId,
        provider_key: &GatewayProviderKey,
    ) -> bool {
        billing_scope_id == self.billing_scope_id
            && gateway_account_id == self.gateway_account_id
            && gateway_configuration_id == self.gateway_configuration_id
            && provider_key == self.provider_key
    }
}

struct RenewalGatewayAccountSnapshot {
    account_id: GatewayAccountId,
    configuration_id: syrup_rail::GatewayConfigurationId,
    provider_key: GatewayProviderKey,
}

impl RenewalGatewayAccountSnapshot {
    fn as_gateway_snapshot(&self) -> GatewayAccountSnapshot {
        GatewayAccountSnapshot {
            account_id: self.account_id,
            provider_key: self.provider_key.clone(),
        }
    }
}

const fn map_reservation_build_error(
    error: SubscriptionEnrollmentReservationBuildError,
) -> SubscriptionEnrollmentServiceError {
    match error {
        SubscriptionEnrollmentReservationBuildError::GatewayIdentityMismatch => {
            SubscriptionEnrollmentServiceError::ResolvedGatewayIdentityMismatch
        }
        SubscriptionEnrollmentReservationBuildError::AttemptKindMismatch
        | SubscriptionEnrollmentReservationBuildError::InvalidCharge => {
            SubscriptionEnrollmentServiceError::InvalidState(INVALID_SERVICE_STATE)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn subscriber_admission_mapping_preserves_each_error_variant() {
        assert!(map_subscriber_mutation_admission(EndUserMutationAdmissionResult::Allowed).is_ok());
        let retry_after = std::time::Duration::from_secs(7);
        assert!(matches!(
            map_subscriber_mutation_admission(EndUserMutationAdmissionResult::Denied {
                retry_after: syrup_rail::EndUserMutationRetryAfter::new(retry_after)
                    .expect("positive retry-after"),
            }),
            Err(SubscriptionEnrollmentServiceError::AdmissionDenied {
                retry_after: actual
            }) if actual == retry_after
        ));
        assert!(matches!(
            map_subscriber_mutation_admission(EndUserMutationAdmissionResult::Timeout),
            Err(SubscriptionEnrollmentServiceError::AdmissionTimeout)
        ));
        assert!(matches!(
            map_subscriber_mutation_admission(EndUserMutationAdmissionResult::Unavailable),
            Err(SubscriptionEnrollmentServiceError::AdmissionUnavailable)
        ));
    }

    #[test]
    fn subscriber_readiness_failure_preserves_codes_cooldowns_and_diagnostics() {
        for (scope, code, detail) in [
            (
                GatewayMutationCooldownScope::Account,
                PaymentResolutionCode::GatewayAccountMutationCooldownBeforeSubmission,
                "gateway account mutation cooldown is active",
            ),
            (
                GatewayMutationCooldownScope::Provider,
                PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission,
                "gateway provider cooldown is active",
            ),
        ] {
            let failure = SubscriberReadinessFailure::Cooldown(scope);
            assert_eq!(failure.resolution_code(), code);
            assert!(failure.cooldown().is_none());
            assert_eq!(failure.cooldown_error_scope(), Some(scope));
            assert_eq!(failure.into_detail().expose(), detail);
        }

        let provider = SubscriberReadinessFailure::ProviderRateLimited(GatewayDiagnostic::new(
            "provider asked to retry later",
        ));
        assert_eq!(
            provider.resolution_code(),
            PaymentResolutionCode::GatewayProviderRateLimitedBeforeSubmission
        );
        assert!(matches!(
            provider.cooldown(),
            Some(RateLimitCooldown::Provider)
        ));
        assert_eq!(
            provider.cooldown_error_scope(),
            Some(GatewayMutationCooldownScope::Provider)
        );
        assert_eq!(
            provider.into_detail().expose(),
            "provider asked to retry later"
        );

        let readiness = SubscriberReadinessFailure::LiveModeUnavailable;
        assert_eq!(
            readiness.resolution_code(),
            PaymentResolutionCode::GatewayLiveReadinessFailedBeforeSubmission
        );
        assert!(readiness.cooldown().is_none());
        assert!(readiness.cooldown_error_scope().is_none());
        assert_eq!(readiness.into_detail().expose(), LIVE_READINESS_FAILED_TEXT);
    }

    #[test]
    fn expected_gateway_identity_requires_every_resolved_component() {
        let provider_key = GatewayProviderKey::new("nmi").expect("valid provider key");
        let account = GatewayAccountSnapshot {
            account_id: GatewayAccountId::new(uuid::Uuid::from_u128(1)),
            provider_key: provider_key.clone(),
        };
        let billing_scope_id = BillingScopeId::new(uuid::Uuid::from_u128(2));
        let gateway_configuration_id =
            syrup_rail::GatewayConfigurationId::new(uuid::Uuid::from_u128(3));
        let expected = ExpectedGatewayIdentity::for_account(
            billing_scope_id,
            gateway_configuration_id,
            &account,
        );

        assert!(expected.matches_components(
            billing_scope_id,
            account.account_id,
            gateway_configuration_id,
            &provider_key,
        ));
        assert!(!expected.matches_components(
            BillingScopeId::new(uuid::Uuid::from_u128(4)),
            account.account_id,
            gateway_configuration_id,
            &provider_key,
        ));
        assert!(!expected.matches_components(
            billing_scope_id,
            GatewayAccountId::new(uuid::Uuid::from_u128(5)),
            gateway_configuration_id,
            &provider_key,
        ));
        assert!(!expected.matches_components(
            billing_scope_id,
            account.account_id,
            syrup_rail::GatewayConfigurationId::new(uuid::Uuid::from_u128(6)),
            &provider_key,
        ));
        assert!(!expected.matches_components(
            billing_scope_id,
            account.account_id,
            gateway_configuration_id,
            &GatewayProviderKey::new("other_gateway").expect("valid provider key"),
        ));
    }
}