feat: implement background queue system using BullMQ for cron, crawl, and webhook processing

render.yaml

+services:
+  - type: web
+    name: data-crawler-be
+    env: node
+    plan: free
+    region: singapore
+    buildCommand: pnpm install --frozen-lockfile && pnpm prisma:generate && pnpm build
+    startCommand: pnpm start
+    healthCheckPath: /health/liveness
+    envVars:
+      - key: NODE_ENV
+        value: production
+      - key: TRUST_PROXY
+        value: "true"
+      - key: START_CRAWL_WORKER
+        value: "true"
+      - key: DATABASE_URL
+        sync: false
+      - key: JWT_ACCESS_SECRET
+        generateValue: true
+      - key: JWT_REFRESH_SECRET
+        generateValue: true
+      - key: REDIS_ENABLED
+        value: "true"
+      - key: REDIS_URL
+        sync: false
+      - key: STORAGE_DRIVER
+        value: s3
+      - key: S3_ENDPOINT
+        sync: false
+      - key: S3_REGION
+        value: ap-southeast-1
+      - key: S3_BUCKET
+        value: data-crawler-exports
+      - key: S3_ACCESS_KEY_ID
+        sync: false
+      - key: S3_SECRET_ACCESS_KEY
+        sync: false
+      - key: S3_FORCE_PATH_STYLE
+        value: "true"
+      - key: FRONTEND_URL
+        sync: false

src/common/redis/redis-client.ts

 import Redis from "ioredis";
 import { envConfig } from "../../config/env.config";
+import { getRedisClientOptions } from "./redis-connection";
 
 let generalClient: Redis | null = null;
 
