syrup_rail_postgres/
discounts.rs

1use std::fmt;
2
3use async_trait::async_trait;
4use sqlx::{PgConnection, PgPool, Postgres, Row, Transaction, postgres::PgRow};
5use syrup_rail::{
6    BillingScopeId, ChargeAmount, CurrencyCode, DiscountClaimId, DiscountCodeId, IdempotencyKey,
7    LimitedDiscountMonths, PaymentAttemptId, PercentOffBasisPoints, PlanKey, PositiveDiscountCents,
8    SubscriberId, SubscriptionDiscountClaim, SubscriptionDiscountClaimOutcome,
9    SubscriptionDiscountClaimRecord, SubscriptionDiscountClaimStatus,
10    SubscriptionDiscountClearOutcome, SubscriptionDiscountCode, SubscriptionDiscountCodeCreation,
11    SubscriptionDiscountCodeQuote, SubscriptionDiscountCodeRecord, SubscriptionDiscountCodeStatus,
12    SubscriptionDiscountCodeUpdate, SubscriptionDiscountDuration, SubscriptionDiscountError,
13    SubscriptionDiscountKind, SubscriptionDiscountSnapshot, SubscriptionEnrollmentReservation,
14    SubscriptionId, SubscriptionOffer,
15};
16use thiserror::Error;
17use uuid::Uuid;
18
19pub use persistence::saved_subscription_discount_claim_in_transaction;
20use persistence::{
21    blocking_initial_attempt, blocking_initial_attempt_exists, claim_from_row, code_by_id,
22    code_from_row, current_subscription_exists, discount_value, duration_months,
23    expire_saved_claims_for_code, find_active_code, lock_initial_attempt_rows,
24    lock_initial_attempts, lock_offer, lock_subscription_aggregate, quote_for_offer,
25    quote_from_row, saved_subscription_discount_claim_on_connection, set_lock_timeout,
26    validate_discount_cadence,
27};
28
29mod persistence;
30
31const BILLING_ROW_LOCK_TIMEOUT: &str = "250ms";
32const BILLING_OPERATION_TIMEOUT: &str = "5s";
33const INVALID_DISCOUNT_STATE: &str = "canonical subscription discount state is invalid";
34
35/// The lifecycle boundary at which enrollment terms are locked.
36///
37/// A single enrollment reservation crosses both stages. The stage describes
38/// why the host is being consulted; it does not create a new eligibility
39/// subject or attempt identity.
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum SubscriptionEnrollmentOfferStage {
42    /// The durable initial payment attempt is about to be reserved.
43    Reservation,
44    /// The prepared attempt is being revalidated immediately before provider I/O.
45    SubmissionAdmission,
46}
47
48/// Stable identity and lifecycle context for locking an enrollment offer.
49///
50/// The PostgreSQL enrollment pipeline constructs this value from one
51/// [`SubscriptionEnrollmentReservation`] and passes the same scope,
52/// subscriber, plan, attempt, and idempotency identity at both stages. A host
53/// implementation that derives eligibility from payment-attempt history must
54/// exclude [`Self::attempt_id`] so the in-flight reservation cannot disqualify
55/// itself during submission admission.
56#[derive(Clone, Copy)]
57pub struct SubscriptionEnrollmentOfferContext<'a> {
58    billing_scope_id: BillingScopeId,
59    subscriber_id: SubscriberId,
60    plan_key: &'a PlanKey,
61    attempt_id: PaymentAttemptId,
62    idempotency_key: &'a IdempotencyKey,
63    stage: SubscriptionEnrollmentOfferStage,
64}
65
66impl<'a> SubscriptionEnrollmentOfferContext<'a> {
67    pub(crate) const fn from_reservation(
68        reservation: &'a SubscriptionEnrollmentReservation,
69        stage: SubscriptionEnrollmentOfferStage,
70    ) -> Self {
71        let identity = reservation.identity();
72        Self {
73            billing_scope_id: identity.billing_scope_id(),
74            subscriber_id: identity.subscriber_id(),
75            plan_key: reservation.plan_key(),
76            attempt_id: identity.attempt_id(),
77            idempotency_key: reservation.idempotency_key(),
78            stage,
79        }
80    }
81
82    pub const fn billing_scope_id(self) -> BillingScopeId {
83        self.billing_scope_id
84    }
85
86    pub const fn subscriber_id(self) -> SubscriberId {
87        self.subscriber_id
88    }
89
90    pub const fn plan_key(self) -> &'a PlanKey {
91        self.plan_key
92    }
93
94    pub const fn attempt_id(self) -> PaymentAttemptId {
95        self.attempt_id
96    }
97
98    pub const fn idempotency_key(self) -> &'a IdempotencyKey {
99        self.idempotency_key
100    }
101
102    pub const fn stage(self) -> SubscriptionEnrollmentOfferStage {
103        self.stage
104    }
105}
106
107impl fmt::Debug for SubscriptionEnrollmentOfferContext<'_> {
108    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
109        formatter
110            .debug_struct("SubscriptionEnrollmentOfferContext")
111            .field("billing_scope_id", &self.billing_scope_id)
112            .field("subscriber_id", &self.subscriber_id)
113            .field("plan_key", &self.plan_key)
114            .field("attempt_id", &self.attempt_id)
115            .field("has_idempotency_key", &true)
116            .field("stage", &self.stage)
117            .finish()
118    }
119}
120
121#[derive(Error)]
122pub enum SubscriptionDiscountOperationError {
123    #[error("subscription discount storage operation failed")]
124    Sql(#[from] sqlx::Error),
125    #[error("current subscription offer is unavailable")]
126    OfferUnavailable,
127    #[error("current subscription offer does not match the requested plan")]
128    OfferPlanMismatch,
129    #[error("subscription discount configuration is invalid for the current offer")]
130    InvalidConfiguration,
131    #[error("limited-month discounts require a one-calendar-month recurring period")]
132    LimitedDiscountCadence,
133    #[error("{0}")]
134    InvalidState(&'static str),
135}
136
137impl SubscriptionDiscountOperationError {
138    pub fn is_unique_violation(&self) -> bool {
139        matches!(self, Self::Sql(sqlx::Error::Database(error)) if error.is_unique_violation())
140    }
141
142    pub fn constraint(&self) -> Option<&str> {
143        match self {
144            Self::Sql(sqlx::Error::Database(error)) => error.constraint(),
145            _ => None,
146        }
147    }
148}
149
150impl fmt::Debug for SubscriptionDiscountOperationError {
151    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
152        match self {
153            Self::Sql(_) => formatter.write_str("SubscriptionDiscountOperationError::Sql"),
154            Self::OfferUnavailable => {
155                formatter.write_str("SubscriptionDiscountOperationError::OfferUnavailable")
156            }
157            Self::OfferPlanMismatch => {
158                formatter.write_str("SubscriptionDiscountOperationError::OfferPlanMismatch")
159            }
160            Self::InvalidConfiguration => {
161                formatter.write_str("SubscriptionDiscountOperationError::InvalidConfiguration")
162            }
163            Self::LimitedDiscountCadence => {
164                formatter.write_str("SubscriptionDiscountOperationError::LimitedDiscountCadence")
165            }
166            Self::InvalidState(detail) => formatter
167                .debug_tuple("SubscriptionDiscountOperationError::InvalidState")
168                .field(detail)
169                .finish(),
170        }
171    }
172}
173
174#[async_trait]
175pub trait SubscriptionOfferStore: Send + Sync {
176    /// Locks the host-owned row for this exact scope and plan on `connection`.
177    ///
178    /// Implementations must not acquire another connection. The returned row
179    /// must remain protected against host price updates until the caller's
180    /// transaction ends.
181    async fn lock_current_offer(
182        &self,
183        connection: &mut PgConnection,
184        billing_scope_id: BillingScopeId,
185        plan_key: &PlanKey,
186    ) -> Result<Option<SubscriptionOffer>, sqlx::Error>;
187
188    /// Locks the subscriber-aware offer used for enrollment admission.
189    ///
190    /// Overrides must lock any host trial-eligibility state on the supplied
191    /// connection and must not acquire a second connection. Enrollment calls
192    /// this hook at both [`SubscriptionEnrollmentOfferStage`] values for the
193    /// same reservation. Attempt-history eligibility queries must exclude
194    /// [`SubscriptionEnrollmentOfferContext::attempt_id`]; the result may
195    /// change only because locked host policy or eligibility state external to
196    /// that in-flight attempt changed. The subscriber/plan aggregate lock
197    /// serializes reservation and admission before this callback. Implementations
198    /// must not acquire Syrup Rail payment-attempt ledger locks independently;
199    /// doing so would invert admission's aggregate -> attempt -> offer order.
200    async fn lock_enrollment_offer(
201        &self,
202        connection: &mut PgConnection,
203        context: SubscriptionEnrollmentOfferContext<'_>,
204    ) -> Result<Option<SubscriptionOffer>, sqlx::Error> {
205        self.lock_current_offer(connection, context.billing_scope_id(), context.plan_key())
206            .await
207    }
208}
209
210pub async fn list_subscription_discount_codes(
211    pool: &PgPool,
212    billing_scope_id: BillingScopeId,
213    plan_key: &PlanKey,
214) -> Result<Vec<SubscriptionDiscountCodeRecord>, SubscriptionDiscountOperationError> {
215    let mut transaction = pool.begin().await?;
216    let records = list_subscription_discount_codes_in_transaction(
217        &mut transaction,
218        billing_scope_id,
219        plan_key,
220    )
221    .await?;
222    transaction.commit().await?;
223    Ok(records)
224}
225
226/// Lists durable administrative records without reinterpreting them against
227/// the current offer. Quote eligibility is intentionally owned by
228/// [`validate_subscription_discount_code_in_transaction`].
229pub async fn list_subscription_discount_codes_in_transaction(
230    transaction: &mut Transaction<'_, Postgres>,
231    billing_scope_id: BillingScopeId,
232    plan_key: &PlanKey,
233) -> Result<Vec<SubscriptionDiscountCodeRecord>, SubscriptionDiscountOperationError> {
234    set_lock_timeout(transaction).await?;
235    let rows = sqlx::query(
236        r#"
237        SELECT id, billing_scope_id, plan_key, code_normalized, display_code,
238            label, status, discount_kind, amount_off_cents, percent_off_bps,
239            currency, duration, duration_months, created_at, updated_at
240        FROM billing_subscription_discount_codes
241        WHERE billing_scope_id = $1 AND plan_key = $2
242        ORDER BY status, code_normalized, created_at DESC
243        "#,
244    )
245    .bind(billing_scope_id.as_uuid())
246    .bind(plan_key.as_str())
247    .fetch_all(&mut **transaction)
248    .await?;
249    rows.iter().map(code_from_row).collect()
250}
251
252pub async fn validate_subscription_discount_code(
253    pool: &PgPool,
254    offers: &dyn SubscriptionOfferStore,
255    billing_scope_id: BillingScopeId,
256    plan_key: &PlanKey,
257    code: &SubscriptionDiscountCode,
258) -> Result<Option<SubscriptionDiscountCodeQuote>, SubscriptionDiscountOperationError> {
259    let mut transaction = pool.begin().await?;
260    let quote = validate_subscription_discount_code_in_transaction(
261        &mut transaction,
262        offers,
263        billing_scope_id,
264        plan_key,
265        code,
266    )
267    .await?;
268    transaction.commit().await?;
269    Ok(quote)
270}
271
272pub async fn validate_subscription_discount_code_in_transaction(
273    transaction: &mut Transaction<'_, Postgres>,
274    offers: &dyn SubscriptionOfferStore,
275    billing_scope_id: BillingScopeId,
276    plan_key: &PlanKey,
277    code: &SubscriptionDiscountCode,
278) -> Result<Option<SubscriptionDiscountCodeQuote>, SubscriptionDiscountOperationError> {
279    set_lock_timeout(transaction).await?;
280    let offer = lock_offer(transaction, offers, billing_scope_id, plan_key).await?;
281    let row = find_active_code(transaction, billing_scope_id, plan_key, code, false).await?;
282    row.as_ref()
283        .map(|row| quote_from_row(row, &offer))
284        .transpose()
285}
286
287/// Creates an active durable code after validating its terms against the
288/// locked current offer. The returned value is the administrative record; use
289/// [`validate_subscription_discount_code_in_transaction`] to obtain a quote.
290pub async fn create_subscription_discount_code_in_transaction(
291    transaction: &mut Transaction<'_, Postgres>,
292    offers: &dyn SubscriptionOfferStore,
293    creation: &SubscriptionDiscountCodeCreation,
294) -> Result<SubscriptionDiscountCodeRecord, SubscriptionDiscountOperationError> {
295    set_lock_timeout(transaction).await?;
296    let offer = lock_offer(
297        transaction,
298        offers,
299        creation.billing_scope_id(),
300        creation.plan_key(),
301    )
302    .await?;
303    validate_discount_cadence(creation.duration(), &offer)?;
304    syrup_rail::discounted_charge(
305        offer.recurring().charge(),
306        creation.currency(),
307        creation.kind(),
308    )
309    .map_err(|_| SubscriptionDiscountOperationError::InvalidConfiguration)?;
310    let (amount_off_cents, percent_off_bps) = discount_value(creation.kind());
311    let duration_months = duration_months(creation.duration());
312    sqlx::query(
313        r#"
314        INSERT INTO billing_subscription_discount_codes (
315            id, billing_scope_id, plan_key, code_normalized, display_code,
316            label, status, discount_kind, amount_off_cents, percent_off_bps,
317            currency, duration, duration_months
318        )
319        VALUES ($1, $2, $3, $4, $4, $5, 'active', $6, $7, $8, $9, $10, $11)
320        "#,
321    )
322    .bind(creation.id().as_uuid())
323    .bind(creation.billing_scope_id().as_uuid())
324    .bind(creation.plan_key().as_str())
325    .bind(creation.code().as_str())
326    .bind(creation.label())
327    .bind(creation.kind().as_str())
328    .bind(amount_off_cents)
329    .bind(percent_off_bps)
330    .bind(creation.currency().as_str())
331    .bind(creation.duration().as_str())
332    .bind(duration_months)
333    .execute(&mut **transaction)
334    .await?;
335    let row = code_by_id(
336        transaction,
337        creation.billing_scope_id(),
338        creation.plan_key(),
339        creation.id(),
340    )
341    .await?
342    .ok_or(SubscriptionDiscountOperationError::InvalidState(
343        INVALID_DISCOUNT_STATE,
344    ))?;
345    code_from_row(&row)
346}
347
348pub async fn create_subscription_discount_code(
349    pool: &PgPool,
350    offers: &dyn SubscriptionOfferStore,
351    creation: &SubscriptionDiscountCodeCreation,
352) -> Result<SubscriptionDiscountCodeRecord, SubscriptionDiscountOperationError> {
353    let mut transaction = pool.begin().await?;
354    let record =
355        create_subscription_discount_code_in_transaction(&mut transaction, offers, creation)
356            .await?;
357    transaction.commit().await?;
358    Ok(record)
359}
360
361/// Updates a durable administrative record. Active records are validated
362/// against the locked current offer. Disabled records remain administrable
363/// without requiring their historical terms to be quoteable today.
364pub async fn update_subscription_discount_code_in_transaction(
365    transaction: &mut Transaction<'_, Postgres>,
366    offers: &dyn SubscriptionOfferStore,
367    update: &SubscriptionDiscountCodeUpdate,
368) -> Result<Option<SubscriptionDiscountCodeRecord>, SubscriptionDiscountOperationError> {
369    set_lock_timeout(transaction).await?;
370    if update.status() == SubscriptionDiscountCodeStatus::Active {
371        let offer = lock_offer(
372            transaction,
373            offers,
374            update.billing_scope_id(),
375            update.plan_key(),
376        )
377        .await?;
378        validate_discount_cadence(update.duration(), &offer)?;
379        syrup_rail::discounted_charge(offer.recurring().charge(), update.currency(), update.kind())
380            .map_err(|_| SubscriptionDiscountOperationError::InvalidConfiguration)?;
381    }
382    let (amount_off_cents, percent_off_bps) = discount_value(update.kind());
383    let row = sqlx::query(
384        r#"
385        UPDATE billing_subscription_discount_codes
386        SET label = $4, status = $5, discount_kind = $6,
387            amount_off_cents = $7, percent_off_bps = $8, currency = $9,
388            duration = $10, duration_months = $11, updated_at = now()
389        WHERE id = $1 AND billing_scope_id = $2 AND plan_key = $3
390        RETURNING id, billing_scope_id, plan_key, code_normalized, display_code,
391            label, status, discount_kind, amount_off_cents, percent_off_bps,
392            currency, duration, duration_months, created_at, updated_at
393        "#,
394    )
395    .bind(update.id().as_uuid())
396    .bind(update.billing_scope_id().as_uuid())
397    .bind(update.plan_key().as_str())
398    .bind(update.label())
399    .bind(update.status().as_str())
400    .bind(update.kind().as_str())
401    .bind(amount_off_cents)
402    .bind(percent_off_bps)
403    .bind(update.currency().as_str())
404    .bind(update.duration().as_str())
405    .bind(duration_months(update.duration()))
406    .fetch_optional(&mut **transaction)
407    .await?;
408    if row.is_some() && update.status() == SubscriptionDiscountCodeStatus::Disabled {
409        expire_saved_claims_for_code(
410            transaction,
411            update.billing_scope_id(),
412            update.plan_key(),
413            update.id(),
414        )
415        .await?;
416    }
417    row.as_ref().map(code_from_row).transpose()
418}
419
420pub async fn update_subscription_discount_code(
421    pool: &PgPool,
422    offers: &dyn SubscriptionOfferStore,
423    update: &SubscriptionDiscountCodeUpdate,
424) -> Result<Option<SubscriptionDiscountCodeRecord>, SubscriptionDiscountOperationError> {
425    let mut transaction = pool.begin().await?;
426    let record =
427        update_subscription_discount_code_in_transaction(&mut transaction, offers, update).await?;
428    transaction.commit().await?;
429    Ok(record)
430}
431
432/// Disables an administrative record even when its historical terms are no
433/// longer quoteable against the host's current offer.
434pub async fn disable_subscription_discount_code_in_transaction(
435    transaction: &mut Transaction<'_, Postgres>,
436    billing_scope_id: BillingScopeId,
437    plan_key: &PlanKey,
438    discount_code_id: DiscountCodeId,
439) -> Result<Option<SubscriptionDiscountCodeRecord>, SubscriptionDiscountOperationError> {
440    set_lock_timeout(transaction).await?;
441    let row = sqlx::query(
442        r#"
443        UPDATE billing_subscription_discount_codes
444        SET status = 'disabled', updated_at = now()
445        WHERE id = $1 AND billing_scope_id = $2 AND plan_key = $3
446        RETURNING id, billing_scope_id, plan_key, code_normalized, display_code,
447            label, status, discount_kind, amount_off_cents, percent_off_bps,
448            currency, duration, duration_months, created_at, updated_at
449        "#,
450    )
451    .bind(discount_code_id.as_uuid())
452    .bind(billing_scope_id.as_uuid())
453    .bind(plan_key.as_str())
454    .fetch_optional(&mut **transaction)
455    .await?;
456    if row.is_some() {
457        expire_saved_claims_for_code(transaction, billing_scope_id, plan_key, discount_code_id)
458            .await?;
459    }
460    row.as_ref().map(code_from_row).transpose()
461}
462
463pub async fn disable_subscription_discount_code(
464    pool: &PgPool,
465    billing_scope_id: BillingScopeId,
466    plan_key: &PlanKey,
467    discount_code_id: DiscountCodeId,
468) -> Result<Option<SubscriptionDiscountCodeRecord>, SubscriptionDiscountOperationError> {
469    let mut transaction = pool.begin().await?;
470    let record = disable_subscription_discount_code_in_transaction(
471        &mut transaction,
472        billing_scope_id,
473        plan_key,
474        discount_code_id,
475    )
476    .await?;
477    transaction.commit().await?;
478    Ok(record)
479}
480
481pub async fn claim_subscription_discount(
482    pool: &PgPool,
483    offers: &dyn SubscriptionOfferStore,
484    claim: &SubscriptionDiscountClaim,
485) -> Result<SubscriptionDiscountClaimOutcome, SubscriptionDiscountOperationError> {
486    let mut transaction = pool.begin().await?;
487    let outcome =
488        claim_subscription_discount_in_transaction(&mut transaction, offers, claim).await?;
489    transaction.commit().await?;
490    Ok(outcome)
491}
492
493pub async fn claim_subscription_discount_in_transaction(
494    transaction: &mut Transaction<'_, Postgres>,
495    offers: &dyn SubscriptionOfferStore,
496    claim: &SubscriptionDiscountClaim,
497) -> Result<SubscriptionDiscountClaimOutcome, SubscriptionDiscountOperationError> {
498    claim_subscription_discount_on_connection(transaction, offers, claim).await
499}
500
501/// Executes a claim on a connection that is already inside the caller's
502/// transaction.
503///
504/// This is crate-visible for subscriber-owned orchestration that must share a
505/// host-prepared transaction. The caller owns transaction completion.
506pub(crate) async fn claim_subscription_discount_on_connection(
507    connection: &mut PgConnection,
508    offers: &dyn SubscriptionOfferStore,
509    claim: &SubscriptionDiscountClaim,
510) -> Result<SubscriptionDiscountClaimOutcome, SubscriptionDiscountOperationError> {
511    set_lock_timeout(connection).await?;
512    lock_subscription_aggregate(connection, claim.subscriber_id(), claim.plan_key()).await?;
513    let offer = lock_offer(
514        connection,
515        offers,
516        claim.billing_scope_id(),
517        claim.plan_key(),
518    )
519    .await?;
520    if current_subscription_exists(connection, claim).await? {
521        return Ok(SubscriptionDiscountClaimOutcome::BlockedBySubscription);
522    }
523    let Some(code_row) = find_active_code(
524        connection,
525        claim.billing_scope_id(),
526        claim.plan_key(),
527        claim.code(),
528        true,
529    )
530    .await?
531    else {
532        return Ok(SubscriptionDiscountClaimOutcome::NotFound);
533    };
534    let code = code_from_row(&code_row)?;
535    let quote = quote_for_offer(code, &offer)?;
536    let existing = saved_subscription_discount_claim_on_connection(
537        connection,
538        claim.billing_scope_id(),
539        claim.subscriber_id(),
540        claim.plan_key(),
541    )
542    .await?;
543    if let Some(existing) = existing.as_ref()
544        && existing.snapshot().code() == claim.code()
545    {
546        return Ok(SubscriptionDiscountClaimOutcome::Existing(Box::new(
547            existing.clone(),
548        )));
549    }
550    lock_initial_attempts(connection, claim).await?;
551    if blocking_initial_attempt_exists(connection, claim).await? {
552        return Ok(SubscriptionDiscountClaimOutcome::BlockedByInitialAttempt);
553    }
554    if let Some(existing) = existing {
555        let result = sqlx::query(
556            "UPDATE billing_subscription_discount_claims SET status = 'superseded', superseded_at = now() WHERE id = $1 AND status = 'saved'",
557        )
558        .bind(existing.id().as_uuid())
559        .execute(&mut *connection)
560        .await?;
561        if result.rows_affected() != 1 {
562            return Err(SubscriptionDiscountOperationError::InvalidState(
563                INVALID_DISCOUNT_STATE,
564            ));
565        }
566    }
567    let quoted_code = quote.code();
568    let (amount_off_cents, percent_off_bps) = discount_value(quoted_code.kind());
569    let row = sqlx::query(
570        r#"
571        INSERT INTO billing_subscription_discount_claims (
572            id, billing_scope_id, subscriber_id, plan_key, discount_code_id,
573            code_snapshot, label_snapshot, discount_kind, amount_off_cents,
574            percent_off_bps, currency, duration, duration_months,
575            base_amount_cents, discounted_amount_cents, status
576        )
577        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, 'saved')
578        RETURNING id, billing_scope_id, subscriber_id, plan_key,
579            discount_code_id, code_snapshot, label_snapshot, discount_kind,
580            amount_off_cents, percent_off_bps, currency, duration,
581            duration_months, base_amount_cents, discounted_amount_cents,
582            status, claimed_at, applied_at, applied_subscription_id,
583            applied_payment_attempt_id, superseded_at
584        "#,
585    )
586    .bind(claim.id().as_uuid())
587    .bind(claim.billing_scope_id().as_uuid())
588    .bind(claim.subscriber_id().as_uuid())
589    .bind(claim.plan_key().as_str())
590    .bind(quoted_code.id().as_uuid())
591    .bind(quoted_code.code().as_str())
592    .bind(quoted_code.label())
593    .bind(quoted_code.kind().as_str())
594    .bind(amount_off_cents)
595    .bind(percent_off_bps)
596    .bind(quoted_code.currency().as_str())
597    .bind(quoted_code.duration().as_str())
598    .bind(duration_months(quoted_code.duration()))
599    .bind(quote.base_charge().cents())
600    .bind(quote.discounted_charge().cents())
601    .fetch_one(&mut *connection)
602    .await?;
603    Ok(SubscriptionDiscountClaimOutcome::Saved(Box::new(
604        claim_from_row(&row)?,
605    )))
606}
607
608pub async fn clear_subscription_discount(
609    pool: &PgPool,
610    billing_scope_id: BillingScopeId,
611    subscriber_id: SubscriberId,
612    plan_key: &PlanKey,
613) -> Result<SubscriptionDiscountClearOutcome, SubscriptionDiscountOperationError> {
614    let mut transaction = pool.begin().await?;
615    let outcome = clear_subscription_discount_in_transaction(
616        &mut transaction,
617        billing_scope_id,
618        subscriber_id,
619        plan_key,
620    )
621    .await?;
622    transaction.commit().await?;
623    Ok(outcome)
624}
625
626pub async fn clear_subscription_discount_in_transaction(
627    transaction: &mut Transaction<'_, Postgres>,
628    billing_scope_id: BillingScopeId,
629    subscriber_id: SubscriberId,
630    plan_key: &PlanKey,
631) -> Result<SubscriptionDiscountClearOutcome, SubscriptionDiscountOperationError> {
632    clear_subscription_discount_on_connection(
633        transaction,
634        billing_scope_id,
635        subscriber_id,
636        plan_key,
637    )
638    .await
639}
640
641/// Executes a saved-discount clear on a connection that is already inside the
642/// caller's transaction. The caller owns transaction completion.
643pub(crate) async fn clear_subscription_discount_on_connection(
644    connection: &mut PgConnection,
645    billing_scope_id: BillingScopeId,
646    subscriber_id: SubscriberId,
647    plan_key: &PlanKey,
648) -> Result<SubscriptionDiscountClearOutcome, SubscriptionDiscountOperationError> {
649    set_lock_timeout(connection).await?;
650    lock_subscription_aggregate(connection, subscriber_id, plan_key).await?;
651    let existing = saved_subscription_discount_claim_on_connection(
652        connection,
653        billing_scope_id,
654        subscriber_id,
655        plan_key,
656    )
657    .await?;
658    lock_initial_attempt_rows(connection, billing_scope_id, subscriber_id, plan_key).await?;
659    if blocking_initial_attempt(connection, billing_scope_id, subscriber_id, plan_key).await? {
660        return Ok(SubscriptionDiscountClearOutcome::BlockedByInitialAttempt);
661    }
662    let Some(existing) = existing else {
663        return Ok(SubscriptionDiscountClearOutcome::NotFound);
664    };
665    let row = sqlx::query(
666        r#"
667        UPDATE billing_subscription_discount_claims
668        SET status = 'expired'
669        WHERE id = $1 AND billing_scope_id = $2 AND subscriber_id = $3
670            AND plan_key = $4 AND status = 'saved'
671        RETURNING id, billing_scope_id, subscriber_id, plan_key,
672            discount_code_id, code_snapshot, label_snapshot, discount_kind,
673            amount_off_cents, percent_off_bps, currency, duration,
674            duration_months, base_amount_cents, discounted_amount_cents,
675            status, claimed_at, applied_at, applied_subscription_id,
676            applied_payment_attempt_id, superseded_at
677        "#,
678    )
679    .bind(existing.id().as_uuid())
680    .bind(billing_scope_id.as_uuid())
681    .bind(subscriber_id.as_uuid())
682    .bind(plan_key.as_str())
683    .fetch_one(&mut *connection)
684    .await?;
685    Ok(SubscriptionDiscountClearOutcome::Cleared(Box::new(
686        claim_from_row(&row)?,
687    )))
688}
689
690pub async fn saved_subscription_discount_claim(
691    pool: &PgPool,
692    billing_scope_id: BillingScopeId,
693    subscriber_id: SubscriberId,
694    plan_key: &PlanKey,
695) -> Result<Option<SubscriptionDiscountClaimRecord>, SubscriptionDiscountOperationError> {
696    let row = sqlx::query(
697        r#"
698        SELECT id, billing_scope_id, subscriber_id, plan_key,
699            discount_code_id, code_snapshot, label_snapshot, discount_kind,
700            amount_off_cents, percent_off_bps, currency, duration,
701            duration_months, base_amount_cents, discounted_amount_cents,
702            status, claimed_at, applied_at, applied_subscription_id,
703            applied_payment_attempt_id, superseded_at
704        FROM billing_subscription_discount_claims
705        WHERE billing_scope_id = $1 AND subscriber_id = $2 AND plan_key = $3
706            AND status = 'saved'
707        ORDER BY claimed_at DESC, id DESC LIMIT 1
708        "#,
709    )
710    .bind(billing_scope_id.as_uuid())
711    .bind(subscriber_id.as_uuid())
712    .bind(plan_key.as_str())
713    .fetch_optional(pool)
714    .await?;
715    row.as_ref().map(claim_from_row).transpose()
716}
717
718pub async fn mark_subscription_discount_claim_applied_in_transaction(
719    transaction: &mut Transaction<'_, Postgres>,
720    claim_id: DiscountClaimId,
721    billing_scope_id: BillingScopeId,
722    subscriber_id: SubscriberId,
723    plan_key: &PlanKey,
724    subscription_id: SubscriptionId,
725    payment_attempt_id: PaymentAttemptId,
726) -> Result<SubscriptionDiscountClaimRecord, SubscriptionDiscountOperationError> {
727    set_lock_timeout(transaction).await?;
728    let row = sqlx::query(
729        r#"
730        UPDATE billing_subscription_discount_claims claims
731        SET status = 'applied', applied_at = now(),
732            applied_subscription_id = $5, applied_payment_attempt_id = $6
733        WHERE claims.id = $1
734            AND claims.billing_scope_id = $2
735            AND claims.subscriber_id = $3
736            AND claims.plan_key = $4
737            AND claims.status IN ('saved', 'expired')
738            AND EXISTS (
739                SELECT 1 FROM billing_payment_attempts attempts
740                WHERE attempts.id = $6
741                    AND attempts.billing_scope_id = claims.billing_scope_id
742                    AND attempts.subscriber_id = claims.subscriber_id
743                    AND attempts.plan_key = claims.plan_key
744                    AND attempts.subscription_initial_discount_claim_id = claims.id
745                    AND attempts.attempt_kind = 'subscription_initial'
746                    AND attempts.submitted_at IS NOT NULL
747            )
748        RETURNING claims.id, claims.billing_scope_id, claims.subscriber_id,
749            claims.plan_key, claims.discount_code_id, claims.code_snapshot,
750            claims.label_snapshot, claims.discount_kind, claims.amount_off_cents,
751            claims.percent_off_bps, claims.currency, claims.duration,
752            claims.duration_months, claims.base_amount_cents,
753            claims.discounted_amount_cents, claims.status, claims.claimed_at,
754            claims.applied_at, claims.applied_subscription_id,
755            claims.applied_payment_attempt_id, claims.superseded_at
756        "#,
757    )
758    .bind(claim_id.as_uuid())
759    .bind(billing_scope_id.as_uuid())
760    .bind(subscriber_id.as_uuid())
761    .bind(plan_key.as_str())
762    .bind(subscription_id.as_uuid())
763    .bind(payment_attempt_id.as_uuid())
764    .fetch_optional(&mut **transaction)
765    .await?
766    .ok_or(sqlx::Error::RowNotFound)?;
767    claim_from_row(&row)
768}
769
770#[cfg(test)]
771mod tests;
syrup_rail_postgres/discounts.rs

syrup_rail_postgres/
discounts.rs
