Skip to main content

max / makenotwork

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