Messaging System em Escala — entrega garantida, ordenação e E2E encryption

Em 2014 a equipe do WhatsApp publicou um post sobre a arquitetura que suportava 450 milhões de usuários com apenas 32 engenheiros. Um dos engenheiros, Rick Reed, descreveu em detalhe o servidor Erlang que gerenciava 2 milhões de conexões WebSocket numa única máquina — um número que era considerado impossível na época. O segredo não era magia: era a escolha deliberada de Erlang, uma linguagem desenhada especificamente para sistemas de telecomunicações que mantêm milhões de conexões simultâneas. Cada conexão vivia num processo Erlang leve (~2KB de memória), supervisionado por um árbitro que reiniciava automaticamente processos que falhavam. A decisão de usar Erlang foi tomada em 2009, antes de qualquer preocupação com escala — foi feita porque a equipe entendia o problema: um sistema de mensagens é fundamentalmente um problema de conexões persistentes e entrega confiável, e Erlang havia resolvido exatamente isso para telefonia por décadas.

O que torna um sistema de mensagens diferente de uma API REST comum é o problema de estado. Uma API REST é stateless: qualquer servidor pode responder qualquer request. Um sistema de mensagens é stateful: o servidor mantém conexões abertas com clientes, e quando uma mensagem chega para o usuário A, o sistema precisa saber em qual servidor físico o usuário A está conectado para entregar em tempo real. Essa característica — servidores stateful — cria três classes de problemas que este conceito resolve em detalhe: routing de mensagens entre servidores, garantia de entrega quando o destinatário está offline, e consistência de ordenação quando dois usuários enviam mensagens simultaneamente na mesma conversa.

Requisitos e estimation

Requisitos funcionais:

• Enviar e receber mensagens de texto e mídia (imagens, vídeo, áudio)
• Mensagens 1:1 e grupos (até 1024 membros)
• Delivery receipts: enviado ao servidor (✓), entregue ao dispositivo (✓✓), lido (✓✓ azul)
• Mensagens offline: usuário recebe ao reconectar
• Opcional: E2E encryption (server não lê o conteúdo)

Requisitos não-funcionais:

• Latência de entrega <100ms quando destinatário está online
• Disponibilidade 99.99% (tolerância a falha de datacenter)
• Nenhuma mensagem perdida, nenhuma mensagem entregue duas vezes

# Estimation — escala WhatsApp
# Usuários: 2B cadastrados, 500M ativos simultaneamente no pico
# Mensagens: 100B/dia = 1.157.000 msg/s médio; pico 3× = ~3.5M msg/s

# Conexões simultâneas:
# 500M usuários × 1 conexão WebSocket = 500M conexões
# Distribuído em servidores WS com 1M conexões/servidor = 500 servidores WS

# Tamanho médio de mensagem:
# Texto: ~500 bytes; mídia: armazenada separadamente no object storage
# 70% texto + 30% mídia (só URL na mensagem)
# 1.15M msg/s × 500B = 575 MB/s de writes ao storage

# Storage total:
# 100B mensagens/dia × 500B × 365 dias × 3 anos × 3x replicação
# = 100B × 500B × 1095 × 3 = ~164 PB (total 3 anos com replicação)
# Na prática: TTL por tier (SSD ativo 30 dias, HDD arquivo 3 anos)

# Custo de polling vs WebSocket:
# 500M clientes polling a cada 5s = 100M req/s só para verificar novas msgs
# Vs WebSocket: 500M conexões abertas, push quando há mensagem → 1.15M msg/s
# WebSocket elimina 99% dos requests desnecessários

Arquitetura WebSocket com sticky sessions

O problema fundamental de um sistema de mensagens distribuído: quando o usuário B envia uma mensagem para o usuário A, o Message Service precisa saber em qual servidor WebSocket o usuário A está conectado. Esse mapeamento — user_id → WS server — é o coração da arquitetura.

# Registro de sessão no Redis (atualizado a cada heartbeat)
# Quando usuário conecta ao WS-server-3:
SETEX ws:session:{user_id} 90 "ws-server-3"  # TTL = 3× o intervalo de heartbeat

