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

import com.alibaba.fastjson.JSON;
import com.zm.order.tool.server.constants.UpgradeSellerLists;
import com.zm.order.tool.server.dao.OrderPayDetailsDao;
import com.zm.order.tool.server.manager.DistributedCacheManager;
import com.zm.order.tool.server.model.KafkaConsumerMessage;
import com.zm.order.tool.server.model.Order;
import com.zm.order.tool.server.model.OrderPayDetails;
import com.zm.order.tool.server.newdatasource.NewDataSourceOrderDao;
import com.zm.order.tool.server.newdatasource.NewDataSourceOrderPayDetailsDao;
import com.zm.order.tool.server.stream.convert.PayDetailStreamConverter;
import com.zm.util.dto.Result;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
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;

/**
 * binlog监听
 *
 * @author nianchang
 * @date 2021/12/21 2:34 下午
 */
@Component
@SuppressWarnings("all")
public class ResConvertPayDetailsSyncListener {

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

    @Autowired
    private NewDataSourceOrderPayDetailsDao newPayDetailsDao;

    @Autowired
    private OrderPayDetailsDao payDetailsDao;
    
    @Autowired
    private PayDetailStreamConverter converter;
    
    @Autowired
    private NewDataSourceOrderDao orderDao;

    @Autowired
    private DistributedCacheManager cacheManager;

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

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

//    @KafkaListener(topics = "${kafka.bin.log.sync.order.pay.details.topic}")
    public void consumer(List<ConsumerRecord<String, String>> recordList, Acknowledgment ack) {
        try {
            logger.info("[bin_log_sync_order_pay_details_consumer]before！");
            if (CollectionUtils.isEmpty(recordList)) {
                return;
            }
            for (ConsumerRecord<String, String> record : recordList) {
                OrderPayDetails model = parse(record);
                if (model != null) {
                    logger.info("[bin_log_sync_res_pay_detail_consumer_123211123]model:{}", model);
                    if (Objects.equals(1, model.getMasterDataSource())) {
                        return;
                    }
                    if (!UpgradeSellerLists.LIST.contains(model.getSellerId())) {
                        logger.info("Filter res un upgrade seller pay detail. orderNo:{}", model.getOrderNo());
                        return;
                    }
                    Result<String> redisData = cacheManager.get("con_order_pay_detail_sync_2");
                    if (redisData.isStatus() && Objects.nonNull(redisData.getData())) {
                        logger.info("Get cache return pay detail. orderNo:{}", model.getOrderNo());
                        return;
                    }
                    logger.info("[bin_log_sync_res_pay_detail_consumer_21312] model:{}", model.getOrderNo());
                    final Long id = model.getId();
                    final Long userId = model.getUserId();
                    OrderPayDetails pay = newPayDetailsDao.selectOne(id, userId);
                    OrderPayDetails replace = pay == null ? model : payDetailsDao.selectOne(id, userId);

                    if (replace != null) {
                        OrderPayDetails insert = (OrderPayDetails) converter.streamIn(replace);
                        Order order = orderDao.selectByOrderNo(replace.getOrderNo(), userId);
                        if (order != null) {
                            insert.setOrderType(order.getOrderType());
                            insert.setBizType(order.getBizType());
                            insert.setOrderSubType(order.getOrderSubType());
                            insert.setPerformanceType(order.getPerformanceType());
                        }
                        logger.info("res watch pay detail model:{}", insert);
                        newPayDetailsDao.replace(insert);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("ConvertPayDetailsSyncListener[res_pay_detail_consumer_fail]failed！msg:{}, cause by:{}", recordList, e);
        } finally {
            ack.acknowledge();
        }
    }

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