Merge pull request #34 from CS3219-AY2526Sem1/collab/basic-realtime

Basic Implementation of Collaborative Service

Author: kokerinks

.gitignore

@@ -1,3 +1,7 @@
+# Virtual environments
+**/venv/
+
+.pytest_cache/
 **/__pycache__/
 
 # Ignore environment variables file

services/collaboration-service/README.md

@@ -0,0 +1,13 @@
+## To run locally:
+`uvicorn app.main:app --reload`
+## Planning
+
+## APIs to implement:
+
+From Matching (REST)
+[ ] /sessions
+    - 
+
+From UI (Websocket)
+[ ] /
+! make session_id non-determined from user ids (for security purpose; hacker cannot hijack session solely from user data)

services/collaboration-service/app/routers/__init__.py renamed to services/collaboration-service/app/__init__.py

@@ -0,0 +1,79 @@
+from typing import Dict
+import uuid 
+import asyncio
+from app.schemas.messages import CollaboratorConnectMessage, CollaboratorDisconnectMessage, DisplayMessage
+from app.core.errors import SessionNotFoundError, UserNotFoundError, InvalidUserIDsError
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: Dict[str, Dict[str, asyncio.Queue]] = {}
+        self.condition = asyncio.Condition()
+    
+    def init_session(self, user_ids: list[str]) -> str:
+        if len(user_ids) != 2:
+            raise InvalidUserIDsError()
+        session_id = self.generate_uuid(user_ids[0], user_ids[1])
+
+        self.active_connections[session_id] = {}
+        for user_id in user_ids:
+            self.active_connections[session_id][user_id] = None
+        return session_id
+    
+    def _get_collaborator_q(self, session_id, user_id):
+        session_data = self.active_connections[session_id]
+        id = next((uid for uid in session_data if uid != user_id), None)
+        if id is None:
+            return None
+        return session_data[id]
+        
+    async def on_connect(self, session_id: str, user_id: str, out_q: asyncio.Queue):
+        if session_id not in self.active_connections:
+            raise SessionNotFoundError()
+
+        if user_id not in self.active_connections[session_id]:
+            raise UserNotFoundError()
+
+        self.active_connections[session_id][user_id] = out_q
+
+        collaborator_q = self._get_collaborator_q(session_id, user_id)
+
+        if collaborator_q is None:
+            return None
+        
+        ## Notify collaborator that this user has connected
+        await collaborator_q.put(CollaboratorConnectMessage())
+        
+        ## Notify this user that the other user has already connected
+        await out_q.put(CollaboratorConnectMessage())
+
+    async def on_disconnect(self, session_id: str, user_id: str):
+        if session_id not in self.active_connections:
+            raise SessionNotFoundError()
+
+        if user_id not in self.active_connections[session_id]:
+            raise UserNotFoundError()
+
+        self.active_connections[session_id][user_id] = None
+
+        collaborator_q = self._get_collaborator_q(session_id, user_id)
+
+        if collaborator_q is None:
+            return None
+        
+        await collaborator_q.put(CollaboratorDisconnectMessage())
+
+    async def on_message(self, session_id: str, user_id: str, msg: str):
+        collaborator_q = self._get_collaborator_q(session_id, user_id)
+
+        if collaborator_q is None:
+            return None
+
+        await collaborator_q.put(DisplayMessage(msg=msg))
+
+    def generate_uuid(self, user1: str, user2: str) -> str:
+        namespace = uuid.NAMESPACE_DNS  
+        seed = ''.join(sorted([user1, user2]))
+        return str(uuid.uuid5(namespace, seed))
+
+
+connection_manager = ConnectionManager()

services/collaboration-service/app/core/connection_manager.py

@@ -0,0 +1,16 @@
+
+
+class InvalidUserIDsError(Exception):
+    """Exception raised when the provided user IDs are invalid."""
+    msg = "user_ids must contain exactly two user IDs."
+    pass
+
+class UserNotFoundError(Exception):
+    """Exception raised when a user is not found in the system."""
+    msg = "user_id not available in this session"  
+    pass
+
+class SessionNotFoundError(Exception):
+    """Exception raised when a session is not found."""
+    msg = "Session not initialized"
+    pass

services/collaboration-service/app/core/errors.py

@@ -0,0 +1,9 @@
+from enum import Enum
+
+class UserState(Enum):
+    AWAIT_CONNECT = "awaiting_connection"
+    AWAIT_POLLING = "awaiting_polling"
+
+
+
+    

services/collaboration-service/app/core/users.py

@@ -1,8 +1,12 @@
 from fastapi import FastAPI
+from app.ws import sessions as ws_sessions
+from app.rest import sessions
 
 app = FastAPI()
+app.include_router(sessions.router, prefix="/sessions")
+app.include_router(ws_sessions.router, prefix="/ws/sessions")
 
 ## API for testing connection
-@app.get("/ping")
-def ping():
-    return {"message": "Pong"}
+@app.get("/health")
+def health_check():
+    return {"status": "Healthy"}

services/collaboration-service/app/main.py

@@ -0,0 +1,15 @@
+
+from fastapi import APIRouter
+from app.schemas.collaboration import CreateSessionRequest, CreateSessionResponse
+from app.core.connection_manager import connection_manager
+
+router = APIRouter()
+
+@router.post("/", response_model=CreateSessionResponse)
+def create_session(request: CreateSessionRequest):
+    user_ids = request.user_ids
+    session_id = connection_manager.init_session(user_ids)
+    return CreateSessionResponse(session_id=session_id)
+
+
+