1-1 및 그룹 채팅 시스템 설계: 메시지 저장 및 읽음 상태

1. 개요

채팅은 간단해 보입니다. 텍스트 한 줄을 보내면 화면에 표시됩니다. 하지만 수백만 명의 사용자가 있을 때는 더 어려운 질문들에 답해야 합니다. 메시지의 순서를 어떻게 유지할까요? 재전송이 중복 메시지를 만들지 않게 하려면? 누가 어디까지 읽었는지 알기 위해 수십억 개의 “본 메시지” 기록을 저장하지 않으면서?

이 글은 1-1 채팅과 그룹 채팅에 대한 완전한 설계를 다룹니다. 요구 사항, 데이터 모델, API, 처리 흐름 및 어려운 부분들을 포함합니다. 이 글을 읽으면 Spring Boot 3, PostgreSQL, Kafka를 사용해 메시지 저장 및 읽음 상태의 핵심 부분을 구축할 수 있습니다.

2. 요구 사항

2.1. 기능

• 1-1 채팅 및 그룹 채팅(최대 1,000명 멤버).
• 텍스트 메시지 전송. 메시지는 한 대화 내에서 올바른 순서로 표시됨.
• 히스토리 보기, 역방향 스크롤(페이지 단위).
• 대화 목록(받은편지함) 미읽음 메시지 수 포함.
• 읽음 상태: 1-1 채팅에서는 “본 메시지” 표시, 그룹 채팅에서는 “N명이 봤음” 표시.

2.2. 규모 및 제약

• 일일 활성 사용자 1,000만 명. 평균적으로 각 사용자는 하루에 40개 메시지를 보내므로, 하루에 4억 개 메시지, 초당 약 4,600개 메시지. 피크 시간대는 초당 약 30,000개 메시지까지 올라갈 수 있습니다.
• 성공적으로 전송됨이 확인된 메시지를 잃으면 안 됩니다.
• 순서는 한 대화 내에서만 정확하면 됩니다. 전역 순서가 필요 없습니다.
• 클라이언트가 네트워크 문제로 재전송할 수 있습니다. 서버는 멱등성이어야 합니다.
• “읽음” 이벤트가 메시지 전송보다 훨씬 많습니다. 앱을 열 때마다 발생하기 때문입니다.

세 번째 제약이 전체 설계의 핵심입니다. 지역적 순서만 필요하므로, 모든 것을 conversation_id로 샤딩할 수 있습니다.

3. 데이터 모델

핵심 아이디어: 각 대화마다 **증가하는 순서 번호(seq)**가 있습니다. 대화의 N번째 메시지는 seq = N을 가집니다. 한 멤버의 읽음 상태는 단 하나의 숫자, last_read_seq입니다. 이것은 워터마크이지, 각각의 읽은 메시지 목록이 아닙니다.

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

주목할 점:

• 기본 키 (conversation_id, seq)를 사용하면 히스토리를 읽을 때 인덱스의 연속된 범위만 스캔하면 됩니다. 이 키는 샤딩할 때의 자연스러운 shard key이기도 합니다.
• UNIQUE (conversation_id, client_msg_id)는 데이터베이스 계층에서 중복 메시지를 차단합니다.
• 미읽음 메시지 수는 last_seq - last_read_seq입니다. COUNT(*)가 필요 없습니다.
• 그룹에서 “본 메시지”는 last_read_seq >= seq인 멤버 수로 계산됩니다. 수십억 행의 message_receipt 테이블이 필요 없습니다.
• 1-1 채팅의 경우 direct_pair(user_low, user_high, conversation_id)에 unique 제약이 있는 테이블을 추가합니다. 두 사람이 동시에 채팅을 열어도 두 개의 대화가 생기지 않습니다.

주의: 워터마크는 사용자가 순차적으로 읽는다고 가정합니다. 즉, N번 메시지를 읽으면 그 이전의 모든 메시지를 읽은 것으로 간주합니다. 이 가정은 거의 모든 채팅 애플리케이션에서 맞습니다. 특정 메시지가 정확히 열렸는지 알아야 한다면(예: 임시 메시지), 그 유형의 메시지에만 별도의 기록을 저장하세요. 모든 메시지에 적용하지 마세요.

4. API

메서드	엔드포인트	설명
POST	/conversations/{id}/messages	메시지 전송. Body는 clientMsgId와 body 포함
GET	/conversations/{id}/messages?beforeSeq=&limit=	히스토리, seq로 페이지 나누기
POST	/conversations/{id}/read	Body { "seq": 120 }
GET	/inbox?limit=	대화 목록과 미읽음 메시지 수
WS	/ws	서버가 message.new와 read.updated 이벤트 푸시

메시지 전송은 WebSocket 대신 REST를 통해 이루어집니다. 이 방식은 재시도, 타임아웃 및 HTTP 형식의 친숙한 멱등성을 제공합니다. WebSocket은 클라이언트에 푸시하는 용도로만 사용됩니다.

5. 메시지 전송 흐름

필요한 라이브러리:

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

