AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/ProductionReceiptMdpSyncService.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 内部，扁平单表，独立于 S5MdpSyncTransformService 的 KPI 管线）。
///
/// 源：aidopdev NbrMaster(n) + NbrDetail(d)，业务类型 Type='WOI'（生产入库）；
///     扁平 LIST join：n.Domain=d.Domain AND n.Nbr=d.Nbr（一行=一条 NbrDetail 明细）。
///     维表 DepartmentMaster(p) / ItemMaster(i) / LocationMaster(lt,lf) 全 LEFT JOIN。
///     LocationMaster 无 Domain 列（DOP 重建：domain_code/location/descr），按 tenant_id+location join。
/// 链路（Phase 1）：执行器抽 NbrMaster/NbrDetail → mdp_stg_production_receipt → mdp_std_production_receipt。
/// 双模式入站：mdp_entity=S7_NBR_MASTER/S7_NBR_DETAIL → Writer → stg；标准层读 stg；维表仍直连本库。
///
/// 约束：
///   - 只读源/贴源，仅写 mdp_stg_production_receipt / mdp_std_production_receipt；绝不写 NbrMaster/NbrDetail。
///   - 过滤 Type='WOI' AND IsActive=1。
///   - WOI 为 0 行时转换成功完成、处理数为 0，不报错。
/// </summary>
public class ProductionReceiptMdpSyncService : ITransient
{
    private const string JobCode = "S5_PRODUCTION_RECEIPT_MDP_SYNC";
    private const string InboundEntityCode = "S7_NBR_MASTER";
    private const string InboundDetailEntityCode = "S7_NBR_DETAIL";

