handle events with idempotency by ignoring replays

crates/component_subsystem/billing/src/stripe/http_endpoints/stripe_webhook_handler.rs

@@ -30,6 +30,7 @@ use sqlx::MySqlPool;
 use std::collections::HashMap;
 use std::fmt;
 use stripe::{EventObject, EventType, PaymentIntentStatus, Webhook};
+use database_queries::queries::billing::stripe::get_stripe_webhook_event_log_by_id::get_stripe_webhook_event_log_by_id;
 
 #[derive(Serialize)]
 pub struct StripeWebhookSuccessResponse {
@@ -92,6 +93,23 @@ pub async fn stripe_webhook_handler(
     &stripe_event_created_at,
     webhook_payload.pending_webhooks);
 
+  let maybe_previously_played_event = get_stripe_webhook_event_log_by_id(&stripe_event_id, &mysql_pool)
+      .await
+      .map_err(|err| {
+        error!("Could not query previous event by ID ({}): {:?}", &stripe_event_id, err);
+        StripeWebhookError::ServerError
+      })?;
+
+  if let Some(event) = maybe_previously_played_event {
+    // The event is being replayed by Stripe, and we've already handled it.
+    // We'll ignore it so that we remain idempotent.
+    if event.should_ignore_retry {
+      warn!("Stripe is replying event with ID={}. Ignoring it since we have previously handled it.",
+        &stripe_event_id);
+      return report_success();
+    }
+  }
+
   let mut unhandled_event_type = false;
 
   let mut webhook_summary = StripeWebhookSummary {
@@ -310,6 +328,11 @@ pub async fn stripe_webhook_handler(
         StripeWebhookError::ServerError
       })?;
 
+  report_success()
+}
+
+#[inline]
+fn report_success() -> Result<HttpResponse, StripeWebhookError> {
   let response = StripeWebhookSuccessResponse {
     success: true,
   };

crates/lib/database_queries/src/queries/billing/stripe/get_stripe_webhook_event_log_by_id.rs

@@ -0,0 +1,54 @@
+use anyhow::anyhow;
+use chrono::{DateTime, NaiveDateTime, Utc};
+use container_common::anyhow_result::AnyhowResult;
+use crate::helpers::boolean_converters::i8_to_bool;
+use log::warn;
+use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus;
+use sqlx::MySqlPool;
+
+pub struct StripeWebhookEventLog {
+  pub stripe_event_id: String,
+
+  /// This helps us handle idempotency and ignore repeat events that get replayed.
+  /// If set to true, we shouldn't process the same event again.
+  pub should_ignore_retry: bool,
+}
+
+pub async fn get_stripe_webhook_event_log_by_id(
+  stripe_event_id: &str,
+  mysql_pool: &MySqlPool
+) -> AnyhowResult<Option<StripeWebhookEventLog>> {
+
+  let maybe_record = sqlx::query_as!(
+      RawStripeWebhookEventLogFromDb,
+        r#"
+SELECT
+  stripe_event_id,
+  should_ignore_retry
+FROM stripe_webhook_event_logs
+WHERE
+  stripe_event_id = ?
+        "#,
+        stripe_event_id,
+    )
+      .fetch_one(mysql_pool)
+      .await;
+
+  match maybe_record {
+    Err(sqlx::error::Error::RowNotFound) => Ok(None),
+    Err(e) => {
+      Err(anyhow!("mysql query error: {:?}", e))
+    }
+    Ok(r) => {
+      Ok(Some(StripeWebhookEventLog {
+        stripe_event_id: r.stripe_event_id,
+        should_ignore_retry: i8_to_bool(r.should_ignore_retry),
+      }))
+    },
+  }
+}
+
+struct RawStripeWebhookEventLogFromDb {
+  pub stripe_event_id: String,
+  pub should_ignore_retry: i8,
+}

crates/lib/database_queries/src/queries/billing/stripe/mod.rs

@@ -1 +1,2 @@
+pub mod get_stripe_webhook_event_log_by_id;
 pub mod insert_stripe_webhook_event_log;