AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S1Refresh/S1DashboardRebuildWorker.cs

using Admin.NET.Plugin.AiDOP.Order;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

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

public sealed class S1DashboardRebuildWorker : BackgroundService
{
    private static readonly TimeSpan IdleDelay = TimeSpan.FromSeconds(5);
    private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(30);

    private readonly IServiceScopeFactory _scopeFactory;
    private readonly S1DashboardRebuildQueue _queue;
    private readonly ILogger _logger;

    public S1DashboardRebuildWorker(
        IServiceScopeFactory scopeFactory,
        S1DashboardRebuildQueue queue,
        ILoggerFactory loggerFactory)
    {
        _scopeFactory = scopeFactory;
        _queue = queue;
        _logger = loggerFactory.CreateLogger(nameof(S1DashboardRebuildWorker));
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try { await Task.Delay(TimeSpan.FromSeconds(8), stoppingToken); }
        catch (OperationCanceledException) { return; }

        using (var probe = _scopeFactory.CreateScope())
        {
            var capability = probe.ServiceProvider.GetService<Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild.IModuleRebuildCapability>();
            if (capability?.IsEnabled("S1") == true)
            {
                _logger.LogInformation("[S1DashboardRebuildWorker] S1 已迁入通用 ModuleRebuildWorker，本 Worker 不再领取任务");
                try { await Task.Delay(Timeout.InfiniteTimeSpan, stoppingToken); }
                catch (OperationCanceledException) { }
                return;
            }
        }

        await FailStaleAsync(stoppingToken);

        var running = new List<Task>();
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                await FillSlotsAsync(running, stoppingToken);
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "[S1DashboardRebuildWorker] run failed");
            }

            running.RemoveAll(t => t.IsCompleted);
            if (running.Count == 0)
            {
                try
                {
                    using var linked = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
                    linked.CancelAfter(IdleDelay);
                    try { await _queue.Reader.ReadAsync(linked.Token); }
                    catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested) { }
                }
                catch (OperationCanceledException) { break; }
            }
            else
            {
                var pulse = WaitPulseAsync(stoppingToken);
                var completed = await Task.WhenAny(running.Append(pulse));
                if (completed != pulse)
                    running.Remove(completed);
            }
        }

        try { await Task.WhenAll(running); }
        catch (Exception ex) { _logger.LogWarning(ex, "[S1DashboardRebuildWorker] drain failed"); }
    }

    private async Task FillSlotsAsync(List<Task> running, CancellationToken stoppingToken)
    {
        int max = 2;
        try
        {
            max = Math.Clamp(Furion.App.GetConfig<int?>("AiDOP:S1Rebuild:MaxParallelTenants", true) ?? 2, 1, 8);
        }
        catch
        {
            max = 2;
        }
        while (running.Count < max && !stoppingToken.IsCancellationRequested)
        {
            var started = await TryStartOneAsync(stoppingToken);
            if (started == null)
                break;
            running.Add(started);
        }
    }

    private async Task<Task> TryStartOneAsync(CancellationToken stoppingToken)
    {
        using var probe = _scopeFactory.CreateScope();
        var job = await probe.ServiceProvider.GetRequiredService<IS1DashboardRebuildJobStore>()
            .ClaimNextQueuedAsync(stoppingToken);
        if (job == null)
            return null;
        return ExecuteClaimedAsync(job, stoppingToken);
    }

    private async Task ExecuteClaimedAsync(Admin.NET.Plugin.AiDOP.Entity.SmartOps.AdoS1DashboardRebuildJob job, CancellationToken stoppingToken)
    {
        using var scope = _scopeFactory.CreateScope();
        var svc = scope.ServiceProvider.GetRequiredService<S1DashboardRebuildService>();
        var transform = scope.ServiceProvider.GetRequiredService<S1MdpSyncTransformService>();
        using var heartbeatCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
        var heartbeat = KeepJobHeartbeatAsync(job.Id, heartbeatCts.Token);
        try
        {
            await svc.RunClaimedAsync(
                job,
                (runScope, rebuildJobId, report, ct) =>
                    transform.RunFullAsync(runScope, ct, "MANUAL", rebuildJobId, report),
                stoppingToken);
        }
        finally
        {
            heartbeatCts.Cancel();
            try { await heartbeat; } catch (OperationCanceledException) { }
        }
    }

    private async Task KeepJobHeartbeatAsync(long jobId, CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            try
            {
                await Task.Delay(HeartbeatInterval, ct);
                using var scope = _scopeFactory.CreateScope();
                await scope.ServiceProvider.GetRequiredService<S1DashboardRebuildService>().TouchHeartbeatAsync(jobId, CancellationToken.None);
            }
            catch (OperationCanceledException)
            {
                return;
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "[S1DashboardRebuildWorker] job heartbeat failed jobId={JobId}", jobId);
            }
        }
    }

    private async Task WaitPulseAsync(CancellationToken stoppingToken)
    {
        using var linked = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
        linked.CancelAfter(IdleDelay);
        try { await _queue.Reader.ReadAsync(linked.Token); }
        catch (OperationCanceledException) { }
    }

    private async Task FailStaleAsync(CancellationToken stoppingToken)
    {
        try
        {
            using var scope = _scopeFactory.CreateScope();
            await scope.ServiceProvider.GetRequiredService<S1DashboardRebuildService>().FailStaleAsync(stoppingToken);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[S1DashboardRebuildWorker] fail-stale failed");
        }
    }
}