Sistema de chat 1-1 e grupo: armazenando mensagens e estado de leitura

1. Visão geral

Chat parece simples: enviar uma linha de texto e exibi-la. Mas com milhões de usuários, você precisa responder perguntas mais difíceis. Como manter a ordem correta das mensagens? Como fazer retransmissões sem criar mensagens duplicadas? Como saber até onde cada um leu sem gravar bilhões de registros de “visualizado”?

Este artigo percorre um design completo para chat 1-1 e em grupo. O design inclui requisitos, modelo de dados, APIs, fluxos de processamento e desafios. Após a leitura, você será capaz de construir a parte central de armazenamento de mensagens e estado de leitura usando Spring Boot 3, PostgreSQL e Kafka.

2. Requisitos

2.1. Funcionalidades

Chat 1-1 e em grupo (máximo 1.000 membros).
Enviar mensagens de texto. Mensagens exibidas na ordem correta em uma conversa.
Visualizar histórico, rolar para trás com paginação.
Lista de conversas (inbox) com número de mensagens não lidas.
Estado de leitura: chat 1-1 mostra “Visto”, chat em grupo mostra “Visto por N pessoas”.

2.2. Escala e restrições

10 milhões de DAU. Em média, cada pessoa envia 40 mensagens/dia, ou 400 milhões de mensagens/dia, cerca de 4.600 mensagens/segundo. Nos horários de pico, pode chegar a cerca de 30.000 mensagens/segundo.
Não pode perder mensagens que foram confirmadas como enviadas com sucesso.
A ordem precisa estar correta apenas em uma conversa. Não precisa de ordem global.
O cliente pode retransmitir devido a problemas de rede. O servidor deve ser idempotente.
Eventos de “leitura” são mais frequentes do que eventos de envio, pois cada abertura do app gera um.

A terceira restrição é a chave para todo o design. Como você só precisa de ordem local, tudo pode ser shardado por conversation_id.

3. Modelo de dados

A ideia central: cada conversa tem um número de sequência crescente (seq). A mensagem N em uma conversa tem seq = N. O estado de leitura de um membro é apenas um número, last_read_seq. É um watermark, não uma lista de cada mensagem lida.

CREATE TABLE conversation (
    id              BIGINT PRIMARY KEY,
    type            VARCHAR(10) NOT NULL,          -- DIRECT | GROUP
    last_seq        BIGINT      NOT NULL DEFAULT 0,
    last_message_at TIMESTAMPTZ
);

CREATE TABLE conversation_member (
    conversation_id BIGINT NOT NULL REFERENCES conversation(id),
    user_id         BIGINT NOT NULL,
    last_read_seq   BIGINT NOT NULL DEFAULT 0,
    PRIMARY KEY (conversation_id, user_id)
);
CREATE INDEX idx_member_user ON conversation_member(user_id);

CREATE TABLE message (
    conversation_id BIGINT      NOT NULL,
    seq             BIGINT      NOT NULL,
    sender_id       BIGINT      NOT NULL,
    client_msg_id   UUID        NOT NULL,
    body            TEXT        NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (conversation_id, seq),
    UNIQUE (conversation_id, client_msg_id)
);

