package com.ms.biz.service.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.ms.biz.consts.QueueConst;
import com.ms.biz.consts.RedisKeyConst;
import com.ms.biz.service.QueueService;
import com.ms.biz.tool.CommonTool;
import com.ms.dal.mapper.QueueMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Service
@Slf4j
@AllArgsConstructor
public class QueueServiceImpl implements QueueService {

    private final RedisTemplate<String, String> redisTemplate;

    private final QueueMapper queueMapper;

    @Override
    public Map<Long, Long> allocBufferToQueueByShopAvg(String bufferName, String queueName, JSONObject filter, int maxQueueCount, Integer shopLimit) {
        int allowMoveCnt = maxQueueCount - this.getQueueCnt(queueName, filter);
        if (allowMoveCnt < 1) {
            return new HashMap<Long, Long>();
        }
        Map<Long, Long> shopIdAndCntMap = getShopIdAndBufferCntMap(bufferName, filter);
        if (ObjectUtil.isNotNull(shopLimit)) {
            System.out.println("shopIdAndCntMap: " + shopIdAndCntMap);
            shopIdAndCntMap = rebuildShopIdAndCntMapByShopLimit(shopIdAndCntMap, queueName, shopLimit);
        }
        return allotShopMoveBufferCnt(shopIdAndCntMap, allowMoveCnt);
    }

    @Override
    public int unlockTimeoutSolidRedisQueue(String tableName, int heartbeatExpiredSeconds, String status) {
        int c1 = unlockSolidRedisDeadQueue(tableName, heartbeatExpiredSeconds, status);
        int c2 = unlockSolidRedisSkipQueue(tableName, status);
        return c1 + c2;
    }

    public int unlockSolidRedisSkipQueue(String queueName, String status) {
        log.info(String.format("unlockSolidRedisSkipQueue start %s", queueName));
        String queueRedisKey = RedisKeyConst.getQueueRedisKey(queueName, false);
        List<String> redisIds = redisTemplate.opsForList().range(queueRedisKey, 0, -1);
        if (ObjectUtil.isEmpty(redisIds)) {
            redisIds = new ArrayList<>();
        }
        StringBuilder where = new StringBuilder();
        if (StrUtil.isNotBlank(status)) {
            where.append(String.format(" AND status = '%s'", status));
        }
        String queueIdColumnName = QueueConst.getQueueIdColumnName(queueName);
        Long startQueueId = 0L;
        int limit = 100;
        List<Map<String, Object>> allQueueList = new ArrayList<>();
        while (true) {
            StringBuilder queueIdWhere = new StringBuilder(" WHERE locked = 0 ");
            if (startQueueId > 0) {
                queueIdWhere.append(String.format(" AND %s > %s", queueIdColumnName, startQueueId));
            }
            if (ObjectUtil.isNotNull(where) && StrUtil.isNotBlank(where.toString())) {
                queueIdWhere.append(where);
            }
            queueIdWhere.append(String.format(" order by %s asc limit %s", queueIdColumnName, limit));
//            log.info("whereStr: " + queueIdWhere);
            List<Map<String, Object>> queueList = queueMapper.select(queueName, queueIdWhere.toString());
//            log.info("queueList size: " + queueList.size());
            if (ObjectUtil.isEmpty(queueList)) {
                break;
            }
            String queueIdStr = queueList.get(queueList.size() - 1).get(queueIdColumnName).toString();
            startQueueId = Long.valueOf(queueIdStr);
            allQueueList.addAll(queueList);
            if (queueList.size() < limit) {
                break;
            }
        }
        if (ObjectUtil.isEmpty(allQueueList)) {
            return 0;
        }
        List<String> newRedisIds = redisTemplate.opsForList().range(queueRedisKey, 0, -1);
        if (ObjectUtil.isEmpty(newRedisIds)) {
            newRedisIds = new ArrayList<>();
        }
        Set<String> tempRedisIds = new HashSet<>();
        tempRedisIds.addAll(redisIds);
        tempRedisIds.addAll(newRedisIds);
//        log.info("allQueueList count: " + allQueueList.size());
//        log.info("tempRedisIds count: " + tempRedisIds.size());
        for (Map<String, Object> queue : allQueueList) {
            String queueIdStr = queue.get(queueIdColumnName).toString();
            if (tempRedisIds.contains(queueIdStr)) {
                continue;
            }
            log.info("replenish task push queueIdStr: " + queueIdStr);
            redisTemplate.opsForList().leftPush(queueRedisKey, queueIdStr);
        }
        log.info(String.format("unlockSolidRedisSkipQueue end %s", queueName));
        return allQueueList.size();
    }