    // 双源（dopdemorq SQL Server）Phase 1：沿用现有 S7_NBR_* 命名语义（业务=生产入库），加 _SQLSERVER 第二源实体。实体默认 status=0（就位不启用）。
    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
    private const string SqlServerMasterEntityCode = "S7_NBR_MASTER_SQLSERVER";
    private const string SqlServerDetailEntityCode = "S7_NBR_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 ProductionReceiptMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher, TransformRunLogFinalizer runLogFinalizer)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _pullDispatcher = pullDispatcher;
    }

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

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

        try
        {
            // 本库 NbrMaster 含 tenant_id；ctx tenantId=0 仅兜底，Writer 优先取源行值。
            var pullCtx = new MdpPullContext
            {
                TenantId = 0,
                FullRefresh = true,
                TaskCode = "S7_PRODUCTION_RECEIPT_INBOUND",
                BatchId = $"{batchId}_PULL"
            };
            await PopulateStgAsync(pullCtx, cancellationToken);
            result.Rows = 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>
    /// S7 成品入库双模式入站：执行器抽主/明细 → stg，再跑 WOI 标准层（读 stg）。
    /// </summary>
    public async Task<ProductionReceiptInboundResult> 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 = "S7_PRODUCTION_RECEIPT_INBOUND",
            BatchId = $"S7_NBR_IN_{now:yyyyMMddHHmmss}"
        };
        var pull = await PopulateStgAsync(pullCtx, cancellationToken);

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

    /// <summary>
    /// 双源切换 FULL Replace（Phase 1）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 stg，
    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_production_receipt（按 tenant 精确隔离），消除旧源独有业务键残留。
    /// 用 PullAllByEntityCodeAsync 抽尽 + FullRefresh=true；transform 仅当前 source_system；std DELETE+INSERT 同事务，Pull 成功后才进入。
    /// SQLSERVER 实体默认 status=0（就位不启用）；dopdemorq 六表当前空，Phase 1 不实际执行 destructive 切换。
    /// </summary>
    public async Task<ProductionReceiptInboundResult> 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。任一 Pull 失败会抛异常，std 未动。
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S7_PRODUCTION_RECEIPT_INBOUND",
            BatchId = $"S7_NBR_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 = $"S7_NBR_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new ProductionReceiptMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.Rows = await MdpStdFullReplace.ReplaceAsync(
                _db, "mdp_std_production_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 ProductionReceiptInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = master.RowsPulled + detail.RowsPulled,
            RowsWrittenStg = master.RowsWritten + detail.RowsWritten,
            NewCursor = detail.NewCursor ?? master.NewCursor,
            TransformBatchId = batchId,
            StdRows = result.Rows
        };
    }

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

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

    /// <summary>防御式建表（与 UpdateScripts/1.0.214.sql 同构，幂等）。</summary>
    private async Task EnsureTablesAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_production_receipt"));
    }

    /// <summary>
    /// 标准化：stg(NbrMaster/NbrDetail) + 维表(本库) → mdp_std_production_receipt。
    /// </summary>
    private async Task<int> TransformStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
    {
        // 双源：sourceSystem 非空时仅统计/转换当前 source（切源 FULL Replace 用）；为空时保持既有全源行为不变。
        var srcClause = sourceSystem == null ? "" : " AND n.source_system=@Src AND d.source_system=@Src";
        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_production_receipt n
            INNER JOIN mdp_stg_production_receipt d
              ON d.source_table='NbrDetail'
             AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
             AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Nbr')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Nbr'))
            WHERE n.source_table='NbrMaster'
              AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Type'))='WOI'
              AND LOWER(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.IsActive'))) IN ('1','true')
              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) IS NOT NULL{srcClause}
            """, countPars);
        var insertSql =
            $"""
            INSERT INTO mdp_std_production_receipt
            (tenant_id, factory_id, source_system, domain, master_rec_id, detail_rec_id, nbr, line,
             receipt_date, status, status_desc, remark, prod_line, work_ord, erp_work_ord,
             department, department_desc, applicant_name, item_num, item_name, item_spec, um,
             location_to, location_to_desc, lot_serial, qty_rec, qty_to, location_from, location_from_desc,
             ord_nbr, source_biz_key, sync_batch_id, sync_time)
            SELECT
                NULLIF(n.tenant_id, 0), 1, {MdpSourceIdentity.Resolve("n.source_table", "n.source_system")},
                JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.RecID')) AS SIGNED),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) AS SIGNED),
                JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Nbr')),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Line')) AS SIGNED),
                STR_TO_DATE(REPLACE(LEFT(NULLIF(NULLIF(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Date')),'null'),''), 19), 'T', ' '), '%Y-%m-%d %H:%i:%s'),
                UPPER(IFNULL(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Status')), '')),
                UPPER(IFNULL(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Status')), '')),
                JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Remark')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ProdLine')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.WorkOrd')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Address')),
                JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Department')),
                TRIM(CONCAT(IFNULL(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Department')),''), ' ', IFNULL(p.Descr, ''))),
                JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Name')),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum')), i.Descr, i.Descr1, i.UM,
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LocationTo')), lt.descr,
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LotSerial')) AS CHAR),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyRec')) AS DECIMAL(18,5)),
                CAST(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.QtyTo')) AS DECIMAL(18,5)),
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LocationFrom')), lf.descr,
                JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.OrdNbr')),
                CONCAT(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')), '#',
                       JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID'))),
                @BatchId, @Now
            FROM mdp_stg_production_receipt n
            INNER JOIN mdp_stg_production_receipt d
              ON d.source_table='NbrDetail'
             AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
             AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Nbr')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Nbr'))
            LEFT JOIN DepartmentMaster p
              ON p.Domain = JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Domain'))
             AND p.Department = JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Department'))
            LEFT JOIN ItemMaster i
              ON i.Domain = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.Domain'))
             AND i.ItemNum = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))
            LEFT JOIN LocationMaster lt
              ON lt.tenant_id = IFNULL(d.tenant_id, 0)
             AND lt.location = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LocationTo'))
            LEFT JOIN LocationMaster lf
              ON lf.tenant_id = IFNULL(d.tenant_id, 0)
             AND lf.location = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.LocationFrom'))
            WHERE n.source_table='NbrMaster'
              AND JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.Type'))='WOI'
              AND LOWER(JSON_UNQUOTE(JSON_EXTRACT(n.raw_data,'$.IsActive'))) IN ('1','true')
              AND JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.RecID')) IS NOT NULL{srcClause}
              AND NULLIF(n.tenant_id, 0) IS NOT NULL
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), master_rec_id=VALUES(master_rec_id), nbr=VALUES(nbr), line=VALUES(line),
                receipt_date=VALUES(receipt_date), status=VALUES(status), status_desc=VALUES(status_desc), remark=VALUES(remark),
                prod_line=VALUES(prod_line), work_ord=VALUES(work_ord), erp_work_ord=VALUES(erp_work_ord),
                department=VALUES(department), department_desc=VALUES(department_desc), applicant_name=VALUES(applicant_name),
                item_num=VALUES(item_num), item_name=VALUES(item_name), item_spec=VALUES(item_spec), um=VALUES(um),
                location_to=VALUES(location_to), location_to_desc=VALUES(location_to_desc), lot_serial=VALUES(lot_serial),
                qty_rec=VALUES(qty_rec), qty_to=VALUES(qty_to), location_from=VALUES(location_from),
                location_from_desc=VALUES(location_from_desc), ord_nbr=VALUES(ord_nbr),
                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, '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, ProductionReceiptMdpSyncResult 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=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.Rows),
            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 ProductionReceiptMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int Rows { get; set; }
}

/// <summary>S7 成品入库双模式入站结果。</summary>
public sealed class ProductionReceiptInboundResult
{
    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 StdRows { get; set; }
}