Merge pull request #3090 from bluewave-labs/develop

Merge develop into master (Jan 15)

Author: HarshP4585

Servers/database/redis.ts

@@ -1,6 +1,8 @@
 import IORedis from "ioredis";
 
-const redisClient = new IORedis(process.env.REDIS_URL || "redis://localhost:6379/0", {
+export const REDIS_URL = process.env.REDIS_URL || "redis://localhost:6379/0";
+
+const redisClient = new IORedis(REDIS_URL, {
   maxRetriesPerRequest: null,
 });
 

Servers/services/automations/automationProducer.ts

@@ -1,10 +1,10 @@
 import { Queue } from "bullmq";
-import redisClient from "../../database/redis"
+import { REDIS_URL } from "../../database/redis"
 import logger from "../../utils/logger/fileLogger";
 
 // Create a new queue (connected to Redis using environment variable)
 export const automationQueue = new Queue("automation-actions", {
-  connection: redisClient
+  connection: { url: REDIS_URL }
 });
 
 export async function enqueueAutomationAction(

Servers/services/automations/automationWorker.ts

@@ -1,5 +1,5 @@
 import { Worker, Job } from "bullmq";
-import redisClient from "../../database/redis";
+import { REDIS_URL } from "../../database/redis";
 import sendEmail from "./actions/sendEmail";
 import { getTenantHash } from "../../tools/getTenantHash";
 import { getAllOrganizationsQuery } from "../../utils/organization.utils";
@@ -396,7 +396,7 @@ export const createAutomationWorker = () => {
         throw error;
       }
     },
-    { connection: redisClient, concurrency: 10 }
+    { connection: { url: REDIS_URL }, concurrency: 10 }
   );
   automationWorker.on("completed", (job) => {
     console.log(`Job ${job.id} of type ${job.name} has been completed`);

Servers/services/mlflow/mlflowSyncProducer.ts

@@ -1,9 +1,9 @@
 import { Queue } from "bullmq";
-import redisClient from "../../database/redis";
+import { REDIS_URL } from "../../database/redis";
 import logger from "../../utils/logger/fileLogger";
 
 export const mlflowSyncQueue = new Queue("mlflow-sync", {
-  connection: redisClient,
+  connection: { url: REDIS_URL },
 });
 
 const SYNC_JOB_NAME = "mlflow-sync-all-orgs";

Servers/services/mlflow/mlflowSyncWorker.ts

@@ -1,5 +1,5 @@
 import { Job, Worker } from "bullmq";
-import redisClient from "../../database/redis";
+import { REDIS_URL } from "../../database/redis";
 import logger from "../../utils/logger/fileLogger";
 import { MLFlowService } from "../../src/services/mlflow.service";
 import { ValidationException } from "../../domain.layer/exceptions/custom.exception";
@@ -146,7 +146,7 @@ export const createMlflowSyncWorker = () => {
       return [];
     },
     {
-      connection: redisClient,
+      connection: { url: REDIS_URL },
       concurrency: 1,
     }
   );

Servers/services/slack/slackProducer.ts

@@ -1,10 +1,10 @@
 import { Queue } from "bullmq";
 import logger from "../../utils/logger/fileLogger";
-import redisClient from "../../database/redis"
+import { REDIS_URL } from "../../database/redis"
 
 // Create a new queue (connected to Redis using environment variable)
 export const notificationQueue = new Queue("slack-notifications", {
-  connection: redisClient,
+  connection: { url: REDIS_URL },
 });
 
 export async function scheduleDailyNotification() {

Servers/services/slack/slackWorker.ts

@@ -1,5 +1,5 @@
 import { Worker, Job } from "bullmq";
-import redisClient from "../../database/redis";
+import { REDIS_URL } from "../../database/redis";
 
 import { sendPolicyDueSoonNotification } from "./policyDueSoonNotification";
 
@@ -14,7 +14,7 @@ export const createNotificationWorker = () => {
         throw new Error(`Unknown job type: ${job.data.type}`);
       }
     },
-    { connection: redisClient }
+    { connection: { url: REDIS_URL } }
   );
 
   worker.on("completed", (_job) => {