WhatsApp Messaging.

# Pseudocode: How WhatsApp structures connection handling
# (Actual WhatsApp code is Erlang, this shows the concept)

"""
WhatsApp Connection Handler

Each user connection runs as a single Erlang process.
This process:
1. Owns the TCP socket
2. Maintains user session state
3. Routes messages to/from the user
4. Dies cleanly when user disconnects
"""

class UserConnectionProcess:
    """
    Conceptual model of WhatsApp's per-user Erlang process.
    
    In Erlang, this is implemented with gen_server behavior.
    Memory: a few KB per process at spawn
    Lifecycle: Lives as long as user is connected
    """
    
    def __init__(self, socket, user_id):
        self.socket = socket
        self.user_id = user_id
        self.session_key = None
        self.message_queue = []  # For messages while processing
        self.last_seen = now()
        
        # Register this process in the global user registry
        UserRegistry.register(user_id, self.pid)
    
    def handle_message(self, msg):
        """
        Main message handler — called for every incoming message.
        
        Erlang processes have a mailbox — messages queue up
        and are processed one at a time.
        """
        match msg:
            case AuthRequest(credentials):
                self.handle_auth(credentials)
            
            case OutgoingMessage(recipient, encrypted_content):
                self.route_to_recipient(recipient, encrypted_content)
            
            case IncomingMessage(sender, encrypted_content):
                self.deliver_to_socket(sender, encrypted_content)
            
            case PresenceUpdate(status):
                self.broadcast_presence(status)
            
            case Disconnect():
                self.cleanup_and_die()
    
    def route_to_recipient(self, recipient_id, content):
        """
        Route message to another user.
        
        This is the core of WhatsApp's routing:
        1. Look up recipient's process (which server?)
        2. Send message directly to that process
        """
        # Find recipient's process (could be on any server)
        recipient_process = UserRegistry.lookup(recipient_id)
        
        if recipient_process:
            # Direct process-to-process communication
            # Works across servers in Erlang!
            send(recipient_process, IncomingMessage(
                sender=self.user_id,
                content=content
            ))
        else:
            # User offline — store for later
            OfflineStore.queue(recipient_id, content)
        
        # Server has the message: single tick (SENT). DELIVERED is
        # sent only when the recipient's DEVICE acks receipt.
        self.send_ack(MessageStatus.SENT)
    
    def cleanup_and_die(self):
        """
        Clean shutdown when user disconnects.
        
        In Erlang, when a process dies, all its memory
        is automatically reclaimed. No per-connection cleanup code needed.
        """
        UserRegistry.unregister(self.user_id)
        close(self.socket)
        # Process simply terminates — Erlang handles cleanup

# encryption/signal_protocol.py

"""
Simplified illustration of Signal Protocol concepts.

The actual implementation uses:
- Curve25519 for key exchange
- AES-256-CBC for encryption
- HMAC-SHA256 for authentication
"""

from dataclasses import dataclass
from typing import Optional, Tuple
import os
import hashlib
import hmac


@dataclass
class KeyPair:
    """A public/private key pair."""
    public_key: bytes
    private_key: bytes


@dataclass
class SessionState:
    """State for an encrypted session with another user."""
    their_identity_key: bytes
    their_signed_prekey: bytes
    
    # Ratchet state
    root_key: bytes
    chain_key: bytes
    message_number: int = 0


