AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Job/S5MdpRefreshJob.cs

using Admin.NET.Core.Service;
using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.FinishedWarehouse;
using Admin.NET.Plugin.AiDOP.Infrastructure;
using Admin.NET.Plugin.AiDOP.Manufacturing;
using Admin.NET.Plugin.AiDOP.MaterialWarehouse;
using Furion.Schedule;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System.Text.Json;

namespace Admin.NET.Plugin.AiDOP.Job;

/// <summary>
/// T8 基础数据 + S5/S6/S7 KPI 统一刷新编排任务（方案④ owner）。
/// 一个 tick 内：先执行 1 次 T8 基表双模式入站（源→mdp_stg_t8_*→mdp_std_t8_*），
/// 再顺序计算 S5/S6/S7 三个模块的 T8 KPI（均读同一份本轮刷新后的 std，保证强一致）。
/// JobId 保留 job_s5_t8_kpi_refresh（不迁移 Job 身份/历史）；原 S6/S7 独立 Job 已合并至本 Job。
/// 每日 02:00 / 07:00 / 12:00 / 17:00 / 22:00 共 5 次。
///
/// 失败语义（关键，勿机械照抄 Bootstrap）：
///   ① T8 inbound 是整轮硬门——失败则记错误日志并立即结束本轮，
///      禁止继续算 S5/S6/S7（不允许退化为用上一周期 std 静默计算的 eventual consistency）；
///   ② inbound 成功后，S5→S6→S7 采用步级失败隔离（某模块失败记错误、继续下一模块）；
///   ③ OperationCanceledException 保持取消语义直接上抛，不吞。
/// 手工补偿入口 /t8-base-mdp/inbound 仍独立保留。
///
/// 并发：与手动刷新（AidopKanbanController → AidopT8KpiManualRefreshService）共享同一把硬 TTL 刷新锁
///   （RefreshLockKey）。占用时本轮定时跳过（5 次/天，下一 tick 自愈），避免共享入站/std 与手动刷新并发写；
///   本轮结束在 finally 主动释放，即使异常/挂死也由 TTL 到点自动过期，不永久阻塞后续刷新。
/// </summary>
[JobDetail("job_s5_t8_kpi_refresh",
    Description = "T8 基础数据 + S5/S6/S7 KPI 统一刷新编排（inbound 硬门 → S5→S6→S7；5 次/天：02/07/12/17/22）",
    GroupName = "default",
    Concurrent = false)]
[Cron("0 2,7,12,17,22 * * *",
    TriggerId = "trigger_s5_t8_kpi_refresh",
    Description = "每日 02:00/07:00/12:00/17:00/22:00 触发（5 字段：分 时 日 月 周，默认 CronStringFormat.Default）")]
public class S5MdpRefreshJob : IJob
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger _logger;

    public S5MdpRefreshJob(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
    {
        _scopeFactory = scopeFactory;
        _logger = loggerFactory.CreateLogger(nameof(S5MdpRefreshJob));
    }

    public async Task ExecuteAsync(JobExecutingContext context, CancellationToken stoppingToken)
    {
        // 与手动刷新共享同一把硬 TTL 刷新锁：占用时本轮跳过；本轮结束 finally 释放，异常/挂死由 TTL 自动过期兜底。
        using var lockScope = _scopeFactory.CreateScope();
        var cache = lockScope.ServiceProvider.GetRequiredService<SysCacheService>();
        if (cache.ExistKey(AidopT8KpiManualRefreshService.RefreshLockKey))
        {
            _logger.LogWarning("S5MdpRefreshJob 检测到刷新锁被占用（手动刷新或上轮未释放），本轮跳过");
            return;
        }
        cache.Set(AidopT8KpiManualRefreshService.RefreshLockKey, $"AUTO:{DateTime.Now:yyyyMMddHHmmss}",
            TimeSpan.FromSeconds(AidopT8KpiManualRefreshService.RefreshLockTtlSeconds));

        try
        {
            // ① 硬门：T8 基表入站（源→stg→std）必须先成功；失败则本轮结束，不继续下游、不降级用旧 std。
            try
            {
                using var inboundScope = _scopeFactory.CreateScope();
                var inbound = inboundScope.ServiceProvider.GetRequiredService<T8BaseInboundMdpSyncService>();
                // T8 源表无 tenant_id；定时任务显式经账套映射解析真实租户。
                var t8TenantId = AidopSourceTenantMap.ResolveTenantId("pbxfxp", 0);
                var inboundResult = await inbound.RunInboundAsync(
                    tenantId: t8TenantId, fullRefresh: true, entityCode: null, cancellationToken: stoppingToken);
                _logger.LogInformation("S5MdpRefreshJob T8 inbound 完成 {Payload}", JsonSerializer.Serialize(inboundResult));
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                _logger.LogInformation("S5MdpRefreshJob T8 inbound 收到停止信号，本轮结束");
                throw;
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "S5MdpRefreshJob T8 inbound 失败，本轮跳过 S5/S6/S7 KPI 刷新（不使用上一周期 std 继续计算）");
                return;
            }

            // ② 下游：inbound 成功后，S5→S6→S7 顺序执行，步级失败隔离（各自独立 DI scope）。
            await RunStepAsync("S5", stoppingToken, async sp =>
                (object)await sp.GetRequiredService<S5MdpSyncTransformService>().RunFullAsync(stoppingToken, "AUTO"));
            await RunStepAsync("S6", stoppingToken, async sp =>
                (object)await sp.GetRequiredService<S6MdpSyncTransformService>().RunFullAsync(stoppingToken, "AUTO"));
            await RunStepAsync("S7", stoppingToken, async sp =>
                (object)await sp.GetRequiredService<S7MdpSyncTransformService>().RunFullAsync(stoppingToken, "AUTO"));
        }
        finally
        {
            cache.Remove(AidopT8KpiManualRefreshService.RefreshLockKey);
        }
    }

    /// <summary>
    /// 下游模块步级执行：独立 DI scope + 失败隔离（仿 SmartOpsKpiMdpBootstrapJob.RunStepAsync）。
    /// 仅用于 inbound 成功之后的 S5/S6/S7；不得用于包裹 inbound（inbound 失败必须中断本轮）。
    /// </summary>
    private async Task RunStepAsync(string step, CancellationToken stoppingToken, Func<IServiceProvider, Task<object>> action)
    {
        using var scope = _scopeFactory.CreateScope();
        try
        {
            stoppingToken.ThrowIfCancellationRequested();
            var payload = await action(scope.ServiceProvider);
            _logger.LogInformation("S5MdpRefreshJob {Step} 完成 {Payload}", step, JsonSerializer.Serialize(payload));
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            _logger.LogInformation("S5MdpRefreshJob {Step} 收到停止信号", step);
            throw;
        }
        catch (Exception ex)
        {
            // 失败通知由各 service 内部 MarkTransformRunFailedAsync 触发；此处记录并继续下一步
            _logger.LogError(ex, "S5MdpRefreshJob {Step} 执行失败，继续后续步骤", step);
        }
    }
}