WhatsApp / Chat System

Functional Requirements

Non-Functional Requirements

Capacity Estimation

Traffic:
  MAU: 2 billion
  DAU: ~1.5 billion
  Messages/day: 100 billion = ~1.16 million msgs/sec
  Peak: 50 million msgs/sec (during holidays/events)
  Avg message size: 100 bytes text + 100KB media (10% of messages)

Storage (per day):
  Text: 100B msgs × 100 bytes = 10 TB/day
  Media: 10B msgs × 100KB = 1 PB/day (most in S3)
  With 3x replication: 30 TB text, 3 PB media

Connections:
  Concurrent connections: ~500 million
  WebSocket memory: 500M × 10KB = 5 TB RAM for connections

Bandwidth:
  50M msgs/sec × 100 bytes = 5 GB/sec (text only)
  With media: 50 GB/sec peak

Architecture Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                         WHATSAPP CHAT ARCHITECTURE                          │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ┌──────────┐                                                              │
│  │ Sender   │                                                              │
│  │ Device   │                                                              │
│  └────┬─────┘                                                              │
│       │ WebSocket                                                           │
│       ▼                                                                     │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐                  │
│  │ Connection   │───▶│ Chat         │───▶│ Message      │                  │
│  │ Server       │    │ Service      │    │ Queue        │                  │
│  │ (WS Gateway) │    │              │    │ (Kafka)      │                  │
│  └──────────────┘    └──────┬───────┘    └──────┬───────┘                  │
│                             │                    │                          │
│                             ▼                    ▼                          │
│                      ┌──────────────┐    ┌──────────────┐                  │
│                      │ Storage      │    │ Push         │                  │
│                      │ Service      │    │ Notification │                  │
│                      │ (Cassandra)  │    │ Service      │                  │
│                      └──────────────┘    └──────┬───────┘                  │
│                                                  │                          │
│                                                  ▼                          │
│                                          ┌──────────────┐                  │
│                                          │ APNS / FCM   │                  │
│                                          └──────────────┘                  │
│                                                                             │
│  ┌──────────────┐                                                          │
│  │ Receiver     │◀──── Connection Server (if online)                        │
│  │ Device       │◀──── Push Notification (if offline)                       │
│  └──────────────┘                                                          │
└─────────────────────────────────────────────────────────────────────────────┘

Connection Management

class ConnectionManager:
    """Manages WebSocket connections from clients."""

    def __init__(self):
        self.connections: Dict[str, WebSocket] = {}  # user_id → WebSocket
        self.user_servers: Dict[str, str] = {}       # user_id → server_id
        self.redis = RedisCluster()

    async def handle_connection(self, websocket: WebSocket, user_id: str):
        """Handle new WebSocket connection."""
        # Register connection
        self.connections[user_id] = websocket
        self.user_servers[user_id] = self.server_id

        # Register in Redis for routing
        self.redis.set(
            f"connection:{user_id}",
            json.dumps({
                'server_id': self.server_id,
                'connected_at': time.time()
            }),
            ex=300  # 5 min TTL, refreshed by heartbeat
        )

        # Update presence
        await self.update_presence(user_id, 'online')

        # Start heartbeat
        asyncio.create_task(self.heartbeat_loop(user_id, websocket))

        # Deliver queued messages
        await self.deliver_queued_messages(user_id)

    async def heartbeat_loop(self, user_id: str, websocket: WebSocket):
        """Send periodic heartbeats to detect disconnections."""
        while True:
            try:
                await websocket.send_json({'type': 'heartbeat', 'ts': time.time()})
                # Refresh Redis TTL
                self.redis.expire(f"connection:{user_id}", 300)
                await asyncio.sleep(60)  # Every 60 seconds
            except Exception:
                await self.handle_disconnect(user_id)
                break

    async def handle_disconnect(self, user_id: str):
        """Handle user disconnection."""
        self.connections.pop(user_id, None)
        self.user_servers.pop(user_id, None)
        self.redis.delete(f"connection:{user_id}")
        await self.update_presence(user_id, 'offline')

