# RevenueJobWorker.java

    package cloud.wconnect.rateshop.scheduler.revenue;

    import cloud.wconnect.commonlib.cons.FunctionIdConstant;
    import cloud.wconnect.commonlib.cons.SchedulerCons;
    import cloud.wconnect.commonlib.service.CommonService;
    import cloud.wconnect.commonlib.service.MailInternalService;
    import cloud.wconnect.commonlib.util.RateLimitUtils;
    import cloud.wconnect.rateshop.businesstask.revenue.*;
    import cloud.wconnect.rateshop.cons.revenue.DailyRevenueNotifyTypeEnum;
    import cloud.wconnect.rateshop.cons.revenue.RevenueImportCons;
    import cloud.wconnect.rateshop.entity.revenue.RevenueSchedulerJob;
    import cloud.wconnect.rateshop.service.AmazonSqsSchedulerService;
    import cloud.wconnect.rateshop.service.revenue.IRevenueSchedulerJobService;
    import com.alibaba.fastjson.JSONObject;
    import jakarta.annotation.Resource;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    import software.amazon.awssdk.services.sqs.model.Message;

    import java.text.MessageFormat;
    import java.util.List;
    import java.util.Objects;

    /**
     * @author yyt
     * 2024-03-16
     */
    @Slf4j
    @Component
    public class RevenueJobWorker {

        @Autowired
        private AmazonSqsSchedulerService amazonSqsSchedulerService;
        @Autowired
        private MailInternalService mailInternalService;
        @Autowired
        private IRevenueSchedulerJobService schedulerJobService;
        @Autowired
        private RevenueHistoryAndForecastBusinessBlocksTask historyAndForecastBusinessBlocksTask;
        @Autowired
        private RevenueHistoryAndForecastNextMonthTask historyAndForecastNextMonthTask;
        @Autowired
        private RevenueHistoryAndForecastTask historyAndForecastTask;
        @Autowired
        private RevenueReservationForecastNextMonthTask reservationForecastNextMonthTask;
        @Autowired
        private RevenueReservationForecastTask reservationForecastTask;
        @Autowired
        private RevenueReservationStatisticsTask reservationStatisticsTask;
        @Autowired
        private RevenueHistoryAndForecastYesterdayAndNinetyTask revenueHistoryAndForecastYesterdayAndNinetyTask;
        @Autowired
        private RevenueDailySummaryMarketCodeTask revenueDailySummaryMarketCodeTask;
        @Autowired
        private RevenueReservationForecastNext14DaysTask revenueReservationForecastNext14DaysTask;
        @Autowired
        private RevenueOccJumpAlertTask revenueOccJumpAlertTask;
        @Autowired
        private RevenueHistoryAndForecastThisMonthFutureTask revenueHistoryAndForecastThisMonthFutureTask;
        @Autowired
        private RevenueHistoryAndForecastNextMonthFutureTask revenueHistoryAndForecastNextMonthFutureTask;
        @Autowired
        private RevenueHistoryAndForecastThisMonthAfterNextFutureTask revenueHistoryAndForecastThisMonthAfterNextFutureTask;
        @Autowired
        private RevenueReservationForecastNext90DaysTask revenueReservationForecastNext90DaysTask;
        @Autowired
        private RevenueHistoryAndForecastBusinessBlocksNext90DaysTask revenueHistoryAndForecastBusinessBlocksNext90DaysTask;
        @Autowired
        private RevenueDailySummaryManagerFlashTask revenueDailySummaryManagerFlashTask;
        @Autowired
        private RevenueDetailedAvailabilityNext14DaysTask revenueDetailedAvailabilityNext14DaysTask;
        @Autowired
        private RevenueRateReviewTask revenueRateReviewTask;
        @Autowired
        private RevenueDailySummaryGroupPickupTask revenueDailySummaryGroupPickupTask;
        @Autowired
        private RevenueHistoryAndForecastFromThirtyDaysAgoToYesterdayTask historyAndForecastFromThirtyDaysAgoToYesterdayTask;
        @Autowired
        private RevenueReservationForecastThisMonthTask reservationForecastThisMonthTask;
        @Autowired
        private RevenueOccupancySummaryYesterdayTask revenueOccupancySummaryYesterdayTask;
        @Autowired
        private RevenueMarketSegmentSummaryYesterdayTask revenueMarketSegmentSummaryYesterdayTask;
        @Autowired
        private RevenueMarketSegmentSummaryHistory30DaysTask revenueMarketSegmentSummaryHistory30DaysTask;
        @Resource
        private CommonService commonService;

        /**
         * 定时消息的队列 url
         */
        @Value("${sqs.scheduler.queue-url.revenue_scheduler}")
        private String schedulerUrl;

        /**
         * 消费功能队列任务
         *
         * @author yyt
         * @date 2024/3/16
         */
        @Scheduled(cron = "0 0/1 * * * ?", zone = "${default.timezone}")
        public void run() {
            List<Message> messages = amazonSqsSchedulerService.getMessagesBatch(schedulerUrl);
            if (CollectionUtils.isNotEmpty(messages)) {
                for (Message message : messages) {
                    // message为json字符串：{"jobId": "894bd865a742c84d9a74c190b1e3cb92"}, 该jobId 为 revenue_scheduler_job 表的id，该对应记录保存着任务的详细信息
                    // 警告: 执行任务的具体方法应该使用新的线程，而不是在本线程执行（要确保本线程不会被阻塞或异常退出）

                    RevenueSchedulerJob schedulerJob = null;
                    try {
                        JSONObject jsonObject = JSONObject.parseObject(message.body());
                        String jobId = (String) jsonObject.get(SchedulerCons.JOB_ID);
                        schedulerJob = schedulerJobService.getById(jobId);

                        if (Objects.nonNull(schedulerJob)) {
                            if (!commonService.checkPropertyEffectiveAndOpenFunction(schedulerJob.getPropertyId(), FunctionIdConstant.REVENUE)) {
                                return;
                            }

                            //获取分类
                            Integer category = schedulerJob.getCategory();
                            if (category.equals(RevenueImportCons.HISTORY_FORECAST)) {
                                historyAndForecastTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_FORECAST_BUSINESS_BLOCK)) {
                                historyAndForecastBusinessBlocksTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.RESERVATION_STATISTIC)) {
                                reservationStatisticsTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.RESERVATION_FORECAST)) {
                                reservationForecastTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.RESERVATION_FORECAST_NEXT_MONTH)) {
                                reservationForecastNextMonthTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_FORECAST_NEXT_MONTH)) {
                                historyAndForecastNextMonthTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_AND_FORECAST_YESTERDAY_AND_NEXT_NINETY_DAYS)) {
                                revenueHistoryAndForecastYesterdayAndNinetyTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.MARKET_CODE)) {
                                revenueDailySummaryMarketCodeTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.RESERVATION_FORECAST_NEXT_14_DAYS)) {
                                revenueReservationForecastNext14DaysTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(DailyRevenueNotifyTypeEnum.OCC_JUMP_ALERT.getType())) {
                                revenueOccJumpAlertTask.autoNotify(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_FORECAST_THIS_MONTH_FUTURE)) {
                                revenueHistoryAndForecastThisMonthFutureTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_FORECAST_NEXT_MONTH_FUTURE)) {
                                revenueHistoryAndForecastNextMonthFutureTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_FORECAST_THIS_MONTH_AFTER_NEXT_FUTURE)) {
                                revenueHistoryAndForecastThisMonthAfterNextFutureTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.RESERVATION_FORECAST_NEXT_90_DAYS)) {
                                revenueReservationForecastNext90DaysTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.HISTORY_FORECAST_BUSINESS_BLOCK_NEXT_90_DAYS)) {
                                revenueHistoryAndForecastBusinessBlocksNext90DaysTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.MANAGER_FLASH)) {
                                revenueDailySummaryManagerFlashTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.DETAILED_AVAILABILITY_NEXT_14_DAYS)) {
                                revenueDetailedAvailabilityNext14DaysTask.checkReportMissing(schedulerJob);
                            } else if (category.equals(RevenueImportCons.RATE_REVIEW)) {
                                revenueRateReviewTask.rateReviewNotify(schedulerJob);
                            } else if (category.equals(RevenueImportCons.GROUP_PICKUP_YESTERDAY)) {
                                revenueDailySummaryGroupPickupTask.checkReportMissing(schedulerJob);
                            } else if(category.equals(RevenueImportCons.HISTORY_AND_FORECAST_FROM_THIRTY_DAYS_AGO_TO_YESTERDAY)){
                                historyAndForecastFromThirtyDaysAgoToYesterdayTask.checkReportMissing(schedulerJob);
                            } else if(category.equals(RevenueImportCons.RESERVATION_FORECAST_THIS_MONTH)){
                                reservationForecastThisMonthTask.checkReportMissing(schedulerJob);
                            } else if(category.equals(RevenueImportCons.OCCUPANCY_SUMMARY_YESTERDAY)){
                                revenueOccupancySummaryYesterdayTask.checkReportMissing(schedulerJob);
                            } else if(category.equals(RevenueImportCons.MARKET_SEGMENT_SUMMARY_YESTERDAY)){
                                revenueMarketSegmentSummaryYesterdayTask.checkReportMissing(schedulerJob);
                            } else if(category.equals(RevenueImportCons.MARKET_SEGMENT_SUMMARY_HISTORY_30_DAYS)){
                                revenueMarketSegmentSummaryHistory30DaysTask.checkReportMissing(schedulerJob);
                            }

                            //任务执行次数加1
                            schedulerJobService.plusRunTimes(jobId);
                        }
                    } catch (Exception e) {
                        log.error("SQS consume error[RevenueJobWorker] - Message: {}; Exception: {}", message.body(), e.getMessage());
                        if (RateLimitUtils.tryAcquire(this)) {
                            sendAlertEmail(schedulerUrl, message.body(), schedulerJob, e.getMessage());
                        }
                    } finally {
                        // 删除消息
                        amazonSqsSchedulerService.deleteMessage(schedulerUrl, message.receiptHandle());
                    }
                }
                run();
            }
        }

        /**
         * 发送alertEmail
         *
         * @param schedulerUrl 定时任务调度URL
         * @param message      消息
         * @param schedulerJob 定时任务对象
         * @param exception    异常信息
         * @author yyt
         * @date 2024/3/16
         */
        public void sendAlertEmail(String schedulerUrl, String message, RevenueSchedulerJob schedulerJob, String exception) {
            // 主题
            String subject = "[Warning] - process RevenueJobWorker";
            // 邮件内容
            String mailContent = MessageFormat.format("SQS url:{0}<br/>Message:{1}<br/>SchedulerJob:{2}<br/>Exception:{3}",
                    schedulerUrl, message, Objects.nonNull(schedulerJob) ? schedulerJob.toString() : "", exception);
            // 发送 alert
            mailInternalService.sendAlertEmail(subject, mailContent);
        }
    }