AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Manufacturing/IpqcInspectionMdpSyncService.cs

using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
using Admin.NET.Plugin.AiDOP.Infrastructure;

namespace Admin.NET.Plugin.AiDOP.Manufacturing;

/// <summary>
/// S6 过程检验单（IPQC）数据中台只读同步转换服务（DOP 内部，独立于 S6MdpSyncTransformService 的 KPI 管线）。
///
/// 源：aidopdev.qms_gcjyd（过程检验单头，31 列）+ aidopdev.qms_gcjydzb（明细，100 列），归口自旧系统 dopdemo。
/// 链路：执行器 → mdp_stg_ipqc_pull（头 source_table='qms_gcjyd' + 明细 source_table='qms_gcjydzb'，共表）→ std（头/明细各读 pull stg）。
/// 双模式入站：mdp_entity=S6_IPQC_INSPECTION_HEAD / S6_IPQC_INSPECTION_DETAIL → Writer → mdp_stg_ipqc_pull；头/明细标准层均读 pull stg（MdpJsonSql 跨源类型兼容）。
///     SyncHeadStagingAsync / SyncDetailStagingAsync 保留为 vestigial 可观测贴源写入（mdp_stg_ipqc_inspection(_detail)），不再被 transform 消费。
///
/// 归口说明（CTO 拍板方案 A）：
///   qms_gcjyd/qms_gcjydzb 原仅存于旧系统 dopdemo（0 行）、aidopdev 无此表，且 dopdemo 非运行时连接。
///   本服务 EnsureTablesAsync 在 aidopdev 建同构空表（IF NOT EXISTS）作为归口源；
///   后续由上游一次性抽数填充，届时无需改代码即随刷新流通。
///
/// 约束：
///   - 只读 qms_gcjyd/qms_gcjydzb，仅写 mdp_stg_*/mdp_std_*/run_log；绝不 INSERT/UPDATE/DELETE 业务源单据表。
///   - 源表无 tenant/org/factory 列 → 标准层按常量 tenant_id=0, factory_id=1 落库。
///   - 明细头关联：qms_gcjydzb.glid -> qms_gcjyd.id；标准层经 std_head_id 回填 mdp_std_ipqc_inspection.id。
///   - 明细只转换已确认子集列（检验项目/依据/工序/方法/技术标准/上下限/测量仪器/检验人/检验时间/样本量/检验数量/不合格数量/判定/备注等）；
///     j1~j70 动态检验值列语义不明、后置不转换、不落 typed 字段。
///   - result_judgement（jgpd/pd）为字典码，原样透出，语义解码后置。
///   - 源为 0 行时转换成功完成、处理数为 0，不报错。
/// </summary>
public class IpqcInspectionMdpSyncService : ITransient
{
    private const string JobCode = "S6_IPQC_INSPECTION_MDP_SYNC";
    private const string InboundEntityCode = "S6_IPQC_INSPECTION_HEAD";
    // Phase 2：Detail(qms_gcjydzb) 迁通用管线，经执行器灌 mdp_stg_ipqc_pull（source_table='qms_gcjydzb'，与 Head 共表），transform 读 pull stg。
    private const string InboundDetailEntityCode = "S6_IPQC_INSPECTION_DETAIL";

    // 双源（dopdemorq SQL Server）：Head（qms_gcjyd）+ Detail（qms_gcjydzb），均无 UpdateTime → FULL。实体默认 status=0（就位不启用）。
    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
    private const string SqlServerHeadEntityCode = "S6_IPQC_INSPECTION_HEAD_SQLSERVER";
    private const string SqlServerDetailEntityCode = "S6_IPQC_INSPECTION_DETAIL_SQLSERVER";
    /// <summary>qms_gcjyd/qms_gcjydzb 无 tenant_id 列；165 侧亦同，须经账套映射解析。</summary>
    private const string SourceZtid = "pbxfxp";

    private readonly ISqlSugarClient _db;
    private readonly TransformRunLogFinalizer _runLogFinalizer;
    private readonly MdpSourcePullDispatcher _pullDispatcher;
    private readonly List<string> _tenantSkipWarnings = new();

