chie_core/

tier_migration.rs

//! Content migration between storage tiers.
//!
//! This module implements actual file operations for moving content between
//! storage tiers (Hot SSD → Warm HDD → Cold Archive), with proper error
//! handling, retry logic, and progress tracking.
//!
//! # Example
//!
//! ```rust
//! use chie_core::tier_migration::{TierMigration, MigrationConfig};
//! use chie_core::tiered_storage::{TieredStorageManager, TieredStorageConfig, StorageTier};
//! use std::sync::Arc;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! let storage_config = TieredStorageConfig::default();
//! let storage = Arc::new(TieredStorageManager::new(storage_config));
//!
//! let config = MigrationConfig::default();
//! let migration = TierMigration::new(storage.clone(), config);
//!
//! // Execute pending migrations
//! let result = migration.execute_pending_migrations().await?;
//! println!("Migrated {} items ({} bytes)", result.successful, result.bytes_moved);
//! # Ok(())
//! # }
//! ```

use crate::tiered_storage::{PendingMove, StorageTier, TieredStorageManager};
use serde::{Deserialize, Serialize};
use std::io;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Instant;
use tokio::fs;
use tokio::time::Duration;
use tracing::{debug, error, info, warn};

/// Configuration for tier migration.
#[derive(Debug, Clone)]
pub struct MigrationConfig {
    /// Maximum concurrent migrations.
    pub max_concurrent: usize,
    /// Timeout for a single migration (seconds).
    pub migration_timeout_secs: u64,
    /// Maximum retries for failed migrations.
    pub max_retries: u32,
    /// Retry delay (milliseconds).
    pub retry_delay_ms: u64,
    /// Whether to verify data after migration.
    pub verify_after_move: bool,
    /// Whether to keep source file after failed verification.
    pub keep_source_on_error: bool,
    /// Minimum free space required in target tier (bytes).
    pub min_free_space: u64,
}

impl Default for MigrationConfig {
    fn default() -> Self {
        Self {
            max_concurrent: 4,
            migration_timeout_secs: 300, // 5 minutes
            max_retries: 3,
            retry_delay_ms: 1000, // 1 second
            verify_after_move: true,
            keep_source_on_error: true,
            min_free_space: 1024 * 1024 * 1024, // 1 GB
        }
    }
}

/// Migration status for a single content item.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum MigrationStatus {
    /// Migration pending.
    Pending,
    /// Migration in progress.
    InProgress,
    /// Migration completed successfully.
    Completed,
    /// Migration failed.
    Failed(String),
    /// Migration cancelled.
    Cancelled,
}

/// A migration task with tracking.
#[derive(Debug, Clone)]
pub struct MigrationTask {
    /// Content CID.
    pub cid: String,
    /// Source tier.
    pub from: StorageTier,
    /// Target tier.
    pub to: StorageTier,
    /// Content size.
    pub size: u64,
    /// Current status.
    pub status: MigrationStatus,
    /// Retry count.
    pub retries: u32,
    /// Created timestamp (Unix seconds).
    pub created_at: u64,
    /// Last update timestamp.
    pub updated_at: u64,
}

/// Result of a migration operation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MigrationResult {
    /// Number of successful migrations.
    pub successful: usize,
    /// Number of failed migrations.
    pub failed: usize,
    /// Number of cancelled migrations.
    pub cancelled: usize,
    /// Total bytes moved.
    pub bytes_moved: u64,
    /// Total time taken (milliseconds).
    pub duration_ms: u64,
    /// Average migration speed (MB/s).
    pub avg_speed_mbps: f64,
}

/// Tier migration manager.
pub struct TierMigration {
    /// Reference to tiered storage manager.
    storage: Arc<TieredStorageManager>,
    /// Migration configuration.
    config: MigrationConfig,
}

impl TierMigration {
    /// Create a new tier migration manager.
    #[must_use]
    pub fn new(storage: Arc<TieredStorageManager>, config: MigrationConfig) -> Self {
        Self { storage, config }
    }

