chore: fix code rabbit suggestions

Author: coodos

platforms/eReputation-api/package.json

@@ -14,9 +14,11 @@
     },
     "dependencies": {
         "axios": "^1.6.7",
+        "bullmq": "^5.3.0",
         "cors": "^2.8.5",
         "dotenv": "^16.4.5",
         "express": "^4.18.2",
+        "ioredis": "^5.3.2",
         "jsonwebtoken": "^9.0.2",
         "openai": "^4.20.1",
         "pg": "^8.11.3",

platforms/eReputation-api/src/controllers/WebhookController.ts

@@ -15,6 +15,8 @@ import { GroupController } from "./controllers/GroupController";
 import { DashboardController } from "./controllers/DashboardController";
 import { authMiddleware, authGuard } from "./middleware/auth";
 import { adapter } from "./web3adapter/watchers/subscriber";
+import { JobQueueService } from "./services/JobQueueService";
+import { PollReputationWorker } from "./workers/PollReputationWorker";
 
 config({ path: path.resolve(__dirname, "../../../.env") });
 
@@ -49,10 +51,14 @@ app.use(
 app.use(express.json({ limit: "50mb" }));
 app.use(express.urlencoded({ limit: "50mb", extended: true }));
 
+// Initialize job queue and worker
+const jobQueueService = new JobQueueService();
+const pollReputationWorker = new PollReputationWorker(jobQueueService);
+
 // Controllers
 const userController = new UserController();
 const authController = new AuthController();
-const webhookController = new WebhookController();
+const webhookController = new WebhookController(jobQueueService);
 const referenceController = new ReferenceController();
 const calculationController = new CalculationController();
 const platformController = new PlatformController();

platforms/eReputation-api/src/index.ts

@@ -1,7 +1,7 @@
 import type { NextFunction, Request, Response } from "express";
 import { AppDataSource } from "../database/data-source";
 import { User } from "../database/entities/User";
-import { verifyToken } from "../utils/jwt";
+import { verifyToken, AuthTokenPayload } from "../utils/jwt";
 
 export const authMiddleware = async (
     req: Request,
@@ -16,11 +16,7 @@ export const authMiddleware = async (
         }
 
         const token = authHeader.split(" ")[1];
-        const decoded = verifyToken(token) as { userId: string };
-
-        if (!decoded?.userId) {
-            return res.status(401).json({ error: "Invalid token" });
-        }
+        const decoded: AuthTokenPayload = verifyToken(token);
 
         const userRepository = AppDataSource.getRepository(User);
         const user = await userRepository.findOneBy({ id: decoded.userId });

platforms/eReputation-api/src/middleware/auth.ts

@@ -1,4 +1,4 @@
-import { Repository } from "typeorm";
+import { Repository, In } from "typeorm";
 import { AppDataSource } from "../database/data-source";
 import { Group } from "../database/entities/Group";
 import { User } from "../database/entities/User";
@@ -23,8 +23,8 @@ export class GroupService {
         name: string,
         description: string,
         owner: string,
-        adminIds: string[],
-        participantIds: string[],
+        adminIds: string[] = [],
+        participantIds: string[] = [],
         charter?: string
     ): Promise<Group> {
         const group = this.groupRepository.create({
@@ -36,13 +36,13 @@ export class GroupService {
 
         // Add admins
         if (adminIds.length > 0) {
-            const admins = await this.userRepository.findByIds(adminIds);
+            const admins = await this.userRepository.findBy({ id: In(adminIds) });
             group.admins = admins;
         }
 
         // Add participants
         if (participantIds.length > 0) {
-            const participants = await this.userRepository.findByIds(participantIds);
+            const participants = await this.userRepository.findBy({ id: In(participantIds) });
             group.participants = participants;
         }
 

platforms/eReputation-api/src/services/GroupService.ts

@@ -0,0 +1,100 @@
+import { Queue, QueueOptions } from "bullmq";
+import Redis from "ioredis";
+
+export interface PollReputationJobData {
+    pollId: string;
+    eventId: string; // For idempotency
+    groupId: string;
+}
+
+export class JobQueueService {
+    private queue: Queue<PollReputationJobData>;
+    private redis: Redis;
+
+    constructor() {
+        // Create Redis connection
+        this.redis = new Redis({
+            host: process.env.REDIS_HOST || "localhost",
+            port: parseInt(process.env.REDIS_PORT || "6379", 10),
+            maxRetriesPerRequest: null,
+        });
+
+        // Create BullMQ queue
+        const queueOptions: QueueOptions = {
+            connection: {
+                host: process.env.REDIS_HOST || "localhost",
+                port: parseInt(process.env.REDIS_PORT || "6379", 10),
+            },
+            defaultJobOptions: {
+                attempts: 3,
+                backoff: {
+                    type: "exponential",
+                    delay: 2000,
+                },
+                removeOnComplete: {
+                    age: 3600, // Keep completed jobs for 1 hour
+                    count: 1000,
+                },
+                removeOnFail: {
+                    age: 86400, // Keep failed jobs for 24 hours
+                },
+            },
+        };
+
+        this.queue = new Queue<PollReputationJobData>("poll-reputation-calculation", queueOptions);
+    }
+
+    /**
+     * Enqueue a poll reputation calculation job with deduplication
+     */
+    async enqueuePollReputationJob(
+        pollId: string,
+        groupId: string,
+        eventId: string
+    ): Promise<void> {
+        try {
+            // Use pollId as the job ID for deduplication (same poll = same job)
+            const jobId = `poll-reputation:${pollId}`;
+
+            // Check if job already exists or was recently processed
+            const existingJob = await this.queue.getJob(jobId);
+            if (existingJob) {
+                const state = await existingJob.getState();
+                if (state === "active" || state === "waiting" || state === "delayed") {
+                    // Job already queued, skip
+                    return;
+                }
+            }
+
+            // Add job with deduplication
+            await this.queue.add(
+                "calculate-poll-reputation",
+                {
+                    pollId,
+                    groupId,
+                    eventId,
+                },
+                {
+                    jobId, // Use pollId as job ID for deduplication
+                    removeOnComplete: true,
+                    removeOnFail: false,
+                }
+            );
+        } catch (error) {
+            throw error;
+        }
+    }
+
+    /**
+     * Close the queue connection
+     */
+    async close(): Promise<void> {
+        await this.queue.close();
+        await this.redis.quit();
+    }
+
+    getQueue(): Queue<PollReputationJobData> {
+        return this.queue;
+    }
+}
+

platforms/eReputation-api/src/services/JobQueueService.ts

@@ -221,8 +221,21 @@ export class VotingReputationService {
                     ename = member?.ename || null;
                 }
 
-                if (!ename || !item.score || typeof item.score !== 'number' || !item.justification || typeof item.justification !== 'string') {
-                    console.error(`   ❌ Invalid result at index ${index}:`, item);
+                // Explicit validation: ensure ename is a non-empty string
+                if (typeof ename !== 'string' || ename.trim() === '') {
+                    console.error(`   ❌ Invalid result at index ${index}: ename is not a non-empty string`, item);
+                    return;
+                }
+                
+                // Explicit validation: ensure score is a finite number
+                if (typeof item.score !== 'number' || !Number.isFinite(item.score)) {
+                    console.error(`   ❌ Invalid result at index ${index}: score is not a finite number`, item);
+                    return;
+                }
+                
+                // Explicit validation: ensure justification is a string
+                if (typeof item.justification !== 'string') {
+                    console.error(`   ❌ Invalid result at index ${index}: justification is not a string`, item);
                     return;
                 }
 

platforms/eReputation-api/src/services/VotingReputationService.ts

@@ -1,11 +1,29 @@
-import jwt from "jsonwebtoken";
+import jwt, { JwtPayload } from "jsonwebtoken";
 
-const JWT_SECRET = process.env.JWT_SECRET || "your-secret-key";
+// Fail fast if JWT_SECRET is missing
+if (!process.env.JWT_SECRET) {
+    throw new Error("JWT_SECRET environment variable is required but was not provided. Please set JWT_SECRET in your environment configuration.");
+}
 
-export const signToken = (payload: { userId: string }): string => {
+const JWT_SECRET = process.env.JWT_SECRET;
+
+export interface AuthTokenPayload {
+    userId: string;
+}
+
+export const signToken = (payload: AuthTokenPayload): string => {
     return jwt.sign(payload, JWT_SECRET, { expiresIn: "7d" });
 };
 
-export const verifyToken = (token: string): any => {
-    return jwt.verify(token, JWT_SECRET);
+export const verifyToken = (token: string): AuthTokenPayload => {
+    const decoded = jwt.verify(token, JWT_SECRET) as JwtPayload & AuthTokenPayload;
+    
+    // Validate that the decoded token has the required userId field
+    if (!decoded.userId || typeof decoded.userId !== 'string') {
+        throw new Error("Invalid token: missing or invalid userId");
+    }
+    
+    return {
+        userId: decoded.userId
+    };
 };