AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpRebuild/ModuleRebuildStore.cs

using System.Threading.Channels;
using Admin.NET.Plugin.AiDOP.Entity.SmartOps;

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

public sealed class ModuleRebuildQueue : ISingleton
{
    private readonly Channel<byte> _channel = Channel.CreateBounded<byte>(new BoundedChannelOptions(8)
    {
        FullMode = BoundedChannelFullMode.DropOldest,
        SingleReader = true,
        SingleWriter = false
    });

    public void Pulse() => _channel.Writer.TryWrite(0);

    public ChannelReader<byte> Reader => _channel.Reader;
}

public sealed class ModuleRebuildJobAccepted
{
    public bool Ok { get; set; }
    public string ModuleCode { get; set; } = string.Empty;
    public long? JobId { get; set; }
    public string Status { get; set; }
    public string Message { get; set; }
}

public sealed class ModuleRebuildJobDto
{
    public bool Ok { get; set; } = true;
    public string ModuleCode { get; set; } = string.Empty;
    public long JobId { get; set; }
    public string Status { get; set; }
    public string CurrentStage { get; set; }
    public int StageIndex { get; set; }
    public int StageTotal { get; set; }
    public int ProgressPercent { get; set; }
    public string ProgressMessage { get; set; }
    public DateTime? LastProgressAt { get; set; }
    public DateTime? HeartbeatAt { get; set; }
    public string FailedStage { get; set; }
    public DateTime SubmittedAt { get; set; }
    public DateTime? StartedAt { get; set; }
    public DateTime? FinishedAt { get; set; }
    public int? DurationMs { get; set; }
    public string BatchId { get; set; }
    public int StageRows { get; set; }
    public int StandardRows { get; set; }
    public int DwdRows { get; set; }
    public int KpiRows { get; set; }
    public int AtomicRows { get; set; }
    public string DetailJson { get; set; }
    public string ErrorMessage { get; set; }
    public string Message { get; set; }
}

public interface IModuleRebuildJobStore
{
    Task<AdoModuleDashboardRebuildJob> InsertQueuedAsync(AdoModuleDashboardRebuildJob row, CancellationToken ct = default);
    Task<AdoModuleDashboardRebuildJob> FindActiveAsync(string moduleCode, long tenantId, long factoryId, CancellationToken ct = default);
    Task<AdoModuleDashboardRebuildJob> GetByIdAsync(string moduleCode, long id, long tenantId, long factoryId, CancellationToken ct = default);
    Task<AdoModuleDashboardRebuildJob> GetLatestAsync(string moduleCode, long tenantId, long factoryId, CancellationToken ct = default);
    Task<AdoModuleDashboardRebuildJob> ClaimNextQueuedAsync(IReadOnlyCollection<string> enabledModules, CancellationToken ct = default);
    Task UpdateAsync(AdoModuleDashboardRebuildJob row, CancellationToken ct = default);
    Task UpdateProgressAsync(long jobId, string currentStage, int stageIndex, int progressPercent, string message, DateTime now, CancellationToken ct = default);
    Task UpdateStageResultAsync(long jobId, string completedStage, int rows, int progressPercent, string nextMessage, string detailJson, DateTime now, CancellationToken ct = default);
    Task TouchHeartbeatAsync(long jobId, DateTime now, CancellationToken ct = default);
    Task FailStaleRunningAsync(TimeSpan staleAfter, CancellationToken ct = default);
}

public sealed class ModuleRebuildJobStore : IModuleRebuildJobStore, ITransient
{
    private readonly ISqlSugarClient _db;
    public ModuleRebuildJobStore(ISqlSugarClient db) => _db = db;

    public async Task<AdoModuleDashboardRebuildJob> InsertQueuedAsync(AdoModuleDashboardRebuildJob row, CancellationToken ct = default)
    {
        var id = await _db.Insertable(row).ExecuteReturnIdentityAsync();
        row.Id = id;
        return row;
    }

    public Task<AdoModuleDashboardRebuildJob> FindActiveAsync(string moduleCode, long tenantId, long factoryId, CancellationToken ct = default) =>
        _db.Queryable<AdoModuleDashboardRebuildJob>()
            .Where(x => x.ModuleCode == moduleCode && x.TenantId == tenantId && x.FactoryId == factoryId
                && (x.Status == ModuleRebuildStatus.Queued || x.Status == ModuleRebuildStatus.Running))
            .OrderBy(x => x.Id)
            .FirstAsync(ct);

    public Task<AdoModuleDashboardRebuildJob> GetByIdAsync(string moduleCode, long id, long tenantId, long factoryId, CancellationToken ct = default) =>
        _db.Queryable<AdoModuleDashboardRebuildJob>()
            .FirstAsync(x => x.Id == id && x.ModuleCode == moduleCode && x.TenantId == tenantId && x.FactoryId == factoryId, ct);

    public Task<AdoModuleDashboardRebuildJob> GetLatestAsync(string moduleCode, long tenantId, long factoryId, CancellationToken ct = default) =>
        _db.Queryable<AdoModuleDashboardRebuildJob>()
            .Where(x => x.ModuleCode == moduleCode && x.TenantId == tenantId && x.FactoryId == factoryId)
            .OrderBy(x => x.Id, OrderByType.Desc)
            .FirstAsync(ct);