전체 메시지 전송 작업은 하나의 트랜잭션 내에 있습니다:

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. conversation 행 잠금: 같은 대화로 가는 모든 전송이 순차 처리됨
        Long lastSeq = jdbc.queryForObject(
            "SELECT last_seq FROM conversation WHERE id = ? FOR UPDATE",
            Long.class, conversationId);

        // 2. 멱등성: 클라이언트가 재전송하면 이전 결과 반환
        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. 새 seq 할당, 메시지 저장 및 대화 업데이트
        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. 발신자는 자신의 메시지를 당연히 읽은 것으로 표시
        jdbc.update("""
            UPDATE conversation_member SET last_read_seq = GREATEST(last_read_seq, ?)
            WHERE conversation_id = ? AND user_id = ?""", seq, conversationId, senderId);

        // 5. Outbox: 이벤트가 트랜잭션과 함께 기록되며, 데이터와 절대 불일치하지 않음
        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);
    }
}

각 단계 설명:

• 단계 1은 SELECT ... FOR UPDATE를 사용합니다. 같은 대화로 가는 두 요청이 순차적으로 실행되므로 seq가 절대 중복되거나 건너뛰지 않습니다. 다른 대화는 서로 차단하지 않습니다.
• 단계 2는 잠금을 획득한 후 중복을 확인합니다. 따라서 같은 clientMsgId의 두 재시도도 통과할 수 없습니다. 잠금 전에 확인했다면, 두 요청 모두 “아직 없음”을 볼 수 있습니다. 그러면 두 번째 요청이 DuplicateKeyException에 걸립니다.
• 단계 3은 seq를 연속적으로 증가시키며, 간격이 없습니다. 덕분에 클라이언트는 누락된 메시지를 자동으로 감지할 수 있습니다(섹션 7.3 참조).
• 단계 5는 transactional outbox를 적용합니다. 트랜잭션 내에서 직접 Kafka로 전송하면, Kafka가 받았지만 DB가 롤백되거나 그 반대의 경우가 발생할 수 있습니다.

같은 clientMsgId로 send를 두 번 호출하면:

send(1, 10, a1b2..., "그룹에 인사합니다") -> SentMessage[conversationId=1, seq=42, ...]
send(1, 10, a1b2..., "그룹에 인사합니다") -> SentMessage[conversationId=1, seq=42, ...]   // 새 메시지를 만들지 않음

별도의 relay가 outbox 테이블을 id로 읽어서 chat.events 토픽에 푸시한 후, 처리된 행을 삭제합니다. 또는 Debezium을 사용할 수도 있습니다. conversationId를 키로 사용하면 같은 대화의 모든 이벤트가 같은 파티션으로 가고 순서가 유지됩니다.

Consumer 측은 이벤트를 WebSocket 게이트웨이들에 푸시합니다:

@Component
public class FanoutListener {

