Conception d'un système de chat 1-1 et groupe : messages et statuts de lecture

1. Vue d’ensemble

Le chat semble simple : envoyer une ligne de texte et l’afficher. Mais quand il y a des millions d’utilisateurs, il faut répondre à des questions plus difficiles. Comment préserver l’ordre des messages ? Comment renvoyer sans créer de doublons ? Comment savoir qui a lu jusqu’où sans enregistrer des milliards de reçus « vus » ?

Cet article couvre une conception complète pour le chat 1-1 et de groupe. La conception inclut les exigences, le modèle de données, les API, les flux de traitement et les défis. Après la lecture, vous pourrez construire le noyau de stockage des messages et des statuts de lecture avec Spring Boot 3, PostgreSQL et Kafka.

2. Exigences

2.1. Fonctionnalités

• Chat 1-1 et chat de groupe (jusqu’à 1 000 membres).
• Envoyer des messages texte. Les messages s’affichent dans le bon ordre dans une conversation.
• Afficher l’historique, défiler en arrière par page.
• Liste des conversations (inbox) avec le nombre de messages non lus.
• Statuts de lecture : chat 1-1 affiche « Vu », chat de groupe affiche « Vu par N personnes ».

2.2. Échelle et contraintes

• 10 millions d’utilisateurs actifs quotidiens. En moyenne, chaque personne envoie 40 messages/jour, soit 400 millions de messages/jour, environ 4 600 messages/seconde. Les heures de pointe peuvent atteindre environ 30 000 messages/seconde.
• Aucun message confirmé comme envoyé avec succès ne doit être perdu.
• L’ordre ne doit être cohérent que dans une conversation. Pas besoin d’ordre global.
• Le client peut renvoyer en raison de problèmes réseau. Le serveur doit être idempotent.
• Les événements « lu » sont plus nombreux que les événements d’envoi de messages, car chaque ouverture d’application en génère un.

La troisième contrainte est la clé de toute la conception. Seul l’ordre local est requis, donc tout peut être partitionné par conversation_id.

3. Modèle de données

L’idée centrale : chaque conversation a un numéro de séquence croissant (seq). Le message N dans une conversation a seq = N. Le statut de lecture d’un membre est juste un nombre, last_read_seq. C’est un watermark, pas une liste de chaque message lu.

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()
);

Points à noter :

• La clé primaire (conversation_id, seq) permet de lire l’historique en balayant juste une plage contiguë de l’index. Cette clé est aussi la clé de partitionnement naturelle lorsque vous fragmentez.
• UNIQUE (conversation_id, client_msg_id) est une couche de protection contre les doublons au niveau de la base de données.
• Le nombre de messages non lus est last_seq - last_read_seq. Ce calcul n’a pas besoin de COUNT(*).
• « Vu par » dans un groupe est calculé comme le nombre de membres ayant last_read_seq >= seq. Vous n’avez pas besoin d’une table message_receipt avec des milliards de lignes.
• Pour le chat 1-1, ajoutez une table direct_pair(user_low, user_high, conversation_id) avec une contrainte unique. Deux personnes ouvrant le chat en même temps ne créeront pas deux conversations.

Note : Le watermark suppose que les utilisateurs lisent séquentiellement, c’est-à-dire que lire le message N signifie avoir lu tous les messages avant. Cette hypothèse est valable pour presque toutes les applications de chat. Si vous avez besoin de savoir exactement quel message spécifique a été ouvert (par exemple, les messages qui s’autodétruisent), enregistrez un enregistrement séparé pour ce type de message. Ne l’appliquez pas à tous les messages.

4. API

Méthode	Point de terminaison	Description
POST	/conversations/{id}/messages	Envoyer un message. Le corps contient clientMsgId et body
GET	/conversations/{id}/messages?beforeSeq=&limit=	Historique, pagination par seq
POST	/conversations/{id}/read	Corps { "seq": 120 }
GET	/inbox?limit=	Liste des conversations et nombre de messages non lus
WS	/ws	Le serveur pousse les événements message.new et read.updated

Les messages sont envoyés via REST au lieu de WebSocket. Cette approche vous donne les retries, timeouts et idempotence familiers du HTTP. WebSocket est utilisé uniquement pour pousser vers le client.

5. Flux d’envoi de message

Dépendances nécessaires :

<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>

