Merge pull request #70 from CS3219-AY2425S1/collab-backend

Add backend for collab: Inter-service communication (Kafka) and Collab service

Author: SPWwj

.env.example

@@ -15,3 +15,4 @@ DB_PORT=27017
 PEERPREP_QUESTION_INITDB_NAME=peerprepQuestionServiceDB # must match question service .env file and init-mongo.js
 PEERPREP_USER_INITDB_NAME=peerprepUserServiceDB # must match user service .env file and init-mongo.js
 PEERPREP_MATCHING_INITDB_NAME=peerprepMatchingServiceDB # must match user service .env file and init-mongo.js
+PEERPREP_COLLABORATION_INITDB_NAME=peerprepCollaborationServiceDB # must match collab service .env file and init-mongo.js

README.md

@@ -7,9 +7,12 @@ This project follows a microservices architecture with the following services:
 2. **User Service** - Port `3001`
 3. **Question Service** - Port `3002`
 4. **Matching Service** - Port `3003`
-5. **MongoDB** - Port `27017` (Database)
-6. **Nginx API Gateway** - Port `80`
-7. **Redis** - Port `6379`
+5. **Collaboration Service** - Port `3004`
+6. **MongoDB** - Port `27017` (Database)
+7. **Nginx API Gateway** - Port `80`
+8. **Redis** - Port `6379`
+9. **Zookeeper** - Port `2181`
+10. **Kafka** - Port `9092`, Port `29092`
 
 ### Setting up the Project
 Copy and paste the .env.example files in each service. Rename them as .env files.
@@ -19,6 +22,7 @@ Files to do this in:
 3. /backend/user-service
 4. /backend/question-service
 5. /backend/matching-service
+6. /backend/collaboration-service
 Then, run `node -e "console.log(require('crypto').randomBytes(32).toString('hex'))"` twice to generate
 your 2 JWT token secrets. For the first one, paste it into the JWT_ACCESS_TOKEN_SECRET variable of
 the .env files in question-service and user-service. Then, copy the second into the
