Skip to main content

max / makenotwork

14.1 KB · 416 lines History Blame Raw
1 //! Admin user management: listing, suspension, trust status.
2
3 use crate::extractors::ValidatedQuery;
4 use axum::{
5 Form,
6 extract::{Path, State},
7 response::{IntoResponse, Response},
8 };
9 use serde::Deserialize;
10 use sqlx::PgPool;
11
12 use crate::{
13 AppCaches, Billing, Integrations,
14 auth::AdminUser,
15 background::BackgroundTx,
16 db::{self, ModerationActionType, UserId},
17 email::EmailClient,
18 error::{AppError, Result},
19 helpers::get_csrf_token,
20 templates::{AdminUserEntriesTemplate, AdminUsersTemplate},
21 types::AdminUserRow,
22 };
23
24 #[derive(Debug, Deserialize)]
25 pub(super) struct UserFilterQuery {
26 pub status: Option<String>,
27 pub page: Option<i64>,
28 }
29
30 /// Render the admin user management page.
31 #[tracing::instrument(skip_all, name = "admin::admin_users")]
32 pub(super) async fn admin_users(
33 State(db): State<PgPool>,
34 session: tower_sessions::Session,
35 AdminUser(user): AdminUser,
36 ValidatedQuery(query): ValidatedQuery<UserFilterQuery>,
37 ) -> Result<impl IntoResponse> {
38 let csrf_token = get_csrf_token(&session).await;
39 let current_filter = query.status.clone().unwrap_or_default();
40
41 // Upper-clamp page so `OFFSET = (page-1)*per_page` doesn't overflow i64
42 // or produce a sqlx "value out of range" 500. 1e9 pages × 50 per_page is
43 // already 50 billion rows, well past anything the admin panel will ever
44 // reach, and keeps the OFFSET safely inside i64.
45 let page = query.page.unwrap_or(1).clamp(1, 1_000_000_000);
46 let per_page: i64 = 50;
47 let offset = (page - 1) * per_page;
48
49 let (total_users_i64, total_suspended_i64) = db::users::count_users_summary(&db).await?;
50 let total_users = total_users_i64 as usize;
51 let total_suspended = total_suspended_i64 as usize;
52
53 let total_count = match query.status.as_deref() {
54 Some("suspended") => total_suspended_i64,
55 Some("active") => total_users_i64 - total_suspended_i64,
56 Some(f @ ("custom_pages" | "pages_locked")) => db::users::count_users(&db, Some(f)).await?,
57 _ => total_users_i64,
58 };
59 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
60
61 let db_users = db::users::get_all_users(&db, query.status.as_deref(), per_page, offset).await?;
62
63 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
64
65 Ok(AdminUsersTemplate {
66 csrf_token,
67 session_user: Some(user),
68 users,
69 total_users,
70 total_suspended,
71 current_filter,
72 current_page: page,
73 total_pages,
74 admin_active_page: "users",
75 })
76 }
77
78 /// Return filtered user entries as an HTMX partial.
79 #[tracing::instrument(skip_all, name = "admin::admin_user_entries")]
80 pub(super) async fn admin_user_entries(
81 State(db): State<PgPool>,
82 AdminUser(_user): AdminUser,
83 ValidatedQuery(query): ValidatedQuery<UserFilterQuery>,
84 ) -> Result<impl IntoResponse> {
85 let current_filter = query.status.clone().unwrap_or_default();
86 // Upper-clamp page so `OFFSET = (page-1)*per_page` doesn't overflow i64
87 // or produce a sqlx "value out of range" 500. 1e9 pages × 50 per_page is
88 // already 50 billion rows, well past anything the admin panel will ever
89 // reach, and keeps the OFFSET safely inside i64.
90 let page = query.page.unwrap_or(1).clamp(1, 1_000_000_000);
91 let per_page: i64 = 50;
92 let offset = (page - 1) * per_page;
93
94 let total_count = db::users::count_users(&db, query.status.as_deref()).await?;
95 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
96
97 let db_users = db::users::get_all_users(&db, query.status.as_deref(), per_page, offset).await?;
98 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
99 Ok(AdminUserEntriesTemplate {
100 users,
101 current_page: page,
102 total_pages,
103 current_filter,
104 })
105 }
106
107 #[derive(Debug, Deserialize)]
108 pub(super) struct SuspendForm {
109 pub reason: String,
110 }
111
112 /// Send a policy warning to a user without suspending their account.
113 /// Records the warning in moderation history and emails the user.
114 #[tracing::instrument(skip_all, name = "admin::admin_warn_user")]
115 pub(super) async fn admin_warn_user(
116 State(db): State<PgPool>,
117 State(email): State<EmailClient>,
118 admin_user: AdminUser,
119 Path(id): Path<UserId>,
120 Form(form): Form<SuspendForm>,
121 ) -> Result<impl IntoResponse> {
122 let reason = form.reason.trim();
123 if reason.is_empty() {
124 return Err(AppError::validation("Reason is required".to_string()));
125 }
126
127 let db_user = db::users::get_user_by_id(&db, id)
128 .await?
129 .ok_or(AppError::NotFound)?;
130
131 // Record warning in moderation history
132 db::moderation::create_action(
133 &db,
134 id,
135 admin_user.admin_id(),
136 ModerationActionType::Warning,
137 reason,
138 None,
139 )
140 .await?;
141
142 // Send warning email
143 if let Err(e) = email
144 .send_policy_warning(&db_user.email, db_user.display_name.as_deref(), reason)
145 .await
146 {
147 tracing::error!(error = ?e, user_id = %id, "failed to send warning email");
148 }
149
150 tracing::info!(user_id = %id, admin_id = %admin_user.id(), reason = %reason, "admin sent policy warning");
151
152 refresh_user_entries_partial(&db).await
153 }
154
155 /// Suspend a user account and send notification email.
156 #[tracing::instrument(skip_all, name = "admin::admin_suspend_user")]
157 #[allow(clippy::too_many_arguments)]
158 pub(super) async fn admin_suspend_user(
159 State(db): State<PgPool>,
160 State(email): State<EmailClient>,
161 State(bg): State<BackgroundTx>,
162 State(caches): State<AppCaches>,
163 State(payments): State<Billing>,
164 State(integrations): State<Integrations>,
165 admin_user: AdminUser,
166 Path(id): Path<UserId>,
167 Form(form): Form<SuspendForm>,
168 ) -> Result<impl IntoResponse> {
169 let reason = form.reason.trim();
170 if reason.is_empty() {
171 return Err(AppError::validation("Reason is required".to_string()));
172 }
173
174 // Get user for email notification
175 let db_user = db::users::get_user_by_id(&db, id)
176 .await?
177 .ok_or(AppError::NotFound)?;
178
179 // Delegate to the shared moderation service so the web path and the
180 // `mnw-admin` CLI perform the exact same due process (audit record, session
181 // revocation, fan-sub pause, email). Web fans the Stripe calls out on the
182 // background queue and evicts the in-memory session cache.
183 super::moderation_service::suspend_creator(
184 &db,
185 &email,
186 payments.payments.as_ref(),
187 super::moderation_service::FanoutMode::Background {
188 bg: &bg,
189 wam: integrations.wam.clone(),
190 session_cache: &caches.session_cache,
191 },
192 &db_user,
193 admin_user.admin_id(),
194 reason,
195 )
196 .await?;
197
198 refresh_user_entries_partial(&db).await
199 }
200
201 /// Unsuspend a user account (admin override).
202 #[tracing::instrument(skip_all, name = "admin::admin_unsuspend_user")]
203 pub(super) async fn admin_unsuspend_user(
204 State(db): State<PgPool>,
205 State(bg): State<BackgroundTx>,
206 State(caches): State<AppCaches>,
207 State(payments): State<Billing>,
208 State(integrations): State<Integrations>,
209 AdminUser(_admin): AdminUser,
210 Path(id): Path<UserId>,
211 ) -> Result<impl IntoResponse> {
212 // Get user for Stripe account ID before unsuspending
213 let db_user = db::users::get_user_by_id(&db, id)
214 .await?
215 .ok_or(AppError::NotFound)?;
216
217 super::moderation_service::unsuspend_creator(
218 &db,
219 payments.payments.as_ref(),
220 super::moderation_service::FanoutMode::Background {
221 bg: &bg,
222 wam: integrations.wam.clone(),
223 session_cache: &caches.session_cache,
224 },
225 &db_user,
226 )
227 .await?;
228
229 refresh_user_entries_partial(&db).await
230 }
231
232 /// Permanently terminate a user account (enforcement ladder step 4).
233 ///
234 /// The account must already be suspended. Sets `terminated_at`, hides all items,
235 /// cancels subscriptions, and emails the user. The user has 30 days to export
236 /// data before the scheduler deletes the account.
237 #[tracing::instrument(skip_all, name = "admin::admin_terminate_user")]
238 pub(super) async fn admin_terminate_user(
239 State(db): State<PgPool>,
240 State(email): State<EmailClient>,
241 State(bg): State<BackgroundTx>,
242 State(payments): State<Billing>,
243 State(integrations): State<Integrations>,
244 admin_user: AdminUser,
245 Path(id): Path<UserId>,
246 ) -> Result<impl IntoResponse> {
247 let db_user = db::users::get_user_by_id(&db, id)
248 .await?
249 .ok_or(AppError::NotFound)?;
250
251 if !db_user.is_suspended() {
252 return Err(AppError::validation(
253 "Account must be suspended before termination".to_string(),
254 ));
255 }
256
257 if db_user.terminated_at.is_some() {
258 return Err(AppError::validation(
259 "Account is already terminated".to_string(),
260 ));
261 }
262
263 db::users::terminate_user(&db, id).await?;
264
265 // Record moderation action
266 db::moderation::create_action(
267 &db,
268 id,
269 admin_user.admin_id(),
270 ModerationActionType::Termination,
271 db_user
272 .suspension_reason
273 .as_deref()
274 .unwrap_or("Account terminated"),
275 None,
276 )
277 .await?;
278
279 // Cancel all fan subscriptions, both active and paused (suspension already paused them)
280 if let Some(ref stripe) = payments.payments
281 && let Some(ref account_id) = db_user.stripe_account_id
282 {
283 let active_subs = db::subscriptions::get_active_subscriptions_by_creator(&db, id).await?;
284 let paused_subs = db::subscriptions::get_paused_subscriptions_by_creator(&db, id).await?;
285 let ids: Vec<String> = active_subs
286 .into_iter()
287 .chain(paused_subs)
288 .map(|s| s.stripe_subscription_id)
289 .collect();
290 crate::payments::fan_ops::spawn_fan_sub_fanout(
291 &bg,
292 std::sync::Arc::clone(stripe),
293 account_id.clone(),
294 ids,
295 crate::payments::fan_ops::FanSubOp::Cancel,
296 integrations.wam.clone(),
297 );
298 }
299
300 // Send termination email
301 let user_email = db_user.email.clone();
302 let user_name = db_user.display_name.clone();
303 let email = email.clone();
304 bg.spawn("account termination notification", async move {
305 if let Err(e) = email
306 .send_account_termination(&user_email, user_name.as_deref())
307 .await
308 {
309 tracing::error!(error = ?e, "failed to send account termination notification");
310 }
311 });
312
313 tracing::info!(
314 user_id = %id,
315 admin_id = %admin_user.id(),
316 "admin terminated user account (30-day export window started)"
317 );
318
319 refresh_user_entries_partial(&db).await
320 }
321
322 /// Trust a user (uploads auto-publish).
323 #[tracing::instrument(skip_all, name = "admin::admin_trust_user")]
324 pub(super) async fn admin_trust_user(
325 State(db): State<PgPool>,
326 AdminUser(_admin): AdminUser,
327 Path(id): Path<UserId>,
328 headers: axum::http::HeaderMap,
329 ) -> Result<Response> {
330 db::users::set_upload_trusted(&db, id, true).await?;
331 tracing::info!(user_id = %id, "admin trusted user for uploads");
332 refresh_partial_for_target(&db, &headers).await
333 }
334
335 /// Untrust a user (uploads require review).
336 #[tracing::instrument(skip_all, name = "admin::admin_untrust_user")]
337 pub(super) async fn admin_untrust_user(
338 State(db): State<PgPool>,
339 AdminUser(_admin): AdminUser,
340 Path(id): Path<UserId>,
341 headers: axum::http::HeaderMap,
342 ) -> Result<Response> {
343 db::users::set_upload_trusted(&db, id, false).await?;
344 tracing::info!(user_id = %id, "admin untrusted user for uploads");
345 refresh_partial_for_target(&db, &headers).await
346 }
347
348 /// Lock a user's custom pages (moderation kill switch): the editor goes
349 /// read-only and their custom profile/project pages render the platform
350 /// default. The source is preserved, so unlocking restores it. Reversible.
351 #[tracing::instrument(skip_all, name = "admin::admin_lock_custom_pages")]
352 pub(super) async fn admin_lock_custom_pages(
353 State(db): State<PgPool>,
354 admin_user: AdminUser,
355 Path(id): Path<UserId>,
356 headers: axum::http::HeaderMap,
357 ) -> Result<Response> {
358 db::users::set_custom_pages_locked(&db, id, true).await?;
359 db::moderation::create_action(
360 &db,
361 id,
362 admin_user.admin_id(),
363 db::ModerationActionType::ContentRemoval,
364 "custom pages locked",
365 Some("custom-page"),
366 )
367 .await?;
368 tracing::info!(user_id = %id, "admin locked custom pages");
369 refresh_partial_for_target(&db, &headers).await
370 }
371
372 /// Unlock a user's custom pages, restoring their preserved custom source.
373 #[tracing::instrument(skip_all, name = "admin::admin_unlock_custom_pages")]
374 pub(super) async fn admin_unlock_custom_pages(
375 State(db): State<PgPool>,
376 AdminUser(_admin): AdminUser,
377 Path(id): Path<UserId>,
378 headers: axum::http::HeaderMap,
379 ) -> Result<Response> {
380 db::users::set_custom_pages_locked(&db, id, false).await?;
381 tracing::info!(user_id = %id, "admin unlocked custom pages");
382 refresh_partial_for_target(&db, &headers).await
383 }
384
385 /// Return the right partial based on which page triggered the request.
386 async fn refresh_partial_for_target(
387 db: &PgPool,
388 headers: &axum::http::HeaderMap,
389 ) -> Result<Response> {
390 // htmx 4 sends `tag#id` here rather than the bare id it sent in 2.x.
391 let target = headers
392 .get("HX-Target")
393 .and_then(|v| v.to_str().ok())
394 .unwrap_or("");
395 if target == "div#users-table" {
396 Ok(refresh_user_entries_partial(db).await?.into_response())
397 } else {
398 super::uploads::refresh_held_uploads_partial(db).await
399 }
400 }
401
402 /// Re-query users and return the entries partial (page 1, no filter).
403 async fn refresh_user_entries_partial(db: &PgPool) -> Result<AdminUserEntriesTemplate> {
404 let per_page: i64 = 50;
405 let total_count = db::users::count_users(db, None).await?;
406 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
407 let db_users = db::users::get_all_users(db, None, per_page, 0).await?;
408 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
409 Ok(AdminUserEntriesTemplate {
410 users,
411 current_page: 1,
412 total_pages,
413 current_filter: String::new(),
414 })
415 }
416