fix(mdp): 修正 conn_mode 回填遗漏与贴源批量写入的参数重名 | server 1.0.582

一、conn_mode（1.0.582.sql / verify）

1.0.576（F-4）的回填条件是 `IFNULL(db_host,'')='' AND system_code IN (四个原生别名)`。
AIDOP 原生库那一行（source_code=AIDOPDEV_MYSQL / system_code=AIDOP_NATIVE）的 db_host
早已被填成主库自己的地址，于是回填一行都没匹配上（aidopdev 执行于 09-25 08:11，
affected_rows=0），该行留在默认值 EXTERNAL。

MdpSourceScopeFactory.OpenScope 对 EXTERNAL 会强校验 db_host/db_name/db_user 齐全且
明确禁止回退主库，而该行 db_user 为空，于是挂在它下面的 83 个启用实体从 09-25 起
全部同步不了：S7 FQC 结果列表 9/14 后无新单，S2/S3/S4 重算作业在 STAGING 阶段失败。

原生库在任何环境下都是主库本身，conn_mode 恒为 SELF，与 db_host 填没填无关，
故去掉 db_host 条件重跑回填。verify 断言「不存在原生别名却非 SELF 的行」，
每次部署都会执行，即防止同类回填遗漏再犯的守卫。

二、贴源批量写入参数重名（MdpStagingWriter）

修完 conn_mode 后露出第二个缺陷：`Parameter '@now20' has already been defined.`。
BuildUpsertSql 的参数名是「基名 + 行号」直接拼接，第 20 行的 @now 与第 0 行的 @now2
都得到 @now20，凑满 21 行的批次必炸。原有的重名断言只跑 3 行，够不到这个边界。

改为「基名 + 下划线 + 行号」：行号是纯数字，按最后一个下划线切分即可唯一还原
(基名, 行号)，基名含下划线也不冲突。补 BuildUpsertSql_ManyRows_KeepsParameterNamesUnique
压 25 行批次；已验证改回旧写法该用例转红。

本次提交一并纳入贴源批量写入（多值 INSERT ... ON DUPLICATE KEY UPDATE）的实现本体，
此前一直停在工作区未提交。

server/Admin.NET.Web.Entry/Admin.NET.Web.Entry.csproj

@@ -11,9 +11,9 @@
     <GenerateSatelliteAssembliesForCore>true</GenerateSatelliteAssembliesForCore>
     <Copyright>Admin.NET</Copyright>
     <Description>Admin.NET 通用权限开发平台</Description>
-    <AssemblyVersion>1.0.581</AssemblyVersion>
-    <FileVersion>1.0.581</FileVersion>
-    <Version>1.0.581</Version>
+    <AssemblyVersion>1.0.582</AssemblyVersion>
+    <FileVersion>1.0.582</FileVersion>
+    <Version>1.0.582</Version>
   </PropertyGroup>
 
   <ItemGroup>
@@ -940,6 +940,12 @@
     <None Update="UpdateScripts\1.0.581.verify.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
+    <None Update="UpdateScripts\1.0.582.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Update="UpdateScripts\1.0.582.verify.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
     <None Update="UpdateScripts\UAT-PLACEHOLDER-MENU-HIDE.ops.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>

server/Admin.NET.Web.Entry/UpdateScripts/1.0.582.sql

@@ -0,0 +1,20 @@
+-- 1.0.582 —— 修正 1.0.576（F-4）的 conn_mode 回填规则。
+-- 背景：1.0.576 的回填条件是 `IFNULL(db_host,'')='' AND system_code IN (四个原生别名)`。
+--       但 AIDOP 原生库那一行（source_code=AIDOPDEV_MYSQL / system_code=AIDOP_NATIVE）的
+--       db_host 早已被填成主库自己的地址，于是回填一行都没匹配上（1.0.576 在 aidopdev
+--       执行于 2026-09-25 08:11，affected_rows=0），该行留在默认值 EXTERNAL。
+--       MdpSourceScopeFactory.OpenScope 对 EXTERNAL 会强校验 db_host/db_name/db_user 齐全
+--       且明确禁止回退主库，而该行 db_user 为空 —— 结果挂在它下面的 83 个启用实体
+--       从 09-25 起全部同步不了（S7 FQC 结果列表无新单、S4 重算作业 STAGING 阶段失败）。
+--
+-- 修法：原生库在任何环境下都是主库本身，conn_mode 恒为 SELF，与 db_host 填没填无关。
+--       故这里去掉 db_host 条件重跑一次回填。
+--       这四个编码仍然只允许出现在迁移脚本里，运行期代码不得按它们分支。
+--
+-- 幂等：命中 0 行即无副作用，可重复执行。
+
+UPDATE mdp_source
+   SET conn_mode = 'SELF'
+ WHERE conn_mode <> 'SELF'
+   AND UPPER(IFNULL(NULLIF(system_code, ''), source_code)) IN
+       ('AIDOP_NATIVE', 'AIDOPDEV_MYSQL', 'LOCAL_MYSQL', 'AIDOP_MYSQL');

