Upgrade Next.js to 16.2.9

apps/queue/package.json

@@ -27,7 +27,7 @@
         "bullmq": "^4.14.0",
         "express": "^4.18.2",
         "jsdom": "^26.1.0",
-        "liquidjs": "^10.11.1",
+        "liquidjs": "10.26.0",
         "mongodb": "^6.21.0",
         "mongoose": "^8.22.1",
         "nodemailer": "^6.9.2",

apps/queue/src/bullmq.ts

@@ -0,0 +1,85 @@
+import type { Job, Worker, WorkerOptions } from "bullmq";
+import redis from "./redis";
+import { logger } from "./logger";
+import {
+    captureError,
+    captureEvent,
+    getDomainId,
+} from "./observability/posthog";
+
+export const workerOptions: WorkerOptions = {
+    connection: redis,
+    lockDuration: 5 * 60 * 1000,
+    stalledInterval: 30 * 1000,
+    maxStalledCount: 2,
+};
+
+export function registerWorkerEvents(
+    worker: Pick<Worker, "on">,
+    queueName: string,
+) {
+    worker.on("ready", () => {
+        captureEvent({
+            event: "queue_worker_ready",
+            source: `worker.${queueName}.ready`,
+            domainId: getDomainId(),
+            properties: {
+                queue_name: queueName,
+            },
+        });
+    });
+
+    worker.on("closed", () => {
+        captureEvent({
+            event: "queue_worker_closed",
+            source: `worker.${queueName}.closed`,
+            domainId: getDomainId(),
+            properties: {
+                queue_name: queueName,
+            },
+        });
+    });
+
+    worker.on("failed", (job: Job | undefined, err: Error) => {
+        const domainId = getDomainId(job?.data?.domainId || job?.data?.domain);
+        logger.error(err);
+        captureError({
+            error: err,
+            source: `worker.${queueName}.failed`,
+            domainId,
+            context: {
+                queue_name: queueName,
+                job_id: job?.id ? String(job.id) : undefined,
+                job_name: job?.name,
+                failed_reason: job?.failedReason,
+                attempts_made: job?.attemptsMade,
+            },
+        });
+    });
+
+    worker.on("stalled", (jobId: string) => {
+        const err = new Error(`BullMQ job stalled in ${queueName}: ${jobId}`);
+        logger.error(err);
+        captureError({
+            error: err,
+            source: `worker.${queueName}.stalled`,
+            domainId: getDomainId(),
+            context: {
+                queue_name: queueName,
+                job_id: jobId,
+            },
+        });
+    });
+
+    worker.on("error", (err: Error) => {
+        logger.error(err);
+        captureError({
+            error: err,
+            source: `worker.${queueName}.error`,
+            domainId: getDomainId(),
+            context: {
+                queue_name: queueName,
+            },
+        });
+    });
+}

apps/queue/src/domain/process-ongoing-sequences/index.ts

@@ -1,18 +1,18 @@
 import { logger } from "../../logger";
 import { getDueOngoingSequences } from "../queries";
 import { Worker } from "bullmq";
-import redis from "../../redis";
 import sequenceQueue from "../sequence-queue";
 import { processOngoingSequence } from "./process-ongoing-sequence";
 import { captureError, getDomainId } from "../../observability/posthog";
+import { registerWorkerEvents, workerOptions } from "../../bullmq";
 
 interface SequenceWorkerData {
     ongoingSequenceId: string;
     domainId?: string;
 }
 
-if (process.env.NODE_ENV !== "test") {
-    new Worker(
+export function startSequenceWorker() {
+    const worker = new Worker(
         "sequence",
         async (job) => {
             const payload = job.data as SequenceWorkerData | string;
@@ -38,10 +38,15 @@ if (process.env.NODE_ENV !== "test") {
                         ongoing_sequence_id: String(ongoingSequenceId),
                     },
                 });
+                throw err;
             }
         },
-        { connection: redis },
+        workerOptions,
     );
