AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/PurchaseReceiptMdpSyncService.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.MaterialWarehouse;

/// <summary>
/// S5 采购收货单 数据中台只读同步转换服务（DOP 内部，方案 1）。
///
/// 源：aidopdev PurOrdRctDetail(p) + PurOrdRctMaster(d)，业务类型 RctType='rc'（采购收货）；
///     头明细关联 p.Domain=d.Domain AND p.Receiver=d.Receiver；
///     维表 ItemMaster(i) / SuppMaster(s) / ConsigneeAddressMaster(a) / PurOrdDetail(pd,sd) / srm_pr_main(dr) 全 LEFT JOIN。
/// 链路（Phase 1）：执行器抽主/明细 → mdp_stg_purchase_receipt → mdp_std_purchase_receipt(typed)。
/// 双模式入站：mdp_entity=S5_PURCHASE_RECEIPT_DETAIL/MASTER（DB 或 API）→ MdpStagingWriter → stg，
///     标准层读 stg.raw_data；维表 ItemMaster/SuppMaster 等仍直连本库（本地口径补全）。
///
/// 约束：
///   - 只读源/贴源，仅写 mdp_stg_purchase_receipt / mdp_std_purchase_receipt；绝不写源表；不读/不改 S3 mdp_stg_receipt、S4 ado_s4_receipt。
///   - 过滤 RctType='rc'；租户隔离在读 API 侧按 tenant_id。
///   - rc 为 0 行时成功完成、处理数为 0，不报错。
/// </summary>
public class PurchaseReceiptMdpSyncService : ITransient
{
    private const string JobCode = "S5_PURCHASE_RECEIPT_MDP_SYNC";
    private const string InboundEntityCode = "S5_PURCHASE_RECEIPT_DETAIL";
    private const string InboundMasterEntityCode = "S5_PURCHASE_RECEIPT_MASTER";

    // 双源（dopdemorq SQL Server）Phase 1：默认第二源与其入站实体。实体默认 status=0，未启用前 RunSourceSwitchFullAsync 会因找不到启用实体而抛错（就位不启用）。
    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
    private const string SqlServerMasterEntityCode = "S5_PURCHASE_RECEIPT_MASTER_SQLSERVER";
    private const string SqlServerDetailEntityCode = "S5_PURCHASE_RECEIPT_DETAIL_SQLSERVER";
    /// <summary>165 SQL Server 源无 tenant_id 列；切源时须经账套映射解析。</summary>
    private const string SourceZtid = "pbxfxp";

    private readonly ISqlSugarClient _db;
    private readonly TransformRunLogFinalizer _runLogFinalizer;
    private readonly MdpSourcePullDispatcher _pullDispatcher;

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

    /// <summary>全量：本地 DB 执行器灌 stg → 标准层（读 stg）。</summary>
    public async Task<PurchaseReceiptMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsureStgTableAsync();

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