server/Admin.NET.Web.Entry/UpdateScripts/1.0.582.verify.sql

@@ -0,0 +1,9 @@
+-- 表尚未建成时放行。否则要求：不存在「系统码属 AIDOP 原生别名、却不是 SELF」的来源行。
+-- 这条断言就是防止 1.0.576 那类回填遗漏再次发生的守卫，每次部署都会执行。
+SELECT
+  (SELECT COUNT(*) FROM information_schema.TABLES
+    WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='mdp_source') = 0
+  OR (SELECT COUNT(*) FROM mdp_source
+       WHERE conn_mode <> 'SELF'
+         AND UPPER(IFNULL(NULLIF(system_code, ''), source_code)) IN
+             ('AIDOP_NATIVE', 'AIDOPDEV_MYSQL', 'LOCAL_MYSQL', 'AIDOP_MYSQL')) = 0;

server/Plugins/Admin.NET.Plugin.AiDOP.Tests/DataPlatform/MdpStagingWriterContractTests.cs

@@ -55,4 +55,99 @@ public class MdpStagingWriterContractTests
         Assert.Null(MdpStagingWriter.NormalizeFactoryId(tenantId, 0));
         Assert.Equal(8010, MdpStagingWriter.NormalizeFactoryId(tenantId, 8010));
     }
+
+    private static MdpStagingWriter.MdpStagingUpsertRow Row(int n) => new(
+        TenantId: 9,
+        FactoryId: 1,
+        CompanyId: null,
+        SourceSystem: "AIDOPDEV_MYSQL",
+        SourceTable: "src",
+        SourceRowId: $"r{n}",
+        BizKey: $"b{n}",
+        RawJson: "{}",
+        BatchId: "batch");
+
+    [Fact]
+    public void BuildUpsertSql_SingleRow_MatchesLegacyShape()
+    {
+        var columns = CanonicalColumns.ToHashSet(StringComparer.OrdinalIgnoreCase);
+        var (sql, parameters) = MdpStagingWriter.BuildUpsertSql("mdp_stg_so", columns, [Row(1)], DateTime.Now);
+
+        Assert.Contains("ON DUPLICATE KEY UPDATE", sql);
+        Assert.Contains("process_status='PENDING'", sql);
+        Assert.Contains("update_time=VALUES(create_time)", sql);
+        Assert.Contains("`tenant_id`, `factory_id`, `source_system`", sql);
+        Assert.Equal(10, parameters.Count);
+    }
+
+    [Fact]
+    public void BuildUpsertSql_ThreeRows_ProducesThreeValueTuples()
+    {
+        var columns = CanonicalColumns.ToHashSet(StringComparer.OrdinalIgnoreCase);
+        var (sql, parameters) = MdpStagingWriter.BuildUpsertSql(
+            "mdp_stg_so", columns, [Row(1), Row(2), Row(3)], DateTime.Now);
+
+        var valuesAt = sql.IndexOf("VALUES", StringComparison.Ordinal);
+        var updateAt = sql.IndexOf("ON DUPLICATE KEY UPDATE", StringComparison.Ordinal);
+        var tuples = sql[valuesAt..updateAt].Count(c => c == '(');
+        Assert.Equal(3, tuples);
+        Assert.Equal(30, parameters.Count);
+        Assert.Equal(parameters.Count, parameters.Select(p => p.ParameterName).Distinct().Count());
+    }
+
+    /// <summary>
+    /// 参数名曾用「基名 + 行号」直接拼接，第 20 行的 @now 与第 0 行的 @now2 都是 @now20，
+    /// 凑满 21 行的批次就抛「Parameter '@now20' has already been defined」。
+    /// 原有的重名断言只跑 3 行，够不到这个边界，所以这里单独压一个超过 21 行的批次。
+    /// </summary>
+    [Fact]
+    public void BuildUpsertSql_ManyRows_KeepsParameterNamesUnique()
+    {
+        var columns = CanonicalColumns.ToHashSet(StringComparer.OrdinalIgnoreCase);
+        var rows = Enumerable.Range(1, 25).Select(Row).ToList();
+
+        var (_, parameters) = MdpStagingWriter.BuildUpsertSql("mdp_stg_so", columns, rows, DateTime.Now);
+
+        var duplicates = parameters
+            .GroupBy(p => p.ParameterName)
+            .Where(g => g.Count() > 1)
+            .Select(g => g.Key)
+            .ToList();
+
+        Assert.Empty(duplicates);
+    }
+
+    [Fact]
+    public void BuildUpsertSql_OmitsColumnsMissingFromTable()
+    {
+        var columns = CanonicalColumns
+            .Where(x => x != "factory_id")
+            .ToHashSet(StringComparer.OrdinalIgnoreCase);
+
+        var (sql, _) = MdpStagingWriter.BuildUpsertSql("mdp_stg_so", columns, [Row(1)], DateTime.Now);
+
+        Assert.DoesNotContain("`factory_id`", sql);
+        Assert.DoesNotContain("factory_id=VALUES(factory_id)", sql);
+    }
+
+    [Fact]
+    public void UpsertBatchRows_IsWithinMySqlPlaceholderLimit()
+        => Assert.True(MdpStagingWriter.UpsertBatchRows * 12 < 65535);
+
+    [Fact]
+    public void DbPullExecutor_KeepsCursorAdvanceInPerRowLoop()
+    {
+        var dir = new DirectoryInfo(AppContext.BaseDirectory);
+        while (dir != null && !File.Exists(Path.Combine(dir.FullName, "AGENTS.md")))
+            dir = dir.Parent;
+        Assert.NotNull(dir);
+        var source = File.ReadAllText(Path.Combine(
+            dir!.FullName, "server", "Plugins", "Admin.NET.Plugin.AiDOP", "DataPlatform", "Executors", "MdpDbPullExecutor.cs"));
+
+        var loopAt = source.IndexOf("foreach (DataRow row in table.Rows)", StringComparison.Ordinal);
+        var batchAt = source.IndexOf("UpsertBatchAsync(", StringComparison.Ordinal);
+        var cursorAt = source.IndexOf("maxCursor = cursor;", StringComparison.Ordinal);
+        Assert.True(loopAt >= 0 && cursorAt > loopAt && batchAt > cursorAt,
+            "游标推进必须留在逐行循环内、批量写入之前");
+    }
 }

