subscription events seem okay-ish

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<MySql>,
+    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(()),

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<Option<UserSubscription>> {
+  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<MySql>
+) -> AnyhowResult<Option<UserSubscription>> {
 
   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 {

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<MySql>) -> 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(),

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<MySql>,
-  server_environment: ServerEnvironment,
-  webhook_payload: Event,
-  stripe_event_type: &String
-) -> Result<StripeArtcraftWebhookSummary, StripeArtcraftWebhookError> {
-  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)
-}

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<MySql>,
 ) -> Result<StripeArtcraftWebhookSummary, StripeArtcraftWebhookError> {
 
   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

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<MySql>,
 ) -> Result<StripeArtcraftWebhookSummary, StripeArtcraftWebhookError> {
   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);

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<MySql>,
 ) -> Result<StripeArtcraftWebhookSummary, StripeArtcraftWebhookError> {
 
   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

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<MySql>,
+  server_environment: ServerEnvironment,
+  webhook_payload: Event,
+  stripe_event_type: &String
+) -> Result<StripeArtcraftWebhookSummary, StripeArtcraftWebhookError> {
+  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)
+}

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;

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 {