package cn.lili.trigger.interfaces.impl;

import cn.hutool.json.JSONUtil;
import cn.lili.cache.Cache;
import cn.lili.rocketmq.RocketmqSendCallbackBuilder;
import cn.lili.trigger.delay.queue.PromotionDelayQueue;
import cn.lili.trigger.interfaces.TimeTrigger;
import cn.lili.trigger.model.TimeTriggerMsg;
import cn.lili.trigger.util.DelayQueueTools;
import cn.lili.common.utils.DateUtil;
import cn.lili.common.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * 延时任务实现
 *
 * @author paulG
 * @since 2020/11/5
 **/
@Component
@Slf4j
public class RocketmqTimerTrigger implements TimeTrigger {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private Cache<Integer> cache;
    @Autowired
    private PromotionDelayQueue promotionDelayQueue;


    @Override
    public void addDelay(TimeTriggerMsg timeTriggerMsg) {
        //执行器唯一key
        String uniqueKey = timeTriggerMsg.getUniqueKey();
        if (StringUtils.isEmpty(uniqueKey)) {
            uniqueKey = StringUtils.getRandStr(10);
        }
        //执行任务key
        String generateKey = DelayQueueTools.generateKey(timeTriggerMsg.getTriggerExecutor(), timeTriggerMsg.getTriggerTime(), uniqueKey);
        this.cache.put(generateKey, 1);
        //设置延时任务
        if (Boolean.TRUE.equals(promotionDelayQueue.addJob(JSONUtil.toJsonStr(timeTriggerMsg), timeTriggerMsg.getTriggerTime()))) {
            log.info("延时任务标识： {}", generateKey);
            log.info("定时执行在【" + DateUtil.toString(timeTriggerMsg.getTriggerTime(), "yyyy-MM-dd HH:mm:ss") + "】，消费【" + timeTriggerMsg.getParam().toString() + "】");
        } else {
            log.error("延时任务添加失败:{}", timeTriggerMsg);
        }
    }

    @Override
    public void execute(TimeTriggerMsg timeTriggerMsg) {
        this.addExecute(timeTriggerMsg.getTriggerExecutor(),
                timeTriggerMsg.getParam(),
                timeTriggerMsg.getTriggerTime(),
                timeTriggerMsg.getUniqueKey(),
                timeTriggerMsg.getTopic()
        );
    }

    /**
     * 将任务添加到mq，mq异步队列执行。
     * <p>
     * 本系统中redis相当于延时任务吊起机制，而mq才是实际的业务消费，执行任务的存在
     *
     * @param executorName 执行器beanId
     * @param param        执行参数
     * @param triggerTime  执行时间 时间戳 秒为单位
     * @param uniqueKey    如果是一个 需要有 修改/取消 延时任务功能的延时任务，<br/>
     *                     请填写此参数，作为后续删除，修改做为唯一凭证 <br/>
     *                     建议参数为：COUPON_{ACTIVITY_ID} 例如 coupon_123<br/>
     *                     业务内全局唯一
     * @param topic        rocketmq topic
     */
    private void addExecute(String executorName, Object param, Long triggerTime, String uniqueKey, String topic) {

        TimeTriggerMsg timeTriggerMsg = new TimeTriggerMsg(executorName, triggerTime, param, uniqueKey, topic);
        Message<TimeTriggerMsg> message = MessageBuilder.withPayload(timeTriggerMsg).build();
        log.info("延时任务发送信息：{}", message);
        this.rocketMQTemplate.asyncSend(topic, message, RocketmqSendCallbackBuilder.commonCallback());
    }

    @Override
    public void edit(String executorName, Object param, Long oldTriggerTime, Long triggerTime, String uniqueKey, int delayTime, String topic) {
        this.delete(executorName, oldTriggerTime, uniqueKey, topic);
        this.addDelay(new TimeTriggerMsg(executorName, triggerTime, param, uniqueKey, topic));
    }

    @Override
    public void delete(String executorName, Long triggerTime, String uniqueKey, String topic) {
        String generateKey = DelayQueueTools.generateKey(executorName, triggerTime, uniqueKey);
        log.info("删除延时任务{}", generateKey);
        this.cache.remove(generateKey);
    }
}