Skip to main content

max / makenotwork

29.6 KB · 881 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 ///
645 /// The note arrives in the `HX-Prompt` header, which is what the `hx-prompt`
646 /// extension sends. It used to arrive as a form field the template filled from
647 /// `hx-vals='js:...'`, and that never worked: `script-src 'self'` carries no
648 /// `unsafe-eval`, so the expression could not be evaluated and the field was
649 /// always absent. The header is the transport htmx itself defines and needs no
650 /// evaluation.
651 #[tracing::instrument(skip_all, name = "admin::bulk_promote_held")]
652 pub(super) async fn admin_bulk_promote_held(
653 State(db): State<PgPool>,
654 State(storage): State<AppStorage>,
655 State(config): State<Config>,
656 AdminUser(admin): AdminUser,
657 headers: axum::http::HeaderMap,
658 ) -> Result<Response> {
659 let prompt = prompt_header(&headers);
660 let note = prompt.trim();
661 if note.is_empty() {
662 return Err(crate::error::AppError::BadRequest(
663 "Bulk promote requires a note explaining why every held file is safe to clear."
664 .to_string(),
665 ));
666 }
667
668 let held_items = db::scanning::get_held_items(&db).await?;
669 let held_versions = db::scanning::get_held_versions(&db).await?;
670
671 // Bounded-concurrency promote (Run #21 Performance MODERATE). The note is
672 // shared across tasks via Arc; MAX_INFLIGHT caps borrowed connections.
673 const MAX_INFLIGHT: usize = 8;
674 let note: std::sync::Arc<str> = note.into();
675 let mut set: tokio::task::JoinSet<bool> = tokio::task::JoinSet::new();
676 let mut total = 0usize;
677
678 for v in held_versions {
679 if set.len() >= MAX_INFLIGHT
680 && let Some(Ok(ok)) = set.join_next().await
681 {
682 total += ok as usize;
683 }
684 let db = db.clone();
685 let storage = storage.clone();
686 let config = config.clone();
687 let note = note.clone();
688 let admin_id = admin.id;
689 let vid = v.version_id;
690 set.spawn(async move {
691 if let Err(e) = crate::routes::storage::commit_promote_version(&db, &storage, &config, vid).await {
692 tracing::warn!(version_id = %vid, error = ?e, "bulk promote: version promote failed");
693 false
694 } else {
695 db::scan_admin_actions::log_version(
696 &db, vid, admin_id, AdminAction::BulkPromote,
697 Some("held_for_review"), Some("clean"), Some(&note),
698 ).await.unwrap_or_else(|e| tracing::warn!(error = ?e, "failed to record scan admin audit action"));
699 true
700 }
701 });
702 }
703 for i in held_items {
704 if set.len() >= MAX_INFLIGHT
705 && let Some(Ok(ok)) = set.join_next().await
706 {
707 total += ok as usize;
708 }
709 let db = db.clone();
710 let storage = storage.clone();
711 let config = config.clone();
712 let note = note.clone();
713 let admin_id = admin.id;
714 let iid = i.item_id;
715 set.spawn(async move {
716 if let Err(e) =
717 crate::routes::storage::commit_promote_item(&db, &storage, &config, iid).await
718 {
719 tracing::warn!(item_id = %iid, error = ?e, "bulk promote: item promote failed");
720 false
721 } else {
722 db::scan_admin_actions::log_item(
723 &db,
724 iid,
725 admin_id,
726 AdminAction::BulkPromote,
727 Some("held_for_review"),
728 Some("clean"),
729 Some(&note),
730 )
731 .await
732 .unwrap_or_else(
733 |e| tracing::warn!(error = ?e, "failed to record scan admin audit action"),
734 );
735 true
736 }
737 });
738 }
739 while let Some(res) = set.join_next().await {
740 if let Ok(ok) = res {
741 total += ok as usize;
742 }
743 }
744 tracing::warn!(total, admin_id = %admin.id, note = %note, "bulk promote of held queue executed");
745 refresh_held_uploads_partial(&db).await
746 }
747
748 /// The `HX-Prompt` header's value, percent-decoded.
749 ///
750 /// The extension sends `encodeURI(answer)`, so anything the admin types that
751 /// URI syntax reserves arrives escaped. A value that will not decode is passed
752 /// through rather than dropped: the caller only asks whether it is blank.
753 fn prompt_header(headers: &axum::http::HeaderMap) -> String {
754 let raw = headers
755 .get("HX-Prompt")
756 .and_then(|v| v.to_str().ok())
757 .unwrap_or_default();
758 urlencoding::decode(raw).map_or_else(|_| raw.to_string(), std::borrow::Cow::into_owned)
759 }
760
761 // --- Public health endpoint (for PoM) ---
762
763 #[derive(serde::Serialize)]
764 struct LayerHealthJson {
765 layer: String,
766 total_1h: i64,
767 success_1h: i64,
768 error_1h: i64,
769 last_clean_secs_ago: Option<i64>,
770 }
771
772 #[derive(serde::Serialize)]
773 struct ScanPipelineHealth {
774 queue_pending: i64,
775 queue_running: i64,
776 queue_stuck: i64,
777 held_versions: i64,
778 held_items: i64,
779 held_media: i64,
780 held_total: i64,
781 layers: Vec<LayerHealthJson>,
782 scan_spool_free_bytes: u64,
783 scan_spool_file_count: u64,
784 generated_at: String,
785 }
786
787 /// Aggregate scan-pipeline health stats for external monitors (PoM).
788 ///
789 /// **Unauthenticated**, returns counts only, no PII. The data is the same
790 /// shape the admin dashboard's health panel uses; the JSON contract is
791 /// stable for PoM threshold rules. Stuck threshold matches the worker's
792 /// reaper (300s).
793 #[tracing::instrument(skip_all, name = "admin::scan_health_json")]
794 pub(super) async fn scan_health_json(State(db): State<PgPool>) -> Result<impl IntoResponse> {
795 let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
796 let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
797 let queue_stuck = db::scan_jobs::stuck_count(&db, 300).await.unwrap_or(0);
798 let held = db::scanning::held_counts(&db)
799 .await
800 .unwrap_or(db::scanning::HeldCounts {
801 held_versions: 0,
802 held_items: 0,
803 held_media: 0,
804 });
805 let held_total = held.held_versions + held.held_items + held.held_media;
806
807 let rows = db::scanning::layer_health_window(&db, 1)
808 .await
809 .unwrap_or_default();
810 let now = chrono::Utc::now();
811 let layers: Vec<LayerHealthJson> = rows
812 .into_iter()
813 .map(|r| {
814 let success = r.pass_count + r.skip_count;
815 let total = success + r.fail_count + r.error_count;
816 LayerHealthJson {
817 layer: r.layer,
818 total_1h: total,
819 success_1h: success,
820 error_1h: r.error_count,
821 last_clean_secs_ago: r.last_pass_or_skip.map(|t| (now - t).num_seconds()),
822 }
823 })
824 .collect();
825
826 // The spool disk stats hit the filesystem (statvfs + a full directory scan);
827 // run them on the blocking pool so a slow or large spool dir can't stall the
828 // async runtime for every other task on this worker thread.
829 let spool_dir_path = std::path::PathBuf::from(crate::constants::SCAN_SPOOL_DIR);
830 let (scan_spool_free_bytes, scan_spool_file_count) = tokio::task::spawn_blocking(move || {
831 let spool_dir = spool_dir_path.as_path();
832 let free = fs2::available_space(if spool_dir.exists() {
833 spool_dir
834 } else {
835 spool_dir
836 .parent()
837 .unwrap_or_else(|| std::path::Path::new("/"))
838 })
839 .unwrap_or(0);
840 let count = std::fs::read_dir(spool_dir).map_or(0, |rd| {
841 rd.flatten()
842 .filter(|e| e.file_type().is_ok_and(|t| t.is_file()))
843 .count() as u64
844 });
845 (free, count)
846 })
847 .await
848 .unwrap_or((0, 0));
849
850 let body = ScanPipelineHealth {
851 queue_pending,
852 queue_running,
853 queue_stuck,
854 held_versions: held.held_versions,
855 held_items: held.held_items,
856 held_media: held.held_media,
857 held_total,
858 layers,
859 scan_spool_free_bytes,
860 scan_spool_file_count,
861 generated_at: now.to_rfc3339(),
862 };
863 Ok(axum::Json(body))
864 }
865
866 // --- Live partials ---
867
868 /// HTMX partial: current pending + running counts. Polled by the dashboard.
869 #[tracing::instrument(skip_all, name = "admin::queue_summary_partial")]
870 pub(super) async fn admin_queue_summary_partial(
871 State(db): State<PgPool>,
872 AdminUser(_admin): AdminUser,
873 ) -> Result<impl IntoResponse> {
874 let queue_pending = db::scan_jobs::queued_count(&db).await.unwrap_or(0);
875 let queue_running = db::scan_jobs::running_count(&db).await.unwrap_or(0);
876 Ok(AdminQueueSummaryTemplate {
877 queue_pending,
878 queue_running,
879 })
880 }
881