Live Chat System

~13 min · app, chat, rooms, presence

Level 0Poller

0 XP0/60 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

Anatomy of a chat backend

A real chat system is the sum of small, well-defined pieces: rooms (Track 4), message protocol with envelopes (Track 5), connection identity (Track 4), persistence to a database (here), typing indicators, read receipts, and history fetch on connect. None of them are individually hard; the discipline is keeping them composable.

Persistence is REST, real-time is push

The most common production pattern: chat messages POST to a REST endpoint that persists to a database AND fans out via WebSocket. The WebSocket is for liveness; the database is for history. Reconnecting clients fetch history via REST GET /messages?since=..., then receive new messages via WebSocket. Two paths, one source of truth.

Code

Server: chat handler with persistence + fan-out·python

from datetime import datetime
import uuid

class ChatServer:
    def __init__(self, manager, db):
        self.manager = manager
        self.db = db

    async def on_message(self, ws, user, msg):
        t = msg.get('type')
        data = msg.get('data', {})

        if t == 'chat.send':
            saved = {
                'id': str(uuid.uuid4()),
                'room': data['room'],
                'text': data['text'],
                'from': user['id'],
                'ts': datetime.utcnow().isoformat() + 'Z',
            }
            await self.db.messages.insert(saved)
            await self.manager.broadcast(data['room'], {
                'type': 'chat.message',
                'data': saved,
            })

        elif t == 'chat.typing':
            await self.manager.broadcast(data['room'], {
                'type': 'chat.typing',
                'data': {'user': user['id'], 'is_typing': data['is_typing']},
            }, exclude=ws)

        elif t == 'chat.read':
            await self.manager.send_to_user(data['target_user'], {
                'type': 'chat.read_receipt',
                'data': {
                    'reader': user['id'],
                    'message_id': data['message_id'],
                },
            })

        elif t == 'chat.history':
            history = await self.db.messages.last_n(data['room'], 50)
            await ws.send_json({'type': 'chat.history', 'data': history})

External links

Discord engineering: scaling real-time messaging

Exercise

Build the chat server above with SQLite persistence. Two browser tabs as different users in the same room: send messages, typing indicators, read receipts. Verify that messages persist (restart the server, reconnect, fetch history). Verify typing indicator does not echo to the typer.