# Heartbeat do cliente (a cada 30s):
# Client → Server: {type: "ping"}
# Server → Client: {type: "pong"}
# Se 3 heartbeats perdidos (90s), servidor fecha conexão e chave expira

# Fluxo de envio de mensagem:
# 1. Sender POST /messages → Message Service
# 2. Message Service persiste + lookup no Redis:
ws_server = redis.get(f"ws:session:{recipient_id}")
# → "ws-server-3" (online) ou None (offline)

# 3a. Online: publicar via Redis Pub/Sub
if ws_server:
    redis.publish(f"channel:user:{recipient_id}", message_json)
    # Redis entrega ao WS-server-3 que tem o subscriber ativo
    # WS-server-3 injeta na conexão WebSocket do recipient

# 3b. Offline: persistir, disparar push notification
else:
    push_notification_service.send(recipient_id, "new_message")
    # Push é apenas um nudge: "você tem mensagens novas"
    # O conteúdo é buscado pelo cliente quando reconecta

Por que Redis Pub/Sub em vez de HTTP entre servidores WS? Com 500 servidores WS, cada mensagem poderia exigir uma chamada HTTP do Message Service ao WS correto. Redis Pub/Sub inverte o modelo: cada WS server subscreve um canal por usuário conectado; o Message Service publica sem precisar saber o endereço físico de nenhum servidor. O Redis faz o roteamento.

# No WS Server: ao aceitar nova conexão do user_id
def on_connect(user_id, websocket):
    connections[user_id] = websocket
    # Registrar sessão e subscrever canal
    redis.setex(f"ws:session:{user_id}", 90, server_id)
    pubsub = redis.pubsub()
    pubsub.subscribe(f"channel:user:{user_id}")
    asyncio.create_task(listen_and_forward(user_id, pubsub, websocket))

async def listen_and_forward(user_id, pubsub, websocket):
    async for message in pubsub.listen():
        if message["type"] == "message":
            await websocket.send(message["data"])

# No WS Server: ao desconectar
def on_disconnect(user_id):
    connections.pop(user_id, None)
    redis.delete(f"ws:session:{user_id}")
    # O canal Redis se fecha automaticamente sem subscribers

Entrega exatamente uma vez: deduplicação com idempotency keys

O cliente pode enviar a mesma mensagem duas vezes: se a conexão cai após o POST mas antes do ACK, o cliente não sabe se o servidor recebeu e reenvia. Sem deduplicação, o destinatário recebe duplicatas. A solução: o cliente gera um client_message_id antes de enviar — o servidor usa esse ID como chave de idempotência.

# Cliente (antes de enviar):
client_message_id = uuid4()  # gerado pelo cliente, persiste na fila local
# Armazenado localmente até ACK confirmado

# Servidor (Message Service):
def send_message(sender_id, recipient_id, content, client_message_id):
    # 1. Verificar dedup: já processamos esse client_message_id?
    dedup_key = f"msg:dedup:{client_message_id}"
    existing = redis.get(dedup_key)
    if existing:
        # Já processado — retornar o mesmo server_message_id
        return {"server_message_id": existing, "status": "duplicate"}

    # 2. Gerar ID único do servidor (Snowflake: timestamp + machine_id + sequence)
    server_message_id = snowflake.generate()

    # 3. Persistir no Cassandra (atomic write)
    cassandra.execute("""
        INSERT INTO messages
            (conversation_id, message_id, sender_id, content, client_message_id, created_at)
        VALUES (?, ?, ?, ?, ?, ?)
    """, [conversation_id, server_message_id, sender_id, content, client_message_id, now()])

    # 4. Marcar como processado (TTL = 7 dias, janela de retry razoável)
    redis.setex(dedup_key, 604800, str(server_message_id))

    # 5. Entregar ao recipient
    deliver_to_recipient(recipient_id, message_payload)

    return {"server_message_id": server_message_id, "status": "sent"}

# Por que Redis e não o Cassandra para dedup?
# Cassandra: INSERT IF NOT EXISTS é pesado — leva 2 roundtrips (read + write = Paxos leve)
# Redis: GET + SETEX é O(1), sub-millisecond
# Tradeoff: Redis pode perder o cache se reiniciar → janela de dedup potencialmente menor
# Para sistemas críticos: Cassandra com LWT (Lightweight Transactions) como fallback