    public int unlockSolidRedisDeadQueue(String tableName, int heartbeatExpiredSeconds, String status) {
        DateTime deadTime = DateUtil.offsetSecond(new Date(), -heartbeatExpiredSeconds);
        String where = String.format(" WHERE locked > 0 AND gmt_last_heartbeat < '%s' ", DateUtil.formatDateTime(deadTime));
        if (StrUtil.isNotBlank(status)) {
            where = where.concat(String.format(" and `status` = '%s' ", status));
        }
        List<Map<String, Object>> queueList = queueMapper.select(tableName, where);
        if (ObjectUtil.isEmpty(queueList)) {
            return 0;
        }
        String queueIdColumnName = QueueConst.getQueueIdColumnName(tableName);
        int ar1 = 0;
        for (Map<String, Object> queue : queueList) {
            String queueIdStr = queue.get(queueIdColumnName).toString();
            Long queueId = Long.valueOf(queueIdStr);
            if (ObjectUtil.isNotNull(queueId)) {
                ar1 = queueMapper.updateLocked(tableName, queueIdColumnName, queueId, 0, DateUtil.now());
                String queueRedisKey = RedisKeyConst.getQueueRedisKey(tableName, false);
                redisTemplate.opsForList().leftPush(queueRedisKey, queueId.toString());
            }
        }
        return queueList.size();
    }

    private Map<Long, Long> rebuildShopIdAndCntMapByShopLimit(Map<Long, Long> shopIdAndCntMap, String queueName, Integer shopLimit) {
        if (ObjectUtil.isEmpty(shopIdAndCntMap) || ObjectUtil.isNull(shopLimit)) {
            return shopIdAndCntMap;
        }
        Map<Long, Map<String, Object>> dataMap = queueMapper.selectShopIdAndCnt(queueName, null);
        Map<Long, Long> shopIdAndQueueCntMap = new HashMap<>();
        for (Map.Entry<Long, Map<String, Object>> entry : dataMap.entrySet()) {
            String cntStr = entry.getValue().getOrDefault("cnt", 0L).toString();
            Long cnt = Long.valueOf(cntStr);
            shopIdAndQueueCntMap.put(entry.getKey(), cnt);
        }
        Map<Long, Long> resultMap = new HashMap<>();
        for (Map.Entry<Long, Long> entry : shopIdAndCntMap.entrySet()) {
            Long shopQueueCount = 0L;
            if (shopIdAndQueueCntMap.containsKey(entry.getKey())) {
                shopQueueCount = shopIdAndQueueCntMap.get(entry.getKey());
            }
            long shopAllowMoveMaxCnt = shopLimit - shopQueueCount;
            long shopAllowMoveCnt = Math.min(entry.getValue(), shopAllowMoveMaxCnt);
            if (shopAllowMoveCnt < 1) {
                continue;
            }
            resultMap.put(entry.getKey(), shopAllowMoveCnt);
        }
        return resultMap;
    }

    private Map<Long, Long> allotShopMoveBufferCnt(Map<Long, Long> shopIdAndCntMap, int allowMoveCnt) {
        Map<Long, Long> shopIdAndMoveCntMap = new HashMap<>();
        Set<Long> waitSelectShopIds = shopIdAndCntMap.keySet();

        do {
            if (waitSelectShopIds.isEmpty() || allowMoveCnt < 1) {
                break;
            }

            for (Long mallId: waitSelectShopIds) {
                if (allowMoveCnt < 1) {
                    break;
                }

                Long waitMoveCnt = shopIdAndCntMap.get(mallId);
                if (waitMoveCnt < 1) {
                    waitSelectShopIds.remove(mallId);
                    continue;
                }

                shopIdAndMoveCntMap.put(mallId, shopIdAndMoveCntMap.getOrDefault(mallId, 0L) + 1);
                shopIdAndCntMap.put(mallId, shopIdAndCntMap.get(mallId) - 1);
                allowMoveCnt--;
            }
        } while (true);

        return shopIdAndMoveCntMap;
    }

