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

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.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.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.ArrayList;
import java.util.List;
import java.util.Objects;
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 ConvertOrderSyncListener implements InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(ConvertOrderSyncListener.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.new.order.topic}")
    public void consumer(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        try {
            if (CollectionUtils.isEmpty(recordList)) {
                return;
            }
            while (workQueue.remainingCapacity() < 100) {
                logger.info("[BIN_LOG_SYNC_ORDER_CONSUMER_QUEUE_LIMIT]size:{}", workQueue.size());
                Thread.sleep(1000);
            }
            recordList.forEach(e -> {
                executorService.execute(() -> {
                    this.dispatch(e);
                });
            });
        } catch (Exception e) {
            logger.error("ConvertOrderSyncListener[order_fail_consumer]failed！msg:{}, cause by:{}",
                    recordList, Throwables.getStackTraceAsString(e));
        } finally {
            ack.acknowledge();
        }
    }

    private void dispatch(ConsumerRecord<String, String> record) {
        final String threadId = String.valueOf(Thread.currentThread().getId());
        Order order = parse(record);
        if (order != null) {
            logger.info("[order_kafka_consume_sync_write_321321]orderNo:{}", order.getOrderNo());
            if (Objects.equals(0, order.getMasterDataSource())) {
                return;
            }
            if (!UpgradeSellerLists.LIST.contains(order.getSellerId())) {
                logger.info("Filter un upgrade seller order. orderNo:{}", order.getOrderNo());
                return;
            }
            Result<String> redisData = cacheManager.get("res_con_order_sync_1233");
            if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                return;
            }
            logger.info("[bin_log_sync_order_consumer_asc12asd]order:{}", 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("[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);
//                }
            }
        }
    }

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

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

        // 加密与转换
        Order newOrder = (Order) streamConverter.streamOut(replace);
        orderDao.replace(newOrder);

        List<Order> filter = filter(Lists.newArrayList(newOrder));
        if (!CollectionUtils.isEmpty(filter)) {
            searchManager.indexOrder(filter, CommonConstants.ES_NEW_ORDER_INDEX);
        }
    }

    private List<Order> filter(List<Order> converted) {
        List<Order> newList = new ArrayList<>(converted.size());
        for (Order order : converted) {
            Integer bizType = order.getBizType();
            if (bizType == 4 && order.getOrderSubType() == 1) {
                OrderUserRelation relation = relationDao.selectRoot(order.getOrderNo());
                if (relation == null) {
                    newList.add(order);
                    continue;
                }
                if (Objects.equals(order.getUserId(), relation.getUserId())) {
                    newList.add(order);
                }
                continue;
            }
            if ((bizType == 5 || bizType == 6 || bizType == 7) && (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("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(40, 40, 1, TimeUnit.MINUTES,
                workQueue, new NamedThreadFactory("order-consumer"),
                new ThreadPoolExecutor.AbortPolicy());
    }
}
