raex_back/src/main/java/com/izouma/nineth/service/CollectionService.java

package com.izouma.nineth.service;

import com.alibaba.fastjson.JSON;
import com.izouma.nineth.TokenHistory;
import com.izouma.nineth.annotations.Debounce;
import com.izouma.nineth.config.GeneralProperties;
import com.izouma.nineth.config.RedisKeys;
import com.izouma.nineth.converter.LongArrayConverter;
import com.izouma.nineth.domain.Collection;
import com.izouma.nineth.domain.*;
import com.izouma.nineth.dto.*;
import com.izouma.nineth.enums.AuthStatus;
import com.izouma.nineth.enums.CollectionSource;
import com.izouma.nineth.enums.CollectionType;
import com.izouma.nineth.enums.OrderStatus;
import com.izouma.nineth.exception.BusinessException;
import com.izouma.nineth.repo.*;
import com.izouma.nineth.utils.JpaUtils;
import com.izouma.nineth.utils.SecurityUtils;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.Range;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.core.env.Environment;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.RequestParam;

import javax.annotation.PostConstruct;
import javax.persistence.criteria.Predicate;
import javax.transaction.Transactional;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Service
@AllArgsConstructor
@Slf4j
public class CollectionService {

    private CollectionRepo                collectionRepo;
    private LikeRepo                      likeRepo;
    private BlindBoxItemRepo              blindBoxItemRepo;
    private AppointmentRepo               appointmentRepo;
    private UserRepo                      userRepo;
    private TaskScheduler                 taskScheduler;
    private CacheService                  cacheService;
    private RedisTemplate<String, Object> redisTemplate;
    private RocketMQTemplate              rocketMQTemplate;
    private GeneralProperties             generalProperties;
    private Environment                   env;
    private OrderRepo                     orderRepo;
    private TokenHistoryRepo              tokenHistoryRepo;

    private final Map<Long, ScheduledFuture<?>> tasks = new HashMap<>();

    @PostConstruct
    public void init() {
        if (Arrays.asList(env.getActiveProfiles()).contains("dev")) {
            return;
        }
        List<Collection> collections = collectionRepo.findByScheduleSaleTrueAndOnShelfFalseAndStartTimeBeforeAndDelFalse(LocalDateTime.now());
        for (Collection collection : collections) {
            onShelfTask(collection);
        }
    }

    @Cacheable(value = "collectionList", key = "#pageQuery.hashCode()")
    public PageWrapper<Collection> all(PageQuery pageQuery) {
        pageQuery.getQuery().put("del", false);
        String type = MapUtils.getString(pageQuery.getQuery(), "type", "DEFAULT");
        pageQuery.getQuery().remove("type");

        Specification<Collection> specification = JpaUtils.toSpecification(pageQuery, Collection.class);
        PageRequest pageRequest = JpaUtils.toPageRequest(pageQuery);
//        if (pageRequest.getSort().stream().noneMatch(order -> order.getProperty().equals("createdAt"))) {
//            pageRequest = PageRequest.of(pageRequest.getPageNumber(), pageQuery.getSize(),
//                    pageRequest.getSort().and(Sort.by("createdAt").descending()));
//        }

        specification = specification.and((Specification<Collection>) (root, criteriaQuery, criteriaBuilder) -> {
            List<Predicate> and = new ArrayList<>();

            if (StringUtils.isNotEmpty(type) && !"all".equalsIgnoreCase(type)) {
                try {
                    if (type.contains(",")) {
                        and.add(root.get("type")
                                .in(Arrays.stream(type.split(",")).map(s -> Enum.valueOf(CollectionType.class, s))
                                        .collect(Collectors.toList())));
                    } else {
                        and.add(criteriaBuilder.equal(root.get("type"), Enum.valueOf(CollectionType.class, type)));
                    }
                } catch (Exception e) {

                }
            }
            return criteriaBuilder.and(and.toArray(new Predicate[0]));
        });
        Page<Collection> page = collectionRepo.findAll(specification, pageRequest);
        return new PageWrapper<>(page.getContent(), page.getPageable().getPageNumber(),
                page.getPageable().getPageSize(), page.getTotalElements());
    }

