# SchedulerServiceImpl.java

    package cloud.wconnect.frontoffice.service.adjustment.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.*;
    import cloud.wconnect.commonlib.exception.BadRequestException;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.pojo.dto.scheduler.JobBodyDTO;
    import cloud.wconnect.commonlib.pojo.dto.scheduler.JobParameterDTO;
    import cloud.wconnect.commonlib.service.SchedulerInternalService;
    import cloud.wconnect.commonlib.util.*;
    import cloud.wconnect.frontoffice.entity.adjustment.SchedulerJob;
    import cloud.wconnect.frontoffice.mapper.adjustment.SchedulerMapper;
    import cloud.wconnect.frontoffice.pojo.dto.adjustment.SchedulerDTO;
    import cloud.wconnect.frontoffice.service.adjustment.ISchedulerService;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
    import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
    import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;

    import java.text.MessageFormat;
    import java.util.*;
    import java.util.concurrent.Executor;

    @Service
    @DS("logbook")
    public class SchedulerServiceImpl extends ServiceImpl<SchedulerMapper, SchedulerJob> implements ISchedulerService {

        @Autowired
        private BeanMapper beanMapper;
        @Autowired
        private Executor executor;
        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private SchedulerInternalService schedulerInternalService;
        @Value("${default.timezone}")
        private String defaultTimezone;
        @Autowired
        private UpdateLogUtil updateLogUtil;
        private final static Integer MODULE_ID = MoudleIdConstant.FRONT_OFFICE; // front office module id
        private final static Integer FUNCTION_ID = FunctionIdConstant.ADJUSTMENT_REQUESTS; // adjustment feature id

        /**
         * 刷新调度任务同步状态为已同步
         *
         * @param ids 调度任务ID集合
         * @return void
         * @author lichen
         * @date 2024/3/12
         */
        @Override
        public void refreshSyncStateByIds(List<String> ids) {
            baseMapper.refreshSyncStateByIds(ids);
        }

        /**
         * 获取日期区间
         *
         * @param propertyDTO 资产DTO
         * @param type        邮件配置类型
         * @return java.util.Map<java.lang.String, java.lang.Long>
         * @author lichen
         * @date 2024/3/11
         */
        @Override
        public Map<String, Long> getDateRange(PropertyDTO propertyDTO, int type) {
            SchedulerJob schedulerJob = getOne(new LambdaQueryWrapper<SchedulerJob>().eq(SchedulerJob::getPropertyId, propertyDTO.getId())
                    .eq(SchedulerJob::getCategory, type));

            if (Objects.isNull(schedulerJob)) return Collections.emptyMap();

            String parameters = schedulerJob.getParameters();
            if (StringUtils.isBlank(parameters) || parameters.equals("{}")) return Collections.emptyMap();

            //type等于3（报告缺失时），schedulerJob不为空
            JobParameterDTO jobParameterDTO = JsonUtils.convertStringToObject(objectMapper, parameters, JobParameterDTO.class);

            //解析日期表达式
            return JobParameterUtils.parseDateRange(jobParameterDTO.getDateRange(), jobParameterDTO.getDateRangeUnit(), propertyDTO.getTimezone(), null);
        }

        /**
         * 校验调度任务是否存在
         *
         * @param id 主键ID
         * @return cloud.wconnect.frontoffice.entity.adjustment.SchedulerJob
         * @author lichen
         * @date 2024/3/12
         */
        @Override
        public SchedulerJob checkExist(String id) {
            SchedulerJob schedulerJob = getById(id);
            if (Objects.isNull(schedulerJob)) {
                throw new BadRequestException(CommonErrorI18nConstant.DATA_NOT_FOUND,
                        BaseErrorCodeEnum.DATA_VALIDATION_ERROR.getCode(), "adjustment SchedulerJob not found, id = " + id);
            }
            return schedulerJob;
        }

        /**
         * 修改调度任务
         *
         * @param propertyDTO     资产DTO
         * @param oriSchedulerJob 修改前的调度任务
         * @param schedulerDTO    修改后的调度任务DTO
         * @param loginUserId     登录用户id
         * @return job id
         * @author lichen
         * @date 2024/3/12
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public String update(PropertyDTO propertyDTO, SchedulerJob oriSchedulerJob, SchedulerDTO schedulerDTO, int loginUserId) {
            //修改调度任务
            SchedulerJob schedulerJob = beanMapper.map(schedulerDTO, SchedulerJob.class);
            schedulerJob.setRunTimes(0);
            schedulerJob.setUpdateLog(updateLogUtil.append(oriSchedulerJob.getUpdateLog(), loginUserId));
            //设为未同步
            schedulerJob.setSyncState(false);

            String originJobId = oriSchedulerJob.getId();
            // 更新调度任务的 job id
            updateAndGenerateNewJobId(schedulerJob);
            String newJobId = schedulerJob.getId();

            // 同步任务调度中心
            TransactionUtils.afterCommitAsyncExecute(executor, () -> {
                if (schedulerJob.getStatus() == SchedulerCons.STATUS_START) {
                    try {
                        //调用内部接口保存任务
                        callInternalServiceLaunchJob(originJobId, schedulerJob, propertyDTO, defaultTimezone);
                        //指定数据源
                        pushDynamicDataSource();
                        //更新为已同步
                        updateSyncState(newJobId, true);
                    } catch (Exception e) {
                        log.error("Launch adjustment scheduler job[" + newJobId + "] failed", e);
                    } finally {
                        DynamicDataSourceContextHolder.clear();
                    }
                } else {
                    try {
                        //调用内部接口删除任务
                        callInternalServiceDeleteJob(originJobId);
                        //指定数据源
                        pushDynamicDataSource();
                        //更新为已同步
                        updateSyncState(newJobId, true);
                    } catch (Exception e) {
                        log.error("Delete adjustment scheduler job[" + originJobId + "] failed", e);
                    } finally {
                        DynamicDataSourceContextHolder.clear();
                    }
                }
            });
            return newJobId;
        }

        /**
         * 更新调度任务和生成新的 job id
         *
         * @param schedulerJob 调度任务
         * @author zc
         * @date 2024/5/28
         */
        private void updateAndGenerateNewJobId(SchedulerJob schedulerJob) {
            String newJobId = CommonUtils.randomUuidWithoutDashes();
            update(schedulerJob, new LambdaUpdateWrapper<SchedulerJob>()
                    .set(SchedulerJob::getId, newJobId)
                    .eq(SchedulerJob::getId, schedulerJob.getId()));
            schedulerJob.setId(newJobId);
        }

        /**
         * 删除任务
         *
         * @param schedulerJob 调度任务
         * @param propertyDTO  资产DTO
         * @return void
         * @author lichen
         * @date 2024/3/12
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void delete(SchedulerJob schedulerJob, PropertyDTO propertyDTO) {
            //更新为未同步 删除中
            schedulerJob.setSyncState(false).setStatus(SchedulerCons.STATUS_DELETING);
            updateById(schedulerJob);

            // 同步任务调度中心
            TransactionUtils.afterCommitAsyncExecute(executor, () -> {
                try {
                    //调用内部接口删除任务
                    callInternalServiceDeleteJob(schedulerJob.getId());
                    //指定数据源
                    pushDynamicDataSource();
                    //删除记录
                    removeById(schedulerJob.getId());
                } catch (Exception e) {
                    log.error("Delete adjustment scheduler job[" + schedulerJob.getId() + "] failed", e);
                } finally {
                    DynamicDataSourceContextHolder.clear();
                }
            });
        }

        /**
         * 调度任务执行次数递增
         *
         * @param jobId
         * @return void
         * @author lichen
         * @date 2024/3/12
         */
        @Override
        public void plusRunTimes(String jobId) {
            //执行次数递增
            baseMapper.plusRunTimes(jobId);
        }

        /**
         * 更新同步状态
         *
         * @param id        主键ID
         * @param syncState 同步状态
         * @return void
         * @author lichen
         * @date 2024/3/12
         */
        @Override
        public void updateSyncState(String id, boolean syncState) {
            updateById(new SchedulerJob().setId(id).setSyncState(syncState));
        }

        /**
         * 校验分类下最大可配置调度任务的数量
         *
         * @param propertyId 资产Id
         * @param category   分类
         * @param maxNum     最大数量
         * @return void
         * @author lichen
         * @date 2024/3/13
         */
        @Override
        public void checkMaxNum(Integer propertyId, Integer category, int maxNum) {
            long count = count(new LambdaQueryWrapper<SchedulerJob>().eq(SchedulerJob::getPropertyId, propertyId)
                    .eq(SchedulerJob::getCategory, category)
                    .ne(SchedulerJob::getStatus, SchedulerCons.STATUS_DELETING));
            if (count >= maxNum) {
                throw new BadRequestException(CommonErrorI18nConstant.SCHEDULER_COUNT_TRANSFINITE,
                        BaseErrorCodeEnum.DATA_VALIDATION_ERROR.getCode(),
                        MessageFormat.format("adjustment SchedulerJob maximum configuration:{0}, propertyId{1}, categroy:{2}",
                                maxNum, propertyId, category));
            }
        }

        /**
         * 校验执行时间唯一
         *
         * @param propertyId      资产Id
         * @param category        分类
         * @param targetScheduler 目标规则
         * @param ignoreId        忽略的Id
         * @return void
         * @author lichen
         * @date 2024/3/19
         */
        @Override
        public void checkCornRuleSole(Integer propertyId, Integer category, String targetScheduler, String ignoreId) {
            List<SchedulerJob> existScheduler = list(new LambdaQueryWrapper<SchedulerJob>()
                    .eq(SchedulerJob::getPropertyId, propertyId)
                    .eq(SchedulerJob::getCategory, category)
                    .ne(StringUtils.isNotBlank(ignoreId), SchedulerJob::getId, ignoreId)
                    .ne(SchedulerJob::getStatus, SchedulerCons.STATUS_DELETING));
            if (CollectionUtils.isEmpty(existScheduler)) return;

            //校验执行时间是否重复
            CronRuleUtil.checkSchedulerRepeat(objectMapper, existScheduler.stream().map(SchedulerJob::getScheduler).toList(), targetScheduler);
        }

        /**
         * 查询调度任务
         *
         * @param propertyId 资产Id
         * @param category   分类
         * @return java.util.List<cloud.wconnect.frontoffice.entity.adjustment.SchedulerJob>
         * @author lichen
         * @date 2024/3/19
         */
        @Override
        public List<SchedulerJob> listByCategory(int propertyId, Integer category) {
            return list(new LambdaQueryWrapper<SchedulerJob>()
                    .eq(SchedulerJob::getPropertyId, propertyId)
                    .eq(SchedulerJob::getCategory, category)
                    .ne(SchedulerJob::getStatus, SchedulerCons.STATUS_DELETING)
                    .orderByDesc(SchedulerJob::getId));
        }

        /**
         * 创建调度任务
         *
         * @param schedulerDTO 调度任务DTO
         * @param propertyDTO  资产DTO
         * @param loginUserId  登录用户id
         * @return job id
         * @author lichen
         * @date 2024/3/11
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public String createScheduler(SchedulerDTO schedulerDTO, PropertyDTO propertyDTO, int loginUserId) {
            SchedulerJob schedulerJob = beanMapper.map(schedulerDTO, SchedulerJob.class);
            schedulerJob.setRunTimes(0);

            // 如果新任务的status为0-停止，则只创建DB记录，不创建定时任务job（DB记录的sync_state设为1）
            if (schedulerJob.getStatus() == SchedulerCons.STATUS_STOP) {
                schedulerJob.setSyncState(true);
            } else {
                schedulerJob.setSyncState(false);
            }
            schedulerJob.setUpdateLog(updateLogUtil.get(loginUserId));
            save(schedulerJob);

            //同步定时任务
            if (!schedulerJob.getSyncState()) {
                TransactionUtils.afterCommitAsyncExecute(executor, () -> {
                    // 同步任务调度中心
                    try {
                        //调用内部接口保存任务
                        callInternalServiceLaunchJob(StringUtils.EMPTY, schedulerJob, propertyDTO, defaultTimezone);
                        //指定数据源
                        pushDynamicDataSource();
                        //更新为已同步
                        updateSyncState(schedulerJob.getId(), true);
                    } catch (Exception e) {
                        log.error("Launch adjustment scheduler job[" + schedulerJob.getId() + "] failed", e);
                    } finally {
                        DynamicDataSourceContextHolder.clear();
                    }
                });
            }

            return schedulerJob.getId();
        }

        /**
         * 调用内部接口删除任务
         *
         * @param jobId 任务ID
         * @return void
         * @author lichen
         * @date 2024/3/12
         */
        private void callInternalServiceDeleteJob(String jobId) {
            schedulerInternalService.deleteJob(new JobBodyDTO().setJobId(jobId)
                    .setModuleId(MODULE_ID).setFunctionId(FUNCTION_ID));
        }

        /**
         * 调用内部接口保存任务
         *
         * @param originJobId     原调度任务 id
         * @param schedulerJob    调度任务对象
         * @param propertyDTO     资产DTO
         * @param defaultTimezone 默认时区
         * @return void
         * @author lichen
         * @date 2024/3/12
         */
        private void callInternalServiceLaunchJob(String originJobId, SchedulerJob schedulerJob, PropertyDTO propertyDTO, String defaultTimezone) throws JsonProcessingException {
            JobBodyDTO task = new JobBodyDTO()
                    .setJobId(schedulerJob.getId()).setOriginJobId(originJobId)
                    .setModuleId(MODULE_ID).setFunctionId(FUNCTION_ID)
                    .setStartDate(schedulerJob.getStartDate())
                    .setCronRule(CronRuleUtil.generateCronRuleForScheduler(objectMapper.readValue(schedulerJob.getScheduler(), HashMap.class),
                            schedulerJob.getStartDate(), propertyDTO.getTimezone(), defaultTimezone));
            schedulerInternalService.launchJob(task);
        }

        /**
         * 设置数据源
         *
         * @return void
         * @author lichen
         * @date 2024/3/15
         */
        private void pushDynamicDataSource() {
            DynamicDataSourceContextHolder.push("logbook");
        }
    }