package com.zm.order.tool.server.service.impl.routing;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.zm.order.tool.server.dao.OrderItemBatchingDao;
import com.zm.order.tool.server.manager.CommonManager;
import com.zm.order.tool.server.manager.NewHintManager;
import com.zm.order.tool.server.manager.RedisManager;
import com.zm.order.tool.server.manager.RoutingAlgorithm;
import com.zm.order.tool.server.model.OrderItemBatching;
import com.zm.order.tool.server.model.OrderNoUser;
import com.zm.order.tool.server.query.OrderItemQuery;
import com.zm.order.tool.server.service.BatchingRoutingService;
import com.zm.order.tool.server.tempdb.OrderNoUserDao;
import com.zm.order.tool.server.utils.DateUtils;
import com.zm.util.dto.Result;
import org.apache.lucene.util.NamedThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author nianchang
 * @date 2022/01/25 3:57 下午
 */
@Service
@SuppressWarnings("all")
public class BatchingRoutingServiceImpl implements BatchingRoutingService, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(BatchingRoutingServiceImpl.class);

    private BlockingQueue workQueue;

    private ExecutorService executorService;

    @Value("${new.sharding.db.size}")
    private int dbSize;

    @Value("${new.sharding.tb.size}")
    private int tbSize;

    @Value("${sharding.db.size}")
    private int oldDbSize;

    @Value("${sharding.tb.size}")
    private int oldTbSize;

    @Autowired
    private RedisManager redisManager;

    @Autowired
    private NewHintManager hintManager;

    @Autowired
    private CommonManager commonManager;
    @Autowired
    private OrderItemBatchingDao batchingDao;

    @Override
    public void routing() {
        int count = 0;
        OrderItemQuery query = new OrderItemQuery();
//        query.setExcludeSellerId(40927L);
//        query.setUpdatedAtEnd(new Date(1644422400000L));
        query.setSize(200);
        try {
            for (int i = 0; i < oldDbSize; i++) {
                for (int j = 0; j < oldTbSize; j++) {
                    int loop = 0;
                    Long lastId = null;
                    Map<String, List<OrderItemBatching>> map = new HashMap<>(262144);
                    for (;;) {
                        query.setLastId(lastId);
                        long startAt = System.currentTimeMillis();
                        List<OrderItemBatching> list = commonManager.scanBatchingList(query, i, j);
                        long endAt = System.currentTimeMillis() - startAt;
                        if (endAt >= 500) {
                            logger.info("batching_routing_query_time:{}ms", endAt);
                        }
                        if (CollectionUtils.isEmpty(list)) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute(k, v));
                            }
                            break;
                        }
                        for (OrderItemBatching record : list) {
                            groupOrder(record, map);
                        }
                        if (++loop >= 100) {
                            map.forEach((k, v) -> execute(k, v));
                            map.clear();
                            loop = 0;
                        }
                        count += list.size();
                        logger.info("batching_routing_count:{}条", count);
                        if (list.size() < 200) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute(k, v));
                            }
                            break;
                        }
                        lastId = list.get(list.size() - 1).getId();
                        logger.info("batching_routing_last_id:{},i:{},j:{}", lastId, i, j);
                        Result<String> redisData = redisManager.get("routing_batching_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[BATCHING_ROUTING]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[BatchingRoutingServiceImpl]failed!cause by:{0}", e);
            throw new RuntimeException("系统异常");
        }
    }

    private void execute(String shardKey, List<OrderItemBatching> list) {
        while (workQueue.remainingCapacity() < 1) {
            logger.info("workQueue size:{}", workQueue.size());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        executorService.submit(() -> {
            try {
                String[] split = shardKey.split("-");
                for (List<OrderItemBatching> subList : Lists.partition(list, 50)) {
                    hintManager.replaceBatching(subList, Integer.valueOf(split[0]), Integer.valueOf(split[1]));
                }
            } catch (Exception e) {
                logger.error("[execute]failed！cause by:{}", e);
            }
        });
    }

    private void groupOrder(OrderItemBatching record, Map<String, List<OrderItemBatching>> map) {
        // 计算在新数据库中的路由
        final Long userId = record.getUserId();
        int dbIndex = RoutingAlgorithm.routeDB(userId, dbSize);
        int tbIndex = RoutingAlgorithm.routeTB(userId, dbSize, tbSize);
        // 按照路由结果进行分组，用于批量查询
        String key = dbIndex + "-" + tbIndex;
        List<OrderItemBatching> list = null;
        if (!map.containsKey(key)) {
            list = new ArrayList<>();
        } else {
            list = map.get(key);
        }
        list.add(record);
        map.put(key, list);
    }

    @Override
    public void check() {
        int count = 0;
        Date now = new Date();
        AtomicInteger diffCount = new AtomicInteger();
        AtomicInteger sameCount = new AtomicInteger();
        OrderItemQuery query = new OrderItemQuery();
        query.setMaxId(716440803752341506L);
//        query.setUpdatedAtStart(DateUtils.before(now, Calendar.DAY_OF_YEAR, 30));
//        query.setUpdatedAtEnd(now);
        query.setSize(1000);
        try {
            for (int i = 0; i < oldDbSize; i++) {
                for (int j = 0; j < oldTbSize; j++) {
                    int loop = 0;
                    Long lastId = 697264332513206272L;
                    Map<String, List<OrderItemBatching>> map = new HashMap<>(262144);
                    for (;;) {
                        query.setLastId(lastId);
                        List<OrderItemBatching> dataList = commonManager.scanBatchingList(query, i, j);
                        if (CollectionUtils.isEmpty(dataList)) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, sameCount));
                            }
                            break;
                        }
                        for (OrderItemBatching record : dataList) {
                            groupOrder(record, map);
                        }
                        if (++loop >= 100) {
                            map.forEach((k, v) -> execute0(k, v, diffCount, sameCount));
                            map.clear();
                            loop = 0;
                        }
                        count += dataList.size();
                        logger.info("BATCHING_CHECK_COUNT:{}条", count);
                        if (dataList.size() < 1000) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, sameCount));
                            }
                            break;
                        }
                        lastId = dataList.get(dataList.size() - 1).getId();
                        logger.info("batching_check_last_id:{},i:{},j:{}", lastId, i, j);

                        Result<String> redisData = redisManager.get("check_batching_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[BATCHING_CHECK]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[BatchingRoutingServiceImpl]check failed!cause by:{0}", e);
            throw new RuntimeException("系统异常");
        }
    }

    private void execute0(String k, List<OrderItemBatching> list, AtomicInteger diffCount, AtomicInteger sameCount) {
        while (workQueue.remainingCapacity() < 20) {
            logger.info("workQueue size:{}", workQueue.size());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        executorService.submit(() -> {
            try {
                String[] split = k.split("-");
                int dbIndex = Integer.valueOf(split[0]);
                int tbIndex = Integer.valueOf(split[1]);
                List<Long> idList = list.stream().map(OrderItemBatching::getId).collect(Collectors.toList());
                List<OrderItemBatching> newList = hintManager.selectItemBatchingById(idList, dbIndex, tbIndex);
                this.compare11(list, newList, diffCount, sameCount);
                logger.info("COMPARE_BATCHING_COUNT,diffCount:{}条,sameCount:{}条", diffCount, sameCount);
            } catch (Exception e) {
                logger.error("[execute0]failed！cause by:{}", e);
            }
        });
    }
    
    private void compare(List<OrderItemBatching> oldList, List<OrderItemBatching> newList,
                         AtomicInteger diffCount, AtomicInteger sameCount) {
        Map<Long, List<OrderItemBatching>> oldMap = oldList.stream().collect(Collectors.groupingBy(OrderItemBatching::getId));
        for (OrderItemBatching record : newList) {
            List<OrderItemBatching> dataList = oldMap.get(record.getId());
            if (!CollectionUtils.isEmpty(dataList)) {
                String newData = JSON.toJSONString(record);
                String oldData = JSON.toJSONString(dataList.get(0));
                if (!Objects.equals(newData, oldData)) {
                    OrderNoUser noUser = buildParam(record);
                    noUserDao.insert(noUser);
                    diffCount.incrementAndGet();
                    logger.info("BATCHING_FIND_DIFF_ORDER,old:{},new:{}", oldData, newData);
                } else {
                    sameCount.incrementAndGet();
                }
            }
        }
    }

    private void compare11(List<OrderItemBatching> oldList, List<OrderItemBatching> newList,
                         AtomicInteger diffCount, AtomicInteger sameCount) {
        Map<Long, List<OrderItemBatching>> newMap = newList.stream().collect(Collectors.groupingBy(OrderItemBatching::getId));
        for (OrderItemBatching record : oldList) {
            List<OrderItemBatching> dataList = newMap.get(record.getId());
            if (CollectionUtils.isEmpty(dataList)) {
                logger.info("BATCH_EMPTY, orderNo:{}", record.getOrderNo());
                batchingDao.updateUpdatedAt(record.getId(), record.getUserId());
                continue;
            }
            String oldData = JSON.toJSONString(record);
            String newData = JSON.toJSONString(dataList.get(0));
            if (!Objects.equals(newData, oldData)) {
                OrderNoUser noUser = buildParam(record);
                noUserDao.insert(noUser);
                diffCount.incrementAndGet();
                logger.info("BATCHING_FIND_DIFF_ORDER,old:{},new:{}", oldData, newData);
            } else {
                sameCount.incrementAndGet();
            }
        }
    }

    @Override
    public void updateData() {
        int count = 0;
        Date now = new Date();
        OrderItemQuery query = new OrderItemQuery();
        query.setUpdatedAtStart(DateUtils.before(now, Calendar.DAY_OF_YEAR, 7));
        query.setUpdatedAtEnd(now);
        query.setPageSize(1000);
        try {
            for (int i = 0; i < oldDbSize; i++) {
                for (int j = 0; j < oldTbSize; j++) {
                    Long lastId = null;
                    for (;;) {
                        query.setLastId(lastId);
                        List<Long> orderIdList = commonManager.scanBatchingIdList(query, i, j);
                        if (CollectionUtils.isEmpty(orderIdList)) {
                            break;
                        }
                        execute1(orderIdList, i, j);
                        count += orderIdList.size();
                        logger.info("batching_update_data_count:{}条", count);
                        if (orderIdList.size() < 1000) {
                            break;
                        }
                        lastId = orderIdList.get(orderIdList.size() - 1);
                        logger.info("batching_update_data_last_id:{},i:{},j:{}", lastId, i, j);
                        Result<String> redisData = redisManager.get("update_data_batching_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[BATCHING_UPDATE_DATA]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[OrderRoutingServiceImpl]update data failed!cause by:{0}", e);
            throw new RuntimeException("系统异常");
        }
    }

    private void execute1(List<Long> idList, int i, int j) {
        while (workQueue.remainingCapacity() < 10) {
            logger.info("workQueue size:{}", workQueue.size());
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        executorService.submit(() -> {
            try {
                commonManager.batchingBatchUpdateDate(idList, i, j);
            } catch (Exception e) {
                logger.error("[execute]failed！cause by:{}", e);
            }
        });
    }

    @Autowired
    private OrderNoUserDao noUserDao;

//    @Scheduled(fixedDelay = 60000)
    public void discountCheckTask() {
        int count = 0;
        OrderItemQuery query = buildQuery();
        AtomicInteger diff = new AtomicInteger();
        AtomicInteger same = new AtomicInteger();
        try {
            for (int i = 0; i < oldDbSize; i++) {
                for (int j = 0; j < oldTbSize; j++) {
                    List<OrderItemBatching> list = commonManager.scanBatchingList(query, i, j);
                    if (CollectionUtils.isEmpty(list)) {
                        break;
                    }
                    list.forEach(o -> compare0(o, diff, same));
                }
            }
            logger.info("COMPARE0_BATCHING_COUNT, diffCount:{}条, sameCount:{}条", diff, same);
        } catch (Exception e) {
            logger.error("[BATCHING_CHECK_TASK]failed! cause by:{0}", e);
        }
    }

    private void compare0(OrderItemBatching oldData, AtomicInteger diff, AtomicInteger same) {
        List<OrderItemBatching> newList = this.getNewList(oldData);
        if (CollectionUtils.isEmpty(newList)) {
            return;
        }
        String oldNode = JSON.toJSONString(oldData);
        String newNode = JSON.toJSONString(newList.get(0));
        if (!Objects.equals(newNode, oldNode)) {
            OrderNoUser noUser = buildParam(oldData);
            noUserDao.insert(noUser);
            diff.incrementAndGet();
            logger.info("FIND_DIFF_BATCHING0, old:{}, new:{}", oldNode, newNode);
        } else {
            same.incrementAndGet();
        }
    }

    private OrderNoUser buildParam(OrderItemBatching record) {
        OrderNoUser noUser = new OrderNoUser();
        noUser.setOrderNo(record.getOrderNo());
        noUser.setUserId(record.getUserId());
        noUser.setOrderId(record.getId());
        noUser.setType(5);
        return noUser;
    }

    private List<OrderItemBatching> getNewList(OrderItemBatching oldNode) {
        final Long userId = oldNode.getUserId();
        int dbIndex = RoutingAlgorithm.routeDB(userId, dbSize);
        int tbIndex = RoutingAlgorithm.routeTB(userId, dbSize, tbSize);
        return hintManager.selectItemBatchingById(Lists.newArrayList(oldNode.getId()), dbIndex, tbIndex);
    }

    private OrderItemQuery buildQuery() {
        Date nowDate = new Date();
        Date beforeDate0 = DateUtils.before(nowDate, Calendar.MINUTE, 3);
        Date beforeDate1 = DateUtils.before(nowDate, Calendar.MINUTE, 5);
        OrderItemQuery query = new OrderItemQuery();
        query.setUpdatedAtEnd(beforeDate0);
        query.setUpdatedAtStart(beforeDate1);
        query.setSize(1000);
        return query;
    }

    @Override
//    @Scheduled(cron = "0 0 5 * * ?")
    public void checkByOne() {
        int count = 0;
        Date now = new Date();
        AtomicInteger diffCount = new AtomicInteger();
        AtomicInteger sameCount = new AtomicInteger();
        OrderItemQuery query = new OrderItemQuery();
        query.setSortFlag(1);
//        query.setUpdatedAtEnd(new Date(1652112000000L));
//        query.setUpdatedAtStart(new Date(1652104800000L));
//        query.setUpdatedAtEnd(new Date(1652117400000L));
//        query.setUpdatedAtStart(new Date(1652112000000L));

        query.setUpdatedAtEnd(new Date(1652119200000L));
        query.setUpdatedAtStart(new Date(1652117400000L));
        query.setPageSize(1000);
        try {
            for (int i = 0; i < oldDbSize; i++) {
                for (int j = 0; j < oldTbSize; j++) {
                    int loop = 0;
                    int page = 1;
                    Long lastId = null;
                    Map<String, List<OrderItemBatching>> map = new HashMap<>(262144);
                    for (;;) {
                        query.setPageNo(page);
                        List<OrderItemBatching> dataList = commonManager.scanBatchingList(query, i, j);
                        if (CollectionUtils.isEmpty(dataList)) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, sameCount));
                            }
                            break;
                        }
                        for (OrderItemBatching record : dataList) {
                            groupOrder(record, map);
                        }
                        if (++loop >= 100) {
                            map.forEach((k, v) -> execute0(k, v, diffCount, sameCount));
                            map.clear();
                            loop = 0;
                        }
                        page++;
                        count += dataList.size();
                        logger.info("BATCHING_CHECK_BY_ONE_COUNT:{}条", count);
                        if (dataList.size() < 1000) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, sameCount));
                            }
                            break;
                        }
                        lastId = dataList.get(dataList.size() - 1).getId();
                        logger.info("batching_check_by_one_last_id:{},i:{},j:{}", lastId, i, j);

                        Result<String> redisData = redisManager.get("check_batching_by_one_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[BATCHING_CHECK]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[BatchingRoutingServiceImpl]check failed!cause by:{0}", e);
            throw new RuntimeException("系统异常");
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        int coreSize = Runtime.getRuntime().availableProcessors();
        workQueue = new ArrayBlockingQueue(128);
        executorService = new ThreadPoolExecutor(40, 40, 1, TimeUnit.MINUTES,
                workQueue, new NamedThreadFactory("order-routing"),
                new ThreadPoolExecutor.AbortPolicy());
    }
}
