Change collaboration socket to use y-websocket default encoder

Author: SilasTSL

backend/collaboration-service/app.ts

@@ -8,8 +8,8 @@ import collaborationRoutes from "./routes/collaborationRoutes";
 import { errorHandler } from "./middlewares/errorHandler";
 import { connectToDatabase } from "./utils/database";
 import { setUpKafkaSubscribers } from "./utils/kafkaClient";
-import { YDocManager } from './utils/yjs';
-import * as Y from 'yjs';
+const setupWSConnection = require("y-websocket/bin/utils").setupWSConnection;
+
 
 dotenv.config({ path: path.resolve(__dirname, "./.env") });
 
@@ -36,40 +36,43 @@ 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();
+    // // 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);
+    // 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);
+    //     // 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('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();
-    }
+    //     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

@@ -2,7 +2,6 @@
 import { EachMessagePayload } from 'kafkajs';
 import { COLLAB_TOPIC, producer } from '../utils/kafkaClient';
 import Session from '../models/Session';
-import { YDocManager } from '../utils/yjs';
 
 
 export const createSession = async ( matchId: string, userIds: string[]) => {
@@ -12,9 +11,6 @@ export const createSession = async ( matchId: string, userIds: string[]) => {
     // Create new Session document in mongodb
     const newSession = new Session({ sessionId, matchId, userIds, codeContent: '' });
     await newSession.save();
-    
-    // Initialize Yjs document manager for this session
-    YDocManager.initializeDoc(sessionId);
 
     return sessionId;
 };

backend/collaboration-service/package-lock.json

@@ -32,6 +32,7 @@
     "@types/express": "^4.17.21",
     "@types/jsonwebtoken": "^9.0.7",
     "@types/mongoose": "^5.11.97",
+    "@types/node": "^22.8.6",
     "@types/socket.io": "^3.0.1",
     "@types/ws": "^8.5.12",
     "nodemon": "^3.1.1",