Merge pull request #94 from ruiqi7/feature/coding-sandbox

Implement collaboration coding sandbox

Author: guanquann

backend/README.md

@@ -2,11 +2,19 @@
 
 > Before proceeding to each microservice for more instructions:
 
-1. Set up cloud MongoDB if not using docker. We recommend this if you are just testing out each microservice separately to avoid needing to manually set up multiple instances of local MongoDB. Else, if you are using docker-compose.yml to run PeerPrep, check out the READMEs in the different backend microservices to set up the env for the local MongoDB instances.
+1. Set up cloud MongoDB if you are not using Docker. We recommend this if you are just testing out each microservice separately to avoid needing to manually set up multiple instances of local MongoDB. Otherwise, if you are using `docker-compose.yml` to run PeerPrep, check out the READMEs in the different backend microservices to set up the `.env` files for the local MongoDB instances.
 
 2. Set up Firebase.
 
-3. Follow the instructions [here](https://nodejs.org/en/download/package-manager) to set up Node v20.
+3. For the microservices that use Redis, to view the contents stored:
+
+   1. Go to [http://localhost:5540](http://localhost:5540).
+
+   2. Click on "Add Redis Database".
+
+   3. Enter `host.docker.internal` as the Host.
+
+4. Follow the instructions [here](https://nodejs.org/en/download/package-manager) to set up Node v20.
 
 ## Setting-up cloud MongoDB (in production)
 

backend/collab-service/package-lock.json

@@ -14,6 +14,8 @@
   "license": "ISC",
   "description": "",
   "dependencies": {
+    "@codemirror/collab": "^6.1.1",
+    "@codemirror/state": "^6.4.1",
     "axios": "^1.7.7",
     "body-parser": "^1.20.3",
     "cors": "^2.8.5",
@@ -22,7 +24,9 @@
     "redis": "^4.7.0",
     "socket.io": "^4.8.1",
     "swagger-ui-express": "^5.0.1",
-    "yaml": "^2.6.0"
+    "y-protocols": "^1.0.6",
+    "yaml": "^2.6.0",
+    "yjs": "^13.6.20"
   },
   "devDependencies": {
     "@eslint/js": "^9.13.0",

backend/collab-service/package.json

@@ -1,72 +1,170 @@
 import { Socket } from "socket.io";
 import { io } from "../server";
 import redisClient from "../config/redis";
+import { Doc, applyUpdateV2, encodeStateAsUpdateV2 } from "yjs";
 
 enum CollabEvents {
   // Receive
   JOIN = "join",
-  CHANGE = "change",
   LEAVE = "leave",
   DISCONNECT = "disconnect",
+  INIT_DOCUMENT = "init_document",
+  UPDATE_REQUEST = "update_request",
+  UPDATE_CURSOR_REQUEST = "update_cursor_request",
+  RECONNECT_REQUEST = "reconnect_request",
 
   // Send
-  ROOM_FULL = "room_full",
-  CONNECTED = "connected",
-  NEW_USER_CONNECTED = "new_user_connected",
-  CODE_CHANGE = "code_change",
-  PARTNER_LEFT = "partner_left",
-  PARTNER_DISCONNECTED = "partner_disconnected",
+  ROOM_READY = "room_ready",
+  DOCUMENT_READY = "document_ready",
+  UPDATE = "updateV2",
+  UPDATE_CURSOR = "update_cursor",
+  // PARTNER_LEFT = "partner_left",
+  // PARTNER_DISCONNECTED = "partner_disconnected",
 }
 
 const EXPIRY_TIME = 3600;
+const CONNECTION_DELAY = 3000; // time window to allow for page re-renders / refresh
+
+const userConnections = new Map<string, NodeJS.Timeout | null>();
+const collabSessions = new Map<string, Doc>();
+const partnerReadiness = new Map<string, boolean>();
 
 export const handleWebsocketCollabEvents = (socket: Socket) => {
-  socket.on(CollabEvents.JOIN, async ({ roomId }) => {
-    if (!roomId) {
+  socket.on(CollabEvents.JOIN, async (uid: string, roomId: string) => {
+    const connectionKey = `${uid}:${roomId}`;
+    if (userConnections.has(connectionKey)) {
+      clearTimeout(userConnections.get(connectionKey)!);
       return;
     }
+    userConnections.set(connectionKey, null);
 
     const room = io.sockets.adapter.rooms.get(roomId);
-    if (room && room.size >= 2) {
-      socket.emit(CollabEvents.ROOM_FULL);
+    if (room && room?.size >= 2) {
+      socket.emit(CollabEvents.ROOM_READY, false);
       return;
     }
 
     socket.join(roomId);
     socket.data.roomId = roomId;
 
-    // in case of disconnect, send the code to the user when he rejoins
-    const code = await redisClient.get(`collaboration:${roomId}`);
-    socket.emit(CollabEvents.CONNECTED, { code: code ? code : "" });
+    if (
+      io.sockets.adapter.rooms.get(roomId)?.size === 2 &&
+      !collabSessions.has(roomId)
+    ) {
+      createCollabSession(roomId);
+      io.to(roomId).emit(CollabEvents.ROOM_READY, true);
+    }
+  });
+
+  socket.on(CollabEvents.INIT_DOCUMENT, (roomId: string, template: string) => {
+    const doc = getDocument(roomId);
+    const isPartnerReady = partnerReadiness.get(roomId);
 
-    // inform the other user that a new user has joined
-    socket.to(roomId).emit(CollabEvents.NEW_USER_CONNECTED);
+    if (isPartnerReady && doc.getText().length === 0) {
+      doc.transact(() => {
+        doc.getText().insert(0, template);
+      });
+      io.to(roomId).emit(CollabEvents.DOCUMENT_READY);
+    } else {
+      partnerReadiness.set(roomId, true);
+    }
   });
 
-  socket.on(CollabEvents.CHANGE, async ({ roomId, code }) => {
-    if (!roomId || !code) {
-      return;
+  socket.on(
+    CollabEvents.UPDATE_REQUEST,
+    (roomId: string, update: Uint8Array) => {
+      const doc = collabSessions.get(roomId);
+      if (doc) {
+        applyUpdateV2(doc, new Uint8Array(update));
+      } else {
+        // TODO: error handling
+      }
     }
+  );
 
-    await redisClient.set(`collaboration:${roomId}`, code, {
-      EX: EXPIRY_TIME,
-    });
-    socket.to(roomId).emit(CollabEvents.CODE_CHANGE, { code });
-  });
+  socket.on(
+    CollabEvents.UPDATE_CURSOR_REQUEST,
+    (
+      roomId: string,
+      cursor: { uid: string; username: string; from: number; to: number }
+    ) => {
+      socket.to(roomId).emit(CollabEvents.UPDATE_CURSOR, cursor);
+    }
+  );
 
-  socket.on(CollabEvents.LEAVE, ({ roomId }) => {
-    if (!roomId) {
+  socket.on(CollabEvents.LEAVE, (uid: string, roomId: string) => {
+    const connectionKey = `${uid}:${roomId}`;
+    if (!userConnections.has(connectionKey)) {
       return;
     }
 
-    socket.leave(roomId);
-    socket.to(roomId).emit(CollabEvents.PARTNER_LEFT);
+    clearTimeout(userConnections.get(connectionKey)!);
+
+    const connectionTimeout = setTimeout(() => {
+      userConnections.delete(connectionKey);
+      socket.leave(roomId);
+      socket.disconnect();
+
+      const room = io.sockets.adapter.rooms.get(roomId);
+      if (!room || room.size === 0) {
+        removeCollabSession(roomId);
+      }
+    }, CONNECTION_DELAY);
+
+    userConnections.set(connectionKey, connectionTimeout);
   });
 
-  socket.on(CollabEvents.DISCONNECT, () => {
-    const { roomId } = socket.data;
-    if (roomId) {
-      socket.to(roomId).emit(CollabEvents.PARTNER_DISCONNECTED);
+  socket.on(CollabEvents.RECONNECT_REQUEST, async (roomId: string) => {
+    // TODO: Handle recconnection
+    socket.join(roomId);
+
+    const doc = getDocument(roomId);
+    const storeData = await redisClient.get(`collaboration:${roomId}`);
+
+    if (storeData) {
+      const tempDoc = new Doc();
+      const update = Buffer.from(storeData, "base64");
+      applyUpdateV2(tempDoc, new Uint8Array(update));
+      const tempText = tempDoc.getText().toString();
+
+      const text = doc.getText();
+      doc.transact(() => {
+        text.delete(0, text.length);
+        text.insert(0, tempText);
+      });
     }
   });
 };
+
+const createCollabSession = (roomId: string) => {
+  getDocument(roomId);
+  partnerReadiness.set(roomId, false);
+};
+
+const removeCollabSession = (roomId: string) => {
+  collabSessions.get(roomId)?.destroy();
+  collabSessions.delete(roomId);
+  partnerReadiness.delete(roomId);
+};
+
+const getDocument = (roomId: string) => {
+  let doc = collabSessions.get(roomId);
+  if (!doc) {
+    doc = new Doc();
+    doc.on(CollabEvents.UPDATE, (_update) => {
+      saveDocument(roomId, doc!);
+      io.to(roomId).emit(CollabEvents.UPDATE, encodeStateAsUpdateV2(doc!));
+    });
+    collabSessions.set(roomId, doc);
+  }
+
+  return doc;
+};
+
+const saveDocument = async (roomId: string, doc: Doc) => {
+  const docState = encodeStateAsUpdateV2(doc);
+  const docAsString = Buffer.from(docState).toString("base64");
+  await redisClient.set(`collaboration:${roomId}`, docAsString, {
+    EX: EXPIRY_TIME,
+  });
+};

backend/collab-service/src/handlers/websocketHandler.ts

@@ -1,7 +1,7 @@
 import http from "http";
 import app, { allowedOrigins } from "./app.ts";
 import { handleWebsocketCollabEvents } from "./handlers/websocketHandler.ts";
-import { Server } from "socket.io";
+import { Server, Socket } from "socket.io";
 import { connectRedis } from "./config/redis.ts";
 
 const server = http.createServer(app);
@@ -13,7 +13,7 @@ export const io = new Server(server, {
   connectionStateRecovery: {},
 });
 
-io.on("connection", (socket) => {
+io.on("connection", (socket: Socket) => {
   handleWebsocketCollabEvents(socket);
 });