package com.sonic.frog.config;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableMap;
import com.rabbitmq.client.Channel;
import com.sonic.common.AppRuntime;
import com.sonic.common.event.*;
import com.sonic.common.utils.LogUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.RabbitListeners;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
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.context.annotation.Profile;
import org.springframework.core.task.TaskExecutor;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

import java.io.IOException;
import java.nio.charset.Charset;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * @author code
 */
@Slf4j
public class EventConfig {

    /** TODO: 定义 Event.BuildInScene */
    public final static String DEFAULT_SCENE = Event.BuildInScene.SONIC.getCode();
    /** TODO: DEFAULT_SCENE + "_" + appName */
    public final static String DEFAULT_MODULE = DEFAULT_SCENE + "_" + "frog";

    public final static String PIGEON = DEFAULT_SCENE + "_" + "pigeon";

    @Value("${mq.exchange}")
    private String mqExchange;
    @Value("${mq.default.queue}")
    private String defaultQueue;
    @Value("${mq.default.routing-key}")
    private String defaultRoutingKey;

    @Value("${mq.aiImInfo.queue}")
    private String aiImInfoQueue;
    @Value("${mq.aiImInfo.routing-key}")
    private String aiImInfoRoutingKey;

    @Value("${mq.calc-heartbeat-level.queue}")
    private String calcHeartbeatLevelQueue;
    @Value("${mq.calc-heartbeat-level.routing-key}")
    private String calcHeartbeatLevelRoutingKey;

    @Value("${mq.calc-heartbeat-rank.queue}")
    private String calcHeartbeatRankQueue;
    @Value("${mq.calc-heartbeat-rank.routing-key}")
    private String calcHeartbeatRankRoutingKey;

    @Value("${mq.aiChat.queue}")
    private String aiChatQueue;
    @Value("${mq.aiChat.routing-key}")
    private String aiChatRoutingKey;

    @Value("${mq.aiChatToFrog.queue}")
    private String aiChatToFrogQueue;
    @Value("${mq.aiChatToFrog.routing-key}")
    private String aiChatToFrogRoutingKey;

    @Value("${mq.subtract-heartbeat-val.queue}")
    private String subtractHeartbeatValQueue;
    @Value("${mq.subtract-heartbeat-val.routing-key}")
    private String subtractHeartbeatValRoutingKey;

    @Value("${mq.calc-heartbeat-score.queue}")
    private String calcHeartbeatScoreQueue;
    @Value("${mq.calc-heartbeat-score.routing-key}")
    private String calcHeartbeatScoreRoutingKey;

    @Value("${mq.ai-user-stat.queue}")
    private String aiUserStatQueue;
    @Value("${mq.ai-user-stat.routing-key}")
    private String aiUserStatRoutingKey;

    @Value("${mq.user-deduction-stat.queue}")
    private String userDeductionStatQueue;
    @Value("${mq.user-deduction-stat.routing-key}")
    private String userDeductionStatRoutingKey;

    @Value("${mq.user-balance-insufficient-checkout.queue}")
    private String userBalanceInsufficientCheckoutQueue;
    @Value("${mq.user-balance-insufficient-checkout.routing-key}")
    private String userBalanceInsufficientCheckoutRoutingKey;

    @Value("${mq.ai-change.queue}")
    private String aiChangeQueue;
    @Value("${mq.ai-change.routing-key}")
    private String aiChangeRoutingKey;

    @Configuration
    @Profile("!unittest")
    static class Listener {
        @Autowired
        EventConsumer eventConsumer;

