mirror/Checkmate
1
0
Fork 0
mirror of https://github.com/bluewave-labs/Checkmate.git synced 2026-01-24 02:29:35 -06:00
Code Issues Packages Projects Releases Wiki Activity

appSettings -> envSettings

Browse Source
This commit is contained in:
Alex Holliday
2025-07-28 14:37:18 -07:00
parent 140ae1d049
commit f3e50120ee
13 changed files with 27 additions and 1113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
server/index.js
View File

@@ -1,55 +0,0 @@
import { initializeServices } from "./src/config/services.js";
import { initializeControllers } from "./src/config/controllers.js";
import { createApp } from "./src/app.js";
import { initShutdownListener } from "./shutdown.js";
import logger from "./utils/logger.js";
import { fileURLToPath } from "url";
import path from "path";
import fs from "fs";
import SettingsService from "./src/service/system/settingsService.js";
import AppSettings from "./src/db/models/AppSettings.js";
const SERVICE_NAME = "Server";
const startApp = async () => {
// FE path
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
const openApiSpec = JSON.parse(fs.readFileSync(path.join(__dirname, "openapi.json"), "utf8"));
const frontendPath = path.join(__dirname, "public");
// Create services
const settingsService = new SettingsService(AppSettings);
const appSettings = settingsService.loadSettings();
// Initialize services
const services = await initializeServices(appSettings, settingsService);
// Initialize controllers
const controllers = initializeControllers(services);
const app = createApp({
services,
controllers,
appSettings,
frontendPath,
openApiSpec,
});
const port = appSettings.port || 52345;
const server = app.listen(port, () => {
logger.info({ message: `Server started on port:${port}` });
});
initShutdownListener(server, services);
};
startApp().catch((error) => {
logger.error({
message: error.message,
service: SERVICE_NAME,
method: "startApp",
stack: error.stack,
});
process.exit(1);
});

4
server/src/app.js
View File

