use serde::{de::DeserializeOwned, Serialize};
use std::collections::HashMap;
use std::fs::{self, File};
use std::io::{BufReader, Write};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use zeroize::Zeroizing;
use crate::core::errors::{StorageError, StorageResult};
use crate::core::storage_provider::{
AsyncBlobStore, AsyncKeyValueStore, AsyncListStream,
BlobStore, DataValue, KeyValueStore, QueryStore, RateLimiterStore, SqlRow, StorageItemStream,
};
use foundation_core::valtron::Stream;
pub struct JsonFileStorage {
file_path: PathBuf,
data: Arc<Mutex<HashMap<String, Zeroizing<Vec<u8>>>>>,
}
impl JsonFileStorage {
pub fn new<P: AsRef<Path>>(file_path: P) -> StorageResult<Self> {
let file_path = file_path.as_ref().to_path_buf();
let data = if file_path.exists() {
Self::load_from_file(&file_path)?
} else {
HashMap::new()
};
Ok(Self {
file_path,
data: Arc::new(Mutex::new(data)),
})
}
fn load_from_file(file_path: &Path) -> StorageResult<HashMap<String, Zeroizing<Vec<u8>>>> {
let file = File::open(file_path).map_err(StorageError::Io)?;
let reader = BufReader::new(file);
let raw_data: HashMap<String, Vec<u8>> =
serde_json::from_reader(reader).map_err(StorageError::Json)?;
Ok(raw_data
.into_iter()
.map(|(k, v)| (k, Zeroizing::new(v)))
.collect())
}
fn flush_to_disk(&self) -> StorageResult<()> {
let data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
let raw_data: HashMap<String, Vec<u8>> = data
.iter()
.map(|(k, v)| (k.clone(), (**v).clone()))
.collect();
let json_bytes = serde_json::to_vec(&raw_data).map_err(StorageError::Json)?;
let temp_path = self.file_path.with_extension("json.tmp");
let mut temp_file = File::create(&temp_path).map_err(StorageError::Io)?;
temp_file.write_all(&json_bytes).map_err(StorageError::Io)?;
temp_file.sync_all().map_err(StorageError::Io)?;
fs::rename(&temp_path, &self.file_path).map_err(StorageError::Io)?;
Ok(())
}
}
impl KeyValueStore for JsonFileStorage {
fn get<V: DeserializeOwned + Send + 'static>(&self, key: &str) -> StorageResult<Option<V>> {
let data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
match data.get(key) {
Some(bytes) => {
let value: V = serde_json::from_slice(bytes)
.map_err(|e| StorageError::Serialization(e.to_string()))?;
Ok(Some(value))
}
None => Ok(None),
}
}
fn set<V: Serialize>(&self, key: &str, value: V) -> StorageResult<()> {
let bytes =
serde_json::to_vec(&value).map_err(|e| StorageError::Serialization(e.to_string()))?;
let mut data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
data.insert(key.to_string(), Zeroizing::new(bytes));
drop(data);
self.flush_to_disk()
}
fn delete(&self, key: &str) -> StorageResult<()> {
let mut data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
data.remove(key);
drop(data);
self.flush_to_disk()
}
fn exists(&self, key: &str) -> StorageResult<bool> {
let data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
Ok(data.contains_key(key))
}
fn list_keys(&self, prefix: Option<&str>) -> StorageResult<StorageItemStream<'_, String>> {
let data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
let keys: Vec<String> = data
.keys()
.filter(|k| prefix.is_none_or(|p| k.starts_with(p)))
.cloned()
.collect();
Ok(Box::new(keys.into_iter().map(|k| Stream::Next(Ok(k)))))
}
}
impl QueryStore for JsonFileStorage {
fn query(
&self,
_sql: &str,
_params: &[DataValue],
) -> StorageResult<StorageItemStream<'_, SqlRow>> {
Err(StorageError::Generic(
"QueryStore not supported for JsonFileStorage".to_string(),
))
}
fn execute(
&self,
_sql: &str,
_params: &[DataValue],
) -> StorageResult<u64> {
Err(StorageError::Generic(
"QueryStore not supported for JsonFileStorage".to_string(),
))
}
fn execute_batch(&self, _sql: &str) -> StorageResult<()> {
Err(StorageError::Generic(
"QueryStore not supported for JsonFileStorage".to_string(),
))
}
}
impl RateLimiterStore for JsonFileStorage {
fn check_rate_limit(
&self,
_key: &str,
_max_count: u32,
_window_seconds: u64,
) -> StorageResult<bool> {
Err(StorageError::Generic(
"RateLimiterStore not supported for JsonFileStorage".to_string(),
))
}
fn record_rate_limit(&self, _key: &str) -> StorageResult<u32> {
Err(StorageError::Generic(
"RateLimiterStore not supported for JsonFileStorage".to_string(),
))
}
fn reset_rate_limit(&self, _key: &str) -> StorageResult<()> {
Err(StorageError::Generic(
"RateLimiterStore not supported for JsonFileStorage".to_string(),
))
}
}
impl BlobStore for JsonFileStorage {
fn put_blob(&self, key: &str, data: &[u8]) -> StorageResult<()> {
let mut storage = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
storage.insert(key.to_string(), Zeroizing::new(data.to_vec()));
drop(storage); self.flush_to_disk()
}
fn get_blob(&self, key: &str) -> StorageResult<Option<Vec<u8>>> {
let data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
Ok(data.get(key).cloned().map(|z| z.to_vec()))
}
fn delete_blob(&self, key: &str) -> StorageResult<()> {
let mut data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
data.remove(key);
drop(data); self.flush_to_disk()
}
fn blob_exists(&self, key: &str) -> StorageResult<bool> {
let data = self
.data
.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
Ok(data.contains_key(key))
}
}
#[async_trait::async_trait]
impl AsyncKeyValueStore for JsonFileStorage {
async fn get_async<V: DeserializeOwned + Send + 'static>(&self, key: &str) -> StorageResult<Option<V>> {
let data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
let result = match data.get(key) {
Some(bytes) => {
let value: V = serde_json::from_slice(bytes)
.map_err(|e| StorageError::Serialization(e.to_string()))?;
Some(value)
}
None => None,
};
Ok(result)
}
async fn set_async<V: Serialize + Send + 'static>(&self, key: &str, value: V) -> StorageResult<()> {
let bytes =
serde_json::to_vec(&value).map_err(|e| StorageError::Serialization(e.to_string()))?;
let mut data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
data.insert(key.to_string(), Zeroizing::new(bytes));
drop(data);
self.flush_to_disk()
}
async fn delete_async(&self, key: &str) -> StorageResult<()> {
let mut data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
data.remove(key);
drop(data);
self.flush_to_disk()
}
async fn exists_async(&self, key: &str) -> StorageResult<bool> {
let data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
Ok(data.contains_key(key))
}
async fn list_keys_async(&self, prefix: Option<&str>) -> StorageResult<AsyncListStream> {
let data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
let keys: Vec<String> = data.keys()
.filter(|k| prefix.is_none_or(|p| k.starts_with(p)))
.cloned()
.collect();
Ok(AsyncListStream::new(futures_lite::stream::iter(
keys.into_iter().map(Ok)
)))
}
}
#[async_trait::async_trait]
impl AsyncBlobStore for JsonFileStorage {
async fn put_blob_async(&self, key: &str, data: &[u8]) -> StorageResult<()> {
let mut storage = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
storage.insert(key.to_string(), Zeroizing::new(data.to_vec()));
drop(storage);
self.flush_to_disk()
}
async fn get_blob_async(&self, key: &str) -> StorageResult<Option<Vec<u8>>> {
let data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
Ok(data.get(key).cloned().map(|z| z.to_vec()))
}
async fn delete_blob_async(&self, key: &str) -> StorageResult<()> {
let mut data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
data.remove(key);
drop(data);
self.flush_to_disk()
}
async fn blob_exists_async(&self, key: &str) -> StorageResult<bool> {
let data = self.data.lock()
.map_err(|e| StorageError::Backend(format!("Mutex poisoned: {e}")))?;
Ok(data.contains_key(key))
}
}