package com.m2pool.lease.task;

import cn.hutool.json.JSONUtil;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.m2pool.lease.dto.HourIncomeDto;
import com.m2pool.lease.dto.OrderStatusDto;
import com.m2pool.lease.entity.*;
import com.m2pool.lease.mapper.*;
import com.m2pool.lease.mq.message.RabbitmqPoolProxyMessage;
import com.m2pool.lease.service.LeaseOrderInfoService;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.rmi.dgc.Lease;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.m2pool.lease.constant.RabbitmqConstant.POOL_PROXY_QUEUE_NAME;

/**
 * @Description 后台管理系统 定时任务
 * @Date 2025/7/15 15:45
 * @Author yyb
 */
@Configuration
@EnableScheduling
public class OwnProductTask {

    @Resource
    private LeaseUserOwnedProductMapper leaseUserOwnedProductMapper;


    @Resource
    private LeaseProductMachineMapper leaseProductMachineMapper;

    @Resource
    private LeaseOrderItemMapper leaseOrderItemMapper;

    @Resource
    private LeaseOrderInfoService leaseOrderInfoService;

    @Resource
    private LeasePayRecordMessageMapper leasePayRecordMessageMapper;

    @Resource
    private RabbitTemplate rabbitTemplate;

    private static final int THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors();
    private static final int BATCH_SIZE = 1000;
    private final ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
    /**
     * 修改已购表租约过期状态为1 已过期 + 机器表状态为0 未售出 + 修改租约到期订单状态为已完成
     */
    @Scheduled(cron = "0 1,5,7 0 * * ? ")
    //@Scheduled(cron = "0 0/3 * * * ? ")
    @Transactional
    public void updateOwnMachineStateTask(){
        // 获取当天 0 点的 LocalDateTime
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime startOfDay = now.toLocalDate().atStartOfDay();
        //1. 查询当天失效机器
        LambdaQueryWrapper<LeaseUserOwnedProduct> wrapper = new LambdaQueryWrapper<LeaseUserOwnedProduct>()
                .eq(LeaseUserOwnedProduct::getEndTime, startOfDay).eq(LeaseUserOwnedProduct::getStatus, 0);
        List<LeaseUserOwnedProduct> leaseUserOwnedProducts = leaseUserOwnedProductMapper.selectList(wrapper);

        if (!leaseUserOwnedProducts.isEmpty()){
            //2.修改商品机器表中 销售状态为 0 未售出
            List<Long> ids = new ArrayList<>();
            List<Long> isExpireIds = new ArrayList<>();
            List<Long> itemIds = new ArrayList<>();
            for (LeaseUserOwnedProduct leaseUserOwnedProduct : leaseUserOwnedProducts) {
                ids.add(leaseUserOwnedProduct.getId());
                isExpireIds.add(leaseUserOwnedProduct.getProductMachineId());
                itemIds.add(leaseUserOwnedProduct.getOrderItemId());
            }
            leaseUserOwnedProductMapper.update(LeaseUserOwnedProduct.builder().status(1).build(),new LambdaUpdateWrapper<LeaseUserOwnedProduct>()
                    .in(LeaseUserOwnedProduct::getId, ids));
            leaseProductMachineMapper.update(LeaseProductMachine.builder().saleState(0).build(),
                    new LambdaQueryWrapper<LeaseProductMachine>().in(LeaseProductMachine::getId, isExpireIds));
            leaseOrderItemMapper.update(LeaseOrderItem.builder().status(0).build(),new LambdaQueryWrapper<LeaseOrderItem>().in(LeaseOrderItem::getId, itemIds));
            //3.订单所有机器已过期,直接修改订单状态为已完成
            List<Long> orderIds = leaseOrderItemMapper.getOrderIds(itemIds);
            List<OrderStatusDto> orderInfoStatus = leaseOrderItemMapper.getOrderInfoStatus(orderIds);

            List<LeaseOrderInfo> infoList = new ArrayList<>();
            orderInfoStatus.forEach(item->{
                if (item.getOrderInfoStatus() == 8){
                    infoList.add(LeaseOrderInfo.builder().id(item.getOrderId()).status(8).build());
                }
            });
            if (!infoList.isEmpty()){
                leaseOrderInfoService.saveOrUpdateBatch(infoList);
            }

            System.out.println("租约过期机器状态修改成功");
            //3.发送rabbitmq消息给矿池代理删除代理的矿机
            //sendRabbitmqMessage(leaseUserOwnedProducts);
        }
    }