Ordenação por conversa

Mensagens numa conversa precisam de ordenação consistente. O problema: dois usuários enviam mensagens simultaneamente — qual chega primeiro ao servidor? Clock skew entre servidores é inevitável (±100ms é normal). Snowflake ID tem timestamp nos 41 bits mais significativos, o que fornece ordenação aproximada mas não estrita.

# Opção 1: Snowflake ID como ordering (suficiente para a maioria dos casos)
# Snowflake: 41 bits timestamp | 10 bits machine_id | 12 bits sequence
# - 2 mensagens no mesmo milissegundo: ordenadas por machine_id (arbitrário mas consistente)
# - Clock skew de 100ms: as mensagens ficam "fora de ordem" por 100ms → aceitável para UX

# Opção 2: sequence counter por conversa (quando ordenação estrita é necessária)
# Redis INCR é atômico: garante sequence único e crescente por conversa
def assign_sequence(conversation_id):
    return redis.incr(f"conv:seq:{conversation_id}")
# Desvantagem: hot key em conversas muito ativas (10k msg/s na mesma conversa)
# Solução: sharding do counter por slot (counter_slot = message_id % 16)
# → 16 counters independentes, cada um incrementado atomicamente

# Opção 3: Cassandra timeuuid (padrão na prática)
# timeuuid é UUID version 1: incorpora timestamp + clock sequence + node MAC
# Naturalmente ordenável no Cassandra como clustering key
# Cassandra resolve ties por clock sequence → ordenação deterministicamente consistente

# Schema do Cassandra:
# CREATE TABLE messages (
#   conversation_id uuid,
#   message_id timeuuid,       -- ordering nativo, gerado no servidor
#   sender_id bigint,
#   content text,              -- encrypted blob se E2E
#   message_type text,         -- 'text', 'image', 'video', 'audio'
#   media_url text,            -- null para texto; URL no object storage para mídia
#   delivery_status text,      -- 'sent', 'delivered', 'read'
#   client_message_id uuid,    -- para idempotência
#   created_at timestamp,
#   PRIMARY KEY (conversation_id, message_id)
# ) WITH CLUSTERING ORDER BY (message_id DESC);
#
# Por que Cassandra e não PostgreSQL?
# - 100B writes/dia = 1.15M writes/s → PostgreSQL teto ~10k writes/s por instância
# - Queries são sempre por conversation_id (partition key conhecido) → sem full scans
# - Workload é write-heavy: Cassandra LSM tree absorve writes sem B-tree rebalancing
# - Retenção e archival: TTL nativo por linha (ex: mensagens >30 dias vão para S3)

# Paginação de histórico:
# SELECT * FROM messages
# WHERE conversation_id = ?
# AND message_id < ?   -- cursor-based pagination (timeuuid do último item visto)
# ORDER BY message_id DESC
# LIMIT 50;

Entrega offline e push notifications

Quando o destinatário está offline, a mensagem já foi persistida no Cassandra. A questão é: como o cliente descobre que há mensagens novas ao reconectar, e como notificá-lo enquanto está offline?

# Fluxo completo de entrega offline:

# 1. Mensagem chegou, recipient está offline
def handle_offline_recipient(recipient_id, message):
    # Mensagem já persistida no Cassandra — não há risco de perda
    # Enviar push notification como nudge (sem conteúdo por privacidade)
    push_service.send(
        user_id=recipient_id,
        platform=get_platform(recipient_id),  # ios ou android
        payload={
            "type": "new_messages",
            "badge_count": get_unread_count(recipient_id),
            # NÃO incluir conteúdo da mensagem — privacidade + limite de 4KB do APNs
        }
    )

# 2. iOS (APNs) vs Android (FCM):
# APNs (Apple Push Notification service):
#   - Limite: 4KB por notification
#   - Silent push: o app acorda em background e faz pull das mensagens
#   - Reliable: APNs garante entrega mesmo com dispositivo offline (fila)
#   - Token de dispositivo: muda ao reinstalar o app → sistema deve atualizar

