feat: implement grpc client

Author: tituschewxj

apps/matching-service/databases/userqueue.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log"
 	"matching-service/models"
+	"matching-service/servers"
 	"strings"
 
 	"github.com/redis/go-redis/v9"
@@ -30,7 +31,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, servers.MatchmakingQueueRedisKey).Result()
 	if err != nil {
 		log.Println("Error checking queue length:", err)
 		return false, err
@@ -41,15 +42,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, servers.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, servers.MatchmakingQueueRedisKey, 1, username).Err()
 	if err != nil {
 		log.Println("Error dequeuing user:", err)
 		return
@@ -59,7 +60,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, servers.MatchmakingQueueRedisKey, 0).Result()
 	if err != nil {
 		log.Println("Error peeking user from queue:", err)
 		return "", err
@@ -69,7 +70,7 @@ func GetFirstUser(tx *redis.Tx, ctx context.Context) (string, error) {
 
 // Return the usernames of all the queued users.
 func GetAllQueuedUsers(tx *redis.Tx, ctx context.Context) ([]string, error) {
-	users, err := tx.LRange(ctx, MatchmakingQueueRedisKey, 0, -1).Result()
+	users, err := tx.LRange(ctx, servers.MatchmakingQueueRedisKey, 0, -1).Result()
 	if err != nil {
 		log.Println("Error retrieving users from queue:", err)
 		return nil, err
@@ -165,13 +166,13 @@ func RemoveUserDetails(tx *redis.Tx, username string, ctx context.Context) {
 
 func PopAndInsertUser(tx *redis.Tx, username string, ctx context.Context) {
 	// Pop user
-	username, err := tx.LPop(ctx, MatchmakingQueueRedisKey).Result()
+	username, err := tx.LPop(ctx, servers.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, servers.MatchmakingQueueRedisKey, username).Err()
 	if err != nil {
 		log.Println("Error enqueuing user:", err)
 	}

apps/matching-service/handlers/websocket.go

@@ -8,6 +8,7 @@ import (
 	"matching-service/databases"
 	"matching-service/models"
 	"matching-service/processes"
+	"matching-service/servers"
 	"matching-service/utils"
 	"net/http"
 
@@ -26,7 +27,7 @@ var (
 
 // handleConnections manages WebSocket connections and matching logic.
 func HandleWebSocketConnections(w http.ResponseWriter, r *http.Request) {
-	rdb := databases.GetRedisClient()
+	rdb := servers.GetRedisClient()
 	ctx := context.Background()
 
 	// TODO: Parse the authorization header to validate the JWT token and get the user ID claim.
@@ -93,11 +94,11 @@ func readMatchRequest(ws *websocket.Conn) (models.MatchRequest, error) {
 // If user is already removed, then nothing happens.
 // This function is unaffected by the external context.
 func cleanUpUser(username string) {
-	redisClient := databases.GetRedisClient()
+	redisClient := servers.GetRedisClient()
 	ctx := context.Background()
 
 	// Obtain lock with retry
-	lock, err := databases.ObtainRedisLock(ctx)
+	lock, err := servers.ObtainRedisLock(ctx)
 	if err != nil {
 		return
 	}

apps/matching-service/main.go

@@ -3,8 +3,8 @@ package main
 import (
 	"fmt"
 	"log"
-	"matching-service/databases"
 	"matching-service/handlers"
+	"matching-service/servers"
 	"net/http"
 	"os"
 
@@ -13,8 +13,10 @@ import (
 
 func main() {
 	setUpEnvironment()
-	client := databases.SetupRedisClient()
+	client := servers.SetupRedisClient()
 	defer client.Close()
+	grpcClient := servers.InitGrpcServer()
+	defer grpcClient.Close()
 	setupRoutes()
 	startServer()
 }

apps/matching-service/processes/performmatches.go

@@ -8,6 +8,9 @@ import (
 	"log"
 	"matching-service/databases"
 	"matching-service/models"
+	pb "matching-service/proto"
+	"matching-service/servers"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 )
@@ -18,7 +21,7 @@ func PerformMatching(rdb *redis.Client, matchRequest models.MatchRequest, ctx co
 	currentUsername := matchRequest.Username
 
 	// Obtain lock with retry
-	lock, err := databases.ObtainRedisLock(ctx)
+	lock, err := servers.ObtainRedisLock(ctx)
 	if err != nil {
 		return
 	}
@@ -38,27 +41,53 @@ func PerformMatching(rdb *redis.Client, matchRequest models.MatchRequest, ctx co
 		// Find a matching user if any
 		matchFound, err := findMatchingUsers(tx, currentUsername, ctx)
 		if err != nil {
+			if errors.Is(err, models.NoMatchFound) {
+				return nil
+			}
 			log.Println("Error finding matching user:", err)
 			return err
-		} else if matchFound != nil {
+		} else if matchFound != nil && err != models.NoMatchFound {
 			matchedUsername := matchFound.MatchedUser
 			matchedTopics := matchFound.MatchedTopics
 			matchedDifficulties := matchFound.MatchedDifficulties
 
 			// Log down which users got matched
-			log.Printf("Users %s and %s matched on the topics: %v with difficulties: %v", currentUsername, matchedUsername, matchedTopics, matchedDifficulties)
+			log.Printf("Users %s and %s matched on the topics: %v; with difficulties: %v", currentUsername, matchedUsername, matchedTopics, matchedDifficulties)
 
 			// 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)
+			// Query question service to find a question for the match
+			ctx2, cancel := context.WithTimeout(ctx, time.Second)
+			defer cancel()
+
+			question, err := servers.GetGrpcClient().FindMatchingQuestion(ctx2, &pb.MatchQuestionRequest{
+				MatchedTopics:       matchedTopics,
+				MatchedDifficulties: matchedDifficulties,
+			})
+			if err != nil {
+				log.Fatalf("Could not retrieve question from question-service: %v", err)
+			}
+
+			matchQuestionFound := models.MatchQuestionFound{
+				Type:               "match_found",
+				MatchID:            matchFound.MatchID,
+				User:               matchFound.User,
+				MatchedUser:        matchFound.MatchedUser,
+				MatchedTopics:      matchedTopics,
+				QuestionID:         question.QuestionId,
+				QuestionName:       question.QuestionName,
+				QuestionDifficulty: question.QuestionDifficulty,
+				QuestionTopics:     question.QuestionTopics,
+			}
+
+			publishMatch(tx, ctx, currentUsername, matchedUsername, &matchQuestionFound)
+			publishMatch(tx, ctx, matchedUsername, currentUsername, &matchQuestionFound)
 		}
 
 		return nil
 	}); err != nil {
-		// return
 		if errors.Is(err, models.ExistingUserError) {
 			errorChan <- err
 		} else {
@@ -69,7 +98,7 @@ func PerformMatching(rdb *redis.Client, matchRequest models.MatchRequest, ctx co
 }
 
 // Publish a match to the target user's pub/sub channel
-func publishMatch(tx *redis.Tx, ctx context.Context, targetUser string, otherMatchedUser string, matchFound *models.MatchFound) error {
+func publishMatch(tx *redis.Tx, ctx context.Context, targetUser string, otherMatchedUser string, matchFound *models.MatchQuestionFound) error {
 	matchFound.User = targetUser
 	matchFound.MatchedUser = otherMatchedUser
 	msg, err := json.Marshal(matchFound)

apps/matching-service/servers/grpc.go

@@ -0,0 +1,29 @@
+package servers
+
+import (
+	"log"
+	pb "matching-service/proto"
+
+	"google.golang.org/grpc"
+)
+
+var (
+	questionMatchingClient pb.QuestionMatchingServiceClient
+)
+
+func InitGrpcServer() *grpc.ClientConn {
+	// Dial the server
+	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
+	if err != nil {
+		log.Fatalf("Did not connect: %v", err)
+	}
+
+	// Create a new client for the ExampleService
+	questionMatchingClient = pb.NewQuestionMatchingServiceClient(conn)
+
+	return conn
+}
+
+func GetGrpcClient() pb.QuestionMatchingServiceClient {
+	return questionMatchingClient
+}

apps/matching-service/databases/redis.go apps/matching-service/servers/redis.goapps/matching-service/databases/redis.go renamed to apps/matching-service/servers/redis.go

@@ -1,4 +1,4 @@
-package databases
+package servers
 
 import (
 	"context"

apps/question-service/main.go

@@ -129,6 +129,7 @@ func initGrpcServer() {
 }
 
 func (s *server) FindMatchingQuestion(ctx context.Context, req *pb.MatchQuestionRequest) (*pb.QuestionFound, error) {
+	// STUB
 	return &pb.QuestionFound{
 		QuestionId:         1,
 		QuestionName:       "abc",