syrup_rail_postgres/operator_review/
manual_failure.rs

1use super::*;
2
3/// Value-redacted failure returned by the host's manual-failure store.
4#[derive(Debug)]
5pub struct ManualAttemptFailureHostStoreError {
6    source: RedactedHostErrorSource,
7}
8
9impl ManualAttemptFailureHostStoreError {
10    /// Wraps a host error without exposing its value through ordinary error
11    /// formatting or the standard error-source chain.
12    pub fn new(source: impl Error + Send + Sync + 'static) -> Self {
13        Self {
14            source: RedactedHostErrorSource::new(source),
15        }
16    }
17
18    /// Returns the host error for explicit application-level inspection.
19    pub fn into_source(self) -> BoxError {
20        self.source.into_inner()
21    }
22}
23
24impl fmt::Display for ManualAttemptFailureHostStoreError {
25    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
26        formatter.write_str("manual attempt failure host target transition failed")
27    }
28}
29
30impl Error for ManualAttemptFailureHostStoreError {}
31
32#[derive(Clone, Copy, Debug, Eq, PartialEq)]
33pub enum ManualAttemptFailureHostTransitionOutcome {
34    Changed,
35    Unchanged,
36}
37
38#[async_trait]
39pub trait ManualAttemptFailureHostStore: Send + Sync {
40    async fn lock_payment_failure_target(
41        &self,
42        connection: &mut PgConnection,
43        charge: ManualFailureHostCharge,
44    ) -> Result<(), ManualAttemptFailureHostStoreError>;
45
46    async fn mark_payment_failed(
47        &self,
48        connection: &mut PgConnection,
49        charge: ManualFailureHostCharge,
50    ) -> Result<ManualAttemptFailureHostTransitionOutcome, ManualAttemptFailureHostStoreError>;
51}
52
53pub async fn fail_review_required_attempt(
54    pool: &PgPool,
55    coordinator: &dyn BillingTransactionCoordinator,
56    host: &dyn ManualAttemptFailureHostStore,
57    attempt_id: PaymentAttemptId,
58) -> Result<ManualAttemptFailureOutcome, OperatorReviewError> {
59    let Some(preloaded) = payment_attempt_by_id(pool, attempt_id).await? else {
60        return Ok(ManualAttemptFailureOutcome::NotFound);
61    };
62    if !review_required_attempt_can_be_manually_failed(&preloaded) {
63        return Ok(ManualAttemptFailureOutcome::KeptOpen(preloaded));
64    }
65
66    let identity = preloaded.identity();
67    let subject = BillingEventSubject::new(identity.billing_scope_id(), identity.subscriber_id());
68    let mut transaction = coordinator
69        .begin(subject, std::time::Duration::from_millis(250))
70        .await?;
71    let result = async {
72        let connection = transaction.connection();
73        set_enrollment_timeouts(connection).await?;
74        let target = preloaded.request().target();
75        if let Some(plan_key) = target.plan_key() {
76            lock_subscription_aggregate(connection, identity.subscriber_id(), plan_key).await?;
77        } else if let Some(target_id) = target.host_charge_target_id() {
78            host.lock_payment_failure_target(
79                connection,
80                ManualFailureHostCharge::new(
81                    identity.billing_scope_id(),
82                    identity.subscriber_id(),
83                    target_id,
84                ),
85            )
86            .await?;
87        }
88
89        let Some(current) = lock_payment_attempt_by_id_on_connection(
90            connection,
91            identity.billing_scope_id(),
92            attempt_id,
93        )
94        .await?
95        else {
96            return Ok((ManualAttemptFailureOutcome::NotFound, Vec::new()));
97        };
98        if current.identity() != preloaded.identity() || current.request() != preloaded.request() {
99            return Err(OperatorReviewError::InvalidState(INVALID_OPERATOR_STATE));
100        }
101        if !review_required_attempt_can_be_manually_failed(&current)
102            || (current.kind() != PaymentAttemptKind::SubscriptionPaymentMethodUpdate
103                && unresolved_processor_charge_exists(connection, attempt_id).await?)
104        {
105            return Ok((ManualAttemptFailureOutcome::KeptOpen(current), Vec::new()));
106        }
107
108        let evidence = review_required_manual_failure_evidence(&current);
109        let updated = update_attempt_for_manual_failure(connection, &current, &evidence).await?;
110        if updated != 1 {
111            return Err(OperatorReviewError::InvalidState(INVALID_OPERATOR_STATE));
112        }
113
114        if let Some(target_id) = current.request().target().host_charge_target_id() {
115            host.mark_payment_failed(
116                connection,
117                ManualFailureHostCharge::new(
118                    identity.billing_scope_id(),
119                    identity.subscriber_id(),
120                    target_id,
121                ),
122            )
123            .await?;
124        }
125        let attempt = lock_payment_attempt_by_id_on_connection(
126            connection,
127            identity.billing_scope_id(),
128            attempt_id,
129        )
130        .await?
131        .ok_or(OperatorReviewError::InvalidState(INVALID_OPERATOR_STATE))?;
132        let events = if attempt.kind() == PaymentAttemptKind::SubscriptionRenewal
133            && attempt.state().timestamps().submitted_at().is_some()
134            && attempt.state().resolution_code().is_none()
135        {
136            match apply_resolved_automatic_renewal_failure(connection, &attempt).await? {
137                RenewalFailureApplication::Applied { events, .. } => events,
138                RenewalFailureApplication::Noop => Vec::new(),
139            }
140        } else {
141            Vec::new()
142        };
143        Ok((ManualAttemptFailureOutcome::Failed(attempt), events))
144    }
145    .await;
146
147    let (outcome, events) = match result {
148        Ok(value) => value,
149        Err(error) => {
150            let _ = transaction.rollback().await;
151            return Err(error);
152        }
153    };
154    for event in &events {
155        if let Err(error) = transaction.append_event(event).await {
156            let _ = transaction.rollback().await;
157            return Err(error.into());
158        }
159    }
160    transaction.commit().await?;
161    Ok(outcome)
162}
163
164async fn payment_attempt_by_id(
165    pool: &PgPool,
166    attempt_id: PaymentAttemptId,
167) -> Result<Option<PaymentAttempt>, OperatorReviewError> {
168    let query = format!("{} WHERE id = $1", crate::attempts::PAYMENT_ATTEMPT_SELECT);
169    let row = sqlx::query(&query)
170        .bind(attempt_id.as_uuid())
171        .fetch_optional(pool)
172        .await?;
173    row.as_ref()
174        .map(payment_attempt_from_row)
175        .transpose()
176        .map_err(Into::into)
177}
178
179async fn unresolved_processor_charge_exists(
180    connection: &mut PgConnection,
181    attempt_id: PaymentAttemptId,
182) -> Result<bool, sqlx::Error> {
183    sqlx::query_scalar(
184        r#"
185        SELECT EXISTS (
186            SELECT 1
187            FROM billing_processor_charges
188            WHERE attempt_id = $1
189                AND progression_state IN (
190                    'pending', 'reconciliation_required', 'external_reversal_required'
191                )
192        )
193        "#,
194    )
195    .bind(attempt_id.as_uuid())
196    .fetch_one(&mut *connection)
197    .await
198}
199
200async fn update_attempt_for_manual_failure(
201    connection: &mut PgConnection,
202    attempt: &PaymentAttempt,
203    evidence: &ProcessorEvidence,
204) -> Result<u64, sqlx::Error> {
205    let descriptor = evidence.descriptor();
206    let result = sqlx::query(
207        r#"
208        UPDATE billing_payment_attempts
209        SET status = 'failed',
210            gateway_transaction_id = $2,
211            gateway_payment_method_reference = $3,
212            gateway_response = $4,
213            gateway_response_code = $5,
214            gateway_response_text = $6,
215            gateway_condition = $7,
216            payment_type = $8,
217            card_brand = $9,
218            card_last4 = $10,
219            card_exp_month = $11,
220            card_exp_year = $12,
221            resolved_at = clock_timestamp(),
222            updated_at = clock_timestamp()
223        WHERE id = $1 AND status = 'review_required'
224        "#,
225    )
226    .bind(attempt.identity().attempt_id().as_uuid())
227    .bind(evidence.transaction_id().map(GatewayTransactionId::expose))
228    .bind(
229        evidence
230            .payment_method_reference()
231            .map(GatewayPaymentMethodReference::expose),
232    )
233    .bind(evidence.response().map(GatewayDiagnostic::expose))
234    .bind(evidence.response_code().map(GatewayDiagnostic::expose))
235    .bind(evidence.response_text().map(GatewayDiagnostic::expose))
236    .bind(evidence.condition().map(GatewayDiagnostic::expose))
237    .bind(descriptor.payment_type().map(GatewayDiagnostic::expose))
238    .bind(descriptor.card_brand().map(GatewayDiagnostic::expose))
239    .bind(
240        descriptor
241            .card_last_four()
242            .map(syrup_rail::CardLastFour::expose),
243    )
244    .bind(descriptor.card_exp_month())
245    .bind(descriptor.card_exp_year())
246    .execute(&mut *connection)
247    .await?;
248    Ok(result.rows_affected())
249}
syrup_rail_postgres/operator_review/manual_failure.rs

syrup_rail_postgres/operator_review/
manual_failure.rs