class SignalSession:
    """
    Simplified Signal Protocol session.
    
    This illustrates the key concepts — actual implementation
    is more complex with proper cryptographic primitives.
    """
    
    def __init__(
        self,
        my_identity: KeyPair,
        their_identity_public: bytes,
        their_signed_prekey: bytes,
        their_one_time_prekey: Optional[bytes] = None
    ):
        self.my_identity = my_identity
        
        # Perform X3DH key agreement
        self.session = self._x3dh_key_agreement(
            their_identity_public,
            their_signed_prekey,
            their_one_time_prekey
        )
    
    def _x3dh_key_agreement(
        self,
        their_identity: bytes,
        their_signed_prekey: bytes,
        their_one_time_prekey: Optional[bytes]
    ) -> SessionState:
        """
        X3DH: Extended Triple Diffie-Hellman.
        
        Creates a shared secret from multiple DH exchanges.
        """
        # Generate ephemeral key for this session
        ephemeral = self._generate_keypair()
        
        # Perform multiple DH exchanges
        dh1 = self._dh(self.my_identity.private_key, their_signed_prekey)
        dh2 = self._dh(ephemeral.private_key, their_identity)
        dh3 = self._dh(ephemeral.private_key, their_signed_prekey)
        
        if their_one_time_prekey:
            dh4 = self._dh(ephemeral.private_key, their_one_time_prekey)
            shared_secret = self._kdf(dh1 + dh2 + dh3 + dh4)
        else:
            shared_secret = self._kdf(dh1 + dh2 + dh3)
        
        # Derive initial root and chain keys
        root_key, chain_key = self._derive_keys(shared_secret)
        
        return SessionState(
            their_identity_key=their_identity,
            their_signed_prekey=their_signed_prekey,
            root_key=root_key,
            chain_key=chain_key
        )
    
    def encrypt(self, plaintext: bytes) -> Tuple[bytes, int]:
        """
        Encrypt a message using current chain key.
        
        Each message gets a unique key (Double Ratchet).
        """
        # Derive message key from chain key
        message_key, new_chain_key = self._ratchet_chain(
            self.session.chain_key
        )
        
        # Encrypt with message key
        ciphertext = self._aes_encrypt(message_key, plaintext)
        
        # Update chain key (ratchet forward)
        message_number = self.session.message_number
        self.session.chain_key = new_chain_key
        self.session.message_number += 1
        
        return ciphertext, message_number
    
    def decrypt(self, ciphertext: bytes, message_number: int) -> bytes:
        """
        Decrypt a message.
        
        May need to skip ahead if messages arrive out of order.
        """
        # Derive the message key for this message number
        message_key = self._get_message_key(message_number)
        
        # Decrypt
        plaintext = self._aes_decrypt(message_key, ciphertext)
        
        return plaintext
    
    def _ratchet_chain(self, chain_key: bytes) -> Tuple[bytes, bytes]:
        """
        Symmetric ratchet: derive next keys from current.
        
        This is the "forward secrecy" mechanism — old keys
        can't be derived from new keys.
        """
        # Message key for current message
        message_key = hmac.new(chain_key, b"\x01", hashlib.sha256).digest()
        
        # New chain key for next message
        new_chain_key = hmac.new(chain_key, b"\x02", hashlib.sha256).digest()
        
        return message_key, new_chain_key


class GroupEncryption:
    """
    Group messaging uses a different approach.
    
    Instead of pairwise encryption, sender distributes
    a "sender key" to all group members.
    """
    
    def __init__(self, group_id: str, my_user_id: str):
        self.group_id = group_id
        self.my_user_id = my_user_id
        
        # Generate my sender key for this group
        self.my_sender_key = self._generate_sender_key()
        
        # Store other members' sender keys
        self.member_sender_keys = {}
    
    def distribute_sender_key(self, member_sessions: dict):
        """
        Send my sender key to all group members.
        
        Uses pairwise Signal sessions for secure distribution.
        """
        for member_id, session in member_sessions.items():
            # Encrypt sender key using pairwise session
            encrypted_key = session.encrypt(self.my_sender_key)
            # Send to member via normal message path
            yield member_id, encrypted_key
    
    def encrypt_for_group(self, plaintext: bytes) -> bytes:
        """
        Encrypt message using my sender key.
        
        All group members who have my sender key can decrypt.
        Much more efficient than pairwise encryption.
        """
        # Use sender key with chain ratchet
        message_key = self._derive_message_key()
        ciphertext = self._aes_encrypt(message_key, plaintext)
        
        # Ratchet forward
        self._ratchet_sender_key()
        
        return ciphertext
    
    def handle_member_left(self, member_id: str):
        """
        When someone leaves, all sender keys must be regenerated.
        
        This ensures the departed member can't read future messages.
        """
        # Generate new sender key
        self.my_sender_key = self._generate_sender_key()
        
        # Redistribute to remaining members
        # (handled by caller)

# offline/message_store.py