    public void sendRabbitmqMessage( List<LeaseUserOwnedProduct> leaseUserOwnedProducts) {
        leaseUserOwnedProducts.forEach(item -> {
            executorService.execute(()->{
                RabbitmqPoolProxyMessage rabbitmqPoolProxyMessage = RabbitmqPoolProxyMessage.builder()
                        .MethodID(1)
                        .Msg("")
                        .ID(item.getUser() + "-" + item.getMiner())
                        .Address(item.getAddress())
                        .build();
                try {
                    rabbitTemplate.convertAndSend(POOL_PROXY_QUEUE_NAME, rabbitmqPoolProxyMessage);
                    //同步阻塞等待消息发送成功
                    //boolean b = rabbitTemplate.waitForConfirms(10000);
                } catch (Exception e) {

                    System.err.println("发送失败: " + e.getMessage());
                }
            });
        });

    }


    @Scheduled(cron = "0 35 0/1 * * ? ")
    //@Scheduled(cron = "30 0/2 * * * ? ")
    @DSTransactional
    public void updateNexaIncomeTask(){
        List<LeaseUserOwnedProduct> updateList = computeIncome("nexa");
        //批量修改
        batchUpdateLeaseUserOwnedProducts(updateList);

    }

    @Scheduled(cron = "10 35 0/1 * * ? ")
    //@Scheduled(cron = "30 0/2 * * * ? ")
    @DSTransactional
    public void updateGrsIncomeTask(){
        List<LeaseUserOwnedProduct> updateList = computeIncome("grs");
        //批量修改
        batchUpdateLeaseUserOwnedProducts(updateList);
    }

    @Scheduled(cron = "20 35 0/1 * * ? ")
    //@Scheduled(cron = "30 0/2 * * * ? ")
    @DSTransactional
    public void updateRxdIncomeTask(){
        List<LeaseUserOwnedProduct> updateList = computeIncome("rxd");
        //批量修改
        batchUpdateLeaseUserOwnedProducts(updateList);
    }

    @Scheduled(cron = "30 35 0/1 * * ? ")
    //@Scheduled(cron = "30 0/2 * * * ? ")
    @DSTransactional
    public void updateMonaIncomeTask(){
        List<LeaseUserOwnedProduct> updateList = computeIncome("mona");
        //批量修改
        batchUpdateLeaseUserOwnedProducts(updateList);
    }


