Feature/matching service/queue (#165)

* Set up basic queuing

* Abstract logic

* Update env sample

* Unfug the race conditions

* Made the lock a spin lock with multiple retries.
Add test.py to test the correctness of the code.

* Add delete user from queue feature

* Externalise docker variables

* Update env sample

---------

Co-authored-by: Selwyn Ang <[email protected]>

Author: TanPingZhi

.env.sample

@@ -18,3 +18,12 @@ USER_SVC_DB_URI=
 JWT_SECRET=
 EMAIL_ADDRESS=
 EMAIL_PASSWORD=
+
+## Matching service variables
+MATCHING_SVC_PORT=6969
+
+## Redis variables
+REDIS_PORT=6379
+
+## Redisinsight variables
+REDIS_INSIGHT_PORT=5540

docker-compose.yml

@@ -44,9 +44,23 @@ services:
       - $MATCHING_SVC_PORT:$MATCHING_SVC_PORT
     environment:
       - PORT=$MATCHING_SVC_PORT
+      - REDIS_HOST=redis
+      - REDIS_PORT=$REDIS_PORT
+    depends_on:
+      - redis
 
   redis:
     image: redis:7.4-alpine
     restart: always
     ports:
-      - 6379:6379
+      - $REDIS_PORT:$REDIS_PORT
+
+  # access RedisInsight at http://localhost:5540
+  # connect to redis on redis insight at redis:6379
+  redisinsight:
+    image: redis/redisinsight:latest
+    restart: always
+    ports:
+      - $REDIS_INSIGHT_PORT:$REDIS_INSIGHT_PORT  # Expose RedisInsight UI on port 5540
+    depends_on:
+      - redis

matching-service/.gitignore

@@ -1,3 +1,2 @@
 .venv
-app/__pycache__
-app/routers/__pycache__
+**/__pycache__

matching-service/app/__init__.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+# Define the MatchData model
+class MatchData(BaseModel):
+    user1: str
+    user2: str
+    topic: str
+    difficulty: str

matching-service/app/models/__init__.py

@@ -1,8 +1,38 @@
+from utils.redis_utils import get_redis, build_queue_key, listen_for_matches, find_match_else_enqueue, match_channel, remove_user_from_queue
+from models.match import MatchData
 from fastapi import APIRouter
+import asyncio
 
 router = APIRouter()
 
-# Make a dummy endpoint to test the connection
-@router.get("/")
-async def test_connection():
-    return {"message": "Connection successful"}
+# List to store matched pairs
+matched_pairs = []
+
+# Start the match listener in a background task
+
+
+@router.on_event("startup")
+async def startup_event():
+    redis_client = await get_redis()
+    asyncio.create_task(listen_for_matches(redis_client))
+
+# Add a user to the queue
+
+
+@router.post("/queue/{user_id}")
+async def enqueue_user(user_id: str, topic: str, difficulty: str):
+    # redis_client = await get_redis()
+    return await find_match_else_enqueue(user_id, topic, difficulty)
+
+# Get all matched pairs
+
+
+@router.get("/matches")
+async def get_matched_pairs():
+    return {"matched_pairs": matched_pairs}
+
+# Remove a user from the queue
+@router.delete("/queue/{user_id}")
+async def dequeue_user(user_id: str, topic: str, difficulty: str):
+    print("removing user from queue")
+    return await remove_user_from_queue(user_id, topic, difficulty)

matching-service/app/models/match.py

@@ -0,0 +1,111 @@
+import os
+import redis.asyncio as redis
+from models.match import MatchData
+import asyncio
+match_channel = 'match_channel'
+
+# 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
+
+    while retries < max_retries:
+        locked = await redis_client.set(lock_key, "locked", nx=True, px=lock_timeout_ms)
+        if locked:
+            return True
+        else:
+            retries += 1
+            # Convert ms to seconds
+            await asyncio.sleep(retry_interval_ms / 1000)
+
+    return False
+
+
+async def release_lock(redis_client, key) -> None:
+    lock_key = f"{key}:lock"
+    await redis_client.delete(lock_key)
+
+
+# Helper function to build a unique queue key based on topic and difficulty
+def build_queue_key(topic, difficulty):
+    return f"{topic}:{difficulty}"
+
+# Asynchronous task to listen for matches
+
+
+async def listen_for_matches(redis_client):
+    pubsub = redis_client.pubsub()
+    await pubsub.subscribe(match_channel)
+    async for message in pubsub.listen():
+        if message["type"] == "message":
+            print(f"Match notification: {message['data']}")
+
+# Asynchronous matching logic
+
+
+async def find_match_else_enqueue(user_id, topic, difficulty):
+    redis_client = await get_redis()
+    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 = {"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)
+            match_data = MatchData(
+                user1=user_id, user2=matched_user, topic=topic, difficulty=difficulty)
+            await redis_client.publish(match_channel, match_data.json())
+            result = {"message": "Match found", "match": match_data}
+        else:
+            await redis_client.lpush(queue_key, user_id)
+            result = {"message": f"User {
+                user_id} enqueued, waiting for a match"}
+
+    # RELEASE LOCK
+    await release_lock(redis_client, queue_key)
+    return result
+
+
+async def remove_user_from_queue(user_id, topic, difficulty):
+    redis_client = await get_redis()
+    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:
+        await redis_client.lrem(queue_key, 0, user_id)
+        result = {"message": f"User {
+            user_id} removed from the queue"}
+    else:
+        result = {"message": f"User {
+            user_id} is not in the queue"}
+
+    # RELEASE LOCK
+    await release_lock(redis_client, queue_key)
+    return result