Skip to main content

max / makenotwork

Decompose AppState Phase 2: migrate admin to slices Migrate all admin handlers off State<AppState>. Per-handler slices: - comp_codes/signups/waitlist: PgPool - moderation/users: PgPool + varying EmailClient/BackgroundTx/AppCaches (session_cache)/Billing/Integrations per handler - uploads: PgPool + AppStorage/Config (promote) + Scanning (rescan); JoinSet bulk tasks clone individual slices instead of the whole state - mod.rs handlers: admin_mt_provision (PgPool+Integrations), admin_metrics (PgPool+Ops), admin_shutdown_notice (PgPool+EmailClient), etc. Helper fns (fetch_layer_health, attach_last_actions, refresh_* partials, rescan_*_inner, load_comp_code_rows) narrowed to their slice refs. spawn_email! inlined at 3 sites (macro needs both .email and .bg). admin_routes(state) and the require_admin_layer gate keep AppState (structural admin gate needs whole state). No behavior change; admin 33 + user 64 + moderation 4 + waitlist 11 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-14 00:12 UTC
Commit: cf7421bcb0ce36baeec7eb79e53f67705cf1dd8c
Parent: 730c164
7 files changed, +298 insertions, -224 deletions
M server/src/routes/admin/comp_codes.rs +9 -8
@@ -12,6 +12,8 @@ use axum::{
12 12 };
13 13 use serde::Deserialize;
14 14
15 + use sqlx::PgPool;
16 +
15 17 use crate::{
16 18 auth::AdminUser,
17 19 db,
@@ -19,18 +21,17 @@ use crate::{
19 21 helpers::get_csrf_token,
20 22 templates::*,
21 23 types::*,
22 - AppState,
23 24 };
24 25
25 26 /// Render the comp-codes dashboard: mint form plus a status list.
26 27 #[tracing::instrument(skip_all, name = "admin::admin_comp_codes")]
27 28 pub(super) async fn admin_comp_codes(
28 - State(state): State<AppState>,
29 + State(db): State<PgPool>,
29 30 session: tower_sessions::Session,
30 31 AdminUser(user): AdminUser,
31 32 ) -> Result<impl IntoResponse> {
32 33 let csrf_token = get_csrf_token(&session).await;
33 - let comp_codes = load_comp_code_rows(&state).await?;
34 + let comp_codes = load_comp_code_rows(&db).await?;
34 35 Ok(AdminCompCodesTemplate {
35 36 csrf_token,
36 37 session_user: Some(user),
@@ -76,7 +77,7 @@ where
76 77 /// new code appears immediately.
77 78 #[tracing::instrument(skip_all, name = "admin::admin_create_comp_code")]
78 79 pub(super) async fn admin_create_comp_code(
79 - State(state): State<AppState>,
80 + State(db): State<PgPool>,
80 81 AdminUser(admin): AdminUser,
81 82 Form(form): Form<CompCodeForm>,
82 83 ) -> Result<Response> {
@@ -92,7 +93,7 @@ pub(super) async fn admin_create_comp_code(
92 93 .map(|d| chrono::Utc::now() + chrono::Duration::days(d));
93 94
94 95 db::promo_codes::create_platform_promo_code(
95 - &state.db,
96 + &db,
96 97 admin.id,
97 98 &code,
98 99 db::CodePurpose::FreeTrial,
@@ -107,12 +108,12 @@ pub(super) async fn admin_create_comp_code(
107 108
108 109 tracing::info!(code = %code, trial_days = form.trial_days, "minted creator-tier comp code");
109 110
110 - let comp_codes = load_comp_code_rows(&state).await?;
111 + let comp_codes = load_comp_code_rows(&db).await?;
111 112 Ok(AdminCompCodesEntriesTemplate { comp_codes }.into_response())
112 113 }
113 114
114 115 /// Load and format the comp-code rows for the dashboard.
115 - async fn load_comp_code_rows(state: &AppState) -> Result<Vec<AdminCompCodeRow>> {
116 - let codes = db::promo_codes::get_platform_trial_codes(&state.db).await?;
116 + async fn load_comp_code_rows(db: &PgPool) -> Result<Vec<AdminCompCodeRow>> {
117 + let codes = db::promo_codes::get_platform_trial_codes(db).await?;
117 118 Ok(codes.iter().map(AdminCompCodeRow::from_db).collect())
118 119 }
M server/src/routes/admin/mod.rs +23 -18
@@ -18,13 +18,15 @@ use axum::{
18 18 use axum::http::request::Parts;
19 19 use axum::extract::FromRequestParts;
20 20 use serde::Deserialize;
21 + use sqlx::PgPool;
21 22
22 23 use crate::{
23 24 auth::AdminUser,
24 25 csrf::{post_csrf, CsrfRouter},
25 26 db::{self, UserId},
27 + email::EmailClient,
26 28 error::{AppError, Result},
27 - AppState,
29 + AppState, Integrations, Ops,
28 30 };
29 31
30 32 /// `/admin*` routes served WITHOUT the admin gate, by explicit design.
@@ -129,21 +131,22 @@ pub fn admin_routes(state: AppState) -> CsrfRouter<AppState> {
129 131 /// Backfill MT communities for all projects that don't have one yet.
130 132 #[tracing::instrument(skip_all, name = "admin::admin_mt_provision")]
131 133 async fn admin_mt_provision(
132 - State(state): State<AppState>,
134 + State(db): State<PgPool>,
135 + State(integrations): State<Integrations>,
133 136 AdminUser(_admin): AdminUser,
134 137 ) -> Result<Response> {
135 - let Some(ref mt) = state.mt_client else {
138 + let Some(ref mt) = integrations.mt_client else {
136 139 return Err(AppError::validation("MT integration not configured".to_string()));
137 140 };
138 141
139 - let projects = db::projects::get_projects_without_mt_community(&state.db).await?;
142 + let projects = db::projects::get_projects_without_mt_community(&db).await?;
140 143 let total = projects.len();
141 144 let mut provisioned = 0u32;
142 145 let mut failed = 0u32;
143 146
144 147 // Batch-load all project owners in one query instead of N individual lookups
145 148 let owner_ids: Vec<db::UserId> = projects.iter().map(|p| p.user_id).collect();
146 - let owners = db::users::get_users_by_ids(&state.db, &owner_ids).await?;
149 + let owners = db::users::get_users_by_ids(&db, &owner_ids).await?;
147 150 let owner_map: std::collections::HashMap<db::UserId, &db::DbUser> =
148 151 owners.iter().map(|u| (u.id, u)).collect();
149 152
@@ -166,7 +169,7 @@ async fn admin_mt_provision(
166 169 }
167 170
168 171 let mt = mt.clone();
169 - let pool = state.db.clone();
172 + let pool = db.clone();
170 173 let req = crate::mt_client::CreateCommunityRequest {
171 174 name: project.title.clone(),
172 175 slug: project.slug.to_string(),
@@ -220,7 +223,8 @@ pub(super) struct ShutdownNoticeForm {
220 223 /// Send a shutdown notice email to all users.
221 224 #[tracing::instrument(skip_all, name = "admin::admin_shutdown_notice")]
222 225 async fn admin_shutdown_notice(
223 - State(state): State<AppState>,
226 + State(db): State<PgPool>,
227 + State(email): State<EmailClient>,
224 228 AdminUser(_admin): AdminUser,
225 229 Form(form): Form<ShutdownNoticeForm>,
226 230 ) -> Result<Response> {
@@ -229,14 +233,14 @@ async fn admin_shutdown_notice(
229 233 return Err(AppError::validation("Shutdown date is required".to_string()));
230 234 }
231 235
232 - let all_users = db::users::get_all_user_emails(&state.db).await?;
236 + let all_users = db::users::get_all_user_emails(&db).await?;
233 237 let count = all_users.len();
234 238
235 239 // Fan out on a background task with a bounded JoinSet (mirrors the broadcast
236 240 // path). The previous version sent one email at a time, inline, scaling the
237 241 // request's latency with the entire user base and tying up a handler for the
238 242 // whole send. Return immediately; the task logs its own totals.
239 - let email_client = state.email.clone();
243 + let email_client = email.clone();
240 244 let shutdown_date = shutdown_date.to_string();
241 245 tokio::spawn(async move {
242 246 let mut set = tokio::task::JoinSet::new();
@@ -296,10 +300,10 @@ async fn admin_shutdown_notice(
296 300 /// `project_founder_pricing.md` for the full plan.
297 301 #[tracing::instrument(skip_all, name = "admin::admin_close_founder_window")]
298 302 async fn admin_close_founder_window(
299 - State(state): State<AppState>,
303 + State(db): State<PgPool>,
300 304 AdminUser(_admin): AdminUser,
301 305 ) -> Result<Response> {
302 - let locked = db::users::lock_in_founders_with_active_subscriptions(&state.db).await?;
306 + let locked = db::users::lock_in_founders_with_active_subscriptions(&db).await?;
303 307 tracing::info!(locked = locked, "founder window close: users stamped with founder_locked_at");
304 308 Ok((
305 309 axum::http::StatusCode::OK,
@@ -317,13 +321,14 @@ async fn admin_close_founder_window(
317 321 /// Render the admin metrics dashboard with live Prometheus data.
318 322 #[tracing::instrument(skip_all, name = "admin::admin_metrics")]
319 323 async fn admin_metrics(
320 - State(state): State<AppState>,
324 + State(db): State<PgPool>,
325 + State(ops): State<Ops>,
321 326 session: tower_sessions::Session,
322 327 AdminUser(user): AdminUser,
323 328 ) -> Result<impl IntoResponse> {
324 329 let csrf_token = crate::helpers::get_csrf_token(&session).await;
325 330 let uptime = {
326 - let d = state.start_instant.elapsed();
331 + let d = ops.start_instant.elapsed();
327 332 let secs = d.as_secs();
328 333 let days = secs / 86400;
329 334 let hours = (secs % 86400) / 3600;
@@ -333,13 +338,13 @@ async fn admin_metrics(
333 338 else { format!("{mins}m") }
334 339 };
335 340
336 - let pool_size = state.db.size();
337 - let pool_idle = state.db.num_idle() as u32;
341 + let pool_size = db.size();
342 + let pool_idle = db.num_idle() as u32;
338 343 let pool_active = pool_size.saturating_sub(pool_idle);
339 344
340 345 // Parse metrics from the Prometheus handle (if available)
341 346 let (total_requests, error_rate, total_errors, top_routes, error_breakdown) =
342 - if let Some(ref handle) = state.metrics_handle {
347 + if let Some(ref handle) = ops.metrics_handle {
343 348 let snap = crate::metrics::snapshot(handle);
344 349 let rate = if snap.total_requests > 0 {
345 350 snap.total_5xx as f64 / snap.total_requests as f64 * 100.0
@@ -385,7 +390,7 @@ pub(super) struct FileOverrideForm {
385 390 /// POST /api/admin/users/{id}/file-override
386 391 #[tracing::instrument(skip_all, name = "admin::admin_file_override")]
387 392 async fn admin_file_override(
388 - State(state): State<AppState>,
393 + State(db): State<PgPool>,
389 394 AdminUser(_admin): AdminUser,
390 395 Path(user_id): Path<UserId>,
391 396 Form(form): Form<FileOverrideForm>,
@@ -397,7 +402,7 @@ async fn admin_file_override(
397 402 return Err(AppError::BadRequest("Override must be a positive number of bytes".to_string()));
398 403 }
399 404
400 - db::creator_tiers::set_max_file_override(&state.db, user_id, form.max_file_bytes).await?;
405 + db::creator_tiers::set_max_file_override(&db, user_id, form.max_file_bytes).await?;
401 406
402 407 let msg = match form.max_file_bytes {
403 408 Some(bytes) => format!("File override set to {}", crate::helpers::format_bytes(bytes)),
M server/src/routes/admin/moderation.rs +60 -35
@@ -6,15 +6,18 @@ use axum::{
6 6 Form,
7 7 };
8 8 use serde::Deserialize;
9 + use sqlx::PgPool;
9 10
10 11 use crate::{
11 12 auth::AdminUser,
13 + background::BackgroundTx,
12 14 db::{self, AppealDecision, ItemId, ModerationActionType, ReportId, ReportStatus, UserId},
15 + email::EmailClient,
13 16 error::{AppError, Result},
14 - helpers::{get_csrf_token, spawn_email},
17 + helpers::get_csrf_token,
15 18 templates::*,
16 19 types::*,
17 - AppState,
20 + AppCaches, Billing, Integrations,
18 21 };
19 22
20 23 // ── Appeals ──
@@ -22,13 +25,13 @@ use crate::{
22 25 /// Render the admin appeals queue.
23 26 #[tracing::instrument(skip_all, name = "admin::admin_appeals")]
24 27 pub(super) async fn admin_appeals(
25 - State(state): State<AppState>,
28 + State(db): State<PgPool>,
26 29 session: tower_sessions::Session,
27 30 AdminUser(user): AdminUser,
28 31 ) -> Result<impl IntoResponse> {
29 32 let csrf_token = get_csrf_token(&session).await;
30 33
31 - let db_users = db::users::get_pending_appeals(&state.db).await?;
34 + let db_users = db::users::get_pending_appeals(&db).await?;
32 35 let appeals: Vec<AdminAppealRow> = db_users.iter().map(AdminAppealRow::from_db).collect();
33 36
34 37 Ok(AdminAppealsTemplate {
@@ -47,8 +50,14 @@ pub(super) struct AppealDecisionForm {
47 50
48 51 /// Decide an appeal (approve or deny) and send notification email.
49 52 #[tracing::instrument(skip_all, name = "admin::admin_decide_appeal")]
53 + #[allow(clippy::too_many_arguments)]
50 54 pub(super) async fn admin_decide_appeal(
51 - State(state): State<AppState>,
55 + State(db): State<PgPool>,
56 + State(email): State<EmailClient>,
57 + State(bg): State<BackgroundTx>,
58 + State(caches): State<AppCaches>,
59 + State(payments): State<Billing>,
60 + State(integrations): State<Integrations>,
52 61 AdminUser(_admin): AdminUser,
53 62 Path(user_id): Path<UserId>,
54 63 Form(form): Form<AppealDecisionForm>,
@@ -59,7 +68,7 @@ pub(super) async fn admin_decide_appeal(
59 68 }
60 69
61 70 // Get user for email notification
62 - let db_user = db::users::get_user_by_id(&state.db, user_id)
71 + let db_user = db::users::get_user_by_id(&db, user_id)
63 72 .await?
64 73 .ok_or(AppError::NotFound)?;
65 74
@@ -67,13 +76,13 @@ pub(super) async fn admin_decide_appeal(
67 76 // uses): resolve the appeal, resume fan subscriptions on approval, and email
68 77 // the outcome. Web fans the Stripe resume out on the background queue.
69 78 super::moderation_service::decide_appeal(
70 - &state.db,
71 - &state.email,
72 - state.stripe.as_ref(),
79 + &db,
80 + &email,
81 + payments.stripe.as_ref(),
73 82 super::moderation_service::FanoutMode::Background {
74 - bg: &state.bg,
75 - wam: state.wam.clone(),
76 - session_cache: &state.caches.session_cache,
83 + bg: &bg,
84 + wam: integrations.wam.clone(),
85 + session_cache: &caches.session_cache,
77 86 },
78 87 &db_user,
79 88 form.decision,
@@ -82,7 +91,7 @@ pub(super) async fn admin_decide_appeal(
82 91 .await?;
83 92
84 93 // Return updated appeals list
85 - let db_users = db::users::get_pending_appeals(&state.db).await?;
94 + let db_users = db::users::get_pending_appeals(&db).await?;
86 95 let appeals: Vec<AdminAppealRow> = db_users.iter().map(AdminAppealRow::from_db).collect();
87 96 Ok(AdminAppealEntriesTemplate { appeals })
88 97 }
@@ -97,7 +106,7 @@ pub(super) struct ReportFilterQuery {
97 106 /// Render the admin reports queue.
98 107 #[tracing::instrument(skip_all, name = "admin::admin_reports")]
99 108 pub(super) async fn admin_reports(
100 - State(state): State<AppState>,
109 + State(db): State<PgPool>,
101 110 session: tower_sessions::Session,
102 111 AdminUser(user): AdminUser,
103 112 Query(query): Query<ReportFilterQuery>,
@@ -105,14 +114,14 @@ pub(super) async fn admin_reports(
105 114 let csrf_token = get_csrf_token(&session).await;
106 115 let current_filter = query.status.clone().unwrap_or_default();
107 116
108 - let db_stats = db::reports::get_report_stats(&state.db).await?;
117 + let db_stats = db::reports::get_report_stats(&db).await?;
109 118 let stats = ReportStats {
110 119 open: db_stats.open as u32,
111 120 resolved: db_stats.resolved as u32,
112 121 dismissed: db_stats.dismissed as u32,
113 122 };
114 123
115 - let db_reports = db::reports::get_admin_reports(&state.db, query.status.as_deref(), 100, 0).await?;
124 + let db_reports = db::reports::get_admin_reports(&db, query.status.as_deref(), 100, 0).await?;
116 125 let reports: Vec<AdminReportRow> = db_reports.iter().map(AdminReportRow::from_db).collect();
117 126
118 127 Ok(AdminReportsTemplate {
@@ -128,11 +137,11 @@ pub(super) async fn admin_reports(
128 137 /// Return filtered report entries as an HTMX partial.
129 138 #[tracing::instrument(skip_all, name = "admin::admin_report_entries")]
130 139 pub(super) async fn admin_report_entries(
131 - State(state): State<AppState>,
140 + State(db): State<PgPool>,
132 141 AdminUser(_user): AdminUser,
133 142 Query(query): Query<ReportFilterQuery>,
134 143 ) -> Result<impl IntoResponse> {
135 - let db_reports = db::reports::get_admin_reports(&state.db, query.status.as_deref(), 100, 0).await?;
144 + let db_reports = db::reports::get_admin_reports(&db, query.status.as_deref(), 100, 0).await?;
136 145 let reports: Vec<AdminReportRow> = db_reports.iter().map(AdminReportRow::from_db).collect();
137 146
138 147 Ok(AdminReportEntriesTemplate { reports })
@@ -148,7 +157,7 @@ pub(super) struct ReportDecisionForm {
148 157 /// Resolve or dismiss a report.
149 158 #[tracing::instrument(skip_all, name = "admin::admin_resolve_report")]
150 159 pub(super) async fn admin_resolve_report(
151 - State(state): State<AppState>,
160 + State(db): State<PgPool>,
152 161 admin_user: AdminUser,
153 162 Path(id): Path<ReportId>,
154 163 Form(form): Form<ReportDecisionForm>,
@@ -160,7 +169,7 @@ pub(super) async fn admin_resolve_report(
160 169 };
161 170
162 171 db::reports::resolve_report(
163 - &state.db,
172 + &db,
164 173 id,
165 174 status,
166 175 form.admin_notes.trim(),
@@ -170,7 +179,7 @@ pub(super) async fn admin_resolve_report(
170 179 tracing::info!(report_id = %id, decision = %form.decision, "admin resolved report");
171 180
172 181 // Return updated entries (open filter)
173 - let db_reports = db::reports::get_admin_reports(&state.db, Some("open"), 100, 0).await?;
182 + let db_reports = db::reports::get_admin_reports(&db, Some("open"), 100, 0).await?;
174 183 let reports: Vec<AdminReportRow> = db_reports.iter().map(AdminReportRow::from_db).collect();
175 184
176 185 Ok(AdminReportEntriesTemplate { reports })
@@ -188,7 +197,9 @@ pub(super) struct ItemRemovalForm {
188 197 /// Sets `removed_by_admin = true`, hides the item, and emails the creator with the reason.
189 198 #[tracing::instrument(skip_all, name = "admin::admin_remove_item")]
190 199 pub(super) async fn admin_remove_item(
191 - State(state): State<AppState>,
200 + State(db): State<PgPool>,
201 + State(email): State<EmailClient>,
202 + State(bg): State<BackgroundTx>,
192 203 admin_user: AdminUser,
193 204 Path(item_id): Path<ItemId>,
194 205 Form(form): Form<ItemRemovalForm>,
@@ -198,26 +209,32 @@ pub(super) async fn admin_remove_item(
198 209 return Err(AppError::validation("Removal reason is required".to_string()));
199 210 }
200 211
201 - let item = db::items::admin_remove_item(&state.db, item_id, reason).await?;
212 + let item = db::items::admin_remove_item(&db, item_id, reason).await?;
202 213
203 214 // Look up the creator to send notification email
204 - let owner_id = db::items::get_item_owner(&state.db, item_id)
215 + let owner_id = db::items::get_item_owner(&db, item_id)
205 216 .await?
206 217 .ok_or(AppError::NotFound)?;
207 218
208 - if let Ok(Some(owner)) = db::users::get_user_by_id(&state.db, owner_id).await {
219 + if let Ok(Some(owner)) = db::users::get_user_by_id(&db, owner_id).await {
209 220 let owner_email = owner.email.clone();
210 221 let owner_name = owner.display_name.clone();
211 222 let item_title = item.title.clone();
212 223 let reason = reason.to_string();
213 - spawn_email!(state, "content removal notification", |email| {
214 - email.send_content_removal(&owner_email, owner_name.as_deref(), &item_title, &reason)
224 + let email = email.clone();
225 + bg.spawn("content removal notification", async move {
226 + if let Err(e) = email
227 + .send_content_removal(&owner_email, owner_name.as_deref(), &item_title, &reason)
228 + .await
229 + {
230 + tracing::error!(error = ?e, "failed to send content removal notification");
231 + }
215 232 });
216 233 }
217 234
218 235 // Record moderation action against the item owner
219 236 db::moderation::create_action(
220 - &state.db, owner_id, admin_user.admin_id(), ModerationActionType::ContentRemoval, reason, Some(&item_id.to_string()),
237 + &db, owner_id, admin_user.admin_id(), ModerationActionType::ContentRemoval, reason, Some(&item_id.to_string()),
221 238 ).await?;
222 239
223 240 tracing::info!(
@@ -233,28 +250,36 @@ pub(super) async fn admin_remove_item(
233 250 /// Restore a previously admin-removed item (clears removal, creator must re-publish).
234 251 #[tracing::instrument(skip_all, name = "admin::admin_restore_item")]
235 252 pub(super) async fn admin_restore_item(
236 - State(state): State<AppState>,
253 + State(db): State<PgPool>,
254 + State(email): State<EmailClient>,
255 + State(bg): State<BackgroundTx>,
237 256 admin_user: AdminUser,
238 257 Path(item_id): Path<ItemId>,
239 258 ) -> Result<impl IntoResponse> {
240 - let item = db::items::admin_restore_item(&state.db, item_id).await?;
259 + let item = db::items::admin_restore_item(&db, item_id).await?;
241 260
242 261 // Notify creator their item was restored
243 - let owner_id = db::items::get_item_owner(&state.db, item_id)
262 + let owner_id = db::items::get_item_owner(&db, item_id)
244 263 .await?
245 264 .ok_or(AppError::NotFound)?;
246 265
247 - if let Ok(Some(owner)) = db::users::get_user_by_id(&state.db, owner_id).await {
266 + if let Ok(Some(owner)) = db::users::get_user_by_id(&db, owner_id).await {
248 267 let owner_email = owner.email.clone();
249 268 let owner_name = owner.display_name.clone();
250 269 let item_title = item.title.clone();
251 - spawn_email!(state, "content restore notification", |email| {
252 - email.send_content_restored(&owner_email, owner_name.as_deref(), &item_title)
270 + let email = email.clone();
271 + bg.spawn("content restore notification", async move {
272 + if let Err(e) = email
273 + .send_content_restored(&owner_email, owner_name.as_deref(), &item_title)
274 + .await
275 + {
276 + tracing::error!(error = ?e, "failed to send content restore notification");
277 + }
253 278 });
254 279 }
255 280
256 281 // Resolve the content_removal moderation action
257 - db::moderation::resolve_content_removal(&state.db, &item_id.to_string()).await?;
282 + db::moderation::resolve_content_removal(&db, &item_id.to_string()).await?;
258 283
259 284 tracing::info!(
260 285 item_id = %item_id,
M server/src/routes/admin/signups.rs +4 -4
@@ -1,4 +1,5 @@
1 1 use axum::{extract::State, response::IntoResponse};
2 + use sqlx::PgPool;
2 3 use tower_sessions::Session;
3 4
4 5 use crate::{
@@ -8,18 +9,17 @@ use crate::{
8 9 helpers::get_csrf_token,
9 10 templates::AdminSignupsTemplate,
10 11 types::AdminSignupRow,
11 - AppState,
12 12 };
13 13
14 14 /// Admin page listing email signups from the landing page.
15 15 #[tracing::instrument(skip_all, name = "admin::admin_signups")]
16 16 pub(super) async fn admin_signups(
17 - State(state): State<AppState>,
17 + State(db): State<PgPool>,
18 18 session: Session,
19 19 AdminUser(admin): AdminUser,
20 20 ) -> Result<impl IntoResponse> {
21 - let db_signups = db::email_signups::get_all_email_signups(&state.db).await?;
22 - let total = db::email_signups::count_email_signups(&state.db).await?;
21 + let db_signups = db::email_signups::get_all_email_signups(&db).await?;
22 + let total = db::email_signups::count_email_signups(&db).await?;
23 23
24 24 let signups: Vec<AdminSignupRow> = db_signups
25 25 .into_iter()
M server/src/routes/admin/uploads.rs +100 -81
@@ -9,16 +9,18 @@ use axum::{
9 9 extract::{Path, State},
10 10 response::{IntoResponse, Response},
11 11 };
12 + use sqlx::PgPool;
12 13
13 14 use crate::{
14 15 auth::AdminUser,
16 + config::Config,
15 17 db::{self, scan_admin_actions::AdminAction, FileScanStatus, ItemId, UserId, VersionId},
16 18 error::Result,
17 19 helpers::get_csrf_token,
18 20 storage::FileType,
19 21 templates::*,
20 22 types::*,
21 - AppState,
23 + AppStorage, Scanning,
22 24 };
23 25
24 26 const HEALTH_WINDOW_HOURS: i64 = 24;
@@ -28,8 +30,8 @@ const HISTORY_ROW_LIMIT: i64 = 100;
28 30 const AUDIT_LOG_ROW_LIMIT: i64 = 500;
29 31
30 32 /// Build the per-layer health-card vec from the DB rollup.
31 - async fn fetch_layer_health(state: &AppState) -> Result<Vec<LayerHealthCard>> {
32 - let rows = db::scanning::layer_health_window(&state.db, HEALTH_WINDOW_HOURS).await?;
33 + async fn fetch_layer_health(db: &PgPool) -> Result<Vec<LayerHealthCard>> {
34 + let rows = db::scanning::layer_health_window(db, HEALTH_WINDOW_HOURS).await?;
33 35 let mut by_layer: HashMap<String, LayerHealthCard> = HashMap::new();
34 36
35 37 // The canonical layer list — keep cards visible even if a layer hasn't
@@ -106,14 +108,14 @@ fn relative_age(t: chrono::DateTime<chrono::Utc>) -> String {
106 108 /// Render the admin upload review queue.
107 109 #[tracing::instrument(skip_all, name = "admin::admin_uploads")]
108 110 pub(super) async fn admin_uploads(
109 - State(state): State<AppState>,
111 + State(db): State<PgPool>,
110 112 session: tower_sessions::Session,
111 113 AdminUser(user): AdminUser,
112 114 ) -> Result<impl IntoResponse> {
113 115 let csrf_token = get_csrf_token(&session).await;
114 116
115 - let held_items = db::scanning::get_held_items(&state.db).await?;
116 - let held_versions = db::scanning::get_held_versions(&state.db).await?;
117 + let held_items = db::scanning::get_held_items(&db).await?;
118 + let held_versions = db::scanning::get_held_versions(&db).await?;
117 119 // The held queries are capped at 1000 rows each (oldest first). Hitting the
118 120 // cap is an incident-scale backlog; surface it rather than silently truncate.
119 121 if held_items.len() >= 1000 || held_versions.len() >= 1000 {
@@ -130,15 +132,15 @@ pub(super) async fn admin_uploads(
130 132 held_uploads.sort_by(|a, b| a.held_at.cmp(&b.held_at));
131 133
132 134 // Attach the last admin action to each held row in a single batch lookup.
133 - attach_last_actions(&state, &mut held_uploads).await;
135 + attach_last_actions(&db, &mut held_uploads).await;
134 136
135 137 let total_held = held_uploads.len();
136 - let layer_health = fetch_layer_health(&state).await?;
137 - let queue_pending = db::scan_jobs::queued_count(&state.db).await.unwrap_or(0);
138 - let queue_running = db::scan_jobs::running_count(&state.db).await.unwrap_or(0);
138 + let layer_health = fetch_layer_health(&db).await?;
139 + let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
140 + let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
139 141
140 142 let recent_history: Vec<ScanHistoryDisplay> = db::scanning::recent_history(
141 - &state.db, HISTORY_WINDOW_HOURS, HISTORY_ROW_LIMIT,
143 + &db, HISTORY_WINDOW_HOURS, HISTORY_ROW_LIMIT,
142 144 ).await
143 145 .unwrap_or_default()
144 146 .iter()
@@ -162,7 +164,7 @@ pub(super) async fn admin_uploads(
162 164
163 165 /// Batch-fetch the latest admin action for the displayed held rows and
164 166 /// attach each as `last_action` on its row.
165 - async fn attach_last_actions(state: &AppState, rows: &mut [AdminHeldUploadRow]) {
167 + async fn attach_last_actions(db: &PgPool, rows: &mut [AdminHeldUploadRow]) {
166 168 use std::str::FromStr;
167 169 let mut version_ids: Vec<uuid::Uuid> = Vec::new();
168 170 let mut item_ids: Vec<uuid::Uuid> = Vec::new();
@@ -175,9 +177,9 @@ async fn attach_last_actions(state: &AppState, rows: &mut [AdminHeldUploadRow])
175 177 item_ids.push(uuid);
176 178 }
177 179 }
178 - let version_actions = db::scan_admin_actions::latest_per_version(&state.db, &version_ids)
180 + let version_actions = db::scan_admin_actions::latest_per_version(db, &version_ids)
179 181 .await.unwrap_or_default();
180 - let item_actions = db::scan_admin_actions::latest_per_item(&state.db, &item_ids)
182 + let item_actions = db::scan_admin_actions::latest_per_item(db, &item_ids)
181 183 .await.unwrap_or_default();
182 184
183 185 for r in rows.iter_mut() {
@@ -212,7 +214,7 @@ pub(super) struct AuditFilters {
212 214 /// filters are optional; empty/absent means no constraint on that column.
213 215 #[tracing::instrument(skip_all, name = "admin::scan_audit")]
214 216 pub(super) async fn admin_scan_audit(
215 - State(state): State<AppState>,
217 + State(db): State<PgPool>,
216 218 session: tower_sessions::Session,
217 219 AdminUser(user): AdminUser,
218 220 axum::extract::Query(filters): axum::extract::Query<AuditFilters>,
@@ -221,7 +223,7 @@ pub(super) async fn admin_scan_audit(
221 223 let action = filters.action.as_deref().filter(|s| !s.is_empty());
222 224 let admin = filters.admin.as_deref().filter(|s| !s.is_empty());
223 225 let entries: Vec<AdminAuditLogRow> = db::scan_admin_actions::list_filtered(
224 - &state.db, action, admin, filters.since_days, AUDIT_LOG_ROW_LIMIT,
226 + &db, action, admin, filters.since_days, AUDIT_LOG_ROW_LIMIT,
225 227 ).await
226 228 .unwrap_or_default()
227 229 .iter()
@@ -239,9 +241,9 @@ pub(super) async fn admin_scan_audit(
239 241 }
240 242
241 243 /// Re-query held uploads and return the entries partial.
242 - pub(super) async fn refresh_held_uploads_partial(state: &AppState) -> Result<Response> {
243 - let held_items = db::scanning::get_held_items(&state.db).await?;
244 - let held_versions = db::scanning::get_held_versions(&state.db).await?;
244 + pub(super) async fn refresh_held_uploads_partial(db: &PgPool) -> Result<Response> {
245 + let held_items = db::scanning::get_held_items(db).await?;
246 + let held_versions = db::scanning::get_held_versions(db).await?;
245 247
246 248 let mut held_uploads: Vec<AdminHeldUploadRow> = Vec::new();
247 249 held_uploads.extend(held_items.iter().map(AdminHeldUploadRow::from_held_item));
@@ -256,74 +258,78 @@ pub(super) async fn refresh_held_uploads_partial(state: &AppState) -> Result<Res
256 258 /// Promote a held item upload to Clean. Renames the legacy "approve" verb.
257 259 #[tracing::instrument(skip_all, name = "admin::promote_item")]
258 260 pub(super) async fn admin_promote_item(
259 - State(state): State<AppState>,
261 + State(db): State<PgPool>,
262 + State(storage): State<AppStorage>,
263 + State(config): State<Config>,
260 264 AdminUser(admin): AdminUser,
261 265 Path(id): Path<ItemId>,
262 266 ) -> Result<Response> {
263 267 // Copy each held (staging) file to its content key and mark Clean in one
264 268 // step — a held file sits unserved at its staging key, so approving must run
265 269 // the same promote the scan worker's Clean path runs (C1).
266 - crate::routes::storage::commit_promote_item(&state.db, &state.storage, &state.config, id).await?;
270 + crate::routes::storage::commit_promote_item(&db, &storage, &config, id).await?;
267 271 db::scan_admin_actions::log_item(
268 - &state.db, id, admin.id, AdminAction::Promote,
272 + &db, id, admin.id, AdminAction::Promote,
269 273 Some("held_for_review"), Some("clean"),
270 274 None,
271 275 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
272 276 tracing::info!(item_id = %id, admin_id = %admin.id, "item promoted to clean");
273 - refresh_held_uploads_partial(&state).await
277 + refresh_held_uploads_partial(&db).await
274 278 }
275 279
276 280 /// Quarantine a held item upload. Renames the legacy "reject" verb.
277 281 #[tracing::instrument(skip_all, name = "admin::quarantine_item")]
278 282 pub(super) async fn admin_quarantine_item(
279 - State(state): State<AppState>,
283 + State(db): State<PgPool>,
280 284 AdminUser(admin): AdminUser,
281 285 Path(id): Path<ItemId>,
282 286 ) -> Result<Response> {
283 - db::scanning::update_item_scan_status(&state.db, id, FileScanStatus::Quarantined).await?;
287 + db::scanning::update_item_scan_status(&db, id, FileScanStatus::Quarantined).await?;
284 288 db::scan_admin_actions::log_item(
285 - &state.db, id, admin.id, AdminAction::Quarantine,
289 + &db, id, admin.id, AdminAction::Quarantine,
286 290 Some("held_for_review"), Some("quarantined"),
287 291 None,
288 292 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
289 293 tracing::info!(item_id = %id, admin_id = %admin.id, "item quarantined");
290 - refresh_held_uploads_partial(&state).await
294 + refresh_held_uploads_partial(&db).await
291 295 }
292 296
293 297 /// Promote a held version upload to Clean.
294 298 #[tracing::instrument(skip_all, name = "admin::promote_version")]
295 299 pub(super) async fn admin_promote_version(
296 - State(state): State<AppState>,
300 + State(db): State<PgPool>,
301 + State(storage): State<AppStorage>,
302 + State(config): State<Config>,
297 303 AdminUser(admin): AdminUser,
298 304 Path(id): Path<VersionId>,
299 305 ) -> Result<Response> {
300 306 // See admin_promote_item: copy the held (staging) download to its content key
301 307 // and mark Clean together (C1).
302 - crate::routes::storage::commit_promote_version(&state.db, &state.storage, &state.config, id).await?;
308 + crate::routes::storage::commit_promote_version(&db, &storage, &config, id).await?;
303 309 db::scan_admin_actions::log_version(
304 - &state.db, id, admin.id, AdminAction::Promote,
310 + &db, id, admin.id, AdminAction::Promote,
305 311 Some("held_for_review"), Some("clean"),
306 312 None,
307 313 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
308 314 tracing::info!(version_id = %id, admin_id = %admin.id, "version promoted to clean");
309 - refresh_held_uploads_partial(&state).await
315 + refresh_held_uploads_partial(&db).await
310 316 }
311 317
312 318 /// Quarantine a held version upload.
313 319 #[tracing::instrument(skip_all, name = "admin::quarantine_version")]
314 320 pub(super) async fn admin_quarantine_version(
315 - State(state): State<AppState>,
321 + State(db): State<PgPool>,
316 322 AdminUser(admin): AdminUser,
317 323 Path(id): Path<VersionId>,
318 324 ) -> Result<Response> {
319 - db::scanning::update_version_scan_status(&state.db, id, FileScanStatus::Quarantined).await?;
325 + db::scanning::update_version_scan_status(&db, id, FileScanStatus::Quarantined).await?;
320 326 db::scan_admin_actions::log_version(
321 - &state.db, id, admin.id, AdminAction::Quarantine,
327 + &db, id, admin.id, AdminAction::Quarantine,
322 328 Some("held_for_review"), Some("quarantined"),
323 329 None,
324 330 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
325 331 tracing::info!(version_id = %id, admin_id = %admin.id, "version quarantined");
326 - refresh_held_uploads_partial(&state).await
332 + refresh_held_uploads_partial(&db).await
327 333 }
328 334
329 335 // ─── Rescan ──────────────────────────────────────────────────────────────────
@@ -333,25 +339,27 @@ pub(super) async fn admin_quarantine_version(
333 339 /// that a scan is in flight.
334 340 #[tracing::instrument(skip_all, name = "admin::rescan_version")]
335 341 pub(super) async fn admin_rescan_version(
336 - State(state): State<AppState>,
342 + State(db): State<PgPool>,
343 + State(scanning): State<Scanning>,
337 344 AdminUser(admin): AdminUser,
338 345 Path(id): Path<VersionId>,
339 346 ) -> Result<Response> {
340 - rescan_version_inner(&state, id, admin.id, AdminAction::Rescan).await?;
341 - refresh_held_uploads_partial(&state).await
347 + rescan_version_inner(&db, scanning.scanner.as_ref(), id, admin.id, AdminAction::Rescan).await?;
348 + refresh_held_uploads_partial(&db).await
342 349 }
343 350
344 351 async fn rescan_version_inner(
345 - state: &AppState,
352 + db: &PgPool,
353 + scanner: Option<&std::sync::Arc<crate::scanning::ScanPipeline>>,
346 354 id: VersionId,
347 355 admin_id: UserId,
348 356 action: AdminAction,
349 357 ) -> Result<()> {
350 - let v = db::versions::get_version_by_id(&state.db, id).await?
358 + let v = db::versions::get_version_by_id(db, id).await?
351 359 .ok_or(crate::error::AppError::NotFound)?;
352 - let item = db::items::get_item_by_id(&state.db, v.item_id).await?
360 + let item = db::items::get_item_by_id(db, v.item_id).await?
353 361 .ok_or(crate::error::AppError::NotFound)?;
354 - let owner = db::items::get_item_owner(&state.db, item.id).await?
362 + let owner = db::items::get_item_owner(db, item.id).await?
355 363 .ok_or(crate::error::AppError::NotFound)?;
356 364 let s3_key = v.s3_key.clone().ok_or(crate::error::AppError::NotFound)?;
357 365 let size = v.file_size_bytes.unwrap_or(0);
@@ -361,8 +369,8 @@ async fn rescan_version_inner(
361 369 // to call scan_jobs::enqueue + update_*_scan_status directly, which made
362 370 // ordering bugs hard to fence at the type level.
363 371 crate::routes::storage::commit_rescan(
364 - &state.db,
365 - state.scanner.as_ref(),
372 + db,
373 + scanner,
366 374 crate::routes::storage::CommitTarget::Version(id),
367 375 &s3_key,
368 376 FileType::Download,
@@ -370,7 +378,7 @@ async fn rescan_version_inner(
370 378 size,
371 379 ).await?;
372 380 db::scan_admin_actions::log_version(
373 - &state.db, id, admin_id, action,
381 + db, id, admin_id, action,
374 382 Some("held_for_review"), Some("pending"), None,
375 383 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
376 384 tracing::info!(version_id = %id, admin_id = %admin_id, "version re-enqueued for scan");
@@ -380,23 +388,25 @@ async fn rescan_version_inner(
380 388 /// Re-enqueue a single item for scanning.
381 389 #[tracing::instrument(skip_all, name = "admin::rescan_item")]
382 390 pub(super) async fn admin_rescan_item(
383 - State(state): State<AppState>,
391 + State(db): State<PgPool>,
392 + State(scanning): State<Scanning>,
384 393 AdminUser(admin): AdminUser,
385 394 Path(id): Path<ItemId>,
386 395 ) -> Result<Response> {
387 - rescan_item_inner(&state, id, admin.id, AdminAction::Rescan).await?;
388 - refresh_held_uploads_partial(&state).await
396 + rescan_item_inner(&db, scanning.scanner.as_ref(), id, admin.id, AdminAction::Rescan).await?;
397 + refresh_held_uploads_partial(&db).await
389 398 }
390 399
391 400 async fn rescan_item_inner(
392 - state: &AppState,
401 + db: &PgPool,
402 + scanner: Option<&std::sync::Arc<crate::scanning::ScanPipeline>>,
393 403 id: ItemId,
394 404 admin_id: UserId,
395 405 action: AdminAction,
396 406 ) -> Result<()> {
397 - let item = db::items::get_item_by_id(&state.db, id).await?
407 + let item = db::items::get_item_by_id(db, id).await?
398 408 .ok_or(crate::error::AppError::NotFound)?;
399 - let owner = db::items::get_item_owner(&state.db, id).await?
409 + let owner = db::items::get_item_owner(db, id).await?
400 410 .ok_or(crate::error::AppError::NotFound)?;
401 411
402 412 let (s3_key, size, file_type) = if let Some(k) = item.audio_s3_key.clone() {
@@ -408,8 +418,8 @@ async fn rescan_item_inner(
408 418 };
409 419
410 420 crate::routes::storage::commit_rescan(
411 - &state.db,
412 - state.scanner.as_ref(),
421 + db,
422 + scanner,
413 423 crate::routes::storage::CommitTarget::Item(id),
414 424 &s3_key,
415 425 file_type,
@@ -417,7 +427,7 @@ async fn rescan_item_inner(
417 427 size,
418 428 ).await?;
419 429 db::scan_admin_actions::log_item(
420 - &state.db, id, admin_id, action,
430 + db, id, admin_id, action,
421 431 Some("held_for_review"), Some("pending"), None,
422 432 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
423 433 tracing::info!(item_id = %id, admin_id = %admin_id, "item re-enqueued for scan");
@@ -428,7 +438,8 @@ async fn rescan_item_inner(
428 438 /// accumulated under a previous broken pipeline configuration.
429 439 #[tracing::instrument(skip_all, name = "admin::bulk_rescan_held")]
430 440 pub(super) async fn admin_bulk_rescan_held(
431 - State(state): State<AppState>,
441 + State(db): State<PgPool>,
442 + State(scanning): State<Scanning>,
432 443 AdminUser(admin): AdminUser,
433 444 ) -> Result<Response> {
434 445 // Clear the backlog with bounded concurrency instead of one sequential
@@ -438,26 +449,28 @@ pub(super) async fn admin_bulk_rescan_held(
438 449 let mut set: tokio::task::JoinSet<bool> = tokio::task::JoinSet::new();
439 450 let mut total = 0usize;
440 451
441 - for cand in db::scan_jobs::rescan_candidates_versions(&state.db).await? {
452 + for cand in db::scan_jobs::rescan_candidates_versions(&db).await? {
442 453 if set.len() >= MAX_INFLIGHT && let Some(Ok(ok)) = set.join_next().await {
443 454 total += ok as usize;
444 455 }
445 - let state = state.clone();
456 + let db = db.clone();
457 + let scanner = scanning.scanner.clone();
446 458 let admin_id = admin.id;
447 459 set.spawn(async move {
448 - rescan_version_inner(&state, VersionId::from_uuid(cand.version_id), admin_id, AdminAction::BulkRescan)
460 + rescan_version_inner(&db, scanner.as_ref(), VersionId::from_uuid(cand.version_id), admin_id, AdminAction::BulkRescan)
449 461 .await
450 462 .is_ok()
451 463 });
452 464 }
453 - for cand in db::scan_jobs::rescan_candidates_items(&state.db).await? {
465 + for cand in db::scan_jobs::rescan_candidates_items(&db).await? {
454 466 if set.len() >= MAX_INFLIGHT && let Some(Ok(ok)) = set.join_next().await {
455 467 total += ok as usize;
456 468 }
457 - let state = state.clone();
469 + let db = db.clone();
470 + let scanner = scanning.scanner.clone();
458 471 let admin_id = admin.id;
459 472 set.spawn(async move {
460 - rescan_item_inner(&state, ItemId::from_uuid(cand.item_id), admin_id, AdminAction::BulkRescan)
473 + rescan_item_inner(&db, scanner.as_ref(), ItemId::from_uuid(cand.item_id), admin_id, AdminAction::BulkRescan)
461 474 .await
462 475 .is_ok()
463 476 });
@@ -468,7 +481,7 @@ pub(super) async fn admin_bulk_rescan_held(
468 481 }
469 482 }
470 483 tracing::info!(total, admin_id = %admin.id, "bulk rescan of held queue dispatched");
471 - refresh_held_uploads_partial(&state).await
484 + refresh_held_uploads_partial(&db).await
472 485 }
473 486
474 487 /// Bulk-promote every currently-held item and version to Clean. Sticky
@@ -481,7 +494,9 @@ pub(super) async fn admin_bulk_rescan_held(
481 494 /// record of *why*, so we require it.
482 495 #[tracing::instrument(skip_all, name = "admin::bulk_promote_held")]
483 496 pub(super) async fn admin_bulk_promote_held(
484 - State(state): State<AppState>,
497 + State(db): State<PgPool>,
498 + State(storage): State<AppStorage>,
499 + State(config): State<Config>,
485 500 AdminUser(admin): AdminUser,
486 501 axum::Form(form): axum::Form<BulkPromoteForm>,
487 502 ) -> Result<Response> {
@@ -492,8 +507,8 @@ pub(super) async fn admin_bulk_promote_held(
492 507 ));
493 508 }
494 509
495 - let held_items = db::scanning::get_held_items(&state.db).await?;
496 - let held_versions = db::scanning::get_held_versions(&state.db).await?;
510 + let held_items = db::scanning::get_held_items(&db).await?;
511 + let held_versions = db::scanning::get_held_versions(&db).await?;
497 512
498 513 // Bounded-concurrency promote (Run #21 Performance MODERATE). The note is
499 514 // shared across tasks via Arc; MAX_INFLIGHT caps borrowed connections.
@@ -506,17 +521,19 @@ pub(super) async fn admin_bulk_promote_held(
506 521 if set.len() >= MAX_INFLIGHT && let Some(Ok(ok)) = set.join_next().await {
507 522 total += ok as usize;
508 523 }
509 - let state = state.clone();
524 + let db = db.clone();
525 + let storage = storage.clone();
526 + let config = config.clone();
510 527 let note = note.clone();
511 528 let admin_id = admin.id;
512 529 let vid = v.version_id;
513 530 set.spawn(async move {
514 - if let Err(e) = crate::routes::storage::commit_promote_version(&state.db, &state.storage, &state.config, vid).await {
531 + if let Err(e) = crate::routes::storage::commit_promote_version(&db, &storage, &config, vid).await {
515 532 tracing::warn!(version_id = %vid, error = ?e, "bulk promote: version promote failed");
516 533 false
517 534 } else {
518 535 db::scan_admin_actions::log_version(
519 - &state.db, vid, admin_id, AdminAction::BulkPromote,
536 + &db, vid, admin_id, AdminAction::BulkPromote,
520 537 Some("held_for_review"), Some("clean"), Some(&note),
521 538 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
522 539 true
@@ -527,17 +544,19 @@ pub(super) async fn admin_bulk_promote_held(
527 544 if set.len() >= MAX_INFLIGHT && let Some(Ok(ok)) = set.join_next().await {
528 545 total += ok as usize;
529 546 }
530 - let state = state.clone();
547 + let db = db.clone();
548 + let storage = storage.clone();
549 + let config = config.clone();
531 550 let note = note.clone();
532 551 let admin_id = admin.id;
533 552 let iid = i.item_id;
534 553 set.spawn(async move {
535 - if let Err(e) = crate::routes::storage::commit_promote_item(&state.db, &state.storage, &state.config, iid).await {
554 + if let Err(e) = crate::routes::storage::commit_promote_item(&db, &storage, &config, iid).await {
536 555 tracing::warn!(item_id = %iid, error = ?e, "bulk promote: item promote failed");
537 556 false
538 557 } else {
539 558 db::scan_admin_actions::log_item(
540 - &state.db, iid, admin_id, AdminAction::BulkPromote,
559 + &db, iid, admin_id, AdminAction::BulkPromote,
541 560 Some("held_for_review"), Some("clean"), Some(&note),
542 561 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
543 562 true
@@ -550,7 +569,7 @@ pub(super) async fn admin_bulk_promote_held(
550 569 }
551 570 }
552 571 tracing::warn!(total, admin_id = %admin.id, note = %note, "bulk promote of held queue executed");
553 - refresh_held_uploads_partial(&state).await
572 + refresh_held_uploads_partial(&db).await
554 573 }
555 574
556 575 #[derive(serde::Deserialize)]
@@ -592,16 +611,16 @@ struct ScanPipelineHealth {
592 611 /// reaper (300s).
593 612 #[tracing::instrument(skip_all, name = "admin::scan_health_json")]
594 613 pub(super) async fn scan_health_json(
595 - State(state): State<AppState>,
614 + State(db): State<PgPool>,
596 615 ) -> Result<impl IntoResponse> {
597 - let queue_pending = db::scan_jobs::queued_count(&state.db).await.unwrap_or(0);
598 - let queue_running = db::scan_jobs::running_count(&state.db).await.unwrap_or(0);
599 - let queue_stuck = db::scan_jobs::stuck_count(&state.db, 300).await.unwrap_or(0);
600 - let held = db::scanning::held_counts(&state.db).await
616 + let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
617 + let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
618 + let queue_stuck = db::scan_jobs::stuck_count(&db, 300).await.unwrap_or(0);
619 + let held = db::scanning::held_counts(&db).await
601 620 .unwrap_or(db::scanning::HeldCounts { held_versions: 0, held_items: 0, held_media: 0 });
602 621 let held_total = held.held_versions + held.held_items + held.held_media;
603 622
604 - let rows = db::scanning::layer_health_window(&state.db, 1).await.unwrap_or_default();
623 + let rows = db::scanning::layer_health_window(&db, 1).await.unwrap_or_default();
605 624 let now = chrono::Utc::now();
606 625 let layers: Vec<LayerHealthJson> = rows.into_iter().map(|r| {
607 626 let success = r.pass_count + r.skip_count;
@@ -656,10 +675,10 @@ pub(super) async fn scan_health_json(
656 675 /// HTMX partial: current pending + running counts. Polled by the dashboard.
657 676 #[tracing::instrument(skip_all, name = "admin::queue_summary_partial")]
658 677 pub(super) async fn admin_queue_summary_partial(
659 - State(state): State<AppState>,
678 + State(db): State<PgPool>,
660 679 AdminUser(_admin): AdminUser,
661 680 ) -> Result<impl IntoResponse> {
662 - let queue_pending = db::scan_jobs::queued_count(&state.db).await.unwrap_or(0);
663 - let queue_running = db::scan_jobs::running_count(&state.db).await.unwrap_or(0);
681 + let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
682 + let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
664 683 Ok(AdminQueueSummaryTemplate { queue_pending, queue_running })
665 684 }
M server/src/routes/admin/users.rs +86 -62
@@ -6,15 +6,18 @@ use axum::{
6 6 Form,
7 7 };
8 8 use serde::Deserialize;
9 + use sqlx::PgPool;
9 10
10 11 use crate::{
11 12 auth::AdminUser,
13 + background::BackgroundTx,
12 14 db::{self, ModerationActionType, UserId},
15 + email::EmailClient,
13 16 error::{AppError, Result},
14 - helpers::{get_csrf_token, spawn_email},
17 + helpers::get_csrf_token,
15 18 templates::*,
16 19 types::*,
17 - AppState,
20 + AppCaches, Billing, Integrations,
18 21 };
19 22
20 23 #[derive(Debug, Deserialize)]
@@ -26,7 +29,7 @@ pub(super) struct UserFilterQuery {
26 29 /// Render the admin user management page.
27 30 #[tracing::instrument(skip_all, name = "admin::admin_users")]
28 31 pub(super) async fn admin_users(
29 - State(state): State<AppState>,
32 + State(db): State<PgPool>,
30 33 session: tower_sessions::Session,
31 34 AdminUser(user): AdminUser,
32 35 Query(query): Query<UserFilterQuery>,
@@ -42,19 +45,19 @@ pub(super) async fn admin_users(
42 45 let per_page: i64 = 50;
43 46 let offset = (page - 1) * per_page;
44 47
45 - let (total_users_i64, total_suspended_i64) = db::users::count_users_summary(&state.db).await?;
48 + let (total_users_i64, total_suspended_i64) = db::users::count_users_summary(&db).await?;
46 49 let total_users = total_users_i64 as usize;
47 50 let total_suspended = total_suspended_i64 as usize;
48 51
49 52 let total_count = match query.status.as_deref() {
50 53 Some("suspended") => total_suspended_i64,
51 54 Some("active") => total_users_i64 - total_suspended_i64,
52 - Some(f @ ("custom_pages" | "pages_locked")) => db::users::count_users(&state.db, Some(f)).await?,
55 + Some(f @ ("custom_pages" | "pages_locked")) => db::users::count_users(&db, Some(f)).await?,
53 56 _ => total_users_i64,
54 57 };
55 58 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
56 59
57 - let db_users = db::users::get_all_users(&state.db, query.status.as_deref(), per_page, offset).await?;
60 + let db_users = db::users::get_all_users(&db, query.status.as_deref(), per_page, offset).await?;
58 61
59 62 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
60 63
@@ -74,7 +77,7 @@ pub(super) async fn admin_users(
74 77 /// Return filtered user entries as an HTMX partial.
75 78 #[tracing::instrument(skip_all, name = "admin::admin_user_entries")]
76 79 pub(super) async fn admin_user_entries(
77 - State(state): State<AppState>,
80 + State(db): State<PgPool>,
78 81 AdminUser(_user): AdminUser,
79 82 Query(query): Query<UserFilterQuery>,
80 83 ) -> Result<impl IntoResponse> {
@@ -87,10 +90,10 @@ pub(super) async fn admin_user_entries(
87 90 let per_page: i64 = 50;
88 91 let offset = (page - 1) * per_page;
89 92
90 - let total_count = db::users::count_users(&state.db, query.status.as_deref()).await?;
93 + let total_count = db::users::count_users(&db, query.status.as_deref()).await?;
91 94 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
92 95
93 - let db_users = db::users::get_all_users(&state.db, query.status.as_deref(), per_page, offset).await?;
96 + let db_users = db::users::get_all_users(&db, query.status.as_deref(), per_page, offset).await?;
94 97 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
95 98 Ok(AdminUserEntriesTemplate {
96 99 users,
@@ -109,7 +112,8 @@ pub(super) struct SuspendForm {
109 112 /// Records the warning in moderation history and emails the user.
110 113 #[tracing::instrument(skip_all, name = "admin::admin_warn_user")]
111 114 pub(super) async fn admin_warn_user(
112 - State(state): State<AppState>,
115 + State(db): State<PgPool>,
116 + State(email): State<EmailClient>,
113 117 admin_user: AdminUser,
114 118 Path(id): Path<UserId>,
115 119 Form(form): Form<SuspendForm>,
@@ -119,15 +123,15 @@ pub(super) async fn admin_warn_user(
119 123 return Err(AppError::validation("Reason is required".to_string()));
120 124 }
121 125
122 - let db_user = db::users::get_user_by_id(&state.db, id)
126 + let db_user = db::users::get_user_by_id(&db, id)
123 127 .await?
124 128 .ok_or(AppError::NotFound)?;
125 129
126 130 // Record warning in moderation history
127 - db::moderation::create_action(&state.db, id, admin_user.admin_id(), ModerationActionType::Warning, reason, None).await?;
131 + db::moderation::create_action(&db, id, admin_user.admin_id(), ModerationActionType::Warning, reason, None).await?;
128 132
129 133 // Send warning email
130 - if let Err(e) = state.email
134 + if let Err(e) = email
131 135 .send_policy_warning(&db_user.email, db_user.display_name.as_deref(), reason)
132 136 .await
133 137 {
@@ -136,13 +140,19 @@ pub(super) async fn admin_warn_user(
136 140
137 141 tracing::info!(user_id = %id, admin_id = %admin_user.id(), reason = %reason, "admin sent policy warning");
138 142
139 - refresh_user_entries_partial(&state).await
143 + refresh_user_entries_partial(&db).await
140 144 }
141 145
142 146 /// Suspend a user account and send notification email.
143 147 #[tracing::instrument(skip_all, name = "admin::admin_suspend_user")]
148 + #[allow(clippy::too_many_arguments)]
144 149 pub(super) async fn admin_suspend_user(
145 - State(state): State<AppState>,
150 + State(db): State<PgPool>,
151 + State(email): State<EmailClient>,
152 + State(bg): State<BackgroundTx>,
153 + State(caches): State<AppCaches>,
154 + State(payments): State<Billing>,
155 + State(integrations): State<Integrations>,
146 156 admin_user: AdminUser,
147 157 Path(id): Path<UserId>,
148 158 Form(form): Form<SuspendForm>,
@@ -153,7 +163,7 @@ pub(super) async fn admin_suspend_user(
153 163 }
154 164
155 165 // Get user for email notification
156 - let db_user = db::users::get_user_by_id(&state.db, id)
166 + let db_user = db::users::get_user_by_id(&db, id)
157 167 .await?
158 168 .ok_or(AppError::NotFound)?;
159 169
@@ -162,13 +172,13 @@ pub(super) async fn admin_suspend_user(
162 172 // revocation, fan-sub pause, email). Web fans the Stripe calls out on the
163 173 // background queue and evicts the in-memory session cache.
164 174 super::moderation_service::suspend_creator(
165 - &state.db,
166 - &state.email,
167 - state.stripe.as_ref(),
175 + &db,
176 + &email,
177 + payments.stripe.as_ref(),
168 178 super::moderation_service::FanoutMode::Background {
169 - bg: &state.bg,
170 - wam: state.wam.clone(),
171 - session_cache: &state.caches.session_cache,
179 + bg: &bg,
180 + wam: integrations.wam.clone(),
181 + session_cache: &caches.session_cache,
172 182 },
173 183 &db_user,
174 184 admin_user.admin_id(),
@@ -176,34 +186,38 @@ pub(super) async fn admin_suspend_user(
176 186 )
177 187 .await?;
178 188
179 - refresh_user_entries_partial(&state).await
189 + refresh_user_entries_partial(&db).await
180 190 }
181 191
182 192 /// Unsuspend a user account (admin override).
183 193 #[tracing::instrument(skip_all, name = "admin::admin_unsuspend_user")]
184 194 pub(super) async fn admin_unsuspend_user(
185 - State(state): State<AppState>,
195 + State(db): State<PgPool>,
196 + State(bg): State<BackgroundTx>,
197 + State(caches): State<AppCaches>,
198 + State(payments): State<Billing>,
199 + State(integrations): State<Integrations>,
186 200 AdminUser(_admin): AdminUser,
187 201 Path(id): Path<UserId>,
188 202 ) -> Result<impl IntoResponse> {
189 203 // Get user for Stripe account ID before unsuspending
190 - let db_user = db::users::get_user_by_id(&state.db, id)
204 + let db_user = db::users::get_user_by_id(&db, id)
191 205 .await?
192 206 .ok_or(AppError::NotFound)?;
193 207
194 208 super::moderation_service::unsuspend_creator(
195 - &state.db,
196 - state.stripe.as_ref(),
209 + &db,
210 + payments.stripe.as_ref(),
197 211 super::moderation_service::FanoutMode::Background {
198 - bg: &state.bg,
199 - wam: state.wam.clone(),
200 - session_cache: &state.caches.session_cache,
212 + bg: &bg,
213 + wam: integrations.wam.clone(),
214 + session_cache: &caches.session_cache,
201 215 },
202 216 &db_user,
203 217 )
204 218 .await?;
205 219
206 - refresh_user_entries_partial(&state).await
220 + refresh_user_entries_partial(&db).await
207 221 }
208 222
209 223 /// Permanently terminate a user account (enforcement ladder step 4).
@@ -213,11 +227,15 @@ pub(super) async fn admin_unsuspend_user(
213 227 /// data before the scheduler deletes the account.
214 228 #[tracing::instrument(skip_all, name = "admin::admin_terminate_user")]
215 229 pub(super) async fn admin_terminate_user(
216 - State(state): State<AppState>,
230 + State(db): State<PgPool>,
231 + State(email): State<EmailClient>,
232 + State(bg): State<BackgroundTx>,
233 + State(payments): State<Billing>,
234 + State(integrations): State<Integrations>,
217 235 admin_user: AdminUser,
218 236 Path(id): Path<UserId>,
219 237 ) -> Result<impl IntoResponse> {
220 - let db_user = db::users::get_user_by_id(&state.db, id)
238 + let db_user = db::users::get_user_by_id(&db, id)
221 239 .await?
222 240 .ok_or(AppError::NotFound)?;
223 241
@@ -233,41 +251,47 @@ pub(super) async fn admin_terminate_user(
233 251 ));
234 252 }
235 253
236 - db::users::terminate_user(&state.db, id).await?;
254 + db::users::terminate_user(&db, id).await?;
237 255
238 256 // Record moderation action
239 257 db::moderation::create_action(
240 - &state.db, id, admin_user.admin_id(), ModerationActionType::Termination,
258 + &db, id, admin_user.admin_id(), ModerationActionType::Termination,
241 259 db_user.suspension_reason.as_deref().unwrap_or("Account terminated"),
242 260 None,
243 261 ).await?;
244 262
245 263 // Cancel all fan subscriptions — both active and paused (suspension already paused them)
246 - if let Some(ref stripe) = state.stripe
264 + if let Some(ref stripe) = payments.stripe
247 265 && let Some(ref account_id) = db_user.stripe_account_id
248 266 {
249 - let active_subs = db::subscriptions::get_active_subscriptions_by_creator(&state.db, id).await?;
250 - let paused_subs = db::subscriptions::get_paused_subscriptions_by_creator(&state.db, id).await?;
267 + let active_subs = db::subscriptions::get_active_subscriptions_by_creator(&db, id).await?;
268 + let paused_subs = db::subscriptions::get_paused_subscriptions_by_creator(&db, id).await?;
251 269 let ids: Vec<String> = active_subs
252 270 .into_iter()
253 271 .chain(paused_subs)
254 272 .map(|s| s.stripe_subscription_id)
255 273 .collect();
256 274 crate::payments::fan_ops::spawn_fan_sub_fanout(
257 - &state.bg,
275 + &bg,
258 276 std::sync::Arc::clone(stripe),
259 277 account_id.clone(),
260 278 ids,
261 279 crate::payments::fan_ops::FanSubOp::Cancel,
262 - state.wam.clone(),
280 + integrations.wam.clone(),
263 281 );
264 282 }
265 283
266 284 // Send termination email
267 285 let user_email = db_user.email.clone();
268 286 let user_name = db_user.display_name.clone();
269 - spawn_email!(state, "account termination notification", |email| {
270 - email.send_account_termination(&user_email, user_name.as_deref())
287 + let email = email.clone();
288 + bg.spawn("account termination notification", async move {
289 + if let Err(e) = email
290 + .send_account_termination(&user_email, user_name.as_deref())
291 + .await
292 + {
293 + tracing::error!(error = ?e, "failed to send account termination notification");
294 + }
271 295 });
272 296
273 297 tracing::info!(
@@ -276,33 +300,33 @@ pub(super) async fn admin_terminate_user(
276 300 "admin terminated user account (30-day export window started)"
277 301 );
278 302
279 - refresh_user_entries_partial(&state).await
303 + refresh_user_entries_partial(&db).await
280 304 }
281 305
282 306 /// Trust a user (uploads auto-publish).
283 307 #[tracing::instrument(skip_all, name = "admin::admin_trust_user")]
284 308 pub(super) async fn admin_trust_user(
285 - State(state): State<AppState>,
309 + State(db): State<PgPool>,
286 310 AdminUser(_admin): AdminUser,
287 311 Path(id): Path<UserId>,
288 312 headers: axum::http::HeaderMap,
289 313 ) -> Result<Response> {
290 - db::users::set_upload_trusted(&state.db, id, true).await?;
314 + db::users::set_upload_trusted(&db, id, true).await?;
291 315 tracing::info!(user_id = %id, "admin trusted user for uploads");
292 - refresh_partial_for_target(&state, &headers).await
316 + refresh_partial_for_target(&db, &headers).await
293 317 }
294 318
295 319 /// Untrust a user (uploads require review).
296 320 #[tracing::instrument(skip_all, name = "admin::admin_untrust_user")]
297 321 pub(super) async fn admin_untrust_user(
298 - State(state): State<AppState>,
322 + State(db): State<PgPool>,
299 323 AdminUser(_admin): AdminUser,
300 324 Path(id): Path<UserId>,
301 325 headers: axum::http::HeaderMap,
302 326 ) -> Result<Response> {
303 - db::users::set_upload_trusted(&state.db, id, false).await?;
327 + db::users::set_upload_trusted(&db, id, false).await?;
304 328 tracing::info!(user_id = %id, "admin untrusted user for uploads");
305 - refresh_partial_for_target(&state, &headers).await
329 + refresh_partial_for_target(&db, &headers).await
306 330 }
307 331
308 332 /// Lock a user's custom pages (moderation kill switch): the editor goes
@@ -310,14 +334,14 @@ pub(super) async fn admin_untrust_user(
310 334 /// default. The source is preserved, so unlocking restores it. Reversible.
311 335 #[tracing::instrument(skip_all, name = "admin::admin_lock_custom_pages")]
312 336 pub(super) async fn admin_lock_custom_pages(
313 - State(state): State<AppState>,
337 + State(db): State<PgPool>,
314 338 admin_user: AdminUser,
315 339 Path(id): Path<UserId>,
316 340 headers: axum::http::HeaderMap,
317 341 ) -> Result<Response> {
318 - db::users::set_custom_pages_locked(&state.db, id, true).await?;
342 + db::users::set_custom_pages_locked(&db, id, true).await?;
319 343 db::moderation::create_action(
320 - &state.db,
344 + &db,
321 345 id,
322 346 admin_user.admin_id(),
323 347 db::ModerationActionType::ContentRemoval,
@@ -326,38 +350,38 @@ pub(super) async fn admin_lock_custom_pages(
326 350 )
327 351 .await?;
328 352 tracing::info!(user_id = %id, "admin locked custom pages");
329 - refresh_partial_for_target(&state, &headers).await
353 + refresh_partial_for_target(&db, &headers).await
330 354 }
331 355
332 356 /// Unlock a user's custom pages, restoring their preserved custom source.
333 357 #[tracing::instrument(skip_all, name = "admin::admin_unlock_custom_pages")]
334 358 pub(super) async fn admin_unlock_custom_pages(
335 - State(state): State<AppState>,
359 + State(db): State<PgPool>,
336 360 AdminUser(_admin): AdminUser,
337 361 Path(id): Path<UserId>,
338 362 headers: axum::http::HeaderMap,
339 363 ) -> Result<Response> {
340 - db::users::set_custom_pages_locked(&state.db, id, false).await?;
364 + db::users::set_custom_pages_locked(&db, id, false).await?;
341 365 tracing::info!(user_id = %id, "admin unlocked custom pages");
342 - refresh_partial_for_target(&state, &headers).await
366 + refresh_partial_for_target(&db, &headers).await
343 367 }
344 368
345 369 /// Return the right partial based on which page triggered the request.
346 - async fn refresh_partial_for_target(state: &AppState, headers: &axum::http::HeaderMap) -> Result<Response> {
370 + async fn refresh_partial_for_target(db: &PgPool, headers: &axum::http::HeaderMap) -> Result<Response> {
347 371 let target = headers.get("HX-Target").and_then(|v| v.to_str().ok()).unwrap_or("");
348 372 if target == "users-table" {
349 - Ok(refresh_user_entries_partial(state).await?.into_response())
373 + Ok(refresh_user_entries_partial(db).await?.into_response())
350 374 } else {
351 - super::uploads::refresh_held_uploads_partial(state).await
375 + super::uploads::refresh_held_uploads_partial(db).await
352 376 }
353 377 }
354 378
355 379 /// Re-query users and return the entries partial (page 1, no filter).
356 - async fn refresh_user_entries_partial(state: &AppState) -> Result<AdminUserEntriesTemplate> {
380 + async fn refresh_user_entries_partial(db: &PgPool) -> Result<AdminUserEntriesTemplate> {
357 381 let per_page: i64 = 50;
358 - let total_count = db::users::count_users(&state.db, None).await?;
382 + let total_count = db::users::count_users(db, None).await?;
359 383 let total_pages = ((total_count as f64) / (per_page as f64)).ceil() as i64;
360 - let db_users = db::users::get_all_users(&state.db, None, per_page, 0).await?;
384 + let db_users = db::users::get_all_users(db, None, per_page, 0).await?;
361 385 let users: Vec<AdminUserRow> = db_users.iter().map(AdminUserRow::from_db).collect();
362 386 Ok(AdminUserEntriesTemplate {
363 387 users,
M server/src/routes/admin/waitlist.rs +16 -16
@@ -6,6 +6,7 @@ use axum::{
6 6 Form,
7 7 };
8 8 use serde::Deserialize;
9 + use sqlx::PgPool;
9 10
10 11 use crate::{
11 12 auth::AdminUser,
@@ -14,7 +15,6 @@ use crate::{
14 15 helpers::get_csrf_token,
15 16 templates::*,
16 17 types::*,
17 - AppState,
18 18 };
19 19
20 20 #[derive(Debug, Deserialize)]
@@ -25,7 +25,7 @@ pub(super) struct WaitlistFilterQuery {
25 25 /// Render the admin waitlist dashboard with stats and entries.
26 26 #[tracing::instrument(skip_all, name = "admin::admin_waitlist")]
27 27 pub(super) async fn admin_waitlist(
28 - State(state): State<AppState>,
28 + State(db): State<PgPool>,
29 29 session: tower_sessions::Session,
30 30 AdminUser(user): AdminUser,
31 31 Query(query): Query<WaitlistFilterQuery>,
@@ -34,8 +34,8 @@ pub(super) async fn admin_waitlist(
34 34 let csrf_token = get_csrf_token(&session).await;
35 35 let current_filter = query.status.clone().unwrap_or_default();
36 36
37 - let db_stats = db::waitlist::get_waitlist_stats(&state.db).await?;
38 - let total_creators = db::waitlist::count_active_creators(&state.db).await?;
37 + let db_stats = db::waitlist::get_waitlist_stats(&db).await?;
38 + let total_creators = db::waitlist::count_active_creators(&db).await?;
39 39
40 40 let stats = WaitlistStats {
41 41 total_pending: db_stats.pending as u32,
@@ -44,7 +44,7 @@ pub(super) async fn admin_waitlist(
44 44 total_creators: total_creators as u32,
45 45 };
46 46
47 - let db_entries = db::waitlist::get_admin_waitlist(&state.db, query.status.as_deref()).await?;
47 + let db_entries = db::waitlist::get_admin_waitlist(&db, query.status.as_deref()).await?;
48 48
49 49 let entries: Vec<AdminWaitlistRow> = db_entries.iter().map(AdminWaitlistRow::from).collect();
50 50
@@ -61,12 +61,12 @@ pub(super) async fn admin_waitlist(
61 61 /// Return filtered waitlist entries as an HTMX partial.
62 62 #[tracing::instrument(skip_all, name = "admin::admin_waitlist_entries")]
63 63 pub(super) async fn admin_waitlist_entries(
64 - State(state): State<AppState>,
64 + State(db): State<PgPool>,
65 65 AdminUser(_user): AdminUser,
66 66 Query(query): Query<WaitlistFilterQuery>,
67 67 ) -> Result<impl IntoResponse> {
68 68
69 - let db_entries = db::waitlist::get_admin_waitlist(&state.db, query.status.as_deref()).await?;
69 + let db_entries = db::waitlist::get_admin_waitlist(&db, query.status.as_deref()).await?;
70 70
71 71 let entries: Vec<AdminWaitlistRow> = db_entries.iter().map(AdminWaitlistRow::from).collect();
72 72
@@ -76,21 +76,21 @@ pub(super) async fn admin_waitlist_entries(
76 76 /// Approve a waitlist entry and grant creator access to the user.
77 77 #[tracing::instrument(skip_all, name = "admin::admin_approve")]
78 78 pub(super) async fn admin_approve(
79 - State(state): State<AppState>,
79 + State(db): State<PgPool>,
80 80 AdminUser(_user): AdminUser,
81 81 Path(id): Path<WaitlistEntryId>,
82 82 ) -> Result<Response> {
83 83
84 84 // Get entry to find user_id
85 - let entry = db::waitlist::update_waitlist_status(&state.db, id, WaitlistStatus::Approved, Some(SelectionMethod::HandPicked), None).await?;
85 + let entry = db::waitlist::update_waitlist_status(&db, id, WaitlistStatus::Approved, Some(SelectionMethod::HandPicked), None).await?;
86 86
87 87 // Grant creator access
88 - db::waitlist::grant_creator_access(&state.db, entry.user_id).await?;
88 + db::waitlist::grant_creator_access(&db, entry.user_id).await?;
89 89
90 90 tracing::info!(entry_id = %id, user_id = %entry.user_id, "admin approved waitlist entry");
91 91
92 92 // Return updated entries partial
93 - let db_entries = db::waitlist::get_admin_waitlist(&state.db, Some("pending")).await?;
93 + let db_entries = db::waitlist::get_admin_waitlist(&db, Some("pending")).await?;
94 94 let entries: Vec<AdminWaitlistRow> = db_entries.iter().map(AdminWaitlistRow::from).collect();
95 95
96 96 Ok(AdminWaitlistEntriesTemplate { entries }.into_response())
@@ -99,17 +99,17 @@ pub(super) async fn admin_approve(
99 99 /// Flag a waitlist entry as spam.
100 100 #[tracing::instrument(skip_all, name = "admin::admin_spam")]
101 101 pub(super) async fn admin_spam(
102 - State(state): State<AppState>,
102 + State(db): State<PgPool>,
103 103 AdminUser(_user): AdminUser,
104 104 Path(id): Path<WaitlistEntryId>,
105 105 ) -> Result<Response> {
106 106
107 - db::waitlist::update_waitlist_status(&state.db, id, WaitlistStatus::Spam, None, None).await?;
107 + db::waitlist::update_waitlist_status(&db, id, WaitlistStatus::Spam, None, None).await?;
108 108
109 109 tracing::info!(entry_id = %id, "admin flagged waitlist entry as spam");
110 110
111 111 // Return updated entries partial
112 - let db_entries = db::waitlist::get_admin_waitlist(&state.db, Some("pending")).await?;
112 + let db_entries = db::waitlist::get_admin_waitlist(&db, Some("pending")).await?;
113 113 let entries: Vec<AdminWaitlistRow> = db_entries.iter().map(AdminWaitlistRow::from).collect();
114 114
115 115 Ok(AdminWaitlistEntriesTemplate { entries }.into_response())
@@ -123,7 +123,7 @@ pub(super) struct LotteryForm {
123 123 /// Run a creator lottery: create a wave, select random winners, and grant access.
124 124 #[tracing::instrument(skip_all, name = "admin::admin_lottery")]
125 125 pub(super) async fn admin_lottery(
126 - State(state): State<AppState>,
126 + State(db): State<PgPool>,
127 127 AdminUser(_user): AdminUser,
128 128 Form(form): Form<LotteryForm>,
129 129 ) -> Result<Response> {
@@ -133,7 +133,7 @@ pub(super) async fn admin_lottery(
133 133 }
134 134
135 135 // Use a transaction for atomicity
136 - let mut tx = state.db.begin().await?;
136 + let mut tx = db.begin().await?;
137 137
138 138 // Count hand-picks not yet assigned to a wave
139 139 let hand_picked_count = db::waitlist::count_unassigned_handpicks(&mut *tx).await?;