    /**
     * 计算已购生效机器过去一小时收益
     * @param coin
     */
    public List<LeaseUserOwnedProduct> computeIncome(String coin){
        //获取当前时间整点时间
        LocalDateTime now = LocalDateTime.now().withMinute(0).withSecond(0).withNano(0);
        //获取当天开始时间
        //LocalDateTime startOfDay = now.with(LocalTime.MIN);
        LocalDateTime end = now.minusHours(1);
        LocalDateTime start = now.minusHours(2);
        //1. 查询在有效期内的已购机器
        LambdaQueryWrapper<LeaseUserOwnedProduct> wrapper = new LambdaQueryWrapper<LeaseUserOwnedProduct>()
                .eq(LeaseUserOwnedProduct::getCoin, coin)
                .eq(LeaseUserOwnedProduct::getStatus, 0);
        List<LeaseUserOwnedProduct> leaseUserOwnedProducts = leaseUserOwnedProductMapper.selectList(wrapper);

        //2.获取近一个小时矿机的收益 + 并计算实时收益
        List<LeaseUserOwnedProduct> updateList = new ArrayList<>();
        if (!leaseUserOwnedProducts.isEmpty()){
            List<HourIncomeDto> incomeDtos = leaseUserOwnedProductMapper.getHourIncomeList(leaseUserOwnedProducts,coin, start, end);
            Map<String, LeaseUserOwnedProduct> collect = leaseUserOwnedProducts.stream()
                    .collect(Collectors.toMap(item-> item.getUser() + "-" + item.getMiner()+ "-" + item.getCoin(),Function.identity()));
            for (HourIncomeDto hourIncomeDto : incomeDtos) {
               String key = hourIncomeDto.getUser() + "-" + hourIncomeDto.getMiner()+ "-" + coin;
               LeaseUserOwnedProduct leaseUserOwnedProduct = collect.get(key) ;
               leaseUserOwnedProduct.setSettleIncome(hourIncomeDto.getIncome().add(leaseUserOwnedProduct.getSettleIncome()));
               //leaseUserOwnedProduct.setSettleUsdtIncome(hourIncomeDto.getUsdtIncome().add(leaseUserOwnedProduct.getSettleUsdtIncome()));
               updateList.add(leaseUserOwnedProduct);
            }

        }
        return updateList;
    }


