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

using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Admin.NET.Plugin.AiDOP.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; }
    public string BatchId { get; set; } = "";
    /// <summary>强制全量（忽略水位与滚动下界）。</summary>
    public bool FullRefresh { get; set; }
    /// <summary>工厂作用域；&gt;0 时入站写入与源过滤按工厂对齐。</summary>
    public long FactoryId { get; set; }
    /// <summary>为 true 时禁止把无租户源行归给当前上下文租户；源租户不匹配则跳过。</summary>
    public bool RequireMatchingSourceTenant { 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; }

    /// <summary>启用复合 keyset 游标（库存冷链等场景）；未设置时保持既有 OFFSET/单列游标行为。</summary>
    public bool UseKeysetCursor { get; set; }
    /// <summary>时间/主游标列，如 UpdateTime / CreateTime。</summary>
    public string? CursorColumn { get; set; }
    /// <summary>唯一 tie-breaker 列，如 RecID。</summary>
    public string? TieBreakerColumn { get; set; }
    /// <summary>本页起始游标时间（ISO/字面量）。</summary>
    public string? CursorValue { get; set; }
    /// <summary>本页起始 tie-breaker。</summary>
    public string? TieBreakerValue { get; set; }
    /// <summary>冻结上界时间。</summary>
    public string? UpperCursorValue { get; set; }
    /// <summary>冻结上界 RecID。</summary>
    public string? UpperTieBreakerValue { get; set; }
    /// <summary>bootstrap 下界（含），如近 12 个月。</summary>
    public DateTime? BootstrapFrom { get; set; }
    /// <summary>为 true 时本页不写实体 LastCursor，由编排在整轮成功后持久化。</summary>
    public bool DeferCursorPersist { get; set; }
    /// <summary>LocationDetail 等 NULL 时间段：仅拉 CursorColumn IS NULL 且按 TieBreaker 推进。</summary>
    public bool NullTimePhase { get; set; }
    /// <summary>为 true 时不从实体 LastCursor 回填起始 keyset（bootstrap/reconcile 首刷用）。</summary>
    public bool SkipPersistedKeysetCursor { get; set; }
    /// <summary>旧单值水位首次升级：同一时间戳整段重读，由贴源幂等吸收重复行。</summary>
    public bool LegacyCursorInclusive { get; set; }
    /// <summary>方式甲配置入口才强制稳定排序列。既有模块调用缺列时保持原单列水位。</summary>
    public bool RequireConfiguredDbSync { get; set; }
    /// <summary>MANUAL / AUTO。</summary>
    public string TriggerType { get; set; } = "AUTO";

    /// <summary>
    /// 贴源归属租户与业务租户分离（165 库存一次贴源、再按库位投影）。
    /// 为 true 时不要求该租户已登记本实体的来源系统。
    /// </summary>
    public bool StagingOwnerPull { get; set; }
}

public sealed class MdpPullResult
{
    public int RowsPulled { get; set; }
    public int RowsWritten { get; set; }
    public int RowsError { get; set; }
    public string? NewCursor { get; set; }
    public string? Status { 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 query = _db.Queryable<MdpEntity>()
            .Where(x => x.EntityCode == entityCode && x.Status == 1)
            .WhereIF(ctx.TenantId > 0, x => x.TenantId == ctx.TenantId || x.TenantId == 0)
            .WhereIF(ctx.TenantId <= 0, x => x.TenantId == 0)
            .OrderByDescending(x => x.TenantId);
        var entity = await query
            .FirstAsync(cancellationToken)
            ?? throw new InvalidOperationException($"mdp_entity 未找到启用实体：{entityCode}");
        return await PullLoadedAsync(entity, ctx, cancellationToken);
    }

    public async Task<MdpPullResult> PullByEntityIdAsync(long entityId, MdpPullContext ctx, CancellationToken cancellationToken = default)
    {
        var entity = await _db.Queryable<MdpEntity>()
            .Where(x => x.Id == entityId && x.Status == 1)
            .FirstAsync(cancellationToken)
            ?? throw new InvalidOperationException($"mdp_entity 未找到启用实体：{entityId}");
        if (ctx.TenantId > 0 && entity.TenantId != 0 && entity.TenantId != ctx.TenantId)
            throw new InvalidOperationException($"不能运行其他租户的实体 {entity.EntityCode}");
        return await PullLoadedAsync(entity, ctx, cancellationToken);
    }

    public async Task<MdpPullResult> PullAllByEntityIdAsync(
        long entityId,
        MdpPullContext ctx,
        CancellationToken cancellationToken = default,
        int maxPages = 500)
    {
        var entity = await _db.Queryable<MdpEntity>().Where(x => x.Id == entityId).FirstAsync(cancellationToken)
            ?? throw new InvalidOperationException($"mdp_entity 未找到实体：{entityId}");
        if (ctx.TenantId > 0 && entity.TenantId != 0 && entity.TenantId != ctx.TenantId)
            throw new InvalidOperationException($"不能运行其他租户的实体 {entity.EntityCode}");
        return await PullAllAsync(entity.Id, entity.EntityCode, ctx, cancellationToken, maxPages);
    }

