lcpfs 2026.1.102

// Copyright 2025 LunaOS Contributors
// SPDX-License-Identifier: Apache-2.0
//
// Cloud Tiering
// Automatic archival to S3/Azure/GCS with policy-based migration.

use alloc::collections::BTreeMap;
use alloc::format;
use alloc::string::String;
use alloc::vec::Vec;

/// Cloud provider type
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum CloudProvider {
    /// Amazon S3
    AwsS3,
    /// Azure Blob Storage
    AzureBlob,
    /// Google Cloud Storage
    GoogleGcs,
    /// MinIO (S3-compatible)
    Minio,
}

impl CloudProvider {
    /// Get the human-readable name of the cloud provider
    pub fn name(&self) -> &'static str {
        match self {
            CloudProvider::AwsS3 => "AWS S3",
            CloudProvider::AzureBlob => "Azure Blob",
            CloudProvider::GoogleGcs => "Google GCS",
            CloudProvider::Minio => "MinIO",
        }
    }

    /// Get endpoint for provider
    pub fn endpoint(&self, region: &str) -> String {
        match self {
            CloudProvider::AwsS3 => {
                format!("https://s3.{}.amazonaws.com", region)
            }
            CloudProvider::AzureBlob => {
                format!("https://{}.blob.core.windows.net", region)
            }
            CloudProvider::GoogleGcs => "https://storage.googleapis.com".into(),
            CloudProvider::Minio => {
                format!("http://{}:9000", region) // region is hostname
            }
        }
    }
}

/// Storage class/tier in cloud
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum CloudStorageClass {
    /// Frequent access (S3 Standard, Azure Hot)
    Hot,
    /// Infrequent access (S3 IA, Azure Cool)
    Warm,
    /// Rare access (S3 Glacier, Azure Archive)
    Cold,
    /// Long-term archive (S3 Deep Archive)
    Archive,
}

impl CloudStorageClass {
    /// Get the human-readable name of the storage class
    pub fn name(&self) -> &'static str {
        match self {
            CloudStorageClass::Hot => "Hot",
            CloudStorageClass::Warm => "Warm",
            CloudStorageClass::Cold => "Cold",
            CloudStorageClass::Archive => "Archive",
        }
    }

    /// Get relative cost multiplier
    pub fn cost_multiplier(&self) -> f64 {
        match self {
            CloudStorageClass::Hot => 1.0,
            CloudStorageClass::Warm => 0.5,
            CloudStorageClass::Cold => 0.1,
            CloudStorageClass::Archive => 0.04,
        }
    }

    /// Get retrieval latency in seconds
    pub fn retrieval_latency_sec(&self) -> u64 {
        match self {
            CloudStorageClass::Hot => 1,
            CloudStorageClass::Warm => 5,
            CloudStorageClass::Cold => 3600,     // 1 hour
            CloudStorageClass::Archive => 43200, // 12 hours
        }
    }
}

/// Cloud tier policy
#[derive(Debug, Clone)]
pub struct TierPolicy {
    /// Age threshold in days
    pub age_days: u64,
    /// Target storage class
    pub storage_class: CloudStorageClass,
    /// Minimum size in bytes
    pub min_size: u64,
    /// Maximum size in bytes (0 = no limit)
    pub max_size: u64,
    /// Access count threshold (below this = tier)
    pub access_threshold: u64,
}

impl TierPolicy {
    /// Create a new tier policy with the given age threshold and target storage class
    pub fn new(age_days: u64, storage_class: CloudStorageClass) -> Self {
        Self {
            age_days,
            storage_class,
            min_size: 0,
            max_size: 0,
            access_threshold: 0,
        }
    }

    /// Check if object matches policy criteria
    pub fn matches(&self, age_days: u64, size: u64, access_count: u64) -> bool {
        if age_days < self.age_days {
            return false;
        }
        if self.min_size > 0 && size < self.min_size {
            return false;
        }
        if self.max_size > 0 && size > self.max_size {
            return false;
        }
        if self.access_threshold > 0 && access_count >= self.access_threshold {
            return false;
        }
        true
    }
}

/// Cloud object metadata
#[derive(Debug, Clone)]
pub struct CloudObject {
    /// Local dataset ID
    pub dataset_id: u64,
    /// Local block offset
    pub block_offset: u64,
    /// Object size
    pub size: u64,
    /// Cloud provider
    pub provider: CloudProvider,
    /// Bucket/container name
    pub bucket: String,
    /// Object key/path
    pub key: String,
    /// Storage class
    pub storage_class: CloudStorageClass,
    /// Upload timestamp
    pub uploaded_at: u64,
    /// Last access timestamp
    pub last_access: u64,
    /// Access count
    pub access_count: u64,
    /// ETag/version identifier
    pub etag: String,
}

