Bài 70: Typing Indicator — Chat Realtime

• Hiểu tại sao typing indicator cần xử lý riêng (ephemeral, tần suất cao, không persist).
• Triển khai pattern Pub/Sub + TTL key để phát sự kiện và giữ trạng thái.
• Áp dụng debounce client-side và rate limit server-side để chống spam.
• Dùng Sorted Set để query "ai đang gõ" mà không cần KEYS (blocking).
• Hiểu cách fail-safe TTL làm cleanup khi client crash hoặc disconnect.

Typing indicator hoạt động như sau: khi Alice gõ vào chat box, Bob thấy "Alice đang gõ...". Khi Alice dừng gõ vài giây, indicator biến mất. Nếu Alice và Bob cùng gõ một lúc, Carol thấy "Alice và Bob đang gõ...".

Các yêu cầu kỹ thuật đặc trưng của tính năng này:

• Tần suất cao: mỗi keystroke có thể sinh event. Phải throttle trước khi gửi lên server.
• Ephemeral: trạng thái "đang gõ" không có giá trị lịch sử — không cần persist vào DB, không cần replay.
• Self-expiring: nếu client crash hoặc mất kết nối mà không gửi "stop typing", indicator phải tự biến mất sau vài giây.
• Multi-user, multi-room: nhiều room độc lập, mỗi room có thể có nhiều người gõ cùng lúc.
• Latency thấp: người dùng trông đợi indicator xuất hiện gần như tức thì (< 200ms end-to-end).

Vì tính ephemeral và không cần persist, đây là bài toán phù hợp với Redis: Pub/Sub cho phân phối event realtime, và TTL key (hoặc Sorted Set) để giữ state ngắn hạn.

Pattern cơ bản sử dụng hai công cụ Redis kết hợp:

1. TTL key typing:{room_id}:{user_id}: đại diện cho "user này đang gõ trong room này". Key tồn tại = đang gõ, key không có / expire = dừng.
2. Pub/Sub channel chat:room:{room_id}: phát event tới mọi WebSocket gateway đang phục vụ room đó.

import json
import redis

r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

TYPING_TTL = 5  # giây — fail-safe nếu client không gửi stop


def typing_start(user_id: str, room_id: str) -> None:
    # Ghi/refresh key với TTL
    r.setex(f"typing:{room_id}:{user_id}", TYPING_TTL, "1")
    # Phát event cho toàn bộ subscriber của room
    r.publish(
        f"chat:room:{room_id}",
        json.dumps({"type": "typing_start", "user": user_id}),
    )


def typing_stop(user_id: str, room_id: str) -> None:
    r.delete(f"typing:{room_id}:{user_id}")
    r.publish(
        f"chat:room:{room_id}",
        json.dumps({"type": "typing_stop", "user": user_id}),
    )

Luồng hoạt động:

• Alice gõ → client gọi typing_start("alice", "room:42") → key được set với TTL 5s, event publish tới channel.
• Mọi WebSocket gateway subscribe channel đó nhận event, forward tới các client trong room.
• Alice dừng gõ → sau timeout trên client, gọi typing_stop("alice", "room:42") → key bị xóa, event publish.
• Nếu Alice crash mà không gọi typing_stop: key tự expire sau 5s, indicator tự biến mất.

Giá trị của key không quan trọng (dùng "1" cho gọn). Điều duy nhất cần biết là key có tồn tại hay không.

User gõ nhanh có thể sinh 5–10 keystroke mỗi giây. Gửi event lên server mỗi keystroke là lãng phí băng thông và tải Redis không cần thiết. Client cần throttle:

let typingTimer = null;
let isTyping = false;
const STOP_TIMEOUT_MS = 3000;  // dừng gõ sau 3s không có keystroke
const REFRESH_INTERVAL_MS = 4000;  // refresh typing_start mỗi 4s nếu tiếp tục gõ
let refreshTimer = null;

input.addEventListener("input", () => {
  // Nếu chưa đang gõ: gửi typing_start ngay
  if (!isTyping) {
    socket.emit("typing_start", { roomId });
    isTyping = true;

    // Refresh định kỳ để key không expire khi đang gõ liên tục
    refreshTimer = setInterval(() => {
      if (isTyping) socket.emit("typing_start", { roomId });
    }, REFRESH_INTERVAL_MS);
  }

  // Reset timer stop mỗi khi có keystroke mới
  clearTimeout(typingTimer);
  typingTimer = setTimeout(() => {
    socket.emit("typing_stop", { roomId });
    isTyping = false;
    clearInterval(refreshTimer);
    refreshTimer = null;
  }, STOP_TIMEOUT_MS);
});

