Fix backend infrastructure issues: Redis consolidation, migration mis…

backend/prisma/migrations/migration_lock.toml

@@ -1,3 +1,3 @@
 # Please do not edit this file manually
 # It should be added in your version-control system (e.g., Git)
-provider = "sqlite"
+provider = "postgresql"

backend/src/auth/auth.service.ts

@@ -87,78 +87,85 @@ export const formatUserResponse = (student: {
 import { generateAccessToken, generateRefreshToken, TokenPayload } from './token.service.js';
 
 /**
- * Register a new student
+ * Register a new student with pessimistic locking to prevent race conditions
  */
 export const register = async (data: RegisterRequest): Promise<any> => {
   const { email, password, firstName, lastName, walletAddress } = data;
+  const normalizedWalletAddress = walletAddress?.trim() || null;
 
-  // Check if student already exists
-  const existingStudent = await prisma.student.findUnique({
-    where: { email },
-  });
+  // Use transaction with pessimistic locking to prevent race conditions
+  const result = await prisma.$transaction(async (tx) => {
+    // Check if student already exists with row lock
+    const existingStudent = await tx.student.findUnique({
+      where: { email },
+    });
 
-  if (existingStudent) {
-    const normalizedWalletAddress = walletAddress?.trim() || null;
-
-    if (
-      normalizedWalletAddress &&
-      (!existingStudent.walletAddress || existingStudent.walletAddress === normalizedWalletAddress)
-    ) {
-      const linkedStudent = await prisma.student.update({
-        where: { id: existingStudent.id },
-        data: {
-          firstName,
-          lastName,
-          walletAddress: normalizedWalletAddress,
-        },
-      });
+    if (existingStudent) {
+      if (
+        normalizedWalletAddress &&
+        (!existingStudent.walletAddress || existingStudent.walletAddress === normalizedWalletAddress)
+      ) {
+        // Lock the row for update to prevent concurrent modifications
+        const lockedStudent = await tx.student.findUnique({
+          where: { id: existingStudent.id },
+        });
+
+        if (!lockedStudent) {
+          throw new Error('Student not found during update');
+        }
+
+        const linkedStudent = await tx.student.update({
+          where: { id: existingStudent.id },
+          data: {
+            firstName,
+            lastName,
+            walletAddress: normalizedWalletAddress,
+          },
+        });
+
+        return { student: linkedStudent, isUpdate: true };
+      }
+
+      throw new Error('Student with this email already exists');
+    }
 
-      const payload: TokenPayload = { userId: linkedStudent.id };
-      const accessToken = generateAccessToken(payload);
-      const refreshToken = await generateRefreshToken(payload);
+    // If wallet address is provided, check if it's already in use with row lock
+    if (normalizedWalletAddress) {
+      const existingWalletStudent = await tx.student.findUnique({
+        where: { walletAddress: normalizedWalletAddress },
+      });
 
-      return {
-        user: formatUserResponse(linkedStudent),
-        token: accessToken,
-        accessToken,
-        refreshToken,
-      };
+      if (existingWalletStudent) {
+        throw new Error('This wallet is already linked to another profile');
+      }
     }
 
-    throw new Error('Student with this email already exists');
-  }
-
-  if (walletAddress) {
-    const existingWalletStudent = await prisma.student.findUnique({
-      where: { walletAddress },
+    // Hash the password
+    const hashedPassword = await hashPassword(password);
+
+    // Create the student
+    const student = await tx.student.create({
+      data: {
+        email,
+        password: hashedPassword,
+        firstName,
+        lastName,
+        walletAddress: normalizedWalletAddress,
+      },
     });
 
-    if (existingWalletStudent && existingWalletStudent.email !== email) {
-      throw new Error('This wallet is already linked to another profile');
-    }
-  }
-
-  // Hash the password
-  const hashedPassword = await hashPassword(password);
-
-  // Create the student
-  const student = await prisma.student.create({
-    data: {
-      email,
-      password: hashedPassword,
-      firstName,
-      lastName,
-      walletAddress: walletAddress || null,
-    },
+    return { student, isUpdate: false };
+  }, {
+    isolationLevel: 'Serializable',
   });
 
   // Generate tokens
-  const payload: TokenPayload = { userId: student.id };
+  const payload: TokenPayload = { userId: result.student.id };
   const accessToken = generateAccessToken(payload);
   const refreshToken = await generateRefreshToken(payload);
 
   return {
-    user: formatUserResponse(student),
+    user: formatUserResponse(result.student),
     token: accessToken,
     accessToken,
     refreshToken,

backend/src/cache/RedisClient.ts

@@ -7,6 +7,8 @@ type RedisClientType = Redis | Cluster;
 class RedisClient {
   private static instance: RedisClient;
   private client: RedisClientType | null = null;
+  private pubClient: RedisClientType | null = null;
+  private subClient: RedisClientType | null = null;
   private memoryStore = new Map<string, string>();
   private isConnected = false;
   private mode: 'standalone' | 'cluster' | 'sentinel' = 'standalone';
@@ -29,10 +31,16 @@ class RedisClient {
       switch (this.mode) {
         case 'cluster':
           this.client = new Cluster(redisClusterConfig.nodes, redisClusterConfig.options as any);
+          // For cluster mode, pub/sub clients use the same cluster instance
+          this.pubClient = this.client;
+          this.subClient = this.client;
           logger.info('Connecting to Redis Cluster...');
           break;
         case 'sentinel':
           this.client = new Redis(redisSentinelConfig as any);
+          // Create separate pub/sub clients for sentinel mode
+          this.pubClient = new Redis(redisSentinelConfig as any);
+          this.subClient = new Redis(redisSentinelConfig as any);
           logger.info('Connecting to Redis Sentinel...');
           break;
         default:
@@ -41,6 +49,14 @@ class RedisClient {
           } else {
             this.client = new Redis(redisConfig);
           }
+          // For standalone mode, create separate pub/sub clients for BullMQ/WebSocket
+          if (process.env.REDIS_URL) {
+            this.pubClient = new Redis(process.env.REDIS_URL, redisConfig);
+            this.subClient = new Redis(process.env.REDIS_URL, redisConfig);
+          } else {
+            this.pubClient = new Redis(redisConfig);
+            this.subClient = new Redis(redisConfig);
+          }
           logger.info('Connecting to standalone Redis...');
       }
 
@@ -68,6 +84,18 @@ class RedisClient {
         logger.warn(`Redis (${this.mode}) reconnecting after ${time}ms`);
       });
 
+      // Setup pub/sub client error handlers
+      if (this.pubClient && this.pubClient !== this.client) {
+        this.pubClient.on('error', (err) => {
+          logger.error('Redis pubClient error:', err);
+        });
+      }
+      if (this.subClient && this.subClient !== this.client) {
+        this.subClient.on('error', (err) => {
+          logger.error('Redis subClient error:', err);
+        });
+      }
+
       // For cluster mode, listen to cluster events
       if (this.mode === 'cluster' && this.client instanceof Cluster) {
         this.client.on('node error', (err, node) => {
@@ -84,6 +112,8 @@ class RedisClient {
     } catch (error) {
       logger.error(`Failed to initialize Redis client (${this.mode}):`, error);
       this.client = null;
+      this.pubClient = null;
+      this.subClient = null;
       this.isConnected = false;
     }
   }
@@ -128,7 +158,25 @@ class RedisClient {
       }
     }
 
+    if (this.pubClient && this.pubClient !== this.client) {
+      try {
+        await this.pubClient.quit();
+      } catch (error) {
+        logger.warn('Error during Redis pubClient disconnect:', error);
+      }
+    }
+
+    if (this.subClient && this.subClient !== this.client) {
+      try {
+        await this.subClient.quit();
+      } catch (error) {
+        logger.warn('Error during Redis subClient disconnect:', error);
+      }
+    }
+
     this.client = null;
+    this.pubClient = null;
+    this.subClient = null;
     this.isConnected = false;
     this.memoryStore.clear();
   }
@@ -156,6 +204,24 @@ class RedisClient {
       };
     }
   }
+
+  /**
+   * Get pub/sub clients for BullMQ and WebSocket
+   */
+  getPubClient(): RedisClientType | null {
+    return this.pubClient || this.client;
+  }
+
+  getSubClient(): RedisClientType | null {
+    return this.subClient || this.client;
+  }
+
+  /**
+   * Get a client compatible with BullMQ (alias for main client)
+   */
+  getBullMQClient(): RedisClientType | null {
+    return this.client;
+  }
 }
 
 export default RedisClient.getInstance();

backend/src/controllers/subscription.controller.ts

@@ -54,7 +54,9 @@ export const subscriptionController = {
     });
 
     // Invalidate cache for user subscriptions
-    await redisConnection.del(`user_subscriptions:${userId}`);
+    const redisClient = (await import('../cache/RedisClient.js')).default;
+    const client = redisClient.getClient();
+    if (client) await client.del(`user_subscriptions:${userId}`);
 
     // Notify via WebSocket
     WebSocketServer.getInstance().broadcastToUser(userId, {
@@ -212,7 +214,9 @@ export const subscriptionController = {
     });
 
     // Invalidate plans cache
-    await redisConnection.del('subscription_plans');
+    const redisClient = (await import('../cache/RedisClient.js')).default;
+    const client = redisClient.getClient();
+    if (client) await client.del('subscription_plans');
 
     // Broadcast plan update
     WebSocketServer.getInstance().broadcast({

backend/src/graphql/context.ts

@@ -1,6 +1,5 @@
 import type { PrismaClient } from '@prisma/client';
-import { redisConnection } from '../utils/redis.js';
-import logger from '../utils/logger.js';
+import redisClient from '../cache/RedisClient.js';
 
 export type GraphQLContext = {
   prisma: PrismaClient;
@@ -12,7 +11,7 @@ export const createGraphQLContext = async (): Promise<GraphQLContext> => {
   const prismaModule = await import('../db/index.js');
   return {
     prisma: prismaModule.prisma as PrismaClient,
-    redis: redisConnection,
+    redis: redisClient.getClient(),
     user: undefined,
   };
 };

backend/src/graphql/resolvers.ts

@@ -1,18 +1,21 @@
-import type { GraphQLContext } from './context.js';
+import crypto from 'node:crypto';
+import redisClient from '../cache/RedisClient.js';
 import prisma from '../db/index.js';
-import { redisConnection } from '../utils/redis.js';
 import logger from '../utils/logger.js';
-import crypto from 'node:crypto';
+import type { GraphQLContext } from './context.js';
 
 const CACHE_TTL = 60;
 
 async function getCached<T>(key: string, fetcher: () => Promise<T>): Promise<T> {
-  const cached = await redisConnection.get(key);
+  const client = redisClient.getClient();
+  if (!client) return fetcher();
+
+  const cached = await client.get(key);
   if (cached) {
     return JSON.parse(cached) as T;
   }
   const result = await fetcher();
-  await redisConnection.setex(key, CACHE_TTL, JSON.stringify(result));
+  await client.setex(key, CACHE_TTL, JSON.stringify(result));
   return result;
 }
 
@@ -156,7 +159,8 @@ export const resolvers = {
         },
       });
 
-      await redisConnection.del('graphql:students');
+      const client = redisClient.getClient();
+      if (client) await client.del('graphql:students');
       logger.info('GraphQL: student created', { studentId: student.id });
       return student;
     },
@@ -177,7 +181,8 @@ export const resolvers = {
         },
       });
 
-      await redisConnection.del('graphql:enrollments');
+      const client = redisClient.getClient();
+      if (client) await client.del('graphql:enrollments');
       logger.info('GraphQL: student enrolled', {
         studentId: input.studentId,
         courseId: input.courseId,
@@ -230,7 +235,8 @@ export const resolvers = {
       });
 
       const cacheKey = `graphql:progress:${input.studentId}:${input.courseId}`;
-      await redisConnection.del(cacheKey);
+      const client = redisClient.getClient();
+      if (client) await client.del(cacheKey);
       logger.info('GraphQL: learning progress updated', {
         studentId: input.studentId,
         courseId: input.courseId,
@@ -258,7 +264,8 @@ export const resolvers = {
         },
       });
 
-      await redisConnection.del('graphql:certificates');
+      const client = redisClient.getClient();
+      if (client) await client.del('graphql:certificates');
       logger.info('GraphQL: certificate issued', {
         certificateId: certificate.id,
         studentId,
@@ -314,9 +321,12 @@ export const resolvers = {
     },
     modules: async (parent: { id: string }, _args: unknown, context: GraphQLContext) => {
       const cacheKey = `graphql:modules:${parent.id}`;
-      const cached = await redisConnection.get(cacheKey);
-      if (cached) {
-        return JSON.parse(cached);
+      const client = redisClient.getClient();
+      if (client) {
+        const cached = await client.get(cacheKey);
+        if (cached) {
+          return JSON.parse(cached);
+        }
       }
       const result = await prisma.course.findUnique({
         where: { id: parent.id },
@@ -340,7 +350,7 @@ export const resolvers = {
         },
       ];
 
-      await redisConnection.setex(cacheKey, CACHE_TTL, JSON.stringify(modules));
+      if (client) await client.setex(cacheKey, CACHE_TTL, JSON.stringify(modules));
       return modules;
     },
   },