/// Cloud object upload configuration
#[derive(Debug, Clone)]
pub struct CloudUploadConfig {
    /// Local dataset ID
    pub dataset_id: u64,
    /// Local block offset
    pub block_offset: u64,
    /// Object size
    pub size: u64,
    /// Cloud provider
    pub provider: CloudProvider,
    /// Bucket/container name
    pub bucket: String,
    /// Object key/path
    pub key: String,
    /// Storage class
    pub storage_class: CloudStorageClass,
}

impl CloudObject {
    /// Create new cloud object from upload config and timestamp
    pub fn new(config: CloudUploadConfig, timestamp: u64) -> Self {
        let etag = format!("etag-{}-{}", config.dataset_id, config.block_offset);
        Self {
            dataset_id: config.dataset_id,
            block_offset: config.block_offset,
            size: config.size,
            provider: config.provider,
            bucket: config.bucket,
            key: config.key,
            storage_class: config.storage_class,
            uploaded_at: timestamp,
            last_access: timestamp,
            access_count: 0,
            etag,
        }
    }
}

/// Cloud tier statistics
#[derive(Debug, Clone, Default)]
pub struct CloudTierStats {
    /// Objects uploaded
    pub uploaded: u64,
    /// Bytes uploaded
    pub uploaded_bytes: u64,
    /// Objects downloaded
    pub downloaded: u64,
    /// Bytes downloaded
    pub downloaded_bytes: u64,
    /// Objects deleted
    pub deleted: u64,
    /// Policy migrations
    pub migrations: u64,
    /// Failed uploads
    pub upload_failures: u64,
    /// Failed downloads
    pub download_failures: u64,
}

/// Cloud tiering manager
pub struct CloudTierManager {
    /// Cloud objects by (dataset_id, block_offset)
    objects: BTreeMap<(u64, u64), CloudObject>,
    /// Tier policies (ordered by age)
    policies: Vec<TierPolicy>,
    /// Statistics
    stats: CloudTierStats,
}

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

impl CloudTierManager {
    /// Create a new cloud tier manager with empty state
    pub fn new() -> Self {
        Self {
            objects: BTreeMap::new(),
            policies: Vec::new(),
            stats: CloudTierStats::default(),
        }
    }

    /// Add a tier policy
    pub fn add_policy(&mut self, policy: TierPolicy) {
        self.policies.push(policy);
        // Sort by age (highest first) - want most specific policy
        self.policies.sort_by(|a, b| b.age_days.cmp(&a.age_days));
    }

    /// Upload object to cloud
    ///
    /// # Arguments
    /// * `object` - Cloud object with pre-configured metadata
    ///
    /// # Example
    /// ```ignore
    /// let object = CloudObject::new(dataset_id, block_offset, size, provider, bucket, key, class, timestamp);
    /// manager.upload(object)?;
    /// ```
    pub fn upload(&mut self, object: CloudObject) -> Result<(), &'static str> {
        let dataset_id = object.dataset_id;
        let block_offset = object.block_offset;
        let size = object.size;
        let provider_name = object.provider.name();
        let bucket = object.bucket.clone();
        let storage_class_name = object.storage_class.name();

        self.objects.insert((dataset_id, block_offset), object);
        self.stats.uploaded += 1;
        self.stats.uploaded_bytes += size;

        crate::lcpfs_println!(
            "[ CLOUD ] Uploaded dataset {} block 0x{:x} to {} ({} / {})",
            dataset_id,
            block_offset,
            provider_name,
            bucket,
            storage_class_name
        );