        try
        {
            // 本库 PurOrdRct* 含 tenant_id；ctx tenantId=0 仅兜底，Writer 优先取源行值。
            var pullCtx = new MdpPullContext
            {
                TenantId = 0,
                FullRefresh = true,
                TaskCode = "S5_PURCHASE_RECEIPT_INBOUND",
                BatchId = $"{batchId}_PULL"
            };
            await PopulateStgAsync(pullCtx, cancellationToken);
            result.StdRows = await TransformStandardAsync(batchId, now);
            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>
    /// 双模式入站：执行器抽主/明细落 stg，再跑标准层（读 stg）。
    /// </summary>
    public async Task<PurchaseReceiptInboundResult> RunInboundAsync(
        long tenantId = 0,
        bool fullRefresh = false,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsureStgTableAsync();

        var now = DateTime.Now;
        // 入参 tenantId：本库源含 tenant_id 时可传 0（Writer 优先取源行）。
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = fullRefresh,
            TaskCode = "S5_PURCHASE_RECEIPT_INBOUND",
            BatchId = $"S5_PUR_RCT_IN_{now:yyyyMMddHHmmss}"
        };
        var pull = await PopulateStgAsync(pullCtx, cancellationToken);

        var batchId = $"S5_PUR_RCT_STD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "INBOUND");
        var result = new PurchaseReceiptMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.StdRows = await TransformStandardAsync(batchId, now);
            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 PurchaseReceiptInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            PullMessage = pull.Message,
            TransformBatchId = result.BatchId,
            StdRows = result.StdRows
        };
    }

    /// <summary>
    /// 双源切换 FULL Replace（Phase 1）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 stg，
    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_purchase_receipt（按 tenant 精确隔离），消除旧源独有业务键残留。
    ///
    /// 与既有 RunFull/RunInbound 的区别：① 用 PullAllByEntityCodeAsync 抽尽（非单批 PullByEntityCode）；
    /// ② FullRefresh=true 显式全量 bootstrap；③ transform 仅当前 source_system；④ std DELETE+INSERT 同事务，Pull 成功后才进入。
    ///
    /// 切源语义为「部署级单活源」。SQLSERVER 实体默认 status=0，未启用前本方法运行时会因 PullByEntityCode 找不到启用实体抛错——符合 Phase 1「就位不启用」。
    /// dopdemorq 当前六表为空：本方法 FULL Replace 的理论结果是把当前 tenant 对应 std 置空；Phase 1 不实际执行 destructive 切换。
    /// </summary>
    public async Task<PurchaseReceiptInboundResult> RunSourceSwitchFullAsync(
        string sourceCode = SqlServerSourceCode,
        string masterEntityCode = SqlServerMasterEntityCode,
        string detailEntityCode = SqlServerDetailEntityCode,
        long tenantId = 0,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsureStgTableAsync();

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

        var now = DateTime.Now;
        // 1) PullAll 抽尽 + fullRefresh=true（不用单批 PullByEntityCode）。任一 Pull 失败会抛异常，std 未动。
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S5_PURCHASE_RECEIPT_INBOUND",
            BatchId = $"S5_PUR_RCT_SW_{now:yyyyMMddHHmmss}"
        };
        var master = await _pullDispatcher.PullAllByEntityCodeAsync(masterEntityCode, pullCtx, cancellationToken);
        var detail = await _pullDispatcher.PullAllByEntityCodeAsync(detailEntityCode, pullCtx, cancellationToken);

        // 2) FULL Replace：事务内 DELETE 当前 tenant 的 std，再 INSERT 仅当前 source_system 的结果。
        var batchId = $"S5_PUR_RCT_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new PurchaseReceiptMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.StdRows = await MdpStdFullReplace.ReplaceAsync(
                _db, "mdp_std_purchase_receipt", tenantId, extraWhere: null,
                insertScopedAsync: () => TransformStandardAsync(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 PurchaseReceiptInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = master.RowsPulled + detail.RowsPulled,
            RowsWrittenStg = master.RowsWritten + detail.RowsWritten,
            NewCursor = detail.NewCursor ?? master.NewCursor,
            PullMessage = $"{master.Message}; {detail.Message}",
            TransformBatchId = batchId,
            StdRows = result.StdRows
        };
    }

    private async Task<(int RowsPulled, int RowsWritten, string? NewCursor, string? Message)> PopulateStgAsync(
        MdpPullContext pullCtx, CancellationToken cancellationToken)
    {
        var master = await _pullDispatcher.PullByEntityCodeAsync(InboundMasterEntityCode, pullCtx, cancellationToken);
        var detail = await _pullDispatcher.PullByEntityCodeAsync(InboundEntityCode, pullCtx, cancellationToken);
        return (
            master.RowsPulled + detail.RowsPulled,
            master.RowsWritten + detail.RowsWritten,
            detail.NewCursor ?? master.NewCursor,
            $"{master.Message}; {detail.Message}");
    }

    private async Task EnsureStgTableAsync()
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_stg_purchase_receipt (
                id BIGINT PRIMARY KEY AUTO_INCREMENT,
                tenant_id BIGINT NOT NULL,
                source_system VARCHAR(50) NULL,
                source_table VARCHAR(200),
                source_row_id VARCHAR(200),
                source_biz_key VARCHAR(300) NULL,
                raw_data JSON,
                sync_batch_id VARCHAR(100),
                sync_time DATETIME NULL DEFAULT CURRENT_TIMESTAMP,
                process_status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
                process_message VARCHAR(500) NULL,
                create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                update_time DATETIME NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_source_key (source_system, source_table, source_biz_key),
                KEY idx_batch (sync_batch_id),
                KEY idx_src (source_table, source_row_id),
                KEY idx_tenant (tenant_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='S5采购收货贴源层'
            """);
    }

    /// <summary>防御式建表（与 UpdateScripts WIP-S5PR.sql / 正式 1.0.&lt;n&gt;.sql 同构，幂等）。</summary>
    private async Task EnsureTablesAsync()
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            CREATE TABLE IF NOT EXISTS mdp_std_purchase_receipt (
                id BIGINT AUTO_INCREMENT PRIMARY KEY,
                tenant_id BIGINT NOT NULL DEFAULT 0,
                factory_id BIGINT NULL DEFAULT 1,
                source_system VARCHAR(50) NOT NULL DEFAULT 'AIDOP',
                domain VARCHAR(24) NOT NULL,
                receiver VARCHAR(24) NOT NULL,
                line SMALLINT NOT NULL DEFAULT 0,
                rct_date DATETIME NULL,
                supp VARCHAR(20) NULL,
                sort_name VARCHAR(255) NULL,
                item_num VARCHAR(60) NULL,
                item_name VARCHAR(200) NULL,
                item_spec VARCHAR(200) NULL,
                um VARCHAR(8) NULL,
                qty_ordered DECIMAL(18,6) NULL DEFAULT 0,
                qty_received DECIMAL(18,6) NULL DEFAULT 0,
                lot_serial VARCHAR(120) NULL,
                location VARCHAR(8) NULL,
                ord_nbr VARCHAR(24) NULL,
                ord_line SMALLINT NULL,
                blanket_line INT NULL,
                pur_ord VARCHAR(24) NULL,
                pur_line SMALLINT NULL,
                sales_job VARCHAR(200) NULL,
                address1 VARCHAR(200) NULL,
                req VARCHAR(20) NULL,
                req_line INT NULL,
                dop_req VARCHAR(255) NULL,
                source_biz_key VARCHAR(200) NULL,
                sync_batch_id VARCHAR(100) NOT NULL,
                sync_time DATETIME NOT NULL,
                create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
                update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                UNIQUE KEY uk_mdp_std_pur_rct (tenant_id, domain, receiver, line),
                KEY idx_mdp_std_pur_rct_date (tenant_id, rct_date),
                KEY idx_mdp_std_pur_rct_item (tenant_id, item_num),
                KEY idx_mdp_std_pur_rct_supp (tenant_id, supp),
                KEY idx_mdp_std_pur_rct_purord (tenant_id, pur_ord),
                KEY idx_mdp_std_pur_rct_salesjob (tenant_id, sales_job),
                KEY idx_mdp_std_pur_rct_req (tenant_id, req),
                KEY idx_mdp_std_pur_rct_dopreq (tenant_id, dop_req)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S5采购收货单标准层'
            """);
    }

    /// <summary>
    /// 标准化：stg(PurOrdRctDetail/Master) + 维表(本库) → mdp_std_purchase_receipt。
    ///
    /// JSON null 归一：展示/属性列一律经 <see cref="MdpJsonSql"/>（Str/Int/Dec）取值，
    /// 避免 <c>JSON_UNQUOTE</c> 把 JSON null 产出成长度 4 的字面量字符串 <c>'null'</c> 落进标准层（前端 `|| '-'` 对其无效）。
    /// <b>身份列与 JOIN 键刻意保留裸取值</b>：<c>Domain</c> / <c>Receiver</c> / <c>Line</c> 是
    /// <c>uk_mdp_std_pur_rct</c> 的 NOT NULL 组成部分与 source_biz_key 兜底；且头明细 INNER JOIN 依赖 Domain 相等，
    /// 一旦归一为 SQL NULL 则 <c>NULL = NULL</c> 不成立、整批行会从标准层消失。
    /// <c>RctType</c> / <c>OrdType</c> 用于 WHERE 过滤与 CASE 判别，同样不归一。
    /// </summary>
    private async Task<int> TransformStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
    {
        // 双源：sourceSystem 非空时仅统计/转换当前 source（切源 FULL Replace 用）；为空时保持既有全源行为不变。
        var srcClause = sourceSystem == null ? "" : " AND p.source_system=@Src AND d.source_system=@Src";
        var pTenant = MdpJsonSql.TenantFromStg("p");
        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_purchase_receipt p
            INNER JOIN mdp_stg_purchase_receipt d
              ON d.source_table='PurOrdRctMaster'
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Receiver'))
            WHERE p.source_table='PurOrdRctDetail'
              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'{srcClause}
            """, countPars);
        var insertSql =
            $"""
            INSERT INTO mdp_std_purchase_receipt
            (tenant_id, factory_id, source_system, domain, receiver, line, rct_date, supp, sort_name,
             item_num, item_name, item_spec, um, qty_ordered, qty_received, lot_serial, location,
             ord_nbr, ord_line, blanket_line, pur_ord, pur_line, sales_job, address1,
             req, req_line, dop_req, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {pTenant}, 1, IFNULL(NULLIF(p.source_system,''), 'AIDOP'),
                JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')),
                JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Line')) AS SIGNED),
                STR_TO_DATE(REPLACE(LEFT(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RctDate')),'null'),''), 19), 'T', ' '), '%Y-%m-%d %H:%i:%s'),
                {MdpJsonSql.Str("p", "Supp")},
                NULLIF(TRIM(CONCAT(IFNULL({MdpJsonSql.Str("d", "Supp")},''), ' ', IFNULL(s.SortName, ''))), ''),
                {MdpJsonSql.Str("p", "ItemNum")}, i.Descr, i.Descr1,
                {MdpJsonSql.Str("p", "UM")},
                {MdpJsonSql.Dec("p", "QtyOrded", 18, 6)},
                {MdpJsonSql.Dec("p", "QtyReceived", 18, 6)},
                {MdpJsonSql.Str("p", "LotSerial")},
                {MdpJsonSql.Str("p", "Location")},
                {MdpJsonSql.Str("p", "OrdNbr")},
                {MdpJsonSql.Int("p", "OrdLine")},
                pd.BlanketLine, pd.PurOrd, pd.Line, pd.SalesJob, a.Address1,
                (CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdType'))='DO' THEN '' ELSE sd.Req END),
                sd.ReqLine,
                (CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdType'))='DO' THEN sd.Req ELSE dr.pr_billno END),
                IFNULL(NULLIF(p.source_biz_key,''), CONCAT(
                    JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')), '#',
                    JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')), '#',
                    JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Line')))),
                @BatchId, @Now
            FROM mdp_stg_purchase_receipt p
            INNER JOIN mdp_stg_purchase_receipt d
              ON d.source_table='PurOrdRctMaster'
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Receiver')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Receiver'))
            LEFT JOIN ItemMaster i
              ON JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = i.Domain
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.ItemNum')) = i.ItemNum
            LEFT JOIN SuppMaster s
              ON JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain')) = s.Domain
             AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Supp')) = s.Supp
            LEFT JOIN ConsigneeAddressMaster a
              ON JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = a.Domain
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Supp')) = a.Address
             AND a.Typed = 'Supp'
            LEFT JOIN PurOrdDetail pd
              ON JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = pd.Domain
             AND (CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdType'))='DO'
                        AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctNbr')),'')<>''
                       THEN JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctNbr'))
                       ELSE JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.OrdNbr')) END)
                 = (CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdType'))='DO'
                          AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctNbr')),'')<>''
                         THEN pd.Contract ELSE pd.PurOrd END)
             AND (CASE WHEN JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdType'))='DO'
                        AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctNbr')),'')<>''
                       THEN CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.BlanketLine')) AS SIGNED)
                       ELSE CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.OrdLine')) AS SIGNED) END) = pd.Line
            LEFT JOIN PurOrdDetail sd
              ON JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.Domain')) = sd.Domain
             AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.OrdNbr')) = sd.PurOrd
             AND CAST(JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.OrdLine')) AS SIGNED) = sd.Line
            LEFT JOIN srm_pr_main dr
              ON CAST(dr.factory_id AS CHAR) = sd.Domain
             AND dr.SAP_pr_billno = sd.Req
             AND IFNULL(dr.SAP_pr_billno,'')<>''
            WHERE p.source_table='PurOrdRctDetail'
              AND JSON_UNQUOTE(JSON_EXTRACT(p.raw_data,'$.RctType'))='rc'{srcClause}
              AND {MdpJsonSql.TenantGuard(pTenant)}
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), rct_date=VALUES(rct_date), supp=VALUES(supp), sort_name=VALUES(sort_name),
                item_num=VALUES(item_num), item_name=VALUES(item_name), item_spec=VALUES(item_spec), um=VALUES(um),
                qty_ordered=VALUES(qty_ordered), qty_received=VALUES(qty_received), lot_serial=VALUES(lot_serial), location=VALUES(location),
                ord_nbr=VALUES(ord_nbr), ord_line=VALUES(ord_line), blanket_line=VALUES(blanket_line),
                pur_ord=VALUES(pur_ord), pur_line=VALUES(pur_line), sales_job=VALUES(sales_job), address1=VALUES(address1),
                req=VALUES(req), req_line=VALUES(req_line), dop_req=VALUES(dop_req),
                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 _db.Ado.ExecuteCommandAsync(insertSql, insPars);
        return rows;
    }

    private async Task<long> InsertRunLogAsync(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 (0, @JobCode, 'S5采购收货单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, PurchaseReceiptMdpSyncResult 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=0, 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("@StandardRows", result.StdRows),
            new SugarParameter("@Id", runLogId));
    }

    private async Task MarkRunFailedAsync(long runLogId, DateTime startedAt, string message)
    {
        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", 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 PurchaseReceiptMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int StdRows { get; set; }
}

/// <summary>采购收货双模式入站结果（stg 抽数 + std 转换）。</summary>
public sealed class PurchaseReceiptInboundResult
{
    public string PullBatchId { get; set; } = string.Empty;
    public int RowsPulled { get; set; }
    public int RowsWrittenStg { get; set; }
    public string? NewCursor { get; set; }
    public string? PullMessage { get; set; }
    public string TransformBatchId { get; set; } = string.Empty;
    public int StdRows { get; set; }
}