#IM消息可靠性保证

@Service
class MessageService {
    @Autowired
    private MessageDao messageDao;
    @Autowired
    private PushService pushService;
    @Autowired
    private OfflineQueue offlineQueue;

    /**
     * 发送消息
     */
    public Message send(MessageRequest request) {
        // 1. 生成消息 ID
        long messageId = idGenerator.nextId();

        // 2. 持久化消息（最重要的一步）
        Message message = new Message();
        message.setId(messageId);
        message.setFromUserId(request.getFromUserId());
        message.setToUserId(request.getToUserId());
        message.setContent(request.getContent());
        message.setTimestamp(System.currentTimeMillis());
        message.setStatus(MessageStatus.PENDING);
        messageDao.save(message);

        // 3. 更新状态为已存储
        message.setStatus(MessageStatus.STORED);
        messageDao.update(message);

        // 4. 推送给接收方
        if (userService.isOnline(request.getToUserId())) {
            pushService.pushToUser(request.getToUserId(), message);
            message.setStatus(MessageStatus.DELIVERED);
        } else {
            // 接收方不在线，存入离线队列
            offlineQueue.push(request.getToUserId(), message);
            message.setStatus(MessageStatus.QUEUED);
        }

        messageDao.update(message);
        return message;
    }
}

class MessageClient {
    private final Map<Long, Message> pendingMessages = new ConcurrentHashMap<>();

    /**
     * 发送消息并等待 ACK
     */
    public CompletableFuture<Message> send(Message message) {
        pendingMessages.put(message.getId(), message);

        return CompletableFuture.supplyAsync(() -> {
            // 1. 发送消息到服务器
            server.send(message);

            // 2. 等待 ACK
            return waitForAck(message.getId(), 30000); // 30秒超时
        }).thenApply(ack -> {
            pendingMessages.remove(message.getId());
            return message;
        }).exceptionally(ex -> {
            // 发送失败，标记重试
            markForRetry(message);
            throw new RuntimeException(ex);
        });
    }
}

@Service
class IdempotentService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private static final String DEDUP_PREFIX = "msg:dedup:";

    /**
     * 检查消息是否重复
     */
    public boolean isDuplicate(long messageId) {
        String key = DEDUP_PREFIX + messageId;

        // SETNX 原子操作
        Boolean result = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", Duration.ofDays(7));

        return !Boolean.TRUE.equals(result);
    }

    /**
     * 处理消息
     */
    public void handleMessage(Message message) {
        if (isDuplicate(message.getId())) {
            log.info("消息重复: {}", message.getId());
            return;
        }

        // 处理业务逻辑
        doProcess(message);
    }
}

class MessageClient {
    private final Set<Long> receivedMessageIds = ConcurrentHashMap.newKeySet();

    /**
     * 处理收到的消息
     */
    public void onMessageReceived(Message message) {
        if (receivedMessageIds.contains(message.getId())) {
            // 重复消息，忽略
            return;
        }

        receivedMessageIds.add(message.getId());

        // 渲染消息
        renderMessage(message);

        // 发送已读回执
        sendReadReceipt(message.getId());
    }
}

@Service
class MessageService {
    /**
     * 获取会话消息（保证顺序）
     */
    public List<Message> getConversation(Long userId1, Long userId2,
                                        Long lastMessageId, int limit) {
        // 1. 按消息 ID 升序获取
        List<Message> messages = messageDao.findByConversation(
            userId1, userId2, lastMessageId, limit
        );

        // 2. 按时间戳排序（作为保底）
        messages.sort(Comparator.comparingLong(Message::getId));

        return messages;
    }
}

class MessageRenderer {
    private final Map<String, TreeMap<Long, Message>> pendingMessages =
        new ConcurrentHashMap<>();

    /**
     * 渲染消息，处理乱序
     */
    public void renderMessage(Long conversationId, Message message) {
        TreeMap<Long, Message> conversation =
            pendingMessages.computeIfAbsent(
                String.valueOf(conversationId),
                k -> new TreeMap<>()
            );

        conversation.put(message.getId(), message);

        // 渲染连续的消息
        Long lowestKey = conversation.firstKey();
        List<Message> toRender = new ArrayList<>();

        for (Map.Entry<Long, Message> entry : conversation.entrySet()) {
            if (entry.getKey() == lowestKey + toRender.size()) {
                toRender.add(entry.getValue());
            } else {
                break;
            }
        }

        // 渲染并移除已渲染的消息
        for (Message m : toRender) {
            render(m);
            conversation.remove(m.getId());
        }
    }
}

@Service
class OfflineMessageService {
    @Autowired
    private OfflineMessageDao offlineMessageDao;

    /**
     * 拉取离线消息
     */
    public List<Message> pullOfflineMessages(Long userId, Long lastMessageId) {
        // 1. 查询离线消息
        List<OfflineMessage> offlineMessages =
            offlineMessageDao.findByUserIdAfter(userId, lastMessageId);

        List<Message> messages = offlineMessages.stream()
            .map(OfflineMessage::getMessage)
            .collect(Collectors.toList());

        // 2. 清理已拉取的离线消息
        offlineMessageDao.deleteByUserIdAndMessageIdBefore(
            userId, lastMessageId
        );

        return messages;
    }
}

@Service
class IncrementalSyncService {
    /**
     * 增量同步消息（基于时间戳）
     */
    public SyncResult sync(Long userId, Long lastSyncTime) {
        // 1. 查询增量消息
        List<Message> newMessages = messageDao.findAfter(userId, lastSyncTime);

        // 2. 查询已删除消息
        List<Long> deletedIds = deletedMessageDao.findAfter(userId, lastSyncTime);

        return new SyncResult(
            newMessages,
            deletedIds,
            System.currentTimeMillis()
        );
    }
}