Toute l’opération d’envoi de message se fait dans une seule transaction :

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. Verrouiller la ligne conversation : tous les envois à la même conversation sont mis en file
        Long lastSeq = jdbc.queryForObject(
            "SELECT last_seq FROM conversation WHERE id = ? FOR UPDATE",
            Long.class, conversationId);

        // 2. Idempotence : si le client renvoie, retourner le résultat précédent
        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. Attribuer un seq nouveau, enregistrer le message et mettre à jour la conversation
        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. L'expéditeur a évidemment lu son propre message
        jdbc.update("""
            UPDATE conversation_member SET last_read_seq = GREATEST(last_read_seq, ?)
            WHERE conversation_id = ? AND user_id = ?""", seq, conversationId, senderId);

        // 5. Outbox : l'événement est enregistré dans la même transaction, jamais hors de sync avec les données
        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);
    }
}

Explication de chaque étape :

• L’étape 1 utilise SELECT ... FOR UPDATE. Deux requêtes d’envoi à la même conversation s’exécutent séquentiellement, donc seq n’est jamais dupliqué ou sauté. Les conversations différentes ne se bloquent pas mutuellement.
• L’étape 2 vérifie les doublons après avoir acquis le verrou. Ainsi, deux tentatives de relance simultanées du même clientMsgId ne peuvent pas contourner le contrôle. Si vous vérifiez avant le verrou, les deux requêtes pourraient voir « pas encore existant ». La seconde requête déclencherait alors DuplicateKeyException.
• L’étape 3 incrémente seq continuellement, sans lacunes. Grâce à cela, le client peut détecter automatiquement les messages manquants (section 7.3).
• L’étape 5 applique la boîte aux lettres transactionnelle. Si vous envoyez directement à Kafka dans une transaction, il peut arriver que Kafka reçoive le message mais que la BD fasse un rollback, ou l’inverse.

Appeler send deux fois avec le même clientMsgId :

send(1, 10, a1b2..., "Chào nhóm") -> SentMessage[conversationId=1, seq=42, ...]
send(1, 10, a1b2..., "Chào nhóm") -> SentMessage[conversationId=1, seq=42, ...]   // aucun nouveau message créé

Un relais séparé lit la table outbox par id, l’envoie au sujet chat.events avec la clé conversationId, puis supprime les lignes envoyées. Vous pouvez aussi utiliser Debezium pour faire cela. Utiliser la conversation comme clé garantit que tous les événements de la même conversation vont à la même partition et préservent l’ordre.

Le côté consumer pousse les événements vers les passerelles WebSocket :

@Component
public class FanoutListener {

    private final MemberRepository members;   // retourne la liste des userId d'une conversation
    private final GatewayRegistry gateways;   // userId -> les sessions actuellement en ligne

    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());
        }
    }
}

Les utilisateurs hors ligne ne reçoivent pas de push, et c’est normal. Lors de la reconnexion, le client appelle /inbox et charge automatiquement les messages avec seq plus grand que le dernier seq qu’il possède.

6. Statut de lecture

Marquer comme lu nécessite juste une simple UPDATE monotone :

@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 si le watermark progresse vraiment
    }

    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);
    }
}

La requête UPDATE résout trois problèmes à la fois :

• Concurrence : le téléphone envoie seq=50, l’ordinateur portable envoie seq=45 après. La condition last_read_seq < ... empêche la requête ancienne de faire reculer le watermark. Vous n’avez besoin ni de verrou ni de version.
• Données erronées : LEAST(?, c.last_seq) empêche le client d’envoyer un seq dépassant les messages réels.
• Réduction de la charge : quand updated == 0, le watermark n’a pas changé, donc aucun événement read.updated n’est publié. Le fanout inutile est supprimé.

Seulement si markRead retourne true, le service enregistre un événement read.updated dans la boîte aux lettres, similaire à la section 5.

7. Défis et solutions

7.1. Conversations « chaudes » et contention de verrou

FOR UPDATE sérialise les envois de messages dans une conversation. Pour un chat 1-1 ou un petit groupe, ce coût est négligeable. Un groupe de 1 000 personnes très actif peut atteindre quelques dizaines de messages/seconde dans une seule ligne. Chaque transaction ne détient le verrou que quelques millisecondes, donc ce niveau est tolérable.

Si une conversation dépasse quelques centaines de messages/seconde (comme un canal de diffusion), séparez l’attribution de seq dans Redis en utilisant INCR chat:seq:{conversationId}. En contrepartie, vous devez accepter que seq ait des lacunes lors de l’écriture en BD. Le client doit distinguer les « vraies lacunes » des « messages pas encore arrivés ».

