package com.ms.biz.service.impl;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.ms.biz.common.ApiResult;
import com.ms.biz.consts.RedisKeyConst;
import com.ms.biz.consts.TblConst;
import com.ms.biz.service.DsMessageConsumerQueueService;
import com.ms.dal.entity.DsMessageConsumerBuffer;
import com.ms.dal.entity.DsMessageConsumerQueue;
import com.ms.dal.mapper.DsMessageConsumerBufferMapper;
import com.ms.dal.mapper.DsMessageConsumerQueueMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;



/**
 * @author elf
 */
@Service
@Slf4j
@AllArgsConstructor
public class DsMessageConsumerQueueServiceImpl implements DsMessageConsumerQueueService {

    private final DataSourceTransactionManager dataSourceTransactionManager;
    private final TransactionDefinition transactionDefinition;
    private final RedisTemplate<String, String> redisTemplate;

    private DsMessageConsumerQueueMapper dsMessageConsumerQueueMapper;
    private DsMessageConsumerBufferMapper dsMessageConsumerBufferMapper;

    @Override
    public int deleteByPrimaryKey(Long id) {
        return dsMessageConsumerQueueMapper.deleteByPrimaryKey(id);
    }

    @Override
    public int insert(DsMessageConsumerQueue record) {
        return dsMessageConsumerQueueMapper.insert(record);
    }

    @Override
    public int insertSelective(DsMessageConsumerQueue record) {
        return dsMessageConsumerQueueMapper.insertSelective(record);
    }

    @Override
    public DsMessageConsumerQueue selectByPrimaryKey(Long id) {
        return dsMessageConsumerQueueMapper.selectByPrimaryKey(id);
    }

    @Override
    public int updateByPrimaryKeySelective(DsMessageConsumerQueue record) {
        return dsMessageConsumerQueueMapper.updateByPrimaryKeySelective(record);
    }

    @Override
    public int updateByPrimaryKey(DsMessageConsumerQueue record) {
        return dsMessageConsumerQueueMapper.updateByPrimaryKey(record);
    }

    @Override
    public ApiResult moveBufferToSolidRedisQueue(Integer maxQueueCount) {
        Integer queueCount = dsMessageConsumerQueueMapper.getCurrentQueueCount();
        if (queueCount > maxQueueCount) {
            return ApiResult.fail("队列数已满。");
        }

        ApiResult result;

        int needMoveQueueCount = maxQueueCount - queueCount;
        Long startBufferId = 0L;
        int limit = 100;
        while (true) {
            if (needMoveQueueCount <= 0) {
                result = ApiResult.fail("没有需要move的buffer");
                break;
            }
            List<DsMessageConsumerQueue> queues = new ArrayList<>();
            List<Long> deleteBufferIds = new ArrayList<>();
            List<DsMessageConsumerBuffer> buffers = dsMessageConsumerBufferMapper.getListPrimaryKeyBiggerThan(startBufferId, limit);
            log.info("buffers:" + buffers);
            for (DsMessageConsumerBuffer buffer: buffers) {
                startBufferId = buffer.getDsMessageConsumerBufferId();
                deleteBufferIds.add(buffer.getDsMessageConsumerBufferId());
                DsMessageConsumerQueue queue = new DsMessageConsumerQueue();
                queue.setDsMessageId(buffer.getDsMessageId());
                queue.setShopId(buffer.getShopId());
                queue.setLocked(0L);
                queue.setHostname("");
                queue.setGmtLocked(null);
                queue.setGmtCreate(new Date());
                queue.setGmtModified(new Date());
                queues.add(queue);
            }


            TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
            try {
                if (ObjectUtil.isNotEmpty(queues)) {
                    dsMessageConsumerQueueMapper.insertBatch(queues);
                }
                if (ObjectUtil.isNotEmpty(deleteBufferIds)) {
                    dsMessageConsumerBufferMapper.deleteByIdList(deleteBufferIds);
                }
                dataSourceTransactionManager.commit(transactionStatus);
            } catch (Exception e) {
                dataSourceTransactionManager.rollback(transactionStatus);
                log.error("提交数据异常" + e.getMessage());
                result = ApiResult.fail("提交数据异常");
                break;
            }

            log.info("queues:" + queues);
            Long startQueueId = null;
            if (!queues.isEmpty()) {
                startQueueId = queues.get(0).getDsMessageConsumerQueueId();
            }
            if (ObjectUtil.isNotNull(startQueueId)) {
                int addQueueCount = queues.size();
                for (Long queueId = startQueueId; queueId < (addQueueCount + startQueueId); queueId++) {
                    String queueRedisKey = RedisKeyConst.getQueueRedisKey(TblConst.ds_message_consumer_queue, false);
                    log.info("in_queue[" + queueRedisKey + "]: " + queueId);
                    redisTemplate.opsForList().leftPush(queueRedisKey, String.valueOf(queueId));
                }
            }

            needMoveQueueCount -= deleteBufferIds.size();
            if (needMoveQueueCount <= 0 || buffers.size() < limit) {
                result = ApiResult.fail("move buffer 完成");
                break;
            }
        }
        return  result;
    }

    @Override
    public DsMessageConsumerQueue lockSolidRedisQueue(Long timerLockId, boolean isAdminSelf) {
        String queueRedisKey = RedisKeyConst.getQueueRedisKey(TblConst.ds_message_consumer_queue, isAdminSelf);
        String queueIdRedis = redisTemplate.opsForList().rightPop(queueRedisKey);
        log.info("out_queue[" + queueRedisKey + "]" + queueIdRedis);
        if (StrUtil.isBlank(queueIdRedis)) {
            return null;
        }
        DsMessageConsumerQueue row = dsMessageConsumerQueueMapper.queryFirstRowByQueueId(queueIdRedis);
        log.info("out_queue_data: " + row);
        if (ObjectUtil.isEmpty(row)) {
            return null;
        }
        Date date = new Date();
        row.setLocked(timerLockId);
        row.setHostname("");
        row.setGmtLastHeartbeat(date);
        row.setGmtModified(date);
        row.setGmtLocked(date);
        int affectRows = dsMessageConsumerQueueMapper.updateByPrimaryKey(row);
        if (affectRows == 1) {
            return row;
        }
        return null;
    }
}




