Skip to main content

max / makenotwork

20.1 KB · 551 lines History Blame Raw
1 //! Database operations for file scan results.
2
3 use chrono::{DateTime, Utc};
4 use sqlx::{FromRow, PgPool};
5 use uuid::Uuid;
6
7 use super::FileScanStatus;
8 use super::ItemId;
9 use super::UserId;
10 use super::VersionId;
11 use crate::scanning::ScanResult;
12
13 /// An item held for review, joined with creator info and latest scan layers.
14 #[derive(Debug, Clone, FromRow)]
15 pub struct HeldItemRow {
16 pub item_id: ItemId,
17 pub item_title: String,
18 pub s3_key: Option<String>,
19 pub creator_username: String,
20 pub creator_id: UserId,
21 pub upload_trusted: bool,
22 pub held_at: DateTime<Utc>,
23 /// Latest `file_scan_results.scan_layers` JSON for this entity's s3_key,
24 /// or `null` if no scan has run yet. The dashboard renders this as chips.
25 pub scan_layers: Option<serde_json::Value>,
26 }
27
28 /// A version held for review, joined with creator info and latest scan layers.
29 #[derive(Debug, Clone, FromRow)]
30 pub struct HeldVersionRow {
31 pub item_id: ItemId,
32 pub item_title: String,
33 pub version_id: VersionId,
34 pub version_number: String,
35 pub s3_key: Option<String>,
36 pub creator_username: String,
37 pub creator_id: UserId,
38 pub upload_trusted: bool,
39 pub held_at: DateTime<Utc>,
40 pub scan_layers: Option<serde_json::Value>,
41 }
42
43 /// Remove every CDN-served image reference to `s3_key`, across the image
44 /// surfaces that have no per-row scan gate: the gallery tables
45 /// (`item_images.s3_key`, `project_images.s3_key`, `content_insertions.storage_key`)
46 /// and the item *cover* columns on `items` (`cover_s3_key`/`cover_image_url`).
47 /// Returns the number of rows affected.
48 ///
49 /// On quarantine of these kinds, removing the reference IS the primary
50 /// enforcement: it stops the app from ever rendering the (Cloudflare-served) URL
51 /// again, and, critically, makes the key non-live so the durable S3-deletion
52 /// queue will actually purge the object instead of parking it behind the
53 /// `is_s3_key_live` guard. `storage_used` counters self-heal on the weekly
54 /// `recalculate_all_storage_used` pass; we accept a transient over-count for a
55 /// malicious upload rather than join through three ownership paths here.
56 ///
57 /// The item cover is a special case: the bad image lives in columns ON the
58 /// `items` row, so we NULL the three cover columns rather than DELETE the row,
59 /// deleting it would take the legitimate audio/video track down with the
60 /// thumbnail. Both cover references (`cover_s3_key` exact + `cover_image_url`
61 /// suffix) are registered in `S3_KEY_REFS`, so NULLing both makes the key dead.
62 #[tracing::instrument(skip_all)]
63 pub async fn purge_cdn_image_rows_by_key(db: &PgPool, s3_key: &str) -> Result<u64, sqlx::Error> {
64 // Wrap all reference-clearing statements in one transaction: a mid-loop
65 // failure must not leave a partially-purged malicious reference set (some
66 // tables still pointing at a quarantined object while others were cleared).
67 let mut tx = db.begin().await?;
68 let mut removed = 0u64;
69 for sql in [
70 "DELETE FROM item_images WHERE s3_key = $1",
71 "DELETE FROM project_images WHERE s3_key = $1",
72 "DELETE FROM content_insertions WHERE storage_key = $1",
73 ] {
74 removed += sqlx::query(sql)
75 .bind(s3_key)
76 .execute(&mut *tx)
77 .await?
78 .rows_affected();
79 }
80 // Item + project covers: NULL the columns in place, keeping the row (a track
81 // or a project must not be deleted along with its quarantined thumbnail).
82 // Matches on the exact `cover_s3_key`; the cover_image_url suffix is cleared
83 // in the same statement so neither reference keeps the object live. BOTH the
84 // items and projects cover surfaces must be cleared, the read-side
85 // counterpart `set_cdn_image_scan_status_by_key` stamps both, and
86 // `promote_cdn_image_by_key` promotes both, so a purge that skipped projects
87 // would leave a quarantined project cover rendered AND its (public-bucket)
88 // object un-reapable behind the `is_s3_key_live` guard.
89 for sql in [
90 "UPDATE items SET cover_s3_key = NULL, cover_image_url = NULL, \
91 cover_file_size_bytes = NULL, updated_at = NOW() WHERE cover_s3_key = $1",
92 "UPDATE projects SET cover_s3_key = NULL, cover_image_url = NULL, \
93 updated_at = NOW() WHERE cover_s3_key = $1",
94 ] {
95 removed += sqlx::query(sql)
96 .bind(s3_key)
97 .execute(&mut *tx)
98 .await?
99 .rows_affected();
100 }
101 tx.commit().await?;
102 Ok(removed)
103 }
104
105 /// Of the given S3 keys, the subset whose authoritative scan verdict is
106 /// `Quarantined` (confirmed malicious).
107 ///
108 /// `file_scan_results` is the per-object scan record the quarantine worker
109 /// writes; keying on it covers every content surface (audio/cover/video,
110 /// versions, insertions) uniformly. Used to exclude quarantined objects from
111 /// bulk key reads that ultimately serve content to a user, notably content
112 /// export, which collects raw keys and would otherwise hand a creator back their
113 /// own quarantined object that every download path already refuses. Absent rows
114 /// (never scanned) are not quarantined and are not returned.
115 #[tracing::instrument(skip_all)]
116 pub async fn quarantined_s3_keys(
117 pool: &PgPool,
118 keys: &[String],
119 ) -> Result<std::collections::HashSet<String>, sqlx::Error> {
120 if keys.is_empty() {
121 return Ok(std::collections::HashSet::new());
122 }
123 let rows: Vec<(String,)> = sqlx::query_as(
124 "SELECT DISTINCT s3_key FROM file_scan_results \
125 WHERE s3_key = ANY($1) AND scan_status = 'quarantined'",
126 )
127 .bind(keys)
128 .fetch_all(pool)
129 .await?;
130 Ok(rows.into_iter().map(|(k,)| k).collect())
131 }
132
133 /// Set the per-row `scan_status` on every CDN-served image surface keyed by
134 /// `s3_key`, in one transaction. The symmetric read-side counterpart of
135 /// [`purge_cdn_image_rows_by_key`]: where the purge DELETEs (or NULLs) a
136 /// quarantined reference, this stamps a non-quarantine terminal verdict,
137 /// `clean` (renders) or `held` (stays hidden), so the fail-closed gate can
138 /// distinguish "not yet scanned" (`pending`) from "scanned and cleared".
139 ///
140 /// Touches the same surfaces as the purge: the gallery tables
141 /// (`item_images.s3_key`, `project_images.s3_key`,
142 /// `content_insertions.storage_key`) and the cover columns on `items` /
143 /// `projects` (`cover_s3_key`). Wrapped in a single transaction so a mid-loop
144 /// failure can't leave the reference set half-stamped (some surfaces cleared to
145 /// `clean` while others stay `pending`). Returns total rows affected.
146 #[tracing::instrument(skip_all)]
147 pub async fn set_cdn_image_scan_status_by_key(
148 db: &PgPool,
149 s3_key: &str,
150 status: FileScanStatus,
151 ) -> Result<u64, sqlx::Error> {
152 let mut tx = db.begin().await?;
153 let mut affected = 0u64;
154 for sql in [
155 "UPDATE item_images SET scan_status = $1 WHERE s3_key = $2",
156 "UPDATE project_images SET scan_status = $1 WHERE s3_key = $2",
157 "UPDATE content_insertions SET scan_status = $1 WHERE storage_key = $2",
158 "UPDATE items SET cover_scan_status = $1 WHERE cover_s3_key = $2",
159 "UPDATE projects SET cover_scan_status = $1 WHERE cover_s3_key = $2",
160 ] {
161 affected += sqlx::query(sql)
162 .bind(status)
163 .bind(s3_key)
164 .execute(&mut *tx)
165 .await?
166 .rows_affected();
167 }
168 tx.commit().await?;
169 Ok(affected)
170 }
171
172 /// Insert a scan result record for audit trail.
173 #[tracing::instrument(skip_all)]
174 pub async fn insert_scan_result(
175 db: &PgPool,
176 s3_key: &str,
177 result: &ScanResult,
178 ) -> Result<Uuid, sqlx::Error> {
179 let layers_json =
180 serde_json::to_value(&result.layers).unwrap_or_else(|_| serde_json::Value::Array(vec![]));
181
182 let id = sqlx::query_scalar::<_, Uuid>(
183 r"
184 INSERT INTO file_scan_results (s3_key, scan_status, scan_layers, sha256, file_size_bytes)
185 VALUES ($1, $2, $3, $4, $5)
186 RETURNING id
187 ",
188 )
189 .bind(s3_key)
190 .bind(result.status)
191 .bind(&layers_json)
192 .bind(&result.sha256)
193 .bind(result.file_size as i64)
194 .fetch_one(db)
195 .await?;
196
197 Ok(id)
198 }
199
200 /// Update the scan_status column on an item.
201 #[tracing::instrument(skip_all)]
202 pub async fn update_item_scan_status(
203 db: &PgPool,
204 item_id: ItemId,
205 status: FileScanStatus,
206 ) -> Result<(), sqlx::Error> {
207 sqlx::query(
208 r"
209 UPDATE items SET scan_status = $1, updated_at = NOW() WHERE id = $2
210 ",
211 )
212 .bind(status)
213 .bind(item_id)
214 .execute(db)
215 .await?;
216
217 Ok(())
218 }
219
220 /// Update the scan_status column on a version.
221 #[tracing::instrument(skip_all)]
222 pub async fn update_version_scan_status(
223 db: &PgPool,
224 version_id: VersionId,
225 status: FileScanStatus,
226 ) -> Result<(), sqlx::Error> {
227 sqlx::query(
228 r"
229 UPDATE versions SET scan_status = $1 WHERE id = $2
230 ",
231 )
232 .bind(status)
233 .bind(version_id)
234 .execute(db)
235 .await?;
236
237 Ok(())
238 }
239
240 /// Update the scan_status column on a media file.
241 #[tracing::instrument(skip_all)]
242 pub async fn update_media_file_scan_status(
243 db: &PgPool,
244 media_file_id: crate::db::MediaFileId,
245 status: FileScanStatus,
246 ) -> Result<(), sqlx::Error> {
247 let id: uuid::Uuid = media_file_id.into();
248 sqlx::query("UPDATE media_files SET scan_status = $1 WHERE id = $2")
249 .bind(status)
250 .bind(id)
251 .execute(db)
252 .await?;
253 Ok(())
254 }
255
256 /// Most recent content hash recorded for a key, from the scan-result audit
257 /// trail. The scan worker computes and stores the sha256 in `file_scan_results`
258 /// at scan time; the admin-approve promote path (which flips a held file to
259 /// Clean and must copy it to its content-addressed key) has only the staging
260 /// key in scope, so it reads the hash back here. Empty hashes, recorded by a
261 /// degraded/held scan that never fully hashed the object, are skipped, so a
262 /// content key is never derived from a blank digest.
263 #[tracing::instrument(skip_all)]
264 pub async fn latest_sha256_by_key(
265 db: &PgPool,
266 s3_key: &str,
267 ) -> Result<Option<String>, sqlx::Error> {
268 sqlx::query_scalar::<_, String>(
269 "SELECT sha256 FROM file_scan_results \
270 WHERE s3_key = $1 AND sha256 <> '' \
271 ORDER BY scanned_at DESC LIMIT 1",
272 )
273 .bind(s3_key)
274 .fetch_optional(db)
275 .await
276 }
277
278 /// Repoint a GATED entity (Item audio/video, Version, Media, OTA artifact) from
279 /// its staging key to the immutable content key and mark it Clean in one write.
280 ///
281 /// This is the scan-then-promote closing move for the kinds that carry their own
282 /// `scan_status` gate. The `(table, key column)` pair is a compile-time constant
283 /// selected by `(kind, file_type)`, never user input, so formatting it into the
284 /// statement is safe. Uses the runtime `query` (not the `!` macro) so extending
285 /// the promote set needs no offline-cache regeneration. Accepts any executor so
286 /// the caller can run it inside the same transaction as the staging-key delete
287 /// enqueue.
288 pub async fn promote_gated<'e>(
289 executor: impl sqlx::PgExecutor<'e>,
290 kind: crate::db::scan_jobs::ScanTargetKind,
291 file_type: crate::storage::FileType,
292 target_id: Uuid,
293 content_key: &str,
294 ) -> Result<(), sqlx::Error> {
295 use crate::db::scan_jobs::ScanTargetKind as K;
296 use crate::storage::FileType as F;
297 let sql: &'static str = match (kind, file_type) {
298 (K::Item, F::Audio) => {
299 "UPDATE items SET audio_s3_key = $1, scan_status = 'clean', updated_at = NOW() WHERE id = $2"
300 }
301 (K::Item, F::Video) => {
302 "UPDATE items SET video_s3_key = $1, scan_status = 'clean', updated_at = NOW() WHERE id = $2"
303 }
304 (K::Version, _) => "UPDATE versions SET s3_key = $1, scan_status = 'clean' WHERE id = $2",
305 (K::Media, _) => "UPDATE media_files SET s3_key = $1, scan_status = 'clean' WHERE id = $2",
306 (K::OtaArtifact, _) => {
307 "UPDATE ota_artifacts SET s3_key = $1, scan_status = 'clean' WHERE id = $2"
308 }
309 (other_kind, other_ft) => {
310 // A CDN-image kind (or an Item file role with no key column) must go
311 // through `promote_cdn_image_by_key` instead; reaching here is a wiring
312 // bug, not a data condition.
313 return Err(sqlx::Error::Protocol(format!(
314 "promote_gated: {other_kind:?}/{other_ft:?} is not a gated promote target"
315 )));
316 }
317 };
318 sqlx::query(sql)
319 .bind(content_key)
320 .bind(target_id)
321 .execute(executor)
322 .await?;
323 Ok(())
324 }
325
326 /// Repoint every CDN-served image surface that currently references `staging_key`
327 /// to the immutable `content_key` (and its rebuilt public `content_url`), marking
328 /// the row Clean. Mirrors [`set_cdn_image_scan_status_by_key`]'s surface list so
329 /// the promote and the status-stamp can't drift apart. Only the one surface whose
330 /// key matches is touched; the others no-op. `content_insertions` carries no URL
331 /// column (it is served presigned, not straight from the CDN) so only its key and
332 /// status move. Returns the number of rows repointed (expected: exactly 1).
333 pub async fn promote_cdn_image_by_key(
334 conn: &mut sqlx::PgConnection,
335 staging_key: &str,
336 content_key: &str,
337 content_url: &str,
338 ) -> Result<u64, sqlx::Error> {
339 let mut affected = 0u64;
340 // (SQL, binds_content_url), the surfaces with a materialized public URL take
341 // three binds (key, url, where-key); content_insertions takes two.
342 for (sql, has_url) in [
343 (
344 "UPDATE item_images SET s3_key = $1, image_url = $2, scan_status = 'clean' WHERE s3_key = $3",
345 true,
346 ),
347 (
348 "UPDATE project_images SET s3_key = $1, image_url = $2, scan_status = 'clean' WHERE s3_key = $3",
349 true,
350 ),
351 (
352 "UPDATE items SET cover_s3_key = $1, cover_image_url = $2, cover_scan_status = 'clean' WHERE cover_s3_key = $3",
353 true,
354 ),
355 (
356 "UPDATE projects SET cover_s3_key = $1, cover_image_url = $2, cover_scan_status = 'clean' WHERE cover_s3_key = $3",
357 true,
358 ),
359 (
360 "UPDATE content_insertions SET storage_key = $1, scan_status = 'clean' WHERE storage_key = $2",
361 false,
362 ),
363 ] {
364 let q = if has_url {
365 sqlx::query(sql)
366 .bind(content_key)
367 .bind(content_url)
368 .bind(staging_key)
369 } else {
370 sqlx::query(sql).bind(content_key).bind(staging_key)
371 };
372 affected += q.execute(&mut *conn).await?.rows_affected();
373 }
374 Ok(affected)
375 }
376
377 /// Get items held for review, joined with creator info + latest scan layers.
378 /// Oldest first.
379 #[tracing::instrument(skip_all)]
380 pub async fn get_held_items(db: &PgPool) -> Result<Vec<HeldItemRow>, sqlx::Error> {
381 let rows = sqlx::query_as::<_, HeldItemRow>(
382 r"
383 SELECT i.id AS item_id, i.title AS item_title,
384 COALESCE(i.audio_s3_key, i.cover_s3_key) AS s3_key,
385 u.username AS creator_username, u.id AS creator_id,
386 u.upload_trusted, i.updated_at AS held_at,
387 (
388 SELECT fsr.scan_layers FROM file_scan_results fsr
389 WHERE fsr.s3_key = COALESCE(i.audio_s3_key, i.cover_s3_key)
390 ORDER BY fsr.scanned_at DESC LIMIT 1
391 ) AS scan_layers
392 FROM items i
393 JOIN projects p ON p.id = i.project_id
394 JOIN users u ON u.id = p.user_id
395 WHERE i.scan_status = 'held_for_review'
396 ORDER BY i.updated_at ASC
397 LIMIT 1000
398 ",
399 )
400 .fetch_all(db)
401 .await?;
402
403 Ok(rows)
404 }
405
406 /// Get versions held for review, joined with creator info + latest scan layers.
407 /// Oldest first.
408 #[tracing::instrument(skip_all)]
409 pub async fn get_held_versions(db: &PgPool) -> Result<Vec<HeldVersionRow>, sqlx::Error> {
410 let rows = sqlx::query_as::<_, HeldVersionRow>(
411 r"
412 SELECT i.id AS item_id, i.title AS item_title,
413 v.id AS version_id, v.version_number,
414 v.s3_key,
415 u.username AS creator_username, u.id AS creator_id,
416 u.upload_trusted, v.created_at AS held_at,
417 (
418 SELECT fsr.scan_layers FROM file_scan_results fsr
419 WHERE fsr.s3_key = v.s3_key
420 ORDER BY fsr.scanned_at DESC LIMIT 1
421 ) AS scan_layers
422 FROM versions v
423 JOIN items i ON i.id = v.item_id
424 JOIN projects p ON p.id = i.project_id
425 JOIN users u ON u.id = p.user_id
426 WHERE v.scan_status = 'held_for_review'
427 ORDER BY v.created_at ASC
428 LIMIT 1000
429 ",
430 )
431 .fetch_all(db)
432 .await?;
433
434 Ok(rows)
435 }
436
437 /// Per-layer aggregate stats over a window for the admin dashboard.
438 #[derive(Debug, Clone, FromRow)]
439 pub struct LayerHealthRow {
440 pub layer: String,
441 pub pass_count: i64,
442 pub skip_count: i64,
443 pub fail_count: i64,
444 pub error_count: i64,
445 pub last_pass_or_skip: Option<DateTime<Utc>>,
446 }
447
448 /// Compute per-layer health stats over the last N hours.
449 ///
450 /// Reads `file_scan_results.scan_layers` JSONB and rolls up verdict counts
451 /// per layer. `last_pass_or_skip` is the most recent timestamp at which the
452 /// layer returned a non-error, non-fail verdict, the indicator the admin
453 /// panel uses to flag a layer as down.
454 #[tracing::instrument(skip_all)]
455 pub async fn layer_health_window(
456 db: &PgPool,
457 hours: i64,
458 ) -> Result<Vec<LayerHealthRow>, sqlx::Error> {
459 sqlx::query_as::<_, LayerHealthRow>(
460 r"
461 WITH expanded AS (
462 SELECT fsr.scanned_at,
463 (l ->> 'layer') AS layer,
464 (l ->> 'verdict') AS verdict
465 FROM file_scan_results fsr,
466 jsonb_array_elements(fsr.scan_layers) AS l
467 WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
468 )
469 SELECT layer,
470 COUNT(*) FILTER (WHERE verdict = 'pass') AS pass_count,
471 COUNT(*) FILTER (WHERE verdict = 'skip') AS skip_count,
472 COUNT(*) FILTER (WHERE verdict = 'fail') AS fail_count,
473 COUNT(*) FILTER (WHERE verdict = 'error') AS error_count,
474 MAX(scanned_at) FILTER (WHERE verdict IN ('pass', 'skip')) AS last_pass_or_skip
475 FROM expanded
476 GROUP BY layer
477 ORDER BY layer
478 ",
479 )
480 .bind(hours.to_string())
481 .fetch_all(db)
482 .await
483 }
484
485 /// A scan-history row for the dashboard's "Recent" grid.
486 #[derive(Debug, Clone, FromRow)]
487 pub struct ScanHistoryRow {
488 pub scanned_at: DateTime<Utc>,
489 pub s3_key: String,
490 pub scan_status: String,
491 pub sha256: Option<String>,
492 pub file_size_bytes: Option<i64>,
493 pub scan_layers: serde_json::Value,
494 }
495
496 /// Aggregate counts of entities currently in non-clean states. Used by the
497 /// PoM health endpoint to alert on growing review backlogs.
498 #[derive(Debug, Clone)]
499 pub struct HeldCounts {
500 pub held_versions: i64,
501 pub held_items: i64,
502 pub held_media: i64,
503 }
504
505 #[tracing::instrument(skip_all)]
506 pub async fn held_counts(db: &PgPool) -> Result<HeldCounts, sqlx::Error> {
507 let held_versions: i64 =
508 sqlx::query_scalar("SELECT COUNT(*) FROM versions WHERE scan_status = 'held_for_review'")
509 .fetch_one(db)
510 .await?;
511 let held_items: i64 =
512 sqlx::query_scalar("SELECT COUNT(*) FROM items WHERE scan_status = 'held_for_review'")
513 .fetch_one(db)
514 .await?;
515 let held_media: i64 = sqlx::query_scalar(
516 "SELECT COUNT(*) FROM media_files WHERE scan_status = 'held_for_review'",
517 )
518 .fetch_one(db)
519 .await?;
520 Ok(HeldCounts {
521 held_versions,
522 held_items,
523 held_media,
524 })
525 }
526
527 /// Recent scan results across all entities. Newest first, capped at `limit`.
528 /// Used by the Recent History collapsible section. `since_hours` bounds the
529 /// window so the grid renders fast.
530 #[tracing::instrument(skip_all)]
531 pub async fn recent_history(
532 db: &PgPool,
533 since_hours: i64,
534 limit: i64,
535 ) -> Result<Vec<ScanHistoryRow>, sqlx::Error> {
536 sqlx::query_as::<_, ScanHistoryRow>(
537 r"
538 SELECT fsr.scanned_at, fsr.s3_key, fsr.scan_status,
539 fsr.sha256, fsr.file_size_bytes, fsr.scan_layers
540 FROM file_scan_results fsr
541 WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
542 ORDER BY fsr.scanned_at DESC
543 LIMIT $2
544 ",
545 )
546 .bind(since_hours.to_string())
547 .bind(limit)
548 .fetch_all(db)
549 .await
550 }
551