Thiết kế hệ thống chat 1-1 và nhóm: lưu tin nhắn và trạng thái đã đọc

1. Tổng quan

Chat trông đơn giản: gửi một dòng chữ rồi hiển thị nó. Nhưng khi có hàng triệu người dùng, bạn phải trả lời những câu hỏi khó hơn. Làm sao giữ đúng thứ tự tin nhắn? Làm sao để gửi lại không tạo tin trùng? Làm sao biết ai đã đọc tới đâu mà không ghi hàng tỷ bản ghi “đã xem”?

Bài này đi qua một thiết kế hoàn chỉnh cho chat 1-1 và chat nhóm. Thiết kế gồm yêu cầu, mô hình dữ liệu, API, luồng xử lý và các điểm khó. Đọc xong, bạn sẽ dựng được phần lõi lưu tin nhắn và trạng thái đã đọc bằng Spring Boot 3, PostgreSQL và Kafka.

2. Yêu cầu

2.1. Chức năng

• Chat 1-1 và chat nhóm (tối đa 1.000 thành viên).
• Gửi tin văn bản. Tin hiển thị theo đúng thứ tự trong một cuộc hội thoại.
• Xem lịch sử, cuộn ngược theo trang.
• Danh sách hội thoại (inbox) kèm số tin chưa đọc.
• Trạng thái đã đọc: chat 1-1 hiện “Đã xem”, chat nhóm hiện “Đã xem bởi N người”.

2.2. Quy mô và ràng buộc

• 10 triệu DAU. Trung bình mỗi người gửi 40 tin/ngày, tức 400 triệu tin/ngày, khoảng 4.600 tin/giây. Giờ cao điểm có thể lên khoảng 30.000 tin/giây.
• Không được mất tin đã được xác nhận gửi thành công.
• Thứ tự chỉ cần đúng trong một hội thoại. Không cần thứ tự toàn cục.
• Client có thể gửi lại do mạng chập chờn. Server phải idempotent.
• Sự kiện “đã đọc” nhiều hơn sự kiện gửi tin nhiều lần, vì mỗi lần mở app đều phát sinh.

Ràng buộc thứ ba là chìa khoá của cả thiết kế. Chỉ cần thứ tự cục bộ, nên mọi thứ có thể phân mảnh theo conversation_id.

3. Mô hình dữ liệu

Ý tưởng trung tâm: mỗi hội thoại có một số thứ tự tăng dần (seq). Tin nhắn thứ N trong hội thoại có seq = N. Trạng thái đã đọc của một thành viên chỉ là một con số, last_read_seq. Nó là một watermark, không phải danh sách từng tin đã đọc.

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

Các điểm đáng chú ý:

• Khoá chính (conversation_id, seq) giúp đọc lịch sử chỉ cần quét một dải liên tục của index. Khoá này cũng là shard key tự nhiên khi bạn phân mảnh.
• UNIQUE (conversation_id, client_msg_id) là lớp chặn tin trùng ở tầng database.
• Số tin chưa đọc là last_seq - last_read_seq. Phép tính này không cần COUNT(*).
• “Đã xem bởi” trong nhóm được tính bằng số thành viên có last_read_seq >= seq. Bạn không cần bảng message_receipt với hàng tỷ dòng.
• Với chat 1-1, thêm bảng direct_pair(user_low, user_high, conversation_id) có ràng buộc unique. Hai người mở chat cùng lúc sẽ không tạo ra hai hội thoại.

Lưu ý: Watermark giả định người dùng đọc tuần tự, tức đọc tin N thì coi như đã đọc mọi tin trước đó. Giả định này đúng với gần như mọi ứng dụng chat. Nếu bạn cần biết chính xác từng tin cụ thể đã được mở (ví dụ tin nhắn tự huỷ), hãy lưu thêm bản ghi riêng cho loại tin đó. Đừng áp dụng cho mọi tin.

4. API

Method	Endpoint	Mô tả
POST	/conversations/{id}/messages	Gửi tin. Body gồm clientMsgId và body
GET	/conversations/{id}/messages?beforeSeq=&limit=	Lịch sử, phân trang theo seq
POST	/conversations/{id}/read	Body { "seq": 120 }
GET	/inbox?limit=	Danh sách hội thoại và số tin chưa đọc
WS	/ws	Server đẩy sự kiện message.new và read.updated

Gửi tin đi qua REST thay vì WebSocket. Cách này cho bạn retry, timeout và idempotency theo kiểu HTTP quen thuộc. WebSocket chỉ dùng để đẩy xuống client.

