From 64fbb0b04e08946e15bc436f7c02a8ae6a88c4af Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 3 Sep 2025 10:48:36 -0400 Subject: [PATCH] subscription events seem okay-ish --- ...user_record_with_new_stripe_customer_id.rs | 19 +- ..._subscription_by_stripe_subscription_id.rs | 17 +- .../upsert_user_subscription_by_stripe_id.rs | 10 +- .../stripe_artcraft_webhook_handler.rs | 212 +--------------- .../customer_subscription_created_handler.rs | 29 +-- .../customer_subscription_deleted_handler.rs | 13 +- .../customer_subscription_updated_handler.rs | 18 +- .../handle_webhook_payload.rs | 227 ++++++++++++++++++ .../webhook/webhook_event_handlers/mod.rs | 1 + .../stripe_artcraft_webhook_error.rs | 3 + 10 files changed, 307 insertions(+), 242 deletions(-) create mode 100644 crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/handle_webhook_payload.rs diff --git a/crates/schema/database/mysql_queries/src/queries/users/user/update/update_user_record_with_new_stripe_customer_id.rs b/crates/schema/database/mysql_queries/src/queries/users/user/update/update_user_record_with_new_stripe_customer_id.rs index 659ca7f274..fb585d7fe8 100644 --- a/crates/schema/database/mysql_queries/src/queries/users/user/update/update_user_record_with_new_stripe_customer_id.rs +++ b/crates/schema/database/mysql_queries/src/queries/users/user/update/update_user_record_with_new_stripe_customer_id.rs @@ -1,6 +1,6 @@ use anyhow::anyhow; -use sqlx::MySqlPool; - +use sqlx::{MySql, MySqlPool}; +use sqlx::pool::PoolConnection; use errors::AnyhowResult; pub async fn update_user_record_with_new_stripe_customer_id( @@ -8,6 +8,19 @@ pub async fn update_user_record_with_new_stripe_customer_id( user_token: &str, maybe_stripe_customer_id: Option<&str> ) -> AnyhowResult<()> { + let mut mysql_connection = mysql_pool.acquire().await?; + update_user_record_with_new_stripe_customer_id_with_connection( + &mut mysql_connection, + user_token, + maybe_stripe_customer_id + ).await +} + +pub async fn update_user_record_with_new_stripe_customer_id_with_connection( + mysql_connection: &mut PoolConnection, + user_token: &str, + maybe_stripe_customer_id: Option<&str> +) -> AnyhowResult<()> { // TODO: This will overwrite whatever the previous customer_id was. // Should we guard against that? @@ -26,7 +39,7 @@ LIMIT 1 user_token, ); - let query_result = query.execute(mysql_pool).await; + let query_result = query.execute(&mut ** mysql_connection).await; match query_result { Ok(_) => Ok(()), diff --git a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_user_subscription_by_stripe_subscription_id.rs b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_user_subscription_by_stripe_subscription_id.rs index 399b4a3e26..c39cddf06e 100644 --- a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_user_subscription_by_stripe_subscription_id.rs +++ b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_user_subscription_by_stripe_subscription_id.rs @@ -1,9 +1,9 @@ use anyhow::anyhow; use chrono::{DateTime, Utc}; -use sqlx::MySqlPool; - use errors::AnyhowResult; use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlPool}; use crate::helpers::boolean_converters::nullable_i8_to_optional_bool; @@ -31,6 +31,17 @@ pub async fn get_user_subscription_by_stripe_subscription_id( stripe_subscription_id: &str, mysql_pool: &MySqlPool ) -> AnyhowResult> { + let mut mysql_connection = mysql_pool.acquire().await?; + get_user_subscription_by_stripe_subscription_id_with_connection( + stripe_subscription_id, + &mut mysql_connection + ).await +} + +pub async fn get_user_subscription_by_stripe_subscription_id_with_connection( + stripe_subscription_id: &str, + mysql_connection: &mut PoolConnection +) -> AnyhowResult> { let maybe_user_record = sqlx::query_as!( RawUserSubscriptionFromDb, @@ -55,7 +66,7 @@ WHERE "#, stripe_subscription_id, ) - .fetch_one(mysql_pool) + .fetch_one(&mut **mysql_connection) .await; match maybe_user_record { diff --git a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/upsert_user_subscription_by_stripe_id.rs b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/upsert_user_subscription_by_stripe_id.rs index 2f1f0418ea..4454c0a57f 100644 --- a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/upsert_user_subscription_by_stripe_id.rs +++ b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/upsert_user_subscription_by_stripe_id.rs @@ -1,10 +1,11 @@ use anyhow::anyhow; use chrono::NaiveDateTime; -use sqlx::MySqlPool; use enums::common::subscription_namespace::SubscriptionNamespace; use errors::AnyhowResult; use reusable_types::stripe::stripe_recurring_interval::StripeRecurringInterval; use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlPool}; use tokens::tokens::user_subscriptions::UserSubscriptionToken; // TODO: Make a trait with default impls to handle common query concerns. @@ -53,6 +54,11 @@ pub struct UpsertUserSubscription<'a> { impl <'a> UpsertUserSubscription<'a> { pub async fn upsert(&'a self, mysql_pool: &MySqlPool) -> AnyhowResult<()> { + let mut conn = mysql_pool.acquire().await?; + self.upsert_with_connection(&mut conn).await + } + + pub async fn upsert_with_connection(&'a self, mysql_connection: &mut PoolConnection) -> AnyhowResult<()> { let token = UserSubscriptionToken::generate().to_string(); // NB: The following behaviors are intentional @@ -149,7 +155,7 @@ ON DUPLICATE KEY UPDATE self.maybe_canceled_at, ); - let query_result = query.execute(mysql_pool).await; + let query_result = query.execute(&mut **mysql_connection).await; let _record_id = match query_result { Ok(res) => res.last_insert_id(), diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/stripe_artcraft_webhook_handler.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/stripe_artcraft_webhook_handler.rs index 26d3aa8b8f..695a685d3e 100644 --- a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/stripe_artcraft_webhook_handler.rs +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/stripe_artcraft_webhook_handler.rs @@ -28,6 +28,7 @@ use sqlx::{MySql, MySqlConnection, MySqlPool}; use std::sync::Arc; use stripe_webhook::{Event, EventObject, Webhook}; use crate::endpoints::webhook::webhook_event_handlers::customer_subscription::customer_subscription_updated_handler::customer_subscription_updated_handler; +use crate::endpoints::webhook::webhook_event_handlers::handle_webhook_payload::handle_webhook_payload; #[derive(Serialize)] pub struct StripeArtcraftWebhookSuccessResponse { @@ -87,14 +88,6 @@ pub async fn stripe_artcraft_webhook_handler( let stripe_event_type = webhook_payload.type_.as_str().to_string(); - //let stripe_event_type = serde_json::to_string(&webhook_payload.type_) - // .map(|s| s.replace("\"", "")) - // .map_err(|err| { - // let reason = format!("Could not serialize webhook type: {:?}", err); - // error!("{}", &reason); - // StripeArtcraftWebhookError::BadInputWithSimpleMessage(reason) - // })?; - // NB: Whether this was test data or live data let stripe_is_production = webhook_payload.livemode; @@ -110,15 +103,14 @@ pub async fn stripe_artcraft_webhook_handler( .await .map_err(|err| { error!("Could not acquire mysql connection: {:?}", err); - StripeArtcraftWebhookError::ServerError + StripeArtcraftWebhookError::ServerError("database error".to_string()) })?; let maybe_previously_played_event = get_stripe_webhook_event_log_by_id_with_connection(&stripe_event_id, &mut mysql_connection) .await .map_err(|err| { - let reason = format!("Could not query previous event by ID ({}): {:?}", &stripe_event_id, err); - error!("{}", &reason); - Err(StripeArtcraftWebhookError::ServerError) + error!("Could not query previous event by ID ({}): {:?}", &stripe_event_id, err); + StripeArtcraftWebhookError::ServerError("database error".to_string()) })?; if let Some(event) = maybe_previously_played_event { @@ -137,7 +129,7 @@ pub async fn stripe_artcraft_webhook_handler( let webhook_summary = handle_webhook_payload( &mut mysql_connection, **server_environment, - webhook_payload, + webhook_payload, &stripe_event_type ).await?; @@ -158,202 +150,10 @@ pub async fn stripe_artcraft_webhook_handler( .map_err(|err| { let reason = format!("Could not insert Stripe webhook event log: {:?}", err); error!("{}", &reason); - StripeArtcraftWebhookError::ServerError + StripeArtcraftWebhookError::ServerError("database error".to_string()) })?; Ok(Json(StripeArtcraftWebhookSuccessResponse { success: true, })) } - -async fn handle_webhook_payload( - mysql_connection: &mut PoolConnection, - server_environment: ServerEnvironment, - webhook_payload: Event, - stripe_event_type: &String -) -> Result { - let mut unhandled_event_type = false; - - let mut webhook_summary = StripeArtcraftWebhookSummary { - maybe_user_token: None, - maybe_event_entity_id: None, - maybe_stripe_customer_id: None, - action_was_taken: false, - should_ignore_retry: false, - }; - - - match webhook_payload.data.object { - - // =============== CHECKOUT SESSIONS =============== - - EventObject::CheckoutSessionCompleted(checkout_session) => { - webhook_summary = checkout_session_completed_handler(checkout_session)?; - } - - // EventObject::CheckoutSessionAsyncPaymentFailed(_) => {} - // EventObject::CheckoutSessionAsyncPaymentSucceeded(_) => {} - // EventObject::CheckoutSessionExpired(_) => {} - - // =============== CUSTOMERS =============== - - EventObject::CustomerCreated(customer) => { - webhook_summary = customer_created_handler(&customer)?; - } - - EventObject::CustomerUpdated(customer) => { - webhook_summary = customer_updated_handler(&customer)?; - } - - EventObject::CustomerDeleted(customer) => { - webhook_summary = customer_deleted_handler(&customer)?; - } - - // =============== CUSTOMER SUBSCRIPTIONS =============== - - EventObject::CustomerSubscriptionCreated(subscription) => { - webhook_summary = customer_subscription_created_handler( - &subscription, - server_environment, - &mysql_pool).await?; - } - - EventObject::CustomerSubscriptionUpdated(subscription) => { - webhook_summary = customer_subscription_updated_handler( - &subscription, - server_environment, - &mysql_pool).await?; - } - - EventObject::CustomerSubscriptionDeleted(subscription) => { - webhook_summary = customer_subscription_deleted_handler( - &subscription, - server_environment, - &mysql_pool).await?; - } - - // EventObject::CustomerSubscriptionPendingUpdateApplied(_) => {} - // EventObject::CustomerSubscriptionPendingUpdateExpired(_) => {} - // EventObject::CustomerSubscriptionTrialWillEnd(_) => {} - - // =============== INVOICES =============== - - EventObject::InvoiceCreated(_invoice) => { - // TODO: We need to respond to this so we don't hold payments up by 72 hours! - // See: https://stripe.com/docs/billing/subscriptions/webhooks - } - EventObject::InvoiceUpdated(invoice) => { - webhook_summary = invoice_updated_handler(&invoice)?; - } - EventObject::InvoicePaid(invoice) => { - webhook_summary = invoice_paid_handler(&invoice)?; - } - EventObject::InvoicePaymentSucceeded(invoice) => { - webhook_summary = invoice_payment_succeeded_handler(&invoice)?; - } - EventObject::InvoicePaymentFailed(invoice) => { - webhook_summary = invoice_payment_failed_handler(&invoice)?; - } - - // EventObject::InvoiceDeleted(_) => {} - // EventObject::InvoiceFinalizationFailed(_) => {} - // EventObject::InvoiceFinalized(_) => {} - // EventObject::InvoiceItemCreated(_) => {} - // EventObject::InvoiceItemDeleted(_) => {} - // EventObject::InvoiceItemUpdated(_) => {} - // EventObject::InvoiceMarkedUncollectible(_) => {} - // EventObject::InvoicePaymentActionRequired(_) => {} - // EventObject::InvoiceSent(_) => {} - // EventObject::InvoiceUpcoming(_) => {} - // EventObject::InvoiceUpdated(_) => {} - // EventObject::InvoiceVoided(_) => {} - - // =============== PAYMENT INTENTS =============== - - EventObject::PaymentIntentSucceeded(payment_intent) => { - webhook_summary = payment_intent_succeeded_handler(&payment_intent)?; - } - - // EventType::PaymentIntentAmountCapturableUpdated => {} - // EventType::PaymentIntentCanceled => {} - // EventType::PaymentIntentCreated => {} - // EventType::PaymentIntentPartiallyFunded => {} - // EventType::PaymentIntentPaymentFailed => {} - // EventType::PaymentIntentProcessing => {} - // EventType::PaymentIntentRequiresAction => {} - // EventType::PaymentIntentRequiresCapture => {} - - // =============== CHARGES =============== - - EventObject::ChargeSucceeded(charge) => { - webhook_summary = charge_succeeded_handler(&charge)?; - } - - // =============== Ignored =============== - - // Ignoring these types (NOTE: This is from the old async-stripe version): - // Account* (6), - // ApplicationFee* (2), - // BalanceAvailable (1), - // BillingPortal* (2), - // CapabilityCreated (1), - // Charge* (13), - // [CheckoutSession* handled above], - // Coupon* (3), - // CreditNote* (3), - // CustomerDiscount* (3), - // CustomerSource* (4), - // [CustomerSubscription* handled above], - // CreditTax* (3), - // FileCreated (1), - // IdentityVerification* (6), - // [Invoice* handled above], - // IssuingAuthorization* (3), - // IssuingCard* (2), - // IssuingCardholder* (2), - // IssuingDispute* (2), - // IssuingTransaction* (2), - // MandateUpdated (1), - // Order* (6), - // [PaymentIntent* handled above], - // PaymentLink* (2), - // PaymentMethod* (4), - // Payout* (5), - // Person* (3), - // Plan* (3), - // Price* (3), - // Product* (3), - // PromotionCode* (2), - // Quote* (4), - // RadarEarlyFraudWarning* (2), - // Recipient* (3), - // ReportingReport* (3), - // Review* (2), - // SetupIntent* (5), -- N/A; collecting payments in the future we don't yet have information for (eg. crowdfunding, rental car, utility bill) - // SubscriptionSchedule* (7), -- N/A; backdate subscriptions or schedule one to start later (eg. for physical goods, like a newspaper) - // SigmaScheduledQueryRunCreated (1), - // Sku (3), - // Source* (7), - // TaxRate* (2), - // TerminalReader* (2), - // TestHelpersTestClock* (5), - // Topup* (5), - // Transfer* (5), - - _ => { - unhandled_event_type = true; - }, - } - - // To play with the payload contents as JSON: - // let json = serde_json::ser::to_string(&event_payload).unwrap(); - // info!("event payload as json: {}", json); - - if unhandled_event_type { - warn!("Unhandled Stripe webhook event type: {} ({:?})", - &stripe_event_type, - &webhook_payload.type_); - } - - Ok(webhook_summary) -} diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_created_handler.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_created_handler.rs index b5fba140c5..54ac96e821 100644 --- a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_created_handler.rs +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_created_handler.rs @@ -6,11 +6,12 @@ use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_e use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_summary::StripeArtcraftWebhookSummary; use enums::common::subscription_namespace::SubscriptionNamespace; use log::error; -use mysql_queries::queries::users::user::update::update_user_record_with_new_stripe_customer_id::update_user_record_with_new_stripe_customer_id; -use mysql_queries::queries::users::user_subscriptions::get_user_subscription_by_stripe_subscription_id::get_user_subscription_by_stripe_subscription_id; +use mysql_queries::queries::users::user::update::update_user_record_with_new_stripe_customer_id::{update_user_record_with_new_stripe_customer_id, update_user_record_with_new_stripe_customer_id_with_connection}; +use mysql_queries::queries::users::user_subscriptions::get_user_subscription_by_stripe_subscription_id::{get_user_subscription_by_stripe_subscription_id, get_user_subscription_by_stripe_subscription_id_with_connection}; use mysql_queries::queries::users::user_subscriptions::upsert_user_subscription_by_stripe_id::UpsertUserSubscription; use reusable_types::server_environment::ServerEnvironment; -use sqlx::MySqlPool; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlConnection, MySqlPool}; use stripe_shared::Subscription; /// Handle event type: 'customer.subscription.created' @@ -20,7 +21,7 @@ use stripe_shared::Subscription; pub async fn customer_subscription_created_handler( subscription: &Subscription, server_environment: ServerEnvironment, - mysql_pool: &MySqlPool, + mysql_connection: &mut PoolConnection, ) -> Result { let summary = subscription_summary_extractor(subscription) @@ -38,9 +39,8 @@ pub async fn customer_subscription_created_handler( should_ignore_retry: false, }; - // NB: It's possible to receive events out of order. - // We won't want to play a `create` event on top. - let maybe_existing_subscription = get_user_subscription_by_stripe_subscription_id(&summary.stripe_subscription_id, &mysql_pool) + let maybe_existing_subscription = get_user_subscription_by_stripe_subscription_id_with_connection( + &summary.stripe_subscription_id, mysql_connection) .await .map_err(|err| { let reason = format!("Mysql error: {:?}", err); @@ -48,6 +48,8 @@ pub async fn customer_subscription_created_handler( StripeArtcraftWebhookError::ServerError(reason) })?; + // NB: It's possible to receive events out of order. + // We won't want to play a `create` event on top. if maybe_existing_subscription.is_some() { result.should_ignore_retry = true; return Ok(result); @@ -58,9 +60,9 @@ pub async fn customer_subscription_created_handler( let product = match maybe_product { None => { - error!("No matching product for stripe product ID: {}", &summary.stripe_product_id); - result.should_ignore_retry = true; - return Ok(result); + return Err(StripeArtcraftWebhookError::BadRequest( + format!("No matching product for stripe product ID: {}", &summary.stripe_product_id) + )); } Some(product) => product, }; @@ -87,7 +89,7 @@ pub async fn customer_subscription_created_handler( maybe_canceled_at: summary.maybe_canceled_at, }; - let _r = upsert.upsert(mysql_pool) + let _r = upsert.upsert_with_connection(mysql_connection) .await .map_err(|err| { let reason = format!("Mysql error: {:?}", err); @@ -95,11 +97,10 @@ pub async fn customer_subscription_created_handler( StripeArtcraftWebhookError::ServerError(reason) })?; - // TODO: Should we care if a user accidentally gets two stripe customer IDs and this // overwrites one of them? - update_user_record_with_new_stripe_customer_id( - mysql_pool, + update_user_record_with_new_stripe_customer_id_with_connection( + mysql_connection, user_token, Some(&summary.stripe_customer_id)) .await diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_deleted_handler.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_deleted_handler.rs index be5d2268e0..907218f31c 100644 --- a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_deleted_handler.rs +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_deleted_handler.rs @@ -5,20 +5,20 @@ use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_e use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_summary::StripeArtcraftWebhookSummary; use enums::common::subscription_namespace::SubscriptionNamespace; use log::error; -use mysql_queries::queries::users::user_subscriptions::get_user_subscription_by_stripe_subscription_id::get_user_subscription_by_stripe_subscription_id; +use mysql_queries::queries::users::user_subscriptions::get_user_subscription_by_stripe_subscription_id::{get_user_subscription_by_stripe_subscription_id, get_user_subscription_by_stripe_subscription_id_with_connection}; use mysql_queries::queries::users::user_subscriptions::upsert_user_subscription_by_stripe_id::UpsertUserSubscription; use reusable_types::server_environment::ServerEnvironment; use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus; -use sqlx::MySqlPool; +use sqlx::{MySql, MySqlPool}; +use sqlx::pool::PoolConnection; use stripe_shared::Subscription; /// Handle event type: 'customer.subscription.deleted' /// Sent when a customer’s subscription ends. pub async fn customer_subscription_deleted_handler( subscription: &Subscription, - //internal_subscription_product_lookup: &dyn InternalSubscriptionProductLookup, server_environment: ServerEnvironment, - mysql_pool: &MySqlPool, + mysql_connection: &mut PoolConnection, ) -> Result { let summary = subscription_summary_extractor(subscription) .map_err(|err| { @@ -50,7 +50,8 @@ pub async fn customer_subscription_deleted_handler( }; // NB: It's possible to receive events out of order. - let maybe_existing_subscription = get_user_subscription_by_stripe_subscription_id(&summary.stripe_subscription_id, &mysql_pool) + let maybe_existing_subscription = get_user_subscription_by_stripe_subscription_id_with_connection( + &summary.stripe_subscription_id, mysql_connection) .await .map_err(|err| { let reason = format!("Mysql error: {:?}", err); @@ -93,7 +94,7 @@ pub async fn customer_subscription_deleted_handler( maybe_canceled_at: summary.maybe_canceled_at, }; - let _r = upsert.upsert(mysql_pool) + let _r = upsert.upsert_with_connection(mysql_connection) .await .map_err(|err| { let reason = format!("Mysql error: {:?}", err); diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_updated_handler.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_updated_handler.rs index 5eacf2d55b..d18f6906d1 100644 --- a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_updated_handler.rs +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/customer_subscription/customer_subscription_updated_handler.rs @@ -5,19 +5,20 @@ use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_e use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_summary::StripeArtcraftWebhookSummary; use enums::common::subscription_namespace::SubscriptionNamespace; use log::{error, info}; -use mysql_queries::queries::users::user::update::update_user_record_with_new_stripe_customer_id::update_user_record_with_new_stripe_customer_id; -use mysql_queries::queries::users::user_subscriptions::get_user_subscription_by_stripe_subscription_id::get_user_subscription_by_stripe_subscription_id; +use mysql_queries::queries::users::user::update::update_user_record_with_new_stripe_customer_id::{update_user_record_with_new_stripe_customer_id, update_user_record_with_new_stripe_customer_id_with_connection}; +use mysql_queries::queries::users::user_subscriptions::get_user_subscription_by_stripe_subscription_id::{get_user_subscription_by_stripe_subscription_id, get_user_subscription_by_stripe_subscription_id_with_connection}; use mysql_queries::queries::users::user_subscriptions::upsert_user_subscription_by_stripe_id::UpsertUserSubscription; use reusable_types::server_environment::ServerEnvironment; use reusable_types::stripe::stripe_subscription_status::StripeSubscriptionStatus; -use sqlx::MySqlPool; +use sqlx::pool::PoolConnection; +use sqlx::{MySql, MySqlPool}; use stripe_shared::Subscription; /// Handle event type: 'customer.subscription.updated' pub async fn customer_subscription_updated_handler( subscription: &Subscription, server_environment: ServerEnvironment, - mysql_pool: &MySqlPool, + mysql_connection: &mut PoolConnection, ) -> Result { let summary = subscription_summary_extractor(subscription) @@ -50,7 +51,8 @@ pub async fn customer_subscription_updated_handler( }; // NB: It's possible to receive events out of order. - let maybe_existing_subscription = get_user_subscription_by_stripe_subscription_id(&summary.stripe_subscription_id, &mysql_pool) + let maybe_existing_subscription = get_user_subscription_by_stripe_subscription_id_with_connection( + &summary.stripe_subscription_id, mysql_connection) .await .map_err(|err| { let reason = format!("Mysql error: {:?}", err); @@ -92,7 +94,7 @@ pub async fn customer_subscription_updated_handler( maybe_canceled_at: summary.maybe_canceled_at, }; - let _r = upsert.upsert(mysql_pool) + let _r = upsert.upsert_with_connection(mysql_connection) .await .map_err(|err| { let reason = format!("Mysql error: {:?}", err); @@ -104,8 +106,8 @@ pub async fn customer_subscription_updated_handler( // TODO: Should we care if a user accidentally gets two stripe customer IDs and this // overwrites one of them? - update_user_record_with_new_stripe_customer_id( - mysql_pool, + update_user_record_with_new_stripe_customer_id_with_connection( + mysql_connection, user_token, Some(&summary.stripe_customer_id)) .await diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/handle_webhook_payload.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/handle_webhook_payload.rs new file mode 100644 index 0000000000..92a253869e --- /dev/null +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/handle_webhook_payload.rs @@ -0,0 +1,227 @@ +use crate::endpoints::webhook::webhook_event_handlers::charge::charge_succeeded_handler::charge_succeeded_handler; +use crate::endpoints::webhook::webhook_event_handlers::checkout_session::checkout_session_completed_handler::checkout_session_completed_handler; +use crate::endpoints::webhook::webhook_event_handlers::customer::customer_created_handler::customer_created_handler; +use crate::endpoints::webhook::webhook_event_handlers::customer::customer_deleted_handler::customer_deleted_handler; +use crate::endpoints::webhook::webhook_event_handlers::customer::customer_updated_handler::customer_updated_handler; +use crate::endpoints::webhook::webhook_event_handlers::customer_subscription::customer_subscription_created_handler::customer_subscription_created_handler; +use crate::endpoints::webhook::webhook_event_handlers::customer_subscription::customer_subscription_updated_handler::customer_subscription_updated_handler; +use crate::endpoints::webhook::webhook_event_handlers::invoice::invoice_paid_handler::invoice_paid_handler; +use crate::endpoints::webhook::webhook_event_handlers::invoice::invoice_payment_failed::invoice_payment_failed_handler; +use crate::endpoints::webhook::webhook_event_handlers::invoice::invoice_payment_succeeded_handler::invoice_payment_succeeded_handler; +use crate::endpoints::webhook::webhook_event_handlers::invoice::invoice_updated_handler::invoice_updated_handler; +use crate::endpoints::webhook::webhook_event_handlers::payment_intent::payment_intent_succeeded_handler::payment_intent_succeeded_handler; +use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_error::StripeArtcraftWebhookError; +use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_summary::StripeArtcraftWebhookSummary; +use log::warn; +use reusable_types::server_environment::ServerEnvironment; +use sqlx::pool::PoolConnection; +use sqlx::MySql; +use stripe_webhook::{Event, EventObject}; +use crate::endpoints::webhook::webhook_event_handlers::customer_subscription::customer_subscription_deleted_handler::customer_subscription_deleted_handler; + +pub async fn handle_webhook_payload( + mysql_connection: &mut PoolConnection, + server_environment: ServerEnvironment, + webhook_payload: Event, + stripe_event_type: &String +) -> Result { + let mut unhandled_event_type = false; + + let mut webhook_summary = StripeArtcraftWebhookSummary { + maybe_user_token: None, + maybe_event_entity_id: None, + maybe_stripe_customer_id: None, + action_was_taken: false, + should_ignore_retry: false, + }; + + /* + You usually need these: + 1. Subscription events: to track contract state (active, canceled, past_due, trialing, etc.). + 2. Invoice events: to track actual money flow and handle edge cases like failed payments, dunning, or refunds. + + Minimal Set to Cover Payment Lifecycle + - invoice.paid → confirm credits or entitlements + - invoice.payment_failed → trigger dunning or downgrade access + - customer.subscription.deleted → stop service when canceled + - customer.subscription.updated → react to upgrades, downgrades, trial end + - (Optionally) checkout.session.completed → provision on signup + */ + + match webhook_payload.data.object { + + // =============== CHECKOUT SESSIONS =============== + + EventObject::CheckoutSessionCompleted(checkout_session) => { + webhook_summary = checkout_session_completed_handler(checkout_session)?; + } + + // EventObject::CheckoutSessionAsyncPaymentFailed(_) => {} + // EventObject::CheckoutSessionAsyncPaymentSucceeded(_) => {} + // EventObject::CheckoutSessionExpired(_) => {} + + // =============== CUSTOMERS =============== + + EventObject::CustomerCreated(customer) => { + webhook_summary = customer_created_handler(&customer)?; + } + + EventObject::CustomerUpdated(customer) => { + webhook_summary = customer_updated_handler(&customer)?; + } + + EventObject::CustomerDeleted(customer) => { + webhook_summary = customer_deleted_handler(&customer)?; + } + + // =============== CUSTOMER SUBSCRIPTIONS =============== + + EventObject::CustomerSubscriptionCreated(subscription) => { + webhook_summary = customer_subscription_created_handler( + &subscription, + server_environment, + mysql_connection, + ).await?; + } + + EventObject::CustomerSubscriptionUpdated(subscription) => { + webhook_summary = customer_subscription_updated_handler( + &subscription, + server_environment, + mysql_connection, + ).await?; + } + + EventObject::CustomerSubscriptionDeleted(subscription) => { + webhook_summary = customer_subscription_deleted_handler( + &subscription, + server_environment, + mysql_connection, + ).await?; + } + + // EventObject::CustomerSubscriptionPendingUpdateApplied(_) => {} + // EventObject::CustomerSubscriptionPendingUpdateExpired(_) => {} + // EventObject::CustomerSubscriptionTrialWillEnd(_) => {} + + // =============== INVOICES =============== + + EventObject::InvoicePaid(invoice) => { + webhook_summary = invoice_paid_handler(&invoice)?; + } + EventObject::InvoiceCreated(_invoice) => { + // TODO: We need to respond to this so we don't hold payments up by 72 hours! + // See: https://stripe.com/docs/billing/subscriptions/webhooks + } + EventObject::InvoiceUpdated(invoice) => { + webhook_summary = invoice_updated_handler(&invoice)?; + } + EventObject::InvoicePaymentSucceeded(invoice) => { + webhook_summary = invoice_payment_succeeded_handler(&invoice)?; + } + EventObject::InvoicePaymentFailed(invoice) => { + webhook_summary = invoice_payment_failed_handler(&invoice)?; + } + + // EventObject::InvoiceDeleted(_) => {} + // EventObject::InvoiceFinalizationFailed(_) => {} + // EventObject::InvoiceFinalized(_) => {} + // EventObject::InvoiceItemCreated(_) => {} + // EventObject::InvoiceItemDeleted(_) => {} + // EventObject::InvoiceItemUpdated(_) => {} + // EventObject::InvoiceMarkedUncollectible(_) => {} + // EventObject::InvoicePaymentActionRequired(_) => {} + // EventObject::InvoiceSent(_) => {} + // EventObject::InvoiceUpcoming(_) => {} + // EventObject::InvoiceUpdated(_) => {} + // EventObject::InvoiceVoided(_) => {} + + // =============== PAYMENT INTENTS =============== + + EventObject::PaymentIntentSucceeded(payment_intent) => { + webhook_summary = payment_intent_succeeded_handler(&payment_intent)?; + } + + // EventType::PaymentIntentAmountCapturableUpdated => {} + // EventType::PaymentIntentCanceled => {} + // EventType::PaymentIntentCreated => {} + // EventType::PaymentIntentPartiallyFunded => {} + // EventType::PaymentIntentPaymentFailed => {} + // EventType::PaymentIntentProcessing => {} + // EventType::PaymentIntentRequiresAction => {} + // EventType::PaymentIntentRequiresCapture => {} + + // =============== CHARGES =============== + + EventObject::ChargeSucceeded(charge) => { + webhook_summary = charge_succeeded_handler(&charge)?; + } + + // =============== Ignored =============== + + // Ignoring these types (NOTE: This is from the old async-stripe version): + // Account* (6), + // ApplicationFee* (2), + // BalanceAvailable (1), + // BillingPortal* (2), + // CapabilityCreated (1), + // Charge* (13), + // [CheckoutSession* handled above], + // Coupon* (3), + // CreditNote* (3), + // CustomerDiscount* (3), + // CustomerSource* (4), + // [CustomerSubscription* handled above], + // CreditTax* (3), + // FileCreated (1), + // IdentityVerification* (6), + // [Invoice* handled above], + // IssuingAuthorization* (3), + // IssuingCard* (2), + // IssuingCardholder* (2), + // IssuingDispute* (2), + // IssuingTransaction* (2), + // MandateUpdated (1), + // Order* (6), + // [PaymentIntent* handled above], + // PaymentLink* (2), + // PaymentMethod* (4), + // Payout* (5), + // Person* (3), + // Plan* (3), + // Price* (3), + // Product* (3), + // PromotionCode* (2), + // Quote* (4), + // RadarEarlyFraudWarning* (2), + // Recipient* (3), + // ReportingReport* (3), + // Review* (2), + // SetupIntent* (5), -- N/A; collecting payments in the future we don't yet have information for (eg. crowdfunding, rental car, utility bill) + // SubscriptionSchedule* (7), -- N/A; backdate subscriptions or schedule one to start later (eg. for physical goods, like a newspaper) + // SigmaScheduledQueryRunCreated (1), + // Sku (3), + // Source* (7), + // TaxRate* (2), + // TerminalReader* (2), + // TestHelpersTestClock* (5), + // Topup* (5), + // Transfer* (5), + + _ => { + unhandled_event_type = true; + }, + } + + // To play with the payload contents as JSON: + // let json = serde_json::ser::to_string(&event_payload).unwrap(); + // info!("event payload as json: {}", json); + + if unhandled_event_type { + warn!("Unhandled Stripe webhook event type: {} ({:?})", + &stripe_event_type, + &webhook_payload.type_); + } + + Ok(webhook_summary) +} diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/mod.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/mod.rs index c0262f9f3b..53d4bd7d35 100644 --- a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/mod.rs +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/mod.rs @@ -7,3 +7,4 @@ pub(crate) mod customer_subscription; pub(crate) mod invoice; pub(crate) mod payment_intent; pub(crate) mod stripe_artcraft_webhook_summary; +pub mod handle_webhook_payload; diff --git a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/stripe_artcraft_webhook_error.rs b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/stripe_artcraft_webhook_error.rs index e79d9cdebe..fc82c6e4c6 100644 --- a/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/stripe_artcraft_webhook_error.rs +++ b/crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/stripe_artcraft_webhook_error.rs @@ -1,5 +1,6 @@ //! Returned by the webhook endpoint, but also dispatched event handler functions. +use std::error::Error; use std::fmt; use actix_web::http::StatusCode; @@ -13,6 +14,8 @@ pub enum StripeArtcraftWebhookError { ServerError(String), } +impl Error for StripeArtcraftWebhookError {} + impl ResponseError for StripeArtcraftWebhookError { fn status_code(&self) -> StatusCode { match self {