diff --git a/crates/component_subsystem/billing/src/stripe/http_endpoints/stripe_webhook_handler.rs b/crates/component_subsystem/billing/src/stripe/http_endpoints/stripe_webhook_handler.rs index 8e82bb96eb..eada4472b0 100644 --- a/crates/component_subsystem/billing/src/stripe/http_endpoints/stripe_webhook_handler.rs +++ b/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 { let response = StripeWebhookSuccessResponse { success: true, }; diff --git a/crates/lib/database_queries/src/queries/billing/stripe/get_stripe_webhook_event_log_by_id.rs b/crates/lib/database_queries/src/queries/billing/stripe/get_stripe_webhook_event_log_by_id.rs new file mode 100644 index 0000000000..cd7af6e8ef --- /dev/null +++ b/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> { + + 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, +} diff --git a/crates/lib/database_queries/src/queries/billing/stripe/mod.rs b/crates/lib/database_queries/src/queries/billing/stripe/mod.rs index 3a522eced5..8e95f58c36 100644 --- a/crates/lib/database_queries/src/queries/billing/stripe/mod.rs +++ b/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; \ No newline at end of file