using System.Text.Json;
using Admin.NET.Plugin.AiDOP.DataPlatform;
using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Furion.Schedule;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
namespace Admin.NET.Plugin.AiDOP.Job;
///
/// 方式甲动态调度。不标注 JobDetail,避免被内置扫描当成固定作业删掉。
///
public sealed class MdpConfiguredDbSyncJob : IJob
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger _logger;
public MdpConfiguredDbSyncJob(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
{
_scopeFactory = scopeFactory;
_logger = loggerFactory.CreateLogger(nameof(MdpConfiguredDbSyncJob));
}
public async Task ExecuteAsync(JobExecutingContext context, CancellationToken stoppingToken)
{
var payload = JsonSerializer.Deserialize(context.JobDetail.Properties ?? "{}");
if (payload == null || payload.TenantId <= 0 || string.IsNullOrWhiteSpace(payload.TaskCode))
{
_logger.LogWarning("方式甲调度缺少 tenantId 或 taskCode");
return;
}
using var scope = _scopeFactory.CreateScope();
var db = scope.ServiceProvider.GetRequiredService();
var ops = scope.ServiceProvider.GetRequiredService();
var neutralProjection = scope.ServiceProvider.GetRequiredService();
var rebuild = scope.ServiceProvider.GetRequiredService();
var task = await db.Queryable()
.Where(x => x.TaskCode == payload.TaskCode && x.TenantId == payload.TenantId && x.Status == 1)
.FirstAsync(stoppingToken);
if (task == null || !string.Equals(task.TaskType, "DB_SYNC", StringComparison.OrdinalIgnoreCase))
return;
var schedule = await db.Queryable()
.Where(x => x.TaskCode == task.TaskCode && x.TenantId == task.TenantId)
.FirstAsync(stoppingToken);
if (schedule == null || schedule.AutoEnabled != 1)
return;
var entities = await db.Queryable()
.Where(x => x.Status == 1 && (x.TenantId == payload.TenantId || x.TenantId == 0))
.Where(x => x.JobId == task.TaskCode || x.JobId == task.JobCode)
.ToListAsync(stoppingToken);
var failed = 0;
foreach (var entity in entities)
{
stoppingToken.ThrowIfCancellationRequested();
var issues = MdpDbSyncRules.ValidateDbEntity(entity, "DB");
if (issues.Count > 0)
{
failed++;
continue;
}
try
{
var mode = string.IsNullOrWhiteSpace(schedule.SyncWindowType) ? entity.SyncMode : schedule.SyncWindowType;
await ops.RunEntityAsync(entity.Id, mode, "调度自动执行", "AUTO", stoppingToken, payload.TenantId);
}
catch (Exception ex)
{
failed++;
_logger.LogError(ex, "方式甲实体 {EntityCode} 调度失败", entity.EntityCode);
}
}
if (failed > 0)
{
_logger.LogWarning("方式甲任务 {TaskCode} 有 {Failed} 个实体未成功", task.TaskCode, failed);
throw new InvalidOperationException(
$"方式甲任务 {task.TaskCode} 有 {failed} 个实体未成功,本轮不触发模块重算");
}
else if (entities.Count > 0)
{
await neutralProjection.ProjectAsync(entities, payload.TenantId, stoppingToken);
var modules = await rebuild.EnqueueForTaskAsync(
task, payload.TenantId, factoryId: 1, triggerType: ModuleRebuildTriggerType.DbSync, stoppingToken);
_logger.LogInformation(
"方式甲任务 {TaskCode} 抽取完成,已触发模块重算:{Modules}",
task.TaskCode,
modules.Count == 0 ? "无" : string.Join(",", modules));
}
}
private sealed class Payload
{
public long TenantId { get; set; }
public string? TaskCode { get; set; }
}
}