package cn.axzo.workflow.server.mq.outside;

import cn.axzo.framework.rocketmq.BaseListener;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.framework.rocketmq.EventProducer;
import cn.axzo.framework.rocketmq.RocketMQEventProducer;
import cn.axzo.workflow.server.mq.outside.producer.DingtalkSendProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashMap;

/**
 * 钉钉消息 Rocket 配置
 *
 * @author wangli
 * @since 2024-10-25 13:39
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
public class DingtalkRocketConfiguration {
    @Value("${spring.profiles.active:dev}")
    private String activeProfile;
    private static final String DEFAULT_MODULE = "workflowEngine";
    private static final String DEFAULT_EVENT = "topic_third_party_sync_event_";
    private static final String MODULE_NAME_SUFFIX = "_dingtalk_message";

    @Bean
    public DingtalkSendProducer dingtalkSendProducer(RocketMQTemplate rocketMQTemplate) {
        return new DingtalkSendProducer(rocketMQTemplate,
                DEFAULT_MODULE,
                DEFAULT_MODULE + MODULE_NAME_SUFFIX,
                EventProducer.Context.<RocketMQEventProducer.RocketMQMessageMeta>builder()
                        .meta(RocketMQEventProducer.RocketMQMessageMeta.builder()
                                .topic(DEFAULT_EVENT + activeProfile)
                                .build())
                        .headers(new HashMap<>())
                        .syncSending(Boolean.TRUE)
                        .exceptionHandler(context -> {
                            log.error("MQ, send event error: {}, event: {}",
                                    context.getThrowable().getCause().getMessage(),
                                    context.getEvent().toPrettyJsonString(),
                                    context.getThrowable());
                        })
                        .build(),
                null
        );
    }

    @Component
    @ConditionalOnProperty(name = "rocketmq.name-server")
    @RocketMQMessageListener(topic = "topic_third_party_sync_event_${spring.profiles.active}",
            consumerGroup = "GID_${spring.application.name}_riven_${spring.profiles.active}_consumer",
            consumeMode = ConsumeMode.CONCURRENTLY,
            selectorType = SelectorType.TAG,
            selectorExpression = "riven-dingtalk-receive",
            maxReconsumeTimes = 3,
            nameServer = "${rocketmq.name-server}"
    )
    public static class ReplyMessageRocketConsumer extends BaseListener implements RocketMQListener<MessageExt> {
        @Resource
        private EventConsumer eventConsumer;

        @Override
        public void onMessage(MessageExt message) {
            super.onEvent(message, eventConsumer);
        }
    }
}
