From 79abb0fd849b5fe9fd062b4e19c26a8dc26ca1a9 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 14:07:47 -0700 Subject: [PATCH 01/40] rename methods --- server/controllers/queueController.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server/controllers/queueController.js b/server/controllers/queueController.js index 9d87a70f6..be3d764e7 100755 --- a/server/controllers/queueController.js +++ b/server/controllers/queueController.js @@ -23,7 +23,7 @@ class JobQueueController { getJobs = async (req, res, next) => { try { - const jobs = await this.jobQueue.getJobStats(); + const jobs = await this.jobQueue.getJobs(); return res.success({ msg: this.stringService.queueGetMetrics, data: jobs, @@ -60,7 +60,7 @@ class JobQueueController { flushQueue = async (req, res, next) => { try { - const result = await this.jobQueue.flushQueue(); + const result = await this.jobQueue.flushQueues(); return res.success({ msg: this.stringService.jobQueueFlush, data: result, From b471c49cbf415e8e4a25f33536e6bec397f8254a Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 14:08:45 -0700 Subject: [PATCH 02/40] unordered bulk insert --- server/db/mongo/modules/checkModule.js | 2 +- server/db/mongo/modules/distributedCheckModule.js | 3 +-- server/db/mongo/modules/hardwareCheckModule.js | 2 +- server/db/mongo/modules/pageSpeedCheckModule.js | 2 +- 4 files changed, 4 insertions(+), 5 deletions(-) diff --git a/server/db/mongo/modules/checkModule.js b/server/db/mongo/modules/checkModule.js index 2b0eb33ba..2339edbed 100755 --- a/server/db/mongo/modules/checkModule.js +++ b/server/db/mongo/modules/checkModule.js @@ -43,7 +43,7 @@ const createCheck = async (checkData) => { const createChecks = async (checks) => { try { - await Check.insertMany(checks); + await Check.insertMany(checks, { ordered: false }); } catch (error) { error.service = SERVICE_NAME; error.method = "createCheck"; diff --git a/server/db/mongo/modules/distributedCheckModule.js b/server/db/mongo/modules/distributedCheckModule.js index 10fc577b2..7200cba33 100755 --- a/server/db/mongo/modules/distributedCheckModule.js +++ b/server/db/mongo/modules/distributedCheckModule.js @@ -128,9 +128,8 @@ const createDistributedChecks = async (checksData) => { }; }); - // Execute bulk operation await DistributedUptimeCheck.bulkWrite(bulkOps, { - ordered: false, // Allow parallel processing + ordered: false, }); } catch (error) { error.service = SERVICE_NAME; diff --git a/server/db/mongo/modules/hardwareCheckModule.js b/server/db/mongo/modules/hardwareCheckModule.js index 738412c60..868c83958 100755 --- a/server/db/mongo/modules/hardwareCheckModule.js +++ b/server/db/mongo/modules/hardwareCheckModule.js @@ -51,7 +51,7 @@ const createHardwareCheck = async (hardwareCheckData) => { const createHardwareChecks = async (hardwareChecks) => { try { - await HardwareCheck.insertMany(hardwareChecks); + await HardwareCheck.insertMany(hardwareChecks, { ordered: false }); return true; } catch (error) { error.service = SERVICE_NAME; diff --git a/server/db/mongo/modules/pageSpeedCheckModule.js b/server/db/mongo/modules/pageSpeedCheckModule.js index f634fae27..9169a0836 100755 --- a/server/db/mongo/modules/pageSpeedCheckModule.js +++ b/server/db/mongo/modules/pageSpeedCheckModule.js @@ -26,7 +26,7 @@ const createPageSpeedCheck = async (pageSpeedCheckData) => { }; const createPageSpeedChecks = async (pageSpeedChecks) => { try { - await PageSpeedCheck.insertMany(pageSpeedChecks); + await PageSpeedCheck.insertMany(pageSpeedChecks, { ordered: false }); return true; } catch (error) { error.service = SERVICE_NAME; From e6a387b8b23755f5f5716c4f209877df2f01c230 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 14:09:49 -0700 Subject: [PATCH 03/40] refactor job queue --- server/index.js | 36 +- server/service/JobQueue/JobQueue.js | 301 ++++++++ server/service/JobQueue/JobQueueHelper.js | 315 +++++++++ server/service/jobQueue.js | 814 ---------------------- server/service/redisService.js | 96 +-- 5 files changed, 666 insertions(+), 896 deletions(-) create mode 100644 server/service/JobQueue/JobQueue.js create mode 100644 server/service/JobQueue/JobQueueHelper.js delete mode 100755 server/service/jobQueue.js diff --git a/server/index.js b/server/index.js index f299b32e9..96db857dd 100755 --- a/server/index.js +++ b/server/index.js @@ -46,7 +46,8 @@ import DiagnosticRoutes from "./routes/diagnosticRoute.js"; import DiagnosticController from "./controllers/diagnosticController.js"; //JobQueue service and dependencies -import JobQueue from "./service/jobQueue.js"; +import JobQueue from "./service/JobQueue/JobQueue.js"; +import JobQueueHelper from "./service/JobQueue/JobQueueHelper.js"; import { Queue, Worker } from "bullmq"; //Network service and dependencies @@ -112,17 +113,13 @@ const shutdown = async () => { service: SERVICE_NAME, method: "shutdown", }); - // flush Redis - const redisService = ServiceRegistry.get(RedisService.SERVICE_NAME); - await redisService.flushall(); - + await ServiceRegistry.get(RedisService.SERVICE_NAME).flushRedis(); process.exit(1); }, SHUTDOWN_TIMEOUT); try { server.close(); - await ServiceRegistry.get(JobQueue.SERVICE_NAME).obliterate(); + await ServiceRegistry.get(JobQueue.SERVICE_NAME).shutdown(); await ServiceRegistry.get(MongoDB.SERVICE_NAME).disconnect(); - await ServiceRegistry.get(RedisService.SERVICE_NAME).flushall(); logger.info({ message: "Graceful shutdown complete" }); process.exit(0); } catch (error) { @@ -182,24 +179,19 @@ const startApp = async () => { stringService ); - const redisService = await RedisService.createInstance({ - logger, - IORedis, - SettingsService: settingsService, - }); + const redisService = new RedisService({ Redis: IORedis, logger }); - const jobQueue = new JobQueue({ - db, - statusService, - networkService, - notificationService, - settingsService, - stringService, - logger, + const jobQueueHelper = new JobQueueHelper({ + redisService, Queue, Worker, - redisService, + logger, + db, + networkService, + statusService, + notificationService, }); + const jobQueue = new JobQueue({ db, jobQueueHelper, logger, stringService }); // Register services ServiceRegistry.register(JobQueue.SERVICE_NAME, jobQueue); @@ -309,8 +301,6 @@ const startApp = async () => { ); const notificationRoutes = new NotificationRoutes(notificationController); const diagnosticRoutes = new DiagnosticRoutes(diagnosticController); - // Init job queue - await jobQueue.initJobQueue(); // Middleware app.use(responseHandler); app.use( diff --git a/server/service/JobQueue/JobQueue.js b/server/service/JobQueue/JobQueue.js new file mode 100644 index 000000000..84d603df5 --- /dev/null +++ b/server/service/JobQueue/JobQueue.js @@ -0,0 +1,301 @@ +const QUEUE_NAMES = ["uptime", "pagespeed", "hardware", "distributed"]; +const SERVICE_NAME = "JobQueue"; +const HEALTH_CHECK_INTERVAL = 10 * 60 * 1000; // 10 minutes +const QUEUE_LOOKUP = { + hardware: "hardware", + http: "uptime", + ping: "uptime", + port: "uptime", + docker: "uptime", + pagespeed: "pagespeed", + distributed_http: "distributed", +}; +const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`; + +class JobQueue { + static SERVICE_NAME = SERVICE_NAME; + + constructor({ db, jobQueueHelper, logger, stringService }) { + this.db = db; + this.jobQueueHelper = jobQueueHelper; + this.stringService = stringService; + this.logger = logger; + this.queues = {}; + this.workers = []; + this.init(); + } + + async init() { + try { + await this.initQueues(); + await this.initWorkers(); + const monitors = await this.db.getAllMonitors(); + await Promise.all( + monitors + .filter((monitor) => monitor.isActive) + .map(async (monitor) => { + try { + await this.addJob(monitor._id, monitor); + } catch (error) { + this.logger.error({ + message: error.message, + service: SERVICE_NAME, + method: "initJobQueue", + stack: error.stack, + }); + } + }) + ); + + this.healthCheckInterval = setInterval(async () => { + try { + const queueHealthChecks = await this.checkQueueHealth(); + const queueIsStuck = queueHealthChecks.some((healthCheck) => healthCheck.stuck); + if (queueIsStuck) { + console.log("Queue is stuck"); + console.log(queueHealthChecks); + // await this.flushQueues(); + } + } catch (error) { + this.logger.error({ + message: error.message, + service: SERVICE_NAME, + method: "periodicHealthCheck", + stack: error.stack, + }); + } + }, HEALTH_CHECK_INTERVAL); + } catch (error) { + this.logger.error({ + message: error.message, + service: SERVICE_NAME, + method: "initJobQueue", + stack: error.stack, + }); + } + } + + async initQueues() { + const readyPromises = []; + + for (const queueName of QUEUE_NAMES) { + const q = this.jobQueueHelper.createQueue(queueName); + this.queues[queueName] = q; + readyPromises.push(q.waitUntilReady()); + } + await Promise.all(readyPromises); + this.logger.info({ + message: "Queues ready", + service: SERVICE_NAME, + method: "initQueues", + }); + } + + async initWorkers() { + const workerReadyPromises = []; + + for (const queueName of QUEUE_NAMES) { + const worker = this.jobQueueHelper.createWorker(queueName, this.queues[queueName]); + this.workers.push(worker); + workerReadyPromises.push(worker.waitUntilReady()); + } + await Promise.all(workerReadyPromises); + this.logger.info({ + message: "Workers ready", + service: SERVICE_NAME, + method: "initWorkers", + }); + } + + async addJob(jobName, monitor) { + this.logger.info({ + message: `Adding job ${monitor?.url ?? "No URL"}`, + service: SERVICE_NAME, + method: "addJob", + }); + + const queueName = QUEUE_LOOKUP[monitor.type]; + const queue = this.queues[queueName]; + if (typeof queue === "undefined") { + throw new Error(`Queue for ${monitor.type} not found`); + } + const jobTemplate = { + name: jobName, + data: monitor, + opts: { + attempts: 1, + backoff: { + type: "exponential", + delay: 1000, + }, + removeOnComplete: true, + removeOnFail: false, + timeout: 1 * 60 * 1000, + }, + }; + + const schedulerId = getSchedulerId(monitor); + await queue.upsertJobScheduler( + schedulerId, + { every: monitor?.interval ?? 60000 }, + jobTemplate + ); + } + + async deleteJob(monitor) { + try { + const queue = this.queues[QUEUE_LOOKUP[monitor.type]]; + const schedulerId = getSchedulerId(monitor); + const wasDeleted = await queue.removeJobScheduler(schedulerId); + + if (wasDeleted === true) { + this.logger.info({ + message: this.stringService.jobQueueDeleteJob, + service: SERVICE_NAME, + method: "deleteJob", + details: `Deleted job ${monitor._id}`, + }); + return true; + } else { + this.logger.error({ + message: this.stringService.jobQueueDeleteJob, + service: SERVICE_NAME, + method: "deleteJob", + details: `Failed to delete job ${monitor._id}`, + }); + return false; + } + } catch (error) { + error.service === undefined ? (error.service = SERVICE_NAME) : null; + error.method === undefined ? (error.method = "deleteJob") : null; + throw error; + } + } + + async getJobs() { + try { + let stats = {}; + await Promise.all( + QUEUE_NAMES.map(async (name) => { + const queue = this.queues[name]; + const jobs = await queue.getJobs(); + const ret = await Promise.all( + jobs.map(async (job) => { + const state = await job.getState(); + return { url: job.data.url, state, progress: job.progress }; + }) + ); + stats[name] = { jobs: ret }; + }) + ); + return stats; + } catch (error) { + error.service === undefined ? (error.service = SERVICE_NAME) : null; + error.method === undefined ? (error.method = "getJobStats") : null; + throw error; + } + } + + async getMetrics() { + try { + let metrics = {}; + + await Promise.all( + QUEUE_NAMES.map(async (name) => { + const queue = this.queues[name]; + const [waiting, active, failed, delayed, repeatableJobs] = await Promise.all([ + queue.getWaitingCount(), + queue.getActiveCount(), + queue.getFailedCount(), + queue.getDelayedCount(), + queue.getRepeatableJobs(), + ]); + + metrics[name] = { + waiting, + active, + failed, + delayed, + repeatableJobs: repeatableJobs.length, + }; + }) + ); + + return metrics; + } catch (error) { + this.logger.error({ + message: error.message, + service: SERVICE_NAME, + method: "getMetrics", + stack: error.stack, + }); + } + } + + async checkQueueHealth() { + const res = []; + for (const queueName of QUEUE_NAMES) { + const q = this.queues[queueName]; + await q.waitUntilReady(); + + const lastJobProcessedTime = q.lastJobProcessedTime; + const currentTime = Date.now(); + const timeDiff = currentTime - lastJobProcessedTime; + + // Check for jobs + const jobCounts = await q.getJobCounts(); + const hasJobs = Object.values(jobCounts).some((count) => count > 0); + + res.push({ + queueName, + timeSinceLastJob: timeDiff, + stuck: hasJobs && timeDiff > 10000, + jobCounts, + }); + } + return res; + } + + async flushQueues() { + try { + this.logger.warn({ + message: "Flushing queues", + method: "flushQueues", + service: SERVICE_NAME, + }); + for (const worker of this.workers) { + await worker.close(); + } + this.workers = []; + + for (const queue of Object.values(this.queues)) { + await queue.obliterate(); + } + this.queue = {}; + await this.init(); + return true; + } catch (error) { + this.logger.warn({ + message: `${error.message} - Flushing redis manually`, + service: SERVICE_NAME, + method: "flushQueues", + }); + return await this.jobQueueHelper.flushRedis(); + } + } + + async shutdown() { + if (this.healthCheckInterval) { + clearInterval(this.healthCheckInterval); + this.healthCheckInterval = null; + } + for (const worker of this.workers) { + await worker.close(); + } + for (const queue of Object.values(this.queues)) { + await queue.obliterate(); + } + } +} + +export default JobQueue; diff --git a/server/service/JobQueue/JobQueueHelper.js b/server/service/JobQueue/JobQueueHelper.js new file mode 100644 index 000000000..f58f9cef7 --- /dev/null +++ b/server/service/JobQueue/JobQueueHelper.js @@ -0,0 +1,315 @@ +const SERVICE_NAME = "JobQueueHelper"; + +class JobQueueHelper { + constructor({ + redisService, + Queue, + Worker, + logger, + db, + networkService, + statusService, + notificationService, + }) { + this.db = db; + this.redisService = redisService; + this.Queue = Queue; + this.Worker = Worker; + this.logger = logger; + this.networkService = networkService; + this.statusService = statusService; + this.notificationService = notificationService; + } + + createQueue(queueName) { + const connection = this.redisService.getNewConnection(); + const q = new this.Queue(queueName, { + connection, + }); + q.lastJobProcessedTime = Date.now(); + q.on("cleaned", (jobs, type) => { + this.logger.debug({ + message: `Queue ${queueName} is cleaned with jobs: ${jobs} and type: ${type}`, + service: SERVICE_NAME, + method: "createQueue:cleaned", + }); + }); + q.on("error", (err) => { + this.logger.error({ + message: `Queue ${queueName} is error with msg: ${err}`, + service: SERVICE_NAME, + method: "createQueue:error", + }); + }); + q.on("ioredis:close", () => { + this.logger.debug({ + message: `Queue ${queueName} is ioredis:close`, + service: SERVICE_NAME, + method: "createQueue:ioredis:close", + }); + }); + q.on("paused", () => { + this.logger.debug({ + message: `Queue ${queueName} is paused`, + service: SERVICE_NAME, + method: "createQueue:paused", + }); + }); + q.on("progress", (job, progress) => { + this.logger.debug({ + message: `Queue ${queueName} is progress with msg: ${progress}`, + service: SERVICE_NAME, + method: "createQueue:progress", + }); + }); + q.on("removed", (job) => { + this.logger.debug({ + message: `Queue ${queueName} is removed with msg: ${job}`, + service: SERVICE_NAME, + method: "createQueue:removed", + }); + }); + q.on("resumed", () => { + this.logger.debug({ + message: `Queue ${queueName} is resumed`, + service: SERVICE_NAME, + method: "createQueue:resumed", + }); + }); + q.on("waiting", () => { + this.logger.debug({ + message: `Queue ${queueName} is waiting`, + service: SERVICE_NAME, + method: "createQueue:waiting", + }); + }); + return q; + } + + createWorker(queueName, queue) { + const connection = this.redisService.getNewConnection({ + maxRetriesPerRequest: null, + }); + const worker = new this.Worker(queueName, this.createJobHandler(queue), { + connection, + concurrency: 50, + }); + worker.on("active", (job) => { + this.logger.debug({ + message: `Worker ${queueName} is active`, + service: SERVICE_NAME, + method: "createWorker:active", + }); + }); + + worker.on("closed", () => { + this.logger.debug({ + message: `Worker ${queueName} is closed`, + service: SERVICE_NAME, + method: "createWorker:closed", + }); + }); + + worker.on("closing", (msg) => { + this.logger.debug({ + message: `Worker ${queueName} is closing with msg: ${msg}`, + service: SERVICE_NAME, + method: "createWorker:closing", + }); + }); + + worker.on("completed", (job) => { + this.logger.debug({ + message: `Worker ${queueName} is completed`, + service: SERVICE_NAME, + method: "createWorker:completed", + }); + }); + + worker.on("drained", () => { + this.logger.debug({ + message: `Worker ${queueName} is drained`, + service: SERVICE_NAME, + method: "createWorker:drained", + }); + }); + + worker.on("error", (failedReason) => { + this.logger.error({ + message: `Worker ${queueName} is error with msg: ${failedReason}`, + service: SERVICE_NAME, + method: "createWorker:error", + }); + }); + + worker.on("failed", (job, error, prev) => { + this.logger.warn({ + message: `Worker ${queueName} is failed with msg: ${error}`, + service: SERVICE_NAME, + method: "createWorker:failed", + }); + }); + + worker.on("ioredis:close", () => { + this.logger.debug({ + message: `Worker ${queueName} is ioredis:close`, + service: SERVICE_NAME, + method: "createWorker:ioredis:close", + }); + }); + + worker.on("paused", () => { + this.logger.debug({ + message: `Worker ${queueName} is paused`, + service: SERVICE_NAME, + method: "createWorker:paused", + }); + }); + + worker.on("progress", (job, progress) => { + this.logger.debug({ + message: `Worker ${queueName} is progress with msg: ${progress}`, + service: SERVICE_NAME, + method: "createWorker:progress", + }); + }); + + worker.on("ready", () => { + this.logger.debug({ + message: `Worker ${queueName} is ready`, + service: SERVICE_NAME, + method: "createWorker:ready", + }); + }); + + worker.on("resumed", () => { + this.logger.debug({ + message: `Worker ${queueName} is resumed`, + service: SERVICE_NAME, + method: "createWorker:resumed", + }); + }); + + worker.on("stalled", () => { + this.logger.warn({ + message: `Worker ${queueName} is stalled`, + service: SERVICE_NAME, + method: "createWorker:stalled", + }); + }); + return worker; + } + + async isInMaintenanceWindow(monitorId) { + const maintenanceWindows = await this.db.getMaintenanceWindowsByMonitorId(monitorId); + // Check for active maintenance window: + const maintenanceWindowIsActive = maintenanceWindows.reduce((acc, window) => { + if (window.active) { + const start = new Date(window.start); + const end = new Date(window.end); + const now = new Date(); + const repeatInterval = window.repeat || 0; + + // If start is < now and end > now, we're in maintenance + if (start <= now && end >= now) return true; + + // If maintenance window was set in the past with a repeat, + // we need to advance start and end to see if we are in range + + while (start < now && repeatInterval !== 0) { + start.setTime(start.getTime() + repeatInterval); + end.setTime(end.getTime() + repeatInterval); + if (start <= now && end >= now) { + return true; + } + } + return false; + } + return acc; + }, false); + return maintenanceWindowIsActive; + } + + createJobHandler(q) { + return async (job) => { + try { + // Update the last job processed time for this queue + q.lastJobProcessedTime = Date.now(); + // Get all maintenance windows for this monitor + await job.updateProgress(0); + const monitorId = job.data._id; + const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId); + // If a maintenance window is active, we're done + + if (maintenanceWindowActive) { + await job.updateProgress(100); + this.logger.info({ + message: `Monitor ${monitorId} is in maintenance window`, + service: SERVICE_NAME, + method: "createWorker", + }); + return false; + } + + // Get the current status + await job.updateProgress(30); + const networkResponse = await this.networkService.getStatus(job); + if ( + job.data.type === "distributed_http" || + job.data.type === "distributed_test" + ) { + return; + } + // Handle status change + await job.updateProgress(60); + const { monitor, statusChanged, prevStatus } = + await this.statusService.updateStatus(networkResponse); + // Handle notifications + await job.updateProgress(80); + this.notificationService + .handleNotifications({ + ...networkResponse, + monitor, + prevStatus, + statusChanged, + }) + .catch((error) => { + this.logger.error({ + message: error.message, + service: SERVICE_NAME, + method: "createJobHandler", + details: `Error sending notifications for job ${job.id}: ${error.message}`, + stack: error.stack, + }); + }); + await job.updateProgress(100); + return true; + } catch (error) { + this.logger.error({ + message: error.message, + service: error.service ?? SERVICE_NAME, + method: error.method ?? "createJobHandler", + details: `Error processing job ${job.id}: ${error.message}`, + stack: error.stack, + }); + throw error; + } + }; + } + async flushRedis() { + try { + const connection = this.redisService.getNewConnection(); + const flushResult = await connection.flushall(); + return flushResult; + } catch (error) { + this.logger.warn({ + message: error.message, + service: SERVICE_NAME, + method: "flushRedis", + }); + return false; + } + } +} + +export default JobQueueHelper; diff --git a/server/service/jobQueue.js b/server/service/jobQueue.js deleted file mode 100755 index 8f16f910d..000000000 --- a/server/service/jobQueue.js +++ /dev/null @@ -1,814 +0,0 @@ -const QUEUE_NAMES = ["uptime", "pagespeed", "hardware", "distributed"]; -const SERVICE_NAME = "JobQueue"; -const JOBS_PER_WORKER = 5; -const HEALTH_CHECK_INTERVAL = 10 * 60 * 1000; // 10 minutes -const QUEUE_LOOKUP = { - hardware: "hardware", - http: "uptime", - ping: "uptime", - port: "uptime", - docker: "uptime", - pagespeed: "pagespeed", - distributed_http: "distributed", -}; -const getSchedulerId = (monitor) => `scheduler:${monitor.type}:${monitor._id}`; - -class NewJobQueue { - static SERVICE_NAME = SERVICE_NAME; - - constructor({ - db, - statusService, - networkService, - notificationService, - settingsService, - stringService, - logger, - Queue, - Worker, - redisService, - }) { - this.connection = redisService.getConnection(); - this.queues = {}; - this.workers = {}; - this.lastJobProcessedTime = {}; - - this.db = db; - this.networkService = networkService; - this.statusService = statusService; - this.notificationService = notificationService; - this.settingsService = settingsService; - this.logger = logger; - this.Worker = Worker; - this.stringService = stringService; - - QUEUE_NAMES.forEach((name) => { - const q = new Queue(name, { connection: this.connection }); - this.lastJobProcessedTime[q.name] = Date.now(); - q.on("error", (error) => { - this.logger.error({ - message: error.message, - service: SERVICE_NAME, - method: "queue:error", - stack: error.stack, - }); - }); - this.queues[name] = q; - - this.workers[name] = []; - }); - - this.healthCheckInterval = setInterval(async () => { - try { - const health = await this.checkQueueHealth(); - if (health.stuck === true) { - this.logger.error({ - message: `Queue is stuck: ${health.stuckQueues.join(", ")}`, - service: SERVICE_NAME, - method: "healthCheckInterval", - }); - await this.flushQueue(); - } - } catch (error) { - this.logger.error({ - message: error.message, - service: SERVICE_NAME, - method: "periodicHealthCheck", - stack: error.stack, - }); - } - }, HEALTH_CHECK_INTERVAL); - } - - /** - * Initializes job queues by adding jobs for all active monitors - * @async - * @function initJobQueue - * @description Retrieves all monitors from the database and adds jobs for active ones to their respective queues - * @throws {Error} If there's an error retrieving monitors or adding jobs - * @returns {Promise} - */ - async initJobQueue() { - await this.connection.flushall(); - const monitors = await this.db.getAllMonitors(); - await Promise.all( - monitors - .filter((monitor) => monitor.isActive) - .map(async (monitor) => { - try { - await this.addJob(monitor._id, monitor); - } catch (error) { - this.logger.error({ - message: `Failed to add job for monitor ${monitor._id}:`, - service: SERVICE_NAME, - method: "initJobQueue", - stack: error.stack, - }); - } - }) - ); - } - - /** - * Checks if a monitor is currently in a maintenance window - * @async - * @param {string} monitorId - The ID of the monitor to check - * @returns {Promise} Returns true if the monitor is in an active maintenance window, false otherwise - * @throws {Error} If there's an error retrieving maintenance windows from the database - * @description - * Retrieves all maintenance windows for a monitor and checks if any are currently active. - * A maintenance window is considered active if: - * 1. The window is marked as active AND - * 2. Either: - * - Current time falls between start and end times - * - For repeating windows: Current time falls between any repeated interval - */ - async isInMaintenanceWindow(monitorId) { - const maintenanceWindows = await this.db.getMaintenanceWindowsByMonitorId(monitorId); - // Check for active maintenance window: - const maintenanceWindowIsActive = maintenanceWindows.reduce((acc, window) => { - if (window.active) { - const start = new Date(window.start); - const end = new Date(window.end); - const now = new Date(); - const repeatInterval = window.repeat || 0; - - // If start is < now and end > now, we're in maintenance - if (start <= now && end >= now) return true; - - // If maintenance window was set in the past with a repeat, - // we need to advance start and end to see if we are in range - - while (start < now && repeatInterval !== 0) { - start.setTime(start.getTime() + repeatInterval); - end.setTime(end.getTime() + repeatInterval); - if (start <= now && end >= now) { - return true; - } - } - return false; - } - return acc; - }, false); - return maintenanceWindowIsActive; - } - - /** - * Creates a job processing handler for monitor checks - * @function createJobHandler - * @returns {Function} An async function that processes monitor check jobs - * @description - * Creates and returns a job handler that: - * 1. Checks if monitor is in maintenance window - * 2. If not in maintenance, performs network status check - * 3. Updates monitor status in database - * 4. Triggers notifications if status changed - * - * @param {Object} job - The job to process - * @param {Object} job.data - The monitor data - * @param {string} job.data._id - Monitor ID - * @param {string} job.id - Job ID - * - * @throws {Error} Logs errors but doesn't throw them to prevent job failure - * @returns {Promise} Resolves when job processing is complete - */ - createJobHandler() { - return async (job) => { - try { - // Update the last job processed time for this queue - this.lastJobProcessedTime[job.queue.name] = Date.now(); - // Get all maintenance windows for this monitor - await job.updateProgress(0); - const monitorId = job.data._id; - const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId); - // If a maintenance window is active, we're done - - if (maintenanceWindowActive) { - await job.updateProgress(100); - this.logger.info({ - message: `Monitor ${monitorId} is in maintenance window`, - service: SERVICE_NAME, - method: "createWorker", - }); - return false; - } - - // Get the current status - await job.updateProgress(30); - const networkResponse = await this.networkService.getStatus(job); - if ( - job.data.type === "distributed_http" || - job.data.type === "distributed_test" - ) { - return; - } - // Handle status change - await job.updateProgress(60); - const { monitor, statusChanged, prevStatus } = - await this.statusService.updateStatus(networkResponse); - // Handle notifications - await job.updateProgress(80); - this.notificationService - .handleNotifications({ - ...networkResponse, - monitor, - prevStatus, - statusChanged, - }) - .catch((error) => { - this.logger.error({ - message: error.message, - service: SERVICE_NAME, - method: "createJobHandler", - details: `Error sending notifications for job ${job.id}: ${error.message}`, - stack: error.stack, - }); - }); - await job.updateProgress(100); - return true; - } catch (error) { - this.logger.error({ - message: error.message, - service: error.service ?? SERVICE_NAME, - method: error.method ?? "createJobHandler", - details: `Error processing job ${job.id}: ${error.message}`, - stack: error.stack, - }); - throw error; - } - }; - } - - /** - * Creates a new worker for processing jobs in a queue - * @param {Queue} queue - The BullMQ queue to create a worker for - * @returns {Worker} A new BullMQ worker instance - * @description - * Creates and configures a new worker with: - * - Queue-specific job handler - * - Redis connection settings - * - Default worker options - * The worker processes jobs from the specified queue using the job handler - * created by createJobHandler() - * - * @throws {Error} If worker creation fails or connection is invalid - */ - createWorker(queue) { - try { - const worker = new this.Worker(queue.name, this.createJobHandler(), { - connection: this.connection, - concurrency: 5, - stalledInterval: 10000, - maxStalledCount: 1, - lockDuration: 60000, - }); - - worker.on("failed", (job, err) => { - this.logger.error({ - message: `Job ${job.id} failed: ${err.message}`, - service: SERVICE_NAME, - method: "worker:failed", - stack: err.stack, - jobData: job.data, - }); - }); - worker.on("error", (job, err) => { - this.logger.error({ - message: `Job ${job.id} error: ${err.message}`, - service: SERVICE_NAME, - method: "worker:error", - stack: err.stack, - jobData: job.data, - }); - }); - - worker.on("stalled", (jobId) => { - this.logger.warn({ - message: `Job ${jobId} stalled`, - service: SERVICE_NAME, - method: "worker:stalled", - }); - }); - - return worker; - } catch (error) { - this.logger.error({ - message: error.message, - service: SERVICE_NAME, - method: "createWorker", - stack: error.stack, - }); - error.service = SERVICE_NAME; - error.method = "createWorker"; - throw error; - } - } - - /** - * Gets stats related to the workers - * This is used for scaling workers right now - * In the future we will likely want to scale based on server performance metrics - * CPU Usage & memory usage, if too high, scale down workers. - * When to scale up? If jobs are taking too long to complete? - * @async - * @returns {Promise} - Returns the worker stats - */ - async getWorkerStats(queue) { - try { - const jobs = await queue.getRepeatableJobs(); - const load = jobs.length / this.workers[queue.name].length; - return { jobs, load }; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "getWorkerStats") : null; - throw error; - } - } - - /** - * Scales workers up or down based on queue load - * @async - * @param {Object} workerStats - Statistics about current worker load - * @param {number} workerStats.load - Current load per worker - * @param {Array} workerStats.jobs - Array of current jobs - * @param {Queue} queue - The BullMQ queue to scale workers for - * @returns {Promise} True if scaling occurred, false if no scaling was needed - * @throws {Error} If no workers array exists for the queue - * @description - * Scales workers based on these rules: - * - Maintains minimum of 5 workers - * - Adds workers if load exceeds JOBS_PER_WORKER - * - Removes workers if load is below JOBS_PER_WORKER - * - Creates initial workers if none exist - * Worker scaling is calculated based on excess jobs or excess capacity - */ - async scaleWorkers(workerStats, queue) { - const workers = this.workers[queue.name]; - if (workers === undefined) { - throw new Error(`No workers found for ${queue.name}`); - } - - if (workers.length === 0) { - // There are no workers, need to add one - for (let i = 0; i < 5; i++) { - const worker = this.createWorker(queue); - workers.push(worker); - } - return true; - } - if (workerStats.load > JOBS_PER_WORKER) { - // Find out how many more jobs we have than current workers can handle - const excessJobs = workerStats.jobs.length - workers.length * JOBS_PER_WORKER; - // Divide by jobs/worker to find out how many workers to add - const workersToAdd = Math.ceil(excessJobs / JOBS_PER_WORKER); - for (let i = 0; i < workersToAdd; i++) { - const worker = this.createWorker(queue); - workers.push(worker); - } - return true; - } - - if (workerStats.load < JOBS_PER_WORKER) { - // Find out how much excess capacity we have - const workerCapacity = workers.length * JOBS_PER_WORKER; - const excessCapacity = workerCapacity - workerStats.jobs.length; - // Calculate how many workers to remove - let workersToRemove = Math.floor(excessCapacity / JOBS_PER_WORKER); // Make sure there are always at least 5 - while (workersToRemove > 0 && workers.length > 5) { - const worker = workers.pop(); - workersToRemove--; - await worker.close().catch((error) => { - // Catch the error instead of throwing it - this.logger.error({ - message: error.message, - service: SERVICE_NAME, - method: "scaleWorkers", - stack: error.stack, - }); - }); - } - return true; - } - return false; - } - - /** - * Gets all jobs in the queue. - * - * @async - * @returns {Promise>} - * @throws {Error} - Throws error if getting jobs fails - */ - async getJobs(queue) { - try { - const jobs = await queue.getRepeatableJobs(); - return jobs; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "getJobs") : null; - throw error; - } - } - - /** - * Retrieves detailed statistics about jobs and workers for all queues - * @async - * @returns {Promise} Queue statistics object - * @throws {Error} If there's an error retrieving job information - * @description - * Returns an object with statistics for each queue including: - * - List of jobs with their URLs and current states - * - Number of workers assigned to the queue - */ - async getJobStats() { - try { - let stats = {}; - await Promise.all( - QUEUE_NAMES.map(async (name) => { - const queue = this.queues[name]; - const jobs = await queue.getJobs(); - const ret = await Promise.all( - jobs.map(async (job) => { - const state = await job.getState(); - return { url: job.data.url, state, progress: job.progress }; - }) - ); - stats[name] = { jobs: ret, workers: this.workers[name].length }; - }) - ); - return stats; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "getJobStats") : null; - throw error; - } - } - - /** - * Adds both immediate and repeatable jobs to the appropriate queue - * @async - * @param {string} jobName - Name identifier for the job - * @param {Object} monitor - Job data and configuration - * @param {string} monitor.type - Type of monitor/queue ('uptime', 'pagespeed', 'hardware') - * @param {string} [monitor.url] - URL to monitor (optional) - * @param {number} [monitor.interval=60000] - Repeat interval in milliseconds - * @param {string} monitor._id - Monitor ID - * @throws {Error} If queue not found for payload type - * @throws {Error} If job addition fails - * @description - * 1. Identifies correct queue based on payload type - * 2. Adds immediate job execution - * 3. Adds repeatable job with specified interval - * 4. Scales workers based on updated queue load - * Jobs are configured with exponential backoff, single attempt, - * and automatic removal on completion - */ - async addJob(jobName, monitor) { - try { - this.logger.info({ - message: `Adding job ${monitor?.url ?? "No URL"}`, - service: SERVICE_NAME, - method: "addJob", - }); - - // Find the correct queue - - const queue = this.queues[QUEUE_LOOKUP[monitor.type]]; - if (queue === undefined) { - throw new Error(`Queue for ${monitor.type} not found`); - } - - const jobTemplate = { - name: jobName, - data: monitor, - opts: { - attempts: 1, - backoff: { - type: "exponential", - delay: 1000, - }, - removeOnComplete: true, - removeOnFail: false, - timeout: 1 * 60 * 1000, - }, - }; - - const schedulerId = getSchedulerId(monitor); - await queue.upsertJobScheduler( - schedulerId, - { every: monitor?.interval ?? 60000 }, - jobTemplate - ); - - const workerStats = await this.getWorkerStats(queue); - await this.scaleWorkers(workerStats, queue); - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "addJob") : null; - throw error; - } - } - - /** - * Deletes a repeatable job from its queue and adjusts worker scaling - * @async - * @param {Object} monitor - Monitor object containing job details - * @param {string} monitor._id - ID of the monitor/job to delete - * @param {string} monitor.type - Type of monitor determining queue selection - * @param {number} monitor.interval - Job repeat interval in milliseconds - * @throws {Error} If queue not found for monitor type - * @throws {Error} If job deletion fails - * @description - * 1. Identifies correct queue based on monitor type - * 2. Removes repeatable job using monitor ID and interval - * 3. Logs success or failure of deletion - * 4. Updates worker scaling based on new queue load - * Returns void but logs operation result - */ - async deleteJob(monitor) { - try { - const queue = this.queues[QUEUE_LOOKUP[monitor.type]]; - const schedulerId = getSchedulerId(monitor); - const wasDeleted = await queue.removeJobScheduler(schedulerId); - - if (wasDeleted === true) { - this.logger.info({ - message: this.stringService.jobQueueDeleteJob, - service: SERVICE_NAME, - method: "deleteJob", - details: `Deleted job ${monitor._id}`, - }); - const workerStats = await this.getWorkerStats(queue); - await this.scaleWorkers(workerStats, queue); - } else { - this.logger.error({ - message: this.stringService.jobQueueDeleteJob, - service: SERVICE_NAME, - method: "deleteJob", - details: `Failed to delete job ${monitor._id}`, - }); - } - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "deleteJob") : null; - throw error; - } - } - - /** - * Retrieves comprehensive metrics for all queues - * @async - * @returns {Promise>} Object with metrics for each queue - * @throws {Error} If metrics retrieval fails - * @description - * Collects the following metrics for each queue: - * - Number of waiting jobs - * - Number of active jobs - * - Number of completed jobs - * - Number of failed jobs - * - Number of delayed jobs - * - Number of repeatable jobs - * - Number of active workers - * - * @typedef {Object} QueueMetrics - * @property {number} waiting - Count of jobs waiting to be processed - * @property {number} active - Count of jobs currently being processed - * @property {number} completed - Count of successfully completed jobs - * @property {number} failed - Count of failed jobs - * @property {number} delayed - Count of delayed jobs - * @property {number} repeatableJobs - Count of repeatable job patterns - * @property {number} workers - Count of active workers for this queue - */ - async getMetrics() { - try { - let metrics = {}; - - await Promise.all( - QUEUE_NAMES.map(async (name) => { - const queue = this.queues[name]; - const workers = this.workers[name]; - const [waiting, active, completed, failed, delayed, repeatableJobs] = - await Promise.all([ - queue.getWaitingCount(), - queue.getActiveCount(), - queue.getCompletedCount(), - queue.getFailedCount(), - queue.getDelayedCount(), - queue.getRepeatableJobs(), - ]); - - metrics[name] = { - waiting, - active, - completed, - failed, - delayed, - repeatableJobs: repeatableJobs.length, - workers: workers.length, - }; - }) - ); - - return metrics; - } catch (error) { - this.logger.error({ - message: error.message, - service: SERVICE_NAME, - method: "getMetrics", - stack: error.stack, - }); - } - } - - /** - * @async - * @returns {Promise} - Returns true if obliteration is successful - */ - - async obliterate() { - try { - this.logger.info({ - message: "Attempting to obliterate job queue...", - service: SERVICE_NAME, - method: "obliterate", - }); - - if (this.healthCheckInterval) { - clearInterval(this.healthCheckInterval); - this.healthCheckInterval = null; - } - - await Promise.all( - QUEUE_NAMES.map(async (name) => { - const queue = this.queues[name]; - await queue.pause(); - const jobs = await this.getJobs(queue); - - // Remove all repeatable jobs - for (const job of jobs) { - await queue.removeRepeatableByKey(job.key); - await queue.remove(job.id); - } - }) - ); - - // Close workers - await Promise.all( - QUEUE_NAMES.map(async (name) => { - const workers = this.workers[name]; - await Promise.all( - workers.map(async (worker) => { - await worker.close(); - }) - ); - }) - ); - - QUEUE_NAMES.forEach(async (name) => { - const queue = this.queues[name]; - await queue.obliterate(); - }); - - const metrics = await this.getMetrics(); - this.logger.info({ - message: this.stringService.jobQueueObliterate, - service: SERVICE_NAME, - method: "obliterate", - details: metrics, - }); - return true; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "obliterate") : null; - throw error; - } - } - - // ************************** - // Queue Health Checks - // ************************** - async getKeyValuePairs() { - try { - // Get all keys - const keys = await this.connection.keys("*"); - - if (keys.length === 0) { - return {}; // Return an empty object if no keys are found - } - - // Get values for all keys - const values = await this.connection.mget(keys); - - // Combine keys and values into an object - const keyValuePairs = keys.reduce((result, key, index) => { - result[key] = values[index]; - return result; - }, {}); - this.logger.info({ - message: "Redis key-value", - service: SERVICE_NAME, - method: "flushQueue", - details: keyValuePairs, - }); - return keyValuePairs; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "getKeyValuePairs") : null; - throw error; - } - } - - async flushQueue() { - try { - const keyValuePairs = await this.getKeyValuePairs(); - this.logger.info({ - message: "Before flush", - service: SERVICE_NAME, - method: "flushQueue", - details: keyValuePairs, - }); - const flushResult = await this.connection.flushall(); - const keyValuePairsAfter = await this.getKeyValuePairs(); - this.logger.info({ - message: "After flush", - service: SERVICE_NAME, - method: "flushQueue", - details: keyValuePairsAfter, - }); - if (flushResult !== "OK") { - throw new Error("Failed to flush queue"); - } - await this.initJobQueue(); - return { - keyValuePairs, - flush: flushResult, - keyValuePairsAfter, - init: true, - }; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "flushQueue") : null; - throw error; - } - } - - /** - * Gets metrics for a specific queue - * @async - * @function getQueueHealthMetrics - * @param {Queue} queue - The queue to get metrics for - * @returns {Promise} Queue metrics - */ - async getQueueHealthMetrics(queue) { - return await queue.getJobCounts(); - } - - getQueueIdleTimes() { - const now = Date.now(); - const idleTimes = {}; - Object.entries(this.lastJobProcessedTime).forEach(([queueName, lastProcessed]) => { - idleTimes[queueName] = now - lastProcessed; - }); - return idleTimes; - } - async checkQueueHealth() { - try { - const currentTime = Date.now(); - const stuckQueues = []; - const idleTimes = this.getQueueIdleTimes(); - for (const queueName of QUEUE_NAMES) { - const queue = this.queues[queueName]; - - const jobCounts = await this.getQueueHealthMetrics(queue); - const hasJobs = Object.values(jobCounts).some((count) => count > 0); - - const timeSinceLastProcessed = currentTime - this.lastJobProcessedTime[queueName]; - const isStuck = hasJobs && timeSinceLastProcessed > HEALTH_CHECK_INTERVAL; - - if (isStuck) { - stuckQueues.push(queueName); - } - } - - const queueHealth = { stuck: false, stuckQueues, idleTimes }; - - if (stuckQueues.length > 0) { - queueHealth.stuck = true; - } - - this.logger.info({ - message: "Queue health check", - service: SERVICE_NAME, - method: "checkQueueHealth", - details: queueHealth, - }); - return queueHealth; - } catch (error) { - error.service === undefined ? (error.service = SERVICE_NAME) : null; - error.method === undefined ? (error.method = "checkQueueHealth") : null; - throw error; - } - } -} - -export default NewJobQueue; diff --git a/server/service/redisService.js b/server/service/redisService.js index d2d76002c..12c1c04cf 100644 --- a/server/service/redisService.js +++ b/server/service/redisService.js @@ -1,71 +1,49 @@ +const SERVICE_NAME = "RedisService"; + class RedisService { - static SERVICE_NAME = "RedisService"; - - constructor({ logger, IORedis, SettingsService }) { + static SERVICE_NAME = SERVICE_NAME; + constructor({ Redis, logger }) { + this.Redis = Redis; + this.connections = new Set(); this.logger = logger; - this.IORedis = IORedis; - this.SettingsService = SettingsService; - this.connection = null; } - static async createInstance({ logger, IORedis, SettingsService }) { - const instance = new RedisService({ logger, IORedis, SettingsService }); - await instance.connect(); - return instance; - } - - async connect() { - const settings = this.SettingsService.getSettings(); - const { redisUrl } = settings; - this.connection = new this.IORedis(redisUrl, { - maxRetriesPerRequest: null, + getNewConnection(options = {}) { + const connection = new this.Redis(process.env.REDIS_URL, { retryStrategy: (times) => { - if (times >= 5) { - throw new Error("Failed to connect to Redis"); - } - this.logger.debug({ - message: "Retrying Redis connection", - service: RedisService.SERVICE_NAME, - details: { times }, - }); - return Math.min(times * 100, 2000); + return null; }, + ...options, }); - - await new Promise((resolve, reject) => { - let errorOccurred = false; - - this.connection.on("ready", () => { - if (!errorOccurred) { - this.logger.info({ - message: "Redis connection established", - service: RedisService.SERVICE_NAME, - }); - resolve(); - } - }); - - this.connection.on("error", (err) => { - errorOccurred = true; - this.logger.error({ - message: "Redis connection error", - service: RedisService.SERVICE_NAME, - error: err, - }); - setTimeout(() => reject(err), 5000); - }); - }); - } - async flushall() { - this.logger.debug({ - message: "Flushing all Redis data", - service: RedisService.SERVICE_NAME, - }); - await this.connection.flushall(); + this.connections.add(connection); + return connection; } - getConnection() { - return this.connection; + async closeAllConnections() { + const closePromises = Array.from(this.connections).map((conn) => + conn.quit().catch((err) => { + console.error("Error closing Redis connection:", err); + }) + ); + + await Promise.all(closePromises); + this.connections.clear(); + console.log("All Redis connections closed"); + } + + async flushRedis() { + this.logger.info({ + message: "Flushing Redis", + service: SERVICE_NAME, + method: "flushRedis", + }); + const flushPromises = Array.from(this.connections).map((conn) => conn.flushall()); + await Promise.all(flushPromises); + this.logger.info({ + message: "Redis flushed", + service: SERVICE_NAME, + method: "flushRedis", + }); } } From 9d69bd6782ccd5eb47d7809e3f83ce79dd90ead2 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 14:18:17 -0700 Subject: [PATCH 04/40] update package-lock --- server/package-lock.json | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/server/package-lock.json b/server/package-lock.json index f236139d8..7bf4bfc78 100755 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -26,8 +26,9 @@ "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", - "multer": "1.4.5-lts.1", + "multer": "^1.4.5-lts.1", "nodemailer": "^6.9.14", + "papaparse": "^5.5.2", "ping": "0.4.4", "sharp": "0.33.5", "ssl-checker": "2.0.10", @@ -5499,6 +5500,12 @@ "integrity": "sha512-UEZIS3/by4OC8vL3P2dTXRETpebLI2NiI5vIrjaD/5UtrkFX/tNbwjTSRAGC/+7CAo2pIcBaRgWmcBBHcsaCIw==", "license": "BlueOak-1.0.0" }, + "node_modules/papaparse": { + "version": "5.5.2", + "resolved": "https://registry.npmjs.org/papaparse/-/papaparse-5.5.2.tgz", + "integrity": "sha512-PZXg8UuAc4PcVwLosEEDYjPyfWnTEhOrUfdv+3Bx+NuAb+5NhDmXzg5fHWmdCh1mP5p7JAZfFr3IMQfcntNAdA==", + "license": "MIT" + }, "node_modules/parent-module": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/parent-module/-/parent-module-1.0.1.tgz", @@ -7290,9 +7297,9 @@ "license": "MIT" }, "node_modules/undici": { - "version": "6.21.1", - "resolved": "https://registry.npmjs.org/undici/-/undici-6.21.1.tgz", - "integrity": "sha512-q/1rj5D0/zayJB2FraXdaWxbhWiNKDvu8naDT2dl1yTlvJp4BLtOcp2a5BvgGNQpYYJzau7tf1WgKv3b+7mqpQ==", + "version": "6.21.3", + "resolved": "https://registry.npmjs.org/undici/-/undici-6.21.3.tgz", + "integrity": "sha512-gBLkYIlEnSp8pFbT64yFgGE6UIB9tAkhukC23PmMDCe5Nd+cRqKxSjw5y54MK2AZMgZfJWMaNE4nYUHgi1XEOw==", "license": "MIT", "engines": { "node": ">=18.17" From 0906e883c52ec054dad22cd5da5ab81e445341b8 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 15:01:06 -0700 Subject: [PATCH 05/40] add test email controller method --- server/controllers/settingsController.js | 58 +++++++++++++++++++++++- 1 file changed, 56 insertions(+), 2 deletions(-) diff --git a/server/controllers/settingsController.js b/server/controllers/settingsController.js index 21f0050ba..58e8ce9f4 100755 --- a/server/controllers/settingsController.js +++ b/server/controllers/settingsController.js @@ -1,13 +1,14 @@ import { updateAppSettingsBodyValidation } from "../validation/joi.js"; import { handleValidationError, handleError } from "./controllerUtils.js"; - +import { sendTestEmailBodyValidation } from "../validation/joi.js"; const SERVICE_NAME = "SettingsController"; class SettingsController { - constructor(db, settingsService, stringService) { + constructor({ db, settingsService, stringService, emailService }) { this.db = db; this.settingsService = settingsService; this.stringService = stringService; + this.emailService = emailService; } getAppSettings = async (req, res, next) => { @@ -55,6 +56,59 @@ class SettingsController { next(handleError(error, SERVICE_NAME, "updateAppSettings")); } }; + + sendTestEmail = async (req, res, next) => { + try { + await sendTestEmailBodyValidation.validateAsync(req.body); + } catch (error) { + next(handleValidationError(error, SERVICE_NAME)); + return; + } + + try { + const { + to, + systemEmailHost, + systemEmailPort, + systemEmailAddress, + systemEmailPassword, + systemEmailUser, + systemEmailConnectionHost, + } = req.body; + + const subject = this.stringService.testEmailSubject; + const context = { testName: "Monitoring System" }; + + const messageId = await this.emailService.buildAndSendEmail( + "testEmailTemplate", + context, + to, + subject, + { + systemEmailHost, + systemEmailPort, + systemEmailUser, + systemEmailAddress, + systemEmailPassword, + systemEmailConnectionHost, + } + ); + + if (!messageId) { + return res.error({ + msg: "Failed to send test email.", + }); + } + + return res.success({ + msg: this.stringService.sendTestEmail, + data: { messageId }, + }); + } catch (error) { + next(handleValidationError(error, SERVICE_NAME)); + return; + } + }; } export default SettingsController; From bea8ddb330e22780bceac3717698bb0832050848 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 15:01:34 -0700 Subject: [PATCH 06/40] add a test email route to settings routes --- server/routes/monitorRoute.js | 4 ++-- server/routes/settingsRoute.js | 5 +++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/server/routes/monitorRoute.js b/server/routes/monitorRoute.js index f9ef460c6..6d1db90ad 100755 --- a/server/routes/monitorRoute.js +++ b/server/routes/monitorRoute.js @@ -4,8 +4,8 @@ import multer from "multer"; import { fetchMonitorCertificate } from "../controllers/controllerUtils.js"; const upload = multer({ - storage: multer.memoryStorage() // Store file in memory as Buffer - }); + storage: multer.memoryStorage(), // Store file in memory as Buffer +}); class MonitorRoutes { constructor(monitorController) { diff --git a/server/routes/settingsRoute.js b/server/routes/settingsRoute.js index a5501aaf3..f293fd660 100755 --- a/server/routes/settingsRoute.js +++ b/server/routes/settingsRoute.js @@ -15,6 +15,11 @@ class SettingsRoutes { isAllowed(["admin", "superadmin"]), this.settingsController.updateAppSettings ); + this.router.post( + "/test-email", + isAllowed(["admin", "superadmin"]), + this.settingsController.sendTestEmail + ); } getRouter() { From d71181e65b21b4094f8e0eb569704899eb921402 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 15:01:45 -0700 Subject: [PATCH 07/40] add test-email body validation --- server/validation/joi.js | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/server/validation/joi.js b/server/validation/joi.js index 3dfee80c4..c16843fcd 100755 --- a/server/validation/joi.js +++ b/server/validation/joi.js @@ -591,6 +591,16 @@ const createAnnouncementValidation = joi.object({ userId: joi.string().required(), }); +const sendTestEmailBodyValidation = joi.object({ + to: joi.string().required(), + systemEmailHost: joi.string(), + systemEmailPort: joi.number(), + systemEmailAddress: joi.string(), + systemEmailPassword: joi.string(), + systemEmailUser: joi.string(), + systemEmailConnectionHost: joi.string(), +}); + export { roleValidatior, loginValidation, @@ -654,4 +664,5 @@ export { triggerNotificationBodyValidation, webhookConfigValidation, createAnnouncementValidation, + sendTestEmailBodyValidation, }; From f6a2a663867c90f8285fde9626b4774e19622a24 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 15:02:07 -0700 Subject: [PATCH 08/40] allow optional transport params --- server/service/emailService.js | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/server/service/emailService.js b/server/service/emailService.js index b3e087046..6e6a70684 100755 --- a/server/service/emailService.js +++ b/server/service/emailService.js @@ -89,8 +89,16 @@ class EmailService { * @param {string} subject - The subject of the email. * @returns {Promise} A promise that resolves to the messageId of the sent email. */ - buildAndSendEmail = async (template, context, to, subject) => { + buildAndSendEmail = async (template, context, to, subject, transportConfig) => { // TODO - Consider an update transporter method so this only needs to be recreated when smtp settings change + let config; + + if (typeof transportConfig !== "undefined") { + config = transportConfig; + } else { + config = this.settingsService.getDBSettings(); + } + const { systemEmailHost, systemEmailPort, @@ -98,7 +106,7 @@ class EmailService { systemEmailAddress, systemEmailPassword, systemEmailConnectionHost, - } = await this.settingsService.getDBSettings(); + } = config; const baseEmailConfig = { host: systemEmailHost, From 5872879c9ab45fbcd64be4d7b082eab458f17008 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Fri, 16 May 2025 15:02:26 -0700 Subject: [PATCH 09/40] inject email service into settings controller --- server/index.js | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/server/index.js b/server/index.js index f299b32e9..192f25c5a 100755 --- a/server/index.js +++ b/server/index.js @@ -241,11 +241,12 @@ const startApp = async () => { ServiceRegistry.get(EmailService.SERVICE_NAME) ); - const settingsController = new SettingsController( - ServiceRegistry.get(MongoDB.SERVICE_NAME), - ServiceRegistry.get(SettingsService.SERVICE_NAME), - ServiceRegistry.get(StringService.SERVICE_NAME) - ); + const settingsController = new SettingsController({ + db: ServiceRegistry.get(MongoDB.SERVICE_NAME), + settingsService: ServiceRegistry.get(SettingsService.SERVICE_NAME), + stringService: ServiceRegistry.get(StringService.SERVICE_NAME), + emailService: ServiceRegistry.get(EmailService.SERVICE_NAME), + }); const checkController = new CheckController( ServiceRegistry.get(MongoDB.SERVICE_NAME), From cf344b1809f55c14c912d752e502797066fb312d Mon Sep 17 00:00:00 2001 From: Owaise Imdad Date: Sat, 17 May 2025 11:51:31 +0530 Subject: [PATCH 10/40] Email test button added in details page. --- .../MonitorDetailsControlHeader/index.jsx | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/client/src/Components/MonitorDetailsControlHeader/index.jsx b/client/src/Components/MonitorDetailsControlHeader/index.jsx index c4dcf1ffc..5dcb56e84 100644 --- a/client/src/Components/MonitorDetailsControlHeader/index.jsx +++ b/client/src/Components/MonitorDetailsControlHeader/index.jsx @@ -5,12 +5,14 @@ import Button from "@mui/material/Button"; import SettingsOutlinedIcon from "@mui/icons-material/SettingsOutlined"; import PauseOutlinedIcon from "@mui/icons-material/PauseOutlined"; import PlayArrowOutlinedIcon from "@mui/icons-material/PlayArrowOutlined"; +import EmailIcon from '@mui/icons-material/Email'; // Utils import PropTypes from "prop-types"; import { useNavigate } from "react-router-dom"; import { useTheme } from "@mui/material/styles"; import { usePauseMonitor } from "../../Hooks/useMonitorControls"; +import { useSendTestEmail } from "../../Hooks/useSendTestEmail"; /** * MonitorDetailsControlHeader component displays the control header for monitor details. @@ -39,6 +41,8 @@ const MonitorDetailsControlHeader = ({ triggerUpdate, }); + const [isSending, emailError, sendTestEmail] = useSendTestEmail(); + if (isLoading) { return ; } @@ -54,6 +58,15 @@ const MonitorDetailsControlHeader = ({ direction="row" gap={theme.spacing(2)} > +