|
Please check the job for more details.
`
});
await sendTaskEmail({
to: recipient,
subject,
type: "html",
html: emailBody
});
devDebugLogger(
`Sent consolidated email to ${recipient} for jobId ${jobId} with ${messages.length} updates`
);
await pubClient.del(userKey);
await pubClient.del(detailsKey);
}
}
await pubClient.del(recipientsSet);
await pubClient.del(`email:${devKey}:consolidate:${jobId}`);
} catch (err) {
logger.log(`email-queue-consolidation-error`, "ERROR", "notifications", "api", {
message: err?.message,
stack: err?.stack
});
throw err;
} finally {
await pubClient.del(lockKey);
}
} else {
devDebugLogger(`Skipped email consolidation for jobId ${jobId} - lock held by another worker`);
}
},
{
prefix,
connection: pubClient,
concurrency: 1,
limiter: { max: 1, duration: RATE_LIMITER_DURATION }
}
);
// Event handlers for workers
emailAddWorker.on("completed", (job) => devDebugLogger(`Email add job ${job.id} completed`));
emailConsolidateWorker.on("completed", (job) => devDebugLogger(`Email consolidate job ${job.id} completed`));
emailAddWorker.on("failed", (job, err) =>
logger.log(`add-email-queue-failed`, "ERROR", "notifications", "api", {
message: err?.message,
stack: err?.stack
})
);
emailConsolidateWorker.on("failed", (job, err) =>
logger.log(`email-consolidation-job-failed`, "ERROR", "notifications", "api", {
message: err?.message,
stack: err?.stack
})
);
// Register cleanup task instead of direct process listeners
const shutdown = async () => {
devDebugLogger("Closing email queue workers...");
await Promise.all([emailAddWorker.close(), emailConsolidateWorker.close()]);
devDebugLogger("Email queue workers closed");
};
registerCleanupTask(shutdown);
}
return emailAddQueue;
};
/**
* Retrieves the initialized `emailAddQueue` instance.
*
* @returns {Queue} The `emailAddQueue` instance for adding notifications.
* @throws {Error} If `emailAddQueue` is not initialized.
*/
const getQueue = () => {
if (!emailAddQueue) {
throw new Error("Email add queue not initialized. Ensure loadEmailQueue is called during bootstrap.");
}
return emailAddQueue;
};
/**
* Dispatches email notifications to the `emailAddQueue` for processing.
*
* @param {Object} options - Options for dispatching notifications.
* @param {Array} options.emailsToDispatch - Array of email notification objects.
* @param {Object} options.logger - Logger instance for logging dispatch events.
* @returns {Promise} Resolves when all notifications are added to the queue.
*/
// eslint-disable-next-line no-unused-vars
const dispatchEmailsToQueue = async ({ emailsToDispatch, logger }) => {
const emailAddQueue = getQueue();
for (const email of emailsToDispatch) {
const { jobId, jobRoNumber, bodyShopName, bodyShopTimezone, body, recipients } = email;
if (!jobId || !jobRoNumber || !bodyShopName || !body || !recipients.length) {
devDebugLogger(
`Skipping email dispatch for jobId ${jobId} due to missing data: ` +
`jobRoNumber=${jobRoNumber || "N/A"}, bodyShopName=${bodyShopName}, body=${body}, recipients=${recipients.length}`
);
continue;
}
await emailAddQueue.add(
"add-email-notification",
{ jobId, jobRoNumber, bodyShopName, bodyShopTimezone, body, recipients },
{ jobId: `${jobId}:${Date.now()}` }
);
devDebugLogger(`Added email notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
}
};
module.exports = { loadEmailQueue, getQueue, dispatchEmailsToQueue };
|