Skip to main content

max / makenotwork

6.3 KB · 176 lines History Blame Raw
1 //! Periodic integrity checks: sales count drift, stale subscriptions, bounce spikes,
2 //! storage recalculation, and post-grace enforcement.
3
4 use crate::AppState;
5 use crate::db;
6
7 /// Weekly storage drift correction, batch recalculates storage_used_bytes for all creators.
8 #[tracing::instrument(skip_all, name = "scheduler::recalculate_all_storage_used")]
9 pub(super) async fn recalculate_all_storage_used(state: &AppState) {
10 match db::creator_tiers::recalculate_all_storage_batch(&state.db).await {
11 Ok(corrected) => {
12 if corrected > 0 {
13 tracing::info!(
14 corrected = corrected,
15 "weekly storage drift correction complete"
16 );
17 }
18 }
19 Err(e) => {
20 tracing::error!(error = ?e, "weekly storage drift correction failed");
21 }
22 }
23 }
24
25 /// Enforce post-grace item hiding for creators whose cancellation grace period has expired.
26 #[tracing::instrument(skip_all, name = "scheduler::enforce_post_grace_hiding")]
27 pub(super) async fn enforce_post_grace_hiding(state: &AppState) {
28 let user_ids = match db::creator_tiers::get_expired_grace_creators(&state.db).await {
29 Ok(ids) => ids,
30 Err(e) => {
31 tracing::error!(error = ?e, "failed to query expired grace creators");
32 return;
33 }
34 };
35 if user_ids.is_empty() {
36 return;
37 }
38
39 // Set-based hide + mark in two statements instead of 2N per-creator round-trips
40 // inline on the lock-held tick (Perf-S4, Run 9). Both are idempotent, so if the
41 // mark fails after the hide the next sweep re-hides (a no-op) and re-marks.
42 let hidden = match db::items::hide_all_items_for_users(&state.db, &user_ids).await {
43 Ok(count) => count,
44 Err(e) => {
45 tracing::error!(error = ?e, "failed to hide items for post-grace enforcement");
46 return;
47 }
48 };
49 if let Err(e) = db::creator_tiers::mark_grace_enforced_batch(&state.db, &user_ids).await {
50 tracing::error!(error = ?e, "failed to mark grace enforced");
51 return;
52 }
53 if hidden > 0 {
54 tracing::info!(
55 creators = user_ids.len(),
56 items_hidden = hidden,
57 "post-grace enforcement: items hidden"
58 );
59 }
60 }
61
62 /// Detect items where denormalized sales_count has drifted from actual transaction count.
63 #[tracing::instrument(skip_all, name = "scheduler::check_sales_count_drift")]
64 pub(super) async fn check_sales_count_drift(state: &AppState) {
65 // Pre-filter to items that either have a non-zero recorded count OR have
66 // at least one completed transaction. The previous query GROUPed every
67 // item in the platform's history; on a mature DB this multi-minute scan
68 // pins a pool connection. `EXISTS ... LIMIT 1` short-circuits per item.
69 let rows = match sqlx::query_as::<_, (db::ItemId, i32, i64)>(
70 r"
71 SELECT i.id, i.sales_count, COUNT(t.id)
72 FROM items i
73 LEFT JOIN transactions t ON t.item_id = i.id AND t.status = 'completed'
74 WHERE i.sales_count > 0
75 OR EXISTS (SELECT 1 FROM transactions WHERE item_id = i.id AND status = 'completed' LIMIT 1)
76 GROUP BY i.id
77 HAVING i.sales_count != COUNT(t.id)
78 LIMIT 50
79 ",
80 )
81 .fetch_all(&state.db)
82 .await
83 {
84 Ok(r) if r.is_empty() => return,
85 Ok(r) => r,
86 Err(e) => {
87 tracing::error!(error = ?e, "sales count drift check failed");
88 return;
89 }
90 };
91
92 tracing::warn!(count = rows.len(), "sales count drift detected");
93
94 if let Some(ref wam) = state.wam {
95 let items: Vec<String> = rows
96 .iter()
97 .map(|(id, cached, actual)| format!(" {id}: cached={cached}, actual={actual}"))
98 .collect();
99 let body = format!("Items with drifted sales_count:\n{}", items.join("\n"));
100 wam.create_ticket(
101 &format!("Sales count drift: {} items", rows.len()),
102 Some(&body),
103 "medium",
104 "sales-count-drift",
105 None,
106 )
107 .await;
108 }
109 }
110
111 /// Find subscriptions stuck in past_due for >7 days (possible missed webhook).
112 #[tracing::instrument(skip_all, name = "scheduler::check_stale_subscriptions")]
113 pub(super) async fn check_stale_subscriptions(state: &AppState) {
114 let count: i64 = match sqlx::query_scalar(
115 r"
116 SELECT COUNT(*) FROM (
117 SELECT 1 FROM creator_subscriptions WHERE status = 'past_due' AND current_period_end < NOW() - INTERVAL '7 days'
118 UNION ALL
119 SELECT 1 FROM subscriptions WHERE status = 'past_due' AND current_period_end < NOW() - INTERVAL '7 days'
120 ) stale
121 ",
122 )
123 .fetch_one(&state.db)
124 .await
125 {
126 Ok(c) => c,
127 Err(e) => {
128 tracing::error!(error = ?e, "stale subscription check failed");
129 return;
130 }
131 };
132
133 if count > 0 {
134 tracing::warn!(count, "stale past_due subscriptions detected");
135 if let Some(ref wam) = state.wam {
136 wam.create_ticket(
137 &format!("{count} subscriptions past_due >7 days"),
138 Some("Subscriptions stuck in past_due for over 7 days. A Stripe webhook may have been missed. Check the Stripe dashboard."),
139 "medium",
140 "subscription-stale-past-due",
141 None,
142 ).await;
143 }
144 }
145 }
146
147 /// Detect email bounce/complaint spikes (>10 suppressions in 24h).
148 #[tracing::instrument(skip_all, name = "scheduler::check_email_bounce_spike")]
149 pub(super) async fn check_email_bounce_spike(state: &AppState) {
150 let count: i64 = match sqlx::query_scalar(
151 "SELECT COUNT(*) FROM email_suppressions WHERE created_at > NOW() - INTERVAL '24 hours'",
152 )
153 .fetch_one(&state.db)
154 .await
155 {
156 Ok(c) => c,
157 Err(e) => {
158 tracing::error!(error = ?e, "email bounce spike check failed");
159 return;
160 }
161 };
162
163 if count > 10 {
164 tracing::warn!(count, "email bounce/complaint spike");
165 if let Some(ref wam) = state.wam {
166 wam.create_ticket(
167 &format!("Email bounce spike: {count} suppressions in 24h"),
168 Some("Elevated bounce/complaint rate may indicate a deliverability problem. Check Postmark dashboard."),
169 "high",
170 "email-bounce-spike",
171 None,
172 ).await;
173 }
174 }
175 }
176