first iteration of memory/disk/mixed cleaning tick

Cargo.toml

@@ -7,21 +7,21 @@ edition = "2021"
 
 [dependencies]
 futures = "0.3"
-tokio = { version = "1.38", features = ["full", "tracing"] }
+tokio = { version = "1.35", features = ["full", "tracing"] }
 tracing = "0.1"
 tracing-subscriber = "0.3"
 tokio-stream = "0.1"
 tokio-util = "0.7"
 axum = { version = "0.7" }
-reqwest = { version = "0.12", features = ["stream"] }
+reqwest = { version = "0.11", features = ["stream"] }
 anyhow = "1.0"
-sentry = "0.34"
+sentry = "0.32"
 dotenvy = "0.15"
 axum-macros = "0.4"
 magic-crypt = "3.1"
 async-trait = "0.1"
 serde = "1.0"
 serde_yaml = "0.9"
-base64 = "0.22"
+base64 = "0.21"
 regex = "1.10"
 serde_regex = "1.1"

Dockerfile

@@ -1,19 +0,0 @@
-FROM rust:1.79 AS builder
-
-WORKDIR /app
-
-COPY src/ src/
-COPY Cargo.lock .
-COPY Cargo.toml .
-
-RUN cargo build --release
-
-
-FROM rust:1.79 AS runner
-
-WORKDIR /app
-COPY --from=builder /app/target/release/imgproxy-rs .
-
-RUN chmod +x imgproxy-rs
-
-CMD [ "./imgproxy-rs" ]

src/config.rs

@@ -6,20 +6,20 @@ use serde::Deserialize;
 #[derive(Deserialize, Clone)]
 pub struct MemoryStorageConfig {
     pub max_size: usize,
-    pub ttl: usize,
+    pub ttl: u64,
 }
 
 #[derive(Deserialize, Clone)]
 pub struct DiskStorageConfig {
     pub path: PathBuf,
-    pub ttl: usize,
+    pub ttl: u64,
     pub max_size: usize,
 }
 
 #[derive(Deserialize, Clone)]
 pub struct MixedStorageConfig {
     pub path: PathBuf,
-    pub ttl: usize,
+    pub ttl: u64,
     pub max_size: usize,
 }
 
