Skip to main content

max / makenotwork

21.7 KB · 542 lines History Blame Raw
1 //! Scheduled cleanup jobs: sandbox expiry, terminated accounts, content
2 //! removal, IP scrubbing, stale pending transactions, orphaned uploads, cart
3 //! items, soft-deleted item purges, and pending S3 deletion retries.
4
5 use crate::constants;
6 use crate::db;
7 use crate::AppState;
8
9 /// Hourly: purge `scan_jobs` rows in a terminal state older than the
10 /// retention window. Queued/running rows are not touched.
11 pub(super) async fn purge_old_scan_jobs(state: &AppState) {
12 let window = chrono::Duration::days(constants::SCAN_JOB_RETENTION_DAYS as i64);
13 match db::scan_jobs::purge_old_terminal(&state.db, window).await {
14 Ok(n) => {
15 if n > 0 {
16 tracing::info!(purged = n, "scan_jobs retention sweep");
17 }
18 let _ = db::scheduler_jobs::record_job_run(
19 &state.db, "scan_jobs_retention", n as i64,
20 ).await;
21 }
22 Err(e) => tracing::error!(error = ?e, "scan_jobs retention sweep failed"),
23 }
24 }
25
26 /// Delete expired sandbox accounts and their S3 objects.
27 pub(super) async fn cleanup_sandbox_accounts(state: &AppState) {
28 let expired_ids = match db::users::get_expired_sandbox_ids(&state.db).await {
29 Ok(ids) => ids,
30 Err(e) => {
31 tracing::error!(error = ?e, "failed to query expired sandbox accounts");
32 return;
33 }
34 };
35
36 // Drain the expired list with bounded concurrency (4 in-flight). Each
37 // cleanup holds a few pool conns + does S3 delete-prefix; a serial loop
38 // is fine for handfuls but pins the tick for tens of expired accounts.
39 // Cap of 4 leaves the remaining ~21 pool conns for request handlers.
40 const CLEANUP_PARALLELISM: usize = 4;
41 let mut set = tokio::task::JoinSet::new();
42 let mut iter = expired_ids.iter().copied();
43 let mut deleted = 0i64;
44 loop {
45 while set.len() < CLEANUP_PARALLELISM {
46 match iter.next() {
47 Some(uid) => {
48 let state = state.clone();
49 set.spawn(async move {
50 cleanup_user_s3_and_delete(
51 &state, uid, "sandbox_expired", "sandbox",
52 ).await
53 });
54 }
55 None => break,
56 }
57 }
58 match set.join_next().await {
59 Some(Ok(true)) => deleted += 1,
60 Some(Ok(false)) => {}
61 Some(Err(e)) => tracing::warn!(error = ?e, "sandbox_cleanup task panicked"),
62 None => break,
63 }
64 }
65 let _ = db::scheduler_jobs::record_job_run(&state.db, "sandbox_cleanup", deleted).await;
66 }
67
68 /// Clean up a user's S3 objects, git repos, and CASCADE-delete the user row.
69 ///
70 /// Shared between sandbox, terminated, and content-removal account cleanup.
71 /// S3 objects are deleted first (before CASCADE removes the DB rows that reference them).
72 async fn cleanup_user_s3_and_delete(state: &AppState, user_id: db::UserId, event: &str, label: &str) -> bool {
73 // Resolve everything we need from the DB once — the enqueue list and the
74 // delete list must come from the same snapshot, or a project/app created
75 // between calls will be enqueued but not deleted (or vice versa).
76 let project_ids = db::projects::get_project_ids_for_user(&state.db, user_id)
77 .await
78 .unwrap_or_default();
79 let sync_apps = db::synckit::get_sync_apps_by_creator(&state.db, user_id)
80 .await
81 .unwrap_or_default();
82
83 let user_prefix = format!("{user_id}/");
84 let mut keys: Vec<(String, String)> = Vec::new();
85 keys.push((user_prefix.clone(), "main".to_string()));
86 for pid in &project_ids {
87 keys.push((format!("projects/{pid}/"), "main".to_string()));
88 }
89 for app in &sync_apps {
90 keys.push((format!("{}/", app.id), "synckit".to_string()));
91 keys.push((format!("ota/{}/", app.id), "synckit".to_string()));
92 }
93
94 // Enqueue all keys before any destructive work
95 if let Err(e) = db::pending_s3_deletions::enqueue_deletions(&state.db, &keys, label).await {
96 tracing::error!(error = ?e, %user_id, "{label}: failed to enqueue S3 deletions, aborting cleanup");
97 return false;
98 }
99
100 if let Some(ref s3) = state.s3 {
101 if let Err(e) = s3.delete_prefix(&user_prefix).await {
102 tracing::warn!(error = ?e, %user_id, "{label}: failed to delete user S3 objects");
103 }
104 for pid in &project_ids {
105 let proj_prefix = format!("projects/{pid}/");
106 if let Err(e) = s3.delete_prefix(&proj_prefix).await {
107 tracing::warn!(error = ?e, %user_id, %pid, "{label}: failed to delete project S3 objects");
108 }
109 }
110 }
111
112 if let Some(ref synckit_s3) = state.synckit_s3 {
113 for app in &sync_apps {
114 let blob_prefix = format!("{}/", app.id);
115 if let Err(e) = synckit_s3.delete_prefix(&blob_prefix).await {
116 tracing::warn!(error = ?e, %user_id, app_id = %app.id, "{label}: failed to delete SyncKit blobs");
117 }
118 let ota_prefix = format!("ota/{}/", app.id);
119 if let Err(e) = synckit_s3.delete_prefix(&ota_prefix).await {
120 tracing::warn!(error = ?e, %user_id, app_id = %app.id, "{label}: failed to delete OTA artifacts");
121 }
122 }
123 }
124
125 // Git repos on disk
126 if let Some(ref git_root) = state.config.git_repos_path
127 && let Ok(Some(user)) = db::users::get_user_by_id(&state.db, user_id).await
128 {
129 cleanup_git_repos_on_disk(git_root, &user.username, user_id).await;
130 }
131
132 // CASCADE delete user row
133 if let Err(e) = db::users::delete_user(&state.db, user_id).await {
134 tracing::error!(error = ?e, %user_id, "{label}: failed to delete account");
135 false
136 } else {
137 tracing::info!(%user_id, event, "{label}: account cleaned up");
138 true
139 }
140 }
141
142 /// Remove a user's bare git repositories from disk.
143 ///
144 /// Must be called before `delete_user` (which CASCADE-deletes the git_repos rows).
145 /// Best-effort: logs warnings on failure but does not block account deletion.
146 /// Runs blocking I/O on a dedicated thread to avoid stalling the Tokio runtime.
147 pub(super) async fn cleanup_git_repos_on_disk(git_repos_path: &str, username: &str, user_id: db::UserId) {
148 let user_git_dir = std::path::Path::new(git_repos_path).join(username);
149 if user_git_dir.exists() {
150 let path = user_git_dir.clone();
151 match tokio::task::spawn_blocking(move || std::fs::remove_dir_all(&path)).await {
152 Ok(Ok(())) => tracing::info!(%user_id, path = %user_git_dir.display(), "deleted git repos from disk"),
153 Ok(Err(e)) => tracing::warn!(error = ?e, %user_id, path = %user_git_dir.display(), "failed to delete git repos from disk"),
154 Err(e) => tracing::warn!(error = ?e, %user_id, path = %user_git_dir.display(), "git repo cleanup task panicked"),
155 }
156 }
157 }
158
159 /// Delete accounts that were terminated >30 days ago.
160 ///
161 /// Per-user S3 sweeps + git-repo removal + CASCADE delete are spawned off the
162 /// scheduler's advisory-lock-held tick so a backlog of expired accounts
163 /// (5 creators × 20 projects of multi-page `delete_prefix` calls) doesn't
164 /// extend the lock hold past `TICK_DURATION_ALERT_SECS`. Each step in
165 /// `cleanup_user_s3_and_delete` is idempotent: `delete_prefix` on a missing
166 /// prefix is a no-op, the CASCADE `delete_user` on a non-existent row affects
167 /// zero rows. If a second tick fires before all spawned cleanups complete
168 /// (extremely large accounts), the next `get_expired_terminated_ids` query
169 /// excludes already-deleted rows; the worst-case race is one extra harmless
170 /// no-op call against an in-flight target.
171 pub(super) async fn delete_expired_terminated_accounts(state: &AppState) {
172 spawn_expired_account_cleanups(
173 state,
174 db::users::get_expired_terminated_ids(&state.db).await,
175 "termination_expired",
176 "terminated account",
177 "terminated_account_cleanup",
178 )
179 .await;
180 }
181
182 /// Delete creator accounts whose 90-day content removal grace period has expired.
183 /// Off-lock spawn pattern — see `delete_expired_terminated_accounts`.
184 pub(super) async fn delete_expired_content_removal_accounts(state: &AppState) {
185 spawn_expired_account_cleanups(
186 state,
187 db::users::get_expired_content_removal_ids(&state.db).await,
188 "content_removal_expired",
189 "content removal",
190 "content_removal_cleanup",
191 )
192 .await;
193 }
194
195 async fn spawn_expired_account_cleanups(
196 state: &AppState,
197 fetched: Result<Vec<db::UserId>, crate::error::AppError>,
198 event: &'static str,
199 label: &'static str,
200 job_name: &'static str,
201 ) {
202 let expired_ids = match fetched {
203 Ok(ids) if ids.is_empty() => {
204 let _ = db::scheduler_jobs::record_job_run(&state.db, job_name, 0).await;
205 return;
206 }
207 Ok(ids) => ids,
208 Err(e) => {
209 tracing::error!(error = ?e, %job_name, "failed to query expired accounts");
210 return;
211 }
212 };
213
214 let scheduled = expired_ids.len() as i64;
215 for user_id in expired_ids {
216 let state_for_cleanup = state.clone();
217 tokio::spawn(async move {
218 cleanup_user_s3_and_delete(&state_for_cleanup, user_id, event, label).await;
219 });
220 }
221 // Record the count *scheduled*, not *deleted*: the deletes finish out of band.
222 // Operators read this metric to confirm the scheduler saw the work, not to
223 // count completed sweeps. cleanup_user_s3_and_delete logs its own success/
224 // failure per user.
225 let _ = db::scheduler_jobs::record_job_run(&state.db, job_name, scheduled).await;
226 }
227
228 /// Delete pending transactions older than 25 hours and release promo code reservations.
229 pub(super) async fn cleanup_stale_pending_transactions(state: &AppState) {
230 let promo_ids = match db::transactions::cleanup_stale_pending(
231 &state.db,
232 chrono::Duration::hours(25),
233 )
234 .await
235 {
236 Ok(ids) => ids,
237 Err(e) => {
238 tracing::error!(error = ?e, "failed to clean up stale pending transactions");
239 return;
240 }
241 };
242
243 // Cart checkouts produce N pending-tx rows that share a promo_code_id;
244 // release once per reservation, not once per row.
245 let unique_promo_ids: std::collections::HashSet<_> = promo_ids.into_iter().flatten().collect();
246
247 let mut released = 0i64;
248 for pc_id in unique_promo_ids {
249 if let Err(e) = db::promo_codes::release_use_count(&state.db, pc_id).await {
250 tracing::warn!(promo_code_id = %pc_id, error = ?e, "failed to release promo code use count");
251 } else {
252 released += 1;
253 }
254 }
255
256 if released > 0 {
257 tracing::info!(released, "released promo code reservations from stale pending transactions");
258 }
259 let _ = db::scheduler_jobs::record_job_run(&state.db, "stale_pending_cleanup", released).await;
260 }
261
262 /// NULL out IP addresses older than 30 days in user_sessions.
263 pub(super) async fn scrub_stale_ip_addresses(state: &AppState) {
264 let cutoff = chrono::Utc::now() - chrono::Duration::days(30);
265
266 match sqlx::query(
267 "UPDATE user_sessions SET ip_address = NULL WHERE ip_address IS NOT NULL AND created_at < $1",
268 )
269 .bind(cutoff)
270 .execute(&state.db)
271 .await
272 {
273 Ok(r) => {
274 if r.rows_affected() > 0 {
275 tracing::info!(scrubbed = r.rows_affected(), "scrubbed stale IP addresses (30-day retention)");
276 }
277 let _ = db::scheduler_jobs::record_job_run(&state.db, "ip_scrub", r.rows_affected() as i64).await;
278 }
279 Err(e) => tracing::error!(error = ?e, "failed to scrub IPs from user_sessions"),
280 }
281 }
282
283 /// Permanently delete items that were soft-deleted more than 7 days ago.
284 /// Cleans up S3 objects (item files + version files) and decrements storage
285 /// before DB deletion to prevent orphaned storage and accounting drift.
286 pub(super) async fn purge_expired_deleted_items(state: &AppState) {
287 // Collect S3 keys from items AND their versions before CASCADE delete destroys the data
288 let mut all_s3_keys: Vec<(String, String)> = Vec::new();
289
290 match db::items::get_expired_deleted_item_s3_keys(&state.db).await {
291 Ok(keys) => {
292 for key in &keys {
293 all_s3_keys.push((key.clone(), "main".to_string()));
294 }
295 }
296 Err(e) => {
297 tracing::error!(error = ?e, "failed to query item S3 keys for items pending purge");
298 }
299 }
300
301 match db::items::get_expired_deleted_item_version_s3_keys(&state.db).await {
302 Ok(keys) => {
303 for key in &keys {
304 all_s3_keys.push((key.clone(), "main".to_string()));
305 }
306 }
307 Err(e) => {
308 tracing::error!(error = ?e, "failed to query version S3 keys for items pending purge");
309 }
310 }
311
312 // Enqueue all keys as a durable safety net before any destructive work
313 if !all_s3_keys.is_empty()
314 && let Err(e) = db::pending_s3_deletions::enqueue_deletions(&state.db, &all_s3_keys, "purge_deleted_items").await
315 {
316 tracing::error!(error = ?e, "failed to enqueue S3 deletions for purged items, aborting purge");
317 return;
318 }
319
320 if let Some(ref s3) = state.s3
321 && !all_s3_keys.is_empty()
322 {
323 let keys_only: Vec<String> = all_s3_keys.iter().map(|(k, _)| k.clone()).collect();
324 if let Err(e) = s3.delete_objects(&keys_only).await {
325 tracing::warn!(error = ?e, "batch S3 delete failed for purged items; pending_s3_deletions queue will retry");
326 }
327 tracing::info!(count = all_s3_keys.len(), "deleted S3 objects for purged items");
328 }
329
330 // Decrement storage for each affected user before CASCADE delete
331 match db::items::get_expired_deleted_item_storage_by_user(&state.db).await {
332 Ok(user_sizes) => {
333 for (user_id, total_bytes) in &user_sizes {
334 if *total_bytes > 0
335 && let Err(e) = db::creator_tiers::decrement_storage_used(&state.db, *user_id, *total_bytes).await
336 {
337 tracing::warn!(user_id = %user_id, bytes = total_bytes, error = ?e,
338 "failed to decrement storage for purged items");
339 }
340 }
341 }
342 Err(e) => {
343 tracing::error!(error = ?e, "failed to query storage sizes for items pending purge");
344 }
345 }
346
347 match db::items::purge_expired_deleted_items(&state.db).await {
348 Ok(0) => {
349 let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", 0).await;
350 }
351 Ok(n) => {
352 tracing::info!(deleted = n, "purged expired soft-deleted items");
353 let _ = db::scheduler_jobs::record_job_run(&state.db, "soft_delete_purge", n as i64).await;
354 }
355 Err(e) => {
356 tracing::error!(error = ?e, "failed to purge expired soft-deleted items");
357 }
358 }
359 }
360
361 /// Delete S3 objects from presigned uploads that were never confirmed (>24h old).
362 pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
363 let stale = match db::pending_uploads::get_stale_pending_uploads(
364 &state.db,
365 chrono::Duration::hours(24),
366 )
367 .await
368 {
369 Ok(rows) if rows.is_empty() => {
370 let _ = db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", 0).await;
371 return;
372 }
373 Ok(rows) => rows,
374 Err(e) => {
375 tracing::error!(error = ?e, "failed to query stale pending uploads");
376 return;
377 }
378 };
379
380 let mut cleaned = 0i64;
381 let mut keys_to_delete: Vec<String> = Vec::with_capacity(stale.len());
382
383 for (s3_key, bucket) in &stale {
384 let s3_client = match bucket.as_str() {
385 "synckit" => state.synckit_s3.as_ref(),
386 _ => state.s3.as_ref(),
387 };
388 if let Some(s3) = s3_client {
389 match s3.delete_object(s3_key).await {
390 Ok(()) => {
391 cleaned += 1;
392 keys_to_delete.push(s3_key.clone());
393 }
394 Err(e) => {
395 tracing::warn!(s3_key = %s3_key, bucket = %bucket, error = ?e, "failed to delete orphaned S3 object");
396 // Still remove the DB record so we don't retry forever
397 keys_to_delete.push(s3_key.clone());
398 }
399 }
400 } else {
401 // S3 not configured for this bucket; remove the DB record anyway
402 keys_to_delete.push(s3_key.clone());
403 }
404 }
405
406 if !keys_to_delete.is_empty()
407 && let Err(e) = db::pending_uploads::delete_pending_uploads(&state.db, &keys_to_delete).await
408 {
409 tracing::error!(error = ?e, "failed to delete pending upload records");
410 }
411
412 if cleaned > 0 {
413 tracing::info!(cleaned, "cleaned up orphaned presigned uploads");
414 }
415 let _ = db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", cleaned).await;
416 }
417
418 /// Remove stale cart items (>30 days old) and items that became unavailable.
419 pub(super) async fn cleanup_cart_items(state: &AppState) {
420 match db::cart::cleanup_stale_cart_items(&state.db, chrono::Duration::days(30)).await {
421 Ok(n) if n > 0 => tracing::info!(removed = n, "cleaned up stale cart items"),
422 Err(e) => tracing::error!(error = ?e, "failed to clean up stale cart items"),
423 _ => {}
424 }
425 match db::cart::cleanup_unavailable_cart_items(&state.db).await {
426 Ok(n) if n > 0 => tracing::info!(removed = n, "cleaned up unavailable cart items"),
427 Err(e) => tracing::error!(error = ?e, "failed to clean up unavailable cart items"),
428 _ => {}
429 }
430 }
431
432 /// Retry stale pending S3 deletions (older than 10 minutes, batch of 100).
433 pub(super) async fn retry_pending_s3_deletions(state: &AppState) {
434 let stale = match db::pending_s3_deletions::get_stale_pending(
435 &state.db,
436 chrono::Duration::minutes(10),
437 100,
438 ).await {
439 Ok(rows) => rows,
440 Err(e) => {
441 tracing::error!(error = ?e, "failed to fetch stale pending S3 deletions");
442 return;
443 }
444 };
445
446 if stale.is_empty() {
447 let _ = db::scheduler_jobs::record_job_run(&state.db, "s3_deletion_retry", 0).await;
448 return;
449 }
450
451 let mut completed_ids = Vec::new();
452 for row in &stale {
453 if row.attempts >= 10 {
454 tracing::error!(s3_key = %row.s3_key, bucket = %row.bucket, source = %row.source, attempts = row.attempts,
455 "S3 deletion dead-lettered after 10 attempts — removing from queue");
456 completed_ids.push(row.id);
457 continue;
458 } else if row.attempts >= 5 {
459 tracing::warn!(s3_key = %row.s3_key, bucket = %row.bucket, source = %row.source, attempts = row.attempts,
460 "S3 deletion stuck after 5+ attempts");
461 }
462
463 let s3 = if row.bucket == "synckit" {
464 state.synckit_s3.as_ref()
465 } else {
466 state.s3.as_ref()
467 };
468
469 if let Some(s3) = s3 {
470 // Delete-then-reupload race: a fresh upload may have reclaimed this
471 // s3_key after the deletion was enqueued. Skip and dequeue if any
472 // live row still references the key. Prefix deletes (user-scoped
473 // cascade cleanups) bypass the check.
474 if !row.s3_key.ends_with('/') {
475 match db::pending_s3_deletions::is_s3_key_live(&state.db, &row.bucket, &row.s3_key).await {
476 Ok(true) => {
477 tracing::info!(s3_key = %row.s3_key, bucket = %row.bucket,
478 "S3 deletion skipped — key reclaimed by a live row (delete-then-reupload)");
479 completed_ids.push(row.id);
480 continue;
481 }
482 Ok(false) => {}
483 Err(e) => {
484 tracing::warn!(s3_key = %row.s3_key, error = ?e,
485 "live-key check failed; proceeding with delete attempt");
486 }
487 }
488 }
489
490 let result = if row.s3_key.ends_with('/') {
491 s3.delete_prefix(&row.s3_key).await
492 } else {
493 s3.delete_object(&row.s3_key).await
494 };
495 match result {
496 Ok(()) => completed_ids.push(row.id),
497 Err(e) => {
498 tracing::warn!(s3_key = %row.s3_key, error = ?e, "retry S3 deletion failed");
499 }
500 }
501 } else {
502 // S3 not configured — remove from queue (can't delete what doesn't exist)
503 completed_ids.push(row.id);
504 }
505 }
506
507 if !completed_ids.is_empty() {
508 if let Err(e) = db::pending_s3_deletions::remove_completed(&state.db, &completed_ids).await {
509 tracing::error!(error = ?e, "failed to dequeue completed S3 deletions");
510 } else {
511 tracing::info!(completed = completed_ids.len(), total = stale.len(), "retried pending S3 deletions");
512 }
513 let _ = db::scheduler_jobs::record_job_run(&state.db, "s3_deletion_retry", completed_ids.len() as i64).await;
514 }
515 }
516
517 #[cfg(test)]
518 mod tests {
519 use super::*;
520
521 #[tokio::test]
522 async fn cleanup_git_repos_removes_directory() {
523 let tmp = tempfile::tempdir().unwrap();
524 let git_root = tmp.path();
525 let user_dir = git_root.join("testuser");
526 std::fs::create_dir_all(user_dir.join("repo.git")).unwrap();
527 std::fs::write(user_dir.join("repo.git/HEAD"), "ref: refs/heads/main\n").unwrap();
528
529 let user_id = db::UserId::nil();
530 cleanup_git_repos_on_disk(git_root.to_str().unwrap(), "testuser", user_id).await;
531
532 assert!(!user_dir.exists(), "user git directory should be deleted");
533 }
534
535 #[tokio::test]
536 async fn cleanup_git_repos_noop_if_missing() {
537 let tmp = tempfile::tempdir().unwrap();
538 let user_id = db::UserId::nil();
539 cleanup_git_repos_on_disk(tmp.path().to_str().unwrap(), "nonexistent", user_id).await;
540 }
541 }
542