AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Infrastructure/MdpRefreshDispatcher.cs

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.Infrastructure;

/// <summary>
/// 数据中台全量刷新后台派发器。
/// 将评审 / 采购管线末尾的全量 ETL 从「同步 await」改为「后台异步执行」，
/// 并按 key（如 "S1"、"S4"）做「单飞 + 合并去抖」：
///   · 该 key 空闲 → 立即后台执行一轮；
///   · 正在执行 → 只置 pending，不排队叠加；
///   · 一轮结束若 pending 为真 → 再补跑一轮（吸收运行期间累计的触发）。
/// 说明：进程内单飞；多实例部署时跨实例合并依赖各模块的定时刷新任务兜底。
/// </summary>
public class MdpRefreshDispatcher : ISingleton
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger<MdpRefreshDispatcher> _logger;
    private readonly object _gate = new();
    private readonly Dictionary<string, RefreshState> _states = new();

    public MdpRefreshDispatcher(IServiceScopeFactory scopeFactory, ILogger<MdpRefreshDispatcher> logger)
    {
        _scopeFactory = scopeFactory;
        _logger = logger;
    }

    /// <summary>
    /// 提交一次后台刷新（立即返回，不等待执行完成）。
    /// </summary>
    /// <param name="key">去抖分组键，同一 key 同一时刻最多 1 个在跑。</param>
    /// <param name="job">在独立 DI scope 内执行的刷新委托。</param>
    public void Enqueue(string key, Func<IServiceProvider, CancellationToken, Task> job)
    {
        lock (_gate)
        {
            if (_states.TryGetValue(key, out var state) && state.Running)
            {
                state.Pending = true; // 合并：运行期间的新触发只置 pending
                return;
            }
            _states[key] = new RefreshState { Running = true, Pending = false };
        }

        // fire-and-forget：独立后台任务，异常在内部吞掉并记日志
        _ = RunLoopAsync(key, job);
    }

    private async Task RunLoopAsync(string key, Func<IServiceProvider, CancellationToken, Task> job)
    {
        while (true)
        {
            try
            {
                using var scope = _scopeFactory.CreateScope();
                await job(scope.ServiceProvider, CancellationToken.None);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "MdpRefreshDispatcher[{Key}] 后台全量刷新失败", key);
            }

            lock (_gate)
            {
                var state = _states[key];
                if (!state.Pending)
                {
                    state.Running = false;
                    return;
                }
                state.Pending = false; // 吸收运行期间累计的触发，再补跑一轮
            }
        }
    }

    private sealed class RefreshState
    {
        public bool Running;
        public bool Pending;
    }
}