package cn.axzo.msg.center.mq;

import cn.axzo.framework.rocketmq.BaseListener;
import cn.axzo.framework.rocketmq.DefaultEventConsumer;
import cn.axzo.framework.rocketmq.EventConsumer;
import cn.axzo.framework.rocketmq.EventHandlerRepository;
import cn.axzo.framework.rocketmq.EventProducer;
import cn.axzo.framework.rocketmq.EventProducer.Context;
import cn.axzo.framework.rocketmq.RocketMQEventProducer;
import cn.axzo.framework.rocketmq.RocketMQEventProducer.RocketMQMessageMeta;
import cn.axzo.framework.rocketmq.utils.TraceUtils;
import cn.axzo.msg.center.api.mq.MqMessage;
import cn.axzo.msg.center.inside.notices.config.PendingMessageBizConfig;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
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.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;
import java.util.function.Consumer;

/**
 * @author yanglin
 */
@Slf4j
@Configuration
public class RocketMQConfig {

    public static final String APP_NAME = "MSG-CENTER";

    @Value("topic_msg_center_${spring.profiles.active}")
    private String topic;

    @Bean
    EventProducer<MqMessage> eventProducer(RocketMQTemplate rocketMQTemplate) {
        RocketMQMessageMeta topicInfo = RocketMQMessageMeta.builder()
                .topic(topic)
                .build();
        Context<RocketMQMessageMeta> context = Context.<RocketMQMessageMeta>builder()
                .meta(topicInfo)
                .build();
        //noinspection unchecked
        return new RocketMQEventProducer(rocketMQTemplate, APP_NAME, APP_NAME, context, null);
    }

    @Bean
    EventConsumer eventConsumer(EventHandlerRepository eventHandlerRepository) {
        Consumer<EventConsumer.EventWrapper> callback = (eventWrapper) -> {
            if (eventWrapper.isHandled()) {
                // 只收集被App真正消费的消息.
                //String topic = (String) eventWrapper.getExt().get(EVENT_TOPIC_KEY);

            }
        };
        return new DefaultEventConsumer(APP_NAME, eventHandlerRepository, callback);
    }

    @Slf4j
    @Component
    @RocketMQMessageListener(topic = "topic_im_center_${spring.profiles.active}",
            consumerGroup = "GID_topic_im_center_msg_center_${spring.profiles.active}",
            consumeMode = ConsumeMode.ORDERLY,
            nameServer = "${rocketmq.name-server}"
    )
    public static class ImCenterListener extends BaseListener implements RocketMQListener<MessageExt> {

        @Autowired
        private EventConsumer eventConsumer;

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

    @Slf4j
    @Component
    @RequiredArgsConstructor
    @RocketMQMessageListener(topic = "topic_msg_center_${spring.profiles.active}",
            consumerGroup = "GID_topic_msg_center_self_consume_${spring.profiles.active}",
            consumeMode = ConsumeMode.ORDERLY,
            nameServer = "${rocketmq.name-server}"
    )
    public static class MsgCenterListener extends BaseListener implements RocketMQListener<MessageExt> {

        private final EventConsumer eventConsumer;
        private final PendingMessageBizConfig cfg;

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

        /**
         * 拷贝下来改一下, 设置最大处理时间
         */
        public void onEvent(MessageExt message, EventConsumer eventConsumer) {
            String topic = message.getTopic();
            String value = new String(message.getBody());
            Map<String, String> headers = message.getProperties();
            MDC.put(TraceUtils.CTX_LOG_ID, headers.get(TraceUtils.TRACE_ID));
            MDC.put(TraceUtils.TRACE_ID, headers.get(TraceUtils.TRACE_ID));
            MDC.put(TraceUtils.TRACE_ID_IN_MDC, headers.get(TraceUtils.TRACE_ID));
            if (log.isDebugEnabled()) {
                log.debug("received message, topic={}, headers={}, value={}", topic, headers, value);
            }

            //当前消息所在分区的lag. 而不是整个topic
            Long partitionLag = Optional.ofNullable(message.getProperties().get(MessageConst.PROPERTY_MAX_OFFSET))
                    .map(e -> Long.parseLong(e) - message.getQueueOffset()).orElse(0L);

            eventConsumer.onEvent(value, EventConsumer.Context.builder()
                    .msgId(message.getMsgId())
                    .ext(ImmutableMap.of("topic", topic))
                    .headers(Maps.transformValues(headers, header -> Optional.ofNullable(header).map(String::getBytes).orElse(new byte[] {})))
                    .lagSupplier(() -> partitionLag)
                    .maxAllowElapsedMillis(cfg.getMsgCenterMqSelfConsumeMaxExecMs())
                    .build());
        }
    }

    @Bean
    EventHandlerRepository eventHandlerRepository() {
        return new EventHandlerRepository((ex, logText) -> {
            log.warn("MQ, handle warning {}", logText, ex);
        });
    }

}