Design Amazon's Real-Time Messaging System

Difficulty: Mid/Senior Level

Understanding the Problem

🎯 What is Amazon's Real-Time Messaging System?

Amazon's Real-Time Messaging System is a scalable, distributed communication platform that enables instant message delivery across millions of concurrent users worldwide. Similar to WhatsApp or Telegram, this system must handle real-time message exchange, offline message delivery, file attachments, and maintain strong consistency guarantees while scaling to support Amazon's massive user base.

This system represents one of the most technically challenging distributed systems problems, requiring sophisticated solutions for message ordering, delivery guarantees, connection management, and horizontal scaling. The platform must process billions of messages daily while maintaining sub-second delivery times and ensuring no message loss under any circumstances.

As we'll explore in this comprehensive breakdown, we'll walk through this complex system step by step, covering real-time communication patterns, distributed system challenges, and scalability solutions. While we'll cover more technical depth than typically required in a single interview session, this detailed analysis helps build a thorough understanding of large-scale messaging architectures.

Complete Amazon Real-Time Messaging Platform System Architecture Overview

Functional Requirements

Core Requirements

• Users should be able to send real-time messages to other users with guaranteed delivery
• Users should be able to receive messages when offline and sync them upon reconnection
• Users should be able to send and receive file attachments (images, documents, videos) within messages
• Users should be able to view message delivery status (sent, delivered, read) in real-time

Out of Scope

• End-to-end encryption and advanced security features
• Voice and video calling capabilities
• Message search and advanced filtering
• Advanced group administration features (complex permissions, moderation tools)

Non-Functional Requirements

When discussing system requirements with your interviewer, it's essential to establish scale expectations upfront. These requirements will drive fundamental architectural decisions and help prioritize which technical challenges to address first.

Core Requirements

• The system should scale to handle 500 million concurrent users globally
• The system should provide real-time message delivery with sub-second latency
• The system should guarantee message durability - no messages can be lost under any circumstances
• The system should ensure strong message ordering and consistency for conversation threads

Out of Scope

• Advanced analytics and user behavior tracking
• Comprehensive audit logging and compliance features
• Multi-region disaster recovery and backup systems

Core Entities of our System

Before diving into system architecture, we need to establish the fundamental data models that will drive our messaging platform. We'll start with a conceptual overview and progressively add implementation details as we develop our API contracts. These entities form the backbone of our entire system design.

Our messaging platform centers around five essential entities that fulfill our core functional requirements:

• User: Represents individuals using the messaging platform, including authentication credentials, profile information, and connection status
• Chat: Defines conversation contexts between users, encompassing both direct messages and metadata for conversation management
• Message: Captures individual message content with delivery tracking, timestamps, and attachment references
• Client: Represents user devices and applications, enabling multi-device support and connection management
• Attachment: Handles file uploads and downloads with metadata, storage references, and lifecycle management

The decision to separate these entities provides several architectural benefits:

• Independent Scaling: Each entity can be optimized for distinct access patterns and throughput requirements
• Multi-Device Support: Client separation enables seamless synchronization across user devices
• Attachment Isolation: File handling can be scaled and optimized independently from message processing
• Conversation Context: Chat entities enable efficient message grouping and conversation-level features

During interviews, both normalized and denormalized approaches are valid - the critical aspect is articulating your design rationale and trade-offs clearly.

Let's define these entities with specific fields to guide our implementation:

// Detailed entity definitions for implementation
interface User {
  userId: string;
  phoneNumber: string;
  name: string;
  profilePicUrl?: string;
  status: string;
  createdAt: Date;
  lastSeen: Date;
  isActive: boolean;
}

interface Chat {
  chatId: string;
  name?: string;
  createdBy: string;
  createdAt: Date;
  lastActivity: Date;
  isGroupChat: boolean;
  adminUserIds?: string[];
}

interface Message {
  chatId: string;
  messageId: string;
  senderUserId: string;
  content: string;
  attachmentIds?: string[];
  sentAt: Date;
  deliveredToCount: number;
  readByCount: number;
}

interface Client {
  userId: string;
  clientId: string;
  deviceType: 'MOBILE' | 'DESKTOP' | 'WEB';
  lastActive: Date;
  pushToken?: string;
  platform: string;
  isOnline: boolean;
}

interface Attachment {
  attachmentId: string;
  ownerId: string;
  fileName: string;
  fileType: string;
  fileSize: number;
  blobUrl: string;
  hash: string;
  uploadedAt: Date;
  expiresAt: Date;
}

API or System Interface

Before diving into architectural details, we need to establish clear API contracts that define how clients interact with our messaging system. These interfaces will guide our implementation and ensure we address all functional requirements systematically.

For sending messages, we need a POST endpoint that handles message creation and delivery:

POST /chats/:chatId/messages -> Message
Authorization: Bearer <jwt_token>
{
  "content": "Hey! How are you doing?",
  "attachmentIds": []
}

// Response
{
  "messageId": "msg_def456",
  "chatId": "chat_xyz789",
  "status": "SENT",
  "sentAt": "2024-06-15T14:30:00Z",
  "deliveryStatus": {
    "pending": 1,
    "delivered": 0,
    "read": 0
  }
}

For real-time message delivery, we need WebSocket connections for live updates:

// WebSocket connection for real-time messaging
WebSocket /ws/messages

// Incoming message event (server sends to client)
{
  "type": "newMessage",
  "data": {
    "messageId": "msg_def456",
    "chatId": "chat_xyz789",
    "senderUserId": "user_abc123",
    "content": "Hey! How are you doing?",
    "sentAt": "2024-06-15T14:30:00Z"
  }
}

// Delivery acknowledgment
{
  "type": "messageDelivered",
  "data": {
    "messageId": "msg_def456",
    "chatId": "chat_xyz789"
  }
}

For attachment handling, we need endpoints for upload and download:

