-- 消息表結(jié)構(gòu)
CREATE TABLE message (
    id VARCHAR(32) PRIMARY KEY,
    content TEXT NOT NULL,
    topic VARCHAR(100) NOT NULL,
    status VARCHAR(20) NOT NULL COMMENT 'INIT, SENDING, SENT, FAILED',
    retry_count INT DEFAULT 0,
    next_retry_time DATETIME,
    create_time DATETIME NOT NULL,
    update_time DATETIME NOT NULL,
    INDEX idx_status (status),
    INDEX idx_next_retry_time (next_retry_time)
);

-- 業(yè)務(wù)表示例(如訂單表)
CREATE TABLE orders (
    id VARCHAR(32) PRIMARY KEY,
    user_id VARCHAR(32) NOT NULL,
    amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(20) NOT NULL,
    create_time DATETIME NOT NULL,
    update_time DATETIME NOT NULL
);

public enum MessageStatus {
    INIT("初始化"),
    SENDING("發(fā)送中"),
    SENT("已發(fā)送"),
    FAILED("發(fā)送失敗");

    private final String description;

    MessageStatus(String description) {
        this.description = description;
    }

    public String getDescription() {
        return description;
    }
}

public interface MessageRepository extends JpaRepository<Message, String> {

    List<Message> findByStatusAndNextRetryTimeLessThanEqual(String status, LocalDateTime now);

    @Transactional
    @Modifying
    @Query("UPDATE Message m SET m.status = ?2, m.updateTime = ?3 WHERE m.id = ?1")
    int updateStatus(String id, String status, LocalDateTime updateTime);

    @Transactional
    @Modifying
    @Query("UPDATE Message m SET m.status = ?2, m.retryCount = m.retryCount + 1, " +
            "m.nextRetryTime = ?3, m.updateTime = ?3 WHERE m.id = ?1")
    int updateForRetry(String id, String status, LocalDateTime nextRetryTime);

    @Transactional
    @Modifying
    @Query("DELETE FROM Message m WHERE m.createTime < ?1")
    int deleteOlderThan(LocalDateTime threshold);
}

@Service
public class MessageServiceImpl implements MessageService {

    @Autowired
    private MessageRepository messageRepository;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${message.retry.interval:60000}")
    private long retryInterval;

    @Value("${message.max.retry:10}")
    private int maxRetry;

    @Value("${message.retention.days:30}")
    private int retentionDays;

    @Override
    @Transactional
    public Message saveMessage(String content, String topic) {
        Message message = new Message();
        message.setId(UUID.randomUUID().toString().replace("-", ""));
        message.setContent(content);
        message.setTopic(topic);
        message.setStatus(MessageStatus.INIT.name());
        message.setCreateTime(LocalDateTime.now());
        message.setUpdateTime(LocalDateTime.now());
        return messageRepository.save(message);
    }

    @Override
    public void sendMessage(Message message) {
        try {
            // 更新消息狀態(tài)為發(fā)送中
            messageRepository.updateStatus(message.getId(), MessageStatus.SENDING.name(), LocalDateTime.now());
            
            // 發(fā)送消息到RabbitMQ
            rabbitTemplate.convertAndSend(message.getTopic(), message.getContent());
            
            // 更新消息狀態(tài)為已發(fā)送
            messageRepository.updateStatus(message.getId(), MessageStatus.SENT.name(), LocalDateTime.now());
        } catch (Exception e) {
            // 發(fā)送失敗，更新重試信息
            handleSendFailure(message, e);
        }
    }

    @Override
    public void processPendingMessages() {
        List<Message> pendingMessages = messageRepository.findByStatusAndNextRetryTimeLessThanEqual(
                MessageStatus.INIT, LocalDateTime.now());
        
        for (Message message : pendingMessages) {
            sendMessage(message);
        }
    }

    @Override
    public void retryFailedMessages() {
        List<Message> failedMessages = messageRepository.findByStatusAndNextRetryTimeLessThanEqual(
                MessageStatus.FAILED, LocalDateTime.now());
        
        for (Message message : failedMessages) {
            if (message.getRetryCount() < maxRetry) {
                try {
                    // 更新消息狀態(tài)為發(fā)送中
                    messageRepository.updateStatus(message.getId(), MessageStatus.SENDING.name(), LocalDateTime.now());
                    
                    // 重試發(fā)送消息
                    rabbitTemplate.convertAndSend(message.getTopic(), message.getContent());
                    
                    // 更新消息狀態(tài)為已發(fā)送
                    messageRepository.updateStatus(message.getId(), MessageStatus.SENT.name(), LocalDateTime.now());
                } catch (Exception e) {
                    // 重試失敗，更新重試信息
                    handleSendFailure(message, e);
                }
            } else {
                // 超過最大重試次數(shù)，記錄日志并標(biāo)記為永久失敗
                // 可以添加告警機(jī)制
                System.err.println("Message exceeded max retry count: " + message.getId());
            }
        }
    }

