package com.gxwebsoft.common.mq.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
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;

/**
 * RabbitMQ 配置类
 */
@Configuration
@ConditionalOnProperty(name = "sync.mq.enabled", havingValue = "true", matchIfMissing = true)
public class RabbitMQConfig {

    // ==================== 常量定义 ====================
    public static final String SYNC_EXCHANGE = "sync.topic.exchange";
    public static final String SYNC_QUEUE = "sync.queue";
    public static final String SYNC_ROUTING_KEY = "sync.message";

    // 死信队列
    public static final String DLX_EXCHANGE = "sync.dlx.exchange";
    public static final String DLQ_QUEUE = "sync.dlq";
    public static final String DLQ_ROUTING_KEY = "sync.dlq";

    @Value("${spring.rabbitmq.host:localhost}")
    private String host;

    @Value("${spring.rabbitmq.port:5672}")
    private int port;

    @Value("${spring.rabbitmq.username:guest}")
    private String username;

    @Value("${spring.rabbitmq.password:guest}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:/}")
    private String virtualHost;

    // ==================== Connection Factory ====================

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        // 开启publisher-confirm确认模式
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        // 开启publisher-return确认模式
        connectionFactory.setPublisherReturns(true);
        return connectionFactory;
    }

    // ==================== Message Converter ====================

    @Bean
    public MessageConverter messageConverter(ObjectMapper objectMapper) {
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    // ==================== RabbitTemplate ====================

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);
        // 设置Mandatory为true，才能触发ReturnCallback
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
            ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(messageConverter);
        // 设置并发数
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(5);
        // 设置手动ack
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 预取数量
        factory.setPrefetchCount(10);
        return factory;
    }

    // ==================== 交换机 ====================

    /**
     * 用户同步 Topic Exchange
     * 使用 Topic 类型，支持按 targetSystem 路由到不同队列
     * routing key 格式: user.sync.{targetSystem}
     * 各子系统可以绑定自己的队列来消费消息
     */
    @Bean
    public TopicExchange syncExchange() {
        return new TopicExchange(SYNC_EXCHANGE, true, false);
    }

    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange(DLX_EXCHANGE, true, false);
    }

    // ==================== 队列 ====================

    /**
     * 注意：core 系统只负责发送消息，不消费消息
     * 各子系统（websopy等）需要在自己的系统中配置消费者和队列
     * 
     * 如果 core 系统也需要消费某些消息，可以在这里添加对应的队列
     */

    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_QUEUE).build();
    }

    // ==================== 绑定 ====================

    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(deadLetterQueue())
                .to(deadLetterExchange())
                .with(DLQ_ROUTING_KEY);
    }
}