POST /attachments/upload -> Attachment
{
  "fileName": "vacation_photo.jpg",
  "fileType": "image/jpeg",
  "fileSize": 2048000
}

// Response
{
  "attachmentId": "att_ghi789",
  "uploadUrl": "https://s3.amazonaws.com/...",
  "expiresAt": "2024-06-15T15:00:00Z"
}

GET /attachments/:attachmentId/download -> AttachmentDownload
// Response
{
  "downloadUrl": "https://s3.amazonaws.com/...",
  "fileName": "vacation_photo.jpg",
  "fileSize": 2048000,
  "expiresAt": "2024-06-15T15:00:00Z"
}

High-Level Design

1) Users should be able to send real-time messages to other users with guaranteed delivery

The foundation of our messaging system begins with real-time message delivery capabilities. Users initiate this process through WebSocket connections, providing immediate bidirectional communication channels for instant message exchange.

Our initial architecture establishes sophisticated message routing and delivery mechanisms. The "Chat Server" serves as our primary real-time component, managing WebSocket connections and coordinating message delivery across distributed infrastructure. This service specializes in connection management, message routing, and real-time notification systems.

Real-Time Message Sending Flow - Chat Server Architecture

Detailed Message Sending Flow

When User A sends a message to User B, our system executes a sophisticated multi-step process ensuring both real-time delivery and guaranteed durability:

Step 1: Message Creation and Validation

When User A (sender) sends a message to User B (recipient):

• Client Construction: User A's client constructs a message payload containing chatId, message content, and any attachments
• WebSocket Transmission: The message travels via WebSocket to the Chat Server that User A (sender) is connected to
• Participant Validation: The server validates that User A is a legitimate participant in the specified chat by querying the ChatParticipants table using composite key (ChatId=chat123, UserId=userA_id)
• Active Status Verification: The server verifies User A's active status by checking the IsActive field in the retrieved record

Step 2: Message Storage Transaction

The server begins a DynamoDB transaction that handles multiple write operations atomically:

Messages Table Write:

• Generates a unique message ID with timestamp for ordering: msg-164901234567-a1b2c3d4
• Creates an item in the Messages table with:
  • ChatId (partition key): The 1:1 chat identifier between User A and User B
  • MessageId (sort key): The generated timestamp-based ID
  • SenderUserId: User A's ID
  • Content: The actual message text
  • AttachmentIds: Array of attachment IDs (empty if no attachments)
  • SentAt: Current timestamp in milliseconds
  • DeliveredToCount: Set to 0 initially
  • ReadByCount: Set to 0 initially

Recipient Client Lookup:

• Queries the Clients table to find all registered devices for User B
• Uses the UserId index on the Clients table: UserId=userB_id
• This retrieves all User B's devices (mobile, desktop, web) as multiple clients may exist

Inbox Table Writes:

• For each of User B's client devices retrieved above, the server creates an entry in the Inbox table:
  • ClientId (partition key): Each client device ID of User B
  • MessageId (sort key): The generated message ID
  • ChatId: The chat identifier
  • DeliveryStatus: Set to "PENDING"
  • CreatedAt: Current timestamp
  • ExpiresAt: Current timestamp + 30 days (TTL attribute for auto-deletion)

Chat Activity Update:

• Updates the LastActivity field in the Chats table for the specific chat
• This ensures chat sorting by recent activity works correctly

All the above operations are bundled in a single DynamoDB transaction, ensuring that either all succeed or all fail, maintaining data consistency.

Step 3: Determining Target User's Shard

After storing the message, the server needs to route it to the correct server handling User B (recipient):

Shard Calculation:

• The server calculates User B's shard using consistent hashing: userShard = hash(userB_id) % TOTAL_SHARDS (e.g., 1000 total shards)
• This maps User B to a specific shard (e.g., shard-456)

Constructing Delivery Payload:

• The server creates a delivery payload containing:
  • recipientId: User B's ID
  • command: "newMessage"
  • payload: Object containing message details (messageId, chatId, content, senderId, timestamp, attachments)

Step 4: Publishing to Redis Shard Channel

The server publishes the message to the appropriate Redis channel:

Channel Selection:

• Uses the calculated shard: messages:shard:456
• This ensures the message goes only to servers handling User B's shard

Message Publishing:

• Publishes the JSON-serialized payload to the Redis channel
• Redis immediately forwards this to all servers subscribed to this channel
• Only servers assigned to shard-456 will be subscribed to this channel

Acknowledgment to Sender:

• After publishing to Redis, the server sends a success acknowledgment back to User A
• Includes the generated messageId so User A's client can track the message

Step 5: Message Reception at Target Server

The server responsible for User B's shard processes the incoming Redis message:

Message Reception:

• Server handling shard-456 receives the message from Redis
• Deserializes the JSON payload to extract recipientId and message details

Connection Check:

• Checks its in-memory connectedUsers map to see if User B is currently connected
• This map maintains userId -> {clientId, connectionObject} for all connected users
• Lookup is O(1) operation for efficient message routing

For Connected Clients:

• If User B has active connections, the server iterates through each one
• For each connection, sends the message payload via the WebSocket
• Updates the Inbox entry for each client to DeliveryStatus="DELIVERED"
• Increments the DeliveredToCount counter in the Messages table

For Offline Clients:

• If User B isn't connected or has some devices offline, the corresponding Inbox entries remain with DeliveryStatus="PENDING"
• These will be delivered when those clients connect later

This comprehensive flow ensures that messages are delivered in real-time to online users while maintaining a reliable queue for offline delivery, all while preserving message ordering and consistency across the distributed system.

