razeri_plugin_api/

storage.rs

// Copyright (C) 2026 Gabriel Shea
// Licensed under either of the Apache License, Version 2.0 or the MIT
// license, at your option.

//! Storage provider factory trait and registry.
//!
//! [`StorageProviderFactory`] is the extension point for custom storage backends.
//! [`StorageRegistry`] provides dual-key lookup for built-in and third-party providers.

use crate::error::StorageError;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use razeri_plugin_contract::StoragePlatform;
use std::collections::HashMap;
use std::sync::Arc;

/// Entry returned by listing a directory.
#[derive(Debug, Clone)]
pub struct StorageListEntry {
    /// File or directory name (just the last segment).
    pub name: String,
    /// Size in bytes (0 for directories).
    pub size_bytes: u64,
    /// Whether this entry is a directory/prefix.
    pub is_directory: bool,
}

/// Metadata about a single file/object in storage.
#[derive(Debug, Clone)]
pub struct StorageMetadata {
    /// Size in bytes.
    pub size_bytes: u64,
    /// Last modification time, if available.
    pub modified: Option<DateTime<Utc>>,
    /// MIME content type, if available.
    pub content_type: Option<String>,
    /// Checksum (e.g. `ETag`, MD5), if available.
    pub checksum: Option<String>,
}

/// Factory that creates storage operation instances from configuration.
///
/// Third-party storage backends (e.g., Kafka, HDFS, proprietary systems)
/// implement this trait to integrate with Razeri's file management.
#[async_trait]
pub trait StorageProviderFactory: Send + Sync {
    /// Machine-readable provider name (e.g., "s3", "`azure_blob`", "kafka").
    fn name(&self) -> &str;

    /// Human-readable display name.
    fn display_name(&self) -> &str;

    /// Known enum variant, if this is a built-in platform.
    fn known_platform(&self) -> Option<StoragePlatform>;

    /// Create a storage operations instance from a key-value configuration.
    async fn create(
        &self,
        config: &HashMap<String, String>,
    ) -> Result<Box<dyn StorageOps>, StorageError>;

    /// Validate configuration without creating a connection.
    fn validate_config(&self, config: &HashMap<String, String>) -> Vec<String>;

    /// List required configuration keys.
    fn required_config_keys(&self) -> Vec<&str>;
}

/// Minimal storage operations interface for plugin-API use.
///
/// This parallels the full `StorageOperations` in `razeri-runtime` but is
/// defined here so provider plugins don't depend on `razeri-runtime`.
#[async_trait]
pub trait StorageOps: Send + Sync {
    /// List entries at the given path.
    async fn list(&self, path: &str) -> Result<Vec<StorageListEntry>, StorageError>;

    /// Check if a path exists.
    async fn exists(&self, path: &str) -> Result<bool, StorageError>;

    /// Read file contents.
    async fn read(&self, path: &str) -> Result<Vec<u8>, StorageError>;

    /// Write content to a path.
    async fn write(&self, path: &str, content: &[u8]) -> Result<(), StorageError>;

    /// Get file metadata (size, modified time, content type, checksum).
    async fn metadata(&self, path: &str) -> Result<StorageMetadata, StorageError>;

    /// Move/rename a file (atomic where supported, copy+delete where not).
    async fn move_file(&self, src: &str, dst: &str) -> Result<(), StorageError>;

    /// Delete a path.
    async fn delete(&self, path: &str) -> Result<(), StorageError>;

    /// Create a directory/prefix (no-op for object stores).
    async fn mkdir(&self, path: &str) -> Result<(), StorageError>;
}

/// Registry for storage providers with dual-key lookup.
pub struct StorageRegistry {
    known: HashMap<StoragePlatform, Arc<dyn StorageProviderFactory>>,
    extensions: HashMap<String, Arc<dyn StorageProviderFactory>>,
}

impl StorageRegistry {
    /// Create an empty registry.
    #[must_use]
    pub fn new() -> Self {
        Self { known: HashMap::new(), extensions: HashMap::new() }
    }

    /// Register a built-in storage provider (dual-keyed).
    pub fn register_known(
        &mut self,
        platform: StoragePlatform,
        factory: Arc<dyn StorageProviderFactory>,
    ) {
        let name = factory.name().to_string();
        self.known.insert(platform, Arc::clone(&factory));
        self.extensions.insert(name, factory);
    }

    /// Register a third-party storage extension (string-keyed only).
    pub fn register_extension(&mut self, name: String, factory: Arc<dyn StorageProviderFactory>) {
        self.extensions.insert(name, factory);
    }

    /// Resolve by enum key (fast path for built-in platforms).
    #[must_use]
    pub fn get(&self, platform: &StoragePlatform) -> Option<&Arc<dyn StorageProviderFactory>> {
        self.known.get(platform)
    }

    /// Resolve by string name (extension path, also works for built-in).
    #[must_use]
    pub fn get_by_name(&self, name: &str) -> Option<&Arc<dyn StorageProviderFactory>> {
        self.extensions.get(name)
    }

    /// Resolve by string — covers both known and extension lookups.
    #[must_use]
    pub fn resolve(&self, target: &str) -> Option<Arc<dyn StorageProviderFactory>> {
        self.extensions.get(target).cloned()
    }

