collab: Rework Stripe event processing (#15510)

This PR reworks how we process Stripe events for reconciliation
purposes.

The previous approach in #15480 turns out to not be workable, on account
of the Stripe event IDs not being strictly in order. This meant that we
couldn't reliably compare two arbitrary event IDs and determine which
one was more recent.

This new approach leans on the guidance that Stripe provides for
webhooks events:

> Webhook endpoints might occasionally receive the same event more than
once. You can guard against duplicated event receipts by logging the
[event IDs](https://docs.stripe.com/api/events/object#event_object-id)
you’ve processed, and then not processing already-logged events.
>
> https://docs.stripe.com/webhooks#handle-duplicate-events

We now record processed Stripe events in the `processed_stripe_events`
table and use this to filter out events that have already been
processed, so we do not process them again.

When retrieving events from the Stripe events API we now buffer the
unprocessed events so that we can sort them by their `created` timestamp
and process them in (roughly) the order they occurred.

Release Notes:

- N/A

crates/collab/src/db/queries.rs

@@ -14,6 +14,7 @@ pub mod extensions;
 pub mod hosted_projects;
 pub mod messages;
 pub mod notifications;
+pub mod processed_stripe_events;
 pub mod projects;
 pub mod rate_buckets;
 pub mod rooms;

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

@@ -4,14 +4,12 @@ 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 {
@@ -45,7 +43,6 @@ impl Database {
                 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)

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

@@ -1,5 +1,3 @@
-use sea_orm::IntoActiveValue;
-
 use crate::db::billing_subscription::StripeSubscriptionStatus;
 
 use super::*;
@@ -9,7 +7,6 @@ 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)]
@@ -17,7 +14,6 @@ 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 {
@@ -31,7 +27,6 @@ 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)
@@ -54,7 +49,6 @@ impl Database {
                 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()
             })
             .exec(&*tx)

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

@@ -0,0 +1,69 @@
+use super::*;
+
+#[derive(Debug)]
+pub struct CreateProcessedStripeEventParams {
+    pub stripe_event_id: String,
+    pub stripe_event_type: String,
+    pub stripe_event_created_timestamp: i64,
+}
+
+impl Database {
+    /// Creates a new processed Stripe event.
+    pub async fn create_processed_stripe_event(
+        &self,
+        params: &CreateProcessedStripeEventParams,
+    ) -> Result<()> {
+        self.transaction(|tx| async move {
+            processed_stripe_event::Entity::insert(processed_stripe_event::ActiveModel {
+                stripe_event_id: ActiveValue::set(params.stripe_event_id.clone()),
+                stripe_event_type: ActiveValue::set(params.stripe_event_type.clone()),
+                stripe_event_created_timestamp: ActiveValue::set(
+                    params.stripe_event_created_timestamp,
+                ),
+                ..Default::default()
+            })
+            .exec_without_returning(&*tx)
+            .await?;
+
+            Ok(())
+        })
+        .await
+    }
+
+    /// Returns the processed Stripe event with the specified event ID.
+    pub async fn get_processed_stripe_event_by_event_id(
+        &self,
+        event_id: &str,
+    ) -> Result<Option<processed_stripe_event::Model>> {
+        self.transaction(|tx| async move {
+            Ok(processed_stripe_event::Entity::find_by_id(event_id)
+                .one(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    /// Returns the processed Stripe events with the specified event IDs.
+    pub async fn get_processed_stripe_events_by_event_ids(
+        &self,
+        event_ids: &[&str],
+    ) -> Result<Vec<processed_stripe_event::Model>> {
+        self.transaction(|tx| async move {
+            Ok(processed_stripe_event::Entity::find()
+                .filter(
+                    processed_stripe_event::Column::StripeEventId.is_in(event_ids.iter().copied()),
+                )
+                .all(&*tx)
+                .await?)
+        })
+        .await
+    }
+
+    /// Returns whether the Stripe event with the specified ID has already been processed.
+    pub async fn already_processed_stripe_event(&self, event_id: &str) -> Result<bool> {
+        Ok(self
+            .get_processed_stripe_event_by_event_id(event_id)
+            .await?
+            .is_some())
+    }
+}

crates/collab/src/db/tables.rs

@@ -25,6 +25,7 @@ pub mod notification;
 pub mod notification_kind;
 pub mod observed_buffer_edits;
 pub mod observed_channel_messages;
+pub mod processed_stripe_event;
 pub mod project;
 pub mod project_collaborator;
 pub mod rate_buckets;

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

@@ -9,7 +9,6 @@ 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,7 +10,6 @@ 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/tables/processed_stripe_event.rs

@@ -0,0 +1,16 @@
+use sea_orm::entity::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
+#[sea_orm(table_name = "processed_stripe_events")]
+pub struct Model {
+    #[sea_orm(primary_key)]
+    pub stripe_event_id: String,
+    pub stripe_event_type: String,
+    pub stripe_event_created_timestamp: i64,
+    pub processed_at: DateTime,
+}
+
+#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
+pub enum Relation {}
+
+impl ActiveModelBehavior for ActiveModel {}

crates/collab/src/db/tests.rs

@@ -9,6 +9,7 @@ mod embedding_tests;
 mod extension_tests;
 mod feature_flag_tests;
 mod message_tests;
+mod processed_stripe_event_tests;
 
 use super::*;
 use gpui::BackgroundExecutor;

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

@@ -29,7 +29,6 @@ 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();
@@ -39,7 +38,6 @@ 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();
@@ -65,7 +63,6 @@ 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();
@@ -75,7 +72,6 @@ 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();

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

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+
+use crate::test_both_dbs;
+
+use super::{CreateProcessedStripeEventParams, Database};
+
+test_both_dbs!(
+    test_already_processed_stripe_event,
+    test_already_processed_stripe_event_postgres,
+    test_already_processed_stripe_event_sqlite
+);
+
+async fn test_already_processed_stripe_event(db: &Arc<Database>) {
+    let unprocessed_event_id = "evt_1PiJOuRxOf7d5PNaw2zzWiyO".to_string();
+    let processed_event_id = "evt_1PiIfMRxOf7d5PNakHrAUe8P".to_string();
+
+    db.create_processed_stripe_event(&CreateProcessedStripeEventParams {
+        stripe_event_id: processed_event_id.clone(),
+        stripe_event_type: "customer.created".into(),
+        stripe_event_created_timestamp: 1722355968,
+    })
+    .await
+    .unwrap();
+
+    assert_eq!(
+        db.already_processed_stripe_event(&processed_event_id)
+            .await
+            .unwrap(),
+        true,
+        "Expected {processed_event_id} to already be processed"
+    );
+
+    assert_eq!(
+        db.already_processed_stripe_event(&unprocessed_event_id)
+            .await
+            .unwrap(),
+        false,
+        "Expected {unprocessed_event_id} to be unprocessed"
+    );
+}