content: package com.aizuda.snailjob.server.job.task.support.schedule;
  
  import cn.hutool.core.collection.CollUtil;
  import com.aizuda.snailjob.common.core.enums.JobTaskBatchStatusEnum;
  import com.aizuda.snailjob.common.core.util.JsonUtil;
  import com.aizuda.snailjob.common.core.util.StreamUtils;
  import com.aizuda.snailjob.common.log.SnailJobLog;
  import com.aizuda.snailjob.server.common.Lifecycle;
  import com.aizuda.snailjob.server.common.config.SystemProperties;
  import com.aizuda.snailjob.server.common.dto.JobTaskBatchReason;
  import com.aizuda.snailjob.server.common.enums.SyetemTaskTypeEnum;
  import com.aizuda.snailjob.server.common.schedule.AbstractSchedule;
  import com.aizuda.snailjob.server.common.triple.Pair;
  import com.aizuda.snailjob.template.datasource.persistence.dataobject.JobBatchSummaryResponseDO;
  import com.aizuda.snailjob.template.datasource.persistence.mapper.JobMapper;
  import com.aizuda.snailjob.template.datasource.persistence.mapper.JobSummaryMapper;
  import com.aizuda.snailjob.template.datasource.persistence.mapper.JobTaskBatchMapper;
  import com.aizuda.snailjob.template.datasource.persistence.po.JobSummary;
  import com.aizuda.snailjob.template.datasource.persistence.po.JobTaskBatch;
  import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
  import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
  import com.google.common.collect.Lists;
  import lombok.RequiredArgsConstructor;
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Component;
  
  import java.time.Duration;
  import java.time.LocalDate;
  import java.time.LocalDateTime;
  import java.time.LocalTime;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.stream.Collectors;
  
  /**
   * Job Dashboard
   *
   * @author: wodeyangzipingpingwuqi
   * @date : 2023-11-21 11:15
   * @since 2.5.0
   */
  @Component
  @RequiredArgsConstructor
  public class JobSummarySchedule extends AbstractSchedule implements Lifecycle {
      private final JobTaskBatchMapper jobTaskBatchMapper;
      private final JobSummaryMapper jobSummaryMapper;
      private final JobMapper jobMapper;
      private final SystemProperties systemProperties;
  
      @Override
      public String lockName() {
          return "jobSummaryDashboard";
      }
  
      @Override
      public String lockAtMost() {
          return "PT1M";
      }
  
      @Override
      public String lockAtLeast() {
          return "PT20S";
      }
  
      @Override
      protected void doExecute() {
          try {
              for (int i = 0; i < systemProperties.getSummaryDay(); i++) {
  
                  // 定时按日实时查询统计数据（00:00:00 - 23:59:59）
                  LocalDateTime todayFrom = LocalDateTime.of(LocalDate.now(), LocalTime.MIN).plusDays(-i);
                  LocalDateTime todayTo = LocalDateTime.of(LocalDate.now(), LocalTime.MAX).plusDays(-i);
                  LambdaQueryWrapper<JobTaskBatch> wrapper = new LambdaQueryWrapper<JobTaskBatch>()
                          .eq(JobTaskBatch::getSystemTaskType, SyetemTaskTypeEnum.JOB.getType())
                          .between(JobTaskBatch::getCreateDt, todayFrom, todayTo)
                          .groupBy(JobTaskBatch::getNamespaceId, JobTaskBatch::getGroupName,
                                  JobTaskBatch::getJobId, JobTaskBatch::getTaskBatchStatus, JobTaskBatch::getOperationReason);
                  List<JobBatchSummaryResponseDO> summaryResponseDOList = jobTaskBatchMapper.selectJobBatchSummaryList(wrapper);
                  if (CollUtil.isEmpty(summaryResponseDOList)) {
                      continue;
                  }
  
                  // insertOrUpdate
                  List<JobSummary> jobSummaryList = jobSummaryList(todayFrom, summaryResponseDOList);
  
                  List<JobSummary> jobSummaries = jobSummaryMapper.selectList(new LambdaQueryWrapper<JobSummary>()
                          .eq(JobSummary::getTriggerAt, todayFrom)
                          .eq(JobSummary::getSystemTaskType, SyetemTaskTypeEnum.JOB.getType())
                          .in(JobSummary::getBusinessId, StreamUtils.toSet(jobSummaryList, JobSummary::getBusinessId)));
  
                  Map<Pair<Long, LocalDateTime>, JobSummary> summaryMap = StreamUtils.toIdentityMap(jobSummaries,
                          jobSummary -> Pair.of(jobSummary.getBusinessId(), jobSummary.getTriggerAt()));
  
                  List<JobSummary> waitInserts = Lists.newArrayList();
                  List<JobSummary> waitUpdates = Lists.newArrayList();
                  for (final JobSummary jobSummary : jobSummaryList) {
                      if (Objects.isNull(
                              summaryMap.get(Pair.of(jobSummary.getBusinessId(), jobSummary.getTriggerAt())))) {
                          waitInserts.add(jobSummary);
                      } else {
                          waitUpdates.add(jobSummary);
                      }
                  }
  
                  int updateTotalJobSummary = 0;
                  if (CollUtil.isNotEmpty(waitUpdates)) {
                      updateTotalJobSummary = jobSummaryMapper.updateBatch(waitUpdates);
                  }
  
                  int insertTotalJobSummary = 0;
                  if (CollUtil.isNotEmpty(waitInserts)) {
                      insertTotalJobSummary = jobSummaryMapper.insertBatch(waitInserts);
                  }
  
                  SnailJobLog.LOCAL.debug(
                          "job summary dashboard success todayFrom:[{}] todayTo:[{}] updateTotalJobSummary:[{}] insertTotalJobSummary:[{}]",
                          todayFrom, todayTo, updateTotalJobSummary, insertTotalJobSummary);
              }
          } catch (Exception e) {
              SnailJobLog.LOCAL.error("job summary dashboard log error", e);
          }
      }
  
      private List<JobSummary> jobSummaryList(LocalDateTime triggerAt,
                                              List<JobBatchSummaryResponseDO> summaryResponseDOList) {
          List<JobSummary> jobSummaryList = new ArrayList<>();
          Map<Long, List<JobBatchSummaryResponseDO>> jobIdListMap = summaryResponseDOList.parallelStream()
                  .collect(Collectors.groupingBy(JobBatchSummaryResponseDO::getJobId));
          for (Map.Entry<Long, List<JobBatchSummaryResponseDO>> job : jobIdListMap.entrySet()) {
              JobSummary jobSummary = new JobSummary();
              jobSummary.setBusinessId(job.getKey());
              jobSummary.setTriggerAt(triggerAt);
              jobSummary.setNamespaceId(job.getValue().get(0).getNamespaceId());
              jobSummary.setGroupName(job.getValue().get(0).getGroupName());
              jobSummary.setSystemTaskType(SyetemTaskTypeEnum.JOB.getType());
              jobSummary.setSuccessNum(job.getValue().stream().mapToInt(JobBatchSummaryResponseDO::getSuccessNum).sum());
              jobSummary.setFailNum(job.getValue().stream().mapToInt(JobBatchSummaryResponseDO::getFailNum).sum());
              jobSummary.setStopNum(job.getValue().stream().mapToInt(JobBatchSummaryResponseDO::getStopNum).sum());
              jobSummary.setCancelNum(job.getValue().stream().mapToInt(JobBatchSummaryResponseDO::getCancelNum).sum());
  
              jobSummary.setFailReason(
                      JsonUtil.toJsonString(jobTaskBatchReasonList(JobTaskBatchStatusEnum.FAIL.getStatus(), job.getValue())));
              jobSummary.setStopReason(
                      JsonUtil.toJsonString(jobTaskBatchReasonList(JobTaskBatchStatusEnum.STOP.getStatus(), job.getValue())));
              jobSummary.setCancelReason(JsonUtil.toJsonString(
                      jobTaskBatchReasonList(JobTaskBatchStatusEnum.CANCEL.getStatus(), job.getValue())));
              jobSummaryList.add(jobSummary);
          }
          return jobSummaryList;
      }
  
      /**
       * 批次状态查询 (操作原因 && 总数)
       *
       * @param jobTaskBatchStatus
       * @param jobBatchSummaryResponseDOList
       * @return
       */
      private List<JobTaskBatchReason> jobTaskBatchReasonList(int jobTaskBatchStatus,
                                                              List<JobBatchSummaryResponseDO> jobBatchSummaryResponseDOList) {
          List<JobTaskBatchReason> jobTaskBatchReasonArrayList = new ArrayList<>();
          List<JobBatchSummaryResponseDO> summaryResponseDOList = jobBatchSummaryResponseDOList.stream()
                  .filter(i -> jobTaskBatchStatus == i.getTaskBatchStatus()).collect(Collectors.toList());
          for (JobBatchSummaryResponseDO jobBatchSummaryResponseDO : summaryResponseDOList) {
              JobTaskBatchReason jobTaskBatchReason = new JobTaskBatchReason();
              jobTaskBatchReason.setReason(jobBatchSummaryResponseDO.getOperationReason());
              jobTaskBatchReason.setTotal(jobBatchSummaryResponseDO.getOperationReasonTotal());
              jobTaskBatchReasonArrayList.add(jobTaskBatchReason);
          }
          return jobTaskBatchReasonArrayList;
      }
  
      @Override
      public void start() {
          taskScheduler.scheduleAtFixedRate(this::execute, Duration.parse("PT1M"));
      }
  
      @Override
      public void close() {
  
      }
  }
  