5. Luồng gửi tin

Thư viện cần dùng:

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

Toàn bộ thao tác gửi tin nằm trong một 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. Khoá dòng conversation: mọi lần gửi vào cùng hội thoại được xếp hàng
        Long lastSeq = jdbc.queryForObject(
            "SELECT last_seq FROM conversation WHERE id = ? FOR UPDATE",
            Long.class, conversationId);

        // 2. Idempotency: client gửi lại thì trả về kết quả cũ
        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. Cấp seq mới, ghi tin và cập nhật hội thoại
        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. Người gửi hiển nhiên đã đọc tin của chính mình
        jdbc.update("""
            UPDATE conversation_member SET last_read_seq = GREATEST(last_read_seq, ?)
            WHERE conversation_id = ? AND user_id = ?""", seq, conversationId, senderId);

        // 5. Outbox: sự kiện được ghi cùng transaction, không bao giờ lệch với dữ liệu
        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);
    }
}

Giải thích từng bước:

• Bước 1 dùng SELECT ... FOR UPDATE. Hai request gửi vào cùng hội thoại sẽ chạy tuần tự, nên seq không bao giờ trùng hay nhảy cóc. Các hội thoại khác nhau không chặn nhau.
• Bước 2 kiểm tra trùng sau khi đã giữ lock. Vì vậy hai lần retry đồng thời của cùng clientMsgId cũng không lọt qua được. Nếu kiểm tra trước khi lock, cả hai request có thể cùng thấy “chưa có”. Request thứ hai khi đó sẽ dính DuplicateKeyException.
• Bước 3 tăng seq liên tục, không có khoảng trống. Nhờ đó client có thể tự phát hiện tin bị thiếu (mục 7.3).
• Bước 5 áp dụng transactional outbox. Nếu bạn gửi thẳng lên Kafka trong transaction, có thể xảy ra trường hợp Kafka đã nhận nhưng DB rollback, hoặc ngược lại.

Gọi send hai lần với cùng 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, ...]   // không tạo tin mới

Một relay riêng đọc bảng outbox theo id, đẩy lên topic chat.events với key là conversationId, rồi xoá các dòng đã gửi. Bạn cũng có thể dùng Debezium để làm việc này. Key theo hội thoại đảm bảo mọi sự kiện của cùng hội thoại vào cùng partition và giữ thứ tự.

Phía consumer đẩy sự kiện xuống các gateway WebSocket:

@Component
public class FanoutListener {

    private final MemberRepository members;   // trả về danh sách userId của hội thoại
    private final GatewayRegistry gateways;   // userId -> các session đang 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());
        }
    }
}

Người dùng offline không nhận được push, và điều đó không sao. Khi kết nối lại, client gọi /inbox và tự tải các tin có seq lớn hơn seq cuối cùng nó đang giữ.

6. Trạng thái đã đọc

Đánh dấu đã đọc chỉ cần một câu UPDATE đơn điệu:

@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 nếu watermark thực sự tiến lên
    }

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

Câu UPDATE giải quyết ba vấn đề cùng lúc:

• Tranh chấp: điện thoại gửi seq=50, laptop gửi seq=45 đến sau. Điều kiện last_read_seq < ... khiến request cũ không kéo watermark lùi lại. Bạn không cần lock hay version.
• Dữ liệu rác: LEAST(?, c.last_seq) chặn client gửi seq vượt quá tin thực tế.
• Giảm tải: khi updated == 0, watermark không đổi nên không phát sự kiện read.updated. Fanout vô ích được cắt bớt.

Chỉ khi markRead trả về true, service mới ghi một sự kiện read.updated vào outbox, tương tự mục 5.

7. Các điểm khó và cách giải

7.1. Hội thoại “nóng” và tranh chấp lock

FOR UPDATE tuần tự hoá việc gửi tin trong một hội thoại. Với chat 1-1 hoặc nhóm nhỏ, chi phí này không đáng kể. Nhóm 1.000 người đang sôi nổi có thể lên vài chục tin/giây vào cùng một dòng. Mỗi transaction chỉ giữ lock vài mili-giây, nên mức này vẫn chịu được.

Nếu một hội thoại vượt quá vài trăm tin/giây (kiểu kênh broadcast), hãy tách bộ cấp seq ra Redis bằng INCR chat:seq:{conversationId}. Đổi lại, bạn phải chấp nhận seq có khoảng trống khi ghi DB thất bại. Client cần phân biệt “khoảng trống thật” với “tin chưa tới”.

