Add job to remove expired files

1 files changed, 77 insertions, 0 deletions

diff --git a/src/jobs.rs b/src/jobs.rs
new file mode 100644
index 0000000..a01a70d
--- /dev/null
+++ b/src/jobs.rs
@@ -0,0 +1,77 @@
+use std::collections::HashSet;
+use std::path::Path;
+
+use chrono::Local;
+use tokio::fs;
+use tokio::time::{sleep, Duration};
+use tokio_rusqlite::Connection;
+
+use crate::db;
+
+pub async fn start(db_conn: Connection, files_dir: String) {
+    loop {
+        log::info!("Starting removing expired files");
+        cleanup_expired(&db_conn, &files_dir).await;
+
+        // Sleeping 1 day
+        sleep(Duration::from_secs(24 * 60 * 60)).await;
+    }
+}
+
+async fn cleanup_expired(db_conn: &Connection, files_dir: &String) {
+    let time = Local::now();
+
+    match read_dir(files_dir).await {
+        Err(msg) => log::error!("Listing files: {msg}"),
+        Ok(files) => match db::list_expire_after(db_conn, time).await {
+            Err(msg) => log::error!("Getting non expirable files: {msg}"),
+            Ok(non_expirable) => {
+                let non_expirable = HashSet::<String>::from_iter(non_expirable.iter().cloned());
+                let expired_ids = files.difference(&non_expirable);
+                let count = remove_files(files_dir, expired_ids.cloned()).await;
+                log::info!("Removed {} files", count);
+                if let Err(msg) = db::remove_expire_before(db_conn, time).await {
+                    log::error!("Removing files: {msg}")
+                }
+            }
+        },
+    }
+}
+
+async fn read_dir(files_dir: &String) -> Result<HashSet<String>, String> {
+    match fs::read_dir(files_dir).await {
+        Err(msg) => Err(msg.to_string()),
+        Ok(mut read_dir) => {
+            let mut files = HashSet::<String>::new();
+            loop {
+                let entry = read_dir.next_entry().await;
+                match entry {
+                    Ok(Some(entry)) => match entry.file_name().into_string() {
+                        Ok(filename) => {
+                            files.insert(filename.clone());
+                        }
+                        Err(_) => log::error!("Decoding filename"),
+                    },
+                    Ok(None) => break,
+                    Err(msg) => log::error!("File entry: {msg}"),
+                }
+            }
+            Ok(files)
+        }
+    }
+}
+
+async fn remove_files<I>(files_dir: &String, ids: I) -> i32
+where
+    I: Iterator<Item = String>,
+{
+    let mut count = 0;
+    for id in ids {
+        let path = Path::new(&files_dir).join(id.clone());
+        match fs::remove_file(path).await {
+            Err(msg) => log::error!("Removing file: {msg}"),
+            Ok(_) => count += 1
+        }
+    }
+    count
+}