AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/Executors/MdpDbPullExecutor.cs

using System.Data;
using System.Globalization;
using System.Text.Json;
using System.Text.Json.Serialization;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using SqlSugar;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.Executors;

/// <summary>
/// 方式甲：从 mdp_source DB 源按 mdp_entity.source_table_name 增量抽数 → target_table_name（贴源）。
/// </summary>
public sealed class MdpDbPullExecutor : IMdpSourcePullExecutor, ITransient
{
    public string SupportedType => "DB_SYNC";

    /// <summary>
    /// 贴源 raw_data 的时间统一按 MySQL 字面量格式输出；各域转换层用
    /// STR_TO_DATE(..., '%Y-%m-%d %H:%i:%s.%f') 解析，ISO 8601 的 T 分隔符会解析失败。
    /// </summary>
    private static readonly JsonSerializerOptions RawDataJsonOptions = new()
    {
        Converters =
        {
            new MysqlLiteralDateTimeConverter(),
            new MysqlLiteralDateTimeOffsetConverter()
        }
    };

    private readonly MdpSourceScopeFactory _scopeFactory;
    private readonly ISqlSugarClient _db;
    private readonly MdpStagingWriter _writer;
    private readonly MdpPullFieldMappingService _fieldMappings;

    public MdpDbPullExecutor(
        MdpSourceScopeFactory scopeFactory,
        ISqlSugarClient db,
        MdpStagingWriter writer,
        MdpPullFieldMappingService fieldMappings)
    {
        _scopeFactory = scopeFactory;
        _db = db;
        _writer = writer;
        _fieldMappings = fieldMappings;
    }

