Skip to main content

max / makenotwork

8.9 KB · 288 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 /// Insert a scan result record for audit trail.
44 #[tracing::instrument(skip_all)]
45 pub async fn insert_scan_result(
46 db: &PgPool,
47 s3_key: &str,
48 result: &ScanResult,
49 ) -> Result<Uuid, sqlx::Error> {
50 let layers_json = serde_json::to_value(&result.layers)
51 .unwrap_or_else(|_| serde_json::Value::Array(vec![]));
52
53 let id = sqlx::query_scalar::<_, Uuid>(
54 r#"
55 INSERT INTO file_scan_results (s3_key, scan_status, scan_layers, sha256, file_size_bytes)
56 VALUES ($1, $2, $3, $4, $5)
57 RETURNING id
58 "#,
59 )
60 .bind(s3_key)
61 .bind(result.status)
62 .bind(&layers_json)
63 .bind(&result.sha256)
64 .bind(result.file_size as i64)
65 .fetch_one(db)
66 .await?;
67
68 Ok(id)
69 }
70
71 /// Update the scan_status column on an item.
72 #[tracing::instrument(skip_all)]
73 pub async fn update_item_scan_status(
74 db: &PgPool,
75 item_id: ItemId,
76 status: FileScanStatus,
77 ) -> Result<(), sqlx::Error> {
78 sqlx::query(
79 r#"
80 UPDATE items SET scan_status = $1, updated_at = NOW() WHERE id = $2
81 "#,
82 )
83 .bind(status)
84 .bind(item_id)
85 .execute(db)
86 .await?;
87
88 Ok(())
89 }
90
91 /// Update the scan_status column on a version.
92 #[tracing::instrument(skip_all)]
93 pub async fn update_version_scan_status(
94 db: &PgPool,
95 version_id: VersionId,
96 status: FileScanStatus,
97 ) -> Result<(), sqlx::Error> {
98 sqlx::query(
99 r#"
100 UPDATE versions SET scan_status = $1 WHERE id = $2
101 "#,
102 )
103 .bind(status)
104 .bind(version_id)
105 .execute(db)
106 .await?;
107
108 Ok(())
109 }
110
111 /// Update the scan_status column on a media file.
112 #[tracing::instrument(skip_all)]
113 pub async fn update_media_file_scan_status(
114 db: &PgPool,
115 media_file_id: crate::db::MediaFileId,
116 status: FileScanStatus,
117 ) -> Result<(), sqlx::Error> {
118 let id: uuid::Uuid = media_file_id.into();
119 sqlx::query("UPDATE media_files SET scan_status = $1 WHERE id = $2")
120 .bind(status)
121 .bind(id)
122 .execute(db)
123 .await?;
124 Ok(())
125 }
126
127 /// Get items held for review, joined with creator info + latest scan layers.
128 /// Oldest first.
129 #[tracing::instrument(skip_all)]
130 pub async fn get_held_items(db: &PgPool) -> Result<Vec<HeldItemRow>, sqlx::Error> {
131 let rows = sqlx::query_as::<_, HeldItemRow>(
132 r#"
133 SELECT i.id AS item_id, i.title AS item_title,
134 COALESCE(i.audio_s3_key, i.cover_s3_key) AS s3_key,
135 u.username AS creator_username, u.id AS creator_id,
136 u.upload_trusted, i.updated_at AS held_at,
137 (
138 SELECT fsr.scan_layers FROM file_scan_results fsr
139 WHERE fsr.s3_key = COALESCE(i.audio_s3_key, i.cover_s3_key)
140 ORDER BY fsr.scanned_at DESC LIMIT 1
141 ) AS scan_layers
142 FROM items i
143 JOIN projects p ON p.id = i.project_id
144 JOIN users u ON u.id = p.user_id
145 WHERE i.scan_status = 'held_for_review'
146 ORDER BY i.updated_at ASC
147 "#,
148 )
149 .fetch_all(db)
150 .await?;
151
152 Ok(rows)
153 }
154
155 /// Get versions held for review, joined with creator info + latest scan layers.
156 /// Oldest first.
157 #[tracing::instrument(skip_all)]
158 pub async fn get_held_versions(db: &PgPool) -> Result<Vec<HeldVersionRow>, sqlx::Error> {
159 let rows = sqlx::query_as::<_, HeldVersionRow>(
160 r#"
161 SELECT i.id AS item_id, i.title AS item_title,
162 v.id AS version_id, v.version_number,
163 v.s3_key,
164 u.username AS creator_username, u.id AS creator_id,
165 u.upload_trusted, v.created_at AS held_at,
166 (
167 SELECT fsr.scan_layers FROM file_scan_results fsr
168 WHERE fsr.s3_key = v.s3_key
169 ORDER BY fsr.scanned_at DESC LIMIT 1
170 ) AS scan_layers
171 FROM versions v
172 JOIN items i ON i.id = v.item_id
173 JOIN projects p ON p.id = i.project_id
174 JOIN users u ON u.id = p.user_id
175 WHERE v.scan_status = 'held_for_review'
176 ORDER BY v.created_at ASC
177 "#,
178 )
179 .fetch_all(db)
180 .await?;
181
182 Ok(rows)
183 }
184
185 /// Per-layer aggregate stats over a window for the admin dashboard.
186 #[derive(Debug, Clone, FromRow)]
187 pub struct LayerHealthRow {
188 pub layer: String,
189 pub pass_count: i64,
190 pub skip_count: i64,
191 pub fail_count: i64,
192 pub error_count: i64,
193 pub last_pass_or_skip: Option<DateTime<Utc>>,
194 }
195
196 /// Compute per-layer health stats over the last N hours.
197 ///
198 /// Reads `file_scan_results.scan_layers` JSONB and rolls up verdict counts
199 /// per layer. `last_pass_or_skip` is the most recent timestamp at which the
200 /// layer returned a non-error, non-fail verdict — the indicator the admin
201 /// panel uses to flag a layer as down.
202 #[tracing::instrument(skip_all)]
203 pub async fn layer_health_window(
204 db: &PgPool,
205 hours: i64,
206 ) -> Result<Vec<LayerHealthRow>, sqlx::Error> {
207 sqlx::query_as::<_, LayerHealthRow>(
208 r#"
209 WITH expanded AS (
210 SELECT fsr.scanned_at,
211 (l ->> 'layer') AS layer,
212 (l ->> 'verdict') AS verdict
213 FROM file_scan_results fsr,
214 jsonb_array_elements(fsr.scan_layers) AS l
215 WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
216 )
217 SELECT layer,
218 COUNT(*) FILTER (WHERE verdict = 'pass') AS pass_count,
219 COUNT(*) FILTER (WHERE verdict = 'skip') AS skip_count,
220 COUNT(*) FILTER (WHERE verdict = 'fail') AS fail_count,
221 COUNT(*) FILTER (WHERE verdict = 'error') AS error_count,
222 MAX(scanned_at) FILTER (WHERE verdict IN ('pass', 'skip')) AS last_pass_or_skip
223 FROM expanded
224 GROUP BY layer
225 ORDER BY layer
226 "#,
227 )
228 .bind(hours.to_string())
229 .fetch_all(db)
230 .await
231 }
232
233 /// A scan-history row for the dashboard's "Recent" grid.
234 #[derive(Debug, Clone, FromRow)]
235 pub struct ScanHistoryRow {
236 pub scanned_at: DateTime<Utc>,
237 pub s3_key: String,
238 pub scan_status: String,
239 pub sha256: Option<String>,
240 pub file_size_bytes: Option<i64>,
241 pub scan_layers: serde_json::Value,
242 }
243
244 /// Aggregate counts of entities currently in non-clean states. Used by the
245 /// PoM health endpoint to alert on growing review backlogs.
246 #[derive(Debug, Clone)]
247 pub struct HeldCounts {
248 pub held_versions: i64,
249 pub held_items: i64,
250 pub held_media: i64,
251 }
252
253 #[tracing::instrument(skip_all)]
254 pub async fn held_counts(db: &PgPool) -> Result<HeldCounts, sqlx::Error> {
255 let held_versions: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM versions WHERE scan_status = 'held_for_review'")
256 .fetch_one(db).await?;
257 let held_items: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM items WHERE scan_status = 'held_for_review'")
258 .fetch_one(db).await?;
259 let held_media: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM media_files WHERE scan_status = 'held_for_review'")
260 .fetch_one(db).await?;
261 Ok(HeldCounts { held_versions, held_items, held_media })
262 }
263
264 /// Recent scan results across all entities. Newest first, capped at `limit`.
265 /// Used by the Recent History collapsible section. `since_hours` bounds the
266 /// window so the grid renders fast.
267 #[tracing::instrument(skip_all)]
268 pub async fn recent_history(
269 db: &PgPool,
270 since_hours: i64,
271 limit: i64,
272 ) -> Result<Vec<ScanHistoryRow>, sqlx::Error> {
273 sqlx::query_as::<_, ScanHistoryRow>(
274 r#"
275 SELECT fsr.scanned_at, fsr.s3_key, fsr.scan_status,
276 fsr.sha256, fsr.file_size_bytes, fsr.scan_layers
277 FROM file_scan_results fsr
278 WHERE fsr.scanned_at > NOW() - ($1 || ' hours')::interval
279 ORDER BY fsr.scanned_at DESC
280 LIMIT $2
281 "#,
282 )
283 .bind(since_hours.to_string())
284 .bind(limit)
285 .fetch_all(db)
286 .await
287 }
288