AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Supply/PurchaseOrderCompletionMdpSyncService.cs

using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using SqlSugar;
using System.Data;
using System.Text;

namespace Admin.NET.Plugin.AiDOP.Supply;

/// <summary>
/// S8 Stage-3 采购订单完成态标准事实层投影服务。
///
/// <para>链路：Source B（WMS/MES 的 <c>PurOrdDetail</c>）→ <c>mdp_std_purchase_order_completion</c>。
/// 目的是让 S8 Stage-3 能在【不覆盖】<c>mdp_std_purchase_order.status</c> 的前提下拿到
/// 「这条采购行收货是否已关闭」这一 WMS 侧事实。</para>
///
/// <para><b>Grain = 一条采购行</b>（<c>tenant_id + po_no + po_line</c>）。
/// <c>domain</c> / <c>potype</c> / <c>source_row_id</c> / <c>source_id</c> 是 provenance，不进唯一键。</para>
///
/// <para><b>当前状态，不是事件</b>：上游收货过程的写法是
/// <c>Status = (case when 收满 then 'C' else '' end)</c>，退货会把 <c>'C'</c> 打回空串。
/// 因此本服务的 UPSERT 必须无条件刷新结论，允许 COMPLETED → NOT_COMPLETED 回退。</para>
///
/// <para><b>Full Reconciliation 是正确性主路径</b>：HotWatch 只覆盖活动中的采购单，
/// 实测 5 个 <c>PUR_ORDER</c> 关注里 3 个已提前终止（且无一是因采购完成而终止），
/// 98 条源行里 88 条根本没有对应关注。所以单靠增量必然漏判，周期性全量对账不可省。</para>
///
/// <para><b>刻意不做</b>：不读 IQC、不算 ObjectCompleted、不做订单级聚合、
/// 不用数量重新推导完成（唯一 Authority 是源侧 Status）、不写任何源表、不改 Source A 的任何列。</para>
/// </summary>
public class PurchaseOrderCompletionMdpSyncService : ITransient
{
    private const string JobCode = "S3_PO_COMPLETION_MDP_SYNC";
    private const string JobName = "S3采购订单完成态标准事实层投影";

    /// <summary>完成结论取值域。</summary>
    public const string StatusCompleted = "COMPLETED";
    public const string StatusNotCompleted = "NOT_COMPLETED";
    public const string StatusUnknown = "UNKNOWN";

    private readonly ISqlSugarClient _db;
    private readonly TransformRunLogFinalizer _runLogFinalizer;
    private readonly DataPlatform.MdpSourceScopeFactory _scopeFactory;
    private readonly AidopPoCompletionOptions _opt;
    private readonly IHostEnvironment _env;
    private readonly ILogger<PurchaseOrderCompletionMdpSyncService> _logger;

    public PurchaseOrderCompletionMdpSyncService(
        ISqlSugarClient db,
        DataPlatform.MdpSourceScopeFactory scopeFactory,
        IOptions<AidopPoCompletionOptions> opt,
        IHostEnvironment env,
        ILogger<PurchaseOrderCompletionMdpSyncService> logger,
        TransformRunLogFinalizer runLogFinalizer)
    {
        _db = db;
        _runLogFinalizer = runLogFinalizer;
        _scopeFactory = scopeFactory;
        _opt = opt.Value;
        _env = env;
        _logger = logger;
    }

    /// <summary>
    /// 完成态映射。<b>这是全链路唯一的状态映射入口</b>——HotWatch 增量若接入，必须复用本方法，
    /// 不得另写一套。
    ///
    /// <para><c>'C'</c> → COMPLETED；空串 → NOT_COMPLETED；
    /// <c>null</c> 与任何其它非空值 → UNKNOWN。</para>
    ///
    /// <para>注意这里刻意<b>不</b>写成 <c>IFNULL(status,'') == ""</c>：
    /// NULL 表示「源侧没给值」，未知值表示「源侧给了我们不认识的值」，
    /// 两者都不等于「已确认尚未完成」，滑成 NOT_COMPLETED 会让下游把未知当成确定结论。</para>
    /// </summary>
    public static string MapCompletionStatus(string? rawStatus)
    {
        if (rawStatus is null) return StatusUnknown;
        var s = rawStatus.Trim();
        if (s.Length == 0) return StatusNotCompleted;
        return string.Equals(s, "C", StringComparison.OrdinalIgnoreCase)
            ? StatusCompleted
            : StatusUnknown;
    }

