AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/MaterialWarehouse/OutsourceIssueMdpSyncService.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 内部，head-detail 双 std 专表，独立于 S5MdpSyncTransformService 的 KPI 管线）。
///
/// B-① 迁通用管线：源(NbrMaster/NbrDetail, Type='CA') 经执行器灌 mdp_stg_outsource_issue_pull（头/明细共表，source_table 区分）
///   → transform 读 pull stg（MdpJsonSql 跨源类型兼容）→ mdp_std_outsource_issue(_detail)（typed，明细回填 std_head_id）。
///   维表 DepartmentMaster(d) 仍本库 LEFT JOIN（富化 department_name，业务 code 由事实表保留）。
///   注：canonical 执行器 stg 用 *_pull 命名（沿用 mdp_stg_ipqc_pull 惯例），与 legacy bespoke mdp_stg_outsource_issue(_detail) 区分；后者迁移后为死支路、不再消费、不动。
///
/// 双源：本库实体 S5_OUTSOURCE_ISSUE_MASTER/_DETAIL（AIDOPDEV_MYSQL, status=1）；
///   dopdemorq 第二源 S5_OUTSOURCE_ISSUE_MASTER_SQLSERVER/_DETAIL_SQLSERVER（DOPDEMORQ_SQLSERVER, status=0 就位不启用）。
///
/// 约束：
///   - 只读源/贴源，仅写 mdp_stg_outsource_issue_pull / mdp_std_outsource_issue(_detail)；绝不 INSERT/UPDATE/DELETE NbrMaster/NbrDetail。
///   - 过滤 NbrMaster.Type='CA' AND IsActive=1；NbrDetail.Type='CA'。
///   - status_desc 规则本批仅：Status='C' -> '关闭'，其他 -> ''。department_name 经 Domain+Department 关联 DepartmentMaster.Descr。
///   - 明细只转换已确认 9 列（ItemNum/ItemName/UM/QtyOrd/LocationFrom/LocationTo/Line/Status/Remark）；发料数量/已发数/批次号 3 候选列本批后置。
///   - tenant_id 从 raw_data 保留（本库源含该列 → std 租户不回归；SQL Server 源无此列 → 回落入站上下文 tenant）。
///   - CA 为 0 行时转换成功完成、处理数为 0，不报错。切源 FULL Replace 头/明细各一次单事务、仅按 tenant、Pull 成功后才 destructive。
/// </summary>
public class OutsourceIssueMdpSyncService : ITransient
{
    private const string JobCode = "S5_OUTSOURCE_ISSUE_MDP_SYNC";
    private const string InboundMasterEntityCode = "S5_OUTSOURCE_ISSUE_MASTER";
    private const string InboundDetailEntityCode = "S5_OUTSOURCE_ISSUE_DETAIL";

    private const string SqlServerSourceCode = "DOPDEMORQ_SQLSERVER";
    private const string SqlServerMasterEntityCode = "S5_OUTSOURCE_ISSUE_MASTER_SQLSERVER";
    private const string SqlServerDetailEntityCode = "S5_OUTSOURCE_ISSUE_DETAIL_SQLSERVER";
    /// <summary>165 SQL Server 源无 tenant_id 列；切源时须经账套映射解析。</summary>
    private const string SourceZtid = "pbxfxp";

    private const string StdHeadTable = "mdp_std_outsource_issue";
    private const string StdDetailTable = "mdp_std_outsource_issue_detail";

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

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

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

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

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

