juiceback/jobs/
cleanup.rs

1//! Background job that runs on a timer to clean up expired files.
2
3use std::sync::Arc;
4use std::time::Duration;
5
6use crate::db;
7use crate::error::AppError;
8use crate::state::AppState;
9use chrono::Utc;
10
11/// Run the cleanup loop forever, sleeping the configured interval between runs.
12pub async fn run_cleanup_loop(state: Arc<AppState>) {
13    let interval = Duration::from_secs(state.config.cleanup_interval_minutes * 60);
14    let mut ticker = tokio::time::interval(interval);
15    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
16
17    loop {
18        ticker.tick().await;
19        run_cleanup_once(&state).await;
20    }
21}
22
23async fn run_cleanup_once(state: &Arc<AppState>) {
24    tracing::info!("cleanup: starting");
25
26    let state2 = Arc::clone(state);
27    let expired = match tokio::task::spawn_blocking(move || -> Result<Vec<db::FileRecord>, AppError> {
28        let db = state2.db.get().map_err(|e| AppError::DbPoolError(e.to_string()))?;
29        db::list_expired(&db).map_err(AppError::DatabaseError)
30    })
31    .await
32    {
33        Ok(Ok(records)) => records,
34        Ok(Err(e)) => {
35            tracing::error!("cleanup: db query failed: {:?}", e);
36            return;
37        }
38        Err(e) => {
39            tracing::error!("cleanup: spawn_blocking panicked: {}", e);
40            return;
41        }
42    };
43
44    if expired.is_empty() {
45        tracing::info!("cleanup: nothing to clean");
46        return;
47    }
48
49    tracing::info!("cleanup: found {} expired files", expired.len());
50
51    let ids_to_purge: Vec<String> = expired.iter().map(|r| r.id.clone()).collect();
52
53    // Purge expired DB records
54    let state3 = Arc::clone(state);
55    let ids_for_purge = ids_to_purge.clone();
56    let purge_result = tokio::task::spawn_blocking(move || -> Result<usize, AppError> {
57        let db = state3.db.get().map_err(|e| AppError::DbPoolError(e.to_string()))?;
58        db::delete_files_by_ids(&db, &ids_for_purge).map_err(AppError::DatabaseError)
59    })
60    .await
61    .map_err(|e| tracing::error!("cleanup: purge task panicked: {}", e));
62
63    match purge_result {
64        Ok(Ok(count)) => tracing::info!("cleanup: purged {} rows from db", count),
65        Ok(Err(e)) => tracing::error!("cleanup: failed to purge db rows: {:?}", e),
66        Err(_) => {}
67    }
68
69    // Notify juicehost to delete expired files
70    for record in &expired {
71        let host_opt = if record.storage_host.is_empty() { None } else { Some(record.storage_host.as_str()) };
72        crate::juicehost::delete_file_on_juicehost(state, &record.id, host_opt).await;
73    }
74
75    // TUS cleanup: remove abandoned uploads older than 1 hour
76    let now = Utc::now().timestamp();
77    let stale_ids: Vec<String> = state
78        .tus
79        .iter()
80        .filter(|e| now - e.created_at > 3600)
81        .map(|e| e.id.clone())
82        .collect();
83    for id in &stale_ids {
84        if let Some((_, upload)) = state.tus.remove(id) {
85            tracing::info!("tus cleanup: removing abandoned upload id={}", id);
86            let _ = tokio::fs::remove_file(&upload.storage_path).await;
87        }
88    }
89    if !stale_ids.is_empty() {
90        tracing::info!("tus cleanup: removed {} abandoned uploads", stale_ids.len());
91    }
92}
juiceback/jobs/cleanup.rs

juiceback/jobs/
cleanup.rs