    public async Task<MdpPullResult> PullAsync(MdpSource source, MdpEntity entity, MdpPullContext ctx, CancellationToken cancellationToken = default)
    {
        if (string.IsNullOrWhiteSpace(entity.SourceTableName))
            throw new InvalidOperationException($"实体 {entity.EntityCode} 未配置 source_table_name");
        if (string.IsNullOrWhiteSpace(entity.TargetTableName))
            throw new InvalidOperationException($"实体 {entity.EntityCode} 未配置 target_table_name");

        var startedAt = DateTime.Now;
        var scope = await _scopeFactory.GetScopeAsync(source, cancellationToken);
        var batchSize = entity.BatchSize > 0 ? entity.BatchSize : 1000;
        var isSqlServer = string.Equals(source.DbType, "SQLSERVER", StringComparison.OrdinalIgnoreCase)
            || string.Equals(source.DbType, "MSSQL", StringComparison.OrdinalIgnoreCase);

        var useKeyset = ctx.UseKeysetCursor
            && !string.IsNullOrWhiteSpace(ctx.CursorColumn)
            && !string.IsNullOrWhiteSpace(ctx.TieBreakerColumn);

        string? tenantCol = null;
        string? factoryCol = null;
        if (ctx.TenantId > 0)
        {
            var cols = await TryGetSourceColumnsAsync(scope, entity.SourceTableName!, isSqlServer, cancellationToken);
            tenantCol = FindColumn(cols, "tenant_id", "TenantId", "TenantID");
            if (ctx.FactoryId > 0)
                factoryCol = FindColumn(cols, "factory_id", "FactoryId");
        }

        var sql = useKeyset
            ? BuildKeysetSelectSql(entity, isSqlServer, batchSize, ctx, tenantCol, factoryCol)
            : BuildSelectSql(entity, isSqlServer, batchSize, ctx, tenantCol, factoryCol);
        var parameters = new List<SugarParameter>();
        if (NeedsTenantParameter(tenantCol, factoryCol))
            parameters.Add(new SugarParameter("@scopeTenantId", ctx.TenantId));
        if (factoryCol != null)
            parameters.Add(new SugarParameter("@scopeFactoryId", ctx.FactoryId));
        if (useKeyset)
        {
            AddKeysetParameters(parameters, ctx);
        }
        else
        {
            if (ctx.WindowFrom.HasValue)
                parameters.Add(new SugarParameter("@windowFrom", ctx.WindowFrom.Value));
            if (!ctx.FullRefresh
                && !string.Equals(ctx.SyncWindowType, "FULL", StringComparison.OrdinalIgnoreCase)
                && !string.IsNullOrWhiteSpace(entity.IncrColumn)
                && !string.IsNullOrWhiteSpace(entity.LastCursor)
                && !string.Equals(ctx.SyncWindowType, "ROLLING", StringComparison.OrdinalIgnoreCase))
                parameters.Add(new SugarParameter("@cursor", entity.LastCursor));
            // ROLLING：用窗口下界；若已有水位且水位晚于下界，则进一步用水位收窄
            if (!ctx.FullRefresh
                && string.Equals(ctx.SyncWindowType, "ROLLING", StringComparison.OrdinalIgnoreCase)
                && !string.IsNullOrWhiteSpace(entity.IncrColumn)
                && !string.IsNullOrWhiteSpace(entity.LastCursor)
                && ctx.WindowFrom.HasValue
                && DateTime.TryParse(entity.LastCursor, out var cursorDt)
                && cursorDt > ctx.WindowFrom.Value)
                parameters.Add(new SugarParameter("@cursor", entity.LastCursor));
        }

        var table = await scope.Ado.GetDataTableAsync(sql, parameters);
        cancellationToken.ThrowIfCancellationRequested();

        var written = 0;
        var rowErrors = 0;
        var errorSamples = new List<string>();
        string? maxCursor = entity.LastCursor;
        var now = DateTime.Now;
        string? lastKeysetCursor = null;
        string? lastKeysetTie = null;
        var pending = new List<(IDictionary<string, object?> Row, string RawJson, string SourceRowId)>();
        var mappings = await _fieldMappings.GetAsync(entity.Id, cancellationToken);

        foreach (DataRow row in table.Rows)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var dict = new Dictionary<string, object?>(StringComparer.OrdinalIgnoreCase);
            foreach (DataColumn col in table.Columns)
                dict[col.ColumnName] = row[col] == DBNull.Value ? null : row[col];

            var sourceRowId = ResolveSourceRowId(dict);
            var mapped = MdpDbSyncRules.Apply(dict, mappings);
            if (mapped.Errors.Count > 0)
            {
                rowErrors++;
                if (errorSamples.Count < 20)
                    errorSamples.Add(MdpDbSyncRules.Redact(string.Join("；", mapped.Errors)));
                continue;
            }
            var rawJson = JsonSerializer.Serialize(mapped.Values, RawDataJsonOptions);

            // 游标推进必须留在逐行循环里：它依赖遍历顺序，挪到批量写入之后会算错。
            if (useKeyset)
            {
                lastKeysetCursor = FormatCursorValue(dict, ctx.CursorColumn!);
                lastKeysetTie = FormatCursorValue(dict, ctx.TieBreakerColumn!);
                if (!string.IsNullOrEmpty(lastKeysetTie))
                    maxCursor = EncodeKeysetCursor(lastKeysetCursor, lastKeysetTie);
            }
            else if (!string.IsNullOrWhiteSpace(entity.IncrColumn) && dict.TryGetValue(entity.IncrColumn, out var incrVal) && incrVal != null)
            {
                var cursor = incrVal is DateTime dt ? dt.ToString("yyyy-MM-dd HH:mm:ss.fff") : incrVal.ToString();
                if (!string.IsNullOrEmpty(cursor) && (maxCursor == null || string.CompareOrdinal(cursor, maxCursor) > 0))
                    maxCursor = cursor;
            }

            pending.Add((mapped.Values, rawJson, sourceRowId));
        }

        if (pending.Count > 0)
        {
            var results = await _writer.UpsertBatchAsync(
                source,
                entity,
                ResolveTransformSourceTable(entity),
                pending,
                ctx);
            written += results.Sum();
        }

        if (useKeyset && !string.IsNullOrEmpty(lastKeysetTie))
        {
            ctx.CursorValue = lastKeysetCursor;
            ctx.TieBreakerValue = lastKeysetTie;
        }

        if (!ctx.DeferCursorPersist
            && !string.IsNullOrEmpty(maxCursor)
            && maxCursor != entity.LastCursor)
        {
            await _db.Updateable<MdpEntity>()
                .SetColumns(x => new MdpEntity
                {
                    LastCursor = maxCursor,
                    LastSyncTo = now,
                    UpdateTime = now
                })
                .Where(x => x.Id == entity.Id)
                .ExecuteCommandAsync(cancellationToken);
        }

