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; } } }