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

import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.zm.order.tool.server.constants.CommonConstants;
import com.zm.order.tool.server.dao.CompareRecordDao;
import com.zm.order.tool.server.dao.OrderDao;
import com.zm.order.tool.server.dao.OrderUserRelationDao;
import com.zm.order.tool.server.enums.CompareRecordStatusEnum;
import com.zm.order.tool.server.enums.SyncFlagEnum;
import com.zm.order.tool.server.manager.CommonManager;
import com.zm.order.tool.server.manager.DingTalkManager;
import com.zm.order.tool.server.manager.OrderManager;
import com.zm.order.tool.server.manager.SearchManager;
import com.zm.order.tool.server.model.CompareRecord;
import com.zm.order.tool.server.model.Order;
import com.zm.order.tool.server.query.OrderQuery;
import com.zm.order.tool.server.service.ExceptionConstants;
import com.zm.order.tool.server.service.OrderSyncService;
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.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


/**
 * @Description 订单同步服务实现
 * @Author: ph
 * @Date: 2019/11/1 11:02
 */
@Service
@SuppressWarnings("all")
public class OrderSyncServiceImpl implements OrderSyncService, InitializingBean {

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

    @Autowired
    private OrderDao orderDao;

    @Autowired
    private OrderManager orderManager;

    @Autowired
    private SearchManager searchManager;

    @Autowired
    private CompareRecordDao compareRecordDao;

    @Autowired
    private DingTalkManager dingTalkManager;

    private ExecutorService executorService;

    @Autowired
    private CommonManager commonManager;

    @Autowired
    private OrderUserRelationDao relationDao;

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

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

    @Value("${application.server.profile}")
    private String profile;

    @Override
    public Result sync(String orderNo, Long userId, Integer version) {
        Result result = new Result();
        try {
            // 如果订单为空，可能接收到消息的时候还没有落库，没关系，这里不处理，会有定时任务去补偿
            Order order = orderDao.selectByOrderNo(orderNo, userId);
            if (Objects.nonNull(order)) {
                // 只有当数据库version=消息version时才处理
                if (Objects.equals(version, order.getVersion())) {
                    if (!searchManager.indexOrder(order, null)) {
                        logger.error("sync es failed, orderNo:{}", orderNo);
                        result.setException(ExceptionConstants.SYNC_ES_ERROR);
                        return result;
                    }
                }
                orderDao.resetSyncFlag(order);
            } else {
                logger.error("sync order(orderNo:{}, userId:{}) not exists",
                        orderNo, userId);
            }
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("OrderSyncServiceImpl[sync] failed, request:[orderNo:{}, userId:{}, version:{}], cause by:{}",
                    orderNo, userId, version, Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SYNC_ORDER_ES_ERROR);
        }
        return result;
    }

