package com.qmth.exam.reserve.mq;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.qmth.boot.core.concurrent.service.ConcurrentService;
import com.qmth.boot.core.exception.StatusException;
import com.qmth.exam.reserve.bean.Constants;
import com.qmth.exam.reserve.bean.apply.ApplyRecordCacheBean;
import com.qmth.exam.reserve.cache.CacheConstants;
import com.qmth.exam.reserve.cache.impl.ApplyTaskCacheService;
import com.qmth.exam.reserve.entity.StudentApplyEntity;
import com.qmth.exam.reserve.enums.EventType;
import com.qmth.exam.reserve.service.OperateLogService;
import com.qmth.exam.reserve.service.StudentApplyService;
import com.qmth.exam.reserve.util.JsonHelper;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(
        consumerGroup = "${rocketmq.consumer.group}",
        topic = "${rocketmq.topic}",
        selectorExpression = MQConstants.TAG_STUDENT_APPLY,
        consumeMode = ConsumeMode.ORDERLY,
        messageModel = MessageModel.CLUSTERING
)
public class ExamReserveMQConsumer implements RocketMQListener<MessageExt> {

    private static final Logger log = LoggerFactory.getLogger(ExamReserveMQConsumer.class);

    @Autowired
    private StudentApplyService studentApplyService;

    @Autowired
    private OperateLogService operateLogService;

    @Autowired
    private ApplyTaskCacheService applyTaskCacheService;

    @Autowired
    private ConcurrentService concurrentService;

    @Override
    public void onMessage(MessageExt message) {
        ApplyRecordCacheBean queueBean = JsonHelper.toObj(message.getBody(), ApplyRecordCacheBean.class);
        if (queueBean == null) {
            log.warn("【考生预约队列】MQ消息转换失败！queueOffset:{} msgId:{}", message.getQueueOffset(), message.getMsgId());
            throw new RuntimeException("MQ消息转换失败！");
        }

        // 考生预约操作锁
        String studentApplyLockKey = String.format(CacheConstants.LOCK_STUDENT_APPLY, queueBean.getStudentId());
        RLock studentApplyLock = (RLock) concurrentService.getLock(studentApplyLockKey);
        try {
            if (!studentApplyLock.tryLock()) {
                log.warn("获取锁失败，同一个考生不允许同时操作预约！lockKey:{}", studentApplyLockKey);
                throw new StatusException(Constants.SYSTEM_BUSY);
            }
            log.info("获取锁成功！lockKey:{}", studentApplyLockKey);

            ApplyRecordCacheBean cacheBean = applyTaskCacheService.getStudentApplyRecord(queueBean.getStudentId(),
                    queueBean.getExamSiteId(), queueBean.getTimePeriodId());

            final ApplyRecordCacheBean bean;
            if (cacheBean == null) {
                log.warn("【考生预约队列】MQ队列元素对应的缓存记录不存在！bizId:{} msgId:{}", queueBean.getBizId(), message.getMsgId());

                // 若缓存数据不存在，则采用队列数据持久化
                bean = queueBean;
                // 更新缓存
                applyTaskCacheService.saveStudentApplyRecord(bean);
            } else {
                if (cacheBean.getBizId() != null && cacheBean.getBizId() >= queueBean.getBizId()) {
                    log.info("【考生预约队列】MQ消息消费！bizId:{} cacheBizId:{} msgId:{}", queueBean.getBizId(),
                            cacheBean.getBizId(), message.getMsgId());

                    // 默认采用最新预约缓存数据持久化（极端情况下队列消息顺序异常）
                    bean = cacheBean;
                } else {
                    log.warn("【考生预约队列】缓存中的队列业务流水号无效！bizId:{} cacheBizId:{} msgId:{}", queueBean.getBizId(),
                            cacheBean.getBizId(), message.getMsgId());

                    // 其它情况，采用队列数据持久化
                    bean = queueBean;
                }
            }

            this.saveOrUpdate(bean);
        } finally {
            try {
                // 解锁前检查当前线程是否持有该锁
                if (studentApplyLock.isLocked() && studentApplyLock.isHeldByCurrentThread()) {
                    studentApplyLock.unlock();
                    log.info("解锁成功！lockKey:{}", studentApplyLockKey);
                }
            } catch (Exception e) {
                log.warn("解锁失败！lockKey:{} err:{}", studentApplyLockKey, e.getMessage());
            }
        }
    }

    public void saveOrUpdate(ApplyRecordCacheBean bean) {
        LambdaQueryWrapper<StudentApplyEntity> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(StudentApplyEntity::getExamSiteId, bean.getExamSiteId());
        wrapper.eq(StudentApplyEntity::getTimePeriodId, bean.getTimePeriodId());
        wrapper.eq(StudentApplyEntity::getStudentId, bean.getStudentId());
        StudentApplyEntity data = studentApplyService.getOne(wrapper);

        String logContent = String.format("%s_%s_%s_%s", bean.getStudentId(), bean.getExamSiteId(), bean.getTimePeriodId(), bean.getCancel());
        if (data != null) {
            // 存在预约记录，则更新
            data.setCancel(bean.getCancel());
            data.setOperateId(bean.getOperateId());
            data.setUpdateTime(bean.getOperateTime());

            LambdaUpdateWrapper<StudentApplyEntity> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.set(StudentApplyEntity::getCancel, bean.getCancel());
            updateWrapper.set(StudentApplyEntity::getOperateId, bean.getOperateId());
            updateWrapper.set(StudentApplyEntity::getUpdateTime, bean.getOperateTime());
            updateWrapper.eq(StudentApplyEntity::getId, data.getId());
            studentApplyService.update(updateWrapper);
            log.warn("【考生预约】DB更新！{} dbId:{} bizId:{}", logContent, data.getId(), bean.getBizId());
        } else {
            // 不存在预约记录，则新增
            data = new StudentApplyEntity();
            data.setStudentId(bean.getStudentId());
            data.setExamSiteId(bean.getExamSiteId());
            data.setTimePeriodId(bean.getTimePeriodId());
            data.setCancel(bean.getCancel());
            data.setOperateId(bean.getOperateId());
            data.setCreateTime(bean.getOperateTime());
            data.setUpdateTime(bean.getOperateTime());
            studentApplyService.save(data);
            log.warn("【考生预约】DB新增！{} dbId:{} bizId:{}", logContent, data.getId(), bean.getBizId());
        }

        if (data.getCancel()) {
            // 保存“取消预约”操作记录
            operateLogService.insertOperateLog(bean.getOperateId(), EventType.CANCEL_APPLY, JsonHelper.toJson(data));
        }
    }

}
