max / makenotwork

8.2 KB · 216 lines History Blame Raw

1	//! Shared moderation service: the single implementation of creator suspension,
2	//! unsuspension, and appeal decisions, called by BOTH the web admin handlers
3	//! (`routes::admin`) and the `mnw-admin` CLI.
4	//!
5	//! Before this module existed the CLI (`cmd_suspend` / `cmd_unsuspend` /
6	//! `cmd_decide`) called `db::users::suspend_user` directly and skipped the audit
7	//! record, the notification email, session revocation, and the Stripe
8	//! fan-subscription pause/resume that the web path performs. A CLI-issued
9	//! suspension therefore produced no dashboard-visible reason, no email, and left
10	//! fans being charged, silently diverging from the documented due-process
11	//! guarantees. Routing both callers through these functions makes that divergence
12	//! unrepresentable: the due-process steps live in one place.
13	//!
14	//! The only runtime difference between the two callers is captured by
15	//! [`FanoutMode`]: a web request fans the per-subscription Stripe calls out on the
16	//! background queue (and evicts revoked sessions from the in-memory cache), while
17	//! the CLI runs them inline (and has no session cache to touch).
18
19	use std::sync::Arc;
20	use std::time::Instant;
21
22	use dashmap::DashMap;
23
24	use crate::auth::AdminId;
25	use crate::background::BackgroundTx;
26	use crate::db::{self, DbUser, ModerationActionType, StripeAccountId, UserSessionId};
27	use crate::email::EmailClient;
28	use crate::error::Result;
29	use crate::payments::PaymentProvider;
30	use crate::payments::fan_ops::{self, FanSubOp};
31	use crate::wam_client::WamClient;
32
33	/// How to run the per-fan-subscription Stripe fan-out, and whether an in-memory
34	/// session cache exists to evict from.
35	pub enum FanoutMode<'a> {
36	/// Server request/handler context. Stripe ops fan out on the bounded
37	/// background queue so the handler returns without blocking on N round-trips;
38	/// revoked sessions are also evicted from the live in-memory cache to close
39	/// the session-touch-cache window.
40	Background {
41	bg: &'a BackgroundTx,
42	wam: Option<WamClient>,
43	session_cache: &'a Arc<DashMap<UserSessionId, Instant>>,
44	},
45	/// CLI context (`mnw-admin`). No background queue: Stripe ops run inline
46	/// (blocking is fine off any hot path). No in-memory cache to evict, deleting
47	/// the session rows in the DB is sufficient, since the running server
48	/// re-validates against the DB within the touch-cache TTL.
49	Inline,
50	}
51
52	impl FanoutMode<'_> {
53	/// Fan `op` across `sub_ids` for `account_id` using this mode's mechanism.
54	async fn apply_subscription_op(
55	&self,
56	stripe: &Arc<dyn PaymentProvider>,
57	account_id: &StripeAccountId,
58	sub_ids: Vec<String>,
59	op: FanSubOp,
60	) {
61	match self {
62	FanoutMode::Background { bg, wam, .. } => fan_ops::spawn_fan_sub_fanout(
63	bg,
64	Arc::clone(stripe),
65	account_id.clone(),
66	sub_ids,
67	op,
68	wam.clone(),
69	),
70	FanoutMode::Inline => {
71	fan_ops::run_fan_sub_fanout(stripe, account_id, &sub_ids, op).await;
72	}
73	}
74	}
75
76	/// Evict revoked session tracking IDs from the in-memory cache (server only;
77	/// no-op for the CLI, which has no such cache).
78	fn evict_sessions(&self, revoked: &[UserSessionId]) {
79	if let FanoutMode::Background { session_cache, .. } = self {
80	for sid in revoked {
81	session_cache.remove(sid);
82	}
83	}
84	}
85	}
86
87	/// Suspend a creator: record the reason, revoke sessions, write the audit action,
88	/// pause fan subscriptions (DB + Stripe), and email the creator.
89	///
90	/// `target` must be the already-fetched account. The caller is responsible for the
91	/// "already suspended?" short-circuit and for validating a non-empty `reason`.
92	pub async fn suspend_creator(
93	db: &sqlx::PgPool,
94	email: &EmailClient,
95	stripe: Option<&Arc<dyn PaymentProvider>>,
96	fanout: FanoutMode<'_>,
97	target: &DbUser,
98	admin_id: AdminId,
99	reason: &str,
100	) -> Result<()> {
101	db::users::suspend_user(db, target.id, reason).await?;
102
103	// Revoke every session so the suspended creator can't keep acting during the
104	// touch-cache window.
105	let revoked = db::sessions::delete_all_sessions_for_user(db, target.id).await?;
106	fanout.evict_sessions(&revoked);
107	if !revoked.is_empty() {
108	tracing::info!(user_id = %target.id, revoked = revoked.len(), "revoked sessions on suspension");
109	}
110
111	// Append-only audit record, this is what surfaces the reason on the creator's
112	// own dashboard.
113	db::moderation::create_action(
114	db,
115	target.id,
116	admin_id,
117	ModerationActionType::Suspension,
118	reason,
119	None,
120	)
121	.await?;
122
123	// Pause fan subscriptions so fans aren't charged while the creator is
124	// suspended: DB state plus the matching Stripe pause.
125	if let (Some(stripe), Some(account_id)) = (stripe, target.stripe_account_id.as_ref()) {
126	let subs = db::subscriptions::get_active_subscriptions_by_creator(db, target.id).await?;
127	let ids: Vec<String> = subs.into_iter().map(\|s\| s.stripe_subscription_id).collect();
128	fanout
129	.apply_subscription_op(stripe, account_id, ids, FanSubOp::Pause)
130	.await;
131	let paused = db::subscriptions::pause_subscriptions_for_creator(db, target.id).await?;
132	tracing::info!(user_id = %target.id, db_paused = paused, "paused fan subscriptions for suspended creator");
133	}
134
135	if let Err(e) = email
136	.send_suspension_notification(&target.email, target.display_name.as_deref(), reason)
137	.await
138	{
139	tracing::error!(error = ?e, user_id = %target.id, "failed to send suspension email");
140	}
141
142	tracing::info!(user_id = %target.id, admin_id = %admin_id.get(), reason = %reason, "suspended creator");
143	Ok(())
144	}
145
146	/// Lift a creator's suspension: clear it, resolve the audit action, and resume fan
147	/// subscriptions (DB + Stripe). No email, mirroring the web admin path.
148	pub async fn unsuspend_creator(
149	db: &sqlx::PgPool,
150	stripe: Option<&Arc<dyn PaymentProvider>>,
151	fanout: FanoutMode<'_>,
152	target: &DbUser,
153	) -> Result<()> {
154	db::users::unsuspend_user(db, target.id).await?;
155	db::moderation::resolve_actions_by_type(db, target.id, ModerationActionType::Suspension)
156	.await?;
157
158	if let (Some(stripe), Some(account_id)) = (stripe, target.stripe_account_id.as_ref()) {
159	let resumed = db::subscriptions::resume_subscriptions_for_creator(db, target.id).await?;
160	let ids: Vec<String> = resumed
161	.into_iter()
162	.map(\|s\| s.stripe_subscription_id)
163	.collect();
164	fanout
165	.apply_subscription_op(stripe, account_id, ids, FanSubOp::Resume)
166	.await;
167	}
168
169	tracing::info!(user_id = %target.id, "unsuspended creator");
170	Ok(())
171	}
172
173	/// Resolve a suspension appeal. On approval this clears the suspension (via
174	/// `resolve_appeal`) and resumes fan subscriptions; either way the creator is
175	/// emailed the outcome. The caller validates a non-empty `response`.
176	pub async fn decide_appeal(
177	db: &sqlx::PgPool,
178	email: &EmailClient,
179	stripe: Option<&Arc<dyn PaymentProvider>>,
180	fanout: FanoutMode<'_>,
181	target: &DbUser,
182	decision: db::AppealDecision,
183	response: &str,
184	) -> Result<()> {
185	db::users::resolve_appeal(db, target.id, decision, response).await?;
186
187	if decision == db::AppealDecision::Approved
188	&& let (Some(stripe), Some(account_id)) = (stripe, target.stripe_account_id.as_ref())
189	{
190	let resumed = db::subscriptions::resume_subscriptions_for_creator(db, target.id).await?;
191	let ids: Vec<String> = resumed
192	.into_iter()
193	.map(\|s\| s.stripe_subscription_id)
194	.collect();
195	fanout
196	.apply_subscription_op(stripe, account_id, ids, FanSubOp::Resume)
197	.await;
198	}
199
200	let decision_str = decision.to_string();
201	if let Err(e) = email
202	.send_appeal_decision(
203	&target.email,
204	target.display_name.as_deref(),
205	&decision_str,
206	response,
207	)
208	.await
209	{
210	tracing::error!(error = ?e, user_id = %target.id, "failed to send appeal decision email");
211	}
212
213	tracing::info!(user_id = %target.id, decision = %decision_str, "decided appeal");
214	Ok(())
215	}
216