Skip to main content

max / makenotwork

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