# FCM (Firebase Cloud Messaging):
#   - Limite: 4KB para data messages, 2KB para notification messages
#   - Data message: app processa quando acorda
#   - Notification message: sistema operacional exibe sem accordar o app
#   - Para mensagens: data message preferível (app controla a UI)

# 3. Reconexão: sync diferencial
def on_reconnect(user_id, last_sync_state):
    # Cliente envia: {conversation_id: last_message_id} para cada conversa ativa
    # Servidor responde com mensagens novas desde o último sync

    unread = {}
    for conversation_id, last_message_id in last_sync_state.items():
        new_messages = cassandra.execute("""
            SELECT * FROM messages
            WHERE conversation_id = ?
            AND message_id > ?
            ORDER BY message_id ASC
            LIMIT 100
        """, [conversation_id, last_message_id])
        if new_messages:
            unread[conversation_id] = new_messages

    # Enviar via WebSocket em batches
    websocket.send({"type": "sync_response", "messages": unread})

    # Atualizar delivery_status = 'delivered' para as mensagens entregues
    update_delivery_status(user_id, unread)

Delivery receipts: os dois ticks

# Três estados de entrega:
# ✓ (cinza): mensagem chegou ao servidor (ACK do POST /messages)
# ✓✓ (cinza): mensagem entregue ao dispositivo do destinatário
# ✓✓ (azul): mensagem lida pelo destinatário

# Implementação via WebSocket ACK:

# Recipient → quando recebe a mensagem via WS:
websocket.send({"type": "ack", "message_id": msg_id, "status": "delivered"})

# Recipient → quando o usuário abre a conversa (mensagem visível na tela):
websocket.send({"type": "ack", "message_id": msg_id, "status": "read"})

# WS Server → encaminha ACK para Message Service
# Message Service:
def handle_ack(recipient_id, message_id, status):
    # Atualizar no Cassandra
    cassandra.execute("""
        UPDATE messages SET delivery_status = ?
        WHERE conversation_id = ? AND message_id = ?
    """, [status, conversation_id, message_id])

    # Notificar o sender (que pode estar em outro WS server)
    sender_id = get_sender(message_id)
    redis.publish(f"channel:user:{sender_id}", {
        "type": "receipt",
        "message_id": message_id,
        "status": status,
        "timestamp": now()
    })

# Para grupos: "lido" é por membro → matrix esparsa
# CREATE TABLE message_receipts (
#   message_id timeuuid,
#   user_id bigint,
#   status text,
#   timestamp timestamp,
#   PRIMARY KEY (message_id, user_id)
# );
# WhatsApp simplifica: ✓✓ azul = TODOS leram; ✓✓ cinza = pelo menos um não leu

Fan-out de grupos

Uma mensagem de grupo precisa ser entregue a N membros. Para grupos pequenos (WhatsApp, até 1024 membros), o fan-out acontece no servidor. Para canais grandes (Slack public channels, até 100k+ membros), fan-out on write é proibitivo.

# Fan-out para grupos:

SMALL_GROUP_THRESHOLD = 500  # ajustável por carga

async def deliver_group_message(message, group_id):
    members = await group_member_cache.get(group_id)
    # Cache de membros com TTL 5min — evita query ao DB a cada mensagem

    if len(members) <= SMALL_GROUP_THRESHOLD:
        # Fan-out on write: entregar a cada membro individualmente
        tasks = []
        for member_id in members:
            if member_id == message.sender_id:
                continue  # não entregar ao remetente
            ws_server = redis.get(f"ws:session:{member_id}")
            if ws_server:
                # Online: pub/sub
                tasks.append(
                    redis.publish(f"channel:user:{member_id}", message.to_json())
                )
            else:
                # Offline: enfileirar push notification
                tasks.append(
                    push_queue.enqueue(member_id, group_id, message.id)
                )
        await asyncio.gather(*tasks)

    else:
        # Fan-out on read: publicar uma vez no canal do grupo
        # Todos os WS servers subscrevem canais de grupos para seus usuários conectados
        redis.publish(f"channel:group:{group_id}", message.to_json())
        # Cada WS server que tem membros do grupo online recebe e entrega