    @Override
    public void cleanOldMessages() {
        LocalDateTime threshold = LocalDateTime.now().minusDays(retentionDays);
        int deletedCount = messageRepository.deleteOlderThan(threshold);
        System.out.println("Deleted " + deletedCount + " old messages");
    }

    @Override
    public void updateMessageStatus(String messageId, MessageStatus status) {
        messageRepository.updateStatus(messageId, status.name(), LocalDateTime.now());
    }

    private void handleSendFailure(Message message, Exception e) {
        // 計(jì)算下一次重試時(shí)間
        LocalDateTime nextRetryTime = LocalDateTime.now().plusSeconds(retryInterval);
        
        // 更新消息狀態(tài)為失敗并增加重試次數(shù)
        messageRepository.updateForRetry(
                message.getId(), 
                MessageStatus.FAILED.name(), 
                nextRetryTime
        );
        
        // 記錄錯(cuò)誤日志
        System.err.println("Failed to send message: " + message.getId() + ", error: " + e.getMessage());
    }
}

@Component
public class MessageScheduler {

    @Autowired
    private MessageService messageService;


    /**
     * 定時(shí)處理待發(fā)送的消息
     */
    @Scheduled(fixedRate = 10000) // 每10秒執(zhí)行一次
    public void processPendingMessages() {
        messageService.processPendingMessages();
    }

    /**
     * 定時(shí)重試失敗的消息
     */
    @Scheduled(fixedRate = 30000) // 每30秒執(zhí)行一次
    public void retryFailedMessages() {
        messageService.retryFailedMessages();
    }

    /**
     * 定時(shí)清理過期消息
     */
    @Scheduled(fixedRateString = "${message.clean.interval}")
    public void cleanOldMessages() {
        messageService.cleanOldMessages();
    }
}

@Service
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrderRepository orderRepository;

    @Autowired
    private MessageService messageService;

    @Override
    @Transactional
    public Order createOrder(String userId, Double amount) {
        // 創(chuàng)建訂單
        Order order = new Order();
        order.setId(UUID.randomUUID().toString().replace("-", ""));
        order.setUserId(userId);
        order.setAmount(amount);
        order.setStatus("CREATED");
        order.setCreateTime(LocalDateTime.now());
        order.setUpdateTime(LocalDateTime.now());
        
        // 保存訂單
        orderRepository.save(order);
        
        // 在同一事務(wù)中保存消息（關(guān)鍵：確保訂單創(chuàng)建和消息記錄在同一個(gè)本地事務(wù)中）
        String messageContent = String.format("{\"orderId\":\"%s\",\"userId\":\"%s\",\"amount\":%.2f}", 
                order.getId(), userId, amount);
        messageService.saveMessage(messageContent, "order.created");
        
        return order;
    }
}

@Component
public class OrderCreatedConsumer {

    @Autowired
    private PaymentService paymentService;

    @Autowired
    private MessageService messageService;

    @Autowired
    private ObjectMapper objectMapper;

    @RabbitListener(queues = "order.created")
    public void handleOrderCreated(String message) {
        try {
            // 解析消息內(nèi)容
            Map<String, Object> messageData = objectMapper.readValue(message, Map.class);
            String orderId = (String) messageData.get("orderId");
            String userId = (String) messageData.get("userId");
            Double amount = Double.parseDouble(messageData.get("amount").toString());
            
            // 處理訂單創(chuàng)建事件（例如：創(chuàng)建支付記錄）
            paymentService.createPayment(orderId, userId, amount);
            
            // 可以在這里添加其他業(yè)務(wù)邏輯，如扣減庫存、發(fā)送通知等
            
        } catch (Exception e) {
            // 消費(fèi)失敗，記錄日志
            System.err.println("Failed to process order created message: " + e.getMessage());
            // 注意：RabbitMQ默認(rèn)會重試，可能導(dǎo)致消息重復(fù)消費(fèi)，需要在業(yè)務(wù)層處理冪等性
        }
    }
}

@Service
public class PaymentServiceImpl implements PaymentService {

    @Override
    @Transactional
    public void createPayment(String orderId, String userId, Double amount) {
        // 實(shí)現(xiàn)支付邏輯
        // 例如：創(chuàng)建支付記錄、調(diào)用支付網(wǎng)關(guān)、更新訂單狀態(tài)等
        
        System.out.println("Creating payment for order: " + orderId + ", amount: " + amount);
        
        // 這里只是示例，實(shí)際應(yīng)用中需要根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)具體邏輯
    }
}

//一鎖：先加一個(gè)分布式鎖
@DistributeLock(scene = "OEDER", keyExpression = "#request.identifier", expire = 3000)
public OrderResponse apply(OrderRequest request) {
    OrderResponse response = new OrderResponse();   
    //二判：判斷請求是否執(zhí)行成功過
    OrderDTO orderDTO = orderService.queryOrder(request.getProduct(), request.getIdentifier());
    if (orderDTO != null) {
        response.setSuccess(true);
        response.setResponseCode("DUPLICATED");
        return response;
    }
    //三更新：執(zhí)行更新的業(yè)務(wù)邏輯   
    return orderService.order(request);
}