// 下单成功后发消息
@Service
public class OrderService {

    @Transactional
    public void createOrder(OrderRequest req) {
        orderMapper.insert(req.toOrder());           // 1. 写 MySQL
        rabbitTemplate.convertAndSend(                // 2. 发消息
            "order.exchange", "order.created", req);
    }
}

rabbitTemplate.convertAndSend("order.exchange", "order.created", msg);
// 这一行执行完，消息真的到 RabbitMQ 了吗？不一定。
// 网络抖动、Exchange 不存在——这行代码不抛异常也不报错。

spring:
  rabbitmq:
    # 开启发送端确认
    publisher-confirm-type: correlated
    # 开启发送端回退——当消息没有路由到任何队列时通知生产者
    publisher-returns: true

@Configuration
public class RabbitConfirmConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(
            ConnectionFactory factory,
            Jackson2JsonMessageConverter converter) {

        RabbitTemplate template = new RabbitTemplate(factory);
        template.setMessageConverter(converter);

        // ===== 1. Confirm 回调：消息是否到达 Exchange =====
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                // Broker 确认收到消息
                String msgId = correlationData != null ? correlationData.getId() : "null";
                log.info("消息已到达 Broker，消息ID: {}", msgId);
            } else {
                // Broker 拒绝（Exchange 不存在等）
                log.error("消息发送失败！原因: {}", cause);
                // 补偿逻辑：写 DB 重试表、发告警...
            }
        });

        // ===== 2. Return 回调：消息到达 Exchange 但没路由到任何队列 =====
        template.setReturnsCallback(returned -> {
            String msg = new String(returned.getMessage().getBody());
            log.error("消息路由失败！exchange={}, routingKey={}, body={}",
                    returned.getExchange(), returned.getRoutingKey(), msg);
            // 补偿逻辑...
        });

        return template;
    }
}

@Service
public class ReliableOrderSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendOrderCreated(OrderMessage msg) {
        // CorrelationData 关联业务 ID，Confirm 回调中可以拿到
        CorrelationData correlationData = new CorrelationData(
                "order:" + msg.getOrderId() + ":" + UUID.randomUUID()
        );

        rabbitTemplate.convertAndSend(
            "order.direct",
            "order.created",
            msg,
            correlationData       // ← 带上这个，回调中能拿到
        );
    }
}

public void sendReliable(OrderMessage msg) {
    CorrelationData cd = new CorrelationData("order:" + msg.getOrderId());
    // 三次重试
    for (int i = 0; i < 3; i++) {
        rabbitTemplate.convertAndSend("order.direct", "order.created", msg, cd);
        try {
            // 同步等待确认（超时 5s）
            Confirm confirm = rabbitTemplate.waitForConfirms(5000);
            if (confirm != null && confirm.isAck()) {
                return;  // 成功，退出
            }
        } catch (Exception e) {
            log.warn("第{}次发送失败: {}", i + 1, e.getMessage());
        }
    }
    // 三次都失败 → 写 DB 重试表，定时任务补偿
    saveToRetryTable(msg);
}

@Configuration
public class DurableConfig {

    // Exchange 持久化
    @Bean
    public DirectExchange durableExchange() {
        return new DirectExchange("order.durable", true, false);
        //                         名称,              durable, autoDelete
    }

    // Queue 持久化
    @Bean
    public Queue durableQueue() {
        return QueueBuilder.durable("queue.durable").build();
    }

    // Binding 持久化（随 Queue 和 Exchange 自动持久化）
    @Bean
    public Binding durableBinding() {
        return BindingBuilder
                .bind(durableQueue())
                .to(durableExchange())
                .with("order.created");
    }
}

// 方式一：发送时指定
rabbitTemplate.convertAndSend("order.durable", "order.created", msg,
    message -> {
        // deliveryMode=2 即 PERSISTENT
        message.getMessageProperties().setDeliveryMode(
                MessageDeliveryMode.PERSISTENT);
        return message;
    }
);

// 方式二：RabbitTemplate 全局默认持久化
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
    RabbitTemplate template = new RabbitTemplate(factory);
    // 所有消息默认持久化——但会降低吞吐量
    template.setMessageConverter(new Jackson2JsonMessageConverter());
    return template;
}

// ❌ 自动 ACK：消息推给消费者后立即从队列中删除
//    消费者处理到一半 JVM 崩了 → 消息丢了
@RabbitListener(queues = "queue.order.create", ackMode = "AUTO")
public void handle(OrderMessage msg) { ... }

// ✅ 手动 ACK：消费者处理完显式确认
@RabbitListener(queues = "queue.order.create", ackMode = "MANUAL")
public void handle(OrderMessage msg, Channel channel,
                   @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
    try {
        processOrder(msg);
        channel.basicAck(tag, false);   // 处理成功 → 确认
    } catch (Exception e) {
        channel.basicNack(tag, false, true);  // 处理失败 → 重新入队
    }
}

