Presence Features: Who's Online in Real-Time
Presence features—showing who's online, who's typing, when they were last active—dramatically improve user experience in real-time apps. A video call platform needs to know if the recipient is available; a collaborative editor must show active collaborators; a customer support tool highlights available agents. This article adds presence tracking to the room-based chat server, introducing server-side heartbeats, activity timeouts, and typed events that broadcast without being persisted messages.
Presence Data Model and User Status
Extend the user object to track status and activity:
from datetime import datetime, timedelta
from typing import Literal
import asyncio
class PresenceManager:
def __init__(self):
self.rooms: Dict[str, Dict[str, dict]] = {}
self.heartbeat_interval = 30 # seconds
self.inactivity_timeout = 90 # seconds
async def connect(self, room_id: str, client_id: str, username: str, websocket: WebSocket):
self.rooms.setdefault(room_id, {})
self.rooms[room_id][client_id] = {
"username": username,
"websocket": websocket,
"status": "online",
"last_activity": datetime.now(),
"typing": False,
"joined_at": datetime.now()
}
await websocket.accept()
# Notify room of presence update
await self.broadcast_presence(room_id, {
"type": "presence_update",
"online_users": self._get_online_users(room_id),
"user_joined": username
})
# Start heartbeat monitoring for this user
asyncio.create_task(self._heartbeat_monitor(room_id, client_id))
def _get_online_users(self, room_id: str):
if room_id not in self.rooms:
return []
return [
{
"username": u["username"],
"status": u["status"],
"typing": u["typing"],
"last_activity": u["last_activity"].isoformat()
}
for u in self.rooms[room_id].values()
]
async def _heartbeat_monitor(self, room_id: str, client_id: str):
"""Periodically check for inactive users and remove them."""
while client_id in self.rooms.get(room_id, {}):
await asyncio.sleep(self.heartbeat_interval)
user = self.rooms.get(room_id, {}).get(client_id)
if not user:
break
elapsed = (datetime.now() - user["last_activity"]).total_seconds()
if elapsed > self.inactivity_timeout:
# User is inactive; remove and notify
username = user["username"]
del self.rooms[room_id][client_id]
await self.broadcast_presence(room_id, {
"type": "user_inactive",
"username": username,
"online_users": self._get_online_users(room_id)
})
break
async def broadcast_presence(self, room_id: str, message: dict):
"""Broadcast presence events (not stored as chat messages)."""
if room_id not in self.rooms:
return
disconnected = []
for client_id, user in self.rooms[room_id].items():
try:
await user["websocket"].send_json(message)
except Exception:
disconnected.append(client_id)
for client_id in disconnected:
self.rooms[room_id].pop(client_id, None)
async def handle_typing(self, room_id: str, client_id: str, typing: bool):
"""Notify room that user is typing or stopped typing."""
if room_id not in self.rooms or client_id not in self.rooms[room_id]:
return
user = self.rooms[room_id][client_id]
user["typing"] = typing
user["last_activity"] = datetime.now()
await self.broadcast_presence(room_id, {
"type": "typing_indicator",
"username": user["username"],
"typing": typing,
"active_typists": [
u["username"] for u in self.rooms[room_id].values() if u["typing"]
]
})
async def handle_message(self, room_id: str, client_id: str, text: str):
"""Record message and clear typing status."""
if room_id not in self.rooms or client_id not in self.rooms[room_id]:
return
user = self.rooms[room_id][client_id]
user["last_activity"] = datetime.now()
user["typing"] = False
# Broadcast as chat message (separate from presence)
message = {
"type": "chat_message",
"username": user["username"],
"text": text,
"timestamp": datetime.now().isoformat()
}
await self._broadcast_message(room_id, message)
# Notify that typing stopped
await self.broadcast_presence(room_id, {
"type": "typing_indicator",
"username": user["username"],
"typing": False,
"active_typists": [
u["username"] for u in self.rooms[room_id].values() if u["typing"]
]
})
async def _broadcast_message(self, room_id: str, message: dict):
"""Broadcast chat messages to all users in room."""
if room_id not in self.rooms:
return
disconnected = []
for client_id, user in self.rooms[room_id].items():
try:
await user["websocket"].send_json(message)
except Exception:
disconnected.append(client_id)
for client_id in disconnected:
self.rooms[room_id].pop(client_id, None)
manager = PresenceManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, room: str = Query(...), username: str = Query(...), client_id: str = Query(...)):
await manager.connect(room, client_id, username, websocket)
try:
while True:
data = await websocket.receive_text()
msg = json.loads(data) if data.startswith("{") else {"type": "message", "text": data}
if msg.get("type") == "typing":
await manager.handle_typing(room, client_id, msg.get("typing", True))
else:
await manager.handle_message(room, client_id, msg.get("text") or data)
except Exception:
# User disconnect or error
pass
Heartbeat Mechanism: Server-Side Activity Monitoring
The _heartbeat_monitor() coroutine runs in the background for each user, waking every 30 seconds to check if the user has sent activity (message, typing event) in the past 90 seconds. If silent for too long—network failure, browser crash—the server assumes disconnection and removes them, broadcasting a user_inactive event. The client never needs to send pings explicitly; the server knows you're alive if you're active, and evicts you if you're not.
This is more efficient than ping/pong frames because many applications already send messages or activity regularly, making explicit pings redundant. Set inactivity_timeout based on your tolerance: 90 seconds is typical for chat; video calls might use 30 seconds; background analytics might tolerate 300 seconds.
Typing Indicators: Non-Persistent Events
When a user presses a key, send a {"type": "typing", "typing": true} message; when they pause for 2 seconds, send {"type": "typing", "typing": false}. The server broadcasts these to peers, but doesn't store them in chat history—only display them as transient UI (e.g., "Alice is typing..."). This separates presence events from chat messages, allowing different handling:
const ws = new WebSocket(wsUrl);
let typingTimeout;
document.getElementById('msg').addEventListener('keydown', () => {
clearTimeout(typingTimeout);
ws.send(JSON.stringify({type: "typing", typing: true}));
typingTimeout = setTimeout(() => {
ws.send(JSON.stringify({type: "typing", typing: false}));
}, 2000);
});
ws.onmessage = (event) => {
const msg = JSON.parse(event.data);
if (msg.type === 'typing_indicator') {
if (msg.typing) {
document.getElementById('typists').innerText = msg.active_typists.join(', ') + ' is/are typing...';
}
}
};
Last-Seen Timestamp Tracking
Each user object stores last_activity, updated whenever they send a message or typing event. You can expose this via a REST endpoint or presence update:
@app.get("/api/rooms/{room_id}/users")
async def get_room_users(room_id: str):
return {
"room_id": room_id,
"users": [
{
"username": u["username"],
"status": u["status"],
"last_activity": u["last_activity"].isoformat(),
"joined_at": u["joined_at"].isoformat()
}
for u in manager.rooms.get(room_id, {}).values()
]
}
Clients can use last_activity to sort user lists, showing recently active users first, or marking inactive users with a gray indicator.
Handling Rapid Disconnect/Reconnect
If a network hiccup disconnects and reconnects a client within seconds, you might want to preserve their session instead of treating them as a new user. Implement a grace period:
class PresenceManager:
def __init__(self):
# ... existing code ...
self.grace_period = 10 # seconds
self.pending_disconnects = {} # client_id -> (room_id, username, disconnect_time)
def disconnect(self, room_id: str, client_id: str):
if room_id in self.rooms and client_id in self.rooms[room_id]:
user = self.rooms[room_id].pop(client_id)
# Mark as pending disconnect instead of removing immediately
self.pending_disconnects[client_id] = (room_id, user["username"], datetime.now())
return user["username"]
return None
async def connect(self, room_id: str, client_id: str, username: str, websocket: WebSocket):
# Check if this is a reconnection within grace period
if client_id in self.pending_disconnects:
old_room, old_username, disconnect_time = self.pending_disconnects[client_id]
if (datetime.now() - disconnect_time).total_seconds() < self.grace_period:
del self.pending_disconnects[client_id]
# Restore session
self.rooms[old_room][client_id] = {...} # reuse old entry
return
# New user or grace period expired
del self.pending_disconnects.get(client_id, None)
# Normal connect flow
Key Takeaways
- Presence tracking separates user status (online, typing, last-seen) from chat messages, allowing independent update rates.
- Server-side heartbeats monitor inactivity and evict stale connections automatically, improving reliability.
- Typing indicators are transient events broadcast but not persisted, reducing message history bloat.
- Grace period reconnection logic tolerates brief network failures without losing user session.
- Last-activity timestamps enable sorting and filtering, improving UI/UX.
Frequently Asked Questions
How do I prevent typing indicators from spamming the room?
Implement client-side debouncing: only send a typing: true event once, and resend typing: false after 2 seconds of silence. Server-side, you could also throttle: only broadcast a typing update if the typers set changed since the last broadcast.
Can I show a "User X read message Y" feature?
Yes. When the client's UI renders a message, send a {"type": "read", "message_id": "..."} event. The server broadcasts this to the message sender, showing a "read by" indicator. This is separate from typing or presence.
What if presence data becomes stale due to server bugs?
Add a full resync endpoint: GET /api/rooms/{room_id}/presence returns the authoritative presence state. Clients can call this periodically (e.g., every 10 minutes) to reconcile with the server. This is a safety net for long-lived connections.
How do I integrate presence with a database for persistence?
On connect(), insert a row into a user_presence table; on disconnect, update the last_seen timestamp. Query this table to show "Bob was last seen 2 hours ago" after disconnection. Use a background job to clean up old entries (older than 30 days).