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 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().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().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); } }