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.commons.lang.util.JsonUtils;
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.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;

import static com.liquidnet.commons.lang.util.DateUtil.DTF_YMD_HMS;

@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) {
        log.debug("CONSUMER ORDER_CLOSE [streamKey:{},messageId:{},stream:{},body:{}]",
                this.getRedisStreamKey(), message.getId(), message.getStream(), message.getValue());
        boolean result = this.consumerOrderCloseHandler(message.getValue());
        log.info("CONSUMER ORDER_CLOSE RESULT:{} ==> MESSAGE_ID:{}", result, message.getId());

        try {
            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
            stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
        } catch (Exception e) {
            log.error("#CONSUMER SQL RESULT:{} ==> DEL_REDIS_QUEUE_MSG_EXCEPTION[MESSAGE_ID:{},MSG:{}]", result, message.getId(), JsonUtils.toJson(message), e);
        } finally {
            try {
                stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
                stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
            } catch (Exception ignored) {
            }
        }
    }

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

            long durationToMillis = Duration.between(createdAt.plusMinutes(5), LocalDateTime.now()).toMillis();

            if (durationToMillis > 0) {
                return checkOrderTime(orderCode, type);
            } else {
                try {
                    Thread.sleep(Math.abs(durationToMillis));
                } catch (InterruptedException ignored) {
                }
                return consumerOrderCloseHandler(messageMap);
            }
        } catch (Exception e) {
            log.error("CONSUMER ORDER_CLOSE FAIL ==> {}", e.getMessage(), e);
            stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(messageMap).withStreamKey(this.getRedisStreamKey()));
            return false;
        }
    }

    protected abstract String getRedisStreamKey();

    protected abstract String getRedisStreamGroup();

    public boolean checkOrderTime(String valueData, String type) {
        LocalDateTime now = LocalDateTime.now();
        if (type.equals("GOBLIN")) {
            String[] orderIds = getMasterCode(valueData);
            for (String orderId : orderIds) {
                GoblinStoreOrderVo orderVo = getGoblinOrder(orderId);
                if (LocalDateTime.parse(orderVo.getCreatedAt(), DTF_YMD_HMS).plusMinutes(5).isBefore(now)) {
                    continue;
                }
                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));
                }
            }
        }
        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();
    }


}