Skip to main content

max / makenotwork

19.3 KB · 560 lines History Blame Raw
1 //! Admin routes for creator waitlist management, user moderation, and platform operations.
2
3 mod comp_codes;
4 mod mail_caps;
5 mod moderation;
6 pub mod moderation_service;
7 mod signups;
8 mod uploads;
9 mod users;
10 mod waitlist;
11
12 use axum::extract::FromRequestParts;
13 use axum::http::request::Parts;
14 use axum::{
15 Form,
16 extract::{Path, Request, State},
17 middleware::{Next, from_fn_with_state},
18 response::{IntoResponse, Response},
19 routing::get,
20 };
21 use serde::Deserialize;
22 use sqlx::PgPool;
23
24 use crate::{
25 AppState, Integrations, Ops,
26 auth::AdminUser,
27 csrf::{CsrfRouter, post_csrf},
28 db::{self, UserId},
29 email::EmailClient,
30 error::{AppError, Result},
31 };
32
33 /// `/admin*` routes served WITHOUT the admin gate, by explicit design.
34 ///
35 /// Kept as a sealed allowlist so the blanket [`require_admin_layer`] stays
36 /// honest: every route under `admin_routes()` requires admin EXCEPT these, and
37 /// `admin_gate_covers_every_route` asserts the set is exactly this. Mirrors the
38 /// CSRF skip-list discipline, an exemption is a visible, tested decision, not an
39 /// accidental gap.
40 pub(crate) const ADMIN_PUBLIC_PATHS: &[&str] = &[
41 // Scan-queue health JSON, intentionally unauthenticated (counts only, no
42 // PII) so PoM can scrape it against stable threshold rules.
43 "/admin/uploads/health.json",
44 ];
45
46 /// Blanket admin gate over the whole `/admin*` subtree. Runs the same
47 /// [`AdminUser`] check every admin handler ran individually, so a handler that
48 /// forgets the extractor cannot ship unauthenticated. Non-admins get 404,
49 /// keeping admin routes hidden; the
50 /// only bypass is the explicit [`ADMIN_PUBLIC_PATHS`] allowlist. Handlers keep
51 /// their own `AdminUser` where they need the `AdminId` write-witness; the
52 /// duplicate check is a cached session touch.
53 async fn require_admin_layer(
54 State(state): State<AppState>,
55 req: Request,
56 next: Next,
57 ) -> Result<Response> {
58 if ADMIN_PUBLIC_PATHS.contains(&req.uri().path()) {
59 return Ok(next.run(req).await);
60 }
61 let (mut parts, body): (Parts, _) = req.into_parts();
62 // The AdminUser rejection (404 for non-admins) becomes the response.
63 let _admin = AdminUser::from_request_parts(&mut parts, &state).await?;
64 let req = Request::from_parts(parts, body);
65 Ok(next.run(req).await)
66 }
67
68 /// Register admin routes for waitlist management, user moderation, and lottery.
69 pub fn admin_routes(state: AppState) -> CsrfRouter<AppState> {
70 CsrfRouter::new()
71 // Waitlist
72 .route_get("/admin/mail-caps", get(mail_caps::admin_mail_caps))
73 .route(
74 "/api/admin/mail-caps/{id}/grant",
75 post_csrf(mail_caps::admin_mail_cap_grant),
76 )
77 .route(
78 "/api/admin/mail-caps/{id}/deny",
79 post_csrf(mail_caps::admin_mail_cap_deny),
80 )
81 .route_get("/admin/waitlist", get(waitlist::admin_waitlist))
82 .route_get(
83 "/admin/waitlist/entries",
84 get(waitlist::admin_waitlist_entries),
85 )
86 .route(
87 "/api/admin/waitlist/{id}/approve",
88 post_csrf(waitlist::admin_approve),
89 )
90 .route(
91 "/api/admin/waitlist/{id}/spam",
92 post_csrf(waitlist::admin_spam),
93 )
94 .route("/api/admin/lottery", post_csrf(waitlist::admin_lottery))
95 // User management
96 .route_get("/admin/users", get(users::admin_users))
97 .route_get("/admin/users/entries", get(users::admin_user_entries))
98 .route(
99 "/api/admin/users/{id}/warn",
100 post_csrf(users::admin_warn_user),
101 )
102 .route(
103 "/api/admin/users/{id}/suspend",
104 post_csrf(users::admin_suspend_user),
105 )
106 .route(
107 "/api/admin/users/{id}/unsuspend",
108 post_csrf(users::admin_unsuspend_user),
109 )
110 .route(
111 "/api/admin/users/{id}/terminate",
112 post_csrf(users::admin_terminate_user),
113 )
114 // Upload review queue
115 .route_get("/admin/uploads", get(uploads::admin_uploads))
116 .route(
117 "/api/admin/uploads/items/{id}/promote",
118 post_csrf(uploads::admin_promote_item),
119 )
120 .route(
121 "/api/admin/uploads/items/{id}/quarantine",
122 post_csrf(uploads::admin_quarantine_item),
123 )
124 .route(
125 "/api/admin/uploads/items/{id}/rescan",
126 post_csrf(uploads::admin_rescan_item),
127 )
128 .route(
129 "/api/admin/uploads/versions/{id}/promote",
130 post_csrf(uploads::admin_promote_version),
131 )
132 .route(
133 "/api/admin/uploads/versions/{id}/quarantine",
134 post_csrf(uploads::admin_quarantine_version),
135 )
136 .route(
137 "/api/admin/uploads/versions/{id}/rescan",
138 post_csrf(uploads::admin_rescan_version),
139 )
140 .route(
141 "/api/admin/uploads/bulk/rescan",
142 post_csrf(uploads::admin_bulk_rescan_held),
143 )
144 .route(
145 "/api/admin/uploads/bulk/promote",
146 post_csrf(uploads::admin_bulk_promote_held),
147 )
148 .route_get(
149 "/admin/uploads/queue-summary",
150 get(uploads::admin_queue_summary_partial),
151 )
152 .route_get("/admin/uploads/audit", get(uploads::admin_scan_audit))
153 .route_get("/admin/uploads/health.json", get(uploads::scan_health_json))
154 .route(
155 "/api/admin/users/{id}/trust",
156 post_csrf(users::admin_trust_user),
157 )
158 .route(
159 "/api/admin/users/{id}/untrust",
160 post_csrf(users::admin_untrust_user),
161 )
162 .route(
163 "/api/admin/users/{id}/lock-pages",
164 post_csrf(users::admin_lock_custom_pages),
165 )
166 .route(
167 "/api/admin/users/{id}/unlock-pages",
168 post_csrf(users::admin_unlock_custom_pages),
169 )
170 // Appeals
171 .route_get("/admin/appeals", get(moderation::admin_appeals))
172 .route(
173 "/api/admin/appeals/{user_id}/decide",
174 post_csrf(moderation::admin_decide_appeal),
175 )
176 // Email signups
177 .route_get("/admin/signups", get(signups::admin_signups))
178 // Reports
179 .route_get("/admin/reports", get(moderation::admin_reports))
180 .route_get(
181 "/admin/reports/entries",
182 get(moderation::admin_report_entries),
183 )
184 .route(
185 "/api/admin/reports/{id}/resolve",
186 post_csrf(moderation::admin_resolve_report),
187 )
188 // Per-item content removal
189 .route(
190 "/api/admin/items/{id}/remove",
191 post_csrf(moderation::admin_remove_item),
192 )
193 .route(
194 "/api/admin/items/{id}/restore",
195 post_csrf(moderation::admin_restore_item),
196 )
197 // MT provisioning
198 .route("/api/admin/mt/provision", post_csrf(admin_mt_provision))
199 // Storage overrides
200 .route(
201 "/api/admin/users/{id}/file-override",
202 post_csrf(admin_file_override),
203 )
204 // Shutdown
205 .route(
206 "/api/admin/shutdown-notice",
207 post_csrf(admin_shutdown_notice),
208 )
209 // Founder pricing
210 .route(
211 "/api/admin/founder-window/close",
212 post_csrf(admin_close_founder_window),
213 )
214 .route_get("/admin/comp-codes", get(comp_codes::admin_comp_codes))
215 .route(
216 "/api/admin/comp-codes/create",
217 post_csrf(comp_codes::admin_create_comp_code),
218 )
219 // Metrics
220 .route_get("/admin/metrics", get(admin_metrics))
221 // Structural admin gate over every route above (Run 20). A forgotten
222 // per-handler AdminUser can no longer expose an admin route.
223 .route_layer(from_fn_with_state(state, require_admin_layer))
224 }
225
226 // --- MT Provisioning ---
227
228 /// Backfill MT communities for all projects that don't have one yet.
229 #[tracing::instrument(skip_all, name = "admin::admin_mt_provision")]
230 async fn admin_mt_provision(
231 State(db): State<PgPool>,
232 State(integrations): State<Integrations>,
233 AdminUser(_admin): AdminUser,
234 ) -> Result<Response> {
235 let Some(ref mt) = integrations.mt_client else {
236 return Err(AppError::validation(
237 "MT integration not configured".to_string(),
238 ));
239 };
240
241 let projects = db::projects::get_projects_without_mt_community(&db).await?;
242 let total = projects.len();
243 let mut provisioned = 0u32;
244 let mut failed = 0u32;
245
246 // Batch-load all project owners in one query instead of N individual lookups
247 let owner_ids: Vec<db::UserId> = projects.iter().map(|p| p.user_id).collect();
248 let owners = db::users::get_users_by_ids(&db, &owner_ids).await?;
249 let owner_map: std::collections::HashMap<db::UserId, &db::DbUser> =
250 owners.iter().map(|u| (u.id, u)).collect();
251
252 // Fan the per-project MT HTTP call (5s timeout each) out over a bounded
253 // JoinSet instead of awaiting them strictly serially, provisioning latency
254 // was MT-latency x project-count over the whole table (Perf-S3, Run 9). Cap
255 // concurrency so a large backfill can't open hundreds of MT connections at
256 // once; mirror admin_shutdown_notice's bound.
257 let parallelism = crate::constants::BROADCAST_PARALLELISM;
258 let mut set: tokio::task::JoinSet<bool> = tokio::task::JoinSet::new();
259
260 for project in &projects {
261 let Some(user) = owner_map.get(&project.user_id) else {
262 failed += 1;
263 continue;
264 };
265
266 if set.len() >= parallelism
267 && let Some(res) = set.join_next().await
268 {
269 if res.unwrap_or(false) {
270 provisioned += 1;
271 } else {
272 failed += 1;
273 }
274 }
275
276 let mt = mt.clone();
277 let pool = db.clone();
278 let req = crate::mt_client::CreateCommunityRequest {
279 name: project.title.clone(),
280 slug: project.slug.to_string(),
281 description: project.description.clone(),
282 owner_mnw_id: *user.id,
283 owner_username: user.username.to_string(),
284 owner_display_name: user.display_name.clone(),
285 };
286 let project_id = project.id;
287 let slug = project.slug.to_string();
288 set.spawn(async move {
289 match mt.create_community(&req).await {
290 Ok(resp) => match db::projects::set_mt_community_id(&pool, project_id, resp.community_id).await {
291 Ok(()) => true,
292 Err(e) => {
293 tracing::error!(error = ?e, project_id = %project_id, "failed to store MT community ID");
294 false
295 }
296 },
297 Err(e) => {
298 tracing::error!(error = ?e, slug = %slug, "MT community provisioning failed");
299 false
300 }
301 }
302 });
303 }
304
305 while let Some(res) = set.join_next().await {
306 if res.unwrap_or(false) {
307 provisioned += 1;
308 } else {
309 failed += 1;
310 }
311 }
312
313 tracing::info!(total, provisioned, failed, "MT community backfill complete");
314
315 Ok((
316 axum::http::StatusCode::OK,
317 format!("MT provisioning: {provisioned} of {total} projects provisioned ({failed} failed)"),
318 )
319 .into_response())
320 }
321
322 // --- Shutdown ---
323
324 #[derive(Debug, Deserialize)]
325 pub(super) struct ShutdownNoticeForm {
326 pub shutdown_date: String,
327 }
328
329 /// Send a shutdown notice email to all users.
330 #[tracing::instrument(skip_all, name = "admin::admin_shutdown_notice")]
331 async fn admin_shutdown_notice(
332 State(db): State<PgPool>,
333 State(email): State<EmailClient>,
334 AdminUser(_admin): AdminUser,
335 Form(form): Form<ShutdownNoticeForm>,
336 ) -> Result<Response> {
337 let shutdown_date = form.shutdown_date.trim();
338 if shutdown_date.is_empty() {
339 return Err(AppError::validation(
340 "Shutdown date is required".to_string(),
341 ));
342 }
343
344 let all_users = db::users::get_all_user_emails(&db).await?;
345 let count = all_users.len();
346
347 // Fan out on a background task with a bounded JoinSet (mirrors the broadcast
348 // path). The previous version sent one email at a time, inline, scaling the
349 // request's latency with the entire user base and tying up a handler for the
350 // whole send. Return immediately; the task logs its own totals.
351 let email_client = email.clone();
352 let shutdown_date = shutdown_date.to_string();
353 tokio::spawn(async move {
354 let mut set = tokio::task::JoinSet::new();
355 let chunk_delay =
356 std::time::Duration::from_millis(crate::constants::BROADCAST_CHUNK_DELAY_MS);
357 let sent = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0));
358 let failed = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0));
359
360 for (email, display_name) in all_users {
361 if set.len() >= crate::constants::BROADCAST_PARALLELISM {
362 let _ = set.join_next().await;
363 }
364 let email_client = email_client.clone();
365 let shutdown_date = shutdown_date.clone();
366 let sent = std::sync::Arc::clone(&sent);
367 let failed = std::sync::Arc::clone(&failed);
368 set.spawn(async move {
369 use std::sync::atomic::Ordering;
370 if let Err(e) = email_client
371 .send_shutdown_notice(&email, display_name.as_deref(), &shutdown_date)
372 .await
373 {
374 tracing::error!(error = ?e, email = %email, "failed to send shutdown notice");
375 failed.fetch_add(1, Ordering::Relaxed);
376 } else {
377 sent.fetch_add(1, Ordering::Relaxed);
378 }
379 });
380 tokio::time::sleep(chunk_delay).await;
381 }
382 while set.join_next().await.is_some() {}
383
384 use std::sync::atomic::Ordering;
385 tracing::info!(
386 sent = sent.load(Ordering::Relaxed),
387 failed = failed.load(Ordering::Relaxed),
388 shutdown_date = %shutdown_date,
389 "shutdown notices sent"
390 );
391 });
392
393 Ok((
394 axum::http::StatusCode::OK,
395 [("HX-Redirect", "/admin/users")],
396 format!("Sending shutdown notices to {count} users."),
397 )
398 .into_response())
399 }
400
401 // --- Founder pricing ---
402
403 /// Close the founder pricing window. Idempotent: stamps `founder_locked_at`
404 /// on every user currently flagged `is_founder` who has an active
405 /// creator-tier subscription, and skips anyone already locked. Operators
406 /// should also flip `CREATOR_FOUNDER_WINDOW_OPEN=false` in the env
407 /// separately so new signups stop getting founder prices; this route only
408 /// performs the snapshot, it doesn't change config. The snapshot is one-way:
409 /// nothing un-stamps `founder_locked_at`.
410 #[tracing::instrument(skip_all, name = "admin::admin_close_founder_window")]
411 async fn admin_close_founder_window(
412 State(db): State<PgPool>,
413 AdminUser(_admin): AdminUser,
414 ) -> Result<Response> {
415 let locked = db::users::lock_in_founders_with_active_subscriptions(&db).await?;
416 tracing::info!(
417 locked = locked,
418 "founder window close: users stamped with founder_locked_at"
419 );
420 Ok((
421 axum::http::StatusCode::OK,
422 format!(
423 "Founder window snapshot complete: {} user{} locked in. \
424 Remember to flip CREATOR_FOUNDER_WINDOW_OPEN=false in the env.",
425 locked,
426 if locked == 1 { "" } else { "s" }
427 ),
428 )
429 .into_response())
430 }
431
432 // --- Metrics ---
433
434 /// Render the admin metrics dashboard with live Prometheus data.
435 #[tracing::instrument(skip_all, name = "admin::admin_metrics")]
436 async fn admin_metrics(
437 State(db): State<PgPool>,
438 State(ops): State<Ops>,
439 session: tower_sessions::Session,
440 AdminUser(user): AdminUser,
441 ) -> Result<impl IntoResponse> {
442 let csrf_token = crate::helpers::get_csrf_token(&session).await;
443 let uptime = {
444 let d = ops.start_instant.elapsed();
445 let secs = d.as_secs();
446 let days = secs / 86400;
447 let hours = (secs % 86400) / 3600;
448 let mins = (secs % 3600) / 60;
449 if days > 0 {
450 format!("{days}d {hours}h {mins}m")
451 } else if hours > 0 {
452 format!("{hours}h {mins}m")
453 } else {
454 format!("{mins}m")
455 }
456 };
457
458 let pool_size = db.size();
459 let pool_idle = db.num_idle() as u32;
460 let pool_active = pool_size.saturating_sub(pool_idle);
461
462 // Parse metrics from the Prometheus handle (if available)
463 let (total_requests, error_rate, total_errors, top_routes, error_breakdown) =
464 if let Some(ref handle) = ops.metrics_handle {
465 let snap = crate::metrics::snapshot(handle);
466 let rate = if snap.total_requests > 0 {
467 snap.total_5xx as f64 / snap.total_requests as f64 * 100.0
468 } else {
469 0.0
470 };
471 let routes = snap
472 .top_routes
473 .into_iter()
474 .map(
475 |(method, path, status, count)| crate::templates::RouteMetric {
476 method,
477 path,
478 status,
479 count,
480 },
481 )
482 .collect();
483 let errors = snap
484 .error_breakdown
485 .into_iter()
486 .map(|(kind, count)| crate::templates::ErrorMetric { kind, count })
487 .collect();
488 (snap.total_requests, rate, snap.total_errors, routes, errors)
489 } else {
490 (0, 0.0, 0, vec![], vec![])
491 };
492
493 Ok(crate::templates::AdminMetricsTemplate {
494 csrf_token,
495 session_user: Some(user),
496 admin_active_page: "metrics",
497 uptime,
498 total_requests,
499 error_rate,
500 total_errors,
501 pool_max: pool_size,
502 pool_active,
503 pool_idle,
504 top_routes,
505 error_breakdown,
506 })
507 }
508
509 // --- File Override ---
510
511 #[derive(Debug, Deserialize)]
512 pub(super) struct FileOverrideForm {
513 pub max_file_bytes: Option<i64>,
514 }
515
516 /// Set or clear the admin per-file size override for a user.
517 ///
518 /// POST /api/admin/users/{id}/file-override
519 #[tracing::instrument(skip_all, name = "admin::admin_file_override")]
520 async fn admin_file_override(
521 State(db): State<PgPool>,
522 AdminUser(_admin): AdminUser,
523 Path(user_id): Path<UserId>,
524 Form(form): Form<FileOverrideForm>,
525 ) -> Result<impl IntoResponse> {
526 // Validate: if provided, must be positive
527 if let Some(bytes) = form.max_file_bytes
528 && bytes <= 0
529 {
530 return Err(AppError::BadRequest(
531 "Override must be a positive number of bytes".to_string(),
532 ));
533 }
534
535 db::creator_tiers::set_max_file_override(&db, user_id, form.max_file_bytes).await?;
536
537 let msg = match form.max_file_bytes {
538 Some(bytes) => format!(
539 "File override set to {}",
540 crate::helpers::format_bytes(bytes)
541 ),
542 None => "File override cleared".to_string(),
543 };
544
545 Ok(crate::helpers::htmx_toast_response(&msg, "success"))
546 }
547
548 #[cfg(test)]
549 mod tests {
550 use super::ADMIN_PUBLIC_PATHS;
551
552 /// Seal the admin-gate exemption allowlist. Adding a route here is a
553 /// deliberate decision to serve it unauthenticated under `/admin*`; this
554 /// test forces that decision to be explicit rather than a silent gap.
555 #[test]
556 fn admin_public_paths_is_exactly_the_health_endpoint() {
557 assert_eq!(ADMIN_PUBLIC_PATHS, &["/admin/uploads/health.json"]);
558 }
559 }
560