Design WhatsApp - Messaging Platform

public class MessageDeliveryService {
    private WebSocketConnectionPool wsPool;
    private MessageQueue offlineQueue;
    private CassandraClient cassandra;
    private RedisCache redis;
    private KafkaProducer kafkaProducer;
    /**
     * Sends a message from sender to receiver.
     * Delivers immediately if online, queues if offline.
     */
    public MessageResponse sendMessage(SendMessageRequest request) {
        // 1. Generate unique message ID
        String messageId = UUID.randomUUID().toString();
        long timestamp = System.currentTimeMillis();
        // 2. Encrypt message content (end-to-end encryption)
        byte[] encryptedContent = encryptMessage(
            request.getContent(),
            request.getReceiverId()
        );
        // 3. Create message object
        Message message = new Message(
            messageId,
            request.getSenderId(),
            request.getReceiverId(),
            request.getMessageType(),
            encryptedContent,
            timestamp
        );
        // 4. Store message in Cassandra (persistent storage)
        cassandra.insertMessage(message);
        // 5. Check if receiver is online
        boolean isReceiverOnline = redis.get("user:online:" + request.getReceiverId());
        if (isReceiverOnline) {
            // 6a. Receiver is ONLINE - send via WebSocket immediately
            WebSocketConnection receiverConnection = wsPool.getConnection(
                request.getReceiverId()
            );
            if (receiverConnection != null && receiverConnection.isOpen()) {
                receiverConnection.send(message);
                message.setStatus("delivered");
                // Update status in database
                cassandra.updateMessageStatus(messageId, "delivered");
            } else {
                // Connection dropped, queue for later
                queueOfflineMessage(request.getReceiverId(), messageId);
                message.setStatus("sent");
            }
        } else {
            // 6b. Receiver is OFFLINE - queue message
            queueOfflineMessage(request.getReceiverId(), messageId);
            message.setStatus("sent");
        }
        // 7. Send delivery receipt to sender
        notifySender(request.getSenderId(), messageId, message.getStatus());
        // 8. Publish event to Kafka for analytics
        kafkaProducer.send("message-sent", new MessageEvent(
            messageId, request.getSenderId(), request.getReceiverId(), timestamp
        ));
        return new MessageResponse(messageId, message.getStatus(), timestamp);
    }
    /**
     * Handles user coming online - delivers queued messages.
     */
    public void handleUserOnline(long userId) {
        System.out.println("User " + userId + " came online");
        // 1. Mark user as online in Redis
        redis.set("user:online:" + userId, true, 3600); // 1 hour TTL
        // 2. Get all queued messages for this user
        List<String> queuedMessageIds = offlineQueue.getMessages(userId);
        if (queuedMessageIds.isEmpty()) {
            return;
        }
        System.out.println("Delivering " + queuedMessageIds.size() +
                           " queued messages to user " + userId);
        // 3. Get WebSocket connection
        WebSocketConnection connection = wsPool.getConnection(userId);
        if (connection == null || !connection.isOpen()) {
            System.err.println("Failed to get connection for user " + userId);
            return;
        }
        // 4. Fetch messages from Cassandra and deliver
        for (String messageId : queuedMessageIds) {
            Message message = cassandra.getMessage(messageId);
            if (message != null) {
                // Send message via WebSocket
                connection.send(message);
                // Update status to delivered
                cassandra.updateMessageStatus(messageId, "delivered");
                // Notify sender about delivery
                notifySender(message.getSenderId(), messageId, "delivered");
            }
            // Remove from queue
            offlineQueue.removeMessage(userId, messageId);
        }
    }
    /**
     * Queues message for offline user.
     */
    private void queueOfflineMessage(long userId, String messageId) {
        offlineQueue.addMessage(userId, messageId);
        // Also store in Redis sorted set with timestamp for quick access
        redis.zadd("offline:messages:" + userId,
                   System.currentTimeMillis(),
                   messageId);
    }
    /**
     * Encrypts message using receiver's public key (end-to-end encryption).
     */
    private byte[] encryptMessage(String content, long receiverId) {
        // In reality: use Signal Protocol or similar E2E encryption
        // For demo, simplified
        PublicKey receiverPublicKey = getPublicKey(receiverId);
        return CryptoUtil.encrypt(content.getBytes(), receiverPublicKey);
    }
    /**
     * Notifies sender about message status update.
     */
    private void notifySender(long senderId, String messageId, String status) {
        WebSocketConnection senderConnection = wsPool.getConnection(senderId);
        if (senderConnection != null && senderConnection.isOpen()) {
            senderConnection.send(new StatusUpdate(messageId, status));
        }
    }
}