    /// <summary>写入本表时使用的 raw_status 归一：保留 NULL，其余 TRIM。</summary>
    public static string? NormalizeRawStatus(string? rawStatus) => rawStatus?.Trim();

    /// <summary>
    /// 全量对账（Full Reconciliation）。
    /// <paramref name="tenantId"/> 为 0 表示全租户；&gt;0 时只写/只淘汰该租户，绝不触碰其它租户。
    /// </summary>
    public async Task<PurchaseOrderCompletionSyncResult> RunFullAsync(
        long tenantId = 0,
        string triggerType = "AUTO",
        CancellationToken cancellationToken = default)
    {
        cancellationToken.ThrowIfCancellationRequested();

        var now = DateTime.Now;
        var batchId = $"S3_PO_COMPL_{(tenantId > 0 ? tenantId + "_" : "")}{now:yyyyMMddHHmmss}";
        var runLogId = await InsertRunLogAsync(batchId, now, triggerType, tenantId);
        var result = new PurchaseOrderCompletionSyncResult { BatchId = batchId, RunLogId = runLogId };

        try
        {
            // ① 源绑定：非开发环境必须显式绑定真实 WMS 源，绝不静默回落
            var binding = await ResolveSourceBindingAsync(cancellationToken);
            result.SourceCode = binding.SourceCode;

            var remote = await _scopeFactory.GetScopeAsync(binding.SourceCode, cancellationToken);

            // 作用域工厂对「本库样板源」与「未配账号的源」会直接返回主库连接。
            // 那会让本服务把 Source A 的本地表当成 WMS 执行态读进来，结论会整体错误，
            // 所以这里必须结构性挡住，而不是依赖配置写对。
            if (ReferenceEquals(remote, _db))
                throw new InvalidOperationException(
                    $"源 {binding.SourceCode} 解析结果是主库连接，说明它是本库样板源或未配置独立账号，"
                    + "不能承担 S8_PO_COMPLETION_SOURCE 角色");

            // ② 读源：只取本角色范围内的采购行
            var sourceRows = await ReadSourceRowsAsync(remote, cancellationToken);
            result.SourceRows = sourceRows.Count;

            // ③ 归属与校验
            var tenantIndex = await LoadLocalTenantIndexAsync(cancellationToken);
            var poLineIndex = await LoadLocalPoLineIndexAsync(cancellationToken);
            var pending = new List<CompletionRow>(sourceRows.Count);

            foreach (var row in sourceRows)
            {
                cancellationToken.ThrowIfCancellationRequested();

                if (string.IsNullOrWhiteSpace(row.PoNo) || string.IsNullOrWhiteSpace(row.PoLine))
                {
                    result.SkippedRows++;
                    continue;
                }

                var resolved = ResolveTenant(tenantIndex, row.PoNo, row.Domain);
                if (resolved <= 0)
                {
                    result.SkippedRows++;
                    result.TenantUnresolvedRows++;
                    _logger.LogWarning(
                        "[PoCompletion] 跳过：采购单 {PurOrd} 行 {Line} 无法唯一解析租户（候选数 {Count}）",
                        row.PoNo, row.PoLine, CountTenantCandidates(tenantIndex, row.PoNo, row.Domain));
                    continue;
                }

                // 单租户刷新：其它租户的源行本轮不参与，也因此不能被本轮淘汰
                if (tenantId > 0 && resolved != tenantId)
                    continue;

                var matches = poLineIndex.TryGetValue(LineKey(resolved, row.PoNo, row.PoLine), out var c) ? c : 0;
                if (matches > 1)
                    throw new InvalidOperationException(
                        $"数据契约破坏：租户 {resolved} 的采购行 {row.PoNo}#{row.PoLine} 在本库命中 {matches} 条，"
                        + "唯一键 uk_po_line 应当保证唯一");
                if (matches == 0)
                {
                    result.SkippedRows++;
                    result.UnmatchedRows++;
                    _logger.LogWarning(
                        "[PoCompletion] 跳过：源行 {PurOrd}#{Line} 在本库租户 {Tenant} 下无对应采购行",
                        row.PoNo, row.PoLine, resolved);
                    continue;
                }

                result.MatchedRows++;
                var raw = NormalizeRawStatus(row.RawStatus);
                var completion = MapCompletionStatus(row.RawStatus);
                if (completion == StatusCompleted) result.CompletedRows++;
                else if (completion == StatusNotCompleted) result.NotCompletedRows++;
                else result.UnknownStatusRows++;

                pending.Add(new CompletionRow
                {
                    TenantId = resolved,
                    PoNo = row.PoNo.Trim(),
                    PoLine = row.PoLine.Trim(),
                    Domain = string.IsNullOrWhiteSpace(row.Domain) ? null : row.Domain.Trim(),
                    Potype = string.IsNullOrWhiteSpace(row.Potype) ? null : row.Potype.Trim(),
                    RawStatus = raw,
                    CompletionStatus = completion,
                    SourceSystem = binding.SourceCode,
                    SourceId = binding.SourceId,
                    SourceRowId = string.IsNullOrWhiteSpace(row.SourceRowId) ? null : row.SourceRowId.Trim(),
                    SourceUpdateTime = row.SourceUpdateTime,
                    SourceUpdateUser = string.IsNullOrWhiteSpace(row.SourceUpdateUser) ? null : row.SourceUpdateUser.Trim(),
                });
            }

            // ④ UPSERT：无条件刷新结论，支持 COMPLETED → NOT_COMPLETED 回退
            result.WrittenRows = await UpsertAsync(pending, batchId, now, cancellationToken);

            // ⑤ 淘汰：源侧已不存在的行回到 NOT_OBSERVED（本表的表达方式就是「没有这一行」）
            //    只在本轮源侧读取成功后执行；任何异常都会在上面抛出并跳过这里。
            result.StaleRemoved = await RetireStaleAsync(batchId, tenantId, cancellationToken);

            await CompleteRunLogAsync(runLogId, result, now);
            return result;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "[PoCompletion] 批次 {BatchId} 失败", batchId);
            // 宿主关停不是转换失败：交给 finally 收口为 ABORTED，不污染 FAILED 语义。
            if (!_runLogFinalizer.IsHostStopping)
                await FailRunLogAsync(runLogId, ex.Message);
            throw;
        }
        finally
        {
            await _runLogFinalizer.FinalizeIfHostStoppingAsync(runLogId, now);
        }
    }

    // ── 源绑定与 fail-closed ────────────────────────────────────────────────────────

    /// <summary>
    /// 解析并校验源绑定。非开发环境下，未配置、配置为空、或仍指向 DEV/UAT 默认源，
    /// 一律直接失败（fail-closed），不得静默继续。
    /// </summary>
    private async Task<SourceBinding> ResolveSourceBindingAsync(CancellationToken ct)
    {
        var code = (_opt.SourceCode ?? string.Empty).Trim();
        var isDev = _env.IsDevelopment();

        if (code.Length == 0)
            throw new InvalidOperationException(
                "AiDOP:PoCompletion:SourceCode 未配置，无法确定承担 S8_PO_COMPLETION_SOURCE 角色的数据源");

        if (!isDev && (_opt.DevOnlySourceCodes ?? Array.Empty<string>())
                .Any(x => string.Equals((x ?? string.Empty).Trim(), code, StringComparison.OrdinalIgnoreCase)))
            throw new InvalidOperationException(
                $"当前环境 {_env.EnvironmentName} 非开发环境，禁止使用 DEV/UAT 数据源 {code}；"
                + "请在 AiDOP:PoCompletion:SourceCode 显式绑定真实 WMS/MES 源");

        var src = await _db.Ado.SqlQuerySingleAsync<SourceBindingRow>(
            """
            SELECT id AS Id, source_type AS SourceType, IFNULL(db_user,'') AS DbUser
              FROM mdp_source
             WHERE source_code=@Code AND status=1
             LIMIT 1
            """,
            new List<SugarParameter> { new("@Code", code) });

        if (src == null)
            throw new InvalidOperationException($"mdp_source 未找到启用源：{code}");
        if (!string.Equals(src.SourceType, "DB", StringComparison.OrdinalIgnoreCase))
            throw new InvalidOperationException($"源 {code} 的 source_type={src.SourceType}，不是 DB");
        if (string.IsNullOrWhiteSpace(src.DbUser))
            throw new InvalidOperationException(
                $"源 {code} 未配置数据库账号，连接会回落到主库，不能承担 S8_PO_COMPLETION_SOURCE 角色");

        return new SourceBinding { SourceCode = code, SourceId = src.Id };
    }

    // ── 源侧读取 ────────────────────────────────────────────────────────────────────

    /// <summary>
    /// 读 Source B 的采购行。范围 = 与自建单推送口径一致的 Potype，
    /// 不做无差别全表扫，也不顺带拉无关列。
    /// </summary>
    private async Task<List<SourceRow>> ReadSourceRowsAsync(ISqlSugarClient remote, CancellationToken ct)
    {
        var top = _opt.MaxSourceRows > 0 ? _opt.MaxSourceRows : 50000;
        var potype = (_opt.SourcePotype ?? string.Empty).Trim();

        var sql =
            $"""
            SELECT TOP {top}
                   RecID, Domain, Potype, PurOrd, Line, Status, UpdateTime, UpdateUser
              FROM PurOrdDetail
             WHERE (@Potype = '' OR LOWER(LTRIM(RTRIM(ISNULL(Potype,'')))) = LOWER(@Potype))
            """;

        var table = await remote.Ado.GetDataTableAsync(sql, new List<SugarParameter> { new("@Potype", potype) });
        var list = new List<SourceRow>(table.Rows.Count);
        foreach (DataRow r in table.Rows)
        {
            ct.ThrowIfCancellationRequested();
            list.Add(new SourceRow
            {
                SourceRowId = Str(r, "RecID"),
                Domain = Str(r, "Domain"),
                Potype = Str(r, "Potype"),
                PoNo = Str(r, "PurOrd"),
                PoLine = Str(r, "Line"),
                RawStatus = Str(r, "Status"),
                SourceUpdateTime = Dt(r, "UpdateTime"),
                SourceUpdateUser = Str(r, "UpdateUser"),
            });
        }
        return list;
    }

    // ── 租户归属 ────────────────────────────────────────────────────────────────────

    /// <summary>
    /// 本地采购单 → 租户候选集。Domain 只用于放宽匹配（自建单本库不填 Domain），
    /// <b>绝不</b>用 Domain 反推租户 —— 实测同一个 Domain 8010 横跨多个 Ai-DOP 租户。
    /// </summary>
    private async Task<Dictionary<string, List<PoTenantCandidate>>> LoadLocalTenantIndexAsync(CancellationToken ct)
    {
        var rows = await _db.Ado.SqlQueryAsync<PoTenantCandidate>(
            """
            SELECT PurOrd AS PoNo, IFNULL(Domain,'') AS Domain, tenant_id AS TenantId
              FROM PurOrdDetail
             WHERE IFNULL(tenant_id,0) > 0
             GROUP BY PurOrd, IFNULL(Domain,''), tenant_id
            """);

        var map = new Dictionary<string, List<PoTenantCandidate>>(StringComparer.OrdinalIgnoreCase);
        foreach (var r in rows)
        {
            if (string.IsNullOrWhiteSpace(r.PoNo)) continue;
            var key = r.PoNo.Trim();
            if (!map.TryGetValue(key, out var bucket))
                map[key] = bucket = new List<PoTenantCandidate>();
            bucket.Add(r);
        }
        return map;
    }

    /// <summary>
    /// 候选租户集合。Domain 只用于<b>放宽</b>匹配（自建单本库不填 Domain，空 Domain 视为通配），
    /// <b>绝不</b>反过来用 Domain 推租户 —— 实测同一个 Domain 8010 横跨多个 Ai-DOP 租户。
    /// </summary>
    public static IEnumerable<long> TenantCandidates(
        IEnumerable<PoTenantCandidate> candidates, string poNo, string? domain)
    {
        var po = (poNo ?? string.Empty).Trim();
        var dom = (domain ?? string.Empty).Trim();
        return candidates
            .Where(x => string.Equals((x.PoNo ?? string.Empty).Trim(), po, StringComparison.OrdinalIgnoreCase))
            .Where(x => (x.Domain ?? string.Empty).Trim().Length == 0
                        || string.Equals((x.Domain ?? string.Empty).Trim(), dom, StringComparison.OrdinalIgnoreCase))
            .Select(x => x.TenantId)
            .Distinct();
    }

    /// <summary>
    /// 唯一才用：0 个（UNRESOLVED）或 &gt;1 个（AMBIGUOUS）一律返回 0，交由调用方跳过。
    /// <b>禁止</b>退化成「取第一个」—— 那会在多租户重名时把事实写到别人名下。
    /// </summary>
    public static long ResolveTenantId(
        IEnumerable<PoTenantCandidate> candidates, string poNo, string? domain)
    {
        var hits = TenantCandidates(candidates, poNo, domain).Take(2).ToList();
        return hits.Count == 1 ? hits[0] : 0;
    }

    private static long ResolveTenant(
        Dictionary<string, List<PoTenantCandidate>> index, string poNo, string? domain)
        => index.TryGetValue(poNo.Trim(), out var bucket) ? ResolveTenantId(bucket, poNo, domain) : 0;

    private static int CountTenantCandidates(
        Dictionary<string, List<PoTenantCandidate>> index, string poNo, string? domain)
        => index.TryGetValue(poNo.Trim(), out var bucket)
            ? TenantCandidates(bucket, poNo, domain).Count()
            : 0;

    // ── 本地采购行校验 ──────────────────────────────────────────────────────────────

    private async Task<Dictionary<string, int>> LoadLocalPoLineIndexAsync(CancellationToken ct)
    {
        var rows = await _db.Ado.SqlQueryAsync<PoLineKeyRow>(
            "SELECT tenant_id AS TenantId, po_no AS PoNo, po_line AS PoLine FROM mdp_std_purchase_order");

        var map = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
        foreach (var r in rows)
        {
            var key = LineKey(r.TenantId, r.PoNo ?? string.Empty, r.PoLine ?? string.Empty);
            map[key] = map.TryGetValue(key, out var n) ? n + 1 : 1;
        }
        return map;
    }

    private static string LineKey(long tenantId, string poNo, string poLine)
        => $"{tenantId}|{poNo.Trim()}|{poLine.Trim()}";

    // ── 写入 ────────────────────────────────────────────────────────────────────────

    /// <summary>
    /// 幂等 UPSERT。结论列一律 <c>VALUES(...)</c> 无条件覆盖——这是支持
    /// COMPLETED → NOT_COMPLETED 回退的前提，绝不能改成「仅当新值为 C 才更新」。
    /// </summary>
    private async Task<int> UpsertAsync(
        List<CompletionRow> rows, string batchId, DateTime now, CancellationToken ct)
    {
        if (rows.Count == 0) return 0;

        var size = _opt.UpsertBatchSize > 0 ? _opt.UpsertBatchSize : 200;
        var written = 0;

        foreach (var chunk in Chunk(rows, size))
        {
            ct.ThrowIfCancellationRequested();

            var sb = new StringBuilder();
            sb.Append("""
                INSERT INTO mdp_std_purchase_order_completion
                (tenant_id, factory_id, po_no, po_line, domain, potype,
                 raw_status, completion_status,
                 source_system, source_id, source_row_id, source_update_time, source_update_user,
                 observed_at, sync_batch_id, sync_time)
                VALUES
                """);

            var pars = new List<SugarParameter> { new("@Observed", now), new("@Batch", batchId), new("@Now", now) };
            for (var i = 0; i < chunk.Count; i++)
            {
                var r = chunk[i];
                if (i > 0) sb.Append(',');
                sb.Append($"(@t{i},1,@p{i},@l{i},@d{i},@y{i},@r{i},@c{i},@s{i},@i{i},@x{i},@u{i},@w{i},@Observed,@Batch,@Now)");
                pars.Add(new SugarParameter($"@t{i}", r.TenantId));
                pars.Add(new SugarParameter($"@p{i}", r.PoNo));
                pars.Add(new SugarParameter($"@l{i}", r.PoLine));
                pars.Add(new SugarParameter($"@d{i}", (object?)r.Domain ?? DBNull.Value));
                pars.Add(new SugarParameter($"@y{i}", (object?)r.Potype ?? DBNull.Value));
                pars.Add(new SugarParameter($"@r{i}", (object?)r.RawStatus ?? DBNull.Value));
                pars.Add(new SugarParameter($"@c{i}", r.CompletionStatus));
                pars.Add(new SugarParameter($"@s{i}", r.SourceSystem));
                pars.Add(new SugarParameter($"@i{i}", (object?)r.SourceId ?? DBNull.Value));
                pars.Add(new SugarParameter($"@x{i}", (object?)r.SourceRowId ?? DBNull.Value));
                pars.Add(new SugarParameter($"@u{i}", (object?)r.SourceUpdateTime ?? DBNull.Value));
                pars.Add(new SugarParameter($"@w{i}", (object?)r.SourceUpdateUser ?? DBNull.Value));
            }

            sb.Append("""
                 ON DUPLICATE KEY UPDATE
                    factory_id=VALUES(factory_id), domain=VALUES(domain), potype=VALUES(potype),
                    raw_status=VALUES(raw_status), completion_status=VALUES(completion_status),
                    source_system=VALUES(source_system), source_id=VALUES(source_id),
                    source_row_id=VALUES(source_row_id),
                    source_update_time=VALUES(source_update_time),
                    source_update_user=VALUES(source_update_user),
                    observed_at=VALUES(observed_at), sync_batch_id=VALUES(sync_batch_id),
                    sync_time=VALUES(sync_time), update_time=CURRENT_TIMESTAMP
                """);

            await _db.Ado.ExecuteCommandAsync(sb.ToString(), pars);
            written += chunk.Count;
        }

        return written;
    }

    /// <summary>
    /// 淘汰本轮未覆盖的行 —— 它们在源侧已不存在，应回到 NOT_OBSERVED（即本表无此行）。
    ///
    /// <para>只在本方法被调用时执行，而调用点在源侧读取与写入全部成功之后；
    /// 读取失败会在更早处抛出，因此「源侧抓取失败却把旧事实删掉」不可能发生。</para>
    ///
    /// <para>单租户刷新只淘汰该租户 —— 租户 A 的刷新绝不允许删掉租户 B 的事实。</para>
    /// </summary>
    private async Task<int> RetireStaleAsync(string batchId, long tenantId, CancellationToken ct)
    {
        ct.ThrowIfCancellationRequested();

        if (tenantId > 0)
        {
            return await _db.Ado.ExecuteCommandAsync(
                """
                DELETE FROM mdp_std_purchase_order_completion
                 WHERE tenant_id=@TenantId AND IFNULL(sync_batch_id,'') <> @BatchId
                """,
                new List<SugarParameter> { new("@TenantId", tenantId), new("@BatchId", batchId) });
        }

        return await _db.Ado.ExecuteCommandAsync(
            """
            DELETE FROM mdp_std_purchase_order_completion
             WHERE tenant_id > 0 AND IFNULL(sync_batch_id,'') <> @BatchId
            """,
            new List<SugarParameter> { new("@BatchId", batchId) });
    }

    // ── run log ─────────────────────────────────────────────────────────────────────

    private async Task<long> InsertRunLogAsync(string batchId, DateTime startedAt, string triggerType, long tenantId)
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            INSERT INTO mdp_transform_run_log
            (tenant_id, job_code, job_name, trigger_type, batch_id, status, start_time)
            VALUES (@TenantId, @JobCode, @JobName, @TriggerType, @BatchId, 'RUNNING', @StartTime)
            """,
            new SugarParameter("@TenantId", tenantId),
            new SugarParameter("@JobCode", JobCode),
            new SugarParameter("@JobName", JobName),
            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 CompleteRunLogAsync(long runLogId, PurchaseOrderCompletionSyncResult r, DateTime startedAt)
    {
        var endedAt = DateTime.Now;

        // summary_json 是 JSON 列，必须写合法 JSON —— 纯文本会被 MySQL 直接拒绝。
        // 这里只放计数与源编码，不放任何连接信息。
        var summary = System.Text.Json.JsonSerializer.Serialize(new
        {
            sourceCode = r.SourceCode,
            sourceRows = r.SourceRows,
            matchedRows = r.MatchedRows,
            writtenRows = r.WrittenRows,
            completedRows = r.CompletedRows,
            notCompletedRows = r.NotCompletedRows,
            unknownStatusRows = r.UnknownStatusRows,
            skippedRows = r.SkippedRows,
            tenantUnresolvedRows = r.TenantUnresolvedRows,
            unmatchedRows = r.UnmatchedRows,
            staleRemoved = r.StaleRemoved,
        });

        await _db.Ado.ExecuteCommandAsync(
            """
            UPDATE mdp_transform_run_log
               SET status='SUCCESS', stage_rows=@SrcRows, standard_rows=@StdRows, end_time=@EndTime,
                   duration_ms=@Duration, summary_json=@Summary, update_time=CURRENT_TIMESTAMP
             WHERE id=@Id
            """,
            new SugarParameter("@SrcRows", r.SourceRows),
            new SugarParameter("@StdRows", r.WrittenRows),
            new SugarParameter("@EndTime", endedAt),
            new SugarParameter("@Duration", (int)(endedAt - startedAt).TotalMilliseconds),
            // 刻意不截断：这是固定形状的计数 JSON（长度有界），截断会产生非法 JSON 并让整轮失败
            new SugarParameter("@Summary", summary),
            new SugarParameter("@Id", runLogId));
    }

    private async Task FailRunLogAsync(long runLogId, string message)
    {
        await _db.Ado.ExecuteCommandAsync(
            """
            UPDATE mdp_transform_run_log
               SET status='FAILED', end_time=@EndTime, error_message=@Msg, update_time=CURRENT_TIMESTAMP
             WHERE id=@Id
            """,
            new SugarParameter("@EndTime", DateTime.Now),
            new SugarParameter("@Msg", Truncate(message, 900)),
            new SugarParameter("@Id", runLogId));
    }

    // ── 工具 ────────────────────────────────────────────────────────────────────────

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

    private static string Truncate(string? s, int max)
        => string.IsNullOrEmpty(s) ? string.Empty : (s.Length <= max ? s : s[..max]);

    private static string? Str(DataRow row, string col)
    {
        if (!row.Table.Columns.Contains(col)) return null;
        var v = row[col];
        return v == null || v == DBNull.Value ? null : Convert.ToString(v);
    }

    private static DateTime? Dt(DataRow row, string col)
    {
        if (!row.Table.Columns.Contains(col)) return null;
        var v = row[col];
        if (v == null || v == DBNull.Value) return null;
        return Convert.ToDateTime(v);
    }

    private static IEnumerable<List<T>> Chunk<T>(List<T> source, int size)
    {
        for (var i = 0; i < source.Count; i += size)
            yield return source.GetRange(i, Math.Min(size, source.Count - i));
    }

    // ── 内部模型 ────────────────────────────────────────────────────────────────────

    private sealed class SourceRow
    {
        public string? SourceRowId { get; set; }
        public string? Domain { get; set; }
        public string? Potype { get; set; }
        public string PoNo { get; set; } = string.Empty;
        public string PoLine { get; set; } = string.Empty;
        public string? RawStatus { get; set; }
        public DateTime? SourceUpdateTime { get; set; }
        public string? SourceUpdateUser { get; set; }
    }

    private sealed class CompletionRow
    {
        public long TenantId { get; set; }
        public string PoNo { get; set; } = string.Empty;
        public string PoLine { get; set; } = string.Empty;
        public string? Domain { get; set; }
        public string? Potype { get; set; }
        public string? RawStatus { get; set; }
        public string CompletionStatus { get; set; } = StatusUnknown;
        public string SourceSystem { get; set; } = string.Empty;
        public long? SourceId { get; set; }
        public string? SourceRowId { get; set; }
        public DateTime? SourceUpdateTime { get; set; }
        public string? SourceUpdateUser { get; set; }
    }

    private sealed class PoLineKeyRow
    {
        public long TenantId { get; set; }
        public string? PoNo { get; set; }
        public string? PoLine { get; set; }
    }

    private sealed class SourceBinding
    {
        public string SourceCode { get; set; } = string.Empty;
        public long? SourceId { get; set; }
    }

    private sealed class SourceBindingRow
    {
        public long Id { get; set; }
        public string? SourceType { get; set; }
        public string? DbUser { get; set; }
    }
}

/// <summary>
/// 本地采购单 → 租户候选。<c>Domain</c> 为空表示自建单（本库不填 165 账套），视为通配。
/// 仅用于把 Source B 的行归到某个 Ai-DOP 租户，<b>不是</b>租户边界本身。
/// </summary>
public sealed class PoTenantCandidate
{
    public string PoNo { get; set; } = string.Empty;
    public string Domain { get; set; } = string.Empty;
    public long TenantId { get; set; }
}

/// <summary>S3 采购完成态投影结果。</summary>
public sealed class PurchaseOrderCompletionSyncResult
{
    public string BatchId { get; set; } = string.Empty;
    public long RunLogId { get; set; }

    /// <summary>承担本次角色的数据源编码（不含任何连接信息）。</summary>
    public string SourceCode { get; set; } = string.Empty;

    /// <summary>源侧读到的行数。</summary>
    public int SourceRows { get; set; }

    /// <summary>唯一命中本地采购行的行数。</summary>
    public int MatchedRows { get; set; }

    /// <summary>实际写入（新增或刷新）的行数。</summary>
    public int WrittenRows { get; set; }

    /// <summary>被跳过的源行数（租户不可解析 + 本库无对应采购行 + 缺关键字段）。</summary>
    public int SkippedRows { get; set; }

    /// <summary>其中：租户无法唯一解析。</summary>
    public int TenantUnresolvedRows { get; set; }

    /// <summary>其中：本库无对应采购行。</summary>
    public int UnmatchedRows { get; set; }

    public int CompletedRows { get; set; }
    public int NotCompletedRows { get; set; }

    /// <summary>源侧给了 NULL 或无法识别的状态值。</summary>
    public int UnknownStatusRows { get; set; }

    /// <summary>本轮源侧已不存在、被回收为 NOT_OBSERVED 的行数。</summary>
    public int StaleRemoved { get; set; }
}