"""
Offline Message Storage

WhatsApp's approach:
- Messages stored only until delivered
- After delivery confirmation, deleted from server
- This minimizes server storage and privacy exposure
"""

import asyncio
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta


@dataclass
class QueuedMessage:
    """A message waiting for delivery."""
    message_id: str
    sender_id: str
    recipient_id: str
    encrypted_content: bytes
    timestamp: datetime
    expiry: datetime  # Messages expire after 30 days
    
    # Delivery tracking
    delivery_attempts: int = 0
    last_attempt: Optional[datetime] = None


class OfflineMessageStore:
    """
    Stores messages for offline users.
    
    Design principles:
    - In-memory for speed (Mnesia ETS tables)
    - Replicated for durability
    - Partitioned by recipient for scalability
    - Messages deleted after delivery
    """
    
    def __init__(self, db, max_queue_size: int = 10000):
        self.db = db  # Mnesia in actual WhatsApp
        self.max_queue_size = max_queue_size
        
        # TTL for stored messages
        self.message_ttl = timedelta(days=30)
    
    async def queue_message(
        self,
        sender_id: str,
        recipient_id: str,
        encrypted_content: bytes
    ) -> QueuedMessage:
        """
        Store message for offline recipient.
        """
        message = QueuedMessage(
            message_id=generate_id(),
            sender_id=sender_id,
            recipient_id=recipient_id,
            encrypted_content=encrypted_content,
            timestamp=datetime.utcnow(),
            expiry=datetime.utcnow() + self.message_ttl
        )
        
        # Check queue size limits
        queue_size = await self.db.count_for_recipient(recipient_id)
        if queue_size >= self.max_queue_size:
            # Drop oldest message (or reject new one)
            await self.db.delete_oldest(recipient_id)
        
        await self.db.save(message)
        return message
    
    async def get_pending_messages(
        self,
        recipient_id: str
    ) -> List[QueuedMessage]:
        """
        Get all pending messages for a user who just came online.
        
        Called when user establishes connection.
        """
        messages = await self.db.get_by_recipient(
            recipient_id,
            order_by="timestamp",
            limit=1000  # Batch size
        )
        
        # Filter expired messages
        now = datetime.utcnow()
        valid_messages = [m for m in messages if m.expiry > now]
        
        return valid_messages
    
    async def mark_delivered(
        self,
        message_ids: List[str]
    ):
        """
        Delete messages that have been delivered.
        
        This is called after receiving delivery ACK from recipient.
        """
        for message_id in message_ids:
            await self.db.delete(message_id)
    
    async def cleanup_expired(self):
        """
        Background job: Delete expired messages.
        
        Runs periodically to clean up messages that
        were never delivered (user never came back online).
        """
        now = datetime.utcnow()
        expired = await self.db.get_expired(now)
        
        for message in expired:
            await self.db.delete(message.message_id)
            
            # Optionally notify sender that message expired
            # (WhatsApp doesn't do this)


class WriteBackCache:
    """
    WhatsApp uses write-back caching for offline messages.
    
    - Messages stored in memory first (fast)
    - Asynchronously persisted to disk
    - Handles disk I/O slowdowns gracefully
    """
    
    def __init__(self, memory_store, disk_store):
        self.memory = memory_store   # ETS table
        self.disk = disk_store       # Mnesia disk table
        self.pending_writes = []
    
    async def store(self, message: QueuedMessage):
        """
        Store in memory immediately, persist async.
        """
        # Immediate: memory storage
        await self.memory.put(message.message_id, message)
        
        # Async: queue for disk persistence
        self.pending_writes.append(message)
        
        # Trigger background write (non-blocking)
        asyncio.create_task(self._flush_writes())
    
    async def _flush_writes(self):
        """
        Background: persist to disk.
        
        Batched for efficiency.
        """
        if len(self.pending_writes) >= 100:
            batch = self.pending_writes[:100]
            self.pending_writes = self.pending_writes[100:]
            
            await self.disk.batch_write(batch)

# presence/manager.py

"""
Presence Management

Optimizations a WhatsApp-scale presence system needs (a likely design):
1. Batching: Aggregate multiple presence updates
2. Rate limiting: Don't send typing every keystroke
3. TTL: Presence expires automatically
4. Selective delivery: Only send to active chats
"""

