# DbSchedulerServiceImpl.java

    package cloud.wconnect.workperformance.service.database.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.*;
    import cloud.wconnect.commonlib.exception.BadRequestException;
    import cloud.wconnect.commonlib.exception.ForbiddenException;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.pojo.dto.scheduler.JobBodyDTO;
    import cloud.wconnect.commonlib.service.SchedulerInternalService;
    import cloud.wconnect.commonlib.util.CommonUtils;
    import cloud.wconnect.commonlib.util.CronRuleUtil;
    import cloud.wconnect.commonlib.util.UpdateLogUtil;
    import cloud.wconnect.workperformance.entity.database.DbScheduler;
    import cloud.wconnect.workperformance.mapper.database.DbSchedulerMapper;
    import cloud.wconnect.workperformance.pojo.dto.database.notification.DbSchedulerDTO;
    import cloud.wconnect.workperformance.service.database.IDbSchedulerService;
    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.core.conditions.update.UpdateWrapper;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import com.fasterxml.jackson.core.JsonProcessingException;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import jakarta.annotation.Resource;
    import org.apache.commons.lang3.StringUtils;
    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.HashMap;
    import java.util.List;
    import java.util.Objects;
    import java.util.concurrent.Executor;

    /**
     * <p>
     * Database scheduler 服务实现类
     * </p>
     *
     * @author zc
     * @since 2025-12-02
     */
    @Service
    @DS("database")
    public class DbSchedulerServiceImpl extends ServiceImpl<DbSchedulerMapper, DbScheduler> implements IDbSchedulerService {
        @Resource
        private SchedulerInternalService schedulerInternalService;
        @Resource
        private UpdateLogUtil updateLogUtil;
        @Resource
        private BeanMapper beanMapper;
        @Resource
        private Executor executor;
        @Resource
        private ObjectMapper objectMapper;
        @Value("${default.timezone}")
        private String defaultTimezone;

        private final static Integer MODULE_ID = MoudleIdConstant.WORK_PERFORMANCE;
        private final static Integer FUNCTION_ID = FunctionIdConstant.DATABASE;

        /**
         * 查询调度任务的配置
         *
         * @param propertyId 资产 id
         * @param category   分类
         * @return 调度任务
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public List<DbScheduler> list(int propertyId, int category) {
            return list(new LambdaQueryWrapper<DbScheduler>()
                    .eq(DbScheduler::getPropertyId, propertyId)
                    .eq(DbScheduler::getCategory, category)
                    .ne(DbScheduler::getStatus, SchedulerCons.STATUS_DELETING));
        }

        /**
         * 校验是否存在
         *
         * @param id 调度任务 id
         * @return DbScheduler
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public DbScheduler checkExist(String id) {
            DbScheduler scheduler = getById(id);
            if (Objects.isNull(scheduler)) {
                throw new BadRequestException(CommonErrorI18nConstant.DATA_NOT_FOUND,
                        BaseErrorCodeEnum.DATA_VALIDATION_ERROR.getCode(),
                        "The Database scheduler is not found, id = " + id);
            }
            return scheduler;
        }

        /**
         * 校验归属
         *
         * @param propertyId 资产 id
         * @param scheduler  调度任务
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public void checkBelong(int propertyId, DbScheduler scheduler) {
            if (!scheduler.getPropertyId().equals(propertyId)) {
                throw new ForbiddenException(BaseErrorCodeEnum.DATA_FORBIDDEN.getI18nCode(),
                        BaseErrorCodeEnum.DATA_FORBIDDEN.getCode(),
                        MessageFormat.format("The Database scheduler not belong current property," +
                                        "schedulerId:{0},schedulerBelongPropertyId:{1},currentPropertyId:{2}",
                                scheduler.getId(), scheduler.getPropertyId(), propertyId));
            }
        }

        /**
         * 创建调度任务
         *
         * @param schedulerDTO 调度任务DTO
         * @param propertyDTO  资产DTO
         * @param operatorId   操作人 id
         * @return 调度任务id
         * @author zc
         * @date 2025/12/02
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public String save(DbSchedulerDTO schedulerDTO, PropertyDTO propertyDTO, int operatorId) {
            DbScheduler scheduler = beanMapper.map(schedulerDTO, DbScheduler.class, item -> {
                item.setRunTimes(BaseConstant.ZERO);
                item.setSyncState(schedulerDTO.getStatus() == SchedulerCons.STATUS_STOP);
                item.setUpdateLog(updateLogUtil.get(operatorId));
            });
            // 保存调度任务
            save(scheduler);
            String jobId = scheduler.getId();

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

        /**
         * 更新同步状态
         *
         * @param id        主键ID
         * @param syncState 同步状态
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public void updateSyncState(String id, boolean syncState) {
            updateById(new DbScheduler().setId(id).setSyncState(syncState));
        }

        /**
         * 修改调度任务
         *
         * @param propertyDTO     资产DTO
         * @param originScheduler 修改前的调度任务
         * @param schedulerDTO    修改后的调度任务DTO
         * @param operatorId      操作人 id
         * @return job id
         * @author zc
         * @date 2025/12/02
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public String update(PropertyDTO propertyDTO, DbScheduler originScheduler,
                             DbSchedulerDTO schedulerDTO, int operatorId) {
            DbScheduler scheduler = beanMapper.map(schedulerDTO, DbScheduler.class, item -> {
                item.setRunTimes(BaseConstant.ZERO);
                item.setSyncState(Boolean.FALSE);
                item.setUpdateLog(updateLogUtil.append(originScheduler.getUpdateLog(), operatorId));
            });

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

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

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

        /**
         * 校验分类下最大可配置调度任务的数量
         *
         * @param propertyId 资产Id
         * @param category   分类
         * @param maxNum     最大数量
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public void checkMaxNum(int propertyId, int category, int maxNum) {
            long count = count(new LambdaQueryWrapper<DbScheduler>()
                    .eq(DbScheduler::getPropertyId, propertyId)
                    .eq(DbScheduler::getCategory, category)
                    .ne(DbScheduler::getStatus, SchedulerCons.STATUS_DELETING));
            if (count >= maxNum) {
                throw new BadRequestException(CommonErrorI18nConstant.DATA_EXISTED,
                        BaseErrorCodeEnum.DATA_VALIDATION_ERROR.getCode(),
                        MessageFormat.format("The Database scheduler job maximum configuration:{0}, propertyId{1}, category:{2}",
                                maxNum, propertyId, category));
            }
        }

        /**
         * 刷新调度任务同步状态为已同步
         *
         * @param ids 调度任务ID集合
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public void refreshSyncStateByIds(List<String> ids) {
            update(new UpdateWrapper<DbScheduler>().lambda()
                    .in(DbScheduler::getId, ids)
                    .set(DbScheduler::getSyncState, Boolean.TRUE));
        }

        /**
         * 调度任务执行次数递增
         *
         * @param jobId 主键ID
         * @author zc
         * @date 2025/12/02
         */
        @Override
        public void plusRunTimes(String jobId) {
            baseMapper.plusRunTimes(jobId);
        }

        /**
         * 调用内部接口删除任务
         *
         * @param jobId 任务ID
         * @author zc
         * @date 2025/12/02
         */
        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 默认时区
         * @author zc
         * @date 2025/12/02
         */
        private void callInternalServiceLaunchJob(String originJobId, DbScheduler 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);
        }

        /**
         * 设置数据源
         *
         * @author zc
         * @date 2025/12/02
         */
        private void pushDynamicDataSource() {
            DynamicDataSourceContextHolder.push("database");
        }
    }