Merge branch 'frontend-websocket' of https://github.com/CS3219-AY2425S1/cs3219-ay2425s1-project-g24 into frontend-websocket

Author: chiaryan

apps/matching-service/README.md

@@ -29,7 +29,7 @@ go mod tidy
 - `MATCH_TIMEOUT`: The time in seconds to wait for a match before timing out.
 - `REDIS_URL`: The URL for the Redis server. Default is `localhost:6379`.
 
-4. Start a local redis server:
+4. Start a local Redis server:
 
 ```bash
 docker run -d -p 6379:6379 redis
@@ -76,8 +76,7 @@ Client sends matching parameters:
   "type": "match_request",
   "topics": ["Algorithms", "Arrays"],
   "difficulties": ["Easy", "Medium"],
-  "username": "Jane Doe",
-  "email": "[email protected]" // possible to change to user ID in mongodb
+  "username": "Jane Doe"
 }
 ```
 
@@ -86,9 +85,11 @@ Server response on successful match:
 ```json
 {
   "type": "match_found",
-  "matchID": "aa41c004e589642402215c2c0a3a165a",
-  "partnerID": 54321, // currently identifying via partner's websocket port --> change to user ID in mongodb
-  "partnerName": "John Doe" // partner username
+  "matchId": "1c018916a34c5bee21af0b2670bd6156",
+  "user": "zkb4px",
+  "matchedUser": "JohnDoe",
+  "topic": "Algorithms",
+  "difficulty": "Medium"
 }
 ```
 