from dataclasses import dataclass
from typing import Dict, Set, Optional
from datetime import datetime, timedelta
from enum import Enum


class PresenceState(Enum):
    ONLINE = "online"
    OFFLINE = "offline"
    TYPING = "typing"


@dataclass
class UserPresence:
    """Current presence state for a user."""
    user_id: str
    state: PresenceState
    last_seen: datetime
    typing_in_chat: Optional[str] = None


class PresenceManager:
    """
    Manages presence state and notifications.
    
    Key optimizations:
    1. In-memory only (no persistence needed)
    2. TTL-based expiry (online expires in 60s)
    3. Rate limiting (typing indicator throttled)
    4. Subscription-based (only notify interested parties)
    """
    
    def __init__(self, redis_cluster, notification_service):
        self.redis = redis_cluster
        self.notifier = notification_service
        
        # TTLs
        self.online_ttl = timedelta(seconds=60)
        self.typing_ttl = timedelta(seconds=5)
    
    async def set_online(self, user_id: str):
        """
        Mark user as online.
        
        Uses Redis with TTL — automatically expires to offline.
        """
        presence = UserPresence(
            user_id=user_id,
            state=PresenceState.ONLINE,
            last_seen=datetime.utcnow()
        )
        
        # Store with TTL (auto-expires to offline)
        await self.redis.set(
            f"presence:{user_id}",
            presence.to_json(),
            ex=int(self.online_ttl.total_seconds())
        )
        
        # Notify subscribers
        await self._notify_subscribers(user_id, presence)
    
    async def set_typing(self, user_id: str, chat_id: str):
        """
        Mark user as typing in a specific chat.
        
        Rate limited: max 1 update per second per chat.
        """
        # Rate limit check
        rate_key = f"typing_rate:{user_id}:{chat_id}"
        if await self.redis.exists(rate_key):
            return  # Already sent recently, skip
        
        # Set rate limit marker (1 second)
        await self.redis.set(rate_key, "1", ex=1)
        
        # Update typing state
        await self.redis.set(
            f"typing:{user_id}:{chat_id}",
            "1",
            ex=int(self.typing_ttl.total_seconds())
        )
        
        # Notify only the other participant(s) of this chat
        await self._notify_chat_participants(
            chat_id, 
            user_id, 
            PresenceState.TYPING
        )
    
    async def get_presence(self, user_id: str) -> Optional[UserPresence]:
        """
        Get current presence for a user.
        
        Used when opening a chat to show current status.
        """
        data = await self.redis.get(f"presence:{user_id}")
        
        if data:
            return UserPresence.from_json(data)
        else:
            # No presence record = offline
            # Get last seen from persistent storage
            last_seen = await self._get_last_seen(user_id)
            return UserPresence(
                user_id=user_id,
                state=PresenceState.OFFLINE,
                last_seen=last_seen
            )
    
    async def _notify_subscribers(
        self,
        user_id: str,
        presence: UserPresence
    ):
        """
        Notify users who care about this user's presence.
        
        Optimization: Only notify users who have this user's
        chat currently open (not all contacts).
        """
        # Get list of users with this chat open
        subscribers = await self.redis.smembers(
            f"presence_subscribers:{user_id}"
        )
        
        for subscriber_id in subscribers:
            await self.notifier.send_presence_update(
                subscriber_id,
                presence
            )
    
    async def subscribe_to_presence(
        self,
        subscriber_id: str,
        target_user_id: str
    ):
        """
        Subscribe to presence updates for a user.
        
        Called when user opens a chat.
        """
        await self.redis.sadd(
            f"presence_subscribers:{target_user_id}",
            subscriber_id
        )
        
        # Auto-expire subscription after 10 minutes
        await self.redis.expire(
            f"presence_subscribers:{target_user_id}",
            600
        )
    
    async def unsubscribe_from_presence(
        self,
        subscriber_id: str,
        target_user_id: str
    ):
        """
        Unsubscribe from presence updates.
        
        Called when user closes a chat.
        """
        await self.redis.srem(
            f"presence_subscribers:{target_user_id}",
            subscriber_id
        )