    /// List all registered providers as (name, `display_name`) pairs.
    #[must_use]
    pub fn list(&self) -> Vec<(&str, &str)> {
        self.extensions.values().map(|f| (f.name(), f.display_name())).collect()
    }
}

impl Default for StorageRegistry {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    struct MockStorageOps;

    #[async_trait]
    impl StorageOps for MockStorageOps {
        async fn list(&self, _path: &str) -> Result<Vec<StorageListEntry>, StorageError> {
            Ok(vec![StorageListEntry {
                name: "file1.csv".to_string(),
                size_bytes: 1024,
                is_directory: false,
            }])
        }

        async fn exists(&self, _path: &str) -> Result<bool, StorageError> {
            Ok(true)
        }

        async fn read(&self, _path: &str) -> Result<Vec<u8>, StorageError> {
            Ok(b"test data".to_vec())
        }

        async fn write(&self, _path: &str, _content: &[u8]) -> Result<(), StorageError> {
            Ok(())
        }

        async fn metadata(&self, _path: &str) -> Result<StorageMetadata, StorageError> {
            Ok(StorageMetadata {
                size_bytes: 42,
                modified: None,
                content_type: None,
                checksum: None,
            })
        }

        async fn move_file(&self, _src: &str, _dst: &str) -> Result<(), StorageError> {
            Ok(())
        }

        async fn delete(&self, _path: &str) -> Result<(), StorageError> {
            Ok(())
        }

        async fn mkdir(&self, _path: &str) -> Result<(), StorageError> {
            Ok(())
        }
    }

    struct MockStorageFactory {
        name: &'static str,
        display_name: &'static str,
        known_platform: Option<StoragePlatform>,
    }

    impl MockStorageFactory {
        fn s3() -> Self {
            Self {
                name: "s3",
                display_name: "Amazon S3",
                known_platform: Some(StoragePlatform::S3),
            }
        }

        fn kafka() -> Self {
            Self { name: "kafka", display_name: "Apache Kafka", known_platform: None }
        }
    }

    #[async_trait]
    impl StorageProviderFactory for MockStorageFactory {
        fn name(&self) -> &str {
            self.name
        }

        fn display_name(&self) -> &str {
            self.display_name
        }

        fn known_platform(&self) -> Option<StoragePlatform> {
            self.known_platform
        }

        async fn create(
            &self,
            _config: &HashMap<String, String>,
        ) -> Result<Box<dyn StorageOps>, StorageError> {
            Ok(Box::new(MockStorageOps))
        }

        fn validate_config(&self, config: &HashMap<String, String>) -> Vec<String> {
            let mut errors = Vec::new();
            for key in self.required_config_keys() {
                if !config.contains_key(key) {
                    errors.push(format!("missing required key: {key}"));
                }
            }
            errors
        }

        fn required_config_keys(&self) -> Vec<&str> {
            vec!["bucket", "region"]
        }
    }

    #[tokio::test]
    async fn storage_factory_creates_operations() {
        let factory = MockStorageFactory::s3();
        let config = HashMap::from([
            ("bucket".to_string(), "test-bucket".to_string()),
            ("region".to_string(), "us-east-1".to_string()),
        ]);
        let ops = factory.create(&config).await.unwrap();
        assert!(ops.exists("test").await.unwrap());
    }

    #[test]
    fn storage_registry_dual_key_lookup() {
        let mut registry = StorageRegistry::new();
        registry.register_known(StoragePlatform::S3, Arc::new(MockStorageFactory::s3()));

        assert!(registry.get(&StoragePlatform::S3).is_some());
        assert!(registry.get_by_name("s3").is_some());
    }

    #[test]
    fn storage_config_validation_catches_missing_keys() {
        let factory = MockStorageFactory::s3();
        let config = HashMap::new();
        let errors = factory.validate_config(&config);
        assert_eq!(errors.len(), 2);
        assert!(errors[0].contains("bucket"));
    }

    #[test]
    fn storage_extension_registration() {
        let mut registry = StorageRegistry::new();
        registry.register_extension("kafka".to_string(), Arc::new(MockStorageFactory::kafka()));
        assert!(registry.get_by_name("kafka").is_some());
    }

    #[test]
    fn storage_registry_list_shows_all_providers() {
        let mut registry = StorageRegistry::new();
        registry.register_known(StoragePlatform::S3, Arc::new(MockStorageFactory::s3()));
        registry.register_extension("kafka".to_string(), Arc::new(MockStorageFactory::kafka()));
        assert_eq!(registry.list().len(), 2);
    }

    #[test]
    fn storage_registry_resolve_covers_both_known_and_extension() {
        let mut registry = StorageRegistry::new();
        registry.register_known(StoragePlatform::S3, Arc::new(MockStorageFactory::s3()));
        registry.register_extension("kafka".to_string(), Arc::new(MockStorageFactory::kafka()));

        assert!(registry.resolve("s3").is_some());
        assert!(registry.resolve("kafka").is_some());
        assert!(registry.resolve("nonexistent").is_none());
    }
}