Message Flow

class ChatService:
    """Core message handling service."""

    def __init__(self):
        self.connection_manager = ConnectionManager()
        self.message_store = MessageStore()
        self.message_queue = KafkaProducer()
        self.notification_service = NotificationService()

    async def send_message(self, sender_id: str, message: Message) -> str:
        """Send a message from sender to receiver(s)."""
        # Step 1: Validate sender
        if not await self.validate_sender(sender_id):
            raise UnauthorizedError("Invalid sender")

        # Step 2: Generate message ID and sequence number
        message_id = str(uuid.uuid4())
        sequence = await self.get_next_sequence(
            message.conversation_id
        )

        # Step 3: Create message record
        msg_record = {
            'message_id': message_id,
            'sender_id': sender_id,
            'conversation_id': message.conversation_id,
            'content': message.content,
            'content_type': message.content_type,
            'sequence': sequence,
            'timestamp': time.time(),
            'status': 'sent'
        }

        # Step 4: Persist to storage (async)
        await self.message_store.save(msg_record)

        # Step 5: Route to recipient(s)
        recipients = await self.get_recipients(
            message.conversation_id, sender_id
        )

        for recipient_id in recipients:
            # Check if recipient is online
            connection_info = self.connection_manager.get_connection(
                recipient_id
            )

            if connection_info:
                # Online: deliver directly via WebSocket
                await self.deliver_message(recipient_id, msg_record)
            else:
                # Offline: queue for delivery + push notification
                await self.message_queue.send(
                    topic='offline_messages',
                    key=recipient_id,
                    value=json.dumps(msg_record)
                )
                await self.notification_service.send_push(
                    recipient_id,
                    msg_record
                )

        # Step 6: Send delivery acknowledgment to sender
        return message_id

    async def deliver_message(self, user_id: str, message: dict):
        """Deliver message to an online user."""
        connection = self.connection_manager.get_connection(user_id)
        if connection:
            await connection.send_json({
                'type': 'message',
                'data': message
            })
            # Update status to 'delivered'
            await self.message_store.update_status(
                message['message_id'], 'delivered'
            )

Message Storage (Cassandra)

-- Messages table: partition by conversation_id, cluster by timestamp
CREATE TABLE messages (
    conversation_id UUID,
    message_id UUID,
    sender_id UUID,
    content TEXT,
    content_type TEXT,  -- 'text', 'image', 'video', 'document'
    sequence BIGINT,
    timestamp TIMESTAMP,
    status TEXT,        -- 'sent', 'delivered', 'read'
    PRIMARY KEY (conversation_id, timestamp, message_id)
) WITH CLUSTERING ORDER BY (timestamp DESC)
  AND default_time_to_live = 31536000;  -- 1 year retention

-- Conversation index: track user's conversations
CREATE TABLE user_conversations (
    user_id UUID,
    conversation_id UUID,
    last_message_time TIMESTAMP,
    last_message_preview TEXT,
    unread_count INT,
    PRIMARY KEY (user_id, last_message_time, conversation_id)
) WITH CLUSTERING ORDER BY (last_message_time DESC);

-- Message status tracking
CREATE TABLE message_status (
    message_id UUID,
    user_id UUID,
    status TEXT,       -- 'sent', 'delivered', 'read'
    status_time TIMESTAMP,
    PRIMARY KEY (message_id, user_id)
);

Message Queue (Kafka)

# Kafka configuration for offline message delivery
KAFKA_CONFIG = {
    'bootstrap_servers': ['kafka-1:9092', 'kafka-2:9092', 'kafka-3:9092'],
    'topic': 'offline_messages',
    'partitions': 128,  # Parallelism for offline delivery
    'replication_factor': 3,
    'retention_hours': 24,  # Keep offline messages for 24h
    'consumer_groups': {
        'offline_delivery': {
            'group_id': 'offline-delivery-workers',
            'auto_offset_reset': 'latest'
        }
    }
}