@@ -39,11 +43,14 @@ Once the containers are up:
 - User Service: [http://localhost:3001](http://localhost:3001)
 - Question Service: [http://localhost:3002](http://localhost:3002)
 - Matching Service: [http://localhost:3003](http://localhost:3003)
+- Collaboration Service: [http://localhost:3004](http://localhost:3004)
 - MongoDB: [http://localhost:27017](http://localhost:27017)
 - Nginx API Gateway: [http://localhost:80](http://localhost:80)
 - Redis: [http://localhost:6379](http://localhost:6379)
+- Zookeeper: [http://localhost:2181](http://localhost:2181)
+- Kafka: [http://localhost:9092](http://localhost:9092)
 
-Note that even after docker says that everything is up and running, there is a risk that they aren't when you load the frontend. 
+Note that even after docker says that everything is up and running, there is a risk that they aren't when you load the frontend.  Wait for the frontend logs to show up in the docker logs.
 In this event, wait for about a minute before trying again. If that still doesn't work and there are network errors, try
 rebuilding the services by running `docker-compose up --build` again.
 

backend/collaboration-service/.env.example

@@ -0,0 +1,29 @@
+# Database Connection String Details
+DATABASE_NAME=peerprepCollaborationServiceDB
+
+# Port to run service on
+PORT=3004
+
+# Kafka configuration
+KAFKA_HOST=localhost
+KAFKA_PORT=9092
+
+# Copy root .env
+# If using mongoDB containerization, set to true. Else set to false (i.e local testing)
+DB_REQUIRE_AUTH=true
+
+# mongoDB auth variables
+MONGO_USER=user
+MONGO_PASSWORD=password
+MONGO_INITDB_ROOT_USERNAME=admin # must match docker-compose.yml and init-mongo.js
+MONGO_INITDB_ROOT_PASSWORD=password # must match docker-compose.yml and init-mongo.js
+
+# mongoDB connection string variables
+DB_HOST=localhost
+DB_PORT=27017
+
+# Service database names
+PEERPREP_QUESTION_INITDB_NAME=peerprepQuestionServiceDB # must match question service .env file and init-mongo.js
+PEERPREP_USER_INITDB_NAME=peerprepUserServiceDB # must match user service .env file and init-mongo.js
+PEERPREP_MATCHING_INITDB_NAME=peerprepMatchingServiceDB # must match user service .env file and init-mongo.js
+PEERPREP_COLLABORATION_INITDB_NAME=peerprepCollaborationServiceDB # must match collab service .env file and init-mongo.js

backend/collaboration-service/Dockerfile

@@ -0,0 +1,24 @@
+FROM node:20-alpine AS base
+WORKDIR /app
+COPY package*.json ./
+RUN npm install
+COPY . .
+
+# For development environment
+FROM base AS dev
+EXPOSE 3004
+CMD ["npm", "run", "dev"]
+
+# For production environment
+FROM base AS build
+RUN npm run build
+
+FROM node:20-alpine AS prod
+WORKDIR /app
+# Copy only the built files and necessary files for production
+COPY --from=build /app/dist ./dist
+COPY package*.json ./
+# Install only production dependencies
+RUN npm install --only=production
+EXPOSE 3004
+CMD ["npm", "start"]

backend/collaboration-service/app.ts

@@ -0,0 +1,78 @@
+import cors from "cors";
+import express, { Express } from "express";
+import dotenv from "dotenv";
+import path from "path";
+import http from "http";
+import { WebSocket } from "ws";
+import collaborationRoutes from "./routes/collaborationRoutes";
+import { errorHandler } from "./middlewares/errorHandler";
+import { connectToDatabase } from "./utils/database";
+import { setUpKafkaSubscribers } from "./utils/kafkaClient";
+const setupWSConnection = require("y-websocket/bin/utils").setupWSConnection;
+
+
+dotenv.config({ path: path.resolve(__dirname, "./.env") });
+
+connectToDatabase();
+
+const port = process.env.PORT || 3004;
+const app: Express = express();
+
+const server = http.createServer(app);
+
+app.use(express.json());
+
+app.use("/api/collab", collaborationRoutes);
+
+app.use(
+    cors({
+        origin: "*",
+        credentials: true,
+    })
+);
+
+app.use(errorHandler);
+
+const wss = new WebSocket.Server({ server });
+
+wss.on("connection", (ws, req) => {
+    // // Extract sessionId from the URL for connecting to the Yjs document
+    // const sessionId = req.url?.split('/').pop();
+
+    // console.log(`A user has connected to session: ${sessionId}!`)
+
+    // if (sessionId) {
+    //     // Initialize or get the Yjs document from YDocManager
+    //     const ydoc = YDocManager.getDoc(sessionId) || YDocManager.initializeDoc(sessionId);
+    
+    //     // Set up the document synchronization logic
+    //     const encoder = Y.encodeStateAsUpdate(ydoc);
+    //     ws.send(encoder);
+    
+    //     ws.on('message', (message) => {
+    //         const update = new Uint8Array(message instanceof ArrayBuffer ? message : (message as Buffer).buffer);
+    //         Y.applyUpdate(ydoc, update);
+    //         wss.clients.forEach((client) => {
+    //             if (client !== ws && client.readyState === WebSocket.OPEN) {
+    //                 client.send(update);
+    //             }
+    //         });
+    //     });
+    
+    //     ws.on('close', () => {
+    //         console.log(`Connection closed for session: ${sessionId}`);
+    //     });
+
+    // } else {
+    //     console.error("No session ID found in WebSocket connection URL.");
+    //     ws.close();
+    // }
+    setupWSConnection(ws, req);
+});
+
+// Start the server
+server.listen(port, async () => {
+    console.log(`Collaboration service running on http://localhost:${port}`);
+
+    await setUpKafkaSubscribers();
+});

backend/collaboration-service/controllers/collaborationController.ts

@@ -0,0 +1,76 @@
+// controllers/collaborationController.js
+import { EachMessagePayload } from 'kafkajs';
+import { COLLAB_TOPIC, producer } from '../utils/kafkaClient';
+import Session from '../models/Session';
+
+
+export const createSession = async ( matchId: string, userIds: string[]) => {
+    // Create unique sessionId:
+    const sessionId = `session-${Date.now()}-${userIds.join('-')}`;
+
+    // Create new Session document in mongodb
+    const newSession = new Session({ sessionId, matchId, userIds, codeContent: '' });
+    await newSession.save();
+
+    return sessionId;
+};
+
+/**
+ * Process the message from the Kafka topic and create a new session.
+ * Emits a Kafka message back to the matching service with the session ID.
+ * @param message - Kafka message payload
+ */
+export async function handleMatchNotification(message: EachMessagePayload) {
+    /**
+     * message contains all the info from the kafka message from matching-service. 
+     * message looks like this:
+     * {
+     *  topic: 'match_topic',
+     *  message: {
+     *      key: matchId,
+     *      value: {
+     *         user1: { userId: user1.userId, socketId: user1.socketId },
+                user2: { userId: user2.userId, socketId: user2.socketId },
+                category: 'algo',
+                difficulty: 'easy',
+                matchId: 123 (mongoDB Id for match),
+     *      }
+     *  }
+     * }
+     */
+    console.log("Collab service creating session");
+
+    // Validation for message format
+    const matchId = message.message.key?.toString();
+    if (!matchId || !message.message.value) {
+        console.error("No match ID/value found in message.");
+        return;
+    }
+
+    // Create session and get sessionId
+    const messageValue = JSON.parse(message.message.value.toString());
+
+    const user1Id = messageValue.user1?.userId;
+    const user2Id = messageValue.user2?.userId;
+
+    if (!user1Id || !user2Id) {
+        console.error("User IDs not found in message.");
+        return;
+    }
+
+    const sessionId = await createSession(matchId, [user1Id, user2Id]);
+
+    // Send the session ID back to the matching service
+    const messageBody = JSON.stringify({ sessionId });
+    await producer.send({
+        topic: COLLAB_TOPIC,
+        messages: [
+            {
+                key: matchId,
+                value: messageBody,
+            },
+        ],
+    });
+
+    console.log(`Sent Session ID ${sessionId} to collab service.`); // modify
+}

backend/collaboration-service/middlewares/errorHandler.ts

@@ -0,0 +1,11 @@
+import { Request, Response, NextFunction } from "express";
+
+export function errorHandler(err: any, req: Request, res: Response, next: NextFunction) {
+    if (err instanceof Error) {
+        console.error("Error:", err.message);
+        return res.status(400).json({ message: err.message });
+    } else {
+        console.error("Unexpected error:", err);
+        return res.status(500).json({ message: "An unexpected error occurred." });
+    }
+}

backend/collaboration-service/models/Session.ts

@@ -0,0 +1,12 @@
+import mongoose, { Types } from "mongoose";
+
+const sessionSchema = new mongoose.Schema({
+    matchId: { type: String, required: true },
+    sessionId: { type: String, required: true },
+    codeContent: { type: String, default: "" },
+    lastUpdated: { type: Date, default: Date.now }
+});
+
+const Session = mongoose.model('session', sessionSchema);
+
+export default Session;