package com.zm.order.tool.server.listener.reverse;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.shade.com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.zm.order.tool.server.cipher.OrderEncryptionHandler;
import com.zm.order.tool.server.constants.CommonConstants;
import com.zm.order.tool.server.constants.UpgradeSellerLists;
import com.zm.order.tool.server.dao.OrderDao;
import com.zm.order.tool.server.dao.OrderUserRelationDao;
import com.zm.order.tool.server.manager.DistributedCacheManager;
import com.zm.order.tool.server.manager.InnerApiManager;
import com.zm.order.tool.server.manager.SearchManager;
import com.zm.order.tool.server.model.KafkaConsumerMessage;
import com.zm.order.tool.server.model.Order;
import com.zm.order.tool.server.model.OrderUserRelation;
import com.zm.order.tool.server.newdatasource.NewDataSourceOrderDao;
import com.zm.order.tool.server.stream.convert.OrderStreamConverter;
import com.zm.order.tool.server.utils.JacksonUtils;
import com.zm.util.dto.Result;
import org.apache.kafka.clients.consumer.ConsumerRecord;
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.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * binlog监听
 *
 * 订阅老的topic， 写入到新的节点
 *
 * @author nianchang
 * @date 2021/12/21 2:34 下午
 */
@Component
@SuppressWarnings("all")
public class ResConvertOrderSyncListener implements InitializingBean {

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

    @Autowired
    private NewDataSourceOrderDao newOrderDao;

    @Autowired
    private OrderDao orderDao;

    @Autowired
    private DistributedCacheManager cacheManager;

    private ArrayBlockingQueue workQueue;

    private ExecutorService executorService;

    @Autowired
    private OrderStreamConverter streamConverter;

    @Autowired
    private SearchManager searchManager;

    private final String PREFIX_KEY = "binlog-sync-";

    @Autowired
    private OrderUserRelationDao relationDao;

    @Autowired
    private OrderEncryptionHandler orderEncryption;

    // 根据品牌id做数据双写过滤
    @KafkaListener(topics = "${kafka.bin.log.sync.order.topic}")
    public void consumer(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        try {
            if (CollectionUtils.isEmpty(recordList)) {
                logger.error("ConvertOrderSyncListener[res_order_consumer_empty] recordList:{}",
                        JacksonUtils.toJsonString(recordList));
                return;
            }
            while (workQueue.remainingCapacity() < 100) {
                logger.info("[bin_log_sync_res_order_consumer_queue_limit]size:{}", workQueue.size());
                Thread.sleep(1000);
            }
            for (ConsumerRecord<String, String> record : recordList) {
                executorService.execute(() -> { this.dispatch(record); });
            }
        } catch (Exception e) {
            logger.error("ConvertOrderSyncListener[res_order_consumer_fail]failed！msg:{}, cause by:{}",
                    recordList, Throwables.getStackTraceAsString(e));
        } finally {
            ack.acknowledge();
        }
    }

//
    @Autowired
    private InnerApiManager innerApiManager;

