mirror of
https://github.com/storytold/artcraft.git
synced 2026-11-03 00:47:53 +00:00
Brandon Thomas authored and 
GitHub committed
1 parent
3e4f998599
commit
d99da56df0
38 files changed
+2022
No files matched your search
+58
@@ -0,0 +1,58 @@
|
||||
{
|
||||
"db_name": "MySQL",
|
||||
"query": "\nSELECT\n payment_occurred_at AS `payment_occurred_at!: DateTime<Utc>`,\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<Utc>",
|
||||
"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"
|
||||
}
|
||||
+113
@@ -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"
|
||||
}
|
||||
+12
@@ -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"
|
||||
}
|
||||
+25
@@ -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"
|
||||
}
|
||||
+25
@@ -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"
|
||||
}
|
||||
+12
@@ -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"
|
||||
}
|
||||
+47
@@ -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<Utc>`\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<Utc>",
|
||||
"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"
|
||||
}
|
||||
Generated
+30
@@ -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"
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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";
|
||||
@@ -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;
|
||||
|
||||
+76
@@ -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<Utc>,
|
||||
/// Exclusive end of the UTC day (start + 1 day).
|
||||
pub range_end: DateTime<Utc>,
|
||||
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<Vec<DailySpendAggregate>, 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)
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
pub mod aggregate_daily_spends_for_date;
|
||||
pub mod upsert_user_daily_spend;
|
||||
+73
@@ -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(())
|
||||
}
|
||||
+34
@@ -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<Vec<PaymentsNamespace>, 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)
|
||||
}
|
||||
+55
@@ -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<Utc>,
|
||||
pub amount_usd_cents: i64,
|
||||
pub event_type: PaymentEventType,
|
||||
pub maybe_credits_granted: Option<u32>,
|
||||
}
|
||||
|
||||
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<Vec<UserSpendEventRow>, 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<Utc>`,
|
||||
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)
|
||||
}
|
||||
+46
@@ -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<Vec<UserToken>, 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)
|
||||
}
|
||||
@@ -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;
|
||||
+129
@@ -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<DateTime<Utc>>,
|
||||
pub first_spend_usd_cents: u64,
|
||||
pub maybe_last_payment_at: Option<DateTime<Utc>>,
|
||||
pub last_spend_usd_cents: u64,
|
||||
pub maybe_days_since_first_payment: Option<u32>,
|
||||
pub maybe_days_since_last_payment: Option<u32>,
|
||||
|
||||
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<u32>,
|
||||
|
||||
pub is_active_subscriber: bool,
|
||||
pub maybe_subscription_interval: Option<&'e str>,
|
||||
pub maybe_reengagement_score: Option<u32>,
|
||||
|
||||
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(())
|
||||
}
|
||||
+55
@@ -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<String>,
|
||||
pub maybe_stripe_recurring_interval: Option<String>,
|
||||
pub subscription_expires_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
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<Option<UserSubscriptionStatusRow>, 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<Utc>`
|
||||
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)
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
[package]
|
||||
name = "user-spend-analytics-job"
|
||||
edition = "2021"
|
||||
version = "0.0.1"
|
||||
authors = [
|
||||
"Brandon Thomas <bt@brand.io>",
|
||||
"Brandon Thomas <echelon@gmail.com>",
|
||||
]
|
||||
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
|
||||
+22
@@ -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
|
||||
+127
@@ -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<Arc<HttpServerSharedState>>,
|
||||
) -> Result<HttpResponse, HealthCheckError> {
|
||||
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))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
pub mod health_check_handler;
|
||||
+10
@@ -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,
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
pub mod endpoints;
|
||||
pub mod http_server_shared_state;
|
||||
pub mod run_http_server;
|
||||
@@ -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<Server> {
|
||||
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(())
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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<usize, sqlx::Error> {
|
||||
let range_start: DateTime<Utc> =
|
||||
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")
|
||||
}
|
||||
@@ -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<Utc>) -> 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<Utc>,
|
||||
) -> 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<DateTime<Utc>>,
|
||||
first_spend_usd_cents: u64,
|
||||
maybe_last_payment_at: Option<DateTime<Utc>>,
|
||||
last_spend_usd_cents: u64,
|
||||
maybe_days_since_first_payment: Option<u32>,
|
||||
maybe_days_since_last_payment: Option<u32>,
|
||||
net_spend_this_year_usd_cents: i64,
|
||||
consecutive_active_weeks: u32,
|
||||
consecutive_inactive_weeks: u32,
|
||||
maybe_weeks_since_last_spend: Option<u32>,
|
||||
}
|
||||
|
||||
/// 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<Utc>) -> 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<i64> = 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<Utc>) -> 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));
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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<u32>,
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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<NaiveDate>,
|
||||
/// Look back this many days from today (UTC) if no explicit start date.
|
||||
pub maybe_backfill_days: Option<i64>,
|
||||
|
||||
/// 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,
|
||||
}
|
||||
@@ -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<NaiveDate> = 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<i64> = 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(())
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
pub mod build_pager;
|
||||
Reference in new issue
Block a user