package com.sonic.cow.config;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
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.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * @author chenjun
 */
@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 + "_" + "sonic";

    /** TODO: DEFAULT_SCENE + "_" + appName */
    public final static String PIGEON = DEFAULT_SCENE + "_" + "pigeon";

    /** TODO: DEFAULT_SCENE + "_" + appName */
    public final static String FROG = DEFAULT_SCENE + "_" + "frog";

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

    @Value("${mq.ai-text-gen-image.queue}")
    private String aiTextGenImageQueue;
    @Value("${mq.ai-text-gen-image.routing-key}")
    private String aiTextGenImageRoutingKey;

    @Value("${mq.ai-image-gen-image.queue}")
    private String aiImageGenImageQueue;
    @Value("${mq.ai-text-gen-image.routing-key}")
    private String aiImageGenImageRoutingKey;

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

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

    @Value("${mq.emotion-score.queue}")
    private String emotionScoreQueue;
    @Value("${mq.emotion-score.routing-key}")
    private String emotionScoreRoutingKey;

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

    @Value("${mq.voice-call-deduction-dead.queue}")
    public String voiceCallDeductionDeadQueue;
    @Value("${mq.voice-call-deduction-dead.routing-key}")
    public String voiceCallDeductionDeadRoutingKey;

    @Value("${mq.voice-call-deduction.queue}")
    public String voiceCallDeductionQueue;
    @Value("${mq.voice-call-deduction.routing-key}")
    public String voiceCallDeductionRoutingKey;

    @Value("${mq.voice-call-webhook.queue}")
    public String voiceCallWebhookQueue;
    @Value("${mq.voice-call-webhook.routing-key}")
    public String voiceCallWebhookRoutingKey;

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

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

        @RabbitListeners({
                @RabbitListener(queues = {"${mq.default.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.ai-text-gen-image.queue}"}, concurrency = "6"),
                @RabbitListener(queues = {"${mq.ai-image-gen-image.queue}"}, concurrency = "3"),
                @RabbitListener(queues = {"${mq.aiChat.queue}"}, concurrency = "5"),
                @RabbitListener(queues = {"${mq.emotion-score.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.voice-call-deduction.queue}"}, concurrency = "2"),
                @RabbitListener(queues = {"${mq.voice-call-webhook.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
    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
    RabbitmqEventProducer.RabbitmqMessageMeta aiTextGenImageMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiTextGenImageRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta aiImageGenImageMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, aiImageGenImageRoutingKey);
    }

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

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

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta emotionScoreMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, emotionScoreRoutingKey);
    }

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


    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta voiceCallDeductionDeadMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, voiceCallDeductionDeadRoutingKey);
    }

    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta voiceCallDeductionMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, voiceCallDeductionRoutingKey);
    }
    @Bean
    RabbitmqEventProducer.RabbitmqMessageMeta voiceCallWebhookMeta() {
        return RabbitmqEventProducer.RabbitmqMessageMeta.build(mqExchange, voiceCallWebhookRoutingKey);
    }

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



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


    @Bean
    public Binding bindingAiTextGenImageQueueExchange(DirectExchange exchange, Queue aiTextGenImageQueue) {
        return bindingExchange(exchange, aiTextGenImageQueue, aiTextGenImageRoutingKey);
    }

    @Bean
    public Binding bindingAiImageGenImageQueueExchange(DirectExchange exchange, Queue aiTextGenImageQueue) {
        return bindingExchange(exchange, aiTextGenImageQueue, aiImageGenImageRoutingKey);
    }

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

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

    @Bean
    public Binding bindingUserDeductionStatQueueExchange(DirectExchange exchange, Queue userDeductionStatQueue) {
        return bindingExchange(exchange, userDeductionStatQueue, userDeductionStatRoutingKey);
    }

    @Bean
    public Binding bindingVoiceCallDeductionDeadQueueExchange(DirectExchange exchange) {
        return bindingExchange(exchange, voiceCallDeductionDeadQueue(), voiceCallDeductionDeadRoutingKey);
    }


    @Bean
    public Binding bindingVoiceCallDeductionQueueExchange(DirectExchange exchange) {
        return bindingExchange(exchange, voiceCallDeductionQueue(), voiceCallDeductionRoutingKey);
    }

    @Bean
    public Binding bindingVoiceCallWebhookQueueExchange(DirectExchange exchange) {
        return bindingExchange(exchange, voiceCallWebhookQueue(), voiceCallWebhookRoutingKey);
    }

    @Bean
    public Binding bindingUserBalanceInsufficientCheckoutQueueExchange(DirectExchange exchange, Queue userBalanceInsufficientCheckoutQueue) {
        return bindingExchange(exchange, userBalanceInsufficientCheckoutQueue, userBalanceInsufficientCheckoutRoutingKey);
    }



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

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

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

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

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

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

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

    @Bean
    public Queue voiceCallDeductionDeadQueue() {
        Map map = Maps.newHashMap();
        //死信对列过期后转发的交换机
        map.put("x-dead-letter-exchange", mqExchange);
        //死信对列过期后转发后的路由键
        map.put("x-dead-letter-routing-key", voiceCallDeductionRoutingKey);
        return new Queue(voiceCallDeductionDeadQueue, true, false, false, map);
    }

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

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

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

}