    public Collection create(Collection record) {
        User minter = userRepo.findById(record.getMinterId()).orElse(SecurityUtils.getAuthenticatedUser());
        record.setMinter(minter.getNickname());
        record.setMinterId(minter.getId());
        record.setMinterAvatar(minter.getAvatar());
        record.setOwner(minter.getNickname());
        record.setOwnerId(minter.getId());
        record.setOwnerAvatar(minter.getAvatar());
        record.setStock(record.getTotal());
        record.setSale(0);
        record.setVipQuota(record.getTotalQuota());
        if (record.isScheduleSale()) {
            if (record.getStartTime() == null) {
                throw new BusinessException("请填写定时发布时间");
            }
            if (record.getStartTime().isBefore(LocalDateTime.now())) {
                record.setOnShelf(true);
                record.setSalable(true);
                record.setStartTime(null);
            }
        }
        record = collectionRepo.save(record);
        onShelfTask(record);
        redisTemplate.opsForValue().set(RedisKeys.COLLECTION_STOCK + record.getId(), record.getStock());
        redisTemplate.opsForValue().set(RedisKeys.COLLECTION_SALE + record.getId(), record.getSale());
        return record;
    }

    public Collection update(Collection record) {
        collectionRepo.update(record.getId(), record.isOnShelf(), record.isSalable(),
                record.getStartTime(), record.isScheduleSale(), record.getSort(),
                record.getDetail(), JSON.toJSONString(record.getPrivileges()),
                JSON.toJSONString(record.getProperties()), JSON.toJSONString(record.getModel3d()),
                record.getMaxCount(), record.getCountId(), record.isScanCode(), record.isNoSoldOut(),
                record.getAssignment(), record.isCouponPayment(), record.getShareBg(), record.getRegisterBg(),
                record.getVipQuota(), record.getTimeDelay(), record.getSaleTime(), record.getHoldDays(),
                record.getOpenQuota(), record.getTotalQuota(), record.getMinimumCharge());

        record = collectionRepo.findById(record.getId()).orElseThrow(new BusinessException("无记录"));
        onShelfTask(record);
        return record;
    }

    private void onShelfTask(Collection record) {
        ScheduledFuture<?> task = tasks.get(record.getId());
        if (task != null) {
            if (!task.cancel(true)) {
                return;
            }
        }
        if (record.isScheduleSale()) {
            if (record.getStartTime().minusSeconds(2).isAfter(LocalDateTime.now())) {
                Date date = Date.from(record.getStartTime().atZone(ZoneId.systemDefault()).toInstant());
                ScheduledFuture<?> future = taskScheduler.schedule(() -> {
                    collectionRepo.scheduleOnShelf(record.getId(), !record.isScanCode());
                    tasks.remove(record.getId());
                }, date);
                tasks.put(record.getId(), future);
            } else {
                collectionRepo.scheduleOnShelf(record.getId(), !record.isScanCode());
            }
        }
    }

    public CollectionDTO toDTO(Collection collection) {
        return toDTO(collection, true, false);
    }

    public CollectionDTO toDTO(Collection collection, boolean join, boolean showVip) {
        CollectionDTO collectionDTO = new CollectionDTO();
        BeanUtils.copyProperties(collection, collectionDTO);
        if (join) {
            User user = SecurityUtils.getAuthenticatedUser();
            if (user != null) {
                List<Like> list = likeRepo.findByUserIdAndCollectionId(user.getId(),
                        collection.getId());
                collectionDTO.setLiked(!list.isEmpty());
                if (collection.getType() == CollectionType.BLIND_BOX) {
                    collectionDTO.setAppointment(appointmentRepo.findFirstByBlindBoxId(collection.getId()).isPresent());
                }
                if (showVip && collection.getAssignment() > 0 && user.getVipPurchase() > 0) {
                    int purchase = orderRepo.countByUserIdAndCollectionIdAndVipTrueAndStatusIn(user.getId(), collection.getId(), Arrays.asList(OrderStatus.FINISH, OrderStatus.NOT_PAID, OrderStatus.PROCESSING));
                    collectionDTO.setVipSurplus(user.getVipPurchase() - purchase);
                }
            }
        }
        return collectionDTO;
    }