    /// Execute pending migrations from storage manager.
    ///
    /// This reads pending moves from the storage manager and executes them
    /// with proper error handling and retry logic.
    pub async fn execute_pending_migrations(&self) -> Result<MigrationResult, MigrationError> {
        let pending = self.storage.get_pending_moves();
        if pending.is_empty() {
            return Ok(MigrationResult {
                successful: 0,
                failed: 0,
                cancelled: 0,
                bytes_moved: 0,
                duration_ms: 0,
                avg_speed_mbps: 0.0,
            });
        }

        info!("Starting migration of {} pending items", pending.len());
        let start = Instant::now();

        let mut tasks: Vec<MigrationTask> =
            pending.into_iter().map(|pm| self.create_task(pm)).collect();

        // Execute migrations with concurrency limit
        let mut successful = 0;
        let mut failed = 0;
        let cancelled = 0;
        let mut bytes_moved = 0u64;

        let semaphore = Arc::new(tokio::sync::Semaphore::new(self.config.max_concurrent));

        let mut handles = Vec::new();
        for task in tasks.iter_mut() {
            let permit = semaphore.clone().acquire_owned().await.unwrap();
            let storage = self.storage.clone();
            let config = self.config.clone();
            let mut task_clone = task.clone();

            let handle = tokio::spawn(async move {
                let result = execute_migration(&storage, &config, &mut task_clone).await;
                drop(permit);
                result
            });

            handles.push((handle, task));
        }

        // Collect results
        for (handle, task) in handles {
            match handle.await {
                Ok(Ok(size)) => {
                    successful += 1;
                    bytes_moved += size;
                    task.status = MigrationStatus::Completed;
                }
                Ok(Err(e)) => {
                    failed += 1;
                    task.status = MigrationStatus::Failed(e.to_string());
                    warn!("Migration failed for {}: {}", task.cid, e);
                }
                Err(e) => {
                    failed += 1;
                    task.status = MigrationStatus::Failed(format!("Task panic: {}", e));
                    error!("Migration task panicked for {}: {}", task.cid, e);
                }
            }
        }

        let duration = start.elapsed();
        let duration_ms = duration.as_millis() as u64;
        let avg_speed_mbps = if duration_ms > 0 {
            (bytes_moved as f64 / 1_000_000.0) / (duration_ms as f64 / 1000.0)
        } else {
            0.0
        };

        info!(
            "Migration complete: {} successful, {} failed, {} MB moved in {} ms ({:.2} MB/s)",
            successful,
            failed,
            bytes_moved / 1_000_000,
            duration_ms,
            avg_speed_mbps
        );

        Ok(MigrationResult {
            successful,
            failed,
            cancelled,
            bytes_moved,
            duration_ms,
            avg_speed_mbps,
        })
    }

    /// Migrate a single content item.
    pub async fn migrate_content(
        &self,
        cid: &str,
        target_tier: StorageTier,
    ) -> Result<u64, MigrationError> {
        let location = self
            .storage
            .get_location(cid)
            .ok_or_else(|| MigrationError::ContentNotFound(cid.to_string()))?;

        if location.tier == target_tier {
            return Err(MigrationError::AlreadyInTargetTier(cid.to_string()));
        }

        let mut task = MigrationTask {
            cid: cid.to_string(),
            from: location.tier,
            to: target_tier,
            size: location.size,
            status: MigrationStatus::Pending,
            retries: 0,
            created_at: current_timestamp(),
            updated_at: current_timestamp(),
        };

        execute_migration(&self.storage, &self.config, &mut task).await
    }

    /// Cancel pending migrations.
    #[must_use]
    #[inline]
    pub fn cancel_pending(&self) -> usize {
        // Note: In a real implementation, this would cancel in-flight migrations
        // For now, we just return the count of pending moves
        self.storage.get_pending_moves().len()
    }

    /// Get migration statistics.
    #[must_use]
    #[inline]
    pub fn config(&self) -> &MigrationConfig {
        &self.config
    }

    /// Create a migration task from pending move.
    #[must_use]
    pub fn create_task(&self, pm: PendingMove) -> MigrationTask {
        let now = current_timestamp();
        MigrationTask {
            cid: pm.cid,
            from: pm.from,
            to: pm.to,
            size: pm.size,
            status: MigrationStatus::Pending,
            retries: 0,
            created_at: now,
            updated_at: now,
        }
    }
}

/// Execute a single migration with retry logic.
async fn execute_migration(
    storage: &TieredStorageManager,
    config: &MigrationConfig,
    task: &mut MigrationTask,
) -> Result<u64, MigrationError> {
    task.status = MigrationStatus::InProgress;
    task.updated_at = current_timestamp();

    let timeout = Duration::from_secs(config.migration_timeout_secs);
    let migration_future = perform_migration(storage, config, task);

    match tokio::time::timeout(timeout, migration_future).await {
        Ok(result) => result,
        Err(_) => {
            task.status = MigrationStatus::Failed("Timeout".to_string());
            Err(MigrationError::Timeout(task.cid.clone()))
        }
    }
}

