Merge pull request #4 from msalinas92/feature/delete-from-cache

Feature/delete from cache

Author: msalinas92

Cargo.lock

@@ -46,7 +46,7 @@ sysinfo = { version = "0.35.1", features = ["multithread"] }
 aws-config = { version = "1.6.3", features = ["behavior-version-latest"] }
 aws-sdk-s3 = "1.90.0"
 aws-types = "1.3.7"
-azure_storage = { version = "0.21.0", default-features = false, features = ["enable_reqwest_rustls"] }
+azure_storage = { version = "0.21.0", default-features = false, features = ["enable_reqwest_rustls", "hmac_rust"] }
 azure_storage_blobs = { version = "0.21.0", default-features = false }
 hex = "0.4.3"
 bincode = "2.0.1"
@@ -55,6 +55,8 @@ regex = "1.11.1"
 metrics = "0.24.2"
 metrics-macros = "0.7.1"
 metrics-exporter-prometheus = "0.17.0"
+futures = "0.3"
+
 
 
 [dev-dependencies]

Cargo.toml

@@ -245,6 +245,35 @@ CacheBolt exposes Prometheus-compatible metrics at the `/metrics` endpoint on po
 - `cachebolt_fallback_miss_total`  
   Count of failover attempts that missed both memory and persistent storage.
 
+---
+## 🧹 Cache Invalidation
+
+You can clear the entire cache (both in-memory and persistent storage) using the `/cache?backend=true` endpoint. This is useful when deploying major updates or invalidating stale content globally.
+
+- When `backend=true`, CacheBolt will delete all cache entries stored in:
+  - 🟢 Amazon S3
+  - 🔵 Google Cloud Storage
+  - 🔶 Azure Blob Storage
+  - 💽 Local Filesystem
+
+### ❌ Pattern-based deletion limitations
+
+If you attempt to use `pattern=...` with `backend=true`, CacheBolt will return an error. Pattern-based invalidation is **only supported in-memory**, not in persistent storage.
+
+### ✅ Example: Full cache invalidation
+
+```bash
+curl -X DELETE "http://localhost:3000/cache?backend=true"
+```
+
+This will:
+
+Clear all in-memory cache
+
+Batch-delete all objects under the prefix cache/{app_id}/ from the configured storage backend
+
+On S3, it uses optimized DeleteObjects requests (up to 1000 keys per request)
+
 ---
 
 ## 📄 License

README.md

@@ -0,0 +1,58 @@
+use crate::memory::memory::MEMORY_CACHE;
+use axum::{extract::Query, http::StatusCode, response::IntoResponse, Json};
+use serde::{Deserialize, Serialize};
+
+// Individual and full backend deletion
+use crate::storage::azure::delete_all_from_cache as delete_all_azure;
+use crate::storage::gcs::delete_all_from_cache as delete_all_gcs;
+use crate::storage::local::delete_all_from_cache as delete_all_local;
+use crate::storage::s3::delete_all_from_cache as delete_all_s3;
+
+#[derive(Deserialize)]
+pub struct InvalidateParams {
+    pub backend: Option<bool>,
+}
+
+#[derive(Serialize)]
+struct SuccessResponse {
+    message: String,
+}
+
+/// DELETE /cache?backend=true
+pub async fn invalidate_handler(Query(params): Query<InvalidateParams>) -> impl IntoResponse {
+    let backend_enabled = params.backend.unwrap_or(false);
+
+    // 🧠 Clear memory cache
+    let mut memory = MEMORY_CACHE.write().await;
+    let count = memory.len();
+    memory.clear();
+    tracing::info!("🧨 Cleared all {count} entries from in-memory cache");
+
+    // ☁️ Optionally clear all backends
+    if backend_enabled {
+        let futures = vec![
+            tokio::spawn(async { delete_all_azure().await }),
+            tokio::spawn(async { delete_all_gcs().await }),
+            tokio::spawn(async { delete_all_s3().await }),
+            tokio::spawn(async { delete_all_local().await }),
+        ];
+
+        for task in futures {
+            if let Err(e) = task.await {
+                tracing::warn!("⚠️ A backend deletion task failed: {:?}", e);
+            }
+        }
+
+        tracing::info!("🧹 Requested full deletion from all persistent backends");
+    }
+
+    let body = Json(SuccessResponse {
+        message: if backend_enabled {
+            format!("Cleared in-memory cache and requested deletion from all backends")
+        } else {
+            format!("Cleared in-memory cache only")
+        },
+    });
+
+    (StatusCode::OK, body)
+}

src/admin/clean.rs

@@ -0,0 +1,15 @@
+// Copyright (C) 2025 Matías Salinas ([email protected])
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod clean;

src/admin/mod.rs

@@ -4,4 +4,5 @@ pub mod memory;
 pub mod proxy;
 pub mod rules;
 pub mod storage;
+pub mod admin;
 

src/lib.rs

@@ -23,18 +23,20 @@ mod memory;
 mod proxy;
 mod rules;
 mod storage;
+mod admin;
 
 // ----------------------
 // External dependencies
 // ----------------------
-use axum::{Router, routing::get}; // Axum: Web framework for routing and request handling
+use axum::{Router, routing::get, routing::delete}; // Axum: Web framework for routing and request handling
 use hyper::Server; // Hyper: High-performance HTTP server
 use std::{net::SocketAddr, process::exit}; // Network + system utilities
 
 use clap::Parser; // CLI argument parsing (via `--config`)
 use tracing::{error, info, warn}; // Structured logging macros
 use tracing_subscriber::EnvFilter; // Log filtering via LOG_LEVEL
 
+use crate::admin::clean::invalidate_handler;
 // ----------------------
 // Internal dependencies
 // ----------------------