class ChatServer:
    def __init__(self, server_id, redis_client, dynamodb_client):
        self.server_id = server_id
        self.redis = redis_client
        self.dynamodb = dynamodb_client
        self.connected_users = {}  # userId -> {clientId, connection}
        
    async def handle_send_message(self, websocket, message_data, authenticated_user_id):
        """Handle incoming message from client"""
        try:
            # 1. Extract sender ID from authenticated JWT token (NOT from request body)
            sender_id = authenticated_user_id  # Extracted from JWT token by auth middleware
            chat_id = message_data['chatId']
            
            if not await self.validate_chat_participant(sender_id, chat_id):
                await websocket.send_json({
                    'type': 'error',
                    'message': 'Not authorized for this chat'
                })
                return
            
            # 2. Generate ordered message ID
            message_id = self.generate_message_id(chat_id)
            
            # 3. Store message in database transaction
            await self.store_message_transaction(
                message_id, chat_id, sender_id, 
                message_data['content'], message_data.get('attachmentIds', [])
            )
            
            # 4. Route message to recipients
            await self.route_message_to_recipients(chat_id, {
                'messageId': message_id,
                'chatId': chat_id,
                'senderUserId': sender_id,
                'content': message_data['content'],
                'sentAt': datetime.utcnow().isoformat()
            })
            
            # 5. Acknowledge to sender
            await websocket.send_json({
                'type': 'messageSent',
                'messageId': message_id,
                'status': 'DELIVERED'
            })
            
        except Exception as e:
            await self.handle_error(websocket, e)
    
    async def store_message_transaction(self, message_id, chat_id, sender_id, content, attachment_ids):
        """Store message with atomic transaction"""
        # Get all recipients for this chat
        recipients = await self.get_chat_participants(chat_id, exclude_user=sender_id)
        
        # Prepare transaction items
        transaction_items = []
        
        # 1. Store message
        transaction_items.append({
            'Put': {
                'TableName': 'Messages',
                'Item': {
                    'ChatId': chat_id,
                    'MessageId': message_id,
                    'SenderUserId': sender_id,
                    'Content': content,
                    'AttachmentIds': attachment_ids,
                    'SentAt': int(time.time() * 1000),
                    'DeliveredToCount': 0,
                    'ReadByCount': 0
                }
            }
        })
        
        # 2. Create inbox entries for all recipient clients
        for recipient_id in recipients:
            client_devices = await self.get_user_clients(recipient_id)
            for client in client_devices:
                transaction_items.append({
                    'Put': {
                        'TableName': 'Inbox',
                        'Item': {
                            'ClientId': client['ClientId'],
                            'MessageId': message_id,
                            'ChatId': chat_id,
                            'DeliveryStatus': 'PENDING',
                            'CreatedAt': int(time.time() * 1000),
                            'ExpiresAt': int(time.time() + 30 * 24 * 3600)  # 30 days TTL
                        }
                    }
                })
        
        # 3. Update chat last activity
        transaction_items.append({
            'Update': {
                'TableName': 'Chats',
                'Key': {'ChatId': chat_id},
                'UpdateExpression': 'SET LastActivity = :timestamp',
                'ExpressionAttributeValues': {
                    ':timestamp': int(time.time() * 1000)
                }
            }
        })
        
        # Execute transaction
        await self.dynamodb.transact_write_items(TransactItems=transaction_items)

2) Users should be able to receive messages when offline and sync them upon reconnection

Offline message delivery represents one of the most complex challenges in distributed messaging systems. Our architecture must ensure that users receive all messages sent during their offline periods, delivered in the correct order, and synchronized across all their devices upon reconnection.

Offline Message Sync - Reconnection and Message Delivery Flow

Comprehensive Offline Message Delivery Flow:

Scenario: User B was Offline and Reconnects Later

When User B opens the app after being offline, our system executes a sophisticated reconnection and synchronization process:

Step 1: Connection and Authentication

When User B opens the app after being offline:

Initial Connection:

• User B's client connects to the load balancer with authentication token
• Load balancer verifies the token and extracts User B's ID

Shard Determination:

• Load balancer calculates User B's shard: shard-456 = hash(userB_id) % TOTAL_SHARDS

Server Lookup in ZooKeeper:

• Load balancer queries ZooKeeper: /shards/456
• Retrieves JSON: {"primary": "server-27", "backups": ["server-35", "server-42"]}

Connection Routing:

• Load balancer redirects User B's connection to server-27 (primary for shard-456)
• If server-27 is down, routes to one of the backups

Step 2: Connection Registration

When server-27 receives the connection:

Connection Validation:

• Server verifies User B's authentication and shard assignment
• Confirms it is indeed responsible for User B's shard

Connection Storage:

• Registers User B's connection in its in-memory map: connectedUsers.set(userB_id, {clientId: "client123", connection: websocketObj})
• Updates User B's status in the Clients table: IsOnline=true, LastActive=current_timestamp

Client Sync Initiation:

• Sends confirmation to User B's client that connection is established
• Begins process to sync pending messages

Step 3: Pending Message Retrieval

The server fetches all pending messages for User B:

Inbox Query:

• Queries the Inbox table for pending messages for User B's clients: ClientId=client123, DeliveryStatus="PENDING"
• This retrieves all messages that arrived while User B was offline

Optimization Through Batching:

• Groups messages by ChatId to enable efficient batch retrieval
• For each chat, creates a batch of up to 25 message IDs (DynamoDB batch limit)

Message Content Retrieval:

• For each batch, queries the Messages table using BatchGetItem
• Retrieves full message content for all pending messages
• Orders messages by timestamp for correct sequencing

Step 4: Message Delivery

The server delivers the pending messages to User B:

Sequential Delivery:

• Sends messages to User B's client in chronological order
• Groups messages by chat for client-side organization

Delivery Status Update:

• For each message sent, updates the Inbox entry: DeliveryStatus="DELIVERED"
• Increments the DeliveredToCount counter in the Messages table

Client Processing:

• User B's client processes incoming messages in order
• Renders messages in appropriate chat threads
• Sends acknowledgments for received messages