# Worker processes offline messages
class OfflineMessageWorker:
    def process(self, message):
        data = json.loads(message.value)
        recipient_id = data['recipient_id']

        # Check if user is now online
        connection = self.connection_manager.get_connection(recipient_id)
        if connection:
            # User came back online — deliver directly
            await connection.send_json({
                'type': 'message',
                'data': data
            })
        else:
            # Still offline — keep in queue for later
            # Message already persisted to Cassandra
            pass

WebSocket Protocol

┌─────────────────────────────────────────────────────────────────────┐
│                    WEBSOCKET COMMUNICATION FLOW                     │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  Client                          Server                            │
│    │                               │                               │
    │──── HTTP Upgrade ─────────────▶│                               │
    │    GET /ws?token=xxx           │                               │
    │                               │                               │
    │◀──── 101 Switching Protocols ──│                               │
    │                               │                               │
    │◀══════ WebSocket Connection ═══│                               │
    │                               │                               │
    │──── { type: "auth" } ────────▶│  Authenticate                 │
    │◀─── { type: "auth_ok" } ──────│                               │
    │                               │                               │
    │──── { type: "message", } ────▶│  Send message                 │
    │    { content: "Hello" }       │                               │
    │                               │───▶ Route to recipient        │
    │                               │                               │
    │◀─── { type: "delivery_ack" } ─│  Acknowledge delivery         │
    │                               │                               │
    │◀══════ { type: "message" } ════│  Receive message              │
    │    { content: "Hi!" }         │                               │
    │                               │                               │
    │──── { type: "read_ack" } ────▶│  Mark as read                 │
    │                               │                               │
    │──── { type: "typing" } ──────▶│  Typing indicator             │
    │◀══════ { type: "typing" } ═════│  Other user typing            │
    │                               │                               │
    │──── { type: "heartbeat" } ───▶│  Keep alive                   │
    │◀─── { type: "heartbeat_ack" } │                               │

Message Types

type WSMessage =
  | { type: "auth"; token: string; device_id: string }
  | { type: "auth_ok"; user_id: string; server_time: number }
  | { type: "message"; conversation_id: string; content: string; content_type: string; client_msg_id: string }
  | { type: "message_ack"; message_id: string; status: "sent" | "delivered" | "read" }
  | { type: "delivery_ack"; message_id: string; server_msg_id: string; timestamp: number }
  | { type: "typing"; conversation_id: string; is_typing: boolean }
  | { type: "read_receipt"; conversation_id: string; last_read_sequence: number }
  | { type: "presence"; user_id: string; status: "online" | "offline"; last_seen: number }
  | { type: "heartbeat"; ts: number }
  | { type: "heartbeat_ack"; ts: number }
  | { type: "error"; code: string; message: string }

Message Ordering with Sequence Numbers

class MessageSequencer:
    """Ensures strict per-conversation message ordering."""

    def __init__(self):
        self.redis = RedisCluster()

    async def get_next_sequence(self, conversation_id: str) -> int:
        """Get next sequence number using Redis atomic increment."""
        key = f"seq:{conversation_id}"
        # Atomic increment — guaranteed unique even with concurrent requests
        sequence = self.redis.incr(key)
        # Set TTL in case conversation becomes inactive
        self.redis.expire(key, 86400 * 30)  # 30 days
        return sequence

    def get_client_sequence(self, client_msg_id: str) -> int:
        """Map client-side message ID to server sequence."""
        # Client generates message_id locally before network
        # Server maps it to server-side sequence on receipt
        mapping = self.redis.get(f"client_seq:{client_msg_id}")
        if mapping:
            return int(mapping)
        return None

Client-Side Optimistic UI