@@ -190,7 +192,8 @@ async fn main() {
     // ------------------------------------------------------
     let app = Router::new()
         .route("/metrics", get(move || async move { handle.render() }))
-        .route("/*path", get(proxy::proxy_handler));
+        .route("/*path", get(proxy::proxy_handler))
+        .route("/cache", delete(invalidate_handler));
 
     // ------------------------------------------------------
     // 8. Bind the server to all interfaces on port 3000

src/main.rs

@@ -25,6 +25,8 @@ use crate::config::CONFIG;
 use serde::{Serialize, Deserialize};
 use base64::engine::general_purpose::STANDARD;
 use base64::Engine;
+use std::error::Error;
+use futures::StreamExt;
 
 /// Structure used to store a cached object in Azure Blob Storage.
 /// - `body`: base64-encoded content (response body).
@@ -180,3 +182,52 @@ pub async fn load_from_cache(key: &str) -> Option<(Bytes, Vec<(String, String)>)
         }
     }
 }
+
+/// Deletes all cached entries from Azure Blob Storage (prefix: "cache/{app_id}/").
+///
+/// # Returns
+/// - `Ok(count)` with number of blobs deleted on success.
+/// - `Err(...)` if listing or deletion fails.
+/// use futures::StreamExt;
+pub async fn delete_all_from_cache() -> Result<usize, Box<dyn Error + Send + Sync>> {
+    let client = AZURE_CLIENT
+        .get()
+        .ok_or("Azure client not initialized")?;
+
+    let config = CONFIG
+        .get()
+        .ok_or("CONFIG not initialized")?;
+
+    let container = &config.azure_container;
+    let container_client = client.container_client(container.clone());
+
+    // List all blobs, no prefix
+    let mut stream = container_client
+        .list_blobs()
+        .into_stream();
+
+    let mut deleted = 0;
+
+    while let Some(result) = stream.next().await {
+        let result = result?;
+        let blobs = result.blobs.blobs();
+
+        for blob in blobs {
+            let blob_name = blob.name.clone();
+            let blob_client = container_client.blob_client(blob_name.clone());
+
+            match blob_client.delete().into_future().await {
+                Ok(_) => {
+                    deleted += 1;
+                    info!("🗑️ Deleted blob '{}' from container '{}'", blob_name, container);
+                }
+                Err(e) => {
+                    warn!("⚠️ Failed to delete blob '{}': {}", blob_name, e);
+                }
+            }
+        }
+    }
+
+    info!("✅ Azure: Deleted {deleted} blobs from container '{}'", container);
+    Ok(deleted)
+}

src/storage/azure.rs

@@ -17,13 +17,11 @@
 use google_cloud_storage::{
     client::Client,
     http::objects::{
-        download::Range,
-        get::GetObjectRequest,
-        upload::{Media, UploadObjectRequest, UploadType},
+        delete::DeleteObjectRequest, download::Range, get::GetObjectRequest, upload::{Media, UploadObjectRequest, UploadType}
     },
 };
 use bytes::Bytes;
-use std::{borrow::Cow};
+use std::{borrow::Cow, error::Error};
 use std::sync::OnceLock;
 use flate2::write::GzEncoder;
 use flate2::read::GzDecoder;
@@ -34,6 +32,7 @@ use crate::config::CONFIG;
 use serde::{Serialize, Deserialize};
 use base64::engine::general_purpose::STANDARD;
 use base64::Engine;
+use google_cloud_storage::http::objects::list::ListObjectsRequest;
 
 /// Global singleton GCS client instance, initialized at runtime.
 pub static GCS_CLIENT: OnceLock<Client> = OnceLock::new();
@@ -185,3 +184,67 @@ pub async fn load_from_cache(key: &str) -> Option<(Bytes, Vec<(String, String)>)
         }
     }
 }
+
+/// Deletes all cached entries from GCS under `cache/{app_id}/`.
+///
+/// # Returns
+/// - `Ok(count)` with number of objects deleted on success.
+/// - `Err(...)` if listing or deletion fails.
+pub async fn delete_all_from_cache() -> Result<usize, Box<dyn Error + Send + Sync>> {
+    let client = GCS_CLIENT
+        .get()
+        .ok_or("GCS client is not initialized")?;
+
+    let config = CONFIG
+        .get()
+        .ok_or("CONFIG is not initialized")?;
+
+    let app_id = &config.app_id;
+    let bucket = &config.gcs_bucket;
+    let prefix = format!("cache/{app_id}/");
+
+    let mut page_token: Option<String> = None;
+    let mut deleted = 0;
+
+    loop {
+        let list_req = ListObjectsRequest {
+            bucket: bucket.clone(),
+            prefix: Some(prefix.clone()),
+            page_token: page_token.clone(),
+            ..Default::default()
+        };
+
+        let objects = client.list_objects(&list_req).await?;
+
+        for obj in objects.items.unwrap_or_default() {
+            let name = &obj.name;
+            let req = DeleteObjectRequest {
+                bucket: bucket.clone(),
+                object: name.clone(),
+                ..Default::default()
+            };
+
+            match client.delete_object(&req).await {
+                Ok(_) => {
+                    deleted += 1;
+                    info!("🗑️ Deleted '{name}' from bucket '{bucket}'");
+                }
+                Err(e) => {
+                    warn!("⚠️ Failed to delete '{name}' from bucket '{bucket}': {e}");
+                }
+            }
+        }
+
+        if let Some(token) = objects.next_page_token {
+            if token.is_empty() {
+                break;
+            }
+            page_token = Some(token);
+        } else {
+            break;
+        }
+    }
+
+    info!("✅ Completed deletion of {deleted} objects under prefix '{prefix}'");
+    Ok(deleted)
+}