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

using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Admin.NET.Plugin.AiDOP.Entity.SmartOps;
using Admin.NET.Plugin.AiDOP.Order;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;

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

public sealed class S1DashboardRebuildService : ITransient
{
    private readonly IS1DashboardRebuildJobStore _jobs;
    private readonly S1DashboardRebuildQueue _queue;
    private readonly ModuleRebuildService _moduleRebuild;
    private readonly IModuleRebuildCapability _capability;
    private readonly ILogger _logger;

    public S1DashboardRebuildService(IS1DashboardRebuildJobStore jobs, S1DashboardRebuildQueue queue)
        : this(jobs, queue, null, new AlwaysOnModuleRebuildCapability(), NullLoggerFactory.Instance)
    {
    }

    public S1DashboardRebuildService(
        IS1DashboardRebuildJobStore jobs,
        S1DashboardRebuildQueue queue,
        ILoggerFactory loggerFactory)
        : this(jobs, queue, null, new AlwaysOnModuleRebuildCapability(), loggerFactory)
    {
    }

    public S1DashboardRebuildService(
        IS1DashboardRebuildJobStore jobs,
        S1DashboardRebuildQueue queue,
        ModuleRebuildService moduleRebuild,
        IModuleRebuildCapability capability,
        ILoggerFactory loggerFactory)
    {
        _jobs = jobs;
        _queue = queue;
        _moduleRebuild = moduleRebuild;
        _capability = capability;
        _logger = loggerFactory.CreateLogger(nameof(S1DashboardRebuildService));
    }

    private bool UseUnified => _moduleRebuild != null && _capability.IsEnabled("S1");

    public async Task<(int StatusCode, S1RebuildJobAccepted Body)> EnqueueAsync(
        long tenantId,
        long factoryId,
        long? requestedBy,
        string triggerType = "MANUAL",
        CancellationToken ct = default)
    {
        if (UseUnified)
        {
            var (status, body) = await _moduleRebuild.EnqueueAsync("S1", tenantId, factoryId, requestedBy, triggerType, ct);
            return (status, new S1RebuildJobAccepted
            {
                Ok = body.Ok,
                JobId = body.JobId,
                Status = body.Status,
                Message = body.Message
            });
        }
        var scope = S1MdpRunScope.Create(tenantId, factoryId);
        var active = await _jobs.FindActiveAsync(scope.TenantId, scope.FactoryId, ct);
        if (active != null)
        {
            return (409, new S1RebuildJobAccepted
            {
                Ok = false,
                JobId = active.Id,
                Status = active.Status,
                Message = "S1 数据重算正在执行，请勿重复提交"
            });
        }

        var now = DateTime.Now;
        var created = await _jobs.InsertQueuedAsync(new AdoS1DashboardRebuildJob
        {
            TenantId = scope.TenantId,
            FactoryId = scope.FactoryId,
            Status = S1DashboardRebuildStatus.Queued,
            CurrentStage = S1MdpRebuildStage.Queued,
            StageIndex = 0,
            StageTotal = S1MdpRebuildStage.StageTotal,
            ProgressPercent = 0,
            ProgressMessage = "已入队",
            LastProgressAt = now,
            TriggerType = string.IsNullOrWhiteSpace(triggerType) ? "MANUAL" : triggerType,
            RequestedBy = requestedBy,
            SubmittedAt = now,
            CreateTime = now,
            UpdateTime = now
        }, ct);
        var oldest = await _jobs.FindActiveAsync(scope.TenantId, scope.FactoryId, ct);
        if (oldest != null && oldest.Id != created.Id)
        {
            created.Status = S1DashboardRebuildStatus.Failed;
            created.CurrentStage = S1MdpRebuildStage.Failed;
            created.FinishedAt = DateTime.Now;
            created.ErrorMessage = "S1 数据重算正在执行，请勿重复提交";
            created.UpdateTime = DateTime.Now;
            await _jobs.UpdateAsync(created, ct);
            return (409, new S1RebuildJobAccepted
            {
                Ok = false,
                JobId = oldest.Id,
                Status = oldest.Status,
                Message = "S1 数据重算正在执行，请勿重复提交"
            });
        }

        _queue.Pulse();
        return (202, new S1RebuildJobAccepted
        {
            Ok = true,
            JobId = created.Id,
            Status = S1DashboardRebuildStatus.Queued,
            Message = "S1 数据重算已排队"
        });
    }