Inbox Cleanup:

• As User B's client acknowledges messages, server removes corresponding entries from the Inbox table
• This prevents redelivery of already received messages

class OfflineMessageSync:
    def __init__(self, dynamodb_client, redis_client):
        self.dynamodb = dynamodb_client
        self.redis = redis_client
    
    async def handle_user_reconnection(self, user_id: str, client_id: str, websocket):
        """Handle user reconnection and sync offline messages"""
        try:
            # 1. Register connection
            await self.register_user_connection(user_id, client_id, websocket)
            
            # 2. Fetch pending messages
            pending_messages = await self.fetch_pending_messages(client_id)
            
            # 3. Group and sort messages
            organized_messages = self.organize_messages_by_chat(pending_messages)
            
            # 4. Deliver messages sequentially
            for chat_id, messages in organized_messages.items():
                await self.deliver_chat_messages(websocket, chat_id, messages)
            
            # 5. Clean up delivered messages
            await self.cleanup_delivered_messages(client_id, pending_messages)
            
        except Exception as e:
            await self.handle_sync_error(websocket, e)
    
    async def fetch_pending_messages(self, client_id: str):
        """Fetch all pending messages for a client"""
        # Query inbox for pending messages
        response = await self.dynamodb.query(
            TableName='Inbox',
            KeyConditionExpression='ClientId = :clientId',
            FilterExpression='DeliveryStatus = :status',
            ExpressionAttributeValues={
                ':clientId': client_id,
                ':status': 'PENDING'
            }
        )
        
        inbox_entries = response['Items']
        
        # Batch fetch message content
        message_ids = [entry['MessageId'] for entry in inbox_entries]
        messages = await self.batch_get_messages(message_ids)
        
        # Combine inbox metadata with message content
        return self.merge_inbox_with_messages(inbox_entries, messages)
    
    async def batch_get_messages(self, message_ids: list):
        """Efficiently fetch message content in batches"""
        messages = {}
        
        # Process in batches of 25 (DynamoDB limit)
        for i in range(0, len(message_ids), 25):
            batch = message_ids[i:i+25]
            
            # Create batch request
            request_items = {
                'Messages': {
                    'Keys': [{'MessageId': msg_id} for msg_id in batch]
                }
            }
            
            response = await self.dynamodb.batch_get_item(RequestItems=request_items)
            
            # Process response
            for item in response['Responses']['Messages']:
                messages[item['MessageId']] = item
        
        return messages
    
    def organize_messages_by_chat(self, messages: list):
        """Group messages by chat and sort chronologically"""
        chat_messages = defaultdict(list)
        
        for message in messages:
            chat_id = message['ChatId']
            chat_messages[chat_id].append(message)
        
        # Sort messages within each chat by timestamp
        for chat_id in chat_messages:
            chat_messages[chat_id].sort(key=lambda m: m['SentAt'])
        
        return dict(chat_messages)
    
    async def deliver_chat_messages(self, websocket, chat_id: str, messages: list):
        """Deliver messages for a specific chat"""
        for message in messages:
            await websocket.send_json({
                'type': 'offlineMessage',
                'data': {
                    'messageId': message['MessageId'],
                    'chatId': chat_id,
                    'senderUserId': message['SenderUserId'],
                    'content': message['Content'],
                    'sentAt': message['SentAt'],
                    'attachmentIds': message.get('AttachmentIds', [])
                }
            })
            
            # Update delivery status
            await self.update_delivery_status(message['ClientId'], message['MessageId'])
    
    async def update_delivery_status(self, client_id: str, message_id: str):
        """Update message delivery status in inbox"""
        await self.dynamodb.update_item(
            TableName='Inbox',
            Key={'ClientId': client_id, 'MessageId': message_id},
            UpdateExpression='SET DeliveryStatus = :status',
            ExpressionAttributeValues={':status': 'DELIVERED'}
        )

3) Users should be able to send and receive file attachments (images, documents, videos) within messages

File attachment handling introduces additional complexity around large file uploads, storage management, and efficient delivery mechanisms. Our system must support various file types while maintaining security, performance, and cost-effectiveness.

File Attachment Handling - Upload and Download Architecture

Comprehensive Attachment Handling Flow:

Sending a Message with Attachment

Step 1: Attachment Pre-Upload Process

Before sending the message, User A needs to upload the attachment:

Upload URL Request:

• User A's client sends: {command: "getAttachmentUploadUrl", data: {fileName: "beach.jpg", fileType: "image/jpeg", fileSize: 2048000}}
• Server validates the request (file size, type, etc.)

Attachment Record Creation:

• Server generates unique attachment ID: att-164901234567-e5f6g7h8
• Creates record in Attachments table:
  • AttachmentId (partition key): The generated ID
  • OwnerId: User A's ID
  • FileName: Original filename ("beach.jpg")
  • FileType: MIME type ("image/jpeg")
  • FileSize: Size in bytes (2048000)
  • BlobUrl: null (will be updated after upload)
  • Hash: null (will be updated after upload)
  • UploadedAt: Current timestamp
  • ExpiresAt: Current timestamp + 30 days (TTL)

Pre-signed URL Generation:

• Server generates pre-signed S3 URL with 15-minute expiration
• URL path includes user and attachment identifiers for organization
• Returns to client: {attachmentId: "att-164901234567-e5f6g7h8", uploadUrl: "https://..."}

Step 2: Direct File Upload

User A's client uploads the file directly to blob storage:

Client-side Upload:

• User A's client sends HTTP PUT request to the pre-signed URL
• Uploads file data directly to S3/blob storage
• Calculates file hash (SHA-256) for integrity verification

Upload Completion Notification:

• After successful upload, client notifies server: {command: "attachmentUploaded", data: {attachmentId: "att-164901234567-e5f6g7h8", hash: "2fd4e1c67a2d28fced849ee1bb76e7391b93eb12"}}

