AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Supply/S3MdpSyncTransformService.cs

using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Admin.NET.Plugin.AiDOP.SmartOps;

namespace Admin.NET.Plugin.AiDOP.Supply;

/// <summary>
/// S3 首批 MDP 同步和标准化转换服务。
/// Phase2：源→stg 经统一执行器；std/DWD/KPI 仍读 stg。
/// </summary>
public class S3MdpSyncTransformService : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly TransformRunLogFinalizer _runLogFinalizer;
    private readonly SmartOpsKpiAtomicBuildService _atomicBuild;
    private readonly MdpModuleStagingPuller _stagingPuller;
    private readonly IKpiTargetResolver _kpiTargetResolver;
    private readonly S9CompositeKpiWriter _s9Composite;
    private MdpRebuildScope _runScope = null!;
    private static readonly SemaphoreSlim GlobalRunGate = new(1, 1);

    public S3MdpSyncTransformService(
        ISqlSugarClient db,
        SmartOpsKpiAtomicBuildService atomicBuild,
        MdpModuleStagingPuller stagingPuller,
        IKpiTargetResolver kpiTargetResolver,
        TransformRunLogFinalizer runLogFinalizer,
        S9CompositeKpiWriter s9Composite)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _atomicBuild = atomicBuild;
        _stagingPuller = stagingPuller;
        _kpiTargetResolver = kpiTargetResolver;
        _s9Composite = s9Composite;
    }

    public Task<S3MdpSyncTransformResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO") =>
        throw new InvalidOperationException("S3 全量重算必须指定租户/工厂作用域，请走 ModuleRebuildService.Enqueue");

    public async Task<S3MdpSyncTransformResult> RunFullAsync(
        MdpRebuildScope scope,
        CancellationToken cancellationToken = default,
        string triggerType = "AUTO",
        long? rebuildJobId = null,
        Func<ModuleProgressUpdate, Task>? reportProgress = null)
    {
        await GlobalRunGate.WaitAsync(cancellationToken);
        try
        {
            _ = rebuildJobId;
            _runScope = MdpRebuildScope.Create("S3", scope.TenantId, scope.FactoryId);
            cancellationToken.ThrowIfCancellationRequested();
            var now = DateTime.Now;
            var batchId = $"S3_MDP_FULL_{_runScope.TenantId}_{_runScope.FactoryId}_{now:yyyyMMddHHmmss}";
            var runLogId = await InsertTransformRunLogAsync(batchId, now, triggerType);
            var result = new S3MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };

            try
            {
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Preparing, 0, 5, "准备运行环境"));
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Staging, 1, 10, "正在拉取源数据"));
                result.StageRows = await SyncStagingAsync(batchId, now, cancellationToken);
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Staging, 1, 35, "拉取源数据完成", result.StageRows, ModuleRebuildStages.Staging));

                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Standard, 2, 35, "正在标准化数据"));
                result.StandardRows = await TransformStandardAsync(batchId, now, retireStale: true, cancellationToken);
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Standard, 2, 50, "标准化数据完成", result.StandardRows, ModuleRebuildStages.Standard));

                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Dwd, 3, 50, "正在生成 DWD 明细"));
                result.DwdRows = await BuildDwdAsync(batchId, now, retireStale: true, result, cancellationToken);
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Dwd, 3, 65, "生成 DWD 明细完成", result.DwdRows, ModuleRebuildStages.Dwd));

                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Kpi, 4, 65, "正在重算 KPI"));
                result.KpiRows = await BuildS3KpiValuesAsync(now, cancellationToken);
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Kpi, 4, 80, "重算 KPI 完成", result.KpiRows, ModuleRebuildStages.Kpi));

                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Atomic, 5, 82, "正在重建原子数据"));
                result.AtomicRows = await _atomicBuild.BuildSupplyPurchaseDomainForAllDatesAsync(
                    _runScope.TenantId, _runScope.FactoryId, batchId, cancellationToken);
                result.AtomicRows += await _atomicBuild.BuildInventoryDomainForAllDatesAsync(
                    _runScope.TenantId, _runScope.FactoryId, batchId, cancellationToken);
                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Atomic, 5, 98, "重建原子数据完成", result.AtomicRows, ModuleRebuildStages.Atomic));

                await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Finalizing, 5, 99, "正在收尾"));
                result.KpiRows += await _s9Composite.WriteRecentAsync(_runScope.TenantId, _runScope.FactoryId, now, cancellationToken);
                await MarkTransformRunSuccessAsync(runLogId, now, result);
                return result;
            }
            catch (Exception ex)
            {
                // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
                if (!_runLogFinalizer.IsHostStopping)
                    await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
                throw;
            }
            finally
            {
                await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
            }
        }
        finally
        {
            GlobalRunGate.Release();
        }
    }

    private static async Task ReportAsync(Func<ModuleProgressUpdate, Task>? report, ModuleProgressUpdate update)
    {
        if (report == null) return;
        await report(update);
    }

    public async Task<S3MaterialRefreshResult> RefreshMaterialReadinessAsync(string? batchId = null, CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        var now = DateTime.Now;
        batchId ??= $"S3_MATERIAL_{now:yyyyMMddHHmmss}";

        _db.Ado.BeginTran();
        try
        {
            var stdCount = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM mdp_std_material_readiness WHERE IFNULL(work_order, '') <> ''");
            var readinessCount = await UpsertMaterialReadinessDwdAsync(batchId, now);
            await DeleteCurrentShortageAsync(now.Date);
            var shortageCount = await InsertMaterialShortageAsync(batchId, now);
            _db.Ado.CommitTran();

            return new S3MaterialRefreshResult
            {
                BatchId = batchId,
                StdCount = stdCount,
                ReadinessCount = readinessCount,
                ShortageCount = shortageCount
            };
        }
        catch
        {
            _db.Ado.RollbackTran();
            throw;
        }
    }

    private async Task<int> SyncStagingAsync(string batchId, DateTime now, CancellationToken cancellationToken)
    {
        _ = now;
        EnsureRunScope();
        return await _stagingPuller.PullEntitiesAsync(
            S3MdpEntityConfig.All.Select(x => x.EntityCode),
            batchId, _runScope.TenantId, fullRefresh: true, taskCode: "S3_MDP_INBOUND", cancellationToken,
            factoryId: _runScope.FactoryId, requireMatchingSourceTenant: true);
    }

    public async Task<S3MdpSyncTransformResult> RunInboundAsync(
        IEnumerable<string>? entityCodes = null,
        long tenantId = 0,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        if (tenantId <= 0)
            throw new InvalidOperationException("S3 inbound 必须指定有效 tenantId");
        _runScope = MdpRebuildScope.Create("S3", tenantId, 1);
        var now = DateTime.Now;
        var batchId = $"S3_MDP_IN_{_runScope.TenantId}_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertTransformRunLogAsync(batchId, now, "INBOUND");
        var result = new S3MdpSyncTransformResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            var codes = entityCodes?.ToList() ?? S3MdpEntityConfig.All.Select(x => x.EntityCode).ToList();
            result.StageRows = await _stagingPuller.PullEntitiesAsync(
                codes, batchId, tenantId, fullRefresh, "S3_MDP_INBOUND", cancellationToken);
            result.StandardRows = await TransformStandardAsync(batchId, now, retireStale: false, cancellationToken);
            result.DwdRows = await BuildDwdAsync(batchId, now, retireStale: false, result, cancellationToken);
            result.KpiRows = await BuildS3KpiValuesAsync(now, cancellationToken);
            result.AtomicRows = await _atomicBuild.BuildSupplyPurchaseDomainForAllDatesAsync(
                _runScope.TenantId, _runScope.FactoryId, batchId, cancellationToken);
            result.AtomicRows += await _atomicBuild.BuildInventoryDomainForAllDatesAsync(
                _runScope.TenantId, _runScope.FactoryId, batchId, cancellationToken);
            await MarkTransformRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkTransformRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    [Obsolete("Phase3: use MdpModuleStagingPuller / executors")]
    private async Task<int> SyncOneEntityAsync(S3MdpEntityConfig entity, string batchId, DateTime now)
    {
        var entityRow = await _db.Ado.SqlQuerySingleAsync<S3MdpEntityRow>(
            "SELECT id AS Id, entity_name AS EntityName FROM mdp_entity WHERE tenant_id=0 AND entity_code=@EntityCode LIMIT 1",
            new SugarParameter("@EntityCode", entity.EntityCode));
        if (entityRow == null) throw Oops.Oh($"未找到 MDP 实体配置：{entity.EntityCode}");

        var columns = await _db.Ado.SqlQueryAsync<S3ColumnRow>(
            """
            SELECT COLUMN_NAME AS ColumnName
            FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@TableName
            ORDER BY ORDINAL_POSITION
            """,
            new SugarParameter("@TableName", entity.SourceTable));
        if (columns.Count == 0) throw Oops.Oh($"未找到源表：{entity.SourceTable}");

        var names = columns.Select(u => u.ColumnName).ToList();
        var tenantExpr = names.Any(u => string.Equals(u, "tenant_id", StringComparison.OrdinalIgnoreCase))
            ? $"IFNULL(s.`{FindColumn(names, "tenant_id")}`,0)"
            : "0";
        var sourceRowExpr = names.Any(u => string.Equals(u, entity.SourceRowIdExpression, StringComparison.OrdinalIgnoreCase))
            ? $"s.`{FindColumn(names, entity.SourceRowIdExpression)}`"
            : entity.SourceRowIdExpression;
        var rawDataExpr = BuildJsonObjectExpression(names);

        var rowsRead = await _db.Ado.GetIntAsync($"SELECT COUNT(1) FROM `{entity.SourceTable}`");
        var logId = await InsertSyncLogAsync(entityRow.Id, entityRow.EntityName, batchId, rowsRead);
        var started = DateTime.Now;

        try
        {
            var affected = await _db.Ado.ExecuteCommandAsync(
                $"""
                INSERT INTO `{entity.TargetTable}`
                (tenant_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
                SELECT
                    {tenantExpr},
                    COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = @SourceTable LIMIT 1), NULLIF('',''), ''),
                    @SourceTable,
                    CAST({sourceRowExpr} AS CHAR),
                    CAST(COALESCE({entity.SourceBizKeyExpression}, CAST({sourceRowExpr} AS CHAR)) AS CHAR),
                    @BatchId,
                    @Now,
                    'PENDING',
                    {rawDataExpr}
                FROM `{entity.SourceTable}` s
                ON DUPLICATE KEY UPDATE
                    source_row_id=VALUES(source_row_id),
                    sync_batch_id=VALUES(sync_batch_id),
                    sync_time=VALUES(sync_time),
                    process_status=VALUES(process_status),
                    raw_data=VALUES(raw_data),
                    update_time=CURRENT_TIMESTAMP
                """,
                new SugarParameter("@SourceTable", entity.SourceTable),
                new SugarParameter("@BatchId", batchId),
                new SugarParameter("@Now", now));

            await MarkSyncLogSuccessAsync(logId, started, affected);
            return rowsRead;
        }
        catch (Exception ex)
        {
            await MarkSyncLogFailedAsync(logId, started, ex.Message);
            throw;
        }
    }

    private async Task<int> TransformStandardAsync(
        string batchId, DateTime now, bool retireStale, CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var command in BuildStandardCommands(batchId, now, retireStale))
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
        }
        return total;
    }

    private async Task<int> BuildDwdAsync(
        string batchId, DateTime now, bool retireStale, S3MdpSyncTransformResult result,
        CancellationToken cancellationToken)
    {
        var total = 0;
        foreach (var command in BuildDwdCommands(batchId, now, retireStale))
        {
            cancellationToken.ThrowIfCancellationRequested();
            total += await _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
        }
        total += await UpsertMaterialReadinessDwdAsync(batchId, now);
        await DeleteCurrentShortageAsync(now.Date);
        total += await InsertMaterialShortageAsync(batchId, now);

        // 发布证据：必须在 DWD 写入**与同日淘汰**都完成之后。
        // 只有全量路径发布 —— 增量批次只含变更行，把它指认成「当前快照」会让读者看到残缺集合。
        if (retireStale)
        {
            cancellationToken.ThrowIfCancellationRequested();
            result.Publication = await PublishCurrentSnapshotAsync(batchId, now);
        }

        return total;
    }

    /// <summary>
    /// 复核并记录本轮发布的当前快照，供 S8 Rule01 精确指认「当前态是哪一批」。
    ///
    /// <para><b>为什么不能靠 MAX(stat_date) 之类的时间约定</b>：源侧被合法清空的那一轮
    /// 产出 0 行 DWD，于是当天根本不存在 <c>stat_date</c> 分区，同日淘汰也无行可删
    /// （它只能删「今天」）。读者按最新日期兜底就会**静默回落到上一天**，把一批早已作废的行
    /// 当成当前态。实测：租户 838257212780613 源侧清零、STD 归 0、跑批 SUCCESS，
    /// 而 DWD 当前快照仍返回前一日的 10 行。</para>
    ///
    /// <para>返回的 <c>CurrentRows = 0</c> 是<b>合法结果</b> —— 成功发布了一个空快照。
    /// 把它记进运行日志，下游才能把「正常的空」与「没发布成」区分开：前者健康且允许恢复，
    /// 后者无证据、判 UNKNOWN 并拦下恢复。</para>
    ///
    /// <para>行数用<b>复核查询</b>得到，不用 INSERT 的影响行数：ODKU 的影响行数
    /// （插入计 1、更新计 2、无变化计 0）从来不等于「当前有几行」。</para>
    /// </summary>
    private async Task<S3MdpSnapshotPublication> PublishCurrentSnapshotAsync(string batchId, DateTime now)
    {
        var scope = RequireScope();
        var currentRows = await _db.Ado.GetIntAsync(
            """
            SELECT COUNT(*) FROM dwd_supplier_delivery
             WHERE tenant_id=@TenantId
               AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
               AND stat_date=@StatDate
               AND sync_batch_id=@BatchId
            """,
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId),
            new SugarParameter("@StatDate", now.Date),
            new SugarParameter("@BatchId", batchId));

        return new S3MdpSnapshotPublication
        {
            Table = "dwd_supplier_delivery",
            BatchId = batchId,
            StatDate = now.Date,
            FactoryId = scope.FactoryId,
            CurrentRows = currentRows
        };
    }

    private async Task<int> BuildS3KpiValuesAsync(DateTime now, CancellationToken cancellationToken)
    {
        var affected = 0;
        for (var dayOffset = 13; dayOffset >= 0; dayOffset--)
        {
            var statDate = now.Date.AddDays(-dayOffset);
            var rows = await CalculateS3KpiValuesAsync(statDate);
            foreach (var row in rows)
            {
                cancellationToken.ThrowIfCancellationRequested();
                affected += await UpsertS3KpiValueAsync(row, statDate, now);
            }
        }
        return affected;
    }

    private async Task<List<S3KpiCalcRow>> CalculateS3KpiValuesAsync(DateTime statDate)
    {
        var scope = RequireScope();
        var rows = await _db.Ado.SqlQueryAsync<S3KpiCalcRow>(MdpSqlScope.InjectTenantFactory(
            """
            SELECT tenant_id AS TenantId, @FactoryId AS FactoryId, 'S3_L2_001' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, request_date, send_date) / 24), 4) AS MetricValue
            FROM mdp_std_purchase_request
            WHERE tenant_id=@TenantId AND (@FactoryId=1 OR COALESCE(NULLIF(factory_id,0),1)=@FactoryId)
              AND request_date IS NOT NULL AND send_date IS NOT NULL AND send_date >= request_date
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, @FactoryId AS FactoryId, 'S3_L2_002' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN TIMESTAMPDIFF(HOUR, request_date, send_date) <= 72 THEN 1 ELSE 0 END)
                         / NULLIF(COUNT(1),0), 4) AS MetricValue
            FROM mdp_std_purchase_request
            WHERE tenant_id=@TenantId AND (@FactoryId=1 OR COALESCE(NULLIF(factory_id,0),1)=@FactoryId)
              AND request_date IS NOT NULL AND send_date IS NOT NULL AND send_date >= request_date
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, @FactoryId AS FactoryId, 'S3_L2_003' AS MetricCode,
                   ROUND(COUNT(1) / NULLIF((
                       SELECT COUNT(1) FROM mdp_std_employee p
                       WHERE p.tenant_id=@TenantId AND p.employment_status='ACTIVE' AND p.position_code='PLANNER' AND (@sourceSystem='' OR p.source_system=@sourceSystem) AND (p.source_system<>'T8' OR p.domain=@sourceDomain)
                   ),0), 4) AS MetricValue
            FROM mdp_std_purchase_request
            WHERE tenant_id=@TenantId AND (@FactoryId=1 OR COALESCE(NULLIF(factory_id,0),1)=@FactoryId)
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L2_004' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, request_date, submit_date) / 24), 4) AS MetricValue
            FROM mdp_std_delivery_schedule
            WHERE request_date IS NOT NULL AND submit_date IS NOT NULL AND submit_date >= request_date
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L2_005' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN IFNULL(schedule_qty,0) >= IFNULL(order_qty,0) AND IFNULL(order_qty,0) > 0 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM dwd_supplier_delivery
            WHERE stat_date=@StatDate AND IFNULL(order_qty,0) > 0
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, @FactoryId AS FactoryId, 'S3_L2_006' AS MetricCode,
                   ROUND(COUNT(1) / NULLIF((
                       SELECT COUNT(1) FROM mdp_std_employee p
                       WHERE p.tenant_id=@TenantId AND p.employment_status='ACTIVE' AND p.position_code='PLANNER' AND (@sourceSystem='' OR p.source_system=@sourceSystem) AND (p.source_system<>'T8' OR p.domain=@sourceDomain)
                   ),0), 4) AS MetricValue
            FROM mdp_std_delivery_schedule
            WHERE tenant_id=@TenantId AND (@FactoryId=1 OR COALESCE(NULLIF(factory_id,0),1)=@FactoryId)
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L1_001' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, request_date, submit_date) / 24), 4) AS MetricValue
            FROM mdp_std_delivery_schedule
            WHERE request_date IS NOT NULL AND submit_date IS NOT NULL AND submit_date >= request_date
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L1_002' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN IFNULL(schedule_qty,0) >= IFNULL(order_qty,0) AND IFNULL(order_qty,0) > 0 THEN 1 ELSE 0 END) / COUNT(1), 4) AS MetricValue
            FROM dwd_supplier_delivery
            WHERE stat_date=@StatDate AND IFNULL(order_qty,0) > 0
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, @FactoryId AS FactoryId, 'S3_L3_001' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, request_date, send_date) / 24), 4) AS MetricValue
            FROM mdp_std_purchase_request
            WHERE tenant_id=@TenantId AND (@FactoryId=1 OR COALESCE(NULLIF(factory_id,0),1)=@FactoryId)
              AND request_date IS NOT NULL AND send_date IS NOT NULL AND send_date >= request_date
            GROUP BY tenant_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L3_002' AS MetricCode,
                   ROUND(AVG(TIMESTAMPDIFF(HOUR, order_date, COALESCE(need_date, due_date)) / 24), 4) AS MetricValue
            FROM mdp_std_purchase_order
            WHERE order_date IS NOT NULL AND COALESCE(need_date, due_date) IS NOT NULL
              AND COALESCE(need_date, due_date) >= order_date
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L3_003' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN ready_status='READY' THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM dwd_material_readiness
            WHERE stat_date=@StatDate
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L3_004' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN submit_date IS NOT NULL AND request_date IS NOT NULL
                        AND TIMESTAMPDIFF(HOUR, request_date, submit_date) <= 72 THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM mdp_std_delivery_schedule
            WHERE request_date IS NOT NULL
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L3_005' AS MetricCode,
                   ROUND(100 * SUM(CASE WHEN risk_level IN ('HIGH','MEDIUM') OR IFNULL(remaining_qty,0) > 0 THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0), 4) AS MetricValue
            FROM dwd_supplier_delivery
            WHERE stat_date=@StatDate
            GROUP BY tenant_id, factory_id
            UNION ALL
            SELECT tenant_id AS TenantId, factory_id AS FactoryId, 'S3_L3_006' AS MetricCode,
                   ROUND(100 * (1 - SUM(CASE WHEN IFNULL(shortage_qty,0) > 0 THEN 1 ELSE 0 END) / NULLIF(COUNT(1), 0)), 4) AS MetricValue
            FROM dwd_material_readiness
            WHERE stat_date=@StatDate
            GROUP BY tenant_id, factory_id
            """),
            new SugarParameter("@StatDate", statDate),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId),
            new SugarParameter("@sourceDomain", "pbxfxp"),
            new SugarParameter("@sourceSystem", ""));

        var plannedLines = await _db.Ado.GetIntAsync(
            """
            SELECT COUNT(1)
            FROM dwd_material_readiness
            WHERE tenant_id=@TenantId
              AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
              AND stat_date=@StatDate
            """,
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId),
            new SugarParameter("@StatDate", statDate));
        var plannerCount = await _db.Ado.GetIntAsync(
            """
            SELECT COUNT(1)
            FROM mdp_std_employee
            WHERE tenant_id=@TenantId
              AND employment_status='ACTIVE'
              AND position_code='PLANNER'
              AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
            """,
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@sourceDomain", "pbxfxp"),
            new SugarParameter("@sourceSystem", ""));
        if (plannedLines > 0 && plannerCount > 0)
        {
            rows.Add(new S3KpiCalcRow
            {
                TenantId = scope.TenantId,
                FactoryId = scope.FactoryId,
                MetricCode = "S3_L1_003",
                MetricValue = Math.Round((decimal)plannedLines / plannerCount, 4)
            });
        }

        var materialTurnoverDays = await _db.Ado.GetDecimalAsync(
            """
            SELECT ROUND(
                ((SUM(receipt_qty) + SUM(GREATEST(receipt_qty - issue_qty, 0))) / 2)
                / NULLIF(SUM(issue_qty), 0) * 30,
                4
            )
            FROM (
                SELECT item_num,
                       SUM(CASE WHEN biz_doc_type='PUR_RECEIPT' THEN IFNULL(qty_change,0) ELSE 0 END) AS receipt_qty,
                       SUM(CASE WHEN biz_doc_type='PROD_ISSUE' THEN IFNULL(qty_change,0) ELSE 0 END) AS issue_qty
                FROM mdp_std_inv_trans
                WHERE tenant_id=@TenantId
                  AND summary_flag=0 AND void_flag=0 AND approved_flag=1 AND (@sourceSystem='' OR source_system=@sourceSystem) AND (source_system<>'T8' OR domain=@sourceDomain)
                  AND biz_doc_type IN ('PUR_RECEIPT','PROD_ISSUE')
                  AND approved_time >= DATE_FORMAT(DATE_SUB(@StatDate, INTERVAL 1 MONTH), '%Y-%m-01')
                  AND approved_time < DATE_FORMAT(@StatDate, '%Y-%m-01')
                GROUP BY item_num
            ) material_flow
            HAVING SUM(issue_qty) > 0
            """,
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@StatDate", statDate),
            new SugarParameter("@sourceDomain", "pbxfxp"),
            new SugarParameter("@sourceSystem", ""));
        if (materialTurnoverDays > 0)
        {
            rows.Add(new S3KpiCalcRow
            {
                TenantId = scope.TenantId,
                FactoryId = scope.FactoryId,
                MetricCode = "S3_L1_004",
                MetricValue = materialTurnoverDays
            });
        }

        return rows;
    }

    private async Task<int> UpsertS3KpiValueAsync(S3KpiCalcRow row, DateTime statDate, DateTime now)
    {
        var meta = await _db.Ado.SqlQuerySingleAsync<S3KpiMetaRow>(
            """
            SELECT MetricLevel, Direction, YellowThreshold, RedThreshold
            FROM ado_smart_ops_kpi_master
            WHERE TenantId=@TenantId AND ModuleCode='S3' AND MetricCode=@MetricCode AND IsEnabled=1
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@MetricCode", row.MetricCode));
        if (meta == null || row.MetricValue == null)
            return 0;

        var table = ResolveKpiValueTable(meta.MetricLevel);
        var current = await _db.Ado.SqlQuerySingleAsync<S3KpiValueRow>(
            $"""
            SELECT id AS Id, metric_value AS MetricValue, target_value AS TargetValue
            FROM {table}
            WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S3'
              AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0
            ORDER BY id
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@FactoryId", row.FactoryId),
            new SugarParameter("@MetricCode", row.MetricCode),
            new SugarParameter("@BizDate", statDate));
        var prior = await _db.Ado.SqlQuerySingleAsync<S3KpiValueRow>(
            $"""
            SELECT id AS Id, metric_value AS MetricValue, target_value AS TargetValue
            FROM {table}
            WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S3'
              AND metric_code=@MetricCode AND biz_date<@BizDate AND is_deleted=0
            ORDER BY biz_date DESC, id DESC
            LIMIT 1
            """,
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@FactoryId", row.FactoryId),
            new SugarParameter("@MetricCode", row.MetricCode),
            new SugarParameter("@BizDate", statDate));

        var actual = Math.Round(row.MetricValue.Value, 4);
        var snap = await _kpiTargetResolver.ResolveAsync(row.TenantId, row.FactoryId, row.MetricCode, "S3", statDate);
        var target = snap.TargetValue;
        var status = AidopS4KpiMerge.AchievementLevel(actual, target, meta.Direction, meta.YellowThreshold, meta.RedThreshold);
        var trend = ResolveTrendFlag(actual, prior?.MetricValue);

        if (current != null)
        {
            return await _db.Ado.ExecuteCommandAsync(
                $"""
                UPDATE {table}
                SET metric_value=@MetricValue, target_value=@TargetValue, status_color=@StatusColor, trend_flag=@TrendFlag,
                    target_config_id=@TargetConfigId, target_source=@TargetSource, target_resolved_at=@TargetResolvedAt,
                    is_active=1, status='ACTIVE', calc_time=@CalcTime, update_time=@CalcTime
                WHERE tenant_id=@TenantId AND factory_id=@FactoryId AND module_code='S3'
                  AND metric_code=@MetricCode AND biz_date=@BizDate AND is_deleted=0
                """,
                new SugarParameter("@MetricValue", actual),
                new SugarParameter("@TargetValue", KpiTargetSnapshotSql.ValueOrDbNull(snap)),
                new SugarParameter("@StatusColor", status),
                new SugarParameter("@TrendFlag", trend),
                new SugarParameter("@TargetConfigId", KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
                new SugarParameter("@TargetSource", KpiTargetSnapshotSql.SourceOrDbNull(snap)),
                new SugarParameter("@TargetResolvedAt", snap.ResolvedAt),
                new SugarParameter("@CalcTime", now),
                new SugarParameter("@TenantId", row.TenantId),
                new SugarParameter("@FactoryId", row.FactoryId),
                new SugarParameter("@MetricCode", row.MetricCode),
                new SugarParameter("@BizDate", statDate));
        }

        var nextId = Yitter.IdGenerator.YitIdHelper.NextId();
        return await _db.Ado.ExecuteCommandAsync(
            $"""
            INSERT INTO {table}
            (id, tenant_id, factory_id, status, biz_date, create_time, update_time, is_deleted, is_active,
             module_code, metric_code, metric_value, target_value, status_color, trend_flag, calc_time,
             target_config_id, target_source, target_resolved_at)
            VALUES
            (@Id, @TenantId, @FactoryId, 'ACTIVE', @BizDate, @CalcTime, @CalcTime, 0, 1,
             'S3', @MetricCode, @MetricValue, @TargetValue, @StatusColor, @TrendFlag, @CalcTime,
             @TargetConfigId, @TargetSource, @TargetResolvedAt)
            """,
            new SugarParameter("@Id", nextId),
            new SugarParameter("@TenantId", row.TenantId),
            new SugarParameter("@FactoryId", row.FactoryId),
            new SugarParameter("@BizDate", statDate),
            new SugarParameter("@CalcTime", now),
            new SugarParameter("@MetricCode", row.MetricCode),
            new SugarParameter("@MetricValue", actual),
            new SugarParameter("@TargetValue", KpiTargetSnapshotSql.ValueOrDbNull(snap)),
            new SugarParameter("@StatusColor", status),
            new SugarParameter("@TrendFlag", trend),
            new SugarParameter("@TargetConfigId", KpiTargetSnapshotSql.ConfigIdOrDbNull(snap)),
            new SugarParameter("@TargetSource", KpiTargetSnapshotSql.SourceOrDbNull(snap)),
            new SugarParameter("@TargetResolvedAt", snap.ResolvedAt));
    }

    private IEnumerable<S3MdpSqlCommand> BuildStandardCommands(string batchId, DateTime now, bool retireStale)
    {
        yield return Cmd(
            """
            INSERT INTO mdp_std_supplier
            (tenant_id, factory_id, company_id, source_system, supplier_code, supplier_name, supplier_type, status, contact, address, currency_type, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, factory_id, company_id, COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Supp')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.SortName')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Type')),
                   CASE WHEN JSON_EXTRACT(raw_data,'$.IsActive') IN (1, true) THEN 'ACTIVE' ELSE 'INACTIVE' END,
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.contact')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.address')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Curr')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_supplier
            WHERE source_table='SuppMaster' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Supp')), '') <> ''
            ON DUPLICATE KEY UPDATE supplier_name=VALUES(supplier_name), supplier_type=VALUES(supplier_type), status=VALUES(status),
                contact=VALUES(contact), address=VALUES(address), currency_type=VALUES(currency_type), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_item
            (tenant_id, factory_id, company_id, source_system, item_code, item_name, model, unit, item_type, status, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, factory_id, company_id, COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number'))),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Descr')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.name')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_name'))),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Drawing')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.model'))),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.UM')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.unit'))),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemType')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_type'))),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Status')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.is_active'))),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_item
            WHERE COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), '') <> ''
            ON DUPLICATE KEY UPDATE item_name=VALUES(item_name), model=VALUES(model), unit=VALUES(unit), item_type=VALUES(item_type),
                status=VALUES(status), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_supplier_item
            (tenant_id, factory_id, company_id, source_system, item_code, supplier_code, supplier_name, supplier_type, quota_rate, lead_time, min_qty, packaging_qty, price, currency_type, effective_date, expire_date, status, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, factory_id, company_id, COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_name'))), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_name')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_type')),
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.quota_rate')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.quota_rate')) AS DECIMAL(18,6)) END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.lead_time')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.lead_time')) AS DECIMAL(18,6)) END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty_min')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty_min')) AS DECIMAL(18,6)) END,
                   CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.packaging_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.packaging_qty')) AS DECIMAL(18,6)) END,
                   COALESCE(
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.netpurchase_price')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.netpurchase_price')) AS DECIMAL(18,6)) END,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.order_price')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.order_price')) AS DECIMAL(18,6)) END
                   ),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.currency_type')),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.effective_date')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.expiring_date')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.is_active')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_source_list
            WHERE source_table='srm_purchase' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')), '') <> ''
              AND IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.icitem_name'))), '') <> ''
            ON DUPLICATE KEY UPDATE supplier_name=VALUES(supplier_name), quota_rate=VALUES(quota_rate), lead_time=VALUES(lead_time),
                min_qty=VALUES(min_qty), packaging_qty=VALUES(packaging_qty), price=VALUES(price), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_supply_demand
            (tenant_id, factory_id, company_id, source_system, demand_no, demand_line, demand_type, item_code, item_name, required_qty, required_date, supplier_code, status, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, factory_id, company_id, COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.number')), source_row_id),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.line')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Line'))),
                   source_table,
                   IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_number')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum'))), ''),
                   IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemname')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.item_name')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Descr'))), ''),
                   COALESCE(
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qty')) AS DECIMAL(18,6)) END,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.required_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.required_qty')) AS DECIMAL(18,6)) END,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shortqty')) AS DECIMAL(18,6)) END,
                       0),
                   COALESCE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.required_date')), 'null'), ''), NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.needdate')), 'null'), ''), NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.NeedDate')), 'null'), '')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier_number')),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.status')), JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Status'))),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_supply_demand
            WHERE IFNULL(source_biz_key,'') <> ''
            ON DUPLICATE KEY UPDATE item_code=VALUES(item_code), item_name=VALUES(item_name), required_qty=VALUES(required_qty),
                required_date=VALUES(required_date), supplier_code=VALUES(supplier_code), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_purchase_request
            (tenant_id, factory_id, company_id, source_system, pr_no, pr_line, work_order, sales_order_entry_id, item_code, item_name, supplier_code, request_qty, request_date, send_date, arrive_date, status, source_biz_key, sync_batch_id, sync_time)
            SELECT g.tenant_id, g.factory_id, g.company_id, COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.pr_billno')), g.source_row_id),
                   JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.line_no')),
                   -- Stage-3 订单归因第 1 段：PR 所属工单号（源 srm_pr_main.pr_mono）。
                   -- 必须把 JSON 的字面量 'null' 归一成真 NULL —— 否则下游 JOIN 会去匹配
                   -- 一个叫 "null" 的工单号（既有 mdp_std_purchase_order.work_order 正是踩了
                   -- 这个坑：451 行里 208 行是字面量 'null'）。
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.pr_mono')), 'null'), ''),
                   -- Stage-3 订单归因第 2 段：由工单号派生订单明细行 Id。
                   -- 刻意不读源表 srm_pr_main.sentry_id —— 该字段源侧 1012/1012 全 NULL，
                   -- 生产代码从未写过；这里改从 S2 标准层（1.0.532 建立的桥）派生。
                   -- LEFT JOIN：S2 跑批未覆盖的租户得到 NULL，而不是让 PR 整行消失。
                   wo.sales_order_entry_id,
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.item_number')), ''),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.item_name')), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.supplier_number')),
                   COALESCE(
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.qty')) AS DECIMAL(18,6)) END,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.request_qty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.request_qty')) AS DECIMAL(18,6)) END,
                       0),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.create_time')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.send_date')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.arrive_date')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.status')),
                   g.source_biz_key, @BatchId, @Now
            FROM mdp_stg_supply_demand g
            -- 订单行 Id 派生。ON 里必须【自带】租户谓词：MdpSqlScope.InjectTenantFactory
            -- 只改写筛选子句，结构上够不到 JOIN...ON；少了它工单号就成了跨租户字符串匹配。
            -- （这里刻意不写出那个关键字：注入器对 SQL 注释同样生效，写了会被就地改写。）
            -- 同租户同工单号在 mdp_std_work_order_schedule 内唯一（实测重复组 = 0），故不会扇出。
            LEFT JOIN mdp_std_work_order_schedule wo
                   ON wo.tenant_id = g.tenant_id
                  AND wo.work_order = NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(g.raw_data,'$.pr_mono')), 'null'), '')
                  AND wo.doc_type='PROD_TASK'
            WHERE g.source_table='srm_pr_main'
            ON DUPLICATE KEY UPDATE factory_id=VALUES(factory_id),
                work_order=VALUES(work_order), sales_order_entry_id=VALUES(sales_order_entry_id),
                item_code=VALUES(item_code), item_name=VALUES(item_name), supplier_code=VALUES(supplier_code),
                request_qty=VALUES(request_qty), request_date=VALUES(request_date), send_date=VALUES(send_date),
                arrive_date=VALUES(arrive_date), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
                update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_purchase_order
            (tenant_id, factory_id, company_id, source_system, po_no, po_line, purchase_request_no, po_type, supplier_code, item_code, item_name, order_qty, received_qty, returned_qty, due_date, need_date, order_date, status, buyer, work_order, source_biz_key, sync_batch_id, sync_time)
            SELECT d.tenant_id, d.factory_id, d.company_id, COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd')),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Line')) AS CHAR),
                   -- Stage-3 订单归因主键路径：PO 行 → 采购申请单号（源 PurOrdDetail.Req，实测 1:1）。
                   -- 归因走这里，【不走】下面的 work_order —— 后者只在历史/UAT 批量导入的工单上
                   -- 有值，而源侧 PurOrdDetail.SalesJob 实测 0/448、BusinessID 实测全 0。
                   -- 共享采购（同物料跨多工单）因此仍能按 PO 行精确切分到各自订单。
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Req')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Potype')),
                   JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Supp')),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum')), ''),
                   IFNULL(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(i.raw_data,'$.Descr')), JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Descr'))), ''),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyOrded')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyOrded')) AS DECIMAL(18,6)) END, 0),
                   COALESCE(
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReceived')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReceived')) AS DECIMAL(18,6)) END,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RctQty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RctQty')) AS DECIMAL(18,6)) END,
                       0),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReturned')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyReturned')) AS DECIMAL(18,6)) END, 0),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.DueDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.NeedDate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.OrdDate')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Status')),
                   JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.Buyer')),
                   -- 归一字面量 'null'：改造前 451 行中 208 行落的是字符串 "null"，
                   -- 任何按 work_order 的 JOIN 都会被它污染。本列只作参考，不作归因。
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.WorkOrd')), 'null'), ''),
                   d.source_biz_key, @BatchId, @Now
            FROM mdp_stg_purchase_order d
            -- ── enrichment JOIN 的两个约束，缺一不可 ────────────────────────────────────
            --
            -- ① 必须自带租户谓词。MdpSqlScope.InjectTenantFactory 只把 WHERE 改写成
            --    `WHERE d.tenant_id=@TenantId AND ...`（实现是 Regex.Replace(sql, @"\bWHERE\b", ...)），
            --    它**结构上够不到 JOIN...ON**。少了它，采购单号与物料号就成了跨租户的字符串匹配：
            --    实测 mdp_stg_item 里 ItemMaster 的 item_code 有 5529 组跨租户重名（最多横跨 3 个租户）。
            --
            -- ② 必须按 (租户, 键) 去重取最新一行。贴源层同时存在两代 source_biz_key
            --    （旧 `Domain:PurOrd:Line` 与新的裸 RecID，因 uk_source_key 含 source_system 与
            --    biz_key，两代永远不会互相 ODKU 覆盖），同一个 PurOrd/ItemNum 因此有多行。
            --    不去重则同租户内仍会扇出 —— 单租户实测 870/220 = 3.95 倍。
            --
            -- 扇出被 uk_po_line(tenant_id,po_no,po_line) 的 ODKU 吸收，所以症状不是行数膨胀，
            -- 而是 item_name / supplier_code / order_date / buyer 这些值由 MySQL 任取一行决定。
            -- 这里不按 @BatchId 收窄 enrichment：增量路径下未变更的主数据不在本批次里，
            -- 收窄会把它们打成 NULL。取最新一行才是「当前已知主数据」的正确语义。
            -- 去重用 ROW_NUMBER 而不是 GROUP_CONCAT+SUBSTRING_INDEX：raw_data 是 JSON 大字段，
            -- 而 group_concat_max_len 实测仅 1024，拼接必然截断且首元素本身就可能超限。
            --
            -- ③ 派生表里的每个 WHERE 都必须**自带** @TenantId。这不是冗余，是在关掉注入器：
            --    Inject 只在 WHERE 后 240 字符内出现 @TenantId 时才跳过，否则一律改写。
            --    而 ResolveSourcePrefix 沿 SQL 反向找最近的同层 `FROM x 别名` 来定别名前缀，
            --    派生表外层是 `FROM (` —— 别名正则匹配不到括号，于是回退成**空前缀**，
            --    生成裸 `factory_id`；派生表的 SELECT 列表里没有这一列，
            --    整轮转换直接报 Unknown column 'factory_id' 失败（本地沙箱实测复现）。
            --    （此处刻意不写出该报错的完整英文原文：注入器对 SQL 注释同样生效，
            --      原文里的 where 一词会被当成关键字改写，徒增排查噪音。）
            --    自带 @TenantId 后注入被跳过，作用域完全由这里的字面 SQL 决定。
            --    只按租户不按工厂：驱动行 d 已被注入器限定到 (@TenantId,@FactoryId)，
            --    而主数据（PurOrdMaster / ItemMaster）的 factory_id 未必与明细一致，
            --    多加一层工厂过滤会把 supplier_code / item_name 静默打成 NULL。
            LEFT JOIN (
                SELECT x.tenant_id, x.pur_ord, x.raw_data FROM (
                    SELECT x0.tenant_id, x0.raw_data,
                           JSON_UNQUOTE(JSON_EXTRACT(x0.raw_data,'$.PurOrd')) AS pur_ord,
                           ROW_NUMBER() OVER (
                               PARTITION BY x0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(x0.raw_data,'$.PurOrd'))
                               ORDER BY x0.sync_time DESC, x0.id DESC) AS rn
                      FROM mdp_stg_purchase_order x0
                     WHERE x0.tenant_id=@TenantId AND x0.source_table='PurOrdMaster') x
                 WHERE x.tenant_id=@TenantId AND x.rn=1
            ) m ON m.tenant_id = d.tenant_id
                AND m.pur_ord = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd'))
            LEFT JOIN (
                SELECT y.tenant_id, y.item_num, y.raw_data FROM (
                    SELECT y0.tenant_id, y0.raw_data,
                           JSON_UNQUOTE(JSON_EXTRACT(y0.raw_data,'$.ItemNum')) AS item_num,
                           ROW_NUMBER() OVER (
                               PARTITION BY y0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(y0.raw_data,'$.ItemNum'))
                               ORDER BY y0.sync_time DESC, y0.id DESC) AS rn
                      FROM mdp_stg_item y0
                     WHERE y0.tenant_id=@TenantId AND y0.source_table='ItemMaster') y
                 WHERE y.tenant_id=@TenantId AND y.rn=1
            ) i ON i.tenant_id = d.tenant_id
                AND i.item_num = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))
            -- 驱动行按本轮批次收窄：staging 与本转换共用同一个 @BatchId
            -- （RunFullAsync 生成一次，依次传给 SyncStagingAsync / TransformStandardAsync / BuildDwdAsync），
            -- 因此「本批次里的行」就是「本轮源侧的行」，比"最新批次"子查询更确定。
            -- 少了它，本 INSERT 会把整张贴源历史重新物化，并把死行盖上最新批次戳
            -- —— 实测租户 797 因此在 STD 沉淀了 336 行源侧已不存在的采购订单行（518 行中的 64.9%）。
            -- 驱动行的租户/工厂作用域在这里写死，不再交给注入器。
            -- 上面两个派生表插进 `FROM ... d` 与本筛选子句之间以后，ResolveSourcePrefix 反向找
            -- 最近的同层 `FROM 表 别名` 时会先撞上派生表的 `FROM (`，别名正则匹配不到括号 →
            -- 回退成空前缀，生成裸 `tenant_id` —— 而派生表 m/i 也各自暴露 tenant_id，
            -- 该列随即变成歧义列。本 SQL 因此必须自带前缀，且注入对它应当是恒等变换。
            -- （注释里刻意避开 WHERE 一词：注入器对注释同样生效，只有紧跟 @TenantId 时才跳过。）
            WHERE d.tenant_id=@TenantId AND COALESCE(NULLIF(d.factory_id,0),1)=@FactoryId
              AND d.source_table='PurOrdDetail' AND d.sync_batch_id=@BatchId
              AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd')), '') <> ''
            ON DUPLICATE KEY UPDATE purchase_request_no=VALUES(purchase_request_no),
                po_type=VALUES(po_type), supplier_code=VALUES(supplier_code), item_code=VALUES(item_code),
                item_name=VALUES(item_name), order_qty=VALUES(order_qty), received_qty=VALUES(received_qty), returned_qty=VALUES(returned_qty),
                due_date=VALUES(due_date), need_date=VALUES(need_date), order_date=VALUES(order_date), status=VALUES(status),
                buyer=VALUES(buyer), work_order=VALUES(work_order), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
                update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        // ── 每轮淘汰：让标准层重新成为源的镜像 ────────────────────────────────────────
        //
        // 只加批次过滤是不够的。过滤只让死行「不再被刷新」，而运行期链路里没有任何 DELETE，
        // 死行会带着旧批次号永远留下 —— 那只是把问题从「认不出」变成「认得出但仍在」。
        //
        // 判据：本轮 INSERT 已给所有存活行盖上 @BatchId，
        // 因此该租户下 sync_batch_id 不等于 @BatchId 的行，就是本轮源侧已不存在的行。
        //
        // 安全闸门用 mdp_sync_log 而不是「贴源层有没有行」：后者对「源侧真的清空了」的合法租户
        // 会永久拒绝淘汰、把幽灵行锁死；前者精确表达「本轮拉取对该租户成功了」——
        // 实测 0 行租户同样留下 SUCCESS 记录（如 1300000000777：90 次运行 / 90 次 0 行 / 90 次 SUCCESS），
        // 因此源侧合法清空的租户能被正确清干净。
        //
        // 仅在全量路径执行：增量批次只含变更行，按它淘汰会把未变更的存量行全删掉。
        if (retireStale)
        {
            yield return Cmd(
                """
                DELETE s FROM mdp_std_purchase_order s
                 WHERE s.tenant_id=@TenantId
                   AND COALESCE(NULLIF(s.factory_id,0),1)=@FactoryId
                   AND IFNULL(s.sync_batch_id,'') <> @BatchId
                   AND EXISTS (SELECT 1 FROM mdp_sync_log l
                                WHERE l.tenant_id=@TenantId
                                  AND l.sync_batch_id=@BatchId
                                  AND l.status='SUCCESS')
                """, batchId, now);
        }

        yield return Cmd(
            """
            INSERT INTO mdp_std_delivery_schedule
            (tenant_id, factory_id, source_system, delivery_plan_no, po_no, po_line, item_code, supplier_code, supplier_name, schedule_qty, sent_qty, rest_qty, return_qty, request_date, need_date, submit_date, last_sent_date, status, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.dsnum')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ponumber')),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.poline')) AS CHAR),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.suppliercode')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.supplier')),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.schedqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.schedqty')) AS DECIMAL(18,6)) END,0),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sentqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sentqty')) AS DECIMAL(18,6)) END,0),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.restqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.restqty')) AS DECIMAL(18,6)) END,0),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.returnqty')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.returnqty')) AS DECIMAL(18,6)) END,0),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.requestdate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.needdate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.submitdate')), 'null'), ''),
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.lastsentdate')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.status')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_delivery
            WHERE source_table='srm_polist_ds' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.dsnum')), '') <> ''
            ON DUPLICATE KEY UPDATE factory_id=VALUES(factory_id), po_no=VALUES(po_no), po_line=VALUES(po_line), item_code=VALUES(item_code), supplier_code=VALUES(supplier_code),
                supplier_name=VALUES(supplier_name), schedule_qty=VALUES(schedule_qty), sent_qty=VALUES(sent_qty), rest_qty=VALUES(rest_qty),
                return_qty=VALUES(return_qty), request_date=VALUES(request_date), need_date=VALUES(need_date),
                submit_date=VALUES(submit_date), last_sent_date=VALUES(last_sent_date), status=VALUES(status), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_delivery_result
            (tenant_id, factory_id, source_system, delivery_no, delivery_line, delivery_plan_no, po_no, po_line, item_code, delivery_qty, receipt_qty, return_qty, receipt_date, qc_status, event_time, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.shddh')), source_row_id),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.id')) AS CHAR),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.dsnum')),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.po_bill')),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.po_billline')) AS CHAR),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.itemnum')), ''),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sh_delivery_quantity')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.sh_delivery_quantity')) AS DECIMAL(18,6)) END,0),
                   0,
                   0,
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.receipt_date')), 'null'), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.qc_status')),
                   COALESCE(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.updatetime')), 'null'), ''), NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.createtime')), 'null'), '')),
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_delivery
            WHERE source_table IN ('scm_shd','scm_shdzb')
            ON DUPLICATE KEY UPDATE delivery_qty=VALUES(delivery_qty), receipt_qty=VALUES(receipt_qty), return_qty=VALUES(return_qty),
                receipt_date=VALUES(receipt_date), event_time=VALUES(event_time), sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
                update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO mdp_std_material_readiness
            (tenant_id, factory_id, source_system, work_order, op_code, item_code, component_item_code, required_qty, issued_qty, received_qty, available_qty, in_transit_qty, incoming_qty, shortage_qty, ready_status, need_date, supplier_code, source_biz_key, sync_batch_id, sync_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), COALESCE((SELECT r.source_system FROM mdp_source_table_registry r WHERE r.source_table = source_table LIMIT 1), NULLIF(source_system,''), ''),
                   JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')),
                   CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.Op')) AS CHAR),
                   IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.PMBOM')), ''),
                   COALESCE(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.ItemNum')), source_row_id),
                   COALESCE(
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyRequired')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyRequired')) AS DECIMAL(18,6)) END,
                       CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReq')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReq')) AS DECIMAL(18,6)) END,
                       0),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyIssued')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyIssued')) AS DECIMAL(18,6)) END, 0),
                   COALESCE(CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReceived')) REGEXP '^-?[0-9]+(\\.[0-9]+)?$' THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.QtyReceived')) AS DECIMAL(18,6)) END, 0),
                   0, 0, 0,
                   0,
                   'READY',
                   NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.DueDate')), 'null'), ''),
                   NULL,
                   source_biz_key, @BatchId, @Now
            FROM mdp_stg_work_order_material
            WHERE source_table='WorkOrdDetail' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(raw_data,'$.WorkOrd')), '') <> ''
            ON DUPLICATE KEY UPDATE factory_id=VALUES(factory_id), required_qty=VALUES(required_qty), issued_qty=VALUES(issued_qty),
                received_qty=VALUES(received_qty), available_qty=VALUES(available_qty), in_transit_qty=VALUES(in_transit_qty),
                incoming_qty=VALUES(incoming_qty), shortage_qty=VALUES(shortage_qty), ready_status=VALUES(ready_status),
                need_date=VALUES(need_date), supplier_code=VALUES(supplier_code), sync_batch_id=VALUES(sync_batch_id),
                sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        // ── 已移除：RoutingOpDetail → mdp_std_process_outsource_order 的 STD 物化 ──────────
        //
        // 这条 INSERT 的驱动集是 `WHERE source_table='RoutingOpDetail'`，**只**服务那一个来源，
        // 因此随 S3_ROUTING_OUTSOURCE 入站注册一并移除（见文件末尾 S3MdpEntityConfig.All 的说明）。
        //
        // 它把 S0 主数据（标准工艺路线明细）当成委外交付事实物化：源侧没有 po_no / po_line，
        // 只能硬编码成 NULL，而下游 dwd_process_outsource_delivery 的 uk_work_op_stat 含这两列 ——
        // NULL-distinct 让 ODKU 永不命中，每轮全量追加，实测 3244 万行。
        //
        // mdp_std_process_outsource_order 的另一个写入方是 Supply/ProcessOutsourceOrderService.cs
        // （PurOrdMaster Potype='PW' JOIN PurOrdDetail，字段齐全），**不在本文件内、不受影响**，
        // 移除后它成为该表唯一的写入方。
    }

    private IEnumerable<S3MdpSqlCommand> BuildDwdCommands(string batchId, DateTime now, bool retireStale)
    {
        yield return Cmd(
            """
            INSERT INTO dwd_supply_demand
            (tenant_id, factory_id, stat_date, demand_no, demand_line, demand_type, item_code, item_name, supplier_code, required_qty, fulfilled_qty, shortage_qty, required_date, demand_status, source_system, sync_batch_id, calc_time)
            SELECT tenant_id, COALESCE(NULLIF(factory_id,0),1), @StatDate, demand_no, IFNULL(demand_line,''), demand_type, IFNULL(item_code,''), IFNULL(item_name,''), supplier_code,
                   IFNULL(required_qty,0), 0, IFNULL(required_qty,0), required_date, status, source_system, @BatchId, @Now
            FROM mdp_std_supply_demand
            WHERE IFNULL(demand_no,'') <> ''
            ON DUPLICATE KEY UPDATE item_code=VALUES(item_code), item_name=VALUES(item_name), supplier_code=VALUES(supplier_code),
                required_qty=VALUES(required_qty), fulfilled_qty=VALUES(fulfilled_qty), shortage_qty=VALUES(shortage_qty),
                demand_status=VALUES(demand_status), sync_batch_id=VALUES(sync_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO dwd_supplier_delivery
            (tenant_id, factory_id, stat_date, po_no, po_line, po_type, supplier_code, supplier_name, item_code, item_name, order_qty, schedule_qty, delivery_qty, receipt_qty, return_qty, remaining_qty, due_date, need_date, last_delivery_date, last_receipt_date, delivery_status, risk_level, source_system, sync_batch_id, calc_time)
            SELECT po.tenant_id, COALESCE(NULLIF(po.factory_id,0),1), @StatDate, po.po_no, po.po_line, po.po_type, po.supplier_code, IFNULL(s.supplier_name, ds.supplier_name),
                   IFNULL(po.item_code,''), IFNULL(po.item_name,''), IFNULL(po.order_qty,0), IFNULL(ds.schedule_qty,0), IFNULL(dr.delivery_qty,0),
                   IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0), IFNULL(po.returned_qty,0) + IFNULL(dr.return_qty,0),
                   GREATEST(IFNULL(po.order_qty,0) - IFNULL(po.received_qty,0) - IFNULL(dr.receipt_qty,0) - IFNULL(po.returned_qty,0), 0),
                   po.due_date, COALESCE(ds.need_date, po.need_date), dr.event_time, dr.receipt_date,
                   CASE
                       WHEN IFNULL(po.order_qty,0) <= IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0) THEN 'COMPLETED'
                       WHEN COALESCE(ds.need_date, po.need_date, po.due_date) < @Now THEN 'DELAYED'
                       ELSE 'OPEN'
                   END,
                   CASE
                       WHEN COALESCE(ds.need_date, po.need_date, po.due_date) < @Now
                            AND IFNULL(po.order_qty,0) > IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0) THEN 'HIGH'
                       WHEN IFNULL(po.order_qty,0) > IFNULL(po.received_qty,0) + IFNULL(dr.receipt_qty,0) THEN 'MEDIUM'
                       ELSE 'LOW'
                   END,
                   'AIDOP_NATIVE', @BatchId, @Now
            FROM mdp_std_purchase_order po
            LEFT JOIN (
                SELECT tenant_id, factory_id, po_no, po_line, MIN(supplier_name) AS supplier_name, SUM(IFNULL(schedule_qty,0)) AS schedule_qty, MIN(need_date) AS need_date
                FROM mdp_std_delivery_schedule
                WHERE IFNULL(po_no,'') <> '' AND IFNULL(po_line,'') <> ''
                GROUP BY tenant_id, factory_id, po_no, po_line
            ) ds ON po.tenant_id=ds.tenant_id AND po.factory_id=ds.factory_id AND po.po_no=ds.po_no AND po.po_line=ds.po_line
            LEFT JOIN (
                SELECT tenant_id, factory_id, po_no, po_line, SUM(IFNULL(delivery_qty,0)) AS delivery_qty, SUM(IFNULL(receipt_qty,0)) AS receipt_qty,
                       SUM(IFNULL(return_qty,0)) AS return_qty, MAX(event_time) AS event_time, MAX(receipt_date) AS receipt_date
                FROM mdp_std_delivery_result
                WHERE IFNULL(po_no,'') <> '' AND IFNULL(po_line,'') <> ''
                GROUP BY tenant_id, factory_id, po_no, po_line
            ) dr ON po.tenant_id=dr.tenant_id AND po.factory_id=dr.factory_id AND po.po_no=dr.po_no AND po.po_line=dr.po_line
            LEFT JOIN (
                SELECT tenant_id, factory_id, supplier_code, MAX(supplier_name) AS supplier_name
                FROM mdp_std_supplier
                WHERE IFNULL(supplier_code,'') <> ''
                GROUP BY tenant_id, factory_id, supplier_code
            ) s ON po.tenant_id=s.tenant_id AND po.factory_id=s.factory_id AND po.supplier_code=s.supplier_code
            WHERE IFNULL(po.po_no,'') <> '' AND IFNULL(po.po_line,'') <> ''
            ON DUPLICATE KEY UPDATE supplier_name=VALUES(supplier_name), item_code=VALUES(item_code), item_name=VALUES(item_name),
                order_qty=VALUES(order_qty), schedule_qty=VALUES(schedule_qty), delivery_qty=VALUES(delivery_qty), receipt_qty=VALUES(receipt_qty),
                return_qty=VALUES(return_qty), remaining_qty=VALUES(remaining_qty), delivery_status=VALUES(delivery_status),
                risk_level=VALUES(risk_level), sync_batch_id=VALUES(sync_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        // ── 同日快照淘汰：让 MAX(stat_date) 成为「单一批次」而不是「某一天」 ──────────────
        //
        // Rule01 的 Provider 取 stat_date 最大的那一天，然后读该天全部行。
        // 但 DWD 是按 (tenant, factory, po_no, po_line, stat_date) 唯一键的 upsert，
        // 同一天被跑两轮时，前一轮产出而后一轮没产出的行会**保留旧批次号留在同一天**——
        // 于是 MAX(stat_date) 取到的是「半个旧批 + 半个新批」。实测已发生 3 次
        // （如 2026-06-07 的 178 行 + 7 行分属两个批次）。
        //
        // 本轮 INSERT 已给所有应存在的行盖上 @BatchId，故同日里批次号不等于 @BatchId 的行
        // 要么是本轮源侧已不存在的，要么是旁路写入方留下的（它们用 S3_DELIVERY_ / S3_OUTSOURCE_ 前缀）。
        //
        // 放在 INSERT **之后**而不是之前：先删后插会让快照短暂为空，
        // 而这里没有事务，并发的 Rule01 取数会读到一个空快照并把全部在办异常判成已恢复。
        if (retireStale)
        {
            yield return Cmd(
                """
                DELETE w FROM dwd_supplier_delivery w
                 WHERE w.tenant_id=@TenantId
                   AND COALESCE(NULLIF(w.factory_id,0),1)=@FactoryId
                   AND w.stat_date=@StatDate
                   AND IFNULL(w.sync_batch_id,'') <> @BatchId
                """, batchId, now);
        }

        yield return Cmd("DELETE FROM dwd_supplier_risk WHERE stat_date=@StatDate", batchId, now);
        yield return Cmd(
            """
            INSERT INTO dwd_supplier_risk
            (tenant_id, factory_id, stat_date, supplier_code, supplier_name, item_code, risk_type, risk_level, risk_count, risk_qty, risk_reason, source_system, calc_batch_id, calc_time)
            SELECT tenant_id, factory_id, stat_date, IFNULL(supplier_code,''), supplier_name, IFNULL(item_code,''),
                   'DELIVERY_DELAY', risk_level, COUNT(1), SUM(IFNULL(remaining_qty,0)),
                   '供应交付存在延期或未完成风险', 'AIDOP_NATIVE', @BatchId, @Now
            FROM dwd_supplier_delivery
            WHERE stat_date=@StatDate AND risk_level IN ('HIGH','MEDIUM') AND IFNULL(supplier_code,'') <> ''
            GROUP BY tenant_id, factory_id, stat_date, supplier_code, supplier_name, item_code, risk_level
            ON DUPLICATE KEY UPDATE risk_level=VALUES(risk_level), risk_count=VALUES(risk_count), risk_qty=VALUES(risk_qty),
                risk_reason=VALUES(risk_reason), calc_batch_id=VALUES(calc_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);

        yield return Cmd(
            """
            INSERT INTO dwd_process_outsource_delivery
            (tenant_id, factory_id, stat_date, work_order, op_code, routing_code, supplier_code, supplier_name, po_no, po_line, order_qty, completed_qty, remaining_qty, due_date, delivery_status, risk_level, source_system, calc_batch_id, calc_time)
            -- po_no 与 po_line 都必须 IFNULL 成空串：uk_work_op_stat 含这两列，
            -- 而 MySQL 唯一索引是 NULL-distinct 的 —— 任一列留 NULL 都会让 ODKU 永不命中、
            -- 每轮跑批全量追加（实测曾膨胀到 3244 万行）。两列保护必须对称，缺一即失效。
            SELECT o.tenant_id, o.factory_id, @StatDate, o.work_order, o.op_code, o.routing_code, o.supplier_code, s.supplier_name, IFNULL(o.po_no,''), IFNULL(o.po_line,''),
                   IFNULL(o.order_qty,0), IFNULL(o.completed_qty,0), GREATEST(IFNULL(o.order_qty,0)-IFNULL(o.completed_qty,0),0),
                   o.due_date,
                   CASE WHEN IFNULL(o.completed_qty,0) >= IFNULL(o.order_qty,0) AND IFNULL(o.order_qty,0) > 0 THEN 'COMPLETED'
                        WHEN o.due_date < @Now THEN 'DELAYED' ELSE IFNULL(o.status,'OPEN') END,
                   CASE WHEN o.due_date < @Now AND IFNULL(o.completed_qty,0) < IFNULL(o.order_qty,0) THEN 'HIGH'
                        WHEN IFNULL(o.completed_qty,0) < IFNULL(o.order_qty,0) THEN 'MEDIUM' ELSE 'LOW' END,
                   'AIDOP_NATIVE', @BatchId, @Now
            FROM mdp_std_process_outsource_order o
            LEFT JOIN (
                SELECT tenant_id, factory_id, supplier_code, MAX(supplier_name) AS supplier_name
                FROM mdp_std_supplier
                WHERE IFNULL(supplier_code,'') <> ''
                GROUP BY tenant_id, factory_id, supplier_code
            ) s ON o.tenant_id=s.tenant_id AND o.factory_id=s.factory_id AND o.supplier_code=s.supplier_code
            WHERE (IFNULL(o.work_order,'') <> '' OR IFNULL(o.routing_code,'') <> '')
            ON DUPLICATE KEY UPDATE supplier_code=VALUES(supplier_code), supplier_name=VALUES(supplier_name), order_qty=VALUES(order_qty),
                completed_qty=VALUES(completed_qty), remaining_qty=VALUES(remaining_qty), due_date=VALUES(due_date),
                delivery_status=VALUES(delivery_status), risk_level=VALUES(risk_level), calc_batch_id=VALUES(calc_batch_id),
                calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
            """, batchId, now);
    }

    private async Task<int> UpsertMaterialReadinessDwdAsync(string batchId, DateTime now)
    {
        var scope = RequireScope();
        return await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO dwd_material_readiness
            (tenant_id, factory_id, stat_date, work_order, op_code, parent_item_code, component_item_code, component_item_name, required_qty, cumulative_required_qty, stock_available_qty, qc_pending_qty, in_transit_qty, delivery_reply_qty, available_qty, shortage_qty, ready_status, supplier_code, supplier_name, need_date, calc_batch_id, calc_time)
            SELECT s.tenant_id, s.factory_id, @StatDate, s.work_order, s.op_code, s.item_code, s.component_item_code, IFNULL(i.item_name, ''),
                   IFNULL(s.required_qty, 0), IFNULL(s.required_qty, 0), IFNULL(s.available_qty, 0), 0, IFNULL(s.in_transit_qty, 0),
                   IFNULL(s.incoming_qty, 0), IFNULL(s.available_qty, 0), IFNULL(s.shortage_qty, 0),
                   CASE WHEN IFNULL(s.shortage_qty, 0) > 0 THEN 'SHORTAGE' ELSE 'READY' END,
                   s.supplier_code, IFNULL(sp.supplier_name, ''), s.need_date, @BatchId, @Now
            FROM mdp_std_material_readiness s
            LEFT JOIN mdp_std_item i ON s.tenant_id=i.tenant_id AND s.factory_id=i.factory_id AND s.component_item_code=i.item_code
            LEFT JOIN mdp_std_supplier sp ON s.tenant_id=sp.tenant_id AND s.factory_id=sp.factory_id AND s.supplier_code=sp.supplier_code
            WHERE s.tenant_id = @TenantId
              AND COALESCE(NULLIF(s.factory_id, 0), 1) = @FactoryId
              AND IFNULL(s.work_order, '') <> ''
            ON DUPLICATE KEY UPDATE component_item_name=VALUES(component_item_name), required_qty=VALUES(required_qty),
                cumulative_required_qty=VALUES(cumulative_required_qty), stock_available_qty=VALUES(stock_available_qty),
                qc_pending_qty=VALUES(qc_pending_qty), in_transit_qty=VALUES(in_transit_qty), delivery_reply_qty=VALUES(delivery_reply_qty),
                available_qty=VALUES(available_qty), shortage_qty=VALUES(shortage_qty), ready_status=VALUES(ready_status),
                supplier_code=VALUES(supplier_code), supplier_name=VALUES(supplier_name), need_date=VALUES(need_date),
                calc_batch_id=VALUES(calc_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
            """,
            new SugarParameter("@StatDate", now.Date),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));
    }

    private async Task<int> DeleteCurrentShortageAsync(DateTime statDate)
    {
        var scope = RequireScope();
        return await _db.Ado.ExecuteCommandAsync(
            """
            DELETE s
            FROM dwd_material_shortage s
            JOIN (
                SELECT DISTINCT work_order
                FROM mdp_std_material_readiness
                WHERE tenant_id = @TenantId
                  AND COALESCE(NULLIF(factory_id, 0), 1) = @FactoryId
                  AND IFNULL(work_order, '') <> ''
            ) w ON s.work_order = w.work_order
            WHERE s.tenant_id = @TenantId
              AND COALESCE(NULLIF(s.factory_id, 0), 1) = @FactoryId
              AND s.stat_date = @StatDate
            """,
            new SugarParameter("@StatDate", statDate),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));
    }

    private async Task<int> InsertMaterialShortageAsync(string batchId, DateTime now)
    {
        var scope = RequireScope();
        return await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO dwd_material_shortage
            (tenant_id, factory_id, stat_date, work_order, op_code, component_item_code, shortage_qty, shortage_reason, expected_supply_date, supplier_code, related_po_no, risk_level, calc_batch_id, calc_time)
            SELECT s.tenant_id, s.factory_id, @StatDate, s.work_order, s.op_code, s.component_item_code, IFNULL(s.shortage_qty, 0),
                   '标准层缺料数量大于 0', s.need_date, s.supplier_code, NULL,
                   CASE WHEN IFNULL(s.shortage_qty, 0) > IFNULL(s.required_qty, 0) THEN 'HIGH'
                        WHEN IFNULL(s.shortage_qty, 0) > 0 THEN 'MEDIUM' ELSE 'LOW' END,
                   @BatchId, @Now
            FROM mdp_std_material_readiness s
            WHERE s.tenant_id = @TenantId
              AND COALESCE(NULLIF(s.factory_id, 0), 1) = @FactoryId
              AND IFNULL(s.work_order, '') <> ''
              AND IFNULL(s.shortage_qty, 0) > 0
            """,
            new SugarParameter("@StatDate", now.Date),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId));
    }

    private async Task<long> InsertSyncLogAsync(long entityId, string entityName, string batchId, int rowsRead)
    {
        EnsureRunScope();
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_sync_log
            (tenant_id, entity_id, source_code, entity_name, sync_batch_id, sync_type, trigger_type, sync_start, rows_read, status)
            VALUES (@TenantId, @EntityId, 'AIDOPDEV_MYSQL', @EntityName, @BatchId, 'FULL', 'AUTO', NOW(), @RowsRead, 'RUNNING')
            """,
            new SugarParameter("@TenantId", _runScope.TenantId),
            new SugarParameter("@EntityId", entityId),
            new SugarParameter("@EntityName", entityName),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@RowsRead", rowsRead));
        return await _db.Ado.GetLongAsync(
            "SELECT id FROM mdp_sync_log WHERE sync_batch_id=@BatchId AND entity_id=@EntityId ORDER BY id DESC LIMIT 1",
            new List<SugarParameter>
            {
                new("@BatchId", batchId),
                new("@EntityId", entityId)
            });
    }

    private async Task MarkSyncLogSuccessAsync(long logId, DateTime started, int affected)
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            UPDATE mdp_sync_log
            SET sync_end=NOW(), duration_ms=@DurationMs, rows_insert=@RowsInsert, rows_update=0, rows_skip=0, rows_error=0, status='SUCCESS'
            WHERE id=@Id
            """,
            new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
            new SugarParameter("@RowsInsert", affected),
            new SugarParameter("@Id", logId));
    }

    private async Task MarkSyncLogFailedAsync(long logId, DateTime started, string message)
    {
        try
        {
            await _db.Ado.ExecuteCommandAsync(
                """
                UPDATE mdp_sync_log
                SET sync_end=NOW(), duration_ms=@DurationMs, rows_error=1, status='FAILED', error_msg=@ErrorMsg
                WHERE id=@Id
                """,
                new SugarParameter("@DurationMs", (int)(DateTime.Now - started).TotalMilliseconds),
                new SugarParameter("@ErrorMsg", message.Length <= 1000 ? message : message[..1000]),
                new SugarParameter("@Id", logId));
        }
        catch (Exception ex)
        {
            // 写库自身失败兜底：避免再抛掩盖原异常；遗留 RUNNING 行可由运维手动清理
            Console.Error.WriteLine($"[S3MdpSyncTransform] MarkSyncLogFailed write failed (syncLogId={logId}): {ex.Message}");
        }
    }

    private async Task<long> InsertTransformRunLogAsync(string batchId, DateTime startedAt, string triggerType)
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_transform_run_log
            (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
            VALUES (@TenantId, 'S3_MDP_SYNC_TRANSFORM', 'S3 MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            new SugarParameter("@TenantId", RequireScope().TenantId),
            new SugarParameter("@TriggerType", NormalizeTriggerType(triggerType)),
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@StartTime", startedAt));
        return await _db.Ado.GetLongAsync(
            "SELECT id FROM mdp_transform_run_log WHERE batch_id=@BatchId ORDER BY id DESC LIMIT 1",
            new List<SugarParameter> { new("@BatchId", batchId) });
    }

    private async Task MarkTransformRunSuccessAsync(long runLogId, DateTime startedAt, S3MdpSyncTransformResult result)
    {
        var finishedAt = DateTime.Now;
        await _db.Ado.ExecuteCommandAsync(
            """
            UPDATE mdp_transform_run_log
            SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
                stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=@DwdRows,
                summary_json=@SummaryJson, update_time=CURRENT_TIMESTAMP
            WHERE id=@Id
            """,
            new SugarParameter("@EndTime", finishedAt),
            new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
            new SugarParameter("@StageRows", result.StageRows),
            new SugarParameter("@StandardRows", result.StandardRows),
            new SugarParameter("@DwdRows", result.DwdRows),
            new SugarParameter("@SummaryJson", BuildRunSummaryJson(result)),
            new SugarParameter("@Id", runLogId));
    }

    private async Task MarkTransformRunFailedAsync(long runLogId, DateTime startedAt, string message)
    {
        try
        {
            var finishedAt = DateTime.Now;
            await _db.Ado.ExecuteCommandAsync(
                """
                UPDATE mdp_transform_run_log
                SET status='FAILED', end_time=@EndTime, duration_ms=@DurationMs,
                    error_message=@ErrorMessage, update_time=CURRENT_TIMESTAMP
                WHERE id=@Id
                """,
                new SugarParameter("@EndTime", finishedAt),
                new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
                new SugarParameter("@ErrorMessage", Truncate(message, 2000)),
                new SugarParameter("@Id", runLogId));
        }
        catch (Exception ex)
        {
            // 写库自身失败兜底（典型场景：远端 MySQL 瞬断导致 MarkFailed 自身也连不上）
            Console.Error.WriteLine($"[S3MdpSyncTransform] MarkTransformRunFailed write failed (runLogId={runLogId}): {ex.Message}");
        }
    }

    private S3MdpSqlCommand Cmd(string sql, string batchId, DateTime now)
    {
        var scope = RequireScope();
        return new S3MdpSqlCommand(MdpSqlScope.InjectTenantFactory(sql), new[]
        {
            new SugarParameter("@BatchId", batchId),
            new SugarParameter("@Now", now),
            new SugarParameter("@StatDate", now.Date),
            new SugarParameter("@TenantId", scope.TenantId),
            new SugarParameter("@FactoryId", scope.FactoryId)
        });
    }

    private MdpRebuildScope RequireScope()
    {
        EnsureRunScope();
        return _runScope;
    }

    private void EnsureRunScope()
    {
        if (_runScope == null)
            throw new InvalidOperationException("S3 全量重算必须指定有效 tenantId/factoryId");
    }

    private static string BuildJsonObjectExpression(IEnumerable<string> columns)
    {
        var parts = columns.SelectMany(c => new[] { $"'{c.Replace("'", "''")}'", $"s.`{c}`" });
        return $"JSON_OBJECT({string.Join(",", parts)})";
    }

    private static string FindColumn(IEnumerable<string> columns, string expected)
    {
        return columns.First(u => string.Equals(u, expected, StringComparison.OrdinalIgnoreCase));
    }

    private static string NormalizeTriggerType(string? triggerType)
    {
        return string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
    }

    /// <summary>
    /// 运行摘要。改用 JsonSerializer 而非字符串插值：batchId 今天不可能含引号，
    /// 但拼 JSON 是会被后来者继承的脆弱写法，且这里要加的 publish 是嵌套对象。
    /// </summary>
    private static string BuildRunSummaryJson(S3MdpSyncTransformResult result)
    {
        return System.Text.Json.JsonSerializer.Serialize(new
        {
            batchId = result.BatchId,
            stageRows = result.StageRows,
            standardRows = result.StandardRows,
            dwdRows = result.DwdRows,
            kpiRows = result.KpiRows,
            publish = result.Publication is null
                ? null
                : new
                {
                    table = result.Publication.Table,
                    batchId = result.Publication.BatchId,
                    statDate = result.Publication.StatDate.ToString("yyyy-MM-dd"),
                    factoryId = result.Publication.FactoryId,
                    currentRows = result.Publication.CurrentRows
                }
        });
    }

    private static string ResolveKpiValueTable(int metricLevel)
    {
        return metricLevel switch
        {
            1 => "ado_s9_kpi_value_l1_day",
            2 => "ado_s9_kpi_value_l2_day",
            3 => "ado_s9_kpi_value_l3_day",
            4 => "ado_s9_kpi_value_l4_day",
            _ => "ado_s9_kpi_value_l2_day"
        };
    }

    private static decimal DefaultS3Target(string metricCode) => LegacyKpiCodeTargets.GetOrZero(metricCode);

    private static string ResolveKpiStatus(decimal actual, decimal target, string? direction, decimal? yellowThreshold, decimal? redThreshold)
    {
        if (target <= 0) return "gray";

        var ratio = actual / target * 100m;
        if (string.Equals(direction, "lower_is_better", StringComparison.OrdinalIgnoreCase))
        {
            if (actual <= target) return "green";
            if (ratio <= (yellowThreshold ?? 110m)) return "yellow";
            return ratio >= (redThreshold ?? 120m) ? "red" : "yellow";
        }

        if (actual >= target) return "green";
        if (ratio >= (yellowThreshold ?? 95m)) return "yellow";
        return ratio <= (redThreshold ?? 80m) ? "red" : "yellow";
    }

    private static string ResolveTrendFlag(decimal actual, decimal? previous)
    {
        if (previous == null) return "flat";
        if (actual > previous.Value) return "up";
        if (actual < previous.Value) return "down";
        return "flat";
    }

    private static string Truncate(string? raw, int maxLength)
    {
        if (string.IsNullOrEmpty(raw)) return string.Empty;
        return raw.Length <= maxLength ? raw : raw[..maxLength];
    }

    private sealed class S3ColumnRow
    {
        public string ColumnName { get; set; } = string.Empty;
    }

    private sealed class S3MdpEntityRow
    {
        public long Id { get; set; }
        public string EntityName { get; set; } = string.Empty;
    }

    private sealed class S3KpiCalcRow
    {
        public long TenantId { get; set; }
        public long FactoryId { get; set; }
        public string MetricCode { get; set; } = string.Empty;
        public decimal? MetricValue { get; set; }
    }

    private sealed class S3KpiMetaRow
    {
        public int MetricLevel { get; set; }
        public string Direction { get; set; } = "higher_is_better";
        public decimal? YellowThreshold { get; set; }
        public decimal? RedThreshold { get; set; }
    }

    private sealed class S3KpiValueRow
    {
        public long Id { get; set; }
        public decimal? MetricValue { get; set; }
        public decimal? TargetValue { get; set; }
    }
}

public sealed class S3MdpSyncTransformResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int StageRows { get; set; }
    public int StandardRows { get; set; }
    public int DwdRows { get; set; }
    public int KpiRows { get; set; }
    public int AtomicRows { get; set; }

    /// <summary>
    /// 本轮发布的当前快照。<b>null 表示本轮没有发布</b>（增量路径，或全量路径中途抛错）。
    /// <para>失败方向是保守的：抛错时调用方写 FAILED，本字段保持 null，
    /// 于是运行日志里不会出现「成功了但发布证据是半批」的记录。</para>
    /// </summary>
    public S3MdpSnapshotPublication? Publication { get; set; }
}

/// <summary>
/// 一次当前快照发布的结果。与 <c>status='SUCCESS'</c> 写在同一条 UPDATE 里
/// （见 <c>MarkTransformRunSuccessAsync</c>），因此不存在「已成功但没有发布证据」的中间态。
/// </summary>
public sealed class S3MdpSnapshotPublication
{
    /// <summary>被发布为当前快照的表。</summary>
    public string Table { get; set; } = string.Empty;

    /// <summary>本轮批次号。读者据此确认「当前快照」出自哪一次运行。</summary>
    public string BatchId { get; set; } = string.Empty;

    /// <summary>本轮快照所在的 stat_date。与 BatchId 一起构成读者的精确定位键。</summary>
    public DateTime StatDate { get; set; }

    /// <summary>发布作用域的工厂号。运行日志表本身没有工厂列，只能由这里承载。</summary>
    public long FactoryId { get; set; }

    /// <summary>发布后该批次的当前行数。<b>0 是合法值</b>，表示成功发布了一个空快照。</summary>
    public int CurrentRows { get; set; }
}

public sealed class S3MaterialRefreshResult
{
    public string BatchId { get; set; } = string.Empty;
    public int StdCount { get; set; }
    public int ReadinessCount { get; set; }
    public int ShortageCount { get; set; }
}

internal sealed record S3MdpSqlCommand(string Sql, SugarParameter[] Parameters);

internal sealed record S3MdpEntityConfig(
    string EntityCode,
    string SourceTable,
    string TargetTable,
    string SourceRowIdExpression,
    string SourceBizKeyExpression)
{
    public static readonly IReadOnlyList<S3MdpEntityConfig> All = new List<S3MdpEntityConfig>
    {
        new("S3_SUPPLIER", "SuppMaster", "mdp_stg_supplier", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Supp`,''))"),
        new("S3_CONSIGNEE_SUPPLIER", "ConsigneeAddressMaster", "mdp_stg_supplier", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Address`,''))"),
        new("S3_ITEM_ERP", "ItemMaster", "mdp_stg_item", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`ItemNum`,''))"),
        new("S3_ITEM_NEW", "ic_item", "mdp_stg_item", "Id", "CONCAT(IFNULL(s.`tenant_id`,0), ':', IFNULL(s.`number`,''))"),
        new("S3_SOURCE_LIST", "srm_purchase", "mdp_stg_source_list", "Id", "CONCAT(IFNULL(s.`number`,''), ':', IFNULL(s.`supplier_number`,''))"),
        new("S3_SUPPLY_DEMAND", "ic_demandschedule", "mdp_stg_supply_demand", "Id", "CAST(s.`Id` AS CHAR)"),
        new("S3_PURCHASE_REQUEST", "srm_pr_main", "mdp_stg_supply_demand", "Id", "COALESCE(s.`pr_billno`, CAST(s.`Id` AS CHAR))"),
        new("S3_PURCHASE_ORDER_MASTER", "PurOrdMaster", "mdp_stg_purchase_order", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`PurOrd`,''))"),
        new("S3_PURCHASE_ORDER_DETAIL", "PurOrdDetail", "mdp_stg_purchase_order", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`PurOrd`,''), ':', IFNULL(s.`Line`,''))"),
        new("S3_DELIVERY_PLAN", "srm_polist_ds", "mdp_stg_delivery", "Id", "s.`dsnum`"),
        new("S3_SHIPPER_MASTER", "scm_shd", "mdp_stg_delivery", "id", "COALESCE(s.`shddh`, CAST(s.`id` AS CHAR))"),
        new("S3_SHIPPER_DETAIL", "scm_shdzb", "mdp_stg_delivery", "id", "CONCAT(IFNULL(s.`glid`,''), ':', IFNULL(s.`id`,''))"),
        new("S3_RECEIPT_MASTER", "PurOrdRctMaster", "mdp_stg_receipt", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Receiver`,''))"),
        new("S3_RECEIPT_DETAIL", "PurOrdRctDetail", "mdp_stg_receipt", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`Receiver`,''), ':', IFNULL(s.`Line`,''))"),
        new("S3_WORK_ORDER_MASTER", "WorkOrdMaster", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''))"),
        new("S3_WORK_ORDER_DETAIL", "WorkOrdDetail", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`Op`,''), ':', IFNULL(s.`ItemNum`,''))"),
        new("S3_WORK_ORDER_ROUTING", "WorkOrdRouting", "mdp_stg_work_order_material", "RecID", "CONCAT(IFNULL(s.`Domain`,''), ':', IFNULL(s.`WorkOrd`,''), ':', IFNULL(s.`OP`,''))")

        // ── 已移除：S3_ROUTING_OUTSOURCE（源表 RoutingOpDetail）──────────────────────────
        //
        // RoutingOpDetail 是「S0 标准工艺路线明细」，属**主数据**，已由 S0 维度层建模为
        // bridge_item_routing_op（S0DimCatalog.cs），S3 这份属重复建模。
        //
        // 它没有 work_order / po_no / po_line / 订单量 / 完成量 / 交期，SupplierCode 100% 为空，
        // 只有 0.48% 的行真的是委外工序；却被当成委外交付事实写进 mdp_std_process_outsource_order，
        // 其中 po_no / po_line 只能硬编码为 NULL。而 dwd_process_outsource_delivery 的
        // uk_work_op_stat 含这两列 —— MySQL 唯一索引的 NULL-distinct 语义使 ODKU 永不命中，
        // 于是每轮跑批全量追加，实测膨胀至 3244 万行。
        //
        // 委外交付的唯一真实事实来自 PurOrdMaster(Potype='PW') JOIN PurOrdDetail，
        // 由 Supply/ProcessOutsourceOrderService.cs 写入 std，字段齐全，不受本次移除影响。
    };
}