1対1およびグループチャットシステムの設計：メッセージ保存と既読状態

1. 概要

チャットは単純に見えます：テキストを送信してから表示するだけです。でも、数百万のユーザーがいると、より難しい質問に直面します。メッセージの順序をどうやって正しく保つ？再送信が重複を作らないようにする？誰が何まで読んだかを知るために、数十億の「既読」レコードを記録することなく、どうやって知る？

この記事は、1対1チャットとグループチャットの完全な設計を説明します。この設計は、要件、データモデル、API、処理フロー、および困難な点を含みます。この記事を読み終わると、Spring Boot 3、PostgreSQL、およびKafkaを使用して、メッセージ保存と既読状態管理のコア部分を構築できるようになります。

2. 要件

2.1. 機能要件

• 1対1チャットおよびグループチャット（最大1,000メンバー）。
• テキストメッセージの送信。メッセージは1つの会話内で正しい順序で表示されます。
• 履歴の表示、ページ単位での遡行スクロール。
• 会話リスト（インボックス）と未読メッセージ数。
• 既読状態：1対1チャットでは「既読」を表示、グループチャットでは「N人に既読」を表示。

2.2. スケールと制約

• 1,000万DAU。平均して1人あたり1日40メッセージ、つまり1日4億メッセージ、秒あたり約4,600メッセージ。ピーク時間は秒あたり約30,000メッセージまで増加する可能性があります。
• 送信成功が確認されたメッセージを失ってはいけません。
• 順序は1つの会話内でのみ正しければよい。グローバルな順序は必要ありません。
• クライアントはネットワークの不安定さにより再送信できます。サーバーはべき等である必要があります。
• 「既読」イベントはメッセージ送信イベントより頻繁です。アプリを開くたびに生成されるためです。

3番目の制約がこの設計全体のカギです。ローカルな順序だけで十分なので、すべてがconversation_idでシャーディング可能です。

3. データモデル

中心的な考え方：各会話には増分されるシーケンス番号（seq）があります。会話内のN番目のメッセージはseq = Nを持ちます。メンバーの既読状態は単なる1つの数値、last_read_seqです。これはwatermarkであり、個々のメッセージの既読リストではありません。

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制約付きで追加します。2人が同時にチャットを開いても、2つの会話は作成されません。

注記： Watermarkは、ユーザーが順序どおりに読むことを前提としています。つまり、メッセージNを読むと、その前のすべてのメッセージを読んだと見なします。この仮定はほぼすべてのチャットアプリケーションに当てはまります。特定のメッセージが正確に開かれたかどうかを知る必要がある場合（例えば、自動削除メッセージ）、そのタイプのメッセージに対して別のレコードを保存してください。すべてのメッセージに適用しないでください。

4. API