    public IpqcInspectionMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher, TransformRunLogFinalizer runLogFinalizer)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _pullDispatcher = pullDispatcher;
    }

    /// <summary>
    /// 全量：执行器灌头 pull-stg + 明细贴源 → 标准化(头/明细读 stg)。
    /// </summary>
    public async Task<IpqcInspectionMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        var now = DateTime.Now;
        var batchId = $"S6_IPQC_INSP_FULL_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
        var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        _tenantSkipWarnings.Clear();

        try
        {
            // 源表无 tenant_id；ctx 须传入经账套映射解析的真实租户。
            var pullCtx = new MdpPullContext
            {
                TenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, 0),
                FullRefresh = true,
                TaskCode = "S6_IPQC_INSPECTION_INBOUND",
                BatchId = $"{batchId}_PULL"
            };
            var pullHead = await _pullDispatcher.PullByEntityCodeAsync(InboundEntityCode, pullCtx, cancellationToken);
            var pullDetail = await _pullDispatcher.PullByEntityCodeAsync(InboundDetailEntityCode, pullCtx, cancellationToken);
            result.HeadStageRows = pullHead.RowsWritten;
            result.DetailStageRows = pullDetail.RowsWritten;
            // 保留富贴源头/明细表写入（vestigial 可观测）；头/明细 std 均以 pull stg 为准
            await SyncHeadStagingAsync(batchId, now);
            await SyncDetailStagingAsync(batchId, now);
            result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now);
            result.DetailStandardRows = await TransformDetailStandardAsync(batchId, now);
            result.TenantSkipWarnings = _tenantSkipWarnings.ToList();
            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    /// <summary>本库 MySQL 贴源 source_system（与 MdpDbPullExecutor 落 stg 的取值一致，决定 uk_source_key 归属）。</summary>
    private const string NativeSourceSystem = "AIDOPDEV_MYSQL";

    /// <summary>targeted sync 的单据作用域（租户 + 检验单 id）。</summary>
    private sealed class IpqcBillScope
    {
        public long TenantId { get; init; }
        public long InspectionId { get; init; }
    }

    /// <summary>
    /// 单据级 targeted 同步（业务写入后立即可见），替代把 Full/Inbound 全表扫描挂到高频提交路径。
    ///
    /// 只处理 inspectionId 这一张过程检验单的<b>头表</b>：
    ///   qms_gcjyd (id=inspectionId AND tenant_id=tenantId) → mdp_stg_ipqc_pull → mdp_std_ipqc_inspection
    ///
    /// <b>刻意不同步明细</b>（qms_gcjydzb → mdp_std_ipqc_inspection_detail）：该表全仓无任何应用
    /// INSERT/UPDATE/DELETE，源不变则标准层不可能因业务写入而滞后，纳入即为无依据的额外开销。
    ///
    /// 边界：只读业务源、只写 mdp_stg_ipqc_pull / mdp_std_ipqc_inspection / run_log；
    ///   租户双重收口（源查询 + transform 均按 tenantId）；纯 upsert，无 DELETE / 无 orphan purge /
    ///   无 MdpStdFullReplace / 不复用 RunFullAsync 的 pbxfxp 账套映射租户。
    /// </summary>
    public async Task<IpqcInspectionMdpSyncResult> SyncInspectionAsync(
        long inspectionId, long tenantId, string triggerType = "BILL", CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        if (inspectionId <= 0 || tenantId <= 0)
            return new IpqcInspectionMdpSyncResult { BatchId = "SKIPPED_INVALID_SCOPE" };

        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        var now = DateTime.Now;
        var batchId = $"S6_IPQC_BILL_{now:yyyyMMddHHmmssfff}_{inspectionId}";
        var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
        var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        _tenantSkipWarnings.Clear();

        try
        {
            var scope = new IpqcBillScope { TenantId = tenantId, InspectionId = inspectionId };
            // ① 贴源：把该单据的源行刷进 pull stg（transform 一律读 stg，不先刷就还是旧快照）
            result.HeadStageRows = await UpsertHeadStgFromSourceAsync(scope, batchId, now);
            // ② 标准层：复用全量同一套 transform，仅加单据作用域，字段映射零分叉
            result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now, null, scope);
            result.TenantSkipWarnings = _tenantSkipWarnings.ToList();
            await MarkRunSuccessAsync(runLogId, now, result);
            return result;
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    /// <summary>
    /// 单据级头表贴源 upsert：qms_gcjyd 源行 → mdp_stg_ipqc_pull。
    /// raw_data 按源表<b>全列</b>动态生成 JSON_OBJECT，与 MdpDbPullExecutor/MdpStagingWriter 的信封形状一致
    /// （JSON 键 = 源列名；source_row_id = id；source_biz_key = djbh，空则回落 id，与 MdpStagingWriter.BuildBizKey
    /// 的 "biz_key_expr=djbh，取不到则回落 sourceRowId" 语义一致），因此命中既有 uk_source_key 行原地更新，
    /// 不产生重复 stg 行，也不会因 transform 将来新增字段而漂移。
    /// </summary>
    private async Task<int> UpsertHeadStgFromSourceAsync(IpqcBillScope scope, string batchId, DateTime now)
    {
        var cols = await _db.Ado.SqlQueryAsync<string>(
            "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='qms_gcjyd' ORDER BY ORDINAL_POSITION");
        // 列名取自 information_schema 且强制标识符白名单，非用户输入，无注入面
        cols = cols.Where(c => !string.IsNullOrWhiteSpace(c) && System.Text.RegularExpressions.Regex.IsMatch(c, "^[A-Za-z0-9_]+$")).ToList();
        if (cols.Count == 0) return 0;

        var jsonObj = "JSON_OBJECT(" + string.Join(", ", cols.Select(c => $"'{c}', s.`{c}`")) + ")";
        return await MdpSchemaAligner.ExecuteAsync(_db, 
            $"""
            INSERT INTO mdp_stg_ipqc_pull
            (tenant_id, source_system, source_table, source_row_id, source_biz_key, raw_data, sync_batch_id, sync_time, process_status, create_time)
            SELECT s.tenant_id, @SrcSys, 'qms_gcjyd', CAST(s.id AS CHAR),
                   IFNULL(NULLIF(s.djbh,''), CAST(s.id AS CHAR)), {jsonObj},
                   @BatchId, @Now, 'PENDING', @Now
            FROM qms_gcjyd s
            WHERE s.id=@Id AND s.tenant_id=@TenantId
            ON DUPLICATE KEY UPDATE
                tenant_id=VALUES(tenant_id), source_row_id=VALUES(source_row_id), raw_data=VALUES(raw_data),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time),
                process_status='PENDING', update_time=CURRENT_TIMESTAMP
            """,
            new List<SugarParameter>
            {
                new("@SrcSys", NativeSourceSystem),
                new("@BatchId", batchId),
                new("@Now", now),
                new("@Id", scope.InspectionId),
                new("@TenantId", scope.TenantId),
            });
    }

    /// <summary>
    /// 双模式入站：执行器抽数落 pull-stg，再跑标准层（头读 pull stg）。
    /// </summary>
    public async Task<IpqcInspectionInboundResult> RunInboundAsync(
        long tenantId = 0,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        var now = DateTime.Now;
        // 源表无 tenant_id；显式 tenantId≤0 时经账套映射解析真实租户。
        var resolvedTenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId);
        var pullCtx = new MdpPullContext
        {
            TenantId = resolvedTenantId,
            FullRefresh = fullRefresh,
            TaskCode = "S6_IPQC_INSPECTION_INBOUND",
            BatchId = $"S6_IPQC_IN_{now:yyyyMMddHHmmss}"
        };
        var pullHead = await _pullDispatcher.PullByEntityCodeAsync(InboundEntityCode, pullCtx, cancellationToken);
        var pullDetail = await _pullDispatcher.PullByEntityCodeAsync(InboundDetailEntityCode, pullCtx, cancellationToken);

        var batchId = $"S6_IPQC_STD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "INBOUND");
        var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        _tenantSkipWarnings.Clear();
        try
        {
            result.HeadStageRows = pullHead.RowsWritten;
            result.DetailStageRows = pullDetail.RowsWritten;
            await SyncHeadStagingAsync(batchId, now);
            await SyncDetailStagingAsync(batchId, now);
            result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now);
            result.DetailStandardRows = await TransformDetailStandardAsync(batchId, now);
            result.TenantSkipWarnings = _tenantSkipWarnings.ToList();
            await MarkRunSuccessAsync(runLogId, now, result);
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }

        return new IpqcInspectionInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pullHead.RowsPulled + pullDetail.RowsPulled,
            RowsWrittenStg = pullHead.RowsWritten + pullDetail.RowsWritten,
            NewCursor = pullDetail.NewCursor ?? pullHead.NewCursor,
            TransformBatchId = result.BatchId,
            HeadStageRows = result.HeadStageRows,
            HeadStandardRows = result.HeadStandardRows,
            DetailStageRows = result.DetailStageRows,
            DetailStandardRows = result.DetailStandardRows
        };
    }

    /// <summary>
    /// 双源切换 FULL Replace（Phase 1，仅 Head）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 pull-stg，
    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_ipqc_inspection（按 tenant 精确隔离），消除旧源独有业务键残留。
    /// qms_gcjyd 无 UpdateTime → FULL；用 PullAllByEntityCodeAsync + FullRefresh=true。
    /// <b>只处理 Head</b>：不调用 SyncHeadStagingAsync（vestigial 观测写入）/ SyncDetailStagingAsync / TransformDetailStandardAsync；
    /// IPQC Detail(qms_gcjydzb) 属 bespoke，留 Phase 2。
    /// SQLSERVER 实体默认 status=0（就位不启用）；dopdemorq 该表当前空，Phase 1 不实际执行 destructive 切换。
    /// </summary>
    public async Task<IpqcInspectionInboundResult> RunHeadSourceSwitchFullAsync(
        string sourceCode = SqlServerSourceCode,
        string headEntityCode = SqlServerHeadEntityCode,
        long tenantId = 0,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        // 165 SQL Server 源无 tenant_id；显式 tenantId≤0 时经账套映射解析。
        tenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId);

        var now = DateTime.Now;
        // 1) PullAll 抽尽 + fullRefresh=true。Pull 失败会抛异常，std 未动。
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S6_IPQC_INSPECTION_INBOUND",
            BatchId = $"S6_IPQC_SW_{now:yyyyMMddHHmmss}"
        };
        var pull = await _pullDispatcher.PullAllByEntityCodeAsync(headEntityCode, pullCtx, cancellationToken);

        // 2) FULL Replace（仅头 std）：事务内 DELETE 当前 tenant，再 INSERT 仅当前 source_system 的头结果。
        var batchId = $"S6_IPQC_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId, HeadStageRows = pull.RowsWritten };
        try
        {
            result.HeadStandardRows = await MdpStdFullReplace.ReplaceAsync(
                _db, "mdp_std_ipqc_inspection", tenantId, extraWhere: null,
                insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, sourceCode),
                cancellationToken);
            await MarkRunSuccessAsync(runLogId, now, result);
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }

        return new IpqcInspectionInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            TransformBatchId = batchId,
            HeadStageRows = result.HeadStageRows,
            HeadStandardRows = result.HeadStandardRows,
            DetailStageRows = 0,
            DetailStandardRows = 0
        };
    }

    /// <summary>
    /// 双源切换 FULL Replace（Detail）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 pull-stg（source_table='qms_gcjydzb'），
    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_ipqc_inspection_detail（按 tenant 精确隔离），消除旧源独有业务键残留。
    /// qms_gcjydzb 无 UpdateTime → FULL；用 PullAllByEntityCodeAsync + FullRefresh=true。std_head_id 回填依赖 Head std 已存在（LEFT JOIN，缺则 NULL）。
    /// <b>只处理 Detail</b>：不碰 Head std（切 Head 用 RunHeadSourceSwitchFullAsync）。
    /// SQLSERVER 实体默认 status=0（就位不启用）；dopdemorq 该表当前空，Phase 2 不实际执行 destructive 切换。
    /// </summary>
    public async Task<IpqcInspectionInboundResult> RunDetailSourceSwitchFullAsync(
        string sourceCode = SqlServerSourceCode,
        string detailEntityCode = SqlServerDetailEntityCode,
        long tenantId = 0,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsurePullStgTableAsync();

        // 165 SQL Server 源无 tenant_id；显式 tenantId≤0 时经账套映射解析。
        tenantId = AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId);

        var now = DateTime.Now;
        // 1) PullAll 抽尽 + fullRefresh=true。Pull 失败会抛异常，std 未动。
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S6_IPQC_INSPECTION_INBOUND",
            BatchId = $"S6_IPQC_DTL_SW_{now:yyyyMMddHHmmss}"
        };
        var pull = await _pullDispatcher.PullAllByEntityCodeAsync(detailEntityCode, pullCtx, cancellationToken);

        // 2) FULL Replace（仅明细 std）：事务内 DELETE 当前 tenant，再 INSERT 仅当前 source_system 的明细结果。
        var batchId = $"S6_IPQC_DTL_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new IpqcInspectionMdpSyncResult { BatchId = batchId, RunLogId = runLogId, DetailStageRows = pull.RowsWritten };
        try
        {
            result.DetailStandardRows = await MdpStdFullReplace.ReplaceAsync(
                _db, "mdp_std_ipqc_inspection_detail", tenantId, extraWhere: null,
                insertScopedAsync: () => TransformDetailStandardAsync(batchId, now, sourceCode),
                cancellationToken);
            await MarkRunSuccessAsync(runLogId, now, result);
        }
        catch (Exception ex)
        {
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await MarkRunFailedAsync(runLogId, now, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }

        return new IpqcInspectionInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            TransformBatchId = batchId,
            HeadStageRows = 0,
            HeadStandardRows = 0,
            DetailStageRows = result.DetailStageRows,
            DetailStandardRows = result.DetailStandardRows
        };
    }

    private async Task EnsurePullStgTableAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_ipqc_pull"));
    }

    /// <summary>防御式建表（幂等）：归口源空表 qms_gcjyd/qms_gcjydzb + 贴源层(头/明细) + 标准层(头/明细)。</summary>
    private async Task EnsureTablesAsync()
    {
        // 归口源表（方案 A）：同构自旧系统 dopdemo.qms_gcjyd 头表，空表，待上游抽数填充。
        // id 不设 AUTO_INCREMENT：抽数将携带旧系统主键。
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            CREATE TABLE IF NOT EXISTS qms_gcjyd (
                id BIGINT NOT NULL PRIMARY KEY COMMENT '主表id',
                tenant_id BIGINT NOT NULL COMMENT 'tenant_id',
                djbh VARCHAR(255) NULL COMMENT '单据编号',
                cplx VARCHAR(255) NULL COMMENT '产品型号',
                scph VARCHAR(255) NULL COMMENT '生产批号',
                jyrq VARCHAR(255) NULL COMMENT '检验日期',
                lydjbh VARCHAR(255) NULL COMMENT '来源单据编号',
                jgpd BIGINT NULL COMMENT '结果判定',
                jyhgsl DECIMAL(10,0) NULL COMMENT '检验合格数量',
                jybhgsl DECIMAL(10,0) NULL COMMENT '检验不合格数量',
                fj TEXT NULL COMMENT '附件',
                bz TEXT NULL COMMENT '备注',
                jyr VARCHAR(255) NULL COMMENT '检验人',
                gxbm VARCHAR(255) NULL COMMENT '工序编码',
                gxmc VARCHAR(255) NULL COMMENT '工序名称',
                sczyry VARCHAR(255) NULL COMMENT '生产作业人员',
                ybl DECIMAL(10,0) NULL COMMENT '样本量',
                bdbh VARCHAR(255) NULL COMMENT '表单编号',
                bbh VARCHAR(255) NULL COMMENT '版本号',
                sxrq DATETIME NULL COMMENT '生效日期',
                wlbm VARCHAR(255) NULL COMMENT '物料编码',
                wlmc VARCHAR(255) NULL COMMENT '物料名称',
                lysl DECIMAL(10,0) NULL COMMENT '留样数量',
                phsl DECIMAL(10,0) NULL COMMENT '破坏数量',
                jygfid VARCHAR(255) NULL COMMENT '检验规范',
                jygfbbm VARCHAR(255) NULL COMMENT '没有用',
                jgbb VARCHAR(255) NULL COMMENT '检规版本',
                jgbh VARCHAR(255) NULL COMMENT '检规编号',
                title VARCHAR(255) NULL,
                jyr1 VARCHAR(255) NULL,
                ChangeStatus INT NULL,
                status VARCHAR(255) NULL COMMENT '检验状态'
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='过程检验单(头) 归口自旧系统dopdemo，空表待上游抽数'
            """);
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_ipqc_inspection"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_ipqc_inspection"));
        // 归口源明细表（方案 A）：同构自旧系统 dopdemo.qms_gcjydzb（100 列），空表，待上游抽数填充。
        // 关联键 glid -> qms_gcjyd.id。id 不设 AUTO_INCREMENT：抽数将携带旧系统主键。
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            CREATE TABLE IF NOT EXISTS qms_gcjydzb (
                id BIGINT NOT NULL PRIMARY KEY COMMENT '子表id',
                tenant_id BIGINT NOT NULL COMMENT 'tenant_id',
                glid BIGINT NULL COMMENT '关联id',
                dwsj TEXT NULL, jyxm TEXT NULL, jybz TEXT NULL, jyyj TEXT NULL,
                j1 TEXT NULL, j2 TEXT NULL, j3 TEXT NULL, j4 TEXT NULL, j5 TEXT NULL,
                ljbh TEXT NULL, sczyry TEXT NULL, fj TEXT NULL, bz TEXT NULL,
                j6 TEXT NULL, j7 TEXT NULL, j8 TEXT NULL, j9 TEXT NULL, j10 TEXT NULL,
                j11 TEXT NULL, j12 TEXT NULL, j13 TEXT NULL, j14 TEXT NULL, j15 TEXT NULL,
                j16 TEXT NULL, j17 TEXT NULL, j18 TEXT NULL, j19 TEXT NULL, j20 TEXT NULL,
                j21 TEXT NULL, j22 TEXT NULL, j23 TEXT NULL, j24 TEXT NULL, j25 TEXT NULL,
                j26 TEXT NULL, j27 TEXT NULL, j28 TEXT NULL, j29 TEXT NULL, j30 TEXT NULL,
                j31 TEXT NULL, j32 TEXT NULL, j33 TEXT NULL, j34 TEXT NULL, j35 TEXT NULL,
                j36 TEXT NULL, j37 TEXT NULL, j38 TEXT NULL, j39 TEXT NULL, j40 TEXT NULL,
                pd BIGINT NULL COMMENT '结果判定',
                jysl DECIMAL(10,0) NULL COMMENT '检验数量',
                bhgsl DECIMAL(10,0) NULL COMMENT '不合格数量',
                j41 TEXT NULL, j42 TEXT NULL, j43 TEXT NULL, j44 TEXT NULL, j45 TEXT NULL,
                j46 TEXT NULL, j47 TEXT NULL, j48 TEXT NULL, j49 TEXT NULL, j50 TEXT NULL,
                j51 TEXT NULL, j52 TEXT NULL, j53 TEXT NULL, j54 TEXT NULL, j55 TEXT NULL,
                j56 TEXT NULL, j57 TEXT NULL, j58 TEXT NULL, j59 TEXT NULL, j60 TEXT NULL,
                j61 TEXT NULL, j62 TEXT NULL, j63 TEXT NULL, j64 TEXT NULL, j65 TEXT NULL,
                j66 TEXT NULL, j67 TEXT NULL, j68 TEXT NULL, j69 TEXT NULL, j70 TEXT NULL,
                nature TEXT NULL, jsbz TEXT NULL, jypc TEXT NULL, ggbh TEXT NULL,
                ProcessCode TEXT NULL, ProcessName TEXT NULL, jyff TEXT NULL,
                ybl INT NULL COMMENT '样本量',
                NonNumericalTypeOK INT NULL COMMENT '非数值型OK',
                NonNumericalTypeNG INT NULL COMMENT '非数值型NG',
                Inspector TEXT NULL,
                InspectionTime DATETIME NULL COMMENT '检验时间',
                jgid BIGINT NULL,
                AbnormalNumber TEXT NULL, MeasurementInstrument TEXT NULL, sx TEXT NULL, xx TEXT NULL,
                KEY idx_qms_gcjydzb_glid (glid)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='过程检验单(明细) 归口自旧系统dopdemo，空表待上游抽数'
            """);
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_stg_ipqc_inspection_detail"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_ipqc_inspection_detail"));
    }

    /// <summary>贴源头：qms_gcjyd -> mdp_stg_ipqc_inspection（raw_data JSON 快照）。跳过无有效 tenant_id 的行。返回写入行数。</summary>
    private async Task<int> SyncHeadStagingAsync(string batchId, DateTime now)
    {
        var total = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjyd");
        var skipped = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjyd m WHERE NULLIF(m.tenant_id, 0) IS NULL");
        if (skipped > 0)
            _tenantSkipWarnings.Add($"qms_gcjyd 贴源跳过 {skipped} 行：无法解析 tenant_id");
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_stg_ipqc_inspection
            (tenant_id, factory_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
            SELECT
                NULLIF(m.tenant_id, 0), 1, 'AIDOP_NATIVE', 'qms_gcjyd',
                CAST(m.id AS CHAR), CAST(m.djbh AS CHAR), @BatchId, @Now, 'PENDING',
                JSON_OBJECT(
                    'id', m.id, 'djbh', m.djbh, 'cplx', m.cplx, 'scph', m.scph, 'lydjbh', m.lydjbh,
                    'jgpd', m.jgpd, 'fj', CAST(m.fj AS CHAR), 'bz', CAST(m.bz AS CHAR), 'jyr', m.jyr,
                    'gxbm', m.gxbm, 'gxmc', m.gxmc, 'sczyry', m.sczyry, 'ybl', m.ybl,
                    'bdbh', m.bdbh, 'bbh', m.bbh, 'sxrq', m.sxrq, 'wlbm', m.wlbm, 'wlmc', m.wlmc,
                    'jgbb', m.jgbb, 'jgbh', m.jgbh, 'status', m.status
                )
            FROM qms_gcjyd m
            WHERE NULLIF(m.tenant_id, 0) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                source_biz_key=VALUES(source_biz_key), 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("@BatchId", batchId),
            new SugarParameter("@Now", now));
        return total - skipped;
    }

    /// <summary>标准化头：mdp_stg_ipqc_pull(qms_gcjyd) → mdp_std_ipqc_inspection。</summary>
    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null, IpqcBillScope? scope = null)
    {
        // 双源：sourceSystem 非空时仅统计/转换当前 source（切源 FULL Replace 用）；为空时保持既有全源行为不变。
        var countPars = new List<SugarParameter>();
        var countSrc = "";
        var insertSrc = "";
        if (sourceSystem != null)
        {
            countSrc = " AND source_system=@Src";
            insertSrc = " AND m.source_system=@Src";
            countPars.Add(new SugarParameter("@Src", sourceSystem));
        }
        var mTenant = MdpJsonSql.TenantFromStg("m");
        // 单据作用域（targeted sync）：按 source_row_id 收敛到单张检验单，并强制租户相等。
        // 租户显式来自业务单，不依赖 RunFullAsync 的 pbxfxp 账套映射上下文。scope 为 null 时全量行为逐字不变。
        if (scope != null)
        {
            var scopeSql = $" AND m.source_row_id=@ScopeRowId AND ({mTenant})=@ScopeTenant";
            countSrc += scopeSql;
            insertSrc += scopeSql;
            countPars.Add(new SugarParameter("@ScopeRowId", scope.InspectionId.ToString()));
            countPars.Add(new SugarParameter("@ScopeTenant", scope.TenantId));
        }
        var rows = await _db.Ado.GetIntAsync(
            $"SELECT COUNT(1) FROM mdp_stg_ipqc_pull m WHERE m.source_table='qms_gcjyd'{countSrc} AND {MdpJsonSql.TenantGuard(mTenant)}", countPars);
        var skipped = await _db.Ado.GetIntAsync(
            $"SELECT COUNT(1) FROM mdp_stg_ipqc_pull m WHERE m.source_table='qms_gcjyd'{countSrc} AND NOT {MdpJsonSql.TenantGuard(mTenant)}", countPars);
        if (skipped > 0)
            _tenantSkipWarnings.Add($"mdp_std_ipqc_inspection 跳过 {skipped} 行：无法解析 tenant_id");
        var insertSql =
            $"""
            INSERT INTO mdp_std_ipqc_inspection
            (tenant_id, factory_id, source_system, bill_no, product_model, production_batch_no, production_work_order,
             result_judgement, attachment, remark, inspector, process_code, process_name, production_person,
             sample_qty, form_no, version_no, effective_date, material_code, material_name,
             inspec_standard_version, inspec_standard_code, inspection_status,
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {mTenant}, 1, {MdpSourceIdentity.Resolve("m.source_table", "m.source_system")},
                IFNULL({MdpJsonSql.Str("m", "djbh")}, {MdpJsonSql.Str("m", "id")}),
                {MdpJsonSql.Str("m", "cplx")},
                {MdpJsonSql.Str("m", "scph")},
                {MdpJsonSql.Str("m", "lydjbh")},
                {MdpJsonSql.Str("m", "jgpd")},
                {MdpJsonSql.Str("m", "fj")},
                {MdpJsonSql.Str("m", "bz")},
                {MdpJsonSql.Str("m", "jyr")},
                {MdpJsonSql.Str("m", "gxbm")},
                {MdpJsonSql.Str("m", "gxmc")},
                {MdpJsonSql.Str("m", "sczyry")},
                {MdpJsonSql.Dec("m", "ybl", 18, 6)},
                {MdpJsonSql.Str("m", "bdbh")},
                {MdpJsonSql.Str("m", "bbh")},
                {MdpJsonSql.DateTimeSec("m", "sxrq")},
                {MdpJsonSql.Str("m", "wlbm")},
                {MdpJsonSql.Str("m", "wlmc")},
                {MdpJsonSql.Str("m", "jgbb")},
                {MdpJsonSql.Str("m", "jgbh")},
                {MdpJsonSql.Str("m", "status")},
                IFNULL(m.source_row_id, JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.id'))),
                IFNULL(NULLIF(m.source_biz_key,''), JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.djbh'))),
                @BatchId, @Now
            FROM mdp_stg_ipqc_pull m
            WHERE m.source_table='qms_gcjyd'{insertSrc} AND {MdpJsonSql.TenantGuard(mTenant)}
            ON DUPLICATE KEY UPDATE
                bill_no=VALUES(bill_no), product_model=VALUES(product_model), production_batch_no=VALUES(production_batch_no),
                production_work_order=VALUES(production_work_order), result_judgement=VALUES(result_judgement),
                attachment=VALUES(attachment), remark=VALUES(remark), inspector=VALUES(inspector),
                process_code=VALUES(process_code), process_name=VALUES(process_name), production_person=VALUES(production_person),
                sample_qty=VALUES(sample_qty), form_no=VALUES(form_no), version_no=VALUES(version_no),
                effective_date=VALUES(effective_date), material_code=VALUES(material_code), material_name=VALUES(material_name),
                inspec_standard_version=VALUES(inspec_standard_version), inspec_standard_code=VALUES(inspec_standard_code),
                inspection_status=VALUES(inspection_status),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """;
        var insPars = new List<SugarParameter>
        {
            new("@BatchId", batchId),
            new("@Now", now)
        };
        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
        if (scope != null)
        {
            insPars.Add(new SugarParameter("@ScopeRowId", scope.InspectionId.ToString()));
            insPars.Add(new SugarParameter("@ScopeTenant", scope.TenantId));
        }
        await MdpSchemaAligner.ExecuteAsync(_db, insertSql, insPars);
        return rows;
    }

    /// <summary>贴源明细：qms_gcjydzb -> mdp_stg_ipqc_inspection_detail（raw_data JSON 快照，确定子集列）。跳过无有效 tenant_id 的行。</summary>
    private async Task<int> SyncDetailStagingAsync(string batchId, DateTime now)
    {
        var total = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjydzb");
        var skipped = await _db.Ado.GetIntAsync("SELECT COUNT(1) FROM qms_gcjydzb d WHERE NULLIF(d.tenant_id, 0) IS NULL");
        if (skipped > 0)
            _tenantSkipWarnings.Add($"qms_gcjydzb 贴源跳过 {skipped} 行：无法解析 tenant_id");
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_stg_ipqc_inspection_detail
            (tenant_id, factory_id, source_system, source_table, source_row_id, source_biz_key, sync_batch_id, sync_time, process_status, raw_data)
            SELECT
                NULLIF(d.tenant_id, 0), 1, 'AIDOP_NATIVE', 'qms_gcjydzb',
                CAST(d.id AS CHAR), CAST(d.glid AS CHAR), @BatchId, @Now, 'PENDING',
                JSON_OBJECT(
                    'id', d.id, 'glid', d.glid, 'jyxm', CAST(d.jyxm AS CHAR), 'jyyj', CAST(d.jyyj AS CHAR),
                    'jsbz', CAST(d.jsbz AS CHAR), 'jyff', CAST(d.jyff AS CHAR), 'sx', CAST(d.sx AS CHAR), 'xx', CAST(d.xx AS CHAR),
                    'ProcessCode', CAST(d.ProcessCode AS CHAR), 'ProcessName', CAST(d.ProcessName AS CHAR),
                    'MeasurementInstrument', CAST(d.MeasurementInstrument AS CHAR), 'Inspector', CAST(d.Inspector AS CHAR),
                    'InspectionTime', d.InspectionTime, 'ybl', d.ybl, 'jysl', d.jysl, 'bhgsl', d.bhgsl,
                    'pd', d.pd, 'NonNumericalTypeOK', d.NonNumericalTypeOK, 'NonNumericalTypeNG', d.NonNumericalTypeNG,
                    'bz', CAST(d.bz AS CHAR), 'fj', CAST(d.fj AS CHAR)
                )
            FROM qms_gcjydzb d
            WHERE NULLIF(d.tenant_id, 0) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                source_biz_key=VALUES(source_biz_key), 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("@BatchId", batchId),
            new SugarParameter("@Now", now));
        return total - skipped;
    }

    /// <summary>
    /// 标准化明细：mdp_stg_ipqc_pull(qms_gcjydzb) → mdp_std_ipqc_inspection_detail（回填 std_head_id）。
    /// 全 typed 列经 MdpJsonSql 跨源类型兼容（ybl/jysl/bhgsl/NonNumerical* Dec/Int 加 NULL 守卫、InspectionTime 兼容 ISO-T）。
    /// sourceSystem 非空时仅统计/转换当前 source（切源 FULL Replace 用）；为空时保持既有全源行为。
    /// </summary>
    private async Task<int> TransformDetailStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
    {
        var srcClause = sourceSystem == null ? "" : " AND d.source_system=@Src";
        var dTenant = MdpJsonSql.TenantFromStg("d");

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        var rows = await _db.Ado.GetIntAsync(
            $"SELECT COUNT(1) FROM mdp_stg_ipqc_pull d WHERE d.source_table='qms_gcjydzb'{srcClause} AND {MdpJsonSql.TenantGuard(dTenant)}", countPars);
        var skipped = await _db.Ado.GetIntAsync(
            $"SELECT COUNT(1) FROM mdp_stg_ipqc_pull d WHERE d.source_table='qms_gcjydzb'{srcClause} AND NOT {MdpJsonSql.TenantGuard(dTenant)}", countPars);
        if (skipped > 0)
            _tenantSkipWarnings.Add($"mdp_std_ipqc_inspection_detail 跳过 {skipped} 行：无法解析 tenant_id");

        var insertSql =
            $"""
            INSERT INTO mdp_std_ipqc_inspection_detail
            (tenant_id, std_head_id, source_head_id, inspection_item, inspection_basis, technical_standard, inspection_method,
             upper_limit, lower_limit, process_code, process_name, measurement_instrument, inspector, inspection_time,
             sample_qty, inspection_qty, unqualified_qty, result_judgement, non_numeric_ok, non_numeric_ng, remark, attachment,
             source_row_id, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {dTenant}, h.id,
                {MdpJsonSql.Str("d", "glid")},
                {MdpJsonSql.Str("d", "jyxm")},
                {MdpJsonSql.Str("d", "jyyj")},
                {MdpJsonSql.Str("d", "jsbz")},
                {MdpJsonSql.Str("d", "jyff")},
                {MdpJsonSql.Str("d", "sx")},
                {MdpJsonSql.Str("d", "xx")},
                {MdpJsonSql.Str("d", "ProcessCode")},
                {MdpJsonSql.Str("d", "ProcessName")},
                {MdpJsonSql.Str("d", "MeasurementInstrument")},
                {MdpJsonSql.Str("d", "Inspector")},
                {MdpJsonSql.DateTimeSec("d", "InspectionTime")},
                {MdpJsonSql.Int("d", "ybl")},
                {MdpJsonSql.Dec("d", "jysl", 18, 6)},
                {MdpJsonSql.Dec("d", "bhgsl", 18, 6)},
                {MdpJsonSql.Str("d", "pd")},
                {MdpJsonSql.Int("d", "NonNumericalTypeOK")},
                {MdpJsonSql.Int("d", "NonNumericalTypeNG")},
                {MdpJsonSql.Str("d", "bz")},
                {MdpJsonSql.Str("d", "fj")},
                IFNULL(d.source_row_id, {MdpJsonSql.Str("d", "id")}),
                IFNULL(NULLIF(d.source_biz_key,''), {MdpJsonSql.Str("d", "glid")}),
                @BatchId, @Now
            FROM mdp_stg_ipqc_pull d
            LEFT JOIN mdp_std_ipqc_inspection h
              ON h.tenant_id = {dTenant}
             AND h.source_row_id = {MdpJsonSql.Str("d", "glid")}
            WHERE d.source_table='qms_gcjydzb'{srcClause} AND {MdpJsonSql.TenantGuard(dTenant)}
            ON DUPLICATE KEY UPDATE
                std_head_id=VALUES(std_head_id), source_head_id=VALUES(source_head_id), inspection_item=VALUES(inspection_item),
                inspection_basis=VALUES(inspection_basis), technical_standard=VALUES(technical_standard), inspection_method=VALUES(inspection_method),
                upper_limit=VALUES(upper_limit), lower_limit=VALUES(lower_limit), process_code=VALUES(process_code), process_name=VALUES(process_name),
                measurement_instrument=VALUES(measurement_instrument), inspector=VALUES(inspector), inspection_time=VALUES(inspection_time),
                sample_qty=VALUES(sample_qty), inspection_qty=VALUES(inspection_qty), unqualified_qty=VALUES(unqualified_qty),
                result_judgement=VALUES(result_judgement), non_numeric_ok=VALUES(non_numeric_ok), non_numeric_ng=VALUES(non_numeric_ng),
                remark=VALUES(remark), attachment=VALUES(attachment),
                sync_batch_id=VALUES(sync_batch_id), sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
            """;
        var insPars = new List<SugarParameter> { new("@BatchId", batchId), new("@Now", now) };
        if (sourceSystem != null) insPars.Add(new SugarParameter("@Src", sourceSystem));
        await MdpSchemaAligner.ExecuteAsync(_db, insertSql, insPars);
        return rows;
    }

    private async Task<long> InsertRunLogAsync(string batchId, DateTime startedAt, string triggerType)
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            INSERT INTO mdp_transform_run_log
            (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
            VALUES (0, @JobCode, 'S6过程检验单MDP同步与标准化转换', @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            new SugarParameter("@JobCode", JobCode),
            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 MarkRunSuccessAsync(long runLogId, DateTime startedAt, IpqcInspectionMdpSyncResult result)
    {
        var finishedAt = DateTime.Now;
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_transform_run_log
            SET status='SUCCESS', end_time=@EndTime, duration_ms=@DurationMs,
                stage_rows=@StageRows, standard_rows=@StandardRows, dwd_rows=0, update_time=CURRENT_TIMESTAMP
            WHERE id=@Id
            """,
            new SugarParameter("@EndTime", finishedAt),
            new SugarParameter("@DurationMs", (int)(finishedAt - startedAt).TotalMilliseconds),
            new SugarParameter("@StageRows", result.HeadStageRows + result.DetailStageRows),
            new SugarParameter("@StandardRows", result.HeadStandardRows + result.DetailStandardRows),
            new SugarParameter("@Id", runLogId));
    }

    private async Task MarkRunFailedAsync(long runLogId, DateTime startedAt, string message)
    {
        var finishedAt = DateTime.Now;
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            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", message.Length > 2000 ? message[..2000] : message),
            new SugarParameter("@Id", runLogId));
    }

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

/// <summary>过程检验单 MDP 同步转换结果。</summary>
public sealed class IpqcInspectionMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int HeadStageRows { get; set; }
    public int HeadStandardRows { get; set; }
    public int DetailStageRows { get; set; }
    public int DetailStandardRows { get; set; }
    public List<string> TenantSkipWarnings { get; set; } = new();
}

/// <summary>S6 IPQC 双模式入站结果。</summary>
public sealed class IpqcInspectionInboundResult
{
    public string PullBatchId { get; set; } = string.Empty;
    public int RowsPulled { get; set; }
    public int RowsWrittenStg { get; set; }
    public string? NewCursor { get; set; }
    public string TransformBatchId { get; set; } = string.Empty;
    public int HeadStageRows { get; set; }
    public int HeadStandardRows { get; set; }
    public int DetailStageRows { get; set; }
    public int DetailStandardRows { get; set; }
}