    ///**
    // * 计算已购生效机器过去一小时收益
    // * @param coin
    // */
    //public List<LeaseUserOwnedProduct> computeIncome(String coin){
    //    //获取当前时间整点时间
    //    LocalDateTime now = LocalDateTime.now().withMinute(0).withSecond(0).withNano(0);
    //    //获取当天开始时间
    //    LocalDateTime startOfDay = now.with(LocalTime.MIN);
    //    LocalDateTime end = now.minusHours(1);
    //    LocalDateTime start = now.minusHours(2);
    //    //1. 查询在有效期内的已购机器
    //    LambdaQueryWrapper<LeaseUserOwnedProduct> wrapper = new LambdaQueryWrapper<LeaseUserOwnedProduct>()
    //            .eq(LeaseUserOwnedProduct::getCoin, coin)
    //            .eq(LeaseUserOwnedProduct::getStatus, 0);
    //    List<LeaseUserOwnedProduct> leaseUserOwnedProducts = leaseUserOwnedProductMapper.selectList(wrapper);
    //    System.out.println("个人收益--矿机: " + JSONUtil.toJsonPrettyStr(leaseUserOwnedProducts));
    //    //2.获取近一个小时矿机的收益 + 并计算实时收益
    //    List<LeaseUserOwnedProduct> updateList = new ArrayList<>();
    //    if (!leaseUserOwnedProducts.isEmpty()){
    //        Map<String, LeaseUserOwnedProduct> collect = new HashMap<>();
    //        Set<Long> orderInfoIds = new HashSet<>();
    //        for (LeaseUserOwnedProduct item : leaseUserOwnedProducts) {
    //            collect.put(item.getUser() + "-" + item.getMiner() + "-" + item.getCoin(), item);
    //            orderInfoIds.add(item.getOrderId());
    //        }
    //
    //        //2.1 查询当天已支付订单
    //        //查询订单详情对应的信息formAddress + fromChain + fromSymbol + toAddress + toChain + toSymbol
    //        List<LeaseOrderItem> leaseOrderItems = leaseOrderItemMapper.getOrderItemByOrderIds(orderInfoIds);
    //        Map<String, LeaseOrderItem> collect1 = leaseOrderItems.stream().collect(Collectors.toMap(item -> item.getUser() + "-" + item.getMiner() + "-" + item.getCoin(), Function.identity()));
    //
    //        List<HourIncomeDto> incomeDtos = leaseUserOwnedProductMapper.getHourIncomeList(leaseUserOwnedProducts,coin, start, end);
    //        for (HourIncomeDto hourIncomeDto : incomeDtos) {
    //           String key = hourIncomeDto.getUser() + "-" + hourIncomeDto.getMiner()+ "-" + coin;
    //            System.out.println("个人收益--矿机详情: " + key);
    //            LeaseUserOwnedProduct leaseUserOwnedProduct = collect.get(key) ;
    //            LeaseOrderItem leaseOrderItem = collect1.get(key);
    //            System.out.println("个人收益--订单详情: " + JSONUtil.toJsonPrettyStr(leaseOrderItem));
    //            if (leaseOrderItem != null && leaseOrderItem.getStatus() == 1){
    //                BigDecimal coinIncome = hourIncomeDto.getIncome().add(leaseUserOwnedProduct.getCurrentIncome());
    //                BigDecimal usdtIncome = hourIncomeDto.getUsdtIncome().add(leaseUserOwnedProduct.getCurrentUsdtIncome());
    //                //当日待结算收益不为0，把待结算收益加到当前收益中
    //                if (leaseUserOwnedProduct.getSettleIncome().compareTo(BigDecimal.ZERO) > 0){
    //                    coinIncome =  coinIncome.add(leaseUserOwnedProduct.getSettleIncome());
    //                    usdtIncome = usdtIncome.add(leaseUserOwnedProduct.getSettleUsdtIncome());
    //                    leaseUserOwnedProduct.setSettleIncome(BigDecimal.ZERO);
    //                    leaseUserOwnedProduct.setSettleUsdtIncome(BigDecimal.ZERO);
    //                }
    //                leaseUserOwnedProduct.setCurrentIncome(coinIncome);
    //                leaseUserOwnedProduct.setCurrentUsdtIncome(usdtIncome);
    //
    //            }else{
    //                leaseUserOwnedProduct.setSettleIncome(hourIncomeDto.getIncome().add(leaseUserOwnedProduct.getSettleIncome()));
    //                leaseUserOwnedProduct.setSettleUsdtIncome(hourIncomeDto.getUsdtIncome().add(leaseUserOwnedProduct.getSettleUsdtIncome()));
    //            }
    //            if (now.equals(startOfDay)){
    //                leaseUserOwnedProduct.setSettleIncome(BigDecimal.ZERO);
    //                leaseUserOwnedProduct.setSettleUsdtIncome(BigDecimal.ZERO) ;
    //            }
    //
    //            updateList.add(leaseUserOwnedProduct);
    //        }
    //    }
    //    return updateList;
    //}

    /**
     * 批量修改已购机器的收益
     * @param updateList
     */
    public void batchUpdateLeaseUserOwnedProducts(List<LeaseUserOwnedProduct> updateList) {
        // 用于记录执行过程中的异常
        AtomicReference<Exception> exceptionRef = new AtomicReference<>();
        List<CompletableFuture<Void>> futures = new ArrayList<>();

        for (int i = 0; i < updateList.size(); i += BATCH_SIZE) {
            int toIndex = Math.min(i + BATCH_SIZE, updateList.size());
            List<LeaseUserOwnedProduct> subList = updateList.subList(i, toIndex);

            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                try {
                    if (!subList.isEmpty()){
                        leaseUserOwnedProductMapper.updateBatchIncome(subList);
                    }
                } catch (Exception e) {
                    exceptionRef.set(e);
                    throw new RuntimeException(e);
                }
            }, executorService);

            futures.add(future);
        }

        // 等待所有任务完成
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        try {
            allFutures.get();
        } catch (InterruptedException | ExecutionException e) {
            // 如果有异常，将异常信息设置到 exceptionRef 中
            exceptionRef.set(e);
        }

        // 如果有异常，抛出异常触发事务回滚
        if (exceptionRef.get() != null) {
            throw new RuntimeException("批量更新数据时发生错误", exceptionRef.get());
        }
    }

}
