Integrate execution collaboration (#19)

* Add POST API for execution service

* Rewrite execution service to running code only

* WIP: latest changes before merging dev

* Modify docker-compose.yml

* Merge collaboration service and change execution-service

* feat(collaboration-service-frontend):

- Integrate collaboration service backend with into Code Editor panel
on /practice/:id pages.
- Add placeholder panels for Question, Communication, Code Execution.

* fix(code-editor-panel): dynamic loading monaco editor on client

* Error handling when worker can't connect to PistonAPI

* Add Agora Token generator

* Add CI workflow for execution service

* Integrate collaboration and execution, todo: broadcast to FE, and send code from FE

* Integrate code execution to collab FE

* Fix rabbitMQ startup error

* Add sync error output when execution dies

* Fix lock file

* Clean execution service

* Clean execution service and add .env example

* Add .env example

* Update README to include installation instructions for Piston language packages on Linux/Mac and Windows

* Fix error message

---------

Co-authored-by: k1b1t0 <[email protected]>
Co-authored-by: Nguyen Cao Duy <[email protected]>
Co-authored-by: TramMinhMan <[email protected]>

Author: 4 people

README.md

@@ -109,11 +109,22 @@ docker exec mongodb mongorestore --username admin --password password /data/back
 - The teaching team should be given access to the repositories as we may require viewing the history of the repository in case of any disputes or disagreements. 
 
 ## Note for execution service
-- After run docker compose-up, run the next commands in terminal
+- After run docker compose-up, run the next commands in terminal to install language packages for Piston
+
+### Linux/Mac (Bash)
 ```bash
 chmod +x init-piston-languages.sh
-# Install language packages for Piston
-./init-piston-languages.sh 
+./init-piston-languages.sh
+```
+
+### Windows (PowerShell)
+```powershell
+powershell -ExecutionPolicy Bypass -File .\init-piston-languages.ps1
+```
+
+Alternatively, if your PowerShell execution policy allows it:
+```powershell
+.\init-piston-languages.ps1
 ```
 - APIs
 ```bash

backend/collaboration-service/package.json

@@ -15,6 +15,7 @@
   "packageManager": "[email protected]",
   "dependencies": {
     "@y/websocket-server": "^0.1.1",
+    "amqplib": "^0.10.9",
     "cors": "^2.8.5",
     "express": "^5.1.0",
     "jsonwebtoken": "^9.0.2",

backend/collaboration-service/src/http/codeExecutionRoutes.js

@@ -0,0 +1,148 @@
+import express from 'express'
+import amqp from 'amqplib'
+import roomManager from '../websocket/roomManager.js'
+
+const router = express.Router();
+
+const RABBITMQ_URL = 'amqp://user:password@rabbitmq';
+const QUEUE_NAME = 'execution_jobs';
+
+const EXECUTION_TIMEOUT_MS = 30000 // 30s
+const activeTimers = new Map()
+
+let mqChannel = null
+
+
+export async function initRabbitMQ() {
+    try {
+        const connection = await amqp.connect(RABBITMQ_URL)
+        mqChannel = await connection.createChannel()
+
+        await mqChannel.assertQueue(QUEUE_NAME, {durable: true})
+
+        console.log("RabbitMQ Producer connected")
+    } catch (error) {
+        console.log("Producer can't connect to RabbitMQ")
+        console.log(error.message)
+        process.exit(1)
+    }
+}
+
+async function sendJob(job) {
+    let connection
+    try {
+        mqChannel.sendToQueue(
+            QUEUE_NAME,
+            Buffer.from(JSON.stringify(job)),
+            {persistent: true}
+        )
+
+        console.log('>>> Sent job: ', job.room_id)
+    } catch (error) {
+        console.log ("Error while sending job: ", error.message)
+    }
+}
+
+/**
+ * SUBMIT
+ * POST /api/v1/code/submit-code
+ * get code from FE and push a job to MQ
+ */
+router.post("/submit-code", async (req, res) => {
+    try {
+        // export job
+        const {room_id, language, source_code} = req.body
+        const job = {
+            room_id: room_id,
+            language: language,
+            source_code: source_code
+        }
+
+        console.log(">>> Got code ", job)
+
+        // delete old timers
+        if (activeTimers.has(room_id)) {
+            clearTimeout(activeTimers.get(room_id))
+            activeTimers.delete(room_id)
+        }
+
+        const newTimer = setTimeout(() => {
+            console.log('>>> Job timeout set for ', room_id)
+
+            activeTimers.delete(room_id)
+
+            const timeoutMessage = {
+                type: "code-execution-result",
+                data: {
+                    isError: true,
+                    output: "Execution timed out. Please try again"
+                }
+            }
+
+            roomManager.broadcastToRoom(room_id, timeoutMessage)
+        }, EXECUTION_TIMEOUT_MS)
+
+        activeTimers.set(room_id, newTimer)
+
+        // send job to MQ
+        await sendJob(job)
+        res.status(202).json({
+            message: "Job accepted"
+        })
+
+    } catch (error) {
+        console.error('Error during submission process:', error.message)
+        res.status(500).json({
+            error: 'Internal Server Error' 
+        })
+    }
+})
+
+/**
+ * CALLBACK
+ * POST /api/v1/code/execute-callback
+ * get result and push to FE
+ */
+router.post("/execute-callback", async (req, res) => {
+    try {
+        // get result
+        const {room_id, isError, output} = req.body
+        
+        // delete the timeout or ignore if the job is timeout
+        if (activeTimers.has(room_id)) {
+            clearTimeout(activeTimers.get(room_id))
+            activeTimers.delete(room_id)
+        } else {
+            console.log(`>>> Received STALE callback for ${room_id}`)
+            return res.status(200).json({message: 'Job already timeout'})
+        }
+
+        // send result to FE
+        const message = {
+            type: "code-execution-result",
+            data: {
+                isError: isError,
+                output: output
+            }
+        }
+
+        roomManager.broadcastToRoom(room_id, message)
+
+        console.log('>>> Callback, broadcasted to room')
+        console.log('room_id: ', room_id)
+        console.log('isError: ', isError)
+        console.log('output:', output)
+
+        res.status(200).json({
+            message: 'Got result and broadcasted to room'
+        })
+
+    } catch (error) {
+        console.error('Error during submission process:', error.message)
+        res.status(500).json({
+            error: 'Internal Server Error' 
+        })
+    }
+})
+
+export default router

backend/collaboration-service/src/http/httpServer.js

@@ -3,6 +3,7 @@ import cors from "cors";
 import http from "http";
 import config from "../config.js";
 import roomRoutes from "./roomRoutes.js";
+import codeExecutionRoutes from "./codeExecutionRoutes.js"
 
 class HttpServer {
   constructor() {
@@ -29,6 +30,7 @@ class HttpServer {
    * Set up API routes
    */
   setupRoutes() {
+    this.app.use("/api/v1/code", codeExecutionRoutes)
     this.app.use("/api/v1/rooms", roomRoutes);
 
     this.app.use((req, res) => {

backend/collaboration-service/src/server.js

@@ -1,17 +1,21 @@
 import db from "./db.js";
 import httpServer from "./http/httpServer.js";
 import webSocketServer from "./websocket/websocketServer.js";
+import {initRabbitMQ} from "./http/codeExecutionRoutes.js";
 import { setupRoomCreationConsumer } from "./consumers/roomCreationConsumer.js";
 
 async function startServer() {
   try {
     // Step 1: Connect to MongoDB
     await db.connect();
 
-    // Step 2: Start HTTP server
+    // Step 2: Connect to RabbitMQ
+    await initRabbitMQ();
+
+    // Step 3: Start HTTP server
     await httpServer.start();
 
-    // Step 3: Start WebSocket server
+    // Step 4: Start WebSocket server
     await webSocketServer.start();
 
     // Step 4: Start Kafka consumer for room creation

backend/execution-service/.env example

@@ -0,0 +1,4 @@
+RABBITMQ_URL="amqp://user:password@rabbitmq"
+QUEUE_NAME="execution_jobs"
+CALLBACK_URL="http://collaboration-service:8004/api/v1/code/execute-callback"
+PISTON_URL="http://piston-api:2000/api/v2/execute"

backend/execution-service/.env.example

@@ -2,26 +2,21 @@
   "name": "execution-service",
   "version": "1.0.0",
   "description": "",
-  "main": "src/api.js",
+  "main": "src/worker.js",
   "scripts": {
-    "start:api": "node src/api.js",
-    "start:worker": "node src/worker.js",
-    "dev:api": "nodemon src/api.js",
+    "start": "node src/worker.js",
     "dev:worker": "nodemon src/worker.js",
-    "dev": "concurrently \"npm:dev:api\" \"npm:dev:worker\"",
     "test": "echo \"Error: no test specified\" && exit 1"
   },
   "keywords": [],
   "author": "",
   "license": "ISC",
   "packageManager": "[email protected]",
   "dependencies": {
-    "axios": "^1.12.2",
-    "express": "^5.1.0",
-    "mongoose": "^8.19.1"
+    "amqplib": "^0.10.9",
+    "axios": "^1.12.2"
   },
   "devDependencies": {
-    "concurrently": "^9.2.1",
     "nodemon": "^3.1.10"
   }
 }