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

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;

/// <summary>
/// 方式甲动态调度。不标注 JobDetail，避免被内置扫描当成固定作业删掉。
/// </summary>
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<Payload>(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<ISqlSugarClient>();
        var ops = scope.ServiceProvider.GetRequiredService<MdpDbEntityOperationService>();
        var neutralProjection = scope.ServiceProvider.GetRequiredService<MdpDbSyncNeutralProjection>();
        var rebuild = scope.ServiceProvider.GetRequiredService<MdpDbSyncRebuildTrigger>();
        var task = await db.Queryable<MdpSyncTask>()
            .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<MdpSyncTaskSchedule>()
            .Where(x => x.TaskCode == task.TaskCode && x.TenantId == task.TenantId)
            .FirstAsync(stoppingToken);
        if (schedule == null || schedule.AutoEnabled != 1)
            return;

        var entities = await db.Queryable<MdpEntity>()
            .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; }
    }
}