datatrash/src/deleter.rs

use async_std::{fs, path::PathBuf, sync::Receiver, task};
use chrono::{prelude::*, Duration};
use futures::future::FutureExt;
use sqlx::postgres::PgPool;

pub(crate) async fn delete_old_files(receiver: Receiver<()>, db: PgPool, files_dir: PathBuf) {
    loop {
        wait_for_file_expiry(&receiver, &db).await;
        let now = Local::now().naive_local();
        let expired_files =
            sqlx::query!("SELECT file_id FROM files WHERE files.valid_till < $1", now)
                .fetch_all(&db)
                .await
                .unwrap();
        for expired_file in expired_files {
            let mut path = files_dir.clone();
            path.push(&expired_file.file_id);
            if path.exists().await {
                log::info!("delete file {}", expired_file.file_id);
                fs::remove_file(&path).await.expect("could not delete file");
            }
        }
        sqlx::query!("DELETE FROM files WHERE valid_till < $1", now)
            .execute(&db)
            .await
            .expect("could not delete expired files from database");
    }
}

async fn wait_for_file_expiry(receiver: &Receiver<()>, db: &PgPool) {
    let row = sqlx::query!("SELECT MIN(valid_till) as min from files")
        .fetch_one(db)
        .await
        .expect("could not fetch expiring file from database");
    let next_timeout = match row.min {
        Some(min) => min.signed_duration_since(Local::now().naive_local()),
        None => Duration::days(1),
    };
    let positive_timeout = next_timeout
        .to_std()
        .unwrap_or_else(|_| std::time::Duration::from_secs(0));
    futures::select! {
        _ = task::sleep(positive_timeout).fuse() => {}
        _ = receiver.recv().fuse() => {}
    }
}