AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP.Tests/DataPlatform/DbSyncIsolatedAcceptanceTests.cs

using System.Text.RegularExpressions;
using Admin.NET.Core;
using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Logging.Abstractions;
using SqlSugar;
using Xunit;

namespace Admin.NET.Plugin.AiDOP.Tests.DataPlatform;

/// <summary>
/// 隔离验收：只碰 DBSYNC_TEST_MYSQL / DBSYNC_TEST_SQLSERVER 与 dbsync_test_po。
/// 默认跳过。设置 AIDOP_DBSYNC_ACCEPT=1 后执行。
/// </summary>
public class DbSyncIsolatedAcceptanceTests
{
    private const long MysqlTenant = 990594001;
    private const long SqlServerTenant = 990594002;

    [Fact]
    public async Task Isolated_sources_cover_keyset_update_full_and_reset()
    {
        if (Environment.GetEnvironmentVariable("AIDOP_DBSYNC_ACCEPT") != "1")
            return;

        var root = FindRepoRoot();
        using var db = OpenAidopdev(root);
        await RunSourceAsync(db, "DBSYNC_TEST_MYSQL", MysqlTenant, null);
        await RunNeutralProjectionAsync(db);
        await PrepareSqlServerSourceAsync(db, root);
        await RunSourceAsync(db, "DBSYNC_TEST_SQLSERVER", SqlServerTenant, OpenSqlServer(db, root));
    }