public class GroupMessagingService {
    private MessageDeliveryService messageDelivery;
    private CassandraClient cassandra;
    private RedisCache redis;
    /**
     * Sends a message to a group chat.
     * Fans out message to all group members.
     */
    public GroupMessageResponse sendGroupMessage(GroupMessageRequest request) {
        String groupId = request.getGroupId();
        long senderId = request.getSenderId();
        // 1. Verify sender is a member of the group
        Group group = getGroup(groupId);
        if (!group.getMembers().contains(senderId)) {
            throw new UnauthorizedException("User not in group");
        }
        // 2. Generate message ID and timestamp
        String messageId = UUID.randomUUID().toString();
        long timestamp = System.currentTimeMillis();
        // 3. Store group message once
        GroupMessage groupMessage = new GroupMessage(
            messageId,
            groupId,
            senderId,
            request.getContent(),
            timestamp
        );
        cassandra.insertGroupMessage(groupMessage);
        // 4. Fan out to all group members (except sender)
        List<Long> recipients = group.getMembers().stream()
            .filter(memberId -> memberId != senderId)
            .collect(Collectors.toList());
        System.out.println("Fanning out message to " + recipients.size() + " members");
        // 5. Use parallel processing for large groups
        if (recipients.size() > 50) {
            // Large group: use async processing
            fanOutAsync(messageId, recipients, groupMessage);
        } else {
            // Small group: send synchronously
            fanOutSync(messageId, recipients, groupMessage);
        }
        // 6. Return success to sender
        return new GroupMessageResponse(messageId, "sent", timestamp);
    }
    /**
     * Synchronous fan-out for small groups.
     */
    private void fanOutSync(String messageId, List<Long> recipients,
                            GroupMessage message) {
        for (Long recipientId : recipients) {
            try {
                // Check if recipient is online
                boolean isOnline = redis.get("user:online:" + recipientId);
                if (isOnline) {
                    // Deliver via WebSocket
                    WebSocketConnection conn = wsPool.getConnection(recipientId);
                    if (conn != null && conn.isOpen()) {
                        conn.send(message);
                    }
                } else {
                    // Queue for later delivery
                    offlineQueue.addMessage(recipientId, messageId);
                }
            } catch (Exception e) {
                System.err.println("Failed to deliver to " + recipientId + ": " +
                                   e.getMessage());
            }
        }
    }
    /**
     * Asynchronous fan-out for large groups (>50 members).
     */
    private void fanOutAsync(String messageId, List<Long> recipients,
                             GroupMessage message) {
        // Split recipients into batches of 100
        int batchSize = 100;
        List<List<Long>> batches = Lists.partition(recipients, batchSize);
        // Process batches in parallel using thread pool
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (List<Long> batch : batches) {
            executor.submit(() -> {
                fanOutSync(messageId, batch, message);
            });
        }
        executor.shutdown();
    }
    /**
     * Gets group information from cache or database.
     */
    private Group getGroup(String groupId) {
        // Try cache first
        String cacheKey = "group:" + groupId;
        Group cached = redis.get(cacheKey);
        if (cached != null) {
            return cached;
        }
        // Cache miss - fetch from database
        Group group = cassandra.getGroup(groupId);
        if (group != null) {
            // Cache for 1 hour
            redis.setWithExpiry(cacheKey, group, 3600);
        }
        return group;
    }
    /**
     * Adds member to group.
     */
    public void addMemberToGroup(String groupId, long userId, long adminId) {
        Group group = getGroup(groupId);
        // Verify admin permissions
        if (!group.getAdmins().contains(adminId)) {
            throw new UnauthorizedException("Only admins can add members");
        }
        // Add member to group
        group.getMembers().add(userId);
        cassandra.updateGroup(group);
        // Invalidate cache
        redis.delete("group:" + groupId);
        // Send system message to group
        sendSystemMessage(groupId, userId + " joined the group");
    }
    /**
     * Sends system message to group (e.g., "Alice added Bob").
     */
    private void sendSystemMessage(String groupId, String content) {
        GroupMessageRequest systemMsg = new GroupMessageRequest(
            groupId,
            0, // System sender
            "system",
            content
        );
        sendGroupMessage(systemMsg);
    }
}