Giải thích:

• Gửi typing_start một lần khi bắt đầu gõ, không gửi lại mỗi keystroke.
• Refresh mỗi 4s: TTL key trên Redis là 5s. Nếu user gõ liên tục hơn 5s mà không refresh, key expire và indicator biến mất nhầm. Refresh interval phải nhỏ hơn TTL.
• Stop sau 3s im lặng: đủ để coi là đã dừng gõ.

Kết quả: với user gõ bình thường (1–2 từ/s), server chỉ nhận 1 event typing_start + vài event refresh mỗi 4s, thay vì hàng chục event mỗi giây.

Trường hợp khó xử nhất là client crash hoặc mất mạng đột ngột mà không có cơ hội gửi typing_stop. Nếu không xử lý, indicator sẽ hiển thị mãi mãi.

TTL key là cơ chế fail-safe chính: key typing:{room_id}:{user_id} chỉ sống tối đa 5s. Dù client không gửi typing_stop, key tự expire và state được cleanup.

Server cần xử lý sự kiện disconnect WebSocket để cleanup chủ động hơn:

def on_websocket_disconnect(user_id: str, room_id: str) -> None:
    """
    Gọi khi WebSocket của user đóng lại.
    Cleanup ngay thay vì đợi TTL expire.
    """
    r.delete(f"typing:{room_id}:{user_id}")
    r.publish(
        f"chat:room:{room_id}",
        json.dumps({"type": "typing_stop", "user": user_id}),
    )

Với Socket.IO (Node.js), hook tương đương là sự kiện disconnect:

socket.on("disconnect", () => {
  // user_id và room_id đã lưu trên socket session
  redisClient.del(`typing:${roomId}:${userId}`);
  redisClient.publish(
    `chat:room:${roomId}`,
    JSON.stringify({ type: "typing_stop", user: userId }),
  );
});

TTL là lớp bảo vệ thứ hai: nếu disconnect handler bị bỏ qua, lỗi, hoặc server process crash trước khi handler chạy, TTL vẫn đảm bảo cleanup trong 5s.

Khi cần biết danh sách ai đang gõ trong một room (ví dụ khi user mới join, hoặc để refresh state định kỳ), dùng SCAN với pattern, không dùng KEYS.

def who_is_typing(room_id: str) -> list[str]:
    """
    Trả về danh sách user_id đang gõ trong room.
    Dùng SCAN thay vì KEYS để không block Redis.
    """
    pattern = f"typing:{room_id}:*"
    users = []
    cursor = 0
    while True:
        cursor, keys = r.scan(cursor, match=pattern, count=100)
        for key in keys:
            # key = "typing:{room_id}:{user_id}"
            user_id = key.split(":")[-1]
            users.append(user_id)
        if cursor == 0:
            break
    return users

Tại sao không dùng KEYS typing:room42:*? Lệnh KEYS quét toàn bộ keyspace và block Redis single-thread trong suốt thời gian quét. Với vài triệu key, KEYS có thể mất vài giây và gây latency spike cho toàn bộ hệ thống. SCAN quét theo batch và nhường CPU giữa các lần lặp.

Hàm who_is_typing thường được gọi khi:

• User mới join room — cần biết ai đang gõ lúc đó.
• Frontend poll định kỳ (~5s) để đồng bộ state nếu có sự kiện Pub/Sub bị mất.

Pattern 1 (TTL key riêng biệt cho mỗi user) hoạt động tốt, nhưng có một bất tiện: để query danh sách người gõ, phải SCAN toàn keyspace tìm theo pattern. Với hệ thống nhiều room, SCAN vẫn là O(N) trên keyspace.

Một cách khác là dùng Sorted Set per-room, với score = unix timestamp khi hết hạn:

import time

TYPING_TTL = 5  # giây


def typing_start_v2(user_id: str, room_id: str) -> None:
    expire_at = int(time.time()) + TYPING_TTL
    r.zadd(f"typing:room:{room_id}", {user_id: expire_at})
    r.publish(
        f"chat:room:{room_id}",
        json.dumps({"type": "typing_start", "user": user_id}),
    )


