d5939db
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use bytes::Bytes;
use futures::TryStreamExt;
use object_store::ObjectStoreExt;
use object_store::local::LocalFileSystem;
use object_store::path::Path;
use object_store::{GetResult, ObjectMeta, ObjectStore, PutResult};
use url::Url;
use crate::config::StorageSettings;
use crate::repo::{StorageError, StorageRepository};
#[derive(Debug)]
pub struct LocalStorage {
store: Arc<LocalFileSystem>,
}
impl LocalStorage {
pub fn from_config(settings: &StorageSettings) -> Result<Self, StorageError> {
let root_dir = settings.get_local().get_root_dir();
std::fs::create_dir_all(root_dir).map_err(|e| {
StorageError::Config(format!(
"failed to create local storage directory {}: {e}",
root_dir.display()
))
})?;
let store = LocalFileSystem::new_with_prefix(root_dir).map_err(|e| {
StorageError::Config(format!(
"failed to initialize local filesystem at {}: {e}",
root_dir.display()
))
})?;
Ok(Self {
store: Arc::new(store),
})
}
}
#[async_trait]
impl StorageRepository for LocalStorage {
async fn put(&self, path: &Path, data: Bytes) -> Result<PutResult, StorageError> {
Ok(self.store.put(path, data.into()).await?)
}
async fn get(&self, path: &Path) -> Result<GetResult, StorageError> {
Ok(self.store.get(path).await?)
}
async fn delete(&self, path: &Path) -> Result<(), StorageError> {
Ok(self.store.delete(path).await?)
}
async fn head(&self, path: &Path) -> Result<ObjectMeta, StorageError> {
Ok(self.store.head(path).await?)
}
async fn exists(&self, path: &Path) -> Result<bool, StorageError> {
Ok(self.store.head(path).await.is_ok())
}
async fn list(&self, prefix: Option<&Path>) -> Result<Vec<ObjectMeta>, StorageError> {
let stream = self.store.list(prefix);
Ok(stream.try_collect::<Vec<_>>().await?)
}
async fn copy(&self, from: &Path, to: &Path) -> Result<(), StorageError> {
Ok(self.store.copy(from, to).await?)
}
async fn presigned_upload(
&self,
_path: &Path,
_expires_in: Duration,
) -> Result<Url, StorageError> {
Err(StorageError::Unsupported(
"presigned upload URLs are not supported by the local backend".to_string(),
))
}
async fn presigned_download(
&self,
_path: &Path,
_expires_in: Duration,
) -> Result<Url, StorageError> {
Err(StorageError::Unsupported(
"presigned download URLs are not supported by the local backend".to_string(),
))
}
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use bytes::Bytes;
use object_store::path::Path;
use tempfile::tempdir;
use crate::config::{LocalStorageSettings, StorageSettings};
use crate::repo::{StorageError, from_config};
fn local_settings(dir: &std::path::Path) -> StorageSettings {
let mut local = LocalStorageSettings::default();
local.set_root_dir(dir.to_path_buf());
let mut settings = StorageSettings::default();
settings.set_local(local);
settings
}
#[tokio::test]
async fn local_roundtrip() {
let dir = tempdir().unwrap();
let settings = local_settings(dir.path());
let repo = from_config(&settings).unwrap();
let path = Path::from("test/file.txt");
let payload = Bytes::from_static(b"hello world");
repo.put(&path, payload.clone()).await.unwrap();
assert!(repo.exists(&path).await.unwrap());
let meta = repo.head(&path).await.unwrap();
assert_eq!(meta.size, payload.len() as u64);
let got = repo.get(&path).await.unwrap().bytes().await.unwrap();
assert_eq!(got, payload);
let dest = Path::from("test/copy.txt");
repo.copy(&path, &dest).await.unwrap();
assert!(repo.exists(&dest).await.unwrap());
let listed = repo.list(Some(&Path::from("test"))).await.unwrap();
assert_eq!(listed.len(), 2);
let all = repo.list(None).await.unwrap();
assert_eq!(all.len(), 2);
repo.delete(&path).await.unwrap();
assert!(!repo.exists(&path).await.unwrap());
dir.close().unwrap();
}
#[tokio::test]
async fn local_presign_unsupported() {
let dir = tempdir().unwrap();
let settings = local_settings(dir.path());
let repo = from_config(&settings).unwrap();
let path = Path::from("test/file.txt");
assert!(matches!(
repo.presigned_upload(&path, Duration::from_secs(60)).await,
Err(StorageError::Unsupported(_))
));
assert!(matches!(
repo.presigned_download(&path, Duration::from_secs(60))
.await,
Err(StorageError::Unsupported(_))
));
dir.close().unwrap();
}
}