admin/bodyshop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/IO-3096-GlobalNotifications - Checkpoint, App Queue

Browse Source
This commit is contained in:
Dave Richer
2025-02-18 17:37:24 -05:00
parent 00005c881e
commit 2a81517104
4 changed files with 190 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

263
server/notifications/queues/appQueue.js
View File

@@ -1,103 +1,57 @@
const { Queue, Worker } = require("bullmq");
let appQueue;
let addQueue;
let consolidateQueue;
const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
if (!appQueue) {
logger.logger.info("Initializing Notifications App Queue");
appQueue = new Queue("notificationsApp", {
if (!addQueue || !consolidateQueue) {
logger.logger.info("Initializing Notifications Queues");
addQueue = new Queue("notificationsAdd", {
connection: pubClient,
prefix: "{BULLMQ}"
prefix: "{BULLMQ}",
defaultJobOptions: { removeOnComplete: true, removeOnFail: true }
});
const worker = new Worker(
"notificationsApp",
consolidateQueue = new Queue("notificationsConsolidate", {
connection: pubClient,
prefix: "{BULLMQ}",
defaultJobOptions: { removeOnComplete: true, removeOnFail: true }
});
const addWorker = new Worker(
"notificationsAdd",
async (job) => {
const { jobId, bodyShopId, key, variables, recipients } = job.data;
logger.logger.info(`Processing app job ${job.id} for jobId ${jobId}`);
const { jobId, key, variables, recipients } = job.data;
logger.logger.info(`Adding notifications for jobId ${jobId}`);
const redisKey = `app:notifications:${jobId}`;
const lastSentKey = `${redisKey}:lastSent`;
const lockKey = `lock:send-notifications:${jobId}`;
const recurringFlagKey = `app:recurring:${jobId}`;
const redisKeyPrefix = `app:notifications:${jobId}`;
const notification = { key, variables, timestamp: Date.now() };
if (job.name === "add-notification") {
const notification = { key, variables, timestamp: Date.now() };
for (const recipient of recipients) {
const { user } = recipient;
const userKey = `${redisKey}:${user}`;
const existingNotifications = await pubClient.get(userKey);
const notifications = existingNotifications ? JSON.parse(existingNotifications) : [];
notifications.push(notification);
await pubClient.set(userKey, JSON.stringify(notifications), "EX", 40);
}
} else if (job.name === "send-notifications") {
let hasNewNotifications = false;
const lastSent = parseInt((await pubClient.get(lastSentKey)) || "0", 10);
for (const recipient of recipients) {
const { user } = recipient;
const userKey = `${redisKeyPrefix}:${user}`;
const existingNotifications = await pubClient.get(userKey);
const notifications = existingNotifications ? JSON.parse(existingNotifications) : [];
notifications.push(notification);
await pubClient.set(userKey, JSON.stringify(notifications), "EX", 40);
logger.logger.debug(`Stored notification for ${user} under ${userKey}: ${JSON.stringify(notifications)}`);
}
for (const recipient of recipients) {
const { user, bodyShopId: recipientBodyShopId } = recipient;
const userKey = `${redisKey}:${user}`;
const notifications = await pubClient.get(userKey);
if (notifications) {
const parsedNotifications = JSON.parse(notifications);
const newNotifications = parsedNotifications.filter((n) => n.timestamp > lastSent);
if (newNotifications.length > 0) {
hasNewNotifications = true;
const socketIds = await redisHelpers.getUserSocketMapping(user);
if (socketIds && socketIds[bodyShopId]?.socketIds) {
socketIds[bodyShopId].socketIds.forEach((socketId) => {
ioRedis.to(socketId).emit("notification", {
jobId,
bodyShopId: recipientBodyShopId,
notifications: newNotifications
});
});
logger.logger.info(`Sent ${newNotifications.length} new notifications to ${user} for jobId ${jobId}`);
} else {
logger.logger.warn(`No socket IDs found for ${user} in bodyShopId ${bodyShopId}`);
}
await pubClient.del(userKey);
}
}
}
const consolidateKey = `app:consolidate:${jobId}`;
const flagSet = await pubClient.setnx(consolidateKey, "pending");
logger.logger.debug(`Consolidation flag set for jobId ${jobId}: ${flagSet}`);
if (hasNewNotifications) {
await pubClient.set(lastSentKey, Date.now(), "EX", 300);
} else {
const activeJobs = await appQueue.getActive();
const hasPendingAdds = activeJobs.some((j) => j.name === "add-notification" && j.data.jobId === jobId);
if (!hasPendingAdds) {
const lockAcquired = await pubClient.set(lockKey, "locked", "NX", "EX", 10);
if (lockAcquired) {
const recurringJobKey = `send-notifications:${jobId}`;
const repeatableJobs = await appQueue.getRepeatableJobs();
const jobExists = repeatableJobs.some((j) => j.key === recurringJobKey);
if (jobExists) {
await appQueue.removeRepeatableByKey(recurringJobKey);
// Drain all remaining send-notifications jobs for this jobId
await appQueue.drain(false); // false to not force removal of active jobs
logger.logger.info(
`Successfully removed recurring send-notifications job and drained queue for jobId ${jobId} with key ${recurringJobKey}`
);
} else {
logger.logger.info(
`No recurring send-notifications job found for jobId ${jobId} with key ${recurringJobKey} - processing leftover scheduled instance`
);
}
await pubClient.del(lockKey);
await pubClient.del(recurringFlagKey);
} else {
logger.logger.info(
`Skipped removal of send-notifications for jobId ${jobId} - lock held by another worker`
);
}
} else {
logger.logger.info(
`Skipping removal of send-notifications for jobId ${jobId} - pending add-notification jobs exist`
);
}
}
if (flagSet) {
await consolidateQueue.add(
"consolidate-notifications",
{ jobId, recipients },
{ jobId: `consolidate:${jobId}`, delay: 5000 }
);
logger.logger.info(`Scheduled consolidation for jobId ${jobId}`);
await pubClient.expire(consolidateKey, 300);
} else {
logger.logger.debug(`Consolidation already scheduled for jobId ${jobId}`);
}
},
{
@@ -107,61 +61,112 @@ const loadAppQueue = async ({ pubClient, logger, redisHelpers, ioRedis }) => {
}
);
worker.on("completed", async (job) => {
if (job.name === "add-notification") {
const { jobId } = job.data;
const recurringJobKey = `send-notifications:${jobId}`;
const recurringFlagKey = `app:recurring:${jobId}`;
const flagSet = await pubClient.setnx(recurringFlagKey, "active");
if (flagSet) {
const existingJobs = await appQueue.getRepeatableJobs();
if (!existingJobs.some((j) => j.key === recurringJobKey)) {
await appQueue.add(
"send-notifications",
{ jobId, bodyShopId: job.data.bodyShopId, recipients: job.data.recipients },
{
repeat: {
every: 30 * 1000,
limit: 10
},
jobId: recurringJobKey
const consolidateWorker = new Worker(
"notificationsConsolidate",
async (job) => {
const { jobId, recipients } = job.data;
logger.logger.info(`Consolidating notifications for jobId ${jobId}`);
const redisKeyPrefix = `app:notifications:${jobId}`;
const lockKey = `lock:consolidate:${jobId}`;
const lockAcquired = await pubClient.set(lockKey, "locked", "NX", "EX", 10);
logger.logger.debug(`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}`);
for (const user of uniqueUsers) {
const userKey = `${redisKeyPrefix}:${user}`;
const notifications = await pubClient.get(userKey);
logger.logger.debug(`Retrieved notifications for ${user}: ${notifications}`);
if (notifications) {
const parsedNotifications = JSON.parse(notifications);
const userRecipients = recipients.filter((r) => r.user === user);
for (const { bodyShopId } of userRecipients) {
allNotifications[user] = allNotifications[user] || {};
allNotifications[user][bodyShopId] = parsedNotifications;
}
await pubClient.del(userKey);
logger.logger.debug(`Deleted Redis key ${userKey}`);
} else {
logger.logger.warn(`No notifications found for ${user} under ${userKey}`);
}
);
logger.logger.info(`Scheduled 30s notification send for jobId ${jobId} with key ${recurringJobKey}`);
await pubClient.expire(recurringFlagKey, 300);
}
logger.logger.debug(`Consolidated notifications: ${JSON.stringify(allNotifications)}`);
for (const [user, bodyShopData] of Object.entries(allNotifications)) {
const userMapping = await redisHelpers.getUserSocketMapping(user);
logger.logger.debug(`User socket mapping for ${user}: ${JSON.stringify(userMapping)}`);
for (const [bodyShopId, notifications] of Object.entries(bodyShopData)) {
if (userMapping && userMapping[bodyShopId]?.socketIds) {
userMapping[bodyShopId].socketIds.forEach((socketId) => {
logger.logger.debug(
`Emitting to socket ${socketId}: ${JSON.stringify({ jobId, bodyShopId, notifications })}`
);
ioRedis.to(socketId).emit("notification", {
jobId,
bodyShopId,
notifications
});
});
logger.logger.info(
`Sent ${notifications.length} consolidated notifications to ${user} for jobId ${jobId}`
);
} else {
logger.logger.warn(`No socket IDs found for ${user} in bodyShopId ${bodyShopId}`);
}
}
}
await pubClient.del(`app:consolidate:${jobId}`);
} catch (err) {
logger.logger.error(`Consolidation error for jobId ${jobId}: ${err.message}`, { error: err });
throw err; // Re-throw to trigger failed event
} finally {
await pubClient.del(lockKey);
}
} else {
logger.logger.info(`Skipped consolidation for jobId ${jobId} - lock held by another worker`);
}
},
{
connection: pubClient,
prefix: "{BULLMQ}",
concurrency: 1,
limiter: { max: 1, duration: 5000 }
}
logger.logger.info(`Job ${job.id} completed`);
});
);
worker.on("failed", (job, err) => {
logger.logger.error(`Job ${job.id} failed: ${err.message}`, { error: err });
});
worker.on("error", (err) => {
logger.logger.error("Worker error:", { error: err });
});
addWorker.on("completed", (job) => logger.logger.info(`Add job ${job.id} completed`));
consolidateWorker.on("completed", (job) => logger.logger.info(`Consolidate job ${job.id} completed`));
addWorker.on("failed", (job, err) =>
logger.logger.error(`Add job ${job.id} failed: ${err.message}`, { error: err })
);
consolidateWorker.on("failed", (job, err) =>
logger.logger.error(`Consolidate job ${job.id} failed: ${err.message}`, { error: err })
);
const shutdown = async () => {
if (worker) {
logger.logger.info("Closing app queue worker...");
await worker.close();
logger.logger.info("App queue worker closed");
}
logger.logger.info("Closing app queue workers...");
await Promise.all([addWorker.close(), consolidateWorker.close()]);
logger.logger.info("App queue workers closed");
};
process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);
}
return appQueue;
return addQueue; // Return the add queue for dispatching
};
const getQueue = () => {
if (!appQueue) {
throw new Error("App queue not initialized. Ensure loadAppQueue is called during bootstrap.");
}
return appQueue;
if (!addQueue) throw new Error("Add queue not initialized. Ensure loadAppQueue is called during bootstrap.");
return addQueue;
};
const dispatchAppsToQueue = async ({ appsToDispatch, logger }) => {
@@ -174,7 +179,7 @@ const dispatchAppsToQueue = async ({ appsToDispatch, logger }) => {
{ jobId, bodyShopId, key, variables, recipients },
{ jobId: `${jobId}:${Date.now()}` }
);
logger.logger.info(`Added app notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
logger.logger.info(`Added notification to queue for jobId ${jobId} with ${recipients.length} recipients`);
}
};