    public List<CollectionDTO> toDTO(List<Collection> collections) {
        List<Like> likes = new ArrayList<>();
        List<Appointment> appointments = new ArrayList<>();
        if (SecurityUtils.getAuthenticatedUser() != null) {
            likes.addAll(likeRepo.findByUserIdAndCollectionIdIsNotNull(SecurityUtils.getAuthenticatedUser().getId()));
            appointments.addAll(appointmentRepo.findByUserId(SecurityUtils.getAuthenticatedUser().getId()));
        }
        return collections.stream().parallel().map(collection -> {
            CollectionDTO dto = toDTO(collection, false, false);
            if (!likes.isEmpty()) {
                dto.setLiked(likes.stream().anyMatch(l -> l.getCollectionId().equals(collection.getId())));
            }
            if (!appointments.isEmpty()) {
                dto.setAppointment(appointments.stream().anyMatch(a -> a.getBlindBoxId().equals(collection.getId())));
            }
            return dto;
        }).collect(Collectors.toList());
    }

    public Page<CollectionDTO> toDTO(Page<Collection> collections) {
        List<CollectionDTO> userDTOS = toDTO(collections.getContent());
        return new PageImpl<>(userDTOS, collections.getPageable(), collections.getTotalElements());
    }

    @Transactional
    public Collection createBlindBox(CreateBlindBox createBlindBox) {
        Collection blindBox = createBlindBox.getBlindBox();
        if (blindBox.getId() != null) {
            throw new BusinessException("无法完成此操作");
        }

        List<Collection> list =
                collectionRepo.findAllById(createBlindBox.getItems().stream().map(BlindBoxItem::getCollectionId)
                        .collect(Collectors.toSet()));
        for (BlindBoxItem item : createBlindBox.getItems()) {
            Collection collection = list.stream().filter(i -> i.getId().equals(item.getCollectionId())).findAny()
                    .orElseThrow(new BusinessException("所选藏品不存在"));
            if (item.getTotal() > collection.getStock()) {
                throw new BusinessException("所选藏品库存不足:" + collection.getName());
            }
        }

        User user = userRepo.findById(blindBox.getMinterId()).orElse(SecurityUtils.getAuthenticatedUser());
        blindBox.setMinter(user.getNickname());
        blindBox.setMinterId(user.getId());
        blindBox.setMinterAvatar(user.getAvatar());
        blindBox.setOwner(user.getNickname());
        blindBox.setOwnerId(user.getId());
        blindBox.setOwnerAvatar(user.getAvatar());
        blindBox.setTotal(createBlindBox.getItems().stream().mapToInt(BlindBoxItem::getTotal).sum());
        blindBox.setStock(blindBox.getTotal());
        blindBox.setSale(0);
        collectionRepo.save(blindBox);
        createBlindBox.getItems().stream().parallel().forEach(item -> {
            Collection collection = list.stream().filter(i -> i.getId().equals(item.getCollectionId())).findAny()
                    .orElseThrow(new BusinessException("所选藏品不存在"));
            decreaseStock(collection.getId(), item.getTotal());
            BlindBoxItem blindBoxItem = new BlindBoxItem();
            BeanUtils.copyProperties(collection, blindBoxItem);
            blindBoxItem.setId(null);
            blindBoxItem.setCollectionId(item.getCollectionId());
            blindBoxItem.setSale(0);
            blindBoxItem.setTotal(item.getTotal());
            blindBoxItem.setStock(item.getTotal());
            blindBoxItem.setRare(item.isRare());
            blindBoxItem.setBlindBoxId(blindBox.getId());
            blindBoxItemRepo.saveAndFlush(blindBoxItem);
            log.info("createBlindBoxItemSuccess" + blindBoxItem.getId());
        });
        return blindBox;
    }