        @RabbitListeners({
                @RabbitListener(queues = {"${mq.default.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.aiImInfo.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.calc-heartbeat-level.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.subtract-heartbeat-val.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.calc-heartbeat-score.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.ai-user-stat.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.calc-heartbeat-rank.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.user-deduction-stat.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.user-balance-insufficient-checkout.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.ai-change.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.aiChatToFrog.queue}"}, concurrency = "2"),
        })
        public void process(@Payload Message message, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) {
            try {
                LogUtils.setTraceId();
                byte[] msgBody = message.getBody();
                String contentEncoding = message.getMessageProperties().getContentEncoding();
                String bodyStr = new String(msgBody, Charset.forName(ObjectUtils.defaultIfNull(contentEncoding, "UTF-8")));
                MessageProperties pro = message.getMessageProperties();
                if (log.isDebugEnabled()) {
                    log.info("receive msg routingKey:{}->consumerQueue:{}->redelivered:{}->body:{}",
                            pro.getReceivedRoutingKey(), pro.getConsumerQueue(), pro.getRedelivered(), bodyStr);
                }
                eventConsumer.onEvent(bodyStr, ImmutableMap.of("record", JSON.toJSONString(message)));

                //消息确认，multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息
                channel.basicAck(deliveryTag, false);
            } catch (Exception e) {
                log.error("===> mq handler error, message: {}", JSON.toJSONString(message), e);
                //消息拒绝，//requeue=true,表示将消息重新放入到队列中，false：表示直接从队列中删除，此时和basicAck(long deliveryTag, false)的效果一样
                try {
                    channel.basicReject(deliveryTag,true);
                } catch (IOException ioException) {
                    log.error("channel.basicReject error. message: {}, deliveryTag: {}", JSON.toJSONString(message), deliveryTag, ioException);
                }
            } finally {
                LogUtils.removeTraceId();
            }
        }

    }