    private async Task<MdpPullResult> PullLoadedAsync(MdpEntity entity, MdpPullContext ctx, CancellationToken cancellationToken)
    {
        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 (entity.TenantId == 0 && ctx.TenantId > 0 && !ctx.StagingOwnerPull)
        {
            var neutral = MdpSourceIdentity.NeutralCode(source.SystemCode, source.SourceCode);
            if (!string.Equals(neutral, MdpSourceIdentity.Native, StringComparison.OrdinalIgnoreCase)
                && !string.Equals(neutral, "AIDOPDEV_MYSQL", StringComparison.OrdinalIgnoreCase))
            {
                var allowed = await _db.Ado.GetIntAsync(
                    """
                    SELECT COUNT(*) FROM mdp_tenant_std_source
                    WHERE tenant_id=@t AND source_system=@s
                    """,
                    new SugarParameter("@t", ctx.TenantId),
                    new SugarParameter("@s", neutral));
                if (allowed == 0)
                    throw new InvalidOperationException(
                        $"租户 {ctx.TenantId} 未登记来源 {neutral}，拒绝拉取全局实体 {entity.EntityCode}");
            }
        }

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

        ApplyDbSyncDefaults(entity, source, ctx);
        await ApplyScheduleWindowAsync(ctx, cancellationToken);
        MdpSyncWindowResolver.Apply(ctx);

        if (string.Equals(source.SourceType, "FILE_EXCEL", StringComparison.OrdinalIgnoreCase))
            throw new InvalidOperationException("文件源不支持 Pull，请使用文件导入接口");

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

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

    /// <summary>
    /// 游标没有前进时必须改走 OFFSET，否则空游标增量会反复读第一页。
    /// 全量本来就用 OFFSET。keyset 自己带翻页谓词，不叠加 OFFSET。
    /// </summary>
    internal static bool ShouldPageByOffset(bool fullRefresh, bool useKeyset, string? cursorAtPageStart, string? newCursor)
    {
        if (useKeyset)
            return false;
        if (fullRefresh)
            return true;
        return string.IsNullOrWhiteSpace(newCursor)
            || string.Equals(newCursor, cursorAtPageStart, StringComparison.Ordinal);
    }

    /// <summary>
    /// 分页抽尽：FullRefresh 用 OFFSET；增量依赖实体 LastCursor 在页间推进；
    /// 游标为空或本页没有把游标推前时退化为 OFFSET，避免空游标只抽到第一页或反复读同一页。
    /// UseKeysetCursor 时用复合 keyset，达 maxPages 且末页仍满批则抛错。
    /// </summary>
    public Task<MdpPullResult> PullAllByEntityCodeAsync(
        string entityCode,
        MdpPullContext ctx,
        CancellationToken cancellationToken = default,
        int maxPages = 500)
        => PullAllAsync(null, entityCode, ctx, cancellationToken, maxPages);

    private async Task<MdpEntity?> FindPagedEntityAsync(
        long? entityId,
        string? entityCode,
        long tenantId,
        CancellationToken cancellationToken)
    {
        if (entityId is > 0)
            return await _db.Queryable<MdpEntity>()
                .Where(x => x.Id == entityId && x.Status == 1)
                .FirstAsync(cancellationToken);

        return await _db.Queryable<MdpEntity>()
            .Where(x => x.EntityCode == entityCode && x.Status == 1)
            .WhereIF(tenantId > 0, x => x.TenantId == tenantId || x.TenantId == 0)
            .WhereIF(tenantId <= 0, x => x.TenantId == 0)
            .OrderByDescending(x => x.TenantId)
            .FirstAsync(cancellationToken);
    }

    private async Task<MdpPullResult> PullAllAsync(
        long? entityId,
        string? entityCode,
        MdpPullContext ctx,
        CancellationToken cancellationToken,
        int maxPages)
    {
        var totalPulled = 0;
        var totalWritten = 0;
        string? lastCursor = null;
        string? lastMessage = null;
        ctx.Offset = 0;

        if (ctx.UseKeysetCursor
            && !ctx.SkipPersistedKeysetCursor
            && string.IsNullOrWhiteSpace(ctx.CursorValue)
            && string.IsNullOrWhiteSpace(ctx.TieBreakerValue)
            && !ctx.NullTimePhase)
        {
            var entity0 = await FindPagedEntityAsync(entityId, entityCode, ctx.TenantId, cancellationToken);
            if (entity0 != null
                && MdpDbPullExecutor.TryDecodeKeysetCursor(entity0.LastCursor, out var c, out var t))
            {
                ctx.CursorValue = c;
                ctx.TieBreakerValue = t;
            }
        }

        for (var page = 0; page < maxPages; page++)
        {
            cancellationToken.ThrowIfCancellationRequested();
            var cursorAtPageStart = lastCursor;
            var pageResult = entityId is > 0
                ? await PullByEntityIdAsync(entityId.Value, ctx, cancellationToken)
                : await PullByEntityCodeAsync(entityCode!, ctx, cancellationToken);
            totalPulled += pageResult.RowsPulled;
            totalWritten += pageResult.RowsWritten;
            lastCursor = pageResult.NewCursor ?? lastCursor;
            lastMessage = pageResult.Message;

            var entity = await FindPagedEntityAsync(entityId, entityCode, ctx.TenantId, cancellationToken);
            var batchSize = entity?.BatchSize > 0 ? entity.BatchSize : 1000;

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

            if (ctx.UseKeysetCursor)
            {
                if (page == maxPages - 1)
                    throw new InvalidOperationException(
                        $"实体 {entity?.EntityCode ?? entityCode} keyset 拉取达到 maxPages={maxPages} 且末页仍满批，禁止静默截断");
                // 下一页游标已由执行器写回 ctx.CursorValue/TieBreakerValue
                continue;
            }

            if (ShouldPageByOffset(ctx.FullRefresh, ctx.UseKeysetCursor, cursorAtPageStart, pageResult.NewCursor))
                ctx.Offset += pageResult.RowsPulled;
            // 游标已前进：实体 LastCursor 已在执行器内更新，下一页用 > cursor，Offset 保持 0
        }

        if (ctx.UseKeysetCursor && ctx.DeferCursorPersist && !string.IsNullOrEmpty(lastCursor))
        {
            var entity = await FindPagedEntityAsync(entityId, entityCode, ctx.TenantId, cancellationToken);
            if (entity != null)
            {
                var now = DateTime.Now;
                await _db.Updateable<MdpEntity>()
                    .SetColumns(x => new MdpEntity
                    {
                        LastCursor = lastCursor,
                        LastSyncTo = now,
                        UpdateTime = now
                    })
                    .Where(x => x.Id == entity.Id)
                    .ExecuteCommandAsync(cancellationToken);
            }
        }

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

    internal static void ApplyDbSyncDefaults(MdpEntity entity, MdpSource source, MdpPullContext ctx)
    {
        var isDb = string.Equals(source.SourceType, "DB", StringComparison.OrdinalIgnoreCase)
            && string.IsNullOrWhiteSpace(entity.SourceApiPath);
        if (!isDb)
            return;

        if (ctx.UseKeysetCursor)
        {
            if (Conflicts(ctx.CursorColumn, entity.IncrColumn) || Conflicts(ctx.TieBreakerColumn, entity.TieBreakerColumn))
                throw new InvalidOperationException(
                    $"实体 {entity.EntityCode} 的增量列或稳定排序列与调用方不一致，已拒绝执行");
            return;
        }

        var mode = ctx.FullRefresh
            ? "FULL"
            : (ctx.SyncWindowType ?? entity.SyncMode ?? "INCR").Trim().ToUpperInvariant();
        if (mode is "INCR" or "ROLLING")
        {
            if (ctx.RequireConfiguredDbSync)
            {
                MdpDbSyncRules.RequireIdentifier(entity.IncrColumn, "incr_column");
                MdpDbSyncRules.RequireIdentifier(entity.TieBreakerColumn, "tie_breaker_column");
            }
            if (string.IsNullOrWhiteSpace(entity.IncrColumn) || string.IsNullOrWhiteSpace(entity.TieBreakerColumn))
                return;
            ctx.UseKeysetCursor = true;
            ctx.CursorColumn = entity.IncrColumn.Trim();
            ctx.TieBreakerColumn = entity.TieBreakerColumn.Trim();
            HydrateCursor(entity, ctx);
        }
        else if (mode == "FULL" && ctx.RequireConfiguredDbSync)
        {
            MdpDbSyncRules.RequireIdentifier(entity.TieBreakerColumn, "tie_breaker_column");
        }
    }

    private static bool Conflicts(string? caller, string? configured) =>
        !string.IsNullOrWhiteSpace(caller)
        && !string.IsNullOrWhiteSpace(configured)
        && !string.Equals(caller.Trim(), configured.Trim(), StringComparison.OrdinalIgnoreCase);

    private static void HydrateCursor(MdpEntity entity, MdpPullContext ctx)
    {
        if (ctx.SkipPersistedKeysetCursor || ctx.NullTimePhase)
            return;
        if (!string.IsNullOrWhiteSpace(ctx.CursorValue) || !string.IsNullOrWhiteSpace(ctx.TieBreakerValue))
        {
            if (!string.IsNullOrWhiteSpace(ctx.TieBreakerValue))
                ctx.LegacyCursorInclusive = false;
            return;
        }
        if (MdpDbPullExecutor.TryDecodeKeysetCursor(entity.LastCursor, out var cursor, out var tie))
        {
            ctx.CursorValue = cursor;
            ctx.TieBreakerValue = tie;
            ctx.LegacyCursorInclusive = false;
            return;
        }
        if (!string.IsNullOrWhiteSpace(entity.LastCursor))
        {
            ctx.CursorValue = entity.LastCursor;
            ctx.LegacyCursorInclusive = true;
        }
    }
}