sanket3280/code-execution

1/**2 * Submit Worker - BullMQ3 * Processes code submission jobs4 */5 6const { Worker } = require('bullmq');7const { config } = require('../config/env.config');8const logger = require('../utils/logger');9const codeEvaluationService = require('../services/codeEvaluation.service');10const UserCode = require('../models/UserCode');11const User = require('../models/User');12const Problem = require('../models/Problem');13const { markProblemAsSolved } = require('../utils/updateUserProgress');14 15// Create worker connection16const connection = config.redis.url 17  ? { url: config.redis.url }18  : {19      host: config.redis.host,20      port: config.redis.port,21      password: config.redis.password || undefined22    };23 24const worker = new Worker('submit', async (job) => {25  // Handle different job types26  if (job.name === 'deleteOldCode') {27    return handleDeleteOldCode(job);28  }29  30  const { userId, problemId, sourceCode, language, testCases } = job.data;31  32  logger.info(`Processing submit job ${job.id} for user ${userId}`);33  34  try {35    // Execute test cases36    const results = await codeEvaluationService.executeTestCases(37      sourceCode, 38      language, 39      testCases,40      {41        topic: job.data.category,42        runAllTestCases: true,43        verifierType: job.data.verifierType,44        verifierMetadata: job.data.validationRules || {}45      }46    );47    48    const total = results.summary.total;49    const passed = results.summary.passed;50    const verdict = (passed === total) ? 'accepted' : 'wrong_answer';51    52    // Calculate score (0-100)53    const score = total > 0 ? Math.round((passed / total) * 100) : 0;54    55    // Save submission (using Submission model)56    const Submission = require('../models/Submission');57    const submission = new Submission({58      user: userId,59      problem: problemId,60      challengeId: job.data.challengeId || null, // Associate with challenge61      language,62      code: sourceCode,63      status: verdict,64      score, // Add calculated score65      executionTime: results.executionTime || 0,66      memory: results.memory || 0,67      testCasesPassed: passed,68      totalTestCases: total,69      errorMessage: results.error || '',70      submittedAt: new Date(),71    });72    await submission.save();73    74    logger.info(`💾 Submission saved: ${verdict} (${passed}/${total}) score: ${score} for user ${userId}`);75    76    // Update user stats atomically77    const user = await User.findById(userId);78    if (user) {79      user.totalSubmissions = (user.totalSubmissions || 0) + 1;80      81      if (passed === total && total > 0) {82        // Mark problem as solved and save83        await markProblemAsSolved(userId, problemId);84        85        // If this is a challenge submission, track it separately86        if (job.data.challengeId) {87          const alreadySolved = user.solvedChallengeProblems?.some(88            scp => scp.challengeId?.toString() === job.data.challengeId && 89                   scp.problemId?.toString() === problemId90          );91          92          if (!alreadySolved) {93            if (!user.solvedChallengeProblems) user.solvedChallengeProblems = [];94            user.solvedChallengeProblems.push({95              challengeId: job.data.challengeId,96              problemId: problemId,97              solvedAt: new Date()98            });99            100            // Count unique challenges101            const uniqueChallenges = new Set(102              user.solvedChallengeProblems.map(scp => scp.challengeId?.toString())103            );104            user.challengesSolved = uniqueChallenges.size;105            106            await user.save();107            logger.info(`✅ Challenge problem tracked: ${user.challengesSolved} challenges solved`);108          }109        }110      } else {111        // Just save the submission count update112        await user.save();113      }114    }115    116    // Emit progress event for real-time updates117    const { dispatchProgressUpdate } = require('../utils/progressEvents');118    dispatchProgressUpdate({ 119      userId, 120      problemId, 121      solved: verdict === 'accepted',122      timestamp: Date.now()123    });124    125    // Trigger real-time leaderboard update if this is a challenge submission126    if (job.data.challengeId) {127      try {128        const { broadcastLeaderboardUpdate } = require('../sockets/challengeLeaderboard.socket');129        130        // Use global io instance131        if (global.io) {132          logger.info(`🔔 Broadcasting leaderboard update for challenge ${job.data.challengeId}`);133          134          // Broadcast leaderboard update with submission data135          await broadcastLeaderboardUpdate(global.io, job.data.challengeId, {136            userId,137            problemId,138            status: verdict,139            score,140            submissionId: submission._id,141            timestamp: new Date()142          });143          144          logger.info(`✅ Leaderboard broadcast sent for challenge ${job.data.challengeId}`);145        } else {146          logger.warn('⚠️ Socket.IO instance not available for leaderboard broadcast');147        }148      } catch (broadcastError) {149        logger.error('❌ Error broadcasting leaderboard update:', broadcastError);150        // Don't fail the job if broadcast fails151      }152    }153    154    logger.info(`Submit job ${job.id} completed: ${verdict} (${passed}/${total})`);155    156    return {157      submissionId: submission._id,158      verdict,159      passed,160      total,161      results162    };163  } catch (error) {164    logger.error(`Submit job ${job.id} failed:`, error);165    throw error;166  }167}, { connection, concurrency: 5 }); // Process 5 jobs concurrently168 169// Handler for auto-delete old code job170async function handleDeleteOldCode(job) {171  const { userId, problemId, language } = job.data;172  173  logger.info(`Processing deleteOldCode job ${job.id} for user ${userId}, problem ${problemId}, language ${language}`);174  175  try {176    // Find UserCode document177    const userCode = await UserCode.findOne({ user: userId, problem: problemId });178    179    if (!userCode) {180      logger.warn(`UserCode not found for deletion job ${job.id}`);181      return { success: false, message: 'UserCode not found' };182    }183    184    // Find and mark language entry as deleted185    const langIndex = userCode.codes.findIndex(c => c.language === language && !c.deleted);186    187    if (langIndex < 0) {188      logger.warn(`Language code not found for deletion job ${job.id}`);189      return { success: false, message: 'Language code not found' };190    }191    192    // Mark as deleted instead of removing (for audit trail)193    userCode.codes[langIndex].deleted = true;194    await userCode.save();195    196    logger.info(`Successfully deleted ${language} code for user ${userId}, problem ${problemId}`);197    198    return { success: true, message: 'Code deleted successfully' };199  } catch (error) {200    logger.error(`DeleteOldCode job ${job.id} failed:`, error);201    throw error;202  }203}204 205worker.on('completed', (job) => {206  logger.info(`Job ${job.id} completed`);207});208 209worker.on('failed', (job, err) => {210  logger.error(`Job ${job.id} failed:`, err);211});212 213logger.info('Submit worker started');214 215module.exports = worker;216