Live Chat 시스템

~13 min · app, chat, rooms, presence

Level 0Poller

0 XP0/60 lessons0/10 achievements

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

Chat backend 의 해부

실전 chat 시스템 = 작고 잘 정의된 조각의 합: room (Track 4), envelope 가진 message protocol (Track 5), connection identity (Track 4), DB 영속화 (여기), typing indicator, read receipt, connect 시 history fetch. 개별로 어려운 거 없음; 규율은 composable 유지.

영속화는 REST, real-time 은 push

가장 흔한 production 패턴: chat message 가 REST endpoint 로 POST → DB 저장 + WebSocket fan-out. WebSocket 은 liveness 용; DB 가 history 용. 재연결 클라가 REST GET /messages?since=... 로 history fetch 후 WebSocket 으로 새 message 받음. 두 path, 한 source of truth.

Code

서버: 영속화 + fan-out 가진 chat handler·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: real-time messaging scaling

Exercise

위 SQLite 영속화 가진 chat 서버 짜. 같은 room 에 다른 user 로 두 브라우저 탭: message, typing indicator, read receipt 보내. message 영속 확인 (서버 재시작, reconnect, history fetch). typing indicator 가 typer 한테 echo 안 되는 거 확인.

Progress

Progress is local-only — sign in to sync across devices.

← 📦 Socket.IO & Libraries퀴즈 · 4 questions Next →Collaborative Editing

이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

🔔 답글 알림 (로그인 필요)

로그인 — 댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.