@@ -10,8 +10,8 @@ import { handleErrors } from "./middleware/handleErrors.js";
import { setupRoutes } from "./config/routes.js";
import { generalApiLimiter } from "./middleware/rateLimiter.js";
export const createApp = ({ services, controllers, appSettings, frontendPath, openApiSpec }) => {
const allowedOrigin = appSettings.clientHost;
export const createApp = ({ services, controllers, envSettings, frontendPath, openApiSpec }) => {
const allowedOrigin = envSettings.clientHost;
const app = express();
app.use(generalApiLimiter);

12
server/src/config/services.js
View File

@@ -9,7 +9,6 @@ import BufferService from "../service/infrastructure/bufferService.js";
import StatusService from "../service/infrastructure/statusService.js";
import NotificationUtils from "../service/infrastructure/notificationUtils.js";
import NotificationService from "../service/infrastructure/notificationService.js";
import RedisService from "../service/data/redisService.js";
import ErrorService from "../service/infrastructure/errorService.js";
import SuperSimpleQueueHelper from "../service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js";
import SuperSimpleQueue from "../service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.js";
@@ -19,7 +18,6 @@ import DiagnosticService from "../service/business/diagnosticService.js";
import InviteService from "../service/business/inviteService.js";
import MaintenanceWindowService from "../service/business/maintenanceWindowService.js";
import MonitorService from "../service/business/monitorService.js";
import IORedis from "ioredis";
import papaparse from "papaparse";
import axios from "axios";
import ping from "ping";
@@ -35,19 +33,19 @@ import mjml2html from "mjml";
import jwt from "jsonwebtoken";
import crypto from "crypto";
export const initializeServices = async (appSettings, settingsService) => {
export const initializeServices = async (envSettings, settingsService) => {
const translationService = new TranslationService(logger);
await translationService.initialize();
const stringService = new StringService(translationService);
// Create DB
const db = new MongoDB({ appSettings });
const db = new MongoDB({ envSettings });
await db.connect();
const networkService = new NetworkService(axios, ping, logger, http, Docker, net, stringService, settingsService);
const emailService = new EmailService(settingsService, fs, path, compile, mjml2html, nodemailer, logger);
const bufferService = new BufferService({ db, logger });
const bufferService = new BufferService({ db, logger, envSettings });
const statusService = new StatusService({ db, logger, buffer: bufferService });
const notificationUtils = new NotificationUtils({
@@ -64,7 +62,6 @@ export const initializeServices = async (appSettings, settingsService) => {
notificationUtils,
});
const redisService = new RedisService({ Redis: IORedis, logger });
const errorService = new ErrorService();
const superSimpleQueueHelper = new SuperSimpleQueueHelper({
@@ -76,7 +73,7 @@ export const initializeServices = async (appSettings, settingsService) => {
});
const superSimpleQueue = await SuperSimpleQueue.create({
appSettings,
envSettings,
db,
logger,
helper: superSimpleQueueHelper,
@@ -135,7 +132,6 @@ export const initializeServices = async (appSettings, settingsService) => {
bufferService,
statusService,
notificationService,
redisService,
jobQueue: superSimpleQueue,
userService,
checkService,

8
server/src/db/mongo/MongoDB.js
View File

@@ -70,8 +70,8 @@ import * as diagnosticModule from "./modules/diagnosticModule.js";
class MongoDB {
static SERVICE_NAME = "MongoDB";
constructor({ appSettings }) {
this.appSettings = appSettings;
constructor({ envSettings }) {
this.envSettings = envSettings;
Object.assign(this, userModule);
Object.assign(this, inviteModule);
Object.assign(this, recoveryModule);
@@ -92,9 +92,9 @@ class MongoDB {
connect = async () => {
try {
const connectionString = this.appSettings.dbConnectionString || "mongodb://localhost:27017/uptime_db";
const connectionString = this.envSettings.dbConnectionString || "mongodb://localhost:27017/uptime_db";
await mongoose.connect(connectionString);
// If there are no AppSettings, create one
// If there are no AppSettings, create one // TODO why is this here?
await AppSettings.findOneAndUpdate(
{}, // empty filter to match any document
{}, // empty update

8
server/src/index.js
View File

@@ -20,10 +20,10 @@ const startApp = async () => {
const frontendPath = path.join(__dirname, "public");
// Create services
const settingsService = new SettingsService(AppSettings);
const appSettings = settingsService.loadSettings();
const envSettings = settingsService.loadSettings();
// Initialize services
const services = await initializeServices(appSettings, settingsService);
const services = await initializeServices(envSettings, settingsService);
// Initialize controllers
const controllers = initializeControllers(services);
@@ -31,12 +31,12 @@ const startApp = async () => {
const app = createApp({
services,
controllers,
appSettings,
envSettings,
frontendPath,
openApiSpec,
});
const port = appSettings.port || 52345;
const port = envSettings.port || 52345;
const server = app.listen(port, () => {
logger.info({ message: `Server started on port:${port}` });
});

64
server/src/service/data/redisService.js
View File

@@ -1,64 +0,0 @@
const SERVICE_NAME = "RedisService";
class RedisService {
static SERVICE_NAME = SERVICE_NAME;
constructor({ Redis, logger }) {
this.Redis = Redis;
this.connections = new Set();
this.logger = logger;
}
get serviceName() {
return RedisService.SERVICE_NAME;
}
getNewConnection(options = {}) {
const connection = new this.Redis(process.env.REDIS_URL, {
retryStrategy: (times) => {
return null;
},
...options,
});
this.connections.add(connection);
return connection;
}
async closeAllConnections() {
const closePromises = Array.from(this.connections).map((conn) =>
conn.quit().catch((err) => {
this.logger.error({
message: "Error closing Redis connection",
service: SERVICE_NAME,
method: "closeAllConnections",
details: { error: err },
});
})
);
await Promise.all(closePromises);
this.connections.clear();
this.logger.info({
message: "All Redis connections closed",
service: SERVICE_NAME,
method: "closeAllConnections",
});
}
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",
});
}
}
export default RedisService;

321
server/src/service/infrastructure/JobQueue/JobQueue.js
View File

@@ -1,321 +0,0 @@
const QUEUE_NAMES = ["uptime", "pagespeed", "hardware"];
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",
};
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 = [];
}
get serviceName() {
return JobQueue.SERVICE_NAME;
}
static async create({ db, jobQueueHelper, logger, stringService }) {
const instance = new JobQueue({ db, jobQueueHelper, logger, stringService });
await instance.init();
return instance;
}
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) {
this.logger.warn({
message: "Queue is stuck",
service: SERVICE_NAME,
method: "periodicHealthCheck",
details: 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",
});
}
pauseJob = async (monitor) => {
this.deleteJob(monitor);
};
resumeJob = async (monitor) => {
this.addJob(monitor._id, monitor);
};
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 updateJob(monitor) {
await this.deleteJob(monitor);
await this.addJob(monitor._id, monitor);
}
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;

308
server/src/service/infrastructure/JobQueue/JobQueueHelper.js
View File

@@ -1,308 +0,0 @@
const SERVICE_NAME = "JobQueueHelper";
class JobQueueHelper {
static SERVICE_NAME = SERVICE_NAME;
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;
}
get serviceName() {
return JobQueueHelper.SERVICE_NAME;
}
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.error({
message: `Worker ${queueName} is failed with msg: ${error.message}`,
service: error?.service ?? SERVICE_NAME,
method: error?.method ?? "createWorker:failed",
stack: error?.stack,
});
});
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 monitor = job.data;
const networkResponse = await this.networkService.getStatus(monitor);
// If the network response is not found, we're done
if (!networkResponse) {
await job.updateProgress(100);
return false;
}
// Handle status change
await job.updateProgress(60);
const { monitor: updatedMonitor, statusChanged, prevStatus } = await this.statusService.updateStatus(networkResponse);
// Handle notifications
await job.updateProgress(80);
this.notificationService
.handleNotifications({
...networkResponse,
monitor: updatedMonitor,
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) {
await job.updateProgress(100);
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;

213
server/src/service/infrastructure/PulseQueue/PulseQueue.js
View File

@@ -1,213 +0,0 @@
import { Pulse } from "@pulsecron/pulse";
const SERVICE_NAME = "JobQueue";
class PulseQueue {
static SERVICE_NAME = SERVICE_NAME;
constructor({ appSettings, db, pulseQueueHelper, logger }) {
this.db = db;
this.appSettings = appSettings;
this.pulseQueueHelper = pulseQueueHelper;
this.logger = logger;
}
get serviceName() {
return PulseQueue.SERVICE_NAME;
}
static async create({ appSettings, db, pulseQueueHelper, logger }) {
const instance = new PulseQueue({ appSettings, db, pulseQueueHelper, logger });
await instance.init();
return instance;
}
// ****************************************
// Core methods
// ****************************************
init = async () => {
try {
const mongoConnectionString = this.appSettings.dbConnectionString || "mongodb://localhost:27017/uptime_db";
this.pulse = new Pulse({ db: { address: mongoConnectionString } });
await this.pulse.start();
this.pulse.define("monitor-job", this.pulseQueueHelper.getMonitorJob(), {});
const monitors = await this.db.getAllMonitors();
for (const monitor of monitors) {
await this.addJob(monitor._id, monitor);
}
return true;
} catch (error) {
this.logger.error({
message: "Failed to initialize PulseQueue",
service: SERVICE_NAME,
method: "init",
details: error,
});
return false;
}
};
addJob = async (monitorId, monitor) => {
this.logger.debug({
message: `Adding job ${monitor?.url ?? "No URL"}`,
service: SERVICE_NAME,
method: "addJob",
});
const intervalInSeconds = monitor.interval / 1000;
const job = this.pulse.create("monitor-job", {
monitor,
});
job.unique({ "data.monitor._id": monitor._id });
job.attrs.jobId = monitorId.toString();
job.repeatEvery(`${intervalInSeconds} seconds`);
if (monitor.isActive === false) {
job.disable();
}
await job.save();
};
deleteJob = async (monitor) => {
this.logger.debug({
message: `Deleting job ${monitor?.url ?? "No URL"}`,
service: SERVICE_NAME,
method: "deleteJob",
});
await this.pulse.cancel({
"data.monitor._id": monitor._id,
});
};
pauseJob = async (monitor) => {
const result = await this.pulse.disable({
"data.monitor._id": monitor._id,
});
if (result.length < 1) {
throw new Error("Failed to pause monitor");
}
this.logger.debug({
message: `Paused monitor ${monitor._id}`,
service: SERVICE_NAME,
method: "pauseJob",
});
};
resumeJob = async (monitor) => {
const result = await this.pulse.enable({
"data.monitor._id": monitor._id,
});
if (result.length < 1) {
throw new Error("Failed to resume monitor");
}
this.logger.debug({
message: `Resumed monitor ${monitor._id}`,
service: SERVICE_NAME,
method: "resumeJob",
});
};
updateJob = async (monitor) => {
const jobs = await this.pulse.jobs({
"data.monitor._id": monitor._id,
});
const job = jobs[0];
if (!job) {
throw new Error("Job not found");
}
const intervalInSeconds = monitor.interval / 1000;
job.repeatEvery(`${intervalInSeconds} seconds`);
job.attrs.data.monitor = monitor;
await job.save();
};
shutdown = async () => {
this.logger.info({
message: "Shutting down JobQueue",
service: SERVICE_NAME,
method: "shutdown",
});
await this.pulse.stop();
};
// ****************************************
// Diagnostic methods
// ****************************************
getMetrics = async () => {
const jobs = await this.pulse.jobs();
const metrics = jobs.reduce(
(acc, job) => {
acc.totalRuns += job.attrs.runCount || 0;
acc.totalFailures += job.attrs.failCount || 0;
acc.jobs++;
if (job.attrs.failCount > 0 && job.attrs.failedAt >= job.attrs.lastFinishedAt) {
acc.failingJobs++;
}
if (job.attrs.lockedAt) {
acc.activeJobs++;
}
if (job.attrs.failCount > 0) {
acc.jobsWithFailures.push({
monitorId: job.attrs.data.monitor._id,
monitorUrl: job.attrs.data.monitor.url,
monitorType: job.attrs.data.monitor.type,
failedAt: job.attrs.failedAt,
failCount: job.attrs.failCount,
failReason: job.attrs.failReason,
});
}
return acc;
},
{
jobs: 0,
activeJobs: 0,
failingJobs: 0,
jobsWithFailures: [],
totalRuns: 0,
totalFailures: 0,
}
);
return metrics;
};
getJobs = async () => {
const jobs = await this.pulse.jobs();
return jobs.map((job) => {
return {
monitorId: job.attrs.data.monitor._id,
monitorUrl: job.attrs.data.monitor.url,
monitorType: job.attrs.data.monitor.type,
active: !job.attrs.disabled,
lockedAt: job.attrs.lockedAt,
runCount: job.attrs.runCount || 0,
failCount: job.attrs.failCount || 0,
failReason: job.attrs.failReason,
lastRunAt: job.attrs.lastRunAt,
lastFinishedAt: job.attrs.lastFinishedAt,
lastRunTook: job.attrs.lockedAt ? null : job.attrs.lastFinishedAt - job.attrs.lastRunAt,
lastFailedAt: job.attrs.failedAt,
};
});
};
flushQueues = async () => {
const cancelRes = await this.pulse.cancel();
await this.pulse.stop();
const initRes = await this.init();
return {
flushedJobs: cancelRes,
initSuccess: initRes,
};
};
obliterate = async () => {
await this.flushQueues();
};
}
export default PulseQueue;

104
server/src/service/infrastructure/PulseQueue/PulseQueueHelper.js
View File

@@ -1,104 +0,0 @@
const SERVICE_NAME = "PulseQueueHelper";
class PulseQueueHelper {
static SERVICE_NAME = SERVICE_NAME;
constructor({ db, logger, networkService, statusService, notificationService }) {
this.db = db;
this.logger = logger;
this.networkService = networkService;
this.statusService = statusService;
this.notificationService = notificationService;
}
get serviceName() {
return PulseQueueHelper.SERVICE_NAME;
}
getMonitorJob = () => {
return async (job) => {
try {
const monitor = job.attrs.data.monitor;
const monitorId = job.attrs.data.monitor._id;
if (!monitorId) {
throw new Error("No monitor id");
}
const maintenanceWindowActive = await this.isInMaintenanceWindow(monitorId);
if (maintenanceWindowActive) {
this.logger.info({
message: `Monitor ${monitorId} is in maintenance window`,
service: SERVICE_NAME,
method: "getMonitorJob",
});
return;
}
const networkResponse = await this.networkService.getStatus(monitor);
if (!networkResponse) {
throw new Error("No network response");
}
const { monitor: updatedMonitor, statusChanged, prevStatus } = await this.statusService.updateStatus(networkResponse);
this.notificationService
.handleNotifications({
...networkResponse,
monitor: updatedMonitor,
prevStatus,
statusChanged,
})
.catch((error) => {
this.logger.error({
message: error.message,
service: SERVICE_NAME,
method: "getMonitorJob",
details: `Error sending notifications for job ${job.id}: ${error.message}`,
stack: error.stack,
});
});
} catch (error) {
this.logger.warn({
message: error.message,
service: error.service || SERVICE_NAME,
method: error.method || "getMonitorJob",
stack: error.stack,
});
throw error;
}
};
};
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;
}
}
export default PulseQueueHelper;

10
server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.js
View File

@@ -4,8 +4,8 @@ const SERVICE_NAME = "JobQueue";
class SuperSimpleQueue {
static SERVICE_NAME = SERVICE_NAME;
constructor({ appSettings, db, logger, helper }) {
this.appSettings = appSettings;
constructor({ envSettings, db, logger, helper }) {
this.envSettings = envSettings;
this.db = db;
this.logger = logger;
this.helper = helper;
@@ -15,8 +15,8 @@ class SuperSimpleQueue {
return SuperSimpleQueue.SERVICE_NAME;
}
static async create({ appSettings, db, logger, helper }) {
const instance = new SuperSimpleQueue({ appSettings, db, logger, helper });
static async create({ envSettings, db, logger, helper }) {
const instance = new SuperSimpleQueue({ envSettings, db, logger, helper });
await instance.init();
return instance;
}
@@ -28,7 +28,7 @@ class SuperSimpleQueue {
// storeType: "redis",
logLevel: "debug",
debug: true,
// dbUri: this.appSettings.dbConnectionString,
// dbUri: this.envSettings.dbConnectionString,
});
this.scheduler.start();

8
server/src/service/infrastructure/bufferService.js
View File

@@ -1,5 +1,4 @@
const SERVICE_NAME = "BufferService";
const BUFFER_TIMEOUT = process.env.NODE_ENV === "development" ? 5000 : 1000 * 60 * 1; // 1 minute
const TYPE_MAP = {
http: "checks",
ping: "checks",
@@ -11,7 +10,8 @@ const TYPE_MAP = {
class BufferService {
static SERVICE_NAME = SERVICE_NAME;
constructor({ db, logger }) {
constructor({ db, logger, envSettings }) {
this.BUFFER_TIMEOUT = envSettings.nodeEnv === "development" ? 5000 : 1000 * 60 * 1; // 1 minute
this.db = db;
this.logger = logger;
this.SERVICE_NAME = SERVICE_NAME;
@@ -28,7 +28,7 @@ class BufferService {
this.scheduleNextFlush();
this.logger.info({
message: `Buffer service initialized, flushing every ${BUFFER_TIMEOUT / 1000}s`,
message: `Buffer service initialized, flushing every ${this.BUFFER_TIMEOUT / 1000}s`,
service: this.SERVICE_NAME,
method: "constructor",
});
@@ -66,7 +66,7 @@ class BufferService {
// Schedule the next flush only after the current one completes
this.scheduleNextFlush();
}
}, BUFFER_TIMEOUT);
}, this.BUFFER_TIMEOUT);
}
async flushBuffers() {
let items = 0;

25
server/src/service/system/settingsService.js
View File

@@ -3,20 +3,9 @@ const SERVICE_NAME = "SettingsService";
const envConfig = {
nodeEnv: process.env.NODE_ENV,
logLevel: process.env.LOG_LEVEL,
systemEmailHost: process.env.SYSTEM_EMAIL_HOST,
systemEmailPort: process.env.SYSTEM_EMAIL_PORT,
systemEmailUser: process.env.SYSTEM_EMAIL_USER,
systemEmailAddress: process.env.SYSTEM_EMAIL_ADDRESS,
systemEmailPassword: process.env.SYSTEM_EMAIL_PASSWORD,
jwtSecret: process.env.JWT_SECRET,
jwtTTL: process.env.TOKEN_TTL,
clientHost: process.env.CLIENT_HOST,
dbConnectionString: process.env.DB_CONNECTION_STRING,
redisUrl: process.env.REDIS_URL,
callbackUrl: process.env.CALLBACK_URL,
port: process.env.PORT,
pagespeedApiKey: process.env.PAGESPEED_API_KEY,
uprockApiKey: process.env.UPROCK_API_KEY,
clientHost: process.env.CLIENT_HOST, //*
dbConnectionString: process.env.DB_CONNECTION_STRING, //*
port: process.env.PORT, //*
};
/**
* SettingsService
@@ -46,13 +35,7 @@ class SettingsService {
loadSettings() {
return this.settings;
}
/**
* Reload settings by calling loadSettings.
* @returns {Promise<Object>} The reloaded settings.
*/
reloadSettings() {
return this.loadSettings();
}
/**
* Get the current settings.
* @returns {Object} The current settings.