代码优先策略下，部署阶段Document DB（Cosmos）操作管理咨询（类EF迁移）

要实现类似EF Migrations的Up/Down模式来管理Cosmos等文档数据库的操作，核心是自定义迁移版本控制系统，通过追踪已执行的变更、提供正向/反向操作逻辑，将数据库变更纳入部署流程。以下是具体实现方案：

一、核心设计思路 #

1. 迁移元数据追踪：在Cosmos中创建专门的元数据容器（如__MigrationHistory），记录每个已执行迁移的唯一ID、执行时间、状态（成功/失败）等信息。
2. 迁移单元标准化：每个迁移单元对应一个独立的逻辑块，包含Up（执行变更）和Down（回滚变更）方法，且具备唯一标识（如时间戳+描述的ID）。
3. 迁移执行引擎：编写工具/程序，负责扫描所有待执行迁移，对比元数据判断哪些未执行，自动执行Up；如需回滚，则根据元数据执行指定迁移的Down方法。

二、具体实现步骤 #

1. 初始化迁移元数据容器 #

首先在Cosmos中创建用于追踪迁移的容器，示例SQL（Cosmos SQL API）：

-- 创建__MigrationHistory容器（按需调整分区键）
CREATE CONTAINER IF NOT EXISTS __MigrationHistory WITH PARTITION KEY = "/id"

该容器的文档结构示例：

{
  "id": "202405201000_create_products_container",
  "appliedAt": "2024-05-20T10:00:00Z",
  "status": "Success",
  "description": "Create products container with composite index"
}

2. 定义迁移接口与模板 #

以C#为例，定义统一的迁移接口，规范每个迁移的结构：

public interface ICosmosMigration
{
    // 迁移唯一ID，建议用"时间戳_描述"格式
    string Id { get; }
    // 正向执行变更
    Task UpAsync(CosmosClient client);
    // 反向回滚变更
    Task DownAsync(CosmosClient client);
}

然后创建具体的迁移类，比如创建容器的迁移：

public class _202405201000_CreateProductsContainer : ICosmosMigration
{
    public string Id => "202405201000_create_products_container";

    public async Task UpAsync(CosmosClient client)
    {
        var database = client.GetDatabase("MyAppDB");
        // 创建products容器，指定索引策略
        await database.CreateContainerIfNotExistsAsync(
            id: "products",
            partitionKeyPath: "/category",
            throughput: 400,
            containerProperties: new ContainerProperties
            {
                IndexingPolicy = new IndexingPolicy
                {
                    CompositeIndexes = new List<IList<CompositePath>>
                    {
                        new List<CompositePath>
                        {
                            new CompositePath { Path = "/category", Order = CompositePathSortOrder.Ascending },
                            new CompositePath { Path = "/price", Order = CompositePathSortOrder.Descending }
                        }
                    }
                }
            });
    }

    public async Task DownAsync(CosmosClient client)
    {
        var database = client.GetDatabase("MyAppDB");
        // 删除products容器（回滚操作）
        await database.DeleteContainerAsync("products");
    }
}

3. 实现迁移执行引擎 #

编写控制台程序或类库，负责扫描所有ICosmosMigration实现类，对比元数据执行迁移：

public class CosmosMigrationEngine
{
    private readonly CosmosClient _client;
    private readonly Container _migrationsContainer;
    private readonly IEnumerable<ICosmosMigration> _migrations;

    public CosmosMigrationEngine(CosmosClient client, IEnumerable<ICosmosMigration> migrations)
    {
        _client = client;
        _migrationsContainer = client.GetDatabase("MyAppDB").GetContainer("__MigrationHistory");
        _migrations = migrations.OrderBy(m => m.Id);
    }

    // 执行所有未应用的迁移
    public async Task ApplyMigrationsAsync()
    {
        var appliedMigrations = await GetAppliedMigrationsAsync();
        var pendingMigrations = _migrations.Where(m => !appliedMigrations.Contains(m.Id));

        foreach (var migration in pendingMigrations)
        {
            try
            {
                await migration.UpAsync(_client);
                await RecordMigrationAsync(migration.Id, "Success");
                Console.WriteLine($"Applied migration: {migration.Id}");
            }
            catch (Exception ex)
            {
                await RecordMigrationAsync(migration.Id, "Failed", ex.Message);
                Console.WriteLine($"Failed to apply migration {migration.Id}: {ex.Message}");
                throw; // 终止部署，避免部分执行
            }
        }
    }

