using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.FinishedWarehouse;
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;
///
/// 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 仍独立保留。
///
[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)
{
// ① 硬门:T8 基表入站(源→stg→std)必须先成功;失败则本轮结束,不继续下游、不降级用旧 std。
try
{
using var inboundScope = _scopeFactory.CreateScope();
var inbound = inboundScope.ServiceProvider.GetRequiredService();
var inboundResult = await inbound.RunInboundAsync(
tenantId: 0, 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().RunFullAsync(stoppingToken, "AUTO"));
await RunStepAsync("S6", stoppingToken, async sp =>
(object)await sp.GetRequiredService().RunFullAsync(stoppingToken, "AUTO"));
await RunStepAsync("S7", stoppingToken, async sp =>
(object)await sp.GetRequiredService().RunFullAsync(stoppingToken, "AUTO"));
}
///
/// 下游模块步级执行:独立 DI scope + 失败隔离(仿 SmartOpsKpiMdpBootstrapJob.RunStepAsync)。
/// 仅用于 inbound 成功之后的 S5/S6/S7;不得用于包裹 inbound(inbound 失败必须中断本轮)。
///
private async Task RunStepAsync(string step, CancellationToken stoppingToken, Func> 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);
}
}
}