package com.gxwebsoft.common.mq.producer.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.gxwebsoft.common.mq.config.RabbitMQConfig;
import com.gxwebsoft.common.mq.message.SyncMessage;
import com.gxwebsoft.common.mq.producer.SyncMessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;


import java.util.HashMap;
import java.util.Map;

/**
 * RabbitMQ 消息生产者实现
 */
@Slf4j
@Component
@ConditionalOnProperty(name = "sync.mq.enabled", havingValue = "true", matchIfMissing = true)
public class RabbitMQSyncProducer implements SyncMessageProducer, RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    private final RabbitTemplate rabbitTemplate;
    private final MessageConverter messageConverter;
    private final ObjectMapper objectMapper;

    public RabbitMQSyncProducer(RabbitTemplate rabbitTemplate, MessageConverter messageConverter, ObjectMapper objectMapper) {
        this.rabbitTemplate = rabbitTemplate;
        this.messageConverter = messageConverter;
        this.objectMapper = objectMapper;
        // 设置确认回调
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void sendSyncMessage(SyncMessage message) {
        sendSyncMessage(message, null);
    }

    @Override
    public void sendSyncMessage(SyncMessage message, SendCallback callback) {
        try {
            log.info("发送MQ消息: messageId={}, type={}, event={}, target={}",
                    message.getMessageId(), message.getMessageType(),
                    message.getEventType(), message.getTargetSystem());

            CorrelationData correlationData = new CorrelationData(message.getMessageId());

            if (callback != null) {
                correlationData.getFuture().addCallback(
                        result -> {
                            if (result.isAck()) {
                                callback.onSuccess(message.getMessageId());
                            } else {
                                callback.onFailure(message.getMessageId(),
                                        new RuntimeException("消息发送未被确认"));
                            }
                        },
                        ex -> callback.onFailure(message.getMessageId(), ex)
                );
            }

            // 使用 targetSystem 作为 routing key
            // 格式: user.sync.{targetSystem}
            // 各子系统绑定队列时使用 pattern: user.sync.{systemName}
            String routingKey = buildRoutingKey(message.getTargetSystem());
            
            rabbitTemplate.convertAndSend(
                    RabbitMQConfig.SYNC_EXCHANGE,
                    routingKey,
                    message,
                    correlationData
            );

        } catch (Exception e) {
            log.error("发送MQ消息失败: messageId={}, error={}", message.getMessageId(), e.getMessage(), e);
            if (callback != null) {
                callback.onFailure(message.getMessageId(), e);
            }
        }
    }

    /**
     * 构建 routing key
     * 格式: user.sync.{targetSystem}
     */
    private String buildRoutingKey(String targetSystem) {
        if (targetSystem == null || targetSystem.isEmpty()) {
            return "user.sync.all";
        }
        return "user.sync." + targetSystem.toLowerCase();
    }

    @Override
    public void sendUserSyncMessage(String targetSystem, String eventType, Object userData) {
        try {
            Map<String, Object> dataMap;
            if (userData instanceof Map) {
                dataMap = (Map<String, Object>) userData;
            } else {
                // 转换为Map
                dataMap = objectMapper.convertValue(userData, Map.class);
            }

            SyncMessage message = new SyncMessage("USER_SYNC", eventType, targetSystem, dataMap);
            sendSyncMessage(message);
        } catch (Exception e) {
            log.error("发送用户同步消息失败: targetSystem={}, eventType={}, error={}",
                    targetSystem, eventType, e.getMessage(), e);
        }
    }

    /**
     * 确认回调 - Broker确认收到消息
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String messageId = correlationData.getId();
        if (ack) {
            log.debug("消息确认成功: messageId={}", messageId);
        } else {
            log.warn("消息确认失败: messageId={}, cause={}", messageId, cause);
        }
    }

    /**
     * Return回调 - 消息无法路由时回调
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText,
                                 String exchange, String routingKey) {
        log.error("消息无法路由: exchange={}, routingKey={}, replyCode={}, replyText={}",
                exchange, routingKey, replyCode, replyText);
    }
}
