From 70b097c5e710ebfb5cdcb14c94d61f134b445011 Mon Sep 17 00:00:00 2001 From: Sagar Date: Wed, 1 Jul 2026 22:23:29 +0530 Subject: [PATCH] Add Agora integration and enhance WebSocket connection management - Introduced Agora app ID and certificate configuration in `config.py`. - Added Agora-related functionality in a new `agora.py` helper file. - Updated `Dockerfile` to set `PYTHONUNBUFFERED` environment variable. - Refactored `connection_manager.py` to improve connection handling and thread subscription management. - Updated dependencies in `pyproject.toml` and `uv.lock` to include `agora-token-builder`. --- Dockerfile | 1 + handlers/__init__.py | 236 +++++++++++++++++++++++++++++++--- helpers/agora.py | 25 ++++ helpers/config.py | 5 +- helpers/connection_manager.py | 215 +++++++++++++------------------ helpers/constants.py | 11 +- helpers/wsobjs.py | 100 +++++++++++++- main.py | 2 + pyproject.toml | 1 + uv.lock | 8 ++ 10 files changed, 457 insertions(+), 147 deletions(-) create mode 100644 helpers/agora.py diff --git a/Dockerfile b/Dockerfile index 56f44e6..6fa80be 100644 --- a/Dockerfile +++ b/Dockerfile @@ -18,5 +18,6 @@ RUN mkdir -p /home/appuser/.cache/uv && \ USER 1000 ENV PATH="/app/.venv/bin:$PATH" +ENV PYTHONUNBUFFERED=1 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8081", "--ws-ping-interval", "300", "--ws-ping-timeout", "300"] diff --git a/handlers/__init__.py b/handlers/__init__.py index 8a8c20c..0e18786 100644 --- a/handlers/__init__.py +++ b/handlers/__init__.py @@ -1,15 +1,22 @@ -from helpers.connection_manager import ConnectionManager +from helpers.connection_manager import ConnectionManager from helpers.validator import is_valid_uuid4 from helpers.validator import is_valid_id from helpers.wsobjs import WSObjects +from helpers.config import Config +from helpers.agora import build_rtc_token, ROLE_PUBLISHER, uid_from_uuid +from helpers.database.mongo import Database from helpers.constants import ( WS_TYPE_PING, WS_TYPE_MARK_READ, WS_CHAT_SCREEN_OPEN, WS_CHAT_SCREEN_CLOSE, + WS_FETCH_CHANNEL_USERS, WS_ACTION_START, WS_ACTION_END, - + WS_TYPE_JOIN_LIVE, + WS_TYPE_GET_AGORA, + WS_TYPE_TOPIC_SUBSCRIBE, + WS_TYPE_UPDATE_CHANNEL_TYPE, ACTION_BROWSING, ACTION_CHATTING, @@ -32,30 +39,28 @@ from ._refresh_viewer import refresh_user_viewing - - from .community import on_ws_ndc_event async def handle_message(data: dict, manager: ConnectionManager, uid: str, isAdmin: bool, ws): t: dict = data.get("t") - o: dict = data.get("o") - print(data) - if t is None or o is None: + o: dict = data.get("o") or {} + if t is None: return - if data["t"] == WS_TYPE_PING: + print(f"[WS MSG] uid={uid} t={t} o={o}") + + if t == WS_TYPE_PING: await manager.answer(WSObjects.Pong(), ws) await refresh_user_viewing(uid) return if not data["o"].get("id") or not is_valid_id(data["o"].get("id")): - await manager.answer(WSObjects.WSError(1, "No ID of request"), ws) - return - + print(f"[WS MSG] No/invalid ID for t={t}, o.id={data['o'].get('id')!r}") ndcId: int = o.get("ndcId") chatId: str = o.get("threadId") + ws_req_id = o.get("id") targetChatId: str = o.get("target", "").split("/")[-1] actions: list = o.get("actions") @@ -63,7 +68,6 @@ async def handle_message(data: dict, manager: ConnectionManager, uid: str, isAdm if ndcId: await on_ws_ndc_event(uid, ndcId) - if ( t == WS_TYPE_MARK_READ and o.get("markHasRead", None) is not None @@ -77,7 +81,27 @@ async def handle_message(data: dict, manager: ConnectionManager, uid: str, isAdm and ndcId is not None and chatId is not None ): + manager.subscribe_thread(uid, chatId) await on_chat_screen_open(uid, chatId, ndcId, manager) + await manager.answer({"t": 101, "o": {"id": ws_req_id, "ndcId": ndcId, "threadId": chatId}}, ws) + channel_members_now = manager.get_channel_members(chatId) + if channel_members_now: + channel_user_info = [{"channelUid": m["channelUid"], "joinRole": m["joinRole"], "isOffline": False, "userProfile": {"uid": m["uid"]}} for m in channel_members_now] + await manager.answer({"t": 102, "o": {"threadId": chatId, "userList": channel_user_info}}, ws) + active_channel_type = manager.get_channel_type(chatId) + if not active_channel_type and channel_members_now: + try: + _db = await Database().init() + _chat_info = await _db.get(f"x{ndcId}", "Chats").find_one({"id": chatId}, {"channelType": 1}) or {} + await _db.close() + _ct = _chat_info.get("channelType", 0) + if _ct: + manager.set_channel_type(chatId, _ct) + active_channel_type = _ct + except Exception: + pass + if active_channel_type: + await manager.answer({"t": 111, "o": {"threadId": chatId, "channelType": active_channel_type, "status": 0}}, ws) elif ( @@ -85,7 +109,9 @@ async def handle_message(data: dict, manager: ConnectionManager, uid: str, isAdm and ndcId is not None and chatId is not None ): + manager.unsubscribe_thread(uid, chatId) await on_chat_screen_close(uid, chatId, ndcId, manager) + await manager.answer({"t": 104, "o": {"id": ws_req_id, "ndcId": ndcId}}, ws) @@ -100,7 +126,7 @@ async def handle_message(data: dict, manager: ConnectionManager, uid: str, isAdm await on_chatting(uid, targetChatId, ndcId, manager) - elif actions == [ACTION_BROWSING]: #TODO + elif actions == [ACTION_BROWSING]: pass @@ -111,8 +137,188 @@ async def handle_message(data: dict, manager: ConnectionManager, uid: str, isAdm await on_chat_message_typing_end(uid, targetChatId, ndcId, manager) elif actions == [ACTION_CHATTING] and targetChatId: await on_chatting_end(uid, targetChatId, ndcId, manager) + elif actions == [ACTION_BROWSING]: + pass - elif actions == [ACTION_BROWSING]: #TODO - pass + elif t == WS_TYPE_JOIN_LIVE and ndcId is not None and chatId is not None: + join_role = o.get("joinRole", 1) + agora_uid = uid_from_uuid(uid) + + if join_role == 0: + members_before = manager.get_channel_members(chatId) + was_host = any(m["uid"] == uid and m["joinRole"] == 1 for m in members_before) + manager.remove_channel_member(chatId, uid) + remaining = manager.get_channel_members(chatId) + + await manager.broadcast_to_thread(chatId, WSObjects.ChannelUserLeave(chatId, uid)) + + if was_host or not remaining: + for m in remaining: + await manager.broadcast_to_thread(chatId, WSObjects.ChannelUserLeave(chatId, m["uid"])) + await manager.broadcast_to_thread(chatId, {"t": 102, "o": {"threadId": chatId, "userList": []}}) + manager.clear_channel_members(chatId) + import time as _time + from datetime import datetime, UTC + from uuid import uuid4 as _uuid4 + try: + db = await Database().init() + chat_table = await db.get(f"x{ndcId}", "Chats") + msg_collection = await db.get(f"x{ndcId}", f"_Chat:{chatId}") + msg_id = str(_uuid4()) + now_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + now_ms = int(_time.time() * 1000) + await msg_collection.insert_one({ + "messageId": msg_id, + "authorId": uid, + "messageType": 110, + "clientRefId": 0, + "content": None, + "mediaType": 0, + "mediaValue": None, + "timestamp": now_ms, + "extensions": {}, + "createdTime": now_iso, + }) + await chat_table.update_one( + {"id": chatId}, + {"$set": {"channelType": 0, "lastMessageId": msg_id, "lastMessageTimestamp": now_ms}} + ) + chat_info = await chat_table.find_one({"id": chatId}) or {} + await db.close() + members_list = chat_info.get("memberList", []) + chat_info.get("invitedList", []) + await manager.selective_broadcast({ + "t": 1000, + "o": { + "ndcId": ndcId, + "chatMessage": { + "messageId": msg_id, + "threadId": chatId, + "ndcId": ndcId, + "uid": uid, + "type": 110, + "content": None, + "mediaType": 0, + "mediaValue": None, + "createdTime": now_iso, + "clientRefId": 0, + "extensions": {}, + "includedInSummary": True, + "isHidden": False, + }, + "alertOption": 1, + "membershipStatus": 1, + }, + }, members_list) + except Exception as e: + print(f"[WS] Failed to end channel on host leave: {e}") + await manager.broadcast_to_thread(chatId, WSObjects.ChannelTypeUpdate(chatId, 0)) + else: + manager.subscribe_thread(uid, chatId) + manager.add_channel_member(chatId, uid, join_role, channel_uid=agora_uid) + await manager.answer(WSObjects.LiveChatJoin(ws_req_id, ndcId, chatId, join_role, uid, agora_uid), ws) + await manager.broadcast_to_thread(chatId, WSObjects.ChannelUserJoin(chatId, uid, join_role), exclude_uid=uid) + + + elif t == WS_FETCH_CHANNEL_USERS and ndcId is not None and chatId is not None: + members = manager.get_channel_members(chatId) + user_info_list = [{"channelUid": m["channelUid"], "joinRole": m["joinRole"], "isOffline": False, "userProfile": {"uid": m["uid"]}} for m in members] + await manager.answer({"t": 102, "o": {"id": ws_req_id, "ndcId": ndcId, "threadId": chatId, "userList": user_info_list}}, ws) + + + elif t == WS_TYPE_UPDATE_CHANNEL_TYPE and ndcId is not None and chatId is not None: + import time as _time + from datetime import datetime, UTC + from uuid import uuid4 as _uuid4 + from helpers.processors.cache import CacheProcessor + channel_type = o.get("channelType", 0) + MSG_TYPE = 107 if channel_type == 1 else 110 + + try: + db = await Database().init() + chat_table = await db.get(f"x{ndcId}", "Chats") + msg_collection = await db.get(f"x{ndcId}", f"_Chat:{chatId}") + msg_id = str(_uuid4()) + now_iso = datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%SZ") + now_ms = int(_time.time() * 1000) + system_msg = { + "messageId": msg_id, + "authorId": uid, + "messageType": MSG_TYPE, + "clientRefId": 0, + "content": None, + "mediaType": 0, + "mediaValue": None, + "timestamp": now_ms, + "extensions": {}, + "createdTime": now_iso, + } + await msg_collection.insert_one(system_msg) + + await chat_table.update_one( + {"id": chatId}, + {"$set": { + "channelType": channel_type, + "lastMessageId": msg_id, + "lastMessageTimestamp": now_ms, + }} + ) + + ws_msg_payload = { + "t": 1000, + "o": { + "ndcId": ndcId, + "chatMessage": { + "messageId": msg_id, + "threadId": chatId, + "ndcId": ndcId, + "uid": uid, + "type": MSG_TYPE, + "content": None, + "mediaType": 0, + "mediaValue": None, + "createdTime": now_iso, + "clientRefId": 0, + "extensions": {}, + "includedInSummary": True, + "isHidden": False, + }, + "alertOption": 1, + "membershipStatus": 1, + }, + } + chat_info = await chat_table.find_one({"id": chatId}) or {} + members = chat_info.get("memberList", []) + chat_info.get("invitedList", []) + await db.close() + await manager.selective_broadcast(ws_msg_payload, members) + except Exception as e: + print(f"[WS] Failed to insert system message for channelType change: {e}") + + if channel_type: + manager.set_channel_type(chatId, channel_type) + else: + manager.clear_channel_members(chatId) + + await manager.broadcast_to_thread(chatId, WSObjects.ChannelTypeUpdate(chatId, channel_type), exclude_uid=uid) + await manager.answer(WSObjects.ChannelTypeResponse(ws_req_id, ndcId, channel_type), ws) + + + elif t == WS_TYPE_GET_AGORA and ndcId is not None and chatId is not None: + import time as _time + expire_seconds = 3600 + agora_uid = uid_from_uuid(uid) + token = build_rtc_token( + Config.AGORA_APP_ID, + Config.AGORA_APP_CERTIFICATE, + chatId, + agora_uid, + ROLE_PUBLISHER, + expire_seconds, + ) + expired_time = int(_time.time()) + expire_seconds + await manager.answer(WSObjects.AgoraChannel(ws_req_id, ndcId, token, chatId, agora_uid, expired_time), ws) + + + elif t == WS_TYPE_TOPIC_SUBSCRIBE: + pass diff --git a/helpers/agora.py b/helpers/agora.py new file mode 100644 index 0000000..8904552 --- /dev/null +++ b/helpers/agora.py @@ -0,0 +1,25 @@ +import time +import zlib + +from agora_token_builder import RtcTokenBuilder + +ROLE_PUBLISHER = 1 +ROLE_SUBSCRIBER = 2 + + +def uid_from_uuid(user_uuid: str) -> int: + return zlib.crc32(user_uuid.encode()) & 0xFFFFFFFF + + +def build_rtc_token( + app_id: str, + app_certificate: str, + channel_name: str, + uid: int, + role: int, + expire_seconds: int = 3600, +) -> str: + expire_ts = int(time.time()) + expire_seconds + return RtcTokenBuilder.buildTokenWithUid( + app_id, app_certificate, channel_name, uid, role, expire_ts + ) diff --git a/helpers/config.py b/helpers/config.py index f7bfe1a..40af609 100644 --- a/helpers/config.py +++ b/helpers/config.py @@ -1,4 +1,4 @@ -from os import environ +from os import environ class Config: @@ -42,3 +42,6 @@ class Config: WS_ADMIN_VERIFY = environ.get("WS_ADMIN_VERIFY") PASSWORD_SALT = environ.get("PASSWORD_SALT") + + AGORA_APP_ID = environ.get("AGORA_APP_ID", "") + AGORA_APP_CERTIFICATE = environ.get("AGORA_APP_CERTIFICATE", "") diff --git a/helpers/connection_manager.py b/helpers/connection_manager.py index 6d4eebf..efa9115 100644 --- a/helpers/connection_manager.py +++ b/helpers/connection_manager.py @@ -1,162 +1,121 @@ +from fastapi import WebSocket +from typing import List, Dict, Optional, Set import asyncio import json -from typing import Dict, List, Optional from helpers.database.redis import get as get_redis -from fastapi import WebSocket -from redis import asyncio as aioredis -from starlette.websockets import WebSocketState -from .config import Config -PING_INTERVAL = 30 -PING_TIMEOUT = 10 class ConnectionManager: - CHANNEL_CMD = "ws:commands" - def __init__(self): self.active_connections: Dict[str, List[WebSocket]] = {} - self._ping_tasks: Dict[WebSocket, asyncio.Task] = {} - - self.pubsub_redis = None - self.pubsub = None - self._listener_task: Optional[asyncio.Task] = None - self._stopping = False - - async def start(self): - self._stopping = False - self._listener_task = asyncio.create_task(self._listen_forever()) - - async def stop(self): - self._stopping = True - if self._listener_task: - self._listener_task.cancel() - if self.pubsub: - await self.pubsub.unsubscribe() - await self.pubsub.close() - if self.pubsub_redis: - await self.pubsub_redis.close() + # thread_id -> set of uids subscribed to that thread + self.thread_subscribers: Dict[str, Set[str]] = {} + # thread_id -> list of active voice-channel members {uid, joinRole, channelUid} + self.channel_members: Dict[str, List[dict]] = {} + # thread_id -> active channelType (1=voice, 4=video, 5=screenroom); absent means no live + self.channel_types: Dict[str, int] = {} async def connect(self, websocket: WebSocket, uid: str): await websocket.accept() if uid not in self.active_connections: self.active_connections[uid] = [] self.active_connections[uid].append(websocket) - task = asyncio.create_task(self._ping_loop(websocket, uid)) - self._ping_tasks[websocket] = task def disconnect(self, websocket: WebSocket, uid: str): - task = self._ping_tasks.pop(websocket, None) - if task: - task.cancel() - - conns = self.active_connections.get(uid) - if conns is None: - return - try: - conns.remove(websocket) - except ValueError: - pass - if not conns: + self.active_connections[uid].remove(websocket) + if not self.active_connections[uid]: del self.active_connections[uid] - - async def _ping_loop(self, websocket: WebSocket, uid: str): - try: - while True: - await asyncio.sleep(PING_INTERVAL) - if websocket.client_state != WebSocketState.CONNECTED: - break - try: - await asyncio.wait_for( - websocket.send_json({"t": "ping"}), - timeout=PING_TIMEOUT, - ) - except (asyncio.TimeoutError, Exception): - try: - await websocket.close(code=1001) - except Exception: - pass - break - except asyncio.CancelledError: - pass + # Remove uid from all thread subscriptions if no more connections + if uid not in self.active_connections: + for subs in self.thread_subscribers.values(): + subs.discard(uid) + # Remove from voice channel members too + for members in self.channel_members.values(): + members[:] = [m for m in members if m["uid"] != uid] + + def subscribe_thread(self, uid: str, thread_id: str): + if thread_id not in self.thread_subscribers: + self.thread_subscribers[thread_id] = set() + self.thread_subscribers[thread_id].add(uid) + + def unsubscribe_thread(self, uid: str, thread_id: str): + if thread_id in self.thread_subscribers: + self.thread_subscribers[thread_id].discard(uid) + + def add_channel_member(self, thread_id: str, uid: str, join_role: int = 1, channel_uid: int = 0): + if thread_id not in self.channel_members: + self.channel_members[thread_id] = [] + # Remove any existing entry for this uid first + self.channel_members[thread_id] = [m for m in self.channel_members[thread_id] if m["uid"] != uid] + self.channel_members[thread_id].append({"uid": uid, "joinRole": join_role, "channelUid": channel_uid}) + + def remove_channel_member(self, thread_id: str, uid: str): + if thread_id in self.channel_members: + self.channel_members[thread_id] = [m for m in self.channel_members[thread_id] if m["uid"] != uid] + if not self.channel_members[thread_id]: + del self.channel_members[thread_id] + + def clear_channel_members(self, thread_id: str): + self.channel_members.pop(thread_id, None) + self.channel_types.pop(thread_id, None) + + def get_channel_members(self, thread_id: str) -> List[dict]: + return list(self.channel_members.get(thread_id, [])) + + def set_channel_type(self, thread_id: str, channel_type: int): + if channel_type: + self.channel_types[thread_id] = channel_type + else: + self.channel_types.pop(thread_id, None) + + def get_channel_type(self, thread_id: str) -> int: + return self.channel_types.get(thread_id, 0) async def answer(self, message: dict, websocket: WebSocket): await websocket.send_json(message) - async def _local_broadcast(self, message: dict): - for connections in self.active_connections.values(): - for connection in connections: - try: - await connection.send_json(message) - except Exception: - pass + async def start(self): + pass # no-op; kept for compatibility with main.py lifespan + + async def stop(self): + pass # no-op - async def _local_selective_broadcast(self, message: dict, uids: List[str]): + async def broadcast_to_thread(self, thread_id: str, message: dict, exclude_uid: str = None): + uids = self.thread_subscribers.get(thread_id, set()) tasks = [] for uid in uids: - if uid in self.active_connections: - for connection in self.active_connections[uid]: - tasks.append(connection.send_json(message)) + if uid == exclude_uid: + continue + for conn in self.active_connections.get(uid, []): + tasks.append(conn.send_json(message)) if tasks: await asyncio.gather(*tasks, return_exceptions=True) - async def _connect_pubsub(self): - self.pubsub_redis = aioredis.from_url( - Config.REDIS_CONNECTION_STRING, - decode_responses=True, - socket_timeout=None, - socket_connect_timeout=5, - health_check_interval=25, - retry_on_timeout=True, - ) - self.pubsub = self.pubsub_redis.pubsub() - await self.pubsub.subscribe(self.CHANNEL_CMD) - - async def _listen_forever(self): - backoff = 1 - while not self._stopping: - try: - await self._connect_pubsub() - print("redis pub/sub connected") - backoff = 1 - await self._listen() - except asyncio.CancelledError: - raise - except Exception as e: - print(f"[WS listener error] {e!r}, reconnecting in {backoff}s") - try: - if self.pubsub: - await self.pubsub.close() - if self.pubsub_redis: - await self.pubsub_redis.close() - except Exception: - pass - await asyncio.sleep(backoff) - backoff = min(backoff * 2, 30) - - async def _listen(self): - async for raw in self.pubsub.listen(): - if raw["type"] != "message": - continue - try: - cmd = json.loads(raw["data"]) - except (TypeError, ValueError): - print("decode error") - continue - - message = cmd.get("message") - uids = cmd.get("uids") + async def selective_broadcast(self, message: dict, uids: List[str]): + tasks = [] + for uid in uids: + for conn in self.active_connections.get(uid, []): + tasks.append(conn.send_json(message)) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) - if uids: - await self._local_selective_broadcast(message, uids) - else: - await self._local_broadcast(message) +# Module-level helper used by handlers/chat.py (mirrors the Redis-based version +# in the upstream but falls back to a direct broadcast via the singleton manager). +_manager_ref: Optional["ConnectionManager"] = None async def broadcast_ws_message(message: dict, uids: Optional[List[str]] = None): - redis = get_redis() - payload = {"message": message} + """Send a WS message to specific uids (or all if uids is None).""" + if _manager_ref is None: + return if uids is not None: - payload["uids"] = uids - await redis.publish(ConnectionManager.CHANNEL_CMD, json.dumps(payload)) \ No newline at end of file + await _manager_ref.selective_broadcast(message, uids) + else: + for connections in _manager_ref.active_connections.values(): + for conn in connections: + try: + await conn.send_json(message) + except Exception: + pass diff --git a/helpers/constants.py b/helpers/constants.py index 08cf753..3556df2 100644 --- a/helpers/constants.py +++ b/helpers/constants.py @@ -1,4 +1,4 @@ -# WebSocket message types +# WebSocket message types WS_ERROR_MESSAGE = 1 WS_EXCEPTION_MESSAGE = 305 WS_NOTIFICATION_MESSAGE = 10 @@ -10,6 +10,12 @@ WS_CHAT_SCREEN_CLOSE = 103 #close chat WS_ACTION_START = 304 WS_ACTION_END = 306 +WS_FETCH_CHANNEL_USERS = 105 # FETCH_THREAD_CHANNEL_USER_LIST → responds t=102 +WS_TYPE_JOIN_LIVE = 112 +WS_TYPE_UPDATE_CHANNEL_TYPE = 108 # host starts/stops live (channelType=1=voice,0=off) +WS_TYPE_GET_AGORA = 200 +WS_TYPE_LEAVE_LIVE = 107 +WS_TYPE_TOPIC_SUBSCRIBE = 300 @@ -33,4 +39,5 @@ TTL_VIEWER = 40 TTL_CHATTING = 120 TTL_COMMUNITY_ONLINE = 300 - \ No newline at end of file +CHAT_VIEWER_TTL_SECONDS = 40 + diff --git a/helpers/wsobjs.py b/helpers/wsobjs.py index 6d463b8..4d42615 100644 --- a/helpers/wsobjs.py +++ b/helpers/wsobjs.py @@ -1,4 +1,4 @@ -from datetime import datetime, UTC +from datetime import datetime, UTC from orjson import dumps from helpers.constants import WS_TYPE_ERROR @@ -56,3 +56,101 @@ def InternalWSError(): @staticmethod def Pong(ws_req_id: str | None = None) -> dict: return {"t": 117, "o": {"id": ws_req_id, "threadChannelUserInfoList": []}} + + @staticmethod + def LiveChatJoin(ws_req_id, ndcId: int, thread_id: str, join_role: int, uid: str, channel_uid: int) -> dict: + # Response to UPDATE_USER_ROLE_REQUEST (0x70=112) ÔåÆ UPDATE_USER_ROLE_RESPONSE (0x71=113) + # threadId: required so SignallingService.onWsMessage(t=113) can call getChannelByThread(threadId) + # user.channelUid: sets SignallingChannel.channelUid (local Agora uid) + # user.userProfile.uid: used by uid() for deduplication in userList + return { + "t": 113, + "o": { + "id": ws_req_id, + "ndcId": ndcId, + "threadId": thread_id, + "user": { + "joinRole": join_role, + "channelUid": channel_uid, + "isHost": join_role == 1, + "isOffline": False, + "userProfile": {"uid": uid}, + }, + }, + } + + @staticmethod + def ChannelUserJoin(thread_id: str, user_uid: str, join_role: int, channel_uid: int = 0) -> dict: + return { + "t": 106, + "o": { + "threadId": thread_id, + "user": { + "channelUid": channel_uid, + "isHost": join_role == 1, + "isOffline": False, + "joinRole": join_role, + "userProfile": {"uid": user_uid}, + }, + }, + } + + @staticmethod + def ChannelUserLeave(thread_id: str, user_uid: str) -> dict: + return { + "t": 107, + "o": { + "threadId": thread_id, + "user": { + "channelUid": 0, + "isOffline": True, + "joinRole": 0, + "userProfile": {"uid": user_uid}, + }, + }, + } + + @staticmethod + def ChannelTypeUpdate(thread_id: str, channel_type: int, status: int = 0) -> dict: + return { + "t": 111, + "o": { + "threadId": thread_id, + "channelType": channel_type, + "status": status, + }, + } + + @staticmethod + def ChannelTypeResponse(ws_req_id, ndcId: int, channel_type: int) -> dict: + # Response to UPDATE_THREAD_CHANNEL_REQUEST (0x6c=108) ÔåÆ UPDATE_THREAD_CHANNEL_RESPONSE (0x6d=109) + return { + "t": 109, + "o": { + "id": ws_req_id, + "ndcId": ndcId, + "channelType": channel_type, + }, + } + + @staticmethod + def AgoraChannel( + ws_req_id, + ndcId: int, + channel_key: str, + channel_name: str, + channel_uid: int, + expired_time: int, + ) -> dict: + # Response to AGORA_TOKEN_REQUEST (0xc8=200) ÔåÆ AGORA_TOKEN_RESPONSE (0xc9=201) + return { + "t": 201, + "o": { + "id": ws_req_id, + "ndcId": ndcId, + "channelKey": channel_key, + "channelName": channel_name, + "channelUid": channel_uid, + "expiredTime": expired_time, + }, + } diff --git a/main.py b/main.py index e4e1926..8946737 100644 --- a/main.py +++ b/main.py @@ -4,8 +4,10 @@ from objects.errors import Errors from handlers import handle_message from helpers.connection_manager import ConnectionManager +import helpers.connection_manager as _cm_module manager = ConnectionManager() +_cm_module._manager_ref = manager @asynccontextmanager async def lifespan(app: FastAPI): diff --git a/pyproject.toml b/pyproject.toml index df3b751..a956e28 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "redis>=7.4.0", "uvicorn[standard]>=0.42.0", "websockets>=16.0", + "agora-token-builder>=1.0.0", ] [dependency-groups] diff --git a/uv.lock b/uv.lock index a3e6c00..7a2162f 100644 --- a/uv.lock +++ b/uv.lock @@ -2,6 +2,12 @@ version = 1 revision = 3 requires-python = ">=3.13" +[[package]] +name = "agora-token-builder" +version = "1.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b9/55/6741101e6169f7c44388f1184b845db956ccfaac23356f49077418042885/agora_token_builder-1.0.0.tar.gz", hash = "sha256:6a22a35baf748dc80ca054ba7e293ba16c22dc7eae85396ffb41e80f13f9c2f1", size = 4813, upload-time = "2021-12-29T02:53:13.415Z" } + [[package]] name = "annotated-doc" version = "0.0.4" @@ -572,6 +578,7 @@ name = "websocket" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "agora-token-builder" }, { name = "fastapi" }, { name = "marshmallow" }, { name = "motor" }, @@ -589,6 +596,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "agora-token-builder", specifier = ">=1.0.0" }, { name = "fastapi", specifier = ">=0.135.2" }, { name = "marshmallow", specifier = "<4.0" }, { name = "motor", specifier = ">=3.7.1" },