def typing_stop_v2(user_id: str, room_id: str) -> None:
    r.zrem(f"typing:room:{room_id}", user_id)
    r.publish(
        f"chat:room:{room_id}",
        json.dumps({"type": "typing_stop", "user": user_id}),
    )


def who_is_typing_v2(room_id: str) -> list[str]:
    now = int(time.time())
    key = f"typing:room:{room_id}"
    # Xóa các member đã expire (score <= now)
    r.zremrangebyscore(key, 0, now)
    # Lấy danh sách còn lại — score > now
    return r.zrange(key, 0, -1)

So sánh hai pattern:

Tiêu chí	TTL key riêng	Sorted Set
Cleanup tự động	Redis tự expire từng key	Lazy: cleanup lúc query
Query "ai đang gõ"	SCAN (O(N) keyspace)	ZRANGE (O(log N + M))
Memory	Nhiều key nhỏ	1 key per room
Độ phức tạp implement	Đơn giản hơn	Cần quản lý score/cleanup
Redis 7.4+ Hash field TTL	—	Có thể dùng Hash thay Zset

Khi số room không nhiều và mỗi room ít người, cả hai đều ổn. Sorted Set có lợi thế khi cần query danh sách thường xuyên.

Redis 7.4+ giới thiệu Hash field expiration, cho phép đặt TTL riêng cho từng field trong một Hash key. Đây là option thứ ba nếu đang dùng Redis 7.4 trở lên.

Với pattern Sorted Set, Redis không tự xóa member khi score vượt quá timestamp hiện tại. Cleanup phải được thực hiện thủ công. Có hai cách:

Lazy cleanup: mỗi lần gọi who_is_typing_v2, chạy ZREMRANGEBYSCORE trước khi đọc danh sách. Cách này đơn giản nhất và đủ dùng vì:

• Hàm who_is_typing được gọi đủ thường xuyên (mỗi khi user join room, hoặc periodic poll).
• Kể cả nếu không cleanup ngay, dữ liệu expired chỉ tồn tại trong Sorted Set — không được trả về client nếu gọi who_is_typing đúng cách.

def who_is_typing_v2(room_id: str) -> list[str]:
    now = int(time.time())
    key = f"typing:room:{room_id}"
    # Xóa expired trước khi đọc
    r.zremrangebyscore(key, "-inf", now)
    return r.zrange(key, 0, -1)

Hai lệnh này có thể bọc trong Lua script hoặc pipeline để chạy atomic:

lua_who_is_typing = r.register_script("""
local key = KEYS[1]
local now = tonumber(ARGV[1])
redis.call('ZREMRANGEBYSCORE', key, '-inf', now)
return redis.call('ZRANGE', key, 0, -1)
""")

def who_is_typing_atomic(room_id: str) -> list[str]:
    return lua_who_is_typing(
        keys=[f"typing:room:{room_id}"],
        args=[int(time.time())],
    )

Lua script đảm bảo cleanup và read xảy ra trong cùng một atomic operation, tránh race condition nếu có hai server cùng cleanup một lúc (cleanup idempotent nên race condition ở đây không gây lỗi, nhưng script gọn hơn về số roundtrip).

Throttle client-side giảm tải nhưng không thể tin tưởng hoàn toàn — client có thể bị bypass hoặc là malicious. Server cần rate limit để chặn user gửi quá nhiều typing_start.

Pattern đơn giản nhất: SET NX EX — nếu set được thì cho phép, nếu key đã tồn tại thì từ chối:

def can_send_typing(user_id: str, room_id: str) -> bool:
    """
    Trả về True nếu user được phép gửi typing event.
    Cho phép tối đa 1 lần mỗi 2 giây per user per room.
    """
    key = f"throttle:typing:{user_id}:{room_id}"
    # SET key "1" EX 2 NX: set nếu chưa có key, expire sau 2s
    return r.set(key, "1", nx=True, ex=2) is not None


def handle_typing_start(user_id: str, room_id: str) -> None:
    if not can_send_typing(user_id, room_id):
        return  # bỏ qua — quá nhiều event trong khoảng thời gian ngắn
    typing_start(user_id, room_id)

Logic: key throttle:typing:{user_id}:{room_id} tồn tại trong 2s sau lần gọi gần nhất. Nếu key còn tồn tại, request bị bỏ qua. Khi key expire, user lại được phép gửi event mới.

