Add queue leaving

Author: jq1836

frontend/components/matching/find-match.tsx

@@ -6,6 +6,9 @@ import { SelectionSummary } from "@/components/matching/selection-summary";
 import { useToast } from "@/components/hooks/use-toast";
 import { useAuth } from "@/app/auth/auth-context";
 import { joinMatchQueue } from "@/lib/join-match-queue";
+import { leaveMatchQueue } from "@/lib/leave-match-queue";
+import { matchingServiceWebSockUri } from "@/lib/api-uri";
+import { subscribeMatch } from "@/lib/subscribe-match";
 
 export default function FindMatch() {
   const [selectedDifficulty, setSelectedDifficulty] = useState<string>("");
@@ -71,17 +74,20 @@ export default function FindMatch() {
       selectedDifficulty
     );
     switch (response.status) {
-      case 200:
+      case 201:
         toast({
           title: "Matched",
           description: "Successfully matched",
           variant: "success",
         });
         return;
       case 202:
+      case 304:
         setIsSearching(true);
-        const ws = new WebSocket(
-          `ws://localhost:6969/match/subscribe/${auth?.user?.id}/${selectedTopic}/${selectedDifficulty}`
+        const ws = await subscribeMatch(
+          auth?.user.id,
+          selectedTopic,
+          selectedDifficulty
         );
         ws.onmessage = () => {
           setIsSearching(false);
@@ -90,7 +96,16 @@ export default function FindMatch() {
             description: "Successfully matched",
             variant: "success",
           });
+          ws.onclose = () => null;
         };
+        ws.onclose = () => {
+          setIsSearching(false);
+          toast({
+            title: "Matching Stopped",
+            description: "Matching has been stopped",
+            variant: "destructive",
+          });
+        }
         setWebsocket(ws);
         return;
       default:
@@ -103,9 +118,54 @@ export default function FindMatch() {
     }
   };
 
-  const handleCancel = () => {
-    setIsSearching(false);
-    setWaitTime(0);
+  const handleCancel = async () => {
+    if (!selectedDifficulty || !selectedTopic) {
+      toast({
+        title: "Invalid Selection",
+        description: "Please select both a difficulty level and a topic",
+        variant: "destructive",
+      });
+      return;
+    }
+
+    if (!auth || !auth.token) {
+      toast({
+        title: "Access denied",
+        description: "No authentication token found",
+        variant: "destructive",
+      });
+      return;
+    }
+
+    if (!auth.user) {
+      toast({
+        title: "Access denied",
+        description: "Not logged in",
+        variant: "destructive",
+      });
+      return;
+    }
+
+    const response = await leaveMatchQueue(auth.token, auth.user?.id, selectedTopic, selectedDifficulty);
+    switch (response.status) {
+      case 200:
+        setIsSearching(false);
+        setWaitTime(0);
+        setWebsocket(undefined);
+        toast({
+          title: "Success",
+          description: "Successfully left queue",
+          variant: "success",
+        });
+        return;
+      default:
+        toast({
+          title: "Unknown Error",
+          description: "An unexpected error has occured",
+          variant: "destructive",
+        });
+        return;
+    }
   };
 
   return (

frontend/lib/api-uri.ts

@@ -1,9 +1,15 @@
 const constructUri = (baseUri: string, port: string | undefined) =>
   `http://${process.env.NEXT_PUBLIC_BASE_URI || baseUri}:${port}`;
 
+const constructWebSockUri = (baseUri: string, port: string | undefined) =>
+  `ws://${process.env.NEXT_PUBLIC_BASE_URI || baseUri}:${port}`;
+
 export const userServiceUri: (baseUri: string) => string = (baseUri) =>
   constructUri(baseUri, process.env.NEXT_PUBLIC_USER_SVC_PORT);
 export const questionServiceUri: (baseUri: string) => string = (baseUri) =>
   constructUri(baseUri, process.env.NEXT_PUBLIC_QUESTION_SVC_PORT);
 export const matchingServiceUri: (baseUri: string) => string = (baseUri) =>
   constructUri(baseUri, process.env.NEXT_PUBLIC_MATCHING_SVC_PORT);
+
+export const matchingServiceWebSockUri: (baseUri: string) => string = (baseUri) =>
+  constructWebSockUri(baseUri, process.env.NEXT_PUBLIC_MATCHING_SVC_PORT);

frontend/lib/leave-match-queue.ts

@@ -0,0 +1,24 @@
+import { matchingServiceUri } from "@/lib/api-uri";
+
+export const leaveMatchQueue = async (
+  jwtToken: string,
+  userId: string,
+  category: string,
+  complexity: string
+) => {
+  const params = new URLSearchParams({
+    topic: category,
+    difficulty: complexity,
+  }).toString();
+  const response = await fetch(
+    `${matchingServiceUri(window.location.hostname)}/match/queue/${userId}?${params}`,
+    {
+      method: "DELETE",
+      headers: {
+        Authorization: `Bearer ${jwtToken}`,
+        "Content-Type": "application/json",
+      },
+    }
+  );
+  return response;
+};

frontend/lib/subscribe-match.ts

@@ -0,0 +1,15 @@
+import { matchingServiceWebSockUri } from "@/lib/api-uri";
+
+export const subscribeMatch = async (
+  userId: string,
+  category: string,
+  complexity: string
+) => {
+  const params = new URLSearchParams({
+    topic: category,
+    difficulty: complexity,
+  })
+  return new WebSocket(
+    `${matchingServiceWebSockUri(window.location.hostname)}/match/subscribe/${userId}?${params}`
+  );
+};

matching-service/app/logic/matching.py

@@ -1,78 +1,62 @@
+from fastapi.responses import JSONResponse, Response
 from typing import Union
 
 from models.match import MatchModel, MessageModel
-from utils.redis import get_redis, acquire_lock, release_lock
+from utils.redis import acquire_lock, redis_client, release_lock
 from utils.socketmanager import manager
 
 async def find_match_else_enqueue(
     user_id: str,
     topic: str,
     difficulty: str
-) -> Union[MessageModel, MatchModel]:
-    redis_client = await get_redis()
+) -> Union[Response, JSONResponse]:
     queue_key = _build_queue_key(topic, difficulty)
-
-    result = None
-
-    # ACQUIRE LOCK
     islocked = await acquire_lock(redis_client, queue_key)
 
     if not islocked:
         raise Exception("Could not acquire lock")
 
     # Check if the user is already in the queue
-    user_in_queue = await redis_client.lrange(queue_key, 0, -1)
-    if user_id in user_in_queue:
-        result = MessageModel(
-            message=f"User {user_id} is already in the queue, waiting for a match"
-        )
-    else:
-        queue_length = await redis_client.llen(queue_key)
-        if queue_length > 0:
-            matched_user = await redis_client.rpop(queue_key)
-            result = MatchModel(
-                user1=user_id,
-                user2=matched_user,
-                topic=topic,
-                difficulty=difficulty,
-            )
-            await manager.broadcast(matched_user, topic, difficulty, result.json())
-            # await manager.disconnect(matched_user, topic, difficulty)
-        else:
-            await redis_client.lpush(queue_key, user_id)
-            result = MessageModel(
-                message=f"User {user_id} enqueued, waiting for a match"
-            )
+    if user_id in await redis_client.lrange(queue_key, 0, -1):
+        await release_lock(redis_client, queue_key)
+        return Response(status_code=304)
 
-    # RELEASE LOCK
+    # Check if there are no other users in the queue
+    if await redis_client.llen(queue_key) == 0:
+        await redis_client.lpush(queue_key, user_id)
+        await release_lock(redis_client, queue_key)
+        return Response(status_code=202)
+    
+    # There is a user in the queue
+    matched_user = await redis_client.rpop(queue_key)
     await release_lock(redis_client, queue_key)
-    return result
+    response = MatchModel(
+        user1=matched_user,
+        user2=user_id,
+        topic=topic,
+        difficulty=difficulty,
+    )
+    await manager.broadcast(matched_user, topic, difficulty, response.json())
+    await manager.disconnect_all(matched_user, topic, difficulty)
+    return JSONResponse(status_code=201, content=response.json())
 
 async def remove_user_from_queue(
     user_id: str,
     topic: str,
     difficulty: str
-) -> MessageModel:
-    redis_client = await get_redis()
+) -> Response:
     queue_key = _build_queue_key(topic, difficulty)
-
-    # ACQUIRE LOCK
     islocked = await acquire_lock(redis_client, queue_key)
 
     if not islocked:
         raise Exception("Could not acquire lock")
 
-    # Check if the user is already in the queue
-    user_in_queue = await redis_client.lrange(queue_key, 0, -1)
-    if user_id in user_in_queue:
+    if user_id in await redis_client.lrange(queue_key, 0, -1):
         await redis_client.lrem(queue_key, 0, user_id)
-        result = MessageModel(message=f"User {user_id} removed from the queue")
-    else:
-        result = MessageModel(message=f"User {user_id} is not in the queue")
 
-    # RELEASE LOCK
     await release_lock(redis_client, queue_key)
-    return result
+    await manager.disconnect_all(user_id, topic, difficulty)
+    return Response(status_code=200)
 
 '''
 Helper functions for matching.

matching-service/app/routers/match.py

@@ -1,6 +1,5 @@
 import asyncio
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
-from fastapi.responses import JSONResponse
 from typing import Union
 
 from logic.matching import find_match_else_enqueue, remove_user_from_queue
@@ -16,11 +15,7 @@ async def enqueue_user(
     topic: str,
     difficulty: str
 ):
-    response = await find_match_else_enqueue(user_id, topic, difficulty)
-    if isinstance(response, MatchModel):
-        return JSONResponse(status_code=200, content=response.json())
-    elif isinstance(response, MessageModel):
-        return JSONResponse(status_code=202, content=response.json())
+    return await find_match_else_enqueue(user_id, topic, difficulty)
 
 # Remove a user from the queue
 @router.delete("/queue/{user_id}")
@@ -29,12 +24,9 @@ async def dequeue_user(
     topic: str,
     difficulty: str
 ):
-    return JSONResponse(
-        status_code=200,
-        content=await remove_user_from_queue(user_id, topic, difficulty)
-    )
+    return await remove_user_from_queue(user_id, topic, difficulty)
 
-@router.websocket("/subscribe/{user_id}/{topic}/{difficulty}")
+@router.websocket("/subscribe/{user_id}")
 async def subscribe(
     websocket: WebSocket,
     user_id: str,
@@ -47,4 +39,4 @@ async def subscribe(
         while True:
             await websocket.receive_text()
     except WebSocketDisconnect:
-        pass
+        manager.disconnect(user_id, topic, difficulty, websocket)

matching-service/app/utils/redis.py

@@ -2,13 +2,6 @@
 import os
 import redis.asyncio as redis
 
-# Initialize Redis client
-async def get_redis():
-    redis_host = os.getenv("REDIS_HOST", "localhost")
-    redis_port = os.getenv("REDIS_PORT", "6379")
-    return redis.Redis(host=redis_host, port=int(redis_port), db=0, decode_responses=True)
-
-
 async def acquire_lock(redis_client, key, lock_timeout_ms=30000, retry_interval_ms=100, max_retries=100) -> bool:
     lock_key = f"{key}:lock"
     retries = 0
@@ -27,3 +20,10 @@ async def acquire_lock(redis_client, key, lock_timeout_ms=30000, retry_interval_
 async def release_lock(redis_client, key) -> None:
     lock_key = f"{key}:lock"
     await redis_client.delete(lock_key)
+
+redis_client = redis.Redis(
+    host=os.environ.get("REDIS_HOST", "localhost"),
+    port=int(os.environ.get("REDIS_PORT", "6379")),
+    db=0,
+    decode_responses=True
+)

matching-service/app/utils/socketmanager.py

@@ -37,22 +37,40 @@ async def connect(
     '''
     Disconnects all connections associated with (user_id, topic, complexity)
     '''
-    async def disconnect(
+    async def disconnect_all(
         self,
         user_id: str,
         topic: str,
         complexity: str,
-        websocket: WebSocket,
     ) -> None:
         key: Tuple[str, str, str] = (user_id, topic, complexity)
-        if not key in self.connection_map:
+        if key not in self.connection_map:
             return
 
         await asyncio.gather(
-            *[websocket for websocket in self.connection_map[key]]
+            *[websocket.close() for websocket in self.connection_map[key]]
         )
         del self.connection_map[key]
 
+    '''
+    Disconnects the single connection.
+    '''
+    async def disconnect(
+        self,
+        user_id: str,
+        topic: str,
+        complexity: str,
+        websocket: WebSocket,
+    ):
+        key: Tuple[str, str, str] = (user_id, topic, complexity)
+        if key not in self.connection_map:
+            return
+
+        self.connection_map[key].remove(websocket)
+        if len(self.connection_map[key]) == 0:
+            del self.connections_map[key]
+        websocket.close()
+
     '''
     Data is sent to through all connections associated with
     (user_id, topic, complexity)
@@ -65,7 +83,7 @@ async def broadcast(
         data: str,
     ):
         key: Tuple[str, str, str] = (user_id, topic, complexity)
-        if not key in self.connection_map:
+        if key not in self.connection_map:
             return
 
         await asyncio.gather(