    // 回滚指定迁移
    public async Task RollbackMigrationAsync(string migrationId)
    {
        var migration = _migrations.FirstOrDefault(m => m.Id == migrationId);
        if (migration == null)
            throw new InvalidOperationException($"Migration {migrationId} not found");

        var appliedMigration = await _migrationsContainer.ReadItemAsync<MigrationRecord>(migrationId, new PartitionKey(migrationId));
        if (appliedMigration.Resource.Status != "Success")
            throw new InvalidOperationException($"Cannot rollback failed migration {migrationId}");

        try
        {
            await migration.DownAsync(_client);
            await _migrationsContainer.DeleteItemAsync<MigrationRecord>(migrationId, new PartitionKey(migrationId));
            Console.WriteLine($"Rolled back migration: {migrationId}");
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Failed to rollback migration {migrationId}: {ex.Message}");
            throw;
        }
    }

    private async Task<IEnumerable<string>> GetAppliedMigrationsAsync()
    {
        var query = _migrationsContainer.GetItemQueryIterator<MigrationRecord>("SELECT c.id FROM c WHERE c.status = 'Success'");
        var results = new List<string>();
        while (query.HasMoreResults)
        {
            var response = await query.ReadNextAsync();
            results.AddRange(response.Select(r => r.id));
        }
        return results;
    }

    private async Task RecordMigrationAsync(string id, string status, string errorMessage = null)
    {
        var record = new MigrationRecord
        {
            id = id,
            appliedAt = DateTime.UtcNow,
            status = status,
            errorMessage = errorMessage
        };
        await _migrationsContainer.UpsertItemAsync(record, new PartitionKey(id));
    }

    private class MigrationRecord
    {
        public string id { get; set; }
        public DateTime appliedAt { get; set; }
        public string status { get; set; }
        public string errorMessage { get; set; }
    }
}

三、集成到部署流程 #

1. 控制台程序执行：将迁移引擎打包为控制台应用，在部署流水线（如Azure DevOps Pipeline、GitHub Actions）中作为前置步骤执行：

   # GitHub Actions示例步骤
   - name: Run Cosmos Migrations
     run: dotnet run --project ./CosmosMigrations/CosmosMigrations.csproj
     env:
       COSMOS_CONNECTION_STRING: ${{ secrets.COSMOS_CONNECTION_STRING }}
   

2. 应用启动时执行（需谨慎）：在Web应用启动时调用迁移引擎，但需通过配置开关控制（生产环境建议仅手动触发或流水线执行，避免意外变更）：

   // Program.cs示例
   var builder = WebApplication.CreateBuilder(args);
   if (builder.Configuration.GetValue<bool>("RunMigrationsOnStartup"))
   {
       var cosmosClient = new CosmosClient(builder.Configuration["CosmosConnectionString"]);
       var migrations = typeof(Program).Assembly.GetTypes()
           .Where(t => typeof(ICosmosMigration).IsAssignableFrom(t) && !t.IsInterface)
           .Select(Activator.CreateInstance)
           .Cast<ICosmosMigration>();
       var migrationEngine = new CosmosMigrationEngine(cosmosClient, migrations);
       await migrationEngine.ApplyMigrationsAsync();
   }
   

四、最佳实践 #

• 迁移幂等性：确保Up方法可重复执行（如用CreateContainerIfNotExists而非CreateContainer），避免重复执行导致错误。
• Down操作谨慎：涉及数据删除的回滚操作需严格测试，生产环境建议仅回滚结构变更，数据回滚需单独处理。
• 版本控制：每个迁移类需提交到Git，与业务代码同步迭代，避免遗漏变更。
• 环境隔离：在 staging 环境先执行迁移，验证通过后再部署到生产。
• 监控与日志：迁移执行过程需记录详细日志，失败时触发告警，便于快速排查问题。

内容的提问来源于stack exchange，提问作者andrew