@Configuration
public class DeadLetterConfig {

    // ========== 死信交换机 + 死信队列 ==========
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("dlx.exchange", true, false);
    }

    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable("queue.dlx").build();
    }

    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder
                .bind(deadLetterQueue())
                .to(deadLetterExchange())
                .with("order.dead");
    }

    // ========== 业务队列 —— 指定死信交换机 ==========
    @Bean
    public Queue orderBusinessQueue() {
        return QueueBuilder.durable("queue.order.business")
                // 死信交换机——这个队列中的消息变成死信后发给 dlx.exchange
                .deadLetterExchange("dlx.exchange")
                // 死信 RoutingKey——发到 dlx.exchange 时使用的 RoutingKey
                .deadLetterRoutingKey("order.dead")
                // 消息在队列中的 TTL（30 分钟——超时未消费也进死信）
                .ttl(30 * 60 * 1000)
                // 队列最大长度（防积压爆内存）
                .maxLength(10000)
                .build();
    }

    // 绑定业务队列到普通的 Direct Exchange
    @Bean
    public Binding orderBusinessBinding() {
        return BindingBuilder
                .bind(orderBusinessQueue())
                .to(new DirectExchange("order.direct", true, false))
                .with("order.created");
    }
}

@Component
public class OrderBusinessListener {

    @RabbitListener(queues = "queue.order.business", ackMode = "MANUAL")
    public void handle(OrderMessage msg, Channel channel,
                       @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {

        // 从消息头中拿重试次数（需要发送时设置或利用 RabbitMQ 的 x-death header）
        int maxRetries = 3;

        try {
            log.info("处理订单消息: orderId={}", msg.getOrderId());
            // 业务处理...
            processOrder(msg);

            // 成功 → ACK
            channel.basicAck(tag, false);

        } catch (Exception e) {
            log.error("处理消息失败: orderId={}, error={}", msg.getOrderId(), e.getMessage());

            // 检查重试次数
            Integer retryCount = getRetryCount(msg);  // 从消息头或 Redis 中取

            if (retryCount < maxRetries) {
                // 还没到上限 → 重新入队重试
                incrementRetryCount(msg, retryCount + 1);
                channel.basicNack(tag, false, true);
                log.info("重新入队, 重试次数: {}/{}", retryCount + 1, maxRetries);
            } else {
                // 达到上限 → 不重新入队，消息变成死信进入 DLQ
                channel.basicNack(tag, false, false);
                log.warn("已达最大重试次数, 消息进入死信队列: orderId={}", msg.getOrderId());
                // 同时发告警
                alertService.sendAlert("消息处理失败进死信: " + msg.getOrderId());
            }
        }
    }
}

@Component
public class DeadLetterMonitor {

    @RabbitListener(queues = "queue.dlx")
    public void handleDeadLetter(Message message, Channel channel,
                                  @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        String body = new String(message.getBody());
        Map<String, Object> headers = message.getMessageProperties().getHeaders();

        // x-death 是 RabbitMQ 自动添加的 header，记录死信来源
        List<Map<String, Object>> deathInfo = (List<Map<String, Object>>)
                headers.get("x-death");

        if (deathInfo != null && !deathInfo.isEmpty()) {
            Map<String, Object> death = deathInfo.get(0);
            log.error("死信消息: body={}, 来源队列={}, 原因={}, 原始RoutingKey={}",
                    body,
                    death.get("queue"),       // 从哪个队列来的
                    death.get("reason"),      // 为什么变成死信（rejected/expired/maxlen）
                    death.get("routing-keys") // 原始 RoutingKey
            );
        }

        // 手动 ACK——不要让死信消息永远留在死信队列里
        try {
            channel.basicAck(tag, false);
        } catch (IOException e) {
            log.error("ACK 死信消息失败", e);
        }
    }
}

@Component
public class IdempotentOrderListener {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @RabbitListener(queues = "queue.order.business", ackMode = "MANUAL")
    public void handle(OrderMessage msg, Channel channel,
                       @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {

        // 幂等 Key：订单ID + 操作类型
        String idempotentKey = "consumed:order:" + msg.getOrderId() + ":" + msg.getAction();

        // 尝试写入 Redis——SETNX，成功返回 true 表示第一次处理
        Boolean firstTime = redisTemplate.opsForValue()
                .setIfAbsent(idempotentKey, "1", Duration.ofHours(24));

        if (Boolean.FALSE.equals(firstTime)) {
            log.warn("重复消息，跳过: orderId={}, action={}",
                    msg.getOrderId(), msg.getAction());
            // 直接 ACK，不处理
            channel.basicAck(tag, false);
            return;
        }

        try {
            // 真正的业务处理
            processOrder(msg);
            channel.basicAck(tag, false);
        } catch (Exception e) {
            // 处理失败 → 删除幂等标记，让消息可以重试
            redisTemplate.delete(idempotentKey);
            channel.basicNack(tag, false, true);
        }
    }
}