Skip to main content

max / makenotwork

22.4 KB · 586 lines History Blame Raw
1 //! Admin upload review queue + scan-pipeline dashboard.
2 //!
3 //! Backs the routes registered under `/admin/uploads` and `/api/admin/uploads`.
4 //! See `docs/scan-pipeline-audit.md` § 5 for the layout and behavior spec.
5
6 use std::collections::HashMap;
7
8 use axum::{
9 extract::{Path, State},
10 response::{IntoResponse, Response},
11 };
12
13 use crate::{
14 auth::AdminUser,
15 db::{self, scan_admin_actions::AdminAction, FileScanStatus, ItemId, UserId, VersionId},
16 error::Result,
17 helpers::get_csrf_token,
18 storage::FileType,
19 templates::*,
20 types::*,
21 AppState,
22 };
23
24 const HEALTH_WINDOW_HOURS: i64 = 24;
25 const STALE_LAYER_THRESHOLD_HOURS: i64 = 1;
26 const HISTORY_WINDOW_HOURS: i64 = 24 * 7;
27 const HISTORY_ROW_LIMIT: i64 = 100;
28 const AUDIT_LOG_ROW_LIMIT: i64 = 500;
29
30 /// 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 let mut by_layer: HashMap<String, LayerHealthCard> = HashMap::new();
34
35 // The canonical layer list — keep cards visible even if a layer hasn't
36 // run at all in the window. The dashboard depends on seeing "clamav: ✗
37 // not running" rather than the layer just being absent.
38 for name in ["content_type", "structural", "archive", "yara", "signing_macos", "signing_windows", "signing_linux", "clamav", "malwarebazaar", "urlhaus", "metadefender"] {
39 by_layer.insert(name.to_string(), LayerHealthCard {
40 layer: name.to_string(),
41 total: 0,
42 success_rate_pct: 0,
43 error_rate_pct: 0,
44 fail_count: 0,
45 status_badge: "down",
46 last_seen: "never".to_string(),
47 });
48 }
49
50 for row in rows {
51 let total = row.pass_count + row.skip_count + row.fail_count + row.error_count;
52 let success = row.pass_count + row.skip_count;
53 let success_rate = if total > 0 { (100 * success / total) as i32 } else { 0 };
54 let error_rate = if total > 0 { (100 * row.error_count / total) as i32 } else { 0 };
55
56 let stale_cutoff = chrono::Utc::now() - chrono::Duration::hours(STALE_LAYER_THRESHOLD_HOURS);
57 let recent = row.last_pass_or_skip.is_some_and(|t| t > stale_cutoff);
58 let status_badge = if total == 0 {
59 "down"
60 } else if !recent && row.error_count > 0 {
61 "down"
62 } else if error_rate > 10 {
63 "degraded"
64 } else {
65 "ok"
66 };
67
68 let last_seen = match row.last_pass_or_skip {
69 Some(t) => relative_age(t),
70 None => "never".to_string(),
71 };
72
73 by_layer.insert(row.layer.clone(), LayerHealthCard {
74 layer: row.layer,
75 total,
76 success_rate_pct: success_rate,
77 error_rate_pct: error_rate,
78 fail_count: row.fail_count,
79 status_badge,
80 last_seen,
81 });
82 }
83
84 // Stable ordering matches the pipeline's actual execution order.
85 let mut out: Vec<LayerHealthCard> = Vec::with_capacity(11);
86 for name in ["content_type", "structural", "archive", "yara", "signing_macos", "signing_windows", "signing_linux", "clamav", "malwarebazaar", "urlhaus", "metadefender"] {
87 if let Some(card) = by_layer.remove(name) {
88 out.push(card);
89 }
90 }
91 Ok(out)
92 }
93
94 fn relative_age(t: chrono::DateTime<chrono::Utc>) -> String {
95 let now = chrono::Utc::now();
96 let delta = now - t;
97 if delta.num_seconds() < 60 {
98 format!("{}s ago", delta.num_seconds().max(0))
99 } else if delta.num_minutes() < 60 {
100 format!("{}m ago", delta.num_minutes())
101 } else if delta.num_hours() < 24 {
102 format!("{}h ago", delta.num_hours())
103 } else {
104 format!("{}d ago", delta.num_days())
105 }
106 }
107
108 /// Render the admin upload review queue.
109 #[tracing::instrument(skip_all, name = "admin::admin_uploads")]
110 pub(super) async fn admin_uploads(
111 State(state): State<AppState>,
112 session: tower_sessions::Session,
113 AdminUser(user): AdminUser,
114 ) -> Result<impl IntoResponse> {
115 let csrf_token = get_csrf_token(&session).await;
116
117 let held_items = db::scanning::get_held_items(&state.db).await?;
118 let held_versions = db::scanning::get_held_versions(&state.db).await?;
119
120 let mut held_uploads: Vec<AdminHeldUploadRow> = Vec::new();
121 held_uploads.extend(held_items.iter().map(AdminHeldUploadRow::from_held_item));
122 held_uploads.extend(held_versions.iter().map(AdminHeldUploadRow::from_held_version));
123 held_uploads.sort_by(|a, b| a.held_at.cmp(&b.held_at));
124
125 // Attach the last admin action to each held row in a single batch lookup.
126 attach_last_actions(&state, &mut held_uploads).await;
127
128 let total_held = held_uploads.len();
129 let layer_health = fetch_layer_health(&state).await?;
130 let queue_pending = db::scan_jobs::queued_count(&state.db).await.unwrap_or(0);
131 let queue_running = db::scan_jobs::running_count(&state.db).await.unwrap_or(0);
132
133 let recent_history: Vec<ScanHistoryDisplay> = db::scanning::recent_history(
134 &state.db, HISTORY_WINDOW_HOURS, HISTORY_ROW_LIMIT,
135 ).await
136 .unwrap_or_default()
137 .iter()
138 .map(ScanHistoryDisplay::from_row)
139 .collect();
140 let history_total = recent_history.len();
141
142 Ok(AdminUploadsTemplate {
143 csrf_token,
144 session_user: Some(user),
145 held_uploads,
146 total_held,
147 admin_active_page: "uploads",
148 layer_health,
149 queue_pending,
150 queue_running,
151 recent_history,
152 history_total,
153 })
154 }
155
156 /// Batch-fetch the latest admin action for the displayed held rows and
157 /// attach each as `last_action` on its row.
158 async fn attach_last_actions(state: &AppState, rows: &mut [AdminHeldUploadRow]) {
159 use std::str::FromStr;
160 let mut version_ids: Vec<uuid::Uuid> = Vec::new();
161 let mut item_ids: Vec<uuid::Uuid> = Vec::new();
162 for r in rows.iter() {
163 if let Some(ref vid) = r.version_id
164 && let Ok(uuid) = uuid::Uuid::from_str(vid)
165 {
166 version_ids.push(uuid);
167 } else if let Ok(uuid) = uuid::Uuid::from_str(&r.item_id) {
168 item_ids.push(uuid);
169 }
170 }
171 let version_actions = db::scan_admin_actions::latest_per_version(&state.db, &version_ids)
172 .await.unwrap_or_default();
173 let item_actions = db::scan_admin_actions::latest_per_item(&state.db, &item_ids)
174 .await.unwrap_or_default();
175
176 for r in rows.iter_mut() {
177 let summary = if let Some(ref vid) = r.version_id {
178 uuid::Uuid::from_str(vid).ok().and_then(|u| version_actions.get(&u))
179 } else {
180 uuid::Uuid::from_str(&r.item_id).ok().and_then(|u| item_actions.get(&u))
181 };
182 if let Some(s) = summary {
183 r.last_action = Some(LastAction {
184 action: s.action.clone(),
185 admin_username: s.admin_username.clone(),
186 when: relative_age(s.created_at),
187 });
188 }
189 }
190 }
191
192 #[derive(serde::Deserialize, Default)]
193 pub(super) struct AuditFilters {
194 #[serde(default)]
195 pub action: Option<String>,
196 #[serde(default)]
197 pub admin: Option<String>,
198 #[serde(default)]
199 pub since_days: Option<i64>,
200 }
201
202 /// Full audit-log page with query-string filters.
203 ///
204 /// `?action=promote&admin=max&since_days=30` is the canonical example. All
205 /// filters are optional; empty/absent means no constraint on that column.
206 #[tracing::instrument(skip_all, name = "admin::scan_audit")]
207 pub(super) async fn admin_scan_audit(
208 State(state): State<AppState>,
209 session: tower_sessions::Session,
210 AdminUser(user): AdminUser,
211 axum::extract::Query(filters): axum::extract::Query<AuditFilters>,
212 ) -> Result<impl IntoResponse> {
213 let csrf_token = get_csrf_token(&session).await;
214 let action = filters.action.as_deref().filter(|s| !s.is_empty());
215 let admin = filters.admin.as_deref().filter(|s| !s.is_empty());
216 let entries: Vec<AdminAuditLogRow> = db::scan_admin_actions::list_filtered(
217 &state.db, action, admin, filters.since_days, AUDIT_LOG_ROW_LIMIT,
218 ).await
219 .unwrap_or_default()
220 .iter()
221 .map(AdminAuditLogRow::from_db)
222 .collect();
223 Ok(AdminScanAuditTemplate {
224 csrf_token,
225 session_user: Some(user),
226 admin_active_page: "uploads",
227 entries,
228 filter_action: filters.action.unwrap_or_default(),
229 filter_admin: filters.admin.unwrap_or_default(),
230 filter_since_days: filters.since_days.map(|d| d.to_string()).unwrap_or_default(),
231 })
232 }
233
234 /// Re-query held uploads and return the entries partial.
235 pub(super) async fn refresh_held_uploads_partial(state: &AppState) -> Result<Response> {
236 let held_items = db::scanning::get_held_items(&state.db).await?;
237 let held_versions = db::scanning::get_held_versions(&state.db).await?;
238
239 let mut held_uploads: Vec<AdminHeldUploadRow> = Vec::new();
240 held_uploads.extend(held_items.iter().map(AdminHeldUploadRow::from_held_item));
241 held_uploads.extend(held_versions.iter().map(AdminHeldUploadRow::from_held_version));
242 held_uploads.sort_by(|a, b| a.held_at.cmp(&b.held_at));
243
244 Ok(AdminUploadEntriesTemplate { held_uploads }.into_response())
245 }
246
247 // ─── Per-row actions ─────────────────────────────────────────────────────────
248
249 /// Promote a held item upload to Clean. Renames the legacy "approve" verb.
250 #[tracing::instrument(skip_all, name = "admin::promote_item")]
251 pub(super) async fn admin_promote_item(
252 State(state): State<AppState>,
253 AdminUser(admin): AdminUser,
254 Path(id): Path<ItemId>,
255 ) -> Result<Response> {
256 db::scanning::update_item_scan_status(&state.db, id, FileScanStatus::Clean).await?;
257 db::scan_admin_actions::log_item(
258 &state.db, id, admin.id, AdminAction::Promote,
259 Some("held_for_review"), Some("clean"),
260 None,
261 ).await.ok();
262 tracing::info!(item_id = %id, admin_id = %admin.id, "item promoted to clean");
263 refresh_held_uploads_partial(&state).await
264 }
265
266 /// Quarantine a held item upload. Renames the legacy "reject" verb.
267 #[tracing::instrument(skip_all, name = "admin::quarantine_item")]
268 pub(super) async fn admin_quarantine_item(
269 State(state): State<AppState>,
270 AdminUser(admin): AdminUser,
271 Path(id): Path<ItemId>,
272 ) -> Result<Response> {
273 db::scanning::update_item_scan_status(&state.db, id, FileScanStatus::Quarantined).await?;
274 db::scan_admin_actions::log_item(
275 &state.db, id, admin.id, AdminAction::Quarantine,
276 Some("held_for_review"), Some("quarantined"),
277 None,
278 ).await.ok();
279 tracing::info!(item_id = %id, admin_id = %admin.id, "item quarantined");
280 refresh_held_uploads_partial(&state).await
281 }
282
283 /// Promote a held version upload to Clean.
284 #[tracing::instrument(skip_all, name = "admin::promote_version")]
285 pub(super) async fn admin_promote_version(
286 State(state): State<AppState>,
287 AdminUser(admin): AdminUser,
288 Path(id): Path<VersionId>,
289 ) -> Result<Response> {
290 db::scanning::update_version_scan_status(&state.db, id, FileScanStatus::Clean).await?;
291 db::scan_admin_actions::log_version(
292 &state.db, id, admin.id, AdminAction::Promote,
293 Some("held_for_review"), Some("clean"),
294 None,
295 ).await.ok();
296 tracing::info!(version_id = %id, admin_id = %admin.id, "version promoted to clean");
297 refresh_held_uploads_partial(&state).await
298 }
299
300 /// Quarantine a held version upload.
301 #[tracing::instrument(skip_all, name = "admin::quarantine_version")]
302 pub(super) async fn admin_quarantine_version(
303 State(state): State<AppState>,
304 AdminUser(admin): AdminUser,
305 Path(id): Path<VersionId>,
306 ) -> Result<Response> {
307 db::scanning::update_version_scan_status(&state.db, id, FileScanStatus::Quarantined).await?;
308 db::scan_admin_actions::log_version(
309 &state.db, id, admin.id, AdminAction::Quarantine,
310 Some("held_for_review"), Some("quarantined"),
311 None,
312 ).await.ok();
313 tracing::info!(version_id = %id, admin_id = %admin.id, "version quarantined");
314 refresh_held_uploads_partial(&state).await
315 }
316
317 // ─── Rescan ──────────────────────────────────────────────────────────────────
318
319 /// Re-enqueue a single version for scanning. Worker picks it up on its next
320 /// claim cycle. The entity is moved back to Pending so the dashboard reflects
321 /// that a scan is in flight.
322 #[tracing::instrument(skip_all, name = "admin::rescan_version")]
323 pub(super) async fn admin_rescan_version(
324 State(state): State<AppState>,
325 AdminUser(admin): AdminUser,
326 Path(id): Path<VersionId>,
327 ) -> Result<Response> {
328 rescan_version_inner(&state, id, admin.id, AdminAction::Rescan).await?;
329 refresh_held_uploads_partial(&state).await
330 }
331
332 async fn rescan_version_inner(
333 state: &AppState,
334 id: VersionId,
335 admin_id: UserId,
336 action: AdminAction,
337 ) -> Result<()> {
338 let v = db::versions::get_version_by_id(&state.db, id).await?
339 .ok_or(crate::error::AppError::NotFound)?;
340 let item = db::items::get_item_by_id(&state.db, v.item_id).await?
341 .ok_or(crate::error::AppError::NotFound)?;
342 let owner = db::items::get_item_owner(&state.db, item.id).await?
343 .ok_or(crate::error::AppError::NotFound)?;
344 let s3_key = v.s3_key.clone().ok_or(crate::error::AppError::NotFound)?;
345 let size = v.file_size_bytes.unwrap_or(0);
346
347 // Route through commit_rescan so the enqueue → flip-to-Pending order is
348 // the same as the chronic-disease commit_upload seal — admin paths used
349 // to call scan_jobs::enqueue + update_*_scan_status directly, which made
350 // ordering bugs hard to fence at the type level.
351 crate::routes::storage::commit_rescan(
352 state,
353 crate::routes::storage::CommitTarget::Version(id),
354 &s3_key,
355 FileType::Download,
356 owner,
357 size,
358 ).await?;
359 db::scan_admin_actions::log_version(
360 &state.db, id, admin_id, action,
361 Some("held_for_review"), Some("pending"), None,
362 ).await.ok();
363 tracing::info!(version_id = %id, admin_id = %admin_id, "version re-enqueued for scan");
364 Ok(())
365 }
366
367 /// Re-enqueue a single item for scanning.
368 #[tracing::instrument(skip_all, name = "admin::rescan_item")]
369 pub(super) async fn admin_rescan_item(
370 State(state): State<AppState>,
371 AdminUser(admin): AdminUser,
372 Path(id): Path<ItemId>,
373 ) -> Result<Response> {
374 rescan_item_inner(&state, id, admin.id, AdminAction::Rescan).await?;
375 refresh_held_uploads_partial(&state).await
376 }
377
378 async fn rescan_item_inner(
379 state: &AppState,
380 id: ItemId,
381 admin_id: UserId,
382 action: AdminAction,
383 ) -> Result<()> {
384 let item = db::items::get_item_by_id(&state.db, id).await?
385 .ok_or(crate::error::AppError::NotFound)?;
386 let owner = db::items::get_item_owner(&state.db, id).await?
387 .ok_or(crate::error::AppError::NotFound)?;
388
389 let (s3_key, size, file_type) = if let Some(k) = item.audio_s3_key.clone() {
390 (k, item.audio_file_size_bytes.unwrap_or(0), FileType::Audio)
391 } else if let Some(k) = item.cover_s3_key.clone() {
392 (k, item.cover_file_size_bytes.unwrap_or(0), FileType::Cover)
393 } else {
394 return Err(crate::error::AppError::NotFound);
395 };
396
397 crate::routes::storage::commit_rescan(
398 state,
399 crate::routes::storage::CommitTarget::Item(id),
400 &s3_key,
401 file_type,
402 owner,
403 size,
404 ).await?;
405 db::scan_admin_actions::log_item(
406 &state.db, id, admin_id, action,
407 Some("held_for_review"), Some("pending"), None,
408 ).await.ok();
409 tracing::info!(item_id = %id, admin_id = %admin_id, "item re-enqueued for scan");
410 Ok(())
411 }
412
413 /// Bulk-rescan every currently-held item and version. Used to clear backlogs
414 /// accumulated under a previous broken pipeline configuration.
415 #[tracing::instrument(skip_all, name = "admin::bulk_rescan_held")]
416 pub(super) async fn admin_bulk_rescan_held(
417 State(state): State<AppState>,
418 AdminUser(admin): AdminUser,
419 ) -> Result<Response> {
420 let mut total = 0usize;
421
422 for cand in db::scan_jobs::rescan_candidates_versions(&state.db).await? {
423 let id = VersionId::from_uuid(cand.version_id);
424 if rescan_version_inner(&state, id, admin.id, AdminAction::BulkRescan).await.is_ok() {
425 total += 1;
426 }
427 }
428 for cand in db::scan_jobs::rescan_candidates_items(&state.db).await? {
429 let id = ItemId::from_uuid(cand.item_id);
430 if rescan_item_inner(&state, id, admin.id, AdminAction::BulkRescan).await.is_ok() {
431 total += 1;
432 }
433 }
434 tracing::info!(total, admin_id = %admin.id, "bulk rescan of held queue dispatched");
435 refresh_held_uploads_partial(&state).await
436 }
437
438 /// Bulk-promote every currently-held item and version to Clean. Sticky
439 /// decision; requires a non-empty note for audit-trail attribution. Use
440 /// after a manual review pass where the admin has eyes on the held set and
441 /// has confirmed each is safe.
442 ///
443 /// Distinct from bulk-rescan: rescan re-runs the pipeline (system decides);
444 /// promote bypasses the pipeline (admin decides). The note is the only
445 /// record of *why*, so we require it.
446 #[tracing::instrument(skip_all, name = "admin::bulk_promote_held")]
447 pub(super) async fn admin_bulk_promote_held(
448 State(state): State<AppState>,
449 AdminUser(admin): AdminUser,
450 axum::Form(form): axum::Form<BulkPromoteForm>,
451 ) -> Result<Response> {
452 let note = form.note.trim();
453 if note.is_empty() {
454 return Err(crate::error::AppError::BadRequest(
455 "Bulk promote requires a note explaining why every held file is safe to clear.".to_string(),
456 ));
457 }
458
459 let held_items = db::scanning::get_held_items(&state.db).await?;
460 let held_versions = db::scanning::get_held_versions(&state.db).await?;
461 let mut total = 0usize;
462
463 for v in &held_versions {
464 if db::scanning::update_version_scan_status(&state.db, v.version_id, FileScanStatus::Clean).await.is_ok() {
465 db::scan_admin_actions::log_version(
466 &state.db, v.version_id, admin.id, AdminAction::BulkPromote,
467 Some("held_for_review"), Some("clean"), Some(note),
468 ).await.ok();
469 total += 1;
470 }
471 }
472 for i in &held_items {
473 if db::scanning::update_item_scan_status(&state.db, i.item_id, FileScanStatus::Clean).await.is_ok() {
474 db::scan_admin_actions::log_item(
475 &state.db, i.item_id, admin.id, AdminAction::BulkPromote,
476 Some("held_for_review"), Some("clean"), Some(note),
477 ).await.ok();
478 total += 1;
479 }
480 }
481 tracing::warn!(total, admin_id = %admin.id, note = %note, "bulk promote of held queue executed");
482 refresh_held_uploads_partial(&state).await
483 }
484
485 #[derive(serde::Deserialize)]
486 pub(super) struct BulkPromoteForm {
487 pub note: String,
488 }
489
490 // ─── Public health endpoint (for PoM) ────────────────────────────────────────
491
492 #[derive(serde::Serialize)]
493 struct LayerHealthJson {
494 layer: String,
495 total_1h: i64,
496 success_1h: i64,
497 error_1h: i64,
498 last_clean_secs_ago: Option<i64>,
499 }
500
501 #[derive(serde::Serialize)]
502 struct ScanPipelineHealth {
503 queue_pending: i64,
504 queue_running: i64,
505 queue_stuck: i64,
506 held_versions: i64,
507 held_items: i64,
508 held_media: i64,
509 held_total: i64,
510 layers: Vec<LayerHealthJson>,
511 scan_spool_free_bytes: u64,
512 scan_spool_file_count: u64,
513 generated_at: String,
514 }
515
516 /// Aggregate scan-pipeline health stats for external monitors (PoM).
517 ///
518 /// **Unauthenticated** — returns counts only, no PII. The data is the same
519 /// shape the admin dashboard's health panel uses; the JSON contract is
520 /// stable for PoM threshold rules. Stuck threshold matches the worker's
521 /// reaper (300s).
522 #[tracing::instrument(skip_all, name = "admin::scan_health_json")]
523 pub(super) async fn scan_health_json(
524 State(state): State<AppState>,
525 ) -> Result<impl IntoResponse> {
526 let queue_pending = db::scan_jobs::queued_count(&state.db).await.unwrap_or(0);
527 let queue_running = db::scan_jobs::running_count(&state.db).await.unwrap_or(0);
528 let queue_stuck = db::scan_jobs::stuck_count(&state.db, 300).await.unwrap_or(0);
529 let held = db::scanning::held_counts(&state.db).await
530 .unwrap_or(db::scanning::HeldCounts { held_versions: 0, held_items: 0, held_media: 0 });
531 let held_total = held.held_versions + held.held_items + held.held_media;
532
533 let rows = db::scanning::layer_health_window(&state.db, 1).await.unwrap_or_default();
534 let now = chrono::Utc::now();
535 let layers: Vec<LayerHealthJson> = rows.into_iter().map(|r| {
536 let success = r.pass_count + r.skip_count;
537 let total = success + r.fail_count + r.error_count;
538 LayerHealthJson {
539 layer: r.layer,
540 total_1h: total,
541 success_1h: success,
542 error_1h: r.error_count,
543 last_clean_secs_ago: r.last_pass_or_skip.map(|t| (now - t).num_seconds()),
544 }
545 }).collect();
546
547 let spool_dir = std::path::Path::new(crate::constants::SCAN_SPOOL_DIR);
548 let scan_spool_free_bytes = fs2::available_space(if spool_dir.exists() {
549 spool_dir
550 } else {
551 spool_dir.parent().unwrap_or(std::path::Path::new("/"))
552 })
553 .unwrap_or(0);
554 let scan_spool_file_count = std::fs::read_dir(spool_dir)
555 .map(|rd| rd.flatten().filter(|e| e.file_type().map(|t| t.is_file()).unwrap_or(false)).count() as u64)
556 .unwrap_or(0);
557
558 let body = ScanPipelineHealth {
559 queue_pending,
560 queue_running,
561 queue_stuck,
562 held_versions: held.held_versions,
563 held_items: held.held_items,
564 held_media: held.held_media,
565 held_total,
566 layers,
567 scan_spool_free_bytes,
568 scan_spool_file_count,
569 generated_at: now.to_rfc3339(),
570 };
571 Ok(axum::Json(body))
572 }
573
574 // ─── Live partials ───────────────────────────────────────────────────────────
575
576 /// HTMX partial: current pending + running counts. Polled by the dashboard.
577 #[tracing::instrument(skip_all, name = "admin::queue_summary_partial")]
578 pub(super) async fn admin_queue_summary_partial(
579 State(state): State<AppState>,
580 AdminUser(_admin): AdminUser,
581 ) -> Result<impl IntoResponse> {
582 let queue_pending = db::scan_jobs::queued_count(&state.db).await.unwrap_or(0);
583 let queue_running = db::scan_jobs::running_count(&state.db).await.unwrap_or(0);
584 Ok(AdminQueueSummaryTemplate { queue_pending, queue_running })
585 }
586