@@ -101,6 +102,15 @@ If no match is found after a set period of time, the server will send a timeout
 }
 ```
 
+If user has an existing websocket connection and wants to initiate another match, the server will reject the request:
+
+```json
+{
+  "type": "match_rejected",
+  "message": "You are already in a matchmaking queue. Please disconnect before reconnecting."
+}
+```
+
 If the server encounters an issue during the WebSocket connection or processing, the connection will be closed without any error message. The client should treat the unexpected closing as an error.
 
 ## Testing

apps/matching-service/handlers/websocket.go

@@ -7,16 +7,27 @@ import (
 	"matching-service/processes"
 	"matching-service/utils"
 	"net/http"
+	"strings"
+	"sync"
 
 	"github.com/gorilla/websocket"
 )
 
-var upgrader = websocket.Upgrader{
-	CheckOrigin: func(r *http.Request) bool {
-		// Allow all connections by skipping the origin check (set more restrictions in production)
-		return true
-	},
-}
+var (
+	upgrader = websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool {
+			// Allow all connections by skipping the origin check (set more restrictions in production)
+			return true
+		},
+	}
+	// A map to hold active WebSocket connections per username
+	activeConnections = make(map[string]*websocket.Conn)
+	// A map to hold user's match ctx cancel function
+	matchContexts = make(map[string]context.CancelFunc)
+	// A map to hold user's match channels
+	matchFoundChannels = make(map[string]chan models.MatchFound)
+	mu                 sync.Mutex // Mutex for thread-safe access to activeConnections
+)
 
 // handleConnections manages WebSocket connections and matching logic.
 func HandleWebSocketConnections(w http.ResponseWriter, r *http.Request) {
@@ -42,25 +53,48 @@ func HandleWebSocketConnections(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Store WebSocket connection in the activeConnections map.
+	mu.Lock()
+	// Checks if user is already an existing websocket connection
+	if _, exists := activeConnections[matchRequest.Username]; exists {
+		mu.Unlock()
+		log.Printf("User %s is already connected, rejecting new connection.", matchRequest.Username)
+		ws.WriteJSON(models.MatchRejected{
+			Type:    "match_rejected",
+			Message: "You are already in a matchmaking queue. Please disconnect before reconnecting.",
+		})
+		ws.Close()
+		return
+	}
+	activeConnections[matchRequest.Username] = ws
+	matchCtx, matchCancel := context.WithCancel(context.Background())
+	matchContexts[matchRequest.Username] = matchCancel
+
+	matchFoundChan := make(chan models.MatchFound)
+	matchFoundChannels[matchRequest.Username] = matchFoundChan
+	mu.Unlock()
+
 	// Create a context for cancellation
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel() // Ensure cancel is called to release resources
 
+	processes.EnqueueUser(processes.GetRedisClient(), matchRequest.Username, ctx)
+	processes.AddUserToTopicSets(processes.GetRedisClient(), matchRequest, ctx)
+	processes.StoreUserDetails(processes.GetRedisClient(), matchRequest, ctx)
+
 	timeoutCtx, timeoutCancel, err := createTimeoutContext()
 	if err != nil {
 		log.Printf("Error creating timeout context: %v", err)
 		return
 	}
 	defer timeoutCancel()
 
-	matchFoundChan := make(chan models.MatchFound)
-
 	// Start goroutines for handling messages and performing matching.
 	go processes.ReadMessages(ws, ctx, cancel)
-	go processes.PerformMatching(matchRequest, ctx, matchFoundChan) // Perform matching
+	go processes.PerformMatching(matchRequest, context.Background(), matchFoundChannels) // Perform matching
 
 	// Wait for a match, timeout, or cancellation.
-	waitForResult(ws, ctx, timeoutCtx, matchFoundChan)
+	waitForResult(ws, ctx, timeoutCtx, matchCtx, matchFoundChan, matchRequest.Username)
 }
 
 // readMatchRequest reads the initial match request from the WebSocket connection.
@@ -69,8 +103,13 @@ func readMatchRequest(ws *websocket.Conn) (models.MatchRequest, error) {
 	if err := ws.ReadJSON(&matchRequest); err != nil {
 		return matchRequest, err
 	}
-	matchRequest.Port = ws.RemoteAddr().String()
-	log.Printf("Received match request: %v", matchRequest)
+	// Get the remote address (client's IP and port)
+	clientAddr := ws.RemoteAddr().String()
+
+	// Extract the port (after the last ':')
+	clientPort := clientAddr[strings.LastIndex(clientAddr, ":")+1:]
+
+	log.Printf("Received match request: %v from client port: %s", matchRequest, clientPort)
 	return matchRequest, nil
 }
 
@@ -84,26 +123,55 @@ func createTimeoutContext() (context.Context, context.CancelFunc, error) {
 	return ctx, cancel, nil
 }
 
+// Cleans up the data associated with the user before ending the websocket connection.
+// If user is already removed, then nothing happens.
+func cleanUpUser(username string) {
+	// Cleanup Redis
+	processes.CleanUpUser(processes.GetRedisClient(), username, context.Background())
+
+	// Removes the match context and active
+	if cancelFunc, exists := matchContexts[username]; exists {
+		cancelFunc()
+		delete(matchContexts, username)
+	}
+	if _, exists := activeConnections[username]; exists {
+		delete(activeConnections, username)
+	}
+	if _, exists := matchFoundChannels[username]; exists {
+		delete(matchFoundChannels, username)
+	}
+}
+
 // waitForResult waits for a match result, timeout, or cancellation.
-func waitForResult(ws *websocket.Conn, ctx, timeoutCtx context.Context, matchFoundChan chan models.MatchFound) {
+func waitForResult(ws *websocket.Conn, ctx, timeoutCtx, matchCtx context.Context, matchFoundChan chan models.MatchFound, username string) {
 	select {
 	case <-ctx.Done():
 		log.Println("Matching cancelled")
+		cleanUpUser(username)
 		return
 	case <-timeoutCtx.Done():
 		log.Println("Connection timed out")
 		sendTimeoutResponse(ws)
+		cleanUpUser(username)
+		return
+	case <-matchCtx.Done():
+		log.Println("Match found for user: " + username)
+
+		// NOTE: user is already cleaned-up in the other process,
+		// so there is no need to clean up again.
 		return
 	case result, ok := <-matchFoundChan:
 		if !ok {
 			// Channel closed without a match, possibly due to context cancellation
 			log.Println("Match channel closed without finding a match")
 			return
 		}
-		log.Println("Match found")
-		if err := ws.WriteJSON(result); err != nil {
-			log.Printf("write error: %v", err)
-		}
+		log.Println("Match found for user: " + username)
+		// Notify the user about the match
+		notifyMatches(result.User, result)
+
+		// NOTE: user and other user are already cleaned up in a separate matching algorithm process
+		// so no clean up is required here.
 		return
 	}
 }
@@ -118,3 +186,16 @@ func sendTimeoutResponse(ws *websocket.Conn) {
 		log.Printf("write error: %v", err)
 	}
 }
+
+// Notify matches
+func notifyMatches(username string, result models.MatchFound) {
+	mu.Lock()
+	defer mu.Unlock()
+
+	// Send message to matched user
+	if userConn, userExists := activeConnections[username]; userExists {
+		if err := userConn.WriteJSON(result); err != nil {
+			log.Printf("Error sending message to user %s: %v\n", username, err)
+		}
+	}
+}

apps/matching-service/main.go

@@ -19,31 +19,17 @@ func main() {
 	if err != nil {
 		log.Fatalf("err loading: %v", err)
 	}
-	port := os.Getenv("PORT")
-
-	// Set up link with redis server
-	redisAddr := os.Getenv("REDIS_URL")
-	client := redis.NewClient(&redis.Options{
-		Addr:     redisAddr,
-		Password: "", // no password set
-		DB:       0,  // use default DB
-	})
-
-	// Ping the redis server
-	_, err = client.Ping(context.Background()).Result()
-	if err != nil {
-		log.Fatalf("Could not connect to Redis: %v", err)
-	} else {
-		log.Println("Connected to Redis at the following address: " + redisAddr)
-	}
-
-	// Pass the connect redis client to processes in match
-	processes.SetRedisClient(client)
+	
+	// Setup redis client
+	processes.SetupRedisClient()
 
+	// Run a goroutine that matches users
+	
 	// Routes
 	http.HandleFunc("/match", handlers.HandleWebSocketConnections)
-
+	
 	// Start the server
+	port := os.Getenv("PORT")
 	log.Println(fmt.Sprintf("Server starting on :%s", port))
 	err = http.ListenAndServe(fmt.Sprintf(":%s", port), nil)
 	if err != nil {

apps/matching-service/models/complexity.go

@@ -0,0 +1,67 @@
+package models
+
+import (
+	"log"
+	"strings"
+)
+
+// Get the highest common difficulty (aka complexity) between two users,
+// If no common difficulty found, choose the min of the 2 arrays.
+func GetCommonDifficulty(userArr []string, matchedUserArr []string) string {
+	commonDifficulties := make([]int, 3)
+	for i := range commonDifficulties {
+		commonDifficulties[i] = 0
+	}
+
+	for _, difficulty := range userArr {
+		formattedDifficulty := strings.ToLower(difficulty)
+		switch formattedDifficulty {
+		case "easy":
+			commonDifficulties[0]++
+		case "medium":
+			commonDifficulties[1]++
+		case "hard":
+			commonDifficulties[2]++
+		default:
+			log.Println("Unknown difficulty specified: " + difficulty)
+		}
+	}
+
+	for _, difficulty := range matchedUserArr {
+		formattedDifficulty := strings.ToLower(difficulty)
+		switch formattedDifficulty {
+		case "easy":
+			commonDifficulties[0]++
+		case "medium":
+			commonDifficulties[1]++
+		case "hard":
+			commonDifficulties[2]++
+		default:
+			log.Println("Unknown difficulty specified: " + difficulty)
+		}
+	}
+
+	lowest := "Hard"
+	for i := 2; i >= 0; i-- {
+		if commonDifficulties[i] == 2 {
+			switch i {
+			case 0:
+				return "Easy"
+			case 1:
+				return "Medium"
+			case 2:
+				return "Hard"
+			}
+		} else if commonDifficulties[i] > 0 {
+			switch i {
+			case 0:
+				lowest = "Easy"
+			case 1:
+				lowest = "Medium"
+			case 2:
+				lowest = "Hard"
+			}
+		}
+	}
+	return lowest
+}

apps/matching-service/models/match.go

@@ -5,18 +5,23 @@ type MatchRequest struct {
 	Topics       []string `json:"topics"`
 	Difficulties []string `json:"difficulties"`
 	Username     string   `json:"username"`
-	Email        string   `json:"email"`
-	Port         string   `json:"port"`
 }
 
 type MatchFound struct {
 	Type        string `json:"type"`
 	MatchID     string `json:"matchId"`
-	PartnerID   int64  `json:"partnerId"`
-	PartnerName string `json:"partnerName"`
+	User        string `json:"user"`        // username
+	MatchedUser string `json:"matchedUser"` // matched username
+	Topic       string `json:"topic"`       // matched topic
+	Difficulty  string `json:"difficulty"`  // matched difficulty
 }
 
 type Timeout struct {
 	Type    string `json:"type"`
 	Message string `json:"message"`
 }
+
+type MatchRejected struct {
+	Type    string `json:"type"`
+	Message string `json:"message"`
+}

apps/matching-service/processes/constants.go

@@ -0,0 +1,17 @@
+package processes
+
+import (
+	"context"
+	"sync"
+
+	"github.com/redis/go-redis/v9"
+)
+
+const matchmakingQueueRedisKey = "matchmaking_queue"
+
+var (
+	redisClient          *redis.Client
+	matchingRoutineMutex sync.Mutex // Mutex to ensure only one matchmaking goroutine is running
+	redisAccessMutex     sync.Mutex // Mutex for Redis access for concurrency safety
+	ctx                  = context.Background()
+)