loco-rs 1.1.0

//! # Storage Module
//!
//! This module defines a generic storage abstraction represented by the
//! [`Storage`] struct. It provides methods for performing common storage
//! operations such as upload, download, delete, rename, copy, exists, list,
//! and stat.
//!
//! ## Storage Strategy
//!
//! The [`Storage`] struct is designed to work with different storage
//! strategies. A storage strategy defines the behavior of the storage
//! operations. Strategies implement the [`strategies::StorageStrategy`].
//! The selected strategy can be dynamically changed at runtime.
mod contents;
pub mod drivers;
pub mod strategies;
pub mod stream;
use std::{
    collections::BTreeMap,
    path::{Path, PathBuf},
};

use bytes::Bytes;

use self::{
    drivers::{ListEntry, StoreDriver},
    stream::BytesStream,
};

#[derive(thiserror::Error, Debug)]
#[allow(clippy::module_name_repetitions)]
#[non_exhaustive]
pub enum StorageError {
    #[error("store not found by the given key: {0}")]
    StoreNotFound(String),

    #[error(transparent)]
    Store(#[from] Box<opendal::Error>),

    #[error("Unable to read data from file {}", path.display().to_string())]
    UnableToReadBytes { path: PathBuf },

    #[error("secondaries errors")]
    Multi(BTreeMap<String, String>),

    #[error(transparent)]
    Any(#[from] Box<dyn std::error::Error + Send + Sync>),
}

pub type StorageResult<T> = std::result::Result<T, StorageError>;

impl From<opendal::Error> for StorageError {
    fn from(val: opendal::Error) -> Self {
        Self::Store(Box::new(val))
    }
}

pub struct Storage {
    pub stores: BTreeMap<String, Box<dyn StoreDriver>>,
    pub strategy: Box<dyn strategies::StorageStrategy>,
}

impl Storage {
    /// Creates a new storage instance with a single store and the default
    /// strategy.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    ///
    /// let storage = storage::Storage::single(storage::drivers::mem::new());
    /// ```
    #[must_use]
    pub fn single(store: Box<dyn StoreDriver>) -> Self {
        let default_key = "store";
        Self {
            strategy: Box::new(strategies::single::SingleStrategy::new(default_key)),
            stores: BTreeMap::from([(default_key.to_string(), store)]),
        }
    }

    /// Creates a new storage instance with the provided stores and strategy.
    #[must_use]
    pub fn new(
        stores: BTreeMap<String, Box<dyn StoreDriver>>,
        strategy: Box<dyn strategies::StorageStrategy>,
    ) -> Self {
        Self { stores, strategy }
    }

    /// Uploads content to the storage at the specified path.
    ///
    /// This method uses the selected strategy for the upload operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn upload() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     let content = "Loco!";
    ///     let result = storage.upload(path, &Bytes::from(content)).await;
    ///     assert!(result.is_ok());
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the upload operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn upload(&self, path: &Path, content: &Bytes) -> StorageResult<()> {
        self.upload_with_strategy(path, content, &*self.strategy)
            .await
    }

    /// Uploads content to the storage at the specified path using a specific
    /// strategy.
    ///
    /// This method allows specifying a custom strategy for the upload
    /// operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the upload operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn upload_with_strategy(
        &self,
        path: &Path,
        content: &Bytes,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<()> {
        strategy.upload(self, path, content).await
    }

    /// Downloads content from the storage at the specified path.
    ///
    /// This method uses the selected strategy for the download operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     let content = "Loco!";
    ///     storage.upload(path, &Bytes::from(content)).await;
    ///
    ///     let result: String = storage.download(path).await.unwrap();
    ///     assert_eq!(result, "Loco!");
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the download operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn download<T: TryFrom<contents::Contents>>(&self, path: &Path) -> StorageResult<T> {
        self.download_with_policy(path, &*self.strategy).await
    }

    /// Downloads content from the storage at the specified path using a
    /// specific strategy.
    ///
    /// This method allows specifying a custom strategy for the download
    /// operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the download operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn download_with_policy<T: TryFrom<contents::Contents>>(
        &self,
        path: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<T> {
        let res = strategy.download(self, path).await?;
        contents::Contents::from(res).try_into().map_or_else(
            |_| {
                Err(StorageError::UnableToReadBytes {
                    path: path.to_path_buf(),
                })
            },
            |content| Ok(content),
        )
    }

    /// Deletes content from the storage at the specified path.
    ///
    /// This method uses the selected strategy for the delete operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     let content = "Loco!";
    ///     storage.upload(path, &Bytes::from(content)).await;
    ///
    ///     let result = storage.delete(path).await;
    ///     assert!(result.is_ok());
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the delete operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn delete(&self, path: &Path) -> StorageResult<()> {
        self.delete_with_policy(path, &*self.strategy).await
    }

    /// Deletes content from the storage at the specified path using a specific
    /// strategy.
    ///
    /// This method allows specifying a custom strategy for the delete
    /// operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the delete operation fails or if there
    /// is an issue with the strategy configuration.    
    pub async fn delete_with_policy(
        &self,
        path: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<()> {
        strategy.delete(self, path).await
    }

    /// Renames content from one path to another in the storage.
    ///
    /// This method uses the selected strategy for the rename operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     let content = "Loco!";
    ///     storage.upload(path, &Bytes::from(content)).await;
    ///     
    ///     let new_path = Path::new("new_path.txt");
    ///     let store = storage.as_store("default").unwrap();
    ///     assert!(storage.rename(&path, &new_path).await.is_ok());
    ///     assert!(!store.exists(&path).await.unwrap());
    ///     assert!(store.exists(&new_path).await.unwrap());
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the rename operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn rename(&self, from: &Path, to: &Path) -> StorageResult<()> {
        self.rename_with_policy(from, to, &*self.strategy).await
    }

    /// Renames content from one path to another in the storage using a specific
    /// strategy.
    ///
    /// This method allows specifying a custom strategy for the rename
    /// operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the rename operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn rename_with_policy(
        &self,
        from: &Path,
        to: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<()> {
        strategy.rename(self, from, to).await
    }

    /// Copies content from one path to another in the storage.
    ///
    /// This method uses the selected strategy for the copy operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     let content = "Loco!";
    ///     storage.upload(path, &Bytes::from(content)).await;
    ///     
    ///     let new_path = Path::new("new_path.txt");
    ///     let store = storage.as_store("default").unwrap();
    ///     assert!(storage.copy(&path, &new_path).await.is_ok());
    ///     assert!(store.exists(&path).await.unwrap());
    ///     assert!(store.exists(&new_path).await.unwrap());
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the copy operation fails or if there is
    /// an issue with the strategy configuration.
    pub async fn copy(&self, from: &Path, to: &Path) -> StorageResult<()> {
        self.copy_with_policy(from, to, &*self.strategy).await
    }

    /// Copies content from one path to another in the storage using a specific
    /// strategy.
    ///
    /// This method allows specifying a custom strategy for the copy operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the copy operation fails or if there is
    /// an issue with the strategy configuration.
    pub async fn copy_with_policy(
        &self,
        from: &Path,
        to: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<()> {
        strategy.copy(self, from, to).await
    }

    /// Checks whether content exists at the specified path.
    ///
    /// This method uses the selected strategy for the exists check.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn check_exists() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     storage.upload(path, &Bytes::from("Loco!")).await.unwrap();
    ///
    ///     assert!(storage.exists(path).await.unwrap());
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the exists check fails or if there is
    /// an issue with the strategy configuration.
    pub async fn exists(&self, path: &Path) -> StorageResult<bool> {
        self.exists_with_policy(path, &*self.strategy).await
    }

    /// Checks whether content exists at the specified path using a specific
    /// strategy.
    ///
    /// This method allows specifying a custom strategy for the exists check.
    ///
    /// # Errors
    ///
    /// This method returns an error if the exists check fails or if there is
    /// an issue with the strategy configuration.
    pub async fn exists_with_policy(
        &self,
        path: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<bool> {
        strategy.exists(self, path).await
    }

    /// Lists entries whose paths start with the given prefix.
    ///
    /// This method uses the selected strategy for the listing operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn list_entries() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("dir/example.txt");
    ///     storage.upload(path, &Bytes::from("Loco!")).await.unwrap();
    ///
    ///     let entries = storage.list(Path::new("dir"), true).await.unwrap();
    ///     assert_eq!(entries.len(), 1);
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the listing operation fails or if
    /// there is an issue with the strategy configuration.
    pub async fn list(&self, path: &Path, recursive: bool) -> StorageResult<Vec<ListEntry>> {
        self.list_with_policy(path, recursive, &*self.strategy)
            .await
    }

    /// Lists entries whose paths start with the given prefix using a specific
    /// strategy.
    ///
    /// This method allows specifying a custom strategy for the listing
    /// operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the listing operation fails or if
    /// there is an issue with the strategy configuration.
    pub async fn list_with_policy(
        &self,
        path: &Path,
        recursive: bool,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<Vec<ListEntry>> {
        strategy.list(self, path, recursive).await
    }

    /// Retrieves metadata for a single path without downloading its content.
    ///
    /// This method uses the selected strategy for the stat operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn stat_entry() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("example.txt");
    ///     storage.upload(path, &Bytes::from("Loco!")).await.unwrap();
    ///
    ///     let entry = storage.stat(path).await.unwrap();
    ///     assert_eq!(entry.content_length, Some(5));
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the stat operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn stat(&self, path: &Path) -> StorageResult<ListEntry> {
        self.stat_with_policy(path, &*self.strategy).await
    }

    /// Retrieves metadata for a single path using a specific strategy.
    ///
    /// This method allows specifying a custom strategy for the stat
    /// operation.
    ///
    /// # Errors
    ///
    /// This method returns an error if the stat operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn stat_with_policy(
        &self,
        path: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<ListEntry> {
        strategy.stat(self, path).await
    }

    /// Returns a reference to the store with the specified name if exists.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     assert!(storage.as_store("default").is_some());
    ///     assert!(storage.as_store("store_2").is_none());
    /// }
    /// ```
    ///
    /// # Returns
    /// Return None if the given name not found.
    #[must_use]
    pub fn as_store(&self, name: &str) -> Option<&dyn StoreDriver> {
        self.stores.get(name).map(|s| &**s)
    }

    /// Returns a reference to the store with the specified name.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// use bytes::Bytes;
    /// pub async fn download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     assert!(storage.as_store_err("default").is_ok());
    ///     assert!(storage.as_store_err("store_2").is_err());
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// Return an error if the given store name not exists
    // REVIEW(nd): not sure bout the name 'as_store_err' -- it returns result
    pub fn as_store_err(&self, name: &str) -> StorageResult<&dyn StoreDriver> {
        self.as_store(name)
            .ok_or(StorageError::StoreNotFound(name.to_string()))
    }

    /// Downloads content from storage as a stream, enabling efficient
    /// handling of large files without loading them entirely into memory.
    ///
    /// This method uses the selected strategy for the download operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// pub async fn stream_download() {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("large_file.mp4");
    ///     
    ///     let stream = storage.download_stream(path).await.unwrap();
    ///     // Stream can be converted to axum Body for HTTP response
    ///     // let body = stream.into_body();
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the download operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn download_stream(&self, path: &Path) -> StorageResult<BytesStream> {
        self.download_stream_with_policy(path, &*self.strategy)
            .await
    }

    /// Downloads content from storage as a stream using a specific strategy.
    ///
    /// # Errors
    ///
    /// This method returns an error if the download operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn download_stream_with_policy(
        &self,
        path: &Path,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<BytesStream> {
        strategy.download_stream(self, path).await
    }

    /// Uploads content from a stream to storage, enabling efficient
    /// handling of large files without loading them entirely into memory.
    ///
    /// This method uses the selected strategy for the upload operation.
    ///
    /// # Examples
    ///```
    /// use loco_rs::storage;
    /// use std::path::Path;
    /// pub async fn stream_upload(stream: storage::stream::BytesStream) {
    ///     let storage = storage::Storage::single(storage::drivers::mem::new());
    ///     let path = Path::new("large_file.mp4");
    ///     
    ///     storage.upload_stream(path, stream).await.unwrap();
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// This method returns an error if the upload operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn upload_stream(&self, path: &Path, stream: BytesStream) -> StorageResult<()> {
        self.upload_stream_with_policy(path, stream, &*self.strategy)
            .await
    }

    /// Uploads content from a stream using a specific strategy.
    ///
    /// # Errors
    ///
    /// This method returns an error if the upload operation fails or if there
    /// is an issue with the strategy configuration.
    pub async fn upload_stream_with_policy(
        &self,
        path: &Path,
        stream: BytesStream,
        strategy: &dyn strategies::StorageStrategy,
    ) -> StorageResult<()> {
        strategy.upload_stream(self, path, stream).await
    }
}