@@ -25,15 +26,10 @@ export function getRedisClient(): Redis | null {
 
   if (!generalClient) {
     try {
-      generalClient = new Redis({
-        host: envConfig.redis.host,
-        port: envConfig.redis.port,
-        maxRetriesPerRequest: 1,
-        lazyConnect: true,
-        connectTimeout: 2000,
-        retryStrategy: () => null,
-        enableOfflineQueue: false,
-      });
+      const conn = getRedisClientOptions();
+      generalClient = conn.url
+        ? new Redis(conn.url, conn.options)
+        : new Redis(conn.options);
 
       generalClient.on("error", () => {
         // Suppress unhandled crash logs on reconnect/timeout

src/common/redis/redis-connection.ts

+import { envConfig } from "../../config/env.config";
+import type { ConnectionOptions } from "bullmq";
+import type { RedisOptions } from "ioredis";
+
+/**
+ * Cung cấp tùy chọn kết nối Redis cho ioredis (hỗ trợ cả REDIS_URL và host/port/password)
+ */
+export function getRedisClientOptions(customOpts: RedisOptions = {}): {
+  url?: string;
+  options: RedisOptions;
+} {
+  const commonOpts: RedisOptions = {
+    maxRetriesPerRequest: 1,
+    lazyConnect: true,
+    connectTimeout: 5000,
+    retryStrategy: () => null,
+    enableOfflineQueue: false,
+    ...customOpts,
+  };
+
+  if (envConfig.redis.url) {
+    return {
+      url: envConfig.redis.url,
+      options: commonOpts,
+    };
+  }
+
+  return {
+    options: {
+      host: envConfig.redis.host,
+      port: envConfig.redis.port,
+      password: envConfig.redis.password,
+      ...commonOpts,
+    },
+  };
+}
+
+/**
+ * Cung cấp ConnectionOptions cho BullMQ Queues và Workers (hỗ trợ cả REDIS_URL và host/port/password)
+ */
+export function getBullMQConnection(
+  extraOpts: Record<string, unknown> = {},
+): ConnectionOptions {
+  if (envConfig.redis.url) {
+    return {
+      url: envConfig.redis.url,
+      ...extraOpts,
+    };
+  }
+
+  return {
+    host: envConfig.redis.host,
+    port: envConfig.redis.port,
+    password: envConfig.redis.password,
+    ...extraOpts,
+  };
+}

src/common/redis/redis-pubsub.ts

 import Redis from "ioredis";
 import { envConfig } from "../../config/env.config";
+import { getRedisClientOptions } from "./redis-connection";
 
 let publisherClient: Redis | null = null;
 let subscriberClient: Redis | null = null;
@@ -9,15 +10,10 @@ export function getRedisPublisher(): Redis | null {
 
   if (!publisherClient) {
     try {
-      publisherClient = new Redis({
-        host: envConfig.redis.host,
-        port: envConfig.redis.port,
-        maxRetriesPerRequest: 1,
-        lazyConnect: true,
-        connectTimeout: 2000,
-        retryStrategy: () => null,
-        enableOfflineQueue: false,
-      });
+      const conn = getRedisClientOptions();
+      publisherClient = conn.url
+        ? new Redis(conn.url, conn.options)
+        : new Redis(conn.options);
 
       publisherClient.on("error", () => {
         // Suppress unhandled redis error crashes
@@ -35,15 +31,10 @@ export function getRedisSubscriber(): Redis | null {
 
   if (!subscriberClient) {
     try {
-      subscriberClient = new Redis({
-        host: envConfig.redis.host,
-        port: envConfig.redis.port,
-        maxRetriesPerRequest: 1,
-        lazyConnect: true,
-        connectTimeout: 2000,
-        retryStrategy: () => null,
-        enableOfflineQueue: false,
-      });
+      const conn = getRedisClientOptions();
+      subscriberClient = conn.url
+        ? new Redis(conn.url, conn.options)
+        : new Redis(conn.options);
 
       subscriberClient.on("error", () => {
         // Suppress unhandled redis error crashes

src/config/env.config.ts

@@ -52,9 +52,12 @@ export const envConfig = {
     ),
   },
   redis: {
+    url: process.env.REDIS_URL || "",
     host: process.env.REDIS_HOST || "127.0.0.1",
     port: parseInt(process.env.REDIS_PORT || "6379", 10),
-    enabled: process.env.REDIS_ENABLED === "true",
+    password: process.env.REDIS_PASSWORD || undefined,
+    enabled:
+      process.env.REDIS_ENABLED === "true" || Boolean(process.env.REDIS_URL),
   },
   rateLimit: {
     windowMs: parseInt(process.env.RATE_LIMIT_WINDOW_MS || "900000", 10),

src/queues/crawl.queue.ts

 import { Queue } from "bullmq";
 import { envConfig } from "../config/env.config";
+import { getBullMQConnection } from "../common/redis/redis-connection";
 
 export const crawlQueue = envConfig.redis.enabled
   ? new Queue("crawl-jobs", {
-      connection: {
-        host: envConfig.redis.host,
-        port: envConfig.redis.port,
+      connection: getBullMQConnection({
         enableOfflineQueue: false,
         lazyConnect: true,
-      },
+      }),
       defaultJobOptions: {
         attempts: 3,
         backoff: { type: "exponential", delay: 10000 },

src/queues/crawl.worker.ts

 import "dotenv/config";
 import { Worker } from "bullmq";
 import { envConfig } from "../config/env.config";
+import { getBullMQConnection } from "../common/redis/redis-connection";
 import { processCrawlJob, withTimeout } from "./crawl.worker.processor";
 
-if (!envConfig.redis.enabled) {
-  console.log(
-    "[Worker] REDIS_ENABLED is not set to true. Worker will not start.",
-  );
-  process.exit(0);
-}
+export let crawlWorker: Worker | null = null;
 
-const worker = new Worker(
+if (envConfig.redis.enabled) {
+  crawlWorker = new Worker(
     "crawl-jobs",
     (job) =>
       withTimeout(
@@ -19,28 +16,34 @@ const worker = new Worker(
         job.data.jobId,
       ),
     {
-    connection: {
-      host: envConfig.redis.host,
-      port: envConfig.redis.port,
+      connection: getBullMQConnection({
         maxRetriesPerRequest: null,
-    },
+      }),
       concurrency: envConfig.worker.concurrency,
       maxStalledCount: envConfig.worker.maxStalledCount,
     },
-);
+  );
 
-worker.on("failed", (job, err) => {
+  crawlWorker.on("failed", (job, err) => {
     console.error(`[Worker] Job ${job?.id} failed: ${err.message}`);
-});
+  });
+
+  console.log("[Worker] Crawl worker started");
+} else {
+  console.log(
+    "[Worker] REDIS_ENABLED is not set to true. Worker will not start.",
+  );
+}
 
 async function gracefulShutdown(signal: string) {
+  if (crawlWorker) {
     console.log(`[Worker] Received ${signal}, closing worker gracefully...`);
-  await worker.close();
+    await crawlWorker.close();
     console.log("[Worker] Worker closed");
+  }
   process.exit(0);
 }
 
 process.on("SIGTERM", async () => gracefulShutdown("SIGTERM"));
 process.on("SIGINT", async () => gracefulShutdown("SIGINT"));
 
-console.log("[Worker] Crawl worker started");

src/queues/cron.queue.ts

 import { Queue } from "bullmq";
 import { envConfig } from "../config/env.config";
+import { getBullMQConnection } from "../common/redis/redis-connection";
 import {
   CRON_QUEUE_NAME,
   CronJobName,
@@ -13,12 +14,10 @@ export class CronQueueService {
   constructor() {
     if (envConfig.redis.enabled) {
       this.queue = new Queue(CRON_QUEUE_NAME, {
-        connection: {
-          host: envConfig.redis.host,
-          port: envConfig.redis.port,
+        connection: getBullMQConnection({
           enableOfflineQueue: false,
           lazyConnect: true,
-        },
+        }),
         defaultJobOptions: {
           attempts: 3,
           backoff: { type: "exponential", delay: 5000 },

src/queues/cron.worker.ts

 import "dotenv/config";
 import { Worker } from "bullmq";
 import { envConfig } from "../config/env.config";
+import { getBullMQConnection } from "../common/redis/redis-connection";
 import { cronService } from "../modules/cron/cron.service";
 import {
   CRON_QUEUE_NAME,
@@ -55,11 +56,9 @@ export const cronWorker = new Worker(
     }
   },
   {
-    connection: {
-      host: envConfig.redis.host,
-      port: envConfig.redis.port,
+    connection: getBullMQConnection({
       maxRetriesPerRequest: null,
-    },
+    }),
     concurrency: 2,
   },
 );

src/queues/webhook.queue.ts

 import { Queue } from "bullmq";
 import { envConfig } from "../config/env.config";
+import { getBullMQConnection } from "../common/redis/redis-connection";
 
 export const webhookQueue = envConfig.redis.enabled
   ? new Queue(envConfig.webhook.queueName, {
-      connection: {
-        host: envConfig.redis.host,
-        port: envConfig.redis.port,
+      connection: getBullMQConnection({
         enableOfflineQueue: false,
         lazyConnect: true,
-      },
+      }),
       defaultJobOptions: {
         attempts: 3,
         backoff: { type: "exponential", delay: 5000 },

src/queues/webhook.worker.ts

 import "dotenv/config";
 import { Worker } from "bullmq";
 import { envConfig } from "../config/env.config";
+import { getBullMQConnection } from "../common/redis/redis-connection";
 import { WebhookDeliveryService } from "../modules/webhooks/webhook-delivery.service";
 
 import { getErrorMessage } from "../common/helpers/error-mapping.helper";
@@ -49,11 +50,9 @@ export const webhookWorker = new Worker(
     }
   },
   {
-    connection: {
-      host: envConfig.redis.host,
-      port: envConfig.redis.port,
+    connection: getBullMQConnection({
       maxRetriesPerRequest: null,
-    },
+    }),
     concurrency: 5,
   },
 );

src/server.ts

@@ -70,6 +70,11 @@ async function bootstrap() {
 
     await import("./queues/cron.worker");
     console.log("[Server] Cron worker initialized in background.");
+
+    if (process.env.START_CRAWL_WORKER !== "false") {
+      await import("./queues/crawl.worker");
+      console.log("[Server] Crawl worker initialized in background.");
+    }
   }
 
   app.listen(envConfig.port, () => {