public class MessageStatusService {
    private CassandraClient cassandra;
    private WebSocketConnectionPool wsPool;
    private RedisCache redis;
    /**
     * Updates message status when user receives/reads a message.
     */
    public void updateMessageStatus(String messageId, long userId,
                                     MessageStatus newStatus) {
        // 1. Fetch message from database
        Message message = cassandra.getMessage(messageId);
        if (message == null) {
            System.err.println("Message not found: " + messageId);
            return;
        }
        // 2. Verify user is the receiver
        if (message.getReceiverId() != userId) {
            System.err.println("User " + userId + " not authorized for message " +
                               messageId);
            return;
        }
        // 3. Update status (only allow forward progression)
        MessageStatus currentStatus = message.getStatus();
        if (!canTransition(currentStatus, newStatus)) {
            System.err.println("Invalid status transition: " + currentStatus +
                               " -> " + newStatus);
            return;
        }
        // 4. Update in database
        message.setStatus(newStatus);
        message.setStatusUpdatedAt(System.currentTimeMillis());
        cassandra.updateMessage(message);
        // 5. Send status update to sender (blue checkmarks!)
        notifySenderOfStatusChange(message.getSenderId(), messageId, newStatus);
        // 6. Update last seen if status is "read"
        if (newStatus == MessageStatus.READ) {
            updateLastSeen(userId);
        }
    }
    /**
     * Checks if status transition is valid.
     * Status progression: SENT -> DELIVERED -> READ
     */
    private boolean canTransition(MessageStatus current, MessageStatus next) {
        // Status hierarchy
        int currentLevel = getStatusLevel(current);
        int nextLevel = getStatusLevel(next);
        // Can only move forward, not backward
        return nextLevel > currentLevel;
    }
    private int getStatusLevel(MessageStatus status) {
        switch (status) {
            case SENT: return 1;
            case DELIVERED: return 2;
            case READ: return 3;
            default: return 0;
        }
    }
    /**
     * Sends status update notification to sender.
     * This is how the blue checkmarks appear!
     */
    private void notifySenderOfStatusChange(long senderId, String messageId,
                                             MessageStatus status) {
        WebSocketConnection senderConn = wsPool.getConnection(senderId);
        if (senderConn != null && senderConn.isOpen()) {
            // Send status update message
            StatusUpdateNotification notification = new StatusUpdateNotification(
                messageId,
                status.toString().toLowerCase(),
                System.currentTimeMillis()
            );
            senderConn.send(notification);
            System.out.println("Sent status update to sender " + senderId + ": " +
                               messageId + " is now " + status);
        } else {
            // Sender offline - they'll see the updated status when they come online
            System.out.println("Sender " + senderId + " offline, will sync later");
        }
    }
    /**
     * Marks multiple messages as read at once (bulk operation).
     */
    public void markConversationAsRead(long userId, String conversationId) {
        // 1. Get all unread messages in conversation
        List<Message> unreadMessages = cassandra.getUnreadMessages(
            conversationId,
            userId
        );
        if (unreadMessages.isEmpty()) {
            return;
        }
        System.out.println("Marking " + unreadMessages.size() +
                           " messages as read in conversation " + conversationId);
        // 2. Batch update all messages to READ status
        List<String> messageIds = unreadMessages.stream()
            .map(Message::getMessageId)
            .collect(Collectors.toList());
        cassandra.batchUpdateStatus(messageIds, MessageStatus.READ);
        // 3. Notify senders (group by sender to avoid duplicate notifications)
        Map<Long, List<String>> messagesBySender = unreadMessages.stream()
            .collect(Collectors.groupingBy(
                Message::getSenderId,
                Collectors.mapping(Message::getMessageId, Collectors.toList())
            ));
        for (Map.Entry<Long, List<String>> entry : messagesBySender.entrySet()) {
            long senderId = entry.getKey();
            List<String> senderMessageIds = entry.getValue();
            // Send single notification with all message IDs
            notifyBulkStatusChange(senderId, senderMessageIds, MessageStatus.READ);
        }
        // 4. Update last seen
        updateLastSeen(userId);
    }
    /**
     * Sends bulk status update for multiple messages.
     */
    private void notifyBulkStatusChange(long senderId, List<String> messageIds,
                                         MessageStatus status) {
        WebSocketConnection conn = wsPool.getConnection(senderId);
        if (conn != null && conn.isOpen()) {
            BulkStatusUpdate update = new BulkStatusUpdate(
                messageIds,
                status.toString().toLowerCase(),
                System.currentTimeMillis()
            );
            conn.send(update);
        }
    }
    /**
     * Updates user's last seen timestamp.
     */
    private void updateLastSeen(long userId) {
        long timestamp = System.currentTimeMillis();
        // Update in Redis for fast access
        redis.set("user:last_seen:" + userId, timestamp);
        // Also update in database (async)
        CompletableFuture.runAsync(() -> {
            cassandra.updateUserLastSeen(userId, timestamp);
        });
    }
    /**
     * Gets user's online status and last seen.
     */
    public UserStatus getUserStatus(long userId) {
        // Check if online
        boolean isOnline = redis.get("user:online:" + userId);
        if (isOnline) {
            return new UserStatus(userId, true, "online");
        } else {
            // Get last seen from cache
            Long lastSeen = redis.get("user:last_seen:" + userId);
            if (lastSeen == null) {
                // Cache miss - fetch from database
                lastSeen = cassandra.getUserLastSeen(userId);
                if (lastSeen != null) {
                    redis.set("user:last_seen:" + userId, lastSeen);
                }
            }
            return new UserStatus(userId, false, formatLastSeen(lastSeen));
        }
    }
    /**
     * Formats last seen timestamp (e.g., "5 minutes ago", "yesterday").
     */
    private String formatLastSeen(Long timestamp) {
        if (timestamp == null) {
            return "last seen a long time ago";
        }
        long now = System.currentTimeMillis();
        long diffSeconds = (now - timestamp) / 1000;
        if (diffSeconds < 60) {
            return "last seen just now";
        } else if (diffSeconds < 3600) {
            return "last seen " + (diffSeconds / 60) + " minutes ago";
        } else if (diffSeconds < 86400) {
            return "last seen " + (diffSeconds / 3600) + " hours ago";
        } else {
            return "last seen " + (diffSeconds / 86400) + " days ago";
        }
    }
    enum MessageStatus {
        SENT,
        DELIVERED,
        READ
    }
}