datatrash/src/deleter.rs

52 lines
2.0 KiB
Rust
Raw Normal View History

use async_std::{channel::Receiver, fs, path::PathBuf, task};
2020-07-09 17:27:24 +00:00
use chrono::{prelude::*, Duration};
use futures::{TryStreamExt, future::FutureExt};
use sqlx::{postgres::PgPool, Row};
2020-07-09 17:27:24 +00:00
2020-07-11 21:27:15 +00:00
pub(crate) async fn delete_old_files(receiver: Receiver<()>, db: PgPool, files_dir: PathBuf) {
2020-07-09 17:27:24 +00:00
loop {
wait_for_file_expiry(&receiver, &db).await;
2020-07-13 13:29:40 +00:00
let now = Local::now().naive_local();
let mut rows = sqlx::query("SELECT file_id FROM files WHERE files.valid_till < $1")
2020-07-13 13:29:40 +00:00
.bind(now)
.fetch(&db);
while let Some(row) = rows.try_next().await.expect("could not load expired files") {
let file_id: String = row.try_get("file_id").expect("we selected this column");
2020-07-11 21:27:15 +00:00
let mut path = files_dir.clone();
path.push(&file_id);
2020-07-09 17:27:24 +00:00
if path.exists().await {
log::info!("delete file {}", file_id);
2020-07-09 17:27:24 +00:00
fs::remove_file(&path).await.expect("could not delete file");
}
}
2020-07-13 13:29:40 +00:00
sqlx::query("DELETE FROM files WHERE valid_till < $1")
2020-07-13 13:29:40 +00:00
.bind(now)
2020-07-09 17:27:24 +00:00
.execute(&db)
.await
.expect("could not delete expired files from database");
}
}
async fn wait_for_file_expiry(receiver: &Receiver<()>, db: &PgPool) {
let mut rows = sqlx::query("SELECT MIN(valid_till) as min from files").fetch(db);
let row = rows
.try_next()
2020-07-09 17:27:24 +00:00
.await
.expect("could not fetch expiring files from database")
.expect("postgres min did not return any row");
let valid_till: Option<NaiveDateTime> = row.get("min");
let next_timeout = match valid_till {
Some(valid_till) => valid_till.signed_duration_since(Local::now().naive_local()),
2020-07-09 17:27:24 +00:00
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() => {}
}
}