    public void appointment(Long id, Long userId) {
        Collection collection = collectionRepo.findById(id).orElseThrow(new BusinessException("无记录"));
        if (collection.getType() != CollectionType.BLIND_BOX) {
            throw new BusinessException("非盲盒，无需预约");
        }

        if (collection.getStartTime().isBefore(LocalDateTime.now())) {
            throw new BusinessException("盲盒已开售，无需预约");
        }

        appointmentRepo.save(Appointment.builder()
                .userId(userId)
                .blindBoxId(id)
                .build());
    }

    public synchronized BlindBoxItem draw(Long collectionId) {
        List<BlindBoxItem> items = blindBoxItemRepo.findByBlindBoxId(collectionId);

        Map<BlindBoxItem, Range<Integer>> randomRange = new HashMap<>();
        int c = 0, sum = 0;
        for (BlindBoxItem item : items) {
            randomRange.put(item, Range.between(c, c + item.getStock()));
            c += item.getStock();
            sum += item.getStock();
        }

        int retry = 0;
        BlindBoxItem winItem = null;
        while (winItem == null) {
            retry++;
            int rand = RandomUtils.nextInt(0, sum + 1);
            for (Map.Entry<BlindBoxItem, Range<Integer>> entry : randomRange.entrySet()) {
                BlindBoxItem item = entry.getKey();
                Range<Integer> range = entry.getValue();
                if (rand >= range.getMinimum() && rand < range.getMaximum()) {
                    int total = items.stream().filter(i -> !i.isRare())
                            .mapToInt(BlindBoxItem::getTotal).sum();
                    int stock = items.stream().filter(i -> !i.isRare())
                            .mapToInt(BlindBoxItem::getStock).sum();
                    if (item.isRare()) {
                        double nRate = stock / (double) total;
                        double rRate = (item.getStock() - 1) / (double) item.getTotal();

                        if (Math.abs(nRate - rRate) < (1 / (double) item.getTotal()) || retry > 1 || rRate == 0) {
                            if (!(nRate > 0.1 && item.getStock() == 1)) {
                                winItem = item;
                            }
                        }
                    } else {
                        double nRate = (stock - 1) / (double) total;
                        double rRate = item.getStock() / (double) item.getTotal();

                        if (Math.abs(nRate - rRate) < 0.2 || retry > 1 || nRate == 0) {
                            winItem = item;
                        }
                    }
                }
            }

            if (retry > 100 && winItem == null) {
                throw new BusinessException("盲盒抽卡失败");
            }
        }
//        winItem.setStock(winItem.getStock() - 1);
//        winItem.setSale(winItem.getSale() + 1);
//        blindBoxItemRepo.saveAndFlush(winItem);
        blindBoxItemRepo.decreaseStockAndIncreaseSale(winItem.getId(), 1);
        blindBoxItemRepo.flush();
        return winItem;
    }

    public synchronized Integer getNextNumber(Long collectionId) {
        collectionRepo.increaseNumber(collectionId, 1);
        return collectionRepo.getCurrentNumber(collectionId).orElse(0);
    }

    public void addStock(Long id, int number) {
        Collection collection = collectionRepo.findById(id).orElseThrow(new BusinessException("无记录"));
        if (collection.getSource() != CollectionSource.OFFICIAL) {
            throw new BusinessException("用户转售无法增发");
        }
        if (collection.getType() == CollectionType.BLIND_BOX) {
            throw new BusinessException("盲盒无法增发");
        }
        increaseStock(id, number);
        collectionRepo.increaseTotal(id, number);
    }