TTL của throttle key (2s) nên nhỏ hơn refresh interval client (4s), để event refresh được phép qua server.

Với deployment nhiều WebSocket gateway instance, typing event cần được phân phối tới tất cả instance để mỗi instance forward tới local clients của mình. Đây chính là bài toán WebSocket backplane đã phân tích ở bài 67.

Typing indicator hoạt động tốt trong mô hình này vì:

• Alice kết nối tới Instance A, Bob kết nối tới Instance B.
• Alice gõ → Instance A nhận event → ghi key TTL + publish lên Redis channel chat:room:42.
• Tất cả instance (A, B, C...) đều subscribe channel đó → Instance B nhận event → forward tới Bob.

import asyncio
import aioredis

async def subscribe_room_events(gateway, room_id: str):
    """
    Mỗi WebSocket gateway instance subscribe channel của room.
    Chạy trong background task.
    """
    pubsub_conn = await aioredis.create_redis("redis://localhost")
    channel = await pubsub_conn.subscribe(f"chat:room:{room_id}")
    channel_obj = channel[0]

    async for raw_msg in channel_obj.iter():
        if raw_msg:
            msg = json.loads(raw_msg)
            # Forward tới tất cả local WebSocket clients trong room
            await gateway.broadcast_to_room(room_id, msg)

Vài điểm cần chú ý khi scale:

• Channel per room, không phải channel global. Nếu dùng 1 channel cho mọi room, mọi instance phải xử lý tất cả event của tất cả room — tạo noise và tốn CPU decode/route.
• Giới hạn số channel: nếu có 100.000 room active, mỗi instance không thể subscribe 100.000 channel. Cần chiến lược subscribe lazily (subscribe khi có user trong room, unsubscribe khi room trống).
• Connection multiplexing: nếu user tham gia nhiều room, dùng một WebSocket connection với room ID trong payload thay vì mở nhiều connection.

Frontend cần duy trì một local set "ai đang gõ" và render dựa trên set đó:

const typingUsers = new Set();

function updateTypingUI() {
  const users = [...typingUsers];
  if (users.length === 0) {
    typingIndicator.textContent = "";
    return;
  }
  if (users.length === 1) {
    typingIndicator.textContent = `${users[0]} đang gõ...`;
  } else if (users.length === 2) {
    typingIndicator.textContent = `${users[0]} và ${users[1]} đang gõ...`;
  } else if (users.length === 3) {
    typingIndicator.textContent = `${users[0]}, ${users[1]} và ${users[2]} đang gõ...`;
  } else {
    typingIndicator.textContent = `${users.length} người đang gõ...`;
  }
}

socket.on("message", (msg) => {
  if (msg.type === "typing_start") {
    typingUsers.add(msg.user);
    updateTypingUI();
  } else if (msg.type === "typing_stop") {
    typingUsers.delete(msg.user);
    updateTypingUI();
  }
});

Một số điểm UX cần quan tâm:

• Giới hạn số tên hiển thị: hiển thị tối đa 2–3 tên rồi thu gọn ("và N người khác"). Tránh UI dài vô hạn.
• Anti-flicker: nếu indicator xuất hiện rồi biến mất trong <500ms, trông giật. Có thể debounce việc ẩn indicator tối thiểu 500ms sau khi nhận typing_stop.
• Animate "...": ba chấm nhấp nháy (CSS animation) truyền tải cảm giác "đang gõ" tốt hơn text tĩnh.
• Không hiển thị chính mình: lọc msg.user !== currentUser trước khi add vào typingUsers.
• Periodic refresh state: gọi GET /rooms/:id/typing mỗi 5–10s để đồng bộ nếu Pub/Sub event bị mất (disconnect/reconnect ngắn).

• Gửi event mỗi keystroke: không có throttle client-side → 10–20 event/giây/user → tải Redis và network không cần thiết.
• Không có TTL và không xử lý disconnect: client crash → indicator hiển thị mãi mãi → UX tệ, state sai.
• Lưu typing state vào DB: typing không có giá trị lịch sử. Ghi vào PostgreSQL hay MongoDB vừa tốn I/O vừa tích lũy rác.
• Dùng KEYS để query ai đang gõ: KEYS typing:room42:* block Redis. Dùng SCAN hoặc chuyển sang Sorted Set.
• Dùng một Pub/Sub channel toàn cục: publish tất cả typing event của mọi room lên một channel duy nhất → subscriber nhận cả những event không liên quan, tốn CPU để filter.
• TTL quá ngắn (<2s): nếu TTL ngắn hơn chu kỳ refresh client, key expire trong khi user vẫn đang gõ → indicator nhấp nháy vô cớ.
• Không rate limit server-side: client bị bypass hoặc lỗi có thể gửi hàng trăm event/giây, gây tải đột biến.

