Design WhatsApp/Chat System

import asyncio
import websockets
import json

# Active connections: user_id → websocket
connections = {}

async def handle_connection(websocket, path):
    user_id = await authenticate(websocket)

    # Connection register করুন
    connections[user_id] = websocket
    await redis.set(f"user:{user_id}:server", SERVER_ID)
    await redis.set(f"user:{user_id}:online", "true")

    try:
        async for raw_msg in websocket:
            msg = json.loads(raw_msg)
            await send_message(
                sender_id=user_id,
                receiver_id=msg['to'],
                content=msg['content']
            )
    finally:
        # Disconnect হলে cleanup
        del connections[user_id]
        await redis.set(f"user:{user_id}:online", "false")
        await redis.set(f"user:{user_id}:last_seen", now())

async def send_message(sender_id, receiver_id, content):
    msg_id = generate_unique_id()

    # 1. DB-তে persist করুন আগে
    await db.save_message(msg_id, sender_id, receiver_id, content)

    # 2. Receiver কোন server-এ? Redis check করুন
    receiver_server = await redis.get(f"user:{receiver_id}:server")

    if receiver_server == SERVER_ID:
        # Same server-এ আছে → direct push
        ws = connections.get(receiver_id)
        if ws:
            await ws.send(json.dumps({"msg_id": msg_id, "content": content}))
    else:
        # Different server → Kafka-তে route করুন
        await kafka.publish(f"chat:{receiver_server}", {
            "receiver_id": receiver_id, "msg_id": msg_id
        })

-- 1-to-1 Messages Table
CREATE TABLE messages (
    chat_id      UUID,
    message_id   TIMEUUID,     -- Time-ordered UUID
    sender_id    UUID,
    receiver_id  UUID,
    content      TEXT,
    msg_type     TEXT,         -- 'text', 'image', 'video'
    status       TEXT,         -- 'sent', 'delivered', 'read'
    created_at   TIMESTAMP,
    PRIMARY KEY (chat_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);
-- Partition by chat_id = same conversation একই partition-এ
-- Latest messages DESC = pagination efficient

-- Group Messages (1 copy per message, not per user)
CREATE TABLE group_messages (
    group_id     UUID,
    message_id   TIMEUUID,
    sender_id    UUID,
    content      TEXT,
    sent_at      TIMESTAMP,
    PRIMARY KEY (group_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

-- Per-user delivery tracking (NOT message copy)
CREATE TABLE message_receipts (
    message_id   UUID,
    user_id      UUID,
    delivered_at TIMESTAMP,
    read_at      TIMESTAMP,
    PRIMARY KEY (message_id, user_id)
);
-- 1000 member group = 1 msg + 999 receipt rows (not 1000 msg copies)

-- Group messages: 1 row per message (not per user)
CREATE TABLE group_messages (
    msg_id       UUID PRIMARY KEY,
    group_id     UUID,
    sender_id    UUID,
    content      TEXT,
    sent_at      TIMESTAMP
);

-- Per-user delivery tracking (not message copy)
CREATE TABLE message_receipts (
    msg_id       UUID,
    user_id      UUID,
    delivered_at TIMESTAMP,
    read_at      TIMESTAMP,
    PRIMARY KEY (msg_id, user_id)
);
-- 1000 member group = 1 msg + 999 receipt rows (not 1000 msg copies)