    public async Task<S1RebuildJobDto> GetAsync(long jobId, long tenantId, long factoryId, CancellationToken ct = default)
    {
        if (UseUnified)
        {
            try
            {
                var dto = await _moduleRebuild.GetAsync("S1", jobId, tenantId, factoryId, ct);
                return Map(dto);
            }
            catch
            {
                // 兼容期内回退旧表
            }
        }
        var scope = S1MdpRunScope.Create(tenantId, factoryId);
        var row = await _jobs.GetByIdAsync(jobId, scope.TenantId, scope.FactoryId, ct);
        if (row == null)
            throw Oops.Oh("任务不存在");
        return ToDto(row);
    }

    public async Task<S1RebuildJobDto> LatestAsync(long tenantId, long factoryId, CancellationToken ct = default)
    {
        if (UseUnified)
        {
            var dto = await _moduleRebuild.LatestAsync("S1", tenantId, factoryId, ct);
            if (dto != null) return Map(dto);
        }
        var scope = S1MdpRunScope.Create(tenantId, factoryId);
        var row = await _jobs.GetLatestAsync(scope.TenantId, scope.FactoryId, ct);
        return row == null ? null : ToDto(row);
    }

    public async Task FailStaleAsync(CancellationToken ct = default) =>
        await _jobs.FailStaleRunningAsync(S1MdpFullRunLock.StaleAfter, ct);

    public async Task TouchHeartbeatAsync(long jobId, CancellationToken ct = default)
    {
        try
        {
            await _jobs.TouchHeartbeatAsync(jobId, DateTime.Now, ct);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[S1Rebuild] heartbeat failed jobId={JobId}", jobId);
        }
    }

    public async Task ApplyProgressAsync(long jobId, S1MdpProgressUpdate update, CancellationToken ct = default)
    {
        try
        {
            var now = DateTime.Now;
            if (update.CompletedStage != null && update.Rows.HasValue)
            {
                await _jobs.UpdateStageResultAsync(
                    jobId, update.CompletedStage, update.Rows.Value, update.ProgressPercent, update.Message, now, ct);
            }
            await _jobs.UpdateProgressAsync(
                jobId, update.Stage, update.StageIndex, update.ProgressPercent, update.Message, now, ct);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[S1Rebuild] progress write failed jobId={JobId} stage={Stage}", jobId, update.Stage);
        }
    }

    public async Task RunNextAsync(
        Func<S1MdpRunScope, long, Func<S1MdpProgressUpdate, Task>, CancellationToken, Task<S1MdpSyncTransformResult>> runFull,
        CancellationToken stoppingToken)
    {
        var job = await _jobs.ClaimNextQueuedAsync(stoppingToken);
        if (job == null)
            return;
        await RunClaimedAsync(job, runFull, stoppingToken);
    }

    public async Task<AdoS1DashboardRebuildJob> ClaimNextAsync(CancellationToken ct = default) =>
        await _jobs.ClaimNextQueuedAsync(ct);

    public async Task RunClaimedAsync(
        AdoS1DashboardRebuildJob job,
        Func<S1MdpRunScope, long, Func<S1MdpProgressUpdate, Task>, CancellationToken, Task<S1MdpSyncTransformResult>> runFull,
        CancellationToken stoppingToken)
    {
        var failedStage = job.CurrentStage;
        try
        {
            var scope = S1MdpRunScope.Create(job.TenantId, job.FactoryId);
            await ApplyProgressAsync(job.Id, new S1MdpProgressUpdate(
                S1MdpRebuildStage.AcquiringLock, 0, 2, "等待现有 S1 全量任务完成"), CancellationToken.None);

            var result = await runFull(
                scope,
                job.Id,
                async update =>
                {
                    failedStage = update.Stage;
                    await ApplyProgressAsync(job.Id, update, CancellationToken.None);
                },
                stoppingToken);

            var now = DateTime.Now;
            job.Status = S1DashboardRebuildStatus.Success;
            job.CurrentStage = S1MdpRebuildStage.Success;
            job.StageIndex = S1MdpRebuildStage.StageTotal;
            job.ProgressPercent = 100;
            job.ProgressMessage = "S1 数据重算已完成";
            job.FailedStage = null;
            job.LastProgressAt = now;
            job.BatchId = result.BatchId;
            job.TransformRunLogId = result.RunLogId;
            job.StageRows = result.StageRows;
            job.StandardRows = result.StandardRows;
            job.DwdRows = result.DwdRows;
            job.KpiRows = result.KpiRows;
            job.AtomicRows = result.AtomicRows;
            job.FinishedAt = now;
            job.HeartbeatAt = now;
            job.DurationMs = job.StartedAt.HasValue ? ToDurationMs(now - job.StartedAt.Value) : null;
            job.UpdateTime = now;
            job.ErrorMessage = null;
            await _jobs.UpdateAsync(job, CancellationToken.None);
        }
        catch (S1MdpAlreadyRunningException)
        {
            var now = DateTime.Now;
            job.Status = S1DashboardRebuildStatus.Queued;
            job.CurrentStage = S1MdpRebuildStage.AcquiringLock;
            job.StageIndex = 0;
            job.ProgressPercent = 2;
            job.ProgressMessage = "等待现有 S1 全量任务完成";
            job.StartedAt = null;
            job.HeartbeatAt = now;
            job.LastProgressAt = now;
            job.UpdateTime = now;
            job.ErrorMessage = null;
            await _jobs.UpdateAsync(job, CancellationToken.None);
            _queue.Pulse();
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            await FailJobAsync(job, "服务停止，任务已取消", S1DashboardRebuildStatus.Cancelled, S1MdpRebuildStage.Cancelled, failedStage);
        }
        catch (Exception ex)
        {
            await FailJobAsync(
                job,
                Truncate(ex.Message, 2000),
                S1DashboardRebuildStatus.Failed,
                S1MdpRebuildStage.Failed,
                failedStage);
        }
    }

