AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/ProductionIssueMdpSyncService.cs

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

namespace Admin.NET.Plugin.AiDOP.MaterialWarehouse;

/// <summary>
/// S5 生产领料单 数据中台只读同步转换服务（DOP 内部，头级单表）。
///
/// B-① 双源通用管线（2026-07-29 由 bespoke 直读 NbrMaster 迁入）：
///   源(NbrMaster) --MdpDbPullExecutor--> mdp_stg_production_issue(raw_data JSON) --transform--> mdp_std_production_issue。
///   入站实体：S5_PRODUCTION_ISSUE_MASTER(源=AIDOPDEV_MYSQL) / S5_PRODUCTION_ISSUE_MASTER_SQLSERVER(源=DOPDEMORQ_SQLSERVER)。
///   维表 DepartmentMaster 仍本库 LEFT JOIN（与 mode-A 采购/生产收货同构；department_desc 富化，部门 code 由事实表保留）。
///
/// 约束：
///   - 只读源/贴源，仅写 mdp_stg_production_issue / mdp_std_production_issue；绝不写 NbrMaster；不碰 WorkOrderPickBillService 写路径。
///   - 头过滤 Type='SM' AND IsActive（领料无 IsReturn 语义）；SM=领料单业务类型（Type SM/WOI/WOD/CA 互斥）。
///   - 跨源类型兼容经 MdpJsonSql（datetime ISO-T / bit true-false / JSON null 归一），不改 MDP 核心。
///   - 切源单 active source + FULL Replace（决策1A）；不做 merge、不做回写/outbox（决策2A）。
///   - SM 为 0 行时转换成功完成、处理数为 0，不报错。
/// </summary>
public class ProductionIssueMdpSyncService : ITransient
{
    private const string JobCode = "S5_PRODUCTION_ISSUE_MDP_SYNC";
    private const string InboundEntityCode = "S5_PRODUCTION_ISSUE_MASTER";

    // 双源（dopdemorq SQL Server）：第二源与其入站实体。实体默认 status=0（就位不启用）。
    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
    private const string SqlServerEntityCode = "S5_PRODUCTION_ISSUE_MASTER_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;
    private readonly MdpNeutralSourceGate _neutralGate;