        // 零行假成功守卫：本页 0 行时，用同谓词、去作用域的探针确认源侧是否本就无数据。
        // 源侧有数据而作用域后 0 行 ⇒ 全量被租户/工厂作用域过滤掉，不得记为无保留的 SUCCESS。
        var scopeMismatch = IsScopeMismatch(
            table.Rows.Count, ctx, tenantCol, factoryCol,
            sourceHasRowsOutsideScope: ShouldProbeScope(table.Rows.Count, ctx, tenantCol, factoryCol)
                && await SourceHasRowsOutsideScopeAsync(scope, entity, ctx, isSqlServer, useKeyset, parameters, tenantCol, cancellationToken));

        var diagnostic = scopeMismatch
            ? $"{ScopeMismatchCode}: 源表 {entity.SourceTableName} 在 tenantId={ctx.TenantId}/factoryId={ctx.FactoryId} 作用域外仍有数据，"
              + "但作用域过滤后 0 行；请核对该实体的工厂口径（factory_id 是否被写成 tenant_id 或真实工厂号与作用域不一致）。"
            : null;
        if (rowErrors > 0)
        {
            var sampleText = errorSamples.Count == 0 ? "" : " 样本：" + string.Join(" | ", errorSamples);
            var rowText = $"行错误 {rowErrors}{sampleText}";
            diagnostic = string.IsNullOrEmpty(diagnostic) ? rowText : diagnostic + "；" + rowText;
        }

        var status = rowErrors > 0 && written == 0
            ? "FAILED"
            : scopeMismatch || rowErrors > 0
                ? SyncStatusPartial
                : SyncStatusSuccess;
        await WriteSyncLogAsync(source, entity, ctx, startedAt, table.Rows.Count, written, rowErrors, diagnostic, status);

