Handle subscription update and create

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

@@ -140,17 +140,17 @@ pub async fn stripe_webhook_handler(
 
     EventType::CustomerSubscriptionCreated => {
       if let EventObject::Subscription(subscription) = webhook_payload.data.object {
-        webhook_summary = customer_subscription_created_handler(&subscription)?;
+        webhook_summary = customer_subscription_created_handler(&subscription, &mysql_pool).await?;
       }
     }
     EventType::CustomerSubscriptionUpdated => {
       if let EventObject::Subscription(subscription) = webhook_payload.data.object {
-        webhook_summary = customer_subscription_updated_handler(&subscription)?;
+        webhook_summary = customer_subscription_updated_handler(&subscription, &mysql_pool).await?;
       }
     }
     EventType::CustomerSubscriptionDeleted => {
       if let EventObject::Subscription(subscription) = webhook_payload.data.object {
-        webhook_summary = customer_subscription_deleted_handler(&subscription)?;
+        webhook_summary = customer_subscription_deleted_handler(&subscription, &mysql_pool).await?;
       }
     }
 
@@ -202,7 +202,7 @@ pub async fn stripe_webhook_handler(
 
     EventType::PaymentIntentSucceeded => {
       if let EventObject::PaymentIntent(payment_intent) = webhook_payload.data.object {
-        let _r = payment_intent_succeeded_handler(&payment_intent)?;
+        webhook_summary = payment_intent_succeeded_handler(&payment_intent)?;
       }
     }
 
@@ -219,7 +219,7 @@ pub async fn stripe_webhook_handler(
 
     EventType::ChargeSucceeded => {
       if let EventObject::Charge(charge) = webhook_payload.data.object {
-        let _r = charge_succeeded_handler(&charge)?;
+        webhook_summary = charge_succeeded_handler(&charge)?;
       }
     }
 

crates/component_subsystem/billing/src/stripe/webhook_event_handlers/customer_subscription/customer_subscription_created_handler.rs

@@ -2,23 +2,56 @@ use crate::stripe::webhook_event_handlers::customer_subscription::subscription_e
 use crate::stripe::webhook_event_handlers::stripe_webhook_error::StripeWebhookError;
 use crate::stripe::webhook_event_handlers::stripe_webhook_summary::StripeWebhookSummary;
 use log::{error, warn};
+use sqlx::MySqlPool;
 use stripe::Subscription;
+use database_queries::queries::billing::subscriptions::upsert_subscription_by_stripe_id::UpsertSubscriptionByStripeId;
 
 /// Handle event type: 'customer.subscription.created'
 /// Sent when the subscription is created. The subscription status may be incomplete if customer
 /// authentication is required to complete the payment or if you set payment_behavior to
 /// default_incomplete. For more details, read about subscription payment behavior.
-pub fn customer_subscription_created_handler(subscription: &Subscription) -> Result<StripeWebhookSummary, StripeWebhookError> {
+pub async fn customer_subscription_created_handler(
+  subscription: &Subscription,
+  mysql_pool: &MySqlPool,
+) -> Result<StripeWebhookSummary, StripeWebhookError> {
+
   let summary = subscription_summary_extractor(subscription)
       .map_err(|err| {
         error!("Error extracting subscription from 'customer.subscription.created' payload: {:?}", err);
         StripeWebhookError::ServerError // NB: This was probably *our* fault.
       })?;
 
+  match summary.user_token.as_deref() {
+    None => {
+      error!("Subscription does not have a user token associated with it: {}",
+        &summary.stripe_subscription_id);
+    }
+    Some(user_token) => {
+      let upsert = UpsertSubscriptionByStripeId {
+        stripe_subscription_id: &summary.stripe_subscription_id,
+        user_token: user_token,
+        subscription_category: "todo",
+        subscription_product_key: "todo",
+        maybe_stripe_product_id: Some(&summary.stripe_product_id),
+        maybe_stripe_customer_id: Some(&summary.stripe_customer_id),
+        maybe_stripe_is_production: Some(summary.stripe_is_production),
+        subscription_created_at: summary.subscription_period_start,
+        subscription_expires_at: summary.subscription_period_end,
+      };
+
+      let _r = upsert.upsert(mysql_pool)
+          .await
+          .map_err(|err| {
+            error!("Mysql error: {:?}", err);
+            StripeWebhookError::ServerError
+          })?;
+    }
+  }
+
   Ok(StripeWebhookSummary {
     maybe_user_token: summary.user_token,
     maybe_event_entity_id: Some(summary.stripe_subscription_id),
     maybe_stripe_customer_id: Some(summary.stripe_customer_id),
-    event_was_handled: false,
+    event_was_handled: true,
   })
 }

crates/component_subsystem/billing/src/stripe/webhook_event_handlers/customer_subscription/customer_subscription_deleted_handler.rs

@@ -2,11 +2,15 @@ use crate::stripe::webhook_event_handlers::customer_subscription::subscription_e
 use crate::stripe::webhook_event_handlers::stripe_webhook_error::StripeWebhookError;
 use crate::stripe::webhook_event_handlers::stripe_webhook_summary::StripeWebhookSummary;
 use log::{error, warn};
+use sqlx::MySqlPool;
 use stripe::Subscription;
 
 /// Handle event type: 'customer.subscription.deleted'
 /// Sent when a customer’s subscription ends.
-pub fn customer_subscription_deleted_handler(subscription: &Subscription) -> Result<StripeWebhookSummary, StripeWebhookError> {
+pub async fn customer_subscription_deleted_handler(
+  subscription: &Subscription,
+  mysql_pool: &MySqlPool,
+) -> Result<StripeWebhookSummary, StripeWebhookError> {
   let summary = subscription_summary_extractor(subscription)
       .map_err(|err| {
         error!("Error extracting subscription from 'customer.subscription.deleted' payload: {:?}", err);

crates/component_subsystem/billing/src/stripe/webhook_event_handlers/customer_subscription/customer_subscription_updated_handler.rs

@@ -1,21 +1,54 @@
 use crate::stripe::webhook_event_handlers::customer_subscription::subscription_event_extractor::subscription_summary_extractor;
 use crate::stripe::webhook_event_handlers::stripe_webhook_error::StripeWebhookError;
 use crate::stripe::webhook_event_handlers::stripe_webhook_summary::StripeWebhookSummary;
+use database_queries::queries::billing::subscriptions::upsert_subscription_by_stripe_id::UpsertSubscriptionByStripeId;
 use log::{error, warn};
+use sqlx::MySqlPool;
 use stripe::Subscription;
 
 /// Handle event type: 'customer.subscription.updated'
-pub fn customer_subscription_updated_handler(subscription: &Subscription) -> Result<StripeWebhookSummary, StripeWebhookError> {
+pub async fn customer_subscription_updated_handler(
+  subscription: &Subscription,
+  mysql_pool: &MySqlPool,
+) -> Result<StripeWebhookSummary, StripeWebhookError> {
+
   let summary = subscription_summary_extractor(subscription)
       .map_err(|err| {
         error!("Error extracting subscription from 'customer.subscription.updated' payload: {:?}", err);
         StripeWebhookError::ServerError // NB: This was probably *our* fault.
       })?;
 
+  match summary.user_token.as_deref() {
+    None => {
+      error!("Subscription does not have a user token associated with it: {}",
+        &summary.stripe_subscription_id);
+    }
+    Some(user_token) => {
+      let upsert = UpsertSubscriptionByStripeId {
+        stripe_subscription_id: &summary.stripe_subscription_id,
+        user_token: user_token,
+        subscription_category: "todo",
+        subscription_product_key: "todo",
+        maybe_stripe_product_id: Some(&summary.stripe_product_id),
+        maybe_stripe_customer_id: Some(&summary.stripe_customer_id),
+        maybe_stripe_is_production: Some(summary.stripe_is_production),
+        subscription_created_at: summary.subscription_period_start,
+        subscription_expires_at: summary.subscription_period_end,
+      };
+
+      let _r = upsert.upsert(mysql_pool)
+          .await
+          .map_err(|err| {
+            error!("Mysql error: {:?}", err);
+            StripeWebhookError::ServerError
+          })?;
+    }
+  }
+
   Ok(StripeWebhookSummary {
     maybe_user_token: summary.user_token,
     maybe_event_entity_id: Some(summary.stripe_subscription_id),
     maybe_stripe_customer_id: Some(summary.stripe_customer_id),
-    event_was_handled: false,
+    event_was_handled: true,
   })
 }

crates/component_subsystem/billing/src/stripe/webhook_event_handlers/payment_intent/payment_intent_succeeded_handler.rs

@@ -2,11 +2,13 @@ use container_common::anyhow_result::AnyhowResult;
 use crate::stripe::helpers::common_metadata_keys::METADATA_USER_TOKEN;
 use crate::stripe::helpers::expand_customer_id::expand_customer_id;
 use crate::stripe::webhook_event_handlers::stripe_webhook_error::StripeWebhookError;
+use crate::stripe::webhook_event_handlers::stripe_webhook_summary::StripeWebhookSummary;
 use log::warn;
 use stripe::{Invoice, PaymentIntent};
 
 // Handle event type: 'payment_intent.succeeded'
-pub fn payment_intent_succeeded_handler(payment_intent: &PaymentIntent) -> Result<(), StripeWebhookError> {
+pub fn payment_intent_succeeded_handler(payment_intent: &PaymentIntent) -> Result<StripeWebhookSummary, StripeWebhookError> {
+  let payment_intent_id = payment_intent.id.to_string();
 
   let payment_intent_status = payment_intent.status;
 
@@ -20,8 +22,10 @@ pub fn payment_intent_succeeded_handler(payment_intent: &PaymentIntent) -> Resul
   let maybe_user_token = payment_intent.metadata.get(METADATA_USER_TOKEN)
       .map(|t| t.to_string());
 
-  warn!(">>> payment_intent.succeeded: {:?}, {:?}, {:?}",
-    payment_intent_status, maybe_stripe_customer_id, maybe_user_token);
-
-  Ok(())
+  Ok(StripeWebhookSummary {
+    maybe_user_token: maybe_user_token,
+    maybe_event_entity_id: Some(payment_intent_id),
+    maybe_stripe_customer_id: maybe_stripe_customer_id,
+    event_was_handled: false,
+  })
 }

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

@@ -1 +1,2 @@
-pub mod stripe;
+pub mod stripe;
+pub mod subscriptions;

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

@@ -0,0 +1 @@
+pub mod upsert_subscription_by_stripe_id;

crates/lib/database_queries/src/queries/billing/subscriptions/upsert_subscription_by_stripe_id.rs

@@ -0,0 +1,96 @@
+use anyhow::anyhow;
+use chrono::NaiveDateTime;
+use container_common::anyhow_result::AnyhowResult;
+use sqlx::MySqlPool;
+use crate::tokens::Tokens;
+
+// TODO: Make a trait with default impls to handle common query concerns.
+
+pub struct UpsertSubscriptionByStripeId <'a> {
+  /// Stripe's assigned ID for the subscription
+  /// This acts as an externally-provided unique key for records in this table.
+  pub stripe_subscription_id: &'a str,
+
+  /// Internal user token
+  pub user_token: &'a str,
+
+  /// The platform key, eg. "fakeyou", "storyteller_stream", "symphonia", etc.
+  pub subscription_category: &'a str,
+
+  /// The name of the product the user is subscribing to within the category.
+  pub subscription_product_key: &'a str,
+
+  pub maybe_stripe_product_id: Option<&'a str>,
+  pub maybe_stripe_customer_id: Option<&'a str>,
+  pub maybe_stripe_is_production: Option<bool>,
+
+  pub subscription_created_at: NaiveDateTime,
+
+  /// When the subscription is set to expire.
+  /// This controls whether it is active or not.
+  pub subscription_expires_at: NaiveDateTime,
+}
+
+impl <'a> UpsertSubscriptionByStripeId <'a> {
+
+  pub async fn upsert(&'a self, mysql_pool: &MySqlPool) -> AnyhowResult<()> {
+    let token = Tokens::new_subscription_token()?;
+
+    // NB: The following behaviors are intentional
+    //  - We only set the token initially; changing it obviously breaks.
+    //  - We only set the user token the first time. Stripe entities may remain coherent,
+    //    but it isn't our business to attempt changes on webhooks.
+    //  - Other "static" fields do not need to change on update, either.
+    let query = sqlx::query!(
+        r#"
+INSERT INTO user_subscriptions
+SET
+  token = ?,
+  user_token = ?,
+  subscription_category = ?,
+  subscription_product_key = ?,
+
+  maybe_stripe_subscription_id = ?,
+
+  maybe_stripe_product_id = ?,
+  maybe_stripe_customer_id = ?,
+  maybe_stripe_is_production = ?,
+
+  subscription_created_at = ?,
+  subscription_expires_at = ?,
+
+  version = version + 1
+
+ON DUPLICATE KEY UPDATE
+  subscription_expires_at = ?,
+
+  version = version + 1
+        "#,
+      token,
+      self.user_token,
+      self.subscription_category,
+      self.subscription_product_key,
+
+      self.stripe_subscription_id,
+
+      self.maybe_stripe_product_id,
+      self.maybe_stripe_customer_id,
+      self.maybe_stripe_is_production,
+
+      self.subscription_created_at,
+      self.subscription_expires_at,
+
+      // Upsert
+      self.subscription_expires_at,
+    );
+
+    let query_result = query.execute(mysql_pool).await;
+
+    let _record_id = match query_result {
+      Ok(res) => res.last_insert_id(),
+      Err(err) => return Err(anyhow!("Error upserting subscription record: {:?}", err)),
+    };
+
+    Ok(())
+  }
+}

crates/lib/database_queries/src/tokens.rs

@@ -92,4 +92,10 @@ impl Tokens {
   pub fn new_twitch_event_rule() -> AnyhowResult<String> {
     random_prefix_crockford_token("TER:", 32, false)
   }
+
+  // ========== [Billing] Tokens not typically visible to users ==========
+
+  pub fn new_subscription_token() -> AnyhowResult<String> {
+    random_prefix_crockford_token("SUB:", 32, false)
+  }
 }

crates/service/storyteller_web/src/main.rs

@@ -23,6 +23,7 @@ pub mod configs;
 pub mod http_server;
 pub mod routes;
 pub mod server_state;
+pub mod subscriptions;
 pub mod threads;
 pub mod util;
 pub mod validations;

crates/service/storyteller_web/src/subscriptions/mod.rs

@@ -0,0 +1,2 @@
+pub mod subscription_plan;
+pub mod subscription_plan_list;

crates/service/storyteller_web/src/subscriptions/subscription_plan.rs

@@ -0,0 +1,33 @@
+
+/**
+
+ User has:
+
+   fakeyou, pro
+   fakeyou, pro
+   stream, basic
+
+*/
+
+#[derive(Clone)]
+pub struct SubscriptionPlan {
+  /// Slug is the unique identifier for the subscription
+  /// It must be human-readability friendly, consist of only lowercase alphanumerics,
+  /// underscores, and hyphens, and be 32 characters or fewer.
+  pub slug: String,
+}
+
+impl SubscriptionPlan {
+
+  pub fn base_tts_priority_level(&self) -> u32 {
+    0
+  }
+
+  pub fn base_api_tts_priority_level(&self) -> Option<u32> {
+    Some(0)
+  }
+
+  pub fn has_feature_foo(&self) -> bool {
+    false
+  }
+}