server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Executors/MdpStagingWriter.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using Admin.NET.Plugin.AiDOP.DataPlatform;
 using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
 using Microsoft.Extensions.Logging;
 using SqlSugar;
@@ -17,9 +18,27 @@ public sealed class MdpStagingWriter : ITransient
         "raw_data", "sync_batch_id", "sync_time", "process_status", "create_time", "update_time"
     ];
 
+    /// <summary>
+    /// 单条多值 INSERT 的最大行数。每行约 12 个占位符，200 行约 2400 个，
+    /// 远低于 MySQL 的 65535 上限，同时让单条语句的 redo 量保持可控。
+    /// </summary>
+    internal const int UpsertBatchRows = 200;
+
     private readonly ISqlSugarClient _db;
     private readonly ILogger<MdpStagingWriter> _logger;
 
+    /// <summary>一行已解析好、可直接拼进 SQL 的贴源值。</summary>
+    internal sealed record MdpStagingUpsertRow(
+        long TenantId,
+        long? FactoryId,
+        long? CompanyId,
+        string SourceSystem,
+        string SourceTable,
+        string SourceRowId,
+        string BizKey,
+        string RawJson,
+        string BatchId);
+
     public MdpStagingWriter(ISqlSugarClient db, ILogger<MdpStagingWriter> logger)
     {
         _db = db;
@@ -78,47 +97,148 @@ public sealed class MdpStagingWriter : ITransient
         }
         var companyValue = TryParseOptionalLong(row, "company_id");
 
-        var insertCols = new List<string>();
-        var insertVals = new List<string>();
-        var parameters = new List<SugarParameter>();
+        var resolved = new MdpStagingUpsertRow(
+            tenantValue.Value,
+            factoryValue,
+            companyValue,
+            MdpSourceIdentity.NeutralCode(source.SystemCode, source.SourceCode),
+            sourceTable,
+            sourceRowId,
+            bizKey,
+            rawJson,
+            ctx.BatchId);
+
+        var (sql, parameters) = BuildUpsertSql(table, columns, [resolved], now);
+        return await _db.Ado.ExecuteCommandAsync(sql, parameters);
+    }
 