        Ok(())
    }

    /// Download object from cloud
    pub fn download(
        &mut self,
        dataset_id: u64,
        block_offset: u64,
        timestamp: u64,
    ) -> Result<(), &'static str> {
        let object = self
            .objects
            .get_mut(&(dataset_id, block_offset))
            .ok_or("Object not found in cloud")?;

        object.last_access = timestamp;
        object.access_count += 1;

        self.stats.downloaded += 1;
        self.stats.downloaded_bytes += object.size;

        crate::lcpfs_println!(
            "[ CLOUD ] Downloaded dataset {} block 0x{:x} from {} (latency: {}s)",
            dataset_id,
            block_offset,
            object.provider.name(),
            object.storage_class.retrieval_latency_sec()
        );

        Ok(())
    }

    /// Delete object from cloud
    pub fn delete(&mut self, dataset_id: u64, block_offset: u64) -> Result<(), &'static str> {
        self.objects
            .remove(&(dataset_id, block_offset))
            .ok_or("Object not found")?;

        self.stats.deleted += 1;

        crate::lcpfs_println!(
            "[ CLOUD ] Deleted dataset {} block 0x{:x} from cloud",
            dataset_id,
            block_offset
        );

        Ok(())
    }

    /// Apply lifecycle policies
    pub fn apply_policies(&mut self, current_time: u64) {
        let mut migrations = Vec::new();

        for (key, object) in &self.objects {
            let age_ms = current_time.saturating_sub(object.uploaded_at);
            let age_days = age_ms / (24 * 3600 * 1000);

            // Find matching policy
            for policy in &self.policies {
                if policy.matches(age_days, object.size, object.access_count) {
                    if policy.storage_class != object.storage_class {
                        migrations.push((*key, policy.storage_class));
                    }
                    break;
                }
            }
        }

        // Apply migrations
        for ((dataset_id, block_offset), new_class) in migrations {
            if let Some(object) = self.objects.get_mut(&(dataset_id, block_offset)) {
                let old_class = object.storage_class;
                object.storage_class = new_class;
                self.stats.migrations += 1;

                crate::lcpfs_println!(
                    "[ CLOUD ] Migrated dataset {} block 0x{:x}: {} -> {}",
                    dataset_id,
                    block_offset,
                    old_class.name(),
                    new_class.name()
                );
            }
        }
    }

    /// Get object metadata
    pub fn get_object(&self, dataset_id: u64, block_offset: u64) -> Option<&CloudObject> {
        self.objects.get(&(dataset_id, block_offset))
    }

    /// List objects by storage class
    pub fn list_by_class(&self, storage_class: CloudStorageClass) -> Vec<&CloudObject> {
        self.objects
            .values()
            .filter(|obj| obj.storage_class == storage_class)
            .collect()
    }

    /// Get statistics
    pub fn get_stats(&self) -> CloudTierStats {
        self.stats.clone()
    }

    /// Get total cloud storage used
    pub fn total_cloud_bytes(&self) -> u64 {
        self.objects.values().map(|obj| obj.size).sum()
    }

    /// Get storage breakdown by class
    pub fn storage_by_class(&self) -> BTreeMap<CloudStorageClass, u64> {
        let mut breakdown = BTreeMap::new();
        for object in self.objects.values() {
            *breakdown.entry(object.storage_class).or_insert(0) += object.size;
        }
        breakdown
    }
}

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

    /// Test helper to create a cloud object quickly
    fn test_obj(
        dataset_id: u64,
        block_offset: u64,
        size: u64,
        provider: CloudProvider,
        bucket: &str,
        key: &str,
        storage_class: CloudStorageClass,
        timestamp: u64,
    ) -> CloudObject {
        CloudObject::new(
            CloudUploadConfig {
                dataset_id,
                block_offset,
                size,
                provider,
                bucket: bucket.into(),
                key: key.into(),
                storage_class,
            },
            timestamp,
        )
    }

    #[test]
    fn test_cloud_provider_endpoint() {
        assert_eq!(
            CloudProvider::AwsS3.endpoint("us-east-1"),
            "https://s3.us-east-1.amazonaws.com"
        );
        assert_eq!(
            CloudProvider::AzureBlob.endpoint("mystorageaccount"),
            "https://mystorageaccount.blob.core.windows.net"
        );
        assert_eq!(
            CloudProvider::GoogleGcs.endpoint(""),
            "https://storage.googleapis.com"
        );
    }

    #[test]
    fn test_storage_class_cost() {
        assert!(
            CloudStorageClass::Hot.cost_multiplier() > CloudStorageClass::Warm.cost_multiplier()
        );
        assert!(
            CloudStorageClass::Warm.cost_multiplier() > CloudStorageClass::Cold.cost_multiplier()
        );
        assert!(
            CloudStorageClass::Cold.cost_multiplier()
                > CloudStorageClass::Archive.cost_multiplier()
        );
    }

    #[test]
    fn test_storage_class_latency() {
        assert!(
            CloudStorageClass::Hot.retrieval_latency_sec()
                < CloudStorageClass::Warm.retrieval_latency_sec()
        );
        assert!(
            CloudStorageClass::Warm.retrieval_latency_sec()
                < CloudStorageClass::Cold.retrieval_latency_sec()
        );
        assert!(
            CloudStorageClass::Cold.retrieval_latency_sec()
                < CloudStorageClass::Archive.retrieval_latency_sec()
        );
    }

    #[test]
    fn test_tier_policy_age() {
        let policy = TierPolicy::new(30, CloudStorageClass::Warm);

        assert!(!policy.matches(29, 1000, 0)); // Too young
        assert!(policy.matches(30, 1000, 0)); // Exactly 30 days
        assert!(policy.matches(31, 1000, 0)); // Older than 30 days
    }

    #[test]
    fn test_tier_policy_size() {
        let mut policy = TierPolicy::new(30, CloudStorageClass::Cold);
        policy.min_size = 1_000_000; // 1 MB
        policy.max_size = 10_000_000; // 10 MB

        assert!(!policy.matches(30, 500_000, 0)); // Too small
        assert!(policy.matches(30, 5_000_000, 0)); // In range
        assert!(!policy.matches(30, 20_000_000, 0)); // Too large
    }

    #[test]
    fn test_tier_policy_access() {
        let mut policy = TierPolicy::new(30, CloudStorageClass::Archive);
        policy.access_threshold = 5; // Tier if accessed < 5 times

        assert!(policy.matches(30, 1000, 0)); // Never accessed
        assert!(policy.matches(30, 1000, 4)); // Low access
        assert!(!policy.matches(30, 1000, 5)); // Threshold met
        assert!(!policy.matches(30, 1000, 10)); // High access
    }

    #[test]
    fn test_upload_download() {
        let mut manager = CloudTierManager::new();

        let obj = test_obj(
            1,
            0x1000,
            4096,
            CloudProvider::AwsS3,
            "my-bucket",
            "data/block-1000",
            CloudStorageClass::Hot,
            1000,
        );
        manager.upload(obj).expect("test: operation should succeed");

        assert_eq!(manager.stats.uploaded, 1);
        assert_eq!(manager.stats.uploaded_bytes, 4096);

        let object = manager
            .get_object(1, 0x1000)
            .expect("test: operation should succeed");
        assert_eq!(object.provider, CloudProvider::AwsS3);
        assert_eq!(object.bucket, "my-bucket");
        assert_eq!(object.storage_class, CloudStorageClass::Hot);

        manager
            .download(1, 0x1000, 2000)
            .expect("test: operation should succeed");
        assert_eq!(manager.stats.downloaded, 1);
        assert_eq!(manager.stats.downloaded_bytes, 4096);

        let object = manager
            .get_object(1, 0x1000)
            .expect("test: operation should succeed");
        assert_eq!(object.access_count, 1);
        assert_eq!(object.last_access, 2000);
    }

    #[test]
    fn test_delete_object() {
        let mut manager = CloudTierManager::new();

        let obj = test_obj(
            1,
            0x1000,
            4096,
            CloudProvider::AwsS3,
            "bucket",
            "key",
            CloudStorageClass::Hot,
            1000,
        );
        manager.upload(obj).expect("test: operation should succeed");

        assert!(manager.get_object(1, 0x1000).is_some());

        manager
            .delete(1, 0x1000)
            .expect("test: operation should succeed");
        assert_eq!(manager.stats.deleted, 1);
        assert!(manager.get_object(1, 0x1000).is_none());
    }

    #[test]
    fn test_policy_migration() {
        let mut manager = CloudTierManager::new();

        // Upload object
        let obj = test_obj(
            1,
            0x1000,
            4096,
            CloudProvider::AwsS3,
            "bucket",
            "key",
            CloudStorageClass::Hot,
            1000,
        );
        manager.upload(obj).expect("test: operation should succeed");

        // Add policy: tier to Warm after 30 days
        let policy = TierPolicy::new(30, CloudStorageClass::Warm);
        manager.add_policy(policy);

        // 29 days later - no migration
        let time_29_days = 1000 + 29 * 24 * 3600 * 1000;
        manager.apply_policies(time_29_days);
        assert_eq!(
            manager
                .get_object(1, 0x1000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Hot
        );
        assert_eq!(manager.stats.migrations, 0);

        // 31 days later - should migrate
        let time_31_days = 1000 + 31 * 24 * 3600 * 1000;
        manager.apply_policies(time_31_days);
        assert_eq!(
            manager
                .get_object(1, 0x1000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Warm
        );
        assert_eq!(manager.stats.migrations, 1);
    }

    #[test]
    fn test_multi_tier_policies() {
        let mut manager = CloudTierManager::new();

        // Upload object
        let obj = test_obj(
            1,
            0x1000,
            4096,
            CloudProvider::AwsS3,
            "bucket",
            "key",
            CloudStorageClass::Hot,
            0,
        );
        manager.upload(obj).expect("test: operation should succeed");

        // Add multi-tier policies
        manager.add_policy(TierPolicy::new(30, CloudStorageClass::Warm));
        manager.add_policy(TierPolicy::new(90, CloudStorageClass::Cold));
        manager.add_policy(TierPolicy::new(365, CloudStorageClass::Archive));

        // 60 days - should be Warm
        let time_60_days = 60 * 24 * 3600 * 1000;
        manager.apply_policies(time_60_days);
        assert_eq!(
            manager
                .get_object(1, 0x1000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Warm
        );

        // 120 days - should be Cold
        let time_120_days = 120 * 24 * 3600 * 1000;
        manager.apply_policies(time_120_days);
        assert_eq!(
            manager
                .get_object(1, 0x1000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Cold
        );

        // 400 days - should be Archive
        let time_400_days = 400 * 24 * 3600 * 1000;
        manager.apply_policies(time_400_days);
        assert_eq!(
            manager
                .get_object(1, 0x1000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Archive
        );

        assert_eq!(manager.stats.migrations, 3);
    }

    #[test]
    fn test_list_by_class() {
        let mut manager = CloudTierManager::new();

        manager
            .upload(test_obj(
                1,
                0x1000,
                4096,
                CloudProvider::AwsS3,
                "b1",
                "k1",
                CloudStorageClass::Hot,
                1000,
            ))
            .expect("test: operation should succeed");
        manager
            .upload(test_obj(
                2,
                0x2000,
                8192,
                CloudProvider::AwsS3,
                "b2",
                "k2",
                CloudStorageClass::Hot,
                1000,
            ))
            .expect("test: operation should succeed");
        manager
            .upload(test_obj(
                3,
                0x3000,
                16384,
                CloudProvider::AwsS3,
                "b3",
                "k3",
                CloudStorageClass::Warm,
                1000,
            ))
            .expect("test: operation should succeed");

        let hot_objects = manager.list_by_class(CloudStorageClass::Hot);
        assert_eq!(hot_objects.len(), 2);

        let warm_objects = manager.list_by_class(CloudStorageClass::Warm);
        assert_eq!(warm_objects.len(), 1);
    }

    #[test]
    fn test_storage_breakdown() {
        let mut manager = CloudTierManager::new();

        manager
            .upload(test_obj(
                1,
                0x1000,
                4096,
                CloudProvider::AwsS3,
                "b",
                "k1",
                CloudStorageClass::Hot,
                1000,
            ))
            .expect("test: operation should succeed");
        manager
            .upload(test_obj(
                2,
                0x2000,
                8192,
                CloudProvider::AwsS3,
                "b",
                "k2",
                CloudStorageClass::Hot,
                1000,
            ))
            .expect("test: operation should succeed");
        manager
            .upload(test_obj(
                3,
                0x3000,
                16384,
                CloudProvider::AwsS3,
                "b",
                "k3",
                CloudStorageClass::Warm,
                1000,
            ))
            .expect("test: operation should succeed");

        let breakdown = manager.storage_by_class();
        assert_eq!(breakdown.get(&CloudStorageClass::Hot), Some(&12288)); // 4096 + 8192
        assert_eq!(breakdown.get(&CloudStorageClass::Warm), Some(&16384));

        assert_eq!(manager.total_cloud_bytes(), 28672); // 4096 + 8192 + 16384
    }

    #[test]
    fn test_access_based_policy() {
        let mut manager = CloudTierManager::new();

        manager
            .upload(test_obj(
                1,
                0x1000,
                4096,
                CloudProvider::AwsS3,
                "b",
                "k",
                CloudStorageClass::Hot,
                0,
            ))
            .expect("test: operation should succeed");

        // Policy: tier to Archive if older than 30 days AND accessed < 5 times
        let mut policy = TierPolicy::new(30, CloudStorageClass::Archive);
        policy.access_threshold = 5;
        manager.add_policy(policy);

        // 31 days, 10 accesses - should NOT tier (high access)
        for _ in 0..10 {
            manager
                .download(1, 0x1000, 1000)
                .expect("test: operation should succeed");
        }
        let time_31_days = 31 * 24 * 3600 * 1000;
        manager.apply_policies(time_31_days);
        assert_eq!(
            manager
                .get_object(1, 0x1000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Hot
        );

        // Upload new object with low access
        manager
            .upload(test_obj(
                2,
                0x2000,
                4096,
                CloudProvider::AwsS3,
                "b",
                "k2",
                CloudStorageClass::Hot,
                0,
            ))
            .expect("test: operation should succeed");
        manager
            .download(2, 0x2000, 1000)
            .expect("test: operation should succeed"); // Only 1 access

        manager.apply_policies(time_31_days);
        assert_eq!(
            manager
                .get_object(2, 0x2000)
                .expect("test: operation should succeed")
                .storage_class,
            CloudStorageClass::Archive
        );
    }
}