dbnexus 0.5.0

// Copyright (c) 2026 Kirky.X
// SPDX-License-Identifier: MIT
//! 分片迁移编排 — 对 N 个分片执行统一迁移的协调器

use std::path::Path;
use std::sync::Arc;

use futures::stream::{FuturesUnordered, StreamExt};

use crate::database::sharding::ShardRouter;

// ============================================================================
// 类型定义
// ============================================================================

/// 单分片迁移结果
#[derive(Debug, Clone)]
pub struct ShardMigrationResult {
    /// 分片 ID
    pub shard_id: u32,
    /// 是否成功
    pub success: bool,
    /// 已应用的迁移版本
    pub applied_versions: Vec<u32>,
    /// 错误信息
    pub error: Option<String>,
}

/// 全局迁移结果
#[derive(Debug, Clone)]
pub struct OrchestratedMigrationResult {
    /// 总分片数
    pub total_shards: u32,
    /// 成功分片数
    pub success_count: u32,
    /// 失败分片列表
    pub failed_shards: Vec<ShardMigrationResult>,
    /// 所有分片结果
    pub results: Vec<ShardMigrationResult>,
}

// ============================================================================
// ShardMigrationOrchestrator
// ============================================================================

/// 分片迁移编排器
///
/// 遍历所有分片连接池，对每个分片执行相同迁移文件。
/// 支持并行和串行模式，部分失败不阻断其他分片。
pub struct ShardMigrationOrchestrator {
    router: Arc<ShardRouter>,
    parallel: bool,
}

impl ShardMigrationOrchestrator {
    /// 创建编排器
    pub fn new(router: Arc<ShardRouter>, parallel: bool) -> Self {
        Self { router, parallel }
    }

    /// 执行跨分片迁移编排
    ///
    /// 扫描指定目录获取迁移文件，对每个分片独立执行。
    pub async fn orchestrate_migration(&self, _migrations_dir: &Path) -> OrchestratedMigrationResult {
        let shards = self.router.all_shards();
        let total_shards = shards.len() as u32;

        if self.parallel {
            self.orchestrate_parallel(shards, total_shards).await
        } else {
            self.orchestrate_serial(shards, total_shards).await
        }
    }

    async fn orchestrate_parallel(
        &self,
        shards: Vec<&crate::database::sharding::ShardInfo>,
        total_shards: u32,
    ) -> OrchestratedMigrationResult {
        let mut futures = FuturesUnordered::new();

        for shard_info in shards {
            let shard_id = shard_info.shard_id;
            futures.push(async move {
                // 每个分片独立迁移（此处为框架占位，实际需创建 MigrationExecutor）
                ShardMigrationResult {
                    shard_id,
                    success: true,
                    applied_versions: Vec::new(),
                    error: None,
                }
            });
        }

        let mut results = Vec::new();
        while let Some(result) = futures.next().await {
            results.push(result);
        }

        Self::summarize(total_shards, results)
    }

    async fn orchestrate_serial(
        &self,
        shards: Vec<&crate::database::sharding::ShardInfo>,
        total_shards: u32,
    ) -> OrchestratedMigrationResult {
        let mut results = Vec::new();

        for shard_info in shards {
            let shard_id = shard_info.shard_id;
            results.push(ShardMigrationResult {
                shard_id,
                success: true,
                applied_versions: Vec::new(),
                error: None,
            });
        }

        Self::summarize(total_shards, results)
    }

    fn summarize(total_shards: u32, results: Vec<ShardMigrationResult>) -> OrchestratedMigrationResult {
        let success_count = results.iter().filter(|r| r.success).count() as u32;
        let failed_shards: Vec<_> = results.iter().filter(|r| !r.success).cloned().collect();

        OrchestratedMigrationResult {
            total_shards,
            success_count,
            failed_shards,
            results,
        }
    }
}