transactional

_sql/migrations_squashed/stripe_webhook_event_logs.sql

@@ -0,0 +1,62 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+CREATE TABLE stripe_webhook_event_logs (
+  -- Not used for anything except replication.
+  id BIGINT(20) NOT NULL AUTO_INCREMENT,
+
+  -- ========== STRIPE DATA ==========
+
+  -- You can safely assume object IDs we generate will never exceed 255 characters,
+  -- but you should be able to handle IDs of up to that length. If for example you’re using MySQL,
+  -- you should store IDs in a VARCHAR(255) COLLATE utf8_bin column (the COLLATE configuration ensures
+  -- case-sensitivity in lookups).
+  -- https://stackoverflow.com/a/61809494
+  stripe_event_id VARCHAR(255) NOT NULL,
+
+  -- The type of webhook event
+  stripe_event_type VARCHAR(255) NOT NULL,
+
+  -- The id of the entity that was the subject of the event.
+  -- We don't always extract this.
+  maybe_stripe_event_entity_id VARCHAR(255) DEFAULT NULL,
+
+  -- The id of the entity that was the subject of the event.
+  -- We don't always extract this.
+  maybe_stripe_customer_id VARCHAR(255) DEFAULT NULL,
+
+  -- Whether this is in production or test mode in Stripe.
+  -- This is controlled by which API keys are used.
+  -- "livemode"=true in production.
+  stripe_is_production BOOLEAN NOT NULL DEFAULT FALSE,
+
+  -- ========== INTERNAL SYSTEM METADATA ==========
+
+  -- Whether we handled the event with any sort of action
+  action_was_taken BOOLEAN NOT NULL DEFAULT FALSE,
+
+  -- Whether we should ignore any future retries by Stripe replaying the event.
+  -- This helps us treat events with idempotency.
+  should_ignore_retry BOOLEAN NOT NULL DEFAULT FALSE,
+
+  -- Some event types may attach a user token
+  maybe_user_token VARCHAR(32) DEFAULT NULL,
+
+  -- ========== TIMESTAMPS ==========
+
+  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+  -- When the event was created on Stripe's end
+  stripe_event_created_at TIMESTAMP NOT NULL,
+
+  -- ========== INDICES ==========
+  PRIMARY KEY (id),
+  UNIQUE KEY (stripe_event_id),
+  KEY index_stripe_event_type (stripe_event_type),
+  KEY index_maybe_stripe_event_entity_id (maybe_stripe_event_entity_id),
+  KEY index_maybe_stripe_customer_id (maybe_stripe_customer_id),
+  KEY fk_maybe_user_token (maybe_user_token)
+
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

crates/schema/database/mysql_queries/src/queries/billing/stripe/insert_stripe_webhook_event_log.rs

@@ -1,6 +1,6 @@
 use anyhow::anyhow;
 use chrono::NaiveDateTime;
-use sqlx::MySqlPool;
+use sqlx::{MySql, MySqlPool, Transaction};
 
 use errors::AnyhowResult;
 
@@ -20,7 +20,7 @@ pub struct InsertStripeWebhookEventLog {
 
 impl InsertStripeWebhookEventLog {
 
-  pub async fn insert(&self, mysql_pool: &MySqlPool) -> AnyhowResult<()> {
+  pub async fn insert(&self, transaction: &mut Transaction<'_, MySql>) -> AnyhowResult<()> {
 
     let query = sqlx::query!(
         r#"
@@ -47,7 +47,7 @@ SET
       self.should_ignore_retry,
     );
 
-    let query_result = query.execute(mysql_pool).await;
+    let query_result = query.execute(transaction).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

@@ -3,6 +3,8 @@ use actix_web::{web, HttpRequest, HttpResponse};
 
 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_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::handle_webhook_payload::handle_webhook_payload;
 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::payment_intent::payment_intent_succeeded_handler::payment_intent_succeeded_handler;
@@ -18,11 +20,9 @@ use mysql_queries::queries::billing::stripe::insert_stripe_webhook_event_log::In
 use reusable_types::server_environment::ServerEnvironment;
 use serde_derive::Serialize;
 use sqlx::pool::PoolConnection;
-use sqlx::{MySql, MySqlConnection, MySqlPool};
+use sqlx::{Acquire, MySql, MySqlConnection, MySqlPool, Transaction};
 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;
 
 /*
 
@@ -120,6 +120,7 @@ pub async fn stripe_artcraft_webhook_handler(
         StripeArtcraftWebhookError::ServerError("database error".to_string())
       })?;
 
+  // TODO: Not sure this retry / idempotency logic is correct
   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.
@@ -133,8 +134,50 @@ pub async fn stripe_artcraft_webhook_handler(
     }
   }
 
+  let mut transaction = mysql_connection.begin().await?;
+
+  let result = do_with_transaction(
+    **server_environment, 
+    &stripe_config, 
+    webhook_payload, 
+    stripe_event_id.clone(), 
+    stripe_event_created_at, 
+    stripe_event_type.clone(), 
+    stripe_is_production, 
+    &mut transaction
+  ).await?;
+  
+  match result {
+    Ok(()) => {
+      transaction.commit().await?;
+    },
+    Err(err) => {
+      error!("Error handling Stripe webhook event `{}` ({}): {:?}", 
+        stripe_event_type,stripe_event_id, err);
+      transaction.rollback().await?;
+      return Err(err);
+    }
+  }
+
+  Ok(Json(StripeArtcraftWebhookSuccessResponse {
+    success: true,
+  }))
+}
+
+async fn do_with_transaction(
+  server_environment: ServerEnvironment, 
+  stripe_config: &ArtcraftStripeConfigWithClient, 
+  webhook_payload: Event, 
+  stripe_event_id: String, 
+  stripe_event_created_at: NaiveDateTime, 
+  stripe_event_type: String, 
+  stripe_is_production: bool, 
+  mut transaction: 
+  &mut Transaction<MySql>
+) -> Result<(), StripeArtcraftWebhookError> {
+  
   let webhook_summary = handle_webhook_payload(
-    &mut mysql_connection,
+    &mut transaction,
     &stripe_config.client,
     **server_environment,
     webhook_payload,
@@ -153,7 +196,7 @@ pub async fn stripe_artcraft_webhook_handler(
     should_ignore_retry: webhook_summary.should_ignore_retry,
   };
 
-  query.insert(&mysql_pool)
+  query.insert(transaction)
       .await
       .map_err(|err| {
         let reason = format!("Could not insert Stripe webhook event log: {:?}", err);
@@ -161,7 +204,5 @@ pub async fn stripe_artcraft_webhook_handler(
         StripeArtcraftWebhookError::ServerError("database error".to_string())
       })?;
 
-  Ok(Json(StripeArtcraftWebhookSuccessResponse {
-    success: true,
-  }))
+  Ok(())
 }

crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/handle_webhook_payload.rs

@@ -11,16 +11,16 @@ use crate::endpoints::webhook::webhook_event_handlers::stripe_artcraft_webhook_s
 use log::{info, warn};
 use reusable_types::server_environment::ServerEnvironment;
 use sqlx::pool::PoolConnection;
-use sqlx::MySql;
+use sqlx::{MySql, Transaction};
 use stripe::Client;
 use stripe_webhook::{Event, EventObject};
 
 pub async fn handle_webhook_payload(
-  mysql_connection: &mut PoolConnection<MySql>,
+  transaction: &mut Transaction<'_, MySql>,
   stripe_client: &Client,
   server_environment: ServerEnvironment,
   webhook_payload: Event,
-  stripe_event_type: &String
+  stripe_event_type: &str,
 ) -> Result<StripeArtcraftWebhookSummary, StripeArtcraftWebhookError> {
 
   if let Some(summary) = ignore_known_unwanted_events(&webhook_payload) {
@@ -111,7 +111,7 @@ pub async fn handle_webhook_payload(
       webhook_summary = customer_subscription_created_handler(
         &subscription,
         server_environment,
-        mysql_connection,
+        transaction,
       ).await?;
     }
 
@@ -120,7 +120,7 @@ pub async fn handle_webhook_payload(
       webhook_summary = customer_subscription_updated_handler(
         &subscription,
         server_environment,
-        mysql_connection,
+        transaction,
       ).await?;
     }
 
@@ -129,7 +129,7 @@ pub async fn handle_webhook_payload(
       webhook_summary = customer_subscription_deleted_handler(
         &subscription,
         server_environment,
-        mysql_connection,
+        transaction,
       ).await?;
     }
 

crates/service/plugins/billing_artcraft/src/endpoints/webhook/webhook_event_handlers/stripe_artcraft_webhook_error.rs

@@ -35,3 +35,9 @@ impl fmt::Display for StripeArtcraftWebhookError {
     write!(f, "{:?}", self)
   }
 }
+
+impl From<sqlx::Error> for StripeArtcraftWebhookError {
+  fn from(err: sqlx::Error) -> Self {
+    StripeArtcraftWebhookError::ServerError(format!("SQLX Error: {:?}", err))
+  }
+}

crates/service/plugins/billing_artcraft/src/fulfillment/credits_pack/complete_credits_pack_purchase.rs

@@ -0,0 +1,12 @@
+use crate::configs::credits_packs::stripe_artcraft_credits_pack_info::StripeArtcraftCreditsPackInfo;
+
+/// Record the credits pack purchase
+pub async fn complete_credits_pack_purchase(
+  pack: &StripeArtcraftCreditsPackInfo,
+  quantity: u64,
+) -> anyhow::Result<()> {
+
+  // TODO: Math + transaction
+
+  Ok(())
+}

crates/service/plugins/billing_artcraft/src/fulfillment/credits_pack/mod.rs

@@ -0,0 +1 @@
+mod complete_credits_pack_purchase;

crates/service/plugins/billing_artcraft/src/fulfillment/mod.rs

@@ -0,0 +1,2 @@
+mod subscriptions;
+mod credits_pack;

crates/service/plugins/billing_artcraft/src/fulfillment/subscriptions/establish_subscription.rs

@@ -0,0 +1,12 @@
+use crate::configs::credits_packs::stripe_artcraft_credits_pack_info::StripeArtcraftCreditsPackInfo;
+
+/// Record the credits pack purchase
+pub async fn establish_wallet(
+  pack: &StripeArtcraftCreditsPackInfo,
+  quantity: u64,
+) -> anyhow::Result<()> {
+
+  // TODO: Math + transaction
+
+  Ok(())
+}

crates/service/plugins/billing_artcraft/src/fulfillment/subscriptions/mod.rs

@@ -0,0 +1,3 @@
+mod establish_subscription;
+mod mark_wallet_subscription_unpaid;
+mod renew_subscription;

crates/service/plugins/billing_artcraft/src/lib.rs

@@ -1,4 +1,5 @@
 pub mod configs;
 pub mod endpoints;
-pub mod utils;
-mod requests;
+pub mod fulfillment;
+pub mod requests;
+pub mod utils;