# Por que fan-out on write para grupos pequenos?
# - Cada membro pode estar em WS server diferente
# - Fan-out on read exige que cada WS server descubra se tem membros do grupo
# - Para grupos pequenos, o overhead de N publishes é aceitável vs complexidade de tracking

# Por que fan-out on read para grupos grandes?
# - 100k membros × 1 msg/s no canal = 100k publishes/s só para um canal
# - Com fan-out on read: 1 publish → cada WS server entrega para os membros conectados que ele tem
# - WS server mantém: {group_id: [user_id, ...]} para usuários conectados → entrega direta

E2E Encryption: Signal Protocol

E2E encryption significa que nem o servidor consegue ler o conteúdo das mensagens. O servidor armazena blobs opacos e serve de intermediário para troca de chaves públicas. O Signal Protocol, usado pelo WhatsApp e pelo próprio Signal, resolve um problema difícil: como estabelecer um canal seguro sem que os dois usuários estejam online ao mesmo tempo.

# Signal Protocol — conceitos fundamentais:

# 1. Identity Keys (permanentes, geradas no dispositivo)
identity_key_pair = generate_ed25519_keypair()
# private_key: NUNCA sai do dispositivo
# public_key: publicada no servidor para que remetentes possam iniciar sessão

# 2. Pre-Keys: chaves efêmeras pré-publicadas no servidor
# Permitem que Alice envie a primeira mensagem a Bob sem Bob estar online

signed_prekey = generate_signed_prekey(identity_key_pair)  # médio prazo, ~semanas
one_time_prekeys = [generate_one_time_prekey() for _ in range(100)]  # single-use
# Publicados no servidor: {identity_public, signed_prekey, one_time_prekeys[]}
# Servidor deleta one_time_prekey após entregá-la a Alice (não pode ser reutilizada)

# 3. Session Establishment: X3DH (Extended Triple Diffie-Hellman)
# Alice quer enviar para Bob (Bob está offline):
def establish_session_x3dh(alice_identity, bob_bundle):
    # bob_bundle = {identity_key, signed_prekey, one_time_prekey} do servidor

    # 4 trocas DH independentes:
    dh1 = dh(alice_identity.private, bob_bundle.signed_prekey.public)
    dh2 = dh(alice_ephemeral.private, bob_bundle.identity_key.public)
    dh3 = dh(alice_ephemeral.private, bob_bundle.signed_prekey.public)
    dh4 = dh(alice_ephemeral.private, bob_bundle.one_time_prekey.public)  # single-use

    # Derivar master secret dos 4 DH outputs
    master_secret = hkdf(dh1 + dh2 + dh3 + dh4)

    # Inicializar Double Ratchet com o master_secret
    session = DoubleRatchet.initialize(master_secret, bob_bundle.signed_prekey.public)
    return session

# 4. Double Ratchet: forward secrecy + break-in recovery
# Cada mensagem usa uma chave diferente, derivada por duas funções "ratchet":

class DoubleRatchet:
    def encrypt_message(self, plaintext):
        # Symmetric ratchet: avança a chain key, deriva message key
        self.chain_key, message_key = kdf(self.chain_key)
        ciphertext = aes_gcm_encrypt(message_key, plaintext)
        # message_key é deletado após uso → forward secrecy
        # (mesmo com chain_key atual, mensagens passadas não são descriptografáveis)
        del message_key
        return ciphertext, self.dh_ratchet_public  # inclui public key atual para Bob

    def decrypt_message(self, ciphertext, sender_dh_public):
        # DH ratchet: ambas as partes avançam o estado DH periodicamente
        # Isso "cura" comprometimentos — mesmo que uma chave vaze,
        # a sessão se recupera quando o DH ratchet avança
        if sender_dh_public != self.last_sender_dh_public:
            # Bob avançou o DH → recalcular root key
            dh_output = dh(self.dh_private, sender_dh_public)
            self.root_key, self.chain_key = kdf(self.root_key, dh_output)
            self.rotate_dh_keypair()

        self.chain_key, message_key = kdf(self.chain_key)
        plaintext = aes_gcm_decrypt(message_key, ciphertext)
        del message_key
        return plaintext

