feat(plus): user subscription transitions (#415)

Record user subscription changes into pg table.

Cargo.toml

@@ -27,11 +27,20 @@ actix-session = { version = "0.8", features = ["cookie-session"] }
 actix-web-httpauth = "0.8"
 actix-web-lab = "0.19"
 
-diesel = { version = "2", features = ["postgres", "uuid", "r2d2", "chrono", "serde_json"] }
+diesel = { version = "2", features = [
+    "postgres",
+    "uuid",
+    "r2d2",
+    "chrono",
+    "serde_json",
+] }
 diesel_migrations = "2"
 diesel-derive-enum = { version = "2", features = ["postgres"] }
 pgvector = { version = "0.2", features = ["sqlx"] }
-sqlx = { version = "0.7", features = [ "runtime-tokio-rustls", "postgres"], default-features = false }
+sqlx = { version = "0.7", features = [
+    "runtime-tokio-rustls",
+    "postgres",
+], default-features = false }
 
 elasticsearch = "7.17.7-alpha.1"
 harsh = "0.2"
@@ -43,15 +52,19 @@ jsonwebtoken = "8"
 
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-serde_with = { version = "3", features = ["base64"]}
+serde_with = { version = "3", features = ["base64"] }
 serde_urlencoded = "0.7"
 form_urlencoded = "1"
 serde_path_to_error = "0.1"
 percent-encoding = "2"
 
 config = "0.13"
 hostname = "0.3"
-slog = { version = "2", features = ["max_level_trace", "release_max_level_info", "dynamic-keys"] }
+slog = { version = "2", features = [
+    "max_level_trace",
+    "release_max_level_info",
+    "dynamic-keys",
+] }
 slog-async = "2"
 slog-envlogger = "2"
 slog-mozlog-json = "0.1"
@@ -90,4 +103,3 @@ sha2 = "0.10"
 stubr = "0.6"
 stubr-attributes = "0.6"
 assert-json-diff = "2"
-tokio = { version = "1", features = ["io-util"] }

migrations/2024-02-14-165616_user_subscription_transitions/down.sql

@@ -0,0 +1 @@
+DROP TABLE user_subscription_transitions;

migrations/2024-02-14-165616_user_subscription_transitions/up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE user_subscription_transitions (
+    id BIGSERIAL PRIMARY KEY,
+    user_id BIGSERIAL REFERENCES users (id) ON DELETE CASCADE,
+    old_subscription_type subscription_type NOT NULL,
+    new_subscription_type subscription_type NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT now()
+);

src/db/fxa_webhook.rs

@@ -18,6 +18,7 @@ use diesel::prelude::*;
 use diesel::ExpressionMethods;
 use serde_json::json;
 
+use super::model::SubscriptionChangeInsert;
 use super::types::{FxaEventStatus, Subscription};
 
 pub fn log_failed_webhook_event(
@@ -191,7 +192,7 @@ pub async fn update_subscription_state_from_webhook(
             };
             if subscription == Subscription::Core {
                 // drop permissions
-                if let Some(basket) = &**basket {
+                if let Some(basket) = basket.get_ref() {
                     if let Err(e) = newsletter::unsubscribe(&mut conn, &user, basket).await {
                         error!("error unsubscribing user: {}", e);
                     }
@@ -217,7 +218,7 @@ pub async fn update_subscription_state_from_webhook(
                     )
                     .set(schema::webhook_events::status.eq(FxaEventStatus::Processed))
                     .execute(&mut conn)?;
-                    return Ok(());
+                    // return Ok(());
                 }
                 Err(e) => {
                     diesel::update(
@@ -228,6 +229,21 @@ pub async fn update_subscription_state_from_webhook(
                     return Err(e.into());
                 }
             }
+            // Record the subscription state change in its table.
+            let old_subscription = user.get_subscription_type();
+            if let Some(old_subscription) = old_subscription {
+                // We have the user id, the old and new subscription, store it.
+                let subscription_change = SubscriptionChangeInsert {
+                    user_id: user.id,
+                    old_subscription_type: old_subscription,
+                    new_subscription_type: subscription,
+                    created_at: update.change_time.naive_utc(),
+                };
+                insert_into(schema::user_subscription_transitions::table)
+                    .values(subscription_change)
+                    .execute(&mut conn)?;
+            }
+            return Ok(());
         }
     }
 

src/db/model.rs

@@ -294,3 +294,22 @@ pub struct AIHelpHistoryMessage {
     pub request: Value,
     pub response: Value,
 }
+
+#[derive(Insertable)]
+#[diesel(table_name = user_subscription_transitions)]
+pub struct SubscriptionChangeInsert {
+    pub user_id: i64,
+    pub old_subscription_type: Subscription,
+    pub new_subscription_type: Subscription,
+    pub created_at: NaiveDateTime,
+}
+
+#[derive(Queryable, Debug)]
+#[diesel(table_name = user_subscription_transitions)]
+pub struct SubscriptionChangeQuery {
+    pub id: i64,
+    pub user_id: i64,
+    pub old_subscription_type: Subscription,
+    pub new_subscription_type: Subscription,
+    pub created_at: NaiveDateTime,
+}

src/db/schema.rs

@@ -255,6 +255,20 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    use diesel::sql_types::*;
+    use crate::db::types::*;
+    use super::sql_types::SubscriptionType;
+
+    user_subscription_transitions (id) {
+        id -> Int8,
+        user_id -> Int8,
+        old_subscription_type -> SubscriptionType,
+        new_subscription_type -> SubscriptionType,
+        created_at -> Timestamp,
+    }
+}
+
 diesel::table! {
     use diesel::sql_types::*;
     use crate::db::types::*;
@@ -307,6 +321,7 @@ diesel::joinable!(collection_items -> users (user_id));
 diesel::joinable!(multiple_collections -> users (user_id));
 diesel::joinable!(playground -> users (user_id));
 diesel::joinable!(settings -> users (user_id));
+diesel::joinable!(user_subscription_transitions -> users (user_id));
 
 diesel::allow_tables_to_appear_in_same_query!(
     activity_pings,
@@ -324,6 +339,7 @@ diesel::allow_tables_to_appear_in_same_query!(
     playground,
     raw_webhook_events_tokens,
     settings,
+    user_subscription_transitions,
     users,
     webhook_events,
 );