mirror of
https://github.com/fromchat-messenger/web.git
synced 2026-09-22 19:15:08 +03:00
Implement real-time online status and typing indicator
This commit is contained in:
@@ -4,6 +4,8 @@ from pathlib import Path
|
||||
import os
|
||||
import re
|
||||
import uuid
|
||||
import asyncio
|
||||
import time
|
||||
from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect, UploadFile, File, Form
|
||||
from fastapi.responses import FileResponse
|
||||
from fastapi.security import HTTPAuthorizationCredentials
|
||||
@@ -661,11 +663,19 @@ class MessaggingSocketManager:
|
||||
def __init__(self) -> None:
|
||||
self.connections: list[WebSocket] = []
|
||||
self.user_by_ws: dict[WebSocket, int] = {}
|
||||
self.online_users: set[int] = set()
|
||||
self.typing_users: dict[int, float] = {} # user_id -> timestamp
|
||||
self.dm_typing_users: dict[int, dict[int, float]] = {} # user_id -> {recipient_id -> timestamp}
|
||||
self.ws_subscriptions: dict[WebSocket, set[int]] = {} # websocket -> set of subscribed user_ids
|
||||
self._cleanup_task = None
|
||||
|
||||
async def send_error(self, websocket: WebSocket, type: str, e: HTTPException):
|
||||
await websocket.send_json({"type": type, "error": {"code": e.status_code, "detail": e.detail}})
|
||||
|
||||
async def handle_connection(self, websocket: WebSocket, db: Session):
|
||||
# Initialize subscriptions for this connection
|
||||
self.ws_subscriptions[websocket] = set()
|
||||
|
||||
while True:
|
||||
data = await websocket.receive_json()
|
||||
type = data["type"]
|
||||
@@ -687,6 +697,14 @@ class MessaggingSocketManager:
|
||||
current_user = get_current_user_inner()
|
||||
if current_user:
|
||||
self.user_by_ws[websocket] = current_user.id
|
||||
# Set user online in DB
|
||||
current_user.online = True
|
||||
current_user.last_seen = datetime.now()
|
||||
db.commit()
|
||||
# Add to online users
|
||||
self.online_users.add(current_user.id)
|
||||
# Broadcast status change
|
||||
await self.broadcast_status_change(current_user.id, True, current_user.last_seen.isoformat())
|
||||
else:
|
||||
await websocket.send_json({
|
||||
"type": "ping",
|
||||
@@ -1040,6 +1058,137 @@ class MessaggingSocketManager:
|
||||
await websocket.send_json({"type": "call_screen_share_toggle", "data": {"status": "ok"}})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
elif type == "subscribeStatus":
|
||||
try:
|
||||
current_user = get_current_user_inner()
|
||||
if not current_user:
|
||||
raise HTTPException(401)
|
||||
|
||||
user_id_to_subscribe = int(data["data"]["userId"])
|
||||
self.ws_subscriptions[websocket].add(user_id_to_subscribe)
|
||||
|
||||
# Get current status of the user
|
||||
target_user = db.query(User).filter(User.id == user_id_to_subscribe).first()
|
||||
if target_user:
|
||||
await websocket.send_json({
|
||||
"type": "statusUpdate",
|
||||
"data": {
|
||||
"userId": user_id_to_subscribe,
|
||||
"online": target_user.online,
|
||||
"lastSeen": target_user.last_seen.isoformat()
|
||||
}
|
||||
})
|
||||
else:
|
||||
await websocket.send_json({
|
||||
"type": "subscribeStatus",
|
||||
"data": {"status": "error", "error": "User not found"}
|
||||
})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
elif type == "unsubscribeStatus":
|
||||
try:
|
||||
current_user = get_current_user_inner()
|
||||
if not current_user:
|
||||
raise HTTPException(401)
|
||||
|
||||
user_id_to_unsubscribe = int(data["data"]["userId"])
|
||||
self.ws_subscriptions[websocket].discard(user_id_to_unsubscribe)
|
||||
|
||||
await websocket.send_json({"type": "unsubscribeStatus", "data": {"status": "ok"}})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
elif type == "typing":
|
||||
try:
|
||||
current_user = get_current_user_inner()
|
||||
if not current_user:
|
||||
raise HTTPException(401)
|
||||
|
||||
import time
|
||||
self.typing_users[current_user.id] = time.time()
|
||||
|
||||
# Broadcast to all connected users
|
||||
await self.broadcast({
|
||||
"type": "typing",
|
||||
"data": {
|
||||
"userId": current_user.id,
|
||||
"username": current_user.username
|
||||
}
|
||||
})
|
||||
|
||||
await websocket.send_json({"type": "typing", "data": {"status": "ok"}})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
elif type == "stopTyping":
|
||||
try:
|
||||
current_user = get_current_user_inner()
|
||||
if not current_user:
|
||||
raise HTTPException(401)
|
||||
|
||||
if current_user.id in self.typing_users:
|
||||
del self.typing_users[current_user.id]
|
||||
|
||||
# Broadcast to all connected users
|
||||
await self.broadcast({
|
||||
"type": "stopTyping",
|
||||
"data": {
|
||||
"userId": current_user.id,
|
||||
"username": current_user.username
|
||||
}
|
||||
})
|
||||
|
||||
await websocket.send_json({"type": "stopTyping", "data": {"status": "ok"}})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
elif type == "dmTyping":
|
||||
try:
|
||||
current_user = get_current_user_inner()
|
||||
if not current_user:
|
||||
raise HTTPException(401)
|
||||
|
||||
recipient_id = int(data["data"]["recipientId"])
|
||||
import time
|
||||
|
||||
if current_user.id not in self.dm_typing_users:
|
||||
self.dm_typing_users[current_user.id] = {}
|
||||
self.dm_typing_users[current_user.id][recipient_id] = time.time()
|
||||
|
||||
# Send only to recipient
|
||||
await self.send_to_user(recipient_id, {
|
||||
"type": "dmTyping",
|
||||
"data": {
|
||||
"userId": current_user.id,
|
||||
"username": current_user.username
|
||||
}
|
||||
})
|
||||
|
||||
await websocket.send_json({"type": "dmTyping", "data": {"status": "ok"}})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
elif type == "stopDmTyping":
|
||||
try:
|
||||
current_user = get_current_user_inner()
|
||||
if not current_user:
|
||||
raise HTTPException(401)
|
||||
|
||||
recipient_id = int(data["data"]["recipientId"])
|
||||
|
||||
if current_user.id in self.dm_typing_users and recipient_id in self.dm_typing_users[current_user.id]:
|
||||
del self.dm_typing_users[current_user.id][recipient_id]
|
||||
if not self.dm_typing_users[current_user.id]:
|
||||
del self.dm_typing_users[current_user.id]
|
||||
|
||||
# Send only to recipient
|
||||
await self.send_to_user(recipient_id, {
|
||||
"type": "stopDmTyping",
|
||||
"data": {
|
||||
"userId": current_user.id,
|
||||
"username": current_user.username
|
||||
}
|
||||
})
|
||||
|
||||
await websocket.send_json({"type": "stopDmTyping", "data": {"status": "ok"}})
|
||||
except HTTPException as e:
|
||||
await self.send_error(websocket, type, e)
|
||||
else:
|
||||
await websocket.send_json({"type": type, "error": {"code": 400, "detail": "Invalid type"}})
|
||||
|
||||
@@ -1057,9 +1206,24 @@ class MessaggingSocketManager:
|
||||
except WebSocketDisconnect as e:
|
||||
logger.info(f"WebSocket disconnected with code {e.code}: {e.reason}")
|
||||
finally:
|
||||
# Cleanup connection
|
||||
self.connections.remove(websocket)
|
||||
if websocket in self.user_by_ws:
|
||||
user_id = self.user_by_ws[websocket]
|
||||
# Set user offline in DB
|
||||
user = db.query(User).filter(User.id == user_id).first()
|
||||
if user:
|
||||
user.online = False
|
||||
user.last_seen = datetime.now()
|
||||
db.commit()
|
||||
# Remove from online users
|
||||
self.online_users.discard(user_id)
|
||||
# Broadcast status change
|
||||
await self.broadcast_status_change(user_id, False, user.last_seen.isoformat())
|
||||
del self.user_by_ws[websocket]
|
||||
# Cleanup subscriptions
|
||||
if websocket in self.ws_subscriptions:
|
||||
del self.ws_subscriptions[websocket]
|
||||
|
||||
async def broadcast(self, message: dict):
|
||||
for websocket in self.connections:
|
||||
@@ -1069,8 +1233,82 @@ class MessaggingSocketManager:
|
||||
for websocket in self.connections:
|
||||
if self.user_by_ws.get(websocket) == user_id:
|
||||
await websocket.send_json(message)
|
||||
|
||||
async def broadcast_status_change(self, user_id: int, online: bool, last_seen: str):
|
||||
"""Broadcast status change to all connections that are subscribed to this user"""
|
||||
message = {
|
||||
"type": "statusUpdate",
|
||||
"data": {
|
||||
"userId": user_id,
|
||||
"online": online,
|
||||
"lastSeen": last_seen
|
||||
}
|
||||
}
|
||||
|
||||
# Send to all connections that have this user in their subscriptions
|
||||
for websocket in self.connections:
|
||||
if websocket in self.ws_subscriptions and user_id in self.ws_subscriptions[websocket]:
|
||||
await websocket.send_json(message)
|
||||
|
||||
async def cleanup_stale_typing_indicators(self):
|
||||
"""Periodically cleanup typing indicators that haven't been updated in 3+ seconds"""
|
||||
while True:
|
||||
try:
|
||||
current_time = time.time()
|
||||
stale_threshold = 3.0 # 3 seconds
|
||||
|
||||
# Cleanup public chat typing indicators
|
||||
stale_public_typing = [
|
||||
user_id for user_id, timestamp in self.typing_users.items()
|
||||
if current_time - timestamp > stale_threshold
|
||||
]
|
||||
|
||||
for user_id in stale_public_typing:
|
||||
del self.typing_users[user_id]
|
||||
# Broadcast stop typing
|
||||
await self.broadcast({
|
||||
"type": "stopTyping",
|
||||
"data": {
|
||||
"userId": user_id,
|
||||
"username": "Unknown" # We don't have username here, frontend will handle
|
||||
}
|
||||
})
|
||||
|
||||
# Cleanup DM typing indicators
|
||||
stale_dm_typing = []
|
||||
for user_id, recipients in self.dm_typing_users.items():
|
||||
for recipient_id, timestamp in list(recipients.items()):
|
||||
if current_time - timestamp > stale_threshold:
|
||||
stale_dm_typing.append((user_id, recipient_id))
|
||||
|
||||
for user_id, recipient_id in stale_dm_typing:
|
||||
if user_id in self.dm_typing_users and recipient_id in self.dm_typing_users[user_id]:
|
||||
del self.dm_typing_users[user_id][recipient_id]
|
||||
if not self.dm_typing_users[user_id]:
|
||||
del self.dm_typing_users[user_id]
|
||||
# Send stop typing to recipient
|
||||
await self.send_to_user(recipient_id, {
|
||||
"type": "stopDmTyping",
|
||||
"data": {
|
||||
"userId": user_id,
|
||||
"username": "Unknown" # We don't have username here, frontend will handle
|
||||
}
|
||||
})
|
||||
|
||||
# Wait 1 second before next cleanup
|
||||
await asyncio.sleep(1.0)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in typing cleanup task: {e}")
|
||||
await asyncio.sleep(1.0)
|
||||
|
||||
def start_cleanup_task(self):
|
||||
"""Start the cleanup task if not already running"""
|
||||
if self._cleanup_task is None or self._cleanup_task.done():
|
||||
self._cleanup_task = asyncio.create_task(self.cleanup_stale_typing_indicators())
|
||||
|
||||
messagingManager = MessaggingSocketManager()
|
||||
# Start the cleanup task
|
||||
messagingManager.start_cleanup_task()
|
||||
|
||||
@router.websocket("/chat/ws")
|
||||
async def chat_websocket(
|
||||
|
||||
Reference in New Issue
Block a user