using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Admin.NET.Plugin.AiDOP.Infrastructure;
///
/// 数据中台全量刷新后台派发器。
/// 将评审 / 采购管线末尾的全量 ETL 从「同步 await」改为「后台异步执行」,
/// 并按 key(如 "S1"、"S4")做「单飞 + 合并去抖」:
/// · 该 key 空闲 → 立即后台执行一轮;
/// · 正在执行 → 只置 pending,不排队叠加;
/// · 一轮结束若 pending 为真 → 再补跑一轮(吸收运行期间累计的触发)。
/// 说明:进程内单飞;多实例部署时跨实例合并依赖各模块的定时刷新任务兜底。
///
public class MdpRefreshDispatcher : ISingleton
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger _logger;
private readonly object _gate = new();
private readonly Dictionary _states = new();
public MdpRefreshDispatcher(IServiceScopeFactory scopeFactory, ILogger logger)
{
_scopeFactory = scopeFactory;
_logger = logger;
}
///
/// 提交一次后台刷新(立即返回,不等待执行完成)。
///
/// 去抖分组键,同一 key 同一时刻最多 1 个在跑。
/// 在独立 DI scope 内执行的刷新委托。
public void Enqueue(string key, Func 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 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;
}
}