# IwcSqsServiceImpl.java

    package cloud.wconnect.rateshop.service.rateshop.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.rateshop.entity.rateshop.Setting;
    import cloud.wconnect.rateshop.entity.rateshop.TrendSourceSetting;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.PriceCacheDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.PriceCacheKeyDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PriceMessage;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.sender.CalculateIwcContentByOTA;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.sender.CalculateIwcContentByPriceChange;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.sender.CalculateIwcMessage;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.sender.MsgMinPrice;
    import cloud.wconnect.rateshop.service.AmazonSqsService;
    import cloud.wconnect.rateshop.service.rateshop.IIwcSqsService;
    import cloud.wconnect.rateshop.service.rateshop.IPropertyHotelService;
    import cloud.wconnect.rateshop.service.rateshop.ISettingService;
    import cloud.wconnect.rateshop.service.rateshop.ITrendSourceSettingService;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.google.common.collect.Lists;
    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.data.redis.core.RedisCallback;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Service;

    import java.math.BigDecimal;
    import java.time.LocalDate;
    import java.time.temporal.ChronoUnit;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.Objects;
    import java.util.stream.Collectors;

    /**
     * IWC SQS服务
     *
     * @author lichen
     * @date 2025/4/2
     */
    @Service
    @DS("rate_shop")
    public class IwcSqsServiceImpl implements IIwcSqsService {

        @Autowired
        private IPropertyHotelService propertyHotelService;
        @Autowired
        private ISettingService settingService;
        @Autowired
        private BeanMapper beanMapper;
        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private AmazonSqsService amazonSqsService;
        @Autowired
        private RedisTemplate<String, String> redisTemplate;
        @Value("${sqs.queue-url.iwc_queue}")
        private String iwcUrl;
        @Autowired
        private ITrendSourceSettingService trendSourceSettingService;

        /**
         * 发送计算IWC消息
         *
         * @param priceMessage               房价消息
         * @param containTaxFlag2LowestPrice 房价消息中的最低价格
         * @return void
         * @author lichen
         * @date 2025/4/2
         */
        @Override
        public void sendCalculateIWCMessage(PriceMessage priceMessage, Map<Integer, List<MsgMinPrice>> containTaxFlag2LowestPrice) {
            //获取酒店作为自身对应的资产ID
            Integer propertyId = propertyHotelService.getPropertyIdBySelfHotelId(priceMessage.getHotelId());
            if (Objects.isNull(propertyId)) {
                return;
            }

            //资产rate shop功能的配置
            Setting setting = settingService.getByPropertyId(propertyId);
            //含税标志
            Integer containTaxFlag = setting.getContainTaxFlag();

            //指定含税标识的最低房价集合
            List<MsgMinPrice> msgMinPrices = containTaxFlag2LowestPrice.get(containTaxFlag);

            ArrayList<PriceCacheDTO> refreshCacheList = Lists.newArrayListWithCapacity(msgMinPrices.size());

            for (MsgMinPrice msgMinPrice : msgMinPrices) {
                //消息中的最低价格
                BigDecimal minPriceByMsg = msgMinPrice.getMinPrice();

                //缓存key对象
                PriceCacheKeyDTO priceCacheKeyDTO = beanMapper.map(msgMinPrice, PriceCacheKeyDTO.class);

                //最低价格缓存key
                String lowestPriceCacheKey = priceCacheKeyDTO.generateLowestRateCacheKey();
                //最低价格缓存value
                String cacheMinPriceStr = redisTemplate.opsForValue().get(lowestPriceCacheKey);
                BigDecimal cacheMinPrice = StringUtils.isNotBlank(cacheMinPriceStr) ? new BigDecimal(cacheMinPriceStr) : null;

                //有效最低价格缓存value
                String validLowestPriceCacheKey = priceCacheKeyDTO.generateValidLowestRateCacheKey();
                String validLowestPriceCacheValueStr = redisTemplate.opsForValue().get(validLowestPriceCacheKey);
                BigDecimal validLowestPriceCacheValue = StringUtils.isNotBlank(validLowestPriceCacheValueStr) ? new BigDecimal(validLowestPriceCacheValueStr) : null;

                //最低价格调涨次数缓存value
                String raiseLowestPriceTimesCacheKey = priceCacheKeyDTO.generateLowestRateRiseTimesCacheKey();
                String raiseLowestPriceTimesCacheValueStr = redisTemplate.opsForValue().get(raiseLowestPriceTimesCacheKey);
                int raiseLowestPriceTimesCacheValue = StringUtils.isNotBlank(raiseLowestPriceTimesCacheValueStr) ? Integer.parseInt(raiseLowestPriceTimesCacheValueStr) : 0;

                //计算入住日期和当前时间的差值
                LocalDate checkInDateLocalDate = DateTimeUtils.strToLocalDate(msgMinPrice.getCheckInDate().toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);
                // 比较两个日期之间的天数差
                long diffDays = ChronoUnit.DAYS.between(checkInDateLocalDate, LocalDate.now());
                //最低房价存储到缓存中的过期时间 = 时间差值 + 1天
                long cacheExpirationTime = Math.abs(diffDays) + 1;

                //缓存中的有效最低价格为null
                if (Objects.isNull(validLowestPriceCacheValue)) {
                    //消息中的价格不为null
                    if (Objects.nonNull(minPriceByMsg)) {
                        //发送计算IWC的消息
                        doSendCalculateIWCMessageByPriceChange(propertyId, msgMinPrice);
                        if (minPriceByMsg.compareTo(BigDecimal.ZERO) > 0) {
                            //更新缓存有效最低房价
                            refreshCacheList.add(new PriceCacheDTO(validLowestPriceCacheKey, minPriceByMsg.toString(), cacheExpirationTime));
                        }
                        //else-消息中的最低价格等于0时表示sold out，不用更新缓存（有效最低房价和调涨次数）
                    }
                    //else-消息中的最低价格为null时，不发送计算IWC的消息，也不用更新缓存（有效最低房价和调涨次数）
                } else {//缓存中的有效最低价格不为null
                    //消息中的价格不为null
                    if (Objects.nonNull(minPriceByMsg)) {
                        //消息中的房价为Sold out
                        if (minPriceByMsg.compareTo(BigDecimal.ZERO) == 0) {
                            //非连续Sold out时（原最低房价为null或大于0，表示当前非连续sold out）
                            if(Objects.isNull(cacheMinPrice) || cacheMinPrice.compareTo(BigDecimal.ZERO) > 0){
                                //发送计算IWC的消息，不用更新缓存（有效最低房价和调涨次数）
                                doSendCalculateIWCMessageByPriceChange(propertyId, msgMinPrice);
                            }
                        }else{
                            //消息中的最低价格大于缓存中的有效最低价格时，不发送计算IWC的消息，只更新缓存
                            if (minPriceByMsg.compareTo(validLowestPriceCacheValue) > 0) {
                                //调涨次数加1
                                raiseLowestPriceTimesCacheValue++;
                                //发送计算IWC的消息
                                doSendCalculateIWCMessageByPriceChange(propertyId, msgMinPrice);
                                //更新缓存有效最低房价
                                refreshCacheList.add(new PriceCacheDTO(validLowestPriceCacheKey, minPriceByMsg.toString(), cacheExpirationTime));
                                //更新缓存中的调涨次数
                                refreshCacheList.add(new PriceCacheDTO(raiseLowestPriceTimesCacheKey, Integer.toString(raiseLowestPriceTimesCacheValue), cacheExpirationTime));
                            } else {
                                //消息中的最低价格小于或等于缓存中的有效最低价格时，不发送计算IWC的消息，只更新缓存中的有效最低价格
                                refreshCacheList.add(new PriceCacheDTO(validLowestPriceCacheKey, minPriceByMsg.toString(), cacheExpirationTime));
                            }
                        }
                    }
                    //else-消息中的最低价格为null时，不用更新缓存（有效最低房价和调涨次数）
                }
            }

            //更新缓存
            if(CollectionUtils.isNotEmpty(refreshCacheList)){
                Map<String, String> cacheKey2Value = refreshCacheList.stream().collect(Collectors.toMap(PriceCacheDTO::getKey, PriceCacheDTO::getValue));
                Map<String, Long> cacheKey2ExpireTime = refreshCacheList.stream().collect(Collectors.toMap(PriceCacheDTO::getKey, PriceCacheDTO::getExpireTime));
                redisTemplate.opsForValue().multiSet(cacheKey2Value);
                //用一次连接，设置过期时间
                redisTemplate.executePipelined((RedisCallback<Void>) connection -> {
                    for (String key : cacheKey2Value.keySet()) {
                        //过期时间单位天
                        Long expirationTime = cacheKey2ExpireTime.get(key);
                        //设置过期时间
                        connection.expire(key.getBytes(), 3600L * 24L * expirationTime);
                    }
                    return null;
                });
            }
        }

        /**
         * 发送计算IWC消息
         *
         * @param trendLocationId 需求趋势的地址ID
         * @param checkInDateList 入住日期集合（yyyyMMdd）
         * @return void
         * @author lichen
         * @date 2025/4/2
         */
        @Override
        public void sendCalculateIWCMessage(Integer trendLocationId, List<Long> checkInDateList) {
            List<TrendSourceSetting> trendSourceSettings = trendSourceSettingService.listByTrendLocationId(trendLocationId);
            if (CollectionUtils.isEmpty(trendSourceSettings)) return;

            //入住日期按照年份分组
            Map<String, List<Long>> year2CheckInDateList = checkInDateList.stream().collect(Collectors.groupingBy(checkInDate -> checkInDate.toString().substring(0, 4)));

            for (Map.Entry<String, List<Long>> entry : year2CheckInDateList.entrySet()) {
                String year = entry.getKey();
                List<Long> checkInDateListByYear = entry.getValue();

                //消息内容
                CalculateIwcContentByOTA calculateIwcContentByOTA = new CalculateIwcContentByOTA(year, trendLocationId, checkInDateListByYear);
                String content = JsonUtils.convertObjToString(objectMapper, calculateIwcContentByOTA);

                for (TrendSourceSetting trendSourceSetting : trendSourceSettings) {
                    //资产ID
                    Integer propertyId = trendSourceSetting.getPropertyId();
                    //消息对象
                    CalculateIwcMessage calculateIwcMessage = new CalculateIwcMessage(propertyId, CalculateIwcMessage.OTA, content);
                    //发送消息
                    amazonSqsService.sendMessage(iwcUrl, JsonUtils.convertObjToString(objectMapper, calculateIwcMessage));
                }
            }
        }

        /**
         * 价格变化时，发送计算IWC的消息
         *
         * @param propertyId     资产ID
         * @param msgMinPriceObj 消息中的价格对象
         * @return void
         * @author lichen
         * @date 2025/4/2
         */
        private void doSendCalculateIWCMessageByPriceChange(Integer propertyId, MsgMinPrice msgMinPriceObj) {
            //消息内容
            CalculateIwcContentByPriceChange calculateIwcContentByPriceChange = beanMapper.map(msgMinPriceObj, CalculateIwcContentByPriceChange.class);
            //消息对象
            CalculateIwcMessage calculateIwcMessage = new CalculateIwcMessage(propertyId, CalculateIwcMessage.PRICE_CHANGE,
                    JsonUtils.convertObjToString(objectMapper, calculateIwcContentByPriceChange));
            //发送消息
            amazonSqsService.sendMessage(iwcUrl, JsonUtils.convertObjToString(objectMapper, calculateIwcMessage));
        }
    }