AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/MdpDbSyncScheduleBinder.cs

using System.Text.Json;
using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
using Admin.NET.Plugin.AiDOP.Job;
using Furion.Schedule;

namespace Admin.NET.Plugin.AiDOP.DataPlatform;

public sealed class MdpDbSyncScheduleBinder : ITransient
{
    private readonly ISqlSugarClient _db;
    private readonly ISchedulerFactory _scheduler;

    public MdpDbSyncScheduleBinder(ISqlSugarClient db, ISchedulerFactory scheduler)
    {
        _db = db;
        _scheduler = scheduler;
    }

    public async Task<string?> BindAsync(MdpSyncTask task, MdpSyncTaskSchedule schedule, long actingTenantId)
    {
        if (!string.Equals(task.TaskType, "DB_SYNC", StringComparison.OrdinalIgnoreCase))
            return schedule.ScheduleJobId;
        var tenantId = task.TenantId == 0 ? actingTenantId : task.TenantId;
        var jobId = MdpDbSyncRules.BuildJobId(tenantId, task.TaskCode)
            ?? throw Oops.Oh("任务编码过长，无法生成调度作业编号");
        var triggerId = MdpDbSyncRules.BuildTriggerId(tenantId, task.TaskCode)
            ?? throw Oops.Oh("任务编码过长，无法生成触发器编号");
        if (schedule.AutoEnabled == 1 && string.IsNullOrWhiteSpace(schedule.CronExpr))
            throw Oops.Oh("启用自动调度时必须填写 Cron");

        var jobType = typeof(MdpConfiguredDbSyncJob);
        var detail = await _db.Queryable<SysJobDetail>().Where(x => x.JobId == jobId).FirstAsync()
            ?? new SysJobDetail { JobId = jobId };
        detail.GroupName = "aidop-dbsync";
        detail.JobType = jobType.FullName;
        detail.AssemblyName = jobType.Assembly.GetName().Name;
        detail.Description = "方式甲数据库同步";
        detail.Concurrent = false;
        detail.IncludeAnnotation = false;
        detail.Properties = JsonSerializer.Serialize(new { tenantId, taskCode = task.TaskCode });
        detail.CreateType = JobCreateTypeEnum.Typed;
        detail.UpdatedTime = DateTime.Now;
        if (detail.Id <= 0)
            await _db.Insertable(detail).ExecuteCommandAsync();
        else
            await _db.Updateable(detail).ExecuteCommandAsync();

        var trigger = await _db.Queryable<SysJobTrigger>().Where(x => x.TriggerId == triggerId).FirstAsync()
            ?? new SysJobTrigger { TriggerId = triggerId };
        trigger.JobId = jobId;
        trigger.TriggerType = "Furion.Schedule.CronTrigger";
        trigger.AssemblyName = "Furion.Pure";
        trigger.Args = JsonSerializer.Serialize(new[] { string.IsNullOrWhiteSpace(schedule.CronExpr) ? "0 0 0 1 1 ?" : schedule.CronExpr.Trim() });
        trigger.Description = schedule.CronDesc;
        trigger.Status = schedule.AutoEnabled == 1 && task.Status == 1 ? TriggerStatus.Ready : TriggerStatus.Pause;
        if (trigger.Id <= 0)
            await _db.Insertable(trigger).ExecuteCommandAsync();
        else
            await _db.Updateable(trigger).ExecuteCommandAsync();

        Attach(detail, trigger, jobType);
        return jobId;
    }

    private void Attach(SysJobDetail detail, SysJobTrigger trigger, Type jobType)
    {
        var jobBuilder = JobBuilder.Create(jobType).LoadFrom(detail).SetJobType(jobType).SetIncludeAnnotations(false);
        var scheduler = _scheduler.GetJob(detail.JobId);
        if (scheduler == null)
            _scheduler.AddJob(jobBuilder);
        else
            scheduler.UpdateDetail(jobBuilder);

        var runtime = _scheduler.GetJob(detail.JobId);
        var triggerBuilder = TriggerBuilder.Create(trigger.TriggerId).LoadFrom(trigger);
        if (runtime?.GetTriggerBuilder(trigger.TriggerId) == null)
            runtime?.AddTrigger(triggerBuilder);
        else
            runtime.UpdateTrigger(triggerBuilder);
    }
}