From d99da56df0b75100d14524fecc100dc11fd06b20 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Mon, 29 Jun 2026 21:21:26 -0400 Subject: [PATCH] payment informatics #2 (#1705) --- ...8c3ce3f5e6ed56d358894cbc496d345566519.json | 58 +++ ...5a22786adde9968dd70490c91100cf21f8023.json | 113 ++++++ ...0a5bcdabdb3316aa407c5ca883bca863ee633.json | 12 + ...763b5b067447d295b002f6e766209ea077f42.json | 25 ++ ...2962b6da773460505222a75408dab1d26b70f.json | 25 ++ ...32a6db63e406cff009325ce92dbdd0932809c.json | 12 + ...12cb8ae152afefd2574385c120c63278893d2.json | 47 +++ Cargo.lock | 30 ++ Cargo.toml | 1 + .../config/rootly_config/src/services.rs | 2 + .../database/mysql_queries/src/queries/mod.rs | 2 + .../aggregate_daily_spends_for_date.rs | 76 ++++ .../src/queries/user_daily_spends/mod.rs | 2 + .../upsert_user_daily_spend.rs | 73 ++++ ...payments_namespaces_with_spend_activity.rs | 34 ++ .../list_user_spend_events_for_user.rs | 55 +++ .../list_user_tokens_with_spend_activity.rs | 46 +++ .../src/queries/user_spend_summaries/mod.rs | 4 + .../upsert_user_spend_summary.rs | 129 +++++++ ...get_active_subscription_status_for_user.rs | 55 +++ .../queries/users/user_subscriptions/mod.rs | 1 + .../job/user_spend_analytics_job/Cargo.toml | 47 +++ .../user-spend-analytics-job.development.env | 22 ++ .../endpoints/health_check_handler.rs | 127 ++++++ .../src/http_server/endpoints/mod.rs | 1 + .../http_server/http_server_shared_state.rs | 10 + .../src/http_server/mod.rs | 3 + .../src/http_server/run_http_server.rs | 76 ++++ .../src/job/alert_on_error.rs | 17 + .../src/job/backfill_daily_spends.rs | 130 +++++++ .../src/job/backfill_user_summaries.rs | 363 ++++++++++++++++++ .../src/job/main_loop.rs | 38 ++ .../user_spend_analytics_job/src/job/mod.rs | 5 + .../src/job/reengagement_score.rs | 86 +++++ .../src/job_dependencies.rs | 45 +++ .../job/user_spend_analytics_job/src/main.rs | 184 +++++++++ .../src/startup/build_pager.rs | 65 ++++ .../src/startup/mod.rs | 1 + 38 files changed, 2022 insertions(+) create mode 100644 .sqlx/query-1c997019b7ed18059bc08d857918c3ce3f5e6ed56d358894cbc496d345566519.json create mode 100644 .sqlx/query-83fc05a3594b379482745f851cd5a22786adde9968dd70490c91100cf21f8023.json create mode 100644 .sqlx/query-b7a5fea76da70c03b8744c985940a5bcdabdb3316aa407c5ca883bca863ee633.json create mode 100644 .sqlx/query-bcd07050d5557bf59d6b2691478763b5b067447d295b002f6e766209ea077f42.json create mode 100644 .sqlx/query-cf8c702126bba73cd3abc2ed7202962b6da773460505222a75408dab1d26b70f.json create mode 100644 .sqlx/query-f4df0d23d95aded2f55d464344b32a6db63e406cff009325ce92dbdd0932809c.json create mode 100644 .sqlx/query-f74615faa4321ec4d583829997e12cb8ae152afefd2574385c120c63278893d2.json create mode 100644 crates/schema/database/mysql_queries/src/queries/user_daily_spends/aggregate_daily_spends_for_date.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_daily_spends/mod.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_daily_spends/upsert_user_daily_spend.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_payments_namespaces_with_spend_activity.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_spend_events_for_user.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_tokens_with_spend_activity.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_spend_summaries/mod.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/user_spend_summaries/upsert_user_spend_summary.rs create mode 100644 crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_active_subscription_status_for_user.rs create mode 100644 crates/service/job/user_spend_analytics_job/Cargo.toml create mode 100644 crates/service/job/user_spend_analytics_job/config/user-spend-analytics-job.development.env create mode 100644 crates/service/job/user_spend_analytics_job/src/http_server/endpoints/health_check_handler.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/http_server/endpoints/mod.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/http_server/http_server_shared_state.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/http_server/mod.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/http_server/run_http_server.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job/alert_on_error.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job/backfill_daily_spends.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job/backfill_user_summaries.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job/main_loop.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job/mod.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job/reengagement_score.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/job_dependencies.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/main.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/startup/build_pager.rs create mode 100644 crates/service/job/user_spend_analytics_job/src/startup/mod.rs diff --git a/.sqlx/query-1c997019b7ed18059bc08d857918c3ce3f5e6ed56d358894cbc496d345566519.json b/.sqlx/query-1c997019b7ed18059bc08d857918c3ce3f5e6ed56d358894cbc496d345566519.json new file mode 100644 index 0000000000..aaa1f470c4 --- /dev/null +++ b/.sqlx/query-1c997019b7ed18059bc08d857918c3ce3f5e6ed56d358894cbc496d345566519.json @@ -0,0 +1,58 @@ +{ + "db_name": "MySQL", + "query": "\nSELECT\n payment_occurred_at AS `payment_occurred_at!: DateTime`,\n amount_usd_cents,\n event_type AS `event_type!: PaymentEventType`,\n maybe_credits_granted AS `maybe_credits_granted: u32`\nFROM user_spend_events\nWHERE maybe_user_token = ?\n AND payments_namespace = ?\n AND is_production = TRUE\nORDER BY payment_occurred_at ASC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "payment_occurred_at!: DateTime", + "type_info": { + "type": "Timestamp", + "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE", + "char_set": 63, + "max_size": 19 + } + }, + { + "ordinal": 1, + "name": "amount_usd_cents", + "type_info": { + "type": "LongLong", + "flags": "NOT_NULL | NO_DEFAULT_VALUE", + "char_set": 63, + "max_size": 20 + } + }, + { + "ordinal": 2, + "name": "event_type!: PaymentEventType", + "type_info": { + "type": "VarString", + "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE", + "char_set": 224, + "max_size": 128 + } + }, + { + "ordinal": 3, + "name": "maybe_credits_granted: u32", + "type_info": { + "type": "Long", + "flags": "UNSIGNED", + "char_set": 63, + "max_size": 10 + } + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + false, + false, + false, + true + ] + }, + "hash": "1c997019b7ed18059bc08d857918c3ce3f5e6ed56d358894cbc496d345566519" +} diff --git a/.sqlx/query-83fc05a3594b379482745f851cd5a22786adde9968dd70490c91100cf21f8023.json b/.sqlx/query-83fc05a3594b379482745f851cd5a22786adde9968dd70490c91100cf21f8023.json new file mode 100644 index 0000000000..66560c7c92 --- /dev/null +++ b/.sqlx/query-83fc05a3594b379482745f851cd5a22786adde9968dd70490c91100cf21f8023.json @@ -0,0 +1,113 @@ +{ + "db_name": "MySQL", + "query": "\nSELECT\n maybe_user_token AS `user_token!: UserToken`,\n payments_namespace AS `payments_namespace!: PaymentsNamespace`,\n CAST(SUM(CASE WHEN amount_usd_cents > 0\n AND event_type IN ('subscription_initial', 'subscription_renewal', 'subscription_proration_upgrade')\n THEN amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `subscription_spend_usd_cents!: u64`,\n CAST(SUM(CASE WHEN amount_usd_cents > 0 AND event_type = 'credit_pack_purchase'\n THEN amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `credits_spend_usd_cents!: u64`,\n CAST(SUM(CASE WHEN amount_usd_cents > 0 THEN amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `gross_spend_usd_cents!: u64`,\n CAST(SUM(CASE WHEN amount_usd_cents < 0 THEN -amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `refund_usd_cents!: u64`,\n CAST(SUM(amount_usd_cents) AS SIGNED) AS `net_spend_usd_cents!: i64`,\n CAST(SUM(CASE WHEN amount_usd_cents > 0 THEN 1 ELSE 0 END) AS UNSIGNED) AS `payment_count!: u64`,\n CAST(COALESCE(SUM(maybe_credits_granted), 0) AS UNSIGNED) AS `credits_granted!: u64`\nFROM user_spend_events\nWHERE maybe_user_token IS NOT NULL\n AND is_production = TRUE\n AND payment_occurred_at >= ?\n AND payment_occurred_at < ?\nGROUP BY maybe_user_token, payments_namespace\nHAVING SUM(CASE WHEN amount_usd_cents <> 0 THEN 1 ELSE 0 END) > 0\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "user_token!: UserToken", + "type_info": { + "type": "VarString", + "flags": "MULTIPLE_KEY | BINARY", + "char_set": 224, + "max_size": 128 + } + }, + { + "ordinal": 1, + "name": "payments_namespace!: PaymentsNamespace", + "type_info": { + "type": "VarString", + "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE", + "char_set": 224, + "max_size": 128 + } + }, + { + "ordinal": 2, + "name": "subscription_spend_usd_cents!: u64", + "type_info": { + "type": "LongLong", + "flags": "UNSIGNED | BINARY", + "char_set": 63, + "max_size": 21 + } + }, + { + "ordinal": 3, + "name": "credits_spend_usd_cents!: u64", + "type_info": { + "type": "LongLong", + "flags": "UNSIGNED | BINARY", + "char_set": 63, + "max_size": 21 + } + }, + { + "ordinal": 4, + "name": "gross_spend_usd_cents!: u64", + "type_info": { + "type": "LongLong", + "flags": "UNSIGNED | BINARY", + "char_set": 63, + "max_size": 21 + } + }, + { + "ordinal": 5, + "name": "refund_usd_cents!: u64", + "type_info": { + "type": "LongLong", + "flags": "UNSIGNED | BINARY", + "char_set": 63, + "max_size": 21 + } + }, + { + "ordinal": 6, + "name": "net_spend_usd_cents!: i64", + "type_info": { + "type": "LongLong", + "flags": "BINARY", + "char_set": 63, + "max_size": 21 + } + }, + { + "ordinal": 7, + "name": "payment_count!: u64", + "type_info": { + "type": "LongLong", + "flags": "UNSIGNED | BINARY", + "char_set": 63, + "max_size": 21 + } + }, + { + "ordinal": 8, + "name": "credits_granted!: u64", + "type_info": { + "type": "LongLong", + "flags": "NOT_NULL | UNSIGNED | BINARY", + "char_set": 63, + "max_size": 21 + } + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + true, + false, + true, + true, + true, + true, + true, + true, + false + ] + }, + "hash": "83fc05a3594b379482745f851cd5a22786adde9968dd70490c91100cf21f8023" +} diff --git a/.sqlx/query-b7a5fea76da70c03b8744c985940a5bcdabdb3316aa407c5ca883bca863ee633.json b/.sqlx/query-b7a5fea76da70c03b8744c985940a5bcdabdb3316aa407c5ca883bca863ee633.json new file mode 100644 index 0000000000..a7c563ccdc --- /dev/null +++ b/.sqlx/query-b7a5fea76da70c03b8744c985940a5bcdabdb3316aa407c5ca883bca863ee633.json @@ -0,0 +1,12 @@ +{ + "db_name": "MySQL", + "query": "\nINSERT INTO user_daily_spends\nSET\n user_token = ?,\n payments_namespace = ?,\n spend_date = ?,\n subscription_spend_usd_cents = ?,\n credits_spend_usd_cents = ?,\n gross_spend_usd_cents = ?,\n refund_usd_cents = ?,\n net_spend_usd_cents = ?,\n payment_count = ?,\n credits_granted = ?\nON DUPLICATE KEY UPDATE\n subscription_spend_usd_cents = VALUES(subscription_spend_usd_cents),\n credits_spend_usd_cents = VALUES(credits_spend_usd_cents),\n gross_spend_usd_cents = VALUES(gross_spend_usd_cents),\n refund_usd_cents = VALUES(refund_usd_cents),\n net_spend_usd_cents = VALUES(net_spend_usd_cents),\n payment_count = VALUES(payment_count),\n credits_granted = VALUES(credits_granted)\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 10 + }, + "nullable": [] + }, + "hash": "b7a5fea76da70c03b8744c985940a5bcdabdb3316aa407c5ca883bca863ee633" +} diff --git a/.sqlx/query-bcd07050d5557bf59d6b2691478763b5b067447d295b002f6e766209ea077f42.json b/.sqlx/query-bcd07050d5557bf59d6b2691478763b5b067447d295b002f6e766209ea077f42.json new file mode 100644 index 0000000000..bf920a5baf --- /dev/null +++ b/.sqlx/query-bcd07050d5557bf59d6b2691478763b5b067447d295b002f6e766209ea077f42.json @@ -0,0 +1,25 @@ +{ + "db_name": "MySQL", + "query": "\nSELECT DISTINCT maybe_user_token AS `user_token!: UserToken`\nFROM user_spend_events\nWHERE maybe_user_token IS NOT NULL\n AND is_production = TRUE\n AND payments_namespace = ?\n AND maybe_user_token > ?\nORDER BY maybe_user_token\nLIMIT ?\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "user_token!: UserToken", + "type_info": { + "type": "VarString", + "flags": "MULTIPLE_KEY | BINARY", + "char_set": 224, + "max_size": 128 + } + } + ], + "parameters": { + "Right": 3 + }, + "nullable": [ + true + ] + }, + "hash": "bcd07050d5557bf59d6b2691478763b5b067447d295b002f6e766209ea077f42" +} diff --git a/.sqlx/query-cf8c702126bba73cd3abc2ed7202962b6da773460505222a75408dab1d26b70f.json b/.sqlx/query-cf8c702126bba73cd3abc2ed7202962b6da773460505222a75408dab1d26b70f.json new file mode 100644 index 0000000000..1680af5667 --- /dev/null +++ b/.sqlx/query-cf8c702126bba73cd3abc2ed7202962b6da773460505222a75408dab1d26b70f.json @@ -0,0 +1,25 @@ +{ + "db_name": "MySQL", + "query": "\nSELECT DISTINCT payments_namespace AS `payments_namespace!: PaymentsNamespace`\nFROM user_spend_events\nWHERE maybe_user_token IS NOT NULL\n AND is_production = TRUE\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "payments_namespace!: PaymentsNamespace", + "type_info": { + "type": "VarString", + "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | NO_DEFAULT_VALUE", + "char_set": 224, + "max_size": 128 + } + } + ], + "parameters": { + "Right": 0 + }, + "nullable": [ + false + ] + }, + "hash": "cf8c702126bba73cd3abc2ed7202962b6da773460505222a75408dab1d26b70f" +} diff --git a/.sqlx/query-f4df0d23d95aded2f55d464344b32a6db63e406cff009325ce92dbdd0932809c.json b/.sqlx/query-f4df0d23d95aded2f55d464344b32a6db63e406cff009325ce92dbdd0932809c.json new file mode 100644 index 0000000000..e954d44dd6 --- /dev/null +++ b/.sqlx/query-f4df0d23d95aded2f55d464344b32a6db63e406cff009325ce92dbdd0932809c.json @@ -0,0 +1,12 @@ +{ + "db_name": "MySQL", + "query": "\nINSERT INTO user_spend_summaries\nSET\n payments_namespace = ?,\n user_token = ?,\n lifetime_gross_spend_usd_cents = ?,\n lifetime_subscription_spend_usd_cents = ?,\n lifetime_credits_spend_usd_cents = ?,\n lifetime_refund_usd_cents = ?,\n lifetime_net_spend_usd_cents = ?,\n lifetime_payment_count = ?,\n lifetime_refund_count = ?,\n first_payment_at = ?,\n first_spend_usd_cents = ?,\n last_payment_at = ?,\n last_spend_usd_cents = ?,\n days_since_first_payment = ?,\n days_since_last_payment = ?,\n net_spend_this_year_usd_cents = ?,\n consecutive_active_weeks = ?,\n consecutive_inactive_weeks = ?,\n weeks_since_last_spend = ?,\n is_active_subscriber = ?,\n maybe_subscription_interval = ?,\n maybe_reengagement_score = ?\nON DUPLICATE KEY UPDATE\n lifetime_gross_spend_usd_cents = VALUES(lifetime_gross_spend_usd_cents),\n lifetime_subscription_spend_usd_cents = VALUES(lifetime_subscription_spend_usd_cents),\n lifetime_credits_spend_usd_cents = VALUES(lifetime_credits_spend_usd_cents),\n lifetime_refund_usd_cents = VALUES(lifetime_refund_usd_cents),\n lifetime_net_spend_usd_cents = VALUES(lifetime_net_spend_usd_cents),\n lifetime_payment_count = VALUES(lifetime_payment_count),\n lifetime_refund_count = VALUES(lifetime_refund_count),\n first_payment_at = VALUES(first_payment_at),\n first_spend_usd_cents = VALUES(first_spend_usd_cents),\n last_payment_at = VALUES(last_payment_at),\n last_spend_usd_cents = VALUES(last_spend_usd_cents),\n days_since_first_payment = VALUES(days_since_first_payment),\n days_since_last_payment = VALUES(days_since_last_payment),\n net_spend_this_year_usd_cents = VALUES(net_spend_this_year_usd_cents),\n consecutive_active_weeks = VALUES(consecutive_active_weeks),\n consecutive_inactive_weeks = VALUES(consecutive_inactive_weeks),\n weeks_since_last_spend = VALUES(weeks_since_last_spend),\n is_active_subscriber = VALUES(is_active_subscriber),\n maybe_subscription_interval = VALUES(maybe_subscription_interval),\n maybe_reengagement_score = VALUES(maybe_reengagement_score)\n ", + "describe": { + "columns": [], + "parameters": { + "Right": 22 + }, + "nullable": [] + }, + "hash": "f4df0d23d95aded2f55d464344b32a6db63e406cff009325ce92dbdd0932809c" +} diff --git a/.sqlx/query-f74615faa4321ec4d583829997e12cb8ae152afefd2574385c120c63278893d2.json b/.sqlx/query-f74615faa4321ec4d583829997e12cb8ae152afefd2574385c120c63278893d2.json new file mode 100644 index 0000000000..1c8f600e7d --- /dev/null +++ b/.sqlx/query-f74615faa4321ec4d583829997e12cb8ae152afefd2574385c120c63278893d2.json @@ -0,0 +1,47 @@ +{ + "db_name": "MySQL", + "query": "\nSELECT\n maybe_stripe_subscription_status,\n maybe_stripe_recurring_interval,\n subscription_expires_at AS `subscription_expires_at!: DateTime`\nFROM user_subscriptions\nWHERE user_token = ?\n AND subscription_namespace = ?\nORDER BY subscription_expires_at DESC\nLIMIT 1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "maybe_stripe_subscription_status", + "type_info": { + "type": "VarString", + "flags": "MULTIPLE_KEY | BINARY", + "char_set": 224, + "max_size": 128 + } + }, + { + "ordinal": 1, + "name": "maybe_stripe_recurring_interval", + "type_info": { + "type": "VarString", + "flags": "BINARY", + "char_set": 224, + "max_size": 128 + } + }, + { + "ordinal": 2, + "name": "subscription_expires_at!: DateTime", + "type_info": { + "type": "Timestamp", + "flags": "NOT_NULL | MULTIPLE_KEY | BINARY | TIMESTAMP", + "char_set": 63, + "max_size": 19 + } + } + ], + "parameters": { + "Right": 2 + }, + "nullable": [ + true, + true, + false + ] + }, + "hash": "f74615faa4321ec4d583829997e12cb8ae152afefd2574385c120c63278893d2" +} diff --git a/Cargo.lock b/Cargo.lock index b7c7eb5289..9ac3bb827b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13704,6 +13704,36 @@ dependencies = [ "url", ] +[[package]] +name = "user-spend-analytics-job" +version = "0.0.1" +dependencies = [ + "actix-web", + "actix_helpers", + "anyhow", + "bootstrap", + "chrono", + "concurrency", + "easyenv", + "enums", + "errors", + "jobs_common", + "log", + "mysql_queries", + "pager", + "rootly_client", + "rootly_config", + "serde", + "serde_derive", + "serde_json", + "server_environment", + "shared_env_var_config", + "sqlx", + "tokens", + "tokio", + "workspace-build-acceleration", +] + [[package]] name = "user_input_common" version = "0.0.1" diff --git a/Cargo.toml b/Cargo.toml index e3d5975b8a..63b990d369 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -136,6 +136,7 @@ members = [ "crates/service/job/grok_api_job", "crates/service/job/seedance2_pro_job", "crates/service/job/video_thumbnail_job", + "crates/service/job/user_spend_analytics_job", "crates/service/job/worldlabs_job", "crates/service/web/dummy_service", "crates/service/web/storyteller_web", diff --git a/crates/schema/config/rootly_config/src/services.rs b/crates/schema/config/rootly_config/src/services.rs index 44eebb5861..adf0eb82fd 100644 --- a/crates/schema/config/rootly_config/src/services.rs +++ b/crates/schema/config/rootly_config/src/services.rs @@ -1,3 +1,5 @@ pub const ROOTLY_SERVICE_ID_STORYTELLER_WEB: &str = "9ebebb09-0c56-4c8a-8f57-d5f0f85f3f16"; pub const ROOTLY_SERVICE_ID_SEEDANCE2_PRO_JOB: &str = "2a78f884-c42c-4745-b826-b7a572780ea0"; pub const ROOTLY_SERVICE_ID_VIDEO_THUMBNAIL_JOB: &str = "9d22997c-4474-47cb-af7b-53119eed1a88"; +// TODO: create the Rootly service for this job and replace this placeholder id. +pub const ROOTLY_SERVICE_ID_USER_SPEND_ANALYTICS_JOB: &str = "00000000-0000-0000-0000-000000000000"; diff --git a/crates/schema/database/mysql_queries/src/queries/mod.rs b/crates/schema/database/mysql_queries/src/queries/mod.rs index 513a6ce889..0614fc530e 100644 --- a/crates/schema/database/mysql_queries/src/queries/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/mod.rs @@ -42,7 +42,9 @@ pub mod uploaded_videos; pub mod user_impersonation_requests; pub mod user_referral_codes; pub mod user_referrals; +pub mod user_daily_spends; pub mod user_spend_events; +pub mod user_spend_summaries; pub mod users; pub mod vocoder; pub mod voice_clone_requests; diff --git a/crates/schema/database/mysql_queries/src/queries/user_daily_spends/aggregate_daily_spends_for_date.rs b/crates/schema/database/mysql_queries/src/queries/user_daily_spends/aggregate_daily_spends_for_date.rs new file mode 100644 index 0000000000..ada545d627 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_daily_spends/aggregate_daily_spends_for_date.rs @@ -0,0 +1,76 @@ +use std::marker::PhantomData; + +use chrono::{DateTime, Utc}; +use sqlx::{Executor, MySql}; + +use enums::common::payments_namespace::PaymentsNamespace; +use tokens::tokens::users::UserToken; + +/// One (user, namespace) aggregate of spend activity for a single UTC day, +/// computed from `user_spend_events`. Only produced for (user, day) pairs that +/// actually had spend or refund activity (sparse — see the HAVING clause). +pub struct DailySpendAggregate { + pub user_token: UserToken, + pub payments_namespace: PaymentsNamespace, + pub subscription_spend_usd_cents: u64, + pub credits_spend_usd_cents: u64, + pub gross_spend_usd_cents: u64, + pub refund_usd_cents: u64, + pub net_spend_usd_cents: i64, + pub payment_count: u64, + pub credits_granted: u64, +} + +pub struct AggregateDailySpendsForDateArgs<'c, E> +where + E: Executor<'c, Database = MySql>, +{ + /// Inclusive start of the UTC day. + pub range_start: DateTime, + /// Exclusive end of the UTC day (start + 1 day). + pub range_end: DateTime, + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +/// Aggregate spend events for one UTC day, grouped by (user, namespace). Rows +/// with neither spend nor refund are excluded so `user_daily_spends` stays +/// sparse. Only attributed events (non-null user) are counted. +pub async fn aggregate_daily_spends_for_date<'c, E>( + args: AggregateDailySpendsForDateArgs<'c, E>, +) -> Result, sqlx::Error> +where + E: Executor<'c, Database = MySql>, +{ + let rows = sqlx::query_as!( + DailySpendAggregate, + r#" +SELECT + maybe_user_token AS `user_token!: UserToken`, + payments_namespace AS `payments_namespace!: PaymentsNamespace`, + CAST(SUM(CASE WHEN amount_usd_cents > 0 + AND event_type IN ('subscription_initial', 'subscription_renewal', 'subscription_proration_upgrade') + THEN amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `subscription_spend_usd_cents!: u64`, + CAST(SUM(CASE WHEN amount_usd_cents > 0 AND event_type = 'credit_pack_purchase' + THEN amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `credits_spend_usd_cents!: u64`, + CAST(SUM(CASE WHEN amount_usd_cents > 0 THEN amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `gross_spend_usd_cents!: u64`, + CAST(SUM(CASE WHEN amount_usd_cents < 0 THEN -amount_usd_cents ELSE 0 END) AS UNSIGNED) AS `refund_usd_cents!: u64`, + CAST(SUM(amount_usd_cents) AS SIGNED) AS `net_spend_usd_cents!: i64`, + CAST(SUM(CASE WHEN amount_usd_cents > 0 THEN 1 ELSE 0 END) AS UNSIGNED) AS `payment_count!: u64`, + CAST(COALESCE(SUM(maybe_credits_granted), 0) AS UNSIGNED) AS `credits_granted!: u64` +FROM user_spend_events +WHERE maybe_user_token IS NOT NULL + AND is_production = TRUE + AND payment_occurred_at >= ? + AND payment_occurred_at < ? +GROUP BY maybe_user_token, payments_namespace +HAVING SUM(CASE WHEN amount_usd_cents <> 0 THEN 1 ELSE 0 END) > 0 + "#, + args.range_start, + args.range_end, + ) + .fetch_all(args.mysql_executor) + .await?; + + Ok(rows) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_daily_spends/mod.rs b/crates/schema/database/mysql_queries/src/queries/user_daily_spends/mod.rs new file mode 100644 index 0000000000..a0cea4118b --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_daily_spends/mod.rs @@ -0,0 +1,2 @@ +pub mod aggregate_daily_spends_for_date; +pub mod upsert_user_daily_spend; diff --git a/crates/schema/database/mysql_queries/src/queries/user_daily_spends/upsert_user_daily_spend.rs b/crates/schema/database/mysql_queries/src/queries/user_daily_spends/upsert_user_daily_spend.rs new file mode 100644 index 0000000000..191929367f --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_daily_spends/upsert_user_daily_spend.rs @@ -0,0 +1,73 @@ +use std::marker::PhantomData; + +use chrono::NaiveDate; +use sqlx::{Executor, MySql}; + +use enums::common::payments_namespace::PaymentsNamespace; +use tokens::tokens::users::UserToken; + +/// Upsert one `user_daily_spends` row (recompute-overwrite on the +/// (user, namespace, date) unique key). +pub struct UpsertUserDailySpendArgs<'e, 'c, E> +where + E: 'e + Executor<'c, Database = MySql>, +{ + pub user_token: &'e UserToken, + pub payments_namespace: PaymentsNamespace, + pub spend_date: NaiveDate, + pub subscription_spend_usd_cents: u64, + pub credits_spend_usd_cents: u64, + pub gross_spend_usd_cents: u64, + pub refund_usd_cents: u64, + pub net_spend_usd_cents: i64, + pub payment_count: u32, + pub credits_granted: u64, + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +pub async fn upsert_user_daily_spend<'e, 'c: 'e, E>( + args: UpsertUserDailySpendArgs<'e, 'c, E>, +) -> Result<(), sqlx::Error> +where + E: 'e + Executor<'c, Database = MySql>, +{ + sqlx::query!( + r#" +INSERT INTO user_daily_spends +SET + user_token = ?, + payments_namespace = ?, + spend_date = ?, + subscription_spend_usd_cents = ?, + credits_spend_usd_cents = ?, + gross_spend_usd_cents = ?, + refund_usd_cents = ?, + net_spend_usd_cents = ?, + payment_count = ?, + credits_granted = ? +ON DUPLICATE KEY UPDATE + subscription_spend_usd_cents = VALUES(subscription_spend_usd_cents), + credits_spend_usd_cents = VALUES(credits_spend_usd_cents), + gross_spend_usd_cents = VALUES(gross_spend_usd_cents), + refund_usd_cents = VALUES(refund_usd_cents), + net_spend_usd_cents = VALUES(net_spend_usd_cents), + payment_count = VALUES(payment_count), + credits_granted = VALUES(credits_granted) + "#, + args.user_token.as_str(), + args.payments_namespace.to_str(), + args.spend_date, + args.subscription_spend_usd_cents, + args.credits_spend_usd_cents, + args.gross_spend_usd_cents, + args.refund_usd_cents, + args.net_spend_usd_cents, + args.payment_count, + args.credits_granted, + ) + .execute(args.mysql_executor) + .await?; + + Ok(()) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_payments_namespaces_with_spend_activity.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_payments_namespaces_with_spend_activity.rs new file mode 100644 index 0000000000..6b21596631 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_payments_namespaces_with_spend_activity.rs @@ -0,0 +1,34 @@ +use std::marker::PhantomData; + +use sqlx::{Executor, MySql}; + +use enums::common::payments_namespace::PaymentsNamespace; + +pub struct ListPaymentsNamespacesWithSpendActivityArgs<'c, E> +where + E: Executor<'c, Database = MySql>, +{ + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +/// The distinct payment namespaces that have any attributed spend activity. +pub async fn list_payments_namespaces_with_spend_activity<'c, E>( + args: ListPaymentsNamespacesWithSpendActivityArgs<'c, E>, +) -> Result, sqlx::Error> +where + E: Executor<'c, Database = MySql>, +{ + let rows = sqlx::query_scalar!( + r#" +SELECT DISTINCT payments_namespace AS `payments_namespace!: PaymentsNamespace` +FROM user_spend_events +WHERE maybe_user_token IS NOT NULL + AND is_production = TRUE + "#, + ) + .fetch_all(args.mysql_executor) + .await?; + + Ok(rows) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_spend_events_for_user.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_spend_events_for_user.rs new file mode 100644 index 0000000000..0544fc53e2 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_spend_events_for_user.rs @@ -0,0 +1,55 @@ +use std::marker::PhantomData; + +use chrono::{DateTime, Utc}; +use sqlx::{Executor, MySql}; + +use enums::by_table::user_spend_events::payment_event_type::PaymentEventType; +use enums::common::payments_namespace::PaymentsNamespace; +use tokens::tokens::users::UserToken; + +/// A single spend event, trimmed to the fields the summary aggregation needs. +pub struct UserSpendEventRow { + pub payment_occurred_at: DateTime, + pub amount_usd_cents: i64, + pub event_type: PaymentEventType, + pub maybe_credits_granted: Option, +} + +pub struct ListUserSpendEventsForUserArgs<'e, 'c, E> +where + E: 'e + Executor<'c, Database = MySql>, +{ + pub user_token: &'e UserToken, + pub payments_namespace: PaymentsNamespace, + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +pub async fn list_user_spend_events_for_user<'e, 'c: 'e, E>( + args: ListUserSpendEventsForUserArgs<'e, 'c, E>, +) -> Result, sqlx::Error> +where + E: 'e + Executor<'c, Database = MySql>, +{ + let rows = sqlx::query_as!( + UserSpendEventRow, + r#" +SELECT + payment_occurred_at AS `payment_occurred_at!: DateTime`, + amount_usd_cents, + event_type AS `event_type!: PaymentEventType`, + maybe_credits_granted AS `maybe_credits_granted: u32` +FROM user_spend_events +WHERE maybe_user_token = ? + AND payments_namespace = ? + AND is_production = TRUE +ORDER BY payment_occurred_at ASC + "#, + args.user_token.as_str(), + args.payments_namespace.to_str(), + ) + .fetch_all(args.mysql_executor) + .await?; + + Ok(rows) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_tokens_with_spend_activity.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_tokens_with_spend_activity.rs new file mode 100644 index 0000000000..11018aa2f8 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/list_user_tokens_with_spend_activity.rs @@ -0,0 +1,46 @@ +use std::marker::PhantomData; + +use sqlx::{Executor, MySql}; + +use enums::common::payments_namespace::PaymentsNamespace; +use tokens::tokens::users::UserToken; + +/// Keyset page of distinct user tokens with spend activity in one namespace. +/// Ordered by `user_token`; start with an empty `after_user_token`. +pub struct ListUserTokensWithSpendActivityArgs<'e, 'c, E> +where + E: 'e + Executor<'c, Database = MySql>, +{ + pub payments_namespace: PaymentsNamespace, + pub after_user_token: &'e str, + pub limit: i64, + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +pub async fn list_user_tokens_with_spend_activity<'e, 'c: 'e, E>( + args: ListUserTokensWithSpendActivityArgs<'e, 'c, E>, +) -> Result, sqlx::Error> +where + E: 'e + Executor<'c, Database = MySql>, +{ + let rows = sqlx::query_scalar!( + r#" +SELECT DISTINCT maybe_user_token AS `user_token!: UserToken` +FROM user_spend_events +WHERE maybe_user_token IS NOT NULL + AND is_production = TRUE + AND payments_namespace = ? + AND maybe_user_token > ? +ORDER BY maybe_user_token +LIMIT ? + "#, + args.payments_namespace.to_str(), + args.after_user_token, + args.limit, + ) + .fetch_all(args.mysql_executor) + .await?; + + Ok(rows) +} diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/mod.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/mod.rs new file mode 100644 index 0000000000..ad67766f20 --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/mod.rs @@ -0,0 +1,4 @@ +pub mod list_payments_namespaces_with_spend_activity; +pub mod list_user_tokens_with_spend_activity; +pub mod list_user_spend_events_for_user; +pub mod upsert_user_spend_summary; diff --git a/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/upsert_user_spend_summary.rs b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/upsert_user_spend_summary.rs new file mode 100644 index 0000000000..c8f75eb36e --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/user_spend_summaries/upsert_user_spend_summary.rs @@ -0,0 +1,129 @@ +use std::marker::PhantomData; + +use chrono::{DateTime, Utc}; +use sqlx::{Executor, MySql}; + +use enums::common::payments_namespace::PaymentsNamespace; +use tokens::tokens::users::UserToken; + +/// Upsert one `user_spend_summaries` row (recompute-overwrite on the +/// (payments_namespace, user_token) unique key). Window/cadence/score fields are +/// all recomputed each run since they are time-relative. +pub struct UpsertUserSpendSummaryArgs<'e, 'c, E> +where + E: 'e + Executor<'c, Database = MySql>, +{ + pub payments_namespace: PaymentsNamespace, + pub user_token: &'e UserToken, + + pub lifetime_gross_spend_usd_cents: u64, + pub lifetime_subscription_spend_usd_cents: u64, + pub lifetime_credits_spend_usd_cents: u64, + pub lifetime_refund_usd_cents: u64, + pub lifetime_net_spend_usd_cents: u64, + pub lifetime_payment_count: u32, + pub lifetime_refund_count: u32, + + pub maybe_first_payment_at: Option>, + pub first_spend_usd_cents: u64, + pub maybe_last_payment_at: Option>, + pub last_spend_usd_cents: u64, + pub maybe_days_since_first_payment: Option, + pub maybe_days_since_last_payment: Option, + + pub net_spend_this_year_usd_cents: i64, + + pub consecutive_active_weeks: u32, + pub consecutive_inactive_weeks: u32, + pub maybe_weeks_since_last_spend: Option, + + pub is_active_subscriber: bool, + pub maybe_subscription_interval: Option<&'e str>, + pub maybe_reengagement_score: Option, + + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +pub async fn upsert_user_spend_summary<'e, 'c: 'e, E>( + args: UpsertUserSpendSummaryArgs<'e, 'c, E>, +) -> Result<(), sqlx::Error> +where + E: 'e + Executor<'c, Database = MySql>, +{ + sqlx::query!( + r#" +INSERT INTO user_spend_summaries +SET + payments_namespace = ?, + user_token = ?, + lifetime_gross_spend_usd_cents = ?, + lifetime_subscription_spend_usd_cents = ?, + lifetime_credits_spend_usd_cents = ?, + lifetime_refund_usd_cents = ?, + lifetime_net_spend_usd_cents = ?, + lifetime_payment_count = ?, + lifetime_refund_count = ?, + first_payment_at = ?, + first_spend_usd_cents = ?, + last_payment_at = ?, + last_spend_usd_cents = ?, + days_since_first_payment = ?, + days_since_last_payment = ?, + net_spend_this_year_usd_cents = ?, + consecutive_active_weeks = ?, + consecutive_inactive_weeks = ?, + weeks_since_last_spend = ?, + is_active_subscriber = ?, + maybe_subscription_interval = ?, + maybe_reengagement_score = ? +ON DUPLICATE KEY UPDATE + lifetime_gross_spend_usd_cents = VALUES(lifetime_gross_spend_usd_cents), + lifetime_subscription_spend_usd_cents = VALUES(lifetime_subscription_spend_usd_cents), + lifetime_credits_spend_usd_cents = VALUES(lifetime_credits_spend_usd_cents), + lifetime_refund_usd_cents = VALUES(lifetime_refund_usd_cents), + lifetime_net_spend_usd_cents = VALUES(lifetime_net_spend_usd_cents), + lifetime_payment_count = VALUES(lifetime_payment_count), + lifetime_refund_count = VALUES(lifetime_refund_count), + first_payment_at = VALUES(first_payment_at), + first_spend_usd_cents = VALUES(first_spend_usd_cents), + last_payment_at = VALUES(last_payment_at), + last_spend_usd_cents = VALUES(last_spend_usd_cents), + days_since_first_payment = VALUES(days_since_first_payment), + days_since_last_payment = VALUES(days_since_last_payment), + net_spend_this_year_usd_cents = VALUES(net_spend_this_year_usd_cents), + consecutive_active_weeks = VALUES(consecutive_active_weeks), + consecutive_inactive_weeks = VALUES(consecutive_inactive_weeks), + weeks_since_last_spend = VALUES(weeks_since_last_spend), + is_active_subscriber = VALUES(is_active_subscriber), + maybe_subscription_interval = VALUES(maybe_subscription_interval), + maybe_reengagement_score = VALUES(maybe_reengagement_score) + "#, + args.payments_namespace.to_str(), + args.user_token.as_str(), + args.lifetime_gross_spend_usd_cents, + args.lifetime_subscription_spend_usd_cents, + args.lifetime_credits_spend_usd_cents, + args.lifetime_refund_usd_cents, + args.lifetime_net_spend_usd_cents, + args.lifetime_payment_count, + args.lifetime_refund_count, + args.maybe_first_payment_at, + args.first_spend_usd_cents, + args.maybe_last_payment_at, + args.last_spend_usd_cents, + args.maybe_days_since_first_payment, + args.maybe_days_since_last_payment, + args.net_spend_this_year_usd_cents, + args.consecutive_active_weeks, + args.consecutive_inactive_weeks, + args.maybe_weeks_since_last_spend, + args.is_active_subscriber, + args.maybe_subscription_interval, + args.maybe_reengagement_score, + ) + .execute(args.mysql_executor) + .await?; + + Ok(()) +} diff --git a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_active_subscription_status_for_user.rs b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_active_subscription_status_for_user.rs new file mode 100644 index 0000000000..9cc794e4ed --- /dev/null +++ b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/get_active_subscription_status_for_user.rs @@ -0,0 +1,55 @@ +use std::marker::PhantomData; + +use chrono::{DateTime, Utc}; +use sqlx::{Executor, MySql}; + +use enums::common::payments_namespace::PaymentsNamespace; +use tokens::tokens::users::UserToken; + +/// The user's most-relevant subscription (latest-expiring), for deriving +/// `is_active_subscriber` and `maybe_subscription_interval` in the analytics job. +pub struct UserSubscriptionStatusRow { + pub maybe_stripe_subscription_status: Option, + pub maybe_stripe_recurring_interval: Option, + pub subscription_expires_at: DateTime, +} + +pub struct GetActiveSubscriptionStatusArgs<'e, 'c, E> +where + E: 'e + Executor<'c, Database = MySql>, +{ + pub user_token: &'e UserToken, + pub subscription_namespace: PaymentsNamespace, + pub mysql_executor: E, + pub phantom: PhantomData<&'c E>, +} + +/// Returns the latest-expiring subscription row for the user (or `None`). The +/// caller decides active-ness from the status + expiry. +pub async fn get_active_subscription_status_for_user<'e, 'c: 'e, E>( + args: GetActiveSubscriptionStatusArgs<'e, 'c, E>, +) -> Result, sqlx::Error> +where + E: 'e + Executor<'c, Database = MySql>, +{ + let row = sqlx::query_as!( + UserSubscriptionStatusRow, + r#" +SELECT + maybe_stripe_subscription_status, + maybe_stripe_recurring_interval, + subscription_expires_at AS `subscription_expires_at!: DateTime` +FROM user_subscriptions +WHERE user_token = ? + AND subscription_namespace = ? +ORDER BY subscription_expires_at DESC +LIMIT 1 + "#, + args.user_token.as_str(), + args.subscription_namespace.to_str(), + ) + .fetch_optional(args.mysql_executor) + .await?; + + Ok(row) +} diff --git a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/mod.rs b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/mod.rs index 2ae715e34b..382653b3f7 100644 --- a/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/mod.rs +++ b/crates/schema/database/mysql_queries/src/queries/users/user_subscriptions/mod.rs @@ -1,3 +1,4 @@ +pub mod get_active_subscription_status_for_user; pub mod find_possibly_inactive_first_subscription_for_owner_user; pub mod find_subscription_for_owner_user; pub mod get_user_subscription_by_stripe_subscription_id; diff --git a/crates/service/job/user_spend_analytics_job/Cargo.toml b/crates/service/job/user_spend_analytics_job/Cargo.toml new file mode 100644 index 0000000000..c71c5491a6 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/Cargo.toml @@ -0,0 +1,47 @@ +[package] +name = "user-spend-analytics-job" +edition = "2021" +version = "0.0.1" +authors = [ + "Brandon Thomas ", + "Brandon Thomas ", +] +publish = false + +[[bin]] +name = "user-spend-analytics-job" +path = "src/main.rs" + +[package.metadata.cargo-udeps.ignore] +normal = ["workspace-build-acceleration"] + +[dependencies] +# cargo-hakari workspace hack (faster compile times) +workspace-build-acceleration.workspace = true + +# Internal +actix_helpers.workspace = true +bootstrap.workspace = true +concurrency.workspace = true +easyenv.workspace = true +enums.workspace = true +errors.workspace = true +jobs_common.workspace = true +mysql_queries.workspace = true +pager = { path = "../../../lib/pager" } +rootly_client.workspace = true +rootly_config.workspace = true +shared_env_var_config.workspace = true +server_environment = { path = "../../../lib/server_environment" } +tokens.workspace = true + +# External +actix-web.workspace = true +anyhow.workspace = true +chrono.workspace = true +log.workspace = true +serde.workspace = true +serde_derive.workspace = true +serde_json.workspace = true +sqlx = { version = "0.7.4", features = [ "mysql", "runtime-tokio-rustls", "chrono" ] } +tokio.workspace = true diff --git a/crates/service/job/user_spend_analytics_job/config/user-spend-analytics-job.development.env b/crates/service/job/user_spend_analytics_job/config/user-spend-analytics-job.development.env new file mode 100644 index 0000000000..59ea673f8b --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/config/user-spend-analytics-job.development.env @@ -0,0 +1,22 @@ +SERVER_ENVIRONMENT=development +RUST_LOG=info + +# Cycle cadence + throttling +SLEEP_BETWEEN_CYCLES_SECONDS=86400 +QUERY_DELAY_MILLIS=50 +ERROR_RECOVERY_SECONDS=30 + +# Daily-spends backfill window (pick ONE; omit both = all-time, with a warning) +BACKFILL_DAYS=3 +# BACKFILL_START_DATE=2026-02-01 + +# Summaries +SUMMARY_USER_PAGE_SIZE=500 + +# Reengagement score tunables +REENGAGEMENT_VALUE_CAP_DOLLARS=1000 +REENGAGEMENT_ACTIVE_THRESHOLD_DAYS=14 +REENGAGEMENT_DECAY_DAYS=350 + +# Paging (off in dev) +ENABLE_PAGING=false diff --git a/crates/service/job/user_spend_analytics_job/src/http_server/endpoints/health_check_handler.rs b/crates/service/job/user_spend_analytics_job/src/http_server/endpoints/health_check_handler.rs new file mode 100644 index 0000000000..9c5f4a8b8f --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/http_server/endpoints/health_check_handler.rs @@ -0,0 +1,127 @@ +use std::sync::Arc; + +use actix_web::error::ResponseError; +use actix_web::http::header::ContentType; +use actix_web::http::StatusCode; +use actix_web::{web, HttpRequest, HttpResponse}; +use log::error; + +use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response; +use pager::notification::notification_details_builder::NotificationDetailsBuilder; +use pager::notification::notification_urgency::NotificationUrgency; + +use crate::http_server::http_server_shared_state::HttpServerSharedState; + +#[derive(Serialize)] +pub struct HealthCheckResponse { + pub success: bool, + pub is_healthy: bool, + pub consecutive_failure_count: u64, + pub consecutive_success_count: u64, + pub total_failure_count: u64, + pub total_success_count: u64, + pub total_failure_ratio: f32, + pub total_success_ratio: f32, +} + +#[derive(Debug, Serialize)] +pub enum HealthCheckError { + ServerError, +} + +impl ResponseError for HealthCheckError { + fn status_code(&self) -> StatusCode { + match *self { + HealthCheckError::ServerError => StatusCode::INTERNAL_SERVER_ERROR, + } + } + + fn error_response(&self) -> HttpResponse { + error_to_json_http_response(self) + } +} + +impl std::fmt::Display for HealthCheckError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self) + } +} + +pub async fn get_health_check_handler( + http_request: HttpRequest, + server_state: web::Data>, +) -> Result { + let job_stats = server_state + .job_stats + .get_status() + .map_err(|e| { + error!("Error serving health check status: {:?}", e); + HealthCheckError::ServerError + })?; + + let total_tries = job_stats + .total_failure_count + .saturating_add(job_stats.total_success_count); + + let total_success_ratio = if total_tries > 0 { + (job_stats.total_success_count as f32) / (total_tries as f32) + } else { + 0.0 + }; + + let total_failure_ratio = if total_tries > 0 { + 1.0 - total_success_ratio + } else { + 0.0 + }; + + let is_healthy = + job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold; + + if !is_healthy { + let notification = NotificationDetailsBuilder::from_title( + format!("Health check unhealthy on {}", server_state.hostname)) + .set_description(Some(format!( + "Health check returned unhealthy.\n\n\ + Hostname: {}\n\ + Consecutive failure count: {}\n\ + Total failure count: {}\n\ + Total success count: {}", + server_state.hostname, + job_stats.consecutive_failure_count, + job_stats.total_failure_count, + job_stats.total_success_count, + ))) + .set_urgency(Some(NotificationUrgency::High)) + .set_http_method(Some(http_request.method().to_string())) + .set_http_path(Some(http_request.path().to_string())) + .build(); + + if let Err(err) = server_state.pager.enqueue_page(notification) { + error!("Failed to enqueue health check alert: {:?}", err); + } + } + + let response = HealthCheckResponse { + success: true, + is_healthy, + consecutive_failure_count: job_stats.consecutive_failure_count, + consecutive_success_count: job_stats.consecutive_success_count, + total_failure_count: job_stats.total_failure_count, + total_success_count: job_stats.total_success_count, + total_failure_ratio, + total_success_ratio, + }; + + let body = serde_json::to_string(&response).map_err(|_e| HealthCheckError::ServerError)?; + + if is_healthy { + Ok(HttpResponse::Ok() + .content_type(ContentType::json()) + .body(body)) + } else { + Ok(HttpResponse::InternalServerError() + .content_type(ContentType::json()) + .body(body)) + } +} diff --git a/crates/service/job/user_spend_analytics_job/src/http_server/endpoints/mod.rs b/crates/service/job/user_spend_analytics_job/src/http_server/endpoints/mod.rs new file mode 100644 index 0000000000..0f408270bf --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/http_server/endpoints/mod.rs @@ -0,0 +1 @@ +pub mod health_check_handler; diff --git a/crates/service/job/user_spend_analytics_job/src/http_server/http_server_shared_state.rs b/crates/service/job/user_spend_analytics_job/src/http_server/http_server_shared_state.rs new file mode 100644 index 0000000000..ef3e9fe208 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/http_server/http_server_shared_state.rs @@ -0,0 +1,10 @@ +use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; + +#[derive(Clone)] +pub struct HttpServerSharedState { + pub job_stats: JobStats, + pub consecutive_failure_unhealthy_threshold: u64, + pub pager: Pager, + pub hostname: String, +} diff --git a/crates/service/job/user_spend_analytics_job/src/http_server/mod.rs b/crates/service/job/user_spend_analytics_job/src/http_server/mod.rs new file mode 100644 index 0000000000..5d888824cc --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/http_server/mod.rs @@ -0,0 +1,3 @@ +pub mod endpoints; +pub mod http_server_shared_state; +pub mod run_http_server; diff --git a/crates/service/job/user_spend_analytics_job/src/http_server/run_http_server.rs b/crates/service/job/user_spend_analytics_job/src/http_server/run_http_server.rs new file mode 100644 index 0000000000..c6dc6d247f --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/http_server/run_http_server.rs @@ -0,0 +1,76 @@ +use std::sync::Arc; + +use actix_web::dev::Server; +use actix_web::middleware::Logger; +use actix_web::{web, App, HttpResponse, HttpServer}; +use log::info; + +use bootstrap::bootstrap::ContainerEnvironment; +use errors::AnyhowResult; +use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; + +use crate::http_server::endpoints::health_check_handler::get_health_check_handler; +use crate::http_server::http_server_shared_state::HttpServerSharedState; + +const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:13339"; +const DEFAULT_NUM_WORKERS: usize = 2; + +pub struct CreateServerArgs { + pub container_environment: ContainerEnvironment, + pub job_stats: JobStats, + pub pager: Pager, +} + +pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult { + let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS); + let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?; + let hostname = args.container_environment.hostname.clone(); + + let server_state = HttpServerSharedState { + job_stats: args.job_stats.clone(), + consecutive_failure_unhealthy_threshold: easyenv::get_env_num( + "CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD", + 3, + )?, + pager: args.pager, + hostname, + }; + + let server_state_arc = web::Data::new(Arc::new(server_state)); + + info!("Starting HTTP service (for k8s health checking)."); + + let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T"; + + let handle = HttpServer::new(move || { + App::new() + .app_data(server_state_arc.clone()) + .wrap(Logger::new(&log_format).exclude("/_status")) + .service( + web::resource("/") + .route(web::get().to(|| HttpResponse::Ok())) + .route(web::head().to(|| HttpResponse::Ok())), + ) + .service( + web::resource("/_status") + .route(web::get().to(get_health_check_handler)) + .route(web::head().to(|| HttpResponse::Ok())), + ) + }) + .bind(&bind_address) + .unwrap_or_else(|err| { + eprintln!("FATAL: Failed to bind to address '{}': {}", bind_address, err); + eprintln!("The address is likely already in use by another process."); + std::process::exit(1); + }) + .workers(num_workers) + .run(); + + Ok(handle) +} + +pub async fn launch_http_server(args: CreateServerArgs) -> AnyhowResult<()> { + run_http_server(args)?.await?; + Ok(()) +} diff --git a/crates/service/job/user_spend_analytics_job/src/job/alert_on_error.rs b/crates/service/job/user_spend_analytics_job/src/job/alert_on_error.rs new file mode 100644 index 0000000000..72599b9410 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job/alert_on_error.rs @@ -0,0 +1,17 @@ +use log::error; +use pager::client::pager::Pager; +use pager::notification::notification_details_builder::NotificationDetailsBuilder; +use pager::notification::notification_urgency::NotificationUrgency; + +/// Enqueue a pager alert for an error WITHOUT interrupting the job. The caller is +/// expected to keep going (next date / user / record) after a recovery sleep. +pub fn alert_pager(pager: &Pager, title: &str, detail: &str) { + let notification = NotificationDetailsBuilder::from_title(title.to_string()) + .set_description(Some(detail.to_string())) + .set_urgency(Some(NotificationUrgency::Medium)) + .build(); + + if let Err(pager_err) = pager.enqueue_page(notification) { + error!("Failed to enqueue pager alert: {:?}", pager_err); + } +} diff --git a/crates/service/job/user_spend_analytics_job/src/job/backfill_daily_spends.rs b/crates/service/job/user_spend_analytics_job/src/job/backfill_daily_spends.rs new file mode 100644 index 0000000000..b0a7978e31 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job/backfill_daily_spends.rs @@ -0,0 +1,130 @@ +use std::marker::PhantomData; + +use chrono::{DateTime, Duration as ChronoDuration, NaiveDate, TimeZone, Utc}; +use log::{error, info, warn}; + +use mysql_queries::queries::user_daily_spends::aggregate_daily_spends_for_date::{ + aggregate_daily_spends_for_date, AggregateDailySpendsForDateArgs, +}; +use mysql_queries::queries::user_daily_spends::upsert_user_daily_spend::{ + upsert_user_daily_spend, UpsertUserDailySpendArgs, +}; + +use crate::job::alert_on_error::alert_pager; +use crate::job_dependencies::JobDependencies; + +const ALL_TIME_START: (i32, u32, u32) = (2020, 1, 1); + +/// Backfill `user_daily_spends` for each UTC day from the configured start date +/// up to today, one date at a time. Days with no spend/refund produce no rows. +pub async fn backfill_daily_spends(deps: &JobDependencies) { + let start_date = resolve_start_date(deps); + let today = Utc::now().date_naive(); + info!("Daily-spends backfill window: {start_date} .. {today} (UTC)."); + + let mut date = start_date; + let mut total_upserted = 0u64; + + while date <= today { + if deps.application_shutdown.get() { + return; + } + match process_date(deps, date).await { + Ok(count) => { + total_upserted += count as u64; + if count > 0 { + info!(" {date}: upserted {count} daily row(s)."); + } + } + Err(err) => { + error!("Daily-spends aggregation failed for {date}: {err:?}"); + alert_pager( + &deps.pager, + "user-spend-analytics: daily aggregation error", + &format!("date={date}: {err:?}"), + ); + let _ = deps.job_stats.increment_failure_count(); + tokio::time::sleep(deps.error_recovery).await; + } + } + + tokio::time::sleep(deps.query_delay).await; + date = match date.succ_opt() { + Some(next) => next, + None => break, + }; + } + + info!("Daily-spends backfill complete: {total_upserted} row(s) upserted."); +} + +async fn process_date(deps: &JobDependencies, date: NaiveDate) -> Result { + let range_start: DateTime = + Utc.from_utc_datetime(&date.and_hms_opt(0, 0, 0).expect("valid midnight")); + let range_end = range_start + ChronoDuration::days(1); + + let aggregates = aggregate_daily_spends_for_date(AggregateDailySpendsForDateArgs { + range_start, + range_end, + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }) + .await?; + + let mut upserted = 0usize; + for aggregate in &aggregates { + if deps.application_shutdown.get() { + break; + } + let result = upsert_user_daily_spend(UpsertUserDailySpendArgs { + user_token: &aggregate.user_token, + payments_namespace: aggregate.payments_namespace, + spend_date: date, + subscription_spend_usd_cents: aggregate.subscription_spend_usd_cents, + credits_spend_usd_cents: aggregate.credits_spend_usd_cents, + gross_spend_usd_cents: aggregate.gross_spend_usd_cents, + refund_usd_cents: aggregate.refund_usd_cents, + net_spend_usd_cents: aggregate.net_spend_usd_cents, + payment_count: u32::try_from(aggregate.payment_count).unwrap_or(u32::MAX), + credits_granted: aggregate.credits_granted, + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }) + .await; + + match result { + Ok(()) => { + upserted += 1; + let _ = deps.job_stats.increment_success_count(); + } + Err(err) => { + error!("Daily upsert failed for user {} on {date}: {err:?}", aggregate.user_token.as_str()); + alert_pager( + &deps.pager, + "user-spend-analytics: daily upsert error", + &format!("user={} date={date}: {err:?}", aggregate.user_token.as_str()), + ); + let _ = deps.job_stats.increment_failure_count(); + tokio::time::sleep(deps.error_recovery).await; + } + } + } + + Ok(upserted) +} + +/// `BACKFILL_START_DATE` wins; else `today - BACKFILL_DAYS`; else all-time (with a loud warning). +fn resolve_start_date(deps: &JobDependencies) -> NaiveDate { + if let Some(start_date) = deps.maybe_backfill_start_date { + return start_date; + } + let today = Utc::now().date_naive(); + if let Some(days) = deps.maybe_backfill_days { + return today - ChronoDuration::days(days.max(0)); + } + warn!( + "⚠️ Neither BACKFILL_START_DATE nor BACKFILL_DAYS is set — backfilling \ + user_daily_spends for ALL TIME. This is expensive; set one to bound the window." + ); + NaiveDate::from_ymd_opt(ALL_TIME_START.0, ALL_TIME_START.1, ALL_TIME_START.2).expect("valid date") +} diff --git a/crates/service/job/user_spend_analytics_job/src/job/backfill_user_summaries.rs b/crates/service/job/user_spend_analytics_job/src/job/backfill_user_summaries.rs new file mode 100644 index 0000000000..46f51c9300 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job/backfill_user_summaries.rs @@ -0,0 +1,363 @@ +use std::collections::HashSet; +use std::marker::PhantomData; + +use chrono::{DateTime, Datelike, Utc}; +use log::{error, info}; + +use enums::by_table::user_spend_events::payment_event_type::PaymentEventType; +use enums::common::payments_namespace::PaymentsNamespace; +use mysql_queries::queries::user_spend_summaries::list_payments_namespaces_with_spend_activity::{ + list_payments_namespaces_with_spend_activity, ListPaymentsNamespacesWithSpendActivityArgs, +}; +use mysql_queries::queries::user_spend_summaries::list_user_spend_events_for_user::{ + list_user_spend_events_for_user, ListUserSpendEventsForUserArgs, UserSpendEventRow, +}; +use mysql_queries::queries::user_spend_summaries::list_user_tokens_with_spend_activity::{ + list_user_tokens_with_spend_activity, ListUserTokensWithSpendActivityArgs, +}; +use mysql_queries::queries::user_spend_summaries::upsert_user_spend_summary::{ + upsert_user_spend_summary, UpsertUserSpendSummaryArgs, +}; +use mysql_queries::queries::users::user_subscriptions::get_active_subscription_status_for_user::{ + get_active_subscription_status_for_user, GetActiveSubscriptionStatusArgs, +}; +use tokens::tokens::users::UserToken; + +use crate::job::alert_on_error::alert_pager; +use crate::job::reengagement_score::reengagement_score; +use crate::job_dependencies::JobDependencies; + +const DAYS_PER_WEEK: i64 = 7; + +/// Recompute `user_spend_summaries` for every (user, namespace) with spend +/// activity. Window/cadence/score fields are time-relative, so all users are +/// refreshed each cycle against a single reference time. We iterate one namespace +/// at a time, keyset-paginating users by `user_token`. +pub async fn backfill_user_summaries(deps: &JobDependencies) { + let now = Utc::now(); + info!("User-summaries backfill starting (reference time {now})."); + + let namespaces = match list_payments_namespaces_with_spend_activity( + ListPaymentsNamespacesWithSpendActivityArgs { + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }, + ) + .await + { + Ok(namespaces) => namespaces, + Err(err) => { + error!("User-summaries: listing namespaces failed: {err:?}"); + alert_pager(&deps.pager, "user-spend-analytics: namespace list error", &format!("{err:?}")); + let _ = deps.job_stats.increment_failure_count(); + return; + } + }; + + let mut total_upserted = 0u64; + for namespace in namespaces { + total_upserted += backfill_namespace(deps, namespace, now).await; + if deps.application_shutdown.get() { + return; + } + } + + info!("User-summaries backfill complete: {total_upserted} user(s) upserted."); +} + +async fn backfill_namespace(deps: &JobDependencies, namespace: PaymentsNamespace, now: DateTime) -> u64 { + let mut after_user_token = String::new(); + let mut upserted = 0u64; + + loop { + if deps.application_shutdown.get() { + return upserted; + } + let user_tokens = match list_user_tokens_with_spend_activity(ListUserTokensWithSpendActivityArgs { + payments_namespace: namespace, + after_user_token: &after_user_token, + limit: deps.summary_user_page_size, + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }) + .await + { + Ok(user_tokens) => user_tokens, + Err(err) => { + error!("User-summaries: listing users for {} failed: {err:?}", namespace.to_str()); + alert_pager(&deps.pager, "user-spend-analytics: user list error", &format!("ns={}: {err:?}", namespace.to_str())); + let _ = deps.job_stats.increment_failure_count(); + return upserted; // can't paginate further this cycle + } + }; + + if user_tokens.is_empty() { + break; + } + + for user_token in &user_tokens { + if deps.application_shutdown.get() { + return upserted; + } + match process_user(deps, namespace, user_token, now).await { + Ok(()) => { + upserted += 1; + let _ = deps.job_stats.increment_success_count(); + } + Err(err) => { + error!("User-summaries: failed for {} ({}): {err:?}", user_token.as_str(), namespace.to_str()); + alert_pager( + &deps.pager, + "user-spend-analytics: summary error", + &format!("user={} ns={}: {err:?}", user_token.as_str(), namespace.to_str()), + ); + let _ = deps.job_stats.increment_failure_count(); + tokio::time::sleep(deps.error_recovery).await; + } + } + tokio::time::sleep(deps.query_delay).await; + } + + after_user_token = user_tokens.last().expect("non-empty page").as_str().to_string(); + } + + upserted +} + +async fn process_user( + deps: &JobDependencies, + namespace: PaymentsNamespace, + user_token: &UserToken, + now: DateTime, +) -> Result<(), sqlx::Error> { + let events = list_user_spend_events_for_user(ListUserSpendEventsForUserArgs { + user_token, + payments_namespace: namespace, + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }) + .await?; + + let aggregate = compute_summary(&events, now); + + // Authoritative subscription state (from user_subscriptions, not the events). + let subscription_row = get_active_subscription_status_for_user(GetActiveSubscriptionStatusArgs { + user_token, + subscription_namespace: namespace, + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }) + .await?; + let (is_active_subscriber, maybe_subscription_interval) = match subscription_row { + Some(row) => { + let active = matches!( + row.maybe_stripe_subscription_status.as_deref(), + Some("active") | Some("trialing") + ) && row.subscription_expires_at > now; + (active, row.maybe_stripe_recurring_interval) + } + None => (false, None), + }; + + let score = reengagement_score( + aggregate.lifetime_net_spend_usd_cents, + aggregate.maybe_days_since_last_payment, + &deps.reengagement, + ); + + upsert_user_spend_summary(UpsertUserSpendSummaryArgs { + payments_namespace: namespace, + user_token, + lifetime_gross_spend_usd_cents: aggregate.lifetime_gross_spend_usd_cents, + lifetime_subscription_spend_usd_cents: aggregate.lifetime_subscription_spend_usd_cents, + lifetime_credits_spend_usd_cents: aggregate.lifetime_credits_spend_usd_cents, + lifetime_refund_usd_cents: aggregate.lifetime_refund_usd_cents, + lifetime_net_spend_usd_cents: aggregate.lifetime_net_spend_usd_cents, + lifetime_payment_count: aggregate.lifetime_payment_count, + lifetime_refund_count: aggregate.lifetime_refund_count, + maybe_first_payment_at: aggregate.maybe_first_payment_at, + first_spend_usd_cents: aggregate.first_spend_usd_cents, + maybe_last_payment_at: aggregate.maybe_last_payment_at, + last_spend_usd_cents: aggregate.last_spend_usd_cents, + maybe_days_since_first_payment: aggregate.maybe_days_since_first_payment, + maybe_days_since_last_payment: aggregate.maybe_days_since_last_payment, + net_spend_this_year_usd_cents: aggregate.net_spend_this_year_usd_cents, + consecutive_active_weeks: aggregate.consecutive_active_weeks, + consecutive_inactive_weeks: aggregate.consecutive_inactive_weeks, + maybe_weeks_since_last_spend: aggregate.maybe_weeks_since_last_spend, + is_active_subscriber, + maybe_subscription_interval: maybe_subscription_interval.as_deref(), + maybe_reengagement_score: Some(score), + mysql_executor: &deps.mysql_pool, + phantom: PhantomData, + }) + .await?; + + Ok(()) +} + +/// Derived summary fields, computed from a user's full event history. +struct SummaryAggregate { + lifetime_gross_spend_usd_cents: u64, + lifetime_subscription_spend_usd_cents: u64, + lifetime_credits_spend_usd_cents: u64, + lifetime_refund_usd_cents: u64, + lifetime_net_spend_usd_cents: u64, + lifetime_payment_count: u32, + lifetime_refund_count: u32, + maybe_first_payment_at: Option>, + first_spend_usd_cents: u64, + maybe_last_payment_at: Option>, + last_spend_usd_cents: u64, + maybe_days_since_first_payment: Option, + maybe_days_since_last_payment: Option, + net_spend_this_year_usd_cents: i64, + consecutive_active_weeks: u32, + consecutive_inactive_weeks: u32, + maybe_weeks_since_last_spend: Option, +} + +/// Aggregate one user's events (assumed ascending by `payment_occurred_at`). +/// "Positive" events are payments; negatives are refunds/chargebacks. Weekly +/// cadence uses 7-day buckets relative to `now` (week 0 = the last 7 days). +fn compute_summary(events: &[UserSpendEventRow], now: DateTime) -> SummaryAggregate { + let current_year = now.year(); + + let mut gross: i128 = 0; + let mut subscription: i128 = 0; + let mut credits: i128 = 0; + let mut refund: i128 = 0; + let mut net: i128 = 0; + let mut net_this_year: i128 = 0; + let mut payment_count: u32 = 0; + let mut refund_count: u32 = 0; + let mut first_payment: Option<&UserSpendEventRow> = None; + let mut last_payment: Option<&UserSpendEventRow> = None; + let mut active_weeks: HashSet = HashSet::new(); + + for event in events { + let amount = event.amount_usd_cents as i128; + net += amount; + if event.payment_occurred_at.year() == current_year { + net_this_year += amount; + } + + if event.amount_usd_cents > 0 { + gross += amount; + payment_count += 1; + match event.event_type { + PaymentEventType::SubscriptionInitial + | PaymentEventType::SubscriptionRenewal + | PaymentEventType::SubscriptionProrationUpgrade => subscription += amount, + PaymentEventType::CreditPackPurchase => credits += amount, + _ => {} + } + if first_payment.is_none() { + first_payment = Some(event); + } + last_payment = Some(event); + + let days_ago = (now - event.payment_occurred_at).num_days(); + if days_ago >= 0 { + active_weeks.insert(days_ago / DAYS_PER_WEEK); + } + } else if event.amount_usd_cents < 0 { + refund += -amount; + refund_count += 1; + } + } + + let maybe_first_payment_at = first_payment.map(|e| e.payment_occurred_at); + let maybe_last_payment_at = last_payment.map(|e| e.payment_occurred_at); + let first_spend_usd_cents = first_payment.map(|e| e.amount_usd_cents.max(0) as u64).unwrap_or(0); + let last_spend_usd_cents = last_payment.map(|e| e.amount_usd_cents.max(0) as u64).unwrap_or(0); + let maybe_days_since_first_payment = + maybe_first_payment_at.map(|t| (now - t).num_days().max(0) as u32); + let maybe_days_since_last_payment = + maybe_last_payment_at.map(|t| (now - t).num_days().max(0) as u32); + + // Cadence from the set of active weeks-ago. + let maybe_weeks_since_last_spend = active_weeks.iter().min().map(|w| *w as u32); + let consecutive_active_weeks = { + let mut week = 0i64; + while active_weeks.contains(&week) { + week += 1; + } + week as u32 + }; + let consecutive_inactive_weeks = maybe_weeks_since_last_spend.unwrap_or(0); + + SummaryAggregate { + lifetime_gross_spend_usd_cents: gross.max(0) as u64, + lifetime_subscription_spend_usd_cents: subscription.max(0) as u64, + lifetime_credits_spend_usd_cents: credits.max(0) as u64, + lifetime_refund_usd_cents: refund.max(0) as u64, + lifetime_net_spend_usd_cents: net.max(0) as u64, + lifetime_payment_count: payment_count, + lifetime_refund_count: refund_count, + maybe_first_payment_at, + first_spend_usd_cents, + maybe_last_payment_at, + last_spend_usd_cents, + maybe_days_since_first_payment, + maybe_days_since_last_payment, + net_spend_this_year_usd_cents: net_this_year.clamp(i64::MIN as i128, i64::MAX as i128) as i64, + consecutive_active_weeks, + consecutive_inactive_weeks, + maybe_weeks_since_last_spend, + } +} + +#[cfg(test)] +mod tests { + use chrono::Duration as ChronoDuration; + + use super::*; + + fn event(days_ago: i64, amount_usd_cents: i64, event_type: PaymentEventType, now: DateTime) -> UserSpendEventRow { + UserSpendEventRow { + payment_occurred_at: now - ChronoDuration::days(days_ago), + amount_usd_cents, + event_type, + maybe_credits_granted: None, + } + } + + #[test] + fn aggregates_lifetime_and_cadence() { + let now = Utc::now(); + // ascending by time => furthest in the past first + let events = vec![ + event(40, 1999, PaymentEventType::SubscriptionInitial, now), // ~6 wks ago + event(20, 999, PaymentEventType::CreditPackPurchase, now), // ~3 wks (week 2) + event(5, 2999, PaymentEventType::SubscriptionRenewal, now), // week 0 + event(3, -999, PaymentEventType::Refund, now), // refund, week 0 + ]; + + let agg = compute_summary(&events, now); + + assert_eq!(agg.lifetime_gross_spend_usd_cents, 1999 + 999 + 2999); + assert_eq!(agg.lifetime_subscription_spend_usd_cents, 1999 + 2999); + assert_eq!(agg.lifetime_credits_spend_usd_cents, 999); + assert_eq!(agg.lifetime_refund_usd_cents, 999); + assert_eq!(agg.lifetime_net_spend_usd_cents, (1999 + 999 + 2999 - 999) as u64); + assert_eq!(agg.lifetime_payment_count, 3); + assert_eq!(agg.lifetime_refund_count, 1); + assert_eq!(agg.first_spend_usd_cents, 1999); + assert_eq!(agg.last_spend_usd_cents, 2999); + assert_eq!(agg.maybe_weeks_since_last_spend, Some(0)); + assert_eq!(agg.consecutive_inactive_weeks, 0); + assert_eq!(agg.consecutive_active_weeks, 1); + } + + #[test] + fn lapsed_user_has_inactive_streak() { + let now = Utc::now(); + let events = vec![event(60, 4999, PaymentEventType::SubscriptionInitial, now)]; + let agg = compute_summary(&events, now); + assert_eq!(agg.maybe_weeks_since_last_spend, Some(60 / 7)); + assert_eq!(agg.consecutive_inactive_weeks, (60 / 7) as u32); + assert_eq!(agg.consecutive_active_weeks, 0); + assert_eq!(agg.maybe_days_since_last_payment, Some(60)); + } +} diff --git a/crates/service/job/user_spend_analytics_job/src/job/main_loop.rs b/crates/service/job/user_spend_analytics_job/src/job/main_loop.rs new file mode 100644 index 0000000000..a4d9f41d57 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job/main_loop.rs @@ -0,0 +1,38 @@ +use std::time::Duration; + +use log::{info, warn}; + +use crate::job::backfill_daily_spends::backfill_daily_spends; +use crate::job::backfill_user_summaries::backfill_user_summaries; +use crate::job_dependencies::JobDependencies; + +const SHUTDOWN_POLL_SLICE: Duration = Duration::from_secs(2); + +/// Run the two backfill tasks in order each cycle, then sleep. Errors inside the +/// tasks are paged and skipped (they never abort the cycle). +pub async fn main_loop(deps: JobDependencies) { + while !deps.application_shutdown.get() { + info!("==== Starting spend-analytics cycle ===="); + + backfill_daily_spends(&deps).await; + if deps.application_shutdown.get() { + break; + } + backfill_user_summaries(&deps).await; + + info!("Cycle complete. Sleeping {:?} before next cycle.", deps.sleep_between_cycles); + sleep_interruptible(&deps, deps.sleep_between_cycles).await; + } + + warn!("User spend analytics job main loop is shut down."); +} + +/// Sleep in small slices so a shutdown signal is honored promptly. +async fn sleep_interruptible(deps: &JobDependencies, total: Duration) { + let mut remaining = total; + while remaining > Duration::ZERO && !deps.application_shutdown.get() { + let nap = remaining.min(SHUTDOWN_POLL_SLICE); + tokio::time::sleep(nap).await; + remaining = remaining.saturating_sub(nap); + } +} diff --git a/crates/service/job/user_spend_analytics_job/src/job/mod.rs b/crates/service/job/user_spend_analytics_job/src/job/mod.rs new file mode 100644 index 0000000000..1ba551ebc1 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job/mod.rs @@ -0,0 +1,5 @@ +pub mod alert_on_error; +pub mod backfill_daily_spends; +pub mod backfill_user_summaries; +pub mod main_loop; +pub mod reengagement_score; diff --git a/crates/service/job/user_spend_analytics_job/src/job/reengagement_score.rs b/crates/service/job/user_spend_analytics_job/src/job/reengagement_score.rs new file mode 100644 index 0000000000..f729fec323 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job/reengagement_score.rs @@ -0,0 +1,86 @@ +use crate::job_dependencies::ReengagementConfig; + +const MAX_SCORE: f64 = 1000.0; +const MAX_VALUE_COMPONENT: f64 = 700.0; +const MAX_RECENCY_COMPONENT: f64 = 300.0; + +/// Re-engagement score in `0..=1000`; higher = better re-engagement target. +/// +/// `0` means "not a target": still active, never paid, or no net spend. Otherwise +/// the score is a value component (dominant, log-scaled lifetime net spend) plus a +/// recency component that decays with time since the user lapsed. A big spender who +/// lapsed a while ago still outranks a small spender who just lapsed. +pub fn reengagement_score( + lifetime_net_spend_usd_cents: u64, + maybe_days_since_last_payment: Option, + config: &ReengagementConfig, +) -> u32 { + let days_since_last = match maybe_days_since_last_payment { + Some(days) => days, + None => return 0, // never paid -> not a re-engagement target + }; + + if lifetime_net_spend_usd_cents == 0 { + return 0; + } + if days_since_last < config.active_threshold_days { + return 0; // still active -> no need to re-engage + } + + let dollars = lifetime_net_spend_usd_cents as f64 / 100.0; + let value_component = (MAX_VALUE_COMPONENT * (1.0 + dollars).ln() + / (1.0 + config.value_cap_dollars).ln()) + .clamp(0.0, MAX_VALUE_COMPONENT); + + let days_lapsed = (days_since_last - config.active_threshold_days) as f64; + let recency_fraction = (1.0 - days_lapsed / config.decay_days).max(0.0); + let recency_component = MAX_RECENCY_COMPONENT * recency_fraction; + + (value_component + recency_component).round().clamp(0.0, MAX_SCORE) as u32 +} + +#[cfg(test)] +mod tests { + use super::*; + + const CONFIG: ReengagementConfig = ReengagementConfig { + value_cap_dollars: 1000.0, + active_threshold_days: 14, + decay_days: 350.0, + }; + + #[test] + fn never_paid_scores_zero() { + assert_eq!(reengagement_score(50_000, None, &CONFIG), 0); + } + + #[test] + fn still_active_scores_zero() { + assert_eq!(reengagement_score(50_000, Some(3), &CONFIG), 0); + } + + #[test] + fn zero_net_spend_scores_zero() { + assert_eq!(reengagement_score(0, Some(90), &CONFIG), 0); + } + + #[test] + fn whale_recently_lapsed_outranks_whale_long_lapsed() { + let recent = reengagement_score(100_000, Some(35), &CONFIG); // $1000, ~3wks + let old = reengagement_score(100_000, Some(379), &CONFIG); // $1000, ~1yr + assert!(recent > old, "recent={recent} old={old}"); + assert!(recent > 900, "recent={recent}"); + } + + #[test] + fn whale_long_lapsed_outranks_small_recent() { + let whale_old = reengagement_score(100_000, Some(379), &CONFIG); // $1000, ~1yr -> ~700 + let small_recent = reengagement_score(2_000, Some(35), &CONFIG); // $20, ~3wks + assert!(whale_old > small_recent, "whale_old={whale_old} small_recent={small_recent}"); + } + + #[test] + fn score_never_exceeds_1000() { + assert!(reengagement_score(100_000_000, Some(15), &CONFIG) <= 1000); + } +} diff --git a/crates/service/job/user_spend_analytics_job/src/job_dependencies.rs b/crates/service/job/user_spend_analytics_job/src/job_dependencies.rs new file mode 100644 index 0000000000..f1027b565c --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/job_dependencies.rs @@ -0,0 +1,45 @@ +use std::time::Duration; + +use chrono::NaiveDate; +use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; +use jobs_common::job_stats::JobStats; +use pager::client::pager::Pager; +use server_environment::ServerEnvironment; +use sqlx::MySqlPool; + +/// Tunables for the `maybe_reengagement_score` (0..=1000). Higher = better +/// re-engagement target. See `crate::job::reengagement_score`. +pub struct ReengagementConfig { + /// Lifetime net spend (USD) at which the value component saturates at its max. + pub value_cap_dollars: f64, + /// A user is only a re-engagement target once lapsed this many days. + pub active_threshold_days: u32, + /// Days after the lapse threshold over which the recency bonus decays to zero. + pub decay_days: f64, +} + +pub struct JobDependencies { + pub mysql_pool: MySqlPool, + pub server_environment: ServerEnvironment, + pub job_stats: JobStats, + + /// How long to sleep after a full cycle (daily by default). + pub sleep_between_cycles: Duration, + /// Throttle between individual units of work (dates / users) so we don't hammer the DB. + pub query_delay: Duration, + /// How long to back off after an error before continuing to the next unit. + pub error_recovery: Duration, + + /// Explicit daily-spends backfill start date (UTC). Takes precedence over `maybe_backfill_days`. + pub maybe_backfill_start_date: Option, + /// Look back this many days from today (UTC) if no explicit start date. + pub maybe_backfill_days: Option, + + /// Page size when iterating users for the summaries backfill. + pub summary_user_page_size: i64, + + pub reengagement: ReengagementConfig, + + pub application_shutdown: RelaxedAtomicBool, + pub pager: Pager, +} diff --git a/crates/service/job/user_spend_analytics_job/src/main.rs b/crates/service/job/user_spend_analytics_job/src/main.rs new file mode 100644 index 0000000000..aeeaa3fa42 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/main.rs @@ -0,0 +1,184 @@ +// Never allow these +#![forbid(private_bounds)] +#![forbid(private_interfaces)] +#![forbid(unused_must_use)] + +// Always allow +#![allow(dead_code)] +#![allow(non_snake_case)] + +#[macro_use] extern crate serde_derive; + +use std::time::Duration; + +use anyhow::anyhow; +use chrono::NaiveDate; +use log::{info, warn}; +use sqlx::mysql::MySqlPoolOptions; +use sqlx::Executor; + +use bootstrap::bootstrap::{bootstrap, BootstrapArgs}; +use concurrency::relaxed_atomic_bool::RelaxedAtomicBool; +use errors::AnyhowResult; +use jobs_common::job_stats::JobStats; +use server_environment::ServerEnvironment; +use shared_env_var_config::logging::DEFAULT_RUST_LOG; +use shared_env_var_config::mysql::env_get_mysql_connection_string_or_default; +use tokio::signal::unix::{signal, SignalKind}; + +use crate::http_server::run_http_server::{launch_http_server, CreateServerArgs}; +use crate::job::main_loop::main_loop; +use crate::job_dependencies::{JobDependencies, ReengagementConfig}; +use crate::startup::build_pager::build_pager; + +pub mod http_server; +pub mod job; +pub mod job_dependencies; +pub mod startup; + +#[tokio::main] +async fn main() -> AnyhowResult<()> { + let container_environment = bootstrap(BootstrapArgs { + app_name: "user-spend-analytics-job", + default_logging_override: Some(DEFAULT_RUST_LOG), + config_search_directories: &[".", "./config", "crates/service/job/user_spend_analytics_job/config"], + ignore_legacy_dot_env_file: true, + })?; + + info!("Hostname: {}", &container_environment.hostname); + + let _k8s_node_name = easyenv::get_env_string_optional("K8S_NODE_NAME"); + let _k8s_pod_name = easyenv::get_env_string_optional("K8S_POD_NAME"); + + let db_connection_string = env_get_mysql_connection_string_or_default(); + + info!("Connecting to database..."); + + // Low connection ceiling + a UTC session so day/week bucketing is deterministic. + let mysql_pool = MySqlPoolOptions::new() + .max_connections(2) + .after_connect(|conn, _meta| { + Box::pin(async move { + conn.execute("SET time_zone = '+00:00'").await?; + Ok(()) + }) + }) + .connect(&db_connection_string) + .await?; + + info!("Connected to MySQL."); + + let server_environment = ServerEnvironment::from_str( + &easyenv::get_env_string_required("SERVER_ENVIRONMENT")?, + ) + .ok_or(anyhow!("invalid server environment"))?; + + // ---- Job timing / throttling ---- + let sleep_between_cycles = easyenv::get_env_duration_seconds_or_default( + "SLEEP_BETWEEN_CYCLES_SECONDS", + Duration::from_secs(60 * 60 * 24), // daily + ); + let query_delay = Duration::from_millis(easyenv::get_env_num("QUERY_DELAY_MILLIS", 50)?); + let error_recovery = easyenv::get_env_duration_seconds_or_default( + "ERROR_RECOVERY_SECONDS", + Duration::from_secs(30), + ); + let summary_user_page_size: i64 = easyenv::get_env_num("SUMMARY_USER_PAGE_SIZE", 500)?; + + // ---- Daily-spends backfill window ---- + let maybe_backfill_start_date: Option = easyenv::get_env_string_optional("BACKFILL_START_DATE") + .and_then(|s| NaiveDate::parse_from_str(s.trim(), "%Y-%m-%d").ok()); + let maybe_backfill_days: Option = easyenv::get_env_string_optional("BACKFILL_DAYS") + .and_then(|s| s.trim().parse().ok()); + + // ---- Reengagement score tunables ---- + let reengagement = ReengagementConfig { + value_cap_dollars: easyenv::get_env_string_optional("REENGAGEMENT_VALUE_CAP_DOLLARS") + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(1000.0), + active_threshold_days: easyenv::get_env_num("REENGAGEMENT_ACTIVE_THRESHOLD_DAYS", 14)?, + decay_days: easyenv::get_env_string_optional("REENGAGEMENT_DECAY_DAYS") + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(350.0), + }; + + let (pager, pager_worker) = build_pager(server_environment, &container_environment.hostname); + + info!("Spawning pager worker."); + + // The pager worker uses a blocking Condvar::wait(); it must run on a dedicated OS thread. + std::thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().expect("pager worker tokio runtime"); + rt.block_on(pager_worker.run()); + }); + + let application_shutdown = RelaxedAtomicBool::new(false); + let job_stats = JobStats::new(); + let pager_for_shutdown = pager.clone(); + + let create_server_args = CreateServerArgs { + container_environment: container_environment.clone(), + job_stats: job_stats.clone(), + pager: pager.clone(), + }; + + let job_dependencies = JobDependencies { + mysql_pool, + server_environment, + job_stats, + sleep_between_cycles, + query_delay, + error_recovery, + maybe_backfill_start_date, + maybe_backfill_days, + summary_user_page_size, + reengagement, + application_shutdown: application_shutdown.clone(), + pager, + }; + + // HTTP server runs on a separate OS thread with its own actix System. + std::thread::spawn(move || { + let actix_runtime = actix_web::rt::System::new(); + let http_server_handle = launch_http_server(create_server_args); + + actix_runtime.block_on(http_server_handle) + .expect("HTTP server should not exit."); + + warn!("HTTP server thread is shut down."); + }); + + // Listen for SIGTERM (k8s pod termination) and SIGINT (Ctrl-C) to trigger + // graceful shutdown of the main loop. + let application_shutdown_for_signal = application_shutdown.clone(); + tokio::spawn(async move { + let mut sigterm = match signal(SignalKind::terminate()) { + Ok(stream) => stream, + Err(err) => { + warn!("Failed to install SIGTERM handler: {:?}", err); + return; + } + }; + let mut sigint = match signal(SignalKind::interrupt()) { + Ok(stream) => stream, + Err(err) => { + warn!("Failed to install SIGINT handler: {:?}", err); + return; + } + }; + tokio::select! { + _ = sigterm.recv() => info!("Received SIGTERM. Shutting down..."), + _ = sigint.recv() => info!("Received SIGINT. Shutting down..."), + } + application_shutdown_for_signal.set(true); + }); + + main_loop(job_dependencies).await; + + info!("Shutting down pager worker..."); + pager_for_shutdown.shutdown_worker(); + + info!("User spend analytics job exiting."); + + Ok(()) +} diff --git a/crates/service/job/user_spend_analytics_job/src/startup/build_pager.rs b/crates/service/job/user_spend_analytics_job/src/startup/build_pager.rs new file mode 100644 index 0000000000..6805590260 --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/startup/build_pager.rs @@ -0,0 +1,65 @@ +use log::{info, warn}; +use pager::client::pager::Pager; +use pager::client::pager_builder::PagerBuilder; +use pager::worker::pager_worker::PagerWorker; +use rootly_client::creds::rootly_api_key::RootlyApiKey; +use rootly_config::services::ROOTLY_SERVICE_ID_USER_SPEND_ANALYTICS_JOB; +use rootly_config::urgencies::{ROOTLY_URGENCY_ID_HIGH, ROOTLY_URGENCY_ID_LOW, ROOTLY_URGENCY_ID_MEDIUM}; +use shared_env_var_config::paging::{env_enable_paging_default_false, env_optional_rootly_api_key, env_optional_rootly_notification_target_id, env_optional_rootly_notification_target_type}; + +pub fn build_pager( + server_environment: server_environment::ServerEnvironment, + hostname: &str, +) -> (Pager, PagerWorker) { + let is_paging_enabled = env_enable_paging_default_false(); + + info!("Paging enabled: {}", is_paging_enabled); + + let environment = if server_environment.is_deployed_in_production() { + "production" + } else { + "development" + }; + + let builder = PagerBuilder::new() + .application_name("user-spend-analytics-job".to_string()) + .environment(environment.to_string()) + .hostname(hostname.to_string()) + .service_id(ROOTLY_SERVICE_ID_USER_SPEND_ANALYTICS_JOB.to_string()); + + // If paging is globally disabled, use a NoOp pager regardless of API key. + if !is_paging_enabled { + warn!("ENABLE_PAGING is false. Pager will be NoOp."); + return builder.build_with_worker(); + } + + let maybe_api_key = env_optional_rootly_api_key(); + + match maybe_api_key { + Some(api_key) => { + info!("Rootly API key found. Configuring pager with Rootly backend."); + build_rootly_pager(builder, api_key) + } + None => { + warn!("ROOTLY_API_KEY not set. Pager will not send real pages."); + builder.build_with_worker() + } + } +} + +fn build_rootly_pager(builder: PagerBuilder, api_key: String) -> (Pager, PagerWorker) { + let mut rootly_builder = builder + .rootly(RootlyApiKey::new(api_key)) + .urgency_id_high(ROOTLY_URGENCY_ID_HIGH.to_string()) + .urgency_id_medium(ROOTLY_URGENCY_ID_MEDIUM.to_string()) + .urgency_id_low(ROOTLY_URGENCY_ID_LOW.to_string()); + + let target_type = env_optional_rootly_notification_target_type(); + let target_id = env_optional_rootly_notification_target_id(); + + if let (Some(t_type), Some(t_id)) = (target_type, target_id) { + rootly_builder = rootly_builder.notification_target(t_type, t_id); + } + + rootly_builder.build_with_worker() +} diff --git a/crates/service/job/user_spend_analytics_job/src/startup/mod.rs b/crates/service/job/user_spend_analytics_job/src/startup/mod.rs new file mode 100644 index 0000000000..d074c6c0ce --- /dev/null +++ b/crates/service/job/user_spend_analytics_job/src/startup/mod.rs @@ -0,0 +1 @@ +pub mod build_pager;