CREATE TABLE outbox (
    id         BIGSERIAL PRIMARY KEY,
    msg_key    VARCHAR(64) NOT NULL,
    payload    JSONB       NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

Pontos a notar:

A chave primária (conversation_id, seq) permite ler histórico apenas varrendo um intervalo contíguo do índice. Essa chave também é a chave de shard natural quando você faz sharding.
UNIQUE (conversation_id, client_msg_id) é a camada de proteção contra duplicação no nível do banco de dados.
O número de mensagens não lidas é last_seq - last_read_seq. Esse cálculo não precisa de COUNT(*).
“Visto por” em grupos é calculado pelo número de membros com last_read_seq >= seq. Você não precisa de uma tabela message_receipt com bilhões de linhas.
Para chat 1-1, adicione a tabela direct_pair(user_low, user_high, conversation_id) com constraint unique. Dois usuários abrindo chat simultaneamente não criarão duas conversas.

Nota: Watermark assume que o usuário lê sequencialmente, ou seja, após ler a mensagem N, assume-se que todas as mensagens anteriores foram lidas. Essa suposição é verdadeira para praticamente todo aplicativo de chat. Se você precisar saber exatamente qual mensagem específica foi aberta (por exemplo, mensagens que se autodestruem), armazene um registro separado para esse tipo de mensagem. Não aplique a todas.

4. API

Method	Endpoint	Descrição
`POST`	`/conversations/{id}/messages`	Enviar mensagem. Body contém `clientMsgId` e `body`
`GET`	`/conversations/{id}/messages?beforeSeq=&limit=`	Histórico, paginado por seq
`POST`	`/conversations/{id}/read`	Body `{ "seq": 120 }`
`GET`	`/inbox?limit=`	Lista de conversas e número de mensagens não lidas
`WS`	`/ws`	Server envia eventos `message.new` e `read.updated`

Enviar mensagens passa por REST em vez de WebSocket. Dessa forma, você obtém retry, timeout e idempotência no estilo HTTP familiar. WebSocket é usado apenas para enviar para o cliente.

5. Fluxo de envio

Dependências necessárias:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <scope>runtime</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Toda a operação de envio de mensagem está em uma única transação:

public record SentMessage(long conversationId, long seq, UUID clientMsgId) {}

@Service
public class MessageService {

    private final JdbcTemplate jdbc;

    public MessageService(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    @Transactional
    public SentMessage send(long conversationId, long senderId, UUID clientMsgId, String body) {
        // 1. Lock na linha conversation: todo envio para a mesma conversa é enfileirado
        Long lastSeq = jdbc.queryForObject(
            "SELECT last_seq FROM conversation WHERE id = ? FOR UPDATE",
            Long.class, conversationId);

        // 2. Idempotência: se o cliente retransmitir, retorna o resultado anterior
        List<Long> existing = jdbc.queryForList(
            "SELECT seq FROM message WHERE conversation_id = ? AND client_msg_id = ?",
            Long.class, conversationId, clientMsgId);
        if (!existing.isEmpty()) {
            return new SentMessage(conversationId, existing.getFirst(), clientMsgId);
        }

        // 3. Atribuir novo seq, escrever mensagem e atualizar conversa
        long seq = lastSeq + 1;
        jdbc.update("""
            INSERT INTO message(conversation_id, seq, sender_id, client_msg_id, body)
            VALUES (?, ?, ?, ?, ?)""", conversationId, seq, senderId, clientMsgId, body);
        jdbc.update("UPDATE conversation SET last_seq = ?, last_message_at = now() WHERE id = ?",
            seq, conversationId);

        // 4. O remetente obviamente já leu sua própria mensagem
        jdbc.update("""
            UPDATE conversation_member SET last_read_seq = GREATEST(last_read_seq, ?)
            WHERE conversation_id = ? AND user_id = ?""", seq, conversationId, senderId);

        // 5. Outbox: eventos são escritos na mesma transação, nunca fora de sincronia com os dados
        jdbc.update("INSERT INTO outbox(msg_key, payload) VALUES (?, ?::jsonb)",
            String.valueOf(conversationId),
            """
            {"type":"message.new","conversationId":%d,"seq":%d,"senderId":%d}"""
                .formatted(conversationId, seq, senderId));

        return new SentMessage(conversationId, seq, clientMsgId);
    }
}

Explicação passo a passo:

Passo 1 usa SELECT ... FOR UPDATE. Dois requests enviados para a mesma conversa serão executados sequencialmente, então seq nunca se duplica ou salta. Conversas diferentes não se bloqueiam.
Passo 2 verifica duplicação depois de adquirir o lock. Portanto, duas retransmissões simultâneas do mesmo clientMsgId também não passam. Se verificássemos antes do lock, ambos os requests poderiam ver “não existe”. O segundo request então sofreria uma DuplicateKeyException.
Passo 3 incrementa seq continuamente, sem lacunas. Isso permite que o cliente detecte automaticamente mensagens faltantes (seção 7.3).
Passo 5 implementa outbox transacional. Se você enviasse diretamente para Kafka dentro da transação, poderia ocorrer que Kafka recebesse, mas o DB fizesse rollback, ou vice-versa.

Chamando send duas vezes com o mesmo clientMsgId:

send(1, 10, a1b2..., "Olá grupo") -> SentMessage[conversationId=1, seq=42, ...]
send(1, 10, a1b2..., "Olá grupo") -> SentMessage[conversationId=1, seq=42, ...]   // não cria mensagem duplicada

Um relay separado lê a tabela outbox por id, envia para o topic chat.events com key sendo conversationId, e depois deleta as linhas enviadas. Você também pode usar Debezium para fazer isso. A chave por conversa garante que todos os eventos da mesma conversa vão para a mesma partition e mantêm a ordem.

O consumer envia eventos para os gateways WebSocket:

@Component
public class FanoutListener {

    private final MemberRepository members;   // retorna lista de userId da conversa
    private final GatewayRegistry gateways;   // userId -> sessões online

    public FanoutListener(MemberRepository members, GatewayRegistry gateways) {
        this.members = members;
        this.gateways = gateways;
    }

    @KafkaListener(topics = "chat.events", groupId = "fanout")
    public void onEvent(ConsumerRecord<String, String> record) {
        long conversationId = Long.parseLong(record.key());
        for (long userId : members.findUserIds(conversationId)) {
            gateways.pushIfOnline(userId, record.value());
        }
    }
}

Usuários offline não recebem push, e tudo bem. Quando reconectam, o cliente chama /inbox e carrega as mensagens com seq maior que o último seq que possui.

6. Estado de leitura

Marcar como lido requer apenas um UPDATE monotônico:

@Service
public class ReadStateService {

    private final JdbcTemplate jdbc;

    public ReadStateService(JdbcTemplate jdbc) {
        this.jdbc = jdbc;
    }

    public boolean markRead(long conversationId, long userId, long seq) {
        int updated = jdbc.update("""
            UPDATE conversation_member m
               SET last_read_seq = LEAST(?, c.last_seq)
              FROM conversation c
             WHERE c.id = m.conversation_id
               AND m.conversation_id = ? AND m.user_id = ?
               AND m.last_read_seq < LEAST(?, c.last_seq)""",
            seq, conversationId, userId, seq);
        return updated == 1;   // true se o watermark realmente avançar
    }

    public long unreadCount(long conversationId, long userId) {
        return jdbc.queryForObject("""
            SELECT c.last_seq - m.last_read_seq
              FROM conversation c JOIN conversation_member m ON m.conversation_id = c.id
             WHERE c.id = ? AND m.user_id = ?""", Long.class, conversationId, userId);
    }
}

O UPDATE resolve três problemas ao mesmo tempo:

Concorrência: telefone envia seq=50, laptop envia seq=45 depois. A condição last_read_seq < ... impede que o request antigo puxe o watermark para trás. Você não precisa de lock ou versionamento.
Dados inválidos: LEAST(?, c.last_seq) bloqueia o cliente de enviar seq além da mensagem real.
Redução de carga: quando updated == 0, o watermark não muda, então o evento read.updated não é enviado. O fanout desnecessário é cortado.

Apenas quando markRead retorna true, o serviço escreve um evento read.updated no outbox, similar à seção 5.

7. Desafios e soluções

7.1. Conversas “quentes” e contenção de lock

FOR UPDATE serializa o envio de mensagens em uma conversa. Para chat 1-1 ou grupos pequenos, esse custo é negligenciável. Um grupo de 1.000 pessoas ativo pode ter dezenas de mensagens/segundo na mesma linha. Cada transação segura o lock por apenas alguns milissegundos, então esse nível ainda é viável.

Se uma conversa exceder algumas centenas de mensagens/segundo (como um canal de broadcast), mova a atribuição de seq para Redis usando INCR chat:seq:{conversationId}. Em troca, você deve aceitar lacunas no seq quando a escrita no DB falha. O cliente precisa distinguir entre “lacunas reais” e “mensagens ainda não chegadas”.

Nota: Não use created_at ou IDs globais (Snowflake) para ordenar mensagens em uma conversa. Desvios de relógio entre servidores de alguns milissegundos são suficientes para inverter a ordem de duas mensagens enviadas próximas uma da outra.

7.2. Consistência entre DB, Kafka e cliente

A cadeia de garantias deste design:

DB é a fonte de verdade. Mensagens só são confirmadas para o cliente após o commit.
Outbox garante at-least-once para Kafka. Consumers podem receber duplicatas.
Cliente deduplicar por (conversationId, seq). Mensagens que já existem são ignoradas.

7.3. Detectando mensagens faltantes

Como seq é contínuo, o cliente pode detectar lacunas automaticamente. Suponha que o cliente tenha seq=40 e receba um push seq=43. Ele sabe que as mensagens 41 e 42 estão faltando, então chama GET /messages?afterSeq=40. Você não precisa de um mecanismo de ACK complexo do servidor para cada mensagem.

7.4. Picos de carga

Fanout de grupo grande: uma mensagem em um grupo de 1.000 pessoas cria 1.000 pushes. Para grupos acima do limite (por exemplo, 200 membros), envie apenas para pessoas online, consultando um Redis set online:{conversationId}. O payload contém apenas uma notificação leve {conversationId, seq}, o cliente busca o conteúdo quando necessário.
Tempestade de “lido”: após uma interrupção de rede, milhões de clientes reconectam e enviam read receipts. Clientes devem debounce cerca de 1-2 segundos e enviar apenas o seq máximo. O servidor pode agregar ainda mais em Redis, mantendo apenas o valor máximo por chave {conversationId}:{userId}, depois flush para DB em lotes.
Kafka: partition por conversationId para manter ordem. Quando a carga aumenta, você só precisa aumentar o número de consumers, no máximo igual ao número de partitions.
Armazenamento: 400 milhões de mensagens/dia, cada mensagem cerca de 300 bytes, equivalente a cerca de 120 GB/dia. Faça particionamento da tabela message por conversation_id (hash) e mova mensagens antigas para armazenamento mais barato como Cassandra ou object storage. Consultas de histórico sempre incluem conversation_id, então qualquer modelo funciona.

8. Testando comportamentos principais

Os dois comportamentos mais críticos a testar são: retransmissões não criam duplicatas e o watermark nunca recua. Você deve testar com PostgreSQL real via Testcontainers porque a lógica depende de FOR UPDATE e GREATEST/LEAST.

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-testcontainers</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>postgresql</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <scope>test</scope>
</dependency>

Coloque o DDL da seção 3 em src/main/resources/schema.sql e ative spring.sql.init.mode=always.

@SpringBootTest(properties = {
    "spring.sql.init.mode=always",
    "spring.kafka.listener.auto-startup=false"
})
@Testcontainers
class ChatServiceTest {

    @Container
    @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16-alpine");

    @Autowired MessageService messages;
    @Autowired ReadStateService reads;
    @Autowired JdbcTemplate jdbc;

    @BeforeEach
    void setUp() {
        jdbc.execute("TRUNCATE message, outbox, conversation_member, conversation");
        jdbc.update("INSERT INTO conversation(id, type) VALUES (1, 'GROUP')");
        jdbc.update("INSERT INTO conversation_member(conversation_id, user_id) VALUES (1, 10), (1, 20)");
    }

    @Test
    void resendingSameClientMessageIdDoesNotCreateDuplicate() {
        UUID clientMsgId = UUID.randomUUID();

        SentMessage first = messages.send(1, 10, clientMsgId, "Olá grupo");
        SentMessage retry = messages.send(1, 10, clientMsgId, "Olá grupo");

        assertThat(retry.seq()).isEqualTo(first.seq()).isEqualTo(1L);
        assertThat(jdbc.queryForObject("SELECT count(*) FROM message", Long.class)).isEqualTo(1L);
    }

    @Test
    void readWatermarkNeverMovesBackward() {
        for (int i = 0; i < 5; i++) {
            messages.send(1, 10, UUID.randomUUID(), "mensagem " + i);
        }

        assertThat(reads.markRead(1, 20, 4)).isTrue();
        assertThat(reads.markRead(1, 20, 2)).isFalse();   // request antigo chega atrasado
        assertThat(reads.markRead(1, 20, 99)).isTrue();   // limitado a last_seq = 5

        assertThat(reads.unreadCount(1, 20)).isZero();
        assertThat(reads.unreadCount(1, 10)).isZero();    // remetente lê automaticamente
    }
}

O segundo teste verifica três situações. Watermark avança para 4. Request seq=2 que chega atrasado é ignorado e retorna false. Seq que ultrapassa o real é limitado a 5, então o número de não lidas é 0. A propriedade spring.kafka.listener.auto-startup=false mantém o teste sem tentar conectar ao Kafka.

9. Conclusão

Todo o design pivota em torno de uma decisão: cada conversa tem um seq contínuo. Com seq, você obtém ordem de mensagens, paginação de histórico, detecção de mensagens faltantes e contagem de não lidas com uma simples subtração. O estado de leitura é um watermark monotônico, atualizado com um UPDATE condicional. Isso elimina tanto contenção quanto enormes tabelas de recibos.

O resto é técnica familiar. clientMsgId mais constraint unique para idempotência. Outbox transacional com key Kafka por conversa para consistência e ordem de eventos. Debounce, envio apenas para usuários online e pull ao reconectar ajuda o sistema a sobreviver a picos de carga. Comece com PostgreSQL e FOR UPDATE. Só mova a atribuição de seq para Redis ou altere o armazenamento quando dados reais indicarem que você realmente precisa.