    private async Task FailJobAsync(AdoS1DashboardRebuildJob job, string message, string status, string currentStage, string failedStage)
    {
        var now = DateTime.Now;
        job.Status = status;
        job.CurrentStage = currentStage;
        job.FailedStage = status == S1DashboardRebuildStatus.Failed ? failedStage : job.FailedStage;
        job.ProgressMessage = status == S1DashboardRebuildStatus.Failed
            ? $"在【{S1MdpRebuildStage.ToChinese(failedStage)}】失败"
            : message;
        job.FinishedAt = now;
        job.HeartbeatAt = now;
        job.LastProgressAt = now;
        job.DurationMs = job.StartedAt.HasValue ? ToDurationMs(now - job.StartedAt.Value) : null;
        job.ErrorMessage = message;
        job.UpdateTime = now;
        await _jobs.UpdateAsync(job, CancellationToken.None);
    }

    public static S1RebuildJobDto ToDto(AdoS1DashboardRebuildJob row) => new()
    {
        JobId = row.Id,
        Status = row.Status,
        CurrentStage = row.CurrentStage,
        StageIndex = row.StageIndex,
        StageTotal = row.StageTotal <= 0 ? S1MdpRebuildStage.StageTotal : row.StageTotal,
        ProgressPercent = row.ProgressPercent,
        ProgressMessage = row.ProgressMessage,
        LastProgressAt = row.LastProgressAt,
        HeartbeatAt = row.HeartbeatAt,
        FailedStage = row.FailedStage,
        SubmittedAt = row.SubmittedAt,
        StartedAt = row.StartedAt,
        FinishedAt = row.FinishedAt,
        DurationMs = row.DurationMs,
        BatchId = row.BatchId,
        StageRows = row.StageRows,
        StandardRows = row.StandardRows,
        DwdRows = row.DwdRows,
        KpiRows = row.KpiRows,
        AtomicRows = row.AtomicRows,
        ErrorMessage = row.ErrorMessage
    };

    private static S1RebuildJobDto Map(Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild.ModuleRebuildJobDto dto) => new()
    {
        JobId = dto.JobId,
        Status = dto.Status,
        CurrentStage = dto.CurrentStage,
        StageIndex = dto.StageIndex,
        StageTotal = dto.StageTotal,
        ProgressPercent = dto.ProgressPercent,
        ProgressMessage = dto.ProgressMessage,
        LastProgressAt = dto.LastProgressAt,
        HeartbeatAt = dto.HeartbeatAt,
        FailedStage = dto.FailedStage,
        SubmittedAt = dto.SubmittedAt,
        StartedAt = dto.StartedAt,
        FinishedAt = dto.FinishedAt,
        DurationMs = dto.DurationMs,
        BatchId = dto.BatchId,
        StageRows = dto.StageRows,
        StandardRows = dto.StandardRows,
        DwdRows = dto.DwdRows,
        KpiRows = dto.KpiRows,
        AtomicRows = dto.AtomicRows,
        ErrorMessage = dto.ErrorMessage
    };

    private static int ToDurationMs(TimeSpan elapsed)
    {
        var ms = elapsed.TotalMilliseconds;
        if (double.IsNaN(ms) || ms <= 0) return 0;
        return ms >= int.MaxValue ? int.MaxValue : (int)ms;
    }

    private static string Truncate(string value, int max) =>
        string.IsNullOrEmpty(value) || value.Length <= max ? value : value[..max];
}