Attachment Record Update:

• Server verifies ownership of the attachment
• Updates Attachments table record:
  • Hash: Client-provided hash for verification
  • BlobUrl: Permanent CDN URL for the attachment

Step 3: Sending Message with Attachment Reference

User A sends a message referencing the uploaded attachment:

Message Construction:

• User A's client includes attachment ID in message: {command: "sendMessage", data: {chatId: "chat123", message: "Check out this beach photo!", attachments: ["att-164901234567-e5f6g7h8"]}}

Message Processing:

• Server follows regular message flow (validation, storage, delivery)
• The message in the Messages table includes the attachment ID in the AttachmentIds field
• The full attachment content is not included in the message payload

Message Delivery:

• Server delivers message with attachment reference to User B
• User B's client receives message with attachment ID, but not yet the attachment content

Step 4: Attachment Download Process

When User B wants to view the attachment:

Download URL Request:

• User B's client sends: {command: "getAttachmentDownloadUrl", data: {attachmentId: "att-164901234567-e5f6g7h8"}}
• Server validates User B has access to this attachment (is in the chat)

Attachment Record Retrieval:

• Server queries Attachments table: AttachmentId="att-164901234567-e5f6g7h8"
• Verifies attachment exists and hasn't expired

Download URL Generation:

• Server generates pre-signed S3 download URL with 1-hour expiration
• Returns metadata to client: {attachmentId, fileName, fileType, fileSize, downloadUrl}

Content Download and Display:

• User B's client downloads attachment content directly from blob storage
• Verifies file hash for integrity (optional)
• Displays attachment appropriately based on file type

Step 5: Attachment Lifecycle

The attachment has a defined lifecycle in the system:

Temporary Storage:

• Attachment remains available for 30 days (controlled by TTL)
• After TTL expiration, DynamoDB automatically removes the record

Blob Storage Cleanup:

• S3/blob storage lifecycle policies delete the actual file after 30 days
• This ensures attachments don't consume storage indefinitely

Access Control:

• Only chat participants can access attachment download URLs
• Pre-signed URLs expire quickly to prevent unauthorized access

class AttachmentService:
    def __init__(self, s3_client, dynamodb_client):
        self.s3 = s3_client
        self.dynamodb = dynamodb_client
        self.bucket_name = "messaging-attachments"
    
    async def request_upload_url(self, user_id: str, file_name: str, file_type: str, file_size: int):
        """Generate pre-signed upload URL for attachment"""
        # Validate file constraints
        if not self.validate_file_constraints(file_type, file_size):
            raise ValueError("File type or size not allowed")
        
        # Generate attachment ID
        attachment_id = self.generate_attachment_id()
        
        # Create attachment record
        await self.create_attachment_record(
            attachment_id, user_id, file_name, file_type, file_size
        )
        
        # Generate pre-signed upload URL
        s3_key = f"attachments/{user_id}/{attachment_id}/{file_name}"
        upload_url = await self.s3.generate_presigned_url(
            'put_object',
            Params={
                'Bucket': self.bucket_name,
                'Key': s3_key,
                'ContentType': file_type
            },
            ExpiresIn=900  # 15 minutes
        )
        
        return {
            'attachmentId': attachment_id,
            'uploadUrl': upload_url
        }
    
    async def confirm_upload(self, attachment_id: str, user_id: str, file_hash: str):
        """Confirm successful file upload and update record"""
        # Verify ownership
        attachment = await self.get_attachment(attachment_id)
        if attachment['OwnerId'] != user_id:
            raise PermissionError("Not authorized to update this attachment")
        
        # Generate permanent CDN URL
        cdn_url = f"https://cdn.messaging.com/attachments/{attachment_id}"
        
        # Update attachment record
        await self.dynamodb.update_item(
            TableName='Attachments',
            Key={'AttachmentId': attachment_id},
            UpdateExpression='SET #hash = :hash, BlobUrl = :url',
            ExpressionAttributeNames={'#hash': 'Hash'},
            ExpressionAttributeValues={
                ':hash': file_hash,
                ':url': cdn_url
            }
        )
    
    async def request_download_url(self, attachment_id: str, user_id: str):
        """Generate pre-signed download URL for attachment"""
        # Get attachment details
        attachment = await self.get_attachment(attachment_id)
        
        # Verify user has access (check if user is in any chat with this attachment)
        if not await self.verify_attachment_access(attachment_id, user_id):
            raise PermissionError("Not authorized to access this attachment")
        
        # Generate pre-signed download URL
        s3_key = f"attachments/{attachment['OwnerId']}/{attachment_id}/{attachment['FileName']}"
        download_url = await self.s3.generate_presigned_url(
            'get_object',
            Params={
                'Bucket': self.bucket_name,
                'Key': s3_key
            },
            ExpiresIn=3600  # 1 hour
        )
        
        return {
            'attachmentId': attachment_id,
            'fileName': attachment['FileName'],
            'fileType': attachment['FileType'],
            'fileSize': attachment['FileSize'],
            'downloadUrl': download_url
        }
    
    def validate_file_constraints(self, file_type: str, file_size: int):
        """Validate file type and size constraints"""
        # File size limit: 100MB
        if file_size > 100 * 1024 * 1024:
            return False
        
        # Allowed file types
        allowed_types = [
            'image/jpeg', 'image/png', 'image/gif',
            'application/pdf', 'text/plain',
            'video/mp4', 'video/quicktime'
        ]
        
        return file_type in allowed_types

4) Users should be able to view message delivery status (sent, delivered, read) in real-time

Message delivery status tracking provides users with real-time feedback about their message lifecycle, ensuring transparency and improving user experience. This feature requires sophisticated state management and real-time updates across distributed systems.

Message Delivery Status Tracking - Real-Time Status Updates