Note : N’utilisez pas created_at ou d’ID globaux (Snowflake) pour trier les messages dans une conversation. Même un décalage d’horloge de quelques millisecondes entre les serveurs est suffisant pour inverser l’ordre de deux messages envoyés l’un après l’autre.

7.2. Cohérence entre BD, Kafka et client

La chaîne de garanties de cette conception :

1. La BD est la source de vérité. Les messages ne sont confirmés au client qu’après commit.
2. La boîte aux lettres garantit at-least-once vers Kafka. Le consumer peut recevoir des doublons.
3. Le client déduplique selon (conversationId, seq). Tout message déjà présent est ignoré.

7.3. Détection de messages manquants

Puisque seq est continu, le client peut détecter les lacunes. Supposons que le client ait seq=40 et reçoit un push pour seq=43. Il sait qu’il manque les messages 41 et 42, donc il appelle GET /messages?afterSeq=40. Vous n’avez pas besoin d’un mécanisme ACK complexe du serveur pour chaque message.

7.4. Pics de charge

• Fanout pour les grands groupes : un message entrant dans un groupe de 1 000 personnes crée 1 000 pushes. Pour les groupes au-delà d’un seuil (par exemple 200 membres), ne poussez que vers les gens en ligne, vérifiez dans un ensemble Redis online:{conversationId}. Le payload contient juste une notification légère {conversationId, seq}, le client récupère le contenu quand il en a besoin.
• Déluge de « lu »: après une panne réseau, des millions de clients se reconnectent et envoient tous un reçu de lecture. Le client devrait debouncer pendant 1-2 secondes et n’envoyer que le seq maximum. Le serveur peut agréger davantage dans Redis, en gardant juste la valeur max par clé {conversationId}:{userId}, puis vider en BD par lots.
• Kafka : partitionnez par conversationId pour préserver l’ordre. Quand la charge augmente, vous avez juste besoin d’augmenter le nombre de consumers, jusqu’à un maximum égal au nombre de partitions.
• Stockage : 400 millions de messages/jour, chaque message environ 300 octets, soit environ 120 GB/jour. Partitionnez la table message par conversation_id (hash) et déplacez les anciens messages vers un stockage moins cher comme Cassandra ou object storage. Les requêtes d’historique incluent toujours conversation_id, donc n’importe quel modèle convient.

8. Test des comportements clés

Les deux comportements les plus importants à tester : envoyer à nouveau ne crée pas de doublons, et le watermark ne recule jamais. Vous devriez tester avec PostgreSQL réel via Testcontainers car la logique dépend de FOR UPDATE et 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>

Mettez le DDL de la section 3 dans src/main/resources/schema.sql et activez 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, "Chào nhóm");
        SentMessage retry = messages.send(1, 10, clientMsgId, "Chào nhóm");

        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(), "tin " + i);
        }

        assertThat(reads.markRead(1, 20, 4)).isTrue();
        assertThat(reads.markRead(1, 20, 2)).isFalse();   // la requête ancienne arrive tard
        assertThat(reads.markRead(1, 20, 99)).isTrue();   // limité à last_seq = 5

        assertThat(reads.unreadCount(1, 20)).isZero();
        assertThat(reads.unreadCount(1, 10)).isZero();    // l'expéditeur a automatiquement lu
    }
}

Le deuxième test vérifie trois situations. Le watermark avance à 4. La requête seq=2 arrive tard et est ignorée, retournant false. Le seq dépassant le réel est limité à 5, donc le nombre de messages non lus est 0. La propriété spring.kafka.listener.auto-startup=false empêche le test de tenter de se connecter à Kafka.

9. Conclusion

Toute la conception tourne autour d’une décision : chaque conversation a un seq continu. À partir de seq, vous obtenez l’ordre des messages, la pagination de l’historique, la détection de messages manquants et le nombre de messages non lus avec juste une soustraction. Le statut de lecture est un watermark monotone, mis à jour avec une UPDATE conditionnelle. Cette approche élimine à la fois la contention et les tables de reçus gigantesques.

Le reste est une technique familière. clientMsgId plus contrainte unique pour l’idempotence. Boîte aux lettres transactionnelle avec Kafka partitionnée par conversation pour la cohérence et l’ordre des événements. Debouncer, ne pousser que vers les gens en ligne et tirer lors de la reconnexion aide le système à survivre aux pics de charge. Commencez avec PostgreSQL et FOR UPDATE. Séparez seulement l’allocation de seq vers Redis ou changez le stockage quand les chiffres réels montrent que vous en avez vraiment besoin.