package com.fs.app.mq.consumer;

import com.fs.app.facade.IConversionService;
import com.fs.common.annotation.DistributeLock;
import com.fs.newAdv.constant.ConversionTrackingMessage;
import com.fs.newAdv.constant.MqTopicConstant;
import com.fs.newAdv.enums.SystemEventTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 转化追踪
 *
 * @author zhangqin
 */
@Slf4j
@Component
@RocketMQMessageListener(
        topic = MqTopicConstant.CONVERSION_TRACKING_TOPIC,
        consumerGroup = MqTopicConstant.CONVERSION_TRACKING_GROUP,
        // 并发消费模式（多线程并发消费，线程数由RocketMQ自动管理）
        consumeMode = ConsumeMode.CONCURRENTLY,
        // 最大重试次数（RocketMQ默认16次）
        maxReconsumeTimes = 3
)
public class ConversionTrackingMessageConsumer implements RocketMQListener<ConversionTrackingMessage> {

    @Autowired
    private IConversionService conversionService;

    /**
     * 消费转化消息
     *
     * @param message 转化消息
     */
    @Override
    @DistributeLock(scene = "mq", keyExpression = "#message.traceId", waitTime = 0, errorMsg = "重复消费")
    public void onMessage(ConversionTrackingMessage message) {
        String traceId = message.getTraceId();
        SystemEventTypeEnum eventType = message.getEventType();

        log.info("转化追踪消息开始消费 | traceId={}, eventType={}",
                traceId, eventType.getDescription());

        //  调用广告平台API回传
        boolean success = conversionService.reportTrackingToAdvertiser(eventType, traceId);

        if (success) {
            log.info("转化回传成功 | traceId={}, eventType={}", traceId, eventType);
        } else {
            log.info("转化回传失败 | traceId={}, eventType={}", traceId, eventType);
        }
    }
}