class ChatClient {
    sendMessage(conversationId, content, contentType) {
        // Generate client message ID before sending
        const clientMsgId = generateUUID();

        // Add to UI immediately (optimistic)
        this.ui.addMessage({
            id: clientMsgId,
            status: 'sending',
            content: content,
            timestamp: Date.now()
        });

        // Send via WebSocket
        this.ws.send({
            type: 'message',
            conversation_id: conversationId,
            content: content,
            content_type: contentType,
            client_msg_id: clientMsgId
        });

        // Handle acknowledgment
        this.on('delivery_ack', (msg) => {
            if (msg.client_msg_id === clientMsgId) {
                this.ui.updateMessageStatus(clientMsgId, 'sent');
                this.ui.updateMessageId(clientMsgId, msg.server_msg_id);
            }
        });

        // Handle failure
        this.on('error', (err) => {
            if (err.client_msg_id === clientMsgId) {
                this.ui.updateMessageStatus(clientMsgId, 'failed');
                // Offer retry
            }
        });
    }
}

Retry & Deduplication

class MessageDeduplicator:
    """Prevent duplicate messages from retries."""

    def __init__(self):
        self.redis = RedisCluster()
        self.DEDUP_TTL = 86400  # 24 hours

    def is_duplicate(self, client_msg_id: str) -> bool:
        """Check if message was already processed."""
        key = f"dedup:{client_msg_id}"
        # SETNX returns True if key was set (not duplicate)
        result = self.redis.setnx(key, '1')
        if result:
            self.redis.expire(key, self.DEDUP_TTL)
            return False  # Not a duplicate
        return True  # Duplicate — skip processing

    def get_server_message_id(self, client_msg_id: str) -> str:
        """Get server-assigned message ID for client's message."""
        return self.redis.get(f"server_msg:{client_msg_id}")

Presence System

┌─────────────────────────────────────────────────────────────────────┐
│                       PRESENCE ARCHITECTURE                         │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌──────────┐    ┌──────────────┐    ┌──────────────┐             │
│  │ Client   │───▶│ Connection   │───▶│ Presence     │             │
│  │          │    │ Server       │    │ Service      │             │
│  └──────────┘    └──────────────┘    └──────┬───────┘             │
│                                              │                      │
│                                              ▼                      │
│                                       ┌──────────────┐             │
│                                       │ Redis        │             │
│                                       │ (Online      │             │
│                                       │  Status)     │             │
│                                       └──────┬───────┘             │
│                                              │                      │
│                                              ▼                      │
│                                       ┌──────────────┐             │
│                                       │ Push to      │             │
│                                       │ Subscribers  │             │
│                                       └──────────────┘             │
└─────────────────────────────────────────────────────────────────────┘

Presence Service Implementation

class PresenceService:
    """Track user online/offline status."""

    def __init__(self):
        self.redis = RedisCluster()
        self.ONLINE_TTL = 300       # 5 minutes — must heartbeat within this
        self.LAST_SEEN_TTL = 86400  # 24 hours for last_seen

    async def user_online(self, user_id: str):
        """Mark user as online."""
        pipe = self.redis.pipeline()

        # Set online status with TTL
        pipe.setex(f"presence:{user_id}", self.ONLINE_TTL, 'online')

        # Record last seen time
        pipe.setex(
            f"last_seen:{user_id}",
            self.LAST_SEEN_TTL,
            str(time.time())
        )

        # Notify subscribers (contacts who care about this user's status)
        subscribers = self.get_presence_subscribers(user_id)
        for sub_id in subscribers:
            pipe.publish(
                f"presence:{sub_id}",
                json.dumps({
                    'user_id': user_id,
                    'status': 'online',
                    'last_seen': time.time()
                })
            )

        pipe.execute()

    async def user_offline(self, user_id: str):
        """Mark user as offline."""
        pipe = self.redis.pipeline()

        # Remove online status
        pipe.delete(f"presence:{user_id}")

        # Update last seen
        pipe.setex(
            f"last_seen:{user_id}",
            self.LAST_SEEN_TTL,
            str(time.time())
        )

        # Notify subscribers
        subscribers = self.get_presence_subscribers(user_id)
        for sub_id in subscribers:
            pipe.publish(
                f"presence:{sub_id}",
                json.dumps({
                    'user_id': user_id,
                    'status': 'offline',
                    'last_seen': time.time()
                })
            )

        pipe.execute()

    async def get_presence(self, user_id: str) -> dict:
        """Get current presence status."""
        status = self.redis.get(f"presence:{user_id}")
        last_seen = self.redis.get(f"last_seen:{user_id}")

        return {
            'user_id': user_id,
            'status': 'online' if status else 'offline',
            'last_seen': float(last_seen) if last_seen else None
        }