        var batchId = $"S5_OUTSRC_ISSUE_STD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "INBOUND");
        var result = new OutsourceIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.HeadStandardRows = await TransformHeadStandardAsync(batchId, now);
            result.DetailStandardRows = await TransformDetailStandardAsync(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 OutsourceIssueInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = pull.RowsPulled,
            RowsWrittenStg = pull.RowsWritten,
            NewCursor = pull.NewCursor,
            TransformBatchId = batchId,
            StdHeadRows = result.HeadStandardRows,
            StdDetailRows = result.DetailStandardRows
        };
    }

    /// <summary>
    /// 双源切换 FULL Replace：从指定 DB 源（默认 DOPDEMORQ_SQLSERVER）PullAll 全量灌 pull stg，
    /// 成功后头/明细各一次单事务 FULL 重建（按 tenant 精确隔离），消除旧源独有业务键残留。
    /// 顺序：先 head（写 std 头，供明细回填 std_head_id），再 detail。两次独立事务，Pull 成功后才 destructive。
    /// SQLSERVER 实体默认 status=0（就位不启用）；dopdemorq 六表当前空，Phase 1 不实际执行 destructive 切换。
    /// </summary>
    public async Task<OutsourceIssueInboundResult> 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;
        var pullCtx = new MdpPullContext
        {
            TenantId = tenantId,
            FullRefresh = true,
            TaskCode = "S5_OUTSOURCE_ISSUE_INBOUND",
            BatchId = $"S5_OUTSRC_ISSUE_SW_{now:yyyyMMddHHmmss}"
        };
        var master = await _pullDispatcher.PullAllByEntityCodeAsync(masterEntityCode, pullCtx, cancellationToken);
        var detail = await _pullDispatcher.PullAllByEntityCodeAsync(detailEntityCode, pullCtx, cancellationToken);

        var batchId = $"S5_OUTSRC_ISSUE_SWSTD_{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, "SOURCE_SWITCH");
        var result = new OutsourceIssueMdpSyncResult { BatchId = batchId, RunLogId = runLogId };
        try
        {
            result.HeadStandardRows = await MdpStdFullReplace.ReplaceAsync(
                _db, StdHeadTable, tenantId, extraWhere: null,
                insertScopedAsync: () => TransformHeadStandardAsync(batchId, now, sourceCode),
                cancellationToken);
            result.DetailStandardRows = await MdpStdFullReplace.ReplaceAsync(
                _db, StdDetailTable, 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 OutsourceIssueInboundResult
        {
            PullBatchId = pullCtx.BatchId,
            RowsPulled = master.RowsPulled + detail.RowsPulled,
            RowsWrittenStg = master.RowsWritten + detail.RowsWritten,
            NewCursor = detail.NewCursor ?? master.NewCursor,
            TransformBatchId = batchId,
            StdHeadRows = result.HeadStandardRows,
            StdDetailRows = result.DetailStandardRows
        };
    }

    private async Task<(int RowsPulled, int RowsWritten, string? NewCursor)> PopulateStgAsync(
        MdpPullContext pullCtx, CancellationToken cancellationToken)
    {
        var master = await _pullDispatcher.PullByEntityCodeAsync(InboundMasterEntityCode, 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_outsource_issue_pull"));
    }

    /// <summary>防御式建表（与 UpdateScripts/1.0.206.sql 同构，幂等）。std 头/明细两专表。</summary>
    private async Task EnsureTablesAsync()
    {
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_outsource_issue"));
        await MdpSchemaAligner.ExecuteAsync(_db, MdpSchemaDefinition.Ddl("mdp_std_outsource_issue_detail"));
    }

    /// <summary>标准化头：pull stg(NbrMaster, CA) + DepartmentMaster(本库) → mdp_std_outsource_issue。返回处理头行数。</summary>
    private async Task<int> TransformHeadStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
    {
        var gateSource = sourceSystem ?? MdpSourceIdentity.Native;
        if (!await _neutralGate.AllowsAsync(0, "OUTSOURCE", gateSource))
            return 0;

        var srcClause = sourceSystem == null ? "" : " AND m.source_system=@Src";
        var mTenant = $"COALESCE({MdpJsonSql.Int("m", "tenant_id")}, NULLIF(m.tenant_id, 0))";
        var statusExpr = MdpJsonSql.Str("m", "Status");
        var headWhere =
            $"""
            m.source_table='NbrMaster'
              AND {MdpJsonSql.Str("m", "Type")}='CA'
              AND {MdpJsonSql.BoolTrue("m", "IsActive")}{srcClause}
              AND {MdpJsonSql.TenantGuard(mTenant)}
            """;

        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_outsource_issue_pull m WHERE {headWhere}", countPars);

        var insertSql =
            $"""
            INSERT INTO mdp_std_outsource_issue
            (tenant_id, factory_id, source_system, bill_no, issue_date, outsource_no, work_order,
             department_code, department_name, issuer, status, status_desc, remark, create_user,
             source_create_time, eff_date, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {mTenant}, 1, {MdpSourceIdentity.Resolve("m.source_table", "m.source_system")},
                {MdpJsonSql.Str("m", "Nbr")}, {MdpJsonSql.DateTimeSec("m", "Date")}, {MdpJsonSql.Str("m", "Address")}, {MdpJsonSql.Str("m", "WorkOrd")},
                {MdpJsonSql.Str("m", "Department")}, d.Descr, {MdpJsonSql.Str("m", "User1")}, {statusExpr},
                CASE WHEN {statusExpr}='C' THEN '关闭' ELSE '' END,
                {MdpJsonSql.Str("m", "Remark")}, {MdpJsonSql.Str("m", "CreateUser")}, {MdpJsonSql.DateTimeSec("m", "CreateTime")}, {MdpJsonSql.DateTimeSec("m", "EffDate")},
                {MdpJsonSql.Str("m", "Nbr")}, @BatchId, @Now
            FROM mdp_stg_outsource_issue_pull m
            LEFT JOIN DepartmentMaster d
              ON d.Domain = {MdpJsonSql.Str("m", "Domain")} AND d.Department = {MdpJsonSql.Str("m", "Department")}
            WHERE {headWhere}
            ON DUPLICATE KEY UPDATE
                issue_date=VALUES(issue_date), outsource_no=VALUES(outsource_no), work_order=VALUES(work_order),
                department_code=VALUES(department_code), department_name=VALUES(department_name), issuer=VALUES(issuer),
                status=VALUES(status), status_desc=VALUES(status_desc), remark=VALUES(remark), create_user=VALUES(create_user),
                source_create_time=VALUES(source_create_time), eff_date=VALUES(eff_date),
                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;
    }

    /// <summary>标准化明细：pull stg(NbrDetail, CA) self-join stg(NbrMaster) by RecID↔NbrRecID → mdp_std_outsource_issue_detail（仅 9 列；回填 std_head_id）。</summary>
    private async Task<int> TransformDetailStandardAsync(string batchId, DateTime now, string? sourceSystem = null)
    {
        var gateSource = sourceSystem ?? MdpSourceIdentity.Native;
        if (!await _neutralGate.AllowsAsync(0, "OUTSOURCE", gateSource))
            return 0;

        var srcClause = sourceSystem == null ? "" : " AND m.source_system=@Src AND d.source_system=@Src";
        var dTenant = $"COALESCE({MdpJsonSql.Int("d", "tenant_id")}, NULLIF(d.tenant_id, 0))";
        var fromJoinWhere =
            $"""
            FROM mdp_stg_outsource_issue_pull d
            JOIN mdp_stg_outsource_issue_pull m
              ON m.source_table='NbrMaster'
             AND {MdpJsonSql.Int("m", "RecID")} = {MdpJsonSql.Int("d", "NbrRecID")}
             AND {MdpJsonSql.Str("m", "Type")}='CA'
             AND {MdpJsonSql.BoolTrue("m", "IsActive")}
            LEFT JOIN mdp_std_outsource_issue h
              ON h.tenant_id = {dTenant} AND h.bill_no = {MdpJsonSql.Str("d", "Nbr")}
            WHERE d.source_table='NbrDetail' AND {MdpJsonSql.Str("d", "Type")}='CA'{srcClause}
              AND {MdpJsonSql.TenantGuard(dTenant)}
            """;

        var countPars = new List<SugarParameter>();
        if (sourceSystem != null) countPars.Add(new SugarParameter("@Src", sourceSystem));
        var rows = await _db.Ado.GetIntAsync($"SELECT COUNT(1) {fromJoinWhere}", countPars);

        var insertSql =
            $"""
            INSERT INTO mdp_std_outsource_issue_detail
            (tenant_id, std_head_id, bill_no, line, item_num, item_name, um, qty_ord,
             location_from, location_to, status, remark, source_biz_key, sync_batch_id, sync_time)
            SELECT
                {dTenant}, h.id, {MdpJsonSql.Str("d", "Nbr")}, {MdpJsonSql.Int("d", "Line")}, {MdpJsonSql.Str("d", "ItemNum")}, {MdpJsonSql.Str("d", "ItemName")}, {MdpJsonSql.Str("d", "UM")}, {MdpJsonSql.Dec("d", "QtyOrd", 18, 6)},
                {MdpJsonSql.Str("d", "LocationFrom")}, {MdpJsonSql.Str("d", "LocationTo")}, {MdpJsonSql.Str("d", "Status")}, {MdpJsonSql.Str("d", "Remark")},
                CONCAT(IFNULL({MdpJsonSql.Str("d", "Nbr")},''), '#', IFNULL({MdpJsonSql.Int("d", "Line")},0)), @BatchId, @Now
            {fromJoinWhere}
            ON DUPLICATE KEY UPDATE
                std_head_id=VALUES(std_head_id), line=VALUES(line), item_num=VALUES(item_num), item_name=VALUES(item_name),
                um=VALUES(um), qty_ord=VALUES(qty_ord), location_from=VALUES(location_from), location_to=VALUES(location_to),
                status=VALUES(status), remark=VALUES(remark), 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, OutsourceIssueMdpSyncResult 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 OutsourceIssueMdpSyncResult
{
    public long RunLogId { get; set; }
    public string BatchId { get; set; } = string.Empty;
    public int HeadStageRows { get; set; }
    public int DetailStageRows { get; set; }
    public int HeadStandardRows { get; set; }
    public int DetailStandardRows { get; set; }
}

/// <summary>S5 委外发料双模式入站结果。</summary>
public sealed class OutsourceIssueInboundResult
{
    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 StdHeadRows { get; set; }
    public int StdDetailRows { get; set; }
}