• Client debounce + periodic refresh: gửi typing_start một lần khi bắt đầu, refresh mỗi 4s nếu vẫn gõ, gửi typing_stop sau 3s im lặng.
• TTL fail-safe: TTL key (5s) phải lớn hơn refresh interval (4s) để không expire nhầm. TTL là lớp bảo vệ khi client không cleanup được.
• Channel per room: granularity đúng — subscriber chỉ nhận event của room mình phục vụ.
• SCAN thay KEYS: với pattern 1 (TTL key riêng), luôn dùng SCAN để query danh sách người gõ.
• Sorted Set với lazy cleanup: với pattern 2, gọi ZREMRANGEBYSCORE 0 now trước mỗi lần đọc danh sách.
• Rate limit server-side: SET NX EX 2 cho phép tối đa 1 event mỗi 2s per user per room.
• Cleanup khi disconnect: xóa key ngay khi WebSocket đóng, không chờ TTL expire.
• Frontend UX: hiển thị tối đa 3 tên, debounce ẩn indicator 500ms, lọc chính mình khỏi danh sách.

Typing indicator kết hợp hai công cụ Redis:

• Pub/Sub: phát event typing_start / typing_stop realtime tới tất cả WebSocket instance đang phục vụ room đó.
• TTL key hoặc Sorted Set: giữ trạng thái ephemeral "ai đang gõ", tự cleanup khi expire — fail-safe cho trường hợp client crash.

Các điểm kỹ thuật quan trọng:

• Throttle client giảm tần suất event xuống mức hợp lý; rate limit server-side ngăn lạm dụng.
• Refresh interval phải nhỏ hơn TTL để key không expire khi user đang gõ liên tục.
• Dùng SCAN hoặc Sorted Set thay vì KEYS để query không block Redis.
• Typing state là ephemeral — không ghi vào DB.

Quiz

1. Tại sao TTL key 5s quan trọng dù đã có typing_stop event?
2. Client refresh typing_start mỗi 4s nhưng TTL chỉ là 3s. Điều gì xảy ra?
3. Giải thích tại sao KEYS typing:room42:* nguy hiểm trong production.
4. Sorted Set pattern dùng score là gì, và ZREMRANGEBYSCORE được gọi khi nào?
5. Với 3 WebSocket instance, làm thế nào Alice (instance A) và Bob (instance B) đều nhận được event typing của Carol (instance C)?

Đáp án gợi ý

1. Client có thể crash hoặc mất mạng trước khi gửi được typing_stop. TTL đảm bảo state tự cleanup sau 5s dù không có event stop.
2. Key sẽ expire sau 3s, trước khi client kịp refresh lần tiếp theo (4s). Indicator biến mất khi user vẫn đang gõ. Refresh interval phải nhỏ hơn TTL, không phải ngược lại.
3. KEYS quét toàn bộ keyspace theo pattern và block Redis trong suốt thời gian quét. Với hàng triệu key, lệnh này có thể mất vài giây, làm tất cả client bị delay.
4. Score là unix timestamp thời điểm hết hạn (time.time() + TTL). ZREMRANGEBYSCORE 0 now được gọi mỗi lần query danh sách để xóa các member đã quá hạn (lazy cleanup).
5. Carol's gateway (instance C) publish event lên Redis channel chat:room:{id}. Instance A và B đều subscribe channel đó → đều nhận được event → forward tới local clients Alice và Bob.

Bài tiếp theo

Bài 71 xây dựng Notification Fanout: broadcast thông báo hiệu quả tới nhiều người nhận — cấu trúc dữ liệu, fanout strategy, và xử lý delivery guarantee.

Tham khảo

• Redis Docs — SETEX
• Redis Docs — SCAN
• Redis Docs — ZADD
• Redis Docs — ZREMRANGEBYSCORE
• Redis Docs — PUBLISH
• Socket.IO Docs — Rooms