Lưu ý: Đừng dùng created_at hay ID toàn cục (Snowflake) để sắp xếp tin trong hội thoại. Đồng hồ giữa các server lệch nhau vài mili-giây cũng đủ đảo thứ tự hai tin gửi sát nhau.

7.2. Nhất quán giữa DB, Kafka và client

Chuỗi đảm bảo của thiết kế này:

1. DB là nguồn sự thật. Tin chỉ được xác nhận cho client sau khi commit.
2. Outbox đảm bảo at-least-once lên Kafka. Consumer có thể nhận trùng.
3. Client khử trùng theo (conversationId, seq). Tin nào đã có thì bỏ qua.

7.3. Phát hiện tin bị thiếu

Seq liên tục nên client tự phát hiện được lỗ hổng. Giả sử client đang có seq=40 và nhận push seq=43. Nó biết thiếu tin 41 và 42, nên gọi GET /messages?afterSeq=40. Bạn không cần cơ chế ACK phức tạp từ server cho từng tin.

7.4. Tải đột biến

• Fanout nhóm lớn: một tin vào nhóm 1.000 người tạo ra 1.000 lần push. Với nhóm trên ngưỡng (ví dụ 200 thành viên), chỉ đẩy tới người đang online, tra trong Redis set online:{conversationId}. Payload chỉ chứa thông báo nhẹ {conversationId, seq}, client tự kéo nội dung về khi cần.
• Bão “đã đọc”: sau sự cố mạng, hàng triệu client kết nối lại và cùng gửi read receipt. Client nên debounce khoảng 1–2 giây và chỉ gửi seq lớn nhất. Server có thể gộp tiếp trong Redis, chỉ giữ giá trị max theo key {conversationId}:{userId}, rồi flush xuống DB theo lô.
• Kafka: partition theo conversationId để giữ thứ tự. Khi tải tăng, bạn chỉ cần tăng số consumer, tối đa bằng số partition.
• Lưu trữ: 400 triệu tin/ngày, mỗi tin khoảng 300 byte, tương đương khoảng 120 GB/ngày. Hãy partition bảng message theo conversation_id (hash) và chuyển tin cũ sang kho rẻ hơn như Cassandra hoặc object storage. Truy vấn lịch sử luôn kèm conversation_id, nên mô hình nào cũng hợp.

8. Kiểm thử hành vi chính

Hai hành vi quan trọng nhất cần test: gửi lại không tạo tin trùng, và watermark không bao giờ lùi. Bạn nên test với PostgreSQL thật qua Testcontainers vì logic phụ thuộc vào FOR UPDATE và 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>

Đặt DDL ở mục 3 vào src/main/resources/schema.sql và bật 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();   // request cũ đến muộn
        assertThat(reads.markRead(1, 20, 99)).isTrue();   // bị chặn về last_seq = 5

        assertThat(reads.unreadCount(1, 20)).isZero();
        assertThat(reads.unreadCount(1, 10)).isZero();    // người gửi tự động đã đọc
    }
}

Test thứ hai kiểm tra ba tình huống. Watermark tiến lên 4. Request seq=2 đến muộn bị bỏ qua và trả về false. Seq vượt quá thực tế bị giới hạn về 5, nên số tin chưa đọc bằng 0. Thuộc tính spring.kafka.listener.auto-startup=false giữ cho test không cố kết nối tới Kafka.

9. Kết luận

Toàn bộ thiết kế xoay quanh một quyết định: mỗi hội thoại có một seq liên tục. Từ seq, bạn có thứ tự tin nhắn, phân trang lịch sử, phát hiện tin thiếu và số tin chưa đọc chỉ bằng một phép trừ. Trạng thái đã đọc là một watermark đơn điệu, cập nhật bằng một câu UPDATE có điều kiện. Cách này loại bỏ được cả tranh chấp lẫn bảng receipt khổng lồ.

Phần còn lại là kỹ thuật quen thuộc. clientMsgId cộng ràng buộc unique cho idempotency. Transactional outbox với Kafka key theo hội thoại cho nhất quán và thứ tự sự kiện. Debounce, chỉ đẩy tới người đang online và pull khi kết nối lại giúp hệ thống sống sót qua tải đột biến. Hãy bắt đầu với PostgreSQL và FOR UPDATE. Chỉ tách bộ cấp seq ra Redis hoặc chuyển kho lưu trữ khi số liệu thực tế cho thấy bạn thật sự cần.