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

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.core.type.TypeReference;
import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.commons.lang.util.HttpUtil;
import com.liquidnet.commons.lang.util.JsonUtils;
import com.liquidnet.service.base.SqlMapping;
import com.liquidnet.service.base.constant.MQConst;
import com.liquidnet.service.consumer.kylin.utils.GoblinNftUtils;
import com.liquidnet.service.goblin.constant.GoblinRedisConst;
import com.liquidnet.service.goblin.constant.GoblinStatusConst;
import com.liquidnet.service.goblin.dto.vo.*;
import com.liquidnet.service.goblin.param.BackCouponParam;
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.beans.factory.annotation.Value;
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.mongodb.core.query.Update;
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 org.springframework.util.MultiValueMap;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;

@Slf4j
public abstract class AbstractOrderCloseReceiver implements StreamListener<String, MapRecord<String, String, String>> {
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private GoblinNftUtils goblinNftUtils;
    @Autowired
    private MongoTemplate mongoTemplate;
    @Autowired
    private MongoConverter mongoConverter;
    @Value("${liquidnet.service.candy.url}")
    private String candyUrl;

    @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) {
                switch (type) {
                    case "GOBLIN":
                        return checkOrderTime(orderCode, type);
                    case "NFT":
                        return checkNftOrderTime(orderCode);
                    default:
                        return false;
                }
            } 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, skuVo.getOrderSkuId(), now, now
                            });
                        }
                        //订单
                        orderVo.setStatus(GoblinStatusConst.Status.ORDER_STATUS_5.getValue());
                        updateGoblinStoreOrderVo(orderVo.getOrderId(), orderVo);
                        setGoblinOrder(orderVo.getOrderId(), orderVo);
                        //店铺券退券
                        if (!(orderVo.getStoreCouponId().equals("") || orderVo.getStoreCouponId() == null)) {
                            BackCouponParam backCouponParam = BackCouponParam.getNew();
                            backCouponParam.setuCouponIds(orderVo.getStoreCouponId());
                            backCouponParam.setUid(orderVo.getUserId());
                            backStoreCoupon(backCouponParam);
                        }
                        //平台券
                        if (!(orderVo.getUcouponId().equals("") || orderVo.getUcouponId() == null)) {
                            backCoupon(orderVo.getUcouponId(), orderVo.getUserId());
                        }
                        //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;
    }

    public boolean checkNftOrderTime(String orderId) {
        LocalDateTime now = LocalDateTime.now();
        try {
            GoblinNftOrderVo nftOrder = goblinNftUtils.getNftOrder(orderId);
            if (nftOrder.getStatus().equals(GoblinStatusConst.NftStatus.ORDER_STATUS_1.getValue())) {
                // 库存购买数量回滚
                goblinNftUtils.decrSkuCountByUid(nftOrder.getUserId(), nftOrder.getSkuId(), nftOrder.getNum());
                LocalDateTime nowTime = LocalDateTime.now();
                if (StringUtils.isEmpty(nftOrder.getBoxSkuId())) {// 购买藏品
                    GoblinListCollectVo goblinListCollectVo = goblinNftUtils.getCollectByNowNext(nowTime, nftOrder.getSkuId());
                    String listId = (null == goblinListCollectVo) ? null : goblinListCollectVo.getListId();
                    log.debug("listId : {}",listId);
                    goblinNftUtils.incrSkuStock(listId, nftOrder.getSkuId(), nftOrder.getNum());
                } else {// 购买盲盒
                    GoblinListCollectVo goblinListCollectVo = goblinNftUtils.getCollectByNowNext(nowTime, nftOrder.getBoxSkuId());
                    String listId = (null == goblinListCollectVo) ? null : goblinListCollectVo.getListId();
                    log.debug("listId : {}",listId);
                    goblinNftUtils.incrSkuStock(listId, nftOrder.getBoxSkuId(), nftOrder.getNum());
                }

                // 订单状态
                nftOrder.setStatus(GoblinStatusConst.NftStatus.ORDER_STATUS_3.getValue());
                nftOrder.setUpdatedAt(now);
                nftOrder.setCancelTime(now);
                nftOrder.setCancelReason("超时支付关闭");
                goblinNftUtils.setNftOrder(nftOrder);
                goblinNftUtils.updateGoblinNftOrderVo(nftOrder);
                LinkedList<String> sqls = CollectionUtil.linkedListString();
                LinkedList<Object[]> sqlDataOrder = CollectionUtil.linkedListObjectArr();
                sqls.add(SqlMapping.get("goblin_nft_order.update.close"));
                sqlDataOrder.add(new Object[]{
                        nftOrder.getStatus(), now, now, "超时支付关闭", nftOrder.getOrderId(), now, now
                });
                sendMsgByRedis(MQConst.GoblinQueue.GOBLIN_ORDER_CLOSE.getKey(),
                        SqlMapping.gets(sqls, sqlDataOrder));
            }
            return true;
        } catch (Exception e) {
            log.error("checkNftOrderTime异常 [valueData:{}, e:{}]", orderId, e);
            return false;
        }
    }

    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));
    }

    public Boolean backStoreCoupon(BackCouponParam params) {
        try {
            List<GoblinUserCouponVo> voList = getUserCouponVos(params.getUid());
            for (GoblinUserCouponVo vo : voList) {
                if (vo.getUcouponId().equals(params.getuCouponIds())) {
                    if (LocalDateTime.now().isBefore(vo.getDuedAt())) {
                        vo.setState(1);
                        vo.setUsedFor("");
                        setUserCouponVos(params.getUid(), voList);
                        GoblinUserCouponBasicVo basicVo = GoblinUserCouponBasicVo.getNew();
                        basicVo.setState(1);
                        basicVo.setUsedFor("");
                        changeCouponVos(vo.getUcouponId(), basicVo);
                        sendMsgByRedis(MQConst.GoblinQueue.SQL_COUPON.getKey(),
                                SqlMapping.get("goblin_user_coupon.updateState", vo.getState(), vo.getUsedFor(), LocalDateTime.now(), vo.getUcouponId()));
                    }
                    break;
                }
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    public void backCoupon(String uCouponId, String uid) {
        try {
            com.liquidnet.service.candy.param.BackCouponParam param = com.liquidnet.service.candy.param.BackCouponParam.getNew();
            param.setuCouponIds(uCouponId);
            param.setUid(uid);

            MultiValueMap<String, String> header = CollectionUtil.linkedMultiValueMapStringString();
            header.add("Accept", "application/json;charset=UTF-8");
            ArrayList<com.liquidnet.service.candy.param.BackCouponParam> params = new ArrayList();
            params.add(param);
            String jsonString = JSON.toJSONString(params);
            String returnData = HttpUtil.postRaw(candyUrl + "/candy-coupon/useBack", jsonString, header);
        } catch (Exception e) {
            log.error("回退券ERROR:{}", e);
        }
    }

    public List<GoblinUserCouponVo> getUserCouponVos(String uid) {
        String rk = GoblinRedisConst.USER_COUPON.concat(uid);
        String valStr = (String) redisUtil.get(rk);
        List<GoblinUserCouponVo> vos;
        if (org.springframework.util.StringUtils.isEmpty(valStr)) {
            return getGoblinUserCouponVo();
        } else {
            vos = JsonUtils.fromJson(valStr, new TypeReference<List<GoblinUserCouponVo>>() {
            });
        }
        return vos;
    }

    public Boolean changeCouponVos(String ucouponId, GoblinUserCouponBasicVo vo) {
        return mongoTemplate.getCollection(GoblinUserCouponBasicVo.class.getSimpleName()).updateOne(
                Query.query(Criteria.where("ucouponId").is(ucouponId)).getQueryObject(),
                Update.update("state", vo.getState())
                        .set("usedFor", vo.getUsedFor())
                        .getUpdateObject()
        ).getModifiedCount() > 0;
    }

    public boolean setUserCouponVos(String uid, List<GoblinUserCouponVo> vos) {
        return redisUtil.set(GoblinRedisConst.USER_COUPON.concat(uid), JsonUtils.toJson(vos));
    }

    private static final BasicDBObject basicDBObject = new BasicDBObject();
    private static final ArrayList<GoblinUserCouponVo> goblinUserCouponVo = new ArrayList<>();


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

    public static ArrayList<GoblinUserCouponVo> getGoblinUserCouponVo() {
        return (ArrayList<GoblinUserCouponVo>) goblinUserCouponVo.clone();
    }

}