/// Perform the actual file migration.
async fn perform_migration(
    storage: &TieredStorageManager,
    config: &MigrationConfig,
    task: &mut MigrationTask,
) -> Result<u64, MigrationError> {
    let source_path = storage
        .get_content_path(&task.cid)
        .ok_or_else(|| MigrationError::SourcePathNotFound(task.cid.clone()))?;

    let target_path = get_target_path(storage, &task.cid, task.to)?;

    // Check if source exists
    if !source_path.exists() {
        return Err(MigrationError::SourceFileNotFound(source_path));
    }

    // Create target directory if needed
    if let Some(parent) = target_path.parent() {
        fs::create_dir_all(parent)
            .await
            .map_err(|e| MigrationError::IoError(format!("Failed to create target dir: {}", e)))?;
    }

    // Check free space
    if !has_free_space(&target_path, task.size, config.min_free_space).await {
        return Err(MigrationError::InsufficientSpace(task.to));
    }

    debug!(
        "Migrating {} from {:?} to {:?} ({} bytes)",
        task.cid, task.from, task.to, task.size
    );

    // Copy file to target
    fs::copy(&source_path, &target_path)
        .await
        .map_err(|e| MigrationError::IoError(format!("Copy failed: {}", e)))?;

    // Verify if enabled
    if config.verify_after_move && !verify_migration(&source_path, &target_path).await? {
        if !config.keep_source_on_error {
            let _ = fs::remove_file(&target_path).await;
        }
        return Err(MigrationError::VerificationFailed(task.cid.clone()));
    }

    // Remove source file
    fs::remove_file(&source_path)
        .await
        .map_err(|e| MigrationError::IoError(format!("Remove source failed: {}", e)))?;

    // Update storage manager
    storage.execute_move(&task.cid, task.to);

    task.status = MigrationStatus::Completed;
    task.updated_at = current_timestamp();

    info!(
        "Successfully migrated {} from {:?} to {:?}",
        task.cid, task.from, task.to
    );

    Ok(task.size)
}

/// Get target path for migration.
fn get_target_path(
    storage: &TieredStorageManager,
    cid: &str,
    target_tier: StorageTier,
) -> Result<PathBuf, MigrationError> {
    let tier_path = storage
        .get_tier_path(target_tier)
        .ok_or_else(|| MigrationError::TargetPathNotFound(cid.to_string()))?;

    Ok(tier_path.join(cid))
}

/// Check if target has sufficient free space.
async fn has_free_space(path: &Path, required: u64, min_free: u64) -> bool {
    // Note: This is a simplified check
    // In production, use platform-specific APIs to check disk space
    // For now, assume sufficient space
    let _ = (path, required, min_free);
    true
}

/// Verify migration by comparing file sizes.
async fn verify_migration(source: &Path, target: &Path) -> Result<bool, MigrationError> {
    let source_metadata = fs::metadata(source)
        .await
        .map_err(|e| MigrationError::IoError(format!("Read source metadata: {}", e)))?;

    let target_metadata = fs::metadata(target)
        .await
        .map_err(|e| MigrationError::IoError(format!("Read target metadata: {}", e)))?;

    Ok(source_metadata.len() == target_metadata.len())
}

/// Get current Unix timestamp.
fn current_timestamp() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Migration errors.
#[derive(Debug, thiserror::Error)]
pub enum MigrationError {
    /// Content not found.
    #[error("Content not found: {0}")]
    ContentNotFound(String),

    /// Already in target tier.
    #[error("Content {0} is already in target tier")]
    AlreadyInTargetTier(String),

    /// Source path not found.
    #[error("Source path not found for content: {0}")]
    SourcePathNotFound(String),

    /// Target path not found.
    #[error("Target path not found for content: {0}")]
    TargetPathNotFound(String),

    /// Source file not found.
    #[error("Source file not found: {0}")]
    SourceFileNotFound(PathBuf),

    /// Insufficient space in target tier.
    #[error("Insufficient space in target tier: {0:?}")]
    InsufficientSpace(StorageTier),

    /// Migration timeout.
    #[error("Migration timeout for content: {0}")]
    Timeout(String),

    /// Verification failed.
    #[error("Migration verification failed for content: {0}")]
    VerificationFailed(String),

    /// IO error.
    #[error("IO error: {0}")]
    IoError(String),
}

impl From<io::Error> for MigrationError {
    fn from(e: io::Error) -> Self {
        MigrationError::IoError(e.to_string())
    }
}

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

    #[test]
    fn test_migration_config_default() {
        let config = MigrationConfig::default();
        assert_eq!(config.max_concurrent, 4);
        assert_eq!(config.max_retries, 3);
        assert!(config.verify_after_move);
    }

    #[test]
    fn test_migration_task_creation() {
        let pm = PendingMove {
            cid: "QmTest123".to_string(),
            from: StorageTier::Warm,
            to: StorageTier::Hot,
            size: 1024,
            priority: 10,
        };

        let storage_config = TieredStorageConfig::default();
        let storage = Arc::new(TieredStorageManager::new(storage_config));
        let migration = TierMigration::new(storage, MigrationConfig::default());

        let task = migration.create_task(pm);
        assert_eq!(task.cid, "QmTest123");
        assert_eq!(task.from, StorageTier::Warm);
        assert_eq!(task.to, StorageTier::Hot);
        assert_eq!(task.size, 1024);
        assert_eq!(task.status, MigrationStatus::Pending);
        assert_eq!(task.retries, 0);
    }

    #[test]
    fn test_migration_status() {
        assert_eq!(MigrationStatus::Pending, MigrationStatus::Pending);
        assert_ne!(
            MigrationStatus::Completed,
            MigrationStatus::Failed("error".to_string())
        );
    }
}