メソッド	エンドポイント	説明
POST	/conversations/{id}/messages	メッセージを送信します。ボディにはclientMsgIdとbodyが含まれます
GET	/conversations/{id}/messages?beforeSeq=&limit=	履歴、seqによるページネーション
POST	/conversations/{id}/read	ボディ{ "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>

メッセージ送信の全操作は1つのトランザクション内に含まれます：

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. 会話行のロック：同じ会話への送信はすべてキューイングされます
        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を使用します。同じ会話への2つのリクエストは順序どおりに実行されるため、seqが重複または飛び出すことはありません。異なる会話は互いにブロックしません。
• ステップ2はロック獲得後に重複をチェックします。したがって、同じclientMsgIdの2つの同時再試行も通過しません。ロック前にチェックすると、両方のリクエストが「未実装」と見える可能性があります。その後、2番目のリクエストはDuplicateKeyExceptionに遭遇します。
• ステップ3はseqを連続的に増加させ、ギャップはありません。これにより、クライアントは失われたメッセージを自動的に検出できます（セクション7.3）。
• ステップ5はトランザクショナルoutboxを適用します。トランザクション内でKafkaに直接送信する場合、Kafkaが受信しても後でDBがロールバックされる場合がある、またはその逆もあります。

clientMsgIdでsendを2回呼び出します：

send(1, 10, a1b2..., "グループへこんにちは") -> SentMessage[conversationId=1, seq=42, ...]
send(1, 10, a1b2..., "グループへこんにちは") -> SentMessage[conversationId=1, seq=42, ...]   // 新しいメッセージは作成されません

別のリレーがID順にoutboxテーブルを読み、conversationIdをキーとしてtopic chat.eventsにプッシュし、送信済みの行を削除します。これを行うためにDebeziumを使用することもできます。会話をキーとすることで、同じ会話のすべてのイベントが同じパーティションに入り、順序が保たれます。

コンシューマー側はイベントを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より大きいseqを持つメッセージを自動的にロードします。

6. 既読状態

既読をマークするには、1つの単調なUPDATEステートメントだけが必要です：

@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;   // watermarkが実際に進んだ場合は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ステートメントは、3つの問題を同時に解決します：

• 競合： 電話はseq=50を送信し、ノートパソコンは後でseq=45を送信します。条件last_read_seq < ...により、古いリクエストはwatermarkを後退させません。ロックやバージョン管理は必要ありません。
• 不正なデータ： LEAST(?, c.last_seq)はクライアントが実際のメッセージを超えるseqを送信するのをブロックします。
• 負荷軽減： updated == 0のとき、watermarkが変わらないため、read.updatedイベントが発生しません。不要なfanoutがカットされます。

markReadがtrueを返した場合のみ、サービスはread.updatedイベントをoutboxに記録します。これはセクション5と同様です。

7. 困難な点と解決策

7.1. “ホット”な会話とロック競合

FOR UPDATEは会話内のメッセージ送信をシリアル化します。1対1チャットまたは小さなグループの場合、このコストは無視できます。アクティブな1,000人のグループは、同じ行に毎秒数十のメッセージを送信できます。各トランザクションはロックを数ミリ秒だけ保持するため、この負荷は耐えられます。

会話が数百メッセージ/秒を超える場合（ブロードキャストチャネルのような）、INCR chat:seq:{conversationId}を使用してseq割り当てをRedisに分離します。その代わりに、DB書き込みが失敗したときにseqのギャップを受け入れる必要があります。クライアントは「実際のギャップ」と「まだ到着していないメッセージ」を区別する必要があります。

注記： 会話内のメッセージを並べ替えるためにcreated_atまたはグローバルID（Snowflake）を使用しないでください。サーバー間のクロックが数ミリ秒オフでも、連続して送信される2つのメッセージの順序を反転させるのに十分です。

7.2. DB、Kafka、クライアント間の一貫性

このデザインの保証チェーン：

1. DBは信頼できるソースです。メッセージはコミット後にのみクライアントに確認されます。
2. Outboxは、Kafkaへのat-least-onceを保証します。コンシューマーは重複を受け取る可能性があります。
3. クライアントは(conversationId, seq)に基づいて重複排除します。既存のメッセージはスキップされます。

7.3. 失われたメッセージの検出

seqが連続しているため、クライアントはギャップを自動的に検出できます。クライアントがseq=40を持っていて、プッシュseq=43を受け取ると、メッセージ41と42が不足していることがわかるため、GET /messages?afterSeq=40を呼び出します。各メッセージに対してサーバーから複雑なACKメカニズムは必要ありません。

7.4. トラフィックの急増

• 大規模グループのfanout： 1,000人のグループへの1つのメッセージは、1,000回のプッシュを生成します。閾値を超えるグループ（例えば200メンバー）の場合、オンラインユーザーにのみプッシュし、Redis set online:{conversationId}で確認します。ペイロードには軽いメッセージ{conversationId, seq}のみが含まれ、クライアントは必要に応じてコンテンツを取得します。
• 「既読」嵐： ネットワーク障害後、数百万クライアントが再接続して同時に既読レシートを送信します。クライアントは約1～2秒debounceし、最大seqのみを送信する必要があります。サーバーはRedis内で集約を続行し、キー{conversationId}:{userId}に対して最大値のみを保持し、その後バッチでDBにフラッシュできます。
• Kafka： 順序を保つためconversationIdでpartitionします。負荷が増加すると、コンシューマーの数を増やすだけで、最大partition数です。
• ストレージ： 1日4億メッセージ、1メッセージあたり約300バイト、1日あたり約120GB相当。messageテーブルをconversation_id（ハッシュ）でpartitionし、古いメッセージをCassandraやオブジェクトストレージなどの安価なストレージに移行してください。履歴クエリには常にconversation_idが含まれるため、どのモデルでも機能します。

8. 主要な動作のテスト

テストする2つの最も重要な動作：再送信が重複を作成しない、およびwatermarkが決して後退しない。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();    // 送信者は自動的に既読です
    }
}

9. 結論

全体的な設計は1つの決定の周りに回転します：各会話に連続したseq。seqから、メッセージの順序、履歴のページネーション、失われたメッセージの検出、および減算による未読メッセージ数が得られます。既読状態は単調なwatermarkであり、条件付きUPDATEステートメントで更新されます。この方法では、競合と巨大なレシートテーブルの両方を排除できます。

残りは慣れた技術です。clientMsgIdとユニーク制約によるべき等性。会話ごとのKafkaキーを使用したトランザクショナルoutboxにより、イベントの一貫性と順序が保証されます。Debounce、オンラインユーザーへのプッシュのみ、再接続時のプル、これらがシステムが予期しないトラフィックを乗り切るのに役立ちます。PostgreSQLとFOR UPDATEから始めてください。実装データが本当に必要なことを示すまでのみ、seqの割り当てをRedisに分離するか、ストレージリポジトリを移行してください。