collab: Keep track of last seen Stripe event for each record (#15480)

This PR improves our Stripe event handling by keeping track of the last
event we've seen for each record.

The `billing_customers` and `billing_subscriptions` tables both have a
new `last_stripe_event_id` column. When we apply an event to one of
these records, we store the event ID that was applied.

Then, when we are going through events we can ignore any event that has
an ID that came before the `last_stripe_event_id` (based on the
lexicographical ordering of the IDs).

Release Notes:

- N/A

crates/collab/src/db/queries/billing_customers.rs

@@ -4,6 +4,14 @@ use super::*;
 pub struct CreateBillingCustomerParams {
     pub user_id: UserId,
     pub stripe_customer_id: String,
+    pub last_stripe_event_id: Option<String>,
+}
+
+#[derive(Debug, Default)]
+pub struct UpdateBillingCustomerParams {
+    pub user_id: ActiveValue<UserId>,
+    pub stripe_customer_id: ActiveValue<String>,
+    pub last_stripe_event_id: ActiveValue<Option<String>>,
 }
 
 impl Database {
@@ -26,6 +34,28 @@ impl Database {
         .await
     }
 
+    /// Updates the specified billing customer.
+    pub async fn update_billing_customer(
+        &self,
+        id: BillingCustomerId,
+        params: &UpdateBillingCustomerParams,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            billing_customer::Entity::update(billing_customer::ActiveModel {
+                id: ActiveValue::set(id),
+                user_id: params.user_id.clone(),
+                stripe_customer_id: params.stripe_customer_id.clone(),
+                last_stripe_event_id: params.last_stripe_event_id.clone(),
+                ..Default::default()
+            })
+            .exec(&*tx)
+            .await?;
+
+            Ok(())
+        })
+        .await
+    }
+
     /// Returns the billing customer for the user with the specified ID.
     pub async fn get_billing_customer_by_user_id(
         &self,

crates/collab/src/db/queries/billing_subscriptions.rs

@@ -1,3 +1,5 @@
+use sea_orm::IntoActiveValue;
+
 use crate::db::billing_subscription::StripeSubscriptionStatus;
 
 use super::*;
@@ -7,6 +9,15 @@ pub struct CreateBillingSubscriptionParams {
     pub billing_customer_id: BillingCustomerId,
     pub stripe_subscription_id: String,
     pub stripe_subscription_status: StripeSubscriptionStatus,
+    pub last_stripe_event_id: Option<String>,
+}
+
+#[derive(Debug, Default)]
+pub struct UpdateBillingSubscriptionParams {
+    pub billing_customer_id: ActiveValue<BillingCustomerId>,
+    pub stripe_subscription_id: ActiveValue<String>,
+    pub stripe_subscription_status: ActiveValue<StripeSubscriptionStatus>,
+    pub last_stripe_event_id: ActiveValue<Option<String>>,
 }
 
 impl Database {
@@ -20,6 +31,7 @@ impl Database {
                 billing_customer_id: ActiveValue::set(params.billing_customer_id),
                 stripe_subscription_id: ActiveValue::set(params.stripe_subscription_id.clone()),
                 stripe_subscription_status: ActiveValue::set(params.stripe_subscription_status),
+                last_stripe_event_id: params.last_stripe_event_id.clone().into_active_value(),
                 ..Default::default()
             })
             .exec_without_returning(&*tx)
@@ -30,24 +42,22 @@ impl Database {
         .await
     }
 
-    /// Upserts the billing subscription by its Stripe subscription ID.
-    pub async fn upsert_billing_subscription_by_stripe_subscription_id(
+    /// Updates the specified billing subscription.
+    pub async fn update_billing_subscription(
         &self,
-        params: &CreateBillingSubscriptionParams,
+        id: BillingSubscriptionId,
+        params: &UpdateBillingSubscriptionParams,
     ) -> Result<()> {
         self.transaction(|tx| async move {
-            billing_subscription::Entity::insert(billing_subscription::ActiveModel {
-                billing_customer_id: ActiveValue::set(params.billing_customer_id),
-                stripe_subscription_id: ActiveValue::set(params.stripe_subscription_id.clone()),
-                stripe_subscription_status: ActiveValue::set(params.stripe_subscription_status),
+            billing_subscription::Entity::update(billing_subscription::ActiveModel {
+                id: ActiveValue::set(id),
+                billing_customer_id: params.billing_customer_id.clone(),
+                stripe_subscription_id: params.stripe_subscription_id.clone(),
+                stripe_subscription_status: params.stripe_subscription_status.clone(),
+                last_stripe_event_id: params.last_stripe_event_id.clone(),
                 ..Default::default()
             })
-            .on_conflict(
-                OnConflict::columns([billing_subscription::Column::StripeSubscriptionId])
-                    .update_columns([billing_subscription::Column::StripeSubscriptionStatus])
-                    .to_owned(),
-            )
-            .exec_with_returning(&*tx)
+            .exec(&*tx)
             .await?;
 
             Ok(())
@@ -68,6 +78,22 @@ impl Database {
         .await
     }
 
+    /// Returns the billing subscription with the specified Stripe subscription ID.
+    pub async fn get_billing_subscription_by_stripe_subscription_id(
+        &self,
+        stripe_subscription_id: &str,
+    ) -> Result<Option<billing_subscription::Model>> {
+        self.transaction(|tx| async move {
+            Ok(billing_subscription::Entity::find()
+                .filter(
+                    billing_subscription::Column::StripeSubscriptionId.eq(stripe_subscription_id),
+                )
+                .one(&*tx)
+                .await?)
+        })
+        .await
+    }
+
     /// Returns all of the billing subscriptions for the user with the specified ID.
     ///
     /// Note that this returns the subscriptions regardless of their status.

crates/collab/src/db/tables/billing_customer.rs

@@ -9,6 +9,7 @@ pub struct Model {
     pub id: BillingCustomerId,
     pub user_id: UserId,
     pub stripe_customer_id: String,
+    pub last_stripe_event_id: Option<String>,
     pub created_at: DateTime,
 }
 

crates/collab/src/db/tables/billing_subscription.rs

@@ -10,6 +10,7 @@ pub struct Model {
     pub billing_customer_id: BillingCustomerId,
     pub stripe_subscription_id: String,
     pub stripe_subscription_status: StripeSubscriptionStatus,
+    pub last_stripe_event_id: Option<String>,
     pub created_at: DateTime,
 }
 

crates/collab/src/db/tests/billing_subscription_tests.rs

@@ -29,6 +29,7 @@ async fn test_get_active_billing_subscriptions(db: &Arc<Database>) {
             .create_billing_customer(&CreateBillingCustomerParams {
                 user_id,
                 stripe_customer_id: "cus_active_user".into(),
+                last_stripe_event_id: None,
             })
             .await
             .unwrap();
@@ -38,6 +39,7 @@ async fn test_get_active_billing_subscriptions(db: &Arc<Database>) {
             billing_customer_id: customer.id,
             stripe_subscription_id: "sub_active_user".into(),
             stripe_subscription_status: StripeSubscriptionStatus::Active,
+            last_stripe_event_id: None,
         })
         .await
         .unwrap();
@@ -63,6 +65,7 @@ async fn test_get_active_billing_subscriptions(db: &Arc<Database>) {
             .create_billing_customer(&CreateBillingCustomerParams {
                 user_id,
                 stripe_customer_id: "cus_past_due_user".into(),
+                last_stripe_event_id: None,
             })
             .await
             .unwrap();
@@ -72,6 +75,7 @@ async fn test_get_active_billing_subscriptions(db: &Arc<Database>) {
             billing_customer_id: customer.id,
             stripe_subscription_id: "sub_past_due_user".into(),
             stripe_subscription_status: StripeSubscriptionStatus::PastDue,
+            last_stripe_event_id: None,
         })
         .await
         .unwrap();