    public synchronized Long increaseStock(Long id, int number) {
        BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps(RedisKeys.COLLECTION_STOCK + id);
        if (ops.get() == null) {
            Boolean success = ops.setIfAbsent(Optional.ofNullable(collectionRepo.getStock(id))
                    .orElse(0), 7, TimeUnit.DAYS);
            log.info("创建redis库存：{}", success);
        }
        Long stock = ops.increment(number);
        rocketMQTemplate.convertAndSend(generalProperties.getUpdateStockTopic(), id);
        return stock;
    }

    public synchronized Integer getStock(Long id) {
        BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps(RedisKeys.COLLECTION_STOCK + id);
        Integer stock = (Integer) ops.get();
        if (stock == null) {
            Boolean success = ops.setIfAbsent(Optional.ofNullable(collectionRepo.getStock(id))
                    .orElse(0), 7, TimeUnit.DAYS);
            log.info("创建redis库存：{}", success);
            return (Integer) ops.get();
        } else {
            return stock;
        }
    }

    public synchronized Long decreaseStock(Long id, int number) {
        return increaseStock(id, -number);
    }

    public synchronized Long increaseSale(Long id, int number) {
        BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps(RedisKeys.COLLECTION_SALE + id);
        if (ops.get() == null) {
            Boolean success = ops.setIfAbsent(Optional.ofNullable(collectionRepo.getSale(id))
                    .orElse(0), 7, TimeUnit.DAYS);
            log.info("创建redis销量：{}", success);
        }
        Long sale = ops.increment(number);
        redisTemplate.opsForHash().increment(RedisKeys.UPDATE_SALE, id.toString(), 1);
//        rocketMQTemplate.convertAndSend(generalProperties.getUpdateSaleTopic(), id);
        return sale;
    }

    public synchronized Long decreaseSale(Long id, int number) {
        return increaseSale(id, -number);
    }

//    @Debounce(key = "#id", delay = 500)
    public void syncStock(Long id) {
        Integer stock = (Integer) redisTemplate.opsForValue().get(RedisKeys.COLLECTION_STOCK + id);
        if (stock != null) {
            log.info("同步库存信息{}", id);
            collectionRepo.updateStock(id, stock);
            cacheService.clearCollection(id);
        }
    }

    //    @Debounce(key = "#id", delay = 500)
    public void syncSale(Long id) {
        Integer sale = (Integer) redisTemplate.opsForValue().get(RedisKeys.COLLECTION_SALE + id);
        if (sale != null) {
            log.info("同步销量信息{}", id);
            collectionRepo.updateSale(id, sale);
            cacheService.clearCollection(id);
        }
    }

    @Debounce(key = "#id", delay = 500)
    public void syncQuota(Long id) {
        Integer quota = (Integer) redisTemplate.opsForValue().get(RedisKeys.COLLECTION_QUOTA + id);
        if (quota != null) {
            log.info("同步额度信息{}", id);
            collectionRepo.updateVipQuota(id, quota);
            cacheService.clearCollection(id);
        }
    }

    public synchronized Long decreaseQuota(Long id, int number) {
        BoundValueOperations<String, Object> ops = redisTemplate.boundValueOps(RedisKeys.COLLECTION_QUOTA + id);
        if (ops.get() == null) {
            Boolean success = ops.setIfAbsent(Optional.ofNullable(collectionRepo.getVipQuota(id))
                    .orElse(0), 7, TimeUnit.DAYS);
            log.info("创建redis额度：{}", success);
        }
        Long stock = ops.increment(-number);
        rocketMQTemplate.convertAndSend(generalProperties.getUpdateQuotaTopic(), id);
        return stock;
    }

