Skip to main content

max / makenotwork

5.7 KB · 145 lines History Blame Raw
1 //! Background fan-out of a Stripe subscription operation across all of a
2 //! creator's fan subscriptions.
3 //!
4 //! Pausing, suspending, or terminating a creator requires one Stripe call per
5 //! fan subscription. Run inline on a request or webhook handler, that serial
6 //! fan-out ties the hot path up for the duration of N Stripe round-trips,
7 //! minutes for a large creator, and on the webhook path long enough to trip
8 //! Stripe's delivery timeout and trigger a retry storm that re-runs the loop
9 //! (ultra-fuzz Run 6 S1/S2/R6-Perf-M1). This module moves the loop onto the
10 //! bounded background queue so the handler returns immediately. The operations
11 //! are fire-and-forget, errors were only logged inline, and still are.
12
13 use std::sync::Arc;
14
15 use crate::background::BackgroundTx;
16 use crate::db::StripeAccountId;
17 use crate::payments::PaymentProvider;
18
19 /// Which Stripe subscription operation to apply to each fan subscription.
20 #[derive(Clone, Copy, Debug)]
21 pub enum FanSubOp {
22 /// Set (`true`) or clear (`false`) cancel_at_period_end, creator pause/unpause.
23 CancelAtPeriodEnd(bool),
24 /// Pause collection, admin suspend.
25 Pause,
26 /// Resume collection, admin unsuspend.
27 Resume,
28 /// Cancel outright, admin terminate.
29 Cancel,
30 }
31
32 impl FanSubOp {
33 fn label(self) -> &'static str {
34 match self {
35 FanSubOp::CancelAtPeriodEnd(true) => "set_cancel_at_period_end",
36 FanSubOp::CancelAtPeriodEnd(false) => "clear_cancel_at_period_end",
37 FanSubOp::Pause => "pause",
38 FanSubOp::Resume => "resume",
39 FanSubOp::Cancel => "cancel",
40 }
41 }
42
43 async fn apply(
44 self,
45 stripe: &Arc<dyn PaymentProvider>,
46 sub_id: &str,
47 account_id: &str,
48 ) -> crate::error::Result<()> {
49 match self {
50 FanSubOp::CancelAtPeriodEnd(c) => {
51 stripe.set_cancel_at_period_end(sub_id, account_id, c).await
52 }
53 FanSubOp::Pause => stripe.pause_subscription(sub_id, account_id).await,
54 FanSubOp::Resume => stripe.resume_subscription(sub_id, account_id).await,
55 FanSubOp::Cancel => stripe.cancel_subscription(sub_id, account_id).await,
56 }
57 }
58 }
59
60 /// Apply `op` to every subscription in `sub_ids` on the background queue. Returns
61 /// immediately; the loop runs off the request/webhook hot path. Per-subscription
62 /// failures are logged and do not abort the rest. No-op for an empty list.
63 ///
64 /// A failed op is not merely logged: any failures open a WAM ticket (when `wam`
65 /// is configured) so a dropped pause/suspend/cancel, a fan still charged after a
66 /// creator pause, or a suspended creator's fans retaining access, is actively
67 /// surfaced for manual reconciliation rather than lost in the logs (audit Run 13
68 /// Resilience: fan-out had no dead-letter). The failed subscription IDs are
69 /// listed in the ticket body.
70 /// Apply `op` to every subscription in `sub_ids`, awaiting each call inline.
71 ///
72 /// For contexts with no background queue, the `mnw-admin` CLI, where blocking on
73 /// the serial Stripe fan-out is fine because there is no request or webhook hot
74 /// path to protect. Per-subscription failures are logged and do not abort the
75 /// rest; returns the number that failed so the caller can surface it. No-op (and
76 /// returns 0) for an empty list.
77 pub async fn run_fan_sub_fanout(
78 stripe: &Arc<dyn PaymentProvider>,
79 account_id: &StripeAccountId,
80 sub_ids: &[String],
81 op: FanSubOp,
82 ) -> usize {
83 let mut failed = 0usize;
84 for sub_id in sub_ids {
85 if let Err(e) = op.apply(stripe, sub_id, account_id.as_str()).await {
86 failed += 1;
87 tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed (inline)");
88 }
89 }
90 if failed > 0 {
91 tracing::warn!(
92 total = sub_ids.len(),
93 failed,
94 op = op.label(),
95 "inline fan subscription fan-out completed with failures"
96 );
97 }
98 failed
99 }
100
101 pub fn spawn_fan_sub_fanout(
102 bg: &BackgroundTx,
103 stripe: Arc<dyn PaymentProvider>,
104 account_id: StripeAccountId,
105 sub_ids: Vec<String>,
106 op: FanSubOp,
107 wam: Option<crate::wam_client::WamClient>,
108 ) {
109 if sub_ids.is_empty() {
110 return;
111 }
112 bg.spawn("fan-sub stripe fan-out", async move {
113 let total = sub_ids.len();
114 let mut failed_ids: Vec<&str> = Vec::new();
115 for sub_id in &sub_ids {
116 if let Err(e) = op.apply(&stripe, sub_id, account_id.as_str()).await {
117 failed_ids.push(sub_id);
118 tracing::warn!(stripe_sub_id = %sub_id, op = op.label(), error = ?e, "fan subscription op failed");
119 }
120 }
121 if failed_ids.is_empty() {
122 tracing::debug!(total, op = op.label(), "fan subscription fan-out completed");
123 return;
124 }
125 let failed = failed_ids.len();
126 tracing::warn!(total, failed, op = op.label(), "fan subscription fan-out completed with failures");
127 if let Some(wam) = wam {
128 let title = format!(
129 "Fan subscription fan-out incomplete: {} of {total} '{}' ops failed",
130 failed,
131 op.label()
132 );
133 let body = format!(
134 "Applying '{}' to a creator's fan subscriptions (account {}) left {failed} of {total} \
135 unreconciled. These subscriptions may still be charging (or retaining access) against \
136 the creator's current state and need manual reconciliation in Stripe.\n\nFailed subscription IDs:\n{}",
137 op.label(),
138 account_id.as_str(),
139 failed_ids.join("\n"),
140 );
141 wam.create_ticket(&title, Some(&body), "high", "fan-fanout-incomplete", Some(account_id.as_str())).await;
142 }
143 });
144 }
145