+
+    registerWorkerEvents(worker, "sequence");
+
+    return worker;
 }
 
 export async function processOngoingSequences(): Promise<void> {

apps/queue/src/domain/worker.ts

@@ -1,39 +1,44 @@
 import { Worker } from "bullmq";
-import redis from "../redis";
 import { logger } from "../logger";
 import { captureError, getDomainId } from "../observability/posthog";
 import { sendMail } from "../mail";
+import { registerWorkerEvents, workerOptions } from "../bullmq";
 
-const worker = new Worker(
-    "mail",
-    async (job) => {
-        const { to, from, subject, body, headers, domainId } = job.data;
+export function startMailWorker() {
+    const worker = new Worker(
+        "mail",
+        async (job) => {
+            const { to, from, subject, body, headers, domainId } = job.data;
 
-        try {
-            await sendMail({
-                from,
-                to,
-                subject,
-                html: body,
-                headers,
-            });
-        } catch (err: any) {
-            logger.error(err);
-            captureError({
-                error: err,
-                source: "worker.mail",
-                domainId: getDomainId(domainId),
-                context: {
-                    queue_name: "mail",
-                    job_id: String(job.id),
-                    error_code: err?.code,
-                    response_code: err?.responseCode,
-                    command: err?.command,
-                },
-            });
-        }
-    },
-    { connection: redis },
-);
+            try {
+                await sendMail({
+                    from,
+                    to,
+                    subject,
+                    html: body,
+                    headers,
+                });
+            } catch (err: any) {
+                logger.error(err);
+                captureError({
+                    error: err,
+                    source: "worker.mail",
+                    domainId: getDomainId(domainId),
+                    context: {
+                        queue_name: "mail",
+                        job_id: String(job.id),
+                        error_code: err?.code,
+                        response_code: err?.responseCode,
+                        command: err?.command,
+                    },
+                });
+                throw err;
+            }
+        },
+        workerOptions,
+    );
 
-export default worker;
+    registerWorkerEvents(worker, "mail");
+
+    return worker;
+}

apps/queue/src/index.ts

@@ -2,13 +2,12 @@ import express from "express";
 import jobRoutes from "./job/routes";
 import sseRoutes from "./sse/routes";
 
-// start workers
-import "./domain/worker";
-import "./notifications/worker/notification";
-import "./notifications/worker/dispatch-notification";
-
 // start loops
 import { startEmailAutomation } from "./start-email-automation";
+import { startSequenceWorker } from "./domain/process-ongoing-sequences";
+import { startMailWorker } from "./domain/worker";
+import { startNotificationWorker } from "./notifications/worker/notification";
+import { startDispatchNotificationWorker } from "./notifications/worker/dispatch-notification";
 import { verifyJWTMiddleware } from "./middlewares/verify-jwt";
 import {
     captureError,
@@ -18,6 +17,37 @@ import {
 import { initPosthogLogs } from "./observability/logs";
 import { logger } from "./logger";
 
+const workers =
+    process.env.NODE_ENV === "test"
+        ? []
+        : [
+              startMailWorker(),
+              startNotificationWorker(),
+              startDispatchNotificationWorker(),
+              startSequenceWorker(),
+          ];
+
+async function closeWorkers() {
+    await Promise.allSettled(workers.map((worker) => worker.close()));
+}
+
+for (const signal of ["SIGINT", "SIGTERM"] as const) {
+    process.once(signal, () => {
+        closeWorkers()
+            .catch((err) => {
+                logger.error(err);
+                captureError({
+                    error: err,
+                    source: "service.shutdown",
+                    domainId: getDomainId(),
+                });
+            })
+            .finally(() => {
+                process.kill(process.pid, signal);
+            });
+    });
+}
+
 const app = express();
 app.use(express.json());
 

apps/queue/src/notifications/worker/dispatch-notification.ts

@@ -4,7 +4,6 @@ import {
     Constants,
     NotificationChannel as NotificationChannelType,
 } from "@courselit/common-models";
-import redis from "../../redis";
 import { logger } from "../../logger";
 import UserModel from "../../domain/model/user";
 import DomainModel from "../../domain/model/domain";
@@ -18,6 +17,7 @@ import {
     NotificationChannel,
 } from "../services/channels/types";
 import { captureError, getDomainId } from "../../observability/posthog";
+import { registerWorkerEvents, workerOptions } from "../../bullmq";
 
 interface DispatchNotificationJob {
     domain: string | mongoose.Types.ObjectId;
@@ -33,31 +33,35 @@ const channelRegistry: Record<string, NotificationChannel> = {
     [Constants.NotificationChannel.EMAIL]: new EmailChannel(),
 };
 
-const worker = new Worker(
-    "dispatch-notification",
-    async (job) => {
-        const payload = job.data as DispatchNotificationJob;
-        try {
-            await processDispatchNotificationJob(payload);
-        } catch (err: any) {
-            logger.error(err);
-            captureError({
-                error: err,
-                source: "worker.dispatch_notification",
-                domainId: getDomainId(payload?.domain),
-                context: {
-                    queue_name: "dispatch-notification",
-                    job_id: String(job.id),
-                    activity_type: payload?.activityType,
-                },
-            });
-            throw err;
-        }
-    },
-    { connection: redis },
-);
+export function startDispatchNotificationWorker() {
+    const worker = new Worker(
+        "dispatch-notification",
+        async (job) => {
+            const payload = job.data as DispatchNotificationJob;
+            try {
+                await processDispatchNotificationJob(payload);
+            } catch (err: any) {
+                logger.error(err);
+                captureError({
+                    error: err,
+                    source: "worker.dispatch_notification",
+                    domainId: getDomainId(payload?.domain),
+                    context: {
+                        queue_name: "dispatch-notification",
+                        job_id: String(job.id),
+                        activity_type: payload?.activityType,
+                    },
+                });
+                throw err;
+            }
+        },
+        workerOptions,
+    );
 
-export default worker;
+    registerWorkerEvents(worker, "dispatch-notification");
+
+    return worker;
+}
 
 async function processDispatchNotificationJob(job: DispatchNotificationJob) {
     if (!Object.values(Constants.ActivityType).includes(job.activityType)) {