Message Status Tracking Flow:

Our system tracks three distinct message states: Sent (message stored in system), Delivered (message received by recipient's device), and Read (message viewed by recipient). Each state transition triggers real-time notifications to the sender.

Status Update Process:

1. Sent Status: Automatically set when message is successfully stored in database
2. Delivered Status: Updated when message reaches recipient's active client device
3. Read Status: Updated when recipient views the message in their chat interface

Real-time Status Broadcasting:

class MessageStatusTracker:
    def __init__(self, redis_client, dynamodb_client):
        self.redis = redis_client
        self.dynamodb = dynamodb_client
    
    async def update_delivery_status(self, message_id: str, client_id: str):
        """Update message delivery status and notify sender"""
        # Update inbox entry
        await self.dynamodb.update_item(
            TableName='Inbox',
            Key={'ClientId': client_id, 'MessageId': message_id},
            UpdateExpression='SET DeliveryStatus = :status',
            ExpressionAttributeValues={':status': 'DELIVERED'}
        )
        
        # Increment delivered count
        await self.dynamodb.update_item(
            TableName='Messages',
            Key={'MessageId': message_id},
            UpdateExpression='ADD DeliveredToCount :inc',
            ExpressionAttributeValues={':inc': 1}
        )
        
        # Notify sender of delivery
        await self.notify_sender_status_change(message_id, 'DELIVERED')
    
    async def update_read_status(self, message_id: str, user_id: str):
        """Update message read status and notify sender"""
        # Update read count
        await self.dynamodb.update_item(
            TableName='Messages',
            Key={'MessageId': message_id},
            UpdateExpression='ADD ReadByCount :inc',
            ExpressionAttributeValues={':inc': 1}
        )
        
        # Update chat participant read pointer
        await self.update_read_pointer(user_id, message_id)
        
        # Notify sender of read receipt
        await self.notify_sender_status_change(message_id, 'READ')

Advanced Technical Deep Dives

Having established our foundational architecture, we can now explore the sophisticated technical challenges that distinguish senior-level system design. The depth of discussion in these areas typically correlates with engineering seniority expectations. The following deep dives represent the most critical architectural decisions for large-scale messaging systems.

1) How can we ensure message ordering and consistency?

Message ordering is critical in messaging systems to maintain conversation coherence and user experience. Our system must guarantee that messages appear in the same order for all participants, even under high concurrency and distributed processing.

Message Ordering Challenges:

• Clock Synchronization: Different servers may have slightly different timestamps
• Network Delays: Messages sent in sequence may arrive out of order
• Distributed Processing: Multiple servers processing messages for the same chat

Solution: Timestamp-Based Message IDs with Sequence Numbers

async def generate_ordered_message_id(chat_id: str):
    """Generate globally ordered message ID"""
    # Get current timestamp in microseconds for precision
    timestamp = int(time.time() * 1000000)
    
    # Get sequence number for this chat to handle concurrent messages
    sequence_key = f"chat:{chat_id}:sequence"
    sequence = await redis.incr(sequence_key)
    
    # Create ordered ID: timestamp + sequence + random component
    random_part = secrets.token_hex(4)
    return f"msg-{timestamp:016d}-{sequence:06d}-{random_part}"

def extract_timestamp_from_message_id(message_id: str) -> int:
    """Extract timestamp for ordering"""
    parts = message_id.split('-')
    return int(parts[1]) if len(parts) >= 3 else 0

2) How can we scale connection management across multiple servers?

Managing millions of concurrent WebSocket connections requires sophisticated load balancing and coordination mechanisms. Our system uses consistent hashing and Redis pub/sub for efficient message routing.

Redis PubSub Channel Architecture

Our Redis channels follow a structured format to enable efficient message routing:

Redis Channel Structure:

• Format: messages:shard:{shardId}
• Limited to manageable number: e.g., 1,000 shards
• Messages include recipient information for targeted delivery:

{
  "recipientId": "userB_id",
  "command": "newMessage",
  "payload": {
    "messageId": "msg-164901234567-a1b2c3d4",
    "chatId": "chat123",
    "senderId": "userA_id",
    "message": "Hello there!",
    "attachments": ["att-164901234567-e5f6g7h8"],
    "timestamp": 164901234567
  }
}

ZooKeeper Coordination Structure

ZooKeeper maintains our distributed system coordination with the following hierarchy:

Server Status Tracking (/servers/{serverId}):

{
  "status": "HEALTHY",
  "lastHeartbeat": 164901234567,
  "connections": 12500,
  "region": "us-east-1"
}

Shard-to-Server Mapping (/shards/{shardId}):

{
  "primary": "server-27",
  "backups": ["server-35", "server-42"]
}

Server Shard Assignments (/servers/{serverId}/shards/{shardId}): Server's specific shard assignments

Detailed Real-Time Connection Flow

Let's trace through a complete user connection scenario:

Step 1: User Connection Initiation

• User client (e.g., mobile app or browser) establishes connection to messaging backend
• Backend computes which shard user belongs to via consistent hashing:

  shardId = hash(user-1) % total_shards  // e.g., shard-123
  

  CopyCopied!

Step 2: Server Determination via ZooKeeper

• Backend consults ZooKeeper to find currently-assigned server for shard-123:

  shard-123 → server-5 (IP: 192.168.10.5)
  

  CopyCopied!
• Backend tells client to establish WebSocket connection to: wss://192.168.10.5

Step 3: WebSocket Connection Establishment

• User client initiates WebSocket handshake: Client(user-1) → Server-5 (WebSocket handshake)
• Server-5 accepts the WebSocket connection

Step 4: Redis PubSub Channel Subscription

• Server-5 subscribes to Redis PubSub shard channel:

  SUBSCRIBE messages:shard:123
  

  CopyCopied!