        var windowHint = useKeyset
            ? (ctx.NullTimePhase ? "KEYSET_NULL" : "KEYSET")
            : (ctx.SyncWindowType ?? (ctx.FullRefresh ? "FULL" : "INCR"));
        return new MdpPullResult
        {
            RowsPulled = table.Rows.Count,
            RowsWritten = written,
            RowsError = rowErrors,
            Status = status,
            NewCursor = maxCursor,
            Message = (scopeMismatch ? $"{ScopeMismatchCode} window=" : "OK window=") + windowHint
                + (ctx.WindowFrom.HasValue ? $" from={ctx.WindowFrom:yyyy-MM-dd HH:mm:ss}" : "")
                + (ctx.BootstrapFrom.HasValue ? $" bootstrapFrom={ctx.BootstrapFrom:yyyy-MM-dd HH:mm:ss}" : "")
        };
    }

    internal const string SyncStatusSuccess = "SUCCESS";

    /// <summary>mdp_sync_log.status 既有 enum 成员（'RUNNING','SUCCESS','PARTIAL','FAILED'），无需改表。</summary>
    internal const string SyncStatusPartial = "PARTIAL";

    internal const string ScopeMismatchCode = "SCOPE_MISMATCH";

    /// <summary>
    /// 是否需要跑作用域探针。仅在 ①本页 0 行 ②确实发射了作用域谓词 ③非 OFFSET 续页 时才探。
    /// 第 ③ 条：ctx.Offset&gt;0 说明本轮前一页已经拉满了作用域内的数据，作用域显然没有过滤掉全部，
    /// 此时的 0 行是正常翻页到底，探针（不带 OFFSET）会误报。
    /// </summary>
    internal static bool ShouldProbeScope(int rowsRead, MdpPullContext ctx, string? tenantCol, string? factoryCol)
        => rowsRead == 0 && ctx.Offset <= 0 && NeedsTenantParameter(tenantCol, factoryCol);

    /// <summary>
    /// 最终判定：本页 0 行且探针证明"去掉作用域后源侧仍有数据" ⇒ 全量被作用域过滤，记 PARTIAL。
    /// 源侧本就无数据（合法空源 / 增量无新数据）时 <paramref name="sourceHasRowsOutsideScope"/> 为 false ⇒ 正常 SUCCESS。
    /// </summary>
    internal static bool IsScopeMismatch(
        int rowsRead, MdpPullContext ctx, string? tenantCol, string? factoryCol, bool sourceHasRowsOutsideScope)
        => ShouldProbeScope(rowsRead, ctx, tenantCol, factoryCol) && sourceHasRowsOutsideScope;

    private static async Task<bool> SourceHasRowsOutsideScopeAsync(
        ISqlSugarClient scope, MdpEntity entity, MdpPullContext ctx,
        bool isSqlServer, bool useKeyset, List<SugarParameter> parameters,
        string? tenantCol, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        try
        {
            var keepTenant = !string.IsNullOrWhiteSpace(tenantCol);
            var probeSql = BuildScopeProbeSql(entity, isSqlServer, ctx, useKeyset, tenantCol);
            var probe = await scope.Ado.GetDataTableAsync(probeSql, StripScopeParameters(parameters, keepTenant));
            return probe.Rows.Count > 0;
        }
        catch
        {
            // 探针纯诊断用途，任何失败都不得影响主流程，按"无异常"处理。
            return false;
        }
    }

    /// <summary>
    /// 探针复用正式查询的参数，剔除探针 SQL 不再引用的作用域参数。
    /// <para><paramref name="keepTenantParameter"/> 必须与
    /// <see cref="BuildScopeProbeSql"/> 是否发射了租户谓词保持一致：
    /// 探针带租户谓词时要留下 <c>@scopeTenantId</c>，否则参数会缺失；
    /// 不带时要剔除，否则会多绑一个 SQL 里不存在的参数。</para>
    /// <para><c>@scopeFactoryId</c> 恒剔除 —— 探针的全部意义就是"去掉工厂口径再看一眼"。</para>
    /// </summary>
    internal static List<SugarParameter> StripScopeParameters(
        List<SugarParameter> parameters, bool keepTenantParameter = false)
        => parameters
            .Where(p => !IsStrippedScopeParameter(p.ParameterName, keepTenantParameter))
            .ToList();

    private static bool IsStrippedScopeParameter(string? name, bool keepTenantParameter)
    {
        var n = (name ?? "").TrimStart('@', ':', '?');
        if (string.Equals(n, "scopeFactoryId", StringComparison.OrdinalIgnoreCase)) return true;
        return !keepTenantParameter
            && string.Equals(n, "scopeTenantId", StringComparison.OrdinalIgnoreCase);
    }

    private static string RequireSourceTable(MdpEntity entity)
    {
        var table = entity.SourceTableName!.Trim();
        // 仅允许简单标识符/schema.table，防注入
        if (!System.Text.RegularExpressions.Regex.IsMatch(table, @"^[A-Za-z0-9_\.\[\]]+$"))
            throw new InvalidOperationException($"非法 source_table_name：{table}");
        return table;
    }

    internal static string ResolveTransformSourceTable(MdpEntity entity)
    {
        var logical = string.IsNullOrWhiteSpace(entity.TransformSourceTable)
            ? entity.SourceTableName
            : entity.TransformSourceTable;
        var value = (logical ?? "").Trim();
        if (!System.Text.RegularExpressions.Regex.IsMatch(value, @"^[A-Za-z0-9_\.\[\]]+$"))
            throw new InvalidOperationException($"非法 transform_source_table：{logical}");
        return value.Contains('.') ? value.Split('.').Last().Trim('[', ']') : value.Trim('[', ']');
    }

    /// <summary>增量/窗口谓词（不含租户/工厂作用域），供正式查询与作用域探针复用。</summary>
    private static List<string> BuildIncrementalPredicates(MdpEntity entity, bool isSqlServer, MdpPullContext ctx, out string orderBy)
    {
        var predicates = new List<string>();
        if (!string.IsNullOrWhiteSpace(entity.IncrColumn))
        {
            var incr = entity.IncrColumn.Trim();
            if (!System.Text.RegularExpressions.Regex.IsMatch(incr, @"^[A-Za-z0-9_]+$"))
                throw new InvalidOperationException($"非法 incr_column：{incr}");
            var incrExpr = isSqlServer ? incr : $"`{incr}`";
            orderBy = incrExpr;

            var isRolling = string.Equals(ctx.SyncWindowType, "ROLLING", StringComparison.OrdinalIgnoreCase);
            if (!ctx.FullRefresh && isRolling && ctx.WindowFrom.HasValue)
                predicates.Add($"{incrExpr} >= @windowFrom");

            var useCursor = !ctx.FullRefresh
                && !string.IsNullOrWhiteSpace(entity.LastCursor)
                && (!isRolling
                    || (ctx.WindowFrom.HasValue
                        && DateTime.TryParse(entity.LastCursor, out var cursorDt)
                        && cursorDt > ctx.WindowFrom.Value));
            if (useCursor)
                predicates.Add($"{incrExpr} > @cursor");
        }
        else
        {
            orderBy = StableOrder(entity, isSqlServer);
        }

        return predicates;
    }

    internal static string BuildSelectSql(MdpEntity entity, bool isSqlServer, int batchSize, MdpPullContext ctx, string? tenantCol = null, string? factoryCol = null)
    {
        var table = RequireSourceTable(entity);
        var predicates = BuildIncrementalPredicates(entity, isSqlServer, ctx, out var orderBy);

        AppendScopePredicates(predicates, isSqlServer, tenantCol, factoryCol);

        var where = predicates.Count > 0 ? " WHERE " + string.Join(" AND ", predicates) : "";
        var offset = ctx.Offset > 0 ? ctx.Offset : 0;
        if (ctx.FullRefresh && !string.IsNullOrWhiteSpace(entity.TieBreakerColumn))
            orderBy = QuoteIdent(RequireIdent(entity.TieBreakerColumn, "TieBreakerColumn"), isSqlServer);
        if (isSqlServer)
        {
            if (string.IsNullOrWhiteSpace(orderBy))
                orderBy = StableOrder(entity, true);
            return $"SELECT * FROM {table}{where} ORDER BY {orderBy} OFFSET {offset} ROWS FETCH NEXT {batchSize} ROWS ONLY";
        }
        return offset > 0
            ? $"SELECT * FROM {table}{where} ORDER BY {orderBy} LIMIT {batchSize} OFFSET {offset}"
            : $"SELECT * FROM {table}{where} ORDER BY {orderBy} LIMIT {batchSize}";
    }

    /// <summary>keyset 谓词（不含租户/工厂作用域），供正式查询与作用域探针复用。</summary>
    internal static List<string> BuildKeysetPredicates(bool isSqlServer, MdpPullContext ctx, out string orderBy)
    {
        var cursorCol = RequireIdent(ctx.CursorColumn, "CursorColumn");
        var tieCol = RequireIdent(ctx.TieBreakerColumn, "TieBreakerColumn");
        var cursorExpr = isSqlServer ? cursorCol : $"`{cursorCol}`";
        var tieExpr = isSqlServer ? tieCol : $"`{tieCol}`";

        var predicates = new List<string>();
        if (ctx.NullTimePhase)
        {
            predicates.Add($"{cursorExpr} IS NULL");
            if (!string.IsNullOrWhiteSpace(ctx.TieBreakerValue))
                predicates.Add($"{tieExpr} > @tieBreaker");
        }
        else
        {
            predicates.Add($"{cursorExpr} IS NOT NULL");
            if (ctx.BootstrapFrom.HasValue)
                predicates.Add($"{cursorExpr} >= @bootstrapFrom");

            // keyset lower bound；旧单值水位首次升级时按主游标包含重读
            predicates.Add(ctx.LegacyCursorInclusive
                ? $"""
                (
                  @cursorTime IS NULL
                  OR {cursorExpr} >= @cursorTime
                )
                """
                : $"""
                (
                  @cursorTime IS NULL
                  OR {cursorExpr} > @cursorTime
                  OR ({cursorExpr} = @cursorTime AND {tieExpr} > @tieBreaker)
                )
                """);

            if (!string.IsNullOrWhiteSpace(ctx.UpperCursorValue))
            {
                predicates.Add($"""
                    (
                      {cursorExpr} < @upperTime
                      OR ({cursorExpr} = @upperTime AND {tieExpr} <= @upperTie)
                    )
                    """);
            }
        }

        orderBy = ctx.NullTimePhase ? tieExpr : $"{cursorExpr}, {tieExpr}";
        return predicates;
    }

    private static string BuildKeysetSelectSql(MdpEntity entity, bool isSqlServer, int batchSize, MdpPullContext ctx, string? tenantCol = null, string? factoryCol = null)
    {
        var table = RequireSourceTable(entity);
        var predicates = BuildKeysetPredicates(isSqlServer, ctx, out var orderBy);

        AppendScopePredicates(predicates, isSqlServer, tenantCol, factoryCol);
        var where = " WHERE " + string.Join(" AND ", predicates);
        if (isSqlServer)
            return $"SELECT * FROM {table}{where} ORDER BY {orderBy} OFFSET 0 ROWS FETCH NEXT {batchSize} ROWS ONLY";
        return $"SELECT * FROM {table}{where} ORDER BY {orderBy} LIMIT {batchSize}";
    }

    /// <summary>
    /// 作用域探针：与正式查询同一组增量/窗口/keyset 谓词，**保留租户谓词、只去掉工厂谓词**。
    /// 正式查询 0 行而探针有行 ⇒ 差异只可能来自工厂口径，可据此判定"本租户的数据被工厂过滤掉了"。
    /// 只做存在性判断（TOP 1 / LIMIT 1），不取数据。
    ///
    /// <para><b>租户谓词必须保留</b>。此前两个作用域谓词一起剥掉，探针退化成
    /// <c>SELECT 1 FROM 源表 LIMIT 1</c> —— 在多租户共享同一张源表的模型下，
    /// 它对**任何**租户都会命中别的租户的行，于是每一个合法空租户都被误报成 SCOPE_MISMATCH。
    /// 实测该误报在共享库上累计 291 条，且全部落在源表本就 0 行的租户上，
    /// 而真正被工厂谓词滤空的那两个租户反而一条都没报出来 —— 信号方向是反的。</para>
    ///
    /// <para>探针要回答的是「同一租户内，是否因工厂口径而取空」，
    /// 不是「整张表有没有数据」。丢掉租户维度，这个问题就问不出来了。</para>
    /// </summary>
    internal static string BuildScopeProbeSql(
        MdpEntity entity, bool isSqlServer, MdpPullContext ctx, bool useKeyset, string? tenantCol = null)
    {
        var table = RequireSourceTable(entity);
        var predicates = useKeyset
            ? BuildKeysetPredicates(isSqlServer, ctx, out _)
            : BuildIncrementalPredicates(entity, isSqlServer, ctx, out _);

        if (!string.IsNullOrWhiteSpace(tenantCol))
            predicates.Add($"{QuoteIdent(tenantCol, isSqlServer)} = @scopeTenantId");

        var where = predicates.Count > 0 ? " WHERE " + string.Join(" AND ", predicates) : "";
        return isSqlServer
            ? $"SELECT TOP 1 1 AS probe FROM {table}{where}"
            : $"SELECT 1 AS probe FROM {table}{where} LIMIT 1";
    }

    private static void AddKeysetParameters(List<SugarParameter> parameters, MdpPullContext ctx)
    {
        if (ctx.BootstrapFrom.HasValue)
            parameters.Add(new SugarParameter("@bootstrapFrom", ctx.BootstrapFrom.Value));

        if (ctx.NullTimePhase)
        {
            parameters.Add(new SugarParameter("@tieBreaker",
                string.IsNullOrWhiteSpace(ctx.TieBreakerValue) ? "0" : ctx.TieBreakerValue));
            return;
        }

        object cursorTime = string.IsNullOrWhiteSpace(ctx.CursorValue)
            ? DBNull.Value
            : (object)ctx.CursorValue!;
        parameters.Add(new SugarParameter("@cursorTime", cursorTime));
        parameters.Add(new SugarParameter("@tieBreaker",
            string.IsNullOrWhiteSpace(ctx.TieBreakerValue) ? "0" : ctx.TieBreakerValue));

        if (!string.IsNullOrWhiteSpace(ctx.UpperCursorValue))
        {
            parameters.Add(new SugarParameter("@upperTime", ctx.UpperCursorValue));
            parameters.Add(new SugarParameter("@upperTie",
                string.IsNullOrWhiteSpace(ctx.UpperTieBreakerValue) ? "0" : ctx.UpperTieBreakerValue));
        }
    }

    /// <summary>
    /// 租户/工厂作用域谓词。工厂维度走 <see cref="MdpFactoryScope"/> 的归一语义
    /// （factory_id 为 NULL / &lt;=0 / 等于 tenant_id 一律视作"无真实工厂"→默认工厂 1），
    /// 与 <see cref="MdpStagingWriter"/> 写入侧完全一致；真实工厂号仍逐值隔离。
    /// </summary>
    internal static void AppendScopePredicates(List<string> predicates, bool isSqlServer, string? tenantCol, string? factoryCol)
    {
        if (!string.IsNullOrWhiteSpace(tenantCol))
            predicates.Add($"{QuoteIdent(tenantCol, isSqlServer)} = @scopeTenantId");
        if (!string.IsNullOrWhiteSpace(factoryCol))
        {
            predicates.Add(MdpFactoryScope.SqlPredicate(
                QuoteIdent(factoryCol, isSqlServer), "@scopeTenantId", "@scopeFactoryId"));
        }
    }

    /// <summary>
    /// 工厂谓词引用 @scopeTenantId，故只要发射了任一作用域谓词就必须绑定该参数
    /// （源表只有 factory_id 而无 tenant_id 时，旧逻辑会漏绑导致 SQL 参数缺失）。
    /// </summary>
    internal static bool NeedsTenantParameter(string? tenantCol, string? factoryCol)
        => !string.IsNullOrWhiteSpace(tenantCol) || !string.IsNullOrWhiteSpace(factoryCol);

    private static string StableOrder(MdpEntity entity, bool isSqlServer)
    {
        if (!string.IsNullOrWhiteSpace(entity.TieBreakerColumn))
            return QuoteIdent(RequireIdent(entity.TieBreakerColumn, "TieBreakerColumn"), isSqlServer);
        return isSqlServer ? "(SELECT NULL)" : "1";
    }

    private static string QuoteIdent(string name, bool isSqlServer) => isSqlServer ? name : $"`{name}`";

    private static string? FindColumn(IReadOnlyCollection<string> columns, params string[] names)
    {
        foreach (var n in names)
        {
            var hit = columns.FirstOrDefault(c => string.Equals(c, n, StringComparison.OrdinalIgnoreCase));
            if (hit != null) return hit;
        }
        return null;
    }

    private static async Task<IReadOnlyCollection<string>> TryGetSourceColumnsAsync(
        ISqlSugarClient scope, string table, bool isSqlServer, CancellationToken cancellationToken)
    {
        try
        {
            var simple = table.Contains('.') ? table.Split('.').Last().Trim('[', ']') : table.Trim('[', ']');
            if (!System.Text.RegularExpressions.Regex.IsMatch(simple, @"^[A-Za-z0-9_]+$"))
                return Array.Empty<string>();
            if (isSqlServer)
            {
                var rows = await scope.Ado.SqlQueryAsync<string>(
                    "SELECT name FROM sys.columns WHERE object_id = OBJECT_ID(@t)",
                    new SugarParameter("@t", simple));
                return rows ?? new List<string>();
            }
            var mysql = await scope.Ado.SqlQueryAsync<string>(
                """
                SELECT COLUMN_NAME FROM information_schema.COLUMNS
                WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@t
                """,
                new SugarParameter("@t", simple));
            return mysql ?? new List<string>();
        }
        catch
        {
            return Array.Empty<string>();
        }
    }

    private static string RequireIdent(string? name, string label)
    {
        var v = (name ?? "").Trim();
        if (!System.Text.RegularExpressions.Regex.IsMatch(v, @"^[A-Za-z0-9_]+$"))
            throw new InvalidOperationException($"非法 {label}：{name}");
        return v;
    }

    private static string? FormatCursorValue(Dictionary<string, object?> dict, string column)
    {
        if (!dict.TryGetValue(column, out var raw) || raw == null || raw is DBNull)
            return null;
        return raw switch
        {
            DateTime dt => dt.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture),
            DateTimeOffset dto => dto.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture),
            _ => raw.ToString()
        };
    }

    internal static string EncodeKeysetCursor(string? cursor, string tieBreaker)
        => System.Text.Json.JsonSerializer.Serialize(new KeysetCursorDto
        {
            Cursor = cursor,
            TieBreaker = tieBreaker
        });

    internal static bool TryDecodeKeysetCursor(string? raw, out string? cursor, out string? tieBreaker)
    {
        cursor = null;
        tieBreaker = null;
        if (string.IsNullOrWhiteSpace(raw))
            return false;
        try
        {
            var dto = System.Text.Json.JsonSerializer.Deserialize<KeysetCursorDto>(raw);
            if (dto == null || string.IsNullOrWhiteSpace(dto.TieBreaker))
                return false;
            cursor = dto.Cursor;
            tieBreaker = dto.TieBreaker;
            return true;
        }
        catch
        {
            return false;
        }
    }

    private sealed class KeysetCursorDto
    {
        public string? Cursor { get; set; }
        public string? TieBreaker { get; set; }
    }

    private static string ResolveSourceRowId(Dictionary<string, object?> dict)
    {
        // InvTransHist 等表常有空字符串 ID，若优先取 ID 会导致 source_row_id="" 唯一键互相覆盖。
        // 跳过 null/空白后，再落到 RecID 等真实键。
        foreach (var key in new[]
                 {
                     "RecID", "RecId", "recid", "Recid",
                     "id", "Id", "ID",
                     "noid", "billno", "BillNo", "djbh", "Djbh"
                 })
        {
            if (!dict.TryGetValue(key, out var v) || v is null || v is DBNull)
                continue;
            var s = v.ToString();
            if (!string.IsNullOrWhiteSpace(s))
                return s;
        }
        return Guid.NewGuid().ToString("N");
    }

    private async Task WriteSyncLogAsync(MdpSource source, MdpEntity entity, MdpPullContext ctx, DateTime startedAt, int pulled, int written, int rowsError, string? error, string? status = null)
    {
        var endedAt = DateTime.Now;
        var syncType = ctx.FullRefresh || string.Equals(ctx.SyncWindowType, "FULL", StringComparison.OrdinalIgnoreCase)
            ? "FULL"
            : "INCR";
        var trigger = string.Equals(ctx.TriggerType, "MANUAL", StringComparison.OrdinalIgnoreCase) ? "MANUAL" : "AUTO";

        try
        {
            await _db.Ado.ExecuteCommandAsync(@"
INSERT INTO mdp_sync_log
  (tenant_id, entity_id, source_code, entity_name, sync_batch_id, sync_type, trigger_type,
   sync_start, sync_end, duration_ms, rows_read, rows_written, rows_error, status, error_message)
VALUES
  (@tenantId, @entityId, @sourceCode, @entityName, @batchId, @syncType, @triggerType,
   @syncStart, @syncEnd, @durationMs, @rowsRead, @rowsWritten, @rowsError, @status, @error)",
                new SugarParameter("@tenantId", ctx.TenantId),
                new SugarParameter("@entityId", entity.Id),
                new SugarParameter("@sourceCode", source.SourceCode),
                new SugarParameter("@entityName", string.IsNullOrWhiteSpace(entity.EntityName) ? entity.EntityCode : entity.EntityName),
                new SugarParameter("@batchId", ctx.BatchId),
                new SugarParameter("@syncType", syncType),
                new SugarParameter("@triggerType", trigger),
                new SugarParameter("@syncStart", startedAt),
                new SugarParameter("@syncEnd", endedAt),
                new SugarParameter("@durationMs", (int)(endedAt - startedAt).TotalMilliseconds),
                new SugarParameter("@rowsRead", pulled),
                new SugarParameter("@rowsWritten", written),
                new SugarParameter("@rowsError", rowsError),
                new SugarParameter("@status", ResolveSyncStatus(status, error)),
                new SugarParameter("@error", error));
        }
        catch (Exception ex)
        {
            throw new InvalidOperationException(
                $"实体 {entity.EntityCode} 数据已处理，但同步日志写入失败：{ex.Message}", ex);
        }
    }

    /// <summary>
    /// 显式 status 优先（用于 PARTIAL 等非成功但非失败的场景）；未指定时沿用"有 error 即 FAILED"。
    /// 注意：SCOPE_MISMATCH 走 status=PARTIAL + error_message 诊断，而不是 FAILED——
    /// 它是"读到 0 行且可疑"的告警，不应把正常调度打成失败。
    /// </summary>
    internal static string ResolveSyncStatus(string? status, string? error)
    {
        if (!string.IsNullOrWhiteSpace(status)) return status!;
        return string.IsNullOrEmpty(error) ? SyncStatusSuccess : "FAILED";
    }

    private sealed class MysqlLiteralDateTimeConverter : JsonConverter<DateTime>
    {
        public override DateTime Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
            => reader.GetDateTime();

        public override void Write(Utf8JsonWriter writer, DateTime value, JsonSerializerOptions options)
            => writer.WriteStringValue(value.ToString("yyyy-MM-dd HH:mm:ss.ffffff", CultureInfo.InvariantCulture));
    }

    private sealed class MysqlLiteralDateTimeOffsetConverter : JsonConverter<DateTimeOffset>
    {
        public override DateTimeOffset Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
            => reader.GetDateTimeOffset();

        public override void Write(Utf8JsonWriter writer, DateTimeOffset value, JsonSerializerOptions options)
            => writer.WriteStringValue(value.ToString("yyyy-MM-dd HH:mm:ss.ffffff", CultureInfo.InvariantCulture));
    }
}