Skip to main content

max / makenotwork

29.4 KB · 865 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 `routes::admin::mod` for the registrations):
5 //!
6 //! - the review queue page, plus `queue-summary` and the `audit` log view;
7 //! - per-target promote / quarantine / rescan for both items and versions;
8 //! - the two bulk verbs, `bulk/rescan` and `bulk/promote`;
9 //! - `health.json`, which is the one admin path served without a session
10 //! (`ADMIN_PUBLIC_PATHS`), so it must never carry per-user data.
11 //!
12 //! Every mutating action writes a row to `db::scan_admin_actions`, which is the
13 //! audit trail for an admin overriding a pipeline verdict.
14
15 use crate::extractors::ValidatedQuery;
16 use std::collections::HashMap;
17
18 use axum::{
19 extract::{Path, State},
20 response::{IntoResponse, Response},
21 };
22 use sqlx::PgPool;
23
24 use crate::{
25 AppStorage, Scanning,
26 auth::AdminUser,
27 config::Config,
28 db::{self, FileScanStatus, ItemId, UserId, VersionId, scan_admin_actions::AdminAction},
29 error::Result,
30 helpers::get_csrf_token,
31 storage::FileType,
32 templates::{
33 AdminQueueSummaryTemplate, AdminScanAuditTemplate, AdminUploadEntriesTemplate,
34 AdminUploadsTemplate, LayerHealthCard,
35 },
36 types::{AdminAuditLogRow, AdminHeldUploadRow, LastAction, ScanHistoryDisplay},
37 };
38
39 const HEALTH_WINDOW_HOURS: i64 = 24;
40 const STALE_LAYER_THRESHOLD_HOURS: i64 = 1;
41 const HISTORY_WINDOW_HOURS: i64 = 24 * 7;
42 const HISTORY_ROW_LIMIT: i64 = 100;
43 const AUDIT_LOG_ROW_LIMIT: i64 = 500;
44
45 /// Build the per-layer health-card vec from the DB rollup.
46 async fn fetch_layer_health(db: &PgPool) -> Result<Vec<LayerHealthCard>> {
47 let rows = db::scanning::layer_health_window(db, HEALTH_WINDOW_HOURS).await?;
48 let mut by_layer: HashMap<String, LayerHealthCard> = HashMap::new();
49
50 // The canonical layer list, keep cards visible even if a layer hasn't
51 // run at all in the window. The dashboard depends on seeing "clamav: not
52 // running" rather than the layer just being absent.
53 for name in [
54 "content_type",
55 "structural",
56 "archive",
57 "yara",
58 "signing_macos",
59 "signing_windows",
60 "signing_linux",
61 "clamav",
62 "malwarebazaar",
63 "urlhaus",
64 "metadefender",
65 ] {
66 by_layer.insert(
67 name.to_string(),
68 LayerHealthCard {
69 layer: name.to_string(),
70 total: 0,
71 success_rate_pct: 0,
72 error_rate_pct: 0,
73 fail_count: 0,
74 status_badge: "down",
75 last_seen: "never".to_string(),
76 },
77 );
78 }
79
80 for row in rows {
81 let total = row.pass_count + row.skip_count + row.fail_count + row.error_count;
82 let success = row.pass_count + row.skip_count;
83 let success_rate = if total > 0 {
84 (100 * success / total) as i32
85 } else {
86 0
87 };
88 let error_rate = if total > 0 {
89 (100 * row.error_count / total) as i32
90 } else {
91 0
92 };
93
94 let stale_cutoff =
95 chrono::Utc::now() - chrono::Duration::hours(STALE_LAYER_THRESHOLD_HOURS);
96 let recent = row.last_pass_or_skip.is_some_and(|t| t > stale_cutoff);
97 let status_badge = if total == 0 || (!recent && row.error_count > 0) {
98 "down"
99 } else if error_rate > 10 {
100 "degraded"
101 } else {
102 "ok"
103 };
104
105 let last_seen = match row.last_pass_or_skip {
106 Some(t) => relative_age(t),
107 None => "never".to_string(),
108 };
109
110 by_layer.insert(
111 row.layer.clone(),
112 LayerHealthCard {
113 layer: row.layer,
114 total,
115 success_rate_pct: success_rate,
116 error_rate_pct: error_rate,
117 fail_count: row.fail_count,
118 status_badge,
119 last_seen,
120 },
121 );
122 }
123
124 // Stable ordering matches the pipeline's actual execution order.
125 let mut out: Vec<LayerHealthCard> = Vec::with_capacity(11);
126 for name in [
127 "content_type",
128 "structural",
129 "archive",
130 "yara",
131 "signing_macos",
132 "signing_windows",
133 "signing_linux",
134 "clamav",
135 "malwarebazaar",
136 "urlhaus",
137 "metadefender",
138 ] {
139 if let Some(card) = by_layer.remove(name) {
140 out.push(card);
141 }
142 }
143 Ok(out)
144 }
145
146 fn relative_age(t: chrono::DateTime<chrono::Utc>) -> String {
147 let now = chrono::Utc::now();
148 let delta = now - t;
149 if delta.num_seconds() < 60 {
150 format!("{}s ago", delta.num_seconds().max(0))
151 } else if delta.num_minutes() < 60 {
152 format!("{}m ago", delta.num_minutes())
153 } else if delta.num_hours() < 24 {
154 format!("{}h ago", delta.num_hours())
155 } else {
156 format!("{}d ago", delta.num_days())
157 }
158 }
159
160 /// Render the admin upload review queue.
161 #[tracing::instrument(skip_all, name = "admin::admin_uploads")]
162 pub(super) async fn admin_uploads(
163 State(db): State<PgPool>,
164 session: tower_sessions::Session,
165 AdminUser(user): AdminUser,
166 ) -> Result<impl IntoResponse> {
167 let csrf_token = get_csrf_token(&session).await;
168
169 let held_items = db::scanning::get_held_items(&db).await?;
170 let held_versions = db::scanning::get_held_versions(&db).await?;
171 // The held queries are capped at 1000 rows each (oldest first). Hitting the
172 // cap is an incident-scale backlog; surface it rather than silently truncate.
173 if held_items.len() >= 1000 || held_versions.len() >= 1000 {
174 tracing::warn!(
175 items = held_items.len(),
176 versions = held_versions.len(),
177 "admin held-uploads list hit the 1000-row display cap (oldest shown first)"
178 );
179 }
180
181 let mut held_uploads: Vec<AdminHeldUploadRow> = Vec::new();
182 held_uploads.extend(held_items.iter().map(AdminHeldUploadRow::from_held_item));
183 held_uploads.extend(
184 held_versions
185 .iter()
186 .map(AdminHeldUploadRow::from_held_version),
187 );
188 held_uploads.sort_by(|a, b| a.held_at.cmp(&b.held_at));
189
190 // Attach the last admin action to each held row in a single batch lookup.
191 attach_last_actions(&db, &mut held_uploads).await;
192
193 let total_held = held_uploads.len();
194 let layer_health = fetch_layer_health(&db).await?;
195 let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
196 let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
197
198 let recent_history: Vec<ScanHistoryDisplay> =
199 db::scanning::recent_history(&db, HISTORY_WINDOW_HOURS, HISTORY_ROW_LIMIT)
200 .await
201 .unwrap_or_default()
202 .iter()
203 .map(ScanHistoryDisplay::from_row)
204 .collect();
205 let history_total = recent_history.len();
206
207 Ok(AdminUploadsTemplate {
208 csrf_token,
209 session_user: Some(user),
210 held_uploads,
211 total_held,
212 admin_active_page: "uploads",
213 layer_health,
214 queue_pending,
215 queue_running,
216 recent_history,
217 history_total,
218 })
219 }
220
221 /// Batch-fetch the latest admin action for the displayed held rows and
222 /// attach each as `last_action` on its row.
223 async fn attach_last_actions(db: &PgPool, rows: &mut [AdminHeldUploadRow]) {
224 use std::str::FromStr;
225 let mut version_ids: Vec<uuid::Uuid> = Vec::new();
226 let mut item_ids: Vec<uuid::Uuid> = Vec::new();
227 for r in rows.iter() {
228 if let Some(ref vid) = r.version_id
229 && let Ok(uuid) = uuid::Uuid::from_str(vid)
230 {
231 version_ids.push(uuid);
232 } else if let Ok(uuid) = uuid::Uuid::from_str(&r.item_id) {
233 item_ids.push(uuid);
234 }
235 }
236 let version_actions = db::scan_admin_actions::latest_per_version(db, &version_ids)
237 .await
238 .unwrap_or_default();
239 let item_actions = db::scan_admin_actions::latest_per_item(db, &item_ids)
240 .await
241 .unwrap_or_default();
242
243 for r in rows.iter_mut() {
244 let summary = if let Some(ref vid) = r.version_id {
245 uuid::Uuid::from_str(vid)
246 .ok()
247 .and_then(|u| version_actions.get(&u))
248 } else {
249 uuid::Uuid::from_str(&r.item_id)
250 .ok()
251 .and_then(|u| item_actions.get(&u))
252 };
253 if let Some(s) = summary {
254 r.last_action = Some(LastAction {
255 action: s.action.clone(),
256 admin_username: s.admin_username.clone(),
257 when: relative_age(s.created_at),
258 });
259 }
260 }
261 }
262
263 #[derive(serde::Deserialize, Default)]
264 pub(super) struct AuditFilters {
265 #[serde(default)]
266 pub action: Option<String>,
267 #[serde(default)]
268 pub admin: Option<String>,
269 #[serde(default)]
270 pub since_days: Option<i64>,
271 }
272
273 /// Full audit-log page with query-string filters.
274 ///
275 /// `?action=promote&admin=max&since_days=30` is the canonical example. All
276 /// filters are optional; empty/absent means no constraint on that column.
277 #[tracing::instrument(skip_all, name = "admin::scan_audit")]
278 pub(super) async fn admin_scan_audit(
279 State(db): State<PgPool>,
280 session: tower_sessions::Session,
281 AdminUser(user): AdminUser,
282 ValidatedQuery(filters): ValidatedQuery<AuditFilters>,
283 ) -> Result<impl IntoResponse> {
284 let csrf_token = get_csrf_token(&session).await;
285 let action = filters.action.as_deref().filter(|s| !s.is_empty());
286 let admin = filters.admin.as_deref().filter(|s| !s.is_empty());
287 let entries: Vec<AdminAuditLogRow> = db::scan_admin_actions::list_filtered(
288 &db,
289 action,
290 admin,
291 filters.since_days,
292 AUDIT_LOG_ROW_LIMIT,
293 )
294 .await
295 .unwrap_or_default()
296 .iter()
297 .map(AdminAuditLogRow::from_db)
298 .collect();
299 Ok(AdminScanAuditTemplate {
300 csrf_token,
301 session_user: Some(user),
302 admin_active_page: "uploads",
303 entries,
304 filter_action: filters.action.unwrap_or_default(),
305 filter_admin: filters.admin.unwrap_or_default(),
306 filter_since_days: filters
307 .since_days
308 .map(|d| d.to_string())
309 .unwrap_or_default(),
310 })
311 }
312
313 /// Re-query held uploads and return the entries partial.
314 pub(super) async fn refresh_held_uploads_partial(db: &PgPool) -> Result<Response> {
315 let held_items = db::scanning::get_held_items(db).await?;
316 let held_versions = db::scanning::get_held_versions(db).await?;
317
318 let mut held_uploads: Vec<AdminHeldUploadRow> = Vec::new();
319 held_uploads.extend(held_items.iter().map(AdminHeldUploadRow::from_held_item));
320 held_uploads.extend(
321 held_versions
322 .iter()
323 .map(AdminHeldUploadRow::from_held_version),
324 );
325 held_uploads.sort_by(|a, b| a.held_at.cmp(&b.held_at));
326
327 Ok(AdminUploadEntriesTemplate { held_uploads }.into_response())
328 }
329
330 // ─── Per-row actions ─────────────────────────────────────────────────────────
331
332 /// Promote a held item upload to Clean. Renames the legacy "approve" verb.
333 #[tracing::instrument(skip_all, name = "admin::promote_item")]
334 pub(super) async fn admin_promote_item(
335 State(db): State<PgPool>,
336 State(storage): State<AppStorage>,
337 State(config): State<Config>,
338 AdminUser(admin): AdminUser,
339 Path(id): Path<ItemId>,
340 ) -> Result<Response> {
341 // Copy each held (staging) file to its content key and mark Clean in one
342 // step, a held file sits unserved at its staging key, so approving must run
343 // the same promote the scan worker's Clean path runs (C1).
344 crate::routes::storage::commit_promote_item(&db, &storage, &config, id).await?;
345 db::scan_admin_actions::log_item(
346 &db,
347 id,
348 admin.id,
349 AdminAction::Promote,
350 Some("held_for_review"),
351 Some("clean"),
352 None,
353 )
354 .await
355 .unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
356 tracing::info!(item_id = %id, admin_id = %admin.id, "item promoted to clean");
357 refresh_held_uploads_partial(&db).await
358 }
359
360 /// Quarantine a held item upload. Renames the legacy "reject" verb.
361 #[tracing::instrument(skip_all, name = "admin::quarantine_item")]
362 pub(super) async fn admin_quarantine_item(
363 State(db): State<PgPool>,
364 AdminUser(admin): AdminUser,
365 Path(id): Path<ItemId>,
366 ) -> Result<Response> {
367 db::scanning::update_item_scan_status(&db, id, FileScanStatus::Quarantined).await?;
368 db::scan_admin_actions::log_item(
369 &db,
370 id,
371 admin.id,
372 AdminAction::Quarantine,
373 Some("held_for_review"),
374 Some("quarantined"),
375 None,
376 )
377 .await
378 .unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
379 tracing::info!(item_id = %id, admin_id = %admin.id, "item quarantined");
380 refresh_held_uploads_partial(&db).await
381 }
382
383 /// Promote a held version upload to Clean.
384 #[tracing::instrument(skip_all, name = "admin::promote_version")]
385 pub(super) async fn admin_promote_version(
386 State(db): State<PgPool>,
387 State(storage): State<AppStorage>,
388 State(config): State<Config>,
389 AdminUser(admin): AdminUser,
390 Path(id): Path<VersionId>,
391 ) -> Result<Response> {
392 // See admin_promote_item: copy the held (staging) download to its content key
393 // and mark Clean together (C1).
394 crate::routes::storage::commit_promote_version(&db, &storage, &config, id).await?;
395 db::scan_admin_actions::log_version(
396 &db,
397 id,
398 admin.id,
399 AdminAction::Promote,
400 Some("held_for_review"),
401 Some("clean"),
402 None,
403 )
404 .await
405 .unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
406 tracing::info!(version_id = %id, admin_id = %admin.id, "version promoted to clean");
407 refresh_held_uploads_partial(&db).await
408 }
409
410 /// Quarantine a held version upload.
411 #[tracing::instrument(skip_all, name = "admin::quarantine_version")]
412 pub(super) async fn admin_quarantine_version(
413 State(db): State<PgPool>,
414 AdminUser(admin): AdminUser,
415 Path(id): Path<VersionId>,
416 ) -> Result<Response> {
417 db::scanning::update_version_scan_status(&db, id, FileScanStatus::Quarantined).await?;
418 db::scan_admin_actions::log_version(
419 &db,
420 id,
421 admin.id,
422 AdminAction::Quarantine,
423 Some("held_for_review"),
424 Some("quarantined"),
425 None,
426 )
427 .await
428 .unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
429 tracing::info!(version_id = %id, admin_id = %admin.id, "version quarantined");
430 refresh_held_uploads_partial(&db).await
431 }
432
433 // ─── Rescan ──────────────────────────────────────────────────────────────────
434
435 /// Re-enqueue a single version for scanning. Worker picks it up on its next
436 /// claim cycle. The entity is moved back to Pending so the dashboard reflects
437 /// that a scan is in flight.
438 #[tracing::instrument(skip_all, name = "admin::rescan_version")]
439 pub(super) async fn admin_rescan_version(
440 State(db): State<PgPool>,
441 State(scanning): State<Scanning>,
442 AdminUser(admin): AdminUser,
443 Path(id): Path<VersionId>,
444 ) -> Result<Response> {
445 rescan_version_inner(
446 &db,
447 scanning.scanner.as_ref(),
448 id,
449 admin.id,
450 AdminAction::Rescan,
451 )
452 .await?;
453 refresh_held_uploads_partial(&db).await
454 }
455
456 async fn rescan_version_inner(
457 db: &PgPool,
458 scanner: Option<&std::sync::Arc<crate::scanning::ScanPipeline>>,
459 id: VersionId,
460 admin_id: UserId,
461 action: AdminAction,
462 ) -> Result<()> {
463 let v = db::versions::get_version_by_id(db, id)
464 .await?
465 .ok_or(crate::error::AppError::NotFound)?;
466 let item = db::items::get_item_by_id(db, v.item_id)
467 .await?
468 .ok_or(crate::error::AppError::NotFound)?;
469 let owner = db::items::get_item_owner(db, item.id)
470 .await?
471 .ok_or(crate::error::AppError::NotFound)?;
472 let s3_key = v.s3_key.clone().ok_or(crate::error::AppError::NotFound)?;
473 let size = v.file_size_bytes.unwrap_or(0);
474
475 // Route through commit_rescan so the enqueue → flip-to-Pending order is
476 // the same as the chronic-disease commit_upload seal, admin paths used
477 // to call scan_jobs::enqueue + update_*_scan_status directly, which made
478 // ordering bugs hard to fence at the type level.
479 crate::routes::storage::commit_rescan(
480 db,
481 scanner,
482 crate::routes::storage::CommitTarget::Version(id),
483 &s3_key,
484 FileType::Download,
485 owner,
486 size,
487 )
488 .await?;
489 db::scan_admin_actions::log_version(
490 db,
491 id,
492 admin_id,
493 action,
494 Some("held_for_review"),
495 Some("pending"),
496 None,
497 )
498 .await
499 .unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
500 tracing::info!(version_id = %id, admin_id = %admin_id, "version re-enqueued for scan");
501 Ok(())
502 }
503
504 /// Re-enqueue a single item for scanning.
505 #[tracing::instrument(skip_all, name = "admin::rescan_item")]
506 pub(super) async fn admin_rescan_item(
507 State(db): State<PgPool>,
508 State(scanning): State<Scanning>,
509 AdminUser(admin): AdminUser,
510 Path(id): Path<ItemId>,
511 ) -> Result<Response> {
512 rescan_item_inner(
513 &db,
514 scanning.scanner.as_ref(),
515 id,
516 admin.id,
517 AdminAction::Rescan,
518 )
519 .await?;
520 refresh_held_uploads_partial(&db).await
521 }
522
523 async fn rescan_item_inner(
524 db: &PgPool,
525 scanner: Option<&std::sync::Arc<crate::scanning::ScanPipeline>>,
526 id: ItemId,
527 admin_id: UserId,
528 action: AdminAction,
529 ) -> Result<()> {
530 let item = db::items::get_item_by_id(db, id)
531 .await?
532 .ok_or(crate::error::AppError::NotFound)?;
533 let owner = db::items::get_item_owner(db, id)
534 .await?
535 .ok_or(crate::error::AppError::NotFound)?;
536
537 let (s3_key, size, file_type) = if let Some(k) = item.audio_s3_key.clone() {
538 (k, item.audio_file_size_bytes.unwrap_or(0), FileType::Audio)
539 } else if let Some(k) = item.cover_s3_key.clone() {
540 (k, item.cover_file_size_bytes.unwrap_or(0), FileType::Cover)
541 } else {
542 return Err(crate::error::AppError::NotFound);
543 };
544
545 crate::routes::storage::commit_rescan(
546 db,
547 scanner,
548 crate::routes::storage::CommitTarget::Item(id),
549 &s3_key,
550 file_type,
551 owner,
552 size,
553 )
554 .await?;
555 db::scan_admin_actions::log_item(
556 db,
557 id,
558 admin_id,
559 action,
560 Some("held_for_review"),
561 Some("pending"),
562 None,
563 )
564 .await
565 .unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
566 tracing::info!(item_id = %id, admin_id = %admin_id, "item re-enqueued for scan");
567 Ok(())
568 }
569
570 /// Bulk-rescan every currently-held item and version. Used to clear backlogs
571 /// accumulated under a previous broken pipeline configuration.
572 #[tracing::instrument(skip_all, name = "admin::bulk_rescan_held")]
573 pub(super) async fn admin_bulk_rescan_held(
574 State(db): State<PgPool>,
575 State(scanning): State<Scanning>,
576 AdminUser(admin): AdminUser,
577 ) -> Result<Response> {
578 // Clear the backlog with bounded concurrency instead of one sequential
579 // round-trip per candidate (Run #21 Performance MODERATE). MAX_INFLIGHT
580 // caps how many connections this admin action borrows from the pool at once.
581 const MAX_INFLIGHT: usize = 8;
582 let mut set: tokio::task::JoinSet<bool> = tokio::task::JoinSet::new();
583 let mut total = 0usize;
584
585 for cand in db::scan_jobs::rescan_candidates_versions(&db).await? {
586 if set.len() >= MAX_INFLIGHT
587 && let Some(Ok(ok)) = set.join_next().await
588 {
589 total += ok as usize;
590 }
591 let db = db.clone();
592 let scanner = scanning.scanner.clone();
593 let admin_id = admin.id;
594 set.spawn(async move {
595 rescan_version_inner(
596 &db,
597 scanner.as_ref(),
598 VersionId::from_uuid(cand.version_id),
599 admin_id,
600 AdminAction::BulkRescan,
601 )
602 .await
603 .is_ok()
604 });
605 }
606 for cand in db::scan_jobs::rescan_candidates_items(&db).await? {
607 if set.len() >= MAX_INFLIGHT
608 && let Some(Ok(ok)) = set.join_next().await
609 {
610 total += ok as usize;
611 }
612 let db = db.clone();
613 let scanner = scanning.scanner.clone();
614 let admin_id = admin.id;
615 set.spawn(async move {
616 rescan_item_inner(
617 &db,
618 scanner.as_ref(),
619 ItemId::from_uuid(cand.item_id),
620 admin_id,
621 AdminAction::BulkRescan,
622 )
623 .await
624 .is_ok()
625 });
626 }
627 while let Some(res) = set.join_next().await {
628 if let Ok(ok) = res {
629 total += ok as usize;
630 }
631 }
632 tracing::info!(total, admin_id = %admin.id, "bulk rescan of held queue dispatched");
633 refresh_held_uploads_partial(&db).await
634 }
635
636 /// Bulk-promote every currently-held item and version to Clean. Sticky
637 /// decision; requires a non-empty note for audit-trail attribution. Use
638 /// after a manual review pass where the admin has eyes on the held set and
639 /// has confirmed each is safe.
640 ///
641 /// Distinct from bulk-rescan: rescan re-runs the pipeline (system decides);
642 /// promote bypasses the pipeline (admin decides). The note is the only
643 /// record of *why*, so we require it.
644 #[tracing::instrument(skip_all, name = "admin::bulk_promote_held")]
645 pub(super) async fn admin_bulk_promote_held(
646 State(db): State<PgPool>,
647 State(storage): State<AppStorage>,
648 State(config): State<Config>,
649 AdminUser(admin): AdminUser,
650 axum::Form(form): axum::Form<BulkPromoteForm>,
651 ) -> Result<Response> {
652 let note = form.note.trim();
653 if note.is_empty() {
654 return Err(crate::error::AppError::BadRequest(
655 "Bulk promote requires a note explaining why every held file is safe to clear."
656 .to_string(),
657 ));
658 }
659
660 let held_items = db::scanning::get_held_items(&db).await?;
661 let held_versions = db::scanning::get_held_versions(&db).await?;
662
663 // Bounded-concurrency promote (Run #21 Performance MODERATE). The note is
664 // shared across tasks via Arc; MAX_INFLIGHT caps borrowed connections.
665 const MAX_INFLIGHT: usize = 8;
666 let note: std::sync::Arc<str> = note.into();
667 let mut set: tokio::task::JoinSet<bool> = tokio::task::JoinSet::new();
668 let mut total = 0usize;
669
670 for v in held_versions {
671 if set.len() >= MAX_INFLIGHT
672 && let Some(Ok(ok)) = set.join_next().await
673 {
674 total += ok as usize;
675 }
676 let db = db.clone();
677 let storage = storage.clone();
678 let config = config.clone();
679 let note = note.clone();
680 let admin_id = admin.id;
681 let vid = v.version_id;
682 set.spawn(async move {
683 if let Err(e) = crate::routes::storage::commit_promote_version(&db, &storage, &config, vid).await {
684 tracing::warn!(version_id = %vid, error = ?e, "bulk promote: version promote failed");
685 false
686 } else {
687 db::scan_admin_actions::log_version(
688 &db, vid, admin_id, AdminAction::BulkPromote,
689 Some("held_for_review"), Some("clean"), Some(&note),
690 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
691 true
692 }
693 });
694 }
695 for i in held_items {
696 if set.len() >= MAX_INFLIGHT
697 && let Some(Ok(ok)) = set.join_next().await
698 {
699 total += ok as usize;
700 }
701 let db = db.clone();
702 let storage = storage.clone();
703 let config = config.clone();
704 let note = note.clone();
705 let admin_id = admin.id;
706 let iid = i.item_id;
707 set.spawn(async move {
708 if let Err(e) =
709 crate::routes::storage::commit_promote_item(&db, &storage, &config, iid).await
710 {
711 tracing::warn!(item_id = %iid, error = ?e, "bulk promote: item promote failed");
712 false
713 } else {
714 db::scan_admin_actions::log_item(
715 &db,
716 iid,
717 admin_id,
718 AdminAction::BulkPromote,
719 Some("held_for_review"),
720 Some("clean"),
721 Some(&note),
722 )
723 .await
724 .unwrap_or_else(
725 |e| tracing::warn!(error = ?e, "failed to record scan admin audit action"),
726 );
727 true
728 }
729 });
730 }
731 while let Some(res) = set.join_next().await {
732 if let Ok(ok) = res {
733 total += ok as usize;
734 }
735 }
736 tracing::warn!(total, admin_id = %admin.id, note = %note, "bulk promote of held queue executed");
737 refresh_held_uploads_partial(&db).await
738 }
739
740 #[derive(serde::Deserialize)]
741 pub(super) struct BulkPromoteForm {
742 pub note: String,
743 }
744
745 // ─── Public health endpoint (for PoM) ────────────────────────────────────────
746
747 #[derive(serde::Serialize)]
748 struct LayerHealthJson {
749 layer: String,
750 total_1h: i64,
751 success_1h: i64,
752 error_1h: i64,
753 last_clean_secs_ago: Option<i64>,
754 }
755
756 #[derive(serde::Serialize)]
757 struct ScanPipelineHealth {
758 queue_pending: i64,
759 queue_running: i64,
760 queue_stuck: i64,
761 held_versions: i64,
762 held_items: i64,
763 held_media: i64,
764 held_total: i64,
765 layers: Vec<LayerHealthJson>,
766 scan_spool_free_bytes: u64,
767 scan_spool_file_count: u64,
768 generated_at: String,
769 }
770
771 /// Aggregate scan-pipeline health stats for external monitors (PoM).
772 ///
773 /// **Unauthenticated**, returns counts only, no PII. The data is the same
774 /// shape the admin dashboard's health panel uses; the JSON contract is
775 /// stable for PoM threshold rules. Stuck threshold matches the worker's
776 /// reaper (300s).
777 #[tracing::instrument(skip_all, name = "admin::scan_health_json")]
778 pub(super) async fn scan_health_json(State(db): State<PgPool>) -> Result<impl IntoResponse> {
779 let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
780 let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
781 let queue_stuck = db::scan_jobs::stuck_count(&db, 300).await.unwrap_or(0);
782 let held = db::scanning::held_counts(&db)
783 .await
784 .unwrap_or(db::scanning::HeldCounts {
785 held_versions: 0,
786 held_items: 0,
787 held_media: 0,
788 });
789 let held_total = held.held_versions + held.held_items + held.held_media;
790
791 let rows = db::scanning::layer_health_window(&db, 1)
792 .await
793 .unwrap_or_default();
794 let now = chrono::Utc::now();
795 let layers: Vec<LayerHealthJson> = rows
796 .into_iter()
797 .map(|r| {
798 let success = r.pass_count + r.skip_count;
799 let total = success + r.fail_count + r.error_count;
800 LayerHealthJson {
801 layer: r.layer,
802 total_1h: total,
803 success_1h: success,
804 error_1h: r.error_count,
805 last_clean_secs_ago: r.last_pass_or_skip.map(|t| (now - t).num_seconds()),
806 }
807 })
808 .collect();
809
810 // The spool disk stats hit the filesystem (statvfs + a full directory scan);
811 // run them on the blocking pool so a slow or large spool dir can't stall the
812 // async runtime for every other task on this worker thread.
813 let spool_dir_path = std::path::PathBuf::from(crate::constants::SCAN_SPOOL_DIR);
814 let (scan_spool_free_bytes, scan_spool_file_count) = tokio::task::spawn_blocking(move || {
815 let spool_dir = spool_dir_path.as_path();
816 let free = fs2::available_space(if spool_dir.exists() {
817 spool_dir
818 } else {
819 spool_dir
820 .parent()
821 .unwrap_or_else(|| std::path::Path::new("/"))
822 })
823 .unwrap_or(0);
824 let count = std::fs::read_dir(spool_dir).map_or(0, |rd| {
825 rd.flatten()
826 .filter(|e| e.file_type().is_ok_and(|t| t.is_file()))
827 .count() as u64
828 });
829 (free, count)
830 })
831 .await
832 .unwrap_or((0, 0));
833
834 let body = ScanPipelineHealth {
835 queue_pending,
836 queue_running,
837 queue_stuck,
838 held_versions: held.held_versions,
839 held_items: held.held_items,
840 held_media: held.held_media,
841 held_total,
842 layers,
843 scan_spool_free_bytes,
844 scan_spool_file_count,
845 generated_at: now.to_rfc3339(),
846 };
847 Ok(axum::Json(body))
848 }
849
850 // ─── Live partials ───────────────────────────────────────────────────────────
851
852 /// HTMX partial: current pending + running counts. Polled by the dashboard.
853 #[tracing::instrument(skip_all, name = "admin::queue_summary_partial")]
854 pub(super) async fn admin_queue_summary_partial(
855 State(db): State<PgPool>,
856 AdminUser(_admin): AdminUser,
857 ) -> Result<impl IntoResponse> {
858 let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
859 let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
860 Ok(AdminQueueSummaryTemplate {
861 queue_pending,
862 queue_running,
863 })
864 }
865