build-flow.api/src/BuildFlow.Host/Configuration/DatabaseInitializer.cs

using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Platform.Infrastructure.Persistence;
using File.Infrastructure.Persistence;
using IM.Infrastructure.Mongo;
using IM.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore.Diagnostics;
using MongoDB.Entities;
using Monitoring.Infrastructure.Persistence;
using Scripting.Infrastructure.Persistence;
using Workflow.Infrastructure.Persistence;
using PlatformSeedData = BuildFlow.Platform.Infrastructure.Persistence.SeedData;
using WorkflowSeedData = Workflow.Infrastructure.Persistence.SeedData;

namespace BuildFlow.Host.Configuration;

/// <summary>
/// 启动期数据库初始化：按固定顺序完成「迁移 → IM MongoDB 建索引 → 种子数据」。
/// 容错策略：单个 DbContext / 单个步骤失败只记录警告并跳过，不影响其他库与主流程启动，
/// 避免任一外部库抖动拖垮整个 Host。日志走 <see cref="ILogger"/>（可被 OTel 采集），
/// 输出格式与 <see cref="Services.BackgroundServices"/> 一致。
/// </summary>
public static class DatabaseInitializer
{
    public static async Task RunAsync(IServiceProvider rootProvider)
    {
        using var scope = rootProvider.CreateScope();
        var sp = scope.ServiceProvider;
        var logger = sp.GetRequiredService<ILoggerFactory>().CreateLogger(nameof(DatabaseInitializer));

        // 1. 各模块 DbContext 自动迁移（顺序：rag → platform → file → im → workflow）
        //    workflow 传 suppressPendingChanges:true：该模块注册 DbContext 时未抑制
        //    PendingModelChangesWarning（有未提交的模型变更但不影响已有 migration），单独处理避免告警。
        await MigrateAsync<RagDbContext>(sp, logger);

        await MigrateAsync<PlatformDbContext>(sp, logger);
        await MigrateAsync<FileDbContext>(sp, logger);
        await MigrateAsync<ImDbContext>(sp, logger);
        await MigrateAsync<WorkflowDbContext>(sp, logger, suppressPendingChanges: true);

        // Scripting 与 Workflow 共用 workflow 库，迁移创建 funcs / func_execution_logs 表
        await MigrateAsync<ScriptingDbContext>(sp, logger, suppressPendingChanges: true);

        await MigrateAsync<MonitoringDbContext>(sp, logger);

        // IM 列补丁：AddPinAndPinnedAtColumns 迁移因 ImDbContext Snapshot 与实体不匹配
        // （PendingModelChangesWarning）被跳过，导致 b_conversation_participants 缺 is_pinned/pinned_at 列，
        // /conversations 查询 500。这里用幂等 raw SQL 兜底补列，迁移修复后可移除。
        await SafeRunAsync(async () =>
        {
            var imDb = sp.GetRequiredService<ImDbContext>();
            await imDb.Database.ExecuteSqlRawAsync(
                "ALTER TABLE b_conversation_participants ADD COLUMN IF NOT EXISTS is_pinned boolean NOT NULL DEFAULT false");
            await imDb.Database.ExecuteSqlRawAsync(
                "ALTER TABLE b_conversation_participants ADD COLUMN IF NOT EXISTS pinned_at timestamp NULL");
        }, logger, "IM is_pinned/pinned_at 列补丁");

        // 2. IM MongoDB 建索引（MongoDB.Entities：用 DB.Index<MessageDocument> 创建两个复合索引）
        //    两个复合索引支撑两种查询路径：按时间倒序拉取最新消息、按 ObjectId 游标分页上拉更早消息。
        await SafeRunAsync(
            async () =>
            {
                await DB.Default.Index<MessageDocument>()
                    .Key(m => m.ConversationId, KeyType.Ascending)
                    .Key(m => m.CreatedAt, KeyType.Descending)
                    .CreateAsync();
                await DB.Default.Index<MessageDocument>()
                    .Key(m => m.ConversationId, KeyType.Ascending)
                    .Key(m => m.ID, KeyType.Descending)
                    .CreateAsync();
            },
            logger,
            "IM MongoDB 建索引");

        // 3. 种子数据：平台（权限/角色/用户/菜单）+ workflow（组件注册）
        await SafeRunAsync(
            async () =>
            {
                var platformDb = sp.GetRequiredService<PlatformDbContext>();
                await PlatformSeedData.SeedAsync(platformDb);

                var wfDb = sp.GetRequiredService<WorkflowDbContext>();
                await WorkflowSeedData.SeedAsync(wfDb);

                // 把存量关系表（nodes/edges）迁移为 X6 JSON 单一事实源（幂等，已迁移的跳过）
                var (defs, snaps, insts) = await WorkflowGraphMigration.RunAsync(wfDb);
                if (defs > 0 || snaps > 0 || insts > 0)
                    logger.LogInformation("Workflow graph migration: {Defs} definitions, {Snaps} snapshots, {Insts} instances backfilled", defs, snaps, insts);
            },
            logger,
            "Seed");
    }

    /// <summary>
    /// 迁移单个 DbContext，失败仅记录警告。<paramref name="suppressPendingChanges"/> 为 true 时，
    /// 用独立的 options（抑制 PendingModelChangesWarning）新建临时 DbContext 完成迁移，避免污染主注册。
    /// </summary>
    private static async Task MigrateAsync<T>(
        IServiceProvider sp,
        ILogger logger,
        bool suppressPendingChanges = false)
        where T : DbContext
    {
        try
        {
            if (suppressPendingChanges)
            {
                var db = sp.GetRequiredService<T>();
                var options = new DbContextOptionsBuilder<T>()
                    .UseNpgsql(db.Database.GetConnectionString())
                    .UseSnakeCaseNamingConvention()
                    .ConfigureWarnings(w => w.Ignore(RelationalEventId.PendingModelChangesWarning))
                    .Options;
                await using var tmp = (T)Activator.CreateInstance(typeof(T), options)!;
                await tmp.Database.MigrateAsync();
            }
            else
            {
                await sp.GetRequiredService<T>().Database.MigrateAsync();
            }
            logger.LogInformation("[ok] {Name} migrated", typeof(T).Name);
        }
        catch (Exception ex)
        {
            logger.LogWarning(ex, "[warn] {Name} migrate 跳过: {Message}", typeof(T).Name, ex.Message);
        }
    }

    /// <summary>包裹一个启动步骤，失败仅记录警告（含步骤名）不中断主流程。</summary>
    private static async Task SafeRunAsync(Func<Task> action, ILogger logger, string stepName)
    {
        try
        {
            await action();
        }
        catch (Exception ex)
        {
            logger.LogWarning(ex, "[warn] {Step} 跳过: {Message}", stepName, ex.Message);
        }
    }
}