@@ -52,6 +52,7 @@ pub enum StorageStrategyConfig {
 #[derive(Deserialize)]
 pub struct StorageConfig {
     pub strategy: StorageStrategyConfig,
+    pub name: Option<String>,
 
     #[serde(with = "serde_regex")]
     pub regex: regex::Regex,

src/crypt.rs

@@ -1,13 +1,13 @@
 use std::io::Write;
 
 use base64::engine::general_purpose;
-use magic_crypt::{MagicCrypt256, MagicCryptTrait};
+use magic_crypt::{new_magic_crypt, MagicCryptTrait};
 
 pub fn decryt<I>(value: I, secret_key: String) -> anyhow::Result<String>
 where
     I: AsRef<str>,
 {
-    let mc = MagicCrypt256::new(&secret_key, None::<&str>);
+    let mc = new_magic_crypt!(&secret_key, 256);
     Ok(mc.decrypt_base64_to_string(value)?)
 }
 

src/main.rs

@@ -1,4 +1,5 @@
 use axum::{routing::get, Router};
+use tokio::sync::Mutex;
 
 use std::sync::Arc;
 
@@ -22,19 +23,38 @@ async fn main() -> anyhow::Result<()> {
     storage_pool.init().await?;
 
     let app_state = AppState::new(storage_pool, secret_key);
+    let routes = match app_state.secret_key {
+        Some(_) => get(routes::handle_secure),
+        None => get(routes::handle_unsecure),
+    };
+
+    let app_state_th = Arc::new(Mutex::new(app_state));
+
+    tokio::spawn(launch_storages_tick(app_state_th.clone()));
 
     let app = Router::new()
-        .route(
+        .route("/*src", routes)
-            "/*src",
+        .with_state(app_state_th);
-            match app_state.secret_key {
-                Some(_) => get(routes::handle_secure),
-                None => get(routes::handle_unsecure),
-            },
-        )
-        .with_state(Arc::new(app_state));
 
-    let listener = tokio::net::TcpListener::bind("0.0.0.0:3100").await?;
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
     axum::serve(listener, app).await?;
 
     Ok(())
 }
+
+async fn launch_storages_tick(app_state: Arc<Mutex<AppState>>) {
+    loop {
+        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+        let mut app_state_locked = app_state.lock().await;
+        let results = app_state_locked.storage_pool.tick().await;
+        let errs = results
+            .into_iter()
+            .filter(anyhow::Result::is_err)
+            .map(anyhow::Result::unwrap_err);
+
+        for err in errs {
+            println!("error while tick {}", err);
+        }
+    }
+}

src/routes.rs

@@ -6,7 +6,6 @@ use axum::{
     response::IntoResponse,
 };
 use futures::TryStreamExt;
-use reqwest::header;
 use tokio::sync::Mutex;
 use tokio_stream::Stream;
 use tokio_util::bytes::Bytes;
@@ -15,36 +14,41 @@ use crate::{crypt, state::AppState};
 
 pub async fn handle_secure(
     Path(data): Path<String>,
-    state: State<Arc<AppState>>,
+    state: State<Arc<Mutex<AppState>>>,
 ) -> Result<impl IntoResponse, MyAnyhow> {
-    let decrypt = crypt::decryt(data, state.secret_key.clone().unwrap())?;
+    let app_state = state.lock().await;
+    let decrypt = crypt::decryt(data, app_state.secret_key.clone().unwrap())?;
+    std::mem::drop(app_state);
 
     let bytes = handle(decrypt, state).await?;
-    Ok(([(header::CACHE_CONTROL, "public, max-age=31919000")], axum::body::Body::from_stream(bytes)))
+    Ok(axum::body::Body::from_stream(bytes))
 }
 
 pub async fn handle_unsecure(
     Path(src): Path<String>,
-    state: State<Arc<AppState>>,
+    state: State<Arc<Mutex<AppState>>>,
 ) -> Result<impl IntoResponse, MyAnyhow> {
     let bytes = handle(src, state).await?;
 
-    Ok(([(header::CACHE_CONTROL, "public, max-age=31919000")], axum::body::Body::from_stream(bytes)))
+    Ok(axum::body::Body::from_stream(bytes))
 }
 
 async fn handle(
     src: String,
-    state: State<Arc<AppState>>,
+    state: State<Arc<Mutex<AppState>>>,
 ) -> anyhow::Result<impl Stream<Item = Result<Bytes, anyhow::Error>> + Send> {
-    let mut storage_pool = state.storage_pool.lock().await;
+    let mut app_state = state.lock().await;
 
-    match storage_pool.retrieve(src.clone()).await {
+    match app_state.storage_pool.retrieve(src.clone()).await {
         Some(stream) => Ok(stream),
         None => {
             let save_stream = fetch(src.clone()).await?;
-            storage_pool.save(src.clone(), save_stream).await?;
+            app_state
+                .storage_pool
+                .save(src.clone(), save_stream)
+                .await?;
 
-            Ok(storage_pool.retrieve(src).await.unwrap())
+            Ok(app_state.storage_pool.retrieve(src).await.unwrap())
         }
     }
 }

src/state.rs

@@ -1,17 +1,15 @@
-use tokio::sync::Mutex;
-
 use crate::storages::StoragePool;
 
 pub struct AppState {
     pub secret_key: Option<String>,
-    pub storage_pool: Mutex<StoragePool>,
+    pub storage_pool: StoragePool,
 }
 
 impl AppState {
     pub fn new(storage_pool: StoragePool, secret_key: Option<String>) -> Self {
         Self {
             secret_key,
-            storage_pool: Mutex::new(storage_pool),
+            storage_pool,
         }
     }
 }

src/storages/disk.rs

@@ -1,4 +1,4 @@
-use std::{path::PathBuf, pin::Pin};
+use std::{collections::HashMap, path::PathBuf, pin::Pin, time::SystemTime};
 
 use crate::config::DiskStorageConfig;
 use async_trait::async_trait;
@@ -9,13 +9,31 @@ use tokio_util::{bytes::Bytes, io::ReaderStream};
 
 use super::Storage;
 
+struct DiskStorageItem {
+    path: PathBuf,
+    created_at: SystemTime,
+}
+
+impl DiskStorageItem {
+    fn new(path: PathBuf, created_at: SystemTime) -> Self {
+        Self {
+            path: path,
+            created_at: created_at,
+        }
+    }
+}
+
 pub struct DiskStorage {
     config: DiskStorageConfig,
+    items: HashMap<String, DiskStorageItem>,
 }
 
 impl DiskStorage {
     pub fn new(config: DiskStorageConfig) -> Self {
-        Self { config }
+        Self {
+            config,
+            items: HashMap::new(),
+        }
     }
 
     pub async fn retrieve_all(
@@ -26,16 +44,10 @@ impl DiskStorage {
             Pin<Box<dyn Stream<Item = Result<Bytes, anyhow::Error>> + Send>>,
         )>,
     > {
-        let mut saved_files = tokio::fs::read_dir(self.config.path.clone()).await?;
         let mut files = Vec::new();
 
-        while let Some(file) = saved_files.next_entry().await? {
+        for key in self.items.keys() {
-            let key = file.file_name().to_string_lossy().to_string();
+            files.push((key.clone(), self.retrieve(key.clone()).await.unwrap()));
-            println!("{}", key);
-
-            let stream = self.retrieve(key.clone()).await.unwrap();
-
-            files.push((key, stream));
         }
 
         Ok(files)
@@ -47,10 +59,34 @@ impl Storage for DiskStorage {
     async fn init(&mut self) -> anyhow::Result<()> {
         tokio::fs::create_dir_all(self.config.path.clone()).await?;
 
+        let mut saved_files = tokio::fs::read_dir(self.config.path.clone()).await?;
+        while let Some(file) = saved_files.next_entry().await? {
+            let key = file.file_name().to_string_lossy().to_string();
+            let path = file.path();
+            let metadata = file.metadata().await?;
+
+            self.items
+                .insert(key, DiskStorageItem::new(path, metadata.created()?));
+        }
+
         Ok(())
     }
 
     async fn tick(&mut self) -> anyhow::Result<()> {
+        let paths_deleted = self
+            .items
+            .iter()
+            .filter(|(_, item)| item.created_at.elapsed().unwrap().as_secs() >= self.config.ttl)
+            .map(|(_, item)| item.path.clone())
+            .collect::<Vec<PathBuf>>();
+
+        self.items
+            .retain(|_, item| item.created_at.elapsed().unwrap().as_secs() < self.config.ttl);
+
+        for path in paths_deleted {
+            tokio::fs::remove_file(path).await?;
+        }
+
         Ok(())
     }
 

src/storages/memory.rs

@@ -1,8 +1,11 @@
+use std::{
+    collections::HashMap,
+    pin::Pin,
+    time::{Duration, SystemTime},
+};
+
 use anyhow::Ok;
 use async_trait::async_trait;
-use std::time::{Duration, SystemTime};
-use std::{collections::HashMap, pin::Pin};
-use tokio::time::Instant;
 use tokio_stream::{Stream, StreamExt};
 use tokio_util::bytes::Bytes;
 
@@ -12,7 +15,16 @@ use super::Storage;
 
 struct MemoryStorageItem {
     data: Vec<u8>,
-    added_at: Instant,
+    created_at: SystemTime,
+}
+
+impl MemoryStorageItem {
+    fn new() -> Self {
+        Self {
+            data: Vec::new(),
+            created_at: SystemTime::now(),
+        }
+    }
 }
 
 pub struct MemoryStorage {
@@ -35,23 +47,23 @@ impl Storage for MemoryStorage {
         Ok(())
     }
 
-    async fn eligible(&self, _src: String) -> bool {
-        true
-    }
-
     async fn tick(&mut self) -> anyhow::Result<()> {
-        let ttl = self.config.ttl;
         self.items
-            .retain(|_, elt| elt.added_at.elapsed().as_secs() < ttl as u64);
+            .retain(|_, item| item.created_at.elapsed().unwrap().as_secs() < self.config.ttl);
 
         Ok(())
     }
 
+    async fn eligible(&self, _src: String) -> bool {
+        true
+    }
+
     async fn delete(&mut self, key: String) -> anyhow::Result<()> {
-        self.items
+        if self.items.get(&key).is_some() {
-            .remove(&key)
+            self.items.remove(&key);
-            .ok_or_else(|| anyhow::anyhow!(""))
+        }
-            .map(|_| ())
+
+        Ok(())
     }
 
     async fn retrieve(
@@ -75,17 +87,9 @@ impl Storage for MemoryStorage {
         mut stream: Pin<Box<dyn Stream<Item = Result<Bytes, anyhow::Error>> + Send>>,
     ) -> anyhow::Result<()> {
         if !self.items.contains_key(&key) {
-            self.items.insert(
+            self.items.insert(key.clone(), MemoryStorageItem::new());
-                key.clone(),
-                MemoryStorageItem {
-                    data: Vec::new(),
-                    added_at: Instant::now(),
-                },
-            );
         }
 
-        println!("save {}", key);
-
         while let Some(chunk) = stream.next().await {
             self.items
                 .get_mut(&key)

src/storages/mixed.rs

@@ -16,7 +16,7 @@ pub struct MixedStorage {
 impl MixedStorage {
     pub fn new(config: MixedStorageConfig) -> Self {
         Self {
-            disk: DiskStorage::new(config.clone().into()),
+            disk: DiskStorage::new(config.to_owned().into()),
             memory: MemoryStorage::new(config.into()),
         }
     }
@@ -37,6 +37,9 @@ impl Storage for MixedStorage {
     }
 
     async fn tick(&mut self) -> anyhow::Result<()> {
+        self.disk.tick().await?;
+        self.memory.tick().await?;
+
         Ok(())
     }
 

src/storages/mod.rs

@@ -19,8 +19,8 @@ use self::{disk::DiskStorage, memory::MemoryStorage, mixed::MixedStorage};
 pub trait Storage {
     async fn init(&mut self) -> anyhow::Result<()>;
     async fn eligible(&self, src: String) -> bool;
-    async fn delete(&mut self, key: String) -> anyhow::Result<()>;
     async fn tick(&mut self) -> anyhow::Result<()>;
+    async fn delete(&mut self, key: String) -> anyhow::Result<()>;
     async fn retrieve(
         &self,
         key: String,
@@ -45,13 +45,22 @@ impl StoragePool {
 
     pub async fn init(&mut self) -> anyhow::Result<()> {
         for item in &mut self.storages {
-            println!("init storage");
             item.init().await?;
         }
 
         Ok(())
     }
 
+    pub async fn tick(&mut self) -> Vec<anyhow::Result<()>> {
+        let mut results = vec![];
+
+        for item in &mut self.storages {
+            results.push(item.tick().await);
+        }
+
+        results
+    }
+
     pub async fn retrieve(
         &self,
         src: String,