    private Map<Long, Long> getShopIdAndBufferCntMap(String bufferName, JSONObject filter) {
        String whereStr = getFilterShopIdWhere(filter.getString("includeShopIds"), filter.getString("excludeShopIds"), filter.getDate("gmtExec"));
        Map<Long, Long> result = new HashMap<>();
        Map<Long, Map<String, Object>> dataMap = queueMapper.selectShopIdAndCnt(bufferName, whereStr);
        for (Map.Entry<Long, Map<String, Object>> entry : dataMap.entrySet()) {
            String shopId = String.valueOf(entry.getKey());
            String bufferCntStr = entry.getValue().getOrDefault("cnt", 0L).toString();
            Long bufferCnt = Long.valueOf(bufferCntStr);
            result.put(Long.valueOf(shopId), bufferCnt);
        }
        return result;
    }

    private int getQueueCnt(String queueName, JSONObject filter) {
        String whereStr = getFilterShopIdWhere(filter.getString("includeShopIds"), filter.getString("excludeShopIds"), null);
        return queueMapper.getQueueCnt(queueName, whereStr);
    }

    private String getFilterShopIdWhere(String includeShopIds, String excludeShopIds, Date gmtExec) {
        StringBuilder extraWhere = new StringBuilder(" 1 = 1");
        if (ObjectUtil.isNotNull(gmtExec)) {
            extraWhere.append(String.format(" and gmt_exec <= '%s' ", DateUtil.formatDateTime(gmtExec)));
        }
        if (StrUtil.isNotBlank(includeShopIds)) {
            return String.format(" shop_id IN (%s) ", CommonTool.stringToSQLIN(CommonTool.splitWithComma(includeShopIds)));
        }
        if (StrUtil.isNotBlank(excludeShopIds)) {
            return String.format(" shop_id NOT IN (%s) ", CommonTool.stringToSQLIN(CommonTool.splitWithComma(excludeShopIds)));
        }
        return extraWhere.toString();
    }

    public int unlockTimeoutQueue(String tableName, int heartbeatExpiredSeconds, String status) {
        DateTime deadTime = DateUtil.offsetSecond(new Date(), -heartbeatExpiredSeconds);
        String where = String.format(" WHERE locked > 0 AND gmt_last_heartbeat < '%s' ", DateUtil.formatDateTime(deadTime));
        if (StrUtil.isNotBlank(status)) {
            where = where.concat(String.format(" and `status` = '%s' ", status));
        }
        List<Map<String, Object>> queueList = queueMapper.select(tableName, where);
        if (ObjectUtil.isEmpty(queueList)) {
            return 0;
        }
        String queueIdColumnName = QueueConst.getQueueIdColumnName(tableName);
        int ar1 = 0;
        for (Map<String, Object> queue : queueList) {
            String queueIdStr = queue.get(queueIdColumnName).toString();
            Long queueId = Long.valueOf(queueIdStr);
            if (ObjectUtil.isNotNull(queueId)) {
                ar1 = queueMapper.updateLocked(tableName, queueIdColumnName, queueId, 0, DateUtil.now());
            }
        }
        return queueList.size();
    }

//    public void lockSolidRedisQueue(Integer timerLockId, String queueName, boolean isAdminSelf) {
//        String queueRedisKey = RedisKeyConst.getQueueRedisKey(queueName, isAdminSelf);
//        String queueId = redisTemplate.opsForList().rightPop(queueRedisKey);
//        if (StrUtil.isBlank(queueId)) {
//            return;
//        }
//        String primaryColumn = QueueConst.getQueueIdColumnName(queueName);
//        queueMapper.queryFirstRow("select * from %b where `locked` = 0 and %b = %i");
//
//    }
}
