feat: add keys to watch

Author: tituschewxj

apps/matching-service/databases/redis.go

@@ -8,7 +8,7 @@ import (
 	"github.com/redis/go-redis/v9"
 )
 
-const matchmakingQueueRedisKey = "matchmaking_queue"
+const MatchmakingQueueRedisKey = "matchmaking_queue"
 
 var redisClient *redis.Client
 

apps/matching-service/databases/userqueue.go

@@ -30,7 +30,7 @@ func PrintMatchingQueue(tx *redis.Tx, status string, ctx context.Context) {
 }
 
 func IsQueueEmpty(tx *redis.Tx, ctx context.Context) (bool, error) {
-	queueLength, err := tx.LLen(ctx, matchmakingQueueRedisKey).Result()
+	queueLength, err := tx.LLen(ctx, MatchmakingQueueRedisKey).Result()
 	if err != nil {
 		log.Println("Error checking queue length:", err)
 		return false, err
@@ -41,15 +41,15 @@ func IsQueueEmpty(tx *redis.Tx, ctx context.Context) (bool, error) {
 
 // Enqueue a user into the matchmaking queue
 func EnqueueUser(tx *redis.Tx, username string, ctx context.Context) {
-	err := tx.LPush(ctx, matchmakingQueueRedisKey, username).Err()
+	err := tx.LPush(ctx, MatchmakingQueueRedisKey, username).Err()
 	if err != nil {
 		log.Println("Error enqueuing user:", err)
 	}
 }
 
 // Remove user from the matchmaking queue
 func DequeueUser(tx *redis.Tx, username string, ctx context.Context) {
-	err := tx.LRem(ctx, matchmakingQueueRedisKey, 1, username).Err()
+	err := tx.LRem(ctx, MatchmakingQueueRedisKey, 1, username).Err()
 	if err != nil {
 		log.Println("Error dequeuing user:", err)
 		return
@@ -59,7 +59,7 @@ func DequeueUser(tx *redis.Tx, username string, ctx context.Context) {
 // Returns the first user's username from the queue.
 func GetFirstUser(tx *redis.Tx, ctx context.Context) (string, error) {
 	// Peek at the user queue
-	username, err := tx.LIndex(ctx, matchmakingQueueRedisKey, 0).Result()
+	username, err := tx.LIndex(ctx, MatchmakingQueueRedisKey, 0).Result()
 	if err != nil {
 		log.Println("Error peeking user from queue:", err)
 		return "", err
@@ -68,7 +68,7 @@ func GetFirstUser(tx *redis.Tx, ctx context.Context) (string, error) {
 }
 
 func GetAllQueuedUsers(tx *redis.Tx, ctx context.Context) ([]string, error) {
-	users, err := tx.LRange(ctx, matchmakingQueueRedisKey, 0, -1).Result()
+	users, err := tx.LRange(ctx, MatchmakingQueueRedisKey, 0, -1).Result()
 	if err != nil {
 		log.Println("Error retrieving users from queue:", err)
 		return nil, err
@@ -189,13 +189,13 @@ func FindMatchingUser(tx *redis.Tx, username string, ctx context.Context) (*mode
 
 func PopAndInsertUser(tx *redis.Tx, username string, ctx context.Context) {
 	// Pop user
-	username, err := tx.LPop(ctx, matchmakingQueueRedisKey).Result()
+	username, err := tx.LPop(ctx, MatchmakingQueueRedisKey).Result()
 	if err != nil {
 		log.Println("Error popping user from queue:", err)
 	}
 
 	// Insert back in queue
-	err = tx.LPush(ctx, matchmakingQueueRedisKey, username).Err()
+	err = tx.LPush(ctx, MatchmakingQueueRedisKey, username).Err()
 	if err != nil {
 		log.Println("Error enqueuing user:", err)
 	}

apps/matching-service/models/errors.go apps/matching-service/models/error.goapps/matching-service/models/errors.go renamed to apps/matching-service/models/error.go

@@ -8,70 +8,87 @@ import (
 	"matching-service/databases"
 	"matching-service/models"
 	"matching-service/utils"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 )
 
 // Performs the matching algorithm at most once starting from the front of the queue of users,
 // until a match is found or no match and the user is enqueued to the queue.
 func PerformMatching(rdb *redis.Client, matchRequest models.MatchRequest, ctx context.Context, errorChan chan error) {
-	err := rdb.Watch(ctx, func(tx *redis.Tx) error {
-		// Log queue before and after matchmaking
-		databases.PrintMatchingQueue(tx, "Before Matchmaking", ctx)
-		defer databases.PrintMatchingQueue(tx, "After Matchmaking", ctx)
+	currentUsername := matchRequest.Username
+	keys := []string{databases.MatchmakingQueueRedisKey, currentUsername}
+	for _, topic := range matchRequest.Topics {
+		keys = append(keys, topic)
+	}
+	for true {
 
-		currentUsername := matchRequest.Username
-		queuedUsernames, err := databases.GetAllQueuedUsers(tx, ctx)
-		if err != nil {
-			return err
-		}
+		err := rdb.Watch(ctx, func(tx *redis.Tx) error {
+			// Log queue before and after matchmaking
+			databases.PrintMatchingQueue(tx, "Before Matchmaking", ctx)
+			defer databases.PrintMatchingQueue(tx, "After Matchmaking", ctx)
 
-		// Check that user is not part of the existing queue
-		for _, username := range queuedUsernames {
-			if username == currentUsername {
-				return models.ExistingUserError
+			queuedUsernames, err := databases.GetAllQueuedUsers(tx, ctx)
+			if err != nil {
+				return err
 			}
-		}
 
-		databases.AddUser(tx, matchRequest, ctx)
-
-		// Find a matching user if any
-		matchFound, err := databases.FindMatchingUser(tx, currentUsername, ctx)
-		if err != nil {
-			log.Println("Error finding matching user:", err)
-			return err
-		}
+			// Check that user is not part of the existing queue
+			for _, username := range queuedUsernames {
+				if username == currentUsername {
+					return models.ExistingUserError
+				}
+			}
 
-		if matchFound != nil {
-			matchedUsername := matchFound.MatchedUser
-			matchedTopic := matchFound.Topic
-			matchedDifficulty := matchFound.Difficulty
+			databases.AddUser(tx, matchRequest, ctx)
 
-			// Generate a random match ID
-			matchId, err := utils.GenerateMatchID()
+			// Find a matching user if any
+			matchFound, err := databases.FindMatchingUser(tx, currentUsername, ctx)
 			if err != nil {
-				log.Println("Unable to randomly generate matchID")
+				log.Println("Error finding matching user:", err)
+				return err
 			}
 
-			// Log down which users got matched
-			matchFound.MatchID = matchId
-			log.Printf("Users %s and %s matched on the topic: %s with difficulty: %s", currentUsername, matchedUsername, matchedTopic, matchedDifficulty)
+			if matchFound != nil {
+				matchedUsername := matchFound.MatchedUser
+				matchedTopic := matchFound.Topic
+				matchedDifficulty := matchFound.Difficulty
 
-			// Clean up redis for this match
-			databases.CleanUpUser(tx, currentUsername, ctx)
-			databases.CleanUpUser(tx, matchedUsername, ctx)
+				// Generate a random match ID
+				matchId, err := utils.GenerateMatchID()
+				if err != nil {
+					log.Println("Unable to randomly generate matchID")
+				}
 
-			publishMatch(tx, ctx, currentUsername, matchedUsername, matchFound)
-			publishMatch(tx, ctx, matchedUsername, currentUsername, matchFound)
-		}
+				// Log down which users got matched
+				matchFound.MatchID = matchId
+				log.Printf("Users %s and %s matched on the topic: %s with difficulty: %s", currentUsername, matchedUsername, matchedTopic, matchedDifficulty)
 
-		return nil
-	})
-	if err != nil {
-		// Handle error (like retry logic could be added here)
-		// return fmt.Errorf("transaction execution failed: %v", err)
-		if errors.Is(err, models.ExistingUserError) {
-			errorChan <- err
+				// Clean up redis for this match
+				databases.CleanUpUser(tx, currentUsername, ctx)
+				databases.CleanUpUser(tx, matchedUsername, ctx)
+
+				publishMatch(tx, ctx, currentUsername, matchedUsername, matchFound)
+				publishMatch(tx, ctx, matchedUsername, currentUsername, matchFound)
+			}
+
+			time.Sleep(time.Duration(time.Second * 1))
+
+			return nil
+		}, keys...)
+		if err != nil {
+			// transaction failed
+			println(err)
+			// Handle error (like retry logic could be added here)
+			// return fmt.Errorf("transaction execution failed: %v", err)
+			if errors.Is(err, models.ExistingUserError) {
+				errorChan <- err
+				break
+			}
+		}
+		if err == nil {
+			// transaction succeeded
+			break
 		}
 	}
 }