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

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

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

/// <summary>统一入站抽数执行器（方式甲 DB / 方式乙 API）。</summary>
public interface IMdpSourcePullExecutor
{
    /// <summary>DB_SYNC / API_PULL</summary>
    string SupportedType { get; }

    Task<MdpPullResult> PullAsync(MdpSource source, MdpEntity entity, MdpPullContext ctx, CancellationToken cancellationToken = default);
}

public sealed class MdpPullContext
{
    public long TenantId { get; set; } = 1300000000001L;
    public string BatchId { get; set; } = "";
    /// <summary>强制全量（忽略水位与滚动下界）。</summary>
    public bool FullRefresh { get; set; }
    /// <summary>关联 mdp_sync_task.task_code，用于读取调度上的同步窗口。</summary>
    public string? TaskCode { get; set; }
    /// <summary>FULL / INCR / ROLLING（可由调度表覆盖）。</summary>
    public string? SyncWindowType { get; set; }
    /// <summary>如 7d / 24h / 2026-01-01。</summary>
    public string? SyncWindowValue { get; set; }
    /// <summary>ROLLING 解析后的下界（含）。</summary>
    public DateTime? WindowFrom { get; set; }
    /// <summary>全量分页偏移（仅 FullRefresh / 无 incr 时由 PullAll 推进）。</summary>
    public int Offset { get; set; }
}

public sealed class MdpPullResult
{
    public int RowsPulled { get; set; }
    public int RowsWritten { get; set; }
    public string? NewCursor { get; set; }
    public string? Message { get; set; }
}

/// <summary>按实体/源类型选择 DB 或 API 执行器。</summary>
public sealed class MdpSourcePullDispatcher : ITransient
{
    private readonly MdpDbPullExecutor _dbExecutor;
    private readonly MdpApiPullExecutor _apiExecutor;
    private readonly ISqlSugarClient _db;

    public MdpSourcePullDispatcher(
        MdpDbPullExecutor dbExecutor,
        MdpApiPullExecutor apiExecutor,
        ISqlSugarClient db)
    {
        _dbExecutor = dbExecutor;
        _apiExecutor = apiExecutor;
        _db = db;
    }

    public async Task<MdpPullResult> PullByEntityCodeAsync(string entityCode, MdpPullContext ctx, CancellationToken cancellationToken = default)
    {
        var entity = await _db.Queryable<MdpEntity>()
            .Where(x => x.EntityCode == entityCode && x.Status == 1)
            .FirstAsync(cancellationToken)
            ?? throw new InvalidOperationException($"mdp_entity 未找到启用实体：{entityCode}");

        var source = await _db.Queryable<MdpSource>()
            .Where(x => x.Id == entity.SourceId && x.Status == 1)
            .FirstAsync(cancellationToken)
            ?? throw new InvalidOperationException($"mdp_source id={entity.SourceId} 未找到或未启用");

        if (string.IsNullOrWhiteSpace(ctx.BatchId))
            ctx.BatchId = $"MDP_PULL_{DateTime.Now:yyyyMMddHHmmss}";

        await ApplyScheduleWindowAsync(ctx, cancellationToken);
        MdpSyncWindowResolver.Apply(ctx);

        IMdpSourcePullExecutor executor =
            !string.IsNullOrWhiteSpace(entity.SourceApiPath)
            || string.Equals(source.SourceType, "API", StringComparison.OrdinalIgnoreCase)
                ? _apiExecutor
                : _dbExecutor;

        return await executor.PullAsync(source, entity, ctx, cancellationToken);
    }

    /// <summary>
    /// 分页抽尽：FullRefresh 用 OFFSET；增量依赖实体 LastCursor 在页间推进。
    /// </summary>
    public async Task<MdpPullResult> PullAllByEntityCodeAsync(
        string entityCode,
        MdpPullContext ctx,
        CancellationToken cancellationToken = default,
        int maxPages = 500)
    {
        var totalPulled = 0;
        var totalWritten = 0;
        string? lastCursor = null;
        string? lastMessage = null;
        ctx.Offset = 0;

        for (var page = 0; page < maxPages; page++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var pageResult = await PullByEntityCodeAsync(entityCode, ctx, cancellationToken);
            totalPulled += pageResult.RowsPulled;
            totalWritten += pageResult.RowsWritten;
            lastCursor = pageResult.NewCursor ?? lastCursor;
            lastMessage = pageResult.Message;

            var entity = await _db.Queryable<MdpEntity>()
                .Where(x => x.EntityCode == entityCode && x.Status == 1)
                .FirstAsync(cancellationToken);
            var batchSize = entity?.BatchSize > 0 ? entity.BatchSize : 1000;

            if (pageResult.RowsPulled <= 0 || pageResult.RowsPulled < batchSize)
                break;

            if (ctx.FullRefresh)
                ctx.Offset += pageResult.RowsPulled;
            // 增量：实体 LastCursor 已在执行器内更新，下一页自动收窄
        }

        return new MdpPullResult
        {
            RowsPulled = totalPulled,
            RowsWritten = totalWritten,
            NewCursor = lastCursor,
            Message = $"OK pages pulled={totalPulled} written={totalWritten}; {lastMessage}"
        };
    }

    /// <summary>若上下文带 TaskCode 且未显式指定窗口，则从 mdp_sync_task_schedule 读取。</summary>
    private async Task ApplyScheduleWindowAsync(MdpPullContext ctx, CancellationToken cancellationToken)
    {
        if (ctx.FullRefresh) return;
        if (string.IsNullOrWhiteSpace(ctx.TaskCode)) return;
        if (!string.IsNullOrWhiteSpace(ctx.SyncWindowType)) return;

        var schedules = await _db.Queryable<MdpSyncTaskSchedule>()
            .Where(x => x.TaskCode == ctx.TaskCode && (x.TenantId == ctx.TenantId || x.TenantId == 0))
            .OrderByDescending(x => x.TenantId)
            .Take(1)
            .ToListAsync(cancellationToken);
        var schedule = schedules.FirstOrDefault();
        if (schedule == null) return;
        ctx.SyncWindowType = schedule.SyncWindowType;
        ctx.SyncWindowValue = schedule.SyncWindowValue;
    }
}