Skip to main content

max / makenotwork

13.2 KB · 352 lines History Blame Raw
1 //! Admin user management: listing, suspension, trust status.
2
3 use axum::{
4 extract::{Path, Query, State},
5 response::{IntoResponse, Response},
6 Form,
7 };
8 use serde::Deserialize;
9
10 use crate::{
11 auth::AdminUser,
12 db::{self, ModerationActionType, UserId},
13 error::{AppError, Result},
14 helpers::{get_csrf_token, spawn_email},
15 templates::*,
16 types::*,
17 AppState,
18 };
19
20 #[derive(Debug, Deserialize)]
21 pub(super) struct UserFilterQuery {
22 pub status: Option<String>,
23 pub page: Option<i64>,
24 }
25
26 /// Render the admin user management page.
27 #[tracing::instrument(skip_all, name = "admin::admin_users")]
28 pub(super) async fn admin_users(
29 State(state): State<AppState>,
30 session: tower_sessions::Session,
31 AdminUser(user): AdminUser,
32 Query(query): Query<UserFilterQuery>,
33 ) -> Result<impl IntoResponse> {
34 let csrf_token = get_csrf_token(&session).await;
35 let current_filter = query.status.clone().unwrap_or_default();
36
37 // Upper-clamp page so `OFFSET = (page-1)*per_page` doesn't overflow i64
38 // or produce a sqlx "value out of range" 500. 1e9 pages × 50 per_page is
39 // already 50 billion rows — well past anything the admin panel will ever
40 // reach, and keeps the OFFSET safely inside i64.
41 let page = query.page.unwrap_or(1).clamp(1, 1_000_000_000);
42 let per_page: i64 = 50;
43 let offset = (page - 1) * per_page;
44
45 let (total_users_i64, total_suspended_i64) = db::users::count_users_summary(&state.db).await?;
46 let total_users = total_users_i64 as usize;
47 let total_suspended = total_suspended_i64 as usize;
48
49 let total_count = match query.status.as_deref() {
50 Some("suspended") => total_suspended_i64,
51 Some("active") => total_users_i64 - total_suspended_i64,
52 _ => total_users_i64,
53 };
54 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
55
56 let db_users = db::users::get_all_users(&state.db, query.status.as_deref(), per_page, offset).await?;
57
58 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
59
60 Ok(AdminUsersTemplate {
61 csrf_token,
62 session_user: Some(user),
63 users,
64 total_users,
65 total_suspended,
66 current_filter,
67 current_page: page,
68 total_pages,
69 admin_active_page: "users",
70 })
71 }
72
73 /// Return filtered user entries as an HTMX partial.
74 #[tracing::instrument(skip_all, name = "admin::admin_user_entries")]
75 pub(super) async fn admin_user_entries(
76 State(state): State<AppState>,
77 AdminUser(_user): AdminUser,
78 Query(query): Query<UserFilterQuery>,
79 ) -> Result<impl IntoResponse> {
80 let current_filter = query.status.clone().unwrap_or_default();
81 // Upper-clamp page so `OFFSET = (page-1)*per_page` doesn't overflow i64
82 // or produce a sqlx "value out of range" 500. 1e9 pages × 50 per_page is
83 // already 50 billion rows — well past anything the admin panel will ever
84 // reach, and keeps the OFFSET safely inside i64.
85 let page = query.page.unwrap_or(1).clamp(1, 1_000_000_000);
86 let per_page: i64 = 50;
87 let offset = (page - 1) * per_page;
88
89 let total_count = db::users::count_users(&state.db, query.status.as_deref()).await?;
90 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
91
92 let db_users = db::users::get_all_users(&state.db, query.status.as_deref(), per_page, offset).await?;
93 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
94 Ok(AdminUserEntriesTemplate {
95 users,
96 current_page: page,
97 total_pages,
98 current_filter,
99 })
100 }
101
102 #[derive(Debug, Deserialize)]
103 pub(super) struct SuspendForm {
104 pub reason: String,
105 }
106
107 /// Send a policy warning to a user without suspending their account.
108 /// Records the warning in moderation history and emails the user.
109 #[tracing::instrument(skip_all, name = "admin::admin_warn_user")]
110 pub(super) async fn admin_warn_user(
111 State(state): State<AppState>,
112 AdminUser(admin): AdminUser,
113 Path(id): Path<UserId>,
114 Form(form): Form<SuspendForm>,
115 ) -> Result<impl IntoResponse> {
116 let reason = form.reason.trim();
117 if reason.is_empty() {
118 return Err(AppError::validation("Reason is required".to_string()));
119 }
120
121 let db_user = db::users::get_user_by_id(&state.db, id)
122 .await?
123 .ok_or(AppError::NotFound)?;
124
125 // Record warning in moderation history
126 db::moderation::create_action(&state.db, id, admin.id, ModerationActionType::Warning, reason, None).await?;
127
128 // Send warning email
129 if let Err(e) = state.email
130 .send_policy_warning(&db_user.email, db_user.display_name.as_deref(), reason)
131 .await
132 {
133 tracing::error!(error = ?e, user_id = %id, "failed to send warning email");
134 }
135
136 tracing::info!(user_id = %id, admin_id = %admin.id, reason = %reason, "admin sent policy warning");
137
138 refresh_user_entries_partial(&state).await
139 }
140
141 /// Suspend a user account and send notification email.
142 #[tracing::instrument(skip_all, name = "admin::admin_suspend_user")]
143 pub(super) async fn admin_suspend_user(
144 State(state): State<AppState>,
145 AdminUser(admin): AdminUser,
146 Path(id): Path<UserId>,
147 Form(form): Form<SuspendForm>,
148 ) -> Result<impl IntoResponse> {
149 let reason = form.reason.trim();
150 if reason.is_empty() {
151 return Err(AppError::validation("Reason is required".to_string()));
152 }
153
154 // Get user for email notification
155 let db_user = db::users::get_user_by_id(&state.db, id)
156 .await?
157 .ok_or(AppError::NotFound)?;
158
159 db::users::suspend_user(&state.db, id, reason).await?;
160
161 // Immediately revoke all sessions and evict from cache so the suspended
162 // user cannot act during the 30-second session-touch cache window.
163 let revoked_ids = db::sessions::delete_all_sessions_for_user(&state.db, id).await?;
164 for sid in &revoked_ids {
165 state.session_cache.remove(sid);
166 }
167 if !revoked_ids.is_empty() {
168 tracing::info!(user_id = %id, revoked = revoked_ids.len(), "revoked sessions on suspension");
169 }
170
171 // Record moderation action for audit trail
172 db::moderation::create_action(&state.db, id, admin.id, ModerationActionType::Suspension, reason, None).await?;
173
174 // Pause fan subscriptions to this creator's projects
175 if let Some(ref stripe) = state.stripe
176 && let Some(ref account_id) = db_user.stripe_account_id
177 {
178 let subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, id).await?;
179 let count = subs.len();
180 for sub in &subs {
181 if let Err(e) = stripe.pause_subscription(&sub.stripe_subscription_id, account_id).await {
182 tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to pause subscription on Stripe");
183 }
184 }
185 let paused = db::subscriptions::pause_subscriptions_for_creator(&state.db, id).await?;
186 tracing::info!(user_id = %id, stripe_paused = count, db_paused = paused, "paused fan subscriptions for suspended creator");
187 }
188
189 // Send notification email (fire-and-forget)
190 if let Err(e) = state.email
191 .send_suspension_notification(&db_user.email, db_user.display_name.as_deref(), reason)
192 .await
193 {
194 tracing::error!(error = ?e, user_id = %id, "failed to send suspension email");
195 }
196
197 tracing::info!(user_id = %id, reason = %reason, "admin suspended user");
198
199 refresh_user_entries_partial(&state).await
200 }
201
202 /// Unsuspend a user account (admin override).
203 #[tracing::instrument(skip_all, name = "admin::admin_unsuspend_user")]
204 pub(super) async fn admin_unsuspend_user(
205 State(state): State<AppState>,
206 AdminUser(_admin): AdminUser,
207 Path(id): Path<UserId>,
208 ) -> Result<impl IntoResponse> {
209 // Get user for Stripe account ID before unsuspending
210 let db_user = db::users::get_user_by_id(&state.db, id)
211 .await?
212 .ok_or(AppError::NotFound)?;
213
214 db::users::unsuspend_user(&state.db, id).await?;
215
216 // Resolve suspension action in moderation history
217 db::moderation::resolve_actions_by_type(&state.db, id, ModerationActionType::Suspension).await?;
218
219 // Resume paused fan subscriptions
220 if let Some(ref stripe) = state.stripe
221 && let Some(ref account_id) = db_user.stripe_account_id
222 {
223 let resumed = db::subscriptions::resume_subscriptions_for_creator(&state.db, id).await?;
224 for sub in &resumed {
225 if let Err(e) = stripe.resume_subscription(&sub.stripe_subscription_id, account_id).await {
226 tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to resume subscription on Stripe");
227 }
228 }
229 tracing::info!(user_id = %id, resumed = resumed.len(), "resumed fan subscriptions for unsuspended creator");
230 }
231
232 tracing::info!(user_id = %id, "admin unsuspended user");
233
234 refresh_user_entries_partial(&state).await
235 }
236
237 /// Permanently terminate a user account (enforcement ladder step 4).
238 ///
239 /// The account must already be suspended. Sets `terminated_at`, hides all items,
240 /// cancels subscriptions, and emails the user. The user has 30 days to export
241 /// data before the scheduler deletes the account.
242 #[tracing::instrument(skip_all, name = "admin::admin_terminate_user")]
243 pub(super) async fn admin_terminate_user(
244 State(state): State<AppState>,
245 AdminUser(admin): AdminUser,
246 Path(id): Path<UserId>,
247 ) -> Result<impl IntoResponse> {
248 let db_user = db::users::get_user_by_id(&state.db, id)
249 .await?
250 .ok_or(AppError::NotFound)?;
251
252 if !db_user.is_suspended() {
253 return Err(AppError::validation(
254 "Account must be suspended before termination".to_string(),
255 ));
256 }
257
258 if db_user.terminated_at.is_some() {
259 return Err(AppError::validation(
260 "Account is already terminated".to_string(),
261 ));
262 }
263
264 db::users::terminate_user(&state.db, id).await?;
265
266 // Record moderation action
267 db::moderation::create_action(
268 &state.db, id, admin.id, ModerationActionType::Termination,
269 db_user.suspension_reason.as_deref().unwrap_or("Account terminated"),
270 None,
271 ).await?;
272
273 // Cancel all fan subscriptions — both active and paused (suspension already paused them)
274 if let Some(ref stripe) = state.stripe
275 && let Some(ref account_id) = db_user.stripe_account_id
276 {
277 let active_subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, id).await?;
278 let paused_subs = db::subscriptions::get_paused_subscriptions_by_creator(&state.db, id).await?;
279 for sub in active_subs.iter().chain(paused_subs.iter()) {
280 if let Err(e) = stripe.cancel_subscription(&sub.stripe_subscription_id, account_id).await {
281 tracing::error!(stripe_sub_id = %sub.stripe_subscription_id, error = ?e, "failed to cancel subscription on termination");
282 }
283 }
284 }
285
286 // Send termination email
287 let user_email = db_user.email.clone();
288 let user_name = db_user.display_name.clone();
289 spawn_email!(state, "account termination notification", |email| {
290 email.send_account_termination(&user_email, user_name.as_deref())
291 });
292
293 tracing::info!(
294 user_id = %id,
295 admin_id = %admin.id,
296 "admin terminated user account (30-day export window started)"
297 );
298
299 refresh_user_entries_partial(&state).await
300 }
301
302 /// Trust a user (uploads auto-publish).
303 #[tracing::instrument(skip_all, name = "admin::admin_trust_user")]
304 pub(super) async fn admin_trust_user(
305 State(state): State<AppState>,
306 AdminUser(_admin): AdminUser,
307 Path(id): Path<UserId>,
308 headers: axum::http::HeaderMap,
309 ) -> Result<Response> {
310 db::users::set_upload_trusted(&state.db, id, true).await?;
311 tracing::info!(user_id = %id, "admin trusted user for uploads");
312 refresh_partial_for_target(&state, &headers).await
313 }
314
315 /// Untrust a user (uploads require review).
316 #[tracing::instrument(skip_all, name = "admin::admin_untrust_user")]
317 pub(super) async fn admin_untrust_user(
318 State(state): State<AppState>,
319 AdminUser(_admin): AdminUser,
320 Path(id): Path<UserId>,
321 headers: axum::http::HeaderMap,
322 ) -> Result<Response> {
323 db::users::set_upload_trusted(&state.db, id, false).await?;
324 tracing::info!(user_id = %id, "admin untrusted user for uploads");
325 refresh_partial_for_target(&state, &headers).await
326 }
327
328 /// Return the right partial based on which page triggered the request.
329 async fn refresh_partial_for_target(state: &AppState, headers: &axum::http::HeaderMap) -> Result<Response> {
330 let target = headers.get("HX-Target").and_then(|v| v.to_str().ok()).unwrap_or("");
331 if target == "users-table" {
332 Ok(refresh_user_entries_partial(state).await?.into_response())
333 } else {
334 super::uploads::refresh_held_uploads_partial(state).await
335 }
336 }
337
338 /// Re-query users and return the entries partial (page 1, no filter).
339 async fn refresh_user_entries_partial(state: &AppState) -> Result<AdminUserEntriesTemplate> {
340 let per_page: i64 = 50;
341 let total_count = db::users::count_users(&state.db, None).await?;
342 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
343 let db_users = db::users::get_all_users(&state.db, None, per_page, 0).await?;
344 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
345 Ok(AdminUserEntriesTemplate {
346 users,
347 current_page: 1,
348 total_pages,
349 current_filter: String::new(),
350 })
351 }
352