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

import com.alibaba.fastjson.JSON;
import com.zm.order.tool.server.cipher.RefundEncryptionHandler;
import com.zm.order.tool.server.constants.UpgradeSellerLists;
import com.zm.order.tool.server.dao.OrderDao;
import com.zm.order.tool.server.dao.RefundDao;
import com.zm.order.tool.server.enums.DeliveryPlatformEnum;
import com.zm.order.tool.server.manager.SearchRefundManager;
import com.zm.order.tool.server.model.KafkaConsumerMessage;
import com.zm.order.tool.server.model.Order;
import com.zm.order.tool.server.model.Refund;
import com.zm.order.tool.server.newdatasource.NewDataSourceRefundDao;
import com.zm.order.tool.server.stream.convert.RefundStreamConverter;
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.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;

/**
 * <p>
 *
 * </p>
 *
 * @author nianchang
 * @date 2023/03/21 1:57 下午
 */
@Component
@SuppressWarnings("all")
public class ConvertRefundSyncListener implements InitializingBean {

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

    @Autowired
    private RefundDao refundDao;

    @Autowired
    private RefundEncryptionHandler refundEncryptionHandler;

    @Autowired
    private NewDataSourceRefundDao newRefundDao;

    @Autowired
    private RefundStreamConverter converter;

    @Autowired
    private SearchRefundManager refundManager;

    private ArrayBlockingQueue workQueue;

    @Autowired
    private OrderDao orderDao;

    private ExecutorService executorService;

//    @KafkaListener(topics = "${kafka.bin.log.sync.new.order.refunds.topic}")
    public void consumer(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        try {
            logger.info("[bin_log_sync_refund_consumer]before！");
            if (CollectionUtils.isEmpty(recordList)) {
                return;
            }
            while (workQueue.remainingCapacity() < 100) {
                logger.info("[bin_log_sync_refund_consumer_queue_limit]size:{}", workQueue.size());
                Thread.sleep(200);
            }
            recordList.forEach(e -> {executorService.execute(() -> { this.doConsumer(e); });});
        } catch (Exception e) {
            logger.error("ConvertRefundSyncListener[refund_fail_consumer]failed！msg:{}, cause by:{}", recordList, e);
        } finally {
            ack.acknowledge();
        }
    }

    private void doConsumer(ConsumerRecord<String, String> record) {
        Refund model = parse(record);
        if (model != null) {
            logger.info("[bin_log_sync_refund_consumer_0]model:{}", model);
            if (Objects.equals(0, model.getMasterDataSource())) {
                return;
            }
            if (!UpgradeSellerLists.LIST.contains(model.getSellerId())) {
                logger.info("Filter un upgrade seller refund. orderNo:{}", model.getOrderNo());
                return;
            }
            logger.info("[bin_log_sync_refund_consumer_xcqwqw]model:{}", model.getRefundNo());
            final Long id = model.getId();
            Refund refund = refundDao.selectById(id);
            Refund replace = refund == null ? model : newRefundDao.selectById(id);
            if (replace != null) {
                Refund insert = (Refund) converter.streamOut(replace);

                // 外卖单单独映射履约平台
                if (insert.getOrderType() == 3) {
                    Order order = orderDao.selectByOrderNo(replace.getOrderNo(), replace.getUserId());
                    if (order != null && !StringUtils.isEmpty(order.getPerformancePlatform())) {
                        DeliveryPlatformEnum byName = DeliveryPlatformEnum.findByName(order.getPerformancePlatform());
                        if (byName != null) {
                            logger.info("[refund_convert_performance_type3] order:{}", order);
                            insert.setPerformanceType(byName.getCode());
                        }
                    }
                }
                refundDao.replace(insert);
//                    refundManager.indexRefund(insert, CommonConstants.ORDER_CENTER_REFUND_TOPIC);
            }
        }
    }

    private Refund 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(), Refund.class);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        workQueue = new ArrayBlockingQueue(1000);
        executorService = new ThreadPoolExecutor(4, 8, 1, TimeUnit.MINUTES,
                workQueue, new NamedThreadFactory("refund-consumer"),
                new ThreadPoolExecutor.AbortPolicy());
    }
}
