Compare commits

...

29 Commits

Author SHA1 Message Date
Dave Richer
2c0eab9366 IO-3096-GlobalNotifications - Correct time zone from footer in notification email 2025-03-14 11:27:28 -04:00
Patrick Fic
b831d8ca8a IO-3096 Add indexes for notifications. 2025-03-13 15:27:20 -07:00
Dave Richer
69da6bccf7 IO-3096-GlobalNotifications - Adjust splits 2025-03-13 17:37:36 -04:00
Dave Richer
e3d7ebd7d8 IO-3096-GlobalNotifications - Verify status reporter is a function and exists prior to calling it in cleanup task 2025-03-13 14:59:58 -04:00
Dave Richer
5f0b63a192 IO-3096-GlobalNotifications - Add in a function to exclude extra logging from production 2025-03-13 13:56:30 -04:00
Dave Richer
7a5ac739ab Merge branch 'feature/IO-3170-HotFixForRedis' into feature/IO-3096-GlobalNotifications 2025-03-13 11:49:31 -04:00
Dave Richer
e2297be0af IO-3170-HotfixFoRedis 2025-03-13 11:47:21 -04:00
Dave Richer
73c4983342 Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2199)
IO-3166-Global-Notifications-Part-2: Remove unused event handler (hasura),
2025-03-13 15:31:28 +00:00
Dave Richer
5fa7377121 Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2196)
IO-3166-Global-Notifications-Part-2: add additional key prefixes for dev v prod
2025-03-13 01:12:33 +00:00
Dave Richer
d56d1f369c Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2193)
IO-3166-Global-Notifications-Part-2: Make sure BULLMQ prefixes do not collide
2025-03-13 00:01:56 +00:00
Dave Richer
6b047418cc Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2190)
Feature/IO-3166 Global Notifications Part 2
2025-03-12 16:06:34 +00:00
Dave Richer
0d80854196 Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2186)
Feature/IO-3166 Global Notifications Part 2
2025-03-11 19:14:02 +00:00
Dave Richer
212fc4a7cc Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2183)
Feature/IO-3166 Global Notifications Part 2
2025-03-11 17:18:10 +00:00
Dave Richer
1fad3968bb Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2179)
IO-3166-Global-Notifications-Part-2: getAwsClusterFix
2025-03-07 20:59:43 +00:00
Dave Richer
a492909ad7 Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2176)
IO-3170-Enhanced-GetRedisEndpointsFromAWS - Fix to prevent breaking
2025-03-07 20:27:22 +00:00
Dave Richer
6e6cabbd63 Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2172)
IO-3166-Global-Notifications-Part-2 - Improved GetRedisNodesFromAWS
2025-03-07 20:11:02 +00:00
Dave Richer
ffadd31a5f Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2168)
IO-3166-Global-Notifications-Part-2 - Small styling change
2025-03-07 18:51:21 +00:00
Dave Richer
ef22ba3d2c Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2165)
IO-3166-Global-Notifications-Part-2 - Checkpoint
2025-03-07 16:04:27 +00:00
Dave Richer
71dd138f2f Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2162)
Feature/IO-3166 Global Notifications Part 2
2025-03-06 22:43:24 +00:00
Dave Richer
3cbcbb92eb Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2159)
Feature/IO-3166 Global Notifications Part 2
2025-03-06 21:06:15 +00:00
Dave Richer
ef695776cd Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2156)
Feature/IO-3166 Global Notifications Part 2
2025-03-06 18:39:46 +00:00
Dave Richer
9b545d6c8c Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2153)
Feature/IO-3166 Global Notifications Part 2
2025-03-05 22:30:11 +00:00
Dave Richer
b4a3960eac Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2150)
Feature/IO-3166 Global Notifications Part 2
2025-03-05 18:54:42 +00:00
Dave Richer
e40e0bbb8f Merged release/2024-03-14 into feature/IO-3096-GlobalNotifications 2025-03-05 16:45:08 +00:00
Dave Richer
8fdd07827e Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2147)
Feature/IO-3166 Global Notifications Part 2
2025-03-05 16:44:40 +00:00
Dave Richer
ac2bb42124 Merged in feature/IO-3096-GlobalNotifications (pull request #2145)
Feature/IO-3096 GlobalNotifications
2025-03-04 22:55:58 +00:00
Dave Richer
b149f70b6f Merged in feature/IO-3166-Global-Notifications-Part-2 (pull request #2144)
Feature/IO-3166 Global Notifications Part 2
2025-03-04 22:55:26 +00:00
Dave Richer
7bbbf5934a Merged in feature/IO-3096-GlobalNotifications (pull request #2143)
Feature/IO-3096 GlobalNotifications
2025-03-04 16:57:30 +00:00
Dave Richer
03863ce838 Merged in feature/IO-3096-GlobalNotifications (pull request #2141)
Feature/IO-3096 GlobalNotifications
2025-03-04 16:21:54 +00:00
17 changed files with 108 additions and 75 deletions

View File

@@ -48,8 +48,6 @@ export function App({ bodyshop, checkUserSession, currentUser, online, setOnline
const { t } = useTranslation();
const navigate = useNavigate();
const scenarioNotificationsOn = client?.getTreatment("Realtime_Notifications_UI") === "on";
useEffect(() => {
if (!navigator.onLine) {
setOnline(false);
@@ -203,12 +201,7 @@ export function App({ bodyshop, checkUserSession, currentUser, online, setOnline
path="/manage/*"
element={
<ErrorBoundary>
<SocketProvider
bodyshop={bodyshop}
navigate={navigate}
currentUser={currentUser}
scenarioNotificationsOn={scenarioNotificationsOn}
>
<SocketProvider bodyshop={bodyshop} navigate={navigate} currentUser={currentUser}>
<PrivateRoute isAuthorized={currentUser.authorized} />
</SocketProvider>
</ErrorBoundary>
@@ -220,12 +213,7 @@ export function App({ bodyshop, checkUserSession, currentUser, online, setOnline
path="/tech/*"
element={
<ErrorBoundary>
<SocketProvider
bodyshop={bodyshop}
navigate={navigate}
currentUser={currentUser}
scenarioNotificationsOn={scenarioNotificationsOn}
>
<SocketProvider bodyshop={bodyshop} navigate={navigate} currentUser={currentUser}>
<PrivateRoute isAuthorized={currentUser.authorized} />
</SocketProvider>
</ErrorBoundary>

View File

@@ -14,6 +14,7 @@ import {
} from "../../graphql/notifications.queries.js";
import { useMutation } from "@apollo/client";
import { useTranslation } from "react-i18next";
import { useSplitTreatments } from "@splitsoftware/splitio-react";
const SocketContext = createContext(null);
@@ -25,11 +26,10 @@ const INITIAL_NOTIFICATIONS = 10;
* @param bodyshop
* @param navigate
* @param currentUser
* @param scenarioNotificationsOn
* @returns {JSX.Element}
* @constructor
*/
const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNotificationsOn }) => {
const SocketProvider = ({ children, bodyshop, navigate, currentUser }) => {
const socketRef = useRef(null);
const [clientId, setClientId] = useState(null);
const [isConnected, setIsConnected] = useState(false);
@@ -37,6 +37,14 @@ const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNot
const userAssociationId = bodyshop?.associations?.[0]?.id;
const { t } = useTranslation();
const {
treatments: { Realtime_Notifications_UI }
} = useSplitTreatments({
attributes: {},
names: ["Realtime_Notifications_UI"],
splitKey: bodyshop?.imexshopid
});
const [markNotificationRead] = useMutation(MARK_NOTIFICATION_READ, {
update: (cache, { data: { update_notifications } }) => {
const timestamp = new Date().toISOString();
@@ -209,7 +217,7 @@ const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNot
const handleNotification = (data) => {
// Scenario Notifications have been disabled, bail.
if (!scenarioNotificationsOn) {
if (Realtime_Notifications_UI?.treatment !== "on") {
return;
}
@@ -329,7 +337,7 @@ const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNot
const handleSyncNotificationRead = ({ notificationId, timestamp }) => {
// Scenario Notifications have been disabled, bail.
if (!scenarioNotificationsOn) {
if (Realtime_Notifications_UI?.treatment !== "on") {
return;
}
@@ -371,7 +379,7 @@ const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNot
const handleSyncAllNotificationsRead = ({ timestamp }) => {
// Scenario Notifications have been disabled, bail.
if (!scenarioNotificationsOn) {
if (Realtime_Notifications_UI?.treatment !== "on") {
return;
}
@@ -462,7 +470,7 @@ const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNot
markAllNotificationsRead,
navigate,
currentUser,
scenarioNotificationsOn,
Realtime_Notifications_UI,
t
]);
@@ -474,7 +482,7 @@ const SocketProvider = ({ children, bodyshop, navigate, currentUser, scenarioNot
isConnected,
markNotificationRead,
markAllNotificationsRead,
scenarioNotificationsOn
scenarioNotificationsOn: Realtime_Notifications_UI?.treatment === "on"
}}
>
{children}

View File

@@ -0,0 +1 @@
DROP INDEX IF EXISTS "public"."notificiations_idx_jobs";

View File

@@ -0,0 +1,2 @@
CREATE INDEX "notificiations_idx_jobs" on
"public"."notifications" using btree ("jobid");

View File

@@ -0,0 +1 @@
DROP INDEX IF EXISTS "public"."notifications_idx_associations";

View File

@@ -0,0 +1,2 @@
CREATE INDEX "notifications_idx_associations" on
"public"."notifications" using btree ("associationid");

View File

@@ -0,0 +1,3 @@
-- Could not auto-generate a down migration.
-- Please write an appropriate down migration for the SQL below:
-- CREATE INDEX idx_notifications_created_at_not_read ON notifications(created_at desc, read) where read is null;

View File

@@ -0,0 +1 @@
CREATE INDEX idx_notifications_created_at_not_read ON notifications(created_at desc, read) where read is null;

View File

@@ -0,0 +1,3 @@
-- Could not auto-generate a down migration.
-- Please write an appropriate down migration for the SQL below:
-- CREATE INDEX idx_notifications_associations_not_read ON notifications(associationid, read) where read is null;

View File

@@ -0,0 +1 @@
CREATE INDEX idx_notifications_associations_not_read ON notifications(associationid, read) where read is null;

View File

@@ -22,7 +22,7 @@ const cookieParser = require("cookie-parser");
const { Server } = require("socket.io");
const { createAdapter } = require("@socket.io/redis-adapter");
const { instrument } = require("@socket.io/admin-ui");
const { isString, isEmpty } = require("lodash");
const { isString, isEmpty, isFunction } = require("lodash");
const logger = require("./server/utils/logger");
const { applyRedisHelpers } = require("./server/utils/redisHelpers");
@@ -393,7 +393,9 @@ const main = async () => {
const StatusReporter = StartStatusReporter();
registerCleanupTask(async () => {
StatusReporter.end();
if (isFunction(StatusReporter?.end)) {
StatusReporter.end();
}
});
try {

View File

@@ -20,6 +20,11 @@ const defaultFooter = () => {
const now = () => moment().format("MM/DD/YYYY @ hh:mm a");
/**
* Generate the email template
* @param strings
* @returns {string}
*/
const generateEmailTemplate = (strings) => {
return (
`

View File

@@ -2,6 +2,7 @@ const { Queue, Worker } = require("bullmq");
const { INSERT_NOTIFICATIONS_MUTATION } = require("../../graphql-client/queries");
const { registerCleanupTask } = require("../../utils/cleanupManager");
const getBullMQPrefix = require("../../utils/getBullMQPrefix");
const devDebugLogger = require("../../utils/devDebugLogger");
const graphQLClient = require("../../graphql-client/graphql-client").client;
// Base time-related constant in minutes, sourced from environment variable or defaulting to 1
@@ -49,7 +50,7 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
const prefix = getBullMQPrefix();
const devKey = process.env?.NODE_ENV === "production" ? "prod" : "dev";
logger.logger.debug(`Initializing Notifications Queues with prefix: ${prefix}`);
devDebugLogger(`Initializing Notifications Queues with prefix: ${prefix}`);
addQueue = new Queue("notificationsAdd", {
prefix,
@@ -67,7 +68,7 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
"notificationsAdd",
async (job) => {
const { jobId, key, variables, recipients, body, jobRoNumber } = job.data;
logger.logger.debug(`Adding notifications for jobId ${jobId}`);
devDebugLogger(`Adding notifications for jobId ${jobId}`);
const redisKeyPrefix = `app:${devKey}:notifications:${jobId}`;
const notification = { key, variables, body, jobRoNumber, timestamp: Date.now() };
@@ -79,12 +80,12 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
const notifications = existingNotifications ? JSON.parse(existingNotifications) : [];
notifications.push(notification);
await pubClient.set(userKey, JSON.stringify(notifications), "EX", NOTIFICATION_STORAGE_EXPIRATION / 1000);
logger.logger.debug(`Stored notification for ${user} under ${userKey}: ${JSON.stringify(notifications)}`);
devDebugLogger(`Stored notification for ${user} under ${userKey}: ${JSON.stringify(notifications)}`);
}
const consolidateKey = `app:${devKey}:consolidate:${jobId}`;
const flagSet = await pubClient.setnx(consolidateKey, "pending");
logger.logger.debug(`Consolidation flag set for jobId ${jobId}: ${flagSet}`);
devDebugLogger(`Consolidation flag set for jobId ${jobId}: ${flagSet}`);
if (flagSet) {
await consolidateQueue.add(
@@ -97,10 +98,10 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
backoff: LOCK_EXPIRATION
}
);
logger.logger.debug(`Scheduled consolidation for jobId ${jobId}`);
devDebugLogger(`Scheduled consolidation for jobId ${jobId}`);
await pubClient.expire(consolidateKey, CONSOLIDATION_FLAG_EXPIRATION / 1000);
} else {
logger.logger.debug(`Consolidation already scheduled for jobId ${jobId}`);
devDebugLogger(`Consolidation already scheduled for jobId ${jobId}`);
}
},
{
@@ -114,24 +115,24 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
"notificationsConsolidate",
async (job) => {
const { jobId, recipients } = job.data;
logger.logger.debug(`Consolidating notifications for jobId ${jobId}`);
devDebugLogger(`Consolidating notifications for jobId ${jobId}`);
const redisKeyPrefix = `app:${devKey}:notifications:${jobId}`;
const lockKey = `lock:${devKey}:consolidate:${jobId}`;
const lockAcquired = await pubClient.set(lockKey, "locked", "NX", "EX", LOCK_EXPIRATION / 1000);
logger.logger.debug(`Lock acquisition for jobId ${jobId}: ${lockAcquired}`);
devDebugLogger(`Lock acquisition for jobId ${jobId}: ${lockAcquired}`);
if (lockAcquired) {
try {
const allNotifications = {};
const uniqueUsers = [...new Set(recipients.map((r) => r.user))];
logger.logger.debug(`Unique users for jobId ${jobId}: ${uniqueUsers}`);
devDebugLogger(`Unique users for jobId ${jobId}: ${uniqueUsers}`);
for (const user of uniqueUsers) {
const userKey = `${redisKeyPrefix}:${user}`;
const notifications = await pubClient.get(userKey);
logger.logger.debug(`Retrieved notifications for ${user}: ${notifications}`);
devDebugLogger(`Retrieved notifications for ${user}: ${notifications}`);
if (notifications) {
const parsedNotifications = JSON.parse(notifications);
@@ -141,13 +142,13 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
allNotifications[user][bodyShopId] = parsedNotifications;
}
await pubClient.del(userKey);
logger.logger.debug(`Deleted Redis key ${userKey}`);
devDebugLogger(`Deleted Redis key ${userKey}`);
} else {
logger.logger.debug(`No notifications found for ${user} under ${userKey}`);
devDebugLogger(`No notifications found for ${user} under ${userKey}`);
}
}
logger.logger.debug(`Consolidated notifications: ${JSON.stringify(allNotifications)}`);
devDebugLogger(`Consolidated notifications: ${JSON.stringify(allNotifications)}`);
// Insert notifications into the database and collect IDs
const notificationInserts = [];
@@ -174,7 +175,7 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
const insertResponse = await graphQLClient.request(INSERT_NOTIFICATIONS_MUTATION, {
objects: notificationInserts
});
logger.logger.debug(
devDebugLogger(
`Inserted ${insertResponse.insert_notifications.affected_rows} notifications for jobId ${jobId}`
);
@@ -208,11 +209,11 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
associationId
});
});
logger.logger.debug(
devDebugLogger(
`Sent ${notifications.length} consolidated notifications to ${user} for jobId ${jobId} with notificationId ${notificationId}`
);
} else {
logger.logger.debug(`No socket IDs found for ${user} in bodyShopId ${bodyShopId}`);
devDebugLogger(`No socket IDs found for ${user} in bodyShopId ${bodyShopId}`);
}
}
}
@@ -228,7 +229,7 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
await pubClient.del(lockKey);
}
} else {
logger.logger.debug(`Skipped consolidation for jobId ${jobId} - lock held by another worker`);
devDebugLogger(`Skipped consolidation for jobId ${jobId} - lock held by another worker`);
}
},
{
@@ -239,8 +240,9 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
}
);
addWorker.on("completed", (job) => logger.logger.debug(`Add job ${job.id} completed`));
consolidateWorker.on("completed", (job) => logger.logger.debug(`Consolidate job ${job.id} completed`));
addWorker.on("completed", (job) => devDebugLogger(`Add job ${job.id} completed`));
consolidateWorker.on("completed", (job) => devDebugLogger(`Consolidate job ${job.id} completed`));
addWorker.on("failed", (job, err) =>
logger.log(`app-queue-notification-error`, "ERROR", "notifications", "api", {
message: err?.message,
@@ -256,9 +258,9 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
// Register cleanup task instead of direct process listeners
const shutdown = async () => {
logger.logger.debug("Closing app queue workers...");
devDebugLogger("Closing app queue workers...");
await Promise.all([addWorker.close(), consolidateWorker.close()]);
logger.logger.debug("App queue workers closed");
devDebugLogger("App queue workers closed");
};
registerCleanupTask(shutdown);
@@ -288,7 +290,7 @@ const dispatchAppsToQueue = async ({ appsToDispatch, logger }) => {
{ jobId, bodyShopId, key, variables, recipients, body, jobRoNumber },
{ jobId: `${jobId}:${Date.now()}` }
);
logger.logger.debug(`Added notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
devDebugLogger(`Added notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
}
};

View File

@@ -4,6 +4,8 @@ const generateEmailTemplate = require("../../email/generateTemplate");
const { InstanceEndpoints } = require("../../utils/instanceMgr");
const { registerCleanupTask } = require("../../utils/cleanupManager");
const getBullMQPrefix = require("../../utils/getBullMQPrefix");
const devDebugLogger = require("../../utils/devDebugLogger");
const moment = require("moment-timezone");
const EMAIL_CONSOLIDATION_DELAY_IN_MINS = (() => {
const envValue = process.env?.EMAIL_CONSOLIDATION_DELAY_IN_MINS;
@@ -38,7 +40,7 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
const prefix = getBullMQPrefix();
const devKey = process.env?.NODE_ENV === "production" ? "prod" : "dev";
logger.logger.debug(`Initializing Email Notification Queues with prefix: ${prefix}`);
devDebugLogger(`Initializing Email Notification Queues with prefix: ${prefix}`);
// Queue for adding email notifications
emailAddQueue = new Queue("emailAdd", {
@@ -58,8 +60,8 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
emailAddWorker = new Worker(
"emailAdd",
async (job) => {
const { jobId, jobRoNumber, bodyShopName, body, recipients } = job.data;
logger.logger.debug(`Adding email notifications for jobId ${jobId}`);
const { jobId, jobRoNumber, bodyShopName, bodyShopTimezone, body, recipients } = job.data;
devDebugLogger(`Adding email notifications for jobId ${jobId}`);
const redisKeyPrefix = `email:${devKey}:notifications:${jobId}`;
@@ -71,9 +73,10 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
const detailsKey = `email:${devKey}:recipientDetails:${jobId}:${user}`;
await pubClient.hsetnx(detailsKey, "firstName", firstName || "");
await pubClient.hsetnx(detailsKey, "lastName", lastName || "");
await pubClient.hsetnx(detailsKey, "bodyShopTimezone", bodyShopTimezone);
await pubClient.expire(detailsKey, NOTIFICATION_EXPIRATION / 1000);
await pubClient.sadd(`email:${devKey}:recipients:${jobId}`, user);
logger.logger.debug(`Stored message for ${user} under ${userKey}: ${body}`);
devDebugLogger(`Stored message for ${user} under ${userKey}: ${body}`);
}
const consolidateKey = `email:${devKey}:consolidate:${jobId}`;
@@ -81,7 +84,7 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
if (flagSet) {
await emailConsolidateQueue.add(
"consolidate-emails",
{ jobId, jobRoNumber, bodyShopName },
{ jobId, jobRoNumber, bodyShopName, bodyShopTimezone },
{
jobId: `consolidate:${jobId}`,
delay: EMAIL_CONSOLIDATION_DELAY,
@@ -89,10 +92,10 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
backoff: LOCK_EXPIRATION
}
);
logger.logger.debug(`Scheduled email consolidation for jobId ${jobId}`);
devDebugLogger(`Scheduled email consolidation for jobId ${jobId}`);
await pubClient.expire(consolidateKey, CONSOLIDATION_KEY_EXPIRATION / 1000);
} else {
logger.logger.debug(`Email consolidation already scheduled for jobId ${jobId}`);
devDebugLogger(`Email consolidation already scheduled for jobId ${jobId}`);
}
},
{
@@ -107,7 +110,7 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
"emailConsolidate",
async (job) => {
const { jobId, jobRoNumber, bodyShopName } = job.data;
logger.logger.debug(`Consolidating emails for jobId ${jobId}`);
devDebugLogger(`Consolidating emails for jobId ${jobId}`);
const lockKey = `lock:${devKey}:emailConsolidate:${jobId}`;
const lockAcquired = await pubClient.set(lockKey, "locked", "NX", "EX", LOCK_EXPIRATION / 1000);
@@ -124,9 +127,11 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
const firstName = details.firstName || "User";
const multipleUpdateString = messages.length > 1 ? "Updates" : "Update";
const subject = `${multipleUpdateString} for job ${jobRoNumber || "N/A"} at ${bodyShopName}`;
const timezone = moment.tz.zone(details?.bodyShopTimezone) ? details.bodyShopTimezone : "UTC";
const emailBody = generateEmailTemplate({
header: `${multipleUpdateString} for Job ${jobRoNumber || "N/A"}`,
subHeader: `Dear ${firstName},`,
dateLine: moment().tz(timezone).format("MM/DD/YYYY hh:mm a"),
body: `
<p>There have been updates to job ${jobRoNumber || "N/A"} at ${bodyShopName}:</p><br/>
<ul>
@@ -141,7 +146,7 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
type: "html",
html: emailBody
});
logger.logger.debug(
devDebugLogger(
`Sent consolidated email to ${recipient} for jobId ${jobId} with ${messages.length} updates`
);
await pubClient.del(userKey);
@@ -160,7 +165,7 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
await pubClient.del(lockKey);
}
} else {
logger.logger.debug(`Skipped email consolidation for jobId ${jobId} - lock held by another worker`);
devDebugLogger(`Skipped email consolidation for jobId ${jobId} - lock held by another worker`);
}
},
{
@@ -172,8 +177,9 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
);
// Event handlers for workers
emailAddWorker.on("completed", (job) => logger.logger.debug(`Email add job ${job.id} completed`));
emailConsolidateWorker.on("completed", (job) => logger.logger.debug(`Email consolidate job ${job.id} completed`));
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,
@@ -189,9 +195,9 @@ const loadEmailQueue = async ({ pubClient, logger }) => {
// Register cleanup task instead of direct process listeners
const shutdown = async () => {
logger.logger.debug("Closing email queue workers...");
devDebugLogger("Closing email queue workers...");
await Promise.all([emailAddWorker.close(), emailConsolidateWorker.close()]);
logger.logger.debug("Email queue workers closed");
devDebugLogger("Email queue workers closed");
};
registerCleanupTask(shutdown);
}
@@ -224,10 +230,10 @@ const dispatchEmailsToQueue = async ({ emailsToDispatch, logger }) => {
const emailAddQueue = getQueue();
for (const email of emailsToDispatch) {
const { jobId, jobRoNumber, bodyShopName, body, recipients } = email;
const { jobId, jobRoNumber, bodyShopName, bodyShopTimezone, body, recipients } = email;
if (!jobId || !jobRoNumber || !bodyShopName || !body || !recipients.length) {
logger.logger.warn(
devDebugLogger(
`Skipping email dispatch for jobId ${jobId} due to missing data: ` +
`jobRoNumber=${jobRoNumber || "N/A"}, bodyShopName=${bodyShopName}, body=${body}, recipients=${recipients.length}`
);
@@ -236,10 +242,10 @@ const dispatchEmailsToQueue = async ({ emailsToDispatch, logger }) => {
await emailAddQueue.add(
"add-email-notification",
{ jobId, jobRoNumber, bodyShopName, body, recipients },
{ jobId, jobRoNumber, bodyShopName, bodyShopTimezone, body, recipients },
{ jobId: `${jobId}:${Date.now()}` }
);
logger.logger.debug(`Added email notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
devDebugLogger(`Added email notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
}
};

View File

@@ -28,6 +28,7 @@ const buildNotification = (data, key, body, variables = {}) => {
jobId: data.jobId,
jobRoNumber: data.jobRoNumber,
bodyShopName: data.bodyShopName,
bodyShopTimezone: data.bodyShopTimezone,
body,
recipients: []
},

View File

@@ -0,0 +1,10 @@
const logger = require("./logger");
const devDebugLogger = (message, meta) => {
if (process.env?.NODE_ENV === "production") {
return;
}
logger.logger.debug(message, meta);
};
module.exports = devDebugLogger;

View File

@@ -1,4 +1,5 @@
const { GET_BODYSHOP_BY_ID } = require("../graphql-client/queries");
const devDebugLogger = require("./devDebugLogger");
const client = require("../graphql-client/graphql-client").client;
const BODYSHOP_CACHE_TTL = 3600; // 1 hour
@@ -87,7 +88,7 @@ const applyRedisHelpers = ({ pubClient, app, logger }) => {
const addUserSocketMapping = async (email, socketId, bodyshopId) => {
const socketMappingKey = getUserSocketMappingKey(email);
try {
logger.log(`Adding socket ${socketId} to user ${email} for bodyshop ${bodyshopId}`, "debug", "redis");
devDebugLogger(`Adding socket ${socketId} to user ${email} for bodyshop ${bodyshopId}`);
// Save the mapping: socketId -> bodyshopId
await pubClient.hset(socketMappingKey, socketId, bodyshopId);
// Set TTL (24 hours) for the mapping hash
@@ -109,7 +110,7 @@ const applyRedisHelpers = ({ pubClient, app, logger }) => {
const exists = await pubClient.exists(socketMappingKey);
if (exists) {
await pubClient.expire(socketMappingKey, 86400);
logger.log(`Refreshed TTL for ${email} socket mapping`, "debug", "redis");
devDebugLogger(`Refreshed TTL for ${email} socket mapping`);
}
} catch (error) {
logger.log(`Error refreshing TTL for ${email}: ${error}`, "ERROR", "redis");
@@ -126,20 +127,16 @@ const applyRedisHelpers = ({ pubClient, app, logger }) => {
const socketMappingKey = getUserSocketMappingKey(email);
try {
logger.log(`Removing socket ${socketId} mapping for user ${email}`, "DEBUG", "redis");
devDebugLogger(`Removing socket ${socketId} mapping for user ${email}`);
// Look up the bodyshopId associated with this socket
const bodyshopId = await pubClient.hget(socketMappingKey, socketId);
if (!bodyshopId) {
logger.log(`Socket ${socketId} not found for user ${email}`, "DEBUG", "redis");
devDebugLogger(`Socket ${socketId} not found for user ${email}`);
return;
}
// Remove the socket mapping
await pubClient.hdel(socketMappingKey, socketId);
logger.log(
`Removed socket ${socketId} (associated with bodyshop ${bodyshopId}) for user ${email}`,
"DEBUG",
"redis"
);
devDebugLogger(`Removed socket ${socketId} (associated with bodyshop ${bodyshopId}) for user ${email}`);
// Refresh TTL if any socket mappings remain
const remainingSockets = await pubClient.hlen(socketMappingKey);
@@ -227,7 +224,7 @@ const applyRedisHelpers = ({ pubClient, app, logger }) => {
await pubClient.set(key, jsonData);
await pubClient.expire(key, BODYSHOP_CACHE_TTL);
logger.log("bodyshop-cache-miss", "DEBUG", "redis", null, {
devDebugLogger("bodyshop-cache-miss", {
bodyshopId,
action: "Fetched from DB and cached"
});
@@ -254,7 +251,7 @@ const applyRedisHelpers = ({ pubClient, app, logger }) => {
if (!values) {
// Invalidate cache by deleting the key
await pubClient.del(key);
logger.log("bodyshop-cache-invalidate", "DEBUG", "api", "redis", {
devDebugLogger("bodyshop-cache-invalidate", {
bodyshopId,
action: "Cache invalidated"
});
@@ -263,7 +260,7 @@ const applyRedisHelpers = ({ pubClient, app, logger }) => {
const jsonData = JSON.stringify(values);
await pubClient.set(key, jsonData);
await pubClient.expire(key, BODYSHOP_CACHE_TTL);
logger.log("bodyshop-cache-update", "DEBUG", "api", "redis", {
devDebugLogger("bodyshop-cache-update", {
bodyshopId,
action: "Cache updated",
values