• However, if another user (e.g., user-2) from same shard (shard-123) previously connected, the subscription already exists. Thus, server-5 does nothing further here.

Step 5: In-Memory Connection Mapping

• Server-5 maintains internal dictionary/hashmap of active WebSocket connections:

  // Internal memory of server-5
  activeUserSockets = {
    "user-1": websocketConnectionObject1,
    "user-2": websocketConnectionObject2,
    ...
  }
  

  CopyCopied!
• Upon user-1 connection: activeUserSockets["user-1"] = websocketConnectionObject1
• This is an O(1) hashmap/dictionary insertion

Step 6: Message Publishing Flow

• When user-9 (from another shard, connected to another server) sends message to user-1
• The server handling user-9 identifies user-1's shard (shard-123), then:

  PUBLISH messages:shard:123 '{"recipientId":"user-1", "message":"Hello!"}'
  

  CopyCopied!

Step 7: Redis Message Delivery

• Redis immediately checks its internal subscriber mapping:

  messages:shard:123 → subscribers [server-5 socket]
  

  CopyCopied!
• Redis immediately pushes this message via TCP socket connection to server-5

Step 8: Server Message Routing

• Server-5 receives payload from Redis:

  {
    "recipientId": "user-1",
    "message": "Hello!"
  }
  

  CopyCopied!
• Server-5 extracts "recipientId": "user-1" and performs O(1) lookup in in-memory dictionary:

  websocketConnectionObject = activeUserSockets["user-1"]
  

  CopyCopied!
• Server immediately pushes message via WebSocket to user-1's browser/mobile client:

  websocketConnectionObject.send("Hello!")
  

  CopyCopied!
• Message instantly appears on user-1's screen

Step 9: User Disconnection Cleanup

• If user-1 disconnects (e.g., closes the app), server-5 simply removes their socket from internal map:

  activeUserSockets.remove("user-1")
  

  CopyCopied!
• If no more users remain connected on server-5 for shard 123, server-5 could unsubscribe from Redis PubSub (optional):

  UNSUBSCRIBE messages:shard:123
  

  CopyCopied!

Connection Sharding Strategy:

The connection sharding strategy is the backbone of our horizontal scaling approach. Each user is deterministically assigned to a specific shard using consistent hashing, ensuring they always connect to the same server (or backup servers) for that shard. This approach provides several key benefits: it enables O(1) user-to-server lookups, ensures efficient Redis subscription patterns, and maintains connection affinity for optimal performance.

The strategy involves calculating the user's shard using a hash function, verifying the server is responsible for that shard, maintaining an in-memory mapping of active connections, and subscribing to only the Redis channels needed for assigned shards. This selective subscription pattern dramatically reduces Redis overhead compared to having every server subscribe to every possible channel.

Connection Manager Implementation:

class ConnectionManager:
    def __init__(self, server_id: str, total_shards: int = 1000):
        self.server_id = server_id
        self.total_shards = total_shards
        self.connected_users = {}  # userId -> websocket mapping
        self.assigned_shards = set()
        self.redis_subscriptions = set()
    
    def get_user_shard(self, user_id: str) -> int:
        """Calculate user's shard using consistent hashing"""
        # Note: In production, use proper consistent hashing algorithms
        hash_value = hashlib.sha256(user_id.encode()).hexdigest()
        return int(hash_value[:8], 16) % self.total_shards
    
    async def register_user_connection(self, user_id: str, websocket):
        """Register user connection on appropriate server"""
        user_shard = self.get_user_shard(user_id)
        
        # Verify this server handles this shard
        if user_shard not in self.assigned_shards:
            raise ValueError(f"User {user_id} should connect to different server")
        
        # Store connection in O(1) hashmap
        self.connected_users[user_id] = {
            'websocket': websocket,
            'connected_at': time.time(),
            'shard': user_shard
        }
        
        # Subscribe to shard channel if not already subscribed
        shard_channel = f"messages:shard:{user_shard}"
        if shard_channel not in self.redis_subscriptions:
            await self.redis.subscribe(shard_channel)
            self.redis_subscriptions.add(shard_channel)
    
    async def handle_user_disconnection(self, user_id: str):
        """Clean up user connection"""
        if user_id in self.connected_users:
            user_shard = self.connected_users[user_id]['shard']
            del self.connected_users[user_id]
            
            # Check if any other users from this shard are still connected
            shard_users = [uid for uid, conn in self.connected_users.items() 
                          if conn['shard'] == user_shard]
            
            # Optionally unsubscribe if no users from this shard remain
            if not shard_users:
                shard_channel = f"messages:shard:{user_shard}"
                await self.redis.unsubscribe(shard_channel)
                self.redis_subscriptions.discard(shard_channel)

3) How can we ensure fault tolerance and message durability?

Message durability is non-negotiable in messaging systems. Our architecture uses multiple layers of redundancy and persistence to guarantee zero message loss.

Multi-Layer Durability Strategy:

Our multi-layer durability approach ensures zero message loss through redundant persistence mechanisms, each optimized for different failure scenarios. This strategy combines the immediate durability of Kafka's distributed log, the long-term reliability of DynamoDB's managed database service, the fast recovery capabilities of Redis caching, and cross-region replication for disaster recovery.

The four layers work in harmony: Kafka provides immediate write-ahead logging with partition-based ordering, DynamoDB handles structured storage with ACID transactions, Redis enables rapid recovery from temporary failures, and cross-region replication protects against regional outages. This redundant approach means that even if multiple layers fail simultaneously, messages remain recoverable from at least one persistence layer.

Each layer serves a specific recovery time objective (RTO) and recovery point objective (RPO). Kafka enables sub-second recovery from server failures, Redis provides instant failover capabilities, DynamoDB ensures long-term data durability, and cross-region replication handles catastrophic regional failures.

Comprehensive Durability Implementation:

class MessageDurabilityManager:
    def __init__(self, kafka_producer, dynamodb_client, redis_client):
        self.kafka = kafka_producer
        self.dynamodb = dynamodb_client
        self.redis = redis_client
    
    async def ensure_message_durability(self, message_data: dict):
        """Implement multi-layer message persistence"""
        message_id = message_data['messageId']
        
        # Layer 1: Immediate Kafka persistence
        await self.kafka.send(
            topic='message-events',
            key=message_data['chatId'],
            value=json.dumps(message_data),
            partition_key=message_data['chatId']
        )
        
        # Layer 2: DynamoDB transaction
        await self.store_message_transaction(message_data)
        
        # Layer 3: Redis backup for fast recovery
        await self.redis.setex(
            f"message_backup:{message_id}",
            3600,  # 1 hour TTL
            json.dumps(message_data)
        )
        
        # Layer 4: Cross-region replication (for critical messages)
        if message_data.get('priority') == 'HIGH':
            await self.replicate_to_backup_region(message_data)

4) How can we optimize for global scale and low latency?

Global messaging requires careful consideration of geographic distribution, CDN usage, and regional data placement to minimize latency while maintaining consistency.

Global Architecture Strategy:

The global architecture strategy addresses latency optimization and user experience across diverse geographic regions. Our approach deploys complete infrastructure stacks in multiple regions while implementing intelligent user routing based on geographic proximity and network conditions. This strategy minimizes latency for the majority of interactions while maintaining global connectivity for cross-region communication.

The architecture balances two competing objectives: minimizing latency through regional affinity and ensuring seamless communication between users in different regions. We achieve this through smart user placement, regional data replication for frequently accessed content, and efficient cross-region synchronization protocols. Each region operates independently for local users while participating in a global mesh for inter-region communication.

User routing decisions consider multiple factors: geographic distance, network latency, regional server capacity, and data residency requirements. The system continuously monitors performance metrics to adjust routing algorithms and ensure optimal user experience. For cross-region messages, we use asynchronous replication to maintain consistency without blocking real-time delivery.

Global Distribution Implementation:

class GlobalMessagingArchitecture:
    def __init__(self):
        self.regions = {
            'us-east-1': {'primary': True, 'capacity': '40%'},
            'eu-west-1': {'primary': False, 'capacity': '25%'},
            'ap-southeast-1': {'primary': False, 'capacity': '20%'},
            'us-west-2': {'primary': False, 'capacity': '15%'}
        }
    
    def route_user_to_region(self, user_location: dict) -> str:
        """Route user to optimal region based on location"""
        # Calculate latency to each region
        region_latencies = {}
        for region, config in self.regions.items():
            latency = self.calculate_latency(user_location, region)
            region_latencies[region] = latency
        
        # Select region with lowest latency and available capacity
        return min(region_latencies.keys(), key=lambda r: region_latencies[r])
    
    async def cross_region_message_sync(self, message_data: dict):
        """Synchronize messages across regions for global users"""
        # Identify users in different regions for this chat
        chat_participants = await self.get_chat_participants(message_data['chatId'])
        
        # Group participants by region
        regional_participants = defaultdict(list)
        for participant in chat_participants:
            user_region = await self.get_user_region(participant['userId'])
            regional_participants[user_region].append(participant)
        
        # Send message to each region
        for region, participants in regional_participants.items():
            await self.send_to_region(region, message_data, participants)

Conclusion

Designing Amazon's Real-Time Messaging System presents a comprehensive exploration of distributed systems challenges that are fundamental to modern communication platforms. The system must balance competing requirements: ensuring message ordering and consistency while achieving global scale, providing real-time delivery while guaranteeing durability, and managing millions of concurrent connections while maintaining sub-second latency.

Here is the final system architecture diagram:

Final Complete Amazon Real-Time Messaging System Architecture

Key Technical Insights:

1. Message Ordering and Consistency: The progression from simple timestamps to sequence-based ordering with Redis coordination demonstrates how distributed systems achieve consistency across multiple servers and time zones.

2. Connection Management at Scale: Using consistent hashing for user-to-server mapping combined with Redis pub/sub for message routing showcases how to efficiently manage millions of concurrent WebSocket connections.

3. Offline Message Delivery: The sophisticated inbox system with TTL-based cleanup and batch synchronization illustrates how to guarantee message delivery regardless of user connectivity patterns.

4. Attachment Handling: The separation of file storage from message processing, using pre-signed URLs and CDN delivery, demonstrates how to handle large files efficiently while maintaining security.

Scalability Strategies:

• Horizontal Sharding: User-based sharding enables linear scaling of connection management and message processing
• Multi-Layer Persistence: Kafka, DynamoDB, and Redis provide redundant durability guarantees
• Geographic Distribution: Regional deployment with cross-region synchronization minimizes global latency
• Asynchronous Processing: Message queues decouple real-time delivery from persistence operations

To Succeed in the Interview:

• Start with User Flows: Begin with concrete user scenarios (send message, offline sync, attachments) before diving into technical details
• Show Progressive Complexity: Demonstrate simple solutions first, then add sophistication for scale and reliability
• Focus on Critical Paths: Identify the most challenging aspects (message ordering, connection management) and spend appropriate time on them
• Consider Operational Concerns: Address monitoring, fault tolerance, and real-world deployment challenges
• Justify Technical Decisions: Explain why specific technologies (DynamoDB, Redis, Kafka) were chosen for each use case

This system design question effectively evaluates a candidate's ability to reason about real-time distributed systems, handle consistency challenges in messaging, design for global scale, and balance competing technical requirements - all essential skills for senior engineering roles at companies like Amazon.

This comprehensive system design question evaluates understanding of real-time messaging systems, distributed consistency patterns, connection management at scale, and the architectural complexity required for Amazon-scale infrastructure.

It has appeared frequently in recent Amazon interviews for L5 (mid-level) roles.