Skip to main content

max / makenotwork

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