-        if (columns.Contains("tenant_id"))
-        {
-            insertCols.Add("tenant_id");
-            insertVals.Add("@tenant");
-            parameters.Add(new SugarParameter("@tenant", tenantValue.Value));
-        }
+    /// <summary>
+    /// 同表、同实体、同批次的多行一次写入。返回与入参等长的逐行结果（0 表示租户无法解析或作用域不匹配已跳过）。
+    /// 超过 <see cref="UpsertBatchRows"/> 时切成多条语句，不把整批拼成一条。
+    /// </summary>
+    public async Task<IReadOnlyList<int>> UpsertBatchAsync(
+        MdpSource source,
+        MdpEntity entity,
+        string sourceTable,
+        IReadOnlyList<(IDictionary<string, object?> Row, string RawJson, string SourceRowId)> rows,
+        MdpPullContext ctx)
+    {
+        var results = new int[rows.Count];
+        if (rows.Count == 0)
+            return results;
+
+        if (string.IsNullOrWhiteSpace(entity.TargetTableName))
+            throw new InvalidOperationException($"实体 {entity.EntityCode} 未配置 target_table_name");
 
-        if (columns.Contains("factory_id"))
+        var table = entity.TargetTableName!.Trim();
+        if (!System.Text.RegularExpressions.Regex.IsMatch(table, @"^[A-Za-z0-9_]+$"))
+            throw new InvalidOperationException($"非法 target_table_name：{table}");
+
+        var columns = await GetTableColumnsAsync(table);
+        ValidateTableContract(table, entity.EntityCode, columns);
+        var now = DateTime.Now;
+        var sourceSystem = MdpSourceIdentity.NeutralCode(source.SystemCode, source.SourceCode);
+
+        var pending = new List<(int Index, MdpStagingUpsertRow Row)>();
+        for (var i = 0; i < rows.Count; i++)
         {
-            insertCols.Add("factory_id");
-            insertVals.Add("@factory");
-            parameters.Add(new SugarParameter("@factory", factoryValue.HasValue ? factoryValue.Value : DBNull.Value));
+            var (row, rawJson, sourceRowId) = rows[i];
+            var tenantValue = ResolveTenantId(row, ctx);
+            if (tenantValue is null)
+            {
+                _logger.LogWarning(
+                    "贴源写入跳过：无法解析 tenant_id，entity={EntityCode}, source={SourceCode}, table={Table}, rowId={RowId}",
+                    entity.EntityCode, source.SourceCode, table, sourceRowId);
+                continue;
+            }
+            if (ctx.TenantId > 0 && tenantValue.Value != ctx.TenantId)
+                continue;
+
+            var factoryValue = NormalizeFactoryId(
+                tenantValue.Value,
+                TryParseOptionalLong(row, "factory_id") ?? TryParseOptionalLong(row, "FactoryId"));
+            if (ctx.FactoryId > 0)
+            {
+                var resolvedFactory = factoryValue is > 0 ? factoryValue.Value : 1;
+                if (resolvedFactory != ctx.FactoryId)
+                    continue;
+                factoryValue = resolvedFactory;
+            }
+
+            pending.Add((i, new MdpStagingUpsertRow(
+                tenantValue.Value,
+                factoryValue,
+                TryParseOptionalLong(row, "company_id"),
+                sourceSystem,
+                sourceTable,
+                sourceRowId,
+                BuildBizKey(entity.BizKeyExpr, row) ?? sourceRowId,
+                rawJson,
+                ctx.BatchId)));
         }
 
-        if (columns.Contains("company_id"))
+        for (var offset = 0; offset < pending.Count; offset += UpsertBatchRows)
         {
-            insertCols.Add("company_id");
-            insertVals.Add("@company");
-            parameters.Add(new SugarParameter("@company", companyValue.HasValue ? companyValue.Value : DBNull.Value));
+            var chunk = pending.Skip(offset).Take(UpsertBatchRows).ToList();
+            var (sql, parameters) = BuildUpsertSql(table, columns, chunk.Select(x => x.Row).ToList(), now);
+            await _db.Ado.ExecuteCommandAsync(sql, parameters);
+            foreach (var (index, _) in chunk)
+                results[index] = 1;
         }
 
+        return results;
+    }
+
+    /// <summary>
+    /// 拼一条多值 INSERT ... ON DUPLICATE KEY UPDATE。
+    /// 参数名是「基名 + 下划线 + 行号」。下划线不可省：行号是纯数字，直接拼接会让
+    /// <c>@now</c> 的第 20 行与 <c>@now2</c> 的第 0 行都得到 <c>@now20</c>，
+    /// 凑满 21 行的批次就抛「Parameter '@now20' has already been defined」。
+    /// 加了分隔符后，按最后一个下划线切分即可唯一还原 (基名, 行号)，基名含下划线也不冲突。
+    /// </summary>
+    internal static (string Sql, List<SugarParameter> Parameters) BuildUpsertSql(
+        string table,
+        IReadOnlySet<string> columns,
+        IReadOnlyList<MdpStagingUpsertRow> rows,
+        DateTime now)
+    {
+        var insertCols = new List<string>();
+        if (columns.Contains("tenant_id")) insertCols.Add("tenant_id");
+        if (columns.Contains("factory_id")) insertCols.Add("factory_id");
+        if (columns.Contains("company_id")) insertCols.Add("company_id");
         insertCols.AddRange(
         [
             "source_system", "source_table", "source_row_id", "source_biz_key",
             "raw_data", "sync_batch_id", "sync_time", "process_status", "create_time"
         ]);
-        insertVals.AddRange(["@sys", "@tbl", "@rid", "@biz", "@raw", "@batch", "@now", "'PENDING'", "@now"]);
-        parameters.AddRange(
-        [
-            new SugarParameter("@sys", source.SourceCode),
-            new SugarParameter("@tbl", sourceTable),
-            new SugarParameter("@rid", sourceRowId),
-            new SugarParameter("@biz", bizKey),
-            new SugarParameter("@raw", rawJson),
-            new SugarParameter("@batch", ctx.BatchId),
-            new SugarParameter("@now", now)
-        ]);
+
+        var parameters = new List<SugarParameter>();
+        var valueGroups = new List<string>();
+        for (var i = 0; i < rows.Count; i++)
+        {
+            var row = rows[i];
+            var vals = new List<string>();
+            void Add(string name, object? value)
+            {
+                var p = $"{name}_{i}";
+                vals.Add(p);
+                parameters.Add(new SugarParameter(p, value ?? DBNull.Value));
+            }
+
+            if (columns.Contains("tenant_id")) Add("@tenant", row.TenantId);
+            if (columns.Contains("factory_id")) Add("@factory", row.FactoryId);
+            if (columns.Contains("company_id")) Add("@company", row.CompanyId);
+            Add("@sys", row.SourceSystem);
+            Add("@tbl", row.SourceTable);
+            Add("@rid", row.SourceRowId);
+            Add("@biz", row.BizKey);
+            Add("@raw", row.RawJson);
+            Add("@batch", row.BatchId);
+            Add("@now", now);
+            vals.Add("'PENDING'");
+            Add("@now2", now);
+            valueGroups.Add($"({string.Join(", ", vals)})");
+        }
 
         var updateParts = new List<string>
         {
@@ -128,26 +248,20 @@ public sealed class MdpStagingWriter : ITransient
             "sync_time=VALUES(sync_time)",
             "process_status='PENDING'"
         };
-
-        if (columns.Contains("tenant_id"))
-            updateParts.Add("tenant_id=VALUES(tenant_id)");
-        if (columns.Contains("factory_id"))
-            updateParts.Add("factory_id=VALUES(factory_id)");
-        if (columns.Contains("company_id"))
-            updateParts.Add("company_id=VALUES(company_id)");
-
-        updateParts.Add("update_time=@now");
+        if (columns.Contains("tenant_id")) updateParts.Add("tenant_id=VALUES(tenant_id)");
+        if (columns.Contains("factory_id")) updateParts.Add("factory_id=VALUES(factory_id)");
+        if (columns.Contains("company_id")) updateParts.Add("company_id=VALUES(company_id)");
+        updateParts.Add("update_time=VALUES(create_time)");
 
         var sql = $"""
             INSERT INTO `{table}`
               ({string.Join(", ", insertCols.Select(c => $"`{c}`"))})
             VALUES
-              ({string.Join(", ", insertVals)})
+              {string.Join(",\n  ", valueGroups)}
             ON DUPLICATE KEY UPDATE
               {string.Join(",\n  ", updateParts)}
             """;
-
-        return await _db.Ado.ExecuteCommandAsync(sql, parameters);
+        return (sql, parameters);
     }
 
     internal static long? NormalizeFactoryId(long tenantId, long? factoryId) =>