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;

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

    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.SourceCode, 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 sql = BuildSelectSql(entity, isSqlServer, batchSize, ctx);
        var parameters = new List<SugarParameter>();
        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;
        string? maxCursor = entity.LastCursor;
        var now = DateTime.Now;

        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 rawJson = JsonSerializer.Serialize(dict, RawDataJsonOptions);

            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;
            }

            written += await _writer.UpsertAsync(
                source, entity, entity.SourceTableName!, dict, rawJson, sourceRowId, ctx);
        }

        if (!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);
        }

        await WriteSyncLogAsync(source, entity, ctx, startedAt, table.Rows.Count, written, null);

        var windowHint = ctx.SyncWindowType ?? (ctx.FullRefresh ? "FULL" : "INCR");
        return new MdpPullResult
        {
            RowsPulled = table.Rows.Count,
            RowsWritten = written,
            NewCursor = maxCursor,
            Message = $"OK window={windowHint}" + (ctx.WindowFrom.HasValue ? $" from={ctx.WindowFrom:yyyy-MM-dd HH:mm:ss}" : "")
        };
    }

    private static string BuildSelectSql(MdpEntity entity, bool isSqlServer, int batchSize, MdpPullContext ctx)
    {
        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}");

        var predicates = new List<string>();
        string orderBy;
        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 = isSqlServer ? "(SELECT NULL)" : "1";
        }

        var where = predicates.Count > 0 ? " WHERE " + string.Join(" AND ", predicates) : "";
        var offset = ctx.Offset > 0 ? ctx.Offset : 0;
        if (isSqlServer)
        {
            // SQL Server：OFFSET 需 ORDER BY；无 incr 时用稳定键兜底
            if (string.IsNullOrWhiteSpace(entity.IncrColumn))
                orderBy = "(SELECT NULL)";
            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}";
    }

    private static string ResolveSourceRowId(Dictionary<string, object?> dict)
    {
        foreach (var key in new[]
                 {
                     "id", "Id", "ID",
                     "RecID", "RecId", "recid", "Recid",
                     "noid", "billno", "BillNo", "djbh", "Djbh"
                 })
        {
            if (dict.TryGetValue(key, out var v) && v != null)
                return v.ToString() ?? Guid.NewGuid().ToString("N");
        }
        return Guid.NewGuid().ToString("N");
    }

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

            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, status, error_message)
VALUES
  (@tenantId, @entityId, @sourceCode, @entityName, @batchId, @syncType, 'AUTO',
   @syncStart, @syncEnd, @durationMs, @rowsRead, @rowsWritten, @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("@syncStart", startedAt),
                new SugarParameter("@syncEnd", endedAt),
                new SugarParameter("@durationMs", (int)(endedAt - startedAt).TotalMilliseconds),
                new SugarParameter("@rowsRead", pulled),
                new SugarParameter("@rowsWritten", written),
                new SugarParameter("@status", string.IsNullOrEmpty(error) ? "SUCCESS" : "FAILED"),
                new SugarParameter("@error", error));
        }
        catch
        {
            // 日志表结构可能与种子不一致时不阻断主流程
        }
    }

    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));
    }
}