    @Bean
    EventProducer eventProducer(AppRuntime appRuntime, RabbitTemplate rabbitTemplate, TaskExecutor taskExecutor) {
        BiConsumer<Event, RabbitmqEventProducer.RabbitmqMessageMeta> callback = (event, meta) -> {};
        return new RabbitmqEventProducer(rabbitTemplate, DEFAULT_MODULE, DEFAULT_SCENE, appRuntime.getAppId(),
                RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, defaultRoutingKey), taskExecutor, callback);
    }


    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta aiImInfoMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiImInfoRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta calcHeartbeatLevelMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, calcHeartbeatLevelRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta aiChatMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiChatRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta aiChatToFrogMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiChatToFrogRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta subtractHeartbeatValMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, subtractHeartbeatValRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta calcHeartbeatScoreMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, calcHeartbeatScoreRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta calcHeartbeatRankMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, calcHeartbeatRankRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta aiUserStatMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiUserStatRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta userDeductionStatMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, userDeductionStatRoutingKey);
    }
    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta userBalanceInsufficientCheckoutMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, userBalanceInsufficientCheckoutRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta aiChangeMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiChangeRoutingKey);
    }


    @Bean
    EventConsumer eventConsumer(AppRuntime appRuntime, EventHandlerRepository eventHandlerRepository) {
        Consumer<EventConsumer.EventWrapper> callback = (eventWrapper) -> {};
        return new DefaultEventConsumer(appRuntime.getAppName(), eventHandlerRepository, callback);
    }

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

    @Bean
    public DirectExchange messageServerExchange(){
        return new DirectExchange(mqExchange);
    }


    @Bean
    public Binding bindingDefaultQueueExchange(DirectExchange exchange, Queue defaultQueue) {
        return bindingExchange(exchange, defaultQueue, defaultRoutingKey);
    }

    @Bean
    public Binding bindingAiImInfoQueueExchange(DirectExchange exchange, Queue aiImInfoQueue) {
        return bindingExchange(exchange, aiImInfoQueue, aiImInfoRoutingKey);
    }

    @Bean
    public Binding bindingAiChatQueueExchange(DirectExchange exchange) {
        return bindingExchange(exchange, aiChatQueue(), aiChatRoutingKey);
    }

    @Bean
    public Binding bindingAiChatToFrogQueueExchange(DirectExchange exchange) {
        return bindingExchange(exchange, aiChatToFrogQueue(), aiChatToFrogRoutingKey);
    }

    @Bean
    public Binding bindingCalcHeartbeatLevelQueueExchange(DirectExchange exchange, Queue calcHeartbeatLevelQueue) {
        return bindingExchange(exchange, calcHeartbeatLevelQueue, calcHeartbeatLevelRoutingKey);
    }

    @Bean
    public Binding bindingCalcHeartbeatRankQueueExchange(DirectExchange exchange, Queue calcHeartbeatRankQueue) {
        return bindingExchange(exchange, calcHeartbeatRankQueue, calcHeartbeatRankRoutingKey);
    }

    @Bean
    public Binding bindingSubtractHeartbeatValQueueExchange(DirectExchange exchange, Queue subtractHeartbeatValQueue) {
        return bindingExchange(exchange, subtractHeartbeatValQueue, subtractHeartbeatValRoutingKey);
    }

    @Bean
    public Binding bindingCalcHeartbeatScoreQueueExchange(DirectExchange exchange, Queue calcHeartbeatScoreQueue) {
        return bindingExchange(exchange, calcHeartbeatScoreQueue, calcHeartbeatScoreRoutingKey);
    }

    @Bean
    public Binding bindingAiUserStatQueueExchange(DirectExchange exchange, Queue aiUserStatQueue) {
        return bindingExchange(exchange, aiUserStatQueue, aiUserStatRoutingKey);
    }
    @Bean
    public Binding bindingUserDeductionStatQueueExchange(DirectExchange exchange, Queue userDeductionStatQueue) {
        return bindingExchange(exchange, userDeductionStatQueue, userDeductionStatRoutingKey);
    }
    @Bean
    public Binding bindingUserBalanceInsufficientCheckoutQueueExchange(DirectExchange exchange, Queue userBalanceInsufficientCheckoutQueue) {
        return bindingExchange(exchange, userBalanceInsufficientCheckoutQueue, userBalanceInsufficientCheckoutRoutingKey);
    }

    @Bean
    public Binding bindingAiChangeQueueExchange(DirectExchange exchange, Queue aiChangeQueue) {
        return bindingExchange(exchange, aiChangeQueue, aiChangeRoutingKey);
    }


    @Bean
    public Queue defaultQueue() {
        return new Queue(defaultQueue,true);
    }

    @Bean
    public Queue aiImInfoQueue() {
        return new Queue(aiImInfoQueue,true);
    }

    @Bean
    public Queue calcHeartbeatLevelQueue() {
        return new Queue(calcHeartbeatLevelQueue,true);
    }

    @Bean
    public Queue calcHeartbeatRankQueue() {
        return new Queue(calcHeartbeatRankQueue,true);
    }

    @Bean
    public Queue aiChatQueue() {
        return new Queue(aiChatQueue,true);
    }

    @Bean
    public Queue aiChatToFrogQueue() {
        return new Queue(aiChatToFrogQueue,true);
    }

    @Bean
    public Queue subtractHeartbeatValQueue() {
        return new Queue(subtractHeartbeatValQueue,true);
    }

    @Bean
    public Queue calcHeartbeatScoreQueue() {
        return new Queue(calcHeartbeatScoreQueue,true);
    }

    @Bean
    public Queue aiUserStatQueue() {
        return new Queue(aiUserStatQueue,true);
    }

    @Bean
    public Queue userDeductionStatQueue() {
        return new Queue(userDeductionStatQueue,true);
    }
    @Bean
    public Queue userBalanceInsufficientCheckoutQueue() {
        return new Queue(userBalanceInsufficientCheckoutQueue,true);
    }

    @Bean
    public Queue aiChangeQueue() {
        return new Queue(aiChangeQueue,true);
    }

    public Binding bindingExchange(DirectExchange exchange, Queue queueMessage, String routingKey) {
        return BindingBuilder.bind(queueMessage).to(exchange).with(routingKey);
    }

}