    @Cacheable(value = "recommendLegacy", key = "#type")
    public List<CollectionDTO> recommendLegacy(@RequestParam String type) {
        return collectionRepo.recommend(type).stream().map(rc -> {
            if (StringUtils.isNotBlank(rc.getRecommend().getPic())) {
                rc.getCollection().setPic(Collections.singletonList(new FileObject(null, rc.getRecommend()
                        .getPic(), null, null)));
            }
            CollectionDTO collectionDTO = new CollectionDTO();
            BeanUtils.copyProperties(rc.getCollection(), collectionDTO);
            return collectionDTO;
        }).collect(Collectors.toList());
    }


    public List<PointDTO> savePoint(Long collectionId, LocalDateTime time) {
        Collection collection = collectionRepo.findById(collectionId).orElseThrow(new BusinessException("无藏品"));
        //库存
//        int stock = collection.getStock();
        //是否开启白名单
        int assignment = collection.getAssignment();
        if (assignment <= 0) {
            return null;
        }
        List<User> users = userRepo.findAllByCollectionId(collectionId);
        //邀请者
        Map<Long, List<User>> userMap = users.stream()
                .filter(user -> ObjectUtils.isNotEmpty(user.getCollectionInvitor()))
                .collect(Collectors.groupingBy(User::getCollectionInvitor));

        AtomicInteger sum = new AtomicInteger();
        AtomicInteger sum1 = new AtomicInteger();
        List<PointDTO> dtos = new ArrayList<>();

        Map<Long, BigDecimal> historyMap = tokenHistoryRepo.userBuy(userMap.keySet())
                .stream()
                .collect(Collectors.groupingBy(TokenHistory::getToUserId, Collectors.reducing(BigDecimal.ZERO,
                        TokenHistory::getPrice,
                        BigDecimal::add)));

        DateTimeFormatter dft = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        userMap.forEach((key, value) -> {
            //邀请达到数量
            if (value.size() >= collection.getAssignment()) {
                value.sort(Comparator.comparing(User::getCreatedAt));
                BigDecimal buy = historyMap.get(key);
                //满足条件的时间
                User user = value.get(collection.getAssignment() - 1);
                //作弊得已屏蔽
                if ((ObjectUtils.isEmpty(buy) || buy.compareTo(BigDecimal.valueOf(500)) < 0) && user.getCreatedAt()
                        .isBefore(time)) {
                    sum1.getAndIncrement();
                    System.out.println(key + "," + dft.format(user.getCreatedAt()) + "," + buy);
                } else {
                    //实名数量
                    long identitySum = value.stream().filter(u -> AuthStatus.SUCCESS.equals(u.getAuthStatus())).count();
                    dtos.add(new PointDTO(key, user.getCreatedAt(), value.size(), (int) identitySum, buy));
                    sum.getAndIncrement();
                }
            }
        });
        log.info("完成任务人数：{}", sum);
        log.info("作弊任务人数：{}", sum1);

        LongArrayConverter longArrayConverter = new LongArrayConverter();

        List<Long> collect = dtos.stream()
                .filter(dto -> time.isBefore(dto.getCreatedAt()))
                .map(PointDTO::getId)
                .collect(Collectors.toList());
        log.info(dft.format(time) + "前完成任务人数：{}", collect.size());
//        log.info("sql: update user set vip_point = 1 where id in ({})", longArrayConverter.convertToDatabaseColumn(collect));

        List<PointDTO> collect1 = dtos.stream()
                .filter(dto -> time.isAfter(dto.getCreatedAt()))
                .collect(Collectors.toList());
        log.info(dft.format(time) + "后完成任务人数：{}", collect1.size());

        List<Long> collect2 = dtos.stream()
                .filter(dto -> dto.getIdentitySum() > 0)
                .map(PointDTO::getId)
                .collect(Collectors.toList());
        log.info("邀请实名认证人量：{}", collect2.size());
//        log.info("sql: update user set vip_point = 1 where id in ({})", longArrayConverter.convertToDatabaseColumn(collect2));

        return dtos;
    }
}