Read Receipts System

class ReadReceiptService:
    """Handle message read status (sent → delivered → read)."""

    def __init__(self):
        self.cassandra = CassandraClient()
        self.redis = RedisCluster()

    async def mark_delivered(self, message_id: str, recipient_id: str):
        """Mark message as delivered (received by recipient's device)."""
        self.cassandra.execute("""
            INSERT INTO message_status (message_id, user_id, status, status_time)
            VALUES (%s, %s, 'delivered', toTimestamp(now()))
        """, (message_id, recipient_id))

        # Notify sender of delivery
        sender_id = self.get_message_sender(message_id)
        await self.notify_status(sender_id, message_id, 'delivered')

    async def mark_read(self, conversation_id: str, reader_id: str, last_sequence: int):
        """Mark all messages up to last_sequence as read."""
        # Get all unread messages in conversation up to sequence
        messages = self.cassandra.execute("""
            SELECT message_id FROM messages
            WHERE conversation_id = %s
            AND sequence <= %s
        """, (conversation_id, last_sequence))

        # Batch update status
        batch = []
        for msg in messages:
            batch.append((msg.message_id, reader_id, 'read'))

        # Batch insert for efficiency
        self.cassandra.execute("""
            INSERT INTO message_status (message_id, user_id, status, status_time)
            VALUES (%s, %s, 'read', toTimestamp(now()))
        """, batch)

        # Update unread count
        self.redis.hset(
            f"unread:{conversation_id}",
            reader_id,
            0
        )

        # Notify conversation participants
        participants = self.get_conversation_participants(conversation_id)
        for participant_id in participants:
            if participant_id != reader_id:
                await self.notify_read_receipt(
                    participant_id,
                    conversation_id,
                    reader_id,
                    last_sequence
                )

    async def get_message_status(self, message_id: str) -> dict:
        """Get delivery/read status for a message."""
        statuses = self.cassandra.execute("""
            SELECT user_id, status, status_time
            FROM message_status
            WHERE message_id = %s
        """, (message_id,))

        result = {
            'sent': True,
            'delivered_to': [],
            'read_by': []
        }

        for row in statuses:
            if row.status == 'delivered':
                result['delivered_to'].append({
                    'user_id': str(row.user_id),
                    'time': row.status_time
                })
            elif row.status == 'read':
                result['read_by'].append({
                    'user_id': str(row.user_id),
                    'time': row.status_time
                })

        return result

Group Messaging

class GroupMessagingService:
    """Handle group chat with fanout delivery."""

    def __init__(self):
        self.cassandra = CassandraClient()
        self.connection_manager = ConnectionManager()
        self.message_store = MessageStore()

    async def send_group_message(
        self,
        sender_id: str,
        group_id: str,
        content: str,
        content_type: str
    ) -> str:
        """Send message to a group."""
        # Step 1: Validate sender is in group
        if not await self.is_group_member(group_id, sender_id):
            raise UnauthorizedError("Not a group member")

        # Step 2: Generate message
        message_id = str(uuid.uuid4())
        sequence = await self.get_next_sequence(f"group:{group_id}")

        msg_record = {
            'message_id': message_id,
            'sender_id': sender_id,
            'conversation_id': group_id,
            'is_group': True,
            'content': content,
            'content_type': content_type,
            'sequence': sequence,
            'timestamp': time.time()
        }

        # Step 3: Persist
        await self.message_store.save(msg_record)

        # Step 4: Fanout to all group members
        members = await self.get_group_members(group_id)

        for member_id in members:
            if member_id == sender_id:
                continue  # Don't send to sender

            connection = self.connection_manager.get_connection(member_id)
            if connection:
                await connection.send_json({
                    'type': 'group_message',
                    'data': msg_record
                })
            else:
                # Queue + push notification
                await self.queue_offline_message(member_id, msg_record)

        return message_id

    async def get_group_members(self, group_id: str) -> List[str]:
        """Get all members of a group."""
        # Cache in Redis for frequent access
        cached = self.redis.smembers(f"group_members:{group_id}")
        if cached:
            return list(cached)

        # Fetch from database
        members = self.cassandra.execute("""
            SELECT user_id FROM group_members
            WHERE group_id = %s
        """, (group_id,))

        member_ids = [str(m.user_id) for m in members]

        # Cache for next time
        if member_ids:
            pipe = self.redis.pipeline()
            pipe.sadd(f"group_members:{group_id}", *member_ids)
            pipe.expire(f"group_members:{group_id}", 3600)
            pipe.execute()

        return member_ids