# 5. Grupos: Sender Keys (mais eficiente que N sessões individuais)
# Para um grupo de 100 membros, usar 100 sessões individuais = 100 encriptações por mensagem
# Sender Keys: cada membro tem uma "sender key" para o grupo
# - Alice gera um sender key específico para o grupo G
# - Distribui o sender key para cada membro via suas sessões individuais (1:1 Signal Protocol)
# - Mensagens ao grupo: encriptadas UMA VEZ com o sender key de Alice
# - Qualquer membro (com o sender key de Alice) pode descriptografar

# O servidor armazena:
# - Blobs encriptados (opaco, não consegue ler)
# - Chaves públicas (identity, signed prekeys, one-time prekeys)
# - Metadados: quem fala com quem, quando, tamanho das mensagens
# Metadados são a única "vulnerabilidade" — servidor sabe o grafo social mesmo sem ler conteúdo

Reconexão e sync diferencial

# Heartbeat e detecção de desconexão:
HEARTBEAT_INTERVAL = 30   # segundos — cliente envia ping
SESSION_TTL = 90           # 3× o intervalo: 3 heartbeats perdidos = desconectado
RECONNECT_BACKOFF = [1, 2, 4, 8, 16, 32, 60]  # exponential backoff em segundos

# No cliente (pseudocódigo):
async def maintain_connection():
    backoff_idx = 0
    while True:
        try:
            ws = await connect(WS_ENDPOINT)
            backoff_idx = 0  # reset on success
            await sync_missed_messages(ws)
            await run_heartbeat_loop(ws)  # lança heartbeats a cada 30s
        except ConnectionError:
            delay = RECONNECT_BACKOFF[min(backoff_idx, len(RECONNECT_BACKOFF)-1)]
            await asyncio.sleep(delay)
            backoff_idx += 1

# Sync diferencial ao reconectar:
async def sync_missed_messages(ws):
    # Cliente mantém localmente: {conversation_id: last_received_message_id}
    sync_state = local_db.get_sync_state()
    await ws.send({"type": "sync_request", "state": sync_state})
    response = await ws.recv()
    # Recebe: {conversation_id: [messages_since_last_id]}
    for conv_id, messages in response["messages"].items():
        local_db.store_messages(conv_id, messages)

# Por que sync diferencial em vez de pull completo?
# Pull completo: buscar TODAS as mensagens não lidas → pode ser gigabytes se offline por semanas
# Sync diferencial: o cliente sabe o último ID que recebeu → server retorna apenas o delta
# O last_received_message_id é persistido localmente no dispositivo (SQLite no mobile)

Arquitetura completa

Client A (sender)
  │ WebSocket (sticky session)
  ▼
WS Gateway (stateful, conexão persistente)
  │ POST /api/messages
  ▼
Message Service (stateless, horizontalmente escalável)
  ├── Redis: verificar dedup (GET msg:dedup:{client_msg_id})
  ├── Cassandra: persistir mensagem
  ├── Redis: SETEX ws:session → lookup WS server do recipient
  └── Redis: PUBLISH channel:user:{recipient_id}
              │
              ▼
        Redis Pub/Sub
              │
              ▼
WS Gateway (servidor do recipient B)
  │ subscriber ativo para channel:user:{B}
  ▼
Client B (recipient) — online: recebe em <100ms

Se B offline:
  Message Service → Push Notification Service
    ├── iOS: APNs (silent push, badge count)
    └── Android: FCM (data message, app acorda e faz sync)

Storage layer:
  Cassandra Cluster (messages, conversation_id como partition key)
  Redis Cluster (sessions, pub/sub, group membership cache, dedup)
  PostgreSQL (usuários, contatos, grupos — dados relacionais estruturados)
  S3 / GCS (mídia: imagens, vídeo, áudio — apenas URL na mensagem)

Para grupos:
  Message Service → Group Fan-out Service
    ├── Pequenos (<500): N publishes individuais (pub/sub por membro)
    └── Grandes (>500): 1 publish no canal do grupo (WS servers subscrevem)

Decisões de engenharia

Perguntas de entrevista

Exercícios práticos