- Tasks Email Queue

Signed-off-by: Dave Richer <dave@imexsystems.ca>
This commit is contained in:
Dave Richer
2024-04-17 19:49:49 -04:00
parent 9f92347936
commit b8c34762ed
5 changed files with 110 additions and 26 deletions

21
package-lock.json generated
View File

@@ -16,6 +16,7 @@
"@opensearch-project/opensearch": "^2.5.0", "@opensearch-project/opensearch": "^2.5.0",
"aws4": "^1.12.0", "aws4": "^1.12.0",
"axios": "^1.6.5", "axios": "^1.6.5",
"better-queue": "^3.8.12",
"bluebird": "^3.7.2", "bluebird": "^3.7.2",
"body-parser": "^1.20.2", "body-parser": "^1.20.2",
"cloudinary": "^2.0.2", "cloudinary": "^2.0.2",
@@ -2860,6 +2861,21 @@
"tweetnacl": "^0.14.3" "tweetnacl": "^0.14.3"
} }
}, },
"node_modules/better-queue": {
"version": "3.8.12",
"resolved": "https://registry.npmjs.org/better-queue/-/better-queue-3.8.12.tgz",
"integrity": "sha512-D9KZ+Us+2AyaCz693/9AyjTg0s8hEmkiM/MB3i09cs4MdK1KgTSGJluXRYmOulR69oLZVo2XDFtqsExDt8oiLA==",
"dependencies": {
"better-queue-memory": "^1.0.1",
"node-eta": "^0.9.0",
"uuid": "^9.0.0"
}
},
"node_modules/better-queue-memory": {
"version": "1.0.4",
"resolved": "https://registry.npmjs.org/better-queue-memory/-/better-queue-memory-1.0.4.tgz",
"integrity": "sha512-SWg5wFIShYffEmJpI6LgbL8/3Dqhku7xI1oEiy6FroP9DbcZlG0ZDjxvPdP9t7hTGW40IpIcC6zVoGT1oxjOuA=="
},
"node_modules/bignumber.js": { "node_modules/bignumber.js": {
"version": "9.1.2", "version": "9.1.2",
"resolved": "https://registry.npmjs.org/bignumber.js/-/bignumber.js-9.1.2.tgz", "resolved": "https://registry.npmjs.org/bignumber.js/-/bignumber.js-9.1.2.tgz",
@@ -5493,6 +5509,11 @@
"node": ">= 0.4.0" "node": ">= 0.4.0"
} }
}, },
"node_modules/node-eta": {
"version": "0.9.0",
"resolved": "https://registry.npmjs.org/node-eta/-/node-eta-0.9.0.tgz",
"integrity": "sha512-mTCTZk29tmX1OGfVkPt63H3c3VqXrI2Kvua98S7iUIB/Gbp0MNw05YtUomxQIxnnKMyRIIuY9izPcFixzhSBrA=="
},
"node_modules/node-fetch": { "node_modules/node-fetch": {
"version": "2.7.0", "version": "2.7.0",
"resolved": "https://registry.npmjs.org/node-fetch/-/node-fetch-2.7.0.tgz", "resolved": "https://registry.npmjs.org/node-fetch/-/node-fetch-2.7.0.tgz",

View File

@@ -26,6 +26,7 @@
"@opensearch-project/opensearch": "^2.5.0", "@opensearch-project/opensearch": "^2.5.0",
"aws4": "^1.12.0", "aws4": "^1.12.0",
"axios": "^1.6.5", "axios": "^1.6.5",
"better-queue": "^3.8.12",
"bluebird": "^3.7.2", "bluebird": "^3.7.2",
"body-parser": "^1.20.2", "body-parser": "^1.20.2",
"cloudinary": "^2.0.2", "cloudinary": "^2.0.2",

View File

@@ -76,7 +76,7 @@ const generateEmailTemplate = (strings) => {
<table> <table>
<tbody> <tbody>
<tr> <tr>
<td><p style="font-size:70%;">${strings?.dateLine || now()}</p></td> <td><p style="font-size:70%; padding-right:10px">${strings?.dateLine || now()}</p></td>
</tr> </tr>
</tbody> </tbody>
</table> </table>

View File

@@ -11,7 +11,7 @@ const client = require("../graphql-client/graphql-client").client;
const queries = require("../graphql-client/queries"); const queries = require("../graphql-client/queries");
const generateEmailTemplate = require("./generateTemplate"); const generateEmailTemplate = require("./generateTemplate");
const moment = require("moment"); const moment = require("moment");
const { UPDATE_TASKS_REMIND_AT_SENT } = require("../graphql-client/queries"); const { taskEmailQueue } = require("./tasksEmailsQueue");
const ses = new aws.SES({ const ses = new aws.SES({
apiVersion: "latest", apiVersion: "latest",
@@ -27,6 +27,9 @@ const transporter = nodemailer.createTransport({
sendingRate: 40 // 40 emails per second. sendingRate: 40 // 40 emails per second.
}); });
// Initialize the Tasks Email Queue
const tasksEmailQueue = taskEmailQueue();
const fromEmails = InstanceManager({ const fromEmails = InstanceManager({
imex: "ImEX Online <noreply@imex.online>", imex: "ImEX Online <noreply@imex.online>",
rome: "Rome Online <noreply@romeonline.io>", rome: "Rome Online <noreply@romeonline.io>",
@@ -76,25 +79,27 @@ const generateTemplateArgs = (title, createdBy, dueDate, taskId) => {
const sendMail = (type, to, subject, html, taskIds, successCallback) => { const sendMail = (type, to, subject, html, taskIds, successCallback) => {
// Push next messages to Nodemailer // Push next messages to Nodemailer
transporter.once("idle", () => { transporter.once("idle", () => {
if (transporter.isIdle()) { // Note: This is commented out because despite being in the documentation, it does not work
transporter.sendMail( // and stackoverflow suggests it is not needed
{ // if (transporter.isIdle()) {
from: fromEmails, transporter.sendMail(
to, {
subject, from: fromEmails,
html to,
}, subject,
(error, info) => { html
if (info) { },
if (typeof successCallback === "function" && taskIds && taskIds.length) { (error, info) => {
successCallback(taskIds); if (info) {
} if (typeof successCallback === "function" && taskIds && taskIds.length) {
} else { successCallback(taskIds);
logger.log(`task-${type}-email-failure`, "error", null, null, error);
} }
} else {
logger.log(`task-${type}-email-failure`, "error", null, null, error);
} }
); }
} );
// }
}); });
}; };
@@ -194,9 +199,8 @@ const tasksRemindEmail = async (req, res) => {
subHeader: `Please sign in to your account to view the Task details.`, subHeader: `Please sign in to your account to view the Task details.`,
body: `<ul> body: `<ul>
${allTasks ${allTasks
.map( .map((task) =>
(task) => `<li><a href="${endPoints}/manage/tasks/alltasks?taskid=${task.id}">${task.title} ${task.due_date ? `- ${formatDate(task.due_date)}` : ""}</a></li>`.trim()
`<li><a href="${endPoints}/manage/tasks/alltasks?taskid=${task.id}">${task.title} - ${formatDate(task.due_date)}</a></li>`
) )
.join("")} .join("")}
</ul>` </ul>`
@@ -206,10 +210,9 @@ const tasksRemindEmail = async (req, res) => {
if (emailData?.subject && emailData?.html) { if (emailData?.subject && emailData?.html) {
// Send Email // Send Email
sendMail("remind", emailData.to, emailData.subject, emailData.html, taskIds, (taskIds) => { sendMail("remind", emailData.to, emailData.subject, emailData.html, taskIds, (taskIds) => {
client.request(UPDATE_TASKS_REMIND_AT_SENT, { for (const taskId of taskIds) {
taskIds, taskEmailQueue.push(taskId);
now: moment().toISOString() }
});
}); });
} }
}); });

View File

@@ -0,0 +1,59 @@
const path = require("path");
require("dotenv").config({
path: path.resolve(process.cwd(), `.env.${process.env.NODE_ENV || "development"}`)
});
const Queue = require("better-queue");
const moment = require("moment/moment");
const { client } = require("../graphql-client/graphql-client");
const { UPDATE_TASKS_REMIND_AT_SENT } = require("../graphql-client/queries");
const logger = require("../utils/logger");
const taskEmailQueue = () =>
new Queue(
(taskIds, cb) => {
console.log("Processing reminds for taskIds: ", taskIds.join(", "));
// Set the remind_at_sent to the current time.
const now = moment().toISOString();
client
.request(UPDATE_TASKS_REMIND_AT_SENT, { taskIds, now })
.then((taskResponse) => {
logger.log("task-remind-email-queue", "info", null, null, taskResponse);
cb(null, taskResponse);
})
.catch((err) => {
logger.log("task-remind-email-queue", "error", null, null, err);
cb(err);
});
},
{
batchSize: 50,
batchDelay: 5000,
// The lower this is, the more likely we are to hit the rate limit.
batchDelayTimeout: 1000
}
);
// Handle process signals
process.on("SIGINT", () => {
taskEmailQueue.destroy();
process.exit(0);
});
process.on("SIGTERM", () => {
taskEmailQueue.destroy();
process.exit(0);
});
process.on("uncaughtException", (err) => {
console.error("Unhandled Exception:", err);
taskEmailQueue.destroy();
// Perform cleanup or log before exit
process.exit(1); // Exit with a 'failure' code
});
process.on("unhandledRejection", (reason, promise) => {
console.error("Unhandled Rejection:", reason);
taskEmailQueue.destroy();
process.exit(1);
});
module.exports = { taskEmailQueue };