End-to-End Encryption (Signal Protocol)

class E2EEncryption:
    """Signal Protocol implementation for E2E encryption."""

    def __init__(self):
        self.key_store = KeyStore()

    async def register_keys(self, user_id: str):
        """Generate and upload identity key + signed pre-key."""
        # Generate identity key pair (long-term)
        identity_key = SignalProtocol.generate_identity_key_pair()

        # Generate signed pre-key (rotated periodically)
        signed_pre_key = SignalProtocol.generate_signed_pre_key(
            identity_key, key_id=1
        )

        # Generate one-time pre-keys (for initial key exchange)
        one_time_pre_keys = [
            SignalProtocol.generate_pre_key(identity_key, i)
            for i in range(100)
        ]

        # Upload to server
        await self.key_store.upload_keys(
            user_id=user_id,
            identity_key=identity_key.public,
            signed_pre_key=signed_pre_key,
            one_time_pre_keys=[k.public for k in one_time_pre_keys]
        )

    async def establish_session(self, sender_id: str, recipient_id: str):
        """Establish encrypted session with recipient."""
        # Fetch recipient's keys from server
        recipient_keys = await self.key_store.get_keys(recipient_id)

        # Perform X3DH key agreement
        session = SignalProtocol.create_session(
            our_identity_key=await self.key_store.get_identity(sender_id),
            our_signed_pre_key=await self.key_store.get_signed_pre_key(sender_id),
            their_identity_key=recipient_keys['identity_key'],
            their_signed_pre_key=recipient_keys['signed_pre_key'],
            their_one_time_pre_key=recipient_keys['one_time_pre_keys'][0]
        )

        return session

    def encrypt_message(self, session, plaintext: str) -> bytes:
        """Encrypt message using Double Ratchet algorithm."""
        return SignalProtocol.encrypt(session, plaintext.encode('utf-8'))

    def decrypt_message(self, session, ciphertext: bytes) -> str:
        """Decrypt received message."""
        return SignalProtocol.decrypt(session, ciphertext).decode('utf-8')

Encryption Flow

┌─────────────────────────────────────────────────────────────────────┐
│                    E2E ENCRYPTION FLOW                              │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  Alice                           Server                          Bob │
│    │                               │                               │
│    │─── Upload Identity Key ──────▶│                               │
│    │─── Upload Signed Pre-Key ────▶│                               │
│    │─── Upload One-Time Pre-Keys ─▶│                               │
│    │                               │                               │
│    │                               │◀─── Request Bob's Keys ───────│
│    │                               │──── Return Bob's Keys ───────▶│
│    │                               │                               │
│    │  X3DH Key Agreement          │  X3DH Key Agreement           │
│    │  (generates shared secret)    │  (generates shared secret)     │
│    │                               │                               │
│    │  Double Ratchet              │  Double Ratchet               │
│    │  (derives per-message keys)  │  (derives per-message keys)   │
│    │                               │                               │
│    │──── Encrypted Message ──────▶│──── Forward (encrypted) ─────▶│
│    │    (server can't read)       │    (server can't read)        │
│    │                               │                               │
│    │                               │◀──── Encrypted Reply ─────────│
│    │◀─── Encrypted Reply ─────────│                               │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Typing Indicators