    private final MemberRepository members;   // 대화의 userId 목록 반환
    private final GatewayRegistry gateways;   // userId -> 온라인 세션들

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

오프라인 사용자는 푸시를 받지 않으며, 문제없습니다. 다시 연결할 때, 클라이언트는 /inbox를 호출하고 자신이 가진 마지막 seq보다 큰 메시지들을 자동으로 로드합니다.

6. 읽음 상태

읽음 표시는 단지 하나의 단조 업데이트 쿼리입니다:

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

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

UPDATE 쿼리는 동시에 세 가지 문제를 해결합니다:

• 경합: 휴대폰이 seq=50을 보내고, 노트북이 seq=45를 나중에 보냅니다. last_read_seq < ... 조건이 오래된 요청이 워터마크를 뒤로 끌어당기는 것을 방지합니다. 잠금이나 버전 관리가 필요 없습니다.
• 잘못된 데이터: LEAST(?, c.last_seq)는 클라이언트가 실제 메시지를 초과하는 seq를 보내는 것을 방지합니다.
• 부하 감소: updated == 0일 때, 워터마크가 변하지 않으므로 read.updated 이벤트를 발행하지 않습니다. 불필요한 fanout이 줄어듭니다.

오직 markRead이 true를 반환할 때만, 서비스가 섹션 5처럼 outbox에 read.updated 이벤트를 기록합니다.

7. 어려운 부분과 해결책

7.1. “핫” 대화와 잠금 경합

FOR UPDATE는 한 대화 내에서 메시지 전송을 순차 처리합니다. 1-1 채팅 또는 작은 그룹에서는 이 비용이 무시할 수 있습니다. 1,000명이 열심히 참여하는 그룹이라도 초당 수십 개 메시지 정도입니다. 각 트랜잭션은 잠금을 단 몇 밀리초만 보유하므로, 이 수준은 충분히 감당할 수 있습니다.

한 대화가 초당 수백 개 메시지를 넘어간다면(예: 방송 채널처럼), seq 발급을 Redis의 INCR chat:seq:{conversationId}로 분리하세요. 대신, 데이터베이스 쓰기에 실패할 때 seq에 간격이 있을 수 있습니다. 클라이언트는 “실제 간격”과 “아직 오지 않은 메시지”를 구분해야 합니다.

주의: 대화 내 메시지 정렬에 created_at이나 전역 ID(Snowflake)를 사용하지 마세요. 서버 간의 시계 차이가 몇 밀리초라도 같은 시간에 보낸 두 메시지의 순서를 뒤바꿀 수 있습니다.

7.2. 데이터베이스, Kafka 및 클라이언트 간 일관성

이 설계의 일관성 보증 체인:

1. 데이터베이스가 진실의 원천입니다. 메시지는 커밋 후에만 클라이언트에게 확인됩니다.
2. Outbox는 Kafka로의 최소 한 번 전달을 보장합니다. Consumer는 중복을 받을 수 있습니다.
3. 클라이언트는 (conversationId, seq)에 따라 중복을 제거합니다. 이미 있는 메시지는 건너뜁니다.

7.3. 누락된 메시지 감지

Seq가 연속이므로 클라이언트가 간격을 자동으로 감지할 수 있습니다. 클라이언트가 seq=40을 가지고 있고 push에서 seq=43을 받으면, 41과 42가 누락되었음을 압니다. 그래서 GET /messages?afterSeq=40을 호출합니다. 복잡한 ACK 메커니즘이 필요 없습니다.

7.4. 급증하는 부하

• 큰 그룹의 fanout: 1,000명 그룹에 한 메시지가 1,000번의 푸시를 생성합니다. 임계값을 초과하는 그룹(예: 200명 이상)의 경우, 온라인 사용자에게만 푸시합니다. Redis set online:{conversationId}에서 조회합니다. Payload는 가벼운 알림 {conversationId, seq}만 포함하며, 클라이언트가 필요할 때 콘텐츠를 직접 가져옵니다.
• “읽음” 폭주: 네트워크 장애 후, 수백만 명의 클라이언트가 다시 연결하고 동시에 읽음 수신을 보냅니다. 클라이언트는 1–2초 동안 debounce하고 가장 큰 seq만 보냅니다. 서버는 Redis에서 한 번 더 병합하여 key {conversationId}:{userId}마다 max 값만 유지한 후, 배치로 데이터베이스에 flush합니다.
• Kafka: conversationId로 파티션하면 순서가 유지됩니다. 부하가 증가하면, consumer를 추가하기만 하면 되며, 최대 파티션 수입니다.
• 저장: 하루에 4억 개의 메시지, 각 메시지 약 300바이트 = 하루에 약 120GB. message 테이블을 conversation_id(hash)로 파티셔닝하고 오래된 메시지를 Cassandra나 object storage 같은 저렴한 저장소로 옮기세요. 히스토리 쿼리에는 항상 conversation_id가 포함되므로, 어떤 모델도 작동합니다.

8. 주요 동작 테스트

두 가지 중요한 동작이 있습니다: 재전송이 중복 메시지를 만들지 않는 것, 그리고 워터마크가 절대 뒤로 가지 않는 것. FOR UPDATE와 GREATEST/LEAST에 의존하므로, Testcontainers를 통해 실제 PostgreSQL로 테스트해야 합니다.

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

섹션 3의 DDL을 src/main/resources/schema.sql에 넣고 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, "그룹에 인사합니다");
        SentMessage retry = messages.send(1, 10, clientMsgId, "그룹에 인사합니다");

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

        assertThat(reads.markRead(1, 20, 4)).isTrue();
        assertThat(reads.markRead(1, 20, 2)).isFalse();   // 오래된 요청이 늦게 도착
        assertThat(reads.markRead(1, 20, 99)).isTrue();   // last_seq = 5로 제한됨

        assertThat(reads.unreadCount(1, 20)).isZero();
        assertThat(reads.unreadCount(1, 10)).isZero();    // 발신자는 자동으로 읽음
    }
}

두 번째 테스트는 세 가지 상황을 확인합니다. 워터마크가 4로 진행합니다. seq=2 요청이 늦게 도착하고 무시되며 false를 반환합니다. 실제를 초과하는 seq는 5로 제한되므로, 미읽음 메시지 수는 0입니다. 속성 spring.kafka.listener.auto-startup=false는 테스트가 Kafka 연결을 시도하지 않도록 유지합니다.

9. 결론

전체 설계는 한 가지 결정을 중심으로 회전합니다: 각 대화마다 연속된 seq가 있습니다. Seq에서 메시지 순서, 히스토리 페이지 나누기, 누락된 메시지 감지 및 미읽음 메시지 수를 단순 뺄셈으로 얻습니다. 읽음 상태는 단조로운 워터마크이며, 조건부 UPDATE 쿼리로 업데이트됩니다. 이 방식은 경합과 거대한 receipt 테이블을 모두 제거합니다.

나머지는 친숙한 기술입니다. clientMsgId에 unique 제약을 더하면 멱등성이 됩니다. Transactional outbox와 대화별 Kafka 키는 일관성과 이벤트 순서를 보장합니다. Debounce, 온라인 사용자에게만 푸시하기 및 다시 연결할 때 풀링하기는 시스템이 급증하는 부하를 견디도록 합니다. PostgreSQL과 FOR UPDATE로 시작하세요. seq 발급을 Redis로 분리하거나 저장소를 옮기는 것은 실제 메트릭이 정말 필요함을 보여줄 때만 하세요.