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.OrderExtendsDao;
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.OrderExtends;
import com.zm.order.tool.server.model.OrderNoUser;
import com.zm.order.tool.server.query.OrderQuery;
import com.zm.order.tool.server.service.OrderExtendsRoutingService;
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 OrderExtendsRoutingServiceImpl implements OrderExtendsRoutingService, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(OrderExtendsRoutingServiceImpl.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;

    @Override
    public void routing() {
        int count = 0;
        OrderQuery query = new OrderQuery();
//        query.setUpdatedAtEnd(new Date(1644768000000L));
        query.setSize(200);
        try {
            for (int i = 0; i < oldDbSize; i++) {
                for (int j = 0; j < oldTbSize; j++) {
                    Long lastId = null;
                    int loop = 0;
                    Map<String, List<OrderExtends>> map = new HashMap<>(262144);
                    for (;;) {
                        query.setLastId(lastId);
                        long startAt = System.currentTimeMillis();
                        List<OrderExtends> list = commonManager.scanExtendsList(query, i, j);
                        long endAt = System.currentTimeMillis() - startAt;
                        if (endAt >= 500) {
                            logger.info("extends_routing_query_time:{}ms", endAt);
                        }
                        if (CollectionUtils.isEmpty(list)) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute(k, v));
                            }
                            break;
                        }
                        for (OrderExtends record : list) {
                            groupOrder(record, map);
                        }
                        if (++loop >= 100) {
                            map.forEach((k, v) -> execute(k, v));
                            map.clear();
                            loop = 0;
                        }
                        count += list.size();
                        logger.info("extends_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("extends_routing_last_id:{},i:{},j:{}", lastId, i, j);
                        Result<String> redisData = redisManager.get("routing_extends_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[EXTENDS_ROUTING]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[OrderExtendsRoutingServiceImpl]failed!cause by:{0}", e);
            throw new RuntimeException("系统异常");
        }
    }

    private void execute(String k, List<OrderExtends> 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 = k.split("-");
                for (List<OrderExtends> subList : Lists.partition(list, 50)) {
                    hintManager.replaceExtends(subList, Integer.valueOf(split[0]), Integer.valueOf(split[1]));
                }
            } catch (Exception e) {
                logger.error("[execute]failed！cause by:{}", e);
            }
        });
    }

    private void groupOrder(OrderExtends record, Map<String, List<OrderExtends>> map) {
        // 计算在新数据库中的路由
        final Long userId = record.getUserId();
        int dbIndex = RoutingAlgorithm.routeDB(userId, dbSize);
        int tbIndex = RoutingAlgorithm.routeTB(userId, dbSize, tbSize);
        // 按照路由结果进行分组，用于批量查询
        String key = dbIndex + "-" + tbIndex;
        List<OrderExtends> 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 same = new AtomicInteger();
        OrderQuery query = new OrderQuery();
        query.setMaxId(716440803752341506L);
//        query.setExcludeSellerId(40927L);
//        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++) {
                    Long lastId = 687264332513206272L;
                    int loop = 0;
                    Map<String, List<OrderExtends>> map = new HashMap<>(262144);
                    for (;;) {
                        query.setLastId(lastId);
                        List<OrderExtends> dataList = commonManager.scanExtendsList(query, i, j);
                        if (CollectionUtils.isEmpty(dataList)) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, same));
                            }
                            break;
                        }
                        for (OrderExtends record : dataList) {
                            groupOrder(record, map);
                        }
                        if (++loop >= 100) {
                            map.forEach((k, v) -> execute0(k, v, diffCount, same));
                            map.clear();
                            loop = 0;
                        }
                        count += dataList.size();
                        logger.info("EXTENDS_CHECK_COUNT:{}条", count);
                        if (dataList.size() < 1000) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, same));
                            }
                            break;
                        }
                        lastId = dataList.get(dataList.size() - 1).getId();
                        logger.info("extends_check_last_id:{},i:{},j:{}", lastId, i, j);

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

    private void execute0(String k, List<OrderExtends> list, AtomicInteger diffCount, AtomicInteger same) {
        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(OrderExtends::getId).collect(Collectors.toList());
                List<OrderExtends> newList = hintManager.selectExtendsById(idList, dbIndex, tbIndex);
                this.compare(list, newList, diffCount, same);
                logger.info("COMPARE_EXTENDS_COUNT, diffCount:{}条, sameCount:{}条", diffCount, same);
            } catch (Exception e) {
                logger.error("[execute0]failed！cause by:{}", e);
            }
        });
    }

    @Autowired
    private OrderExtendsDao extendsDao;

    private void compare(List<OrderExtends> oldList, List<OrderExtends> newList, AtomicInteger diffCount, AtomicInteger same) {
        Map<Long, List<OrderExtends>> newMap = newList.stream().collect(Collectors.groupingBy(OrderExtends::getId));
        for (OrderExtends record : oldList) {
            List<OrderExtends> dataList = newMap.get(record.getId());
            if (CollectionUtils.isEmpty(dataList)) {
                logger.info("EXTEND_EMPTY, orderNo:{}", record.getOrderNo());
                extendsDao.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("EXTENDS_FIND_DIFF_ORDER,old:{},new:{}", oldData, newData);
            } else {
                same.incrementAndGet();
            }
        }
    }

    @Override
    public void updateData() {
        int count = 0;
        Date now = new Date();
        OrderQuery query = new OrderQuery();
        query.setUpdatedAtStart(DateUtils.before(now, Calendar.DAY_OF_YEAR, 7));
        query.setUpdatedAtEnd(now);
        query.setSize(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> idList = commonManager.scanExtendsIdList(query, i, j);
                        if (CollectionUtils.isEmpty(idList)) {
                            break;
                        }
                        execute1(idList, i, j);
                        count += idList.size();
                        logger.info("extend_update_data_count:{}条", count);
                        if (idList.size() < 1000) {
                            break;
                        }
                        lastId = idList.get(idList.size() - 1);
                        logger.info("extend_update_data_last_id:{},i:{},j:{}", lastId, i, j);
                        Result<String> redisData = redisManager.get("update_data_extend_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[EXTEND_UPDATE_DATE]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[DiscountRoutingServiceImpl]failed!cause by:{0}", e);
            throw new RuntimeException("系统异常");
        }
    }

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

    @Autowired
    private OrderNoUserDao noUserDao;

