diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 17ad26903..c2e536da5 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -151,6 +151,7 @@ LOG_LEVEL="debug" - `ORIGIN`: Origin for CORS purposes - `TOKEN_TTL`: Token time to live (in vercel/ms format) - `LOG_LEVEL`: Debug level (debug, info, warn, error) +- `QUEUE_TYPE`: Job queue implementation (`superSimpleQueue` or `lessSimpleQueue`, optional, default: `superSimpleQueue`) Start the backend server: diff --git a/server/.env.example b/server/.env.example index 20c579f5c..92798b56b 100644 --- a/server/.env.example +++ b/server/.env.example @@ -11,6 +11,7 @@ LOG_LEVEL=debug # Database # -------- DB_CONNECTION_STRING=mongodb://localhost:27017/uptime_db +QUEUE_TYPE=superSimpleQueue # JWT Authentication # ------------------ diff --git a/server/.prettierrc b/server/.prettierrc index 681669aa7..932d6330b 100755 --- a/server/.prettierrc +++ b/server/.prettierrc @@ -12,5 +12,14 @@ "quoteProps": "as-needed", "arrowParens": "always", "trailingComma": "es5", - "htmlWhitespaceSensitivity": "css" + "htmlWhitespaceSensitivity": "css", + "plugins": ["prettier-plugin-ini"], + "overrides": [ + { + "files": [".env", ".env.*"], + "options": { + "parser": "ini" + } + } + ] } diff --git a/server/openapi.json b/server/openapi.json index 26c58fa29..233aad746 100644 --- a/server/openapi.json +++ b/server/openapi.json @@ -392,6 +392,12 @@ "type": "string" } }, + "tags": { + "type": "array", + "items": { + "type": "string" + } + }, "secret": { "type": "string" }, @@ -469,6 +475,7 @@ "ignoreTlsErrors", "useAdvancedMatching", "notifications", + "tags", "cpuAlertThreshold", "memoryAlertThreshold", "diskAlertThreshold", @@ -2362,6 +2369,16 @@ "profilePicture": { "type": "string" }, + "password": { + "type": "string", + "minLength": 8, + "pattern": "^(?=.*[a-z])(?=.*[A-Z])(?=.*[0-9])(?=.*[!?@#$%^&*()\\-_=+[\\]{};:'\",.~`|\\\\/])[A-Za-z0-9!?@#$%^&*()\\-_=+[\\]{};:'\",.~`|\\\\/]+$" + }, + "newPassword": { + "type": "string", + "minLength": 8, + "pattern": "^(?=.*[a-z])(?=.*[A-Z])(?=.*[0-9])(?=.*[!?@#$%^&*()\\-_=+[\\]{};:'\",.~`|\\\\/])[A-Za-z0-9!?@#$%^&*()\\-_=+[\\]{};:'\",.~`|\\\\/]+$" + }, "profileImage": { "type": "string", "format": "binary" @@ -5207,7 +5224,8 @@ "type": "array", "items": { "type": "string" - } + }, + "minItems": 1 }, "duration": { "type": "number" @@ -5478,6 +5496,24 @@ "required": false, "name": "filter", "in": "query" + }, + { + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + }, + "required": false, + "name": "tags", + "in": "query" } ], "responses": { @@ -5672,6 +5708,24 @@ "name": "type", "in": "query" }, + { + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "array", + "items": { + "type": "string" + } + } + ] + }, + "required": false, + "name": "tags", + "in": "query" + }, { "schema": { "type": "boolean" @@ -6924,6 +6978,12 @@ "type": "string" } }, + "tags": { + "type": "array", + "items": { + "type": "string" + } + }, "secret": { "type": "string" }, @@ -7586,6 +7646,13 @@ }, "default": [] }, + "tags": { + "type": "array", + "items": { + "type": "string" + }, + "default": [] + }, "secret": { "type": "string" }, @@ -8104,6 +8171,12 @@ "type": "string" } }, + "tags": { + "type": "array", + "items": { + "type": "string" + } + }, "secret": { "type": "string" }, diff --git a/server/package-lock.json b/server/package-lock.json index 540efdacc..131589613 100644 --- a/server/package-lock.json +++ b/server/package-lock.json @@ -31,6 +31,7 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", + "less-simple-scheduler": "2.2.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", @@ -75,6 +76,7 @@ "lint-staged": "^16.2.7", "nodemon": "^3.1.11", "prettier": "^3.3.3", + "prettier-plugin-ini": "1.3.0", "ts-jest": "^29.4.6", "ts-node": "^10.9.2", "tsc-alias": "1.8.16", @@ -5468,16 +5470,42 @@ "license": "MIT" }, "node_modules/axios": { - "version": "1.15.2", - "resolved": "https://registry.npmjs.org/axios/-/axios-1.15.2.tgz", - "integrity": "sha512-wLrXxPtcrPTsNlJmKjkPnNPK2Ihe0hn0wGSaTEiHRPxwjvJwT3hKmXF4dpqxmPO9SoNb2FsYXj/xEo0gHN+D5A==", + "version": "1.16.1", + "resolved": "https://registry.npmjs.org/axios/-/axios-1.16.1.tgz", + "integrity": "sha512-caYkukvroVPO8KrzuJEb50Hm07KwfBZPEC3VeFHTsqWHvKTsy54hjJz9BS/cdaypROE2rH6xvm9mHX4fgWkr3A==", "license": "MIT", "dependencies": { - "follow-redirects": "^1.15.11", + "follow-redirects": "^1.16.0", "form-data": "^4.0.5", + "https-proxy-agent": "^5.0.1", "proxy-from-env": "^2.1.0" } }, + "node_modules/axios/node_modules/agent-base": { + "version": "6.0.2", + "resolved": "https://registry.npmjs.org/agent-base/-/agent-base-6.0.2.tgz", + "integrity": "sha512-RZNwNclF7+MS/8bDg70amg32dyeZGZxiDuQmZxKLAlQjr3jGyLx+4Kkk58UO7D2QdgFIQCovuSuZESne6RG6XQ==", + "license": "MIT", + "dependencies": { + "debug": "4" + }, + "engines": { + "node": ">= 6.0.0" + } + }, + "node_modules/axios/node_modules/https-proxy-agent": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-5.0.1.tgz", + "integrity": "sha512-dFcAjpTQFgoLMzC2VwU+C/CbS7uRL0lWmxDITmqm7C+7F0Odmj6s9l6alZc6AELXhrnggM2CeWSXHGOdX2YtwA==", + "license": "MIT", + "dependencies": { + "agent-base": "6", + "debug": "4" + }, + "engines": { + "node": ">= 6" + } + }, "node_modules/babel-jest": { "version": "30.2.0", "resolved": "https://registry.npmjs.org/babel-jest/-/babel-jest-30.2.0.tgz", @@ -10388,6 +10416,95 @@ "integrity": "sha512-Xq9nH7KlWZmXAtodXDDRE7vs6DU1gTU8zYDHDiWLSip45Egwq3plLHzPn27NgvzL2r1LMPC1vdqh98sQxtqj4A==", "license": "MIT" }, + "node_modules/less-simple-scheduler": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/less-simple-scheduler/-/less-simple-scheduler-2.2.0.tgz", + "integrity": "sha512-0bbFohlkcWZY2wfkGfDiYEbDxkshj0RuskkL0RLUDMn1Y3WjoPqy29PAnKNtnHG+DQqn0hMjZ0e3hVFJ4k76oA==", + "license": "ISC", + "dependencies": { + "mongodb": "7.2.0" + }, + "engines": { + "node": ">=18" + } + }, + "node_modules/less-simple-scheduler/node_modules/@types/whatwg-url": { + "version": "13.0.0", + "resolved": "https://registry.npmjs.org/@types/whatwg-url/-/whatwg-url-13.0.0.tgz", + "integrity": "sha512-N8WXpbE6Wgri7KUSvrmQcqrMllKZ9uxkYWMt+mCSGwNc0Hsw9VQTW7ApqI4XNrx6/SaM2QQJCzMPDEXE058s+Q==", + "license": "MIT", + "dependencies": { + "@types/webidl-conversions": "*" + } + }, + "node_modules/less-simple-scheduler/node_modules/bson": { + "version": "7.2.0", + "resolved": "https://registry.npmjs.org/bson/-/bson-7.2.0.tgz", + "integrity": "sha512-YCEo7KjMlbNlyHhz7zAZNDpIpQbd+wOEHJYezv0nMYTn4x31eIUM2yomNNubclAt63dObUzKHWsBLJ9QcZNSnQ==", + "license": "Apache-2.0", + "engines": { + "node": ">=20.19.0" + } + }, + "node_modules/less-simple-scheduler/node_modules/mongodb": { + "version": "7.2.0", + "resolved": "https://registry.npmjs.org/mongodb/-/mongodb-7.2.0.tgz", + "integrity": "sha512-F/2+BMZtLVhY30ioZp0dAmZ+IRZMBqI+nrv6t5+9/1AIwCa8sMRC3jBf81lpxMhnZgqq8CoUD503Z1oZWq1/sw==", + "license": "Apache-2.0", + "dependencies": { + "@mongodb-js/saslprep": "^1.3.0", + "bson": "^7.2.0", + "mongodb-connection-string-url": "^7.0.0" + }, + "engines": { + "node": ">=20.19.0" + }, + "peerDependencies": { + "@aws-sdk/credential-providers": "^3.806.0", + "@mongodb-js/zstd": "^7.0.0", + "gcp-metadata": "^7.0.1", + "kerberos": "^7.0.0", + "mongodb-client-encryption": ">=7.0.0 <7.1.0", + "snappy": "^7.3.2", + "socks": "^2.8.6" + }, + "peerDependenciesMeta": { + "@aws-sdk/credential-providers": { + "optional": true + }, + "@mongodb-js/zstd": { + "optional": true + }, + "gcp-metadata": { + "optional": true + }, + "kerberos": { + "optional": true + }, + "mongodb-client-encryption": { + "optional": true + }, + "snappy": { + "optional": true + }, + "socks": { + "optional": true + } + } + }, + "node_modules/less-simple-scheduler/node_modules/mongodb-connection-string-url": { + "version": "7.0.1", + "resolved": "https://registry.npmjs.org/mongodb-connection-string-url/-/mongodb-connection-string-url-7.0.1.tgz", + "integrity": "sha512-h0AZ9A7IDVwwHyMxmdMXKy+9oNlF0zFoahHiX3vQ8e3KFcSP3VmsmfvtRSuLPxmyv2vjIDxqty8smTgie/SNRQ==", + "license": "Apache-2.0", + "dependencies": { + "@types/whatwg-url": "^13.0.0", + "whatwg-url": "^14.1.0" + }, + "engines": { + "node": ">=20.19.0" + } + }, "node_modules/leven": { "version": "3.1.0", "resolved": "https://registry.npmjs.org/leven/-/leven-3.1.0.tgz", @@ -12991,6 +13108,16 @@ "url": "https://github.com/prettier/prettier?sponsor=1" } }, + "node_modules/prettier-plugin-ini": { + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/prettier-plugin-ini/-/prettier-plugin-ini-1.3.0.tgz", + "integrity": "sha512-kX1KnPwITCTtYV0MTBZQKuFLxcNcbHOS9r7Ta1jHLLvRvZRTzHHqX0MEZMrCkedjLuxpQSgzr4aC5IRd+R3FPw==", + "dev": true, + "license": "MIT", + "dependencies": { + "prettier": ">=3.0.0-alpha.3" + } + }, "node_modules/pretty-format": { "version": "30.2.0", "resolved": "https://registry.npmjs.org/pretty-format/-/pretty-format-30.2.0.tgz", diff --git a/server/package.json b/server/package.json index 2382c5cc9..ea518a298 100755 --- a/server/package.json +++ b/server/package.json @@ -51,6 +51,7 @@ "jmespath": "^0.16.0", "jsdom": "^26.1.0", "jsonwebtoken": "9.0.2", + "less-simple-scheduler": "2.2.0", "mailersend": "^2.2.0", "mjml": "^5.0.0-alpha.4", "mongoose": "^8.3.3", @@ -95,6 +96,7 @@ "lint-staged": "^16.2.7", "nodemon": "^3.1.11", "prettier": "^3.3.3", + "prettier-plugin-ini": "1.3.0", "ts-jest": "^29.4.6", "ts-node": "^10.9.2", "tsc-alias": "1.8.16", diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 4eb6c1081..04966d6e2 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -7,8 +7,9 @@ import { EmailService, BufferService, GlobalPingService, + LessSimpleQueue, SuperSimpleQueue, - SuperSimpleQueueHelper, + QueueHelper, NotificationsService, TagsService, StatusService, @@ -37,7 +38,7 @@ import { INetworkService, IEmailService, IBufferService, - ISuperSimpleQueue, + IJobQueue, INotificationsService, ITagsService, IStatusService, @@ -131,7 +132,7 @@ export type InitializedServices = { emailService: IEmailService; bufferService: IBufferService; statusService: IStatusService; - jobQueue: ISuperSimpleQueue; + jobQueue: IJobQueue; userService: IUserService; checkService: ICheckService; geoChecksService: IGeoChecksService; @@ -325,7 +326,7 @@ export const initializeServices = async ({ const tagsService = new TagsService(tagsRepository, monitorsRepository); - const superSimpleQueueHelper = new SuperSimpleQueueHelper( + const queueHelper = new QueueHelper( logger, networkService, statusService, @@ -344,7 +345,14 @@ export const initializeServices = async ({ geoChecksRepository ); - const superSimpleQueue = await SuperSimpleQueue.create(logger, superSimpleQueueHelper, monitorsRepository); + let jobQueue: IJobQueue; + switch (envSettings.queueType) { + case "lessSimpleQueue": + jobQueue = await LessSimpleQueue.create(logger, queueHelper, monitorsRepository, envSettings); + break; + default: + jobQueue = await SuperSimpleQueue.create(logger, queueHelper, monitorsRepository); + } // Business services const userService = new UserService({ @@ -353,7 +361,7 @@ export const initializeServices = async ({ settingsService, logger, jwt, - jobQueue: superSimpleQueue, + jobQueue, monitorsRepository, usersRepository, invitesRepository, @@ -373,7 +381,7 @@ export const initializeServices = async ({ maintenanceWindowsRepository, }); const monitorService = new MonitorService({ - jobQueue: superSimpleQueue, + jobQueue, logger, games, monitorsRepository, @@ -393,7 +401,7 @@ export const initializeServices = async ({ emailService, bufferService, statusService, - jobQueue: superSimpleQueue, + jobQueue, userService, checkService, geoChecksService, diff --git a/server/src/service/business/incidentService.ts b/server/src/service/business/incidentService.ts index 2070a8937..1606cc95d 100644 --- a/server/src/service/business/incidentService.ts +++ b/server/src/service/business/incidentService.ts @@ -5,7 +5,7 @@ import { AppError } from "@/utils/AppError.js"; import { getDateForRange } from "@/utils/dataUtils.js"; import type { IIncidentsRepository, IMonitorsRepository, IUsersRepository } from "@/repositories/index.js"; import type { Incident, IncidentSummary, User } from "@/types/index.js"; -import type { MonitorActionDecision } from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js"; +import type { MonitorActionDecision } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import type { INotificationMessageBuilder } from "@/service/infrastructure/notificationMessageBuilder.js"; import type { ILogger } from "@/utils/logger.js"; diff --git a/server/src/service/business/monitorService.ts b/server/src/service/business/monitorService.ts index 2023a1d7a..46b81a2ca 100644 --- a/server/src/service/business/monitorService.ts +++ b/server/src/service/business/monitorService.ts @@ -23,7 +23,7 @@ import type { import demoMonitorsData from "@/utils/demoMonitors.json" with { type: "json" }; import { AppError } from "@/utils/AppError.js"; import type { ImportedMonitor } from "@/validation/monitorValidation.js"; -import { ISuperSimpleQueue } from "../infrastructure/SuperSimpleQueue/SuperSimpleQueue.js"; +import { IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; import { ILogger } from "@/utils/logger.js"; const SERVICE_NAME = "MonitorService"; @@ -96,7 +96,7 @@ export interface IMonitorService { export class MonitorService implements IMonitorService { static SERVICE_NAME = SERVICE_NAME; - private jobQueue: ISuperSimpleQueue; + private jobQueue: IJobQueue; private logger: ILogger; private games: GamesMap; private monitorsRepository: IMonitorsRepository; @@ -117,7 +117,7 @@ export class MonitorService implements IMonitorService { statusPagesRepository, incidentsRepository, }: { - jobQueue: ISuperSimpleQueue; + jobQueue: IJobQueue; logger: ILogger; games: GamesMap; monitorsRepository: IMonitorsRepository; diff --git a/server/src/service/business/userService.ts b/server/src/service/business/userService.ts index 7acca5742..8b2c52159 100644 --- a/server/src/service/business/userService.ts +++ b/server/src/service/business/userService.ts @@ -10,7 +10,7 @@ import type { User } from "@/types/index.js"; import { canManageRole, type UserRole } from "@/types/user.js"; import bcrypt from "bcryptjs"; import { AppError } from "@/utils/AppError.js"; -import { ISuperSimpleQueue } from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.js"; +import { IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; import { IEmailService } from "@/service/infrastructure/emailService.js"; import { EnvConfig, ISettingsService } from "@/service/system/settingsService.js"; import { ILogger } from "@/utils/logger.js"; @@ -55,7 +55,7 @@ export class UserService implements IUserService { private settingsService: ISettingsService; private logger: ILogger; private jwt: JwtType; - private jobQueue: ISuperSimpleQueue; + private jobQueue: IJobQueue; private crypto: CryptoType; private monitorsRepository: IMonitorsRepository; private usersRepository: IUsersRepository; @@ -83,7 +83,7 @@ export class UserService implements IUserService { settingsService: ISettingsService; logger: ILogger; jwt: JwtType; - jobQueue: ISuperSimpleQueue; + jobQueue: IJobQueue; monitorsRepository: IMonitorsRepository; usersRepository: IUsersRepository; invitesRepository: IInvitesRepository; diff --git a/server/src/service/index.ts b/server/src/service/index.ts index 9bac381f6..f94f82a19 100644 --- a/server/src/service/index.ts +++ b/server/src/service/index.ts @@ -10,8 +10,10 @@ export * from "@/service/business/statusPageService.js"; export * from "@/service/business/userService.js"; // Infrastructure services -export * from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.js"; -export * from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js"; +export * from "@/service/infrastructure/JobQueues/IJobQueue.js"; +export * from "@/service/infrastructure/JobQueues/SuperSimpleQueue.js"; +export * from "@/service/infrastructure/JobQueues/LessSimpleQueue.js"; +export * from "@/service/infrastructure/JobQueues/QueueHelper.js"; export * from "@/service/infrastructure/notificationMessageBuilder.js"; export * from "@/service/infrastructure/bufferService.js"; export * from "@/service/infrastructure/emailService.js"; diff --git a/server/src/service/infrastructure/JobQueues/IJobQueue.ts b/server/src/service/infrastructure/JobQueues/IJobQueue.ts new file mode 100644 index 000000000..a97c8eb6a --- /dev/null +++ b/server/src/service/infrastructure/JobQueues/IJobQueue.ts @@ -0,0 +1,52 @@ +import { Monitor } from "@/types/monitor.js"; + +export type QueueJobFailure = { + monitorId: string | number; + monitorUrl: string | null; + monitorType: string | null; + failedAt: number | null; + failCount: number; + failReason: string | null; +}; + +export type QueueMetrics = { + jobs: number; + activeJobs: number; + failingJobs: number; + jobsWithFailures: QueueJobFailure[]; + totalRuns: number; + totalFailures: number; +}; + +export type QueueJobSummary = { + monitorId: string | number; + monitorUrl: string | null; + monitorType: string | null; + monitorInterval: number | null; + monitorGeoInterval: number | null; + monitorActive: boolean | null; + active: boolean; + lockedAt: number | null; + runCount: number; + failCount: number; + failReason: string | null; + lastRunAt: number | null; + lastFinishedAt: number | null; + lastRunTook: number | null; + lastFailedAt: number | null; + repeat: number | null; +}; + +export interface IJobQueue { + readonly serviceName: string; + init(): Promise; + addJob(monitorId: string, monitor: Monitor): Promise; + deleteJob(monitor: Monitor): Promise; + pauseJob(monitor: Monitor): Promise; + resumeJob(monitor: Monitor): Promise; + updateJob(monitor: Monitor): Promise; + shutdown(): Promise; + getMetrics(): Promise; + getJobs(): Promise; + flushQueues(): Promise<{ success: boolean }>; +} diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts new file mode 100644 index 000000000..954f6ca91 --- /dev/null +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -0,0 +1,356 @@ +import { IMonitorsRepository } from "@/repositories/index.js"; +import { ILogger } from "@/utils/logger.js"; +import { MongoStore, Scheduler, IJob } from "less-simple-scheduler"; +import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; +import { Monitor, supportsGeoCheck } from "@/types/monitor.js"; +import { type QueueMetrics, IJobQueue } from "@/service/infrastructure/JobQueues/IJobQueue.js"; +import { EnvConfig } from "@/service/system/settingsService.js"; +const SERVICE_NAME = "JobQueue"; + +// For discriminating job types +type MonitorJob = Omit & { data: Monitor }; +const MONITOR_TEMPLATES = new Set(["monitor-job", "geo-check-job"]); +const isMonitorJob = (job: IJob): job is MonitorJob => MONITOR_TEMPLATES.has(job.template); + +export class LessSimpleQueue implements IJobQueue { + static SERVICE_NAME = SERVICE_NAME; + + private logger: ILogger; + private helper: IQueueHelper; + private monitorsRepository: IMonitorsRepository; + private readonly scheduler: Scheduler; + + constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { + this.logger = logger; + this.helper = helper; + this.monitorsRepository = monitorsRepository; + this.scheduler = scheduler; + } + + get serviceName() { + return LessSimpleQueue.SERVICE_NAME; + } + + private registerListeners = () => { + this.scheduler.on("scheduler:start", () => { + this.logger.info({ + message: "Scheduler started", + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("scheduler:stop", () => { + this.logger.info({ + message: "Scheduler stopped", + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("scheduler:error", (workerId, error) => { + this.logger.error({ + message: `w${workerId} error: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:locked", (workerId, job) => { + this.logger.debug({ + message: `w${workerId} ${job.id} locked`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:abort", (workerId, job, reason) => { + this.logger.warn({ + message: `w${workerId} ${job.id} aborted: ${reason}`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:attempt", (workerId, job, attempt) => { + this.logger.debug({ + message: `w${workerId} ${job.id} attempt ${attempt}`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:complete", (workerId, job) => { + this.logger.debug({ + message: `w${workerId} ${job.id} completed successfully`, + service: SERVICE_NAME, + }); + }); + + this.scheduler.on("job:exhausted", (workerId, job, error) => { + this.logger.error({ + message: `w${workerId} ${job.id} exhausted all retries: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:fail", (workerId, job, error, attempt) => { + this.logger.warn({ + message: `w${workerId} ${job.id} failed on attempt ${attempt}: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + stack: error instanceof Error ? error.stack : undefined, + }); + }); + + this.scheduler.on("job:start", (workerId, job) => { + this.logger.debug({ + message: `w${workerId} ${job.id} started`, + service: SERVICE_NAME, + }); + }); + }; + + static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, envSettings: EnvConfig) { + const store = new MongoStore({ + url: envSettings.dbConnectionString ?? "mongodb://localhost:27017/uptime_db", + }); + const scheduler = new Scheduler(store, { + concurrency: 50, + }); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, scheduler); + await instance.init(); + + return instance; + } + + init = async () => { + try { + this.registerListeners(); + await this.scheduler.start(); + this.scheduler.addTemplate("monitor-job", this.helper.getHeartbeatJob()); + this.scheduler.addTemplate("geo-check-job", this.helper.getHeartbeatGeoJob()); + this.scheduler.addTemplate("cleanup-orphaned", this.helper.getCleanupOrphanedJob()); + this.scheduler.addTemplate("cleanup-retention-job", this.helper.getCleanupRetentionJob()); + const monitors = await this.monitorsRepository.findAll(); + if (!monitors) { + return true; + } + for (const monitor of monitors) { + const randomOffset = Math.floor(Math.random() * 100); + setTimeout(() => { + this.addJob(monitor.id, monitor); + }, randomOffset); + } + + this.scheduler.addJob({ id: "cleanup-orphaned", template: "cleanup-orphaned", active: true, upsert: true }); + this.scheduler.addJob({ id: "cleanup-retention", template: "cleanup-retention-job", active: true, repeat: 24 * 60 * 60 * 1000, upsert: true }); + + return true; + } catch (error: unknown) { + this.logger.error({ + message: `Failed to initialize LessSimpleQueue: ${error instanceof Error ? error.message : String(error)}`, + service: SERVICE_NAME, + method: "init", + }); + return false; + } + }; + + addJob = async (monitorId: string, monitor: Monitor) => { + this.scheduler.addJob({ + id: monitorId, + template: "monitor-job", + repeat: monitor.interval, + active: monitor.isActive, + data: monitor, + upsert: true, + }); + + // Return early if we don't need geo checks + if (!supportsGeoCheck(monitor.type)) { + return; + } + + // Add geo check job if enabled for HTTP monitors + if (monitor.geoCheckEnabled) { + this.scheduler.addJob({ + id: `${monitorId}-geo`, + template: "geo-check-job", + repeat: monitor.geoCheckInterval, + active: monitor.isActive, + data: monitor, + upsert: true, + }); + } + }; + + deleteJob = async (monitor: Monitor) => { + this.scheduler.removeJob(monitor.id); + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) await this.scheduler.removeJob(`${monitor.id}-geo`); + }; + + pauseJob = async (monitor: Monitor) => { + const result = await this.scheduler.pauseJob(monitor.id); + if (result === false) { + throw new Error("Failed to pause monitor"); + } + this.scheduler.updateJob(monitor.id, { data: monitor }); + + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) { + const geoRes = await this.scheduler.pauseJob(`${monitor.id}-geo`); + if (geoRes === false) { + this.logger.error({ + message: `Failed to pause geo check job for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "pauseJob", + }); + } + this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); + } + + this.logger.debug({ + message: `Paused monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "pauseJob", + }); + }; + + resumeJob = async (monitor: Monitor) => { + const result = await this.scheduler.resumeJob(monitor.id); + if (result === false) { + throw new Error("Failed to resume monitor"); + } + this.scheduler.updateJob(monitor.id, { data: monitor }); + + const geoJob = await this.scheduler.getJob(`${monitor.id}-geo`); + if (geoJob) { + const geoRes = await this.scheduler.resumeJob(`${monitor.id}-geo`); + if (geoRes === false) { + this.logger.error({ + message: `Failed to resume geo check job for monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "resumeJob", + }); + } + this.scheduler.updateJob(`${monitor.id}-geo`, { data: monitor }); + } + this.logger.debug({ + message: `Resumed monitor ${monitor.id}`, + service: SERVICE_NAME, + method: "resumeJob", + }); + }; + + private syncGeoJob = async (monitor: Monitor) => { + const geoJobId = `${monitor.id}-geo`; + const existingGeoJob = await this.scheduler.getJob(geoJobId); + + // If geoChecks have been disabled, or the monitor type doesn't support them, remove + if (!monitor.geoCheckEnabled || !supportsGeoCheck(monitor.type)) { + if (existingGeoJob) this.scheduler.removeJob(geoJobId); + return; + } + + // If the job exists, update it + if (existingGeoJob) { + this.scheduler.updateJob(geoJobId, { repeat: monitor.geoCheckInterval, active: monitor.isActive, data: monitor }); + return; + } + + // Otherwise, create it + this.scheduler.addJob({ + id: geoJobId, + template: "geo-check-job", + repeat: monitor.geoCheckInterval, + active: monitor.isActive, + data: monitor, + upsert: true, + }); + }; + + updateJob = async (monitor: Monitor) => { + this.scheduler.updateJob(monitor.id, { repeat: monitor.interval, data: monitor }); + await this.syncGeoJob(monitor); + }; + + shutdown = async () => { + this.scheduler.stop(); + }; + + getMetrics = async () => { + const jobs = await this.scheduler.getJobs(); + const metrics = jobs.reduce( + (acc, job) => { + const runCount = job.runCount ?? 0; + const failCount = job.failCount ?? 0; + const lastFailedAt = job.lastFailedAt ?? 0; + const lastFinishedAt = job.lastFinishedAt ?? 0; + acc.totalRuns += runCount; + acc.totalFailures += failCount; + acc.jobs++; + if (failCount > 0 && lastFailedAt >= lastFinishedAt) { + acc.failingJobs++; + } + + if (job.lockedAt) { + acc.activeJobs++; + } + + if (failCount > 0) { + const monitor = isMonitorJob(job) ? job.data : undefined; + acc.jobsWithFailures.push({ + monitorId: job.id, + monitorUrl: monitor?.url || null, + monitorType: monitor?.type || null, + failedAt: job.lastFailedAt ?? null, + failCount, + failReason: job.lastError ?? null, + }); + } + return acc; + }, + { + jobs: 0, + activeJobs: 0, + failingJobs: 0, + jobsWithFailures: [], + totalRuns: 0, + totalFailures: 0, + } + ); + return metrics; + }; + + getJobs = async () => { + const jobs = await this.scheduler.getJobs(); + return jobs.map((job) => { + const monitor = isMonitorJob(job) ? job.data : undefined; + + return { + monitorId: job.id, + monitorUrl: monitor?.url || null, + monitorType: monitor?.type || null, + monitorInterval: monitor?.interval || null, + monitorGeoInterval: monitor?.geoCheckInterval || null, + monitorActive: monitor?.isActive ?? null, + active: job.active, + repeat: job.repeat ?? null, + lockedAt: job.lockedAt ?? null, + runCount: job.runCount ?? 0, + failCount: job.failCount ?? 0, + failReason: job.lastError ?? null, + lastRunAt: job.lastStartedAt ?? null, + lastFinishedAt: job.lastFinishedAt ?? null, + lastRunTook: job.lockedAt || !job.lastFinishedAt || !job.lastStartedAt ? null : job.lastFinishedAt - job.lastStartedAt, + lastFailedAt: job.lastFailedAt ?? null, + }; + }); + }; + + flushQueues = async () => { + const stopRes = await this.scheduler.stop(); + const flushRes = await this.scheduler.flushJobs(); + const initRes = await this.init(); + return { + success: Boolean(stopRes && flushRes && initRes), + }; + }; +} diff --git a/server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts b/server/src/service/infrastructure/JobQueues/QueueHelper.ts similarity index 98% rename from server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts rename to server/src/service/infrastructure/JobQueues/QueueHelper.ts index 603379190..ea86feae0 100644 --- a/server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts +++ b/server/src/service/infrastructure/JobQueues/QueueHelper.ts @@ -25,7 +25,7 @@ import { import { ILogger } from "@/utils/logger.js"; import { IBufferService } from "@/service/index.js"; -export interface ISuperSimpleQueueHelper { +export interface IQueueHelper { readonly serviceName: string; getHeartbeatJob(): (monitor: Monitor) => Promise; getHeartbeatGeoJob(): (monitor: Monitor) => Promise; @@ -48,7 +48,7 @@ export interface MonitorActionDecision { }; } -export class SuperSimpleQueueHelper implements ISuperSimpleQueueHelper { +export class QueueHelper implements IQueueHelper { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; @@ -105,7 +105,7 @@ export class SuperSimpleQueueHelper implements ISuperSimpleQueueHelper { } get serviceName() { - return SuperSimpleQueueHelper.SERVICE_NAME; + return QueueHelper.SERVICE_NAME; } getHeartbeatJob = () => { diff --git a/server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts similarity index 96% rename from server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.ts rename to server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts index 8ecfecc65..b98ecb2db 100644 --- a/server/src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/SuperSimpleQueue.ts @@ -1,7 +1,7 @@ import { IMonitorsRepository } from "@/repositories/index.js"; import { ILogger } from "@/utils/logger.js"; import Scheduler from "super-simple-scheduler"; -import { ISuperSimpleQueueHelper } from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js"; +import { IQueueHelper } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import { Monitor, MonitorType, supportsGeoCheck } from "@/types/monitor.js"; const SERVICE_NAME = "JobQueue"; @@ -60,11 +60,11 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { static SERVICE_NAME = SERVICE_NAME; private logger: ILogger; - private helper: ISuperSimpleQueueHelper; + private helper: IQueueHelper; private monitorsRepository: IMonitorsRepository; private readonly scheduler: Scheduler; - constructor(logger: ILogger, helper: ISuperSimpleQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { + constructor(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository, scheduler: Scheduler) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; @@ -143,7 +143,7 @@ export class SuperSimpleQueue implements ISuperSimpleQueue { }); }; - static async create(logger: ILogger, helper: ISuperSimpleQueueHelper, monitorsRepository: IMonitorsRepository) { + static async create(logger: ILogger, helper: IQueueHelper, monitorsRepository: IMonitorsRepository) { const scheduler = new Scheduler({ concurrency: 50, // storeType: "mongo", diff --git a/server/src/service/infrastructure/notificationMessageBuilder.ts b/server/src/service/infrastructure/notificationMessageBuilder.ts index 934163b2a..bc1d7ac82 100644 --- a/server/src/service/infrastructure/notificationMessageBuilder.ts +++ b/server/src/service/infrastructure/notificationMessageBuilder.ts @@ -1,5 +1,5 @@ import type { HardwareStatusPayload, Monitor, MonitorStatusResponse } from "@/types/index.js"; -import type { MonitorActionDecision } from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js"; +import type { MonitorActionDecision } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import type { NotificationMessage, NotificationType, diff --git a/server/src/service/infrastructure/notificationsService.ts b/server/src/service/infrastructure/notificationsService.ts index 90e9d637a..d1956a724 100644 --- a/server/src/service/infrastructure/notificationsService.ts +++ b/server/src/service/infrastructure/notificationsService.ts @@ -2,7 +2,7 @@ import type { Monitor, MonitorStatusResponse, Notification } from "@/types/index import type { NotificationMessage } from "@/types/notificationMessage.js"; import { IMonitorsRepository, INotificationsRepository } from "@/repositories/index.js"; import { INotificationProvider } from "./notificationProviders/INotificationProvider.js"; -import type { MonitorActionDecision } from "@/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.js"; +import type { MonitorActionDecision } from "@/service/infrastructure/JobQueues/QueueHelper.js"; import type { ISettingsService } from "@/service/system/settingsService.js"; import { ILogger } from "@/utils/logger.js"; import type { INotificationMessageBuilder } from "@/service/infrastructure/notificationMessageBuilder.js"; diff --git a/server/src/service/system/settingsService.ts b/server/src/service/system/settingsService.ts index 32b95d261..97960df3a 100755 --- a/server/src/service/system/settingsService.ts +++ b/server/src/service/system/settingsService.ts @@ -1,5 +1,5 @@ import { ISettingsRepository } from "@/repositories/index.js"; -import { type DbType, Settings, SettingsUpdate } from "@/types/index.js"; +import { type DbType, type QueueType, Settings, SettingsUpdate } from "@/types/index.js"; import { AppError } from "@/utils/AppError.js"; import { ValidatedEnv } from "@/validation/envValidation.js"; import type { StringValue } from "ms"; @@ -13,6 +13,7 @@ export type EnvConfig = { clientHost: string; dbConnectionString: string; dbType: DbType; + queueType: QueueType; statusPageThemesEnabled: boolean; }; @@ -39,6 +40,7 @@ export class SettingsService implements ISettingsService { clientHost: env.CLIENT_HOST, dbConnectionString: env.DB_CONNECTION_STRING, dbType: env.DB_TYPE, + queueType: env.QUEUE_TYPE, statusPageThemesEnabled: env.STATUS_PAGE_THEMES_ENABLED, }; } diff --git a/server/src/types/settings.ts b/server/src/types/settings.ts index 3d0d0d931..7bbf57401 100644 --- a/server/src/types/settings.ts +++ b/server/src/types/settings.ts @@ -1,6 +1,9 @@ export const DbTypes = ["mongodb"] as const; export type DbType = (typeof DbTypes)[number]; +export const QueueTypes = ["lessSimpleQueue", "superSimpleQueue"] as const; +export type QueueType = (typeof QueueTypes)[number]; + export interface SettingsThresholds { cpu?: number; memory?: number; diff --git a/server/src/validation/envValidation.ts b/server/src/validation/envValidation.ts index 3adf4c90c..7eea70fc5 100644 --- a/server/src/validation/envValidation.ts +++ b/server/src/validation/envValidation.ts @@ -1,5 +1,5 @@ import { z } from "zod"; -import { DbTypes } from "@/types/settings.js"; +import { DbTypes, QueueTypes } from "@/types/settings.js"; import { booleanCoercion } from "./shared.js"; const envSchema = z.object({ @@ -11,6 +11,7 @@ const envSchema = z.object({ // Database DB_CONNECTION_STRING: z.string().min(1, "Database connection string is required"), DB_TYPE: z.enum(DbTypes).default("mongodb"), + QUEUE_TYPE: z.enum(QueueTypes).default("superSimpleQueue"), // JWT Authentication JWT_SECRET: z.string().min(1, "JWT_SECRET is required"), diff --git a/server/test/helpers/heartbeatTestHarness.ts b/server/test/helpers/heartbeatTestHarness.ts index ec0c3714e..9b0592afc 100644 --- a/server/test/helpers/heartbeatTestHarness.ts +++ b/server/test/helpers/heartbeatTestHarness.ts @@ -1,5 +1,5 @@ import { jest } from "@jest/globals"; -import { SuperSimpleQueueHelper } from "../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts"; +import { QueueHelper } from "../../src/service/infrastructure/JobQueues/QueueHelper.ts"; import { StatusService } from "../../src/service/infrastructure/statusService.ts"; import { IncidentService } from "../../src/service/business/incidentService.ts"; import { InMemoryMonitorsRepository } from "./InMemoryMonitorsRepository.ts"; @@ -143,7 +143,7 @@ export function createHeartbeatTestHarness(): HeartbeatTestHarness { const maintenanceWindowsRepo = { findByMonitorId: jest.fn().mockResolvedValue([]) }; - const helper = new SuperSimpleQueueHelper( + const helper = new QueueHelper( logger, networkService as any, statusService as any, diff --git a/server/test/integration/incidentLifecycle.test.ts b/server/test/integration/incidentLifecycle.test.ts index d2429a812..0fe1ddcf8 100644 --- a/server/test/integration/incidentLifecycle.test.ts +++ b/server/test/integration/incidentLifecycle.test.ts @@ -5,7 +5,7 @@ import { createMockLogger } from "../helpers/createMockLogger.ts"; import type { IMonitorsRepository, IUsersRepository } from "../../src/repositories/index.ts"; import type { INotificationMessageBuilder } from "../../src/service/infrastructure/notificationMessageBuilder.ts"; import type { Monitor } from "../../src/types/monitor.ts"; -import type { MonitorActionDecision } from "../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts"; +import type { MonitorActionDecision } from "../../src/service/infrastructure/JobQueues/QueueHelper.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── diff --git a/server/test/unit/services/incidentService.test.ts b/server/test/unit/services/incidentService.test.ts index 0a17cbd79..4274f67d8 100644 --- a/server/test/unit/services/incidentService.test.ts +++ b/server/test/unit/services/incidentService.test.ts @@ -5,7 +5,7 @@ import type { IIncidentsRepository, IMonitorsRepository, IUsersRepository } from import type { INotificationMessageBuilder } from "../../../src/service/infrastructure/notificationMessageBuilder.ts"; import type { Monitor } from "../../../src/types/monitor.ts"; import type { Incident } from "../../../src/types/index.ts"; -import type { MonitorActionDecision } from "../../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts"; +import type { MonitorActionDecision } from "../../../src/service/infrastructure/JobQueues/QueueHelper.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── diff --git a/server/test/unit/services/notificationMessageBuilder.test.ts b/server/test/unit/services/notificationMessageBuilder.test.ts index d10bd1604..4403a8bb3 100644 --- a/server/test/unit/services/notificationMessageBuilder.test.ts +++ b/server/test/unit/services/notificationMessageBuilder.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it, beforeEach } from "@jest/globals"; import { NotificationMessageBuilder } from "../../../src/service/infrastructure/notificationMessageBuilder.ts"; import type { Monitor, MonitorStatusResponse, HardwareStatusPayload } from "../../../src/types/index.ts"; -import type { MonitorActionDecision } from "../../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts"; +import type { MonitorActionDecision } from "../../../src/service/infrastructure/JobQueues/QueueHelper.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── diff --git a/server/test/unit/services/notificationsService.test.ts b/server/test/unit/services/notificationsService.test.ts index cdc1ad320..e8c3a0d97 100644 --- a/server/test/unit/services/notificationsService.test.ts +++ b/server/test/unit/services/notificationsService.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it, jest, beforeEach } from "@jest/globals"; import { NotificationsService } from "../../../src/service/infrastructure/notificationsService.ts"; import { createMockLogger } from "../../helpers/createMockLogger.ts"; import type { Notification, Monitor, MonitorStatusResponse } from "../../../src/types/index.ts"; -import type { MonitorActionDecision } from "../../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts"; +import type { MonitorActionDecision } from "../../../src/service/infrastructure/JobQueues/QueueHelper.ts"; // ── Helpers ────────────────────────────────────────────────────────────────── diff --git a/server/test/unit/services/superSimpleQueue.test.ts b/server/test/unit/services/superSimpleQueue.test.ts index 685dd71c7..660d0b8f7 100644 --- a/server/test/unit/services/superSimpleQueue.test.ts +++ b/server/test/unit/services/superSimpleQueue.test.ts @@ -35,7 +35,7 @@ jest.unstable_mockModule("super-simple-scheduler", () => ({ default: MockScheduler, })); -const { SuperSimpleQueue } = await import("../../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueue.ts"); +const { SuperSimpleQueue } = await import("../../../src/service/infrastructure/JobQueues/SuperSimpleQueue.ts"); const createQueueHelper = () => ({ getHeartbeatJob: jest.fn().mockReturnValue(() => Promise.resolve()), diff --git a/server/test/unit/services/superSimpleQueueHelper.test.ts b/server/test/unit/services/superSimpleQueueHelper.test.ts index 4d8ecfbc0..7075051eb 100644 --- a/server/test/unit/services/superSimpleQueueHelper.test.ts +++ b/server/test/unit/services/superSimpleQueueHelper.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, jest } from "@jest/globals"; -import { SuperSimpleQueueHelper } from "../../../src/service/infrastructure/SuperSimpleQueue/SuperSimpleQueueHelper.ts"; +import { QueueHelper } from "../../../src/service/infrastructure/JobQueues/QueueHelper.ts"; import type { Monitor } from "../../../src/types/monitor.ts"; import { createMockLogger } from "../../helpers/createMockLogger.ts"; @@ -59,7 +59,7 @@ const createHelper = (overrides?: Record) => { ...overrides, }; - const helper = new SuperSimpleQueueHelper( + const helper = new QueueHelper( defaults.logger as any, defaults.networkService as any, defaults.statusService as any, @@ -94,7 +94,7 @@ const makeMonitor = (overrides?: Partial): Monitor => // ── Tests ──────────────────────────────────────────────────────────────────── -describe("SuperSimpleQueueHelper", () => { +describe("QueueHelper", () => { describe("serviceName", () => { it("returns JobQueueHelper", () => { const { helper } = createHelper();