class TypingIndicator:
    """Handle typing indicators with debouncing."""

    def __init__(self):
        self.redis = RedisCluster()
        self.connection_manager = ConnectionManager()
        self.TYPING_TTL = 10  # Auto-expire after 10 seconds

    async def start_typing(self, user_id: str, conversation_id: str):
        """User started typing."""
        # Store in Redis with short TTL
        self.redis.setex(
            f"typing:{conversation_id}:{user_id}",
            self.TYPING_TTL,
            '1'
        )

        # Notify other participants
        participants = self.get_participants(conversation_id)
        for participant_id in participants:
            if participant_id != user_id:
                connection = self.connection_manager.get_connection(
                    participant_id
                )
                if connection:
                    await connection.send_json({
                        'type': 'typing',
                        'conversation_id': conversation_id,
                        'user_id': user_id,
                        'is_typing': True
                    })

    async def stop_typing(self, user_id: str, conversation_id: str):
        """User stopped typing."""
        self.redis.delete(f"typing:{conversation_id}:{user_id}")

        # Notify to hide typing indicator
        participants = self.get_participants(conversation_id)
        for participant_id in participants:
            if participant_id != user_id:
                connection = self.connection_manager.get_connection(
                    participant_id
                )
                if connection:
                    await connection.send_json({
                        'type': 'typing',
                        'conversation_id': conversation_id,
                        'user_id': user_id,
                        'is_typing': False
                    })

System Architecture Summary

┌─────────────────────────────────────────────────────────────────────────────┐
│                          WHATSAPP ARCHITECTURE                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ┌──────────┐                                                              │
│  │  Mobile  │──┐                                                           │
│  │  Client  │  │    ┌──────────────┐                                      │
│  └──────────┘  ├───▶│  Load        │    ┌──────────────┐                  │
│                │    │  Balancer    │───▶│  Connection  │  (WebSocket       │
│  ┌──────────┐  │    └──────────────┘    │  Servers     │   Gateway)        │
│  │  Web     │──┘                        └──────┬───────┘                  │
│  │  Client  │                                  │                           │
│  └──────────┘                    ┌─────────────┼─────────────┐            │
│                                  │             │             │            │
│                                  ▼             ▼             ▼            │
│                           ┌───────────┐ ┌───────────┐ ┌───────────┐     │
│                           │   Chat    │ │ Presence  │ │  Read     │     │
│                           │  Service  │ │  Service  │ │  Receipt  │     │
│                           └─────┬─────┘ └─────┬─────┘ │  Service  │     │
│                                 │             │       └─────┬─────┘     │
│                                 ▼             ▼             ▼           │
│                          ┌───────────┐ ┌───────────┐ ┌───────────┐     │
│                          │  Kafka    │ │  Redis    │ │Cassandra  │     │
│                          │ (Message  │ │ (Online   │ │(Messages  │     │
│                          │  Queue)   │ │  Status)  │ │ & Status) │     │
│                          └─────┬─────┘ └───────────┘ └───────────┘     │
│                                │                                        │
│                          ┌─────▼─────┐                                  │
│                          │  Offline  │                                  │
│                          │  Workers  │                                  │
│                          └─────┬─────┘                                  │
│                                │                                        │
│                          ┌─────▼─────┐                                  │
│                          │ Push      │                                  │
│                          │ (APNS/FCM)│                                  │
│                          └───────────┘                                  │
│                                                                         │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                    E2E ENCRYPTION LAYER                         │   │
│  │  - Signal Protocol (X3DH + Double Ratchet)                      │   │
│  │  - Server never sees plaintext messages                         │   │
│  │  - Keys managed client-side, uploaded to server for exchange     │   │
│  └─────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────────┘