//    @Scheduled(fixedDelay = 60000)
    public void checkTask() {
        int count = 0;
        OrderQuery 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<OrderExtends> list = commonManager.scanExtendsList(query, i, j);
                    if (CollectionUtils.isEmpty(list)) {
                        break;
                    }
                    list.forEach(o -> compare0(o, diff, same));
                }
            }
            logger.info("COMPARE0_EXTEND_COUNT, diffCount:{}条, sameCount:{}条", diff, same);
        } catch (Exception e) {
            logger.error("[EXTEND_CHECK_TASK]failed! cause by:{0}", e);
        }
    }

    private void compare0(OrderExtends oldData, AtomicInteger diff, AtomicInteger same) {
        List<OrderExtends> 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_EXTEND0, old:{}, new:{}", oldNode, newNode);
        } else {
            same.incrementAndGet();
        }
    }

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

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

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

    @Override
//    @Scheduled(cron = "0 0 8 * * ?")
    public void checkByOne() {
        int count = 0;
        Date now = new Date();
        AtomicInteger diffCount = new AtomicInteger();
        AtomicInteger same = new AtomicInteger();
        OrderQuery query = new OrderQuery();
        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++) {
                    Long lastId = null;
                    int loop = 0;
                    int page = 1;
                    Map<String, List<OrderExtends>> map = new HashMap<>(262144);
                    for (;;) {
                        query.setPageNo(page);
                        List<OrderExtends> dataList = commonManager.scanExtendsList(query, i, j);
                        if (CollectionUtils.isEmpty(dataList)) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, same));
                            }
                            break;
                        }
                        for (OrderExtends record : dataList) {
                            groupOrder(record, map);
                        }
                        if (++loop >= 100) {
                            map.forEach((k, v) -> execute0(k, v, diffCount, same));
                            map.clear();
                            loop = 0;
                        }
                        page++;
                        count += dataList.size();
                        logger.info("EXTENDS_CHECK_BY_ONE_COUNT:{}条", count);
                        if (dataList.size() < 1000) {
                            if (!CollectionUtils.isEmpty(map)) {
                                map.forEach((k, v) -> execute0(k, v, diffCount, same));
                            }
                            break;
                        }
                        lastId = dataList.get(dataList.size() - 1).getId();
                        logger.info("extends_check_by_one_last_id:{},i:{},j:{}", lastId, i, j);

                        Result<String> redisData = redisManager.get("check_extends_shut_down_key");
                        if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                            logger.info("[EXTENDS_CHECK]shut down！");
                            return;
                        }
                    }
                }
            }
        } catch (Exception e) {
            logger.error("[OrderExtendsRoutingServiceImpl]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(20, 20, 1, TimeUnit.MINUTES,
                workQueue, new NamedThreadFactory("order-routing"),
                new ThreadPoolExecutor.AbortPolicy());
    }
}