    public async Task<AdoModuleDashboardRebuildJob> ClaimNextQueuedAsync(IReadOnlyCollection<string> enabledModules, CancellationToken ct = default)
    {
        if (enabledModules == null || enabledModules.Count == 0)
            return null;
        var enabled = enabledModules.ToList();
        var row = await _db.Queryable<AdoModuleDashboardRebuildJob>()
            .Where(x => x.Status == ModuleRebuildStatus.Queued && enabled.Contains(x.ModuleCode))
            .Where(x => SqlFunc.Subqueryable<AdoModuleDashboardRebuildJob>()
                .Where(y => y.ModuleCode == x.ModuleCode && y.TenantId == x.TenantId && y.FactoryId == x.FactoryId
                    && y.Status == ModuleRebuildStatus.Running)
                .NotAny())
            .OrderBy(x => x.Id)
            .FirstAsync(ct);
        if (row == null)
            return null;
        var now = DateTime.Now;
        var n = await _db.Updateable<AdoModuleDashboardRebuildJob>()
            .SetColumns(x => new AdoModuleDashboardRebuildJob
            {
                Status = ModuleRebuildStatus.Running,
                CurrentStage = ModuleRebuildStages.AcquiringLock,
                StageIndex = 0,
                ProgressPercent = 2,
                ProgressMessage = $"等待现有 {row.ModuleCode} 全量任务完成",
                LastProgressAt = now,
                StartedAt = now,
                HeartbeatAt = now,
                UpdateTime = now
            })
            .Where(x => x.Id == row.Id && x.Status == ModuleRebuildStatus.Queued)
            .ExecuteCommandAsync(ct);
        if (n <= 0)
            return null;
        row.Status = ModuleRebuildStatus.Running;
        row.CurrentStage = ModuleRebuildStages.AcquiringLock;
        row.ProgressPercent = 2;
        row.StartedAt = now;
        row.HeartbeatAt = now;
        row.UpdateTime = now;
        return row;
    }

    public Task UpdateAsync(AdoModuleDashboardRebuildJob row, CancellationToken ct = default) =>
        _db.Updateable(row).ExecuteCommandAsync(ct);

    public Task UpdateProgressAsync(long jobId, string currentStage, int stageIndex, int progressPercent, string message, DateTime now, CancellationToken ct = default) =>
        _db.Updateable<AdoModuleDashboardRebuildJob>()
            .SetColumns(x => new AdoModuleDashboardRebuildJob
            {
                CurrentStage = currentStage,
                StageIndex = stageIndex,
                ProgressPercent = progressPercent,
                ProgressMessage = message,
                LastProgressAt = now,
                HeartbeatAt = now,
                UpdateTime = now
            })
            .Where(x => x.Id == jobId && x.Status == ModuleRebuildStatus.Running)
            .ExecuteCommandAsync(ct);

    public Task UpdateStageResultAsync(long jobId, string completedStage, int rows, int progressPercent, string nextMessage, string detailJson, DateTime now, CancellationToken ct = default)
    {
        var column = completedStage switch
        {
            ModuleRebuildStages.Staging or ModuleRebuildStages.T8Inbound => "stage_rows",
            ModuleRebuildStages.Standard or ModuleRebuildStages.KpiPreparing => "standard_rows",
            ModuleRebuildStages.Dwd => "dwd_rows",
            ModuleRebuildStages.Kpi or ModuleRebuildStages.KpiCalculating => "kpi_rows",
            ModuleRebuildStages.Atomic => "atomic_rows",
            _ => null
        };
        if (column == null)
            return UpdateProgressAsync(jobId, completedStage, 0, progressPercent, nextMessage, now, ct);
        return _db.Ado.ExecuteCommandAsync(
            $"""
            UPDATE ado_module_dashboard_rebuild_job
            SET `{column}`=@Rows,
                progress_percent=@Pct,
                progress_message=@Msg,
                detail_json=IFNULL(@Detail, detail_json),
                last_progress_at=@Now,
                heartbeat_at=@Now,
                update_time=@Now
            WHERE id=@Id AND status='RUNNING'
            """,
            new SugarParameter("@Rows", rows),
            new SugarParameter("@Pct", progressPercent),
            new SugarParameter("@Msg", nextMessage),
            new SugarParameter("@Detail", (object?)detailJson ?? DBNull.Value),
            new SugarParameter("@Now", now),
            new SugarParameter("@Id", jobId));
    }

    public Task TouchHeartbeatAsync(long jobId, DateTime now, CancellationToken ct = default) =>
        _db.Updateable<AdoModuleDashboardRebuildJob>()
            .SetColumns(x => new AdoModuleDashboardRebuildJob { HeartbeatAt = now, UpdateTime = now })
            .Where(x => x.Id == jobId && x.Status == ModuleRebuildStatus.Running)
            .ExecuteCommandAsync(ct);

    public Task FailStaleRunningAsync(TimeSpan staleAfter, CancellationToken ct = default)
    {
        var cutoff = DateTime.Now - staleAfter;
        var now = DateTime.Now;
        return _db.Updateable<AdoModuleDashboardRebuildJob>()
            .SetColumns(x => new AdoModuleDashboardRebuildJob
            {
                Status = ModuleRebuildStatus.Failed,
                CurrentStage = ModuleRebuildStages.Failed,
                FinishedAt = now,
                LastProgressAt = now,
                UpdateTime = now,
                ErrorMessage = "服务中断，任务未正常结束"
            })
            .Where(x => x.Status == ModuleRebuildStatus.Running && (x.HeartbeatAt == null || x.HeartbeatAt < cutoff))
            .ExecuteCommandAsync(ct);
    }
}