    public ProductionIssueMdpSyncService(ISqlSugarClient db, MdpSourcePullDispatcher pullDispatcher, TransformRunLogFinalizer runLogFinalizer, MdpNeutralSourceGate neutralGate)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _pullDispatcher = pullDispatcher;
        _neutralGate = neutralGate;
    }

    /// <summary>全量：按当前启用入站实体灌 stg → 标准层（WP9 S5 后默认 165）。</summary>
    public async Task<ProductionIssueMdpSyncResult> RunFullAsync(CancellationToken cancellationToken = default, string triggerType = "AUTO")
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsureStgTableAsync();

        var active = await ResolveActiveInboundAsync(cancellationToken);
        var now = DateTime.Now;
        var batchId = $"S5_PROD_ISSUE_FULL_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, triggerType);
        var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };

        try
        {
            // 本库 NbrMaster 含 tenant_id；ctx tenantId=0 仅兜底，Writer 优先取源行值。
            var pullCtx = new MdpPullContext
            {
                TenantId = 0,
                FullRefresh = true,
                TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
                BatchId = $"{batchId}_PULL"
            };
            await PopulateStgAsync(active.EntityCode, pullCtx, cancellationToken);
            if (active.UseSourceFilter)
            {
                result.HeadRows = await MdpStdFullReplace.ReplaceAsync(
                    _db, "mdp_std_production_issue", 0, extraWhere: null,
                    insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, active.SourceCode),
                    cancellationToken);
            }
            else
            {
                result.HeadRows = await TransformHeadStandardAsync(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<ProductionIssueInboundResult> RunInboundAsync(
        long tenantId = 0, bool fullRefresh = false, CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureTablesAsync();
        await EnsureStgTableAsync();

        var active = await ResolveActiveInboundAsync(cancellationToken);
        var now = DateTime.Now;
        // 165 源无 tenant_id 须映射；本库 NbrMaster 含 tenant_id，tenantId=0 时 Writer 优先取源行。
        var ctxTenantId = active.UseSourceFilter
            ? AidopSourceTenantMap.ResolveTenantId(SourceZtid, tenantId)
            : tenantId;
        var pullCtx = new MdpPullContext
        {
            TenantId = ctxTenantId,
            FullRefresh = fullRefresh,
            TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
            BatchId = $"S5_PROD_ISSUE_IN_{now:yyyyMMddHHmmss}"
        };
        var pull = await PopulateStgAsync(active.EntityCode, pullCtx, cancellationToken);

        var batchId = $"S5_PROD_ISSUE_STD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "INBOUND");
        var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            if (active.UseSourceFilter)
            {
                result.HeadRows = await MdpStdFullReplace.ReplaceAsync(
                    _db, "mdp_std_production_issue", ctxTenantId, extraWhere: null,
                    insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, active.SourceCode),
                    cancellationToken);
            }
            else
            {
                result.HeadRows = await TransformHeadStandardAsync(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 ProductionIssueInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            TransformBatchId = result.BatchId,
            StdRows = result.HeadRows
        };
    }

    /// <summary>
    /// WP9 S5：切到 165 权威源并 FULL Replace 标准层；同时确保本库历史 SM 已打 AIDOP_LEGACY。
    /// </summary>
    public async Task<ProductionIssueInboundResult> ActivateSqlServerSourceAndSwitchAsync(
        long tenantId = 0,
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();
        await EnsureNbrMasterSourceSystemColumnAsync();
        await MarkLocalSmLegacyAsync();
        await FlipInboundEntityStatusAsync();
        return await RunSourceSwitchFullAsync(
            SqlServerSourceCode, SqlServerEntityCode, tenantId, cancellationToken);
    }

    private async Task<(string EntityCode, string SourceCode, bool UseSourceFilter)> ResolveActiveInboundAsync(
        CancellationToken cancellationToken)
    {
        var sqlOn = await _db.Queryable<MdpEntity>()
            .Where(x => x.EntityCode == SqlServerEntityCode && x.Status == 1)
            .AnyAsync(cancellationToken);
        if (sqlOn)
            return (SqlServerEntityCode, SqlServerSourceCode, true);

        var mysqlOn = await _db.Queryable<MdpEntity>()
            .Where(x => x.EntityCode == InboundEntityCode && x.Status == 1)
            .AnyAsync(cancellationToken);
        if (mysqlOn)
            return (InboundEntityCode, "AIDOPDEV_MYSQL", false);

        throw Oops.Oh("生产领料同步失败：未启用任何 S5 入站实体，请执行 WP9 S5 切源或检查 mdp_entity。");
    }

    private async Task FlipInboundEntityStatusAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_entity
            SET status = 0, update_time = NOW()
            WHERE entity_code = @Mysql AND status <> 0
            """,
            new SugarParameter("@Mysql", InboundEntityCode));
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE mdp_entity
            SET status = 1, update_time = NOW()
            WHERE entity_code = @Sql AND status <> 1
            """,
            new SugarParameter("@Sql", SqlServerEntityCode));
    }

    private async Task EnsureNbrMasterSourceSystemColumnAsync()
    {
        var exists = await _db.Ado.GetIntAsync(
            """
            SELECT COUNT(1) FROM information_schema.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'NbrMaster' AND COLUMN_NAME = 'source_system'
            """);
        if (exists == 0)
        {
            await MdpSchemaAligner.ExecuteAsync(_db, 
                "ALTER TABLE NbrMaster ADD COLUMN source_system VARCHAR(50) NULL DEFAULT NULL AFTER Remark");
        }
    }

    private async Task MarkLocalSmLegacyAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, 
            """
            UPDATE NbrMaster
            SET source_system = 'AIDOP_NATIVE'
            WHERE Type = 'SM'
              AND IFNULL(source_system, '') IN ('', 'AIDOP', 'AIDOP_LEGACY')
            """);
    }

    /// <summary>
    /// 双源切换 FULL Replace（Phase 1）：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 stg，
    /// 成功后在单事务内以「仅当前源」结果 FULL 重建 mdp_std_production_issue（按 tenant 精确隔离），消除旧源独有业务键残留。
    /// PullAllByEntityCodeAsync 抽尽 + FullRefresh=true；transform 仅当前 source_system；Pull 成功后才进入 destructive replace。
    /// SQLSERVER 实体默认 status=0（就位不启用）；本批 FULL-only，不启自动增量。
    /// </summary>
    public async Task<ProductionIssueInboundResult> RunSourceSwitchFullAsync(
        string sourceCode = SqlServerSourceCode,
        string entityCode = SqlServerEntityCode,
        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;
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S5_PRODUCTION_ISSUE_INBOUND",
            BatchId = $"S5_PROD_ISSUE_SW_{now:yyyyMMddHHmmss}"
        };
        var pull = await _pullDispatcher.PullAllByEntityCodeAsync(entityCode, pullCtx, cancellationToken);

        var batchId = $"S5_PROD_ISSUE_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new ProductionIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.HeadRows = await MdpStdFullReplace.ReplaceAsync(
                _db, "mdp_std_production_issue", 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 ProductionIssueInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            TransformBatchId = batchId,
            StdRows = result.HeadRows
        };
    }

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

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

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

    /// <summary>
    /// 标准化头：stg(NbrMaster,SM) + DepartmentMaster(本库) → mdp_std_production_issue。
    /// sourceSystem 非空时仅转当前源（切源 FULL Replace 用）；为空时全源（本库全量）。跨源类型经 MdpJsonSql 兼容。
    /// </summary>
    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null, long tenantId = 0)
    {
        var gateSource = sourceSystem ?? MdpSourceIdentity.Native;
        if (!await _neutralGate.AllowsAsync(tenantId, "PROD_DOC", gateSource))
            return 0;

        var srcClause = sourceSystem == null ? "" : " AND m.source_system=@Src";
        // 表达式片段（跨源类型兼容）
        var mTenant = MdpJsonSql.TenantFromStgJsonCol("m", MdpJsonSql.Int("m", "tenant_id"));
        var domainE = MdpJsonSql.Str("m", "Domain");
        var deptE = MdpJsonSql.Str("m", "Department");
        var statusUpper = $"UPPER(IFNULL({MdpJsonSql.Str("m", "Status")},''))";
        var pretreatE = MdpJsonSql.Str("m", "PretreatmentState");
        var transTypeE = MdpJsonSql.Str("m", "TransType");

        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_issue m WHERE m.source_table='NbrMaster' " +
            $"AND {MdpJsonSql.Str("m", "Type")}='SM' AND {MdpJsonSql.BoolTrue("m", "IsActive")}{srcClause}",
            countPars);

        var insertSql =
            $"""
            INSERT INTO mdp_std_production_issue
            (tenant_id, factory_id, source_system, domain, rec_id, nbr, issue_date, status, status_desc,
             work_ord, department, department_desc, qty_ord, prod_line, applicant_name, issue_user,
             user1, remark, create_user, source_create_time, pretreatment_state, trans_type, trans_type_text,
             eff_date, address, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {mTenant}, 1, {MdpSourceIdentity.Resolve("m.source_table", "m.source_system")},
                {domainE}, {MdpJsonSql.Int("m", "RecID")}, {MdpJsonSql.Str("m", "Nbr")},
                {MdpJsonSql.DateTimeSec("m", "Date")},
                {statusUpper},
                CASE
                    WHEN IFNULL({pretreatE},'')<>'' THEN {pretreatE}
                    WHEN {statusUpper}='C' THEN '已下架'
                    WHEN {statusUpper}='A' THEN '备料中'
                    ELSE ''
                END,
                {MdpJsonSql.Str("m", "WorkOrd")}, {deptE},
                TRIM(CONCAT(IFNULL({deptE},''), ' ', IFNULL(d.Descr, ''))),
                {MdpJsonSql.Dec("m", "QtyOrd", 18, 5)}, {MdpJsonSql.Str("m", "ProdLine")}, {MdpJsonSql.Str("m", "Name")},
                {MdpJsonSql.Str("m", "User1")},
                {MdpJsonSql.Str("m", "User1")}, {MdpJsonSql.Str("m", "Remark")}, {MdpJsonSql.Str("m", "CreateUser")},
                {MdpJsonSql.DateTimeSec("m", "CreateTime")},
                IFNULL({pretreatE}, ''),
                CASE WHEN {transTypeE}='Z61' THEN '补料' ELSE '正常' END,
                CASE WHEN {transTypeE}='PrevProcess' THEN '需要前处理' ELSE '' END,
                {MdpJsonSql.DateTimeSec("m", "EffDate")}, {MdpJsonSql.Str("m", "Address")},
                IFNULL(NULLIF(m.source_biz_key,''), CONCAT(IFNULL({domainE},''), ':', IFNULL({MdpJsonSql.Str("m", "RecID")},''))),
                @BatchId, @Now
            FROM mdp_stg_production_issue m
            LEFT JOIN DepartmentMaster d ON d.Domain = {domainE} AND d.Department = {deptE}
            WHERE m.source_table='NbrMaster'
              AND {MdpJsonSql.Str("m", "Type")}='SM'
              AND {MdpJsonSql.BoolTrue("m", "IsActive")}{srcClause}
              AND {MdpJsonSql.TenantGuard(mTenant)}
            ON DUPLICATE KEY UPDATE
                factory_id=VALUES(factory_id), source_system=VALUES(source_system), nbr=VALUES(nbr), issue_date=VALUES(issue_date),
                status=VALUES(status), status_desc=VALUES(status_desc), work_ord=VALUES(work_ord),
                department=VALUES(department), department_desc=VALUES(department_desc), qty_ord=VALUES(qty_ord),
                prod_line=VALUES(prod_line), applicant_name=VALUES(applicant_name), issue_user=VALUES(issue_user),
                user1=VALUES(user1), remark=VALUES(remark), create_user=VALUES(create_user),
                source_create_time=VALUES(source_create_time), pretreatment_state=VALUES(pretreatment_state),
                trans_type=VALUES(trans_type), trans_type_text=VALUES(trans_type_text),
                eff_date=VALUES(eff_date), address=VALUES(address),
                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, ProductionIssueMdpSyncResult 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.HeadRows),
            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 ProductionIssueMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int HeadRows { get; set; }
}

/// <summary>生产领料双源入站结果（stg 抽数 + std 转换）。</summary>
public sealed class ProductionIssueInboundResult
{
    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; }
}