    private static async Task RunNeutralProjectionAsync(ISqlSugarClient db)
    {
        const string entityCode = "DBSYNC_TEST_S3_PURCHASE";
        var source = await db.Queryable<MdpSource>()
            .Where(x => x.SourceCode == "DBSYNC_TEST_MYSQL")
            .FirstAsync();
        var neutral = MdpSourceIdentity.NeutralCode(source.SystemCode, source.SourceCode);

        var oldIds = await db.Queryable<MdpEntity>()
            .Where(x => x.TenantId == MysqlTenant && x.EntityCode == entityCode)
            .Select(x => x.Id)
            .ToListAsync();
        if (oldIds.Count > 0)
            await db.Deleteable<MdpFieldMapping>().Where(x => oldIds.Contains(x.EntityId)).ExecuteCommandAsync();
        await db.Deleteable<MdpEntity>()
            .Where(x => x.TenantId == MysqlTenant && x.EntityCode == entityCode)
            .ExecuteCommandAsync();

        var now = DateTime.Now;
        var entity = new MdpEntity
        {
            TenantId = MysqlTenant,
            SourceId = source.Id,
            EntityCode = entityCode,
            EntityName = "隔离验收采购订单",
            EntityType = "TABLE",
            SourceTableName = "dbsync_test_po",
            TransformSourceTable = "S3_PURCHASE_ORDER",
            TargetTableName = "mdp_stg_purchase_order",
            SyncMode = "FULL",
            TieBreakerColumn = "id",
            BizKeyExpr = "PoNo,PoLine",
            BatchSize = 100,
            JobId = "DBSYNC_TEST_S3",
            Status = 1,
            CreateTime = now,
            UpdateTime = now
        };
        entity.Id = await db.Insertable(entity).ExecuteReturnBigIdentityAsync();
        await db.Insertable(new List<MdpFieldMapping>
        {
            new() { EntityId = entity.Id, SourceField = "bill_no", TargetField = "PoNo", FieldType = "DIRECT", IsRequired = 1, SortOrder = 10, CreateTime = now },
            new() { EntityId = entity.Id, SourceField = "line_no", TargetField = "PoLine", FieldType = "DIRECT", IsRequired = 1, SortOrder = 20, CreateTime = now },
            new() { EntityId = entity.Id, SourceField = "qty", TargetField = "OrderQty", FieldType = "DIRECT", IsRequired = 1, SortOrder = 30, CreateTime = now },
            new() { EntityId = entity.Id, SourceField = "update_time", TargetField = "OrderDate", FieldType = "DIRECT", IsRequired = 1, SortOrder = 40, CreateTime = now },
            new() { EntityId = entity.Id, SourceField = "update_time", TargetField = "DueDate", FieldType = "DIRECT", IsRequired = 1, SortOrder = 50, CreateTime = now },
            new() { EntityId = entity.Id, SourceField = "", TargetField = "SupplierCode", FieldType = "CONST", ConstValue = "TEST_SUPPLIER", SortOrder = 60, CreateTime = now },
            new() { EntityId = entity.Id, SourceField = "", TargetField = "ItemCode", FieldType = "CONST", ConstValue = "TEST_ITEM", SortOrder = 70, CreateTime = now }
        }).ExecuteCommandAsync();

        await db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_tenant_std_source
              (tenant_id, std_object, source_system, source_scope, remark)
            VALUES (@t, 'PURCHASE', @s, 'DBSYNC_ACCEPT', '隔离验收')
            ON DUPLICATE KEY UPDATE source_system=VALUES(source_system), source_scope=VALUES(source_scope), remark=VALUES(remark)
            """,
            new SugarParameter("@t", MysqlTenant),
            new SugarParameter("@s", neutral));
        await db.Ado.ExecuteCommandAsync(
            "DELETE FROM mdp_stg_purchase_order WHERE tenant_id=@t AND source_system=@s AND source_table='S3_PURCHASE_ORDER'",
            new SugarParameter("@t", MysqlTenant),
            new SugarParameter("@s", neutral));
        await db.Ado.ExecuteCommandAsync(
            "DELETE FROM mdp_std_purchase_order WHERE tenant_id=@t AND source_system=@s",
            new SugarParameter("@t", MysqlTenant),
            new SugarParameter("@s", neutral));

        var executor = BuildExecutor(db);
        var pull = await PullAllAsync(executor, db, source, entity.Id, full: true);
        Assert.True(pull.Written >= 29, "neutral staging written=" + pull.Written);

        var resolver = new MdpTenantStdSourceResolver(db);
        var gate = new MdpNeutralSourceGate(resolver, db, NullLoggerFactory.Instance);
        var projection = new InboundNeutralProjectionService(db, gate);
        await projection.ProjectAsync("S3_PURCHASE_ORDER", MysqlTenant, neutral);
        var standardRows = await db.Ado.GetIntAsync(
            "SELECT COUNT(1) FROM mdp_std_purchase_order WHERE tenant_id=@t AND source_system=@s",
            new SugarParameter("@t", MysqlTenant),
            new SugarParameter("@s", neutral));
        Assert.True(standardRows >= 29, "neutral standard rows=" + standardRows);
    }

    private static async Task RunSourceAsync(ISqlSugarClient db, string sourceCode, long tenantId, ISqlSugarClient sourceDb)
    {
        await db.Ado.ExecuteCommandAsync(
            "DELETE FROM mdp_stg_dbsync_test_po WHERE source_system=@s",
            new SugarParameter("@s", sourceCode));
        await db.Ado.ExecuteCommandAsync(
            "UPDATE mdp_entity SET last_cursor=NULL WHERE tenant_id=@t AND entity_code='DBSYNC_TEST_PO'",
            new SugarParameter("@t", tenantId));

        var entity = await db.Queryable<MdpEntity>()
            .Where(x => x.TenantId == tenantId && x.EntityCode == "DBSYNC_TEST_PO" && x.Status == 1)
            .FirstAsync();
        var source = await db.Queryable<MdpSource>().Where(x => x.Id == entity.SourceId).FirstAsync();
        var cursorBefore = entity.LastCursor;

        var previewDb = sourceDb ?? db;
        var preview = await previewDb.Ado.GetIntAsync("SELECT COUNT(1) FROM dbsync_test_po");
        var cursorAfterPreview = await db.Queryable<MdpEntity>().Where(x => x.Id == entity.Id).Select(x => x.LastCursor).FirstAsync();
        Assert.True(preview >= 30);
        Assert.Equal(cursorBefore, cursorAfterPreview);

        var executor = BuildExecutor(db);
        var first = await PullAllAsync(executor, db, source, entity.Id, full: false);
        Assert.True(first.Written >= 29, sourceCode + " written=" + first.Written);
        Assert.True(first.Errors >= 1, sourceCode + " errors=" + first.Errors);

        var staged = await db.Ado.GetIntAsync(
            "SELECT COUNT(1) FROM mdp_stg_dbsync_test_po WHERE source_system=@s AND source_table='dbsync_test_po'",
            new SugarParameter("@s", sourceCode));
        var keys = await db.Ado.GetIntAsync(
            "SELECT COUNT(DISTINCT source_biz_key) FROM mdp_stg_dbsync_test_po WHERE source_system=@s AND source_table='dbsync_test_po'",
            new SugarParameter("@s", sourceCode));
        Assert.Equal(staged, keys);

        await TouchRowAsync(previewDb, source.DbType);
        var second = await PullAllAsync(executor, db, source, entity.Id, full: false);
        Assert.True(second.Written >= 1, sourceCode + " update written=" + second.Written);

        var full = await PullAllAsync(executor, db, source, entity.Id, full: true);
        Assert.True(full.Read >= 30, sourceCode + " full read=" + full.Read);

        var current = await db.Queryable<MdpEntity>().Where(x => x.Id == entity.Id).Select(x => x.LastCursor).FirstAsync();
        await db.Ado.ExecuteCommandAsync(
            "UPDATE mdp_entity SET last_cursor=NULL, update_time=NOW() WHERE id=@id AND ((last_cursor IS NULL AND @old IS NULL) OR last_cursor=@old)",
            new SugarParameter("@id", entity.Id),
            new SugarParameter("@old", current));
        await db.Ado.ExecuteCommandAsync(
            @"INSERT INTO mdp_entity_cursor_audit
              (tenant_id, entity_id, entity_code, old_cursor, new_cursor, action_type, reason, create_time)
              VALUES (@t, @id, @code, @old, NULL, 'RESET', '隔离验收重置水位', NOW())",
            new SugarParameter("@t", tenantId),
            new SugarParameter("@id", entity.Id),
            new SugarParameter("@code", entity.EntityCode),
            new SugarParameter("@old", current));

        var third = await PullAllAsync(executor, db, source, entity.Id, full: false);
        Assert.True(third.Read >= 30, sourceCode + " reset read=" + third.Read);

        var logs = await db.Ado.GetIntAsync(
            "SELECT COUNT(1) FROM mdp_sync_log WHERE source_code=@s AND entity_id=@id",
            new SugarParameter("@s", sourceCode),
            new SugarParameter("@id", entity.Id));
        Assert.True(logs >= 3, sourceCode + " logs=" + logs);
    }

    private static async Task TouchRowAsync(ISqlSugarClient sourceDb, string dbType)
    {
        var sql = dbType != null && dbType.Replace("_", "").Contains("SQLSERVER", StringComparison.OrdinalIgnoreCase)
            ? "UPDATE dbsync_test_po SET qty=qty+1, update_time=DATEADD(second,1,(SELECT MAX(update_time) FROM dbsync_test_po)) WHERE id=2"
            : "UPDATE dbsync_test_po SET qty=qty+1, update_time=(SELECT next_time FROM (SELECT DATE_ADD(MAX(update_time), INTERVAL 1 SECOND) next_time FROM dbsync_test_po) x) WHERE id=2";
        await sourceDb.Ado.ExecuteCommandAsync(sql);
    }

    private static async Task<(int Read, int Written, int Errors)> PullAllAsync(
        MdpDbPullExecutor executor, ISqlSugarClient db, MdpSource source, long entityId, bool full)
    {
        var read = 0;
        var written = 0;
        var errors = 0;
        var ctx = new MdpPullContext
        {
            TenantId = source.SourceCode == "DBSYNC_TEST_SQLSERVER" ? SqlServerTenant : MysqlTenant,
            FullRefresh = full,
            SyncWindowType = full ? "FULL" : "INCR",
            RequireConfiguredDbSync = true,
            TriggerType = "MANUAL",
            BatchId = "DBSYNC_ACCEPT_" + DateTime.Now.ToString("yyyyMMddHHmmssfff")
        };
        for (var page = 0; page < 20; page++)
        {
            var entity = await db.Queryable<MdpEntity>().Where(x => x.Id == entityId).FirstAsync();
            if (!full)
            {
                ctx.UseKeysetCursor = false;
                ctx.CursorValue = null;
                ctx.TieBreakerValue = null;
                ctx.Offset = 0;
            }
            MdpSourcePullDispatcher.ApplyDbSyncDefaults(entity, source, ctx);
            var result = await executor.PullAsync(source, entity, ctx);
            read += result.RowsPulled;
            written += result.RowsWritten;
            errors += result.RowsError;
            if (result.RowsPulled <= 0 || result.RowsPulled < entity.BatchSize)
                break;
            if (full)
                ctx.Offset += result.RowsPulled;
        }
        return (read, written, errors);
    }

    private static MdpDbPullExecutor BuildExecutor(ISqlSugarClient db)
    {
        var passwords = new MdpSourcePasswordResolver(NullLogger<MdpSourcePasswordResolver>.Instance);
        var scopes = new MdpSourceScopeFactory(db, passwords, NullLogger<MdpSourceScopeFactory>.Instance);
        var maps = new MdpPullFieldMappingService(db, new MemoryCache(new MemoryCacheOptions()));
        var writer = new MdpStagingWriter(db, NullLogger<MdpStagingWriter>.Instance);
        return new MdpDbPullExecutor(scopes, db, writer, maps);
    }

    private static async Task PrepareSqlServerSourceAsync(ISqlSugarClient mysql, string root)
    {
        using var db = OpenSqlServer(mysql, root);
        await db.Ado.ExecuteCommandAsync("""
            IF OBJECT_ID('dbo.dbsync_test_po','U') IS NULL
            CREATE TABLE dbo.dbsync_test_po (
              id bigint NOT NULL PRIMARY KEY,
              bill_no nvarchar(40) NULL,
              line_no int NOT NULL,
              qty decimal(18,4) NOT NULL,
              update_time datetime2(3) NOT NULL,
              row_status nvarchar(20) NOT NULL)
            """);
        await db.Ado.ExecuteCommandAsync("DELETE FROM dbo.dbsync_test_po");
        await db.Ado.ExecuteCommandAsync("""
            INSERT INTO dbo.dbsync_test_po (id, bill_no, line_no, qty, update_time, row_status) VALUES
            (1,'PO-001',1,1.5,'2026-10-01T08:00:00','OPEN'),
            (2,'PO-002',2,2.5,'2026-10-01T08:00:00','OPEN'),
            (3,'PO-003',3,3.5,'2026-10-01T08:00:00','OPEN'),
            (4,'PO-004',4,4.5,'2026-10-01T08:00:00','OPEN'),
            (5,'PO-005',5,5.5,'2026-10-01T08:00:00','DELETED'),
            (6,'PO-006',6,6.5,'2026-10-01T08:00:00','OPEN'),
            (7,'PO-007',7,7.5,'2026-10-01T08:00:00','OPEN'),
            (8,'PO-008',8,8.5,'2026-10-01T08:00:00','OPEN'),
            (9,'PO-009',9,9.5,'2026-10-01T08:00:00','OPEN'),
            (10,'PO-010',10,10.5,'2026-10-01T08:00:00','OPEN'),
            (11,'PO-011',11,11.5,'2026-10-02T08:00:00','OPEN'),
            (12,'PO-012',12,12.5,'2026-10-02T08:00:00','OPEN'),
            (13,'PO-013',13,13.5,'2026-10-02T08:00:00','OPEN'),
            (14,'PO-014',14,14.5,'2026-10-02T08:00:00','OPEN'),
            (15,'PO-015',15,15.5,'2026-10-02T08:00:00','OPEN'),
            (16,'PO-016',16,16.5,'2026-10-02T08:00:00','OPEN'),
            (17,'PO-017',17,17.5,'2026-10-02T08:00:00','OPEN'),
            (18,'PO-018',18,18.5,'2026-10-02T08:00:00','OPEN'),
            (19,'PO-019',19,19.5,'2026-10-02T08:00:00','OPEN'),
            (20,'PO-020',20,20.5,'2026-10-02T08:00:00','OPEN'),
            (21,'PO-021',21,21.5,'2026-10-03T08:00:00','OPEN'),
            (22,'PO-022',22,22.5,'2026-10-03T08:00:00','OPEN'),
            (23,'PO-023',23,23.5,'2026-10-03T08:00:00','OPEN'),
            (24,'PO-024',24,24.5,'2026-10-03T08:00:00','OPEN'),
            (25,'PO-025',25,25.5,'2026-10-03T08:00:00','OPEN'),
            (26,'PO-026',26,26.5,'2026-10-03T08:00:00','OPEN'),
            (27,'PO-027',27,27.5,'2026-10-03T08:00:00','OPEN'),
            (28,'PO-028',28,28.5,'2026-10-03T08:00:00','OPEN'),
            (29,'PO-029',29,29.5,'2026-10-03T08:00:00','OPEN'),
            (30,NULL,30,30.5,'2026-10-03T08:00:00','OPEN')
            """);
    }

    private static SqlSugarClient OpenSqlServer(ISqlSugarClient mysql, string root)
    {
        var cipher = mysql.Ado.GetString("SELECT db_password_enc FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'");
        var key = Regex.Match(File.ReadAllText(Path.Combine(root, "server", "Admin.NET.Application", "Configuration", "App.json")),
            "\"PrivateKey\"\\s*:\\s*\"([^\"]+)\"").Groups[1].Value;
        string password;
        try
        {
            password = GMUtil.SM2Decrypt(key, cipher);
        }
        catch (Exception)
        {
            password = cipher;
        }
        Environment.SetEnvironmentVariable(MdpSourcePasswordResolver.EnvironmentVariableName("DBSYNC_TEST_SQLSERVER"), password);
        var host = mysql.Ado.GetString("SELECT db_host FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'");
        var port = mysql.Ado.GetInt("SELECT db_port FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'");
        var name = mysql.Ado.GetString("SELECT db_name FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'");
        var user = mysql.Ado.GetString("SELECT db_user FROM mdp_source WHERE source_code='DBSYNC_TEST_SQLSERVER'");
        return new SqlSugarClient(new ConnectionConfig
        {
            ConfigId = "dbsync-test-sqlserver",
            DbType = DbType.SqlServer,
            ConnectionString = $"Server={host},{port};Database={name};User Id={user};Password={password};TrustServerCertificate=true;Encrypt=false",
            IsAutoCloseConnection = true
        });
    }

    private static SqlSugarScope OpenAidopdev(string root)
    {
        var cs = File.ReadLines(Path.Combine(root, "server", "Admin.NET.Application", "Configuration", "Database.json"))
            .Select(x => x.Trim())
            .Where(x => !x.StartsWith("//") && x.Contains("\"ConnectionString\"") && x.Contains("Database=aidopdev"))
            .Select(x => Regex.Match(x, "\"ConnectionString\"\\s*:\\s*\"([^\"]+)\"").Groups[1].Value)
            .First(x => !string.IsNullOrEmpty(x));
        return new SqlSugarScope(new ConnectionConfig
        {
            ConfigId = "dbsync-accept",
            DbType = DbType.MySql,
            ConnectionString = cs,
            IsAutoCloseConnection = true
        });
    }

    private static string FindRepoRoot()
    {
        var dir = new DirectoryInfo(AppContext.BaseDirectory);
        while (dir != null)
        {
            if (File.Exists(Path.Combine(dir.FullName, "server", "Admin.NET.Application", "Configuration", "Database.json")))
                return dir.FullName;
            dir = dir.Parent;
        }
        throw new InvalidOperationException("未找到 Database.json");
    }
}