# NotifySupplementaryPersonnelApprovalTask.java

    package cloud.wconnect.frontoffice.businesstask.adjustment;

    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.cons.FunctionIdConstant;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.pojo.query.OpenFunctionPropertyQuery;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.service.PropertyInternalService;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.frontoffice.entity.adjustment.AdRequest;
    import cloud.wconnect.frontoffice.entity.adjustment.AdjApproveConfig;
    import cloud.wconnect.frontoffice.entity.adjustment.AdjApproveLog;
    import cloud.wconnect.frontoffice.entity.adjustment.AdjOutOfOfficeConfig;
    import cloud.wconnect.frontoffice.service.adjustment.*;
    import com.google.common.collect.Lists;
    import com.google.common.util.concurrent.RateLimiter;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.collections.MapUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Service;

    import java.util.*;
    import java.util.concurrent.TimeUnit;
    import java.util.stream.Collectors;

    import static java.util.stream.Collectors.*;

    /**
     * @author: yyt
     * 2026-03-19
     */
    @Service
    @Slf4j
    public class NotifySupplementaryPersonnelApprovalTask {

        @Autowired
        private IAdjOutOfOfficeConfigService adjOutOfOfficeConfigService;
        @Autowired
        private IAdjApproveConfigService adjApproveConfigService;
        @Autowired
        private IAdRequestService adRequestService;
        @Autowired
        private IAdjApproveLogService adjApproveLogService;
        @Autowired
        private IAdjSendNotifyService adjSendNotifyService;
        @Autowired
        private IAdjCheckUserEffectiveService adjCheckUserEffectiveService;
        @Autowired
        private PropertyInternalService propertyInternalService;
        @Autowired
        private RedisTemplate<String, Object> redisTemplate;
        @Autowired
        private MailInternalService mailInternalService;

        @Value("${default.timezone}")
        private String defaultTimezone;

        private static final String PREFIX_KEY = "wconnect:task:adj-notify-supplementary-personnel:";

        /**
         * 限制速率（每秒5次）
         */
        private final RateLimiter rateLimiter = RateLimiter.create(5.0);

        /**
         * notify supplementary personnel
         *
         * @author yyt
         * @date 2026/3/19
         **/
        @Scheduled(cron = "${schedule.cron.notify.adj-notify-supplementary-personnel}", zone = "${default.timezone}")
        public void notifySupplementaryPersonnel() {
            String currentDateStr = DateTimeUtils.dateNowStr(defaultTimezone, BaseConstant.DATE_FORMAT_YYMMDDHH);
            if (!preventConcurrency(PREFIX_KEY + currentDateStr)) {
                return;
            }

            log.info("NotifySupplementaryPersonnelApprovalTask... start");
            String alertTitle = "[Warning] - Exceptions About Notify Supplementary Personnel Approval";
            StringBuilder warningEmail = new StringBuilder("<p>There is something wrong when notify supplementary personnel approval.</p>");

            try {
                // 查询开通此功能 的资产
                List<PropertyDTO> openModuleProperty = propertyInternalService.getOpenFunctionProperty(OpenFunctionPropertyQuery.builder()
                        .openFunctionIds(Lists.newArrayList(FunctionIdConstant.ADJUSTMENT_REQUESTS)).build())
                        .stream().filter(PropertyDTO::getStatus).toList();
                // 循环处理每个资产的数据
                for (PropertyDTO propertyDTO : openModuleProperty) {
                    Integer propertyId = propertyDTO.getId();

                    // 查询刚进入请假的补位配置，若没有请假的，则 return
                    List<AdjOutOfOfficeConfig> adjOutOfOfficeConfigs = adjOutOfOfficeConfigService.listJustStartedTakingLeaveConfig(propertyDTO);
                    if (CollectionUtils.isEmpty(adjOutOfOfficeConfigs)) {
                        log.info("NotifySupplementaryPersonnelApprovalTask... people who have not taken leave, propertyId: {}", propertyId);
                        continue;
                    }

                    handleNotify(propertyDTO, adjOutOfOfficeConfigs);
                }
            } catch (Exception e) {
                warningEmail.append("<p>Exception: ").append(e.getMessage()).append("</p>");
                mailInternalService.sendAlertEmail(alertTitle, warningEmail.toString());
                log.error("NotifySupplementaryPersonnelApprovalTask... error", e);
            }

            log.info("NotifySupplementaryPersonnelApprovalTask... end");
        }

        /**
         * 处理通知
         *
         * @param propertyDTO 资产
         * @param adjOutOfOfficeConfigs 补位配置
         * @author yyt
         * @date 2026/3/19
         **/
        private void handleNotify(PropertyDTO propertyDTO, List<AdjOutOfOfficeConfig> adjOutOfOfficeConfigs) {
            // 查询现有的审批层级
            Integer propertyId = propertyDTO.getId();
            List<AdjApproveConfig> adjApproveConfigs = adjApproveConfigService.listByUserIds(propertyId,
                    adjOutOfOfficeConfigs.stream().map(AdjOutOfOfficeConfig::getUpdatedBy).toList());
            if (CollectionUtils.isEmpty(adjApproveConfigs)) return;

            // 查询审批层级对应的审批中的 adj
            List<Long> levelIds = adjApproveConfigs.stream().map(AdjApproveConfig::getLevelId).toList();
            List<AdRequest> adRequests = adRequestService.listApprovingByLevels(propertyId, levelIds);
            if (CollectionUtils.isEmpty(adRequests)) return;

            // 查询 adj 对应的当前层级审批日志
            Map<Long, Map<Long, Set<Integer>>> alreadyApprovedUserIdsByLevelId = adjApproveLogService.listApproveRecord(propertyId, adRequests)
                    .stream().collect(Collectors.groupingBy(AdjApproveLog::getLevelId,
                            Collectors.groupingBy(AdjApproveLog::getAdjId,
                                    Collectors.mapping(AdjApproveLog::getApproveBy, toSet()))));

            // 请假人员分组，补位人员对应多个请假人
            Map<Integer, Map<Integer, List<Integer>>> approversBySubstitutePersonnelId = adjOutOfOfficeConfigs.stream()
                    .collect(Collectors.groupingBy(AdjOutOfOfficeConfig::getSubstitutePersonnelId,
                            Collectors.groupingBy(AdjOutOfOfficeConfig::getType,
                                    Collectors.mapping(AdjOutOfOfficeConfig::getUpdatedBy, toList()))));
            // 审批树上的审批人
            Map<Integer, List<AdjApproveConfig>> approverUserIdsByType = adjApproveConfigs.stream()
                    .collect(Collectors.groupingBy(AdjApproveConfig::getType));
            // 不同类型下的 adj
            Map<Integer, Map<Long, List<AdRequest>>> adRequestsByType = adRequests.stream()
                    .collect(Collectors.groupingBy(AdRequest::getRequestType,
                            Collectors.groupingBy(AdRequest::getApproveLevelId)));
            approversBySubstitutePersonnelId.forEach((substitutePersonnelId, approversByType) -> {
                // 获取令牌后执行后续逻辑
                rateLimiter.acquire();

                // 汇总需要通知的 adj
                List<AdRequest> notifyAdRequests = collectNotifyAdRequests(approversByType, approverUserIdsByType, adRequestsByType, alreadyApprovedUserIdsByLevelId);
                if (CollectionUtils.isNotEmpty(notifyAdRequests)) {
                    // 给补位人员发通知
                    notify(propertyDTO, notifyAdRequests, substitutePersonnelId);
                }
            });
        }

        /**
         * 汇总需要通知的 adj
         *
         * @param approversByType                 请假人员对应的审批人
         * @param approverUserIdsByType           审批层级对应的审批人
         * @param adRequestsByType                adj 对应的审批层级
         * @param alreadyApprovedUserIdsByLevelId 已审批的 adj 对应的审批层级
         * @return 需要通知的 adj
         * @author yyt
         * @date 2026/3/19
         **/
        private List<AdRequest> collectNotifyAdRequests(Map<Integer, List<Integer>> approversByType,
                                                        Map<Integer, List<AdjApproveConfig>> approverUserIdsByType,
                                                        Map<Integer, Map<Long, List<AdRequest>>> adRequestsByType,
                                                        Map<Long, Map<Long, Set<Integer>>> alreadyApprovedUserIdsByLevelId) {
            // 根据请假人的审批层级，过滤出对应层级下的 adj，
            // 然后看看当前请假人是否已经审批过，若没有，则汇总此adj给补位人员发通知
            List<AdRequest> notifyAdRequests = new ArrayList<>();
            approversByType.forEach((type, approverUserIds) -> {
                // 当前类型下的审批人
                List<AdjApproveConfig> currentTypeApproverUserIds = approverUserIdsByType.getOrDefault(type, Collections.emptyList())
                        .stream().filter(item -> approverUserIds.contains(item.getUserId())).toList();
                if (CollectionUtils.isEmpty(currentTypeApproverUserIds)) return;
                // 当前类型下的 adj
                Map<Long, List<AdRequest>> adRequestsByLevel = adRequestsByType.getOrDefault(type, Collections.emptyMap());
                if (MapUtils.isEmpty(adRequestsByLevel)) return;

                Map<Long, List<Integer>> approverUserIdsByLevel = currentTypeApproverUserIds.stream()
                        .collect(Collectors.groupingBy(AdjApproveConfig::getLevelId,
                                Collectors.mapping(AdjApproveConfig::getUserId, toList())));
                approverUserIdsByLevel.forEach((levelId, currentLevelApproverUserIds) -> {
                    // 当前层级下的 adj
                    List<AdRequest> currentLevelAdRequests = adRequestsByLevel.getOrDefault(levelId, Collections.emptyList());
                    if (CollectionUtils.isEmpty(currentLevelAdRequests)) return;
                    // 当前层级下 adj 的已审批过的审批人
                    Map<Long, Set<Integer>> alreadyApprovedUserIdsByAdj = alreadyApprovedUserIdsByLevelId.getOrDefault(levelId, Collections.emptyMap());

                    currentLevelAdRequests.forEach(adRequest -> {
                        // 看看当前补位人员对应的审批人是否都已审批过，若都审批过，则当前补位人员无需再审批，不需再给补位人员通知
                        if (!alreadyApprovedUserIdsByAdj.getOrDefault(adRequest.getId(), Collections.emptySet())
                                .containsAll(currentLevelApproverUserIds)) {
                            notifyAdRequests.add(adRequest);
                        }
                    });
                });
            });
            return notifyAdRequests;
        }

        /**
         * 给补位人员发通知
         *
         * @param propertyDTO      资产
         * @param notifyAdRequests adj
         * @author yyt
         * @date 2026/3/19
         **/
        private void notify(PropertyDTO propertyDTO, List<AdRequest> notifyAdRequests, Integer substitutePersonnelId) {
            Map<Integer, List<AdRequest>> adRequestsByType = notifyAdRequests.stream().collect(groupingBy(AdRequest::getRequestType));
            adRequestsByType.forEach((type, adRequests) ->
                    adjSendNotifyService.notifyTargetUser(propertyDTO, type, adRequests, substitutePersonnelId));
        }

        /**
         * 阻止并发操作
         *
         * @param key 缓存key
         * @return java.lang.Boolean
         * @author yyt
         * @date 2026/3/19
         */
        private Boolean preventConcurrency(String key) {
            return redisTemplate.opsForValue().setIfAbsent(key, 0, 1, TimeUnit.DAYS);
        }

    }