    @Override
    public Result dispatch() {
        Result result = new Result();
        try {
            for (int i = 0; i < dbSize; i++) {
                for (int j = 0; j < tbSize; j++) {
                    executorService.submit(new MonitorThread(i, j, this));
                }
            }
            result.setStatus(true);
        } catch (Throwable t) {
            logger.error("OrderSyncServiceImpl[dispatch] failed, cause by:{}", Throwables.getStackTraceAsString(t));
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    @Override
    public Result compare(Integer dbNo, Integer tbNo) {
        Result result = new Result();
        try {
            // 1.获取最近的一条记录，生成新的记录，插入，占领该时间段
            CompareRecord lastRecord = compareRecordDao.selectLast(dbNo, tbNo);
            CompareRecord newRecord = this.genNextCompareRecord(lastRecord, dbNo, tbNo);
            compareRecordDao.insert(newRecord);
            // 2.开始工作
            CompareRecord update = new CompareRecord();
            update.setId(newRecord.getId());
            OrderQuery orderQuery = new OrderQuery();
            orderQuery.setUpdatedAtStart(newRecord.getStartAt());
            orderQuery.setUpdatedAtEnd(newRecord.getEndAt());
            int delayCount = 0;
            int pageNo = 1;
            final int pageSize = 200;
            List<String> failCompensateList = Lists.newArrayList();
            int successCompensateCount = 0;
            while (true) {
                orderQuery.setPageNo(pageNo);
                orderQuery.setPageSize(pageSize);
                orderQuery.setSyncFlag(SyncFlagEnum.SYNC.getValue());
                List<Order> orderList = orderManager.scanOrder(orderQuery, dbNo, tbNo);
                if (!CollectionUtils.isEmpty(orderList)) {
                    logger.error("[SYNC] orderNo:{}", orderList.get(0).getOrderNo());
                    delayCount += orderList.size();
                    List<Order> realOrderList = Lists.newArrayList();
                    orderList.forEach(order -> {
                        Order realOrder = orderDao.selectByOrderNo(order.getOrderNo(), order.getUserId());
                        if (realOrder != null) {
                            realOrderList.add(realOrder);
                        }
                    });
                    List<Order> resetOrderList = Lists.newArrayList();
                    if (searchManager.indexOrder(realOrderList)) {
                        successCompensateCount += orderList.size();
                        resetOrderList.addAll(orderList);
                    } else {
                        failCompensateList.addAll(orderList.stream().map(Order::getOrderNo).collect(Collectors.toList()));
                    }
                    // 重置order表状态
                    this.reset(resetOrderList);
                    if (orderList.size() != pageSize) {
                        break;
                    }
                } else {
                    break;
                }
            }
            // 3.记录工作结果
            if (delayCount > 0) {
                // 告警
                warning(successCompensateCount, failCompensateList.size());
                if (!CollectionUtils.isEmpty(failCompensateList)) {
                    Map<String, Object> extraMap = Maps.newHashMap();
                    extraMap.put("failCompensateList", failCompensateList);
                    update.setExtraMap(extraMap);
                }
            }
            update.setDelayCount(delayCount);
            update.setExecuteAt(new Date());
            update.setStatus(CompareRecordStatusEnum.END.getStatus());
            compareRecordDao.update(update);
            result.setStatus(true);
        } catch (Throwable throwable) {
            logger.error("OrderSyncServiceImpl[compare] failed, cause by:{}", Throwables.getStackTraceAsString(throwable));
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    @Override
    public Result scanSync(int before) {
        int count = 0;
        final int limit = 200;
        Result result = new Result();
        OrderQuery orderQuery = buildOrderQuery(before);
        logger.info("orderQuery:{}", orderQuery.toString());
        List<Integer> deleteTypeList = Lists.newArrayList(0, 1, 2);
        try {
            long startAt = System.currentTimeMillis();
            for (int i = 0; i < dbSize; i++) {
                for (int j = 0; j < tbSize; j++) {
                    List<Long> idList = commonManager.scanOrderId(orderQuery, i, j);
                    logger.info("idList:{}", idList);
                    if (CollectionUtils.isEmpty(idList)) {
                        continue;
                    }
                    for (List<Long> executeList : Lists.partition(idList, limit)) {
                        List<Order> orderList = commonManager.listOrderBySyncFlag(executeList, i, j, 1, deleteTypeList);
                        if (CollectionUtils.isEmpty(orderList)) {
                            continue;
                        }
                        List<Order> filterList = filterSubOrder(orderList);
                        if (CollectionUtils.isEmpty(filterList)) {
                            continue;
                        }
                        searchManager.indexOrder(filterList);
                        List<Long> updateIdList = filterList.stream().map(Order::getId).collect(Collectors.toList());
                        logger.info("updateIdList:{}", updateIdList);
                        commonManager.batchUpdateSyncFlag(updateIdList, 0, i, j);
                        count += filterList.size();
                    }
                    logger.info("scanSync[cycle_task_sync_es]success! orderpro_:{}, orders_:{}, count:{}", i + 1, j, count);
                }
            }
            logger.info("[cycle_task_execute_time]success! time:{}ms", System.currentTimeMillis() - startAt);
            result.setStatus(true);
        } catch (Exception e) {
            logger.error("OrderSyncServiceImpl[scanSync]failed! cause by:{0}", e);
            result.setException(ExceptionConstants.SYSTEM_ERROR);
        }
        return result;
    }

    @Override
    public void retailSync() {
        int count = 0;
        OrderQuery query = build();
        try {
            for (int i = 0; i < dbSize; i++) {
                for (int j = 0; j < tbSize; j++) {
                    List<Order> orderList = commonManager.scanAppointSellerList(query, i, j);
                    if (CollectionUtils.isEmpty(orderList)) {
                        continue;
                    }

                    for (Order order : orderList) {
                        count++;
                        searchManager.indexOrder(order, null);
                    }

                    List<String> orderNoList = orderList.stream().map(Order::getOrderNo).collect(Collectors.toList());
                    logger.info("[retail_order_sync]success. count:{}, orderNoList:{}", count, orderNoList);
                    if (orderList.size() < 200) {
                        continue;
                    }
                }
            }
        } catch (Exception e) {

        }
    }

    private OrderQuery build() {
        OrderQuery query = new OrderQuery();
        query.setPageNo(1);
        query.setPageSize(200);
        query.setSellerId(23200L);
        query.setSortFlag(1);
        Date now = new Date();
        query.setUpdatedAtStart(DateUtils.before(now, Calendar.HOUR_OF_DAY, 1));
        query.setUpdatedAtEnd(now);
        return query;
    }


    private List<Order> filterSubOrder(List<Order> orderList) {
        List<Order> newOrderList = new ArrayList<>();
        for (Order order : orderList) {
            if (isMultiUser(order)) {
                Map<String, Object> extraMap = order.getExtraMap();
                if (CollectionUtils.isEmpty(extraMap) || Objects.isNull(extraMap.get("masterId"))) {
                    newOrderList.add(order);
                    continue;
                }
                Long masterId = Long.valueOf(extraMap.get("masterId").toString());
                if (Objects.equals(masterId, order.getId())) {
                    newOrderList.add(order);
                }
            } else {
                newOrderList.add(order);
            }
        }
        return newOrderList;
    }

    private boolean isMultiUser(Order order) {
        if (order != null) {
            // 单一场景：餐饮-多人点餐业务
            if (CommonConstants.MULTI_BIZ_TYPES.contains(order.getBizType())
                    && Objects.equals(1, order.getOrderSubType())) {
                return true;
            }
        }
        return false;
    }

    private OrderQuery buildOrderQuery(int before1) {
        OrderQuery orderQuery = new OrderQuery();
        Date before = DateUtils.before(new Date(), Calendar.MINUTE, before1 + 1);
        Date before2 = DateUtils.before(new Date(), Calendar.MINUTE, before1 - 1);
        orderQuery.setUpdatedAtStart(DateUtils.getEndDate0(before));
        orderQuery.setUpdatedAtEnd(DateUtils.getStartDate0(before2));
        orderQuery.setSize(3000);
        return orderQuery;
    }

    public static void main(String[] args) {
        System.out.println(DateUtils.before(new Date(), Calendar.HOUR_OF_DAY, 1).getTime());
    }

    /**
     * 告警
     *
     * @param successCompensateCount 补偿成功数量
     * @param failCompensateCount    补偿失败列表
     */
    public void warning(int successCompensateCount, int failCompensateCount) {
        if (failCompensateCount == 0) {
            return;
        }
        StringBuilder sb = new StringBuilder(profile + "订单中心同步ES异常");
        sb.append(",")
                .append("异常数量:")
                .append(successCompensateCount + failCompensateCount)
                .append(",")
                .append("补偿成功数量:")
                .append(successCompensateCount)
                .append(",")
                .append("补偿失败数量:")
                .append(failCompensateCount);
        Result result = dingTalkManager.sendTextMessage(sb.toString());
        if (!result.isStatus()) {
            logger.error("OrderSyncServiceImpl[warning] failed, cause by:{}",
                    result.getMessage());
        }
    }

    /**
     * 生成下一个比较记录
     *
     * @param record
     * @return
     */
    public CompareRecord genNextCompareRecord(CompareRecord record, Integer dbNo, Integer tbNo) {
        CompareRecord nextRecord = new CompareRecord();
        Date startAt;
        int no;
        if (Objects.isNull(record)) {
            no = 1;
            startAt = DateUtils.before(Calendar.SECOND, 10);
        } else {
            no = record.getNo() + 1;
            startAt = record.getEndAt();
        }
        nextRecord.setStatus(CompareRecordStatusEnum.INIT.getStatus());
        nextRecord.setNo(no);
        nextRecord.setDbNo(dbNo);
        nextRecord.setTbNo(tbNo);
        nextRecord.setStartAt(startAt);
        nextRecord.setEndAt(DateUtils.before(new Date(), Calendar.SECOND, 5));
        nextRecord.setDelayCount(0);

        //构建默认值
        if (Objects.isNull(nextRecord.getNo())) {
            nextRecord.setNo(0);
        }
        if (Objects.isNull(nextRecord.getDbNo())) {
            nextRecord.setDbNo(0);
        }
        if (Objects.isNull(nextRecord.getTbNo())) {
            nextRecord.setTbNo(0);
        }
        if (Objects.isNull(nextRecord.getDelayCount())) {
            nextRecord.setDelayCount(0);
        }
        if (Objects.isNull(nextRecord.getStatus())) {

            nextRecord.setStatus(0);
        }
        return nextRecord;
    }

    /**
     * 重置同步标识
     *
     * @param resetOrderList
     */
    private void reset(List<Order> resetOrderList) {
        for (Order order : resetOrderList) {
            Order update = new Order();
            update.setId(order.getId());
            update.setUserId(order.getUserId());
            update.setVersion(order.getVersion());
            orderDao.resetSyncFlag(update);
        }
    }

    public class MonitorThread implements Runnable {

        private Integer dbNo;

        private Integer tbNo;

        private OrderSyncService orderSyncService;

        public MonitorThread(Integer dbNo, Integer tbNo, OrderSyncService orderSyncService) {
            this.dbNo = dbNo;
            this.tbNo = tbNo;
            this.orderSyncService = orderSyncService;
        }

        @Override
        public void run() {
            orderSyncService.compare(dbNo, tbNo);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        int coreSize = Runtime.getRuntime().availableProcessors();
        executorService = new ThreadPoolExecutor(coreSize,
                coreSize * 2,
                1,
                TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(1024),
                new NamedThreadFactory("sync-es"),
                new ThreadPoolExecutor.AbortPolicy());
    }
}
