1use crate::prelude::*;
10use chrono::Duration;
11
12const REAP_BATCH_LIMIT: i64 = 1000;
15
16pub async fn insert_many(
19 conn: &mut PgConnection,
20 exercise_service_slug: &str,
21 uploaded_by_user: Option<Uuid>,
22 file_upload_ids: &[Uuid],
23) -> ModelResult<()> {
24 sqlx::query!(
25 "
26INSERT INTO exercise_spec_uploads (file_upload_id, exercise_service_slug, uploaded_by_user)
27SELECT file_upload_id,
28 $2,
29 $3
30FROM UNNEST($1::uuid []) AS t(file_upload_id)
31",
32 file_upload_ids,
33 exercise_service_slug,
34 uploaded_by_user
35 )
36 .execute(conn)
37 .await?;
38 Ok(())
39}
40
41#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct ReapableUpload {
44 pub id: Uuid,
45 pub file_upload_id: Uuid,
46 pub path: String,
48}
49
50pub async fn get_reapable(conn: &mut PgConnection) -> ModelResult<Vec<ReapableUpload>> {
78 let res = sqlx::query_as!(
79 ReapableUpload,
80 "
81SELECT u.id,
82 u.file_upload_id,
83 f.path
84FROM exercise_spec_uploads AS u
85 JOIN file_uploads AS f ON f.id = u.file_upload_id
86WHERE f.deleted_at IS NULL
87 AND u.created_at < now() - interval '7 days'
88 AND (
89 u.exercise_service_slug = 'playground'
90 OR EXISTS (
91 SELECT 1
92 FROM exercise_services AS s
93 JOIN exercise_service_info AS i ON i.exercise_service_id = s.id
94 WHERE s.slug = u.exercise_service_slug
95 AND s.deleted_at IS NULL
96 AND i.declares_spec_files
97 )
98 )
99 AND NOT EXISTS (
100 SELECT 1
101 FROM exercise_task_spec_files AS t
102 WHERE t.file_upload_id = u.file_upload_id
103 AND t.deleted_at IS NULL
104 )
105 AND NOT EXISTS (
106 SELECT 1
107 FROM page_history_spec_files AS h
108 WHERE h.file_upload_id = u.file_upload_id
109 AND h.deleted_at IS NULL
110 )
111ORDER BY u.created_at
112LIMIT $1
113",
114 REAP_BATCH_LIMIT
115 )
116 .fetch_all(conn)
117 .await?;
118 Ok(res)
119}
120
121pub async fn mark_reaped(conn: &mut PgConnection, id: Uuid) -> ModelResult<bool> {
130 let mut tx = conn.begin().await?;
131 let locked = sqlx::query_scalar!(
132 "
133SELECT id
134FROM exercise_spec_uploads
135WHERE id = $1
136FOR UPDATE
137",
138 id
139 )
140 .fetch_optional(&mut *tx)
141 .await?;
142 if locked.is_none() {
143 tx.rollback().await?;
144 return Ok(false);
145 }
146 let retired = sqlx::query_scalar!(
147 "
148UPDATE exercise_spec_uploads AS u
149SET deleted_at = COALESCE(u.deleted_at, now())
150WHERE u.id = $1
151 AND NOT EXISTS (
152 SELECT 1
153 FROM exercise_task_spec_files AS t
154 WHERE t.file_upload_id = u.file_upload_id
155 AND t.deleted_at IS NULL
156 )
157 AND NOT EXISTS (
158 SELECT 1
159 FROM page_history_spec_files AS h
160 WHERE h.file_upload_id = u.file_upload_id
161 AND h.deleted_at IS NULL
162 )
163RETURNING u.id
164",
165 id
166 )
167 .fetch_optional(&mut *tx)
168 .await?;
169 tx.commit().await?;
170 Ok(retired.is_some())
171}
172
173#[derive(Debug, Clone, PartialEq, Eq)]
175pub struct SpecUpload {
176 pub id: Uuid,
177 pub file_upload_id: Uuid,
178 pub deleted: bool,
179}
180
181pub async fn get_by_file_upload_id(
182 conn: &mut PgConnection,
183 file_upload_id: Uuid,
184) -> ModelResult<Option<SpecUpload>> {
185 let res = sqlx::query!(
186 "
187SELECT id,
188 file_upload_id,
189 deleted_at
190FROM exercise_spec_uploads
191WHERE file_upload_id = $1
192",
193 file_upload_id
194 )
195 .fetch_optional(conn)
196 .await?;
197 Ok(res.map(|row| SpecUpload {
198 id: row.id,
199 file_upload_id: row.file_upload_id,
200 deleted: row.deleted_at.is_some(),
201 }))
202}
203
204pub async fn backdate(
206 conn: &mut PgConnection,
207 file_upload_id: Uuid,
208 age: Duration,
209) -> ModelResult<()> {
210 sqlx::query!(
211 "
212UPDATE exercise_spec_uploads
213SET created_at = now() - $2::interval
214WHERE file_upload_id = $1
215",
216 file_upload_id,
217 age as Duration
218 )
219 .execute(conn)
220 .await?;
221 Ok(())
222}
223
224#[cfg(test)]
225mod test {
226 use super::*;
227 use crate::exercise_task_spec_files::SpecKind;
228 use crate::test_helper::*;
229
230 const DECLARING_SLUG: &str = "declaring-service";
231 const SILENT_SLUG: &str = "silent-service";
232
233 async fn insert_file(tx: &mut PgConnection, name: &str) -> Uuid {
234 crate::file_uploads::insert(
235 tx,
236 name,
237 &format!("{DECLARING_SLUG}/{name}"),
238 "application/octet-stream",
239 None,
240 None,
241 )
242 .await
243 .unwrap()
244 }
245
246 async fn insert_service(tx: &mut PgConnection, slug: &str, declares_spec_files: bool) {
249 let service = crate::exercise_services::insert_exercise_service(
250 tx,
251 &crate::exercise_services::ExerciseServiceNewOrUpdate {
252 name: slug.to_string(),
253 slug: slug.to_string(),
254 public_url: format!("http://{slug}.example.com/api/service-info"),
255 internal_url: None,
256 max_reprocessing_submissions_at_once: 1,
257 },
258 )
259 .await
260 .unwrap();
261 crate::exercise_service_info::insert(
262 tx,
263 &crate::exercise_service_info::PathInfo {
264 exercise_service_id: service.id,
265 user_interface_iframe_path: "/iframe".to_string(),
266 grade_endpoint_path: "/api/grade".to_string(),
267 public_spec_endpoint_path: "/api/public-spec".to_string(),
268 model_solution_spec_endpoint_path: "/api/model-solution".to_string(),
269 has_custom_view: false,
270 supports_native_client: false,
271 produces_file_answers: false,
272 declares_spec_files,
273 },
274 )
275 .await
276 .unwrap();
277 }
278
279 async fn insert_stale_upload(tx: &mut PgConnection, slug: &str, name: &str) -> Uuid {
281 let file_id = insert_file(&mut *tx, name).await;
282 insert_many(&mut *tx, slug, None, &[file_id]).await.unwrap();
283 backdate(&mut *tx, file_id, Duration::days(8))
284 .await
285 .unwrap();
286 file_id
287 }
288
289 fn lists(reapable: &[ReapableUpload], file_upload_id: Uuid) -> bool {
290 reapable
291 .iter()
292 .any(|upload| upload.file_upload_id == file_upload_id)
293 }
294
295 #[tokio::test]
296 async fn lists_an_upload_no_spec_declares() {
297 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:_task);
298 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
299 let file_id = insert_stale_upload(tx.as_mut(), DECLARING_SLUG, "abandoned").await;
300
301 let reapable = get_reapable(tx.as_mut()).await.unwrap();
302
303 assert!(lists(&reapable, file_id));
304 }
305
306 #[tokio::test]
307 async fn spares_an_upload_inside_the_retention_window() {
308 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:_task);
309 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
310 let file_id = insert_file(tx.as_mut(), "fresh").await;
311 insert_many(tx.as_mut(), DECLARING_SLUG, None, &[file_id])
312 .await
313 .unwrap();
314
315 let reapable = get_reapable(tx.as_mut()).await.unwrap();
316
317 assert!(!lists(&reapable, file_id));
318 }
319
320 #[tokio::test]
323 async fn never_lists_an_upload_of_a_service_that_declares_nothing() {
324 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:_task);
325 insert_service(tx.as_mut(), SILENT_SLUG, false).await;
326 let file_id = insert_stale_upload(tx.as_mut(), SILENT_SLUG, "kept-forever").await;
327
328 let reapable = get_reapable(tx.as_mut()).await.unwrap();
329
330 assert!(!lists(&reapable, file_id));
331 }
332
333 #[tokio::test]
336 async fn lists_a_playground_upload_although_no_service_declares() {
337 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:_task);
338 let file_id = insert_stale_upload(tx.as_mut(), "playground", "playground-file").await;
339
340 let reapable = get_reapable(tx.as_mut()).await.unwrap();
341
342 assert!(lists(&reapable, file_id));
343 }
344
345 #[tokio::test]
346 async fn spares_an_upload_a_live_spec_declares() {
347 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:task_id);
348 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
349 let file_id = insert_stale_upload(tx.as_mut(), DECLARING_SLUG, "in-a-spec").await;
350 crate::exercise_task_spec_files::replace_for_exercise_task(
351 tx.as_mut(),
352 task_id,
353 SpecKind::Private,
354 &[file_id],
355 )
356 .await
357 .unwrap();
358
359 let reapable = get_reapable(tx.as_mut()).await.unwrap();
360
361 assert!(!lists(&reapable, file_id));
362 }
363
364 #[tokio::test]
367 async fn spares_an_upload_only_a_derived_spec_declares() {
368 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:task_id);
369 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
370 let file_id = insert_stale_upload(tx.as_mut(), DECLARING_SLUG, "stub-archive").await;
371 crate::exercise_task_spec_files::replace_for_exercise_task(
372 tx.as_mut(),
373 task_id,
374 SpecKind::Public,
375 &[file_id],
376 )
377 .await
378 .unwrap();
379
380 let reapable = get_reapable(tx.as_mut()).await.unwrap();
381
382 assert!(!lists(&reapable, file_id));
383 }
384
385 #[tokio::test]
388 async fn spares_an_upload_only_page_history_declares() {
389 insert_data!(:tx, user:user_id, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:page_id, exercise:_exercise, slide:_slide, task:_task);
390 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
391 let file_id =
392 insert_stale_upload(tx.as_mut(), DECLARING_SLUG, "dropped-but-in-history").await;
393 let history_id = crate::page_history::insert(
394 tx.as_mut(),
395 PKeyPolicy::Generate,
396 page_id,
397 "Snapshot",
398 &crate::page_history::PageHistoryContent {
399 content: serde_json::json!([]),
400 exercises: vec![],
401 exercise_slides: vec![],
402 exercise_tasks: vec![],
403 peer_or_self_review_configs: vec![],
404 peer_or_self_review_questions: vec![],
405 },
406 crate::page_history::HistoryChangeReason::PageSaved,
407 user_id,
408 None,
409 )
410 .await
411 .unwrap();
412 crate::page_history_spec_files::insert_many(tx.as_mut(), history_id, &[file_id])
413 .await
414 .unwrap();
415
416 let reapable = get_reapable(tx.as_mut()).await.unwrap();
417
418 assert!(!lists(&reapable, file_id));
419 }
420
421 #[tokio::test]
424 async fn declines_to_retire_an_upload_declared_after_it_was_listed() {
425 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:task_id);
426 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
427 let file_id = insert_stale_upload(tx.as_mut(), DECLARING_SLUG, "declared-late").await;
428 let listed = get_reapable(tx.as_mut()).await.unwrap();
429 let upload = listed
430 .iter()
431 .find(|upload| upload.file_upload_id == file_id)
432 .expect("listed");
433 crate::exercise_task_spec_files::replace_for_exercise_task(
434 tx.as_mut(),
435 task_id,
436 SpecKind::Private,
437 &[file_id],
438 )
439 .await
440 .unwrap();
441
442 assert!(!mark_reaped(tx.as_mut(), upload.id).await.unwrap());
443 assert_eq!(
444 get_by_file_upload_id(tx.as_mut(), file_id)
445 .await
446 .unwrap()
447 .map(|recorded| recorded.deleted),
448 Some(false)
449 );
450 }
451
452 #[tokio::test]
453 async fn retires_an_upload_nothing_declares() {
454 insert_data!(:tx, user:_user, :org, course:_course, instance:_instance, course_module:_cm, chapter:_chapter, page:_page, exercise:_exercise, slide:_slide, task:_task);
455 insert_service(tx.as_mut(), DECLARING_SLUG, true).await;
456 let file_id = insert_stale_upload(tx.as_mut(), DECLARING_SLUG, "retired").await;
457 let recorded = get_by_file_upload_id(tx.as_mut(), file_id)
458 .await
459 .unwrap()
460 .expect("recorded");
461
462 assert!(mark_reaped(tx.as_mut(), recorded.id).await.unwrap());
463
464 assert_eq!(
465 get_by_file_upload_id(tx.as_mut(), file_id)
466 .await
467 .unwrap()
468 .map(|recorded| recorded.deleted),
469 Some(true),
470 "the row survives soft-deleted, as the audit trail of what was reclaimed"
471 );
472 }
473}