package com.liquidnet.service.consumer.kylin.receiver;

import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.goblin.constant.GoblinRedisConst;
import com.liquidnet.service.goblin.constant.GoblinStatusConst;
import com.liquidnet.service.goblin.dto.vo.GoblinOrderSkuVo;
import com.liquidnet.service.goblin.dto.vo.GoblinStoreOrderVo;
import com.mongodb.BasicDBObject;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.convert.MongoConverter;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;

@Slf4j
public abstract class AbstractOrderCloseReceiver implements StreamListener<String, MapRecord<String, String, String>> {
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private MongoConverter mongoConverter;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String redisStreamKey = this.getRedisStreamKey();
        log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
        boolean result = this.consumerMessageHandler(message.getValue());
        log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());

        try {
            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
        } catch (Exception e) {
            log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
        }
        try {
            stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
        } catch (Exception e) {
            log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
        }
    }

    private boolean consumerMessageHandler(Map<String, String> messageMap) {
        try {
            String orderCode = messageMap.get("id"), type = messageMap.get("type"), time = messageMap.get("time");

            LocalDateTime now = LocalDateTime.now(), closeTime = now.minusMinutes(5);
            LocalDateTime createdAt = StringUtils.isEmpty(time) ? closeTime : LocalDateTime.parse(time);
            long durationToMillis = Duration.between(createdAt, closeTime).toMillis();

            if (durationToMillis >= 0) {
                return checkOrderTime(orderCode, type);
            } else {
                try {
                    Thread.sleep(Math.abs(durationToMillis));
                } catch (InterruptedException ignored) {
                }
                return consumerMessageHandler(messageMap);
            }
        } catch (Exception e) {
            String redisStreamKey = this.getRedisStreamKey();
            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", redisStreamKey, messageMap, e);
            stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(messageMap).withStreamKey(redisStreamKey));
            return false;
        }
    }

    protected abstract String getRedisStreamKey();

    protected abstract String getRedisStreamGroup();

    public boolean checkOrderTime(String valueData, String type) {
        LocalDateTime now = LocalDateTime.now();
        try {
            if (type.equals("GOBLIN")) {
                String[] orderIds = getMasterCode(valueData);
                for (String orderId : orderIds) {
                    GoblinStoreOrderVo orderVo = getGoblinOrder(orderId);
                    if (orderVo.getStatus().equals(GoblinStatusConst.Status.ORDER_STATUS_0.getValue())) {//订单回滚
                        LinkedList<String> sqls = CollectionUtil.linkedListString();
                        LinkedList<Object[]> sqlDataOrder = CollectionUtil.linkedListObjectArr();
                        LinkedList<Object[]> sqlDataSku = CollectionUtil.linkedListObjectArr();
                        sqls.add(SqlMapping.get("goblin_order.close.order"));
                        sqls.add(SqlMapping.get("goblin_order.close.sku"));
                        for (String orderSkuId : orderVo.getOrderSkuVoIds()) {
                            GoblinOrderSkuVo skuVo = getGoblinOrderSkuVo(orderSkuId);
                            //订单详情
                            skuVo.setStatus(GoblinStatusConst.Status.ORDER_STATUS_5.getValue());
                            updateGoblinOrderSkuVo(skuVo.getOrderSkuId(), skuVo);
                            setGoblinOrderSku(skuVo.getOrderSkuId(), skuVo);
                            //库存&限购&&待支付订单
                            String pre = GoblinStatusConst.MarketPreStatus.getPre(skuVo.getSkuId());
                            incrSkuStock(pre, skuVo.getSkuId(), skuVo.getNum());
                            decrSkuCountByUid(orderVo.getUserId(), skuVo.getSkuId(), skuVo.getNum());
                            //mysql
                            sqlDataSku.add(new Object[]{
                                    skuVo.getStatus(), now, orderVo.getOrderId(), now, now
                            });
                        }
                        //订单
                        orderVo.setStatus(GoblinStatusConst.Status.ORDER_STATUS_5.getValue());
                        updateGoblinStoreOrderVo(orderVo.getOrderId(), orderVo);
                        setGoblinOrder(orderVo.getOrderId(), orderVo);
                        //mysql
                        sqlDataOrder.add(new Object[]{
                                orderVo.getStatus(), now, now, "超时关闭", orderVo.getOrderId(), now, now
                        });
                        //执行sql
                        sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_ORDER_CLOSE.getKey(),
                                SqlMapping.gets(sqls, sqlDataOrder, sqlDataSku));
                    }
                }
            }
        }catch (Exception e){
            log.error("库存回滚异常valueData={}",valueData, e);
        }
        return true;
    }

    private UpdateResult updateGoblinStoreOrderVo(String orderId, GoblinStoreOrderVo data) {
        BasicDBObject object = cloneBasicDBObject().append("$set", mongoConverter.convertToMongoType(data));
        return mongoTemplate.getCollection(GoblinStoreOrderVo.class.getSimpleName()).updateOne(
                Query.query(Criteria.where("orderId").is(orderId)).getQueryObject(),
                object);
    }

    private UpdateResult updateGoblinOrderSkuVo(String orderSkuId, GoblinOrderSkuVo data) {
        BasicDBObject object = cloneBasicDBObject().append("$set", mongoConverter.convertToMongoType(data));
        return mongoTemplate.getCollection(GoblinOrderSkuVo.class.getSimpleName()).updateOne(
                Query.query(Criteria.where("orderSkuId").is(orderSkuId)).getQueryObject(),
                object);
    }

    public String[] getMasterCode(String masterCode) {
        String redisKey = GoblinRedisConst.REDIS_GOBLIN_ORDER_MASTER.concat(masterCode);
        Object obj = redisUtil.get(redisKey);
        if (obj == null) {
            return null;
        } else {
            return ((String) obj).split(",");
        }
    }

    // 获取 订单相关vo
    public GoblinStoreOrderVo getGoblinOrder(String orderId) {
        String redisKey = GoblinRedisConst.REDIS_GOBLIN_ORDER.concat(orderId);
        Object obj = redisUtil.get(redisKey);
        if (obj == null) {
            return null;
        } else {
            return (GoblinStoreOrderVo) obj;
        }
    }

    // 获取 订单相关Skuvo
    public GoblinOrderSkuVo getGoblinOrderSkuVo(String orderSkuId) {
        String redisKey = GoblinRedisConst.REDIS_GOBLIN_ORDER_SKU.concat(orderSkuId);
        Object obj = redisUtil.get(redisKey);
        if (obj == null) {
            return null;
        } else {
            return (GoblinOrderSkuVo) obj;
        }
    }

    // 赋值 订单相关Skuvo
    public void setGoblinOrderSku(String orderSkuId, GoblinOrderSkuVo vo) {
        String redisKey = GoblinRedisConst.REDIS_GOBLIN_ORDER_SKU.concat(orderSkuId);
        redisUtil.set(redisKey, vo);
    }

    public int incrSkuStock(String marketPre, String skuId, Integer stock) {
        String rk = GoblinRedisConst.REAL_STOCK_SKU;
        if (marketPre != null && !marketPre.equals("null")) {
            rk = rk.concat(marketPre + ":");
        }
        rk = rk.concat(skuId);
        return (int) redisUtil.incr(rk, stock);
    }

    // 减少 用户sku购买个数
    public int decrSkuCountByUid(String uid, String skuId, int number) {
        String redisKey = GoblinRedisConst.REDIS_GOBLIN_BUY_COUNT.concat(uid + ":skuId:" + skuId);
        return (int) redisUtil.decr(redisKey, number);
    }

    // 赋值 订单相关vo
    public void setGoblinOrder(String orderId, GoblinStoreOrderVo vo) {
        String redisKey = GoblinRedisConst.REDIS_GOBLIN_ORDER.concat(orderId);
        redisUtil.set(redisKey, vo);
    }

    public void sendMsgByRedis(String streamKey, String jsonMsg) {
        HashMap<String, String> map = CollectionUtil.mapStringString();
        map.put("message", jsonMsg);
        stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(streamKey));
    }

    private static final BasicDBObject basicDBObject = new BasicDBObject();

    public static BasicDBObject cloneBasicDBObject() {
        return (BasicDBObject) basicDBObject.clone();
    }


}