    private void dispatch(ConsumerRecord<String, String> record) {
        try {
            final String threadId = String.valueOf(Thread.currentThread().getId());
                Order order = parse(record);
                if (order == null) {
                    logger.info("[bin_log_sync_res_order_consumer_json]model:{}", record.toString());
                }

                if (order != null) {
                    logger.info("[bin_log_sync_res_order_consumer_998] orderNo:{}", order.getOrderNo());
                    if (Objects.equals(1, order.getMasterDataSource())) {
                        return;
                    }
                    if (!UpgradeSellerLists.LIST.contains(order.getSellerId())) {
                        logger.info("Filter res un upgrade seller order. orderNo:{}", order.getOrderNo());
                        return;
                    }
                    Result<String> redisData = cacheManager.get("con_order_sync_3");
                    if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                        logger.info("Get cache return order. orderNo:{}", order.getOrderNo());
                        return;
                    }
                    //如果商家已经开启升级，则不进行同步
                    Map<String, Object> map = new HashMap<>(10);
                    map.put("sellerId", order.getSellerId());
                    map.put("fieldCode", CommonConstants.SELLER_UPGRADE_CONFIG);
                    boolean sellerUpgrade = innerApiManager.configCenterSeller(map);
                    if (sellerUpgrade) {
                        logger.info("bin_log_sync_seller_upgrade, sellerId:{}, orderNo:{}", order.getSellerId(), order.getOrderNo());
                        return;
                    }
                    logger.info("[bin_log_sync_res_order_consumer_12]orderNo:{}", order.getOrderNo());
                    final Long userId = order.getUserId();
                    final String orderNo = order.getOrderNo();
                    final String key = PREFIX_KEY + orderNo + "-" + userId;
                    try {
                        this.doConsumer(threadId, order, userId, orderNo, key);
                    } catch (Exception e) {
                        logger.error("[res_order_convert_consumer]failed! order:{}, cause by:{}", order, e);
                        int retry = 3;
                        try {
                            for (;retry > 0;retry--) {
                                Thread.sleep(10);
                                this.doConsumer(threadId, order, userId, orderNo, key);
                            }
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                    } finally {
                        // 懒得用lua脚本了
//                        Result<String> result = cacheManager.get(key);
//                        if (result.isStatus() && Objects.equals(result.getData(), threadId)) {
//                            cacheManager.delete(key);
//                        }
                    }
                }
        } catch (Exception e) {
            logger.error("ConvertOrderSyncListener[res_order_dispatch_consumer_fail]failed！msg:{}, cause by:{}",
                    record, Throwables.getStackTraceAsString(e));
        }
    }

    private void doConsumer(String threadId, Order order, Long userId, String orderNo, String key) throws IOException {
//        this.lock(key, threadId);

        Order latest = newOrderDao.selectByOrderNo(orderNo, userId);
        Order replace = latest == null ? order : orderDao.selectByOrderNo(orderNo, userId);

        // 加密与转换
        Order newOrder = (Order) streamConverter.streamIn(replace);
        newOrderDao.replace(newOrder);

        List<Order> filter = filter(Lists.newArrayList(newOrder));
        if (!CollectionUtils.isEmpty(filter)) {
            searchManager.indexOrder(filter, CommonConstants.ES_NEW_INDEX_TOPIC);
        }
        logger.info("[bin_log_sync_res_order_consumer_1234as],orderNo:{}", newOrder.getOrderNo());
    }

    private List<Order> filter(List<Order> converted) {
        List<Order> newList = new ArrayList<>(converted.size());
        for (Order order : converted) {
            Integer bizType = order.getBizType();
            if (order.getOrderSubType() == 1 || order.getOrderSubType() == 2) {
                OrderUserRelation relation = relationDao.selectRoot(order.getOrderNo());
                if (relation == null) {
                    newList.add(order);
                    continue;
                }
                if (Objects.equals(order.getUserId(), relation.getUserId())) {
                    newList.add(order);
                }
                continue;
            }
            newList.add(order);
        }
        return newList;
    }

    private void lock(final String key, final String threadId) {
        while (true) {
            Result result = cacheManager.setNxEx(key, threadId, 3);
            if (result.isStatus()) {
                return;
            }
            try {
                logger.info("res concurrent lock failed! key:{}", key);
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    private Order parse(ConsumerRecord<String, String> record) {
        String dataSource = record.value();
        if (StringUtils.isEmpty(dataSource)) {
            return null;
        }
        KafkaConsumerMessage message = JSON.parseObject(dataSource, KafkaConsumerMessage.class);
        return JSON.parseObject(message.getData(), Order.class);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        workQueue = new ArrayBlockingQueue(10000);
        executorService = new ThreadPoolExecutor(64, 100, 1, TimeUnit.MINUTES,
                workQueue, new NamedThreadFactory("res-order-consumer"),
                new ThreadPoolExecutor.AbortPolicy());
    }
}
